diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OpcUaApplicationHost.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OpcUaApplicationHost.cs index e71ee3e5..13e19a15 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OpcUaApplicationHost.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OpcUaApplicationHost.cs @@ -69,8 +69,10 @@ public sealed class OpcUaApplicationHostOptions /// Note this controls admission, not authority. An admitted anonymous session is a /// normal principal (anonymous) whose grants come from NodeAcl rows naming it, /// exactly like a key's principal — there is no anonymous bypass branch in the evaluator. - /// Until per-node enforcement lands, an admitted session of ANY kind can still read the whole - /// address space, because reads carry no authorization check yet. + /// When a wired data-plane authorizer is enabled, per-node ACL checks apply to anonymous + /// sessions, including reads. Reads denied by the node manager return BadUserAccessDenied. + /// With the authorizer disabled or not wired, reads retain the legacy behaviour without + /// per-node ACL enforcement. /// /// public bool AllowAnonymous { get; set; } = true; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs index 05db8ecd..52e0a57c 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -147,6 +147,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers private readonly TimeSpan _staleWindow; private readonly TimeSpan _probeFreshnessWindow; private readonly TimeSpan _healthTickInterval; + private readonly Func _utcNow; private readonly Akka.Cluster.Cluster _cluster = Akka.Cluster.Cluster.Get(Context.System); private readonly ILoggingAdapter _log = Context.GetLogger(); @@ -200,9 +201,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// The window beyond which a peer's OPC UA probe verdict about /// this node is considered stale (and thus given the benefit of the doubt rather than demoting); /// defaults to 30 seconds. - /// The period between self-driven DB-health refresh ticks (each - /// Asks for its cached status); defaults to 5 seconds. No timer is - /// started when is null. + /// The period between freshness ticks; defaults to 5 seconds. + /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness. + /// Legacy nodes without a probe do not start a timer. + /// The UTC clock used for freshness; defaults to . /// Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the /// ServiceLevel DB-reachability input is retired (DbReachable treated constant-true, only /// staleness demotes). No is wired on such a node. @@ -219,7 +221,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers TimeSpan? probeFreshnessWindow = null, TimeSpan? healthTickInterval = null, bool dbLess = false, - IDataPlaneAuthoritySink? dataPlaneAuthority = null) => + IDataPlaneAuthoritySink? dataPlaneAuthority = null, + Func? utcNow = null) => Akka.Actor.Props.Create(() => new OpcUaPublishActor( sink ?? NullOpcUaAddressSpaceSink.Instance, serviceLevel ?? NullServiceLevelPublisher.Instance, @@ -232,7 +235,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers probeFreshnessWindow, healthTickInterval, dbLess, - dataPlaneAuthority)).WithDispatcher(DispatcherId); + dataPlaneAuthority, + utcNow)).WithDispatcher(DispatcherId); /// Test-only Props that omits the pinned-dispatcher requirement and skips the /// DPS subscribe so unit tests can spin up the actor on a vanilla TestKit cluster. @@ -249,9 +253,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// The window beyond which a peer's OPC UA probe verdict about /// this node is considered stale (and thus given the benefit of the doubt rather than demoting); /// defaults to 30 seconds. - /// The period between self-driven DB-health refresh ticks (each - /// Asks for its cached status); defaults to 5 seconds. No timer is - /// started when is null. + /// The period between freshness ticks; defaults to 5 seconds. + /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness. + /// Legacy nodes without a probe do not start a timer. + /// The UTC clock used for freshness; defaults to . /// Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the /// ServiceLevel DB-reachability input is retired (DbReachable treated constant-true, only /// staleness demotes). No is wired on such a node. @@ -269,7 +274,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers TimeSpan? probeFreshnessWindow = null, TimeSpan? healthTickInterval = null, bool dbLess = false, - IDataPlaneAuthoritySink? dataPlaneAuthority = null) => + IDataPlaneAuthoritySink? dataPlaneAuthority = null, + Func? utcNow = null) => Akka.Actor.Props.Create(() => new OpcUaPublishActor( sink ?? NullOpcUaAddressSpaceSink.Instance, serviceLevel ?? NullServiceLevelPublisher.Instance, @@ -282,7 +288,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers probeFreshnessWindow, healthTickInterval, dbLess, - dataPlaneAuthority)); + dataPlaneAuthority, + utcNow)); /// Initializes a new instance of the class. /// The OPC UA address space sink. @@ -298,9 +305,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// The window beyond which a peer's OPC UA probe verdict about /// this node is considered stale (and thus given the benefit of the doubt rather than demoting); /// defaults to 30 seconds. - /// The period between self-driven DB-health refresh ticks (each - /// Asks for its cached status); defaults to 5 seconds. No timer is - /// started when is null. + /// The period between freshness ticks; defaults to 5 seconds. + /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness. + /// Legacy nodes without a probe do not start a timer. + /// The UTC clock used for freshness; defaults to . /// Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the /// ServiceLevel DB-reachability input is retired (DbReachable treated constant-true, only /// staleness demotes). No is wired on such a node. @@ -316,7 +324,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers TimeSpan? probeFreshnessWindow = null, TimeSpan? healthTickInterval = null, bool dbLess = false, - IDataPlaneAuthoritySink? dataPlaneAuthority = null) + IDataPlaneAuthoritySink? dataPlaneAuthority = null, + Func? utcNow = null) { _sink = sink; _dataPlaneAuthority = dataPlaneAuthority; @@ -330,6 +339,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers _staleWindow = staleWindow ?? TimeSpan.FromSeconds(30); _probeFreshnessWindow = probeFreshnessWindow ?? TimeSpan.FromSeconds(30); _healthTickInterval = healthTickInterval ?? TimeSpan.FromSeconds(5); + _utcNow = utcNow ?? (() => DateTime.UtcNow); Receive(HandleAttributeUpdate); Receive(HandleAttributeBatchUpdate); @@ -353,11 +363,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(RedundancyStateTopic, Self)); } - // Production delivery of DB-health: when a probe is wired, kick an immediate refresh + start the - // periodic tick so _lastDbHealth gets populated and kept fresh without any external pump. Gated - // ONLY on the probe being present (independent of the DPS subscribe) so the calculator path works - // in tests too. No probe → stay on the legacy role-only seam (no timer). - if (_dbHealthProbe is not null) + // DB-backed nodes refresh their cached probe status; DB-less nodes still need periodic + // snapshot/probe-verdict age checks even when no new messages arrive. Legacy nodes keep + // their role-only behavior without a timer. + if (_dbLess || _dbHealthProbe is not null) { Self.Tell(HealthTick.Instance); // immediate first refresh Timers.StartPeriodicTimer("db-health", HealthTick.Instance, _healthTickInterval); @@ -1006,7 +1015,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers RecomputeServiceLevel(); } - /// Periodic self-tick: Asks the local for its cached + /// Periodic self-tick: recomputes freshness directly when no DB probe exists; otherwise + /// Asks the local for its cached /// status and pipes the reply back to , where /// caches it + recomputes. A hung/late probe (Ask timeout) pipes a Reachable=false status so /// the node FAIL-SAFE-DEMOTES rather than freezing the last-known-good. The continuation only @@ -1015,11 +1025,17 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// touched here. private void OnHealthTick() { - _dbHealthProbe!.Ask( + if (_dbHealthProbe is null) + { + RecomputeServiceLevel(); + return; + } + + _dbHealthProbe.Ask( DbHealthProbeActor.GetStatus.Instance, TimeSpan.FromSeconds(1)) .ContinueWith(t => t.IsCompletedSuccessfully ? t.Result - : new DbHealthProbeActor.DbHealthStatus(false, DateTime.UtcNow, "db-health ask timeout")) + : new DbHealthProbeActor.DbHealthStatus(false, _utcNow(), "db-health ask timeout")) .PipeTo(Self); } @@ -1032,7 +1048,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers { // The result targets the probed node. If it isn't me, it's about a peer — ignore it. if (_localNode is null || r.NodeId != _localNode.Value) return; - _probeAboutMe = (r.Ok, DateTime.UtcNow); + _probeAboutMe = (r.Ok, _utcNow()); RecomputeServiceLevel(); } @@ -1044,7 +1060,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers private bool OpcUaProbeOk() { if (_probeAboutMe is not { } verdict) return true; - if (DateTime.UtcNow - verdict.At > _probeFreshnessWindow) return true; + if (_utcNow() - verdict.At > _probeFreshnessWindow) return true; return verdict.Ok; } @@ -1081,7 +1097,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers // wins, so a DB-less Detached node stays 0. if (_dbLess) { - var utcNow = DateTime.UtcNow; + var utcNow = _utcNow(); var dbLessInputs = new NodeHealthInputs( MemberState: SafeSelfStatus(), DbReachable: true, @@ -1101,7 +1117,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers return; } - var now = DateTime.UtcNow; + var now = _utcNow(); var inputs = new NodeHealthInputs( MemberState: SafeSelfStatus(), DbReachable: _lastDbHealth.Reachable, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorFreshnessTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorFreshnessTests.cs new file mode 100644 index 00000000..70dde5d2 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorFreshnessTests.cs @@ -0,0 +1,240 @@ +using System.Collections.Concurrent; +using Akka.Actor; +using Akka.TestKit; +using Akka.TestKit.Xunit2; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Health; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa; + +public sealed class OpcUaPublishActorFreshnessTests : RuntimeActorTestBase +{ + [Theory] + [InlineData(false, 240, 200)] + [InlineData(true, 250, 210)] + public void DbLess_timer_demotes_without_incoming_messages(bool primary, byte fresh, byte stale) + { + var publisher = new RecordingPublisher(); + var clock = new ControlledClock(); + var local = NodeId.Parse("freshness-node"); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: local, dbLess: true, + staleWindow: TimeSpan.FromSeconds(1), healthTickInterval: TimeSpan.FromMilliseconds(25), + utcNow: clock.Read)); + actor.Tell(Snapshot(local, primary, clock.Now)); + AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh }), PresenceBudget); + // Equality remains fresh. Observe several real ticks with a fixed clock; no actor Tell. + clock.Advance(TimeSpan.FromSeconds(1)); + var reads = clock.Reads; + AwaitCondition(() => clock.Reads >= reads + 4, PresenceBudget); + // Drain earlier self-enqueued publications without triggering a health recomputation. + actor.Tell(new Identify("freshness-barrier"), TestActor); + ExpectMsg().Subject.ShouldBe(actor); + publisher.Levels.ShouldBe(new[] { fresh }); + clock.Advance(TimeSpan.FromTicks(1)); + // No further Tell: only the production timer can notice age crossing the window. + AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh, stale }), PresenceBudget); + // A new fresh snapshot restores service, after which ticks continue to age that snapshot. + actor.Tell(Snapshot(local, primary, clock.Now)); + AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh, stale, fresh }), PresenceBudget); + clock.Advance(TimeSpan.FromSeconds(1) + TimeSpan.FromTicks(1)); + AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh, stale, fresh, stale }), PresenceBudget); + } + + [Fact] + public void DbLess_timer_expires_negative_peer_verdict_only_after_exact_boundary() + { + var publisher = new RecordingPublisher(); + var clock = new ControlledClock(); + var local = NodeId.Parse("peer-verdict-node"); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: local, dbLess: true, + staleWindow: TimeSpan.FromSeconds(10), probeFreshnessWindow: TimeSpan.FromSeconds(2), + healthTickInterval: TimeSpan.FromMilliseconds(25), utcNow: clock.Read)); + actor.Tell(Snapshot(local, false, clock.Now)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240 }), PresenceBudget); + actor.Tell(new PeerOpcUaProbeActor.OpcUaProbeResult(local, Ok: false)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0 }), PresenceBudget); + clock.Advance(TimeSpan.FromSeconds(2)); + var reads = clock.Reads; + AwaitCondition(() => clock.Reads >= reads + 4, PresenceBudget); + // Drain earlier self-enqueued publications without triggering a health recomputation. + actor.Tell(new Identify("freshness-barrier"), TestActor); + ExpectMsg().Subject.ShouldBe(actor); + publisher.Levels.ShouldBe(new byte[] { 240, 0 }); + clock.Advance(TimeSpan.FromTicks(1)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0, 240 }), PresenceBudget); + } + + [Fact] + public void Stopping_DbLess_actor_cancels_freshness_timer_and_publications() + { + var publisher = new RecordingPublisher(); + var clock = new ControlledClock(); + var local = NodeId.Parse("stopped-node"); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: local, dbLess: true, + staleWindow: TimeSpan.FromSeconds(1), healthTickInterval: TimeSpan.FromMilliseconds(25), + utcNow: clock.Read)); + actor.Tell(Snapshot(local, false, clock.Now)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240 }), PresenceBudget); + var reads = clock.Reads; + AwaitCondition(() => clock.Reads >= reads + 4, PresenceBudget); + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor); + reads = clock.Reads; + clock.Advance(TimeSpan.FromSeconds(2)); + ExpectNoMsg(TimeSpan.FromMilliseconds(150)); // Six timer periods after confirmed termination. + clock.Reads.ShouldBe(reads); + publisher.Levels.ShouldBe(new byte[] { 240 }); + } + + [Fact] + public void DbBacked_timer_refreshes_probe_status_periodically() + { + var publisher = new RecordingPublisher(); + var clock = new ControlledClock(); + var local = NodeId.Parse("db-backed-periodic-node"); + var db = CreateTestProbe(); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: local, dbHealthProbe: db.Ref, + healthTickInterval: TimeSpan.FromMilliseconds(100), utcNow: clock.Read)); + actor.Tell(Snapshot(local, false, clock.Now)); + db.ExpectMsg(); + db.Reply(new DbHealthProbeActor.DbHealthStatus(true, clock.Now, null)); + AwaitAssert(() => publisher.Levels.ShouldContain((byte)240), PresenceBudget); + // A second request comes from the periodic timer, not the immediate startup tick. + db.ExpectMsg(); + db.Reply(new DbHealthProbeActor.DbHealthStatus(false, clock.Now, "unreachable")); + AwaitAssert(() => publisher.Levels.Last().ShouldBe((byte)100), PresenceBudget); + } + + [Fact] + public void DbBacked_timer_ask_timeout_demotes_last_known_healthy_node() + { + var publisher = new RecordingPublisher(); + var clock = new ControlledClock(); + var local = NodeId.Parse("db-backed-timeout-node"); + var db = CreateTestProbe(); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: local, dbHealthProbe: db.Ref, + healthTickInterval: TimeSpan.FromMilliseconds(100), utcNow: clock.Read)); + actor.Tell(Snapshot(local, true, clock.Now)); + db.ExpectMsg(); + db.Reply(new DbHealthProbeActor.DbHealthStatus(true, clock.Now, null)); + AwaitAssert(() => publisher.Levels.ShouldContain((byte)250), PresenceBudget); + // Leave subsequent real Ask requests unanswered: the timeout status must reach the actor. + db.ExpectMsg(); + AwaitAssert(() => publisher.Levels.Last().ShouldBe((byte)110), PresenceBudget); + } + + [Fact] + public void Missing_local_snapshot_entry_retries_zero_on_timer_and_recovers() + { + var publisher = new RetryZeroPublisher(); + var clock = new ControlledClock(); + var local = NodeId.Parse("missing-entry-node"); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: local, dbLess: true, + healthTickInterval: TimeSpan.FromMilliseconds(25), utcNow: clock.Read)); + actor.Tell(Snapshot(local, false, clock.Now)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240 }), PresenceBudget); + + // Only this snapshot Tell requests the first zero. Fail it and the next attempt so + // subsequent zero attempts must come from the periodic timer, exercising the null-entry + // branch repeatedly rather than mistaking the Detached branch for missing-node coverage. + actor.Tell(Snapshot(NodeId.Parse("other-node"), false, clock.Now)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0 }), PresenceBudget); + publisher.ZeroAttempts.ShouldBeGreaterThanOrEqualTo(3); + + actor.Tell(Snapshot(local, false, clock.Now)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0, 240 }), PresenceBudget); + clock.Advance(TimeSpan.FromSeconds(30) + TimeSpan.FromTicks(1)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0, 240, 200 }), PresenceBudget); + } + + [Fact] + public void No_snapshot_startup_does_not_publish_across_virtual_timer_periods() + { + // This actor requires the Cluster extension, but the absent/missing-entry guards do + // not inspect membership. Do not self-join: the ordinary harness waits for Up in its + // constructor, which cannot progress with a frozen scheduler. + using var local = new TestKit(AkkaTestHocon + @" + akka.scheduler.implementation = ""Akka.TestKit.TestScheduler, Akka.TestKit"""); + var scheduler = (TestScheduler)local.Sys.Scheduler; + var publisher = new RetryZeroPublisher(); + var clock = new ControlledClock(); + var actor = local.ActorOfAsTestActorRef(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: NodeId.Parse("no-snapshot-node"), dbLess: true, + healthTickInterval: TimeSpan.FromSeconds(1), utcNow: clock.Read)); + + // TestActorRef uses the calling-thread dispatcher: Advance completes delivery before + // these assertions, including the immediate startup tick and three periodic ticks. + for (var period = 0; period < 3; period++) + scheduler.Advance(TimeSpan.FromSeconds(1)); + publisher.Levels.ShouldBeEmpty(); + publisher.ZeroAttempts.ShouldBe(0); + clock.Reads.ShouldBe(0); + + // Positive control: the same registered timer must subsequently retry a failed zero. + // No whole snapshot at startup means no publication; a received snapshot missing the + // local entry means zero. Keep these two policies explicitly distinct. + actor.Tell(Snapshot(NodeId.Parse("other-node"), false, clock.Now)); + publisher.ZeroAttempts.ShouldBe(1); + scheduler.Advance(TimeSpan.FromSeconds(1)); + publisher.ZeroAttempts.ShouldBe(2); + publisher.Levels.ShouldBeEmpty(); + scheduler.Advance(TimeSpan.FromSeconds(1)); + publisher.ZeroAttempts.ShouldBe(3); + publisher.Levels.ShouldBe(new byte[] { 0 }); + clock.Reads.ShouldBe(0); + } + + private sealed class RetryZeroPublisher : IServiceLevelPublisher + { + private readonly ConcurrentQueue _levels = new(); + private int _zeroAttempts; + public byte[] Levels => _levels.ToArray(); + public int ZeroAttempts => Volatile.Read(ref _zeroAttempts); + public void Publish(byte serviceLevel) + { + if (serviceLevel == 0 && Interlocked.Increment(ref _zeroAttempts) <= 2) + throw new InvalidOperationException("test zero-publication failure"); + _levels.Enqueue(serviceLevel); + } + } + + private sealed class ControlledClock + { + private long _ticks = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc).Ticks; + private int _reads; + public DateTime Now => new(Interlocked.Read(ref _ticks), DateTimeKind.Utc); + public int Reads => Volatile.Read(ref _reads); + public DateTime Read() + { + Interlocked.Increment(ref _reads); + return Now; + } + public void Advance(TimeSpan age) => Interlocked.Add(ref _ticks, age.Ticks); + } + + private static RedundancyStateChanged Snapshot(NodeId local, bool primary, DateTime asOf) => + new(new[] { new NodeRedundancyState(local, + primary ? RedundancyRole.Primary : RedundancyRole.Secondary, + IsClusterLeader: primary, IsDriverPrimary: primary, asOf) }, CorrelationId.NewId()); + + private sealed class RecordingPublisher : IServiceLevelPublisher + { + private readonly ConcurrentQueue _levels = new(); + public byte[] Levels => _levels.ToArray(); + public void Publish(byte serviceLevel) => _levels.Enqueue(serviceLevel); + } +}