fix(localdb): fault live sync session on cap-prune; keepalive + newest-wins sync guard (0.3.2, ScadaBridge arch-review P4-1 + P4-3)

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbSyncService.cs

@@ -21,8 +21,26 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase
     private readonly LocalDbMetrics? _metrics;
     private readonly SyncStatus? _status;
 
-    // 0 = idle, 1 = a stream is active. Interlocked-guarded across concurrent calls (singleton service).
-    private int _active;
+    // The stream currently holding the guard, or null when idle. Guarded by _gate (singleton service).
+    private readonly object _gate = new();
+    private Registration? _active;
+
+    /// <summary>
+    /// How long a displacing call waits for the incumbent it evicted to actually finish before
+    /// giving up and refusing itself. Test seam; the default is generous because eviction only has
+    /// to outrun the operator, not the network.
+    /// </summary>
+    internal TimeSpan TakeoverTimeout { get; set; } = TimeSpan.FromSeconds(10);
+
+    /// <summary>One in-flight <c>Sync</c> handler: its cancellation handle plus a completion signal
+    /// a displacing handler can await, so two sessions never run against the database at once.</summary>
+    private sealed class Registration(CancellationTokenSource cts)
+    {
+        public CancellationTokenSource Cts { get; } = cts;
+
+        public TaskCompletionSource Finished { get; } =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+    }
 
     /// <summary>
     /// Creates the passive sync service. <paramref name="status"/> is taken as the public
@@ -41,21 +59,47 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase
         _status = status as SyncStatus;
     }
 
+    /// <summary>
+    /// Accepts one inbound sync stream, evicting any stream already holding the guard.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <b>Newest wins.</b> The guard exists to SERIALIZE sessions — a node syncs with exactly one
+    /// peer, and two sessions against one database would fight over the same watermarks — not to
+    /// fence out an impostor; authentication is the interceptor's job, and an authenticated caller
+    /// on this key IS the peer by definition. Refusing the newcomer therefore protected nothing and
+    /// cost everything in the case that actually happens: the dialling node crashes or is
+    /// partitioned, this handler stays parked forever inside <c>MoveNext</c> on a half-open stream
+    /// that will never produce another byte, and every reconnect from the rebooted peer is rejected
+    /// <c>AlreadyExists</c> — potentially for as long as the pair stays idle, while this side keeps
+    /// reporting itself connected. The newest dialler is the one with a live link, so it takes over.
+    /// </para>
+    /// <para>
+    /// Eviction is cancel-then-await, not cancel-and-go: the incumbent's cancellation unwinds its
+    /// session and reader pump, and the newcomer waits for that to complete before starting its own,
+    /// so the "one session at a time" property the guard was built for still holds. An incumbent
+    /// that cannot be evicted within <see cref="TakeoverTimeout"/> is the one case where the old
+    /// answer is still right — the newcomer refuses itself rather than double-running.
+    /// </para>
+    /// </remarks>
     public override async Task Sync(
         IAsyncStreamReader<SyncMessage> requestStream,
         IServerStreamWriter<SyncMessage> responseStream,
         ServerCallContext context)
     {
-        if (Interlocked.CompareExchange(ref _active, 1, 0) != 0)
-            throw new RpcException(new Status(StatusCode.AlreadyExists, "a sync stream is already active"));
+        // The linked source exists so the reader pump is deterministically torn down on EVERY exit
+        // path — including a session fault for a local (non-transport) reason, where the call token
+        // never fires and the pump would otherwise keep reading a stream whose handler has exited.
+        // It is also the handle a later call cancels to displace this one.
+        using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken);
+        var mine = new Registration(linkedCts);
 
         try
         {
-            // The linked source exists so the reader pump is deterministically torn down on EVERY
-            // exit path — including a session fault for a local (non-transport) reason, where the
-            // call token never fires and the pump would otherwise keep reading a stream whose
-            // handler has exited.
-            using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken);
+            // Inside the try so the finally below always releases the guard — including when the
+            // takeover itself is abandoned because THIS caller gave up mid-handover.
+            await TakeOverAsync(mine, context.CancellationToken);
+
             var (duplex, readerTask) = GrpcSyncDuplex.Create(requestStream, responseStream, linkedCts.Token);
             var session = SyncSessionFactory.Create(_db, _options, _loggerFactory, _metrics, _status);
             _status?.SessionStarted();
@@ -81,13 +125,66 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase
         {
             // Client disconnect / server shutdown: the normal end of a long-lived streaming RPC.
         }
+        catch (OperationCanceledException)
+        {
+            // Displaced by a newer stream: an orderly handover, not a failure of this node.
+            _logger.LogInformation("Sync stream ended: displaced by a newer stream from the peer.");
+        }
         catch (LocalDbSchemaMismatchException ex)
         {
             throw new RpcException(new Status(StatusCode.FailedPrecondition, ex.Message));
         }
         finally
         {
-            Interlocked.Exchange(ref _active, 0);
+            lock (_gate)
+            {
+                if (ReferenceEquals(_active, mine))
+                    _active = null;
+            }
+            mine.Finished.TrySetResult();
+        }
+    }
+
+    /// <summary>
+    /// Claims the single-stream guard for <paramref name="mine"/>, cancelling and awaiting whatever
+    /// held it. Throws <c>AlreadyExists</c> only when the incumbent refuses to end.
+    /// </summary>
+    private async Task TakeOverAsync(Registration mine, CancellationToken ct)
+    {
+        Registration? displaced;
+        lock (_gate)
+        {
+            displaced = _active;
+            _active = mine;
+        }
+        if (displaced is null)
+            return;
+
+        _logger.LogWarning(
+            "A new sync stream arrived while one was already active; displacing the incumbent (newest wins).");
+        try
+        {
+            // The incumbent may already be disposing its own source between its finally and here.
+            displaced.Cts.Cancel();
+        }
+        catch (ObjectDisposedException) { }
+
+        try
+        {
+            await displaced.Finished.Task.WaitAsync(TakeoverTimeout, ct);
+        }
+        catch (TimeoutException)
+        {
+            // Could not evict it: two live sessions over one database is worse than a refused dial,
+            // so fall back to the historical answer — and hand the guard back to the incumbent.
+            lock (_gate)
+            {
+                if (ReferenceEquals(_active, mine))
+                    _active = displaced;
+            }
+            throw new RpcException(new Status(
+                StatusCode.AlreadyExists,
+                "a sync stream is already active and did not yield within the takeover timeout"));
         }
     }
 }