diff --git a/src/B3.EntryPoint.Client/EntryPointClient.cs b/src/B3.EntryPoint.Client/EntryPointClient.cs index c845bfc..82204a6 100644 --- a/src/B3.EntryPoint.Client/EntryPointClient.cs +++ b/src/B3.EntryPoint.Client/EntryPointClient.cs @@ -29,6 +29,14 @@ public sealed class EntryPointClient : IEntryPointClient, ISubmitOrder, IReplace private readonly Channel _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 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; } diff --git a/src/B3.EntryPoint.Client/Logging/LogMessages.cs b/src/B3.EntryPoint.Client/Logging/LogMessages.cs index 7800568..6599aa8 100644 --- a/src/B3.EntryPoint.Client/Logging/LogMessages.cs +++ b/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, diff --git a/tests/B3.EntryPoint.Client.Tests/ColdStartResumeTests.cs b/tests/B3.EntryPoint.Client.Tests/ColdStartResumeTests.cs index 837021b..7195d84 100644 --- a/tests/B3.EntryPoint.Client.Tests/ColdStartResumeTests.cs +++ b/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() {