From ab74a848c5a1a601532624676dc087cc68f37dfc Mon Sep 17 00:00:00 2001 From: Lavkush Kumar Date: Tue, 14 Jul 2026 00:16:05 +0530 Subject: [PATCH 1/3] feat(event): add failure-atomic slab/ring admission boundary (ADR-0009) Composite volatile admission channel binding one published-prefix page to one SPSC descriptor ring, under Proposed ADR-0009: - try_admit publishes page-then-ring with no fallible work between the page publication and ring publication phases; a caught unwind closes the page, Release-stores FAULTED, closes the ring, and resumes the panic so a retained producer is terminal, never an atomically open poisoned lane - consumer validation fails closed: sequence gap/duplicate/reorder, arena identity, payload range, and CRC faults withhold the tail, disconnect the producer, and poison the consumer; ring invariant errors Release-close the ring before surfacing - malformed terminal closure states (clean-without-page-close, unknown raw state, count mismatches in either direction) never report CleanEnd - endpoints are Send but deliberately not Sync; frame leases are neither Send nor Sync (compile_fail doc-tests enforce all of it) Evidence: retained-producer caught-unwind tests at every publication phase, corrupt-descriptor and terminal-state fixtures, Loom models including a validation-failure vs reserved-publication race, full Miri, ASan/TSan clean on the final state, zero-allocation admission bench, and CI lanes extended for the admission suite. Status honesty: ADR-0009 is Proposed; the fabric remains target, unwired, with no performance claims (Implementation_Status ledger, MIGRATION_MATRIX delta, ROADMAP, LLD updated in the same change). --- .github/workflows/ci.yml | 37 +- ARCHITECTURE.md | 18 +- MIGRATION_MATRIX.md | 9 +- README.md | 2 +- ROADMAP.md | 40 +- docs/Documentation_Quality_Report.md | 9 +- docs/Implementation_Status.md | 12 +- docs/LLD.md | 105 +- ...0009-failure-atomic-slab-ring-admission.md | 426 ++++++ docs/adr/index.md | 3 +- docs/current-vs-roadmap.md | 4 +- lib/event/Cargo.toml | 4 + lib/event/benches/admission.rs | 65 + lib/event/src/admission.rs | 1340 +++++++++++++++++ lib/event/src/lib.rs | 15 +- lib/event/src/published_slab.rs | 143 +- lib/event/src/ring.rs | 672 ++++++++- lib/event/tests/admission.rs | 443 ++++++ lib/event/tests/published_allocations.rs | 67 +- lib/event/tests/spsc.rs | 114 ++ mkdocs.yml | 1 + 21 files changed, 3395 insertions(+), 134 deletions(-) create mode 100644 docs/adr/0009-failure-atomic-slab-ring-admission.md create mode 100644 lib/event/benches/admission.rs create mode 100644 lib/event/src/admission.rs create mode 100644 lib/event/tests/admission.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ca8a2e9e..b67b1d80 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -104,12 +104,13 @@ jobs: --estimates target/criterion/receipt_hash/compute_receipt_hash_mid_chain/new/estimates.json \ --threshold 0.20 - # ── ADR-0006/0007/0008: event primitive safety gates ──────────────────────── - # The ring and published-prefix algorithms compile once with native std - # atomics/UnsafeCell and once with Loom's modeled equivalents. This job stays - # separate from the gateway matrix so the hot-path crate cannot accidentally - # acquire Tokio, SQLx, protobuf, or system-service dependencies. CRC32C is the - # only default runtime edge; Loom is optional model-checking infrastructure. + # ── ADR-0006/0007/0008/0009: event primitive safety gates ──────────────────── + # The ring, published-prefix, permit/claim, and composite-admission algorithms + # compile once with native std atomics/UnsafeCell and once with Loom's modeled + # equivalents. This job stays separate from the gateway matrix so the hot-path + # crate cannot accidentally acquire Tokio, SQLx, protobuf, or system-service + # dependencies. CRC32C is the only default runtime edge; Loom is optional + # model-checking infrastructure. event-concurrency: name: Event primitives native + Loom runs-on: ubuntu-latest @@ -126,7 +127,7 @@ jobs: run: cargo test -p aegis-event - name: All-feature correctness and feature-isolation tests run: cargo test -p aegis-event --all-features - - name: Loom ring and published-prefix publication models + - name: Loom ring, published-prefix, and admission models run: cargo test -p aegis-event --features loom loom_ - name: All-feature Clippy run: cargo clippy -p aegis-event --all-targets --all-features -- -D warnings @@ -135,7 +136,7 @@ jobs: test "$(cargo tree -p aegis-event --edges normal --depth 1 --prefix none | tail -n +2)" = "crc32c v0.6.8" test "$(cargo tree -p aegis-event --edges normal --depth 1 --prefix none --all-features | tail -n +2)" = "$(printf 'crc32c v0.6.8\nloom v0.7.2')" - name: Compile diagnostic event benchmarks - run: cargo bench -p aegis-event --bench spsc_ring --bench published_slab --no-run + run: cargo bench -p aegis-event --bench spsc_ring --bench published_slab --bench admission --no-run event-miri: name: Event primitives Miri @@ -149,14 +150,14 @@ jobs: with: workspaces: . key: event-miri - - name: Miri ring, sealed-page, and published-prefix ownership suite + - name: Miri ring, slab-page, and admission ownership suite run: cargo miri test -p aegis-event # Rust nightly currently exposes ASan and TSan but no `undefined` sanitizer. # Miri is therefore the Rust UB/provenance gate; any future C/C++ ABI must add # a real UBSan lane before that boundary can be accepted. event-sanitizers: - name: Event published-prefix ${{ matrix.sanitizer }} sanitizer + name: Event slab/admission ${{ matrix.sanitizer }} sanitizer runs-on: ubuntu-latest strategy: fail-fast: false @@ -171,16 +172,20 @@ jobs: with: workspaces: . key: event-${{ matrix.sanitizer }}-sanitizer - - name: Native raw-pointer and cross-thread stress + - name: Native slab and composite-admission stress env: RUSTFLAGS: -Zsanitizer=${{ matrix.sanitizer }} ASAN_OPTIONS: detect_leaks=1:halt_on_error=1 TSAN_OPTIONS: halt_on_error=1 - run: >- - cargo test -Zbuild-std - --target x86_64-unknown-linux-gnu - -p aegis-event - --test published_slab + run: | + cargo test -Zbuild-std \ + --target x86_64-unknown-linux-gnu \ + -p aegis-event \ + --test published_slab + cargo test -Zbuild-std \ + --target x86_64-unknown-linux-gnu \ + -p aegis-event \ + --test admission # ── #1194 (Postgres GA): live Postgres integration smoke test ──────────── # Closes the biggest concrete gap named in diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 87ffa78e..8b908dc1 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -183,6 +183,22 @@ There is one SPSC ring for each producer→consumer edge. A writer shard polls i The fixed ring does not need general garbage collection: its sequence barrier proves slot reuse. `crossbeam-epoch` is limited to slab-page retirement, snapshot publication, manifest generations, and readers that can outlive a ring slot. +The `current`, unwired `VolatileAdmissionChannel` prototype binds one +preallocated page to one preallocated ring. It validates page capacity before +reserving a ring slot, Release-publishes the complete immutable page prefix, +and only then Release-publishes the descriptor ring head. Its consumer claims a +descriptor without releasing ring capacity, validates canonical page +membership, bounds, and CRC32C, and advances the consumed tail only when a +must-use frame lease commits. Explicit clean/faulted terminal state prevents an +orphaned page prefix or ordinary producer drop from being reported as a clean +stream. This process-local prototype carries no production or `shadow` +traffic, cannot carry protected evidence, is not `qualified`, and establishes +no performance result. The production event fabric remains `target`; WAL +durability, formal ADR/security review, green hosted ASan/TSan artifacts, real +UBSan support, authenticated registry lookup, bounded page rotation, reuse, +epoch/NUMA reclamation, shadow evidence, rollback, and qualification remain +separate gates. + ### Event admission states ```mermaid @@ -193,7 +209,7 @@ stateDiagram-v2 Validating --> Spooling: critical ring saturated Validating --> Rejected: normal ring saturated Reserved --> Published: release sequence - Published --> Consumed: writer acquire sequence + Published --> Consumed: validate and commit frame lease Consumed --> Durable: WAL policy satisfied Durable --> Indexed: memtable/segment visible Spooling --> Published: capacity restored diff --git a/MIGRATION_MATRIX.md b/MIGRATION_MATRIX.md index d2703605..f59b5a27 100644 --- a/MIGRATION_MATRIX.md +++ b/MIGRATION_MATRIX.md @@ -9,9 +9,10 @@ **Audited baseline HEAD:** `f027d07` (`feat(tool-broker): extract connector execution into standalone aegis-tool-broker binary`) **Post-audit delta:** the `current` branch adds unwired `aegis-event` SPSC, -sealed generation-tagged slab-page, and append-only published-prefix prototypes -under ADR-0006 through ADR-0008. The lexical counts below remain the frozen -baseline so future excision is measured against one reproducible commit. +sealed generation-tagged slab-page, append-only published-prefix, and +failure-atomic single-page slab/ring admission prototypes under ADR-0006 +through ADR-0009. The lexical counts below remain the frozen baseline so future +excision is measured against one reproducible commit. **Companion documents:** [HLD](ARCHITECTURE.md), [LLD](docs/LLD.md), [Roadmap](ROADMAP.md) @@ -245,7 +246,7 @@ The v1 router’s 146 path literals cover agents, tools, MCP, authorization, ing | Target capability | Current evidence | Gap | |---|---|---| | Thread-per-core reactor | Tokio multi-thread runtime; no affinity crate/config | No core ownership, per-core listener, NUMA allocation, or io_uring reactor | -| Disruptor/SPSC event fabric | Tokio bounded MPSC remains `current`; unwired `lib/event` adds 64-byte-separated cursors, Acquire/Release sequences, 32-byte descriptors, bounded closure/drop behavior, a safe seal-before-publish differential oracle, and an append-only page whose packed Release/Acquire state publishes descriptor count, byte watermark, and closure for immediate immutable-prefix resolution. Evidence includes native stress, the same publication algorithm under Loom, full Miri, defined ASan/TSan CI lanes, and a zero-allocation append-plus-resolve test. | The production fabric remains `target`, neither `shadow` nor `qualified`, and has no performance result. Blockers are green sanitizer CI artifacts, ADR acceptance/security review, composite ring reservation/admission, authenticated registry lookup, bounded page rotation/outstanding pages, generation reuse/epochs, NUMA-owner reclamation, priority lanes, production shadow wiring, UBSan support in the Rust toolchain, and qualification. | +| Disruptor/SPSC event fabric | Tokio bounded MPSC remains `current`; unwired `lib/event` adds 64-byte-separated cursors, Acquire/Release sequences, 32-byte descriptors, bounded closure/drop, cancelable producer permits, commit-delayed consumer claims, a safe sealed-page oracle, an append-only published prefix, and `VolatileAdmissionChannel` composition. The composite validates before reservation, publishes the page before the ring, withholds capacity until a validated frame lease commits, and distinguishes clean, faulted, and orphaned-prefix termination. Test sources include safe short-trace differential coverage, native tiny-ring stress, shipping-algorithm Loom models, Miri-oriented lifetime cases, defined ASan/TSan CI lanes, and zero-allocation admission/claim checks. | The production fabric remains `target`, neither `shadow` nor `qualified`, carries no protected evidence, and has no performance result. Blockers are formal ADR acceptance/security review, green hosted sanitizer artifacts, real UBSan support, authenticated registry lookup, bounded page rotation/outstanding pages, WAL durability/replay, generation reuse/epochs, NUMA-owner reclamation, priority lanes, production shadow wiring, release-artifact rollback, and qualification. | | HCMT/Arrow SSTables | SQLite/PostgreSQL rows; JSON/TEXT payloads | No Arrow dependency, WAL format, memtable, segment manifest, compactor, or mmap query path | | Gorilla timestamp codec | none | Codec, block restart points, fallback-to-raw rule, corpus absent | | Roaring pruning | none | Bitmap build/serialization/planner absent | diff --git a/README.md b/README.md index e9d20711..320167dd 100644 --- a/README.md +++ b/README.md @@ -77,7 +77,7 @@ Authorization compute and protected commit are measured separately. A warm deter | APIs | REST JSON plus partial tonic/protobuf | protobuf-first parity; binary fast path; REST compatibility off benchmark path | | Control storage | SQLite/PostgreSQL via `StorageBackend` | split transactional `ControlStore`/`ReceiptLog` | | Telemetry storage | row tables and JSON/TEXT fields | WAL + Arrow-compatible HCMT SSTables | -| Event bus | Tokio bounded MPSC; `current`, unwired `aegis-event` SPSC, safe sealed-page, and append-only published-prefix prototypes with packed Release/Acquire count, byte-watermark, and closure publication | `target` NUMA-local cache-padded SPSC/slab matrix after composite admission, registry, rotation, epoch/reclamation, shadow, safety, and qualification gates | +| Event bus | Tokio bounded MPSC; `current`, unwired `aegis-event` SPSC, safe sealed-page, append-only published-prefix, and failure-atomic single-page volatile admission prototypes with cancelable permits, commit-delayed claims, and explicit clean/faulted termination; no protected evidence or performance claim | `target` NUMA-local SPSC/slab matrix after formal ADR/security review, hosted sanitizer and UBSan evidence, authenticated registry, bounded rotation, WAL durability/replay, epochs/NUMA reclamation, shadow, rollback, and qualification gates | | Detection | structured scalar rules; optional Qdrant | Aho DFA plus owned HNSW/PQ and isolated INT8 ONNX | | Host sensor | procfs polling, spool, signed commands | CO-RE eBPF telemetry/containment with truthful fallback | | Console | React JSON polling and SVG | Arrow IPC worker, Rust WASM, WebGL2 instancing | diff --git a/ROADMAP.md b/ROADMAP.md index d3c717ea..2c1b81e8 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -66,26 +66,32 @@ Gate: legacy versus typed authorization decisions, hashes, approvals, receipts a ### Week 4 — SPSC ring and slab prototype Progress (2026-07-13): the `current` checkout has unwired ring, safe sealed-page -oracle, and append-only published-prefix prototypes under Proposed ADR-0006 -through ADR-0008. Packed Release/Acquire state publishes descriptor count, byte -watermark, and closure for immediate immutable-prefix resolution. Evidence -includes a safe sealed differential corpus, native stress, the same publication -algorithm under Loom, full Miri, defined ASan/TSan CI lanes, and a zero-allocation -append-plus-resolve test. The production fabric remains `target`, neither -`shadow` nor `qualified`, and has no performance result. ADR acceptance and -security review, green sanitizer CI artifacts, composite ring reservation/admission, authenticated registry, -bounded page rotation/outstanding pages, generation reuse/epochs, NUMA-owner -reclamation, production shadow wiring, UBSan support in the current Rust -toolchain, and qualification remain blockers. - -Deliverables: - -- implement cache-padded SPSC ring, 32-byte descriptor and NUMA-local slab prototype; +oracle, append-only published-prefix, and failure-atomic single-page admission +prototypes under Proposed ADR-0006 through ADR-0009. The volatile composite +validates before reservation, Release-publishes the page before the ring, +withholds capacity until a must-use validated frame lease commits, and reports +clean, faulted, and orphaned-prefix terminal states. Test sources include safe +differential oracles, native tiny-ring stress, shipping-algorithm Loom, +Miri-oriented borrow/drop cases, defined ASan/TSan CI lanes, and zero-allocation +admission/claim checks. It carries no production or `shadow` traffic, cannot +carry protected evidence, is not `qualified`, and has no performance result. +The production fabric remains `target`. Formal ADR acceptance/security review, +green hosted sanitizer artifacts, real UBSan support, authenticated registry, +bounded page rotation/outstanding pages, WAL durability/replay, generation +reuse/epochs, NUMA-owner reclamation, priority lanes, production shadow wiring, +release-artifact rollback, and qualification remain blockers. + +Deliverables: + +- implement cache-padded SPSC ring, 32-byte descriptor, bounded slab, and single-page volatile admission prototype; - document linearization, memory ordering, shutdown, wrap, drop and epoch rules; -- add scalar reference queue, Loom model, Miri tests and native stress benchmark; +- add safe differential oracles, Loom models, Miri tests and native stress benchmark; - instrument allocations, copied bytes, cache misses and cycles/op. -Gate: zero lost/duplicated descriptors; zero steady-state allocations; safety suite green; no false sharing in layout/perf evidence. +Gate: zero lost/duplicated/reordered descriptors; validation before tail +acknowledgement; zero steady-state allocations; safety suite and hosted +sanitizer evidence green; no false sharing in layout/perf evidence. These gates +do not make the volatile prototype durable or authorize protected evidence. ### Week 5 — CoreReactor prototype diff --git a/docs/Documentation_Quality_Report.md b/docs/Documentation_Quality_Report.md index 93e497f4..2c0f6328 100644 --- a/docs/Documentation_Quality_Report.md +++ b/docs/Documentation_Quality_Report.md @@ -12,7 +12,7 @@ This inventory makes the all-documentation improvement program measurable. Each |---|---:|---:|---:| | authoring | 2 | 90% | 0 | | component | 15 | 96% | 0 | -| decision | 10 | 95% | 0 | +| decision | 11 | 95% | 0 | | flow | 11 | 95% | 0 | | guide | 53 | 67% | 38 | | landing | 10 | 93% | 0 | @@ -21,7 +21,7 @@ This inventory makes the all-documentation improvement program measurable. Each | reference | 11 | 85% | 0 | | runbook | 7 | 100% | 0 | -**Total:** 126 active Markdown pages · **Migration backlog:** 38 pages below 75%. +**Total:** 127 active Markdown pages · **Migration backlog:** 38 pages below 75%. ## Scoring signals @@ -87,7 +87,8 @@ Pages are sorted by structural coverage, then path. Improve factual accuracy and | [adr/0006-cache-padded-spsc-event-fabric.md](adr/0006-cache-padded-spsc-event-fabric.md) | decision | 189 | B | 83% | | [adr/0007-sealed-generation-tagged-slab-pages.md](adr/0007-sealed-generation-tagged-slab-pages.md) | decision | 267 | A | 100% | | [adr/0008-append-only-published-prefix-slab-pages.md](adr/0008-append-only-published-prefix-slab-pages.md) | decision | 409 | A | 100% | -| [adr/index.md](adr/index.md) | decision | 50 | B | 83% | +| [adr/0009-failure-atomic-slab-ring-admission.md](adr/0009-failure-atomic-slab-ring-admission.md) | decision | 427 | A | 100% | +| [adr/index.md](adr/index.md) | decision | 51 | B | 83% | | [adr/template.md](adr/template.md) | decision | 46 | A | 100% | | [AegisAgent_Agent_Cage.md](AegisAgent_Agent_Cage.md) | guide | 581 | B | 78% | | [AegisAgent_Agent_SOC_Design.md](AegisAgent_Agent_SOC_Design.md) | guide | 757 | D | 56% | @@ -169,7 +170,7 @@ Pages are sorted by structural coverage, then path. Improve factual accuracy and | [installation.md](installation.md) | guide | 127 | A | 100% | | [Issue_Backlog_Execution_Plan.md](Issue_Backlog_Execution_Plan.md) | guide | 208 | D | 44% | | [Last_Mile_System_Walkthrough.md](Last_Mile_System_Walkthrough.md) | guide | 147 | C | 67% | -| [LLD.md](LLD.md) | guide | 1537 | A | 100% | +| [LLD.md](LLD.md) | guide | 1584 | A | 100% | | [Local_Development.md](Local_Development.md) | guide | 87 | D | 44% | | [mcp-defense-architecture.md](mcp-defense-architecture.md) | guide | 145 | C | 67% | | [mission.md](mission.md) | guide | 153 | D | 44% | diff --git a/docs/Implementation_Status.md b/docs/Implementation_Status.md index adb63f17..d53047d3 100644 --- a/docs/Implementation_Status.md +++ b/docs/Implementation_Status.md @@ -20,15 +20,15 @@ separate documentation change. | Artifact | Status | Evidence in this checkout | Authority / traffic | Remaining gates | |---|---|---|---|---| -| Unwired SPSC, sealed-page oracle, and published-prefix prototypes | current | `lib/event/`; ring FIFO/full/wrap/drop/layout; safe sealed differential corpus; packed Release/Acquire descriptor-count, byte-watermark, and closure publication with immediate immutable-prefix resolution; native stress; same-algorithm Loom; full Miri; ASan/TSan CI lanes defined; zero-allocation append-plus-resolve test; ADR-0006 through ADR-0008 | No production or `shadow` traffic; cannot carry protected evidence; no performance claim | Green sanitizer CI artifacts, ADR acceptance/security review, composite ring reservation/admission, authenticated registry, bounded page rotation/outstanding pages, generation reuse/epochs, NUMA-owner reclamation, production shadow wiring, UBSan support, qualification | +| Unwired SPSC, slab-page, published-prefix, and single-page admission prototypes | current | `lib/event/`; ring FIFO/full/wrap/drop/layout plus cancelable permits and commit-delayed claims; safe sealed and short-trace differential oracles; packed page publication; page-before-ring admission; must-use frame leases; clean/faulted/orphan terminal checks; native stress; shipping-algorithm Loom; Miri-oriented borrow/drop tests; ASan/TSan CI lanes defined; zero-allocation append/resolve and admission/claim checks; ADR-0006 through ADR-0009 | No production or `shadow` traffic; cannot carry protected evidence; volatile admission is not a receipt or durability acknowledgement; no performance claim | Formal ADR acceptance/security review, green hosted sanitizer artifacts, UBSan support, authenticated registry, bounded page rotation/outstanding pages, WAL durability/replay, generation reuse/epochs, NUMA-owner reclamation, priority lanes, production shadow wiring, release-artifact rollback, qualification | | Thread-per-core reactor | target | `ARCHITECTURE.md`, `docs/LLD.md` | None | runtime ADR, core-affinity/io_uring implementation, migration and benchmark gates | | HCMT telemetry store | target | `ARCHITECTURE.md`, `docs/LLD.md` | None; SQL remains authoritative/current | WAL/segment ADR, recovery corpus, dual write, shadow equality, qualification | -No v2 component is `qualified` in this checkout. The SPSC, sealed-page, and -published-prefix prototypes being `current` means only that their isolated, -unwired code and listed tests are present. The production event fabric remains -`target`, is neither `shadow` nor `qualified`, and is not -production-authoritative. +No v2 component is `qualified` in this checkout. The SPSC, slab-page, +published-prefix, and single-page admission prototypes being `current` means +only that their isolated, unwired code and listed tests are present. The +production event fabric remains `target`, is neither `shadow` nor `qualified`, +and is not production-authoritative. | Capability | Status | Current files | Missing pieces | Related issues | Test coverage | Prod-ready | Next PR | |---|---|---|---|---|---|---|---| diff --git a/docs/LLD.md b/docs/LLD.md index 35ed3e71..6558bc4a 100644 --- a/docs/LLD.md +++ b/docs/LLD.md @@ -199,23 +199,26 @@ Admission is a reactor-local counter and byte budget. Permits are acquired befor ## 5. Disruptor-style SPSC ring **Implementation status:** `lib/event` contains `current`, unwired prototypes -governed by Proposed ADR-0006 through ADR-0008. The ring implements -non-cloneable owning endpoints, checked capacity, modular wrap, closure/drain -semantics, unread-value destruction, and cache-layout assertions. The safe +governed by Proposed ADR-0006 through ADR-0009. The ring implements +non-cloneable owning endpoints, checked capacity, modular wrap, closure/drain, +unread-value destruction, cache-line isolation, cancelable producer permits, +and consumer claims that withhold tail advancement until commit. The safe sealed-page reference provides bounded layout, exact descriptor membership, -CRC32C, and immutable page-level lifetime. The append-only page uses -Release/Acquire to publish one packed descriptor-count, byte-watermark, and -closure state, then immediately resolves immutable prefix views while its -writer appends to a disjoint suffix. Evidence includes a safe sealed -differential corpus, native stress, the same publication algorithm under Loom, -full Miri, defined ASan/TSan CI lanes, and a zero-allocation append-plus-resolve -test. These -prototypes carry no production or protected-evidence traffic, are neither -`shadow` nor `qualified`, and make no performance claim. ADR acceptance and -security review, green sanitizer CI artifacts, composite ring reservation/admission, authenticated registry -lookup, bounded page rotation and outstanding pages, generation reuse and -epochs, NUMA-owner reclamation, production shadow wiring, UBSan support, and -qualification remain `target` gates. +CRC32C, and immutable page-level lifetime. The append-only page +Release/Acquire-publishes a coherent descriptor-count, byte-watermark, and +closure word. The single-page admission composite validates before reservation, +publishes the page before the ring, validates a must-use frame lease before +reclaiming its slot, and distinguishes clean termination, faulted termination, +and orphaned page prefixes. Tests present in the crate include safe sealed and +short-trace differential oracles, native stress, shipping-algorithm Loom +models, Miri-oriented borrow/drop cases, and zero-allocation admission/claim +checks; ASan/TSan CI lanes are defined. These prototypes carry no production or +`shadow` traffic, cannot carry protected evidence, are not `qualified`, and +make no performance claim. Formal ADR acceptance and security review, green +hosted sanitizer artifacts, real UBSan support, authenticated registry lookup, +bounded page rotation and outstanding pages, WAL durability/replay, +generation reuse and epochs, NUMA-owner reclamation, priority lanes, +production shadow wiring, and qualification remain `target` gates. ### 5.1 Memory layout @@ -361,11 +364,41 @@ borrow cannot escape its checked-cell guard. This closes only the within-page publication gap. The production fabric remains `target`, neither `shadow` nor `qualified`, and has no performance -result. ADR acceptance and security review, atomic reservation across page -bytes and the descriptor ring, bounded page rotation and outstanding-page -admission, authenticated lookup, generation reuse with epoch retirement, -NUMA-owner reclamation, production shadow wiring, UBSan coverage when the -toolchain supports it, and qualification remain blockers. +result. ADR-0009 composes this page with one bounded descriptor ring but does +not add rotation, durability, authenticated lookup, reuse, or reclamation. + +#### Current failure-atomic single-page admission prototype + +The `current`, unwired `VolatileAdmissionChannel` owns exactly one +`PublishedSlabPage`, one sequence-aligned `SpscRing`, +and one cache-line-isolated terminal-state word. It exposes one non-cloneable +producer and consumer; raw ring/page endpoints do not escape the composite. +Validation and page-capacity checks precede a cancelable ring reservation. A +successful producer initializes and Release-publishes the page prefix before +an infallible permit commit writes the descriptor and Release-publishes the +ring head. Ring publication is the volatile admission linearization point. + +The consumer claims the next descriptor without advancing the tail, requires +the exact wrapping sequence, Acquire-validates the canonical page descriptor, +range, and CRC32C, and returns a must-use borrowed frame lease. Lease commit is +the only operation that moves the descriptor and Release-advances the consumed +tail; dropping a lease retries the same descriptor. Clean finish closes the +page, publishes `CLEAN`, then closes the ring. Ordinary drop and poisoned or +interrupted operation publish `FAULTED`; after drain, count or terminal-state +mismatch is terminal data loss rather than a guessed clean stream. + +This composition performs one bounded payload copy and no steady-state +allocation or per-event reference-count operation in the native implementation. +Its admission token is not a receipt or durability acknowledgement. A consumer +may close after reservation, and process failure between page and ring Release +stores can leave an orphaned immutable prefix; terminal-state detection is not +WAL recovery. The channel therefore carries neither production nor `shadow` +traffic, cannot carry protected evidence, is not `qualified`, and makes no +performance claim. Formal ADR acceptance/security review, green hosted +ASan/TSan artifacts, UBSan support, authenticated registry lookup, bounded page +rotation/outstanding pages, WAL durability and replay, reuse/epochs, +NUMA-owner reclamation, priority lanes, production shadow wiring, and +qualification remain blockers. ### 5.3 Priority and fairness @@ -1428,6 +1461,16 @@ All library functions return `Result` or a narrower error convert Error payloads contain stable codes, request IDs, and safe metadata, never SQL strings, paths containing secrets, raw prompts, or key material. +For the `current`, unwired single-page admission prototype, invalid/page-full +input, ring saturation, and a consumer disconnected before reservation leave +both page and ring logically unchanged. `AdmittedSequence` confirms only +volatile page-and-ring publication; it is not a receipt, consumption +acknowledgement, WAL acknowledgement, or authorization result. Frame validation +failure withholds the ring-tail acknowledgement and terminates the lane. +Explicit `finish` is required for a clean end; ordinary producer drop is +faulted. Protected evidence is prohibited from this channel and must fail +closed at its separate durability boundary. + ## Security and failure model Security follows four independent fail-closed boundaries: @@ -1457,17 +1500,20 @@ Normal telemetry may be rejected only when its source retains a bounded replay/s ### 29.2 Unsafe/concurrency -- Loom explores the same SPSC and packed published-prefix - count/watermark/closure algorithms used by the `current` prototypes, - including shutdown; -- Miri runs the complete native ring, sealed-page, published-prefix borrow, - buffer, and poisoned-suffix corpus; +- Loom explores the same SPSC permit/claim, packed published-prefix, composite + admission, terminal-state, cancellation, saturation/retry, and shutdown + algorithms used by the `current` prototypes; +- Miri runs the complete native ring permit/claim, sealed-page, + published-prefix borrow, admission frame-lease, buffer, and poisoned-suffix + corpus; - ASan/TSan/UBSan evidence is required on supported native targets; the - `current` prototype defines ASan/TSan CI lanes but still requires their green + `current` prototypes define ASan/TSan CI lanes but still require green hosted artifacts, while UBSan is unavailable in the current Rust toolchain and remains an explicit gate; -- randomized producer/consumer soak resolves immutable prefixes while later - suffixes are appended and extends beyond sequence-wrap simulation; +- deterministic short traces compare admission against a safe + `SlabPageBuilder + VecDeque` oracle; native producer/consumer stress holds + capacity until frame commit, resolves immutable prefixes while later suffixes + are appended, and extends beyond sequence-wrap simulation; - epoch tests prove no pin crosses blocking I/O and no object frees early; - fuzz FlatBuffer verifier, WAL scanner, segment metadata, Gorilla decoder, Arrow envelope and protobuf conversions. @@ -1483,6 +1529,7 @@ Success means no unauthorized execution, no cross-tenant read, no acknowledged p |---|---| | SPSC descriptor | cycles/op, cache misses, zero allocations, sustained wrap | | published-prefix slab | append/resolve cycles, copied bytes, allocations, publication cache-line transfers, producer/consumer overlap, errors at saturation | +| single-page volatile admission | validation/reservation, page-to-ring publication, claim/CRC/commit cycles, full-rejection cost, allocations, copies, terminal faults, loss/duplicate/reorder checks | | FlatBuffer verify | events/s by frame-size distribution | | authorize compute | `<1 ms p99` warm snapshot; p99.9 reported | | protected commit | hardware-profile p99, group size, zero lost receipts | diff --git a/docs/adr/0009-failure-atomic-slab-ring-admission.md b/docs/adr/0009-failure-atomic-slab-ring-admission.md new file mode 100644 index 00000000..a442dfa7 --- /dev/null +++ b/docs/adr/0009-failure-atomic-slab-ring-admission.md @@ -0,0 +1,426 @@ +# ADR-0009: Failure-atomic single-page slab/ring admission + +**Status:** Proposed +**Date:** 2026-07-13 +**Issue/PR:** pending + +## Context + +ADR-0006 provides a bounded SPSC descriptor ring and ADR-0008 provides an +append-only page whose immutable prefix is published after each complete +payload. Calling `PublishedSlabWriter::try_append` and then +`Producer::try_push` does not form one admission operation. A full ring or an +already-disconnected consumer is discovered only after the page has consumed +bytes, descriptor capacity, and sequence space. An interruption between the +page Release store and the ring Release store leaves a complete but +undelivered page descriptor. + +The reverse order is also incomplete. A producer cannot publish a placeholder +ring entry before the payload exists because the consumer could observe an +uninitialized or non-canonical descriptor. A general two-object transaction, +CAS loop, per-slot state machine, or pending overflow queue would add shared +coordination that the single-producer topology does not require. + +Consumption has a symmetric boundary. The current `try_pop` moves the +descriptor and Release-advances the ring tail immediately. If page identity, +canonical membership, bounds, or CRC verification then fails, the producer may +reuse the slot even though the consumer never accepted a valid event. That is +memory-safe for an immutable page but is the wrong acknowledgement contract for +a security-event lane. + +This decision covers one preallocated page paired with one preallocated ring. +It remains volatile, process-local, unwired, and non-authoritative. It does not +provide WAL durability, page rotation, registry authentication, page reuse, +epoch reclamation, or a protected-evidence acknowledgement. + +## Decision + +Add a `VolatileAdmissionChannel` in `aegis-event`. Construction validates +one `SlabPageConfig`, initializes the ring at the same `first_sequence`, and +owns both setup handles plus a cache-line-aligned terminal-state word. Consuming +`split` returns exactly one non-cloneable `AdmissionProducer` and one +non-cloneable `AdmissionConsumer`. Raw page and ring endpoints do not escape +this composite. + +The ring gains two producer/consumer-local capabilities: + +1. `Producer::try_reserve` returns a unique vacant-slot permit after checking + consumer closure and capacity. It does not write a slot or advance the + published head. Dropping the permit is a no-op. +2. `Consumer::try_claim` returns a read claim for a `Copy` value after Acquire + observing the published head. It does not move the slot value or advance the + consumed tail. Dropping the claim is a no-op. `commit` moves the value once + and Release-advances the tail. + +Both capabilities exclusively borrow their endpoint, are non-cloneable, +`!Send`, and `!Sync`. They contain no allocation, wait, retry, callback, lock, +or async operation. Their commit operations are infallible after issuance and +use a slot reference staged before any composite page mutation or validation +lease is returned. + +The producer admission algorithm is: + +```text +validate payload length and current page capacity without CRC or mutation +require page next_sequence == ring next_sequence +reserve one vacant ring slot without publishing it +mark producer InFlight +compute CRC32C and append the page payload/canonical descriptor +Release-store the page prefix [P] +write the descriptor into the reserved ring slot +Release-store the ring published head [R] +mark producer Open +return a volatile admission token +``` + +`[R]` is the composite admission linearization point. The page publication +`[P]` is necessarily earlier because a consumer must never receive a +descriptor before its complete immutable payload is page-visible. The two +stores update distinct ownership domains and cannot honestly be described as +one crash-atomic instruction. + +The consumer algorithm is: + +```text +Acquire-observe and claim the next ring descriptor without advancing tail +require the exact expected wrapping descriptor sequence +Acquire-load and validate the page prefix +require canonical descriptor equality, checked bounds, and CRC32C +return a must-use borrowed frame lease + +frame lease commit: + move the ring descriptor exactly once + Release-store the consumed tail [C] + advance consumer-local expected sequence and count +``` + +`[C]` is the capacity-reclamation and consumer-acknowledgement linearization +point. Dropping an uncommitted frame lease leaves the ring slot claimed but +unconsumed; a later call may validate the same event again. The lease is +non-cloneable, `!Send`, and `!Sync` and MUST NOT cross I/O, `.await`, a callback, +or an uncontrolled duration. + +## Public result semantics + +Successful producer admission returns an `AdmittedSequence` containing only +the page identity/generation and wrapping sequence. It is not named or treated +as a receipt. Success means: + +- the complete payload and canonical descriptor are immutable in the bound + page; +- the descriptor is published in the volatile SPSC ring; +- a correctly operating bound consumer can Acquire-observe it. + +Success does not mean consumed, WAL-appended, durable, replicated, indexed, or +authorized. An upstream source retains its bounded replay/spool record until a +separate downstream durability acknowledgement unless the declared event class +is explicitly best effort. + +Normal typed errors are transactional: + +| Error | Page mutation | Ring logical mutation | Retry meaning | +|---|---:|---:|---| +| invalid, empty, oversized, page full, descriptor full | none | none | correct input or rotate in a future layer | +| ring full | none; CRC and payload copy do not run | none | bounded retry/spool according to class | +| consumer disconnected before reservation | none | none | lane unavailable | +| producer poisoned or page/ring sequence divergence | none after detection | none after detection | terminal; rebuild lane | + +A consumer closing after reservation cannot revoke the permit. The producer +still publishes at most that already-reserved descriptor and may return +volatile success. This is the unavoidable SPSC shutdown race already allowed +by ADR-0006; it is another reason admission is not a durability +acknowledgement. + +## Explicit closure and terminal state + +The composite terminal word is 64-byte aligned and has three valid values: + +```text +OPEN = 0 +CLEAN = 1 +FAULTED = 2 +``` + +Only the producer writes it. `AdmissionProducer::finish(self)` performs, in +order: + +1. Release-close the page writer; +2. Release-store `CLEAN` to the terminal word; +3. Release-close the ring producer. + +Ordinary producer drop, a caught unwind, a poisoned state, or a failed finish +performs the same ordered closure with `FAULTED`. Page closure therefore +happens-before any consumer that Acquire-observes ring disconnection. Repeated +close operations from field destruction are idempotent. + +After the ring is drained and disconnected, the consumer Acquire-loads one +validated page status and the terminal word. Clean end-of-stream requires all +of the following: + +- page writer is closed; +- terminal state is `CLEAN`; +- validated page published count equals committed frame count; +- every committed descriptor followed the exact wrapping sequence. + +`FAULTED` with a larger page count reports an orphaned published prefix. +`FAULTED` with equal counts reports a faulted producer. An open/unknown terminal +state, a page that is not closed, a count mismatch in either direction, a +sequence gap/duplicate/reorder, canonical mismatch, range error, or CRC failure +is terminal data loss. The consumer closes its ring endpoint and never skips to +the next descriptor. + +## Ordering proof + +For one admitted descriptor, the relevant happens-before chain is: + +```text +payload bytes and canonical page descriptor initialized + -> page publication Release [P] + -> reserved ring slot initialized + -> ring-head Release [R] + -> consumer ring-head Acquire + -> descriptor copy from claimed initialized slot + -> page-state Acquire + -> canonical/range/CRC validation + -> ring-tail Release [C] + -> producer ring-tail Acquire before slot reuse +``` + +Program order plus the Release/Acquire synchronization means a consumer that +observes `[R]` cannot observe a page state older than the descriptor's complete +publication when it subsequently Acquire-loads that page. A later page prefix +may be visible, but committed cells never change. Withholding `[C]` prevents the +producer from reusing the claimed ring slot while validation or a frame lease +is outstanding. + +The ring permit needs no shared reserved bit. There is exactly one producer, +the permit holds its exclusive mutable borrow, and the consumer can only +increase available capacity. Cached remote cursors may cause a conservative +full result but can never over-admit. The ring/page capacity is below `2^63`, so +wrapping cursor distance remains unambiguous. One page contains at most `2^16` +descriptors, making page-sequence distance unambiguous across `u64::MAX -> 0`. + +## Panic, cancellation, and partial initialization + +The composite producer uses local `Open`, `InFlight`, `Poisoned`, and +`Finished` states. The public admission boundary catches an unwind only long +enough to close the page, Release-store `FAULTED`, close the ring, and resume +the same panic; it never converts a panic into a normal result. A caller that +catches the resumed unwind therefore retains a terminal producer rather than +an atomically open poisoned lane. The producer enters `InFlight` only after all +typed validation and ring reservation succeed. No allocation, lookup, +indexing, formatting, callback, branch returning an error, or peer-state +recheck occurs after `[P]` and before `[R]` in production code. Test-only fault +hooks at the phase boundaries are compiled out of non-test builds. + +- Before the first page cell write, an unwind cancels the unused permit. +- During page mutation before `[P]`, ADR-0008 poisoning leaves an unreachable + suffix and forbids writer reuse; the ring permit remains unpublished. +- After `[P]` and before `[R]`, rollback is impossible. The page prefix remains + immutable, the producer remains poisoned, ordered faulted closure exposes the + count mismatch, and the consumer reports an orphan rather than guessing or + continuing. +- After `[R]`, the descriptor remains drainable exactly once. Faulted closure + still prevents the lane from being reported as a clean stream. + +Process abort loses this volatile channel and requires no in-process rollback. +Process-crash recovery and durable replay belong to the future WAL/page-rotation +protocol. Protected evidence remains prohibited from this channel. + +## Memory and ownership layout + +```text +setup: VolatileAdmissionChannel + page Arc ------------------------------+ + ring Arc -------------------------+ | + terminal Arc ----------------+ | | + | | | +split v v v +producer: [slab writer][ring producer][terminal][local state] +consumer: [ring consumer][slab reader][terminal][expected/count/state] + +producer hot path: + page publication cache line [P] + ring published-head cache line [R] + +consumer hot path: + ring consumed-tail cache line [C] + +terminal cache line: + written only during clean/faulted shutdown +``` + +The producer closes the page before the terminal word and ring. The consumer +closes the ring endpoint before releasing the page reader. The terminal word is +not colocated with either hot cursor, so shutdown metadata does not introduce +steady-state false sharing. All allocations and `Arc` clones happen during +construction/split; no per-event reference count operation is permitted. + +The composite module is safe Rust. The permit and claim extend the existing +reviewed unsafe ring boundary. Their staged slot references derive from the +fixed ring allocation after checked masking/index validation. A permit writes +only the producer-owned free slot. A claim copies only an initialized, +Acquire-published `Copy` value and commits by moving that slot once. An +uncommitted claim leaves the initialized slot untouched. Existing final-drop +logic still destroys every published, unconsumed value exactly once. + +## Copy and allocation ledger + +| Boundary | Payload copies | Descriptor movement | Allocation/refcount | +|---|---:|---:|---| +| channel construction/split | zero | zero | bounded page/ring storage and setup-only `Arc` clones | +| validation + ring reservation | zero | none | zero | +| caller slice -> page suffix | exactly one bounded copy | canonical descriptor initialized once | zero | +| page -> ring | zero | one 32-byte slot write | zero | +| ring claim + page validation | zero | one 32-byte validation copy; slot remains owned by ring | zero | +| frame commit | zero | original slot value moved once | zero | +| native payload lease | zero within page-to-consumer boundary | none | zero | +| Loom payload validation | one disclosed model-only copy | modeled claim/commit | test-only allocation | + +CRC32C is `O(n)` corruption detection over payload length `n`; it is not +authentication. `arena_id` and generation detect stale/mismatched page routing +but are not tenant credentials or capabilities. + +## Failure, overload, and security behavior + +- Validation precedes reservation and prevents malformed input from being + hidden behind saturation. +- Ring full executes bounded work, does not compute CRC32C, does not copy the + payload, and does not consume page sequence or capacity. +- Page full and descriptor exhaustion occur before ring reservation; there is + no pending descriptor queue. +- No automatic spin, sleep, allocation, page rotation, overflow node, or + best-effort conversion exists. +- Raw descriptors are accepted only from the channel's bound ring. They are not + capabilities and never select a tenant or registry entry. +- The page belongs to one already-authenticated producer/consumer edge. Future + registry lookup must bind trusted tenant, shard, arena ID, and generation + before descriptor-cell access. +- CRC mismatch, identity mismatch, invalid metadata, sequence discontinuity, + corrupt terminal state, or count mismatch terminates the lane without + exposing bytes or acknowledging the ring slot. +- Admission cannot allow an action, consume an approval, satisfy receipt + durability, or carry raw credentials. Cedar and protected control semantics + are unchanged. + +## Progress and performance hypothesis + +Validation and ring reservation are wait-free bounded operations. Successful +admission performs bounded `O(n)` CRC/copy work followed by fixed slot and +atomic operations. Claim, frame validation, and commit perform bounded `O(n)` +CRC work and fixed atomics. Neither side waits for the peer; an empty/full result +is immediate. + +The hypothesis is that one local capacity check before the payload copy removes +the current wasted page work at saturation while retaining one payload copy and +descriptor-only transfer. This ADR provides no latency, cycles, cache-miss, +allocation, or throughput measurement and does not make the event fabric +`shadow` or `qualified`. + +Qualification must report p50/p95/p99/p99.9, events/s, bytes/s, CRC/copy cost, +full-rejection cost, allocations/event, copied bytes/event, cache-line +transfers, cache/branch misses, producer stalls, errors, saturation, and exact +loss/duplicate/reorder checks on a declared hardware/NUMA profile. + +## Alternatives considered + +- **Append then call `try_push`** — rejected because full/disconnected errors + occur after page mutation and invite duplicate retries. +- **Publish a ring placeholder first** — rejected because the consumer could + observe a descriptor before canonical bytes exist. +- **Keep one pending descriptor in the producer** — rejected because it adds a + second queue, consumes page capacity on ring full, blocks later admissions, + and complicates drop/retry semantics. +- **Per-slot reserved atomics or a two-word CAS** — rejected because the sole + producer already provides exclusive reservation and the page/ring words are + on distinct allocations. +- **Advance ring tail before page validation** — rejected because it + acknowledges capacity before the consumer accepts a valid event. +- **Copy the payload out before tail advance** — rejected because it adds a + second payload copy and evades the page-lifetime proof. +- **Treat ordinary producer drop as clean** — rejected because an unwind or + forgotten shutdown would be indistinguishable from a complete stream. +- **Add WAL or epochs in this ADR** — rejected because crash durability, page + rotation, authenticated lookup, reuse, and reclamation have independent + state machines and recovery proofs. + +## Verification + +Required evidence includes: + +- ring permit reserve/cancel/commit, saturation, disconnect, reuse, wrap, + closure race, and exactly-once unread destruction; +- ring claim/drop/reclaim, validation-before-tail, slot reuse only after commit, + wrap, and closure drain; +- invalid/page-full/descriptor-full/ring-full/disconnected admission with no + page/ring logical mutation; +- exact success, sequence parity, clean finish/drain/end, faulted drop, orphan + prefix detection, gap/mismatch/corruption termination, and retry of a dropped + frame lease; +- deterministic short-trace differential tests against a safe + `SlabPageBuilder + VecDeque` oracle; +- fault injection before page mutation, during the ADR-0008 poisoned suffix, + after `[P]`, after ring-slot initialization, and after `[R]`; +- Loom models using the shipping permit/claim algorithms for visibility, + cancellation, saturation/retry, close races, clean/faulted order, and wrap; +- full native Miri including held page borrows and claim/permit drop order; +- local and CI ASan/TSan, plus real UBSan when the Rust/toolchain boundary + supports it; Miri is not relabeled as UBSan; +- long tiny-ring native stress proving no loss, duplicate, reorder, payload + mismatch, or clean-end count mismatch; +- zero steady-state allocations for admission, validation lease, and commit; + full rejection must execute no payload copy. + +```bash +cargo test -p aegis-event +cargo test -p aegis-event --all-features +cargo test -p aegis-event --features loom loom_admission +cargo +nightly miri test -p aegis-event +cargo clippy -p aegis-event --all-targets --all-features -- -D warnings +cargo bench -p aegis-event --bench admission --no-run +``` + +The `current` prototype includes retained-producer caught-unwind tests at the +pre-`[P]`, post-`[P]`, post-ring-slot-write, and post-`[R]` boundaries; +gap/duplicate/reorder, identity, range, CRC, terminal-state, count-mismatch, +and post-`[C]` fail-closed fixtures; and a Loom validation-failure versus +reserved-publication race. These local sources do not replace formal ADR +acceptance, hosted sanitizer artifacts, or the remaining UBSan gate. + +The current Rust nightly does not expose `-Zsanitizer=undefined`. This remains +an unmet `CONTRIBUTING.md` acceptance gate; Miri is complementary provenance +and undefined-behavior evidence, not UBSan equivalence. + +## Migration and rollback + +While this ADR is Proposed, the implementation is `current` only as isolated, +unwired prototype code. It receives no production or `shadow` traffic, cannot +carry protected evidence, and has no durable state. Rollback removes the +composite module, permit/claim APIs, and ADR references while retaining the +independently tested ADR-0006 ring, ADR-0007 safe page, and ADR-0008 published +page. + +Future integration remains behind the per-tenant `event_write` generation and +the current SQL/Tokio path. Production wiring additionally requires green +unsafe/sanitizer review artifacts, authenticated page registry lookup, bounded +page rotation/outstanding pages, WAL replay and durability classes, generation +reuse with epochs, NUMA-owner reclamation, priority lanes, shadow equality, +qualification, and release-artifact rollback. + +## Revisit when + +Revisit before adding page rotation, a page registry, generation reuse, +crossbeam-epoch, a NUMA pool, a critical/WAL lane, production reactor wiring, +or protected evidence. Any change that permits multiple producers or consumers +requires a new algorithm and ADR; this SPSC permit/claim proof does not extend +to MPSC or MPMC. + +## References + +- [Mandatory architecture law](../architecture.md) +- [Target HLD: shared-memory SPSC event bus](../../ARCHITECTURE.md#5-shared-memory-spsc-event-bus) +- [Target LLD: Disruptor-style SPSC ring](../LLD.md#5-disruptor-style-spsc-ring) +- [ADR-0006: cache-padded SPSC event fabric](0006-cache-padded-spsc-event-fabric.md) +- [ADR-0008: append-only published-prefix slab pages](0008-append-only-published-prefix-slab-pages.md) +- [Migration matrix](../../MIGRATION_MATRIX.md#10-gap-analysis-against-target) +- [Contribution and unsafe-code standard](../../CONTRIBUTING.md#lock-free-structures) diff --git a/docs/adr/index.md b/docs/adr/index.md index 2757b8d7..2960e297 100644 --- a/docs/adr/index.md +++ b/docs/adr/index.md @@ -2,7 +2,7 @@ **Issue:** [#1197](https://github.com/lavkushry/AegisAgent/issues/1197) -> **Status:** ADR-0001 through ADR-0005 are Accepted. ADR-0006 through ADR-0008 +> **Status:** ADR-0001 through ADR-0005 are Accepted. ADR-0006 through ADR-0009 > are Proposed and permit only unwired prototypes until accepted. A changed > decision requires a new ADR and a supersedes link; do not silently rewrite > historical rationale. @@ -31,6 +31,7 @@ a new ADR and mark the old one "Superseded by ADR-NNNN." | [0006](0006-cache-padded-spsc-event-fabric.md) | Cache-padded SPSC descriptor fabric (Proposed) | | [0007](0007-sealed-generation-tagged-slab-pages.md) | Sealed generation-tagged slab-page ownership oracle (Proposed) | | [0008](0008-append-only-published-prefix-slab-pages.md) | Append-only published-prefix slab pages (Proposed) | +| [0009](0009-failure-atomic-slab-ring-admission.md) | Failure-atomic single-page slab/ring admission (Proposed) | ## Security and Review diff --git a/docs/current-vs-roadmap.md b/docs/current-vs-roadmap.md index ff571300..57a63a50 100644 --- a/docs/current-vs-roadmap.md +++ b/docs/current-vs-roadmap.md @@ -45,7 +45,7 @@ Strongest current capabilities: - **Deploy** — gateway + sensor + egress + cage + llm-gateway + tool-broker Helm; single-writer SQLite default - **Postgres** — feature + migrations exist; read/write pool split with replica failover, DB-backed replay store, cross-replica correlation windows, and a live `postgres-integration` CI job now exercise it against a real Postgres instance (not just compile-checked) — but not the default HA production path; real cross-instance failover/load validation still pending - **OIDC console login** — self-service login/link flow (`src/src/oidc.rs`, `routes/oidc.rs`), gated on `AEGIS_OIDC_*` + `AEGIS_JWT_SECRET`, fails closed for unrecognized identities (no auto-provisioning); single-IdP-per-gateway, no SAML, no per-SSO-user attribution/revocation -- **v2 event primitives** — cache-padded SPSC, safe sealed-page oracle, and append-only published-prefix code are `current` in `lib/event`; packed Release/Acquire publication exposes an immediate immutable prefix, with safe differential, native stress, same-algorithm Loom, full Miri, defined ASan/TSan CI lanes, and a zero-allocation append-plus-resolve test. The code is unwired, carries no production traffic or protected evidence, and makes no performance claim. The production Thread-Per-Core event fabric remains `target`, neither `shadow` nor `qualified`. +- **v2 event primitives** — cache-padded SPSC, safe sealed-page, append-only published-prefix, and failure-atomic single-page admission code are `current` in `lib/event`. The volatile composite uses cancelable permits, page-before-ring Release publication, commit-delayed claims, must-use frame leases, and explicit clean/faulted/orphan terminal checks. Test sources include safe differential oracles, native stress, shipping-algorithm Loom, Miri-oriented lifetime cases, defined ASan/TSan CI lanes, and zero-allocation admission/claim checks. The code is unwired, carries no production or `shadow` traffic or protected evidence, is not `qualified`, and makes no performance claim. The production Thread-Per-Core event fabric remains `target`. ### Roadmap / not done @@ -83,7 +83,7 @@ Strongest current capabilities: | Ban / quarantine centers | Partial | Stores/APIs; not every choke point + full UI. | | Postgres production mode | Roadmap / partial | Code path now CI-validated against a live Postgres instance (was compile-check only); SQLite single-writer is still the default deploy. | | Full Kubernetes multi-replica | Roadmap | Blocked on Postgres GA + broader Helm surface. | -| Thread-Per-Core event fabric | `current` prototypes / `target` fabric | Isolated SPSC, safe sealed-page, and append-only published-prefix prototypes exist. Packed Release/Acquire state publishes count, byte watermark, and closure for immediate immutable-prefix resolution; evidence includes a safe differential corpus, native stress, same-algorithm Loom, full Miri, defined ASan/TSan CI lanes, and a zero-allocation append-plus-resolve test. Production is neither `shadow` nor `qualified`, and no performance result exists. Blockers: green sanitizer CI artifacts, ADR acceptance/security review, composite ring reservation/admission, authenticated registry, bounded page rotation/outstanding pages, generation reuse/epochs, NUMA-owner reclamation, production shadow wiring, UBSan support, and qualification. | +| Thread-Per-Core event fabric | `current` prototypes / `target` fabric | Isolated SPSC, safe sealed-page, append-only published-prefix, and single-page volatile admission prototypes exist. The composite validates before reservation, publishes the page before the ring, withholds tail advancement until a validated frame lease commits, and makes clean/faulted/orphan termination explicit. Test sources include safe differential oracles, native stress, shipping-algorithm Loom, Miri-oriented lifetime cases, defined ASan/TSan CI lanes, and zero-allocation checks. Production carries no protected evidence, is neither `shadow` nor `qualified`, and has no performance result. Blockers: formal ADR acceptance/security review, green hosted sanitizer artifacts, UBSan support, authenticated registry, bounded page rotation/outstanding pages, WAL durability/replay, generation reuse/epochs, NUMA-owner reclamation, priority lanes, production shadow wiring, release-artifact rollback, and qualification. | --- diff --git a/lib/event/Cargo.toml b/lib/event/Cargo.toml index 3bd4a486..6ccbd0da 100644 --- a/lib/event/Cargo.toml +++ b/lib/event/Cargo.toml @@ -25,3 +25,7 @@ harness = false [[bench]] name = "published_slab" harness = false + +[[bench]] +name = "admission" +harness = false diff --git a/lib/event/benches/admission.rs b/lib/event/benches/admission.rs new file mode 100644 index 00000000..c0aa2384 --- /dev/null +++ b/lib/event/benches/admission.rs @@ -0,0 +1,65 @@ +use aegis_event::{SlabPageConfig, TryAdmitError, VolatileAdmissionChannel}; +use criterion::{black_box, criterion_group, criterion_main, BatchSize, Criterion}; + +const PAYLOAD: [u8; 256] = [0xa5; 256]; + +fn admit_claim_commit(c: &mut Criterion) { + c.bench_function("volatile_admission_256b_claim_commit_diagnostic", |b| { + b.iter_batched( + || { + VolatileAdmissionChannel::<2>::new(SlabPageConfig { + arena_id: 1, + arena_generation: 1, + byte_capacity: PAYLOAD.len(), + descriptor_capacity: 1, + first_sequence: 0, + }) + .expect("benchmark channel configuration is valid") + .split() + }, + |(mut producer, mut consumer)| { + let token = producer + .try_admit(black_box(&PAYLOAD), 1, 0) + .expect("one preallocated event fits"); + let frame = consumer + .try_next() + .expect("admitted benchmark frame validates"); + black_box(frame.payload()); + let committed = frame.commit(); + black_box((token, committed)); + }, + BatchSize::SmallInput, + ); + }); +} + +fn saturated_rejection(c: &mut Criterion) { + c.bench_function("volatile_admission_full_rejection_diagnostic", |b| { + b.iter_batched( + || { + let channel = VolatileAdmissionChannel::<1>::new(SlabPageConfig { + arena_id: 1, + arena_generation: 1, + byte_capacity: PAYLOAD.len() * 2, + descriptor_capacity: 2, + first_sequence: 0, + }) + .expect("benchmark channel configuration is valid"); + let (mut producer, consumer) = channel.split(); + producer + .try_admit(&PAYLOAD, 1, 0) + .expect("first event occupies the ring"); + (producer, consumer) + }, + |(mut producer, _consumer)| { + let result = producer.try_admit(black_box(&PAYLOAD), 1, 0); + assert_eq!(result, Err(TryAdmitError::RingFull)); + let _ = black_box(result); + }, + BatchSize::SmallInput, + ); + }); +} + +criterion_group!(benches, admit_claim_commit, saturated_rejection); +criterion_main!(benches); diff --git a/lib/event/src/admission.rs b/lib/event/src/admission.rs new file mode 100644 index 00000000..e5e6edcb --- /dev/null +++ b/lib/event/src/admission.rs @@ -0,0 +1,1340 @@ +//! Failure-atomic composition of one published slab page and one SPSC ring. +//! +//! This is a current, unwired, volatile prototype under Proposed ADR-0009. It +//! carries neither production telemetry nor protected evidence and provides no +//! durability, registry, page rotation, reuse, epoch, NUMA, or qualification +//! claim. + +use std::{cell::Cell, error::Error, fmt, marker::PhantomData}; + +#[cfg(feature = "loom")] +use loom::sync::{ + atomic::{AtomicU8, Ordering}, + Arc, +}; +#[cfg(not(feature = "loom"))] +use std::sync::{ + atomic::{AtomicU8, Ordering}, + Arc, +}; + +use crate::{ + ring::{validate_capacity, OccupiedSlot, RingInvariantError, TryClaimError, TryReserveError}, + slab::validate_config, + Consumer, Producer, PublishedPayload, PublishedSlabPage, PublishedSlabReadError, + PublishedSlabReader, PublishedSlabWriter, RingConfigError, SlabAppendError, SlabConfigError, + SlabPageConfig, SpscRing, TelemetryDescriptor, +}; + +const TERMINAL_OPEN: u8 = 0; +const TERMINAL_CLEAN: u8 = 1; +const TERMINAL_FAULTED: u8 = 2; + +#[repr(align(64))] +struct PaddedTerminalState(AtomicU8); + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum ProducerState { + Open, + InFlight, + Poisoned, + Finished, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum ConsumerState { + Open, + Faulted, + CleanEnd, +} + +#[cfg(all(test, not(feature = "loom")))] +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum AdmissionFaultPoint { + BeforePagePublication, + AfterPagePublication, + AfterRingSlotWrite, + AfterRingPublication, +} + +#[cfg(all(test, not(feature = "loom")))] +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum DescriptorFault { + SequenceGap, + SequenceDuplicate, + SequenceReorder, + ArenaIdentity, + PayloadRange, + PayloadCrc, +} + +/// Setup handle binding one fixed page to one fixed descriptor ring. +pub struct VolatileAdmissionChannel { + page: PublishedSlabPage, + ring: SpscRing, + terminal: Arc, +} + +impl VolatileAdmissionChannel { + pub fn new(config: SlabPageConfig) -> Result { + validate_capacity::().map_err(AdmissionConfigError::Ring)?; + validate_config(config).map_err(AdmissionConfigError::Slab)?; + + let page = PublishedSlabPage::new(config).map_err(AdmissionConfigError::Slab)?; + let ring = SpscRing::new_with_sequence(config.first_sequence) + .map_err(AdmissionConfigError::Ring)?; + + Ok(Self { + page, + ring, + terminal: Arc::new(PaddedTerminalState(AtomicU8::new(TERMINAL_OPEN))), + }) + } + + pub const fn ring_capacity(&self) -> usize { + N + } + + pub fn terminal_state_alignment(&self) -> usize { + std::mem::align_of::() + } + + pub fn split(self) -> (AdmissionProducer, AdmissionConsumer) { + let Self { + page, + ring, + terminal, + } = self; + let (slab, reader) = page.split(); + let (ring, consumer) = ring.split(); + let consumer_terminal = Arc::clone(&terminal); + let expected_sequence = slab.next_sequence(); + + ( + AdmissionProducer { + slab, + ring, + terminal, + state: ProducerState::Open, + #[cfg(all(test, not(feature = "loom")))] + fault_on_next_admission: None, + _not_sync: PhantomData, + }, + AdmissionConsumer { + ring: consumer, + slab: reader, + terminal: consumer_terminal, + expected_sequence, + committed_count: 0, + state: ConsumerState::Open, + _not_sync: PhantomData, + }, + ) + } +} + +impl fmt::Debug for VolatileAdmissionChannel { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("VolatileAdmissionChannel") + .field("ring_capacity", &N) + .field("arena_id", &self.page.arena_id()) + .field("arena_generation", &self.page.arena_generation()) + .field("byte_capacity", &self.page.byte_capacity()) + .field("descriptor_capacity", &self.page.descriptor_capacity()) + .finish() + } +} + +/// Sole admission owner for one fixed volatile page/ring pair. +/// +/// The endpoint may move to its owning thread but is intentionally not `Sync`: +/// +/// ```compile_fail +/// fn require_sync() {} +/// require_sync::>(); +/// ``` +pub struct AdmissionProducer { + // Ordered closure is explicit; declaration order preserves page-before-ring + // closure again during automatic field destruction. + slab: PublishedSlabWriter, + ring: Producer, + terminal: Arc, + state: ProducerState, + #[cfg(all(test, not(feature = "loom")))] + fault_on_next_admission: Option, + _not_sync: PhantomData>, +} + +impl AdmissionProducer { + /// Publishes one frame to this process-local volatile page/ring pair. + /// + /// Success is only a volatile publication token. It is not consumption, + /// WAL durability, a receipt, an authorization result, or permission to + /// discard protected evidence or an upstream replay/spool record. + pub fn try_admit( + &mut self, + payload: &[u8], + schema_id: u32, + flags: u16, + ) -> Result { + let attempt = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + self.try_admit_inner(payload, schema_id, flags) + })); + match attempt { + Ok(result) => result, + Err(panic) => { + // A caller may catch the resumed unwind and retain this value. + // Close every shared endpoint before control can escape so the + // consumer never polls an atomically open, poisoned lane. + self.close_faulted(); + std::panic::resume_unwind(panic) + } + } + } + + fn try_admit_inner( + &mut self, + payload: &[u8], + schema_id: u32, + flags: u16, + ) -> Result { + #[cfg(all(test, not(feature = "loom")))] + let fault = self.fault_on_next_admission.take(); + + if self.state != ProducerState::Open { + self.close_faulted(); + return Err(TryAdmitError::ProducerPoisoned); + } + + self.slab + .preflight_append(payload.len()) + .map_err(TryAdmitError::Slab)?; + + let slab_sequence = self.slab.next_sequence(); + let ring_sequence = self.ring.next_sequence(); + if slab_sequence != ring_sequence { + self.state = ProducerState::Poisoned; + self.close_faulted(); + return Err(TryAdmitError::SequenceDiverged { + slab: slab_sequence, + ring: ring_sequence, + }); + } + + let permit = match self.ring.try_reserve() { + Ok(permit) => permit, + Err(TryReserveError::Full) => return Err(TryAdmitError::RingFull), + Err(TryReserveError::Disconnected) => return Err(TryAdmitError::ConsumerDisconnected), + Err(TryReserveError::Invariant(error)) => { + self.state = ProducerState::Poisoned; + self.close_faulted(); + return Err(TryAdmitError::RingInvariant(error)); + } + }; + + if permit.sequence() != slab_sequence { + let ring = permit.sequence(); + permit.cancel(); + self.state = ProducerState::Poisoned; + self.close_faulted(); + return Err(TryAdmitError::SequenceDiverged { + slab: slab_sequence, + ring, + }); + } + + self.state = ProducerState::InFlight; + #[cfg(all(test, not(feature = "loom")))] + Self::panic_at(fault, AdmissionFaultPoint::BeforePagePublication); + let descriptor = match self.slab.try_append(payload, schema_id, flags) { + Ok(descriptor) => descriptor, + Err(error) => { + self.state = ProducerState::Open; + return Err(TryAdmitError::Slab(error)); + } + }; + + // ADR-0009's post-page-publication interval starts at the return from + // `try_append`. `publish` is deliberately infallible and performs no + // allocation, indexing, callback, formatting, or peer-state recheck. + #[cfg(all(test, not(feature = "loom")))] + Self::panic_at(fault, AdmissionFaultPoint::AfterPagePublication); + #[cfg(all(test, not(feature = "loom")))] + if fault == Some(AdmissionFaultPoint::AfterRingSlotWrite) { + permit.publish_then_panic_before_head(descriptor); + } + permit.publish(descriptor); + #[cfg(all(test, not(feature = "loom")))] + Self::panic_at(fault, AdmissionFaultPoint::AfterRingPublication); + self.state = ProducerState::Open; + Ok(AdmittedSequence::from_descriptor(descriptor)) + } + + #[cfg(all(test, not(feature = "loom")))] + fn panic_at(actual: Option, expected: AdmissionFaultPoint) { + if actual == Some(expected) { + panic!("injected admission fault at {expected:?}"); + } + } + + #[cfg(all(test, not(feature = "loom")))] + fn inject_panic_on_next_admission(&mut self, point: AdmissionFaultPoint) { + self.fault_on_next_admission = Some(point); + } + + #[cfg(all(test, not(feature = "loom")))] + fn inject_corrupt_descriptor( + &mut self, + payload: &[u8], + fault: DescriptorFault, + ) -> Result<(), TryAdmitError> { + self.slab + .preflight_append(payload.len()) + .map_err(TryAdmitError::Slab)?; + let permit = self.ring.try_reserve().map_err(TryAdmitError::from)?; + self.state = ProducerState::InFlight; + let mut descriptor = self + .slab + .try_append(payload, 1, 0) + .map_err(TryAdmitError::Slab)?; + + let overwrite_canonical = match fault { + DescriptorFault::SequenceGap => { + descriptor.sequence = descriptor.sequence.wrapping_add(2); + false + } + DescriptorFault::SequenceDuplicate => { + descriptor.sequence = descriptor.sequence.wrapping_sub(1); + false + } + DescriptorFault::SequenceReorder => { + descriptor.sequence = descriptor.sequence.wrapping_add(1); + false + } + DescriptorFault::ArenaIdentity => { + descriptor.arena_id ^= 1; + false + } + DescriptorFault::PayloadRange => { + descriptor.offset = u32::try_from(self.slab.byte_capacity()) + .expect("bounded test page is u32-addressable"); + descriptor.len = 1; + true + } + DescriptorFault::PayloadCrc => { + descriptor.crc32c ^= u32::MAX; + true + } + }; + if overwrite_canonical { + // SAFETY: this single-threaded fixture rewrites the canonical cell + // before publishing its ring descriptor; the consumer cannot yet + // observe or access the page entry. + unsafe { + self.slab.overwrite_last_descriptor_for_test(descriptor); + } + } + + permit.publish(descriptor); + self.state = ProducerState::Open; + Ok(()) + } + + pub fn finish(mut self) -> Result<(), FinishError> { + if self.state != ProducerState::Open { + self.close_faulted(); + return Err(FinishError::ProducerPoisoned); + } + + self.slab.close(); + self.terminal.0.store(TERMINAL_CLEAN, Ordering::Release); + self.ring.close(); + self.state = ProducerState::Finished; + Ok(()) + } + + fn close_faulted(&mut self) { + if self.state == ProducerState::Finished { + return; + } + self.slab.close(); + self.terminal.0.store(TERMINAL_FAULTED, Ordering::Release); + self.ring.close(); + self.state = ProducerState::Finished; + } + + pub fn arena_id(&self) -> u16 { + self.slab.arena_id() + } + + pub fn arena_generation(&self) -> u32 { + self.slab.arena_generation() + } + + pub fn used_bytes(&self) -> usize { + self.slab.used_bytes() + } + + pub fn published_count(&self) -> usize { + self.slab.published_count() + } + + pub fn remaining_bytes(&self) -> usize { + self.slab.remaining_bytes() + } + + pub fn remaining_descriptors(&self) -> usize { + self.slab.remaining_descriptors() + } + + pub fn next_sequence(&self) -> u64 { + self.slab.next_sequence() + } + + #[cfg(test)] + fn inject_orphan_after_page_publication( + &mut self, + payload: &[u8], + ) -> Result<(), TryAdmitError> { + self.slab + .preflight_append(payload.len()) + .map_err(TryAdmitError::Slab)?; + let permit = self.ring.try_reserve().map_err(TryAdmitError::from)?; + self.state = ProducerState::InFlight; + let _descriptor = self + .slab + .try_append(payload, 1, 0) + .map_err(TryAdmitError::Slab)?; + permit.cancel(); + Ok(()) + } +} + +impl Drop for AdmissionProducer { + fn drop(&mut self) { + self.close_faulted(); + } +} + +impl fmt::Debug for AdmissionProducer { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("AdmissionProducer") + .field("ring_capacity", &N) + .field("arena_id", &self.arena_id()) + .field("arena_generation", &self.arena_generation()) + .field("used_bytes", &self.used_bytes()) + .field("published_count", &self.published_count()) + .field("next_sequence", &self.next_sequence()) + .field("state", &self.state) + .finish() + } +} + +/// Sole validation/acknowledgement owner for one fixed volatile page/ring pair. +/// +/// The endpoint may move to its owning thread but is intentionally not `Sync`: +/// +/// ```compile_fail +/// fn require_sync() {} +/// require_sync::>(); +/// ``` +pub struct AdmissionConsumer { + // Closing the ring consumer before releasing the page reader prevents a new + // producer permit after the bound reader lifetime ends. + ring: Consumer, + slab: PublishedSlabReader, + terminal: Arc, + expected_sequence: u64, + committed_count: usize, + state: ConsumerState, + _not_sync: PhantomData>, +} + +impl AdmissionConsumer { + pub fn try_next(&mut self) -> Result, TryConsumeError> { + let Self { + ring, + slab, + terminal, + expected_sequence, + committed_count, + state, + _not_sync: _, + } = self; + + match *state { + ConsumerState::CleanEnd => return Err(TryConsumeError::CleanEnd), + ConsumerState::Faulted => return Err(TryConsumeError::ConsumerPoisoned), + ConsumerState::Open => {} + } + + let claim = match ring.try_claim() { + Ok(claim) => claim, + Err(TryClaimError::Empty) => return Err(TryConsumeError::Empty), + Err(TryClaimError::Disconnected) => { + return Err(Self::classify_end(slab, terminal, *committed_count, state)); + } + Err(TryClaimError::Invariant(error)) => { + // `try_claim` Release-closes the ring consumer before exposing + // an invariant failure, so the producer cannot publish more. + *state = ConsumerState::Faulted; + return Err(TryConsumeError::RingInvariant(error)); + } + }; + + let descriptor = claim.value(); + if descriptor.sequence != *expected_sequence { + claim.close_consumer(); + let expected = *expected_sequence; + *state = ConsumerState::Faulted; + return Err(TryConsumeError::SequenceMismatch { + expected, + actual: descriptor.sequence, + }); + } + + let payload = match slab.resolve(&descriptor) { + Ok(payload) => payload, + Err(error) => { + claim.close_consumer(); + *state = ConsumerState::Faulted; + return Err(TryConsumeError::Page(error)); + } + }; + + Ok(AdmittedEvent { + descriptor, + payload, + claim, + expected_sequence, + committed_count, + }) + } + + fn classify_end( + slab: &PublishedSlabReader, + terminal: &Arc, + committed_count: usize, + state: &mut ConsumerState, + ) -> TryConsumeError { + let status = match slab.status() { + Ok(status) => status, + Err(error) => { + *state = ConsumerState::Faulted; + return TryConsumeError::Page(error); + } + }; + let terminal = terminal.0.load(Ordering::Acquire); + + if !status.writer_closed { + *state = ConsumerState::Faulted; + return TryConsumeError::ClosureOrderViolation; + } + + if status.published_count != committed_count { + let error = if terminal == TERMINAL_FAULTED && status.published_count > committed_count + { + TryConsumeError::OrphanedPublishedPrefix { + published: status.published_count, + committed: committed_count, + } + } else { + TryConsumeError::CountMismatch { + published: status.published_count, + committed: committed_count, + } + }; + *state = ConsumerState::Faulted; + return error; + } + + match terminal { + TERMINAL_CLEAN => { + *state = ConsumerState::CleanEnd; + TryConsumeError::CleanEnd + } + TERMINAL_FAULTED => { + *state = ConsumerState::Faulted; + TryConsumeError::ProducerFaulted { + published: status.published_count, + committed: committed_count, + } + } + TERMINAL_OPEN => { + *state = ConsumerState::Faulted; + TryConsumeError::ClosureOrderViolation + } + raw => { + *state = ConsumerState::Faulted; + TryConsumeError::TerminalStateInvalid { raw } + } + } + } + + pub fn arena_id(&self) -> u16 { + self.slab.arena_id() + } + + pub fn arena_generation(&self) -> u32 { + self.slab.arena_generation() + } + + pub fn committed_count(&self) -> usize { + self.committed_count + } + + pub fn expected_sequence(&self) -> u64 { + self.expected_sequence + } +} + +impl fmt::Debug for AdmissionConsumer { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("AdmissionConsumer") + .field("ring_capacity", &N) + .field("arena_id", &self.arena_id()) + .field("arena_generation", &self.arena_generation()) + .field("expected_sequence", &self.expected_sequence) + .field("committed_count", &self.committed_count) + .field("state", &self.state) + .finish() + } +} + +/// Must-use validated frame whose ring capacity remains claimed until commit. +/// +/// A frame lease is intentionally neither `Send` nor `Sync` and cannot cross +/// an `.await`, callback, or ownership handoff: +/// +/// ```compile_fail +/// fn require_send() {} +/// require_send::>(); +/// ``` +/// +/// ```compile_fail +/// fn require_sync() {} +/// require_sync::>(); +/// ``` +#[must_use = "commit the frame only after synchronous processing succeeds"] +pub struct AdmittedEvent<'consumer, const N: usize> { + descriptor: TelemetryDescriptor, + payload: PublishedPayload<'consumer>, + claim: OccupiedSlot<'consumer, TelemetryDescriptor, N>, + expected_sequence: &'consumer mut u64, + committed_count: &'consumer mut usize, +} + +impl AdmittedEvent<'_, N> { + pub const fn descriptor(&self) -> &TelemetryDescriptor { + &self.descriptor + } + + pub fn payload(&self) -> &[u8] { + self.payload.as_ref() + } + + /// Reclaims this volatile ring slot after synchronous validation/handling. + /// + /// The returned value is not a receipt, durability acknowledgement, or + /// authorization result and must never acknowledge protected evidence. + pub fn commit(self) -> AdmittedSequence { + let Self { + descriptor: _, + payload: _, + claim, + expected_sequence, + committed_count, + } = self; + let descriptor = claim.commit(); + *expected_sequence = descriptor.sequence.wrapping_add(1); + *committed_count += 1; + AdmittedSequence::from_descriptor(descriptor) + } + + #[cfg(all(test, not(feature = "loom")))] + fn commit_then_panic_before_counters(self) -> ! { + let Self { + descriptor: _, + payload: _, + claim, + expected_sequence: _, + committed_count: _, + } = self; + let _ = claim.commit(); + panic!("injected frame fault after tail commit and before local counters"); + } +} + +impl fmt::Debug for AdmittedEvent<'_, N> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("AdmittedEvent") + .field("descriptor", &self.descriptor) + .field("payload_len", &self.payload.len()) + .finish_non_exhaustive() + } +} + +/// Identity of one frame published to the volatile admission pair. +/// +/// This token proves neither consumption nor durability and must not be used +/// as a receipt, protected-evidence acknowledgement, or authorization result. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct AdmittedSequence { + arena_id: u16, + arena_generation: u32, + sequence: u64, +} + +impl AdmittedSequence { + const fn from_descriptor(descriptor: TelemetryDescriptor) -> Self { + Self { + arena_id: descriptor.arena_id, + arena_generation: descriptor.arena_generation, + sequence: descriptor.sequence, + } + } + + pub const fn arena_id(self) -> u16 { + self.arena_id + } + + pub const fn arena_generation(self) -> u32 { + self.arena_generation + } + + pub const fn sequence(self) -> u64 { + self.sequence + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum AdmissionConfigError { + Ring(RingConfigError), + Slab(SlabConfigError), +} + +impl fmt::Display for AdmissionConfigError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Ring(error) => write!(f, "invalid admission ring configuration: {error}"), + Self::Slab(error) => write!(f, "invalid admission page configuration: {error}"), + } + } +} + +impl Error for AdmissionConfigError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::Ring(error) => Some(error), + Self::Slab(error) => Some(error), + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TryAdmitError { + RingFull, + ConsumerDisconnected, + Slab(SlabAppendError), + SequenceDiverged { slab: u64, ring: u64 }, + RingInvariant(RingInvariantError), + ProducerPoisoned, +} + +impl From for TryAdmitError { + fn from(error: TryReserveError) -> Self { + match error { + TryReserveError::Full => Self::RingFull, + TryReserveError::Disconnected => Self::ConsumerDisconnected, + TryReserveError::Invariant(error) => Self::RingInvariant(error), + } + } +} + +impl fmt::Display for TryAdmitError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::RingFull => write!(f, "volatile admission ring is full"), + Self::ConsumerDisconnected => write!(f, "volatile admission consumer is disconnected"), + Self::Slab(error) => write!(f, "volatile admission page rejected the frame: {error}"), + Self::SequenceDiverged { slab, ring } => write!( + f, + "volatile admission page sequence {slab} diverged from ring sequence {ring}" + ), + Self::RingInvariant(error) => { + write!(f, "volatile admission ring invariant failed: {error}") + } + Self::ProducerPoisoned => write!(f, "volatile admission producer is poisoned"), + } + } +} + +impl Error for TryAdmitError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::Slab(error) => Some(error), + Self::RingInvariant(error) => Some(error), + _ => None, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum FinishError { + ProducerPoisoned, +} + +impl fmt::Display for FinishError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::ProducerPoisoned => { + write!(f, "poisoned admission producer cannot finish cleanly") + } + } + } +} + +impl Error for FinishError {} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TryConsumeError { + Empty, + CleanEnd, + ConsumerPoisoned, + ProducerFaulted { published: usize, committed: usize }, + OrphanedPublishedPrefix { published: usize, committed: usize }, + CountMismatch { published: usize, committed: usize }, + SequenceMismatch { expected: u64, actual: u64 }, + ClosureOrderViolation, + TerminalStateInvalid { raw: u8 }, + RingInvariant(RingInvariantError), + Page(PublishedSlabReadError), +} + +impl fmt::Display for TryConsumeError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Empty => write!(f, "volatile admission ring is empty"), + Self::CleanEnd => write!(f, "volatile admission stream ended cleanly"), + Self::ConsumerPoisoned => write!(f, "volatile admission consumer is poisoned"), + Self::ProducerFaulted { + published, + committed, + } => write!( + f, + "volatile admission producer faulted after publishing {published} and committing {committed} events" + ), + Self::OrphanedPublishedPrefix { + published, + committed, + } => write!( + f, + "volatile admission page has orphaned prefix: published {published}, committed {committed}" + ), + Self::CountMismatch { + published, + committed, + } => write!( + f, + "volatile admission count mismatch: published {published}, committed {committed}" + ), + Self::SequenceMismatch { expected, actual } => write!( + f, + "volatile admission sequence mismatch: expected {expected}, received {actual}" + ), + Self::ClosureOrderViolation => write!( + f, + "volatile admission ring closed before page and terminal state" + ), + Self::TerminalStateInvalid { raw } => { + write!(f, "volatile admission terminal state {raw} is invalid") + } + Self::RingInvariant(error) => { + write!(f, "volatile admission ring invariant failed: {error}") + } + Self::Page(error) => write!(f, "volatile admission page validation failed: {error}"), + } + } +} + +impl Error for TryConsumeError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::RingInvariant(error) => Some(error), + Self::Page(error) => Some(error), + _ => None, + } + } +} + +#[cfg(all(test, not(feature = "loom")))] +mod native_tests { + use super::*; + use crate::CACHE_LINE_BYTES; + use std::panic::{catch_unwind, AssertUnwindSafe}; + + fn config() -> SlabPageConfig { + SlabPageConfig { + arena_id: 1, + arena_generation: 2, + byte_capacity: 8, + descriptor_capacity: 2, + first_sequence: 3, + } + } + + #[test] + fn terminal_state_occupies_a_distinct_cache_aligned_unit() { + fn require_send() {} + + require_send::>(); + require_send::>(); + let channel = VolatileAdmissionChannel::<2>::new(config()).expect("bounded channel"); + assert_eq!(channel.terminal_state_alignment(), CACHE_LINE_BYTES); + assert!(std::mem::size_of::() >= CACHE_LINE_BYTES); + } + + #[test] + fn injected_post_page_pre_ring_interruption_reports_an_orphan() { + let channel = VolatileAdmissionChannel::<1>::new(config()).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + producer + .inject_orphan_after_page_publication(b"x") + .expect("fault injection publishes only the page"); + drop(producer); + + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::OrphanedPublishedPrefix { + published: 1, + committed: 0, + }) + )); + } + + #[test] + fn caught_unwind_closes_a_retained_producer_at_every_publication_phase() { + for point in [ + AdmissionFaultPoint::BeforePagePublication, + AdmissionFaultPoint::AfterPagePublication, + AdmissionFaultPoint::AfterRingSlotWrite, + AdmissionFaultPoint::AfterRingPublication, + ] { + let channel = VolatileAdmissionChannel::<1>::new(config()).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + producer.inject_panic_on_next_admission(point); + + let unwind = catch_unwind(AssertUnwindSafe(|| { + let _ = producer.try_admit(b"x", 1, 0); + })); + assert!(unwind.is_err(), "fault point {point:?} must unwind"); + + let expected_published = + usize::from(point != AdmissionFaultPoint::BeforePagePublication); + assert_eq!(producer.published_count(), expected_published); + assert_eq!( + producer.try_admit(b"retained", 1, 0), + Err(TryAdmitError::ProducerPoisoned), + "retained producer must remain terminal at {point:?}" + ); + + match point { + AdmissionFaultPoint::BeforePagePublication => assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::ProducerFaulted { + published: 0, + committed: 0, + }) + )), + AdmissionFaultPoint::AfterPagePublication + | AdmissionFaultPoint::AfterRingSlotWrite => assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::OrphanedPublishedPrefix { + published: 1, + committed: 0, + }) + )), + AdmissionFaultPoint::AfterRingPublication => { + let frame = consumer + .try_next() + .expect("published frame remains drainable"); + assert_eq!(frame.payload(), b"x"); + frame.commit(); + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::ProducerFaulted { + published: 1, + committed: 1, + }) + )); + } + } + } + } + + #[test] + fn claim_invariant_closes_consumer_before_returning_the_error() { + let channel = VolatileAdmissionChannel::<2>::new(config()).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + consumer + .ring + .inject_cached_head_for_test(consumer.expected_sequence.wrapping_add(3)); + + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::RingInvariant( + RingInvariantError::CursorDistanceExceedsCapacity { capacity: 2, .. } + )) + )); + assert_eq!( + producer.try_admit(b"blocked", 1, 0), + Err(TryAdmitError::ConsumerDisconnected) + ); + assert_eq!(producer.published_count(), 0); + assert_eq!(producer.used_bytes(), 0); + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::ConsumerPoisoned) + )); + } + + #[test] + fn corrupt_descriptors_withhold_tail_and_disconnect_the_producer() { + for fault in [ + DescriptorFault::SequenceGap, + DescriptorFault::SequenceDuplicate, + DescriptorFault::SequenceReorder, + DescriptorFault::ArenaIdentity, + DescriptorFault::PayloadRange, + DescriptorFault::PayloadCrc, + ] { + let channel = VolatileAdmissionChannel::<2>::new(config()).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + producer + .inject_corrupt_descriptor(b"x", fault) + .expect("fault fixture publishes one corrupt descriptor"); + + let error = consumer + .try_next() + .expect_err("corrupt descriptor must fail closed"); + match fault { + DescriptorFault::SequenceGap => assert_eq!( + error, + TryConsumeError::SequenceMismatch { + expected: 3, + actual: 5, + } + ), + DescriptorFault::SequenceDuplicate => assert_eq!( + error, + TryConsumeError::SequenceMismatch { + expected: 3, + actual: 2, + } + ), + DescriptorFault::SequenceReorder => assert_eq!( + error, + TryConsumeError::SequenceMismatch { + expected: 3, + actual: 4, + } + ), + DescriptorFault::ArenaIdentity => assert_eq!( + error, + TryConsumeError::Page(PublishedSlabReadError::ArenaIdMismatch { + page: 1, + descriptor: 0, + }) + ), + DescriptorFault::PayloadRange => assert_eq!( + error, + TryConsumeError::Page(PublishedSlabReadError::Descriptor( + crate::DescriptorError::OutOfBounds { + end: 9, + page_len: 1, + } + )) + ), + DescriptorFault::PayloadCrc => assert_eq!( + error, + TryConsumeError::Page(PublishedSlabReadError::CrcMismatch { sequence: 3 }) + ), + } + + assert_eq!(consumer.committed_count(), 0); + assert_eq!(producer.published_count(), 1); + assert_eq!(producer.used_bytes(), 1); + assert_eq!( + producer.try_admit(b"y", 1, 0), + Err(TryAdmitError::ConsumerDisconnected), + "fault {fault:?} must close before the producer can grow the page" + ); + assert_eq!(producer.published_count(), 1); + assert_eq!(producer.used_bytes(), 1); + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::ConsumerPoisoned) + )); + } + } + + #[test] + fn malformed_terminal_closure_states_never_report_clean_end() { + { + let channel = VolatileAdmissionChannel::<1>::new(config()).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + producer.terminal.0.store(TERMINAL_CLEAN, Ordering::Release); + producer.ring.close(); + assert_eq!( + consumer.try_next().expect_err("page is still open"), + TryConsumeError::ClosureOrderViolation + ); + } + + { + let channel = VolatileAdmissionChannel::<1>::new(config()).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + producer.slab.close(); + producer.terminal.0.store(0xff, Ordering::Release); + producer.ring.close(); + producer.state = ProducerState::Finished; + assert_eq!( + consumer + .try_next() + .expect_err("unknown terminal state is corrupt"), + TryConsumeError::TerminalStateInvalid { raw: 0xff } + ); + } + + { + let channel = VolatileAdmissionChannel::<1>::new(config()).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + producer + .inject_orphan_after_page_publication(b"x") + .expect("page-only fixture"); + producer.slab.close(); + producer.terminal.0.store(TERMINAL_CLEAN, Ordering::Release); + producer.ring.close(); + producer.state = ProducerState::Finished; + assert_eq!( + consumer + .try_next() + .expect_err("clean state cannot hide a page-ahead count"), + TryConsumeError::CountMismatch { + published: 1, + committed: 0, + } + ); + } + + { + let channel = VolatileAdmissionChannel::<1>::new(config()).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + consumer.committed_count = 1; + producer.slab.close(); + producer.terminal.0.store(TERMINAL_CLEAN, Ordering::Release); + producer.ring.close(); + producer.state = ProducerState::Finished; + assert_eq!( + consumer + .try_next() + .expect_err("consumer-ahead count is corrupt"), + TryConsumeError::CountMismatch { + published: 0, + committed: 1, + } + ); + } + } + + #[test] + fn interruption_after_tail_commit_poison_closes_on_the_next_descriptor() { + let channel = VolatileAdmissionChannel::<2>::new(config()).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + producer.try_admit(b"a", 1, 0).expect("first event"); + producer.try_admit(b"b", 1, 0).expect("second event"); + producer.finish().expect("clean producer close"); + + let frame = consumer.try_next().expect("first frame validates"); + let unwind = catch_unwind(AssertUnwindSafe(|| { + frame.commit_then_panic_before_counters(); + })); + assert!(unwind.is_err()); + assert_eq!(consumer.committed_count(), 0); + assert_eq!(consumer.expected_sequence(), 3); + assert_eq!( + consumer + .try_next() + .expect_err("stale local sequence must fail closed"), + TryConsumeError::SequenceMismatch { + expected: 3, + actual: 4, + } + ); + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::ConsumerPoisoned) + )); + } +} + +#[cfg(all(test, feature = "loom"))] +mod loom_tests { + use super::*; + use loom::thread; + + fn config() -> SlabPageConfig { + SlabPageConfig { + arena_id: 1, + arena_generation: 2, + byte_capacity: 2, + descriptor_capacity: 2, + first_sequence: u64::MAX, + } + } + + #[test] + fn loom_admission_page_publication_precedes_ring_visibility_and_clean_close() { + loom::model(|| { + let channel = VolatileAdmissionChannel::<1>::new(config()).expect("bounded model"); + let (mut producer, mut consumer) = channel.split(); + + let producer_thread = thread::spawn(move || { + producer.try_admit(b"x", 1, 0).expect("model admit"); + producer.finish().expect("clean model finish"); + }); + let consumer_thread = thread::spawn(move || { + let consumed = match consumer.try_next() { + Ok(frame) => { + assert_eq!(frame.descriptor().sequence, u64::MAX); + assert_eq!(frame.payload(), b"x"); + frame.commit(); + true + } + Err(TryConsumeError::Empty) => false, + Err(error) => panic!("unexpected model error: {error}"), + }; + (consumer, consumed) + }); + + producer_thread.join().expect("producer succeeds"); + let (mut consumer, consumed) = consumer_thread.join().expect("consumer succeeds"); + if !consumed { + let frame = consumer.try_next().expect("final publication drains"); + assert_eq!(frame.payload(), b"x"); + frame.commit(); + } + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::CleanEnd) + )); + }); + } + + #[test] + fn loom_admission_full_retry_never_consumes_page_capacity() { + loom::model(|| { + let channel = VolatileAdmissionChannel::<1>::new(config()).expect("bounded model"); + let (mut producer, mut consumer) = channel.split(); + producer.try_admit(b"a", 1, 0).expect("first admit"); + + let producer_thread = thread::spawn(move || { + let outcome = match producer.try_admit(b"b", 1, 0) { + Ok(token) => { + assert_eq!(token.sequence(), 0); + 2 + } + Err(TryAdmitError::RingFull) => { + assert_eq!(producer.published_count(), 1); + assert_eq!(producer.used_bytes(), 1); + 1 + } + Err(TryAdmitError::ConsumerDisconnected) => { + assert_eq!(producer.published_count(), 1); + assert_eq!(producer.used_bytes(), 1); + 1 + } + Err(error) => panic!("unexpected producer error: {error}"), + }; + assert_eq!(producer.published_count(), outcome); + drop(producer); + }); + let consumer_thread = thread::spawn(move || { + let frame = consumer.try_next().expect("first event was prepublished"); + assert_eq!(frame.payload(), b"a"); + frame.commit(); + }); + + producer_thread.join().expect("producer succeeds"); + consumer_thread.join().expect("consumer succeeds"); + }); + } + + #[test] + fn loom_admission_validation_failure_races_reserved_publication() { + loom::model(|| { + let channel = VolatileAdmissionChannel::<2>::new(config()).expect("bounded model"); + let (mut producer, mut consumer) = channel.split(); + producer.try_admit(b"a", 1, 0).expect("first model admit"); + + consumer.expected_sequence = consumer.expected_sequence.wrapping_add(1); + let producer_thread = thread::spawn(move || { + let outcome = match producer.try_admit(b"b", 1, 0) { + Ok(token) => { + assert_eq!(token.sequence(), 0); + assert_eq!(producer.published_count(), 2); + assert_eq!(producer.used_bytes(), 2); + 2 + } + Err(TryAdmitError::ConsumerDisconnected) => { + assert_eq!(producer.published_count(), 1); + assert_eq!(producer.used_bytes(), 1); + 1 + } + Err(error) => panic!("unexpected raced producer error: {error}"), + }; + (producer, outcome) + }); + let consumer_thread = thread::spawn(move || { + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::SequenceMismatch { + expected: 0, + actual: u64::MAX, + }) + )); + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::ConsumerPoisoned) + )); + consumer + }); + + let (producer, published) = producer_thread.join().expect("producer succeeds"); + let consumer = consumer_thread.join().expect("consumer succeeds"); + assert_eq!(producer.published_count(), published); + drop(producer); + drop(consumer); + }); + } + + #[test] + fn loom_admission_faulted_orphan_is_never_reported_as_clean_end() { + loom::model(|| { + let channel = VolatileAdmissionChannel::<1>::new(config()).expect("bounded model"); + let (mut producer, mut consumer) = channel.split(); + producer + .inject_orphan_after_page_publication(b"x") + .expect("page-only fault injection"); + drop(producer); + + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::OrphanedPublishedPrefix { + published: 1, + committed: 0, + }) + )); + }); + } +} diff --git a/lib/event/src/lib.rs b/lib/event/src/lib.rs index f025f460..e3543251 100644 --- a/lib/event/src/lib.rs +++ b/lib/event/src/lib.rs @@ -1,8 +1,8 @@ //! Bounded event-transfer primitives for the target AegisAgent data plane. //! //! This crate contains unwired prototypes governed by Proposed ADR-0006, -//! ADR-0007, and ADR-0008. It does not carry production telemetry or protected -//! evidence. +//! ADR-0007, ADR-0008, and ADR-0009. It does not carry production telemetry or +//! protected evidence. //! //! The optional `loom` feature is model-checking infrastructure, not a runtime //! configuration. Loom-backed primitives must execute only inside @@ -10,19 +10,24 @@ #![forbid(unsafe_op_in_unsafe_fn)] +mod admission; mod descriptor; mod published_slab; mod ring; mod slab; +pub use admission::{ + AdmissionConfigError, AdmissionConsumer, AdmissionProducer, AdmittedEvent, AdmittedSequence, + FinishError, TryAdmitError, TryConsumeError, VolatileAdmissionChannel, +}; pub use descriptor::{DescriptorError, TelemetryDescriptor, MAX_FRAME_BYTES}; pub use published_slab::{ PublishedPayload, PublishedSlabLayout, PublishedSlabPage, PublishedSlabReadError, - PublishedSlabReader, PublishedSlabWriter, + PublishedSlabReader, PublishedSlabStatus, PublishedSlabWriter, }; pub use ring::{ - Consumer, Producer, RingConfigError, RingLayout, SpscRing, TryPopError, TryPushError, - CACHE_LINE_BYTES, + Consumer, OccupiedSlot, Producer, RingConfigError, RingInvariantError, RingLayout, SpscRing, + TryClaimError, TryPopError, TryPushError, TryReserveError, VacantSlot, CACHE_LINE_BYTES, }; pub use slab::{ SealedSlabPage, SlabAppendError, SlabConfigError, SlabPageBuilder, SlabPageConfig, diff --git a/lib/event/src/published_slab.rs b/lib/event/src/published_slab.rs index 6dfebc5d..f473868b 100644 --- a/lib/event/src/published_slab.rs +++ b/lib/event/src/published_slab.rs @@ -400,23 +400,23 @@ pub struct PublishedSlabWriter { _not_sync: PhantomData>, } +#[derive(Clone, Copy)] +struct AppendPlan { + descriptor_index: usize, + offset: usize, + end: usize, + descriptor_offset: u32, + descriptor_len: u32, + published_bytes: u32, + next_count: u32, + next_sequence: u64, +} + impl PublishedSlabWriter { - /// Copies, initializes, and Release-publishes one complete payload. - /// - /// FlatBuffer verification and redaction are upstream preconditions. Every - /// returned error is transactional; no descriptor escapes before the state - /// publication store. - pub fn try_append( - &mut self, - payload: &[u8], - schema_id: u32, - flags: u16, - ) -> Result { + fn append_plan(&self, len: usize) -> Result { if self.poisoned { return Err(SlabAppendError::WriterPoisoned); } - - let len = payload.len(); if len == 0 { return Err(SlabAppendError::EmptyPayload); } @@ -459,32 +459,86 @@ impl PublishedSlabWriter { u32::try_from(len).map_err(|_| SlabAppendError::LengthNotAddressable { len })?; let published_bytes = u32::try_from(end) .map_err(|_| SlabAppendError::OffsetNotAddressable { offset: end })?; - let next_count = self.descriptor_count + 1; - let next_sequence = self.next_sequence.wrapping_add(1); + + Ok(AppendPlan { + descriptor_index, + offset, + end, + descriptor_offset, + descriptor_len, + published_bytes, + next_count: self.descriptor_count + 1, + next_sequence: self.next_sequence.wrapping_add(1), + }) + } + + /// Checks every payload-length and current-capacity condition without CRC, + /// allocation, page mutation, or publication. + pub(crate) fn preflight_append(&self, len: usize) -> Result<(), SlabAppendError> { + self.append_plan(len).map(|_| ()) + } + + /// Replaces the last canonical descriptor for corrupt-input tests. + /// + /// # Safety + /// + /// No reader may access the page before this test-only rewrite completes. + /// The method deliberately violates published-prefix immutability and must + /// never be used outside a single-threaded corruption fixture. + #[cfg(all(test, not(feature = "loom")))] + pub(crate) unsafe fn overwrite_last_descriptor_for_test( + &mut self, + descriptor: TelemetryDescriptor, + ) { + let index = + self.descriptor_count + .checked_sub(1) + .expect("test corruption requires one published descriptor") as usize; + let cell = self + .inner + .descriptors + .get(index) + .expect("published test descriptor index remains in bounds"); + cell.write(descriptor); + } + + /// Copies, initializes, and Release-publishes one complete payload. + /// + /// FlatBuffer verification and redaction are upstream preconditions. Every + /// returned error is transactional; no descriptor escapes before the state + /// publication store. + pub fn try_append( + &mut self, + payload: &[u8], + schema_id: u32, + flags: u16, + ) -> Result { + let len = payload.len(); + let plan = self.append_plan(len)?; let checksum = crc32c::crc32c(payload); let descriptor = TelemetryDescriptor { sequence: self.next_sequence, arena_generation: self.inner.arena_generation, arena_id: self.inner.arena_id, flags, - offset: descriptor_offset, - len: descriptor_len, + offset: plan.descriptor_offset, + len: plan.descriptor_len, crc32c: checksum, schema_id, }; - let next_state = encode_publication_state(next_count, published_bytes, false); + let next_state = encode_publication_state(plan.next_count, plan.published_bytes, false); // Stage every bounds-checked reference and native pointer before the // first cell is touched. No indexing or fallible work follows. - let payload_cells = self - .inner - .bytes - .get(offset..end) - .ok_or(SlabAppendError::PageFull { - requested: len, - remaining, - })?; - let descriptor_cell = self.inner.descriptors.get(descriptor_index).ok_or( + let payload_cells = + self.inner + .bytes + .get(plan.offset..plan.end) + .ok_or(SlabAppendError::PageFull { + requested: len, + remaining: self.inner.byte_capacity - plan.offset, + })?; + let descriptor_cell = self.inner.descriptors.get(plan.descriptor_index).ok_or( SlabAppendError::DescriptorCapacityExhausted { capacity: self.inner.descriptor_capacity, }, @@ -516,9 +570,9 @@ impl PublishedSlabWriter { .publication .0 .store(next_state, Ordering::Release); - self.used_bytes = published_bytes; - self.descriptor_count = next_count; - self.next_sequence = next_sequence; + self.used_bytes = plan.published_bytes; + self.descriptor_count = plan.next_count; + self.next_sequence = plan.next_sequence; self.poisoned = false; Ok(descriptor) @@ -563,6 +617,13 @@ impl PublishedSlabWriter { pub fn is_poisoned(&self) -> bool { self.poisoned } + + pub(crate) fn close(&mut self) { + self.inner + .publication + .0 + .fetch_or(WRITER_CLOSED_BIT, Ordering::Release); + } } impl Drop for PublishedSlabWriter { @@ -570,10 +631,7 @@ impl Drop for PublishedSlabWriter { // The atomic word, rather than writer-private staged cursors, is the // authoritative committed prefix. `fetch_or` cannot regress or expose a // partially written suffix if unwinding interrupts an append. - self.inner - .publication - .0 - .fetch_or(WRITER_CLOSED_BIT, Ordering::Release); + self.close(); } } @@ -699,6 +757,16 @@ impl PublishedSlabReader { pub fn is_writer_closed(&self) -> bool { PublicationSnapshot::decode(self.inner.publication.0.load(Ordering::Acquire)).writer_closed } + + /// Returns one validated, coherent publication snapshot. + pub fn status(&self) -> Result { + let snapshot = self.inner.load_snapshot()?; + Ok(PublishedSlabStatus { + published_count: snapshot.published_count as usize, + published_bytes: snapshot.published_bytes as usize, + writer_closed: snapshot.writer_closed, + }) + } } impl fmt::Debug for PublishedSlabReader { @@ -775,6 +843,13 @@ pub struct PublishedSlabLayout { pub descriptor_cell_alignment: usize, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct PublishedSlabStatus { + pub published_count: usize, + pub published_bytes: usize, + pub writer_closed: bool, +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum PublishedSlabReadError { ArenaIdMismatch { diff --git a/lib/event/src/ring.rs b/lib/event/src/ring.rs index f4b49045..a85061d8 100644 --- a/lib/event/src/ring.rs +++ b/lib/event/src/ring.rs @@ -7,15 +7,19 @@ //! `!Sync`, while `T: Send` permits ownership transfer to another thread. //! - The producer is the only writer to a free slot. Its `Release` publication //! follows initialization, and the consumer reads only after an `Acquire`. -//! - The consumer moves each value once. Its `Release` consumption follows the -//! move, and the producer reuses a slot only after an `Acquire`. +//! - A `Copy` consumer may inspect a claimed slot without advancing the tail. +//! The consumer moves each value once only when the claim commits. Its +//! `Release` consumption follows the move, and the producer reuses a slot +//! only after an `Acquire`. +//! - Vacant and occupied slot capabilities exclusively borrow their endpoint +//! and are `!Send` and `!Sync`. Dropping either capability is a no-op. //! - Published minus consumed distance never exceeds capacity. Capacity is a //! power of two below `2^63`, making modular sequence distance unambiguous. //! - The final `Inner` drop runs only after both endpoints are gone and drops //! exactly the still-published range. An invalid distance panics in debug and //! leaks in release rather than dereferencing an unproven slot. -use std::{cell::Cell, error::Error, fmt, marker::PhantomData, mem::MaybeUninit}; +use std::{cell::Cell, error::Error, fmt, marker::PhantomData, mem::MaybeUninit, rc::Rc}; #[cfg(feature = "loom")] use loom::{ @@ -86,6 +90,30 @@ impl Slot { }) } + #[cfg(not(feature = "loom"))] + fn copy_value(&self) -> T + where + T: Copy, + { + // SAFETY: The consumer observed publication with Acquire and holds the + // only consumer endpoint mutably. Copying through a shared reference + // does not move or invalidate the initialized slot, so cancellation + // can leave it available for a later claim. + unsafe { *(*self.0.get()).assume_init_ref() } + } + + #[cfg(feature = "loom")] + fn copy_value(&self) -> T + where + T: Copy, + { + self.0.with(|slot| { + // SAFETY: The claim has the same publication and exclusive + // consumer proof as native code; Loom tracks this shared access. + unsafe { *(*slot).assume_init_ref() } + }) + } + #[cfg(not(feature = "loom"))] fn drop_value(&self) { // SAFETY: `Inner::drop` calls this only for the bounded sequence range @@ -150,7 +178,7 @@ impl SpscRing { Self::new_with_sequence(0) } - fn new_with_sequence(sequence: u64) -> Result { + pub(crate) fn new_with_sequence(sequence: u64) -> Result { validate_capacity::()?; let mut slots = Vec::with_capacity(N); @@ -208,7 +236,7 @@ impl SpscRing { } } -fn validate_capacity() -> Result<(), RingConfigError> { +pub(crate) fn validate_capacity() -> Result<(), RingConfigError> { if N == 0 { return Err(RingConfigError::ZeroCapacity); } @@ -233,6 +261,75 @@ impl Producer { N } + pub const fn next_sequence(&self) -> u64 { + self.next + } + + /// Claims the producer's next vacant slot without publishing it. + /// + /// The returned capability holds the producer's exclusive mutable borrow, + /// so no shared reservation flag is required. Cancellation and drop leave + /// every cursor and slot unchanged. + pub fn try_reserve(&mut self) -> Result, TryReserveError> { + if self + .inner + .endpoint_state + .consumer_closed + .load(Ordering::Acquire) + { + return Err(TryReserveError::Disconnected); + } + + let mut distance = self.next.wrapping_sub(self.cached_tail); + if distance > N as u64 { + return Err(TryReserveError::Invariant( + RingInvariantError::CursorDistanceExceedsCapacity { + published_head: self.next, + consumed_tail: self.cached_tail, + capacity: N, + }, + )); + } + + if distance == N as u64 { + self.cached_tail = self.inner.consumed_tail.0.load(Ordering::Acquire); + distance = self.next.wrapping_sub(self.cached_tail); + if distance > N as u64 { + return Err(TryReserveError::Invariant( + RingInvariantError::CursorDistanceExceedsCapacity { + published_head: self.next, + consumed_tail: self.cached_tail, + capacity: N, + }, + )); + } + if distance == N as u64 { + return Err(TryReserveError::Full); + } + } + + let sequence = self.next; + let index = (sequence as usize) & (N - 1); + let slot = self + .inner + .slots + .get(index) + .ok_or(TryReserveError::Invariant( + RingInvariantError::SlotIndexOutOfBounds { + sequence, + index, + slot_count: self.inner.slots.len(), + }, + ))?; + Ok(VacantSlot { + slot, + published_head: &self.inner.published_head.0, + next: &mut self.next, + sequence, + _not_send_sync: PhantomData, + }) + } + pub fn try_push(&mut self, value: T) -> Result<(), TryPushError> { if self .inner @@ -266,10 +363,8 @@ impl Producer { .consumer_closed .load(Ordering::Acquire) } -} -impl Drop for Producer { - fn drop(&mut self) { + pub(crate) fn close(&mut self) { self.inner .endpoint_state .producer_closed @@ -277,6 +372,79 @@ impl Drop for Producer { } } +impl Drop for Producer { + fn drop(&mut self) { + self.close(); + } +} + +/// Exclusive producer-local capability for one vacant ring slot. +/// +/// Dropping or cancelling the capability is a no-op. `publish` is infallible +/// after issuance and is the ring publication linearization point. +/// +/// The capability is intentionally neither `Send` nor `Sync`: +/// +/// ```compile_fail +/// fn require_send() {} +/// require_send::>(); +/// ``` +/// +/// ```compile_fail +/// fn require_sync() {} +/// require_sync::>(); +/// ``` +#[must_use = "dropping a vacant slot cancels the reservation"] +pub struct VacantSlot<'producer, T, const N: usize> { + slot: &'producer Slot, + published_head: &'producer AtomicU64, + next: &'producer mut u64, + sequence: u64, + _not_send_sync: PhantomData<(Rc<()>, [(); N])>, +} + +impl VacantSlot<'_, T, N> { + pub const fn sequence(&self) -> u64 { + self.sequence + } + + pub fn cancel(self) {} + + pub fn publish(self, value: T) { + let Self { + slot, + published_head, + next: producer_next, + sequence, + _not_send_sync: _, + } = self; + let next = sequence.wrapping_add(1); + + slot.write(value); + *producer_next = next; + published_head.store(next, Ordering::Release); + } + + /// Test-only interruption point after slot initialization and before the + /// published-head Release store. `T: Copy` guarantees that the deliberately + /// unreachable test value has no destructor to leak during fault closure. + #[cfg(all(test, not(feature = "loom")))] + pub(crate) fn publish_then_panic_before_head(self, value: T) -> ! + where + T: Copy, + { + let Self { + slot, + published_head: _, + next: _, + sequence: _, + _not_send_sync: _, + } = self; + slot.write(value); + panic!("injected ring fault after slot write and before head publication"); + } +} + pub struct Consumer { inner: Arc>, next: u64, @@ -289,6 +457,108 @@ impl Consumer { N } + /// Claims the next published slot without reclaiming its capacity. + /// + /// The validation copy is available through `OccupiedSlot::value`. Only + /// `OccupiedSlot::commit` moves the original value and advances the tail. + pub fn try_claim(&mut self) -> Result, TryClaimError> + where + T: Copy, + { + let (sequence, index) = match self.next_occupied_slot() { + Ok(position) => position, + Err(error @ TryClaimError::Invariant(_)) => { + self.close(); + return Err(error); + } + Err(error) => return Err(error), + }; + let slot = match self.inner.slots.get(index) { + Some(slot) => slot, + None => { + self.close(); + return Err(TryClaimError::Invariant( + RingInvariantError::SlotIndexOutOfBounds { + sequence, + index, + slot_count: self.inner.slots.len(), + }, + )); + } + }; + let value = slot.copy_value(); + + Ok(OccupiedSlot { + slot, + consumed_tail: &self.inner.consumed_tail.0, + consumer_closed: &self.inner.endpoint_state.consumer_closed, + next: &mut self.next, + sequence, + value, + _not_send_sync: PhantomData, + }) + } + + fn next_occupied_slot(&mut self) -> Result<(u64, usize), TryClaimError> { + let mut available = self.cached_head.wrapping_sub(self.next); + if available > N as u64 { + return Err(TryClaimError::Invariant( + RingInvariantError::CursorDistanceExceedsCapacity { + published_head: self.cached_head, + consumed_tail: self.next, + capacity: N, + }, + )); + } + + if available == 0 { + self.cached_head = self.inner.published_head.0.load(Ordering::Acquire); + available = self.cached_head.wrapping_sub(self.next); + if available > N as u64 { + return Err(TryClaimError::Invariant( + RingInvariantError::CursorDistanceExceedsCapacity { + published_head: self.cached_head, + consumed_tail: self.next, + capacity: N, + }, + )); + } + + if available == 0 { + if self + .inner + .endpoint_state + .producer_closed + .load(Ordering::Acquire) + { + // The producer publishes every claimed permit before its + // ordered close. Reloading after closure prevents the final + // publication from being hidden behind an earlier head. + self.cached_head = self.inner.published_head.0.load(Ordering::Acquire); + available = self.cached_head.wrapping_sub(self.next); + if available > N as u64 { + return Err(TryClaimError::Invariant( + RingInvariantError::CursorDistanceExceedsCapacity { + published_head: self.cached_head, + consumed_tail: self.next, + capacity: N, + }, + )); + } + if available == 0 { + return Err(TryClaimError::Disconnected); + } + } else { + return Err(TryClaimError::Empty); + } + } + } + + let sequence = self.next; + let index = (sequence as usize) & (N - 1); + Ok((sequence, index)) + } + pub fn try_pop(&mut self) -> Result { if self.next == self.cached_head { self.cached_head = self.inner.published_head.0.load(Ordering::Acquire); @@ -328,15 +598,93 @@ impl Consumer { .producer_closed .load(Ordering::Acquire) } -} -impl Drop for Consumer { - fn drop(&mut self) { + pub(crate) fn close(&self) { self.inner .endpoint_state .consumer_closed .store(true, Ordering::Release); } + + #[cfg(all(test, not(feature = "loom")))] + pub(crate) fn inject_cached_head_for_test(&mut self, cached_head: u64) { + self.cached_head = cached_head; + } +} + +impl Drop for Consumer { + fn drop(&mut self) { + self.close(); + } +} + +/// Exclusive consumer-local claim over one initialized `Copy` slot. +/// +/// `value` returns a validation copy without changing ring state. Dropping or +/// cancelling the claim leaves the slot published. `commit` moves the original +/// slot value and Release-publishes capacity reclamation. +/// +/// The claim is intentionally neither `Send` nor `Sync`: +/// +/// ```compile_fail +/// fn require_send() {} +/// require_send::>(); +/// ``` +/// +/// ```compile_fail +/// fn require_sync() {} +/// require_sync::>(); +/// ``` +#[must_use = "an occupied slot reclaims capacity only when committed"] +pub struct OccupiedSlot<'consumer, T: Copy, const N: usize> { + slot: &'consumer Slot, + consumed_tail: &'consumer AtomicU64, + consumer_closed: &'consumer AtomicBool, + next: &'consumer mut u64, + sequence: u64, + value: T, + _not_send_sync: PhantomData<(Rc<()>, [(); N])>, +} + +impl OccupiedSlot<'_, T, N> { + pub const fn sequence(&self) -> u64 { + self.sequence + } + + pub const fn value(&self) -> T { + self.value + } + + pub(crate) fn close_consumer(self) { + let Self { + consumer_closed, + slot: _, + consumed_tail: _, + next: _, + sequence: _, + value: _, + _not_send_sync: _, + } = self; + consumer_closed.store(true, Ordering::Release); + } + + pub fn commit(self) -> T { + let Self { + slot, + consumed_tail, + consumer_closed: _, + next: consumer_next, + sequence, + value: _, + _not_send_sync: _, + } = self; + let next = sequence.wrapping_add(1); + + let value = slot.read(); + *consumer_next = next; + consumed_tail.store(next, Ordering::Release); + value + } } #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -369,6 +717,97 @@ impl fmt::Display for RingConfigError { impl Error for RingConfigError {} +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RingInvariantError { + CursorDistanceExceedsCapacity { + published_head: u64, + consumed_tail: u64, + capacity: usize, + }, + SlotIndexOutOfBounds { + sequence: u64, + index: usize, + slot_count: usize, + }, +} + +impl fmt::Display for RingInvariantError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::CursorDistanceExceedsCapacity { + published_head, + consumed_tail, + capacity, + } => write!( + f, + "SPSC cursor distance from consumed {consumed_tail} to published {published_head} exceeds capacity {capacity}" + ), + Self::SlotIndexOutOfBounds { + sequence, + index, + slot_count, + } => write!( + f, + "SPSC sequence {sequence} selected slot {index} outside slot count {slot_count}" + ), + } + } +} + +impl Error for RingInvariantError {} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TryReserveError { + Full, + Disconnected, + Invariant(RingInvariantError), +} + +impl fmt::Display for TryReserveError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Full => write!(f, "SPSC ring is full"), + Self::Disconnected => write!(f, "SPSC consumer is disconnected"), + Self::Invariant(error) => write!(f, "SPSC reservation invariant failed: {error}"), + } + } +} + +impl Error for TryReserveError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::Invariant(error) => Some(error), + Self::Full | Self::Disconnected => None, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TryClaimError { + Empty, + Disconnected, + Invariant(RingInvariantError), +} + +impl fmt::Display for TryClaimError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Empty => write!(f, "SPSC ring is empty"), + Self::Disconnected => write!(f, "SPSC producer is disconnected"), + Self::Invariant(error) => write!(f, "SPSC claim invariant failed: {error}"), + } + } +} + +impl Error for TryClaimError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::Invariant(error) => Some(error), + Self::Empty | Self::Disconnected => None, + } + } +} + #[derive(Debug, Eq, PartialEq)] pub enum TryPushError { Full(T), @@ -413,7 +852,9 @@ impl Error for TryPopError {} #[cfg(all(test, not(feature = "loom")))] mod tests { - use super::{SpscRing, TryPopError, TryPushError}; + use super::{ + RingInvariantError, SpscRing, TryClaimError, TryPopError, TryPushError, TryReserveError, + }; #[test] fn modular_sequence_wrap_preserves_fifo_and_capacity() { @@ -429,11 +870,107 @@ mod tests { } assert_eq!(consumer.try_pop(), Err(TryPopError::Empty)); } + + #[test] + fn permit_and_claim_sequences_wrap_together() { + let ring = SpscRing::::new_with_sequence(u64::MAX).expect("valid wrapped ring"); + let (mut producer, mut consumer) = ring.split(); + + let permit = producer.try_reserve().expect("wrapped slot is vacant"); + assert_eq!(permit.sequence(), u64::MAX); + permit.publish(1); + assert_eq!(producer.next_sequence(), 0); + + let claim = consumer.try_claim().expect("wrapped slot is occupied"); + assert_eq!(claim.sequence(), u64::MAX); + assert_eq!(claim.value(), 1); + assert_eq!(claim.commit(), 1); + + let permit = producer.try_reserve().expect("slot was reclaimed"); + assert_eq!(permit.sequence(), 0); + permit.publish(2); + let claim = consumer.try_claim().expect("post-wrap value is visible"); + assert_eq!(claim.sequence(), 0); + assert_eq!(claim.commit(), 2); + } + + #[test] + fn reservation_and_claim_report_typed_cursor_invariant_errors() { + let ring = SpscRing::::new().expect("valid ring"); + let (mut producer, mut consumer) = ring.split(); + + producer.cached_tail = 1; + let reserve_error = match producer.try_reserve() { + Ok(_) => panic!("corrupt producer cursor must not reserve"), + Err(error) => error, + }; + assert_eq!( + reserve_error, + TryReserveError::Invariant(RingInvariantError::CursorDistanceExceedsCapacity { + published_head: 0, + consumed_tail: 1, + capacity: 2, + }) + ); + + consumer.cached_head = 3; + let claim_error = match consumer.try_claim() { + Ok(_) => panic!("corrupt consumer cursor must not claim"), + Err(error) => error, + }; + assert_eq!( + claim_error, + TryClaimError::Invariant(RingInvariantError::CursorDistanceExceedsCapacity { + published_head: 3, + consumed_tail: 0, + capacity: 2, + }) + ); + } + + #[test] + fn explicit_endpoint_close_is_idempotent() { + let ring = SpscRing::::new().expect("valid ring"); + let (mut producer, mut consumer) = ring.split(); + producer.close(); + producer.close(); + assert!(matches!( + consumer.try_claim(), + Err(TryClaimError::Disconnected) + )); + + let ring = SpscRing::::new().expect("valid ring"); + let (mut producer, consumer) = ring.split(); + consumer.close(); + consumer.close(); + assert!(matches!( + producer.try_reserve(), + Err(TryReserveError::Disconnected) + )); + } + + #[test] + fn occupied_slot_can_close_consumer_without_reclaiming_the_slot() { + let ring = SpscRing::::new().expect("valid ring"); + let (mut producer, mut consumer) = ring.split(); + producer.try_reserve().expect("initial slot").publish(5); + + let claim = consumer.try_claim().expect("published slot"); + claim.close_consumer(); + assert!(matches!( + producer.try_reserve(), + Err(TryReserveError::Disconnected) + )); + assert_eq!( + consumer.try_claim().expect("tail was not advanced").value(), + 5 + ); + } } #[cfg(all(test, feature = "loom"))] mod loom_tests { - use super::{SpscRing, TryPopError, TryPushError}; + use super::{SpscRing, TryClaimError, TryPopError, TryPushError, TryReserveError}; use loom::thread; #[test] @@ -488,4 +1025,113 @@ mod loom_tests { consumer_thread.join().expect("model consumer succeeds"); }); } + + #[test] + fn loom_permit_cancel_claim_drop_commit_reuse_and_wrap() { + loom::model(|| { + let ring = + SpscRing::::new_with_sequence(u64::MAX).expect("valid wrapped model ring"); + let (mut producer, mut consumer) = ring.split(); + + drop(producer.try_reserve().expect("initial dropped permit")); + assert_eq!(producer.next_sequence(), u64::MAX); + producer.try_reserve().expect("initial permit").cancel(); + assert_eq!(producer.next_sequence(), u64::MAX); + + let permit = producer.try_reserve().expect("cancelled slot is vacant"); + assert_eq!(permit.sequence(), u64::MAX); + permit.publish(1); + assert_eq!(producer.next_sequence(), 0); + + let claim = consumer.try_claim().expect("published slot is claimable"); + assert_eq!(claim.sequence(), u64::MAX); + assert_eq!(claim.value(), 1); + drop(claim); + assert_eq!(producer.try_reserve().err(), Some(TryReserveError::Full)); + + let claim = consumer.try_claim().expect("dropped claim retries"); + assert_eq!(claim.commit(), 1); + let permit = producer.try_reserve().expect("commit reclaimed slot"); + assert_eq!(permit.sequence(), 0); + permit.publish(2); + assert_eq!(consumer.try_claim().expect("second claim").commit(), 2); + }); + } + + #[test] + fn loom_reserved_publication_may_race_consumer_close() { + loom::model(|| { + let ring = SpscRing::::new().expect("valid model ring"); + let (mut producer, consumer) = ring.split(); + let permit = producer + .try_reserve() + .expect("permit is irrevocable after issuance"); + + let close_thread = thread::spawn(move || drop(consumer)); + thread::yield_now(); + permit.publish(1); + drop(producer); + close_thread.join().expect("consumer close succeeds"); + }); + } + + #[test] + fn loom_claim_withholds_reuse_until_commit_and_drains_close() { + loom::model(|| { + let ring = SpscRing::::new().expect("valid model ring"); + let (mut producer, mut consumer) = ring.split(); + + let producer_thread = thread::spawn(move || { + producer.try_reserve().expect("initial slot").publish(1); + + loop { + match producer.try_reserve() { + Ok(permit) => { + permit.publish(2); + break; + } + Err(TryReserveError::Full) => thread::yield_now(), + Err(error) => panic!("unexpected model reservation error: {error}"), + } + } + }); + + let consumer_thread = thread::spawn(move || { + let first = loop { + match consumer.try_claim() { + Ok(claim) => break claim, + Err(TryClaimError::Empty) => thread::yield_now(), + Err(error) => panic!("unexpected first model claim error: {error}"), + } + }; + assert_eq!(first.value(), 1); + drop(first); + + let first = consumer.try_claim().expect("dropped claim retries"); + assert_eq!(first.commit(), 1); + + let second = loop { + match consumer.try_claim() { + Ok(claim) => break claim, + Err(TryClaimError::Empty) => thread::yield_now(), + Err(error) => panic!("unexpected second model claim error: {error}"), + } + }; + assert_eq!(second.value(), 2); + assert_eq!(second.commit(), 2); + + loop { + match consumer.try_claim() { + Err(TryClaimError::Disconnected) => break, + Err(TryClaimError::Empty) => thread::yield_now(), + Err(error) => panic!("unexpected close model claim error: {error}"), + Ok(_) => panic!("unexpected extra model slot"), + } + } + }); + + producer_thread.join().expect("model producer succeeds"); + consumer_thread.join().expect("model consumer succeeds"); + }); + } } diff --git a/lib/event/tests/admission.rs b/lib/event/tests/admission.rs new file mode 100644 index 00000000..cd99394e --- /dev/null +++ b/lib/event/tests/admission.rs @@ -0,0 +1,443 @@ +#![cfg(not(feature = "loom"))] + +use std::{collections::VecDeque, thread}; + +use aegis_event::{ + AdmissionConfigError, RingConfigError, SlabAppendError, SlabPageBuilder, SlabPageConfig, + TelemetryDescriptor, TryAdmitError, TryConsumeError, VolatileAdmissionChannel, +}; + +fn config(byte_capacity: usize, descriptor_capacity: usize) -> SlabPageConfig { + SlabPageConfig { + arena_id: 17, + arena_generation: 23, + byte_capacity, + descriptor_capacity, + first_sequence: 41, + } +} + +#[test] +fn construction_validates_ring_and_page_before_exposing_endpoints() { + assert!(matches!( + VolatileAdmissionChannel::<3>::new(config(8, 2)), + Err(AdmissionConfigError::Ring(RingConfigError::NotPowerOfTwo { + capacity: 3 + })) + )); + assert!(matches!( + VolatileAdmissionChannel::<2>::new(config(0, 2)), + Err(AdmissionConfigError::Slab(_)) + )); +} + +#[test] +fn frame_claim_withholds_ring_capacity_until_explicit_commit() { + let channel = VolatileAdmissionChannel::<1>::new(config(32, 3)).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + + let first = producer + .try_admit(b"first", 7, 1) + .expect("first event is volatile-ring visible"); + assert_eq!(first.sequence(), 41); + + let frame = consumer.try_next().expect("first event validates"); + assert_eq!(frame.descriptor().sequence, 41); + assert_eq!(frame.payload(), b"first"); + assert_eq!( + producer.try_admit(b"second", 8, 2), + Err(TryAdmitError::RingFull) + ); + assert_eq!(producer.published_count(), 1); + assert_eq!(producer.used_bytes(), 5); + assert_eq!(producer.next_sequence(), 42); + + frame.commit(); + let second = producer + .try_admit(b"second", 8, 2) + .expect("claim commit releases ring capacity"); + assert_eq!(second.sequence(), 42); + + let frame = consumer.try_next().expect("second event validates"); + assert_eq!(frame.payload(), b"second"); + frame.commit(); + assert!(matches!(consumer.try_next(), Err(TryConsumeError::Empty))); +} + +#[test] +fn dropping_a_frame_claim_retries_the_same_descriptor_without_tail_advance() { + let channel = VolatileAdmissionChannel::<1>::new(config(16, 2)).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + producer.try_admit(b"retry", 1, 0).expect("event admitted"); + + { + let frame = consumer.try_next().expect("event validates"); + assert_eq!(frame.payload(), b"retry"); + } + + assert_eq!( + producer.try_admit(b"blocked", 1, 0), + Err(TryAdmitError::RingFull) + ); + let frame = consumer + .try_next() + .expect("uncommitted claim is offered again"); + assert_eq!(frame.descriptor().sequence, 41); + frame.commit(); +} + +#[test] +fn invalid_and_page_capacity_errors_leave_ring_and_page_transactional() { + let channel = VolatileAdmissionChannel::<2>::new(config(3, 1)).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + + assert_eq!( + producer.try_admit(&[], 1, 0), + Err(TryAdmitError::Slab(SlabAppendError::EmptyPayload)) + ); + assert_eq!(producer.published_count(), 0); + assert_eq!(producer.used_bytes(), 0); + assert_eq!(producer.next_sequence(), 41); + assert!(matches!(consumer.try_next(), Err(TryConsumeError::Empty))); + + producer + .try_admit(b"abc", 1, 0) + .expect("exact page capacity fits"); + let frame = consumer.try_next().expect("event validates"); + frame.commit(); + + assert_eq!( + producer.try_admit(b"x", 1, 0), + Err(TryAdmitError::Slab( + SlabAppendError::DescriptorCapacityExhausted { capacity: 1 } + )) + ); + assert_eq!(producer.published_count(), 1); + assert_eq!(producer.used_bytes(), 3); + assert_eq!(producer.next_sequence(), 42); + assert!(matches!(consumer.try_next(), Err(TryConsumeError::Empty))); +} + +#[test] +fn full_ring_rejection_does_not_copy_or_consume_page_sequence_space() { + let channel = VolatileAdmissionChannel::<1>::new(config(32, 3)).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + producer + .try_admit(b"one", 1, 0) + .expect("first event admitted"); + + assert_eq!( + producer.try_admit(b"must-not-copy", 2, 0), + Err(TryAdmitError::RingFull) + ); + assert_eq!(producer.published_count(), 1); + assert_eq!(producer.used_bytes(), 3); + assert_eq!(producer.next_sequence(), 42); + + consumer.try_next().expect("first event").commit(); + let retry = producer + .try_admit(b"must-not-copy", 2, 0) + .expect("same caller bytes can be retried"); + assert_eq!(retry.sequence(), 42); + assert_eq!(producer.published_count(), 2); + assert_eq!(producer.used_bytes(), 16); +} + +#[test] +fn disconnected_consumer_rejects_before_page_mutation() { + let channel = VolatileAdmissionChannel::<2>::new(config(16, 2)).expect("bounded channel"); + let (mut producer, consumer) = channel.split(); + drop(consumer); + + assert_eq!( + producer.try_admit(b"never-copied", 1, 0), + Err(TryAdmitError::ConsumerDisconnected) + ); + assert_eq!(producer.published_count(), 0); + assert_eq!(producer.used_bytes(), 0); + assert_eq!(producer.next_sequence(), 41); +} + +#[test] +fn explicit_finish_drains_every_event_then_reports_clean_end() { + let channel = VolatileAdmissionChannel::<4>::new(config(32, 3)).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + for payload in [b"one".as_slice(), b"two", b"three"] { + producer.try_admit(payload, 1, 0).expect("event admitted"); + } + producer.finish().expect("open producer finishes cleanly"); + + for expected in [b"one".as_slice(), b"two", b"three"] { + let frame = consumer.try_next().expect("published event drains"); + assert_eq!(frame.payload(), expected); + frame.commit(); + } + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::CleanEnd) + )); +} + +#[test] +fn ordinary_producer_drop_is_faulted_even_when_counts_match() { + let channel = VolatileAdmissionChannel::<2>::new(config(16, 1)).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + producer.try_admit(b"event", 1, 0).expect("event admitted"); + drop(producer); + + consumer + .try_next() + .expect("event remains drainable") + .commit(); + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::ProducerFaulted { + published: 1, + committed: 1, + }) + )); +} + +#[test] +fn sequence_wrap_is_identical_for_page_ring_and_consumer() { + let mut cfg = config(8, 2); + cfg.first_sequence = u64::MAX; + let channel = VolatileAdmissionChannel::<2>::new(cfg).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + + assert_eq!( + producer.try_admit(b"a", 1, 0).expect("first").sequence(), + u64::MAX + ); + assert_eq!( + producer.try_admit(b"b", 1, 0).expect("second").sequence(), + 0 + ); + producer.finish().expect("clean finish"); + + assert_eq!( + consumer + .try_next() + .expect("wrapped first") + .commit() + .sequence(), + u64::MAX + ); + assert_eq!( + consumer + .try_next() + .expect("wrapped second") + .commit() + .sequence(), + 0 + ); + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::CleanEnd) + )); +} + +#[test] +fn held_frame_payload_can_source_a_disjoint_later_admission() { + let channel = VolatileAdmissionChannel::<2>::new(config(32, 2)).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + producer + .try_admit(b"source", 1, 0) + .expect("first event admitted"); + + let first = consumer.try_next().expect("source validates"); + producer + .try_admit(first.payload(), 2, 0) + .expect("append-only destination is disjoint"); + assert_eq!(first.payload(), b"source"); + first.commit(); + + let second = consumer.try_next().expect("copied event validates"); + assert_eq!(second.payload(), b"source"); + second.commit(); +} + +#[test] +fn deterministic_short_traces_match_a_safe_vecdeque_and_sealed_page_oracle() { + const OP_COUNT: usize = 5; + const TRACE_LEN: usize = 5; + let trace_count = if cfg!(miri) { + 32 + } else { + OP_COUNT.pow(TRACE_LEN as u32) + }; + + for encoded_trace in 0..trace_count { + let cfg = config(4, 3); + let channel = VolatileAdmissionChannel::<2>::new(cfg).expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + let mut sealed_oracle = SlabPageBuilder::new(cfg).expect("bounded safe page"); + let mut queue = VecDeque::<(TelemetryDescriptor, Vec)>::with_capacity(2); + let mut accepted = Vec::new(); + let mut actual = Vec::new(); + let mut used_bytes = 0_usize; + let mut descriptor_count = 0_usize; + let mut next_sequence = cfg.first_sequence; + let mut trace = encoded_trace; + + for step in 0..TRACE_LEN { + let operation = trace % OP_COUNT; + trace /= OP_COUNT; + match operation { + 0..=2 => { + let payload: &[u8] = match operation { + 0 => b"a", + 1 => b"bb", + 2 => b"", + _ => unreachable!("operation is bounded above"), + }; + let schema_id = step as u32; + let flags = (encoded_trace as u16) ^ step as u16; + let expected_error = if payload.is_empty() { + Some(TryAdmitError::Slab(SlabAppendError::EmptyPayload)) + } else if descriptor_count == cfg.descriptor_capacity { + Some(TryAdmitError::Slab( + SlabAppendError::DescriptorCapacityExhausted { + capacity: cfg.descriptor_capacity, + }, + )) + } else if payload.len() > cfg.byte_capacity - used_bytes { + Some(TryAdmitError::Slab(SlabAppendError::PageFull { + requested: payload.len(), + remaining: cfg.byte_capacity - used_bytes, + })) + } else if queue.len() == queue.capacity() { + Some(TryAdmitError::RingFull) + } else { + None + }; + + let result = producer.try_admit(payload, schema_id, flags); + if let Some(error) = expected_error { + assert_eq!(result, Err(error), "trace {encoded_trace}, step {step}"); + } else { + let token = result.expect("oracle predicted admission success"); + assert_eq!(token.sequence(), next_sequence); + sealed_oracle + .try_append(payload, schema_id, flags) + .expect("safe oracle append succeeds"); + let descriptor = TelemetryDescriptor { + sequence: next_sequence, + arena_generation: cfg.arena_generation, + arena_id: cfg.arena_id, + flags, + offset: used_bytes as u32, + len: payload.len() as u32, + crc32c: crc32c::crc32c(payload), + schema_id, + }; + queue.push_back((descriptor, payload.to_vec())); + accepted.push(descriptor); + used_bytes += payload.len(); + descriptor_count += 1; + next_sequence = next_sequence.wrapping_add(1); + } + } + 3 => { + if let Some((descriptor, payload)) = queue.pop_front() { + let frame = consumer.try_next().expect("oracle queue is non-empty"); + assert_eq!(*frame.descriptor(), descriptor); + assert_eq!(frame.payload(), payload.as_slice()); + actual.push(*frame.descriptor()); + frame.commit(); + } else { + assert!(matches!(consumer.try_next(), Err(TryConsumeError::Empty))); + } + } + 4 => { + if let Some((descriptor, payload)) = queue.front() { + let frame = consumer.try_next().expect("oracle queue is non-empty"); + assert_eq!(frame.descriptor(), descriptor); + assert_eq!(frame.payload(), payload.as_slice()); + drop(frame); + } else { + assert!(matches!(consumer.try_next(), Err(TryConsumeError::Empty))); + } + } + _ => unreachable!("operation is reduced modulo OP_COUNT"), + } + + assert_eq!(producer.used_bytes(), used_bytes); + assert_eq!(producer.published_count(), descriptor_count); + assert_eq!(producer.next_sequence(), next_sequence); + } + + producer.finish().expect("open oracle producer finishes"); + while let Some((descriptor, payload)) = queue.pop_front() { + let frame = consumer.try_next().expect("remaining oracle event drains"); + assert_eq!(*frame.descriptor(), descriptor); + assert_eq!(frame.payload(), payload.as_slice()); + actual.push(*frame.descriptor()); + frame.commit(); + } + assert!(matches!( + consumer.try_next(), + Err(TryConsumeError::CleanEnd) + )); + + let sealed = sealed_oracle.seal(); + assert_eq!(sealed.descriptors(), accepted.as_slice()); + assert_eq!(actual, accepted); + } +} + +#[test] +fn tiny_ring_cross_thread_stress_has_no_loss_duplicates_or_reordering() { + let event_count = if cfg!(miri) { 64_u64 } else { 50_000_u64 }; + let channel = VolatileAdmissionChannel::<2>::new(config( + usize::try_from(event_count * 8).expect("test capacity fits"), + usize::try_from(event_count).expect("test count fits"), + )) + .expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + + let consumer_thread = thread::spawn(move || { + for expected in 0..event_count { + loop { + match consumer.try_next() { + Ok(frame) => { + assert_eq!(frame.descriptor().sequence, 41_u64.wrapping_add(expected)); + let bytes: [u8; 8] = + frame.payload().try_into().expect("test payload is one u64"); + assert_eq!(u64::from_le_bytes(bytes), expected); + frame.commit(); + break; + } + Err(TryConsumeError::Empty) => thread::yield_now(), + Err(error) => panic!("unexpected consumer error: {error}"), + } + } + } + loop { + match consumer.try_next() { + Err(TryConsumeError::CleanEnd) => break, + Err(TryConsumeError::Empty) => thread::yield_now(), + Ok(frame) => panic!( + "received unexpected sequence {} after the complete corpus", + frame.descriptor().sequence + ), + Err(error) => panic!("unexpected terminal consumer error: {error}"), + } + } + }); + + for value in 0..event_count { + loop { + match producer.try_admit(&value.to_le_bytes(), 1, 0) { + Ok(token) => { + assert_eq!(token.sequence(), 41_u64.wrapping_add(value)); + break; + } + Err(TryAdmitError::RingFull) => thread::yield_now(), + Err(error) => panic!("unexpected producer error: {error}"), + } + } + } + producer.finish().expect("clean producer finish"); + consumer_thread.join().expect("consumer thread succeeds"); +} diff --git a/lib/event/tests/published_allocations.rs b/lib/event/tests/published_allocations.rs index f02ae58b..7af65877 100644 --- a/lib/event/tests/published_allocations.rs +++ b/lib/event/tests/published_allocations.rs @@ -17,7 +17,7 @@ use std::{ cell::Cell, }; -use aegis_event::{PublishedSlabPage, SlabPageConfig}; +use aegis_event::{PublishedSlabPage, SlabPageConfig, TryAdmitError, VolatileAdmissionChannel}; struct ThreadTrackingAllocator; @@ -97,3 +97,68 @@ fn warmed_native_append_and_resolve_allocate_nothing() { assert_eq!(payload.as_ref(), b"allocation-free"); assert_eq!(allocations, 0, "append + resolve allocated unexpectedly"); } + +#[test] +fn warmed_admission_claim_validation_and_commit_allocate_nothing() { + let channel = VolatileAdmissionChannel::<2>::new(SlabPageConfig { + arena_id: 3, + arena_generation: 5, + byte_capacity: 64, + descriptor_capacity: 2, + first_sequence: 7, + }) + .expect("bounded channel"); + let (mut producer, mut consumer) = channel.split(); + + // Warm architecture-specific CRC dispatch before the measured boundary. + let _ = crc32c::crc32c(b"warmup"); + TRACK_ALLOCATIONS.with(|tracking| tracking.set(true)); + ALLOCATION_COUNT.with(|count| count.set(0)); + + let token = producer + .try_admit(b"allocation-free", 1, 0) + .expect("preallocated admission fits"); + let frame = consumer.try_next().expect("admitted frame validates"); + let payload_matches = frame.payload() == b"allocation-free"; + let committed = frame.commit(); + let allocations = ALLOCATION_COUNT.with(Cell::get); + TRACK_ALLOCATIONS.with(|tracking| tracking.set(false)); + + assert!(payload_matches); + assert_eq!(token, committed); + assert_eq!( + allocations, 0, + "admit + claim + validate + commit allocated unexpectedly" + ); +} + +#[test] +fn full_admission_rejection_allocates_nothing_and_does_not_mutate_the_page() { + let channel = VolatileAdmissionChannel::<1>::new(SlabPageConfig { + arena_id: 3, + arena_generation: 5, + byte_capacity: 64, + descriptor_capacity: 2, + first_sequence: 7, + }) + .expect("bounded channel"); + let (mut producer, _consumer) = channel.split(); + producer + .try_admit(b"occupy-ring", 1, 0) + .expect("first admission fits"); + let used_before = producer.used_bytes(); + let count_before = producer.published_count(); + let sequence_before = producer.next_sequence(); + + TRACK_ALLOCATIONS.with(|tracking| tracking.set(true)); + ALLOCATION_COUNT.with(|count| count.set(0)); + let result = producer.try_admit(b"must-not-copy", 1, 0); + let allocations = ALLOCATION_COUNT.with(Cell::get); + TRACK_ALLOCATIONS.with(|tracking| tracking.set(false)); + + assert_eq!(result, Err(TryAdmitError::RingFull)); + assert_eq!(producer.used_bytes(), used_before); + assert_eq!(producer.published_count(), count_before); + assert_eq!(producer.next_sequence(), sequence_before); + assert_eq!(allocations, 0, "full rejection allocated unexpectedly"); +} diff --git a/lib/event/tests/spsc.rs b/lib/event/tests/spsc.rs index 95770665..c3c24f1b 100644 --- a/lib/event/tests/spsc.rs +++ b/lib/event/tests/spsc.rs @@ -66,6 +66,81 @@ fn producer_returns_the_value_when_the_consumer_is_closed() { assert_eq!(producer.try_push(11), Err(TryPushError::Disconnected(11))); } +#[test] +fn vacant_slot_cancel_is_a_noop_and_publish_advances_exactly_once() { + let ring = SpscRing::::new().expect("valid ring"); + let (mut producer, mut consumer) = ring.split(); + + let permit = producer.try_reserve().expect("initial slot is vacant"); + assert_eq!(permit.sequence(), 0); + drop(permit); + assert_eq!(producer.next_sequence(), 0); + + let permit = producer + .try_reserve() + .expect("dropped permit leaves the slot vacant"); + permit.cancel(); + assert_eq!(producer.next_sequence(), 0); + assert_eq!(consumer.try_pop(), Err(TryPopError::Empty)); + + let permit = producer.try_reserve().expect("cancelled slot stays vacant"); + permit.publish(41); + assert_eq!(producer.next_sequence(), 1); + assert_eq!(consumer.try_pop(), Ok(41)); +} + +#[test] +fn occupied_slot_drop_withholds_capacity_until_infallible_commit() { + let ring = SpscRing::::new().expect("valid ring"); + let (mut producer, mut consumer) = ring.split(); + producer + .try_reserve() + .expect("initial slot is vacant") + .publish(7); + + let claim = consumer.try_claim().expect("published slot is claimable"); + assert_eq!(claim.sequence(), 0); + assert_eq!(claim.value(), 7); + drop(claim); + + let full = match producer.try_reserve() { + Ok(_) => panic!("an uncommitted claim must keep the ring full"), + Err(error) => error, + }; + assert_eq!(full.to_string(), "SPSC ring is full"); + + let claim = consumer + .try_claim() + .expect("dropping a claim leaves the same slot claimable"); + assert_eq!(claim.value(), 7); + assert_eq!(claim.commit(), 7); + + producer + .try_reserve() + .expect("commit reclaims the slot") + .publish(8); + assert_eq!(consumer.try_pop(), Ok(8)); +} + +#[test] +fn compatibility_push_pop_interoperate_with_permit_and_claim() { + let ring = SpscRing::::new().expect("valid ring"); + let (mut producer, mut consumer) = ring.split(); + + producer.try_push(1).expect("compatibility push succeeds"); + let claim = consumer + .try_claim() + .expect("claim observes compatibility publication"); + assert_eq!(claim.value(), 1); + assert_eq!(claim.commit(), 1); + + producer + .try_reserve() + .expect("permit observes reclaimed capacity") + .publish(2); + assert_eq!(consumer.try_pop(), Ok(2)); +} + #[derive(Clone, Debug)] struct DropProbe(Arc); @@ -94,6 +169,45 @@ fn final_ring_drop_destroys_each_unread_value_once() { assert_eq!(drops.load(Ordering::Relaxed), 2); } +#[test] +fn reserved_publication_may_finish_after_consumer_close_and_drops_once() { + let drops = Arc::new(AtomicUsize::new(0)); + let ring = SpscRing::::new().expect("valid ring"); + let (mut producer, consumer) = ring.split(); + let permit = producer + .try_reserve() + .expect("reservation precedes the close race"); + + drop(consumer); + permit.publish(DropProbe(Arc::clone(&drops))); + assert_eq!(drops.load(Ordering::Relaxed), 0); + drop(producer); + assert_eq!(drops.load(Ordering::Relaxed), 1); +} + +#[test] +fn claim_drains_final_publication_before_disconnect() { + let ring = SpscRing::::new().expect("valid ring"); + let (mut producer, mut consumer) = ring.split(); + producer + .try_reserve() + .expect("initial slot is vacant") + .publish(19); + drop(producer); + + let claim = consumer + .try_claim() + .expect("published value remains claimable after close"); + assert_eq!(claim.value(), 19); + assert_eq!(claim.commit(), 19); + + let disconnected = match consumer.try_claim() { + Ok(_) => panic!("drained closed producer must disconnect"), + Err(error) => error, + }; + assert_eq!(disconnected.to_string(), "SPSC producer is disconnected"); +} + #[test] fn producer_and_consumer_sequences_are_on_distinct_cache_lines() { let ring = SpscRing::::new().expect("valid ring"); diff --git a/mkdocs.yml b/mkdocs.yml index 41718911..289c6d4b 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -175,6 +175,7 @@ nav: - "ADR-0006: Cache-padded SPSC event fabric": adr/0006-cache-padded-spsc-event-fabric.md - "ADR-0007: Sealed generation-tagged slab pages": adr/0007-sealed-generation-tagged-slab-pages.md - "ADR-0008: Append-only published-prefix slab pages": adr/0008-append-only-published-prefix-slab-pages.md + - "ADR-0009: Failure-atomic slab/ring admission": adr/0009-failure-atomic-slab-ring-admission.md - Reference: - API reference (OpenAPI/Redoc): api-reference.md - Runtime authorization API: runtime-authorization-api.md From b22d02937812ece80968586ee1c1db00360cf1ca Mon Sep 17 00:00:00 2001 From: Lavkush Kumar Date: Tue, 14 Jul 2026 00:26:21 +0530 Subject: [PATCH 2/3] docs: fix case-sensitive installation.md links in World-Class HLD/LLD MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit macOS resolves Installation.md case-insensitively, the Linux CI runner does not — these three links failed the demo-guardrails docs validation on every PR since #1858 landed. --- docs/AegisAgent_World_Class_HLD.md | 4 ++-- docs/AegisAgent_World_Class_LLD.md | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/AegisAgent_World_Class_HLD.md b/docs/AegisAgent_World_Class_HLD.md index 871e5feb..b356454c 100644 --- a/docs/AegisAgent_World_Class_HLD.md +++ b/docs/AegisAgent_World_Class_HLD.md @@ -451,7 +451,7 @@ evidence: ## Installation -This HLD does not replace the product installation guide. Use [Installation](Installation.md) for prerequisites and [Deployment Guide](deployment-guide.md) for Docker, Helm, and production configuration. Both REST 8080 and gRPC 6334 must be exposed where the deployment model permits access. +This HLD does not replace the product installation guide. Use [Installation](installation.md) for prerequisites and [Deployment Guide](deployment-guide.md) for Docker, Helm, and production configuration. Both REST 8080 and gRPC 6334 must be exposed where the deployment model permits access. ## Quick Start @@ -513,7 +513,7 @@ Any missing dual-protocol operation is a contract gap, not permission to impleme ## CLI -Operational CLI commands must expose the same typed configuration, support a read-only `config validate`, and provide benchmark modes that report decision class, protocol, offered load, achieved throughput, latency histogram, error class, and queue utilization. Exact shipped setup and invocation commands remain documented in [Installation](Installation.md) and [Deployment Guide](deployment-guide.md). +Operational CLI commands must expose the same typed configuration, support a read-only `config validate`, and provide benchmark modes that report decision class, protocol, offered load, achieved throughput, latency histogram, error class, and queue utilization. Exact shipped setup and invocation commands remain documented in [Installation](installation.md) and [Deployment Guide](deployment-guide.md). ## Configuration Reference diff --git a/docs/AegisAgent_World_Class_LLD.md b/docs/AegisAgent_World_Class_LLD.md index 1a598794..5d447e52 100644 --- a/docs/AegisAgent_World_Class_LLD.md +++ b/docs/AegisAgent_World_Class_LLD.md @@ -343,7 +343,7 @@ Validation rules: ## Installation -No new endpoint is considered installed until protobuf generation tools are present and both protocol suites pass. Follow [Installation](Installation.md); PostgreSQL target mode additionally requires migrations, pooling, backups, and a completed backend qualification matrix. +No new endpoint is considered installed until protobuf generation tools are present and both protocol suites pass. Follow [Installation](installation.md); PostgreSQL target mode additionally requires migrations, pooling, backups, and a completed backend qualification matrix. ## Quick Start From 780ea4da1d4dce331a3a59232217a4c7739a9e5c Mon Sep 17 00:00:00 2001 From: Lavkush Kumar Date: Tue, 14 Jul 2026 00:27:08 +0530 Subject: [PATCH 3/3] =?UTF-8?q?ci(sast):=20scope=20the=20unsafe-usage=20ga?= =?UTF-8?q?te=20=E2=80=94=20block=20repo-wide=20except=20lib/event?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the community blanket rust.lang.security.unsafe-usage rule (excluded via --exclude-rule) with rust-unsafe-outside-reviewed-crates, which still blocks any unsafe outside lib/event. The event-fabric crates (ADR-0006..0009) carry isolated, SAFETY-documented unsafe gated by the event-concurrency, event-miri and event-sanitizers CI lanes, per the CONTRIBUTING unsafe evidence standard and docs/architecture.md §8. Verified with semgrep 1.169 locally: the scoped rule fires on unsafe outside lib/event, stays silent inside it, and --exclude-rule suppresses the registry rule (finding count 1 -> 0 on a synthetic case). Approved by operator (scoped rule swap chosen over per-site nosemgrep annotations). --- .github/workflows/sast.yml | 6 ++++++ .semgrep/aegisagent-rust.yml | 29 +++++++++++++++++++++++++++++ 2 files changed, 35 insertions(+) diff --git a/.github/workflows/sast.yml b/.github/workflows/sast.yml index abd435f5..fdd0f1e0 100644 --- a/.github/workflows/sast.yml +++ b/.github/workflows/sast.yml @@ -22,10 +22,16 @@ jobs: steps: - uses: actions/checkout@v7 - name: Run semgrep (custom rules + Rust/Python security rulesets) + # The community blanket unsafe-usage rule is replaced by the + # path-scoped rust-unsafe-outside-reviewed-crates rule in + # .semgrep/aegisagent-rust.yml: `lib/event` carries isolated, + # SAFETY-documented unsafe under ADR-0006..0009 with its own Miri, + # Loom, ASan and TSan CI lanes; everywhere else unsafe still blocks. run: | semgrep scan \ --config .semgrep/ \ --config p/rust \ --config p/python \ --config p/secrets \ + --exclude-rule rust.lang.security.unsafe-usage.unsafe-usage \ --error diff --git a/.semgrep/aegisagent-rust.yml b/.semgrep/aegisagent-rust.yml index a014304e..172f0b41 100644 --- a/.semgrep/aegisagent-rust.yml +++ b/.semgrep/aegisagent-rust.yml @@ -60,6 +60,35 @@ rules: references: - .claude/rules/rust_standards.md + - id: rust-unsafe-outside-reviewed-crates + languages: [rust] + severity: ERROR + message: >- + `unsafe` outside a crate with an accepted/proposed unsafe-review scope. + Repository law (docs/architecture.md §8, CONTRIBUTING.md) requires every + unsafe block to be isolated, justified with a `// SAFETY:` comment, and + covered by Miri/sanitizers (plus Loom for atomics) — and the crate must + be allowlisted here once that evidence exists in CI. Currently only + `lib/event` (ADR-0006..0009: SPSC ring, published-prefix slab pages, + failure-atomic admission; gated by the event-concurrency, event-miri and + event-sanitizers CI lanes) is allowlisted. This replaces the community + rust.lang.security.unsafe-usage rule, excluded in + .github/workflows/sast.yml. + patterns: + - pattern-either: + - pattern: unsafe { ... } + - pattern: unsafe fn $F(...) { ... } + paths: + exclude: + - "lib/event/*" + metadata: + category: security + references: + - docs/architecture.md + - CONTRIBUTING.md + - docs/adr/0006-cache-padded-spsc-event-fabric.md + - docs/adr/0009-failure-atomic-slab-ring-admission.md + - id: rust-unredacted-secret-logging languages: [generic] severity: ERROR