From d3047a9847664485b69cc5ba3ba669d3682a2310 Mon Sep 17 00:00:00 2001 From: Pedro Sakuma Travi <39205549+pedrosakuma@users.noreply.github.com> Date: Sat, 5 Sep 2026 03:22:48 +0000 Subject: [PATCH] fix: preserve FIXP recovery and expose journal sizing metrics Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/EXCHANGE-SIMULATOR.md | 21 +- docs/RUNBOOK.md | 96 +++++++-- src/B3.Exchange.Contracts/Metrics.cs | 15 ++ src/B3.Exchange.Core/MetricsRegistry.cs | 22 +++ .../Persistence/FileFixpOutboundJournal.cs | 1 + .../MetricsRegistryTests.cs | 4 + .../FileFixpOutboundJournalTests.cs | 16 ++ .../GracefulShutdownTests.cs | 187 ++++++++++++++++++ 8 files changed, 344 insertions(+), 18 deletions(-) diff --git a/docs/EXCHANGE-SIMULATOR.md b/docs/EXCHANGE-SIMULATOR.md index 0c1f2f5d..2c987ca1 100644 --- a/docs/EXCHANGE-SIMULATOR.md +++ b/docs/EXCHANGE-SIMULATOR.md @@ -167,6 +167,16 @@ Exposes: trade reference exists yet, the decoder accepts and leaves downstream validation unchanged. Breaches produce `ER_Reject(OrdRejReason=16 Price exceeds current price band)`. +* `tcp.retransmitPersistenceDir` — directory for durable FIXP session-resync + state. Empty/unset disables it, so a process restart forces peers down the + renegotiate path even if channel WAL/snapshots preserved the public book. +* `tcp.maxJournalBytes` — per-session outbound retransmit journal byte budget. + Default `268435456` (256 MiB). Rotation is ACK-watermark-gated: if no + segment is provably safe to delete, the journal is allowed to grow and + metrics/logs alert the operator instead of dropping recoverable frames. +* `tcp.maxJournalRetentionHours` — per-session outbound retransmit journal age + target. Default `24`. Retention is also conservative and never deletes + frames the peer may still request. * `http` — **optional** Kestrel-hosted operability endpoint exposing `/health/live`, `/health/ready`, and `/metrics`. Omit the entire block to disable HTTP. @@ -306,11 +316,12 @@ Exposes: closed). Case-insensitive. * Snapshot evolution and on-disk schema versioning are described in the *Snapshot file format* and *Snapshot schema evolution* sections below. -* Limitations: auction-top throttling state, the UMDF retransmit ring, - snapshot-rotator cursor, and FIXP session counters are not persisted. - Operators that depend on auction throttling should drain in-flight - indicative state before restart. Omit the `persistence` block to keep - the legacy stateless boot. +* Limitations: auction-top throttling state, the UMDF retransmit ring, and + snapshot-rotator cursor are not persisted by the per-channel block. FIXP + session envelope counters and outbound business-frame recovery are persisted + only when the top-level `tcp.retransmitPersistenceDir` is set. Operators that + depend on auction throttling should drain in-flight indicative state before + restart. Omit the `persistence` block to keep the legacy stateless boot. ### FIXP session resync persistence (`tcp.retransmitPersistenceDir`) diff --git a/docs/RUNBOOK.md b/docs/RUNBOOK.md index 6e94ffa0..66338415 100644 --- a/docs/RUNBOOK.md +++ b/docs/RUNBOOK.md @@ -12,9 +12,11 @@ it, and triggering the recovery scenarios that consumers care about. > scope**: running two instances against the same UMDF multicast group > would emit duplicate sequence numbers and ER frames, and the engine > state lives in process memory only. When HA becomes a goal it will -> require external coordination (lease / fencing) plus persistent -> journaling — neither of which exists today. Plan deployments around a -> single primary with manual failover. +> require external coordination (lease / fencing) plus replicated engine +> state, which does not exist today. Plan deployments around a single +> primary with manual failover. The optional FIXP replay journal described +> in §7.5 preserves private session replay across restarts, but it is not +> an active-active fencing or replicated-engine-state mechanism. --- @@ -977,31 +979,99 @@ not a designed-in recovery path. | --- | --- | --- | | Live socket write path | OS socket buffer | Until kernel flushes / connection drops | | `RetransmitBuffer` (per FIXP session) | **In-memory ring**, capacity = `session.outboundRetransmitCapacity` | Until the FIXP session is reaped (`SuspendedTimeoutMs`, default **5 min**) or evicted by a newer record | -| WAL / snapshot | Persisted, but stores **commands**, not the resulting `ExecutionReport`s | Bounded by snapshot cadence | +| FIXP outbound journal (when `tcp.retransmitPersistenceDir` is set) | Durable business-frame journal under `{dir}/journal/session-{sessionId:x8}/segment-*.log` | Quota/retention-managed by `tcp.maxJournalBytes` and `tcp.maxJournalRetentionHours`, but never pruned below the peer ACK watermark | +| Channel WAL / snapshot | Persisted, but stores **commands**, not the resulting `ExecutionReport`s | Bounded by snapshot cadence | **The good news (issue [#217](https://github.com/pedrosakuma/B3MatchingPlatform/issues/217)):** passive `ExecutionReport`s emitted while the owning FIXP session is in `FixpState.Suspended` (transport disconnected but session still alive) are appended to that session's `RetransmitBuffer` rather than dropped. A subsequent `Establish` + `RetransmitRequest` -delivers them with `PossResend = 1` as a normal recovery cycle. +delivers them with `PossResend = 1` as a normal recovery cycle. If +`tcp.retransmitPersistenceDir` is set, the same outbound business frames +are also appended to the durable FIXP journal, so replay can fall back to +disk after a ring eviction or host restart. + +**Graceful process shutdown / rolling restart (issue +[#613](https://github.com/pedrosakuma/B3MatchingPlatform/issues/613)):** +`SIGINT`/`SIGTERM` drive `ExchangeHost.StopAsync`, which marks readiness +`NOT_READY`, stops new accepts, drains inbound and outbound queues, then +sends `Terminate(FINISHED)` with `CloseKind.HostShutdown` to every live +FIXP session. `HostShutdown` is a preserving close kind: it does not +delete the state snapshot or outbound journal and it saves a final +snapshot before the host disposes. On the next boot, if +`tcp.retransmitPersistenceDir` points at the same durable directory, the +host loads the snapshots, seeds `SessionClaimRegistry`, and accepts an +`Establish` for the same `SessionId`/`SessionVerId`; cold +`RetransmitRequest` reads are served from the journal. If +`tcp.retransmitPersistenceDir` is empty, the shutdown is still orderly +on the wire, but FIXP resumability is process-local and a rolling restart +cannot avoid a fresh `Negotiate`/session roll. **The known gaps (none of these are silent — alert on them):** | Scenario | Outcome | Mitigation today | Eventual fix | | --- | --- | --- | --- | | Disconnect window ≤ `SuspendedTimeoutMs`, fills fit in ring | All ERs replayed on `Establish` ✅ | — | — | -| Disconnect window ≤ `SuspendedTimeoutMs`, **fills exceed ring capacity** | Oldest ERs evicted; `Establish` with too-low `NextSeqNo` ⇒ `Establishment Reject` | Size `outboundRetransmitCapacity` for worst-case fill rate × `SuspendedTimeoutMs`; alert on `exch_fixp_retransmit_buffer_utilization` ([#288](https://github.com/pedrosakuma/B3MatchingPlatform/issues/288)) | Persisted `RetransmitBuffer` ([#289](https://github.com/pedrosakuma/B3MatchingPlatform/issues/289)) | -| **Host crash** while session is `Suspended` | Ring is in-memory; if `tcp.retransmitPersistenceDir` is unset, all buffered ERs are lost. With persistence enabled, the per-session ring is mirrored to `{retransmitPersistenceDir}/sessions/session-{sessionId:x8}.ring` and rehydrated on the next boot ([#289](https://github.com/pedrosakuma/B3MatchingPlatform/issues/289)). | Set `tcp.retransmitPersistenceDir` on durable storage; alerts unchanged | Covered (issue [#289](https://github.com/pedrosakuma/B3MatchingPlatform/issues/289)) | +| Disconnect window ≤ `SuspendedTimeoutMs`, **fills exceed ring capacity** | With `tcp.retransmitPersistenceDir` set, cold reads come from the durable outbound journal; without it, oldest ERs are evicted and `Establish` with too-low `NextSeqNo` ⇒ `Establishment Reject`. | Enable FIXP resync persistence, then still size `outboundRetransmitCapacity` for hot-path replay and alert on `exch_fixp_retransmit_buffer_utilization` ([#288](https://github.com/pedrosakuma/B3MatchingPlatform/issues/288)). | Covered by journal-backed resync persistence ([#405](https://github.com/pedrosakuma/B3MatchingPlatform/issues/405)) | +| **Host crash** while session is `Suspended` | If `tcp.retransmitPersistenceDir` is unset, the ring and FIXP envelope state are lost and the peer must renegotiate. With persistence enabled, envelope snapshots re-seed `SessionClaimRegistry` and the outbound journal serves cold `RetransmitRequest` reads, so the peer can reattach with its original `SessionVerId` ([#405](https://github.com/pedrosakuma/B3MatchingPlatform/issues/405)). | Set `tcp.retransmitPersistenceDir` on durable storage mounted with the channel WAL; alert on `fixp_journal_bytes`, `fixp_journal_oldest_age_seconds`, and rotation blockers. | Covered (issue [#405](https://github.com/pedrosakuma/B3MatchingPlatform/issues/405)) | | Disconnect window > `SuspendedTimeoutMs` | `TryReapIfSuspended` removes the session from `SessionRegistry`; ring goes to GC | Size `SuspendedTimeoutMs` to the worst-case operational disconnect for the firm; alert on `exch_fixp_sessions_reaped_total` ([#288](https://github.com/pedrosakuma/B3MatchingPlatform/issues/288)) | Tunable per-firm; protocol-level `OrderMassStatus` on next reconnect (Tier 3) | | Cancel-on-Disconnect (CoD) armed, mode 1/3, window expired | `MassCancel` fires ⇒ no passive fills can happen ⇒ no problem | Configure CoD per firm policy | — | -**Operational implication:** the **dimensioning tuple** that -determines whether a disconnect causes data loss is -`(outboundRetransmitCapacity, SuspendedTimeoutMs, expected fill -rate while disconnected)`. Document the chosen values per firm -and alert on the utilization metrics rather than discovering the -limit when a real disconnect happens. +For Helm deployments, prefer a sibling of the existing durable channel +persistence mount (for example +`/var/lib/b3matching/fixp-sessions`) rather than adding a second PVC. The +chart default leaves `tcp.retransmitPersistenceDir` empty so local/ephemeral +installs keep the old renegotiate-on-restart behavior; production-like +overlays must opt in explicitly. The chart also exposes the FIXP outbound +journal quotas as `tcp.maxJournalBytes` (default 256 MiB per session) and +`tcp.maxJournalRetentionHours` (default 24h); both are conservative targets, +not permission to delete frames the peer has not acknowledged. + +**Cold-journal sizing formula (#615):** size the durable replay window from +observed per-session journal append throughput, not from a speculative global +default. The journal stores each outbound business frame plus 20 bytes of +record overhead: + +```text +record_bytes = wire_frame_bytes + 20 +bytes_per_second = rate(fixp_journal_appended_bytes_total{session=...}[5m]) +frames_per_second = rate(fixp_journal_appended_frames_total{session=...}[5m]) +avg_record_bytes = bytes_per_second / max(frames_per_second, 1) +effective_window_seconds = min(maxJournalRetentionHours * 3600, + maxJournalBytes / bytes_per_second) +required_maxJournalBytes = bytes_per_second * target_window_seconds +``` + +With the defaults, the 24h budget is only `268435456 / 86400 = 3106.9 B/s` +per session, so high-throughput firms can exhaust the byte cap before the age +limit. Do not raise defaults or split quota per firm without production +`fixp_journal_appended_*` history and the effective values deployed in that +environment. + +Suggested 80% PromQL alerts for deployments using the defaults: + +```promql +fixp_journal_bytes > 0.8 * 268435456 +fixp_journal_oldest_age_seconds > 0.8 * 24 * 60 * 60 +rate(fixp_journal_appended_bytes_total[5m]) * 24 * 60 * 60 > 0.8 * 268435456 +increase(fixp_journal_rotation_total{reason="bytes"}[5m]) > 0 +``` + +The first two catch retained journal state approaching the configured limits. +The append-rate projection catches sessions whose current throughput cannot +sustain a 24h replay window under the 256 MiB cap. Treat byte-triggered +rotations as a sizing signal: the effective replay floor has moved forward to +the peer-confirmed pruning watermark. + +**Operational implication:** without durable FIXP resync persistence, the +**dimensioning tuple** that determines whether a disconnect causes data loss is +`(outboundRetransmitCapacity, SuspendedTimeoutMs, expected fill rate while +disconnected)`. With `tcp.retransmitPersistenceDir` set, that tuple still +controls the hot-path ring hit rate, while `tcp.maxJournalBytes` and +`tcp.maxJournalRetentionHours` bound the cold replay store. Document the chosen +values per firm and alert on the utilization metrics rather than discovering +the limit when a real disconnect happens. **What an operator should NOT do:** rely on out-of-band reconciliation (back-office, manual `OrderStatusRequest` sweep) as diff --git a/src/B3.Exchange.Contracts/Metrics.cs b/src/B3.Exchange.Contracts/Metrics.cs index d843dbbc..51c0496d 100644 --- a/src/B3.Exchange.Contracts/Metrics.cs +++ b/src/B3.Exchange.Contracts/Metrics.cs @@ -162,6 +162,15 @@ public void Observe(uint sessionId, long bytes, long oldestAgeSeconds) Interlocked.Exchange(ref s.OldestAgeSeconds, oldestAgeSeconds); } + public void IncAppend(uint sessionId, long bytes) + { + if (bytes < 0) + throw new ArgumentOutOfRangeException(nameof(bytes), "must be >= 0"); + var s = Get(sessionId); + Interlocked.Add(ref s.AppendedBytes, bytes); + Interlocked.Increment(ref s.AppendedFrames); + } + public void IncRotation(uint sessionId, string reason) { var s = Get(sessionId); @@ -180,6 +189,8 @@ public IReadOnlyList Snapshot() kv.Key, Interlocked.Read(ref kv.Value.Bytes), Interlocked.Read(ref kv.Value.OldestAgeSeconds), + Interlocked.Read(ref kv.Value.AppendedBytes), + Interlocked.Read(ref kv.Value.AppendedFrames), Interlocked.Read(ref kv.Value.RotationsBytes), Interlocked.Read(ref kv.Value.RotationsAge))) .OrderBy(s => s.Session, StringComparer.Ordinal) @@ -196,6 +207,8 @@ private sealed class SessionJournalMetrics { public long Bytes; public long OldestAgeSeconds; + public long AppendedBytes; + public long AppendedFrames; public long RotationsBytes; public long RotationsAge; } @@ -205,5 +218,7 @@ public readonly record struct FixpJournalMetricsSnapshot( string Session, long Bytes, long OldestAgeSeconds, + long AppendedBytes, + long AppendedFrames, long RotationsBytes, long RotationsAge); diff --git a/src/B3.Exchange.Core/MetricsRegistry.cs b/src/B3.Exchange.Core/MetricsRegistry.cs index c52ade29..874c666e 100644 --- a/src/B3.Exchange.Core/MetricsRegistry.cs +++ b/src/B3.Exchange.Core/MetricsRegistry.cs @@ -941,6 +941,28 @@ private void EmitFixpJournalMetrics(StringBuilder sb) .Append('\n'); } + sb.Append("# HELP fixp_journal_appended_bytes_total Total bytes durably appended to the FIXP outbound retransmit journal per session, including journal record overhead.\n"); + sb.Append("# TYPE fixp_journal_appended_bytes_total counter\n"); + foreach (var s in snap) + { + sb.Append("fixp_journal_appended_bytes_total{session=\"") + .Append(EscapeLabel(s.Session)) + .Append("\"} ") + .Append(s.AppendedBytes.ToString(CultureInfo.InvariantCulture)) + .Append("\n"); + } + + sb.Append("# HELP fixp_journal_appended_frames_total Total business frames durably appended to the FIXP outbound retransmit journal per session.\n"); + sb.Append("# TYPE fixp_journal_appended_frames_total counter\n"); + foreach (var s in snap) + { + sb.Append("fixp_journal_appended_frames_total{session=\"") + .Append(EscapeLabel(s.Session)) + .Append("\"} ") + .Append(s.AppendedFrames.ToString(CultureInfo.InvariantCulture)) + .Append("\n"); + } + sb.Append("# HELP fixp_journal_rotation_total Total FIXP retransmit journal segment rotations by trigger reason.\n"); sb.Append("# TYPE fixp_journal_rotation_total counter\n"); foreach (var s in snap) diff --git a/src/B3.Exchange.Gateway/Persistence/FileFixpOutboundJournal.cs b/src/B3.Exchange.Gateway/Persistence/FileFixpOutboundJournal.cs index f5872acf..f2101d95 100644 --- a/src/B3.Exchange.Gateway/Persistence/FileFixpOutboundJournal.cs +++ b/src/B3.Exchange.Gateway/Persistence/FileFixpOutboundJournal.cs @@ -159,6 +159,7 @@ public void Append(uint sessionId, uint seq, long timestampNanos, ReadOnlySpan public class GracefulShutdownTests { + private const long Petr = 900_000_000_001L; + private sealed class NullSink : IUmdfPacketSink { public void Publish(byte channelNumber, ReadOnlySpan packet) { } @@ -121,6 +126,114 @@ public async Task StopAsync_IsIdempotent() Assert.True(sw.ElapsedMilliseconds < 500, $"second StopAsync should short-circuit, took {sw.ElapsedMilliseconds}ms"); } + [Fact] + public async Task StopAsync_WithRetransmitPersistenceDir_PreservesFixpStateAndJournal_ForEstablishResume() + { + const uint sessionId = 613; + const ulong sessionVerId = 9; + const uint enteringFirm = 7; + + var root = CreateRepoTestDir("graceful-shutdown-fixp-"); + try + { + var cfg = BuildConfig(); + cfg.Auth = new AuthConfig { DevMode = true, RequireFixpHandshake = true }; + cfg.Tcp.RetransmitPersistenceDir = root; + cfg.Tcp.HeartbeatIntervalMs = 60_000; + cfg.Tcp.IdleTimeoutMs = 60_000; + cfg.Tcp.TestRequestGraceMs = 60_000; + cfg.Firms.Add(new FirmConfig + { + Id = "firm-613", + Name = "Issue 613 regression", + EnteringFirmCode = enteringFirm, + }); + cfg.Sessions.Add(new SessionConfig + { + SessionId = sessionId.ToString(System.Globalization.CultureInfo.InvariantCulture), + FirmId = "firm-613", + }); + + await using (var host = new ExchangeHost(cfg, packetSinkFactory: _ => new NullSink())) + { + await host.StartAsync(); + var ep = host.TcpEndpoint!; + + using var client = new TcpClient(); + await client.ConnectAsync(ep.Address, ep.Port); + var stream = client.GetStream(); + + await WriteNegotiateAsync(stream, sessionId, sessionVerId, enteringFirm); + Assert.Equal(EntryPointFrameReader.TidNegotiateResponse, + (await ReadFrameAsync(stream, TimeSpan.FromSeconds(5))).TemplateId); + + await WriteEstablishAsync(stream, sessionId, sessionVerId, nextSeqNo: 1); + var establishAck = await ReadFrameAsync(stream, TimeSpan.FromSeconds(5)); + Assert.Equal(EntryPointFrameReader.TidEstablishAck, establishAck.TemplateId); + Assert.True(EntryPointFixpFrameCodec.TryDecodeEstablishAck(establishAck.Body, out var ack)); + Assert.Equal(1u, ack.NextSeqNo); + + await stream.WriteAsync(BuildSimpleNewOrder( + clOrdId: 61301, + secId: Petr, + side: '1', + ordType: '2', + tif: '0', + qty: 100, + priceMantissa: 123_400, + sessionId: sessionId, + msgSeqNum: 1)); + + var er = await ReadFrameAsync(stream, TimeSpan.FromSeconds(5)); + Assert.Equal(EntryPointFrameReader.TidExecutionReportNew, er.TemplateId); + + await host.StopAsync(); + } + + using (var persister = new FileFixpSessionStatePersister( + root, + NullLogger.Instance)) + { + var snapshot = persister.Load(sessionId); + Assert.NotNull(snapshot); + Assert.Equal(sessionVerId, snapshot.Value.SessionVerId); + Assert.Equal(1u, snapshot.Value.OutboundMsgSeqNum); + Assert.Equal(1u, snapshot.Value.LastIncomingSeqNo); + } + + using (var journal = new FileFixpOutboundJournal( + root, + NullLogger.Instance)) + { + Assert.Equal(1u, journal.MaxSeq(sessionId)); + Assert.Single(journal.ReadRange(sessionId, fromSeq: 1, count: 10)); + } + + await using (var restarted = new ExchangeHost(cfg, packetSinkFactory: _ => new NullSink())) + { + await restarted.StartAsync(); + var ep = restarted.TcpEndpoint!; + + using var client = new TcpClient(); + await client.ConnectAsync(ep.Address, ep.Port); + var stream = client.GetStream(); + + await WriteEstablishAsync(stream, sessionId, sessionVerId, nextSeqNo: 2); + var resumedAckFrame = await ReadFrameAsync(stream, TimeSpan.FromSeconds(5)); + Assert.Equal(EntryPointFrameReader.TidEstablishAck, resumedAckFrame.TemplateId); + Assert.True(EntryPointFixpFrameCodec.TryDecodeEstablishAck(resumedAckFrame.Body, out var resumedAck)); + Assert.Equal(sessionVerId, resumedAck.SessionVerId); + Assert.Equal(2u, resumedAck.NextSeqNo); + + await restarted.StopAsync(); + } + } + finally + { + TryDelete(root); + } + } + private static ShutdownReadinessProbe GetShutdownProbe(ExchangeHost host) { var f = typeof(ExchangeHost) @@ -159,4 +272,78 @@ private static async Task ReadExactAsync(NetworkStream stream, byte[] buffer, Ca read += n; } } + + private static async Task WriteNegotiateAsync(NetworkStream stream, uint sessionId, ulong sessionVerId, uint enteringFirm) + { + var credentials = Encoding.ASCII.GetBytes($"{{\"auth_type\":\"basic\",\"username\":\"{sessionId}\",\"access_key\":\"\"}}"); + var buffer = new byte[EntryPointFrameReader.MaxInboundMessageLength]; + int length = EntryPointFixpFrameCodec.EncodeNegotiate( + buffer, + sessionId, + sessionVerId, + timestampNanos: 0, + enteringFirm, + onBehalfFirm: null, + credentials, + clientIp: "127.0.0.1"u8, + clientAppName: "test"u8, + clientAppVersion: "1"u8); + await stream.WriteAsync(buffer.AsMemory(0, length)); + } + + private static async Task WriteEstablishAsync(NetworkStream stream, uint sessionId, ulong sessionVerId, uint nextSeqNo) + { + var buffer = new byte[EntryPointFrameReader.MaxInboundMessageLength]; + int length = EntryPointFixpFrameCodec.EncodeEstablish( + buffer, + sessionId, + sessionVerId, + timestampNanos: 0, + keepAliveIntervalMillis: 60_000, + nextSeqNo, + cancelOnDisconnectType: 0, + codTimeoutWindowMillis: 0, + credentials: ReadOnlySpan.Empty); + await stream.WriteAsync(buffer.AsMemory(0, length)); + } + + private static byte[] BuildSimpleNewOrder(ulong clOrdId, long secId, char side, char ordType, + char tif, long qty, long priceMantissa, uint sessionId, uint msgSeqNum) + { + var frame = new byte[EntryPointFrameReader.WireHeaderSize + 82]; + EntryPointFrameReader.WriteHeader(frame.AsSpan(0, EntryPointFrameReader.WireHeaderSize), + messageLength: (ushort)frame.Length, + blockLength: 82, + templateId: EntryPointFrameReader.TidSimpleNewOrder, + version: 2); + + var body = frame.AsSpan(EntryPointFrameReader.WireHeaderSize); + BinaryPrimitives.WriteUInt32LittleEndian(body.Slice(0, 4), sessionId); + BinaryPrimitives.WriteUInt32LittleEndian(body.Slice(4, 4), msgSeqNum); + ulong sendingTimeNanos = checked((ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000UL); + BinaryPrimitives.WriteUInt64LittleEndian(body.Slice(8, 8), sendingTimeNanos); + BinaryPrimitives.WriteUInt64LittleEndian(body.Slice(20, 8), clOrdId); + BinaryPrimitives.WriteInt64LittleEndian(body.Slice(48, 8), secId); + body[56] = (byte)side; + body[57] = (byte)ordType; + body[58] = (byte)tif; + BinaryPrimitives.WriteInt64LittleEndian(body.Slice(60, 8), qty); + BinaryPrimitives.WriteInt64LittleEndian(body.Slice(68, 8), priceMantissa); + return frame; + } + + private static string CreateRepoTestDir(string prefix) + { + var config = TestPaths.ResolveRepoFile("config/instruments-eqt.json"); + var repoRoot = Directory.GetParent(Path.GetDirectoryName(config)!)!.FullName; + var dir = Path.Combine(repoRoot, "artifacts", "test-dirs", prefix + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(dir); + return dir; + } + + private static void TryDelete(string dir) + { + try { if (Directory.Exists(dir)) Directory.Delete(dir, recursive: true); } + catch { } + } }