From ea45ce6a6ed46418396c6bf3a17f30f30371c460 Mon Sep 17 00:00:00 2001 From: Alejandro Canela Date: Tue, 21 Jul 2026 16:25:18 +0200 Subject: [PATCH] feat(writer): async sink adapter for non-blocking chunk/journal I/O --- CHANGELOG.md | 14 + README.md | 5 + docs/ARCHITECTURE.md | 7 +- docs/PERFORMANCE.md | 1 + docs/POST_PROCESSING.md | 4 +- docs/SDK_API.md | 37 +- docs/research/async_writer_live_mode.md | 247 +++++++ sdk/include/vtx/common/vtx_diagnostics.h | 4 + .../vtx/writer/core/vtx_writer_facade.h | 22 + .../vtx/writer/core/vtx_writer_result.h | 6 + sdk/include/vtx/writer/core/writer.h | 49 ++ .../policies/sinks/async_sink_adapter.h | 421 ++++++++++++ .../vtx/writer/policies/sinks/file_sink.h | 55 +- .../writer/policies/sinks/recovery_journal.h | 18 +- .../src/vtx/writer/core/vtx_writer_facade.cpp | 99 +-- tests/CMakeLists.txt | 1 + tests/writer/test_async_sink.cpp | 621 ++++++++++++++++++ tests/writer/test_crash_recovery.cpp | 139 +++- 18 files changed, 1684 insertions(+), 66 deletions(-) create mode 100644 docs/research/async_writer_live_mode.md create mode 100644 sdk/include/vtx/writer/policies/sinks/async_sink_adapter.h create mode 100644 tests/writer/test_async_sink.cpp diff --git a/CHANGELOG.md b/CHANGELOG.md index 501e25e..d5aeb50 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **writer/durability**: **async (non-blocking) recording** -- an opt-in that moves all chunk/journal disk I/O off the caller's (game) thread, so accepting a frame never waits on serialization, compression, `fsync`, or a chunk-boundary flush. New sink decorator **`AsyncSinkAdapter`** (`sdk/include/vtx/writer/policies/sinks/async_sink_adapter.h`) turns `JournalFrame` / `SaveChunk` / `Close` into enqueue operations on a bounded FIFO that a single dedicated I/O worker drains in order, calling the real `ChunkedFileSink`. `OnSessionStart` / `JournalTiming` stay synchronous (session setup and the 'S' record must precede everything). Enabled through `WriterFacadeConfig`: + - `async_io` (default **off**) selects the decorated sink; `async_max_queue_frames` (0 -> `2 * chunk_max_frames`) bounds the queue **by item count** -- a full queue blocks the caller (degrading toward synchronous behavior; never drops a frame, never grows memory unbounded). (Byte-bounding is not implementable today: `FlatBuffersVtxPolicy::GetFrameSize()` returns 0 and journal payload sizes are only known after worker-side serialization.) + - `durable_writes` and `enable_recovery_journal` are now exposed at the facade (previously only on the internal sink config), for both synchronous and async recordings. + - **Ordering + crash-recovery are unchanged by construction**: one FIFO + one worker replays the exact synchronous on-disk sequence (data-before-journal, F-record contiguity, append-only journal + compaction), so `RepairReplayFile` needs no changes. What changes is a bounded **durability lag** -- a frame is crash-recoverable only once its item (or group-commit batch) is durable; a hard kill loses that not-yet-durable suffix, and recovery still yields a clean contiguous prefix. `Drain()` and `Stop()` are the zero-lag synchronization points. + - **Group commit**: the worker `fsync`s the recovery journal once per drained batch instead of once per frame, so a slow disk (HDD, ~10-20 ms/fsync) keeps up with 60 fps instead of the queue filling and async degenerating to blocking. + - **Failure protocol**: the first inner-sink I/O failure latches an error, discards the queue, and closes the file + journal handles WITHOUT a footer (releasing the Windows deny-write share so the partial `.vtx` and its journal are repair-ready immediately, even while the process keeps running). Every blocking wait -- a full-queue producer, `Drain()`, `Stop()`, the destructor -- is failure-aware and returns promptly. It surfaces as the new **`VtxErrorCode::SinkFailed`** on the next `TryRecordFrame` (via a zero-cost `if constexpr HasFailed()` hook on the writer core -- no interface change for existing sinks) and through the facade's new `GetLastError()`; `PipelineReport` counts it in a dedicated **`sink_failed`** field, kept separate from `validation_errors`. + - New facade surface on **`IVtxWriterFacade`**: `Drain()` (durability barrier -- blocks until every accepted frame is durable; **note: under async, `Flush()` only closes and enqueues the current chunk and is NO LONGER a durability barrier -- `Drain()` is**), `GetLastError()`, and `GetQueueDepth()` (backpressure telemetry). + - Supporting primitives (synchronous path byte-for-byte unchanged): `RecoveryJournal::AppendFrame` gains a `sync` flag plus `SyncNow()`, and `ChunkedFileSink` gains `JournalFrameBatched` / `SyncJournal` / `IsJournalActive` / `Good` / `AbortClose`. + - An enqueue issued after the I/O worker had already exited (e.g. recording after `Stop()`) could block forever waiting for room on a queue nobody drains any more; the worker now publishes its exit and every wait predicate honours it, so such a call returns immediately instead of deadlocking. + - Verified in `tests/writer/test_async_sink.cpp` (14 tests) and `tests/writer/test_crash_recovery.cpp`: order equivalence against an instrumented fake inner sink; async-vs-sync **byte identity** through the real sink (FlatBuffers + Protobuf); **no frame or chunk loss** across the durability x compression x journal x chunk-size x queue-cap matrix (frame count, chunk count and per-frame content hashes all compared against a synchronous recording); whole-replay validation of a clean async file; a 2000-frame stream through a one-item queue; worker start/join churn; `Drain()` interleaved under permanent backpressure; the failure protocol (SinkFailed surfacing + handle release); and async-ON variants of the crash-recovery E2E **plus the real `TerminateProcess` kill matrix** (durable / flush-only / mid-compaction / varied progress points), each recovering to a clean contiguous prefix identical to what a synchronous clean run would have produced. + - Race freedom checked under **ThreadSanitizer** (gcc-13, `VTX_SANITIZE=thread`): 97 writer/crash-recovery tests clean. The harness was itself validated by confirming TSan reports a deliberately injected race in the adapter. + ## [0.4.0] - 2026-07-20 ### Added diff --git a/README.md b/README.md index 4466a22..cb30e2c 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ VTX is an open binary format for real-time per-frame state data, plus a C++20 SD - **Both Protobuf and FlatBuffers.** SDK supports both backends out of the box. The file announces which in its magic bytes; readers auto-detect. - **Live streaming transports.** Ingest frames live from an OS pipe (Windows named pipe, POSIX FIFO, stdin) or a WebSocket connection, and emit the `.vtx` byte stream over a TCP socket instead of writing to disk. Same writer API behind each transport -- file or network, sink-agnostic. - **Crash-safe recording.** A write-ahead `.recovery` sidecar (checksummed, append-only, fsync'd per operation by default) makes a recording that dies before `Stop()` recoverable up to the last recorded frame -- exact per-frame timestamps included. `RepairReplayFile()` reconstructs the footer; a hours-long session that crashes is no longer lost. +- **Non-blocking recording.** Opt into `async_io` and chunk/journal disk I/O -- serialization, zstd, `fsync`, the chunk-boundary flush -- moves to a dedicated worker thread, so your capture loop never waits on the disk. Accept/reject stays synchronous and post-processors keep running on your thread; a bounded queue applies backpressure instead of dropping frames. The bytes on disk are identical to a synchronous recording, and crash recovery is unchanged. - **Validation & structured diagnostics.** Validate a schema, entity, frame, or whole replay independently; strict recording rejects frames observably (bad game-time, unresolved entity type). Every failure is a structured `VtxError` -- code, severity, location, expected-vs-provided type -- so automation acts on data, not parsed log lines. - **Engine-independent C++20.** No engine dependency. Language bindings wherever Protobuf or FlatBuffers exist (Python, Go, Rust, Java, JS). - **Open.** Apache-2.0. Spec, reference reader, and tooling all in the repo. @@ -56,6 +57,10 @@ writer->Flush(); writer->Stop(); ``` +#### Optional: keep the disk off your capture loop + +Set `config.async_io = true` and the chunk/journal writes move to a worker thread -- `RecordFrame` no longer pays serialization, compression, `fsync`, or the chunk-boundary flush. The API is otherwise the same, with two things to know: `Flush()` stops being a durability barrier (it queues the write; the new `writer->Drain()` is the barrier), and if the disk cannot keep up the caller blocks on a bounded queue rather than losing frames. Full contract in [`docs/SDK_API.md`](docs/SDK_API.md). + #### Optional: one call with `WriteReplay` If your frames come from an `IFrameDataSource`, `VTX::WriteReplay(config, source)` runs the whole pipeline -- create writer, drain the source, finalize -- and returns a `WriteReplayResult` (frames written / dropped, one warning per dropped frame, elapsed time). The schema can be supplied in memory too: set `config.schema_json_content` (raw JSON) or `config.schema_registry` (a pre-built `std::shared_ptr`) instead of `schema_json_path`. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 678ba84..506fc7e 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -38,13 +38,14 @@ Provides the shared type system, serialization infrastructure, and utilities use Records live frame data into `.vtx` replay files (or streams the same bytes over a socket). -- **Facade**: `IVtxWriterFacade` — `RecordFrame()`, `TryRecordFrame()` (returns a `RecordResult`), `Flush()`, `Stop()`, `SetPostProcessor()` +- **Facade**: `IVtxWriterFacade` — `RecordFrame()`, `TryRecordFrame()` (returns a `RecordResult`), `Flush()`, `Stop()`, `SetPostProcessor()`, plus the async-I/O surface `Drain()` (durability barrier), `GetLastError()`, `GetQueueDepth()` - **Finalization & strict recording**: each frame is validated (entity types resolve to schema structs), content-hashed (after schema fields + post-processor overrides), and **frozen** (stashed mutation handles revoked) before it enters the chunk pipeline. `TryRecordFrame()` makes rejection observable via `RecordResult` (carries a `VtxError`); the writer assigns a monotonic frame index; an opt-in `retain_finalized_snapshot` keeps the last finalized frame queryable via `GetLastFinalizedFrame()` / `FindEntity()`. `RecordPipeline::Run()` returns a `PipelineReport` - **Factory**: `CreateFlatBuffersWriterFacade()` / `CreateProtobufWriterFacade()` (file output); `CreateFlatBuffersNetworkWriterFacade()` / `CreateProtobufNetworkWriterFacade()` (TCP socket output) — same `IVtxWriterFacade` behind both, so user code is sink-agnostic - **One-call pipeline**: `WriteReplay(config, source, format)` (`vtx_write_replay.h`) creates the writer, drains an `IFrameDataSource`, finalizes, and returns a `WriteReplayResult` (frames written / dropped, one `VtxWarning` per dropped frame, elapsed time) -- **Config**: `WriterFacadeConfig` (file path) or `NetworkWriterFacadeConfig` (host + port) — chunk size, compression, and the schema supplied as a JSON path, an in-memory JSON string (`schema_json_content`), or a pre-built `SchemaRegistry` (`schema_registry`); `create_output_dirs` (default on) auto-creates missing parent directories of the output file +- **Config**: `WriterFacadeConfig` (file path) or `NetworkWriterFacadeConfig` (host + port) — chunk size, compression, and the schema supplied as a JSON path, an in-memory JSON string (`schema_json_content`), or a pre-built `SchemaRegistry` (`schema_registry`); `create_output_dirs` (default on) auto-creates missing parent directories of the output file; durability (`durable_writes`, `enable_recovery_journal`) and async I/O (`async_io`, `async_max_queue_frames`) are exposed on the file config - **Policy**: Template-parameterized writer policies select FlatBuffers or Protobuf serialization at compile time - **Sinks**: `ChunkedFileSink` (file), `ChunkedNetworkSink` (TCP stream) — both produce the **same `.vtx` byte sequence**, so a socket receiver only has to concatenate incoming bytes into a file to get a valid replay +- **Sink decorator**: `AsyncSinkAdapter` (`async_sink_adapter.h`) wraps a sink so `JournalFrame` / `SaveChunk` / `Close` become enqueue operations on a bounded FIFO that a single I/O worker drains **in order** — same call sequence, therefore the same bytes and the same crash-recovery invariants. Session setup (`OnSessionStart` / `JournalTiming`) stays synchronous. Selected at runtime by `WriterFacadeConfig::async_io`, so it is just another policy: the writer core is unchanged - **Data-source interface**: `IFrameDataSource` (`Initialize()` / `GetNextFrame()` / `GetExpectedTotalFrames()`). Concrete implementations: `samples/advance_write.cpp` (JSON / Protobuf / FlatBuffers from disk); `PipeFrameDataSource` (stdin, Windows named pipes, POSIX FIFOs — including a **server mode** that creates the pipe and waits, for independent game-injector-style external producers); `WebSocketFrameDataSource` (`ws://` + `wss://` with TLS). Both streaming sources share the `IFramePayloadAdapter` concept, so a single adapter plugs into either transport. A `SharedMemoryFrameDataSource` (zero-copy SPSC ring, pluggable `ISharedMemoryTransport`) is present in-tree but **experimental / WIP and not yet functional** — landed unintentionally, not a supported input path - **Dependencies**: protobuf + flatbuffers (serialization), zstd (chunk compression), IXWebSocket + mbedTLS (WebSocket transport; hidden behind a PIMPL boundary in `websocket_client.cpp` — never leak into the public SDK headers) @@ -91,6 +92,8 @@ Each module exposes a single abstract interface (`IVtxReaderFacade`, `IVtxWriter ## Threading Model +- **Writer (default)**: single-threaded. `RecordFrame()` runs the whole pipeline — validation, post-processor, serialization, chunk/journal I/O — on the caller's thread. +- **Writer async I/O** (`async_io = true`): exactly ONE extra thread per writer, owned by `AsyncSinkAdapter`. It is the only thread that touches the inner sink after session start; the caller thread only enqueues. Frame accept/reject, the post-processor, and the snapshot APIs all stay on the caller's thread, so the writer's single-threaded contract is unchanged from user code's point of view. Shared state is a mutex-guarded FIFO plus an atomic failure latch; backpressure blocks the producer on a bounded queue, and `Drain()` / `Stop()` are the synchronization points. Verified race-free under ThreadSanitizer. - **Reader chunk loading**: `std::async` with `std::stop_token`. Up to 3 concurrent chunk loads. Chunks are loaded in priority order (active chunk first, then the window around it). - **Chunk eviction**: Automatic when the sliding window moves. Eviction fires `OnChunkEvicted` on the `ReaderChunkState`. - **Logger**: Thread-safe singleton. Sinks are called under a lock; formatting happens before the lock. diff --git a/docs/PERFORMANCE.md b/docs/PERFORMANCE.md index f29af30..8563152 100644 --- a/docs/PERFORMANCE.md +++ b/docs/PERFORMANCE.md @@ -98,6 +98,7 @@ Ratio shrinks on big frames because hashing itself becomes non-trivial — but t - **No competitor comparison.** We measure VTX against itself. - **`items_per_second` in google/benchmark uses CPU time, not wall time.** Where wall ≫ CPU (async I/O), that metric overstates user-observable throughput. For customer-facing claims use the wall-time column. - **One known fixture bug** — `BM_AccessorRandomWithinBucket` inflates its own counter 2× due to a duplicate-push in the shuffle setup. Flagged for a follow-up fix; everything else is trustworthy. +- **Synchronous writer only.** Every writer number here is the default (synchronous) sink, where `RecordFrame` pays serialization, compression, `fsync` and the chunk-boundary flush inline. They say nothing about `async_io = true`, which moves that work to an I/O worker thread. **No async figures are published yet**: the async-aware benchmark — per-`RecordFrame` latency histograms (p50/p99/max) measured *inside* the loop, excluding the `Stop()` drain — is not in the suite yet. `BM_WriterDurabilityTier` as it stands includes `Stop()` in its per-iteration time, so it cannot show the difference. Until that lands, treat the async win as *architectural* (the fsync and the boundary hitch leave the capture loop) rather than a measured number. ## Running benchmarks locally diff --git a/docs/POST_PROCESSING.md b/docs/POST_PROCESSING.md index 3b188b7..a665086 100644 --- a/docs/POST_PROCESSING.md +++ b/docs/POST_PROCESSING.md @@ -53,6 +53,8 @@ The writer is **single-threaded by design**: `RecordFrame()` is called sequentia This is different from the reader-side processor model some other VTX features use (e.g. async chunk loads). The writer's simplicity is deliberate. +**Async I/O does not change any of this.** With `async_io = true` the writer still runs validation, the post-processor, and serialization on your calling thread -- only the chunk/journal disk writes move to the I/O worker, and that worker never touches a frame the processor can still see (it consumes an already-serialized copy). `Process` therefore keeps its caller-thread affinity: it is safe to read game state from it under async exactly as it is under the synchronous writer. Preserving this guarantee is precisely why the async cut was made between the writer and the sink, rather than moving the whole pipeline onto a worker. + ## API at a glance ```cpp @@ -324,7 +326,7 @@ Avoid resolving `PropertyKey` per `Process()` call -- the lookup hashes by na - **Re-serializing mutated bytes back to disk on the reader side.** Not this feature. Reader-side mutations would diverge from on-disk wire bytes -- documented contract. - **Mutating raw byte spans.** `GetRawFrameBytes` returns on-disk truth by design. -- **Async processors.** `Process` is synchronous on the writer's calling thread. Long-running work blocks `RecordFrame`. +- **Async processors.** `Process` is synchronous on the writer's calling thread. Long-running work blocks `RecordFrame`. (Not to be confused with the writer's `async_io` option, which offloads only the sink's disk I/O -- it does not move `Process` off your thread, by design.) - **Replay-level metadata emission API.** Processors that produce per-replay artefacts (KeyFrames, stats) expose them via their own getters -- the SDK does not add a metadata channel. ## See also diff --git a/docs/SDK_API.md b/docs/SDK_API.md index 7a48ba5..8af24b8 100644 --- a/docs/SDK_API.md +++ b/docs/SDK_API.md @@ -263,6 +263,8 @@ writer->Stop(); // Write footer and close file If the facade is destroyed without an explicit `Stop()`, its destructor finalizes the replay as a best-effort fallback, so a dropped writer still yields a readable `.vtx`. +With `async_io = true` these two calls keep working, but `Flush()` is no longer a durability barrier -- it closes the current chunk and *enqueues* the write. `Stop()` still blocks until everything is durable, and `Drain()` is the mid-recording barrier. See [Async (non-blocking) recording](#async-non-blocking-recording). + ### Crash recovery A recording that dies before `Stop()` -- process crash or power loss, mid-chunk or mid-frame -- is **recoverable up to the last recorded frame**. While recording, the file sink maintains a write-ahead sidecar **`.vtx.recovery`**: every recorded frame is journaled before it joins the pending batch, and every flushed chunk is committed to the journal *after* its bytes are durable in the `.vtx` (data-before-journal ordering, append-only log, per-record checksums). On a clean `Stop()` the sidecar is deleted -- its presence at open time is the signal of an unclean shutdown. @@ -290,13 +292,44 @@ auto ctx = VTX::OpenReplayFile(path); // a repaired file opens like any One property of salvaged files to know about: the in-flight frames that only existed in the journal are re-appended as **one-frame chunks**, so a recovery with a large pending batch reads sequentially ~3x slower than a cleanly chunked file (measured in `BM_ReaderRecoveredTail`). For hot-path use, transcode the salvaged file with the public API -- open it, drain every frame with its footer times into a fresh writer, `Stop()`. `created_utc` round-trips exactly; `game_time` re-enters through the float-seconds register and can lose 1 tick (100 ns) per frame. -Durability knobs on the file sink (`ChunkedFileSink::Config`; facade users currently get the defaults): +Durability knobs (on `WriterFacadeConfig`, and on `ChunkedFileSink::Config` for direct sink users): | Knob | Default | Meaning | |------|---------|---------| | `durable_writes` | `true` | fsync every chunk and journal record to physical media -- survives **power loss**. Set `false` to only flush to the OS: cheaper, still survives a **process crash**. | | `enable_recovery_journal` | `true` | Maintain the `.recovery` sidecar. Opting out removes any stale sidecar at session start; a crash then leaves an unrecoverable (footerless) file. | -| `journal_compact_threshold_bytes` | `0` (64 MB) | How many superseded journal bytes accrue before the sidecar is compacted (rewritten via an atomic rename). | +| `journal_compact_threshold_bytes` | `0` (64 MB) | How many superseded journal bytes accrue before the sidecar is compacted (rewritten via an atomic rename). Sink config only. | + +### Async (non-blocking) recording + +By default recording is synchronous: `RecordFrame` runs the whole pipeline on the caller's thread, so the frame that crosses a chunk boundary pays the full serialize + zstd + write + `fsync` bill inline, and every journaled frame pays an `fsync`. Set **`async_io = true`** to move all chunk/journal disk I/O to a dedicated worker thread -- accepting a frame then never waits on disk. + +```cpp +VTX::WriterFacadeConfig config; +config.output_filepath = "output.vtx"; +config.schema_json_path = "schema.json"; +config.async_io = true; // opt in: I/O runs on a worker thread +config.async_max_queue_frames = 0; // 0 -> 2 * chunk_max_frames + +auto writer = VTX::CreateFlatBuffersWriterFacade(config); +// ... writer->RecordFrame(...) never blocks on serialization/fsync/chunk flush ... +writer->Drain(); // durability barrier: block until every accepted frame is on disk +writer->Stop(); // drains, writes the footer, joins the worker +``` + +Validation and rejection stay synchronous -- `TryRecordFrame` still accepts or rejects on the caller's thread exactly as before, and a registered post-processor still runs on the caller's thread. Only the chunk/journal I/O is deferred. Ordering and crash-recovery are unchanged: one FIFO drained by one worker reproduces the exact synchronous on-disk sequence, so a repaired async recording is identical to a repaired synchronous one. + +What changes: + +- **Durability lag.** A frame is crash-recoverable only once its queued work is durable. A hard kill loses the not-yet-durable suffix (bounded by the queue); recovery still yields a clean contiguous prefix. **`Drain()`** and `Stop()` are the zero-lag synchronization points. +- **`Flush()` is no longer a durability barrier under async** -- it closes the current chunk and *enqueues* its write, but returns before the bytes are on disk. Use **`Drain()`** when you need durability (e.g. before a checkpoint). +- **Backpressure.** `async_max_queue_frames` bounds the queue by item count. If the disk cannot keep up, the caller *blocks* on a full queue (degrading to synchronous behavior) rather than dropping frames or growing memory without bound. `GetQueueDepth()` reports the current depth. +- **I/O failure.** If the worker hits a durable-write failure it aborts the recording: the queue is discarded, the file + journal handles are released (the partial `.vtx` becomes repair-ready immediately), and the next `TryRecordFrame` returns **`VtxErrorCode::SinkFailed`**. `GetLastError()` returns the latched error; `PipelineReport::sink_failed` counts these separately from validation rejections. + +| Knob | Default | Meaning | +|------|---------|---------| +| `async_io` | `false` | Move chunk/journal I/O to a worker thread. | +| `async_max_queue_frames` | `0` (-> `2 * chunk_max_frames`) | Backpressure bound, in queued items. A full queue blocks the caller. | ### One call: `WriteReplay` diff --git a/docs/research/async_writer_live_mode.md b/docs/research/async_writer_live_mode.md new file mode 100644 index 0000000..4f7ee96 --- /dev/null +++ b/docs/research/async_writer_live_mode.md @@ -0,0 +1,247 @@ +# R&D: "Live" async writer — recording while chunks flush to disk + +Status: IMPLEMENTED (Option A) — design v2, revised after a 3-lens adversarial review +(17 confirmed findings incorporated; see §7). Landed on branch `feature/async-sink-adapter`: +`AsyncSinkAdapter` (async_sink_adapter.h), facade `async_io` knobs, +`Drain()`/`GetLastError()`/`GetQueueDepth()`, group-commit journal primitives, the +`SinkFailed` surfacing hook, and `tests/writer/test_async_sink.cpp`. + +VERIFIED: the full test plan of §5 is implemented and green, including the real- +`TerminateProcess` async-ON kill matrix (§5.3) in `test_crash_recovery.cpp`. Race freedom +was proven under ThreadSanitizer (gcc-13/WSL, `VTX_SANITIZE=thread`): 95 writer/crash tests +clean, and the harness itself was validated by confirming TSan reports a deliberately +injected race in the adapter. One hardening fix came out of the audit: an enqueue after the +worker had exited could block forever on a full queue (see `worker_exited_`). + +Remaining follow-ups (not blocking): the async-aware per-`RecordFrame` latency benchmark +(§5.6); the network-sink decorator (§4.7); and the pre-existing inert `max_bytes` +FlatBuffers chunking fix (§8), all filed separately. +Author: Zenos Interactive. + +## 1. Problem + +The writer is fully synchronous. Every `RecordFrame` call runs the entire pipeline on +the calle +r's thread, and the frame that crosses a chunk boundary pays the whole chunk +bill inline (serialize N frames + zstd + write + fsync + journal commit). Measured on +NVMe (`BM_WriterDurabilityTier`, 200 small frames): + +| caller-thread cost | per frame | +|-------------------------------|-----------| +| no journal | ~19 us | +| journal, flush-only | ~30 us | +| journal, fsync (default) | ~290 us | +| boundary frame (chunk flush) | + several ms to tens of ms | + +Goal: **accepting new frames must not wait for chunk/journal I/O.** Durability and +crash-recovery guarantees must degrade in a bounded, documented way only. + +## 2. Current pipeline (what exactly blocks) + +Per `TryRecordFrame` (writer.h), all on the caller thread, in order: + +1. Timer snapshot + `AddTimeRegistry` + `ResolveGameTimes` — cheap, **stateful, ordered** +2. Post-processor `Process` — user code, mutates frame +3. `NormalizeBucketsToSchema` + `FinalizeFrame` (content_hash) — CPU, moderate +4. `Serializer::FromNative` — CPU, moderate +5. `sink_.JournalFrame`: copy frame -> 1-frame `SerializeChunk` -> zstd -> write -> **fsync** +6. On boundary, before 5: `Flush()` -> `SaveChunk`: `SerializeChunk(N)` -> zstd -> write -> **fsync** -> journal C/T commit -> **fsync** + +Steps 5-6 are ~95%+ of the cost. Steps 1-4 are ~10-15 us. + +## 3. Design options + +### Option A — Async SINK decorator (recommended) + +Cut between the writer core (steps 1-4) and the sink I/O (steps 5-6). A new sink +policy `AsyncSinkAdapter` decorates `ChunkedFileSink`: + +- `JournalFrame` / `SaveChunk` / `Close` become **enqueue** operations on a bounded + FIFO; a single dedicated I/O worker drains the queue in order, calling the inner + sink's real implementations. +- `OnSessionStart` and `JournalTiming` stay **synchronous** (session setup completes + before recording; the 'S' record must precede everything — queuing JournalTiming + would race the first frames). + +**Why it wins over B/C:** synchronous accept/reject preserved; post-processor keeps +its documented caller-thread affinity; `GetLastFinalizedFrame`/`FindEntity` intact; +ordering trivially preserved (one FIFO, one worker); zero writer-core redesign. + +### Option B — Full producer/consumer of native frames + +Worker runs the entire writer. Saves ~14 us/frame over A (0.09% of a 60 fps budget) +at the cost of async rejections, post-processor thread hazards, and cross-thread +snapshot APIs. **Rejected as an SDK feature; layerable app-side on top of A.** + +### Option C — Validate sync, defer the rest + +Inherits B's post-processor problems for ~0 win over A. Rejected. + +## 4. Detailed design (Option A, revised) + +### 4.1 Work items and queue + +``` +struct JournalItem { FrameType frame_copy; int32 index; int64 gt, cu; }; +struct ChunkItem { vector> frames; // MOVED from pending_frames_ + vector batch_utc; int32 start, total; }; +struct CloseItem { int32 total_frames; double duration_seconds; + vector game_times, created_utc; // OWNED plain values + vector gaps, segments; }; +``` + +- **`SessionFooter` is a non-owning view type (four raw vector pointers) and must + never be stored in or moved through the queue.** `Close` deep-copies the four + vectors into the `CloseItem` as owned members; the worker constructs a fresh + `SessionFooter` at inner-`Close` call time, pointing into the item it is currently + processing (address-stable for the duration of the call). +- `SaveChunk` moves the frames vector into the item (no copy; the writer's + subsequent `pending_frames_.clear()` is a no-op on moved-out storage). +- `JournalFrame` copies the FrameType (parity with today's sync sink, which already + copies). **If journaling is disabled or its open failed, the adapter enqueues + nothing** (it knows `enable_recovery_journal` and the inner journal state at + session start) — no dead copies. +- Queue: `std::mutex` + `condition_variable`, SPSC in practice. All waits are + failure-aware (§4.4). + +### 4.2 Ordering & crash-recovery invariants + +The worker executes items strictly FIFO via the inner sink, so the on-disk effect +sequence is exactly today's: data-before-journal, F-record contiguity, append-only +journal + compaction all hold by construction; `RepairReplayFile` needs zero changes. + +**What changes — durability lag, stated honestly:** a frame is crash-recoverable +only once its JournalItem (or containing batch, §4.8) is durable. The lag is bounded +by the queue capacity (§4.3), NOT "a few ms": on a saturated slow disk the queue can +hold up to `async_max_queue_frames` accepted-but-not-yet-durable frames. A hard kill +loses that suffix; contiguity still yields a clean recovered prefix. `Stop()` and +`Drain()` are the zero-lag synchronization points. + +### 4.3 Backpressure + +- **Bound by ITEM COUNT, not bytes**: `async_max_queue_frames` (default + `2 * chunk_max_frames`). Byte-accounting is not implementable today — + `FlatBuffersVtxPolicy::GetFrameSize()` returns 0 (see §8, pre-existing bug), and + journal payload sizes are only known after worker-side serialization. +- Policy on full: **block the caller** (degrade toward sync behavior; never drop, + never unbounded memory) — except on latched failure (§4.4). +- **Oversized-item rule**: a ChunkItem always counts as one item and is admitted + whenever the queue is not full; item-count bounding makes the + "single item larger than the cap" deadlock of byte-bounding structurally + impossible. +- Telemetry: `GetQueueDepth()`; peak-memory formula documented in §6. + +### 4.4 Failure protocol (fully specified) + +On the first inner-sink I/O failure (DurableFile `Good()` latch / exception): + +1. Worker sets `atomic failed_` + stores a `VtxError`, then **notify_all**. +2. Worker **halts consumption and discards the remaining queue** (the recording is + dead; writing more would commit journal records for non-durable data). +3. Worker **closes the inner sink's files and journal handles** — this releases the + deny-write handles so the on-disk state (valid journal + data up to the failure + point) becomes repair-ready immediately, even while the app keeps running. +4. **Every blocking wait is failure-aware**: a producer blocked on a full queue, a + `Drain()`, a `Stop()`, or the destructor wakes on `failed_` and returns promptly + (Stop/Drain report the stored error; the blocked enqueue drops its item). + No wait predicate can hang once the worker stops consuming. +5. Surfacing to the caller: the writer core gains one cheap hook — at the top of + `TryRecordFrame`, `if constexpr (sink has HasFailed())` and it returns true, + return `MadeRejected(VtxErrorCode::SinkFailed, ...)`. (Detection idiom, no + interface change for existing sinks.) `PipelineReport::Account` classifies + `SinkFailed` as its own counter — not `validation_errors`. +6. Facade exposes `GetLastError()`. + +### 4.5 Lifecycle + +- `Close(footer)`: enqueue CloseItem, worker drains, inner `Close` writes footer + + deletes journal, worker exits. `Stop()` blocks until fully durable (or returns + the latched error promptly per §4.4). +- Destructor without Stop: drain queue (more data becomes recoverable), close files + without footer, join worker — failure-aware, cannot hang. +- **`Flush()` semantics CHANGE under async (documented, opt-in flag):** it still + closes the current chunk but now only *enqueues* the write — it is no longer a + durability barrier. The barrier is the new `Drain()`. This must be called out in + SDK_API.md's async section explicitly. + +### 4.6 Config surface + +``` +bool async_io = false; // opt-in v1 +size_t async_max_queue_frames = 2 * chunk_max_frames; +bool durable_writes = true; // exposed at facade at last +bool enable_recovery_journal = true; +``` + +Facade constructs `AsyncSinkAdapter>` vs `ChunkedFileSink

` +(two `WriterFacadeImpl` instantiations, runtime-selected). + +### 4.7 Network sink + +Same decorator applies later (stalled TCP peer stops hitching the game). Out of +scope v1. + +### 4.8 Group commit (required, not optional — makes slow disks close) + +Without batching, the worker inherits fsync-per-frame: on an HDD (~10-20 ms/fsync) +that is 50-100 frames/s < 60 fps — the queue fills and async degenerates to +blocking. Therefore the worker uses **group commit**: it drains everything currently +queued, serializes each JournalItem, appends ALL of them to the journal, and issues +ONE `SyncOrFlush` for the batch (ChunkItems keep their own internal ordering: +chunk-write+fsync, then C/T commit+fsync, as today). Recovery semantics are +unchanged — a crash lands on a batch boundary, still a clean contiguous prefix. +Amortized fsync cost makes HDD steady-state close with large margin, and NVMe lag +drop to sub-ms. + +## 5. Test plan (revised) + +1. **Order equivalence**: instrumented fake inner sink; async call sequence == + sync call sequence (batching allowed to coalesce J-syncs only). +2. **Byte equivalence**: same inputs -> async file == sync file bit-for-bit + (same-second pair technique). This also pins the CloseItem/SessionFooter + reconstruction (§4.1) — a dangling/garbage footer cannot pass it. +3. **Crash**: real TerminateProcess kills with async ON (both durability modes, + compaction cadence 1): recovered file = clean contiguous prefix, exact times. +4. **Backpressure**: queue cap 2 + artificially slow inner sink -> caller blocks, + zero loss, order preserved; then latch a failure while a producer is blocked -> + producer unblocks promptly with SinkFailed; Stop() returns the error; files are + repair-ready (deny-write released) while the process still lives. +5. **Error injection**: failing inner sink mid-session -> journal repairable, + RepairReplayFile succeeds on the prefix. +6. **Benchmark**: new async-aware benchmark measuring per-`RecordFrame` latency + histograms EXCLUDING Stop() (manual timers inside the loop, p50/p99/max), + sync vs async, all durability tiers. (`BM_WriterDurabilityTier` as-is includes + Stop() drain in its per-iteration time and cannot show the difference.) +7. Existing suite green with async OFF; a representative async-ON variant of the + crash-recovery E2E tests (the fabrication-based tests are sink-agnostic and + don't need variants; the raw-writer E2E ones parametrize on sink type). + +## 6. Memory model (documented) + +Peak native-frame memory ~= `pending_frames_` (up to chunk_max_frames frames) ++ queued JournalItem copies (up to async_max_queue_frames) + one moved ChunkItem +in flight — i.e. up to ~3x a chunk of frames with defaults. The v1.5 optimization +(pending_frames_ as `shared_ptr`, journal items share instead of copy) +removes the copy term entirely and is the planned follow-up, not a blocker. + +## 7. Adversarial review disposition (v1 -> v2) + +17 confirmed findings, all incorporated: CloseItem/SessionFooter dangling-view +redesign (4.1); JournalTiming kept synchronous (4.1); no dead copies when journaling +is off (4.1); honest durability-lag statement (4.2); item-count bound replacing the +unenforceable byte bound + oversized-item deadlock removal (4.3); fully specified +failure protocol with failure-aware waits, queue discard, early handle release, the +TryRecordFrame surfacing hook and PipelineReport classification (4.4); Flush() +semantic change called out (4.5); group commit as a requirement with HDD math +(4.8); benchmark methodology fix + failure/backpressure tests (5); peak-memory +formula (6). + +## 8. Collateral pre-existing findings (filed separately, not part of this design) + +- **`FlatBuffersVtxPolicy::GetFrameSize()` returns 0** (flatbuffers_vtx_policy.cpp) + -> `ThresholdChunkPolicy::max_bytes` is INERT for FlatBuffers recordings: chunks + split only by max_frames, so a byte budget is silently ignored. Needs a native- + frame size estimator or serialized-size feedback. +- Consequently `ByteBudgetChunkingCrashRecovery` passes for the wrong reason on the + chunk-count assertion (the >=3 chunks come from pending-frame recovery, not byte + splitting). Test needs its assertion corrected alongside the fix. diff --git a/sdk/include/vtx/common/vtx_diagnostics.h b/sdk/include/vtx/common/vtx_diagnostics.h index 0cd440a..ab19e87 100644 --- a/sdk/include/vtx/common/vtx_diagnostics.h +++ b/sdk/include/vtx/common/vtx_diagnostics.h @@ -61,6 +61,8 @@ namespace VTX { ReplayOpenFailed, ReplayNotReady, + + SinkFailed, ///< Async sink's I/O worker latched a durable-write failure; recording aborted. }; inline std::string_view ToString(VtxErrorCode code) { @@ -101,6 +103,8 @@ namespace VTX { return "ReplayOpenFailed"; case VtxErrorCode::ReplayNotReady: return "ReplayNotReady"; + case VtxErrorCode::SinkFailed: + return "SinkFailed"; } return "Unknown"; } diff --git a/sdk/include/vtx/writer/core/vtx_writer_facade.h b/sdk/include/vtx/writer/core/vtx_writer_facade.h index fa103ba..4aba810 100644 --- a/sdk/include/vtx/writer/core/vtx_writer_facade.h +++ b/sdk/include/vtx/writer/core/vtx_writer_facade.h @@ -31,6 +31,20 @@ namespace VTX { virtual void Stop() = 0; virtual VTX::SchemaRegistry& GetSchema() = 0; + // Durability barrier for async recordings (async_io=true): block until every frame + // accepted so far is durable on disk. Under async, Flush() only closes the current chunk + // and enqueues its write -- it is NOT a durability barrier; Drain() is. For synchronous + // recordings every accepted frame is already durable, so Drain() returns immediately. + // Returns the latched sink error if the async I/O worker failed, else a None error. + virtual VtxError Drain() = 0; + + // The last async-sink I/O error, or a None error if none has occurred (always None for + // synchronous recordings). Pairs with TryRecordFrame returning VtxErrorCode::SinkFailed. + virtual VtxError GetLastError() const = 0; + + // Depth of the async I/O queue (0 for synchronous recordings). Telemetry for backpressure. + virtual size_t GetQueueDepth() const = 0; + // The processor's Process() runs on every RecordFrame() call, // after timer validation and BEFORE serialisation to disk: its // mutations are what end up in the .vtx file. Call before @@ -56,6 +70,14 @@ namespace VTX { bool retain_finalized_snapshot = false; bool create_output_dirs = true; IFileSinkPerfObserver* perf_observer = nullptr; ///< optional file-sink timings; null = no-op. + + // --- Durability / crash-recovery (exposed at the facade at last) --- + bool durable_writes = true; ///< fsync each chunk/journal commit (crash + power-loss safe). + bool enable_recovery_journal = true; ///< maintain the ".recovery" sidecar for crash recovery. + + // --- Async I/O (opt-in) --- + bool async_io = false; ///< move chunk/journal I/O to a worker thread; caller never waits on disk. + size_t async_max_queue_frames = 0; ///< backpressure bound (items). 0 -> 2 * chunk_max_frames. }; enum class SerializationFormat : uint8_t { diff --git a/sdk/include/vtx/writer/core/vtx_writer_result.h b/sdk/include/vtx/writer/core/vtx_writer_result.h index 442ec84..b502337 100644 --- a/sdk/include/vtx/writer/core/vtx_writer_result.h +++ b/sdk/include/vtx/writer/core/vtx_writer_result.h @@ -59,6 +59,7 @@ namespace VTX { size_t skipped = 0; size_t validation_errors = 0; size_t timer_errors = 0; + size_t sink_failed = 0; ///< Rejections caused by a latched async-sink I/O failure. std::vector errors; size_t Total() const { return written + rejected + skipped; } @@ -71,6 +72,11 @@ namespace VTX { ++rejected; if (result.error.code == VtxErrorCode::GameTimeRejected) { ++timer_errors; + } else if (result.error.code == VtxErrorCode::SinkFailed) { + // A dead sink is an I/O condition, not a rejected-frame validation problem: keep + // it in its own bucket so callers can tell "this frame was malformed" from "the + // recording is gone". + ++sink_failed; } else { ++validation_errors; } diff --git a/sdk/include/vtx/writer/core/writer.h b/sdk/include/vtx/writer/core/writer.h index 3e3830e..417daa6 100644 --- a/sdk/include/vtx/writer/core/writer.h +++ b/sdk/include/vtx/writer/core/writer.h @@ -80,6 +80,16 @@ namespace VTX { RecordResult TryRecordFrame(VTX::Frame& native_frame, const VTX::GameTime::GameTimeRegister& game_time_register) { + // Async sinks run their I/O on a worker thread; once it latches an I/O failure the + // recording is dead. Reject cheaply and synchronously here rather than accepting + // frames the worker will never durably write. Compiles out for synchronous sinks. + if constexpr (requires(const SinkPolicy& s) { s.HasFailed(); }) { + if (sink_.HasFailed()) { + return RecordResult::MadeRejected(VtxErrorCode::SinkFailed, + "async sink I/O failed; recording aborted"); + } + } + timer_.CreateSnapshot(); if (!timer_.AddTimeRegistry(game_time_register)) { @@ -190,6 +200,45 @@ namespace VTX { sink_.Close(footer_data); } + // --- Async-sink surface (defaults for synchronous sinks; selected via if constexpr) --- + + // Durability barrier: block until every frame accepted so far is durable on disk. For a + // synchronous sink every accepted frame is already durable, so this is a no-op. + VtxError Drain() { + if constexpr (requires { sink_.Drain(); }) { + return sink_.Drain(); + } else { + return VtxError {}; + } + } + + // The last latched async-sink I/O error, or a default (None) error. + VtxError GetLastError() const { + if constexpr (requires { sink_.GetLastError(); }) { + return sink_.GetLastError(); + } else { + return VtxError {}; + } + } + + // True once an async sink has latched an I/O failure; always false for synchronous sinks. + bool HasSinkFailed() const { + if constexpr (requires { sink_.HasFailed(); }) { + return sink_.HasFailed(); + } else { + return false; + } + } + + // Depth of the async sink's pending-I/O queue (0 for synchronous sinks). + size_t GetQueueDepth() const { + if constexpr (requires { sink_.GetQueueDepth(); }) { + return sink_.GetQueueDepth(); + } else { + return 0; + } + } + VTX::SchemaRegistry& GetRegistry() { return registry_; } const VTX::Frame* GetLastFinalizedFrame() const { return has_last_finalized_ ? &last_finalized_ : nullptr; } diff --git a/sdk/include/vtx/writer/policies/sinks/async_sink_adapter.h b/sdk/include/vtx/writer/policies/sinks/async_sink_adapter.h new file mode 100644 index 0000000..fe3e6d2 --- /dev/null +++ b/sdk/include/vtx/writer/policies/sinks/async_sink_adapter.h @@ -0,0 +1,421 @@ +/** + * @file async_sink_adapter.h + * @brief Sink decorator that moves chunk/journal I/O off the caller (game) thread. + * + * @details The synchronous writer runs its whole pipeline on the caller's thread, so the + * frame that crosses a chunk boundary pays the full serialize + zstd + write + fsync bill + * inline, and every journaled frame pays an fsync. `AsyncSinkAdapter` cuts between + * the writer core and the sink I/O: `JournalFrame` / `SaveChunk` / `Close` become enqueue + * operations on a bounded FIFO, and a single dedicated I/O worker drains the queue in order, + * calling the real `InnerSink` (a `ChunkedFileSink`). `JournalTiming` and `OnSessionStart` + * stay synchronous -- session setup and the 'S' record must precede everything. + * + * Because one worker executes items strictly FIFO through the inner sink, the on-disk effect + * sequence is exactly the synchronous one (data-before-journal, F-record contiguity, + * append-only journal + compaction all hold by construction); the repair path needs zero + * changes. What changes is DURABILITY LAG: a frame is crash-recoverable only once its item + * (or containing group-commit batch) is durable. The lag is bounded by the queue capacity; + * a hard kill loses that not-yet-durable suffix, and crash recovery still yields a clean + * contiguous prefix. `Drain()` and `Stop()` are the zero-lag synchronization points. + * + * Backpressure is bounded by ITEM COUNT (`async_max_queue_frames`), not bytes: the only + * per-frame size hook (`GetFrameSize`) returns 0 for the default FlatBuffers format, and + * journal payload sizes are known only after worker-side serialization. On a full queue the + * caller blocks (degrading toward synchronous behavior -- never dropping, never growing + * memory without bound), except once a failure is latched. + * + * Group commit: the worker drains everything currently queued and, for a run of journal + * frames, appends all their F records and issues ONE journal fsync for the batch. Chunk items + * keep their own internal ordering (chunk write+fsync, then C/T commit+fsync). This amortizes + * the per-frame fsync so a slow disk (HDD ~10-20 ms/fsync) can keep up with 60 fps instead of + * degenerating to blocking. + * + * Failure protocol: on the first inner-sink I/O failure the worker latches an error, discards + * the remaining queue, and closes the inner file + journal handles WITHOUT a footer (releasing + * the deny-write share so the on-disk state becomes repair-ready immediately). Every blocking + * wait -- a full-queue producer, `Drain()`, `Close()`/`Stop()`, the destructor -- is + * failure-aware and returns promptly. The writer surfaces the latch as `SinkFailed` on the + * next `TryRecordFrame` via the `HasFailed()` hook. + * + * Threading: `JournalTiming`/`OnSessionStart` run on the caller thread before the worker + * starts; after that ONLY the worker thread touches `inner_`. The caller thread only enqueues + * (and reads the atomic `HasFailed()` / mutex-guarded `GetLastError()`/`GetQueueDepth()`). + * + * @author Zenos Interactive + */ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "vtx/common/vtx_diagnostics.h" +#include "vtx/common/vtx_types.h" + +namespace VTX { + + template + class AsyncSinkAdapter { + public: + // Forward the type surface the writer core reads off its sink policy. + using SerializerPolicy = typename InnerSink::SerializerPolicy; + using FrameType = typename InnerSink::FrameType; + using SchemaType = typename InnerSink::SchemaType; + using HeaderType = typename InnerSink::HeaderType; + + struct Config { + typename InnerSink::Config inner; + // Max queued items before an enqueue blocks the caller. 0 -> kFallbackQueueItems. + // The facade resolves 0 to 2 * chunk_max_frames. + size_t async_max_queue_frames = 0; + }; + + explicit AsyncSinkAdapter(Config config) + : inner_(std::move(config.inner)) + , max_queue_items_(config.async_max_queue_frames == 0 ? kFallbackQueueItems + : config.async_max_queue_frames) {} + + AsyncSinkAdapter(const AsyncSinkAdapter&) = delete; + AsyncSinkAdapter& operator=(const AsyncSinkAdapter&) = delete; + + ~AsyncSinkAdapter() { + // Close() joins the worker on a clean stop. If we get here with the worker still + // running (no Close, e.g. an exception unwound the writer), shut it down: it drains + // what is queued (making more data recoverable) then exits WITHOUT writing a footer, + // so the recovery journal survives and the file is repairable. Never write a footer + // here -- the caller never asked to finalize. + if (worker_.joinable()) { + { + std::lock_guard lk(mu_); + stop_requested_ = true; + } + not_empty_.notify_all(); + worker_.join(); + } + } + + // --- Synchronous session setup (runs on the caller thread, before the worker) --- + + void JournalTiming(float fps, bool is_increasing) { inner_.JournalTiming(fps, is_increasing); } + + void OnSessionStart(const SchemaType& schema) { + inner_.OnSessionStart(schema); + // Whether the recovery journal is actually live is only known after OnSessionStart + // (it may be disabled by config or fail to open). Cache it so JournalFrame can skip + // enqueuing -- and copying -- frames when there is nowhere to journal them. + journal_active_ = inner_.IsJournalActive(); + StartWorker(); + } + + // --- Enqueue path (runs on the caller thread) --- + + void JournalFrame(const FrameType& frame, int32_t frame_index, int64_t game_time, int64_t created_utc) { + if (!journal_active_) + return; // no dead copies when journaling is off/unavailable + JournalItem item; + item.frame = frame; // copy, exactly as the synchronous sink does + item.index = frame_index; + item.game_time = game_time; + item.created_utc = created_utc; + Enqueue(WorkItem {std::move(item)}); + } + + void SaveChunk(std::vector>& frames, const std::vector& created_utc, + int32_t start_frame, int32_t total_frames) { + ChunkItem item; + item.frames = std::move(frames); // steal the batch; the writer clears the moved-from vector next + item.batch_utc = created_utc; + item.start_frame = start_frame; + item.total_frames = total_frames; + Enqueue(WorkItem {std::move(item)}); + } + + void Close(const SessionFooter& footer) { + // SessionFooter is a non-owning view (four raw vector pointers into the caller's + // timer). Deep-copy the vectors into the item NOW, while they are alive; the worker + // rebuilds a fresh SessionFooter pointing into the item at inner-Close time. + CloseItem item; + item.total_frames = footer.total_frames; + item.duration_seconds = footer.duration_seconds; + if (footer.game_times) + item.game_times = *footer.game_times; + if (footer.created_utc) + item.created_utc = *footer.created_utc; + if (footer.gaps) + item.gaps = *footer.gaps; + if (footer.segments) + item.segments = *footer.segments; + // If the enqueue is dropped (a failure was already latched) the worker is already + // exiting; either way, join it so Close() is a full durability barrier. + Enqueue(WorkItem {std::move(item)}); + JoinWorker(); + } + + // --- Async-only surface (reached through the writer / facade passthroughs) --- + + // Cheap, lock-free check the writer polls at the top of TryRecordFrame. + bool HasFailed() const noexcept { return failed_.load(std::memory_order_acquire); } + + // Block until every item enqueued before this call has been processed AND the journal's + // group-commit batch is durable -- the durability barrier that Flush() no longer is under + // async. Returns the latched error if the sink failed, else a default (None) error. + VtxError Drain() { + uint64_t id = 0; + { + std::unique_lock lk(mu_); + // No worker left to answer a barrier: after a clean close everything it held is + // already durable, so there is nothing to wait for (and waiting would hang). + if (failed_.load(std::memory_order_acquire) || worker_exited_) + return failed_.load(std::memory_order_acquire) ? last_error_ : VtxError {}; + not_full_.wait(lk, [this] { + return queue_.size() < max_queue_items_ || failed_.load(std::memory_order_acquire) || + worker_exited_; + }); + if (failed_.load(std::memory_order_acquire) || worker_exited_) + return failed_.load(std::memory_order_acquire) ? last_error_ : VtxError {}; + id = ++next_barrier_id_; + queue_.push_back(WorkItem {BarrierItem {id}}); + not_empty_.notify_one(); + } + std::unique_lock lk(mu_); + drained_.wait(lk, [this, id] { + return last_barrier_done_ >= id || failed_.load(std::memory_order_acquire) || worker_exited_; + }); + return failed_.load(std::memory_order_acquire) ? last_error_ : VtxError {}; + } + + VtxError GetLastError() const { + std::lock_guard lk(mu_); + return last_error_; + } + + size_t GetQueueDepth() const { + std::lock_guard lk(mu_); + return queue_.size(); + } + + private: + static constexpr size_t kFallbackQueueItems = 2000; // 2 * default chunk_max_frames + + struct JournalItem { + FrameType frame; + int32_t index = 0; + int64_t game_time = 0; + int64_t created_utc = 0; + }; + struct ChunkItem { + std::vector> frames; + std::vector batch_utc; + int32_t start_frame = 0; + int32_t total_frames = 0; + }; + struct CloseItem { + int32_t total_frames = 0; + double duration_seconds = 0.0; + std::vector game_times; + std::vector created_utc; + std::vector gaps; + std::vector segments; + }; + struct BarrierItem { + uint64_t id = 0; + }; + using WorkItem = std::variant; + + static VtxError MakeError(VtxErrorCode code, std::string message) { + VtxError e; + e.code = code; + e.severity = Severity::Error; + e.message = std::move(message); + e.source_api = "AsyncSinkAdapter"; + return e; + } + + void StartWorker() { + if (worker_started_) + return; + worker_started_ = true; + worker_ = std::thread([this] { + WorkerMain(); + // Publish the exit so no producer can block on a queue nobody will drain again + // (see Enqueue/Drain): once the worker is gone, waiting for room is waiting + // forever. Reached on every exit path -- clean close, stop, or latched failure. + { + std::lock_guard lk(mu_); + worker_exited_ = true; + } + not_full_.notify_all(); + drained_.notify_all(); + }); + } + + void JoinWorker() { + if (worker_.joinable()) + worker_.join(); + } + + // Blocks the caller when the queue is full (backpressure). Returns false without + // enqueuing if a failure is latched (the recording is dead, so the item is dropped) or if + // the worker has already exited -- e.g. an enqueue after Stop(), where waiting for room + // would block forever because nothing drains the queue any more. + bool Enqueue(WorkItem&& item) { + std::unique_lock lk(mu_); + not_full_.wait(lk, [this] { + return queue_.size() < max_queue_items_ || failed_.load(std::memory_order_acquire) || worker_exited_; + }); + if (failed_.load(std::memory_order_acquire) || worker_exited_) + return false; + queue_.push_back(std::move(item)); + not_empty_.notify_one(); + return true; + } + + // Run one inner-sink call, catching exceptions into the failure protocol. Returns false + // if the call failed (already latched), telling the worker to stop the batch. + template + bool SafeInner(Fn&& fn) { + try { + fn(); + return true; + } catch (const std::exception& e) { + LatchFailure(MakeError(VtxErrorCode::SinkFailed, std::string("async sink inner threw: ") + e.what())); + return false; + } catch (...) { + LatchFailure(MakeError(VtxErrorCode::SinkFailed, "async sink inner threw unknown exception")); + return false; + } + } + + // First-failure latch: store the error, discard the queue (writing more would commit + // journal records for non-durable data), release the inner handles so the on-disk state + // is repair-ready at once, and wake every waiter. Idempotent. Only the worker calls this, + // so touching inner_ here is race-free. + void LatchFailure(VtxError err) { + const bool first = !failed_.exchange(true, std::memory_order_acq_rel); + { + std::lock_guard lk(mu_); + if (first) + last_error_ = std::move(err); + queue_.clear(); + } + try { + inner_.AbortClose(); + } catch (...) {} + not_full_.notify_all(); + not_empty_.notify_all(); + drained_.notify_all(); + } + + void WorkerMain() { + for (;;) { + std::deque batch; + { + std::unique_lock lk(mu_); + not_empty_.wait(lk, [this] { + return !queue_.empty() || stop_requested_ || failed_.load(std::memory_order_acquire); + }); + if (failed_.load(std::memory_order_acquire)) + return; // failure protocol already ran; nothing left to do + if (queue_.empty() && stop_requested_) + return; // graceful shutdown without a Close: nothing left to write + batch.swap(queue_); // take everything currently queued (the group-commit batch) + } + not_full_.notify_all(); // room freed; unblock producers to refill while we work + + bool journaled_unsynced = false; // F records appended this batch but not yet fsync'd + bool did_close = false; + uint64_t batch_barrier = 0; + + for (WorkItem& item : batch) { + if (failed_.load(std::memory_order_acquire)) + break; + if (auto* j = std::get_if(&item)) { + if (!SafeInner( + [&] { inner_.JournalFrameBatched(j->frame, j->index, j->game_time, j->created_utc); })) + break; + journaled_unsynced = true; + } else if (auto* c = std::get_if(&item)) { + if (!SafeInner( + [&] { inner_.SaveChunk(c->frames, c->batch_utc, c->start_frame, c->total_frames); })) + break; + journaled_unsynced = false; // CommitChunk already fsync'd the journal + } else if (auto* cl = std::get_if(&item)) { + SessionFooter footer; + footer.total_frames = cl->total_frames; + footer.duration_seconds = cl->duration_seconds; + footer.game_times = &cl->game_times; + footer.created_utc = &cl->created_utc; + footer.gaps = &cl->gaps; + footer.segments = &cl->segments; + if (!SafeInner([&] { inner_.Close(footer); })) + break; + journaled_unsynced = false; + did_close = true; + } else if (auto* b = std::get_if(&item)) { + // A Drain() barrier: make everything appended so far durable before it + // reports done. + if (journaled_unsynced) { + if (!SafeInner([&] { inner_.SyncJournal(); })) + break; + journaled_unsynced = false; + } + batch_barrier = b->id; + } + } + + // Group commit: one journal fsync for the batch's F records, unless a chunk / + // close / barrier already synced them. + if (journaled_unsynced && !failed_.load(std::memory_order_acquire)) + SafeInner([&] { inner_.SyncJournal(); }); + + // Primary failure detection: a short write / failed fsync latches the inner file's + // Good() flag without throwing. Catch it here and run the failure protocol. + if (!failed_.load(std::memory_order_acquire) && !inner_.Good()) + LatchFailure(MakeError(VtxErrorCode::SinkFailed, "async sink: durable write to .vtx failed")); + + if (batch_barrier != 0) { + std::lock_guard lk(mu_); + if (batch_barrier > last_barrier_done_) + last_barrier_done_ = batch_barrier; + drained_.notify_all(); + } + + if (failed_.load(std::memory_order_acquire)) + return; // LatchFailure closed the inner sink; worker is done + if (did_close) + return; // clean finalize; worker's job is complete + } + } + + InnerSink inner_; + const size_t max_queue_items_; + + mutable std::mutex mu_; + std::condition_variable not_empty_; // worker waits for work + std::condition_variable not_full_; // producers wait for room + std::condition_variable drained_; // Drain() waits for its barrier + std::deque queue_; + + std::thread worker_; + bool worker_started_ = false; + bool stop_requested_ = false; + bool worker_exited_ = false; ///< set by the worker on exit; guarded by mu_ + bool journal_active_ = false; + + std::atomic failed_ {false}; + VtxError last_error_; + + uint64_t next_barrier_id_ = 0; + uint64_t last_barrier_done_ = 0; + }; + +} // namespace VTX diff --git a/sdk/include/vtx/writer/policies/sinks/file_sink.h b/sdk/include/vtx/writer/policies/sinks/file_sink.h index c23008f..15c3504 100644 --- a/sdk/include/vtx/writer/policies/sinks/file_sink.h +++ b/sdk/include/vtx/writer/policies/sinks/file_sink.h @@ -74,7 +74,7 @@ namespace VTX { int8_t compression_level = 10; bool durable_writes = true; ///< fsync each chunk to physical disk (crash/power-loss safe). bool enable_recovery_journal = true; ///< maintain a ".recovery" sidecar for crash recovery. - uint64_t journal_compact_threshold_bytes = 0; ///< journal compaction trigger; 0 = journal default. + uint64_t journal_compact_threshold_bytes = 0; ///< journal compaction trigger; 0 = journal default. IFileSinkPerfObserver* perf_observer = nullptr; ///< optional perf timings; null = no-op. }; @@ -173,14 +173,39 @@ namespace VTX { // Journal a single recorded frame BEFORE it is flushed as part of a chunk, so // a crash mid-batch can still recover the in-flight frames (and their times). void JournalFrame(const FrameType& frame, int32_t frame_index, int64_t game_time, int64_t created_utc) { - if (!journal_.IsOpen()) - return; - std::vector> one; - one.push_back(std::make_unique(frame)); - std::string payload = SerializerPolicy::SerializeChunk(one, /*chunk_idx*/ 0, config_.b_use_compression); - payload = CompressIfBeneficial(std::move(payload)); - journal_.AppendFrame(frame_index, game_time, created_utc, payload); - batch_times_.push_back({frame_index, game_time, created_utc}); + JournalFrameImpl(frame, frame_index, game_time, created_utc, /*sync=*/true); + } + + // Same as JournalFrame but does NOT force the F record durable; the caller (the async + // sink's I/O worker) batches a run of these and issues one SyncJournal() at the end -- + // group commit. Only WHEN the record becomes durable changes; the append order (and + // therefore crash-recovery contiguity) is identical to the synchronous path. + void JournalFrameBatched(const FrameType& frame, int32_t frame_index, int64_t game_time, int64_t created_utc) { + JournalFrameImpl(frame, frame_index, game_time, created_utc, /*sync=*/false); + } + + // Flush the journal's group-commit batch to durability. No-op if journaling is off. + void SyncJournal() { + if (journal_.IsOpen()) + journal_.SyncNow(); + } + + // True once OnSessionStart has an open recovery journal. Lets the async adapter skip + // enqueuing (and copying) journal frames when journaling is disabled or failed to open. + bool IsJournalActive() const { return journal_.IsOpen(); } + + // True while no write/seek/sync on the main .vtx file has failed. The async worker + // latches its failure protocol on this going false (see AbortClose()). + bool Good() const { return file_.Good(); } + + // Failure/abort path: release the file + journal handles WITHOUT writing a footer and + // WITHOUT deleting the recovery journal. On Windows this drops the deny-write share so + // the partially written .vtx and its journal become repair-ready immediately, even while + // the process keeps running. (Contrast Close(), which finalizes a clean recording.) + void AbortClose() { + file_.Close(); + if (journal_.IsOpen()) + journal_.Close(); } void Close(const SessionFooter& footerData) { @@ -209,6 +234,18 @@ namespace VTX { } private: + void JournalFrameImpl(const FrameType& frame, int32_t frame_index, int64_t game_time, int64_t created_utc, + bool sync) { + if (!journal_.IsOpen()) + return; + std::vector> one; + one.push_back(std::make_unique(frame)); + std::string payload = SerializerPolicy::SerializeChunk(one, /*chunk_idx*/ 0, config_.b_use_compression); + payload = CompressIfBeneficial(std::move(payload)); + journal_.AppendFrame(frame_index, game_time, created_utc, payload, sync); + batch_times_.push_back({frame_index, game_time, created_utc}); + } + void NotifySerialize(std::chrono::steady_clock::time_point start) const { if (config_.perf_observer) config_.perf_observer->OnSerialize( diff --git a/sdk/include/vtx/writer/policies/sinks/recovery_journal.h b/sdk/include/vtx/writer/policies/sinks/recovery_journal.h index 1e4de87..8297bbd 100644 --- a/sdk/include/vtx/writer/policies/sinks/recovery_journal.h +++ b/sdk/include/vtx/writer/policies/sinks/recovery_journal.h @@ -122,7 +122,16 @@ namespace VTX { void SetCompactThresholdBytes(uint64_t bytes) { compact_threshold_ = bytes; } // Append an un-flushed frame (F record) at the tail. Call as each frame is recorded. - void AppendFrame(int32_t index, int64_t game_time, int64_t created_utc, const std::string& chunk_payload) { + // + // @param sync When true (default, synchronous sink) the record is made durable + // immediately. When false the record is only appended to the buffer; + // the caller must issue a later SyncNow() to make this and any other + // un-synced F records durable together (group commit, used by the + // async sink). Batching only defers WHEN the append becomes durable, + // never the append ORDER: records still hit disk in write order, so a + // crash still lands on a clean contiguous prefix. + void AppendFrame(int32_t index, int64_t game_time, int64_t created_utc, const std::string& chunk_payload, + bool sync = true) { // A frame whose record would exceed the read-side sanity bound could not be // parsed back on repair; skip journaling it rather than write an unreadable // record (the frame is still captured once its chunk is flushed to the .vtx). @@ -130,9 +139,14 @@ namespace VTX { return; const std::vector payload = BuildFramePayload(index, game_time, created_utc, chunk_payload); pending_f_bytes_ += WriteRecordTo(file_, 'F', payload); - SyncOrFlush(); + if (sync) + SyncOrFlush(); } + // Make every buffered-but-un-synced record durable. Pairs with AppendFrame(sync=false) + // to implement group commit: N appends, then ONE fsync for the whole batch. + void SyncNow() { SyncOrFlush(); } + // Commit a flushed chunk: durably append the chunk (C) and its frames' times (T). // Call AFTER the chunk is durable in the .vtx. Append-only -- the now-redundant F // records for this batch are left in place (repair dedups them) and reclaimed later diff --git a/sdk/src/vtx_writer/src/vtx/writer/core/vtx_writer_facade.cpp b/sdk/src/vtx_writer/src/vtx/writer/core/vtx_writer_facade.cpp index 454e594..6a475cf 100644 --- a/sdk/src/vtx_writer/src/vtx/writer/core/vtx_writer_facade.cpp +++ b/sdk/src/vtx_writer/src/vtx/writer/core/vtx_writer_facade.cpp @@ -8,6 +8,7 @@ #include "vtx/common/vtx_logger.h" #include "vtx/writer/policies/formatters/flatbuffers_vtx_policy.h" #include "vtx/writer/policies/formatters/protobuff_vtx_policy.h" +#include "vtx/writer/policies/sinks/async_sink_adapter.h" #include "vtx/writer/policies/sinks/file_sink.h" #include "vtx/writer/policies/sinks/network_sink.h" @@ -108,6 +109,17 @@ namespace VTX { VTX::SchemaRegistry& GetSchema() override { return writer_.GetRegistry(); } + VtxError Drain() override { + if (stopped_) { + return VtxError {}; + } + return writer_.Drain(); + } + + VtxError GetLastError() const override { return writer_.GetLastError(); } + + size_t GetQueueDepth() const override { return writer_.GetQueueDepth(); } + void SetPostProcessor(std::shared_ptr processor) override { writer_.SetPostProcessor(std::move(processor)); } @@ -119,6 +131,53 @@ namespace VTX { bool stopped_ = false; }; + namespace { + // Builds a file-writer facade for serialization policy P, selecting the synchronous + // ChunkedFileSink or its AsyncSinkAdapter<> decorator based on config.async_io. Both + // share the same inner sink config; only the durability barrier / queue differ. + template + std::unique_ptr MakeFileWriterFacade(const WriterFacadeConfig& config) { + using InnerSink = ChunkedFileSink

; + + typename InnerSink::Config sink_cfg; + sink_cfg.filename = config.output_filepath; + sink_cfg.header_config.replay_name = config.replay_name; + sink_cfg.header_config.replay_uuid = config.replay_uuid; + sink_cfg.b_use_compression = config.use_compression; + sink_cfg.durable_writes = config.durable_writes; + sink_cfg.enable_recovery_journal = config.enable_recovery_journal; + sink_cfg.perf_observer = config.perf_observer; + + const auto fill_common = [&](auto& internal_cfg) { + internal_cfg.default_fps = config.default_fps; + internal_cfg.is_increasing = config.is_increasing; + internal_cfg.chunker_config.max_frames = config.chunk_max_frames; + internal_cfg.chunker_config.max_bytes = config.chunk_max_bytes; + internal_cfg.schema_json_path = config.schema_json_path; + internal_cfg.schema_json_content = config.schema_json_content; + internal_cfg.schema_registry = config.schema_registry; + internal_cfg.retain_finalized_snapshot = config.retain_finalized_snapshot; + }; + + if (config.async_io) { + using SinkType = AsyncSinkAdapter; + typename ReplayWriter::Config internal_cfg; + fill_common(internal_cfg); + internal_cfg.sink_config.inner = std::move(sink_cfg); + internal_cfg.sink_config.async_max_queue_frames = + config.async_max_queue_frames != 0 + ? config.async_max_queue_frames + : static_cast(2) * static_cast(config.chunk_max_frames); + return std::make_unique>(std::move(internal_cfg)); + } + + typename ReplayWriter::Config internal_cfg; + fill_common(internal_cfg); + internal_cfg.sink_config = std::move(sink_cfg); + return std::make_unique>(std::move(internal_cfg)); + } + } // namespace + std::unique_ptr CreateFlatBuffersWriterFacade(const WriterFacadeConfig& config) { if (!WriterSchemaIsAcceptable(config.schema_json_path, config.schema_json_content, config.schema_registry)) { @@ -127,25 +186,7 @@ namespace VTX { if (config.create_output_dirs) { EnsureOutputDir(config.output_filepath); } - using SinkType = ChunkedFileSink; - - ReplayWriter::Config internal_cfg; - internal_cfg.default_fps = config.default_fps; - internal_cfg.is_increasing = config.is_increasing; - internal_cfg.chunker_config.max_frames = config.chunk_max_frames; - internal_cfg.chunker_config.max_bytes = config.chunk_max_bytes; - - internal_cfg.sink_config.filename = config.output_filepath; - internal_cfg.schema_json_path = config.schema_json_path; - internal_cfg.schema_json_content = config.schema_json_content; - internal_cfg.schema_registry = config.schema_registry; - internal_cfg.retain_finalized_snapshot = config.retain_finalized_snapshot; - internal_cfg.sink_config.header_config.replay_name = config.replay_name; - internal_cfg.sink_config.header_config.replay_uuid = config.replay_uuid; - internal_cfg.sink_config.b_use_compression = config.use_compression; - internal_cfg.sink_config.perf_observer = config.perf_observer; - - return std::make_unique>(internal_cfg); + return MakeFileWriterFacade(config); } std::unique_ptr CreateProtobufWriterFacade(const WriterFacadeConfig& config) { @@ -155,25 +196,7 @@ namespace VTX { if (config.create_output_dirs) { EnsureOutputDir(config.output_filepath); } - using SinkType = VTX::ChunkedFileSink; - - ReplayWriter::Config internal_cfg; - internal_cfg.sink_config.header_config.replay_name = config.replay_name; - internal_cfg.sink_config.header_config.replay_uuid = config.replay_uuid; - internal_cfg.default_fps = config.default_fps; - internal_cfg.is_increasing = config.is_increasing; - internal_cfg.chunker_config.max_frames = config.chunk_max_frames; - internal_cfg.chunker_config.max_bytes = config.chunk_max_bytes; - - internal_cfg.sink_config.filename = config.output_filepath; - internal_cfg.schema_json_path = config.schema_json_path; - internal_cfg.schema_json_content = config.schema_json_content; - internal_cfg.schema_registry = config.schema_registry; - internal_cfg.retain_finalized_snapshot = config.retain_finalized_snapshot; - - internal_cfg.sink_config.b_use_compression = config.use_compression; - internal_cfg.sink_config.perf_observer = config.perf_observer; - return std::make_unique>(internal_cfg); + return MakeFileWriterFacade(config); } std::unique_ptr CreateFlatBuffersNetworkWriterFacade(const NetworkWriterFacadeConfig& config) { diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 328ec0f..2badc77 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -77,6 +77,7 @@ add_executable(vtx_tests writer/test_frame_post_processor.cpp writer/test_frame_finalization.cpp writer/test_crash_recovery.cpp + writer/test_async_sink.cpp writer/test_network_sink.cpp writer/test_pipe_source.cpp writer/test_websocket_source.cpp diff --git a/tests/writer/test_async_sink.cpp b/tests/writer/test_async_sink.cpp new file mode 100644 index 0000000..0adc0d0 --- /dev/null +++ b/tests/writer/test_async_sink.cpp @@ -0,0 +1,621 @@ +// AsyncSinkAdapter tests: the async decorator must move chunk/journal I/O to a worker +// thread WITHOUT changing what lands on disk or the guarantees callers rely on. +// +// Coverage: +// * Order equivalence -- against an instrumented fake inner sink, the async call +// sequence to the inner sink equals the synchronous one. +// * Byte equivalence -- through the REAL sink, an async recording is bit-for-bit +// identical to a synchronous one for the same inputs. +// * Backpressure -- a bounded queue + a slow inner sink blocks the caller but +// never drops or reorders a frame. +// * Failure protocol -- a latched inner I/O failure surfaces as SinkFailed on the +// next TryRecordFrame and through Drain()/GetLastError, aborts +// the recording, and unblocks any producer waiting on the queue. +// * Drain() -- the durability barrier drains the queue in order. + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "vtx_schema_generated.h" // complete fbsvtx/cppvtx types before the policy headers +#include "vtx_schema.pb.h" + +#include "vtx/common/vtx_types.h" +#include "vtx/reader/core/vtx_reader_facade.h" +#include "vtx/reader/core/vtx_replay_validation.h" +#include "vtx/writer/core/vtx_replay_recovery.h" +#include "vtx/writer/core/vtx_writer_facade.h" +#include "vtx/writer/core/writer.h" +#include "vtx/writer/policies/formatters/flatbuffers_vtx_policy.h" +#include "vtx/writer/policies/formatters/protobuff_vtx_policy.h" +#include "vtx/writer/policies/sinks/async_sink_adapter.h" +#include "vtx/writer/policies/sinks/file_sink.h" + +#include "util/test_fixtures.h" + +namespace { + + VTX::Frame BuildFrame(int i) { + VTX::Frame f; + auto& bucket = f.CreateBucket("entity"); + VTX::PropertyContainer pc; + pc.entity_type_id = 0; // Player + pc.string_properties = {"player_0", "Alpha"}; + pc.int32_properties = {1, i, 0}; // Team, Score(=i), Deaths + pc.float_properties = {100.0f, 50.0f}; + bucket.unique_ids.push_back("player_0"); + bucket.entities.push_back(std::move(pc)); + return f; + } + + // --- Instrumented fake inner sink ------------------------------------------- + // + // Records the exact ordered sequence of data operations (journal frame, save chunk, + // close, abort) reaching the inner sink, WITHOUT touching disk. Shared with the test + // through a shared_ptr so it survives being moved into the writer/adapter. It reuses + // the real FlatBuffers serializer policy so it plugs straight into ReplayWriter. + + struct FakeSinkState { + std::mutex mu; + std::vector ops; // normalized op log ("J", "C-", ...) + int save_chunk_calls = 0; + int fail_after_chunks = -1; // >=0: latch a failure on that many SaveChunk calls + std::atomic good {true}; // mirrors DurableFile::Good() + bool journal_active = true; // mirrors an open recovery journal + std::chrono::milliseconds work_delay {0}; // artificial per-op slowness for backpressure + + void Log(std::string s) { + std::lock_guard lk(mu); + ops.push_back(std::move(s)); + } + std::vector Ops() { + std::lock_guard lk(mu); + return ops; + } + }; + + struct FakeSink { + using SerializerPolicy = VTX::FlatBuffersVtxPolicy; + using FrameType = VTX::Frame; + using SchemaType = SerializerPolicy::SchemaType; + using HeaderType = SerializerPolicy::HeaderType; + + struct Config { + std::shared_ptr state; + }; + + explicit FakeSink(Config c) + : state_(std::move(c.state)) {} + + void JournalTiming(float, bool) {} + void OnSessionStart(const SchemaType&) { state_->Log("OnSessionStart"); } + + void JournalFrame(const FrameType&, int32_t idx, int64_t, int64_t) { + Delay(); + state_->Log("J" + std::to_string(idx)); + } + // The async worker calls this instead; it must log identically (sync coalescing aside) + // so the two op sequences line up. + void JournalFrameBatched(const FrameType&, int32_t idx, int64_t, int64_t) { + Delay(); + state_->Log("J" + std::to_string(idx)); + } + void SyncJournal() {} // coalescing of J-syncs is allowed; not logged + bool IsJournalActive() const { return state_->journal_active; } + + void SaveChunk(std::vector>&, const std::vector&, int32_t start, + int32_t total) { + Delay(); + state_->Log("C" + std::to_string(start) + "-" + std::to_string(total)); + const int n = ++state_->save_chunk_calls; + if (state_->fail_after_chunks >= 0 && n >= state_->fail_after_chunks) + state_->good.store(false, std::memory_order_release); // latch, as a short write would + } + + void Close(const VTX::SessionFooter& f) { state_->Log("Close" + std::to_string(f.total_frames)); } + bool Good() const { return state_->good.load(std::memory_order_acquire); } + void AbortClose() { state_->Log("AbortClose"); } + + private: + void Delay() { + if (state_->work_delay.count() > 0) + std::this_thread::sleep_for(state_->work_delay); + } + std::shared_ptr state_; + }; + + using SyncFake = VTX::ReplayWriter; + using AsyncFake = VTX::ReplayWriter>; + + std::unique_ptr MakeSyncFake(std::shared_ptr state, int chunk_max) { + SyncFake::Config cfg; + cfg.sink_config.state = std::move(state); + cfg.schema_json_path = VtxTest::FixturePath("test_schema.json"); + cfg.default_fps = 60.0f; + cfg.chunker_config.max_frames = chunk_max; + return std::make_unique(cfg); + } + + std::unique_ptr MakeAsyncFake(std::shared_ptr state, int chunk_max, size_t queue_cap) { + AsyncFake::Config cfg; + cfg.sink_config.inner.state = std::move(state); + cfg.sink_config.async_max_queue_frames = queue_cap; + cfg.schema_json_path = VtxTest::FixturePath("test_schema.json"); + cfg.default_fps = 60.0f; + cfg.chunker_config.max_frames = chunk_max; + return std::make_unique(cfg); + } + + template + void RecordFrames(Writer& w, int count) { + for (int i = 0; i < count; ++i) { + auto frame = BuildFrame(i); + VTX::GameTime::GameTimeRegister t; + t.game_time = float(i) / 60.0f; + t.created_utc_time = 1'000'000'000'000 + int64_t(i) * 166'667; + w.TryRecordFrame(frame, t); + } + } + +} // namespace + +// The inner sink sees the exact same ordered sequence of data operations whether the +// writer is synchronous or async: one FIFO, one worker preserves order by construction. +TEST(AsyncSink, OrderEquivalenceMatchesSync) { + auto sync_state = std::make_shared(); + auto async_state = std::make_shared(); + + { + auto w = MakeSyncFake(sync_state, /*chunk_max=*/3); + RecordFrames(*w, 10); + w->Stop(); + } + { + auto w = MakeAsyncFake(async_state, /*chunk_max=*/3, /*queue_cap=*/4); + RecordFrames(*w, 10); + w->Stop(); // drains + closes + joins the worker + } + + EXPECT_EQ(async_state->Ops(), sync_state->Ops()); + ASSERT_FALSE(sync_state->Ops().empty()); + EXPECT_EQ(sync_state->Ops().front(), "OnSessionStart"); + EXPECT_EQ(sync_state->Ops().back(), "Close10"); +} + +// A bounded queue with a slow inner sink blocks the caller (never returns early, never +// drops) -- every recorded frame reaches the inner sink, in order. +TEST(AsyncSink, BackpressureBlocksButLosesNothing) { + auto state = std::make_shared(); + state->work_delay = std::chrono::milliseconds(2); // make the worker the bottleneck + + { + auto w = MakeAsyncFake(state, /*chunk_max=*/2, /*queue_cap=*/2); + RecordFrames(*w, 20); + w->Stop(); + } + + const auto ops = state->Ops(); + int journaled = 0; + for (const auto& op : ops) + if (!op.empty() && op[0] == 'J') + ++journaled; + EXPECT_EQ(journaled, 20); // no frame lost under backpressure + EXPECT_EQ(ops.front(), "OnSessionStart"); + EXPECT_EQ(ops.back(), "Close20"); +} + +// A latched inner I/O failure aborts the recording: the next TryRecordFrame is rejected +// with SinkFailed, the queue is discarded, and the inner handles are released (AbortClose). +TEST(AsyncSink, FailureSurfacesAsSinkFailed) { + auto state = std::make_shared(); + state->fail_after_chunks = 1; // fail on the first chunk flush + + auto w = MakeAsyncFake(state, /*chunk_max=*/2, /*queue_cap=*/8); + + // Drive frames until the worker has flushed a chunk and latched the failure, which the + // writer then reports synchronously. Bounded loop so a regression cannot hang the suite. + bool saw_sink_failed = false; + for (int i = 0; i < 200 && !saw_sink_failed; ++i) { + auto frame = BuildFrame(i); + VTX::GameTime::GameTimeRegister t; + t.game_time = float(i) / 60.0f; + t.created_utc_time = 1'000'000'000'000 + int64_t(i) * 166'667; + const auto res = w->TryRecordFrame(frame, t); + if (!res.written && res.error.code == VTX::VtxErrorCode::SinkFailed) + saw_sink_failed = true; + else + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + + EXPECT_TRUE(saw_sink_failed); + EXPECT_TRUE(w->HasSinkFailed()); + EXPECT_EQ(w->GetLastError().code, VTX::VtxErrorCode::SinkFailed); + + // Drain() on a failed sink returns the latched error promptly (no hang). + EXPECT_EQ(w->Drain().code, VTX::VtxErrorCode::SinkFailed); + + w->Stop(); // must not hang; worker already exited + + const auto ops = state->Ops(); + bool aborted = false; + for (const auto& op : ops) + if (op == "AbortClose") + aborted = true; + EXPECT_TRUE(aborted); // handles released -> file becomes repair-ready immediately +} + +// Drain() is the durability barrier: it blocks until every frame enqueued before it has +// been handed to the inner sink, then the queue is empty. +TEST(AsyncSink, DrainFlushesQueueInOrder) { + auto state = std::make_shared(); + state->work_delay = std::chrono::milliseconds(1); + + auto w = MakeAsyncFake(state, /*chunk_max=*/4, /*queue_cap=*/64); + RecordFrames(*w, 12); + + const auto err = w->Drain(); + EXPECT_EQ(err.code, VTX::VtxErrorCode::None); + EXPECT_EQ(w->GetQueueDepth(), 0u); // everything caught up + + int journaled = 0; + for (const auto& op : state->Ops()) + if (!op.empty() && op[0] == 'J') + ++journaled; + EXPECT_EQ(journaled, 12); // all 12 frames journaled before Drain returned + + w->Stop(); +} + +// Recording (or draining) after Stop() is caller misuse, but it must never HANG. The worker +// is gone, so an enqueue that would wait for room has nobody to free it: the adapter reports +// the worker's exit and returns immediately instead of blocking forever. Reaching the end of +// this test IS the assertion -- a regression deadlocks here. +TEST(AsyncSink, EnqueueAfterStopDoesNotHang) { + auto state = std::make_shared(); + auto w = MakeAsyncFake(state, /*chunk_max=*/2, /*queue_cap=*/2); + RecordFrames(*w, 4); + w->Stop(); + + RecordFrames(*w, 50); // far more items than the queue cap + EXPECT_EQ(w->Drain().code, VTX::VtxErrorCode::None); // barrier with no worker: returns, not hangs + EXPECT_NO_FATAL_FAILURE(w->Stop()); // a second Stop is also a no-op + + // Nothing after the close reached the sink: the recording ended at Stop(). + const auto ops = state->Ops(); + ASSERT_FALSE(ops.empty()); + EXPECT_EQ(ops.back(), "Close4"); +} + +// --- Byte equivalence through the REAL sink ------------------------------------ +// +// The async .vtx must be bit-for-bit identical to a synchronous one for the same inputs: +// only WHEN I/O happens changes, never WHAT is written. Explicit per-frame times remove +// any wall-clock nondeterminism so the comparison is exact. + +namespace { + + void WriteRecording(const std::string& path, bool async, bool protobuf, int frames, int chunk_max) { + VTX::WriterFacadeConfig cfg; + cfg.output_filepath = path; + cfg.schema_json_path = VtxTest::FixturePath("test_schema.json"); + cfg.replay_name = "AsyncByteEquiv"; + cfg.default_fps = 60.0f; + cfg.chunk_max_frames = chunk_max; + cfg.use_compression = true; + cfg.async_io = async; + + auto w = protobuf ? VTX::CreateProtobufWriterFacade(cfg) : VTX::CreateFlatBuffersWriterFacade(cfg); + EXPECT_NE(w, nullptr); + for (int i = 0; i < frames; ++i) { + auto frame = BuildFrame(i); + VTX::GameTime::GameTimeRegister t; + t.game_time = float(i) / 60.0f; + t.created_utc_time = 1'000'000'000'000 + int64_t(i) * 166'667; + w->RecordFrame(frame, t); + } + w->Stop(); + w.reset(); // release the handle before reading the bytes back + } + + int64_t NowSecond() { + return std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()) + .count(); + } + + void RunByteEquivalence(bool protobuf, const std::string& tag) { + const std::string sync_path = VtxTest::OutputPath("async_equiv_sync_" + tag + ".vtx"); + const std::string async_path = VtxTest::OutputPath("async_equiv_async_" + tag + ".vtx"); + + // The file header embeds a wall-clock second (FileHeader.timestamp = now()), so two + // independent recordings differ there whenever they straddle a second boundary -- which + // then cascades through zstd and the seek-table offsets. Pair the writes inside a single + // wall-clock second (retrying on the rare boundary crossing) so the ONLY remaining source + // of difference is the async path itself, and full-file byte identity is a valid check. + std::vector sync_bytes; + std::vector async_bytes; + bool same_second = false; + for (int attempt = 0; attempt < 8 && !same_second; ++attempt) { + const int64_t before = NowSecond(); + WriteRecording(sync_path, /*async=*/false, protobuf, /*frames=*/17, /*chunk_max=*/4); + WriteRecording(async_path, /*async=*/true, protobuf, /*frames=*/17, /*chunk_max=*/4); + const int64_t after = NowSecond(); + if (before == after) { + same_second = true; + sync_bytes = VtxTest::ReadAllBytes(sync_path); + async_bytes = VtxTest::ReadAllBytes(async_path); + } + } + ASSERT_TRUE(same_second) << "could not pair the writes within one wall-clock second"; + + ASSERT_FALSE(sync_bytes.empty()); + EXPECT_EQ(async_bytes, sync_bytes) << "async .vtx diverged from sync for " << tag; + + // And the async file is a valid, readable recording with the expected frame count. + auto ctx = VTX::OpenReplayFile(async_path); + ASSERT_TRUE(ctx) << ctx.error; + ctx->WaitUntilReady(); + EXPECT_EQ(ctx->GetTotalFrames(), 17); + } + +} // namespace + +TEST(AsyncSink, ByteEquivalenceFlatBuffers) { + RunByteEquivalence(/*protobuf=*/false, "fb"); +} + +TEST(AsyncSink, ByteEquivalenceProtobuf) { + RunByteEquivalence(/*protobuf=*/true, "pb"); +} + +// --- Integrity: no frame and no chunk may be lost ------------------------------ + +namespace { + + struct MatrixCfg { + bool durable; + bool compression; + bool journal; + int chunk_max; + size_t queue_cap; + const char* tag; + }; + + void WriteMatrix(const std::string& path, bool async, const MatrixCfg& c, int frames) { + VTX::WriterFacadeConfig cfg; + cfg.output_filepath = path; + cfg.schema_json_path = VtxTest::FixturePath("test_schema.json"); + cfg.default_fps = 60.0f; + cfg.chunk_max_frames = c.chunk_max; + cfg.use_compression = c.compression; + cfg.durable_writes = c.durable; + cfg.enable_recovery_journal = c.journal; + cfg.async_io = async; + cfg.async_max_queue_frames = c.queue_cap; + + auto w = VTX::CreateFlatBuffersWriterFacade(cfg); + ASSERT_NE(w, nullptr); + for (int i = 0; i < frames; ++i) { + auto frame = BuildFrame(i); + VTX::GameTime::GameTimeRegister t; + t.game_time = float(i) / 60.0f; + t.created_utc_time = 1'000'000'000'000 + int64_t(i) * 166'667; + w->RecordFrame(frame, t); + } + w->Stop(); + } + +} // namespace + +// Across the sink's whole configuration matrix -- durability, compression, journal on/off, +// chunk sizes, and queue caps down to 1 (permanent backpressure) -- an async recording must +// contain EXACTLY what the synchronous one does: same frame count, same CHUNK count, and the +// right frame with the right content at every index. A dropped, duplicated, or reordered +// queue item would break one of these. +TEST(AsyncSink, NoFrameOrChunkLossAcrossConfigMatrix) { + const MatrixCfg configs[] = { + {true, true, true, 4, 0, "d1c1j1_q0"}, // defaults, queue cap auto (2*chunk) + {true, false, true, 4, 2, "d1c0j1_q2"}, // uncompressed, tight queue + {false, true, true, 1, 1, "d0c1j1_chunk1_q1"}, // a chunk per frame, cap 1: max contention + {true, true, false, 8, 3, "d1c1j0_q3"}, // recovery journal OFF (no journal items) + {false, false, true, 16, 64, "d0c0j1_q16_q64"}, // flush-only, roomy queue + }; + constexpr int kFrames = 137; // deliberately not a multiple of any chunk size + + for (const auto& c : configs) { + SCOPED_TRACE(c.tag); + const std::string sync_path = VtxTest::OutputPath(std::string("mtx_sync_") + c.tag + ".vtx"); + const std::string async_path = VtxTest::OutputPath(std::string("mtx_async_") + c.tag + ".vtx"); + + WriteMatrix(sync_path, /*async=*/false, c, kFrames); + WriteMatrix(async_path, /*async=*/true, c, kFrames); + if (::testing::Test::HasFatalFailure()) + return; + + auto sync_ctx = VTX::OpenReplayFile(sync_path); + ASSERT_TRUE(sync_ctx) << sync_ctx.error; + sync_ctx->WaitUntilReady(); + auto async_ctx = VTX::OpenReplayFile(async_path); + ASSERT_TRUE(async_ctx) << async_ctx.error; + async_ctx->WaitUntilReady(); + + ASSERT_EQ(async_ctx->GetTotalFrames(), kFrames) << "async lost frames"; + ASSERT_EQ(sync_ctx->GetTotalFrames(), kFrames); + EXPECT_EQ(async_ctx->GetSeekTable().size(), sync_ctx->GetSeekTable().size()) << "chunk count diverged"; + + for (int i = 0; i < kFrames; ++i) { + const VTX::Frame* a = async_ctx->GetFrameSync(i); + const VTX::Frame* s = sync_ctx->GetFrameSync(i); + ASSERT_NE(a, nullptr) << "async frame " << i << " missing"; + ASSERT_NE(s, nullptr) << "sync frame " << i << " missing"; + // The right frame landed at the right index (a reorder would shift the score)... + EXPECT_EQ(a->GetBuckets()[0].entities[0].int32_properties[1], i) << "frame " << i; + // ...and its full content is identical to the synchronous recording's. + EXPECT_EQ(a->GetBuckets()[0].entities[0].content_hash, s->GetBuckets()[0].entities[0].content_hash) + << "content mismatch at frame " << i; + } + } +} + +// A cleanly-stopped async recording must pass the SDK's OWN whole-replay validation -- +// embedded schema plus every frame -- not merely read back with the right frame count. +TEST(AsyncSink, CleanAsyncRecordingPassesReplayValidation) { + const std::string path = VtxTest::OutputPath("async_validated.vtx"); + const MatrixCfg c {true, true, true, 5, 4, "validate"}; + WriteMatrix(path, /*async=*/true, c, /*frames=*/64); + ASSERT_FALSE(::testing::Test::HasFatalFailure()); + + const VTX::ValidationReport report = VTX::ValidateReplayFile(path); + EXPECT_FALSE(report.HasErrors()) << report.ToString(); +} + +// Sustained volume through a ONE-item queue: thousands of hand-offs between the recording +// thread and the I/O worker, every one of them contended. A lost, duplicated or reordered +// item surfaces as the wrong frame at a known index. +TEST(AsyncSink, HighVolumeContendedStreamStaysIntact) { + const std::string path = VtxTest::OutputPath("async_stress.vtx"); + const MatrixCfg c {false, true, true, 7, 1, "stress"}; // flush-only for speed, queue cap 1 + constexpr int kFrames = 2000; + WriteMatrix(path, /*async=*/true, c, kFrames); + ASSERT_FALSE(::testing::Test::HasFatalFailure()); + + auto ctx = VTX::OpenReplayFile(path); + ASSERT_TRUE(ctx) << ctx.error; + ctx->WaitUntilReady(); + ASSERT_EQ(ctx->GetTotalFrames(), kFrames); + for (int i : {0, 1, 999, 1000, 1777, kFrames - 1}) { + const VTX::Frame* f = ctx->GetFrameSync(i); + ASSERT_NE(f, nullptr) << "frame " << i; + EXPECT_EQ(f->GetBuckets()[0].entities[0].int32_properties[1], i) << "frame " << i; + } +} + +// Worker start/join lifecycle under churn: many short recordings back to back. A leaked, +// double-joined, or never-started worker (or a hang in Close) shows up here. +TEST(AsyncSink, RapidLifecycleCyclesAreClean) { + constexpr int kRounds = 40; + for (int round = 0; round < kRounds; ++round) { + const std::string path = VtxTest::OutputPath("async_cycle_" + std::to_string(round) + ".vtx"); + const MatrixCfg c {true, true, true, 3, 4, "cycle"}; + WriteMatrix(path, /*async=*/true, c, /*frames=*/5); + if (::testing::Test::HasFatalFailure()) + return; + + EXPECT_FALSE(VTX::ReplayNeedsRecovery(path)) << "round " << round; // clean Stop deleted the journal + auto ctx = VTX::OpenReplayFile(path); + ASSERT_TRUE(ctx) << ctx.error; + ctx->WaitUntilReady(); + ASSERT_EQ(ctx->GetTotalFrames(), 5) << "round " << round; + } +} + +// The facade destructor finalizes an async recording that the caller never Stop()ped: the +// adapter must drain the queue, write the footer and join the worker -- with items still +// queued at destruction time (queue cap 2) and never hanging. Repeated to shake out any +// ordering hazard in that teardown path. +TEST(AsyncSink, FacadeDestructorFinalizesAsyncRecording) { + constexpr int kRounds = 15; + constexpr int kFrames = 9; + for (int round = 0; round < kRounds; ++round) { + const std::string path = VtxTest::OutputPath("async_drop_" + std::to_string(round) + ".vtx"); + std::filesystem::remove(path); + std::filesystem::remove(VTX::RecoveryJournalPath(path)); + { + VTX::WriterFacadeConfig cfg; + cfg.output_filepath = path; + cfg.schema_json_path = VtxTest::FixturePath("test_schema.json"); + cfg.default_fps = 60.0f; + cfg.chunk_max_frames = 4; + cfg.async_io = true; + cfg.async_max_queue_frames = 2; // force backpressure, so items are still queued at drop + auto w = VTX::CreateFlatBuffersWriterFacade(cfg); + ASSERT_NE(w, nullptr); + for (int i = 0; i < kFrames; ++i) { + auto frame = BuildFrame(i); + VTX::GameTime::GameTimeRegister t; + t.game_time = float(i) / 60.0f; + t.created_utc_time = 1'000'000'000'000 + int64_t(i) * 166'667; + w->RecordFrame(frame, t); + } + // Deliberately NO Stop(): the facade destructor must finalize the recording. + } + EXPECT_FALSE(VTX::ReplayNeedsRecovery(path)) << "round " << round; // footer written, journal removed + auto ctx = VTX::OpenReplayFile(path); + ASSERT_TRUE(ctx) << ctx.error << " (round " << round << ")"; + ctx->WaitUntilReady(); + ASSERT_EQ(ctx->GetTotalFrames(), kFrames) << "round " << round; + } +} + +// Drain() interleaved with recording under permanent backpressure: the barrier must never +// deadlock against a blocked producer, and nothing may be lost around it. +TEST(AsyncSink, DrainInterleavedWithRecordingLosesNothing) { + const std::string path = VtxTest::OutputPath("async_drain_interleaved.vtx"); + VTX::WriterFacadeConfig cfg; + cfg.output_filepath = path; + cfg.schema_json_path = VtxTest::FixturePath("test_schema.json"); + cfg.default_fps = 60.0f; + cfg.chunk_max_frames = 3; + cfg.async_io = true; + cfg.async_max_queue_frames = 1; // every enqueue contends with the worker + auto w = VTX::CreateFlatBuffersWriterFacade(cfg); + ASSERT_NE(w, nullptr); + + constexpr int kFrames = 80; + for (int i = 0; i < kFrames; ++i) { + auto frame = BuildFrame(i); + VTX::GameTime::GameTimeRegister t; + t.game_time = float(i) / 60.0f; + t.created_utc_time = 1'000'000'000'000 + int64_t(i) * 166'667; + w->RecordFrame(frame, t); + if (i % 7 == 0) { + const VTX::VtxError err = w->Drain(); // barrier mid-stream, under contention + ASSERT_EQ(err.code, VTX::VtxErrorCode::None) << "Drain failed at frame " << i; + EXPECT_EQ(w->GetQueueDepth(), 0u); + } + } + w->Stop(); + w.reset(); + + auto ctx = VTX::OpenReplayFile(path); + ASSERT_TRUE(ctx) << ctx.error; + ctx->WaitUntilReady(); + ASSERT_EQ(ctx->GetTotalFrames(), kFrames); + for (int i : {0, 13, 41, kFrames - 1}) { + const VTX::Frame* f = ctx->GetFrameSync(i); + ASSERT_NE(f, nullptr) << "frame " << i; + EXPECT_EQ(f->GetBuckets()[0].entities[0].int32_properties[1], i); + } +} + +// A clean async recording leaves no recovery sidecar behind (Stop() finalized the footer). +TEST(AsyncSink, CleanStopRemovesJournal) { + const std::string path = VtxTest::OutputPath("async_clean.vtx"); + VTX::WriterFacadeConfig cfg; + cfg.output_filepath = path; + cfg.schema_json_path = VtxTest::FixturePath("test_schema.json"); + cfg.default_fps = 60.0f; + cfg.chunk_max_frames = 3; + cfg.async_io = true; + + auto w = VTX::CreateFlatBuffersWriterFacade(cfg); + ASSERT_NE(w, nullptr); + for (int i = 0; i < 7; ++i) { + auto frame = BuildFrame(i); + VTX::GameTime::GameTimeRegister t; + t.game_time = float(i) / 60.0f; + t.created_utc_time = 1'000'000'000'000 + int64_t(i) * 166'667; + w->RecordFrame(frame, t); + } + w->Stop(); + w.reset(); + + EXPECT_FALSE(VTX::ReplayNeedsRecovery(path)); +} diff --git a/tests/writer/test_crash_recovery.cpp b/tests/writer/test_crash_recovery.cpp index 1b57934..fe925c1 100644 --- a/tests/writer/test_crash_recovery.cpp +++ b/tests/writer/test_crash_recovery.cpp @@ -45,6 +45,7 @@ #include "vtx/writer/core/writer.h" #include "vtx/writer/policies/formatters/flatbuffers_vtx_policy.h" #include "vtx/writer/policies/formatters/protobuff_vtx_policy.h" +#include "vtx/writer/policies/sinks/async_sink_adapter.h" #include "vtx/writer/policies/sinks/file_sink.h" #include "vtx/writer/policies/sinks/recovery_journal.h" @@ -698,6 +699,32 @@ namespace { return std::make_unique>(cfg); } + // Same raw writer, but with the async decorator in front of the file sink: chunk and + // journal I/O run on a worker thread. Dropping it without Stop() still leaves exactly + // what a crash leaves, so every crash-recovery expectation below applies unchanged -- + // that is the point of these variants. + template + using RawAsyncWriterFor = VTX::ReplayWriter>>; + + template + std::unique_ptr> + MakeRawAsyncWriter(const std::string& path, int32_t chunk_max_frames, bool durable = true, bool compression = true, + bool journal = true, uint64_t compact_threshold = 0, size_t queue_cap = 0, + const std::string& schema_name = "test_schema.json") { + typename RawAsyncWriterFor::Config cfg; + cfg.sink_config.inner.filename = path; + cfg.sink_config.inner.header_config.replay_name = "CrashRecoveryE2E"; + cfg.sink_config.inner.durable_writes = durable; + cfg.sink_config.inner.b_use_compression = compression; + cfg.sink_config.inner.enable_recovery_journal = journal; + cfg.sink_config.inner.journal_compact_threshold_bytes = compact_threshold; + cfg.sink_config.async_max_queue_frames = queue_cap; + cfg.schema_json_path = VtxTest::FixturePath(schema_name); + cfg.default_fps = 60.0f; + cfg.chunker_config.max_frames = chunk_max_frames; + return std::make_unique>(cfg); + } + // Records 8 frames with explicit game_time + created_utc: a UTC jump at frame 3 // (timeline gap) and a game-time reversal at frame 5 (game segment). Chunks of 3 -> // crash state = 2 committed chunks + 2 in-flight frames. @@ -721,8 +748,9 @@ namespace { // Full crash-vs-control comparison through the real writer for one policy. template - void RunRealWriterCrashVsControl(const std::string& prefix) { - // Control: identical inputs, clean Stop(). + void RunRealWriterCrashVsControl(const std::string& prefix, bool async_crash = false) { + // Control: identical inputs, clean Stop(). Always the SYNCHRONOUS writer, so an async + // crash side is held to the exact output a synchronous clean run would have produced. const std::string control_path = VtxTest::OutputPath(prefix + "_e2e_control.vtx"); { auto w = MakeRawWriter(control_path, 3); @@ -734,8 +762,16 @@ namespace { // Crash: same inputs, writer dropped without Stop(). const std::string crash_path = VtxTest::OutputPath(prefix + "_e2e_crash.vtx"); { - auto w = MakeRawWriter(crash_path, 3); - RecordEightFrames(*w); + if (async_crash) { + // The adapter's destructor drains the queue and closes WITHOUT a footer, so an + // in-process drop leaves the same footerless .vtx + journal a crash leaves -- + // and, having drained, must still account for every recorded frame. + auto w = MakeRawAsyncWriter(crash_path, 3); + RecordEightFrames(*w); + } else { + auto w = MakeRawWriter(crash_path, 3); + RecordEightFrames(*w); + } // dropped here -- no Stop(), no footer } ASSERT_TRUE(VTX::ReplayNeedsRecovery(crash_path)); @@ -807,6 +843,18 @@ TEST(CrashRecoveryE2E, RealWriterCrashMatchesCleanStopExactly_Protobuf) { RunRealWriterCrashVsControl("pb"); } +// Same contract with async I/O: the recovered output must be indistinguishable from a clean +// SYNCHRONOUS run -- every frame, every content hash, and every footer time field (per-frame +// times, duration, timeline gaps, game segments). Nothing about the journal or the repair +// path may behave differently just because the writes came off a worker thread. +TEST(CrashRecoveryE2E, AsyncWriterCrashMatchesCleanStopExactly) { + RunRealWriterCrashVsControl("fb_async", /*async_crash=*/true); +} + +TEST(CrashRecoveryE2E, AsyncWriterCrashMatchesCleanStopExactly_Protobuf) { + RunRealWriterCrashVsControl("pb_async", /*async_crash=*/true); +} + namespace { // A frame big enough (~4KB) that its serialized chunk clears the 512-byte floor and @@ -1906,6 +1954,21 @@ TEST(CrashRecoverySweep, CompactedJournalTruncationSweep) { // the only honest way to validate the flush-only (durable_writes=false) claim that // data pushed to the OS survives a process crash. +namespace { + + // Endless recording loop for the child process; terminated externally by the parent. + template + void ChildRecordLoop(Writer& w) { + for (int i = 0;; ++i) { + auto frame = BuildFrame(i); + VTX::GameTime::GameTimeRegister t; + t.game_time = float(i) / 60.0f; + (void)w.TryRecordFrame(frame, t); + } + } + +} // namespace + // Child mode: an endless writer loop. Skipped unless spawned with the env var set. TEST(CrashRecoveryProcess, ChildWriterLoop) { const char* path = std::getenv("VTX_CHILD_WRITE_PATH"); @@ -1915,21 +1978,27 @@ TEST(CrashRecoveryProcess, ChildWriterLoop) { const bool durable = !durable_env || std::string(durable_env) != "0"; const char* compact_env = std::getenv("VTX_CHILD_COMPACT_THRESHOLD"); const uint64_t compact_threshold = compact_env ? std::strtoull(compact_env, nullptr, 10) : 0; + const char* async_env = std::getenv("VTX_CHILD_ASYNC"); + const bool async = async_env && std::string(async_env) == "1"; - auto w = MakeRawWriter(path, /*chunk_max_frames=*/10, durable, - /*compression=*/true, /*journal=*/true, compact_threshold); - for (int i = 0;; ++i) { // terminated externally - auto frame = BuildFrame(i); - VTX::GameTime::GameTimeRegister t; - t.game_time = float(i) / 60.0f; - (void)w->TryRecordFrame(frame, t); + if (async) { + // Chunk/journal I/O on a worker thread. TerminateProcess kills the worker mid-write + // exactly like the main thread -- no unwinding, no drain, handles abandoned. + auto w = + MakeRawAsyncWriter(path, /*chunk_max_frames=*/10, durable, + /*compression=*/true, /*journal=*/true, compact_threshold); + ChildRecordLoop(*w); + } else { + auto w = MakeRawWriter(path, /*chunk_max_frames=*/10, durable, + /*compression=*/true, /*journal=*/true, compact_threshold); + ChildRecordLoop(*w); } } namespace { void RunKilledProcessTest(bool durable, const char* tag, uint64_t compact_threshold = 0, - uint64_t progress_threshold = 20'000) { + uint64_t progress_threshold = 20'000, bool async = false) { const std::string path = VtxTest::OutputPath(std::string("killed_") + tag + ".vtx"); const std::string journal_path = VTX::RecoveryJournalPath(path); std::filesystem::remove(path); @@ -1939,6 +2008,7 @@ namespace { ASSERT_GT(GetModuleFileNameA(nullptr, exe, MAX_PATH), 0u); SetEnvironmentVariableA("VTX_CHILD_WRITE_PATH", path.c_str()); SetEnvironmentVariableA("VTX_CHILD_DURABLE", durable ? "1" : "0"); + SetEnvironmentVariableA("VTX_CHILD_ASYNC", async ? "1" : "0"); if (compact_threshold > 0) SetEnvironmentVariableA("VTX_CHILD_COMPACT_THRESHOLD", std::to_string(compact_threshold).c_str()); std::string cmd = std::string("\"") + exe + "\" --gtest_filter=CrashRecoveryProcess.ChildWriterLoop"; @@ -1949,6 +2019,7 @@ namespace { CreateProcessA(nullptr, cmd.data(), nullptr, nullptr, FALSE, CREATE_NO_WINDOW, nullptr, nullptr, &si, &pi); SetEnvironmentVariableA("VTX_CHILD_WRITE_PATH", nullptr); SetEnvironmentVariableA("VTX_CHILD_DURABLE", nullptr); + SetEnvironmentVariableA("VTX_CHILD_ASYNC", nullptr); SetEnvironmentVariableA("VTX_CHILD_COMPACT_THRESHOLD", nullptr); ASSERT_TRUE(created); @@ -2042,6 +2113,50 @@ TEST(CrashRecoveryProcess, KilledAtVariedProgressPointsSoak) { } } +// --- The same real-kill matrix, with ASYNC I/O --------------------------------- +// +// With async_io the chunk/journal writes happen on a worker thread, so a cold kill lands +// mid-write on a thread the recording loop never synchronized with. The recovery contract +// is unchanged: whatever became durable must recover as a CLEAN CONTIGUOUS PREFIX (frames +// 0..N-1, correct per-frame content, footer times sized to match). Only the durability LAG +// differs -- the not-yet-drained queue is lost, which is equivalent to crashing earlier. + +TEST(CrashRecoveryProcess, KilledMidRecordingDurableAsync) { + RunKilledProcessTest(/*durable=*/true, "durable_async", /*compact_threshold=*/0, + /*progress_threshold=*/20'000, /*async=*/true); +} + +TEST(CrashRecoveryProcess, KilledMidRecordingFlushOnlyAsync) { + RunKilledProcessTest(/*durable=*/false, "flushonly_async", /*compact_threshold=*/0, + /*progress_threshold=*/20'000, /*async=*/true); +} + +TEST(CrashRecoveryProcess, KilledMidRecordingWhileCompactingAsync) { + RunKilledProcessTest(/*durable=*/true, "compacting_async", /*compact_threshold=*/1, + /*progress_threshold=*/20'000, /*async=*/true); +} + +TEST(CrashRecoveryProcess, KilledAtVariedProgressPointsSoakAsync) { + struct Round { + bool durable; + uint64_t compact_threshold; + uint64_t progress_threshold; + const char* tag; + }; + const Round rounds[] = { + {true, 0, 800, "soak_async_early"}, + {false, 0, 3'000, "soak_async_firstchunk"}, + {true, 1, 8'000, "soak_async_compact"}, + {false, 0, 25'000, "soak_async_deep"}, + }; + for (const auto& r : rounds) { + SCOPED_TRACE(r.tag); + RunKilledProcessTest(r.durable, r.tag, r.compact_threshold, r.progress_threshold, /*async=*/true); + if (::testing::Test::HasFatalFailure()) + return; + } +} + #endif // _WIN32 // A record whose length field claims far more than the journal holds must be