Fix #208: skip stale-snapshot resume on SessionVerId bump

src/B3.EntryPoint.Client/EntryPointClient.cs

@@ -29,6 +29,14 @@ public sealed class EntryPointClient : IEntryPointClient, ISubmitOrder, IReplace
     private readonly Channel<EntryPointEvent> _events;
     private TcpClient? _tcp;
     private FixpClientSession? _session;
+
+    // #208: set by TryColdResumeAsync immediately before it bumps SessionVerID
+    // and falls through to a fresh Negotiate (recoverable EstablishReuse
+    // reject). Consumed exactly once by the following HydrateFromSnapshotAsync
+    // call so it does NOT resume outbound/inbound state from the now-stale
+    // (pre-bump) snapshot — a fresh Negotiate starts application sequencing
+    // over on both sides regardless of the persisted counters.
+    private bool _suppressNextSnapshotResume;
     private KeepAliveScheduler? _keepAlive;
     private RetransmitRequestHandler? _retransmit;
     private DateTime _lastInboundUtc;
@@ -513,6 +521,16 @@ private async ValueTask HydrateFromSnapshotAsync(CancellationToken ct)
             _options.Logger.StaleSnapshotIgnored(snapshot.SessionId, _options.SessionId);
             return;
         }
+        if (_suppressNextSnapshotResume)
+        {
+            _suppressNextSnapshotResume = false;
+            // See field doc on _suppressNextSnapshotResume (#208): a cold-start
+            // EstablishReuse reject just bumped SessionVerID and fell through
+            // to a fresh Negotiate — the snapshot's counters belong to the
+            // rejected (pre-bump) SessionVerID and must not be resumed.
+            _options.Logger.StaleSnapshotSessionVerIdIgnored(snapshot.SessionVerId, _options.SessionVerId, _options.SessionId);
+            return;
+        }
         _session!.ResumeOutboundSeqNum(snapshot.LastOutboundSeqNum + 1UL);
         lock (_inboundGapGate)
         {
@@ -856,6 +874,7 @@ internal Task StopActiveSessionForTestingAsync(CancellationToken ct = default)
     internal void HandleInboundEventForTesting(EntryPointEvent evt) => OnInboundEventForPersistence(evt);
     internal void BindRetransmitForTesting(RetransmitRequestHandler handler) => _retransmit = handler;
     internal bool HasActiveSessionForTesting => _session is not null || _tcp is not null;
+    internal ulong PeekNextOutboundSeqNumForTesting() => _session?.PeekNextOutboundSeqNum() ?? 0UL;
     internal (ulong contiguous, ulong highest, int pending, bool gapInFlight) GetInboundGapStateForTesting()
     {
         lock (_inboundGapGate)
@@ -1584,6 +1603,11 @@ private async Task<bool> TryColdResumeAsync(CancellationToken ct)
 
         _options.Logger.ColdResumeFallbackToNegotiate(code, prevVerId);
         _options.SessionVerId = nextVerId;
+        // #208: the fresh Negotiate about to happen starts application
+        // sequencing over; the snapshot we just loaded belongs to the
+        // rejected, pre-bump SessionVerID and must not seed the new session's
+        // outbound/inbound counters.
+        _suppressNextSnapshotResume = true;
         return false;
     }
 

src/B3.EntryPoint.Client/Logging/LogMessages.cs

@@ -93,6 +93,10 @@ internal static partial class LogMessages
         Message = "Cold-start resume rejected (code={Code}); falling back to Negotiate from SessionVerID={SessionVerId}")]
     public static partial void ColdResumeFallbackToNegotiate(this ILogger logger, B3.Entrypoint.Fixp.Sbe.V6.EstablishRejectCode code, uint sessionVerId);
 
+    [LoggerMessage(EventId = 3009, Level = LogLevel.Information,
+        Message = "Persisted snapshot SessionVerID={Persisted} != current {Current} for SessionID={SessionId}; new negotiated session starts fresh, ignoring stale sequence/order state.")]
+    public static partial void StaleSnapshotSessionVerIdIgnored(this ILogger logger, uint persisted, uint current, uint sessionId);
+
     // ---------------- Warning (4000–4999) ----------------
 
     [LoggerMessage(EventId = 4000, Level = LogLevel.Warning,

tests/B3.EntryPoint.Client.Tests/ColdStartResumeTests.cs

@@ -97,6 +97,38 @@ public async Task ColdResume_EstablishRejectedUnnegotiated_FallsBackToNegotiate_
         Assert.Equal(107u, options.SessionVerId);
     }
 
+    [Fact]
+    public async Task ColdResume_EstablishRejectedUnnegotiated_DoesNotResumeOutboundSeqFromStaleSnapshot()
+    {
+        // Regression for #208: on a recoverable EstablishReuse reject, the
+        // client bumps SessionVerId and falls back to a fresh Negotiate.
+        // HydrateFromSnapshotAsync must NOT resume the outbound app seq
+        // counter from the pre-bump snapshot (a new SessionVerId means the
+        // wire-level sequence numbering starts fresh regardless of the old
+        // snapshot's contents).
+        await using var peer = new InProcessFixpTestPeer(new TestPeerOptions
+        {
+            RejectEstablishWithoutPriorNegotiate = true,
+        });
+        peer.Start();
+
+        var options = Options(peer);
+        options.ConnectMode = ConnectMode.EstablishReuseThenNegotiate;
+        // Stale snapshot from the prior (rejected) SessionVerId, with a
+        // large outbound counter that must NOT be carried forward.
+        options.SessionStateStore = new SnapshotStore(Snapshot(sessionVerId: 7u, lastOut: 22u));
+        options.NextSessionVerIdSelector = prev => prev + 1u;
+        await using var client = new EntryPointClient(options);
+        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+
+        await client.ConnectAsync(cts.Token);
+
+        Assert.Equal(8u, options.SessionVerId);
+        // Fresh session (new SessionVerId): outbound app seq must start at 1,
+        // not resume at 23 (stale snapshot's LastOutboundSeqNum + 1).
+        Assert.Equal(1u, client.PeekNextOutboundSeqNumForTesting());
+    }
+
     [Fact]
     public async Task ColdResume_NoSnapshot_PlainNegotiate_NoVerIdChange()
     {