feat(mesh): FetchAndCache apply path — fetch->cache->apply-from-bytes, null=apply-failure

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<ConfigSourceOptions> (fetcher only in FetchAndCache mode); Host registers GrpcDeploymentArtifactFetcher under hasDriver. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
2026-07-22 19:39:26 -04:00
parent 1a7e3f7ea3
commit 648f173b18
4 changed files with 335 additions and 14 deletions
@@ -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<IDeploymentArtifactFetcher, GrpcDeploymentArtifactFetcher>();
+
    // 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.
@@ -72,6 +72,19 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
    /// </summary>
    private readonly IDeploymentArtifactCache? _deploymentArtifactCache;

+    /// <summary>
+    ///     Per-cluster mesh Phase 3. When <see langword="true"/> (<c>ConfigSource:Mode =
+    ///     FetchAndCache</c>) 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 <see langword="false"/> (default <c>Direct</c>) every path reads central SQL as before.
+    /// </summary>
+    private readonly bool _fetchAndCacheMode;
+
+    /// <summary>
+    ///     Node-side gRPC artifact fetcher, non-null only under <see cref="_fetchAndCacheMode"/>.
+    /// </summary>
+    private readonly IDeploymentArtifactFetcher? _artifactFetcher;
+
    /// <summary>
    ///     True once this node has booted a cached artifact because central SQL was unreachable.
    /// </summary>
@@ -382,7 +395,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
        Func<int>? 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));

    /// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
    /// <param name="dbFactory">Database context factory for configuration database access.</param>
@@ -446,11 +461,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
        Func<int>? 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<FetchedForApply>(HandleFetchedForApply);
        Receive<GetDiagnostics>(HandleGetDiagnostics);
        Receive<DriverInstanceActor.AttributeValuePublished>(ForwardToMux);
        Receive<DriverInstanceActor.AttributeAlarmPublished>(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);
    }

+    /// <summary>
+    ///     Per-cluster mesh Phase 3 (<c>FetchAndCache</c>). Kicks off a gRPC artifact fetch and pipes
+    ///     the result back to <see cref="Self"/> as a <see cref="FetchedForApply"/> — 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
+    ///     <see cref="HandleFetchedForApply"/>, mirroring <see cref="ApplyAndAck"/>.
+    /// </summary>
+    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));
+    }
+
+    /// <summary>
+    ///     Completes a <c>FetchAndCache</c> 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.
+    /// </summary>
+    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<string, object?>("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<string, object?>("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<string, object?>("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);
+        }
+    }
+
+    /// <summary>Pipe-back carrier for a <c>FetchAndCache</c> fetch result (null bytes = no answer).</summary>
+    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);
+    }
+
+    /// <summary>
+    ///     The blob-processing half of <see cref="ReconcileDrivers"/>, split out so the
+    ///     <c>FetchAndCache</c> apply path (which already holds the fetched bytes) and the
+    ///     boot-from-cache path can reconcile without a ConfigDb read.
+    /// </summary>
+    /// <returns>The reconciled blob, or <see langword="null"/> when it was empty (#485).</returns>
+    private byte[]? ReconcileDriversFromBlob(DeploymentId deploymentId, byte[]? blob)
+    {
+        blob ??= Array.Empty<byte>();
+
        // 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
@@ -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<IDeploymentArtifactCache>();
+            // 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<IOptions<ConfigSourceOptions>>()?.Value ?? new ConfigSourceOptions();
+            var fetchAndCacheMode = string.Equals(
+                configSourceOptions.Mode, ConfigSourceOptions.ModeFetchAndCache, StringComparison.OrdinalIgnoreCase);
+            var artifactFetcher = fetchAndCacheMode ? resolver.GetService<IDeploymentArtifactFetcher>() : 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<DriverHostActorKey>(driverHost);

@@ -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;
+
+/// <summary>
+///     Per-cluster mesh Phase 3 (Task 5) — the <c>FetchAndCache</c> 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).
+/// </summary>
+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));
+
+    /// <summary>Builds a minimal but parseable artifact (no drivers) — the fetched-bytes stand-in.</summary>
+    private static byte[] ArtifactBytes() =>
+        JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty<object>() });
+
+    [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<string> { "driver" },
+            deploymentArtifactCache: cache,
+            fetchAndCacheMode: true,
+            artifactFetcher: fetcher));
+
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+
+        coordinator.ExpectMsg<ApplyAck>(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<string> { "driver" },
+            deploymentArtifactCache: cache,
+            fetchAndCacheMode: true,
+            artifactFetcher: fetcher));
+
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+        coordinator.ExpectMsg<ApplyAck>(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<ApplyAck>(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<string> { "driver" },
+            deploymentArtifactCache: cache,
+            fetchAndCacheMode: true,
+            artifactFetcher: fetcher));
+
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+        coordinator.ExpectMsg<ApplyAck>(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<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
+
+        Thread.Sleep(200);
+        fetcher.Calls.ShouldBe(1);
+    }
+
+    /// <summary>Records fetch calls and returns canned bytes (or null for "no answer").</summary>
+    private sealed class RecordingFetcher(byte[]? bytes) : IDeploymentArtifactFetcher
+    {
+        private int _calls;
+        public int Calls => Volatile.Read(ref _calls);
+
+        public Task<byte[]?> FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct)
+        {
+            Interlocked.Increment(ref _calls);
+            return Task.FromResult(bytes);
+        }
+    }
+
+    /// <summary>Records every store; GetCurrent* return null (no idempotency shortcut in these tests).</summary>
+    private sealed class RecordingArtifactCache : IDeploymentArtifactCache
+    {
+        private readonly Lock _gate = new();
+        private readonly List<StoreCall> _stores = [];
+
+        public IReadOnlyList<StoreCall> 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<CachedDeploymentArtifact?> GetCurrentAsync(string clusterId, CancellationToken ct = default)
+            => Task.FromResult<CachedDeploymentArtifact?>(null);
+
+        public Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default)
+            => Task.FromResult<CachedDeploymentArtifact?>(null);
+
+        internal sealed record StoreCall(
+            string ClusterId, string DeploymentId, string RevisionHash, byte[] Artifact);
+    }
+}