From 44d20ad3f8595f40a00ba5b9bfd8c9cd12df4085 Mon Sep 17 00:00:00 2001 From: Pedro Sakuma Travi Date: Fri, 7 Aug 2026 03:31:53 +0000 Subject: [PATCH 1/2] fix(fixp): preserve outbound journal on suspended reap Closes #594 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/B3.Exchange.Gateway/CloseKind.cs | 6 +- .../FixpSession.Lifecycle.cs | 26 +- .../FixpSession.Negotiate.cs | 25 +- src/B3.Exchange.Gateway/FixpSession.cs | 25 ++ .../Persistence/IFixpOutboundJournal.cs | 18 +- .../EntryPointListenerReaperTests.cs | 243 ++++++++++++++++++ .../FixpSessionMassActionTakeoverTests.cs | 13 +- .../FixpSessionPersistenceWiringTests.cs | 21 +- 8 files changed, 344 insertions(+), 33 deletions(-) diff --git a/src/B3.Exchange.Gateway/CloseKind.cs b/src/B3.Exchange.Gateway/CloseKind.cs index 9d5473b7..e750cfd4 100644 --- a/src/B3.Exchange.Gateway/CloseKind.cs +++ b/src/B3.Exchange.Gateway/CloseKind.cs @@ -70,8 +70,10 @@ public enum CloseKind /// /// 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. /// SuspendedTimeout, diff --git a/src/B3.Exchange.Gateway/FixpSession.Lifecycle.cs b/src/B3.Exchange.Gateway/FixpSession.Lifecycle.cs index 7422a4b5..30fda8b8 100644 --- a/src/B3.Exchange.Gateway/FixpSession.Lifecycle.cs +++ b/src/B3.Exchange.Gateway/FixpSession.Lifecycle.cs @@ -418,23 +418,29 @@ 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.Remove(SessionId); } catch (Exception ex) diff --git a/src/B3.Exchange.Gateway/FixpSession.Negotiate.cs b/src/B3.Exchange.Gateway/FixpSession.Negotiate.cs index 034987e0..b05075d1 100644 --- a/src/B3.Exchange.Gateway/FixpSession.Negotiate.cs +++ b/src/B3.Exchange.Gateway/FixpSession.Negotiate.cs @@ -84,14 +84,29 @@ private NegotiateStep ProcessNegotiate(ReadOnlySpan fixedBlock, ReadOnlySp return NegotiateStep.Rejected(rejectFrame, $"negotiate-reject (legacy, ALREADY_NEGOTIATED, action={action})"); } + SessionId = req.SessionId; + EnteringFirm = req.EnteringFirm; + SessionVerId = req.SessionVerId; + if (!TrySeedOutboundSeqFromJournal(req.SessionId)) + { + State = FixpState.Idle; + SessionId = _acceptedSessionId; + SessionVerId = 0; + EnteringFirm = _acceptedEnteringFirm; + 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 state reconcile failed)"); + } // 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 +198,8 @@ _sessionRegistry is null } _claimedSessionId = req.SessionId; - if (!TryApplyPendingNegotiateState(req.SessionId)) + if (!TryApplyPendingNegotiateState(req.SessionId) + || !TrySeedOutboundSeqFromJournal(req.SessionId)) { if (evictedByTakeOver is not null) { @@ -204,6 +220,7 @@ _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, diff --git a/src/B3.Exchange.Gateway/FixpSession.cs b/src/B3.Exchange.Gateway/FixpSession.cs index b16eb529..e9c76a7c 100644 --- a/src/B3.Exchange.Gateway/FixpSession.cs +++ b/src/B3.Exchange.Gateway/FixpSession.cs @@ -724,6 +724,31 @@ private bool TryApplyPendingNegotiateState(uint sessionId) } } + private bool TrySeedOutboundSeqFromJournal(uint sessionId) + { + if (_outboundJournal is null) + return true; + + try + { + uint current = (uint)Volatile.Read(ref _msgSeqNum); + uint resumedSeq = Math.Max(current, _outboundJournal.MaxSeq(sessionId)); + if (resumedSeq != (uint)Volatile.Read(ref _msgSeqNum)) + { + Volatile.Write(ref _msgSeqNum, resumedSeq); + _pendingNegotiateStateApplied = true; + } + return true; + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "fixp session {ConnectionId} failed to reconcile outbound state for sessionId={SessionId}", + ConnectionId, sessionId); + return false; + } + } + private void RollbackPendingNegotiateState() { if (!_pendingNegotiateStateApplied) diff --git a/src/B3.Exchange.Gateway/Persistence/IFixpOutboundJournal.cs b/src/B3.Exchange.Gateway/Persistence/IFixpOutboundJournal.cs index 853ad02b..a6f002ff 100644 --- a/src/B3.Exchange.Gateway/Persistence/IFixpOutboundJournal.cs +++ b/src/B3.Exchange.Gateway/Persistence/IFixpOutboundJournal.cs @@ -40,11 +40,13 @@ namespace B3.Exchange.Gateway.Persistence; /// boundary for I/O efficiency. /// /// Terminal removal: is called by -/// FixpSession.CloseLocked only on terminal close events -/// (peer-initiated Terminate(Finished), suspended-timeout -/// reaper). Graceful host shutdown does NOT remove journals — they -/// must survive process restart so a reconnecting peer can recover -/// every event produced while the matching-platform was down. +/// FixpSession.CloseLocked only on terminal close events that +/// intentionally retire the durable replay history itself (for example +/// peer-initiated Terminate(Finished) or daily/session-local +/// teardown). Graceful host shutdown and suspended-timeout reaping do +/// NOT remove journals — they must survive so retention/size policy, +/// rather than wall-clock session-slot lifetime, remains the authority +/// on what replay data is still servable. /// /// Boot rehydration: enumerates /// every session whose journal survived the last process exit. The @@ -129,9 +131,9 @@ public interface IFixpOutboundJournal : IDisposable /// /// Terminal removal: deletes every artifact for /// . Idempotent — safe to call when - /// nothing exists for the session. Called only on peer-initiated - /// Terminate(Finished) and on suspended-timeout reaping. - /// Graceful host shutdown must NOT call this. + /// nothing exists for the session. Called only when the close path + /// intentionally retires the durable replay history itself; not on + /// graceful host shutdown or suspended-timeout reaping. /// void Remove(uint sessionId); diff --git a/tests/B3.Exchange.Gateway.Tests/EntryPointListenerReaperTests.cs b/tests/B3.Exchange.Gateway.Tests/EntryPointListenerReaperTests.cs index b0b91b92..f3e44ce0 100644 --- a/tests/B3.Exchange.Gateway.Tests/EntryPointListenerReaperTests.cs +++ b/tests/B3.Exchange.Gateway.Tests/EntryPointListenerReaperTests.cs @@ -1,7 +1,11 @@ using B3.Exchange.Contracts; using System.Net; using System.Net.Sockets; +using System.Buffers.Binary; +using System.Text; +using B3.EntryPoint.Wire; using B3.Exchange.Gateway; +using B3.Exchange.Gateway.Persistence; using B3.Exchange.Matching; using Microsoft.Extensions.Logging.Abstractions; @@ -30,6 +34,80 @@ public void OnDecodeError(B3.Exchange.Contracts.SessionId session, string error) public void OnSessionClosed(B3.Exchange.Contracts.SessionId session) => Interlocked.Increment(ref SessionClosedCalls); } + private sealed class FakeJournal : IFixpOutboundJournal + { + private readonly Dictionary> _data = new(); + public int RemoveCalls { get; private set; } + + public void Append(uint sessionId, uint seq, long timestampNanos, ReadOnlySpan frame) + { + if (!_data.TryGetValue(sessionId, out var session)) + _data[sessionId] = session = new(); + if (session.Count > 0 && seq <= session.Keys.Max()) + throw new InvalidOperationException( + $"journal append for session 0x{sessionId:x8} seq={seq} is not strictly greater than last persisted seq {session.Keys.Max()}"); + session[seq] = frame.ToArray(); + } + + public void ConfirmPeerAck(uint sessionId, uint uptoSeq) { } + + public IReadOnlyList ReadRange(uint sessionId, uint fromSeq, int count) + { + if (!_data.TryGetValue(sessionId, out var session)) + return Array.Empty(); + var list = new List(count); + for (int i = 0; i < count; i++) + { + uint seq = fromSeq + (uint)i; + if (!session.TryGetValue(seq, out var frame)) + break; + list.Add(new OutboundJournalEntry(seq, 0L, frame)); + } + return list; + } + + public void PruneUpTo(uint sessionId, uint uptoSeq) { } + + public uint MaxSeq(uint sessionId) + => _data.TryGetValue(sessionId, out var session) && session.Count > 0 + ? session.Keys.Max() + : 0u; + + public long EntryCount(uint sessionId) + => _data.TryGetValue(sessionId, out var session) ? session.Count : 0L; + + public void Remove(uint sessionId) + { + RemoveCalls++; + _data.Remove(sessionId); + } + + public IReadOnlyCollection ListSessions() => _data.Keys.ToArray(); + + public void Dispose() { } + } + + private sealed class FakeStatePersister : IFixpSessionStatePersister + { + private readonly Dictionary _data = new(); + public int RemoveCalls { get; private set; } + + public void Save(in FixpSessionStateSnapshot snapshot) => _data[snapshot.SessionId] = snapshot; + + public FixpSessionStateSnapshot? Load(uint sessionId) + => _data.TryGetValue(sessionId, out var snapshot) ? snapshot : null; + + public IReadOnlyCollection LoadAll() => _data.Values.ToArray(); + + public void Remove(uint sessionId) + { + RemoveCalls++; + _data.Remove(sessionId); + } + + public void Dispose() { } + } + /// /// Stand up a listener, accept one client, return the resulting /// driven to @@ -81,6 +159,86 @@ public void OnDecodeError(B3.Exchange.Contracts.SessionId session, string error) return (listener, sink, client, session, closures); } + private readonly record struct ReadFrame(ushort TemplateId, byte[] Body); + + private static async Task ConnectAndEstablishAsync( + EntryPointListener listener, + uint sessionId, + ulong sessionVerId) + { + var client = await ConnectAndSendNegotiateAsync(listener, sessionId, sessionVerId); + var stream = client.GetStream(); + var buffer = new byte[512]; + + Assert.Equal(EntryPointFrameReader.TidNegotiateResponse, + (await ReadOneFrameAsync(stream)).TemplateId); + + int length = EntryPointFixpFrameCodec.EncodeEstablish(buffer, + sessionId: sessionId, + sessionVerId: sessionVerId, + timestampNanos: 0, + keepAliveIntervalMillis: 10_000, + nextSeqNo: 1, + cancelOnDisconnectType: 0, + codTimeoutWindowMillis: 0, + credentials: ReadOnlySpan.Empty); + await stream.WriteAsync(buffer.AsMemory(0, length)); + Assert.Equal(EntryPointFrameReader.TidEstablishAck, + (await ReadOneFrameAsync(stream)).TemplateId); + return client; + } + + private static async Task ConnectAndSendNegotiateAsync( + EntryPointListener listener, + uint sessionId, + ulong sessionVerId) + { + var client = new TcpClient(); + await client.ConnectAsync(IPAddress.Loopback, listener.LocalEndpoint!.Port); + var credentials = Encoding.UTF8.GetBytes( + "{\"auth_type\":\"basic\",\"username\":\"1\",\"access_key\":\"\"}"); + var buffer = new byte[512]; + int length = EntryPointFixpFrameCodec.EncodeNegotiate(buffer, + sessionId: sessionId, + sessionVerId: sessionVerId, + timestampNanos: 0, + enteringFirm: 42, + onBehalfFirm: null, + credentials: credentials, + clientIp: ReadOnlySpan.Empty, + clientAppName: ReadOnlySpan.Empty, + clientAppVersion: ReadOnlySpan.Empty); + await client.GetStream().WriteAsync(buffer.AsMemory(0, length)); + return client; + } + + private static async Task ReadOneFrameAsync(NetworkStream stream) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + var header = new byte[EntryPointFrameReader.WireHeaderSize]; + await ReadExactAsync(stream, header, cts.Token); + ushort messageLength = BinaryPrimitives.ReadUInt16LittleEndian(header.AsSpan(0, 2)); + ushort templateId = BinaryPrimitives.ReadUInt16LittleEndian( + header.AsSpan(EntryPointFrameReader.SofhSize + 2, 2)); + var body = new byte[messageLength - EntryPointFrameReader.WireHeaderSize]; + await ReadExactAsync(stream, body, cts.Token); + return new ReadFrame(templateId, body); + } + + private static async Task ReadExactAsync( + NetworkStream stream, + byte[] buffer, + CancellationToken cancellationToken) + { + int read = 0; + while (read < buffer.Length) + { + int count = await stream.ReadAsync(buffer.AsMemory(read), cancellationToken); + if (count <= 0) throw new EndOfStreamException(); + read += count; + } + } + [Fact] public async Task Reaper_ClosesSession_WhenSuspendedLongerThanTimeout() { @@ -252,4 +410,89 @@ public async Task Reaper_increments_LifecycleMetrics_Reaped_counter() // Sanity: Rebound stays zero (we never re-attached). Assert.Equal(0, metrics.Rebound); } + + [Fact] + public async Task Reaper_preserves_outbound_journal_but_removes_state_snapshot() + { + const uint sessionId = 0x594u; + var sink = new NoOpEngineSink(); + var closures = new List(); + var journal = new FakeJournal(); + var state = new FakeStatePersister(); + journal.Append(sessionId, 1, 0L, new byte[] { 0x59, 0x40 }); + state.Save(new FixpSessionStateSnapshot( + SessionId: sessionId, + SessionVerId: 7UL, + OutboundMsgSeqNum: 1u, + LastIncomingSeqNo: 0u, + EnteringFirm: 11u, + UpdatedAtNanos: 0L)); + var options = new FixpSessionOptions + { + HeartbeatIntervalMs = 60_000, + IdleTimeoutMs = 60_000, + TestRequestGraceMs = 60_000, + SuspendedTimeoutMs = 200, + }; + + await using var listener = new EntryPointListener( + new IPEndPoint(IPAddress.Loopback, 0), + sink, + new SessionRegistry(), + NullLoggerFactory.Instance, + identityFactory: _ => new EntryPointListener.AcceptedConnection(1, EnteringFirm: 11, SessionId: sessionId), + sessionOptions: options, + onSessionClosed: (_, reason) => { lock (closures) closures.Add(reason); }, + outboundJournal: journal, + statePersister: state); + listener.Start(); + + using var client = new TcpClient(); + await client.ConnectAsync(IPAddress.Loopback, listener.LocalEndpoint!.Port); + var registered = await TestUtil.WaitUntilAsync( + () => listener.ActiveSessions.Count == 1, + TimeSpan.FromSeconds(2)); + Assert.True(registered, "listener never registered the accepted session"); + + var session = listener.ActiveSessions[0]; + session.ApplyTransition(FixpEvent.Negotiate); + session.ApplyTransition(FixpEvent.Establish); + client.Close(); + + var reaped = await TestUtil.WaitUntilAsync( + () => + { + lock (closures) + return !session.IsOpen + && session.SuspendedSinceMs is null + && closures.Count == 1; + }, + TimeSpan.FromSeconds(2)); + Assert.True(reaped, "reaper did not fully close the suspended session within 2s"); + + Assert.DoesNotContain(session, listener.ActiveSessions); + Assert.Equal(0, journal.RemoveCalls); + Assert.Equal(1L, journal.EntryCount(sessionId)); + Assert.Single(journal.ReadRange(sessionId, 1, 10)); + Assert.Equal(1, state.RemoveCalls); + Assert.Null(state.Load(sessionId)); + + using var resumedClient = await ConnectAndEstablishAsync(listener, sessionId, sessionVerId: 8); + var resumedRegistered = await TestUtil.WaitUntilAsync( + () => listener.ActiveSessions.Count == 1 && listener.ActiveSessions[0].IsOpen, + TimeSpan.FromSeconds(2)); + Assert.True(resumedRegistered, "listener never registered the replacement session"); + + var resumedSession = listener.ActiveSessions[0]; + var result = resumedSession.WriteOrderMassActionReport( + clOrdIdValue: 7003, + massActionResponse: OrderMassActionReportEncoder.MassActionResponseAccepted, + massActionRejectReason: null, + side: null, + securityId: 0, + transactTimeNanos: 2); + Assert.True(result.IsCommitted); + Assert.Equal(2u, resumedSession.OutboundSeq); + Assert.Equal(new uint[] { 1u, 2u }, journal.ReadRange(sessionId, 1, 10).Select(x => x.Seq).ToArray()); + } } diff --git a/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionTakeoverTests.cs b/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionTakeoverTests.cs index d20645f5..949cf5f4 100644 --- a/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionTakeoverTests.cs +++ b/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionTakeoverTests.cs @@ -408,19 +408,24 @@ public async Task ReplacementCloseBeforeSeal_RestoresVictimWithoutPersistingReje listener, sessionVerId: 3); await journal.MaxSeqEntered.Task.WaitAsync(TimeSpan.FromSeconds(5)); var replacement = listener.ActiveSessions.Single( - session => session.SessionVerId == 3); + session => !ReferenceEquals(session, oldSession)); var closeTask = Task.Run(() => replacement.Close("test-close-before-takeover-seal", CloseKind.TransportError)); Assert.True(await TestUtil.WaitUntilAsync( - () => !replacement.IsLiveTakeOverCandidate, + () => !replacement.IsRegistered + && replacement.LastCloseKind == CloseKind.TransportError, TimeSpan.FromSeconds(5))); journal.ReleaseMaxSeq(); await closeTask.WaitAsync(TimeSpan.FromSeconds(5)); Assert.True(await TestUtil.WaitUntilAsync( - () => listener.ActiveSessions.Contains(oldSession) - && !listener.ActiveSessions.Contains(replacement), + () => !replacement.IsOpen + && registry.TryGet(new SessionId("1"), out var currentSession) + && ReferenceEquals(currentSession, oldSession) + && claims.TryGetActiveClaim(1, out var claimHolder, out var claimVersion) + && ReferenceEquals(claimHolder, oldSession) + && claimVersion == 2UL, TimeSpan.FromSeconds(5))); Assert.True(registry.TryGet(new SessionId("1"), out var current)); Assert.Same(oldSession, current); diff --git a/tests/B3.Exchange.Gateway.Tests/FixpSessionPersistenceWiringTests.cs b/tests/B3.Exchange.Gateway.Tests/FixpSessionPersistenceWiringTests.cs index e48d5c3c..568d1201 100644 --- a/tests/B3.Exchange.Gateway.Tests/FixpSessionPersistenceWiringTests.cs +++ b/tests/B3.Exchange.Gateway.Tests/FixpSessionPersistenceWiringTests.cs @@ -19,9 +19,9 @@ namespace B3.Exchange.Gateway.Tests; /// SessionVerId / EnteringFirm / LastIncomingSeqNo and resumes /// the outbound seq at max(snapshot, journal.MaxSeq). /// CloseKind-gated removal: terminal kinds (PeerTerminate, -/// SuspendedTimeout) erase journal + state; preserving kinds -/// (HostShutdown, TransportError) keep them and save a final -/// snapshot. +/// DailyReset) erase journal + state; suspended-timeout only erases +/// the resumability snapshot; preserving kinds (HostShutdown, +/// TransportError) keep them and save a final snapshot. /// Establish/Suspend save the snapshot opportunistically. /// /// @@ -290,13 +290,21 @@ public async Task PeerTerminate_close_removes_journal_and_state() } [Fact] - public async Task SuspendedTimeout_close_removes_journal_and_state() + public async Task SuspendedTimeout_close_preserves_journal_but_removes_state() { var (listener, serverSide, client) = await ConnectPairAsync(); try { var journal = new FakeJournal(); var state = new FakeStatePersister(); + journal.Append(88, 1, 0L, new byte[] { 0x2A }); + state.Save(new FixpSessionStateSnapshot( + SessionId: 88, + SessionVerId: 1UL, + OutboundMsgSeqNum: 1u, + LastIncomingSeqNo: 0u, + EnteringFirm: 1u, + UpdatedAtNanos: 0L)); var session = new FixpSession( connectionId: 1, enteringFirm: 1, sessionId: 88, stream: serverSide, @@ -307,8 +315,11 @@ public async Task SuspendedTimeout_close_removes_journal_and_state() session.Close("idle reap", CloseKind.SuspendedTimeout); - Assert.Equal(1, journal.RemoveCalls); + Assert.Equal(0, journal.RemoveCalls); + Assert.Equal(1L, journal.EntryCount(88)); + Assert.Single(journal.ReadRange(88, 1, 10)); Assert.Equal(1, state.RemoveCalls); + Assert.Null(state.Load(88)); } finally { From 31c7f685b38a09c5d62c58d61f54345f6bc34663 Mon Sep 17 00:00:00 2001 From: Pedro Sakuma Travi Date: Fri, 7 Aug 2026 04:26:33 +0000 Subject: [PATCH 2/2] fix(fixp): harden journal-preserving session rollovers Closes #594 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/B3.Exchange.Gateway/EntryPointListener.cs | 25 ++ .../FixpSession.Lifecycle.cs | 13 + .../FixpSession.Negotiate.cs | 21 +- src/B3.Exchange.Gateway/FixpSession.cs | 112 +++++--- .../Persistence/FileFixpOutboundJournal.cs | 267 +++++++++++++++++- .../Persistence/IFixpOutboundJournal.cs | 40 +++ src/B3.Exchange.Gateway/SessionRegistry.cs | 36 ++- .../EntryPointListenerReaperTests.cs | 70 ++++- .../FixpSessionMassActionCompletionTests.cs | 4 + .../FixpSessionMassActionTakeoverTests.cs | 69 +++-- ...pSessionOutboundSequenceExhaustionTests.cs | 7 + .../FixpSessionPersistenceWiringTests.cs | 20 ++ .../FixpSessionResyncBootRehydrationTests.cs | 44 ++- .../FileFixpOutboundJournalTests.cs | 85 ++++++ .../SessionRegistryHandoffTests.cs | 4 + .../MassCancelPassiveCommitFailureTests.cs | 12 + 16 files changed, 712 insertions(+), 117 deletions(-) diff --git a/src/B3.Exchange.Gateway/EntryPointListener.cs b/src/B3.Exchange.Gateway/EntryPointListener.cs index 9d15be61..6a0c3d9a 100644 --- a/src/B3.Exchange.Gateway/EntryPointListener.cs +++ b/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 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); + } + } + } + /// /// Spec §4.5.1 (#GAP-09 / issue #47): at start of each trading day the /// gateway resets inbound + outbound MsgSeqNum counters to 1 and diff --git a/src/B3.Exchange.Gateway/FixpSession.Lifecycle.cs b/src/B3.Exchange.Gateway/FixpSession.Lifecycle.cs index 30fda8b8..0ca69e4e 100644 --- a/src/B3.Exchange.Gateway/FixpSession.Lifecycle.cs +++ b/src/B3.Exchange.Gateway/FixpSession.Lifecycle.cs @@ -437,6 +437,19 @@ private void CloseLocked(string reason, CloseKind kind) if (removeStateSnapshot) { _retxBuffer.Dispose(); + 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 diff --git a/src/B3.Exchange.Gateway/FixpSession.Negotiate.cs b/src/B3.Exchange.Gateway/FixpSession.Negotiate.cs index b05075d1..a23ecf6c 100644 --- a/src/B3.Exchange.Gateway/FixpSession.Negotiate.cs +++ b/src/B3.Exchange.Gateway/FixpSession.Negotiate.cs @@ -84,15 +84,9 @@ private NegotiateStep ProcessNegotiate(ReadOnlySpan fixedBlock, ReadOnlySp return NegotiateStep.Rejected(rejectFrame, $"negotiate-reject (legacy, ALREADY_NEGOTIATED, action={action})"); } - SessionId = req.SessionId; - EnteringFirm = req.EnteringFirm; - SessionVerId = req.SessionVerId; - if (!TrySeedOutboundSeqFromJournal(req.SessionId)) + if (!TryPrepareFreshNegotiateState(req.SessionId)) { State = FixpState.Idle; - SessionId = _acceptedSessionId; - SessionVerId = 0; - EnteringFirm = _acceptedEnteringFirm; RollbackPendingNegotiateState(); var rejectFrame = new byte[NegotiateRejectEncoder.Total]; NegotiateRejectEncoder.Encode(rejectFrame, req.SessionId, req.SessionVerId, @@ -100,8 +94,11 @@ private NegotiateStep ProcessNegotiate(ReadOnlySpan fixedBlock, ReadOnlySp B3.Entrypoint.Fixp.Sbe.V6.NegotiationRejectCode.UNSPECIFIED, currentSessionVerId: null); return NegotiateStep.Rejected(rejectFrame, - "negotiate-reject (legacy, outbound state reconcile failed)"); + "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, @@ -198,8 +195,8 @@ _sessionRegistry is null } _claimedSessionId = req.SessionId; - if (!TryApplyPendingNegotiateState(req.SessionId) - || !TrySeedOutboundSeqFromJournal(req.SessionId)) + if (evictedByTakeOver is null + && !TryPrepareFreshNegotiateState(req.SessionId)) { if (evictedByTakeOver is not null) { @@ -227,7 +224,7 @@ _sessionRegistry is 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; @@ -298,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 diff --git a/src/B3.Exchange.Gateway/FixpSession.cs b/src/B3.Exchange.Gateway/FixpSession.cs index e9c76a7c..bacbf5ff 100644 --- a/src/B3.Exchange.Gateway/FixpSession.cs +++ b/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,52 +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; - - 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; - return true; - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "fixp session {ConnectionId} failed to reconcile pending persisted state for sessionId={SessionId}", - ConnectionId, sessionId); - return false; - } - } - - private bool TrySeedOutboundSeqFromJournal(uint sessionId) - { if (_outboundJournal is null) return true; try { - uint current = (uint)Volatile.Read(ref _msgSeqNum); - uint resumedSeq = Math.Max(current, _outboundJournal.MaxSeq(sessionId)); - if (resumedSeq != (uint)Volatile.Read(ref _msgSeqNum)) - { - Volatile.Write(ref _msgSeqNum, resumedSeq); - _pendingNegotiateStateApplied = true; - } + _outboundJournal.RollGeneration(sessionId); return true; } catch (Exception ex) { _logger.LogWarning(ex, - "fixp session {ConnectionId} failed to reconcile outbound state for sessionId={SessionId}", + "fixp session {ConnectionId} failed to prepare fresh negotiate state for sessionId={SessionId}", ConnectionId, sessionId); return false; } @@ -870,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 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; @@ -886,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), @@ -904,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; } } @@ -914,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) diff --git a/src/B3.Exchange.Gateway/Persistence/FileFixpOutboundJournal.cs b/src/B3.Exchange.Gateway/Persistence/FileFixpOutboundJournal.cs index 2fddfb30..f5872acf 100644 --- a/src/B3.Exchange.Gateway/Persistence/FileFixpOutboundJournal.cs +++ b/src/B3.Exchange.Gateway/Persistence/FileFixpOutboundJournal.cs @@ -45,6 +45,7 @@ public sealed class FileFixpOutboundJournal : IFixpOutboundJournal { /// Subdirectory under the configured data directory. public const string JournalSubdir = "journal"; + private const string RetiredSubdir = "retired"; /// Default per-segment cap (16 MiB). public const int DefaultSegmentMaxBytes = 16 * 1024 * 1024; @@ -60,6 +61,7 @@ public sealed class FileFixpOutboundJournal : IFixpOutboundJournal private readonly object _lock = new(); private readonly Dictionary _active = new(); private readonly Dictionary _confirmedPeerAck = new(); + private readonly Dictionary> _retiredPeerAcks = new(); private bool _disposed; /// @@ -103,6 +105,9 @@ public FileFixpOutboundJournal(string dataDir, private string SessionDir(uint sessionId) => Path.Combine(_journalDir, $"session-{sessionId:x8}"); + private string RetiredSessionDir(uint sessionId) + => Path.Combine(_journalDir, RetiredSubdir, $"session-{sessionId:x8}"); + private static string SegmentFileName(uint firstSeq) => $"segment-{firstSeq:x8}.log"; @@ -162,14 +167,15 @@ public void Append(uint sessionId, uint seq, long timestampNanos, ReadOnlySpan _maxRetention.TotalSeconds; } - private void RotateSafeSegmentsLocked(uint sessionId, ActiveSegment active, + private void RotateSafeSegmentsLocked(uint sessionId, string? activePath, long nowNanos, string reason) { uint watermark = _confirmedPeerAck.TryGetValue(sessionId, out var ack) ? ack : 0u; if (watermark == 0) { - LogRotationBlocked(sessionId, reason, watermark, active, nowNanos); + LogRotationBlocked(sessionId, reason, watermark, activePath, nowNanos); return; } @@ -237,7 +243,8 @@ private void RotateSafeSegmentsLocked(uint sessionId, ActiveSegment active, : nextFirst - 1; if (segmentLastSeq == 0) continue; if (segmentLastSeq > watermark) break; - if (string.Equals(path, active.Path, StringComparison.Ordinal)) continue; + if (activePath is not null + && string.Equals(path, activePath, StringComparison.Ordinal)) continue; try { File.Delete(path); @@ -256,15 +263,18 @@ private void RotateSafeSegmentsLocked(uint sessionId, ActiveSegment active, } bool stillOver = reason == "bytes" - ? MaxBytesPerSession > 0 && CurrentSessionBytesLocked(sessionId, active) > MaxBytesPerSession + ? MaxBytesPerSession > 0 && CurrentSessionBytesLocked(sessionId, _active.GetValueOrDefault(sessionId)) > MaxBytesPerSession : IsRetentionExceededLocked(sessionId, nowNanos); if (deleted == 0 || stillOver) - LogRotationBlocked(sessionId, reason, watermark, active, nowNanos); + LogRotationBlocked(sessionId, reason, watermark, activePath, nowNanos); } private void LogRotationBlocked(uint sessionId, string reason, uint watermark, - ActiveSegment active, long nowNanos) + string? activePath, long nowNanos) { + ActiveSegment? active = null; + if (activePath is not null && _active.TryGetValue(sessionId, out var maybeActive)) + active = maybeActive; long bytes = CurrentSessionBytesLocked(sessionId, active); long oldestAge = OldestAgeSecondsLocked(sessionId, nowNanos); _metrics?.Observe(sessionId, bytes, oldestAge); @@ -273,19 +283,91 @@ private void LogRotationBlocked(uint sessionId, string reason, uint watermark, reason, sessionId, watermark, bytes, oldestAge); } - private void ObserveSessionLocked(uint sessionId, ActiveSegment active, long nowNanos) + private void TrimRetiredGenerationsForAgeLocked(uint sessionId, long nowNanos) + { + if (_maxRetention <= TimeSpan.Zero) return; + foreach (var retiredDir in EnumerateRetiredGenerationDirs(RetiredSessionDir(sessionId))) + { + long oldest = OldestTimestampNanos(retiredDir); + if (oldest == 0 || AgeSeconds(nowNanos, oldest) <= _maxRetention.TotalSeconds) + continue; + DeleteRetiredGeneration(sessionId, retiredDir, "age"); + } + } + + private void TrimRetiredGenerationsForBytesLocked(uint sessionId) + { + if (MaxBytesPerSession <= 0) return; + while (CurrentRetainedBytesLocked(sessionId) > MaxBytesPerSession) + { + var retiredDir = EnumerateRetiredGenerationDirs(RetiredSessionDir(sessionId)) + .FirstOrDefault(); + if (retiredDir is null) + break; + DeleteRetiredGeneration(sessionId, retiredDir, "bytes"); + } + } + + private long CurrentRetainedBytesLocked(uint sessionId) + { + long total = CurrentSessionBytesLocked(sessionId, _active.GetValueOrDefault(sessionId)); + foreach (var retiredDir in EnumerateRetiredGenerationDirs(RetiredSessionDir(sessionId))) + { + foreach (var path in Directory.EnumerateFiles(retiredDir, "segment-*.log")) + { + try { total += new FileInfo(path).Length; } + catch (FileNotFoundException) { } + } + } + return total; + } + + private void DeleteRetiredGeneration(uint sessionId, string retiredDir, string reason) + { + try + { + Directory.Delete(retiredDir, recursive: true); + var retiredRoot = Path.GetDirectoryName(retiredDir); + if (retiredRoot is not null + && Directory.Exists(retiredRoot) + && !Directory.EnumerateFileSystemEntries(retiredRoot).Any()) + { + Directory.Delete(retiredRoot); + } + _metrics?.IncRotation(sessionId, $"retired-{reason}"); + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "fixp outbound journal: failed to delete retired generation {Dir} (session=0x{SessionId:x8}, reason={Reason})", + retiredDir, sessionId, reason); + } + } + + private void ObserveSessionLocked(uint sessionId, ActiveSegment? active, long nowNanos) => _metrics?.Observe(sessionId, CurrentSessionBytesLocked(sessionId, active), OldestAgeSecondsLocked(sessionId, nowNanos)); private long OldestAgeSecondsLocked(uint sessionId, long nowNanos) { - var dir = SessionDir(sessionId); + long oldest = OldestTimestampNanos(SessionDir(sessionId)); + foreach (var retiredDir in EnumerateRetiredGenerationDirs(RetiredSessionDir(sessionId))) + { + long retiredOldest = OldestTimestampNanos(retiredDir); + if (retiredOldest != 0 && (oldest == 0 || retiredOldest < oldest)) + oldest = retiredOldest; + } + return oldest == 0 ? 0 : AgeSeconds(nowNanos, oldest); + } + + private long OldestTimestampNanos(string dir) + { if (!Directory.Exists(dir)) return 0; foreach (var (_, path) in EnumerateSegmentsOrdered(dir)) { var info = ScanSegment(path); if (info.EntryCount == 0) continue; - return AgeSeconds(nowNanos, info.FirstTimestampNanos); + return info.FirstTimestampNanos; } return 0; } @@ -505,6 +587,120 @@ public long EntryCount(uint sessionId) return total; } + public void ReleaseActive(uint sessionId) + { + lock (_lock) + { + ObjectDisposedException.ThrowIf(_disposed, this); + if (_active.Remove(sessionId, out var active)) + { + try { active.Dispose(); } catch { } + } + } + } + + public void RollGeneration(uint sessionId) + { + ReleaseActive(sessionId); + lock (_lock) + { + if (_confirmedPeerAck.Remove(sessionId, out var ack) && ack != 0) + { + if (!_retiredPeerAcks.TryGetValue(sessionId, out var retiredAcks)) + _retiredPeerAcks[sessionId] = retiredAcks = new(); + retiredAcks.Push(ack); + } + } + + var dir = SessionDir(sessionId); + if (!Directory.Exists(dir)) + { + _metrics?.Reset(sessionId); + return; + } + + var segments = Directory.EnumerateFiles(dir, "segment-*.log").ToArray(); + if (segments.Length == 0) + { + if (!Directory.EnumerateFileSystemEntries(dir).Any()) + Directory.Delete(dir); + _metrics?.Reset(sessionId); + return; + } + + var retiredRoot = RetiredSessionDir(sessionId); + Directory.CreateDirectory(retiredRoot); + string archiveDir = NextRetiredGenerationDir(retiredRoot); + DirectorySync.Fsync(retiredRoot); + Directory.Move(dir, archiveDir); + _metrics?.Reset(sessionId); + } + + public void RestoreLatestGeneration(uint sessionId) + { + ReleaseActive(sessionId); + + var retiredRoot = RetiredSessionDir(sessionId); + if (!Directory.Exists(retiredRoot)) + return; + + var latest = EnumerateRetiredGenerationDirs(retiredRoot).LastOrDefault(); + if (latest is null) + return; + + var dir = SessionDir(sessionId); + if (Directory.Exists(dir) && !Directory.EnumerateFileSystemEntries(dir).Any()) + Directory.Delete(dir); + Directory.Move(latest, dir); + if (!Directory.EnumerateFileSystemEntries(retiredRoot).Any()) + Directory.Delete(retiredRoot); + + lock (_lock) + { + if (_retiredPeerAcks.TryGetValue(sessionId, out var retiredAcks) + && retiredAcks.Count > 0) + { + uint ack = retiredAcks.Pop(); + if (ack != 0) + _confirmedPeerAck[sessionId] = ack; + if (retiredAcks.Count == 0) + _retiredPeerAcks.Remove(sessionId); + } + } + } + + public void EnforceRetention(uint sessionId, long nowNanos) + { + if (MaxBytesPerSession <= 0 && _maxRetention <= TimeSpan.Zero) + return; + + FlushActiveLocked(sessionId); + lock (_lock) + { + ObjectDisposedException.ThrowIf(_disposed, this); + _active.TryGetValue(sessionId, out var active); + TrimRetiredGenerationsForAgeLocked(sessionId, nowNanos); + TrimRetiredGenerationsForBytesLocked(sessionId); + + bool overBytes = MaxBytesPerSession > 0 + && CurrentSessionBytesLocked(sessionId, active) > MaxBytesPerSession; + bool overAge = IsRetentionExceededLocked(sessionId, nowNanos); + + if (!overBytes && !overAge) + { + ObserveSessionLocked(sessionId, active, nowNanos); + return; + } + + string? activePath = active?.Path; + if (overBytes) + RotateSafeSegmentsLocked(sessionId, activePath, nowNanos, "bytes"); + if (overAge) + RotateSafeSegmentsLocked(sessionId, activePath, nowNanos, "age"); + ObserveSessionLocked(sessionId, active, nowNanos); + } + } + public void Remove(uint sessionId) { lock (_lock) @@ -516,6 +712,7 @@ public void Remove(uint sessionId) } } var dir = SessionDir(sessionId); + var retiredRoot = RetiredSessionDir(sessionId); bool removed = !Directory.Exists(dir); if (!removed) { @@ -531,12 +728,24 @@ public void Remove(uint sessionId) dir); } } + if (Directory.Exists(retiredRoot)) + { + try { Directory.Delete(retiredRoot, recursive: true); } + catch (Exception ex) + { + _logger.LogWarning(ex, + "fixp outbound journal: failed to delete retired session directory {Dir}", + retiredRoot); + removed = false; + } + } if (!removed) return; lock (_lock) { _confirmedPeerAck.Remove(sessionId); + _retiredPeerAcks.Remove(sessionId); } _metrics?.Reset(sessionId); } @@ -554,9 +763,39 @@ public IReadOnlyCollection ListSessions() _logger.LogWarning( "fixp outbound journal: skipping unparseable session directory {Dir}", sub); } + var retiredDir = Path.Combine(_journalDir, RetiredSubdir); + if (!Directory.Exists(retiredDir)) return result; + foreach (var sub in Directory.EnumerateDirectories(retiredDir, "session-*")) + { + var name = Path.GetFileName(sub); + if (TryParseSessionDirName(name, out var sessionId) && !result.Contains(sessionId)) + result.Add(sessionId); + else if (!TryParseSessionDirName(name, out _)) + _logger.LogWarning( + "fixp outbound journal: skipping unparseable retired session directory {Dir}", sub); + } return result; } + private static string NextRetiredGenerationDir(string retiredRoot) + { + long stamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + while (true) + { + string candidate = Path.Combine(retiredRoot, $"generation-{stamp:x16}"); + if (!Directory.Exists(candidate)) return candidate; + stamp++; + } + } + + private static IReadOnlyList EnumerateRetiredGenerationDirs(string retiredRoot) + { + if (!Directory.Exists(retiredRoot)) return Array.Empty(); + return Directory.EnumerateDirectories(retiredRoot, "generation-*") + .OrderBy(path => path, StringComparer.Ordinal) + .ToArray(); + } + private static bool TryParseSessionDirName(string dirName, out uint sessionId) { sessionId = 0; diff --git a/src/B3.Exchange.Gateway/Persistence/IFixpOutboundJournal.cs b/src/B3.Exchange.Gateway/Persistence/IFixpOutboundJournal.cs index a6f002ff..bee14bbc 100644 --- a/src/B3.Exchange.Gateway/Persistence/IFixpOutboundJournal.cs +++ b/src/B3.Exchange.Gateway/Persistence/IFixpOutboundJournal.cs @@ -48,6 +48,12 @@ namespace B3.Exchange.Gateway.Persistence; /// rather than wall-clock session-slot lifetime, remains the authority /// on what replay data is still servable. /// +/// Fresh session-version rollover: a higher +/// sessionVerId accepted via a new Negotiate starts a fresh +/// FIXP sequence generation. Implementations therefore need a way to +/// preserve the prior durable bytes without letting them collide with the +/// new generation's MsgSeqNum space; see . +/// /// Boot rehydration: enumerates /// every session whose journal survived the last process exit. The /// host calls this once at boot, then for each session derives the @@ -128,6 +134,40 @@ public interface IFixpOutboundJournal : IDisposable /// long EntryCount(uint sessionId); + /// + /// Releases any active writer state for + /// without deleting retained journal bytes. Used when a logical session + /// is fully reaped but its durable replay history must remain subject to + /// normal retention policy. + /// + void ReleaseActive(uint sessionId); + + /// + /// Preserves any currently-retained journal bytes for + /// as a retired generation and resets the + /// active append path so a fresh higher-sessionVerId Negotiate + /// can start its outbound sequence from 1 without colliding with prior + /// retained bytes. Idempotent; safe when the session has no journal. + /// + void RollGeneration(uint sessionId); + + /// + /// Restores the most recently retired generation for + /// back to the active append path. Used to + /// roll back a failed higher-sessionVerId takeover after the + /// journal was already rolled forward but the new durable session state + /// could not be committed. + /// + void RestoreLatestGeneration(uint sessionId); + + /// + /// Applies the journal's configured quota / age retention policy to + /// outside the hot append path. Used by + /// background maintenance for abandoned sessions whose journals may + /// never see another append. + /// + void EnforceRetention(uint sessionId, long nowNanos); + /// /// Terminal removal: deletes every artifact for /// . Idempotent — safe to call when diff --git a/src/B3.Exchange.Gateway/SessionRegistry.cs b/src/B3.Exchange.Gateway/SessionRegistry.cs index 17c22920..697ef6e6 100644 --- a/src/B3.Exchange.Gateway/SessionRegistry.cs +++ b/src/B3.Exchange.Gateway/SessionRegistry.cs @@ -366,23 +366,38 @@ internal SessionClaimRegistry.TakeOverFinalizeResult TryCommitTakeOver( return SessionClaimRegistry.TakeOverCommitDecision.RollBack; } - var decision = SessionClaimRegistry.TakeOverCommitDecision.RollBack; - if (!replacement.TryAdoptOutboundStateForTakeOver(previous, () => + bool freshGeneration = replacement.SessionVerId > previous.SessionVerId; + var decision = replacement.TryFinalizeOutboundStateForTakeOver(previous, + freshGeneration, () => { if (!ReferenceEquals(previousRoute.Current, previous) || !ReferenceEquals(replacementRoute.Current, replacement)) - return; + return SessionClaimRegistry.TakeOverCommitDecision.RollBack; if (!replacement.TrySealTakeOverCandidate()) - return; + return SessionClaimRegistry.TakeOverCommitDecision.RollBack; - if (!replacement.TrySaveStateSnapshot()) + if (freshGeneration + && !replacement.TryRollOutboundJournalGenerationForFreshTakeOver( + replacement.SessionId)) { bool rollbackPersisted = previous.TrySaveStateSnapshot(); replacement.RollbackTakeOverSeal(); - decision = rollbackPersisted + return rollbackPersisted + ? SessionClaimRegistry.TakeOverCommitDecision.RollBack + : SessionClaimRegistry.TakeOverCommitDecision.FailClosed; + } + + if (!replacement.TrySaveStateSnapshot()) + { + bool journalRestored = + !freshGeneration + || replacement.TryRestoreRolledOutboundJournalGenerationForTakeOver( + replacement.SessionId); + bool rollbackPersisted = journalRestored && previous.TrySaveStateSnapshot(); + replacement.RollbackTakeOverSeal(); + return rollbackPersisted ? SessionClaimRegistry.TakeOverCommitDecision.RollBack : SessionClaimRegistry.TakeOverCommitDecision.FailClosed; - return; } previousRoute.SetCurrent(replacement); @@ -392,11 +407,8 @@ internal SessionClaimRegistry.TakeOverFinalizeResult TryCommitTakeOver( { _sessions[identity] = replacement; } - decision = SessionClaimRegistry.TakeOverCommitDecision.Commit; - })) - { - return SessionClaimRegistry.TakeOverCommitDecision.RollBack; - } + return SessionClaimRegistry.TakeOverCommitDecision.Commit; + }); return decision; }); diff --git a/tests/B3.Exchange.Gateway.Tests/EntryPointListenerReaperTests.cs b/tests/B3.Exchange.Gateway.Tests/EntryPointListenerReaperTests.cs index f3e44ce0..e624ba33 100644 --- a/tests/B3.Exchange.Gateway.Tests/EntryPointListenerReaperTests.cs +++ b/tests/B3.Exchange.Gateway.Tests/EntryPointListenerReaperTests.cs @@ -37,6 +37,7 @@ public void OnDecodeError(B3.Exchange.Contracts.SessionId session, string error) private sealed class FakeJournal : IFixpOutboundJournal { private readonly Dictionary> _data = new(); + private readonly Dictionary>> _retired = new(); public int RemoveCalls { get; private set; } public void Append(uint sessionId, uint seq, long timestampNanos, ReadOnlySpan frame) @@ -76,6 +77,34 @@ public uint MaxSeq(uint sessionId) public long EntryCount(uint sessionId) => _data.TryGetValue(sessionId, out var session) ? session.Count : 0L; + public IReadOnlyList RetiredSequences(uint sessionId) + => _retired.TryGetValue(sessionId, out var retired) + ? retired.Select(entries => entries.Keys.ToArray()).ToArray() + : Array.Empty(); + + public void RollGeneration(uint sessionId) + { + if (_data.Remove(sessionId, out var session)) + { + if (!_retired.TryGetValue(sessionId, out var retired)) + _retired[sessionId] = retired = new(); + retired.Add(session); + } + } + + public void RestoreLatestGeneration(uint sessionId) + { + if (_retired.TryGetValue(sessionId, out var retired) && retired.Count > 0) + { + _data[sessionId] = retired[^1]; + retired.RemoveAt(retired.Count - 1); + } + } + + public void ReleaseActive(uint sessionId) { } + + public void EnforceRetention(uint sessionId, long nowNanos) { } + public void Remove(uint sessionId) { RemoveCalls++; @@ -492,7 +521,44 @@ public async Task Reaper_preserves_outbound_journal_but_removes_state_snapshot() securityId: 0, transactTimeNanos: 2); Assert.True(result.IsCommitted); - Assert.Equal(2u, resumedSession.OutboundSeq); - Assert.Equal(new uint[] { 1u, 2u }, journal.ReadRange(sessionId, 1, 10).Select(x => x.Seq).ToArray()); + Assert.Equal(1u, resumedSession.OutboundSeq); + Assert.Equal(new uint[] { 1u }, journal.ReadRange(sessionId, 1, 10).Select(x => x.Seq).ToArray()); + Assert.Equal(new uint[] { 1u }, Assert.Single(journal.RetiredSequences(sessionId))); + } + + [Fact] + public async Task RetentionSweep_prunes_expired_abandoned_journal_without_future_append() + { + string dir = Path.Combine(Path.GetTempPath(), "fixp-reaper-retention-" + Guid.NewGuid().ToString("n")); + Directory.CreateDirectory(dir); + try + { + using var journal = new FileFixpOutboundJournal( + dir, + NullLogger.Instance, + segmentMaxBytes: 220, + maxRetention: TimeSpan.FromHours(1)); + const uint sessionId = 0x596u; + journal.Append(sessionId, 1, 1_000_000_000L, new byte[100]); + journal.Append(sessionId, 2, 2_000_000_000L, new byte[100]); + journal.ConfirmPeerAck(sessionId, 2); + + await using var listener = new EntryPointListener( + new IPEndPoint(IPAddress.Loopback, 0), + new NoOpEngineSink(), + new SessionRegistry(), + NullLoggerFactory.Instance, + sessionOptions: new FixpSessionOptions { SuspendedTimeoutMs = 200 }, + outboundJournal: journal); + + listener.SweepOutboundJournalRetentionOnce(8_000_000_000_000L); + + Assert.Empty(journal.ReadRange(sessionId, 1, 10)); + Assert.Equal(new uint[] { 2u }, journal.ReadRange(sessionId, 2, 10).Select(x => x.Seq).ToArray()); + } + finally + { + try { Directory.Delete(dir, recursive: true); } catch { } + } } } diff --git a/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionCompletionTests.cs b/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionCompletionTests.cs index 95e68a2b..6e9d9a90 100644 --- a/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionCompletionTests.cs +++ b/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionCompletionTests.cs @@ -52,6 +52,10 @@ public IReadOnlyList ReadRange(uint sessionId, uint fromSe => Entries.Where(entry => entry.Seq >= fromSeq).Take(count).ToArray(); public uint MaxSeq(uint sessionId) => Entries.Count == 0 ? 0u : Entries.Max(entry => entry.Seq); public long EntryCount(uint sessionId) => Entries.Count; + public void RollGeneration(uint sessionId) => Entries.Clear(); + public void RestoreLatestGeneration(uint sessionId) { } + public void ReleaseActive(uint sessionId) { } + public void EnforceRetention(uint sessionId, long nowNanos) { } public void Remove(uint sessionId) => Entries.Clear(); public IReadOnlyCollection ListSessions() => Array.Empty(); public void Dispose() { } diff --git a/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionTakeoverTests.cs b/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionTakeoverTests.cs index 949cf5f4..17c0be50 100644 --- a/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionTakeoverTests.cs +++ b/tests/B3.Exchange.Gateway.Tests/FixpSessionMassActionTakeoverTests.cs @@ -38,6 +38,7 @@ private sealed class StrictJournal : IFixpOutboundJournal { private readonly object _lock = new(); private readonly SortedDictionary _entries = new(); + private readonly List> _retired = new(); private readonly ManualResetEventSlim _releaseMaxSeq = new(false); public IReadOnlyList Entries @@ -45,7 +46,17 @@ public IReadOnlyList Entries get { lock (_lock) return _entries.Values.ToArray(); } } + public IReadOnlyList RetiredSequences + { + get + { + lock (_lock) + return _retired.Select(entries => entries.Keys.ToArray()).ToArray(); + } + } + public bool BlockMaxSeq { get; set; } + public bool FailRollGeneration { get; set; } public TaskCompletionSource MaxSeqEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -84,6 +95,30 @@ public IReadOnlyList ReadRange(uint sessionId, uint fromSe .ToArray(); } } + public void RollGeneration(uint sessionId) + { + lock (_lock) + { + if (FailRollGeneration) + throw new IOException("simulated journal rollover failure"); + if (_entries.Count == 0) return; + _retired.Add(new SortedDictionary(_entries)); + _entries.Clear(); + } + } + public void RestoreLatestGeneration(uint sessionId) + { + lock (_lock) + { + if (_retired.Count == 0) return; + _entries.Clear(); + foreach (var entry in _retired[^1]) + _entries[entry.Key] = entry.Value; + _retired.RemoveAt(_retired.Count - 1); + } + } + public void ReleaseActive(uint sessionId) { } + public void EnforceRetention(uint sessionId, long nowNanos) { } public void Remove(uint sessionId) { lock (_lock) _entries.Clear(); @@ -321,19 +356,16 @@ public async Task PersistenceEnabledTakeover_ContinuesSequenceAcrossCancelAndTer var report = await ReadOneFrameAsync(replacementClient.GetStream()); Assert.Equal(EntryPointFrameReader.TidOrderMassActionReport, report.TemplateId); - Assert.Equal(2u, BinaryPrimitives.ReadUInt32LittleEndian(report.Body.AsSpan(4, 4))); + Assert.Equal(1u, BinaryPrimitives.ReadUInt32LittleEndian(report.Body.AsSpan(4, 4))); var entries = journal.Entries; - Assert.Equal(new uint[] { 1, 2 }, entries.Select(entry => entry.Seq)); + Assert.Equal(new uint[] { 1 }, entries.Select(entry => entry.Seq)); Assert.Equal( - new[] - { - EntryPointFrameReader.TidExecutionReportCancel, - EntryPointFrameReader.TidOrderMassActionReport, - }, + new[] { EntryPointFrameReader.TidOrderMassActionReport }, entries.Select(entry => BinaryPrimitives.ReadUInt16LittleEndian( entry.Frame.AsSpan(EntryPointFrameReader.SofhSize + 2, 2)))); - Assert.Equal(2u, listener.ActiveSessions.Single( + Assert.Equal(new uint[] { 1 }, Assert.Single(journal.RetiredSequences)); + Assert.Equal(1u, listener.ActiveSessions.Single( session => session.SessionVerId == 3).OutboundSeq); } @@ -388,7 +420,7 @@ public async Task ReplacementCloseAfterSeal_CommitsDurableVersionAndNeverRestore } [Fact] - public async Task ReplacementCloseBeforeSeal_RestoresVictimWithoutPersistingRejectedVersion() + public async Task TakeoverJournalRolloverFailure_RestoresVictimWithoutPersistingRejectedVersion() { var sink = new ControlledSink(); var registry = new SessionRegistry(); @@ -402,26 +434,15 @@ public async Task ReplacementCloseBeforeSeal_RestoresVictimWithoutPersistingReje using var oldClient = await ConnectAndEstablishAsync(listener, sessionVerId: 2); var oldSession = listener.ActiveSessions.Single( session => session.SessionVerId == 2); - journal.BlockMaxSeq = true; + journal.FailRollGeneration = true; using var replacementClient = await ConnectAndSendNegotiateAsync( listener, sessionVerId: 3); - await journal.MaxSeqEntered.Task.WaitAsync(TimeSpan.FromSeconds(5)); - var replacement = listener.ActiveSessions.Single( - session => !ReferenceEquals(session, oldSession)); - var closeTask = Task.Run(() => - replacement.Close("test-close-before-takeover-seal", CloseKind.TransportError)); - Assert.True(await TestUtil.WaitUntilAsync( - () => !replacement.IsRegistered - && replacement.LastCloseKind == CloseKind.TransportError, - TimeSpan.FromSeconds(5))); - - journal.ReleaseMaxSeq(); - await closeTask.WaitAsync(TimeSpan.FromSeconds(5)); + var reject = await ReadOneFrameAsync(replacementClient.GetStream()); + Assert.Equal(EntryPointFrameReader.TidNegotiateReject, reject.TemplateId); Assert.True(await TestUtil.WaitUntilAsync( - () => !replacement.IsOpen - && registry.TryGet(new SessionId("1"), out var currentSession) + () => registry.TryGet(new SessionId("1"), out var currentSession) && ReferenceEquals(currentSession, oldSession) && claims.TryGetActiveClaim(1, out var claimHolder, out var claimVersion) && ReferenceEquals(claimHolder, oldSession) diff --git a/tests/B3.Exchange.Gateway.Tests/FixpSessionOutboundSequenceExhaustionTests.cs b/tests/B3.Exchange.Gateway.Tests/FixpSessionOutboundSequenceExhaustionTests.cs index 312a46ec..9baafa8d 100644 --- a/tests/B3.Exchange.Gateway.Tests/FixpSessionOutboundSequenceExhaustionTests.cs +++ b/tests/B3.Exchange.Gateway.Tests/FixpSessionOutboundSequenceExhaustionTests.cs @@ -119,6 +119,13 @@ public long EntryCount(uint sessionId) ? sessionEntries.Count : 0; + public void RollGeneration(uint sessionId) + => _entries.Remove(sessionId); + public void RestoreLatestGeneration(uint sessionId) { } + public void ReleaseActive(uint sessionId) { } + + public void EnforceRetention(uint sessionId, long nowNanos) { } + public void Remove(uint sessionId) { RemoveCalls++; diff --git a/tests/B3.Exchange.Gateway.Tests/FixpSessionPersistenceWiringTests.cs b/tests/B3.Exchange.Gateway.Tests/FixpSessionPersistenceWiringTests.cs index 568d1201..8ede5f20 100644 --- a/tests/B3.Exchange.Gateway.Tests/FixpSessionPersistenceWiringTests.cs +++ b/tests/B3.Exchange.Gateway.Tests/FixpSessionPersistenceWiringTests.cs @@ -41,6 +41,7 @@ public void OnSessionClosed(SessionId s) { } private sealed class FakeJournal : IFixpOutboundJournal { private readonly Dictionary> _data = new(); + private readonly Dictionary>> _retired = new(); public int RemoveCalls { get; private set; } public void Append(uint sessionId, uint seq, long ts, ReadOnlySpan frame) { @@ -66,6 +67,25 @@ public uint MaxSeq(uint sessionId) => _data.TryGetValue(sessionId, out var s) && s.Count > 0 ? s.Keys.Max() : 0u; public long EntryCount(uint sessionId) => _data.TryGetValue(sessionId, out var s) ? s.Count : 0; + public void RollGeneration(uint sessionId) + { + if (_data.Remove(sessionId, out var entries)) + { + if (!_retired.TryGetValue(sessionId, out var retired)) + _retired[sessionId] = retired = new(); + retired.Add(entries); + } + } + public void RestoreLatestGeneration(uint sessionId) + { + if (_retired.TryGetValue(sessionId, out var retired) && retired.Count > 0) + { + _data[sessionId] = retired[^1]; + retired.RemoveAt(retired.Count - 1); + } + } + public void ReleaseActive(uint sessionId) { } + public void EnforceRetention(uint sessionId, long nowNanos) { } public void Remove(uint sessionId) { RemoveCalls++; diff --git a/tests/B3.Exchange.Gateway.Tests/FixpSessionResyncBootRehydrationTests.cs b/tests/B3.Exchange.Gateway.Tests/FixpSessionResyncBootRehydrationTests.cs index 807b7d05..15d58d11 100644 --- a/tests/B3.Exchange.Gateway.Tests/FixpSessionResyncBootRehydrationTests.cs +++ b/tests/B3.Exchange.Gateway.Tests/FixpSessionResyncBootRehydrationTests.cs @@ -16,8 +16,9 @@ namespace B3.Exchange.Gateway.Tests; /// Issue #405 — boot rehydration end-to-end. Validates that an /// wired with a persisted outbound /// journal + session-state snapshots resumes a previously persisted -/// session through matching Establish or a successfully claimed -/// higher-version Negotiate. Negotiate candidates remain pending until +/// session through matching Establish, while a successfully claimed +/// higher-version Negotiate starts a fresh sequence generation after +/// retiring the old journal. Negotiate candidates remain pending until /// their claim commits. /// public class FixpSessionResyncBootRehydrationTests @@ -71,6 +72,13 @@ public long EntryCount(uint sessionId) lock (_lock) return _sessions.TryGetValue(sessionId, out var entries) ? entries.Count : 0; } + public void RollGeneration(uint sessionId) + { + lock (_lock) _sessions.Remove(sessionId); + } + public void RestoreLatestGeneration(uint sessionId) { } + public void ReleaseActive(uint sessionId) { } + public void EnforceRetention(uint sessionId, long nowNanos) { } public void Remove(uint sessionId) { lock (_lock) _sessions.Remove(sessionId); @@ -102,6 +110,17 @@ public uint MaxSeq(uint sessionId) return _inner.MaxSeq(sessionId); } public long EntryCount(uint sessionId) => _inner.EntryCount(sessionId); + public void RollGeneration(uint sessionId) + { + if (Interlocked.Exchange(ref _failuresRemaining, 0) == 1) + throw new IOException("simulated journal rollover failure"); + _inner.RollGeneration(sessionId); + } + public void RestoreLatestGeneration(uint sessionId) + => _inner.RestoreLatestGeneration(sessionId); + public void ReleaseActive(uint sessionId) => _inner.ReleaseActive(sessionId); + public void EnforceRetention(uint sessionId, long nowNanos) + => _inner.EnforceRetention(sessionId, nowNanos); public void Remove(uint sessionId) => _inner.Remove(sessionId); public IReadOnlyCollection ListSessions() => _inner.ListSessions(); public void Dispose() => _inner.Dispose(); @@ -406,10 +425,11 @@ public async Task ReconnectingNegotiate_AdoptsPersistedEnvelopeOnlyAfterClaim() $"rehydrated session should register (have {listener.ActiveSessions.Count})"); // The higher-version candidate remained pending until its claim - // succeeded, then adopted the recoverable sequence envelope. + // succeeded, then started a fresh FIXP generation rather than + // inheriting the prior SessionVerId's sequence envelope. var session = listener.ActiveSessions.Single(s => s.SessionId == 1); - Assert.Equal(3u, session.OutboundSeq); - Assert.Equal(2u, session.LastIncomingSeqNo); + Assert.Equal(0u, session.OutboundSeq); + Assert.Equal(0u, session.LastIncomingSeqNo); Assert.Equal(42u, session.EnteringFirm); } finally @@ -784,8 +804,8 @@ await replacementClient.GetStream().WriteAsync(BuildNegotiate( TimeSpan.FromSeconds(5))); Assert.True(registry.TryGet(new B3.Exchange.Contracts.SessionId("1"), out var replacement)); - Assert.Equal(3u, replacement.OutboundSeq); - Assert.Equal(3u, journal.MaxSeq(1)); + Assert.Equal(0u, replacement.OutboundSeq); + Assert.Equal(0u, journal.MaxSeq(1)); } [Theory] @@ -960,22 +980,22 @@ public async Task ReconnectingNegotiate_BlocksRoutedPassiveReportUntilEstablishA var session = listener.ActiveSessions.Single(s => s.SessionVerId == 101UL); Assert.True(routed.IsDeferred); Assert.Equal(1, registry.PendingWriteCount(session)); - Assert.Equal(3u, journal.MaxSeq(1)); + Assert.Equal(0u, journal.MaxSeq(1)); await AssertNoFrameAsync(stream); await stream.WriteAsync(BuildEstablish( - sessionId: 1, sessionVerId: 101UL, nextSeqNo: 3u)); + sessionId: 1, sessionVerId: 101UL, nextSeqNo: 1u)); var ack = await ReadFrameAsync(stream); Assert.Equal(EntryPointFrameReader.TidEstablishAck, ack.TemplateId); - Assert.Equal(4u, + Assert.Equal(1u, BinaryPrimitives.ReadUInt32LittleEndian(ack.Body.AsSpan(28, 4))); Assert.True((await routed.Completion.WaitAsync( TimeSpan.FromSeconds(5))).IsTransportEnqueued); var report = await ReadFrameAsync(stream); Assert.Equal(EntryPointFrameReader.TidExecutionReportCancel, report.TemplateId); - Assert.Equal(4u, report.MsgSeqNum); - Assert.Equal(4u, journal.MaxSeq(1)); + Assert.Equal(1u, report.MsgSeqNum); + Assert.Equal(1u, journal.MaxSeq(1)); } [Fact] diff --git a/tests/B3.Exchange.Gateway.Tests/Persistence/FileFixpOutboundJournalTests.cs b/tests/B3.Exchange.Gateway.Tests/Persistence/FileFixpOutboundJournalTests.cs index f02c5976..ef5499b2 100644 --- a/tests/B3.Exchange.Gateway.Tests/Persistence/FileFixpOutboundJournalTests.cs +++ b/tests/B3.Exchange.Gateway.Tests/Persistence/FileFixpOutboundJournalTests.cs @@ -342,6 +342,91 @@ public void Age_quota_triggers_rotation_below_peer_ack_watermark() Assert.True(snap.RotationsAge >= 1); } + [Fact] + public void EnforceRetention_applies_age_policy_without_future_append() + { + using var j = NewJournal(segmentMaxBytes: 220, maxRetention: TimeSpan.FromHours(1)); + const uint sid = 0x434u; + j.Append(sid, 1, 1_000_000_000L, Frame(1, extraBytes: 96)); + j.Append(sid, 2, 2_000_000_000L, Frame(2, extraBytes: 96)); + j.ConfirmPeerAck(sid, 2); + + j.EnforceRetention(sid, 8_000_000_000_000L); + + Assert.Empty(j.ReadRange(sid, 1, 10)); + Assert.Equal(new uint[] { 2 }, j.ReadRange(sid, 2, 10).Select(e => e.Seq).ToArray()); + } + + [Fact] + public void ReleaseActive_allows_retention_to_prune_expired_only_segment() + { + using var j = NewJournal(segmentMaxBytes: 1_024, maxRetention: TimeSpan.FromHours(1)); + const uint sid = 0x436u; + j.Append(sid, 1, 1_000_000_000L, Frame(1, extraBytes: 64)); + j.Append(sid, 2, 2_000_000_000L, Frame(2, extraBytes: 64)); + j.ConfirmPeerAck(sid, 2); + + j.ReleaseActive(sid); + j.EnforceRetention(sid, 8_000_000_000_000L); + + Assert.Equal(0u, j.MaxSeq(sid)); + Assert.Equal(0L, j.EntryCount(sid)); + Assert.Empty(j.ReadRange(sid, 1, 10)); + } + + [Fact] + public void Retired_generation_is_listed_and_pruned_by_retention() + { + using var j = NewJournal(maxRetention: TimeSpan.FromHours(1)); + const uint sid = 0x437u; + j.Append(sid, 1, 1_000_000_000L, Frame(1, extraBytes: 64)); + j.Append(sid, 2, 2_000_000_000L, Frame(2, extraBytes: 64)); + + j.RollGeneration(sid); + Assert.Contains(sid, j.ListSessions()); + + j.EnforceRetention(sid, 8_000_000_000_000L); + + Assert.DoesNotContain(sid, j.ListSessions()); + } + + [Fact] + public void RestoreLatestGeneration_restores_rolled_active_sequence_space() + { + using var j = NewJournal(); + const uint sid = 0x438u; + j.Append(sid, 1, 1_000_000_000L, Frame(1)); + j.Append(sid, 2, 2_000_000_000L, Frame(2)); + j.RollGeneration(sid); + + j.RestoreLatestGeneration(sid); + + Assert.Equal(2u, j.MaxSeq(sid)); + Assert.Equal(new uint[] { 1, 2 }, j.ReadRange(sid, 1, 10).Select(e => e.Seq).ToArray()); + } + + [Fact] + public void RollGeneration_preserves_retired_bytes_and_resets_active_sequence_space() + { + using var j = NewJournal(); + const uint sid = 0x435u; + j.Append(sid, 1, 1_000_000_000L, Frame(1)); + j.Append(sid, 2, 2_000_000_000L, Frame(2)); + + j.RollGeneration(sid); + + Assert.Equal(0u, j.MaxSeq(sid)); + Assert.Equal(0L, j.EntryCount(sid)); + + var retiredDir = Path.Combine(_root, FileFixpOutboundJournal.JournalSubdir, + "retired", $"session-{sid:x8}"); + Assert.True(Directory.Exists(retiredDir)); + Assert.True(Directory.EnumerateFiles(retiredDir, "segment-*.log", SearchOption.AllDirectories).Any()); + + j.Append(sid, 1, 3_000_000_000L, Frame(1)); + Assert.Equal(1u, j.MaxSeq(sid)); + } + [Fact] public void Rotation_never_drops_entries_above_peer_confirmed_sequence() { diff --git a/tests/B3.Exchange.Gateway.Tests/SessionRegistryHandoffTests.cs b/tests/B3.Exchange.Gateway.Tests/SessionRegistryHandoffTests.cs index b189e6d6..a61365a6 100644 --- a/tests/B3.Exchange.Gateway.Tests/SessionRegistryHandoffTests.cs +++ b/tests/B3.Exchange.Gateway.Tests/SessionRegistryHandoffTests.cs @@ -41,6 +41,10 @@ public IReadOnlyList ReadRange(uint sessionId, uint fromSe => _entries.Values.Where(entry => entry.Seq >= fromSeq).Take(count).ToArray(); public uint MaxSeq(uint sessionId) => _entries.Count == 0 ? 0u : _entries.Keys.Max(); public long EntryCount(uint sessionId) => _entries.Count; + public void RollGeneration(uint sessionId) => _entries.Clear(); + public void RestoreLatestGeneration(uint sessionId) { } + public void ReleaseActive(uint sessionId) { } + public void EnforceRetention(uint sessionId, long nowNanos) { } public void Remove(uint sessionId) => _entries.Clear(); public IReadOnlyCollection ListSessions() => _entries.Count == 0 ? Array.Empty() : new[] { 1u }; public void Dispose() { } diff --git a/tests/B3.Exchange.Host.Tests/MassCancelPassiveCommitFailureTests.cs b/tests/B3.Exchange.Host.Tests/MassCancelPassiveCommitFailureTests.cs index cedbfa32..a504970e 100644 --- a/tests/B3.Exchange.Host.Tests/MassCancelPassiveCommitFailureTests.cs +++ b/tests/B3.Exchange.Host.Tests/MassCancelPassiveCommitFailureTests.cs @@ -108,6 +108,18 @@ public void Append(uint sessionId, uint seq, long timestampNanos, ReadOnlySpan