fix(#654): bounded fast retry while the known-cluster gate is unpublished

CentralCommunicationActor: while the PSK provider is unresolved or the first ClusterNode load failed, re-arm RefreshContacts every 2 s (max 15) until the first publish; Information log on first publish; 'NEVER be published' Error now only after the retry budget is exhausted. Corrected false comments. Tests: late provider (Grpc+Dps), failed first load, retries stop, bounded budget; ControlPlaneStarterOrderTests back to the default 60 s refresh. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-08 23:32:38 -04:00
parent ca17d2cf71
commit 9dc6390d50
4 changed files with 184 additions and 26 deletions
@@ -171,7 +171,7 @@ The checked-in `appsettings*.json` files are deliberately thin: they carry only
 | Key | Type | Default | Meaning |
 |---|---|---|---|
 | `Mode` | string | `Dps` | `Dps` = DistributedPubSub, pair-local only — cannot cross an application-`Cluster` boundary. `Grpc` = the `mesh_transport.v1` gRPC contract (below), required on any node carrying a `cluster-{ClusterId}` role. Any other value **refuses to start**, including the retired `ClusterClient` value — a node still configured for it is refused at boot rather than silently demoted to `Dps`, which on a split-topology node would deliver nothing at all. |
-| `ContactRefreshSeconds` | int | `60` | Central-side only: how often central rebuilds its node dispatch set — the gRPC dial targets and the known-cluster PSK gate — from the `ClusterNode` rows. Runs in every mode; under `Dps` it still keeps the inbound-ack endpoint's known-cluster gate current even though nothing dials out. |
+| `ContactRefreshSeconds` | int | `60` | Central-side only: how often central rebuilds its node dispatch set — the gRPC dial targets and the known-cluster PSK gate — from the `ClusterNode` rows. Runs in every mode; under `Dps` it still keeps the inbound-ack endpoint's known-cluster gate current even though nothing dials out. While the known-cluster gate is still unpublished (PSK provider not yet resolvable at boot, or the first `ClusterNode` load failed) central also retries a refresh every 2 s, at most 15 times, until the first publish (#654); steady state adds no traffic. |

 **`Grpc` mode: central dials each node, a node posts its ack to central.** `CentralCommunicationActor` fans one `NodeCommandService.Execute` call per node — the target set is every enabled, non-maintenance `ClusterNode` row that carries a `GrpcPort` (dialled at `Host`+`GrpcPort`), rebuilt every `ContactRefreshSeconds`. There is no receptionist and no single fleet-wide (or per-cluster) client object: it is a plain per-node dial. A node posts its `ApplyAck` with `GrpcApplyAckSender`, which walks `CentralGrpcEndpoints` in order and stops at the first success — an availability list, not a fan-out list, so central normally sees one ack per node per dispatch (the walk as a whole is at-least-once on a slow-then-retried endpoint; `ConfigPublishCoordinator.HandleAck` absorbs a duplicate idempotently). Both directions are **fire-and-forget, with no retry**: a failed dispatch is logged (naming the node and the gRPC status) and dropped in the dispatch second — the deploy path's 2-minute apply deadline and a node's boot-time replay (`DriverHostActor.Bootstrap()` re-applying an orphan `Applying` row) are the only recovery mechanisms, exactly as they were under `ClusterClient`.

@@ -48,6 +48,18 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication;
 public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
 {
    private const string RefreshTimerKey = "mesh-contact-refresh";
+    private const string BootstrapRetryTimerKey = "mesh-contact-bootstrap-retry";
+
+    // While the known-cluster gate is unpublished (PSK provider unresolved, or the first
+    // ClusterNode load failed) a single-shot RefreshContacts is re-armed every
+    // DefaultBootstrapRetryInterval, at most MaxBootstrapRetries times (~30 s), stopping at the
+    // first publish. Constants, not config. Steady state costs nothing extra.
+    private static readonly TimeSpan DefaultBootstrapRetryInterval = TimeSpan.FromSeconds(2);
+    private const int MaxBootstrapRetries = 15;
+    private readonly TimeSpan _bootstrapRetryInterval;
+    private int _bootstrapRetries;
+    private int _refreshAttempts;
+    private bool _gatePublished;

    private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
    private readonly MeshTransportOptions _options;
@@ -64,9 +76,11 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
    // Memoize-on-first-use, NOT PreStart-only: PreStart runs on a dispatcher thread concurrently
    // with the builder's remaining starters, so its resolve can lose a millisecond-wide race
    // against AuditWriterActorKey's registration (starter #4) and Resolve() would swallow that
-    // into a PERMANENT silent null — the known-cluster gate never published, fail-open. These
-    // properties retry on first use instead; the first RefreshContacts arrives as a message,
-    // long after every starter finished, so a lost boot race self-heals there. Actor-thread
+    // into a silent null — the known-cluster gate unpublished, fail-open. These properties
+    // retry on first use, but the first RefreshContacts is sent from PreStart (TimeSpan.Zero) and
+    // can ALSO land inside the starter window (a fast DB load finishes in ms), so first-use is not
+    // enough: while the gate is unpublished a bounded fast retry (ScheduleBootstrapRetry) re-arms
+    // RefreshContacts every couple of seconds until it publishes. Actor-thread
    // only, so ??= needs no synchronisation; both are container singletons, so a repeat resolve
    // is a dictionary hit.
    private IMeshPskProvider? PskProvider =>
@@ -118,6 +132,10 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
    ///     exist (the known-cluster gate) and to drop its cache (so a rotated key takes effect within
    ///     one refresh). Optional for the same reason as <paramref name="dialer"/>.
    /// </param>
+    /// <param name="bootstrapRetryInterval">
+    ///     Interval of the bounded fast retry run while the known-cluster gate is unpublished
+    ///     (default 2 s, at most 15 attempts).
+    /// </param>
    /// <returns>The props.</returns>
    /// <remarks>
    ///     <para>
@@ -145,9 +163,11 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
        MeshTransportOptions options,
        Func<IActorRef?> coordinator,
        Func<IMeshCommandDialer?>? dialer = null,
-        Func<IMeshPskProvider?>? pskProvider = null) =>
+        Func<IMeshPskProvider?>? pskProvider = null,
+        TimeSpan? bootstrapRetryInterval = null) =>
        Akka.Actor.Props.Create(() =>
-            new CentralCommunicationActor(dbFactory, options, coordinator, dialer, pskProvider));
+            new CentralCommunicationActor(
+                dbFactory, options, coordinator, dialer, pskProvider, bootstrapRetryInterval));

    /// <summary>Initializes a new instance of the <see cref="CentralCommunicationActor"/> class.</summary>
    /// <param name="dbFactory">Factory for the config database.</param>
@@ -155,13 +175,16 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
    /// <param name="coordinator">Lazily resolves the deploy-coordinator singleton proxy.</param>
    /// <param name="dialer">Resolves the gRPC command dialer; invoked in <see cref="PreStart"/>.</param>
    /// <param name="pskProvider">Resolves the PSK provider; invoked in <see cref="PreStart"/>.</param>
+    /// <param name="bootstrapRetryInterval">Fast-retry interval while the gate is unpublished; default 2 s.</param>
    public CentralCommunicationActor(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        MeshTransportOptions options,
        Func<IActorRef?> coordinator,
        Func<IMeshCommandDialer?>? dialer = null,
-        Func<IMeshPskProvider?>? pskProvider = null)
+        Func<IMeshPskProvider?>? pskProvider = null,
+        TimeSpan? bootstrapRetryInterval = null)
    {
+        _bootstrapRetryInterval = bootstrapRetryInterval ?? DefaultBootstrapRetryInterval;
        _dbFactory = dbFactory;
        _options = options;
        _coordinator = coordinator;
@@ -172,7 +195,11 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers

        Receive<MeshCommand>(HandleMeshCommand);
        Receive<ApplyAck>(HandleApplyAck);
-        Receive<RefreshContacts>(_ => LoadContactsFromDb());
+        Receive<RefreshContacts>(_ =>
+        {
+            _refreshAttempts++;
+            LoadContactsFromDb();
+        });
        Receive<ContactsLoaded>(HandleContactsLoaded);
        Receive<CommandDispatched>(HandleCommandDispatched);

@@ -185,9 +212,12 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
        // widen the gate nor narrow it, and the previous set is the last thing central actually knew.
        // The second line exists because that freeze is INVISIBLE otherwise, and it is fail-OPEN
        // before the first successful load — a central that has never reached its ConfigDb refuses
-        // no cluster at all.
+        // no cluster at all. Until the first publish a bounded fast retry (above) heals it.
        Receive<Status.Failure>(f =>
        {
+            // Gate still unpublished (boot with ConfigDb briefly down): retry fast, bounded.
+            if (!_gatePublished)
+                ScheduleBootstrapRetry();
            _log.Warning(
                f.Cause,
                "Failed to load ClusterNode contact points; the mesh contact set was NOT refreshed "
@@ -213,8 +243,9 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
        // with the builder's remaining starters and loses a millisecond-wide race against
        // AuditWriterActorKey's registration (starter #4) on essentially every admin boot (live
        // gates: 1A 2/3 boots, 1B 4/4 boots). That is the ORDINARY case, not a malfunction — the
-        // memoized PskProvider/Dialer properties retry on first use, long after every starter
-        // finished, and the lost race self-heals there almost every time. Logging it as
+        // memoized PskProvider/Dialer properties retry on first use, and the bounded bootstrap
+        // retry re-arms RefreshContacts until the gate publishes (the first refresh itself can
+        // land inside the starter window). Logging it as
        // Warning/Error twice a boot is what made a soak error-grep on a healthy rig impossible.
        _dialer = Resolve(_dialerFactory, nameof(IMeshCommandDialer), firstUse: false);
        _pskProvider = Resolve(_pskProviderFactory, nameof(IMeshPskProvider), firstUse: false);
@@ -250,6 +281,15 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
            TimeSpan.FromSeconds(_options.ContactRefreshSeconds));
    }

+    /// <summary>Re-arms a single-shot <see cref="RefreshContacts"/> while the gate is unpublished.</summary>
+    private void ScheduleBootstrapRetry()
+    {
+        if (_bootstrapRetries >= MaxBootstrapRetries)
+            return;
+        _bootstrapRetries++;
+        Timers.StartSingleTimer(BootstrapRetryTimerKey, new RefreshContacts(), _bootstrapRetryInterval);
+    }
+
    /// <summary>
    ///     Invokes a dependency factory, converting a resolution failure into a logged absence.
    /// </summary>
@@ -718,12 +758,17 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
            // Gated on a factory having been supplied at all: a never-supplied factory is a
            // deliberate wiring/test choice Resolve() already stayed silent about (see its
            // remarks), not a resolve failure, so it earns no Error here either.
-            if (_pskProviderFactory is not null && !_pskProviderAbsenceReported)
+            if (_pskProviderFactory is not null && !_gatePublished)
+                ScheduleBootstrapRetry();
+
+            // Error only once the bounded fast retry is exhausted, so a healed boot race stays quiet.
+            if (_pskProviderFactory is not null && !_pskProviderAbsenceReported
+                && _bootstrapRetries >= MaxBootstrapRetries)
            {
                _pskProviderAbsenceReported = true;
                _log.Error(
-                    "No IMeshPskProvider is resolvable for this actor, even after a first-use retry; "
-                    + "the known-cluster gate will NEVER be published and cached per-cluster keys "
+                    "No IMeshPskProvider is resolvable for this actor, even after a first-use retry and "
+                    + "the bounded bootstrap retries; the known-cluster gate will NEVER be published and cached per-cluster keys "
                    + "will NEVER be dropped. Inbound mesh acks stay exposed to the "
                    + "pre-authentication secret-store cost the gate removes, and a rotated key "
                    + "needs a central restart. Check that the ControlPlane registration wires up "
@@ -735,6 +780,15 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers

        pskProvider.SetKnownClusters(msg.ClusterIds);
        pskProvider.InvalidateAll();
+
+        if (!_gatePublished)
+        {
+            _gatePublished = true;
+            Timers.Cancel(BootstrapRetryTimerKey);
+            _log.Info(
+                "Known-cluster gate published ({Count} clusters) after {Attempts} refresh attempt(s)",
+                msg.ClusterIds.Count, _refreshAttempts);
+        }
    }

    /// <summary>Internal tick asking for a contact-point refresh.</summary>
@@ -678,6 +678,107 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
        pskCalls.ShouldBeGreaterThanOrEqualTo(2);
    }

+    /// <summary>Counts contact loads and fails the first <c>failFirst</c> of them.</summary>
+    private sealed class CountingDbFactory(
+        IDbContextFactory<OtOpcUaConfigDbContext> inner, int failFirst = 0)
+        : IDbContextFactory<OtOpcUaConfigDbContext>
+    {
+        private int _calls;
+
+        public int Calls => Volatile.Read(ref _calls);
+
+        public OtOpcUaConfigDbContext CreateDbContext()
+        {
+            if (Interlocked.Increment(ref _calls) <= failFirst)
+                throw new InvalidOperationException("ConfigDb unreachable at boot");
+            return inner.CreateDbContext();
+        }
+
+        public Task<OtOpcUaConfigDbContext> CreateDbContextAsync(CancellationToken ct = default) =>
+            Task.FromResult(CreateDbContext());
+    }
+
+    private static MeshTransportOptions ModeOptions(string mode) => new()
+    {
+        Mode = mode,
+        ContactRefreshSeconds = 60,
+        DispatchTimeoutSeconds = 10,
+    };
+
+    [Theory]
+    [InlineData(MeshTransportOptions.ModeGrpc)]
+    [InlineData(MeshTransportOptions.ModeDps)]
+    public void The_gate_publishes_within_seconds_when_the_provider_resolves_late_under_the_default_refresh(
+        string mode)
+    {
+        // #654: the first RefreshContacts can land before the registry-bound PSK graph is resolvable.
+        // The next PERIODIC tick is 60 s away; the bounded bootstrap retry must publish in seconds.
+        var factory = NewInMemoryDbFactory();
+        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
+        var psk = new RecordingPskProvider();
+        var calls = 0;
+
+        IMeshPskProvider? PskFactory() =>
+            Interlocked.Increment(ref calls) <= 3 ? throw new InvalidOperationException("race") : psk;
+
+        EventFilter.Error().Expect(0, TimeSpan.FromSeconds(1), () =>
+        {
+            Sys.ActorOf(CentralCommunicationActor.Props(
+                factory, ModeOptions(mode), () => null, () => new RecordingDialer(), PskFactory));
+            AwaitCondition(() => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(10));
+        });
+    }
+
+    [Fact]
+    public void A_failed_first_contact_load_retries_quickly_under_the_default_refresh()
+    {
+        var inner = NewInMemoryDbFactory();
+        SeedGrpcCluster(inner, "C1", ("host-a:4053", "host-a", 4056, true, false));
+        var factory = new CountingDbFactory(inner, failFirst: 1);
+        var psk = new RecordingPskProvider();
+
+        EventFilter.Warning().Expect(2, TimeSpan.FromSeconds(10), () =>
+            Sys.ActorOf(CentralCommunicationActor.Props(
+                factory, GrpcMode(), () => null, () => new RecordingDialer(), () => psk)));
+
+        AwaitCondition(() => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(10));
+    }
+
+    [Fact]
+    public void Bootstrap_retries_stop_once_the_gate_is_published()
+    {
+        var inner = NewInMemoryDbFactory();
+        SeedGrpcCluster(inner, "C1", ("host-a:4053", "host-a", 4056, true, false));
+        var factory = new CountingDbFactory(inner);
+        var psk = new RecordingPskProvider();
+        Sys.ActorOf(CentralCommunicationActor.Props(
+            factory, GrpcMode(), () => null, () => new RecordingDialer(), () => psk,
+            bootstrapRetryInterval: TimeSpan.FromMilliseconds(100)));
+
+        // Positive control: the gate did publish before we assert anything about "afterwards".
+        AwaitCondition(() => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(10));
+        Thread.Sleep(1000);
+
+        // Ten retry intervals elapsed with a 60 s periodic cadence: no extra loads, no republish.
+        factory.Calls.ShouldBe(1);
+        psk.KnownClusterPublications.Count.ShouldBe(1);
+    }
+
+    [Fact]
+    public void A_permanently_failing_load_retry_budget_is_bounded()
+    {
+        var inner = NewInMemoryDbFactory();
+        var factory = new CountingDbFactory(inner, failFirst: int.MaxValue);
+        Sys.ActorOf(CentralCommunicationActor.Props(
+            factory, GrpcMode(), () => null, () => new RecordingDialer(), () => new RecordingPskProvider(),
+            bootstrapRetryInterval: TimeSpan.FromMilliseconds(20)));
+
+        EventFilter.Warning().Expect(32, TimeSpan.FromSeconds(10), () => { });
+        Thread.Sleep(1500);
+        // 1 initial + 15 retries; the 60 s periodic tick has not fired again.
+        factory.Calls.ShouldBe(16);
+    }
+
    [Fact]
    public void A_permanently_unresolvable_pskProvider_is_reported_once_per_incarnation_not_once_per_refresh()
    {
@@ -702,14 +803,15 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
        }

        IActorRef actor = null!;
-        EventFilter.Error(contains: "will NEVER be published").ExpectOne(TimeSpan.FromSeconds(5), () =>
+        EventFilter.Error(contains: "will NEVER be published").ExpectOne(TimeSpan.FromSeconds(10), () =>
        {
            actor = Sys.ActorOf(CentralCommunicationActor.Props(
                factory,
                GrpcMode(),
                () => null,
                () => new RecordingDialer(),
-                pskProvider: PskFactory));
+                pskProvider: PskFactory,
+                bootstrapRetryInterval: TimeSpan.FromMilliseconds(50)));

            // pskCalls == 1 is PreStart's own attempt; reaching 2 is PROOF the first contact load
            // ran and PublishKnownClusters's property access performed the first-use retry — not
@@ -717,6 +819,13 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
            AwaitCondition(() => pskCalls >= 2, TimeSpan.FromSeconds(5));
        });

+        // The bootstrap retry is BOUNDED: with a 50 ms interval and a budget of 15 it has long
+        // finished, so the call count plateaus instead of growing with elapsed time.
+        Thread.Sleep(500);
+        var plateau = pskCalls;
+        Thread.Sleep(500);
+        pskCalls.ShouldBe(plateau);
+
        var pskCallsAfterFirstLoad = pskCalls;

        // A SECOND contact load must NOT repeat the Error. The pin only means something if that
@@ -94,8 +94,7 @@ public sealed class ControlPlaneStarterOrderTests
        // walks the same registry-bound chain the production MeshPskProvider does — which makes
        // this fact the race detector for the PreStart-vs-starter-#4 window: a lost race that
        // stayed null forever would never publish, and the TaskCompletionSource below would time
-        // out red. (Memoize-on-first-use in the actor is what heals the lost race at the first
-        // RefreshContacts message.)
+        // out red. (Memoize-on-first-use plus the bounded bootstrap retry heal a lost race.)
        //
        // The signal is a TaskCompletionSource rather than a counter, because PreStart runs on the
        // actor's own thread and the contact load is a piped DB read: there is no instant after
@@ -143,15 +142,11 @@ public sealed class ControlPlaneStarterOrderTests
            .ConfigureServices(services =>
            {
                services.AddSingleton<IOptions<AkkaClusterOptions>>(Options.Create(clusterOptions));
-                // ContactRefreshSeconds = 1 (production: 60). The memoized PSK provider retries on EVERY
-                // refresh, so a first refresh that lands before the registry-bound graph is resolvable
-                // (the lost race the resolution fact exists to heal) publishes on the NEXT tick. At 60 s
-                // that next tick was outside the fact's 20 s budget — a lost race on a loaded CI runner
-                // timed out red exactly like a never-resolved provider (the Dps case failed in 2 of 7
-                // Gitea runs, #606). A 1 s cadence
-                // still proves "resolved and used" — a provider that stayed null forever never publishes.
+                // ContactRefreshSeconds is the production default (60). A first refresh that lands before the
+                // registry-bound graph is resolvable is healed by the bounded bootstrap retry (#654), not
+                // by the next periodic tick, so the 20 s budget below proves production heals fast.
                services.AddSingleton<IOptions<MeshTransportOptions>>(Options.Create(
-                    new MeshTransportOptions { Mode = mode, DispatchTimeoutSeconds = 10, ContactRefreshSeconds = 1 }));
+                    new MeshTransportOptions { Mode = mode, DispatchTimeoutSeconds = 10 }));
                services.AddDbContextFactory<OtOpcUaConfigDbContext>(
                    o => o.UseInMemoryDatabase($"starter-order-{Guid.NewGuid():N}"));
                services.AddSingleton<IClusterRoleInfo>(new StubClusterRoleInfo());