From 0345726a7931ac8299abd9ab2d638f1c15d944ed Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:09:51 -0400 Subject: [PATCH] fix(rust-client): async CA file read; typed ping/session-state/worker-info/drain-events --- clients/rust/Cargo.toml | 2 +- clients/rust/README.md | 12 +++ clients/rust/src/client.rs | 2 +- clients/rust/src/galaxy.rs | 2 +- clients/rust/src/options.rs | 43 ++++++--- clients/rust/src/session.rs | 99 +++++++++++++++++-- clients/rust/tests/client_behavior.rs | 134 +++++++++++++++++++++++++- 7 files changed, 265 insertions(+), 29 deletions(-) diff --git a/clients/rust/Cargo.toml b/clients/rust/Cargo.toml index f0ee824..5f755d0 100644 --- a/clients/rust/Cargo.toml +++ b/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" diff --git a/clients/rust/README.md b/clients/rust/README.md index 963f915..94af6c2 100644 --- a/clients/rust/README.md +++ b/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`; 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 diff --git a/clients/rust/src/client.rs b/clients/rust/src/client.rs index 0388915..f8672e4 100644 --- a/clients/rust/src/client.rs +++ b/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)?; } diff --git a/clients/rust/src/galaxy.rs b/clients/rust/src/galaxy.rs index 8ae97d7..794d4b2 100644 --- a/clients/rust/src/galaxy.rs +++ b/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)?; } diff --git a/clients/rust/src/options.rs b/clients/rust/src/options.rs index 86d5691..7ffb9fd 100644 --- a/clients/rust/src/options.rs +++ b/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, Error> { +pub(crate) async fn build_tls_config( + options: &ClientOptions, +) -> Result, 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 { 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")); } } diff --git a/clients/rust/src/session.rs b/clients/rust/src/session.rs index 7a7dd02..c0e4c3c 100644 --- a/clients/rust/src/session.rs +++ b/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 { + 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 { + 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, 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`. diff --git a/clients/rust/tests/client_behavior.rs b/clients/rust/tests/client_behavior.rs index 1fbeac8..f00ddf5 100644 --- a/clients/rust/tests/client_behavior.rs +++ b/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,