From dd623356f453bd7334ef69879535294849826fd9 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 08:46:29 -0400 Subject: [PATCH] docs(clients): ReadBulk timeout is one shared deadline for the command, not per tag Stream E made the worker wait once for all uncached tags. Fix the Go, Python, Rust, Java and .NET library docs and CLI help text, the proto comment (default is the worker's 1000 ms), and the worker design doc. Regenerated Contracts Generated/ and the Java generated protos; both diffs are comment-only. --- .../MxGatewaySession.cs | 2 +- clients/go/cmd/mxgw-go/main.go | 4 ++-- clients/go/mxgateway/session.go | 5 ++-- .../mxaccess_gateway/v1/MxaccessGateway.java | 24 ++++++++++++------- .../zb/mom/ww/mxgateway/cli/MxGatewayCli.java | 4 ++-- .../ww/mxgateway/client/MxGatewaySession.java | 10 ++++---- .../python/src/zb_mom_ww_mxgateway/session.py | 6 +++-- .../src/zb_mom_ww_mxgateway_cli/commands.py | 2 +- clients/rust/crates/mxgw-cli/src/main.rs | 2 +- clients/rust/src/session.rs | 5 +++- docs/MxAccessWorkerInstanceDesign.md | 2 +- .../Generated/MxaccessGateway.cs | 12 ++++++---- .../Protos/mxaccess_gateway.proto | 12 ++++++---- 13 files changed, 57 insertions(+), 33 deletions(-) diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs index 1aae534..1576a9d 100644 --- a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs @@ -641,7 +641,7 @@ public sealed class MxGatewaySession : IAsyncDisposable /// /// The ServerHandle from register. /// Tag addresses to read (one per result). - /// Per-call timeout for the snapshot lifecycle path; uses the gateway default. + /// One shared snapshot deadline for the whole command, not a per-tag budget: the worker advises every uncached tag, waits once for all of them, then tears them down. uses the worker default (1000 ms). /// Cancellation token. /// One per requested tag, in request order. public async Task> ReadBulkAsync( diff --git a/clients/go/cmd/mxgw-go/main.go b/clients/go/cmd/mxgw-go/main.go index f973d99..90e1d38 100644 --- a/clients/go/cmd/mxgw-go/main.go +++ b/clients/go/cmd/mxgw-go/main.go @@ -571,7 +571,7 @@ func runReadBulk(ctx context.Context, args []string, stdout, stderr io.Writer) e sessionID := flags.String("session-id", "", "gateway session id") serverHandle := flags.Int("server-handle", 0, "MXAccess server handle") items := flags.String("items", "", "comma-separated tag addresses") - timeoutMs := flags.Int("timeout-ms", 0, "per-tag snapshot timeout in milliseconds (0 = worker default)") + timeoutMs := flags.Int("timeout-ms", 0, "snapshot deadline in milliseconds, shared by all tags in the command (0 = worker default, 1000 ms)") if err := flags.Parse(args); err != nil { return err @@ -754,7 +754,7 @@ func runBenchReadBulk(ctx context.Context, args []string, stdout, stderr io.Writ tagStart := flags.Int("tag-start", 1, "first machine number") tagPrefix := flags.String("tag-prefix", "TestMachine_", "tag prefix (machine number appended as %03d)") tagAttribute := flags.String("tag-attribute", "TestChangingInt", "attribute appended to each tag prefix") - timeoutMs := flags.Int("timeout-ms", 1500, "per-tag snapshot timeout in milliseconds") + timeoutMs := flags.Int("timeout-ms", 1500, "snapshot deadline in milliseconds, shared by all tags in each ReadBulk command") if err := flags.Parse(args); err != nil { return err diff --git a/clients/go/mxgateway/session.go b/clients/go/mxgateway/session.go index f2233f3..639a38c 100644 --- a/clients/go/mxgateway/session.go +++ b/clients/go/mxgateway/session.go @@ -549,8 +549,9 @@ func (s *Session) WriteSecured2Bulk(ctx context.Context, serverHandle int32, ent // MXAccess COM has no synchronous Read; the worker satisfies this by returning the // most recent cached OnDataChange value when the tag is already advised (WasCached=true), // or by taking a full AddItem + Advise + wait + UnAdvise + RemoveItem snapshot lifecycle -// otherwise. timeout bounds the wait per tag in the snapshot case; pass zero to use the -// worker default. Per-tag failures (timeout, invalid tag) appear as BulkReadResult entries +// otherwise. timeout is one shared deadline for the whole command, not a per-tag budget: +// the worker advises every uncached tag, waits once for all of them, then tears them down. +// Pass zero to use the worker default (1000 ms). Per-tag failures (timeout, invalid tag) appear as BulkReadResult entries // with WasSuccessful=false; the call never returns an error for per-tag MXAccess failures. // // A non-nil but empty tagAddresses slice is treated as a no-op and returns an empty diff --git a/clients/java/src/main/generated/main/java/mxaccess_gateway/v1/MxaccessGateway.java b/clients/java/src/main/generated/main/java/mxaccess_gateway/v1/MxaccessGateway.java index 1ebe2c7..ffbd891 100644 --- a/clients/java/src/main/generated/main/java/mxaccess_gateway/v1/MxaccessGateway.java +++ b/clients/java/src/main/generated/main/java/mxaccess_gateway/v1/MxaccessGateway.java @@ -49125,11 +49125,15 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile { * reply returns the cached value WITHOUT modifying the existing * subscription (was_cached = true). * - Otherwise the worker takes the snapshot lifecycle itself: AddItem + - * Advise, wait up to `timeout_ms` for the first OnDataChange, then - * UnAdvise + RemoveItem before returning. The session is left exactly - * as it was before the call (was_cached = false). + * Advise, wait for the first OnDataChange, then UnAdvise + RemoveItem + * before returning. The session is left exactly as it was before the + * call (was_cached = false). * - * `timeout_ms == 0` uses the gateway-configured default (1000 ms). + * `timeout_ms` is one deadline shared by the whole command, not a per-tag + * budget: the worker adds and advises every uncached tag, waits once for all + * of them, then tears them all down. A tag that has not delivered by the + * deadline gets its own was_successful = false result. + * `timeout_ms == 0` uses the worker's default (1000 ms). * * * Protobuf type {@code mxaccess_gateway.v1.ReadBulkCommand} @@ -49424,11 +49428,15 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile { * reply returns the cached value WITHOUT modifying the existing * subscription (was_cached = true). * - Otherwise the worker takes the snapshot lifecycle itself: AddItem + - * Advise, wait up to `timeout_ms` for the first OnDataChange, then - * UnAdvise + RemoveItem before returning. The session is left exactly - * as it was before the call (was_cached = false). + * Advise, wait for the first OnDataChange, then UnAdvise + RemoveItem + * before returning. The session is left exactly as it was before the + * call (was_cached = false). * - * `timeout_ms == 0` uses the gateway-configured default (1000 ms). + * `timeout_ms` is one deadline shared by the whole command, not a per-tag + * budget: the worker adds and advises every uncached tag, waits once for all + * of them, then tears them all down. A tag that has not delivered by the + * deadline gets its own was_successful = false result. + * `timeout_ms == 0` uses the worker's default (1000 ms). * * * Protobuf type {@code mxaccess_gateway.v1.ReadBulkCommand} diff --git a/clients/java/zb-mom-ww-mxgateway-cli/src/main/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCli.java b/clients/java/zb-mom-ww-mxgateway-cli/src/main/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCli.java index 6965356..a590730 100644 --- a/clients/java/zb-mom-ww-mxgateway-cli/src/main/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCli.java +++ b/clients/java/zb-mom-ww-mxgateway-cli/src/main/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCli.java @@ -1262,7 +1262,7 @@ public final class MxGatewayCli implements Callable { @Option( names = "--timeout-ms", defaultValue = "0", - description = "Per-tag snapshot timeout in milliseconds (0 = worker default).") + description = "Snapshot deadline in milliseconds, shared by all tags in the command (0 = worker default, 1000 ms).") int timeoutMs; ReadBulkCommand(MxGatewayCliClientFactory clientFactory) { @@ -1510,7 +1510,7 @@ public final class MxGatewayCli implements Callable { @Option( names = "--timeout-ms", defaultValue = "0", - description = "Per-tag snapshot timeout in milliseconds (0 = worker default).") + description = "Snapshot deadline in milliseconds, shared by all tags in the command (0 = worker default, 1000 ms).") int timeoutMs; @Option(names = "--iterations", defaultValue = "10", description = "Number of ReadBulk calls to perform.") diff --git a/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewaySession.java b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewaySession.java index e0c803e..320088b 100644 --- a/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewaySession.java +++ b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewaySession.java @@ -555,15 +555,17 @@ public final class MxGatewaySession implements AutoCloseable { * {@code OnDataChange} value for any tag that is already advised * ({@code wasCached == true}) without modifying the existing subscription, * and falls back to a full AddItem + Advise + wait + UnAdvise + RemoveItem - * snapshot lifecycle otherwise. The supplied {@code timeout} bounds the - * per-tag wait in the snapshot case; pass {@link Duration#ZERO} (or - * {@code null}) to use the worker default (1000 ms). Per-tag failures + * snapshot lifecycle otherwise. The supplied {@code timeout} is one shared + * deadline for the whole command, not a per-tag budget: the worker advises + * every uncached tag, waits once for all of them, then tears them down. + * Pass {@link Duration#ZERO} (or {@code null}) to use the worker default + * (1000 ms). Per-tag failures * appear as {@link BulkReadResult} entries with {@code wasSuccessful == false}; * this method does not throw for per-tag MXAccess failures. * * @param serverHandle the {@code ServerHandle} owning the items * @param tagAddresses the tag addresses to read - * @param timeout per-tag snapshot timeout (zero or null = worker default) + * @param timeout snapshot deadline shared by all tags in the command (zero or null = worker default) * @return a per-tag {@link BulkReadResult} list * @throws MxGatewayException on transport or protocol failure * @throws NullPointerException if {@code tagAddresses} is {@code null} diff --git a/clients/python/src/zb_mom_ww_mxgateway/session.py b/clients/python/src/zb_mom_ww_mxgateway/session.py index 8ff3f9f..24ba73f 100644 --- a/clients/python/src/zb_mom_ww_mxgateway/session.py +++ b/clients/python/src/zb_mom_ww_mxgateway/session.py @@ -448,8 +448,10 @@ class Session: ``OnDataChange`` value for any tag that is already advised (``was_cached = True``) without modifying the existing subscription, and falls back to a full AddItem + Advise + wait + UnAdvise + RemoveItem snapshot lifecycle - otherwise. ``timeout_ms`` bounds the per-tag wait in the snapshot case; - pass ``0`` to use the worker default (1000 ms). + otherwise. ``timeout_ms`` is one shared deadline for the whole command, + not a per-tag budget: the worker advises every uncached tag, waits once + for all of them, then tears them down. Pass ``0`` to use the worker + default (1000 ms). """ if tag_addresses is None: raise TypeError("tag_addresses is required") diff --git a/clients/python/src/zb_mom_ww_mxgateway_cli/commands.py b/clients/python/src/zb_mom_ww_mxgateway_cli/commands.py index e3aae6a..fa05282 100644 --- a/clients/python/src/zb_mom_ww_mxgateway_cli/commands.py +++ b/clients/python/src/zb_mom_ww_mxgateway_cli/commands.py @@ -392,7 +392,7 @@ def unsubscribe_bulk(**kwargs: Any) -> None: @click.option("--server-handle", required=True, type=int, help="MXAccess server handle.") @click.option("--items", required=True, help="Comma-separated MXAccess tag addresses.") @click.option("--timeout-ms", default=0, type=int, show_default=True, - help="Per-tag snapshot timeout in milliseconds. 0 = worker default.") + help="Snapshot deadline in milliseconds, shared by all tags in the command. 0 = worker default (1000 ms).") @click.option("--correlation-id", default="", help="Client correlation id.") @click.option("--json", "output_json", is_flag=True, help="Emit JSON output.") def read_bulk(**kwargs: Any) -> None: diff --git a/clients/rust/crates/mxgw-cli/src/main.rs b/clients/rust/crates/mxgw-cli/src/main.rs index 0dace53..9bb1e55 100644 --- a/clients/rust/crates/mxgw-cli/src/main.rs +++ b/clients/rust/crates/mxgw-cli/src/main.rs @@ -215,7 +215,7 @@ enum Command { server_handle: i32, #[arg(long, value_delimiter = ',')] items: Vec, - /// Per-tag snapshot timeout in milliseconds. `0` uses the worker default (1000 ms). + /// Snapshot deadline in milliseconds, shared by all tags in the command. `0` uses the worker default (1000 ms). #[arg(long, default_value_t = 0)] timeout_ms: u32, #[arg(long)] diff --git a/clients/rust/src/session.rs b/clients/rust/src/session.rs index c0e4c3c..cc7862e 100644 --- a/clients/rust/src/session.rs +++ b/clients/rust/src/session.rs @@ -533,7 +533,10 @@ impl Session { /// 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). + /// `timeout_ms` is one shared deadline for the whole command, not a + /// per-tag budget: the worker advises every uncached tag, waits once for + /// all of them, then tears them down. `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. /// diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index b829364..d53543b 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -996,7 +996,7 @@ has been stale for longer than `HeartbeatStuckCeiling`, the watchdog fires healthy long-running commands: `StaRuntime.PumpPendingMessages()` refreshes `LastStaActivityUtc` (via `MarkActivity()`) every time it runs, and long-hold STA commands invoke it on every wait iteration (`ReadBulk` routes its -per-tag wait through the `pumpStep` wired from `StaRuntime.PumpPendingMessages`). +shared snapshot wait through the `pumpStep` wired from `StaRuntime.PumpPendingMessages`). A command that keeps pumping therefore keeps its activity timestamp fresh and never reaches the ceiling, while a genuinely stuck STA — one that has stopped pumping — accrues staleness and faults correctly. The ceiling is thus the diff --git a/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessGateway.cs b/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessGateway.cs index 0bbd29f..2c81411 100644 --- a/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessGateway.cs +++ b/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessGateway.cs @@ -16205,11 +16205,15 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto { /// reply returns the cached value WITHOUT modifying the existing /// subscription (was_cached = true). /// - Otherwise the worker takes the snapshot lifecycle itself: AddItem + - /// Advise, wait up to `timeout_ms` for the first OnDataChange, then - /// UnAdvise + RemoveItem before returning. The session is left exactly - /// as it was before the call (was_cached = false). + /// Advise, wait for the first OnDataChange, then UnAdvise + RemoveItem + /// before returning. The session is left exactly as it was before the + /// call (was_cached = false). /// - /// `timeout_ms == 0` uses the gateway-configured default (1000 ms). + /// `timeout_ms` is one deadline shared by the whole command, not a per-tag + /// budget: the worker adds and advises every uncached tag, waits once for all + /// of them, then tears them all down. A tag that has not delivered by the + /// deadline gets its own was_successful = false result. + /// `timeout_ms == 0` uses the worker's default (1000 ms). /// [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")] public sealed partial class ReadBulkCommand : pb::IMessage diff --git a/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_gateway.proto b/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_gateway.proto index edfdca6..223eb88 100644 --- a/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_gateway.proto +++ b/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_gateway.proto @@ -524,11 +524,15 @@ message WriteSecured2BulkEntry { // reply returns the cached value WITHOUT modifying the existing // subscription (was_cached = true). // - Otherwise the worker takes the snapshot lifecycle itself: AddItem + -// Advise, wait up to `timeout_ms` for the first OnDataChange, then -// UnAdvise + RemoveItem before returning. The session is left exactly -// as it was before the call (was_cached = false). +// Advise, wait for the first OnDataChange, then UnAdvise + RemoveItem +// before returning. The session is left exactly as it was before the +// call (was_cached = false). // -// `timeout_ms == 0` uses the gateway-configured default (1000 ms). +// `timeout_ms` is one deadline shared by the whole command, not a per-tag +// budget: the worker adds and advises every uncached tag, waits once for all +// of them, then tears them all down. A tag that has not delivered by the +// deadline gets its own was_successful = false result. +// `timeout_ms == 0` uses the worker's default (1000 ms). message ReadBulkCommand { int32 server_handle = 1; repeated string tag_addresses = 2;