Add bulk read/write command family across worker, gateway, and clients

Adds five new MXAccess command kinds (WriteBulk, Write2Bulk,
WriteSecuredBulk, WriteSecured2Bulk, ReadBulk) that ride the existing
"one round-trip, per-entry results" bulk shape used by AddItemBulk and
SubscribeBulk today. MXAccess COM has no native bulk API; the worker
runs each bulk operation as a sequential loop on its STA, returning
one BulkWriteResult / BulkReadResult per requested entry so per-item
MXAccess failures surface as was_successful=false rather than throwing.

ReadBulk has no MXAccess analogue. The worker satisfies it by:

  - Returning the last cached OnDataChange payload (was_cached=true)
    when the requested tag is already in the session''s item registry
    AND advised — the existing subscription is NOT touched, since the
    caller did not create it.
  - Otherwise taking the AddItem + Advise + wait-for-OnDataChange +
    UnAdvise + RemoveItem snapshot lifecycle itself (was_cached=false)
    and leaving the session exactly as it was. The wait pumps Windows
    messages on the STA so the inbound MXAccess event can dispatch
    while the executor still holds the thread.

The new MxAccessValueCache lives on each MxAccessSession, shared with
MxAccessBaseEventSink which populates it on every OnDataChange after
the event clears the outbound queue. Eviction on RemoveItem keeps
reused MXAccess handles from serving stale values from a previous
lifetime.

Gateway-side authorization wires WriteBulk/Write2Bulk to invoke:write,
WriteSecuredBulk/WriteSecured2Bulk to invoke:secure, ReadBulk to
invoke:read. The constraint-filter pipeline is refactored from a single
BulkConstraintPlan record into an abstract base plus three concretes
(SubscribeBulk, WriteBulk, ReadBulk), each owning its own denied-entry
merge so the dispatch site never branches on reply shape. A new
FilterWriteBulkAsync<TEntry> generic over the four write-entry shapes
runs CheckWriteHandleAsync per entry; denied entries surface as the
BulkWriteResult shape, preserving original-index order.

All five language clients (.NET, Go, Rust, Python, Java) gained the
five new methods following their existing bulk pattern, with regenerated
protobufs.

Tests added:
  - MxAccessValueCacheTests (6 cases) — Set/TryGet, Remove resets the
    version, TryWaitForUpdate signals on Set, pump step fires each poll.
  - MxAccessBaseEventSinkTests — OnDataChange populates the cache,
    ValueCache property exposes the bound instance.
  - MxAccessCommandExecutorTests — four bulk-write variants (per-entry
    success/failure, value+timestamp forwarding, secured user ids),
    ReadBulk snapshot lifecycle on uncached tag (timeout surfaces as
    was_successful=false), invalid-payload reply.
  - GatewayGrpcScopeResolverTests — five new MxCommandKind cases.
  - SessionManagerTests — WriteBulk and ReadBulk forwarding through
    FakeWorkerHarness; ReadBulk forwards timeout_ms.
  - Per-client (.NET, Go, Rust, Python, Java) — WriteBulk builds the
    right command and returns per-entry results, ReadBulk forwards the
    timeout and unpacks the was_cached flag.

Cross-language e2e CLI subcommands for the new bulks are deliberately
scoped out of this change (each of the five client CLIs would need
five new subcommands plus matching phases in
scripts/run-client-e2e-tests.ps1); coverage equivalent to the existing
bulk-subscribe coverage is provided by worker + gateway + per-client
unit tests.

Docs updated in the same commit: gateway.md (Public MXAccess Command
Surface), docs/DesignDecisions.md (new "Bulk Command Family" section
with the ReadBulk cache-then-snapshot rationale), and every client
README.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

clients/rust/README.md

@@ -99,6 +99,16 @@ preserving the raw message for parity diagnostics. Command replies whose
 protocol status is not `PROTOCOL_STATUS_CODE_OK` become `Error::Command` and
 retain the raw `MxCommandReply`.
 
+The session also exposes the full bulk family —
+`add_item_bulk`, `advise_item_bulk`, `remove_item_bulk`, `un_advise_item_bulk`,
+`subscribe_bulk`, `unsubscribe_bulk`, `write_bulk`, `write2_bulk`,
+`write_secured_bulk`, `write_secured2_bulk`, and `read_bulk`. Each carries a
+`Vec` of entries in one round-trip and returns one result per entry; per-entry
+MXAccess failures populate `was_successful = false` and never raise. `read_bulk`
+takes a per-tag timeout (`u32` milliseconds, `0` = worker default) and returns
+the cached `OnDataChange` value when the tag is already advised (`was_cached =
+true`) without touching the existing subscription.
+
 ## Galaxy Repository browse
 
 The Galaxy Repository service exposes a read-only browse over the AVEVA System

clients/rust/src/session.rs

@@ -16,10 +16,13 @@ use crate::generated::mxaccess_gateway::v1::mx_command::Payload;
 use crate::generated::mxaccess_gateway::v1::mx_command_reply;
 use crate::generated::mxaccess_gateway::v1::{
     AddItem2Command, AddItemBulkCommand, AddItemCommand, AdviseCommand, AdviseItemBulkCommand,
-    CloseSessionRequest, MxCommand, MxCommandKind, MxCommandReply, MxCommandRequest,
-    MxValue as ProtoMxValue, OpenSessionRequest, RegisterCommand, RemoveItemBulkCommand,
-    RemoveItemCommand, StreamEventsRequest, SubscribeBulkCommand, SubscribeResult, UnAdviseCommand,
-    UnAdviseItemBulkCommand, UnsubscribeBulkCommand, Write2Command, WriteCommand,
+    BulkReadResult, BulkWriteResult, CloseSessionRequest, MxCommand, MxCommandKind, MxCommandReply,
+    MxCommandRequest, MxValue as ProtoMxValue, OpenSessionRequest, ReadBulkCommand,
+    RegisterCommand, RemoveItemBulkCommand, RemoveItemCommand, StreamEventsRequest,
+    SubscribeBulkCommand, SubscribeResult, UnAdviseCommand, UnAdviseItemBulkCommand,
+    UnsubscribeBulkCommand, Write2BulkCommand, Write2BulkEntry, Write2Command, WriteBulkCommand,
+    WriteBulkEntry, WriteCommand, WriteSecured2BulkCommand, WriteSecured2BulkEntry,
+    WriteSecuredBulkCommand, WriteSecuredBulkEntry,
 };
 use crate::value::MxValue;
 
@@ -362,6 +365,147 @@ impl Session {
         bulk_results(reply, BulkReplyKind::Unsubscribe)
     }
 
+    /// Bulk `Write` (sequential MXAccess Write per entry, on the worker's STA).
+    ///
+    /// Per-entry MXAccess failures are reported as `BulkWriteResult` entries
+    /// with `was_successful = false`; the call never errors on per-entry
+    /// failure. Protocol-level failures still surface as [`Error::Command`].
+    ///
+    /// # Errors
+    ///
+    /// Same conditions as [`Session::add_item_bulk`], plus the usual
+    /// transport/status errors.
+    pub async fn write_bulk(
+        &self,
+        server_handle: i32,
+        entries: Vec<WriteBulkEntry>,
+    ) -> Result<Vec<BulkWriteResult>, Error> {
+        ensure_bulk_size("entries", entries.len())?;
+        let reply = self
+            .invoke(
+                MxCommandKind::WriteBulk,
+                Payload::WriteBulk(WriteBulkCommand {
+                    server_handle,
+                    entries,
+                }),
+            )
+            .await?;
+
+        bulk_write_results(reply, BulkWriteReplyKind::Write)
+    }
+
+    /// Bulk `Write2` (timestamped) — see [`Session::write_bulk`].
+    ///
+    /// # Errors
+    ///
+    /// Same conditions as [`Session::write_bulk`].
+    pub async fn write2_bulk(
+        &self,
+        server_handle: i32,
+        entries: Vec<Write2BulkEntry>,
+    ) -> Result<Vec<BulkWriteResult>, Error> {
+        ensure_bulk_size("entries", entries.len())?;
+        let reply = self
+            .invoke(
+                MxCommandKind::Write2Bulk,
+                Payload::Write2Bulk(Write2BulkCommand {
+                    server_handle,
+                    entries,
+                }),
+            )
+            .await?;
+
+        bulk_write_results(reply, BulkWriteReplyKind::Write2)
+    }
+
+    /// Bulk `WriteSecured` — credential-sensitive values follow the same
+    /// redaction contract as the single-item `write_secured` path.
+    ///
+    /// # Errors
+    ///
+    /// Same conditions as [`Session::write_bulk`].
+    pub async fn write_secured_bulk(
+        &self,
+        server_handle: i32,
+        entries: Vec<WriteSecuredBulkEntry>,
+    ) -> Result<Vec<BulkWriteResult>, Error> {
+        ensure_bulk_size("entries", entries.len())?;
+        let reply = self
+            .invoke(
+                MxCommandKind::WriteSecuredBulk,
+                Payload::WriteSecuredBulk(WriteSecuredBulkCommand {
+                    server_handle,
+                    entries,
+                }),
+            )
+            .await?;
+
+        bulk_write_results(reply, BulkWriteReplyKind::WriteSecured)
+    }
+
+    /// Bulk `WriteSecured2` (timestamped) — see [`Session::write_secured_bulk`].
+    ///
+    /// # Errors
+    ///
+    /// Same conditions as [`Session::write_bulk`].
+    pub async fn write_secured2_bulk(
+        &self,
+        server_handle: i32,
+        entries: Vec<WriteSecured2BulkEntry>,
+    ) -> Result<Vec<BulkWriteResult>, Error> {
+        ensure_bulk_size("entries", entries.len())?;
+        let reply = self
+            .invoke(
+                MxCommandKind::WriteSecured2Bulk,
+                Payload::WriteSecured2Bulk(WriteSecured2BulkCommand {
+                    server_handle,
+                    entries,
+                }),
+            )
+            .await?;
+
+        bulk_write_results(reply, BulkWriteReplyKind::WriteSecured2)
+    }
+
+    /// Bulk `Read` — snapshot the current value for each requested tag.
+    ///
+    /// MXAccess COM has no synchronous `Read`; the worker satisfies this by
+    /// returning the most recent cached `OnDataChange` value when the tag is
+    /// already advised (`was_cached = true`), or by taking a full AddItem +
+    /// Advise + wait + UnAdvise + RemoveItem snapshot lifecycle otherwise.
+    /// `timeout_ms == 0` lets the worker pick its default (1000 ms).
+    /// Per-tag failures appear as `BulkReadResult` entries with
+    /// `was_successful = false`; the call never errors on per-tag failure.
+    ///
+    /// # Errors
+    ///
+    /// Same conditions as [`Session::add_item_bulk`].
+    pub async fn read_bulk(
+        &self,
+        server_handle: i32,
+        tag_addresses: Vec<String>,
+        timeout_ms: u32,
+    ) -> Result<Vec<BulkReadResult>, Error> {
+        ensure_bulk_size("tag_addresses", tag_addresses.len())?;
+        let reply = self
+            .invoke(
+                MxCommandKind::ReadBulk,
+                Payload::ReadBulk(ReadBulkCommand {
+                    server_handle,
+                    tag_addresses,
+                    timeout_ms,
+                }),
+            )
+            .await?;
+
+        match reply.payload {
+            Some(mx_command_reply::Payload::ReadBulk(reply)) => Ok(reply.results),
+            _ => Err(Error::MalformedReply {
+                detail: "ReadBulk reply did not carry a BulkReadReply payload".to_owned(),
+            }),
+        }
+    }
+
     /// Run MXAccess `Write` (single-value, no caller-supplied timestamp).
     ///
     /// # Errors
@@ -578,6 +722,39 @@ fn bulk_results(reply: MxCommandReply, kind: BulkReplyKind) -> Result<Vec<Subscr
     }
 }
 
+enum BulkWriteReplyKind {
+    Write,
+    Write2,
+    WriteSecured,
+    WriteSecured2,
+}
+
+fn bulk_write_results(
+    reply: MxCommandReply,
+    kind: BulkWriteReplyKind,
+) -> Result<Vec<BulkWriteResult>, Error> {
+    match (reply.payload, kind) {
+        (Some(mx_command_reply::Payload::WriteBulk(reply)), BulkWriteReplyKind::Write) => {
+            Ok(reply.results)
+        }
+        (Some(mx_command_reply::Payload::Write2Bulk(reply)), BulkWriteReplyKind::Write2) => {
+            Ok(reply.results)
+        }
+        (
+            Some(mx_command_reply::Payload::WriteSecuredBulk(reply)),
+            BulkWriteReplyKind::WriteSecured,
+        ) => Ok(reply.results),
+        (
+            Some(mx_command_reply::Payload::WriteSecured2Bulk(reply)),
+            BulkWriteReplyKind::WriteSecured2,
+        ) => Ok(reply.results),
+        _ => Err(Error::MalformedReply {
+            detail: "bulk write reply did not carry the expected BulkWriteReply payload"
+                .to_owned(),
+        }),
+    }
+}
+
 fn int32_reply_value(value: &ProtoMxValue) -> Option<i32> {
     match value.kind.as_ref()? {
         crate::generated::mxaccess_gateway::v1::mx_value::Kind::Int32Value(value) => Some(*value),

clients/rust/tests/client_behavior.rs

@@ -11,14 +11,16 @@ use futures_util::StreamExt;
 use mxgateway_client::generated::mxaccess_gateway::v1::mx_access_gateway_server::{
     MxAccessGateway, MxAccessGatewayServer,
 };
+use mxgateway_client::generated::mxaccess_gateway::v1::mx_command;
 use mxgateway_client::generated::mxaccess_gateway::v1::mx_command_reply;
 use mxgateway_client::generated::mxaccess_gateway::v1::mx_value::Kind;
 use mxgateway_client::generated::mxaccess_gateway::v1::{
     AcknowledgeAlarmReply, AcknowledgeAlarmRequest, ActiveAlarmSnapshot, AddItemReply,
-    BulkSubscribeReply, CloseSessionReply, CloseSessionRequest, MxCommandKind, MxCommandReply,
-    MxDataType, MxEvent, MxEventFamily, MxStatusCategory, MxStatusProxy, MxStatusSource, MxValue,
-    OpenSessionReply, OpenSessionRequest, ProtocolStatus, ProtocolStatusCode,
-    QueryActiveAlarmsRequest, SessionState, StreamEventsRequest, SubscribeResult,
+    BulkReadReply, BulkReadResult, BulkSubscribeReply, BulkWriteReply, BulkWriteResult,
+    CloseSessionReply, CloseSessionRequest, MxCommandKind, MxCommandReply, MxDataType, MxEvent,
+    MxEventFamily, MxStatusCategory, MxStatusProxy, MxStatusSource, MxValue, OpenSessionReply,
+    OpenSessionRequest, ProtocolStatus, ProtocolStatusCode, QueryActiveAlarmsRequest, SessionState,
+    StreamEventsRequest, SubscribeResult, WriteBulkEntry,
 };
 use mxgateway_client::{
     ApiKey, ClientOptions, CommandError, Error, GatewayClient, MxStatus, MxValue as ClientMxValue,
@@ -107,6 +109,61 @@ async fn subscribe_bulk_builds_one_bulk_command_and_returns_results() {
     assert_eq!(*last_command, Some(MxCommandKind::SubscribeBulk as i32));
 }
 
+#[tokio::test]
+async fn write_bulk_builds_one_bulk_command_and_returns_per_entry_results() {
+    let state = Arc::new(FakeState::default());
+    let endpoint = spawn_fake_gateway(state.clone()).await;
+    let client = GatewayClient::connect(ClientOptions::new(endpoint))
+        .await
+        .unwrap();
+    let session = client.session("session-fixture");
+
+    let results = session
+        .write_bulk(
+            12,
+            vec![
+                WriteBulkEntry {
+                    item_handle: 901,
+                    value: Some(int_value(11)),
+                    user_id: 5,
+                },
+                WriteBulkEntry {
+                    item_handle: 902,
+                    value: Some(int_value(22)),
+                    user_id: 5,
+                },
+            ],
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(results.len(), 2);
+    assert!(results[0].was_successful);
+    assert!(!results[1].was_successful);
+    let last_command = state.last_command_kind.lock().await;
+    assert_eq!(*last_command, Some(MxCommandKind::WriteBulk as i32));
+}
+
+#[tokio::test]
+async fn read_bulk_forwards_timeout_and_unpacks_cached_flag() {
+    let state = Arc::new(FakeState::default());
+    let endpoint = spawn_fake_gateway(state.clone()).await;
+    let client = GatewayClient::connect(ClientOptions::new(endpoint))
+        .await
+        .unwrap();
+    let session = client.session("session-fixture");
+
+    let results = session
+        .read_bulk(12, vec!["Area001.Pump001.Speed".to_owned()], 750)
+        .await
+        .unwrap();
+
+    let entry = &results[0];
+    assert!(entry.was_cached);
+    assert_eq!(entry.value.as_ref().and_then(|v| v.kind.as_ref()), Some(&Kind::Int32Value(99)));
+    assert_eq!(*state.last_read_bulk_timeout_ms.lock().await, Some(750));
+}
+
 #[tokio::test]
 async fn event_stream_preserves_order_and_drop_cancels_server_stream() {
     let state = Arc::new(FakeState::default());
@@ -340,6 +397,7 @@ async fn connect_with_unreadable_ca_file_reports_invalid_endpoint() {
 struct FakeState {
     authorization: Mutex<Option<String>>,
     last_command_kind: Mutex<Option<i32>>,
+    last_read_bulk_timeout_ms: Mutex<Option<u32>>,
     stream_dropped: Arc<AtomicBool>,
     emit_stream_fault: AtomicBool,
 }
@@ -420,6 +478,70 @@ impl MxAccessGateway for FakeGateway {
             }));
         }
 
+        if kind == MxCommandKind::WriteBulk as i32 {
+            // Echo one success and one failure so the test can assert the per-entry
+            // shape and verify the call did not throw on per-entry failure.
+            return Ok(Response::new(MxCommandReply {
+                session_id: request.session_id,
+                correlation_id: "fake-correlation".to_owned(),
+                kind,
+                protocol_status: Some(ok_status("command ok")),
+                payload: Some(mx_command_reply::Payload::WriteBulk(BulkWriteReply {
+                    results: vec![
+                        BulkWriteResult {
+                            server_handle: 12,
+                            item_handle: 901,
+                            was_successful: true,
+                            hresult: None,
+                            statuses: vec![],
+                            error_message: String::new(),
+                        },
+                        BulkWriteResult {
+                            server_handle: 12,
+                            item_handle: 902,
+                            was_successful: false,
+                            hresult: None,
+                            statuses: vec![],
+                            error_message: "invalid handle".to_owned(),
+                        },
+                    ],
+                })),
+                ..MxCommandReply::default()
+            }));
+        }
+
+        if kind == MxCommandKind::ReadBulk as i32 {
+            let read_command = request.command.as_ref().and_then(|c| {
+                if let Some(mx_command::Payload::ReadBulk(ref r)) = c.payload {
+                    Some(r.timeout_ms)
+                } else {
+                    None
+                }
+            });
+            *self.state.last_read_bulk_timeout_ms.lock().await = read_command;
+            return Ok(Response::new(MxCommandReply {
+                session_id: request.session_id,
+                correlation_id: "fake-correlation".to_owned(),
+                kind,
+                protocol_status: Some(ok_status("command ok")),
+                payload: Some(mx_command_reply::Payload::ReadBulk(BulkReadReply {
+                    results: vec![BulkReadResult {
+                        server_handle: 12,
+                        tag_address: "Area001.Pump001.Speed".to_owned(),
+                        item_handle: 34,
+                        was_successful: true,
+                        was_cached: true,
+                        value: Some(int_value(99)),
+                        quality: 192,
+                        source_timestamp: None,
+                        statuses: vec![],
+                        error_message: String::new(),
+                    }],
+                })),
+                ..MxCommandReply::default()
+            }));
+        }
+
         Ok(Response::new(MxCommandReply {
             session_id: request.session_id,
             correlation_id: "fake-correlation".to_owned(),
@@ -527,6 +649,15 @@ async fn spawn_fake_gateway(state: Arc<FakeState>) -> String {
     format!("http://{address}")
 }
 
+fn int_value(value: i32) -> MxValue {
+    MxValue {
+        data_type: MxDataType::Integer as i32,
+        variant_type: "VT_I4".to_owned(),
+        kind: Some(Kind::Int32Value(value)),
+        ..MxValue::default()
+    }
+}
+
 fn ok_status(message: &str) -> ProtocolStatus {
     ProtocolStatus {
         code: ProtocolStatusCode::Ok as i32,