Merge stream I (Rust client: async CA read, typed diagnostics)

clients/rust/Cargo.toml

@@ -40,7 +40,7 @@ prost-types = "0.13.5"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.145"
 thiserror = "2.0.17"
-tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "sync", "time"] }
+tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "sync", "time", "fs"] }
 tokio-stream = { version = "0.1.17", features = ["net"] }
 tonic = { version = "0.13.1", features = ["transport", "tls-ring", "tls-native-roots"] }
 tonic-build = "0.13.1"

clients/rust/README.md

@@ -178,6 +178,18 @@ redaction. Per-item bulk failures are reported inside each result entry
 `invoke_raw` / `client.invoke_raw` escape hatch performs neither check and
 returns the unvalidated reply.
 
+### Diagnostic commands
+
+`Session` wraps the worker diagnostic commands as typed calls: `ping(message)`
+(returns `()` on success), `get_session_state()` (returns `SessionState`),
+`get_worker_info()` (returns `WorkerInfoReply`: pid, version, MXAccess
+ProgID/CLSID) and `drain_events(max_events)` (returns `Vec<MxEvent>`; the reply
+is bounded server-side, so drain until an empty result). A reply missing its
+expected payload surfaces as `Error::MalformedReply`.
+
+TLS note: `ClientOptions::with_ca_file` is read asynchronously (`tokio::fs`)
+during `connect`, so connecting no longer blocks the executor thread.
+
 ## Event Streaming And Reconnect-Replay Gaps
 
 `session.events()` / `session.events_after(after_worker_sequence)` (and the

clients/rust/src/client.rs

@@ -187,7 +187,7 @@ impl GatewayClient {
             })?;
         endpoint = endpoint.connect_timeout(options.connect_timeout());
 
-        if let Some(tls) = build_tls_config(&options)? {
+        if let Some(tls) = build_tls_config(&options).await? {
             endpoint = endpoint.tls_config(tls)?;
         }
 

clients/rust/src/galaxy.rs

@@ -182,7 +182,7 @@ impl GalaxyClient {
             })?;
         endpoint = endpoint.connect_timeout(options.connect_timeout());
 
-        if let Some(tls) = build_tls_config(&options)? {
+        if let Some(tls) = build_tls_config(&options).await? {
             endpoint = endpoint.tls_config(tls)?;
         }
 

clients/rust/src/options.rs

@@ -3,7 +3,6 @@
 //! chain of `with_*` setters; the `Debug` impl redacts the API key.
 
 use std::fmt;
-use std::fs;
 use std::path::PathBuf;
 use std::time::Duration;
 
@@ -235,7 +234,9 @@ pub(crate) fn tls_trust_decision(options: &ClientOptions) -> TlsTrustDecision {
 /// rejects the configuration early with an actionable error. The strict opt-in
 /// instead loads the system trust roots so a certificate chaining to an
 /// already-trusted root validates.
-pub(crate) fn build_tls_config(options: &ClientOptions) -> Result<Option<ClientTlsConfig>, Error> {
+pub(crate) async fn build_tls_config(
+    options: &ClientOptions,
+) -> Result<Option<ClientTlsConfig>, Error> {
     let decision = tls_trust_decision(options);
     if decision == TlsTrustDecision::None {
         return Ok(None);
@@ -248,10 +249,13 @@ pub(crate) fn build_tls_config(options: &ClientOptions) -> Result<Option<ClientT
     match decision {
         TlsTrustDecision::PinnedCa => {
             let ca_file = options.ca_file().expect("PinnedCa implies a CA file");
-            let certificate = fs::read(ca_file).map_err(|source| Error::InvalidEndpoint {
-                endpoint: options.endpoint().to_owned(),
-                detail: format!("failed to read CA file {}: {source}", ca_file.display()),
-            })?;
+            let certificate =
+                tokio::fs::read(ca_file)
+                    .await
+                    .map_err(|source| Error::InvalidEndpoint {
+                        endpoint: options.endpoint().to_owned(),
+                        detail: format!("failed to read CA file {}: {source}", ca_file.display()),
+                    })?;
             tls = tls.ca_certificate(Certificate::from_pem(certificate));
         }
         TlsTrustDecision::SystemRoots => {
@@ -334,11 +338,11 @@ mod tests {
         assert!(!debug.contains("mxgw_secret"));
     }
 
-    #[test]
-    fn plaintext_needs_no_tls() {
+    #[tokio::test]
+    async fn plaintext_needs_no_tls() {
         let options = ClientOptions::new("http://127.0.0.1:5000").with_plaintext(true);
         assert_eq!(tls_trust_decision(&options), TlsTrustDecision::None);
-        assert!(build_tls_config(&options).unwrap().is_none());
+        assert!(build_tls_config(&options).await.unwrap().is_none());
     }
 
     #[test]
@@ -349,8 +353,8 @@ mod tests {
         assert_eq!(tls_trust_decision(&options), TlsTrustDecision::PinnedCa);
     }
 
-    #[test]
-    fn strict_without_ca_uses_system_roots() {
+    #[tokio::test]
+    async fn strict_without_ca_uses_system_roots() {
         // Regression for Client.Rust-031: strict verification with no pinned CA
         // must verify against the system trust roots, not produce a config with
         // zero trust anchors. The trust decision proves roots are consulted; the
@@ -365,15 +369,24 @@ mod tests {
             "strict-no-CA must request the system trust roots"
         );
         assert!(
-            build_tls_config(&options).unwrap().is_some(),
+            build_tls_config(&options).await.unwrap().is_some(),
             "strict-no-CA must build a usable TLS config"
         );
     }
 
-    #[test]
-    fn lenient_without_ca_is_rejected() {
+    #[tokio::test]
+    async fn lenient_without_ca_is_rejected() {
         let options = ClientOptions::new("https://127.0.0.1:5000").with_plaintext(false);
         assert_eq!(tls_trust_decision(&options), TlsTrustDecision::RejectNoCa);
-        assert!(build_tls_config(&options).is_err());
+        assert!(build_tls_config(&options).await.is_err());
+    }
+
+    #[tokio::test]
+    async fn missing_pinned_ca_file_errors_without_blocking_read() {
+        let options = ClientOptions::new("https://127.0.0.1:5000")
+            .with_plaintext(false)
+            .with_ca_file("/nonexistent/definitely-missing-ca.pem");
+        let error = build_tls_config(&options).await.unwrap_err();
+        assert!(error.to_string().contains("failed to read CA file"));
     }
 }

clients/rust/src/session.rs

@@ -17,14 +17,15 @@ use crate::generated::mxaccess_gateway::v1::mx_command_reply;
 use crate::generated::mxaccess_gateway::v1::{
     ActivateCommand, AddBufferedItemCommand, AddItem2Command, AddItemBulkCommand, AddItemCommand,
     AdviseCommand, AdviseItemBulkCommand, AdviseSupervisoryCommand, ArchestrAUserToIdCommand,
-    AuthenticateUserCommand, BulkReadResult, BulkWriteResult, CloseSessionRequest, MxCommand,
-    MxCommandKind, MxCommandReply, MxCommandRequest, MxDataType, MxSparseArray, MxSparseElement,
-    MxValue as ProtoMxValue, OpenSessionRequest, ReadBulkCommand, RegisterCommand,
-    RemoveItemBulkCommand, RemoveItemCommand, SetBufferedUpdateIntervalCommand,
+    AuthenticateUserCommand, BulkReadResult, BulkWriteResult, CloseSessionRequest,
+    DrainEventsCommand, GetSessionStateCommand, GetWorkerInfoCommand, MxCommand, MxCommandKind,
+    MxCommandReply, MxCommandRequest, MxDataType, MxEvent, MxSparseArray, MxSparseElement,
+    MxValue as ProtoMxValue, OpenSessionRequest, PingCommand, ReadBulkCommand, RegisterCommand,
+    RemoveItemBulkCommand, RemoveItemCommand, SessionState, SetBufferedUpdateIntervalCommand,
     StreamEventsRequest, SubscribeBulkCommand, SubscribeResult, SuspendCommand, UnAdviseCommand,
-    UnAdviseItemBulkCommand, UnregisterCommand, UnsubscribeBulkCommand, Write2BulkCommand,
-    Write2BulkEntry, Write2Command, WriteBulkCommand, WriteBulkEntry, WriteCommand,
-    WriteSecured2BulkCommand, WriteSecured2BulkEntry, WriteSecured2Command,
+    UnAdviseItemBulkCommand, UnregisterCommand, UnsubscribeBulkCommand, WorkerInfoReply,
+    Write2BulkCommand, Write2BulkEntry, Write2Command, WriteBulkCommand, WriteBulkEntry,
+    WriteCommand, WriteSecured2BulkCommand, WriteSecured2BulkEntry, WriteSecured2Command,
     WriteSecuredBulkCommand, WriteSecuredBulkEntry, WriteSecuredCommand,
 };
 use crate::value::{MxStatus, MxValue, MxValueProjection};
@@ -960,6 +961,90 @@ impl Session {
             .await
     }
 
+    /// Send a `Ping` diagnostic command to the worker. A successful return
+    /// means the worker answered; `message` is forwarded for log correlation.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`Error::Command`] for non-OK statuses plus transport errors.
+    pub async fn ping(&self, message: &str) -> Result<(), Error> {
+        self.invoke(
+            MxCommandKind::Ping,
+            Payload::Ping(PingCommand {
+                message: message.to_owned(),
+            }),
+        )
+        .await?;
+        Ok(())
+    }
+
+    /// Fetch the gateway-side [`SessionState`] of this session.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`Error::MalformedReply`] when the reply lacks a session-state
+    /// payload, plus the errors of [`invoke`](Self::invoke).
+    pub async fn get_session_state(&self) -> Result<SessionState, Error> {
+        let reply = self
+            .invoke(
+                MxCommandKind::GetSessionState,
+                Payload::GetSessionState(GetSessionStateCommand {}),
+            )
+            .await?;
+        match reply.payload {
+            Some(mx_command_reply::Payload::SessionState(state)) => {
+                Ok(SessionState::try_from(state.state).unwrap_or(SessionState::Unspecified))
+            }
+            _ => Err(Error::MalformedReply {
+                detail: "get_session_state reply did not carry a SessionState payload".to_owned(),
+            }),
+        }
+    }
+
+    /// Fetch worker process details (pid, version, MXAccess ProgID/CLSID).
+    ///
+    /// # Errors
+    ///
+    /// Returns [`Error::MalformedReply`] when the reply lacks a worker-info
+    /// payload, plus the errors of [`invoke`](Self::invoke).
+    pub async fn get_worker_info(&self) -> Result<WorkerInfoReply, Error> {
+        let reply = self
+            .invoke(
+                MxCommandKind::GetWorkerInfo,
+                Payload::GetWorkerInfo(GetWorkerInfoCommand {}),
+            )
+            .await?;
+        match reply.payload {
+            Some(mx_command_reply::Payload::WorkerInfo(info)) => Ok(info),
+            _ => Err(Error::MalformedReply {
+                detail: "get_worker_info reply did not carry a WorkerInfo payload".to_owned(),
+            }),
+        }
+    }
+
+    /// Drain up to `max_events` queued events through the diagnostic command
+    /// (distinct from the streaming [`events`](Self::events) surface). The
+    /// reply is bounded server-side, so call repeatedly until it is empty.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`Error::MalformedReply`] when the reply lacks a drain-events
+    /// payload, plus the errors of [`invoke`](Self::invoke).
+    pub async fn drain_events(&self, max_events: u32) -> Result<Vec<MxEvent>, Error> {
+        let reply = self
+            .invoke(
+                MxCommandKind::DrainEvents,
+                Payload::DrainEvents(DrainEventsCommand { max_events }),
+            )
+            .await?;
+        match reply.payload {
+            Some(mx_command_reply::Payload::DrainEvents(drain)) => Ok(drain.events),
+            _ => Err(Error::MalformedReply {
+                detail: "drain_events reply did not carry a DrainEvents payload".to_owned(),
+            }),
+        }
+    }
+
     /// Issue a raw `Invoke` for an arbitrary command, without filtering on
     /// the protocol status. Useful when callers need the full reply for
     /// commands not yet wrapped by `Session`.

clients/rust/tests/client_behavior.rs

@@ -26,12 +26,12 @@ use zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::{
     AddItem2Reply, AddItemReply, AlarmConditionState, AlarmFeedMessage, AlarmTransitionKind,
     AuthenticateUserCommand, AuthenticateUserReply, BulkReadReply, BulkReadResult,
     BulkSubscribeReply, BulkWriteReply, BulkWriteResult, CloseSessionReply, CloseSessionRequest,
-    MxCommandKind, MxCommandReply, MxDataType, MxEvent, MxEventFamily, MxSparseArray,
-    MxSparseElement, MxStatusCategory, MxStatusProxy, MxStatusSource, MxValue,
+    DrainEventsReply, MxCommandKind, MxCommandReply, MxDataType, MxEvent, MxEventFamily,
+    MxSparseArray, MxSparseElement, MxStatusCategory, MxStatusProxy, MxStatusSource, MxValue,
     OnAlarmTransitionEvent, OpenSessionReply, OpenSessionRequest, ProtocolStatus,
     ProtocolStatusCode, QueryActiveAlarmsRequest, RegisterReply, ReplayGap, SessionState,
-    StreamAlarmsRequest, StreamEventsRequest, SubscribeResult, Write2BulkEntry, WriteBulkEntry,
-    WriteCommand, WriteSecured2BulkEntry, WriteSecuredBulkEntry,
+    SessionStateReply, StreamAlarmsRequest, StreamEventsRequest, SubscribeResult, WorkerInfoReply,
+    Write2BulkEntry, WriteBulkEntry, WriteCommand, WriteSecured2BulkEntry, WriteSecuredBulkEntry,
 };
 use zb_mom_ww_mxgateway_client::{
     next_correlation_id, ApiKey, ClientOptions, CommandError, Error, EventItem, GatewayClient,
@@ -550,6 +550,81 @@ async fn unary_invoke_maps_status_unavailable_to_error_unavailable() {
     );
 }
 
+#[tokio::test]
+async fn ping_round_trips_through_the_fake_gateway() {
+    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");
+
+    session.ping("hello").await.unwrap();
+
+    assert_eq!(
+        *state.last_command_kind.lock().await,
+        Some(MxCommandKind::Ping as i32)
+    );
+}
+
+#[tokio::test]
+async fn get_session_state_returns_typed_state() {
+    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 session_state = session.get_session_state().await.unwrap();
+
+    assert_eq!(session_state, SessionState::Ready);
+}
+
+#[tokio::test]
+async fn get_worker_info_returns_typed_reply() {
+    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 info = session.get_worker_info().await.unwrap();
+
+    assert_eq!(info.worker_process_id, 4242);
+    assert_eq!(info.worker_version, "1.2.3");
+}
+
+#[tokio::test]
+async fn drain_events_returns_events() {
+    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 events = session.drain_events(10).await.unwrap();
+
+    assert_eq!(events.len(), 1);
+}
+
+#[tokio::test]
+async fn get_session_state_malformed_without_payload() {
+    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");
+    *state.invoke_override.lock().await = Some(InvokeOverride::OkWithoutPayload);
+
+    let error = session.get_session_state().await.unwrap_err();
+
+    assert!(matches!(error, Error::MalformedReply { .. }));
+}
+
 #[tokio::test]
 async fn read_bulk_round_trips_through_the_fake_gateway() {
     let state = Arc::new(FakeState::default());
@@ -1337,6 +1412,57 @@ impl MxAccessGateway for FakeGateway {
             }));
         }
 
+        if kind == MxCommandKind::Ping as i32 {
+            return Ok(Response::new(MxCommandReply {
+                session_id: request.session_id,
+                correlation_id: "fake-correlation".to_owned(),
+                kind,
+                protocol_status: Some(ok_status("command ok")),
+                ..MxCommandReply::default()
+            }));
+        }
+
+        if kind == MxCommandKind::GetSessionState as i32 {
+            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::SessionState(SessionStateReply {
+                    state: SessionState::Ready as i32,
+                })),
+                ..MxCommandReply::default()
+            }));
+        }
+
+        if kind == MxCommandKind::GetWorkerInfo as i32 {
+            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::WorkerInfo(WorkerInfoReply {
+                    worker_process_id: 4242,
+                    worker_version: "1.2.3".to_owned(),
+                    ..WorkerInfoReply::default()
+                })),
+                ..MxCommandReply::default()
+            }));
+        }
+
+        if kind == MxCommandKind::DrainEvents as i32 {
+            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::DrainEvents(DrainEventsReply {
+                    events: vec![MxEvent::default()],
+                })),
+                ..MxCommandReply::default()
+            }));
+        }
+
         if kind == MxCommandKind::ReadBulk as i32 {
             return Ok(Response::new(MxCommandReply {
                 session_id: request.session_id,