From 648f173b1894fbda3be328cb40b833bb6187fb47 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 22 Jul 2026 19:39:26 -0400 Subject: [PATCH] =?UTF-8?q?feat(mesh):=20FetchAndCache=20apply=20path=20?= =?UTF-8?q?=E2=80=94=20fetch->cache->apply-from-bytes,=20null=3Dapply-fail?= =?UTF-8?q?ure?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 Task 5. Under ConfigSource:Mode=FetchAndCache, a DispatchDeployment whose revision differs from _currentRevision kicks off a gRPC artifact fetch that PipeTo's its result back to Self as FetchedForApply (never a blocking await in a receive — that would freeze the mailbox for the fetch deadline). HandleFetchedForApply then applies synchronously on the actor thread, mirroring ApplyAndAck: null bytes = apply FAILURE (keep last-known-good, do not advance the revision, ack Failed — #485); non-null bytes reconcile drivers, rebuild the address space and push subscriptions FROM the bytes in hand (OpcUaPublishActor never reads central SQL), then cache them. A faulted fetch task maps to null bytes so it can't escape as an unhandled message. Extracted ReconcileDriversFromBlob from ReconcileDrivers so Direct-read, FetchAndCache and boot-from-cache share one reconcile body (the #485 empty guard moves into it); ApplyCachedArtifact now reuses it instead of duplicating the spawn-plan logic. DI: DriverHostActor.Props gains fetchAndCacheMode + artifactFetcher (LAST, per the positional-forwarding warning); Runtime resolves them from IOptions (fetcher only in FetchAndCache mode); Host registers GrpcDeploymentArtifactFetcher under hasDriver. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs | 7 + .../Drivers/DriverHostActor.cs | 175 ++++++++++++++++-- .../ServiceCollectionExtensions.cs | 14 +- .../DriverHostActorFetchAndCacheTests.cs | 153 +++++++++++++++ 4 files changed, 335 insertions(+), 14 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 5dee7c3b..74fb072b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -30,6 +30,7 @@ using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache; using ZB.MOM.WW.OtOpcUa.Runtime.Historian; using ZB.MOM.WW.OtOpcUa.Runtime.Scripting; using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security; @@ -186,6 +187,12 @@ if (hasDriver) // (initiator) / LocalDb:SyncListenPort (listener) are set. See LocalDbRegistration. builder.Services.AddOtOpcUaLocalDb(builder.Configuration); + // Node-side gRPC artifact fetcher (per-cluster mesh Phase 3). Resolved + used by DriverHostActor + // only under ConfigSource:Mode = FetchAndCache; idle (never invoked) in the default Direct mode, so + // registering it unconditionally on driver nodes is harmless. IDisposable — DI disposes its cached + // h2c channels at shutdown. + builder.Services.AddSingleton(); + // gRPC server plumbing (AddGrpc + the two path-scoped auth interceptors) is registered once, // below, for hasDriver || hasAdmin — the LocalDb passive sync endpoint (driver) and the // ConfigServe artifact endpoint (admin) share one pipeline. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 7a842744..14b89ca9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -72,6 +72,19 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// private readonly IDeploymentArtifactCache? _deploymentArtifactCache; + /// + /// Per-cluster mesh Phase 3. When (ConfigSource:Mode = + /// FetchAndCache) this node obtains its config by fetching the artifact bytes from central + /// over gRPC and applying them from the LocalDb cache, reading NO config from central SQL. + /// When (default Direct) every path reads central SQL as before. + /// + private readonly bool _fetchAndCacheMode; + + /// + /// Node-side gRPC artifact fetcher, non-null only under . + /// + private readonly IDeploymentArtifactFetcher? _artifactFetcher; + /// /// True once this node has booted a cached artifact because central SQL was unreachable. /// @@ -382,7 +395,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers Func? driverMemberCountProvider = null, IDeploymentArtifactCache? deploymentArtifactCache = null, IRedundancyRoleView? redundancyRoleView = null, - string? replicationPeerHost = null) => + string? replicationPeerHost = null, + bool fetchAndCacheMode = false, + IDeploymentArtifactFetcher? artifactFetcher = null) => // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an // expression tree. Six IActorRef? parameters and several interface-typed ones mean a // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds @@ -392,7 +407,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers dbFactory, localNode, ackRouter, driverFactory, localRoles, dependencyMux, opcUaPublishActor, healthPublisher, virtualTagEvaluator, historyWriter, virtualTagHostOverride, loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, - deploymentArtifactCache, redundancyRoleView, replicationPeerHost)); + deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -446,11 +461,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers Func? driverMemberCountProvider = null, IDeploymentArtifactCache? deploymentArtifactCache = null, IRedundancyRoleView? redundancyRoleView = null, - string? replicationPeerHost = null) + string? replicationPeerHost = null, + bool fetchAndCacheMode = false, + IDeploymentArtifactFetcher? artifactFetcher = null) { _deploymentArtifactCache = deploymentArtifactCache; _redundancyRoleView = redundancyRoleView; _replicationPeerHost = replicationPeerHost; + _fetchAndCacheMode = fetchAndCacheMode; + _artifactFetcher = artifactFetcher; _dbFactory = dbFactory; _localNode = localNode; _ackRouter = ackRouter; @@ -726,16 +745,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers { var correlation = CorrelationId.NewId(); - var specs = DeploymentArtifact.ParseDriverInstances(blob, _localNode.Value); - var snapshots = _children.ToDictionary( - kv => kv.Key, - kv => new DriverChildSnapshot(kv.Value.DriverType, kv.Value.LastConfigJson, kv.Value.ResilienceConfig), - StringComparer.Ordinal); - var plan = DriverSpawnPlanner.Compute(snapshots, specs); - - foreach (var id in plan.ToStop) StopChild(id); - foreach (var spec in plan.ToApplyDelta) ApplyChildDelta(spec); - foreach (var spec in plan.ToSpawn) SpawnChild(spec); + ReconcileDriversFromBlob(deploymentId, blob); // Hand the cached blob to the rebuild so it materialises the client-facing address space from // it directly. Passing only the deploymentId would drive a ConfigDb read — unreachable here by @@ -781,6 +791,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers _localNode, msg.DeploymentId, _applyingDeploymentId); Self.Forward(msg); // re-deliver after we transition back }); + // The FetchAndCache fetch pipes its result back here (BeginFetchAndCacheApply ran in Steady, + // then Become(Applying)); completing the apply transitions back to Steady. + Receive(HandleFetchedForApply); Receive(HandleGetDiagnostics); Receive(ForwardToMux); Receive(ForwardNativeAlarm); @@ -1615,9 +1628,132 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers SendAck(msg.DeploymentId, ApplyAckOutcome.Applied, failureReason: null, msg.CorrelationId); return; } + + if (_fetchAndCacheMode) + { + BeginFetchAndCacheApply(msg.DeploymentId, msg.RevisionHash, msg.CorrelationId); + return; + } + ApplyAndAck(msg.DeploymentId, msg.RevisionHash, msg.CorrelationId); } + /// + /// Per-cluster mesh Phase 3 (FetchAndCache). Kicks off a gRPC artifact fetch and pipes + /// the result back to as a — the fetch is I/O + /// and MUST NOT block the mailbox for the fetch deadline, so it never runs as an awaited call + /// inside a receive. The apply itself completes synchronously on the actor thread in + /// , mirroring . + /// + private void BeginFetchAndCacheApply(DeploymentId deploymentId, RevisionHash revision, CorrelationId correlation) + { + _applyingDeploymentId = deploymentId; + Become(Applying); + + // Persist Applying row (idempotent on PK) — same crash-boundary marker as ApplyAndAck. + UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Applying, failureReason: null); + + if (_artifactFetcher is null) + { + // Misconfiguration: FetchAndCache with no fetcher wired. Fail the apply rather than fetch + // nothing forever; the node keeps serving last-known-good. + const string reason = "FetchAndCache mode is set but no artifact fetcher is configured on this node"; + UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Failed, reason); + SendAck(deploymentId, ApplyAckOutcome.Failed, reason, correlation); + _log.Error("DriverHost {Node}: cannot apply {Id} — {Reason}", _localNode, deploymentId, reason); + _applyingDeploymentId = null; + Become(Steady); + return; + } + + _log.Debug("DriverHost {Node}: fetching artifact for {Id} (rev {Rev}) from central over gRPC", + _localNode, deploymentId, revision); + + // A faulted fetch task maps to null bytes — the #485 apply-failure path, exactly like an + // all-endpoints-down fetch — so a fault can never escape as an unhandled actor message. + _artifactFetcher.FetchAsync(deploymentId.ToString(), revision.Value, CancellationToken.None) + .PipeTo( + Self, + success: bytes => new FetchedForApply(deploymentId, revision, correlation, bytes), + failure: _ => new FetchedForApply(deploymentId, revision, correlation, Bytes: null)); + } + + /// + /// Completes a FetchAndCache apply once the fetched bytes are in hand. Null bytes are an + /// apply FAILURE (keep last-known-good, do not advance the revision — #485); non-null bytes are + /// applied from hand (no ConfigDb read) and cached. + /// + private void HandleFetchedForApply(FetchedForApply msg) + { + // Ignore a result for anything but the in-flight apply (a stale/duplicate pipe-back). + if (_applyingDeploymentId is null || msg.DeploymentId != _applyingDeploymentId.Value) + { + _log.Debug("DriverHost {Node}: dropping stale fetch result for {Id} (in-flight={Cur})", + _localNode, msg.DeploymentId, _applyingDeploymentId); + return; + } + + using var span = OtOpcUaTelemetry.StartDeployApplySpan(msg.DeploymentId.ToString()); + span?.SetTag("otopcua.node_id", _localNode.ToString()); + span?.SetTag("otopcua.revision", msg.Revision.ToString()); + span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString()); + var sw = Stopwatch.StartNew(); + + try + { + // #485: null bytes = "no answer" (every endpoint unreachable / NotFound / SHA mismatch). + // Fail the apply and stay on the current revision so a re-dispatch actually retries; keep + // serving the last-known-good address space, drivers and subscriptions throughout. + var appliedBlob = ReconcileDriversFromBlob(msg.DeploymentId, msg.Bytes); + if (appliedBlob is null) + { + const string reason = + "the deployment artifact could not be fetched from central (all endpoints unreachable, NotFound, or SHA-256 mismatch); nothing was applied"; + UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Failed, reason); + SendAck(msg.DeploymentId, ApplyAckOutcome.Failed, reason, msg.Correlation); + OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair("outcome", "reject")); + span?.SetStatus(ActivityStatusCode.Error, reason); + _log.Error( + "DriverHost {Node}: FetchAndCache apply of {Id} FAILED — {Reason}. Staying on revision {Rev} and continuing to serve it; re-dispatch once central is reachable", + _localNode, msg.DeploymentId, reason, _currentRevision); + return; + } + + _currentRevision = msg.Revision; + UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Applied, failureReason: null); + SendAck(msg.DeploymentId, ApplyAckOutcome.Applied, failureReason: null, msg.Correlation); + // Rebuild the address space + push subscriptions FROM the fetched bytes so OpcUaPublishActor + // never reads central SQL (the whole point of FetchAndCache). + _opcUaPublishActor?.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace( + msg.Correlation, msg.DeploymentId, appliedBlob)); + PushDesiredSubscriptionsFromArtifact(msg.DeploymentId, appliedBlob); + // Cache the fetched bytes as this node's last-known-good (idempotent store; skips empty). + CacheAppliedArtifact(msg.DeploymentId, msg.Revision, appliedBlob); + _isRunningFromCache = false; + OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair("outcome", "ack")); + _log.Info("DriverHost {Node}: applied fetched deployment {Id} (rev {Rev}, children={Count})", + _localNode, msg.DeploymentId, msg.Revision, _children.Count); + } + catch (Exception ex) + { + UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Failed, ex.Message); + SendAck(msg.DeploymentId, ApplyAckOutcome.Failed, ex.Message, msg.Correlation); + OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair("outcome", "reject")); + span?.SetStatus(ActivityStatusCode.Error, ex.Message); + _log.Error(ex, "DriverHost {Node}: FetchAndCache apply of {Id} failed", _localNode, msg.DeploymentId); + } + finally + { + OtOpcUaTelemetry.DeploymentApplyDurationSec.Record(sw.Elapsed.TotalSeconds); + _applyingDeploymentId = null; + Become(Steady); + } + } + + /// Pipe-back carrier for a FetchAndCache fetch result (null bytes = no answer). + private sealed record FetchedForApply( + DeploymentId DeploymentId, RevisionHash Revision, CorrelationId Correlation, byte[]? Bytes); + private void ApplyAndAck(DeploymentId deploymentId, RevisionHash revision, CorrelationId correlation) { _applyingDeploymentId = deploymentId; @@ -1732,6 +1868,19 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers return null; } + return ReconcileDriversFromBlob(deploymentId, blob); + } + + /// + /// The blob-processing half of , split out so the + /// FetchAndCache apply path (which already holds the fetched bytes) and the + /// boot-from-cache path can reconcile without a ConfigDb read. + /// + /// The reconciled blob, or when it was empty (#485). + private byte[]? ReconcileDriversFromBlob(DeploymentId deploymentId, byte[]? blob) + { + blob ??= Array.Empty(); + // Issue #485 — no bytes is no answer, not "a configuration with no drivers". Parsing zero bytes // yields zero specs, and DriverSpawnPlanner then plans EVERY running child for StopChild: a missing // deployment row (or a half-written one) silently takes this node's entire field I/O down while the diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index 771e9ae0..e374671a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -282,6 +282,16 @@ public static class ServiceCollectionExtensions // harnesses) rather than given a null-object, so DriverHostActor skips caching outright // instead of pretending to cache into a sink that drops everything. var deploymentArtifactCache = resolver.GetService(); + // Per-cluster mesh Phase 3: ConfigSource:Mode selects where this driver reads config. + // FetchAndCache pulls the artifact from central over gRPC and reads only the LocalDb cache; + // Direct (default) reads central SQL as before. The fetcher is resolved only in FetchAndCache + // mode (registered by the Host under hasDriver); its absence there is a misconfiguration the + // actor fails the apply on rather than fetching nothing forever. + var configSourceOptions = + resolver.GetService>()?.Value ?? new ConfigSourceOptions(); + var fetchAndCacheMode = string.Equals( + configSourceOptions.Mode, ConfigSourceOptions.ModeFetchAndCache, StringComparison.OrdinalIgnoreCase); + var artifactFetcher = fetchAndCacheMode ? resolver.GetService() : null; // Where this actor publishes its Primary-gate verdict for the alarm store-and-forward // drain, which runs on a timer and so cannot read RedundancyStateChanged itself. // Registered by AddAlarmHistorian; absent when no durable sink is configured, in which @@ -455,7 +465,9 @@ public static class ServiceCollectionExtensions invokerFactory: invokerFactory, deploymentArtifactCache: deploymentArtifactCache, redundancyRoleView: redundancyRoleView, - replicationPeerHost: replicationPeerHost), + replicationPeerHost: replicationPeerHost, + fetchAndCacheMode: fetchAndCacheMode, + artifactFetcher: artifactFetcher), DriverHostActorName); registry.Register(driverHost); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheTests.cs new file mode 100644 index 00000000..1bcaea99 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheTests.cs @@ -0,0 +1,153 @@ +using System.Text.Json; +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Per-cluster mesh Phase 3 (Task 5) — the FetchAndCache apply path: on dispatch the node +/// fetches the artifact bytes over gRPC, applies them from bytes in hand, and caches them — +/// reading NO config from central SQL. A null fetch is an apply failure that keeps last-known-good +/// and does not advance the revision (#485). +/// +public sealed class DriverHostActorFetchAndCacheTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + + /// Builds a minimal but parseable artifact (no drivers) — the fetched-bytes stand-in. + private static byte[] ArtifactBytes() => + JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty() }); + + [Fact] + public void FetchReturningGoodBytes_AppliesAndCaches_WithoutReadingCentralSql() + { + // No Deployment row is seeded: if the apply path read ArtifactBlob from SQL it would get empty + // bytes and FAIL (#485). Reaching Applied proves the bytes came from the fetcher, not SQL. + var db = NewInMemoryDbFactory(); + var cache = new RecordingArtifactCache(); + var fetcher = new RecordingFetcher(ArtifactBytes()); + var deploymentId = DeploymentId.NewId(); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: cache, + fetchAndCacheMode: true, + artifactFetcher: fetcher)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + AwaitAssert(() => cache.Stores.Count.ShouldBe(1), duration: TimeSpan.FromSeconds(3)); + cache.Stores[0].RevisionHash.ShouldBe(RevA.Value); + fetcher.Calls.ShouldBe(1); + } + + [Fact] + public void FetchReturningNull_FailsTheApply_KeepsRevisionUnchanged_AndCachesNothing() + { + var db = NewInMemoryDbFactory(); + var cache = new RecordingArtifactCache(); + var fetcher = new RecordingFetcher(bytes: null); // all endpoints down / NotFound / SHA mismatch + var deploymentId = DeploymentId.NewId(); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: cache, + fetchAndCacheMode: true, + artifactFetcher: fetcher)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Failed); + + Thread.Sleep(300); + cache.Stores.ShouldBeEmpty(); + + // The revision was NOT advanced: a re-dispatch of the SAME revision fetches AGAIN rather than + // short-circuiting as "already current". (Two fetch attempts prove #485's keep-last-known-good.) + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Failed); + fetcher.Calls.ShouldBe(2); + } + + [Fact] + public void RedispatchOfTheAppliedRevision_DoesNotFetchAgain() + { + var db = NewInMemoryDbFactory(); + var cache = new RecordingArtifactCache(); + var fetcher = new RecordingFetcher(ArtifactBytes()); + var deploymentId = DeploymentId.NewId(); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: cache, + fetchAndCacheMode: true, + artifactFetcher: fetcher)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + + // Re-dispatch the now-current revision: the _currentRevision guard short-circuits to an + // immediate ACK without a second fetch. + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + + Thread.Sleep(200); + fetcher.Calls.ShouldBe(1); + } + + /// Records fetch calls and returns canned bytes (or null for "no answer"). + private sealed class RecordingFetcher(byte[]? bytes) : IDeploymentArtifactFetcher + { + private int _calls; + public int Calls => Volatile.Read(ref _calls); + + public Task FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct) + { + Interlocked.Increment(ref _calls); + return Task.FromResult(bytes); + } + } + + /// Records every store; GetCurrent* return null (no idempotency shortcut in these tests). + private sealed class RecordingArtifactCache : IDeploymentArtifactCache + { + private readonly Lock _gate = new(); + private readonly List _stores = []; + + public IReadOnlyList Stores + { + get { lock (_gate) { return _stores.ToArray(); } } + } + + public Task StoreAsync(string clusterId, string deploymentId, string revisionHash, + byte[] artifact, CancellationToken ct = default) + { + lock (_gate) { _stores.Add(new StoreCall(clusterId, deploymentId, revisionHash, artifact)); } + return Task.CompletedTask; + } + + public Task GetCurrentAsync(string clusterId, CancellationToken ct = default) + => Task.FromResult(null); + + public Task GetCurrentUnkeyedAsync(CancellationToken ct = default) + => Task.FromResult(null); + + internal sealed record StoreCall( + string ClusterId, string DeploymentId, string RevisionHash, byte[] Artifact); + } +}