fix(fixp): preserve outbound journal on suspended reap

src/B3.Exchange.Gateway/CloseKind.cs

@@ -70,8 +70,10 @@ public enum CloseKind
     /// <summary>
     /// Suspended-state reaper aged the session out after the
     /// configured Suspended-window expired without re-attach. The
-    /// session is considered abandoned; persisted state should be
-    /// removed.
+    /// session slot is considered abandoned and its resumability
+    /// snapshot should be removed, but the durable outbound journal is
+    /// preserved so the journal's own retention policy remains the
+    /// sole authority on when replay bytes are deleted.
     /// </summary>
     SuspendedTimeout,
 

src/B3.Exchange.Gateway/EntryPointListener.cs

@@ -3,6 +3,7 @@
 using System.Net;
 using System.Net.Sockets;
 using B3.Exchange.Contracts;
+using B3.Exchange.Contracts.Time;
 using Microsoft.Extensions.Logging;
 
 namespace B3.Exchange.Gateway;
@@ -158,6 +159,7 @@ private async Task RunSuspendedReaperAsync(CancellationToken ct)
                 try { await Task.Delay(poll, ct).ConfigureAwait(false); }
                 catch (OperationCanceledException) { return; /* expected: reaper cancelled during shutdown */ }
                 ReapSuspendedOnce(Environment.TickCount64);
+                SweepOutboundJournalRetentionOnce((long)SystemNanosTimeSource.Instance.NowNanos());
             }
         }
         catch (Exception ex)
@@ -203,6 +205,29 @@ internal void ReapSuspendedOnce(long nowMs)
         }
     }
 
+    internal void SweepOutboundJournalRetentionOnce(long nowNanos)
+    {
+        if (_outboundJournal is null) return;
+        IReadOnlyCollection<uint> sessions;
+        try { sessions = _outboundJournal.ListSessions(); }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "journal retention sweep failed to enumerate sessions");
+            return;
+        }
+
+        foreach (var sessionId in sessions)
+        {
+            try { _outboundJournal.EnforceRetention(sessionId, nowNanos); }
+            catch (Exception ex)
+            {
+                _logger.LogWarning(ex,
+                    "journal retention sweep failed for sessionId={SessionId}",
+                    sessionId);
+            }
+        }
+    }
+
     /// <summary>
     /// Spec §4.5.1 (#GAP-09 / issue #47): at start of each trading day the
     /// gateway resets inbound + outbound MsgSeqNum counters to 1 and

src/B3.Exchange.Gateway/FixpSession.Lifecycle.cs

@@ -418,23 +418,42 @@ private void CloseLocked(string reason, CloseKind kind)
                     ConnectionId);
             }
         }
-        // Issue #289 / #405 / #416: terminal close ⇒ drop the persisted
-        // retransmit ring file AND the unbounded outbound journal AND
-        // the state snapshot. Scoped to terminal close kinds
-        // so transport-error and host-shutdown closes preserve state for
-        // the reconnecting peer to resync against (SBE 5.2 §1.5
-        // recoverable serverFlow). Non-removing kinds also save the
-        // final state so the resume point on reconnect is accurate.
-        bool removePersistence =
+        // Issue #289 / #405 / #416 / #594: terminal close kinds still
+        // retire the in-memory replay window and the FIXP resumability
+        // snapshot, but SuspendedTimeout no longer preempts the durable
+        // journal's own retention policy. The reaper frees the session slot;
+        // the journal decides when bytes are no longer servable.
+        bool removeStateSnapshot =
             kind == CloseKind.PeerTerminate
             || kind == CloseKind.LocalTerminate
             || kind == CloseKind.KeepaliveLapsed
             || kind == CloseKind.SuspendedTimeout
             || kind == CloseKind.DailyReset;
-        if (removePersistence)
+        bool removeOutboundJournal =
+            kind == CloseKind.PeerTerminate
+            || kind == CloseKind.LocalTerminate
+            || kind == CloseKind.KeepaliveLapsed
+            || kind == CloseKind.DailyReset;
+        if (removeStateSnapshot)
         {
             _retxBuffer.Dispose();
-            if (ownsLogicalSession && _outboundJournal is not null && SessionId != 0)
+            if (!removeOutboundJournal
+                && ownsLogicalSession
+                && _outboundJournal is not null
+                && SessionId != 0)
+            {
+                try { _outboundJournal.ReleaseActive(SessionId); }
+                catch (Exception ex)
+                {
+                    _logger.LogWarning(ex,
+                        "fixp session {ConnectionId} sessionId={SessionId} failed to release active outbound journal state",
+                        ConnectionId, SessionId);
+                }
+            }
+            if (removeOutboundJournal
+                && ownsLogicalSession
+                && _outboundJournal is not null
+                && SessionId != 0)
             {
                 try { _outboundJournal.Remove(SessionId); }
                 catch (Exception ex)

src/B3.Exchange.Gateway/FixpSession.Negotiate.cs

@@ -84,14 +84,26 @@ private NegotiateStep ProcessNegotiate(ReadOnlySpan<byte> fixedBlock, ReadOnlySp
                 return NegotiateStep.Rejected(rejectFrame,
                     $"negotiate-reject (legacy, ALREADY_NEGOTIATED, action={action})");
             }
+            if (!TryPrepareFreshNegotiateState(req.SessionId))
+            {
+                State = FixpState.Idle;
+                RollbackPendingNegotiateState();
+                var rejectFrame = new byte[NegotiateRejectEncoder.Total];
+                NegotiateRejectEncoder.Encode(rejectFrame, req.SessionId, req.SessionVerId,
+                    req.TimestampNanos, enteringFirm: null,
+                    B3.Entrypoint.Fixp.Sbe.V6.NegotiationRejectCode.UNSPECIFIED,
+                    currentSessionVerId: null);
+                return NegotiateStep.Rejected(rejectFrame,
+                    "negotiate-reject (legacy, outbound journal rollover failed)");
+            }
+            SessionId = req.SessionId;
+            EnteringFirm = req.EnteringFirm;
+            SessionVerId = req.SessionVerId;
             // Issue #485: update Identity to stable FIXP SessionId (legacy path).
             UpdateIdentityAfterNegotiate(
                 req.SessionId,
                 replaceRetired: true,
                 out _);
-            SessionId = req.SessionId;
-            EnteringFirm = req.EnteringFirm;
-            SessionVerId = req.SessionVerId;
             var frame = new byte[NegotiateResponseEncoder.Total];
             NegotiateResponseEncoder.Encode(frame, req.SessionId, req.SessionVerId,
                 req.TimestampNanos, req.EnteringFirm,
@@ -183,7 +195,8 @@ _sessionRegistry is null
         }
 
         _claimedSessionId = req.SessionId;
-        if (!TryApplyPendingNegotiateState(req.SessionId))
+        if (evictedByTakeOver is null
+            && !TryPrepareFreshNegotiateState(req.SessionId))
         {
             if (evictedByTakeOver is not null)
             {
@@ -204,13 +217,14 @@ _sessionRegistry is null
             }
             _claims.Release(req.SessionId, this);
             _claimedSessionId = 0;
+            RollbackPendingNegotiateState();
             var rejectFrame = new byte[NegotiateRejectEncoder.Total];
             NegotiateRejectEncoder.Encode(rejectFrame, req.SessionId, req.SessionVerId,
                 req.TimestampNanos, enteringFirm: null,
                 B3.Entrypoint.Fixp.Sbe.V6.NegotiationRejectCode.UNSPECIFIED,
                 currentSessionVerId: null);
             return NegotiateStep.Rejected(rejectFrame,
-                "negotiate-reject (UNSPECIFIED: persisted outbound state reconcile failed)");
+                "negotiate-reject (UNSPECIFIED: outbound journal rollover failed)");
         }
         _ = ApplyTransition(FixpEvent.Negotiate);
         SessionId = req.SessionId;
@@ -281,6 +295,8 @@ _sessionRegistry is null
         }
         if (!committed)
         {
+            if (evictedByTakeOver is null && SessionId != 0)
+                _ = TryRestoreRolledOutboundJournalGenerationForTakeOver(SessionId);
             // Roll back the in-memory claim taken above so the session
             // can be retried by the peer (same SessionVerID, same TCP
             // connection or a new one); without this the second attempt

src/B3.Exchange.Gateway/FixpSession.cs

@@ -506,6 +506,16 @@ public FixpSession(long connectionId, uint enteringFirm, uint sessionId,
             uint resumedSeq = state.OutboundMsgSeqNum;
             if (outboundJournal is not null)
             {
+                if (resumedSeq > 0 && outboundJournal.MaxSeq(state.SessionId) == 0)
+                {
+                    try { outboundJournal.RestoreLatestGeneration(state.SessionId); }
+                    catch (Exception ex)
+                    {
+                        _logger.LogWarning(ex,
+                            "fixp session {ConnectionId} failed to recover latest rolled journal generation for sessionId={SessionId}",
+                            connectionId, state.SessionId);
+                    }
+                }
                 uint journalMax = outboundJournal.MaxSeq(state.SessionId);
                 if (journalMax > resumedSeq) resumedSeq = journalMax;
             }
@@ -698,27 +708,22 @@ private void RollbackMsgSeqNum(uint allocated)
         Volatile.Write(ref _msgSeqNum, allocated - 1u);
     }
 
-    private bool TryApplyPendingNegotiateState(uint sessionId)
+    private bool TryPrepareFreshNegotiateState(uint sessionId)
     {
-        if (_pendingNegotiateState is not { } state)
-            return true;
-        if (state.SessionId != sessionId)
+        if (_pendingNegotiateState is { } state && state.SessionId != sessionId)
             return false;
+        if (_outboundJournal is null)
+            return true;
 
         try
         {
-            uint resumedSeq = state.OutboundMsgSeqNum;
-            if (_outboundJournal is not null)
-                resumedSeq = Math.Max(resumedSeq, _outboundJournal.MaxSeq(sessionId));
-            LastIncomingSeqNo = state.LastIncomingSeqNo;
-            Volatile.Write(ref _msgSeqNum, resumedSeq);
-            _pendingNegotiateStateApplied = true;
+            _outboundJournal.RollGeneration(sessionId);
             return true;
         }
         catch (Exception ex)
         {
             _logger.LogWarning(ex,
-                "fixp session {ConnectionId} failed to reconcile pending persisted state for sessionId={SessionId}",
+                "fixp session {ConnectionId} failed to prepare fresh negotiate state for sessionId={SessionId}",
                 ConnectionId, sessionId);
             return false;
         }
@@ -845,14 +850,15 @@ internal void RollbackTakeOverSeal()
         Interlocked.CompareExchange(ref _takeOverCommitFence, restoredState, 2);
     }
 
-    internal bool TryAdoptOutboundStateForTakeOver(
+    internal SessionClaimRegistry.TakeOverCommitDecision TryFinalizeOutboundStateForTakeOver(
         FixpSession previous,
-        Action commit)
+        bool freshGeneration,
+        Func<SessionClaimRegistry.TakeOverCommitDecision> commit)
     {
         ArgumentNullException.ThrowIfNull(previous);
         ArgumentNullException.ThrowIfNull(commit);
         if (!ReferenceEquals(_outboundJournal, previous._outboundJournal))
-            return false;
+            return SessionClaimRegistry.TakeOverCommitDecision.RollBack;
 
         var first = ConnectionId < previous.ConnectionId ? _outboundLock : previous._outboundLock;
         var second = ReferenceEquals(first, _outboundLock) ? previous._outboundLock : _outboundLock;
@@ -861,9 +867,19 @@ internal bool TryAdoptOutboundStateForTakeOver(
             lock (second)
             {
                 if (!IsLiveTakeOverCandidate || !previous.IsRegistered)
-                    return false;
+                    return SessionClaimRegistry.TakeOverCommitDecision.RollBack;
+
+                if (freshGeneration)
+                {
+                    if (_retxBuffer.Count != 0)
+                        return SessionClaimRegistry.TakeOverCommitDecision.RollBack;
+                    Volatile.Write(ref _msgSeqNum, 0);
+                    ResetBusinessAdmissionLocked();
+                    return commit();
+                }
+
                 if (!previous._retxBuffer.TryCopyRetainedFramesTo(_retxBuffer))
-                    return false;
+                    return SessionClaimRegistry.TakeOverCommitDecision.RollBack;
 
                 uint maxSeq = Math.Max(
                     (uint)Volatile.Read(ref _msgSeqNum),
@@ -879,7 +895,7 @@ internal bool TryAdoptOutboundStateForTakeOver(
                         _logger.LogWarning(ex,
                             "fixp session {ConnectionId} failed to reconcile outbound journal during takeover",
                             ConnectionId);
-                        return false;
+                        return SessionClaimRegistry.TakeOverCommitDecision.RollBack;
                     }
                 }
 
@@ -889,12 +905,49 @@ internal bool TryAdoptOutboundStateForTakeOver(
                 // passive ER or deferred completion that linearizes after the
                 // commit blocks before sequence allocation until EstablishAck.
                 ResetBusinessAdmissionLocked();
-                commit();
-                return true;
+                return commit();
             }
         }
     }
 
+    internal bool TryRollOutboundJournalGenerationForFreshTakeOver(uint sessionId)
+    {
+        if (_outboundJournal is null)
+            return true;
+
+        try
+        {
+            _outboundJournal.RollGeneration(sessionId);
+            return true;
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex,
+                "fixp session {ConnectionId} failed to roll outbound journal generation for sessionId={SessionId}",
+                ConnectionId, sessionId);
+            return false;
+        }
+    }
+
+    internal bool TryRestoreRolledOutboundJournalGenerationForTakeOver(uint sessionId)
+    {
+        if (_outboundJournal is null)
+            return true;
+
+        try
+        {
+            _outboundJournal.RestoreLatestGeneration(sessionId);
+            return true;
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex,
+                "fixp session {ConnectionId} failed to restore rolled outbound journal generation for sessionId={SessionId}",
+                ConnectionId, sessionId);
+            return false;
+        }
+    }
+
     internal bool TryResetOutboundStateForRetiredReplacement(
         FixpSession previous,
         uint sessionId)