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;