From 440c452b6ebeb9a45d1fd2299d84b1a37d18c475 Mon Sep 17 00:00:00 2001
From: meh <no-reply@pathscale.com>
Date: Tue, 8 Sep 2026 02:22:22 +0700
Subject: [PATCH 1/4] Run the formatting check CI was missing

`scripts/ci-local.sh` runs `cargo fmt --all --check` as its first step and CI
did not, which made the local script stricter than CI rather than equal to it,
so formatting drift could reach master unnoticed. Same command, same arguments.

Measured before adding it: the tree already passes, exit 0 with no diffs, so
this job is green as written rather than a red first run to clean up.
---
 .github/workflows/rust.yml | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index ecb2281e..695ace57 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -13,6 +13,18 @@ permissions:
   contents: read
 
 jobs:
+  # `scripts/ci-local.sh` runs `cargo fmt --all --check` as its first step. CI
+  # did not, which made the script stricter than CI instead of equal to it, and
+  # formatting drift reached master unnoticed. Same command, same arguments.
+  fmt:
+    name: Formatting
+    runs-on: ubicloud-standard-2
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v4
+      - name: Formatting (cargo fmt --all --check)
+        run: cargo fmt --all --check
+
   build:
     name: Build and test (${{ matrix.name }})
     runs-on: ubicloud-standard-2

From 89b58609a9ed04e48a37aa733b13f17431186bc3 Mon Sep 17 00:00:00 2001
From: meh <no-reply@pathscale.com>
Date: Tue, 8 Sep 2026 02:22:22 +0700
Subject: [PATCH 2/4] Backfill the change log to 0.3.10

The log carried beta.18 and beta.19 and nothing before them, so the history a
consumer needs in order to judge an upgrade was only in the commits. Entries
are derived from this repository's own history.
---
 CHANGELOG.md | 691 +++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 691 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b37f3eed..98d36fda 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -57,6 +57,697 @@ Change Log
 - Persisted primary/secondary index reconstruction and validation failures that
   could otherwise expose missing, duplicate, or mismatched rows.
 
+## [1.0.0-beta.17]
+
+### Added
+
+- `worktable_dsl`, a standalone crate holding the schema language. A schema can
+  now be read as data and written back, two schemas can be compared and the
+  cost of the difference reported, and declarations can be found across a
+  source tree.
+- Every generated table embeds its own declaration, so the schema is
+  recoverable from the code the macro produced.
+
+### Changed
+
+- Dependency requirements on the index and reclamation crates are carets rather
+  than exact pins, and `ps-reclaim` moved to 0.1.1 taken from the registry.
+- Retirement runs through the reclamation domain rather than through the guard.
+
+## [1.0.0-beta.16]
+
+### Changed
+
+- A batch pins its reclamation domain once instead of once per row, and
+  reclamation goes through `ps-reclaim`.
+- Requires data_bucket 0.5.5.
+
+## [1.0.0-beta.15]
+
+### BC Breaks
+
+- Non-unique index entries are identified and ordered by their `(key, value)`
+  pair, and the discriminator is gone. This is a persisted format change. An
+  index file written by beta.14 or earlier orders entries within a key by
+  discriminator, so it must be reindexed rather than loaded.
+
+### Fixed
+
+- Inserting into a non-unique index no longer scans every entry sharing the
+  key. On a table that puts a whole generation under one key, a one-file update
+  measured 698 ms on beta.13 and 15.1 s on beta.14; the per-row cost is back
+  from 330 us to roughly 9 us.
+- Index pages reconstruct in order of their minimum rather than their node id,
+  so a page that merely ends late no longer sorts ahead of one that starts
+  earlier.
+
+## [1.0.0-beta.14]
+
+### Added
+
+- `insert_many` with all-or-nothing semantics and CDC batch operations, and
+  `reserve_pks` for atomic primary key range reservation, both generated on
+  in-memory and persisted tables.
+- Per-table epoch pin domains. The global reader counter is replaced by
+  epoch-based retirement reclamation, and removed partitions are reclaimed
+  through the shared router under the same grace period.
+- Non-unique Arctic indexes for fixed-width integer keys, generated for
+  in-memory and persisted tables, with a pair-list checkpoint and WAL.
+
+### Changed
+
+- The persistence queue takes batches on a single wakeup, deduplicates page
+  queries when collecting a multi-row batch, and caps rows per group id so the
+  analyzer drain stays linear.
+- The table-global page barrier is narrowed to one barrier per page.
+- Requires WorkTablesIndex 0.0.8 and data_bucket 0.5.4.
+
+### Fixed
+
+- Mutation stripes are acquired as a batch without deadlocking.
+- A unique-collision unwind on a persisted table survives reload.
+
+## [1.0.0-beta.13]
+
+The audit-fix release. Most of it is durability and concurrency correctness
+rather than new surface.
+
+### BC Breaks
+
+- Persisted tables reject any `page_size` other than 16384 instead of writing a
+  file that cannot be read back.
+- A torn table-of-contents page 1 fails loudly instead of silently starting
+  from an empty table.
+- In-place update is rejected on indexed columns rather than leaving the index
+  stale.
+- An exhausted autoincrement generator panics instead of wrapping around and
+  handing out keys that are already in use.
+
+### Fixed
+
+- A failed data write no longer leaves published index keys behind. Insert,
+  update and delete each roll their index changes back.
+- Index pages are written before the table of contents that references them,
+  and table-of-contents key updates are guarded against segment overflow.
+- Data-file accounting: the u32 page-offset wrap when writing the last page's
+  data length, files whose length is an exact page multiple failing to reopen,
+  and non-extending writes being counted into the last page's length.
+- Vacuum no longer panics on a failed row move, no longer counts its scratch
+  pages in `pages_freed`, and never reports a source page fully moved when a
+  row was skipped.
+- A cancelled lock wait releases its registered op-lock, and a row that
+  vanishes mid-update returns `NotFound` instead of panicking.
+- The persistence worker refuses new operations once `Drop` has aborted it,
+  propagates `insert_cdc` serialization failure instead of panicking, and keeps
+  surviving data-only writes when event removal empties a batch.
+- ART checkpoints are atomic and clean up stale temporaries.
+- Arctic returns an empty range for `Excluded` bounds with no neighbour.
+- Row counts include inserts and deletes that reused a slot, and the
+  `PageIsFull` page switch is serialized against racing inserters.
+- A misplaced `persist` or `partition_by` in a declaration now names the
+  position it belongs in.
+
+## [1.0.0-beta.12]
+
+### Added
+
+- `partition_by`: one declared table type, many routed instances, with
+  `partition_ref` for borrowing a partition rather than cloning it.
+
+### Changed
+
+- `system_info` no longer copies every data page, and partition metrics scan
+  and allocate once instead of three times.
+
+### Fixed
+
+- A use-after-free in partition removal.
+- `close` reported success having persisted nothing.
+- One panic inside the router no longer disables the router.
+
+## [1.0.0-beta.11]
+
+### Changed
+
+- Requires the reviewed ART backend releases.
+
+## [1.0.0-beta.10]
+
+### Fixed
+
+- Table-of-contents inserts carry across persisted segments, reload insertion
+  stays on the fast path, and the insert API keeps its previous shape.
+
+## [1.0.0-beta.9]
+
+### Fixed
+
+- Persistence health is preserved across page splits.
+
+## [1.0.0-beta.8]
+
+### Fixed
+
+- Multi-row persistence order is preserved, and overlapping durable row writes
+  are ordered against each other.
+
+## [1.0.0-beta.7]
+
+### Fixed
+
+- The sized indexed update path is preserved.
+
+## [1.0.0-beta.6]
+
+### Changed
+
+- Fixed-width updates stay in place.
+
+### Fixed
+
+- Vacuum revalidates links after row locking.
+
+## [1.0.0-beta.5]
+
+### Added
+
+- Checked offline recovery load.
+
+### Changed
+
+- WorkTablesIndex structural persistence moved off the mutation path.
+- Logical WTI mutation stripes hash with FxHash, reusable data ranges are
+  subtracted in one pass, and full-row updates generate distinct paths.
+
+### Fixed
+
+- Same-size unsized updates apply in place instead of going through reinsert.
+- Full-table scans re-resolve stale links.
+- Vacuumed pages are reusable after reload.
+- Cancelled lock acquirers are cleaned up.
+- A panic while loading a persisted table is contained instead of unwinding
+  into the caller.
+
+## [1.0.0-beta.4]
+
+### Fixed
+
+- Release hardening and torn-store refusal are consolidated, so a torn store
+  refuses cleanly.
+
+## [1.0.0-beta.3]
+
+### Changed
+
+- Depends on the published index dependency chain rather than git revisions.
+- Row publication is concurrency-safe by default.
+- Persistence failures are terminal instead of leaving the table in a state
+  that looks usable.
+
+### Fixed
+
+- Synchronous insert is serialized against row mutations.
+- Page and link reclamation no longer overlap, and vacuum page reuse is
+  deferred through the read grace period.
+- Upsert retry backoff is bounded and its shift is capped, so same-key churn
+  cannot livelock.
+- Fragmented unsized index pages are compacted.
+- A stale multimap removal lookup is avoided.
+
+## [1.0.0-beta.2]
+
+### Added
+
+- Native ART index backends persist.
+
+### Changed
+
+- The temporary rusty-s3 fork is retired in favour of the published crate.
+- Stable index reads use the specialized path by default.
+
+### Fixed
+
+- Same-key upserts linearize.
+- Bounded retry for transient index misses is gated rather than always on.
+- Reused persistence slots coalesce.
+
+## [1.0.0-beta.1]
+
+### Added
+
+- Per-index backend selection in the `worktable!` declaration, with unique-index
+  adapters for Arctic, Congee and a parallel upstream indexset. Persistence is
+  preserved across indexset providers.
+
+## [0.9.4]
+
+### Changed
+
+- Requires data_bucket 0.4.1, and the temporary git patch is retired.
+
+### Fixed
+
+- A torn store refuses cleanly instead of terminating the process by signal.
+
+## [0.9.3]
+
+### Fixed
+
+- `worktable_version!` stays read-only when the primary key is unsized.
+
+## [0.9.2]
+
+### Fixed
+
+- Duplicate-key secondary indexes reconstruct correctly on reload.
+- Nodes sharing a maximum key order correctly, and pages are no longer re-sorted
+  on reload.
+- Space files flush before an operation reports done.
+
+## [0.9.1]
+
+### Changed
+
+- The proc-macro crate is published as `worktable_codegen` again, after a brief
+  release under the name `worktable_macros`.
+
+## [0.9.0]
+
+### Changed
+
+- Moves to WorkTablesIndex 0.0.1 and data_bucket 0.4.0.
+- The unsound lock-free persistence queue is replaced with a mutexed
+  `VecDeque`.
+
+### Fixed
+
+- Row lock acquisition and vacuum no longer race between check and act.
+- `wait_for_ops` no longer returns while a popped operation is still in flight.
+- Upsert retries an existence flip instead of surfacing it to the caller.
+- A multi-row update locks one validated snapshot, predicate included, and
+  delete by non-unique index snapshots validated primary keys.
+- Gapped event streams are never force-applied to the on-disk index, and the
+  whole batch is scanned for event-id gaps rather than the last thirty events.
+- A failed batch sub-operation is reported without cancelling the rest of the
+  work.
+- `save_batch_data` tracks the real maximum created page id.
+- Vacuum persists row moves through CDC, so persisted tables survive
+  defragmentation.
+
+## [0.9.0-beta0.2.3]
+
+### Fixed
+
+- Primary key generator state is preserved across migration reinserts.
+
+## [0.9.0-beta0.2.2]
+
+### Changed
+
+- Range ordering query logic reworked.
+
+## [0.9.0-beta0.2.1]
+
+### Changed
+
+- Update locks spin before returning a `Pending` state.
+
+## [0.9.0-beta0.2.0]
+
+### Added
+
+- Migrations.
+
+## [0.9.0-beta0.1.4]
+
+### Fixed
+
+- Page-not-found bug in the table of contents.
+
+## [0.9.0-beta0.1.1]
+
+### Fixed
+
+- Persistence bug affecting operations that fail.
+
+## [0.9.0-alpha8]
+
+### Changed
+
+- S3 integration moves to a different client crate.
+
+## [0.9.0-alpha7]
+
+### Fixed
+
+- S3 integration bug.
+
+## [0.9.0-alpha6]
+
+### Changed
+
+- Moves to rustls.
+
+## [0.9.0-alpha5]
+
+### Added
+
+- nanoid support for primary keys.
+
+## [0.9.0-alpha4]
+
+### Fixed
+
+- Vacuum logic.
+
+## [0.9.0-alpha3]
+
+### Fixed
+
+- The S3 macro.
+
+## [0.9.0-alpha2]
+
+### Added
+
+- S3 sync feature.
+
+## [0.9.0-alpha1]
+
+### Changed
+
+- Persistence is moved behind separate traits.
+
+## [0.8.23]
+
+### Changed
+
+- `Lock`s are reworked around RAII guards.
+
+## [0.8.22]
+
+### Added
+
+- `MemStat` derive on the generated primary key type.
+
+### Changed
+
+- `DataPages` select is generic over the input link type.
+
+## [0.8.21]
+
+### Changed
+
+- `delete` is generic, matching `insert` and `update`.
+
+## [0.8.20]
+
+### Added
+
+- Vacuum.
+
+## [0.8.19]
+
+### Fixed
+
+- Optional fields in persisted tables.
+
+## [0.8.18]
+
+### Fixed
+
+- Persisted table code failed to compile when the declaration used `optional`
+  fields.
+
+## [0.8.17]
+
+### Changed
+
+- Updated `indexset`.
+
+## [0.8.16]
+
+### Changed
+
+- Dependencies are pinned to exact versions.
+
+## [0.8.15]
+
+### Fixed
+
+- Empty link registry.
+
+## [0.8.13]
+
+### Changed
+
+- Bumped `indexset`.
+
+## [0.8.12]
+
+### Changed
+
+- Bumped `data_bucket` to 0.3.5 and `wt-indexset` to 0.12.11, and the crate now
+  declares its repository.
+
+## [0.8.11]
+
+### Changed
+
+- Bumped `indexset`.
+
+## [0.8.10]
+
+### Changed
+
+- Bumped `data_bucket` to 0.3.3 and `wt-indexset` to 0.12.9.
+
+## [0.8.9]
+
+### Fixed
+
+- Empty node bug.
+
+## [0.8.8]
+
+### Added
+
+- Every `AtomicU*` and `AtomicI*` type is usable as a primary key.
+
+## [0.8.7]
+
+### Changed
+
+- Dependency bumps.
+
+## [0.8.6]
+
+### Fixed
+
+- An `update`-related bug.
+
+## [0.8.5]
+
+### Fixed
+
+- Another `update`-related bug.
+
+## [0.8.4]
+
+### Changed
+
+- Codegen version bump.
+
+## [0.8.3]
+
+### Fixed
+
+- `delete` queries on a table whose primary key is not named `id`.
+- An update bug, by way of an `indexset` update.
+
+## [0.8.1]
+
+### Added
+
+- The macro reports an error when an index names a column that does not exist,
+  and declaration errors are raised as `syn::Error`s with usable messages.
+
+### Fixed
+
+- `UnsizedNode` split.
+
+## [0.8.0]
+
+### Fixed
+
+- Unsized node bug.
+
+## [0.7.2]
+
+### Fixed
+
+- A further `update` bug.
+
+## [0.7.1]
+
+### Fixed
+
+- An update violation.
+
+## [0.7.0]
+
+### Fixed
+
+- Reinsert bug.
+
+## [0.6.14]
+
+### Added
+
+- Ghost inserts. A row is staged invisible and becomes visible only once its
+  index entries are in place, so a concurrent reader never observes a
+  half-inserted row.
+
+## [0.6.13]
+
+### Fixed
+
+- Concurrency bugs in `select`.
+
+## [0.6.12]
+
+### Fixed
+
+- A further locking bug.
+
+## [0.6.11]
+
+### Fixed
+
+- Locking bugs for unsized types, and an `UnsizedNode` bug on `update`.
+
+### Changed
+
+- Dependency bumps.
+
+## [0.6.10]
+
+### Changed
+
+- Republished against `worktable_codegen` 0.6.9. No library change.
+
+## [0.6.9]
+
+### Fixed
+
+- Concurrent persistence issues.
+
+## [0.6.8]
+
+### Fixed
+
+- `wait_for_ops` logic.
+
+## [0.6.7]
+
+### Fixed
+
+- `delete` on persisted tables.
+
+## [0.6.5]
+
+### Added
+
+- Custom derives can be attached to the generated row type.
+
+## [0.6.4]
+
+### Fixed
+
+- `uuid` usage.
+
+## [0.6.3]
+
+### Fixed
+
+- A debug `println!` on the persistence batch path no longer writes to stdout.
+
+## [0.6.2]
+
+### Fixed
+
+- Table-of-contents corrections.
+
+## [0.6.1]
+
+### Added
+
+- `update_in_place`.
+
+### Changed
+
+- The persistence queue is optimized.
+
+### Fixed
+
+- `insert` with an already-existing key.
+- A `use rkyv::Archive` import was required for some declarations.
+- `wait_for_ops`.
+
+## [0.5.6]
+
+### Changed
+
+- Updated `indexset`.
+
+## [0.5.5]
+
+### Fixed
+
+- Array-typed fields.
+
+### Changed
+
+- Moves to the newer Rust edition.
+
+## [0.5.4]
+
+### Added
+
+- Unsized index space, so index keys are no longer limited to fixed-width
+  types.
+- `SystemInfo` for the table and its indexes.
+- `where_by` on `SelectBuilder` for any column, indexed or not.
+- Float columns are usable in indexes, including ranges.
+
+### Fixed
+
+- Re-reading a table from file.
+- Index difference logic for `update` queries.
+
+## [0.5.1]
+
+### Changed
+
+- Persistence I/O is asynchronous.
+
+## [0.5.0]
+
+### Added
+
+- `select_where_{field}` queries for selecting data ranges.
+- `count` on the table.
+- Persist sync logic.
+
+### Changed
+
+- Non-unique indexes are backed by `IndexMultiMap`.
+
+### Fixed
+
+- Secondary index left inconsistent after an update.
+- Diff logic for a full-row update.
+
 ## [0.4.1]
 
 ### Added

From cce51f653e0758c3a76366e90b1fb848fb96bbec Mon Sep 17 00:00:00 2001
From: meh <no-reply@pathscale.com>
Date: Tue, 8 Sep 2026 02:22:22 +0700
Subject: [PATCH 3/4] Name the cause of a persistence event gap, not only its
 symptom

`BatchOperation::validate` refuses an event stream with a hole in it. A hole is
normally transient, and one that survives the whole deferral budget is not, but
the stall message could not tell the two apart: it reported the range and
nothing else. Two very different bugs produce that line. Either an id was
assigned by the index and its event was dropped instead of queued, in which
case nothing will ever deliver it, or the operation was queued and batch
collection keeps assembling batches that exclude it.

Distinguishing them needs a record of what was queued. Assignment happens
inside the index and cannot be hooked from here, but every event reaching
persistence passes through `Queue`, so an id present in the stream and absent
from the ledger was assigned and never queued. That is the leak signature.
Producers are named by `std::panic::Location`, which is why `push_at` takes the
caller location and why the old `push` and `push_many` wrappers are deleted
rather than kept: calling them loses exactly the location the ledger exists to
record.

This observes and does not fix. `docs/TODO.md` records the three leak sites the
instrument points at, all in generated code: the `NotFound` arm of the update
query returns without building an acknowledge while its `AlreadyExists` sibling
builds one, and two `res?;` sites in update and delete propagate before
acknowledging.

The TODO's congee section is corrected while here. The crossbeam-epoch port was
already done in congee-wt 0.4.4, and it was a design change rather than a
rename: per-tree `Domain`, batched retires, guard provenance checked, and a new
`Guard` that is `!Send` and tree-scoped, which constrains future callers.
---
 docs/TODO.md                       | 123 +++--
 src/persistence/event_ledger.rs    | 735 +++++++++++++++++++++++++++++
 src/persistence/mod.rs             |   2 +
 src/persistence/operation/batch.rs |  69 ++-
 src/persistence/task.rs            | 194 +++++++-
 5 files changed, 1061 insertions(+), 62 deletions(-)
 create mode 100644 src/persistence/event_ledger.rs

diff --git a/docs/TODO.md b/docs/TODO.md
index 15824525..8b6f79fb 100644
--- a/docs/TODO.md
+++ b/docs/TODO.md
@@ -3,7 +3,9 @@
 What is known to be unfinished, and enough context to act on it without the
 conversation it came from. Ordered by whether it blocks a release.
 
-Last reviewed 2026-09-04, against `master` after beta.17 publication.
+Last reviewed 2026-09-07. Sections below still describe the repository as of
+beta.17; `master` is now at 1.0.0-beta.19 and this file has not been swept for
+what those two releases closed.
 
 ## Closed, and how
 
@@ -111,43 +113,88 @@ yank beta.16 once beta.17 supersedes it.
 
 ## Not blocking, but wrong today
 
-### `congee-wt` still pulls `crossbeam-epoch`
-
-beta.16 removes crossbeam from WorkTable's own reclamation, not from the build.
-`congee-wt` depends on it directly and re-exports its `Guard`, which
-`src/index/congee.rs:101` names in a signature; `crossbeam-skiplist` also
-arrives under `WorkTablesIndex` and `indexset`.
-
-congee's use is shallow: 34 references, none of them `Atomic<>`, `Owned::` or
-`Shared<>`, and most in tests. It only ever calls `pin()` and passes `Guard`
-around as an opaque token, so porting it to `ps-reclaim` is mechanical. The
-catch is that `Guard` is in congee-wt's public API, so it is a breaking change
-there plus the call sites here.
-
-`arctic-wt` should **not** be ported. It reclaims through `seize`, and is right
-to: a trie with short reads reaches quiescence constantly, which is the exact
-property that makes `seize` wrong for this crate, where `select` holds a read
-guard.
-
-### Persistence stalls on a primary index event gap, rarely
-
-One run of `cargo test --workspace --all-targets --all-features` failed with
-
-    persistence stalled on primary index event gap: last applied Id(1439),
-    next available Id(1455) (attempt 9)
-
-in `tests/persistence/loaded_index_growth.rs`. Not a flaky timeout: the guard
-at `src/persistence/operation/batch.rs:346` is deliberate, added in `c0c06ba`,
-and its comment says a gap that persists past eight deferrals means an event id
-was consumed without its event being queued, which only non-CDC index mutations
-do. The gap is 16 ids wide.
-
-1 failure in 6 full runs on this branch, 0 in 3 on master, 0 in 15
-persistence-only runs, so it needs whole-suite load and is not a beta.16
-regression. Do not start with a repro hunt: instrument `IndexChangeEventId`
-assignment against event queueing so the next occurrence names its own cause.
-Evidence at `~/code/wt-event-gap-2026-09-01.txt`.
-
+### `congee-wt` no longer pulls `crossbeam-epoch`
+
+Corrected 2026-09-07. This section said the port was outstanding and mechanical.
+Both halves were wrong.
+
+`congee-wt` dropped `crossbeam-epoch` in 0.4.4. `Cargo.toml:22` now reads
+`ps-reclaim = { version = "0.1.4", default-features = false, features =
+["libc", "spin"] }`, no `crossbeam_epoch` reference survives in its sources or
+tests, and this repository's `Cargo.lock` already resolves `congee-wt 0.4.4`.
+The two remaining `crossbeam` strings in that crate are attribution comments on
+a seqlock and a backoff loop.
+
+The call site this section worried about needed no change. It has moved to
+`src/index/congee.rs:120` and still reads
+`fn retire_old(pointer: usize, guard: &congee::epoch::Guard) -> Arc<V>`. The
+`congee::epoch::Guard` path was deliberately preserved across the port, and the
+lifetime the new guard carries elides in reference position.
+
+The port was not the mechanical rename described here. A literal swap would have
+kept one global epoch; what shipped gives each tree its own `Domain`, adds a
+bounded pending-retire batch, checks guard provenance so a guard from another
+tree panics rather than corrupting, and drains the tree's own domain on `Drop`.
+Worth knowing because the new `Guard` is `!Send` and tree-scoped, so a guard may
+not be created outside the thread and tree that uses it. Nothing in either
+repository does; every threaded test builds its guard inside the spawned
+closure.
+
+### Persistence event gap: three leak sites found, instrumented not yet fixed
+
+Updated 2026-09-07. This section previously said the cause was unknown and that
+the next step was instrumentation rather than a repro hunt. The instrumentation
+was built, and reading for it found the leaks.
+
+`IndexChangeEventId` is `indexset::cdc::change::Id`, allocated by
+`event_id.fetch_add` in the same statement that stamps the event, so indexset
+never consumes an id without emitting its event. Every leak is on our side: an
+event handed back and then dropped. Three sites, all in generated persisted
+query code, all on secondary streams, all confirmed by reading:
+
+- `codegen/src/generators/persist/queries/update.rs:569`. The
+  `IndexError::NotFound => Err(WorkTableError::NotFound)` arm returns with no
+  acknowledge, while its sibling `AlreadyExists` arm immediately above builds an
+  `Acknowledge` carrying `merged_events` and applies it. The events from
+  `process_difference_insert_cdc` are dropped on the `NotFound` path. This
+  asymmetry between two adjacent arms is the clearest of the three.
+- `codegen/src/generators/persist/queries/update.rs:593`,
+  `gen_process_diffs_remove_on_index`: `let (secondary_keys_events_remove, res)
+  = ...; res?;`. On `Err` the `?` returns before
+  `op.extend_secondary_key_events`, dropping the events bound on the line above.
+- `codegen/src/generators/persist/queries/delete.rs:101`: the same `res?;` shape
+  after `delete_row_cdc`.
+
+Checked and NOT leaks: the rollback arms of `insert_cdc`, `insert_many_cdc` and
+`reinsert_cdc` in `src/table/mod.rs` all merge forward and rollback events into
+an `Acknowledge`, as does the data-delete-failure restore path. Vacuum's
+`update_index_after_move` takes the non-CDC branch only when `persistence` is
+`None`, so the one case commit `c0c06ba` named is closed for persisted tables.
+
+The only structurally possible primary-stream leak is a refused
+`apply_operation` after the index mutation already consumed ids, and more
+generally any `?` between a CDC index mutation and its `apply_operation`.
+
+**The failure text quoted in earlier versions of this section is stale.** It
+said "attempt 9"; `GIVE_UP_AFTER_ATTEMPTS` is now 120, and
+`COLLECT_WHOLE_QUEUE_AFTER_ATTEMPTS` plus its regression test
+`collection_recovers_when_event_order_and_operation_order_disagree` were added
+since, for a symptom that reads identically but is a collection failure rather
+than a leak. Telling those two apart is exactly what the new ledger does.
+
+`src/persistence/event_ledger.rs` records queued, collected, requeued, trimmed
+and applied per stream in a bounded 8192-id window, and the guard's message now
+ends in a verdict: either ASSIGNED BUT NEVER QUEUED with the id range and the
+producer sites either side of the gap, or QUEUED BUT NOT APPLIED with the per-id
+stage history. It says so plainly when part of the gap fell outside the retained
+window, so it never claims "never queued" about an id it cannot answer for.
+Always compiled, gated at run time on `debug_assertions` or `WT_EVENT_LEDGER`,
+which puts it on exactly where the bug appears, since the stall needs a full
+debug `--all-features` run.
+
+Not fixed, and not compiled. The three sites need the same treatment the
+rollback arms already use: build an `Acknowledge` carrying the orphaned events
+before propagating the error.
 ## Housekeeping
 
 - `CHANGELOG.md` stops at 0.4.1, long before the 1.0.0-beta line.
diff --git a/src/persistence/event_ledger.rs b/src/persistence/event_ledger.rs
new file mode 100644
index 00000000..560cb7ee
--- /dev/null
+++ b/src/persistence/event_ledger.rs
@@ -0,0 +1,735 @@
+//! Pairs index change event id *assignment* with event *queueing*, so that a
+//! persistence stall on an event gap names its own cause instead of only its
+//! symptom.
+//!
+//! # The defect this exists for
+//!
+//! `BatchOperation::validate` refuses to apply an event stream with a hole in
+//! it (see the guard there and commit `c0c06ba`). A hole is normally
+//! transient: the operation carrying the missing id has been produced but not
+//! yet batched. A hole that survives the whole deferral budget means something
+//! else, and the old message could not tell the two apart. It reported the
+//! range and nothing more:
+//!
+//! ```text
+//! persistence stalled on primary index event gap: last applied Id(1439),
+//! next available Id(1455) (attempt 9)
+//! ```
+//!
+//! Two very different bugs produce that line:
+//!
+//! 1. **A leak.** An id was assigned by the index and its event was then
+//!    dropped instead of being pushed onto the persistence queue. Nothing will
+//!    ever deliver it and the stream is permanently gapped.
+//! 2. **A collection failure.** The operation carrying the id *was* queued and
+//!    is still sitting in the analyzer, but batch collection keeps assembling
+//!    batches that exclude it.
+//!
+//! Distinguishing those needs a record of what was queued, which is what this
+//! ledger keeps. It is deliberately not a fix for either: it only observes.
+//!
+//! # Where the two sides are observed
+//!
+//! Assignment happens inside the index (`indexset` bumps an `AtomicU64` and
+//! stamps the event in the same commit), which this crate cannot hook. What it
+//! *can* hook is the other end: every event that reaches persistence passes
+//! through [`crate::persistence::task::Queue`], so an id present in the stream
+//! but absent from this ledger was assigned and never queued. That is the leak
+//! signature, and it is what [`EventLedger::gap_report`] reports.
+//!
+//! The producer is named by [`std::panic::Location`], captured with
+//! `#[track_caller]` at the queue push, so a leak points at the call site that
+//! produced its neighbours: the generated query, `insert_many`, an
+//! acknowledge path, or vacuum's `apply_move`. A full backtrace is captured
+//! too, but only when `RUST_BACKTRACE` is set; `Backtrace::capture` is
+//! essentially free otherwise.
+//!
+//! # Gating: `debug_assertions`, not a cargo feature
+//!
+//! Recording is compiled in unconditionally and gated at run time by
+//! [`enabled`], which is true when `debug_assertions` is on or when
+//! `WT_EVENT_LEDGER` is set in the environment.
+//!
+//! `debug_assertions` was chosen over a new cargo feature for one reason: the
+//! stall is only ever observed under a full `cargo test --workspace
+//! --all-targets --all-features` run, and that run is a debug build, so the
+//! instrumentation is on exactly where the bug appears. A feature would also
+//! have been enabled by `--all-features`, but it would have to be declared in
+//! `Cargo.toml`, and a diagnostic that lives behind a flag nobody sets in the
+//! failing configuration is worthless. Release builds fold `enabled()` down
+//! to the environment check and every recording call returns immediately, so
+//! they pay a predictable-branch and nothing else.
+//!
+//! `WT_EVENT_LEDGER=1` exists so a release build can be told to record without
+//! being rebuilt, for the day the stall shows up outside a test run.
+//!
+//! Memory is bounded by [`WINDOW`] ids per stream. The gap sits at the head of
+//! the stream by construction, so a recent window always covers it; the report
+//! states the window it holds so a reader can see that for themselves.
+
+use std::backtrace::Backtrace;
+use std::collections::{BTreeMap, HashMap};
+use std::fmt::Write as _;
+use std::panic::Location;
+use std::sync::LazyLock;
+
+use data_bucket::Link;
+use indexset::cdc::change::ChangeEvent;
+use indexset::core::pair::Pair;
+use parking_lot::Mutex;
+
+use crate::persistence::{OperationId, OperationType};
+
+/// Ids retained per stream. The gap the guard reports is at the head of the
+/// stream, so a window this size covers it many times over: the one observed
+/// stall was 16 ids wide.
+const WINDOW: usize = 8192;
+
+/// Gap ids listed individually in a report before it summarises the rest.
+const MAX_LISTED_GAP_IDS: usize = 64;
+
+static ENABLED: LazyLock<bool> = LazyLock::new(|| {
+    if cfg!(debug_assertions) {
+        return true;
+    }
+    match std::env::var("WT_EVENT_LEDGER") {
+        Ok(value) => !value.is_empty() && value != "0" && !value.eq_ignore_ascii_case("false"),
+        Err(_) => false,
+    }
+});
+
+/// Whether event bookkeeping is recording in this process.
+///
+/// See the module comment for why this is `debug_assertions` plus an
+/// environment override rather than a cargo feature.
+#[inline]
+pub fn enabled() -> bool {
+    *ENABLED
+}
+
+/// Which index's event id sequence a record belongs to.
+///
+/// Every index keeps its own counter, so ids only mean anything relative to a
+/// stream. `Primary` allocates nothing, which keeps the hot path free of
+/// allocation; secondary streams are labelled by the `Debug` rendering of the
+/// table's `AvailableIndexes` value, which is the only name available to
+/// non-generic code here.
+#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub enum EventStream {
+    Primary,
+    Secondary(String),
+}
+
+impl std::fmt::Display for EventStream {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            EventStream::Primary => f.write_str("primary"),
+            EventStream::Secondary(index) => write!(f, "secondary {index}"),
+        }
+    }
+}
+
+/// What has been observed happening to one event id.
+///
+/// Flags, not a state machine: an id is queued, then collected into a batch,
+/// then possibly trimmed back out and requeued, possibly several times over.
+/// The report reads the whole set.
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
+pub struct Stages(u8);
+
+impl Stages {
+    /// Pushed onto the persistence queue by a producer.
+    pub const QUEUED: Stages = Stages(1 << 0);
+    /// Pulled into a `BatchOperation` by the analyzer.
+    pub const COLLECTED: Stages = Stages(1 << 1);
+    /// Handed back to the analyzer's queue after a deferral or a trim.
+    pub const REQUEUED: Stages = Stages(1 << 2);
+    /// Removed from a batch by `remove_operations_from_events`.
+    pub const TRIMMED: Stages = Stages(1 << 3);
+    /// Covered by the applied watermark reported after a batch.
+    pub const APPLIED: Stages = Stages(1 << 4);
+
+    fn insert(&mut self, other: Stages) {
+        self.0 |= other.0;
+    }
+
+    /// Both flag sets at once.
+    pub const fn union(self, other: Stages) -> Stages {
+        Stages(self.0 | other.0)
+    }
+
+    fn contains(self, other: Stages) -> bool {
+        self.0 & other.0 == other.0
+    }
+}
+
+impl std::fmt::Display for Stages {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let mut first = true;
+        for (flag, name) in [
+            (Stages::QUEUED, "queued"),
+            (Stages::COLLECTED, "collected"),
+            (Stages::REQUEUED, "requeued"),
+            (Stages::TRIMMED, "trimmed"),
+            (Stages::APPLIED, "applied"),
+        ] {
+            if self.contains(flag) {
+                if !first {
+                    f.write_str("+")?;
+                }
+                f.write_str(name)?;
+                first = false;
+            }
+        }
+        if first {
+            f.write_str("none")?;
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+struct IdRecord {
+    stages: Stages,
+    /// The operation that carried this id when it was first queued.
+    op_id: Option<OperationId>,
+    op_type: Option<OperationType>,
+    /// Producer call site, from `#[track_caller]` at the queue push.
+    site: Option<&'static Location<'static>>,
+    /// Captured only when `RUST_BACKTRACE` is set; otherwise `Disabled` and
+    /// free. Boxed so that the common empty case costs a pointer instead of an
+    /// inline `Backtrace` in each of the thousands of records held per stream.
+    backtrace: Option<Box<Backtrace>>,
+    collected: u32,
+    requeued: u32,
+}
+
+impl IdRecord {
+    fn new() -> Self {
+        Self {
+            stages: Stages::default(),
+            op_id: None,
+            op_type: None,
+            site: None,
+            backtrace: None,
+            collected: 0,
+            requeued: 0,
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+struct StreamLedger {
+    ids: BTreeMap<u64, IdRecord>,
+    /// Ids below this were evicted by the window and cannot be answered for.
+    evicted_below: u64,
+    /// Highest id ever seen at any stage on this stream.
+    highest_seen: u64,
+    /// Highest id the analyzer reported as applied.
+    applied_upto: u64,
+    queued_total: u64,
+}
+
+impl StreamLedger {
+    fn entry(&mut self, id: u64) -> &mut IdRecord {
+        if id > self.highest_seen {
+            self.highest_seen = id;
+        }
+        self.ids.entry(id).or_insert_with(IdRecord::new)
+    }
+
+    fn trim(&mut self) {
+        while self.ids.len() > WINDOW {
+            // Ids are close to monotonic, so the lowest key is the oldest.
+            if let Some((id, _)) = self.ids.pop_first() {
+                self.evicted_below = self.evicted_below.max(id + 1);
+            } else {
+                break;
+            }
+        }
+    }
+
+    /// Lowest id this ledger can still answer for.
+    fn window_start(&self) -> u64 {
+        self.ids.keys().next().copied().unwrap_or(self.evicted_below)
+    }
+}
+
+/// Per-table record of which index change event ids reached the persistence
+/// queue, and what happened to them afterwards.
+///
+/// Shared by `Arc` between the queue (the producer side), the analyzer, and
+/// the `BatchOperation` whose guard reads it.
+#[derive(Debug)]
+pub struct EventLedger {
+    label: String,
+    /// True when no producer writes to this ledger, so "never queued" here
+    /// means "never recorded", not "leaked". Reports say so rather than
+    /// accusing a producer that was never watched.
+    detached: bool,
+    streams: Mutex<HashMap<EventStream, StreamLedger>>,
+}
+
+impl EventLedger {
+    pub fn new(label: impl Into<String>) -> Self {
+        Self {
+            label: label.into(),
+            detached: false,
+            streams: Mutex::new(HashMap::new()),
+        }
+    }
+
+    /// A ledger attached to nothing, for analyzers and `BatchOperation`s built
+    /// outside `run_engine` (unit tests, defensive callers). It records
+    /// normally; it is simply never shared with a producer, so its reports say
+    /// so instead of reading a missing record as a leak.
+    pub fn detached() -> Self {
+        Self {
+            label: "<detached>".to_owned(),
+            detached: true,
+            streams: Mutex::new(HashMap::new()),
+        }
+    }
+
+    pub fn label(&self) -> &str {
+        &self.label
+    }
+
+    /// Collects the event ids of `evs`, for a later [`EventLedger::record_queued`].
+    ///
+    /// Split from the recording itself because the queue only knows a push was
+    /// accepted after the operation has been moved into it, and a refused push
+    /// must not be recorded as queued.
+    pub fn event_ids<T>(evs: &[ChangeEvent<Pair<T, Link>>]) -> Vec<u64> {
+        if !enabled() {
+            return Vec::new();
+        }
+        evs.iter().map(|ev| ev.id().inner()).collect()
+    }
+
+    /// Records every id in `ids` as queued by `site`.
+    ///
+    /// Called from the persistence queue push, which is the single point every
+    /// operation passes through on its way to the engine. An id in the applied
+    /// stream that never appears here was assigned by the index and dropped
+    /// before it reached persistence.
+    pub fn record_queued(
+        &self,
+        stream: EventStream,
+        ids: &[u64],
+        op_id: OperationId,
+        op_type: OperationType,
+        site: &'static Location<'static>,
+    ) {
+        if !enabled() || ids.is_empty() {
+            return;
+        }
+        // Costs nothing unless RUST_BACKTRACE is set: `capture` returns
+        // `Disabled` without walking any frames. One capture per push, moved
+        // onto the first newly recorded id, because every id in this vector
+        // came from the same producer.
+        let backtrace = Backtrace::capture();
+        let mut backtrace =
+            matches!(backtrace.status(), std::backtrace::BacktraceStatus::Captured).then_some(backtrace);
+        let mut streams = self.streams.lock();
+        let ledger = streams.entry(stream).or_default();
+        for id in ids.iter().copied() {
+            let record = ledger.entry(id);
+            let first_time = !record.stages.contains(Stages::QUEUED);
+            record.stages.insert(Stages::QUEUED);
+            if first_time {
+                record.op_id = Some(op_id);
+                record.op_type = Some(op_type);
+                record.site = Some(site);
+                // One backtrace per push, not per id: every id in this vector
+                // has the same producer, and keeping one each would multiply
+                // the cost of a `RUST_BACKTRACE` run for no extra signal.
+                if record.backtrace.is_none() {
+                    record.backtrace = backtrace.take().map(Box::new);
+                }
+            } else {
+                record.stages.insert(Stages::REQUEUED);
+                record.requeued = record.requeued.saturating_add(1);
+            }
+        }
+        ledger.queued_total = ledger.queued_total.saturating_add(ids.len() as u64);
+        ledger.trim();
+    }
+
+    /// Records a stage transition for a single id already known to a stream.
+    pub fn record_stage(&self, stream: EventStream, id: u64, stage: Stages) {
+        if !enabled() {
+            return;
+        }
+        let mut streams = self.streams.lock();
+        let ledger = streams.entry(stream).or_default();
+        let record = ledger.entry(id);
+        record.stages.insert(stage);
+        if stage.contains(Stages::COLLECTED) {
+            record.collected = record.collected.saturating_add(1);
+        }
+        if stage.contains(Stages::REQUEUED) {
+            record.requeued = record.requeued.saturating_add(1);
+        }
+        ledger.trim();
+    }
+
+    /// Records a stage transition for every id in `evs`.
+    pub fn record_stage_for_events<T>(&self, stream: EventStream, evs: &[ChangeEvent<Pair<T, Link>>], stage: Stages) {
+        if !enabled() || evs.is_empty() {
+            return;
+        }
+        let mut streams = self.streams.lock();
+        let ledger = streams.entry(stream).or_default();
+        for ev in evs {
+            let record = ledger.entry(ev.id().inner());
+            record.stages.insert(stage);
+            if stage.contains(Stages::COLLECTED) {
+                record.collected = record.collected.saturating_add(1);
+            }
+            if stage.contains(Stages::REQUEUED) {
+                record.requeued = record.requeued.saturating_add(1);
+            }
+        }
+        ledger.trim();
+    }
+
+    /// Records the applied watermark reported after a batch was accepted.
+    pub fn record_applied_upto(&self, stream: EventStream, id: u64) {
+        if !enabled() || id == 0 {
+            return;
+        }
+        let mut streams = self.streams.lock();
+        let ledger = streams.entry(stream).or_default();
+        if id > ledger.applied_upto {
+            ledger.applied_upto = id;
+        }
+        if id > ledger.highest_seen {
+            ledger.highest_seen = id;
+        }
+        for (_, record) in ledger.ids.range_mut(..=id) {
+            record.stages.insert(Stages::APPLIED);
+        }
+    }
+
+    /// Explains the gap between `last_applied` and `next_available`.
+    ///
+    /// This is the whole point of the ledger. For every id in the hole it says
+    /// whether the id ever reached the persistence queue, and if it did, what
+    /// happened to it afterwards, so the reader can tell a leaked event from a
+    /// batch collection that keeps missing one.
+    pub fn gap_report(&self, stream: &EventStream, last_applied: u64, next_available: u64) -> String {
+        let mut out = String::new();
+        if !enabled() {
+            let _ = write!(
+                out,
+                " Event bookkeeping is off in this build, so the gap cannot be attributed. \
+                 Re-run with WT_EVENT_LEDGER=1 (and RUST_BACKTRACE=1 for producer backtraces) \
+                 to have the next occurrence name its own cause."
+            );
+            return out;
+        }
+
+        if self.detached {
+            let _ = write!(
+                out,
+                " This analyzer holds detached event bookkeeping: no producer ever wrote to it, \
+                 so the gap cannot be attributed. Only analyzers built outside `run_engine` are detached."
+            );
+            return out;
+        }
+
+        let streams = self.streams.lock();
+        let Some(ledger) = streams.get(stream) else {
+            let _ = write!(
+                out,
+                " Event bookkeeping for table {label} holds no records at all for the {stream} stream, \
+                 which should be impossible while it is applying that stream's events.",
+                label = self.label,
+            );
+            return out;
+        };
+
+        let window_start = ledger.window_start();
+        let first_missing = last_applied.saturating_add(1);
+        if next_available <= first_missing {
+            return out;
+        }
+
+        // Bounded on purpose: this runs while building a panic message, and a
+        // corrupt watermark could otherwise make it walk billions of ids.
+        let scan_end = next_available.min(first_missing.saturating_add(WINDOW as u64));
+        let mut never_queued = Vec::new();
+        let mut queued_not_applied = Vec::new();
+        for id in first_missing..scan_end {
+            match ledger.ids.get(&id) {
+                Some(record) if record.stages.contains(Stages::QUEUED) => queued_not_applied.push((id, record)),
+                _ => never_queued.push(id),
+            }
+        }
+
+        let _ = write!(
+            out,
+            " Bookkeeping for table {label}, {stream} stream: window covers ids {window_start}..={highest}, \
+             applied watermark {applied}, {total} id(s) queued in all, \
+             {gap_len} id(s) in the gap ({scanned} scanned), {never} never queued, {queued} queued.",
+            label = self.label,
+            window_start = window_start,
+            highest = ledger.highest_seen,
+            applied = ledger.applied_upto,
+            total = ledger.queued_total,
+            gap_len = next_available - first_missing,
+            scanned = scan_end - first_missing,
+            never = never_queued.len(),
+            queued = queued_not_applied.len(),
+        );
+
+        if first_missing < window_start {
+            let _ = write!(
+                out,
+                " CAUTION: part of the gap ({first_missing}..{window_start}) fell out of the retained window, \
+                 so those ids are unattributable rather than proven missing."
+            );
+        }
+
+        if !never_queued.is_empty() {
+            let _ = write!(out, " ASSIGNED BUT NEVER QUEUED: {}.", format_ids(&never_queued));
+            let _ = write!(
+                out,
+                " Those ids were consumed by the index and their events never reached the persistence queue, \
+                 so nothing will ever deliver them: this is an event leak upstream of the analyzer, \
+                 not a batch collection problem."
+            );
+            let _ = write!(out, "{}", bracketing_producers(ledger, last_applied, next_available));
+        }
+
+        if !queued_not_applied.is_empty() {
+            let _ = write!(out, " QUEUED BUT NOT APPLIED:");
+            for (id, record) in queued_not_applied.iter().take(MAX_LISTED_GAP_IDS) {
+                let _ = write!(
+                    out,
+                    " [{id}: {stages}, collected {collected}x, requeued {requeued}x, {op_type} op {op_id} from {site}]",
+                    stages = record.stages,
+                    collected = record.collected,
+                    requeued = record.requeued,
+                    op_type = OptionDisplay(record.op_type.as_ref().map(|t| format!("{t:?}"))),
+                    op_id = OptionDisplay(record.op_id.as_ref().map(|id| format!("{id:?}"))),
+                    site = OptionDisplay(record.site.map(|site| site.to_string())),
+                );
+            }
+            if queued_not_applied.len() > MAX_LISTED_GAP_IDS {
+                let _ = write!(out, " and {} more", queued_not_applied.len() - MAX_LISTED_GAP_IDS);
+            }
+            let _ = write!(
+                out,
+                ". Those events did reach the queue, so their operations are still somewhere in the analyzer \
+                 and batch collection is failing to assemble them: the bug is in collection, not in event production."
+            );
+        }
+
+        out
+    }
+}
+
+/// Names the producers on either side of the hole.
+///
+/// A leaked id has no record of its own, so the closest evidence about who
+/// should have produced it is who produced its neighbours. One index allocates
+/// its ids from one counter, so the neighbours are almost always the same call
+/// path.
+fn bracketing_producers(ledger: &StreamLedger, last_applied: u64, next_available: u64) -> String {
+    let mut out = String::new();
+    let before = ledger.ids.range(..=last_applied).next_back();
+    let after = ledger.ids.range(next_available..).next();
+    for (side, entry) in [("before the gap", before), ("after the gap", after)] {
+        let Some((id, record)) = entry else {
+            let _ = write!(out, " No record {side}.");
+            continue;
+        };
+        let _ = write!(
+            out,
+            " Producer {side} (id {id}): {op_type}, op {op_id}, pushed from {site}.",
+            op_type = OptionDisplay(record.op_type.as_ref().map(|t| format!("{t:?}"))),
+            op_id = OptionDisplay(record.op_id.as_ref().map(|id| format!("{id:?}"))),
+            site = OptionDisplay(record.site.map(|site| site.to_string())),
+        );
+        if let Some(backtrace) = &record.backtrace {
+            let _ = write!(out, " Backtrace:\n{backtrace}\n");
+        }
+    }
+    if !ledger.ids.values().any(|record| record.backtrace.is_some()) {
+        let _ = write!(
+            out,
+            " Re-run with RUST_BACKTRACE=1 for the full producer backtraces, which this run did not capture."
+        );
+    }
+    out
+}
+
+impl Default for EventLedger {
+    fn default() -> Self {
+        Self::detached()
+    }
+}
+
+struct OptionDisplay(Option<String>);
+
+impl std::fmt::Display for OptionDisplay {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match &self.0 {
+            Some(value) => f.write_str(value),
+            None => f.write_str("<unrecorded>"),
+        }
+    }
+}
+
+/// Renders a sorted id list compactly, collapsing runs.
+fn format_ids(ids: &[u64]) -> String {
+    let mut out = String::new();
+    let mut listed = 0usize;
+    let mut i = 0usize;
+    while i < ids.len() && listed < MAX_LISTED_GAP_IDS {
+        let start = ids[i];
+        let mut end = start;
+        while i + 1 < ids.len() && ids[i + 1] == end + 1 {
+            i += 1;
+            end = ids[i];
+        }
+        if !out.is_empty() {
+            out.push_str(", ");
+        }
+        if start == end {
+            let _ = write!(out, "{start}");
+        } else {
+            let _ = write!(out, "{start}..={end}");
+        }
+        listed += 1;
+        i += 1;
+    }
+    if i < ids.len() {
+        let _ = write!(out, ", and {} more", ids.len() - i);
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn insert_at(id: u64) -> ChangeEvent<Pair<u64, Link>> {
+        ChangeEvent::InsertAt {
+            event_id: id.into(),
+            max_value: Pair {
+                key: id,
+                value: Link::default(),
+            },
+            value: Pair {
+                key: id,
+                value: Link::default(),
+            },
+            index: 0,
+        }
+    }
+
+    #[track_caller]
+    fn queue(ledger: &EventLedger, ids: &[u64], op_type: OperationType) {
+        let evs = ids.iter().copied().map(insert_at).collect::<Vec<_>>();
+        let ids = EventLedger::event_ids(&evs);
+        ledger.record_queued(
+            EventStream::Primary,
+            &ids,
+            OperationId::Single(uuid::Uuid::from_u128(1)),
+            op_type,
+            Location::caller(),
+        );
+    }
+
+    /// These tests assert on what the ledger recorded, so they are only
+    /// meaningful where it records. That is every normal test run
+    /// (`debug_assertions`); a `--release` test run skips them rather than
+    /// failing on a report that correctly says bookkeeping was off.
+    fn recording() -> bool {
+        enabled()
+    }
+
+    #[test]
+    fn names_ids_that_were_never_queued() {
+        if !recording() {
+            return;
+        }
+        let ledger = EventLedger::new("test/table");
+        queue(&ledger, &[1, 2, 3], OperationType::Insert);
+        // 4 and 5 are assigned by the index and leaked: nothing queues them.
+        queue(&ledger, &[6, 7], OperationType::Update);
+
+        let report = ledger.gap_report(&EventStream::Primary, 3, 6);
+
+        assert!(report.contains("ASSIGNED BUT NEVER QUEUED"), "{report}");
+        assert!(report.contains("4..=5"), "{report}");
+        assert!(report.contains("event leak upstream of the analyzer"), "{report}");
+    }
+
+    #[test]
+    fn distinguishes_a_queued_id_from_a_leaked_one() {
+        if !recording() {
+            return;
+        }
+        let ledger = EventLedger::new("test/table");
+        queue(&ledger, &[1, 2, 3], OperationType::Insert);
+        queue(&ledger, &[4], OperationType::Update);
+        queue(&ledger, &[6], OperationType::Insert);
+
+        let report = ledger.gap_report(&EventStream::Primary, 3, 6);
+
+        assert!(report.contains("QUEUED BUT NOT APPLIED"), "{report}");
+        assert!(report.contains("the bug is in collection"), "{report}");
+        // Only id 5 is missing; 4 was queued.
+        assert!(report.contains("ASSIGNED BUT NEVER QUEUED: 5."), "{report}");
+    }
+
+    #[test]
+    fn reports_a_gapless_stream_as_nothing() {
+        if !recording() {
+            return;
+        }
+        let ledger = EventLedger::new("test/table");
+        queue(&ledger, &[1, 2, 3], OperationType::Insert);
+        assert!(ledger.gap_report(&EventStream::Primary, 3, 4).is_empty());
+    }
+
+    #[test]
+    fn window_eviction_is_reported_rather_than_guessed() {
+        if !recording() {
+            return;
+        }
+        let ledger = EventLedger::new("test/table");
+        let ids = (1..=(WINDOW as u64 + 64)).collect::<Vec<_>>();
+        queue(&ledger, &ids, OperationType::Insert);
+
+        // Ask about a gap far below the retained window.
+        let report = ledger.gap_report(&EventStream::Primary, 1, 20);
+        assert!(report.contains("fell out of the retained window"), "{report}");
+    }
+
+    #[test]
+    fn applied_watermark_marks_everything_behind_it() {
+        if !recording() {
+            return;
+        }
+        let ledger = EventLedger::new("test/table");
+        queue(&ledger, &[1, 2, 3, 5], OperationType::Insert);
+        ledger.record_applied_upto(EventStream::Primary, 3);
+
+        let report = ledger.gap_report(&EventStream::Primary, 3, 5);
+        assert!(report.contains("ASSIGNED BUT NEVER QUEUED: 4."), "{report}");
+    }
+
+    #[test]
+    fn stages_render_every_flag_set() {
+        let mut stages = Stages::default();
+        assert_eq!(stages.to_string(), "none");
+        stages.insert(Stages::QUEUED);
+        stages.insert(Stages::TRIMMED);
+        assert_eq!(stages.to_string(), "queued+trimmed");
+    }
+}
diff --git a/src/persistence/mod.rs b/src/persistence/mod.rs
index b5e03ce3..b5ddb33a 100644
--- a/src/persistence/mod.rs
+++ b/src/persistence/mod.rs
@@ -10,6 +10,7 @@ pub use error::{
     PersistenceError, PersistenceIndexCorruption, PersistenceLoadError, PersistenceResult, PersistenceState,
     load_persisted_state,
 };
+pub use event_ledger::{EventLedger, EventStream, Stages};
 pub use operation::{
     AcknowledgeOperation, DeleteOperation, InsertOperation, Operation, OperationId, OperationType, UpdateOperation,
     validate_events,
@@ -91,6 +92,7 @@ impl<T: 'static> std::error::Error for UnloadFailure<T> {}
 
 mod engine;
 mod error;
+pub mod event_ledger;
 pub mod operation;
 mod readonly_engine;
 mod space;
diff --git a/src/persistence/operation/batch.rs b/src/persistence/operation/batch.rs
index 958bcfc0..ea49a1d8 100644
--- a/src/persistence/operation/batch.rs
+++ b/src/persistence/operation/batch.rs
@@ -2,6 +2,7 @@ use std::collections::HashMap;
 use std::fmt::Debug;
 use std::hash::Hash;
 use std::marker::PhantomData;
+use std::sync::Arc;
 
 use data_bucket::page::PageId;
 use data_bucket::{Link, SizeMeasurable};
@@ -10,6 +11,7 @@ use indexset::core::pair::Pair;
 use worktable_codegen::{MemStat, worktable};
 
 use crate::persistence::OperationType;
+use crate::persistence::event_ledger::{self, EventLedger, EventStream, Stages};
 use crate::persistence::space::{BatchChangeEvent, BatchData};
 use crate::persistence::task::{LastEventIds, QueueInnerRow};
 use crate::prelude::*;
@@ -149,6 +151,10 @@ fn latest_data_writes<PrimaryKeyGenState, PrimaryKey, SecondaryEvents>(
 pub struct BatchOperation<PrimaryKeyGenState, PrimaryKey, SecondaryEvents, AvailableIndexes> {
     ops: Vec<Operation<PrimaryKeyGenState, PrimaryKey, SecondaryEvents>>,
     info_wt: BatchInnerWorkTable,
+    /// Event bookkeeping shared with the queue that produced `ops`, read by
+    /// the event-gap guard in `validate` so a stall names its own cause.
+    /// Diagnostics only, and `None` for batches built outside the analyzer.
+    event_ledger: Option<Arc<EventLedger>>,
     prepared_index_evs: Option<PreparedIndexEvents<PrimaryKey, SecondaryEvents>>,
     phantom_data: PhantomData<AvailableIndexes>,
 }
@@ -173,11 +179,23 @@ where
         Self {
             ops,
             info_wt,
+            event_ledger: None,
             prepared_index_evs: None,
             phantom_data: PhantomData,
         }
     }
 
+    /// Attaches the analyzer's event bookkeeping, so the gap guard below can
+    /// say which ids in a gap ever reached the persistence queue.
+    ///
+    /// A builder method rather than a `new` parameter, so every existing
+    /// caller of `new` keeps working unchanged and the batch stays usable
+    /// without any bookkeeping at all.
+    pub fn with_event_ledger(mut self, ledger: Arc<EventLedger>) -> Self {
+        self.event_ledger = Some(ledger);
+        self
+    }
+
     /// Remove metadata immediately after `self.ops.remove(removed_pos)`.
     ///
     /// At entry, `self.ops.len()` is already one shorter while `info_wt` still
@@ -272,9 +290,54 @@ where
             prepared_evs.secondary_evs.remove(op_secondary);
         }
 
+        self.record_stage(&removed_ops, Stages::TRIMMED);
+
         Ok(removed_ops)
     }
 
+    /// Records `stage` against every event id carried by `ops`.
+    ///
+    /// Diagnostics only. Skipped entirely when bookkeeping is off, which keeps
+    /// the `Debug` formatting of secondary index labels out of release builds.
+    fn record_stage(&self, ops: &[Operation<PrimaryKeyGenState, PrimaryKey, SecondaryEvents>], stage: Stages) {
+        let Some(ledger) = &self.event_ledger else {
+            return;
+        };
+        if !event_ledger::enabled() {
+            return;
+        }
+        for op in ops {
+            if let Some(evs) = op.primary_key_events() {
+                ledger.record_stage_for_events(EventStream::Primary, evs, stage);
+            }
+            // See the matching note in `QueueAnalyzer::record_ops_stage`: the
+            // producer side records primary ids only, so an id observed on a
+            // secondary stream here is known to have been queued.
+            for (index, id) in op.secondary_key_events().iter_event_ids() {
+                ledger.record_stage(
+                    EventStream::Secondary(format!("{index:?}")),
+                    id.inner(),
+                    stage.union(Stages::QUEUED),
+                );
+            }
+        }
+    }
+
+    /// The bookkeeping's account of a gap, or a note saying there is none.
+    fn gap_report(
+        &self,
+        stream: &EventStream,
+        last_applied: IndexChangeEventId,
+        next_available: IndexChangeEventId,
+    ) -> String {
+        match &self.event_ledger {
+            Some(ledger) => ledger.gap_report(stream, last_applied.inner(), next_available.inner()),
+            None => {
+                " This batch was built without event bookkeeping attached, so the gap cannot be attributed.".to_owned()
+            }
+        }
+    }
+
     pub fn get_last_event_ids(&self) -> LastEventIds<AvailableIndexes> {
         let prepared_evs = self
             .prepared_index_evs
@@ -358,8 +421,9 @@ where
                 // that persists is a bug upstream of the analyzer; report it
                 // loudly instead of force-applying and corrupting the file.
                 if attempts > GIVE_UP_AFTER_ATTEMPTS {
+                    let report = self.gap_report(&EventStream::Primary, last_ids.primary_id, id);
                     return Err(eyre::eyre!(
-                        "persistence stalled on primary index event gap: last applied {:?}, next available {:?} after {attempts} attempts, with {} operations queued. Every one of them was collected and the stream is still gapped, so the operation carrying the missing id never reached the queue: an event id was consumed without its event being pushed. The producer is upstream of the analyzer, not here.",
+                        "persistence stalled on primary index event gap: last applied {:?}, next available {:?} after {attempts} attempts, with {} operations queued.{report}",
                         last_ids.primary_id,
                         id,
                         self.ops.len()
@@ -381,8 +445,9 @@ where
                     // stream, defer until the missing event arrives, and report
                     // a persistent gap as the bug it is.
                     if attempts > GIVE_UP_AFTER_ATTEMPTS {
+                        let report = self.gap_report(&EventStream::Secondary(format!("{index:?}")), *last, id);
                         return Err(eyre::eyre!(
-                            "persistence stalled on secondary index {index:?} event gap: last applied {last:?}, next available {id:?} after {attempts} attempts, with {} operations queued. All of them were collected and the stream is still gapped, so the operation carrying the missing id never reached the queue.",
+                            "persistence stalled on secondary index {index:?} event gap: last applied {last:?}, next available {id:?} after {attempts} attempts, with {} operations queued.{report}",
                             self.ops.len()
                         ));
                     }
diff --git a/src/persistence/task.rs b/src/persistence/task.rs
index fc67f09f..edba8f5c 100644
--- a/src/persistence/task.rs
+++ b/src/persistence/task.rs
@@ -2,6 +2,7 @@ use std::collections::{HashMap, HashSet, VecDeque};
 use std::fmt::Debug;
 use std::hash::Hash;
 use std::marker::PhantomData;
+use std::panic::Location;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::time::Duration;
@@ -12,7 +13,8 @@ use tokio::sync::Notify;
 use tokio::task::JoinHandle;
 use worktable_codegen::worktable;
 
-use crate::persistence::operation::{BatchInnerRow, BatchInnerWorkTable, BatchOperation, OperationId};
+use crate::persistence::event_ledger::{self, EventLedger, EventStream, Stages};
+use crate::persistence::operation::{BatchInnerRow, BatchInnerWorkTable, BatchOperation, OperationId, OperationType};
 use crate::persistence::{
     PersistenceEngine, PersistenceError, PersistenceIndexCorruption, PersistenceResult, PersistenceState,
 };
@@ -207,6 +209,12 @@ pub struct QueueAnalyzer<PrimaryKeyGenState, PrimaryKey, SecondaryKeys, Availabl
     /// collection off `attempts` therefore never happened in exactly the case
     /// that needed it. Progress is what should widen the search.
     no_progress: usize,
+    /// Shared with the queue that feeds this analyzer and with every
+    /// `BatchOperation` it builds, so the event-gap guard can say which ids in
+    /// a gap ever reached the queue. Diagnostics only: see
+    /// [`crate::persistence::event_ledger`]. Detached until
+    /// `attach_event_ledger` is called, which unit-test analyzers never do.
+    event_ledger: Arc<EventLedger>,
 }
 
 #[derive(Debug)]
@@ -260,9 +268,19 @@ where
             page_limit: MAX_PAGE_AMOUNT,
             attempts: 0,
             no_progress: 0,
+            event_ledger: Arc::new(EventLedger::detached()),
         }
     }
 
+    /// Shares the feeding queue's event bookkeeping with this analyzer.
+    ///
+    /// Only the producer side records who pushed an event, so the analyzer has
+    /// to read the *same* ledger the queue writes for its gap reports to mean
+    /// anything.
+    pub fn attach_event_ledger(&mut self, ledger: Arc<EventLedger>) {
+        self.event_ledger = ledger;
+    }
+
     pub fn push(&mut self, value: Operation<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>) -> eyre::Result<()> {
         let link = value.link();
         let mut row = QueueInnerRow {
@@ -301,6 +319,39 @@ where
             .map(|(id, _)| id)
     }
 
+    /// Records `stage` against every event id carried by `ops`.
+    ///
+    /// Diagnostics only. The whole body is skipped when bookkeeping is off,
+    /// which keeps the `Debug` formatting of secondary index labels off the
+    /// path of a release build entirely.
+    fn record_ops_stage(&self, ops: &[Operation<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>], stage: Stages)
+    where
+        SecondaryKeys: TableSecondaryIndexEventsOps<AvailableIndexes>,
+    {
+        if !event_ledger::enabled() {
+            return;
+        }
+        for op in ops {
+            if let Some(evs) = op.primary_key_events() {
+                self.event_ledger
+                    .record_stage_for_events(EventStream::Primary, evs, stage);
+            }
+            // `Stages::QUEUED` is unioned in for secondary streams because the
+            // producer side records primary ids only: an operation reaching
+            // the analyzer at all proves it was queued, and without this the
+            // secondary gap report would call every id it knows about leaked.
+            // The cost is that a secondary record carries no producer call
+            // site, which the report prints as `<unrecorded>`.
+            for (index, id) in op.secondary_key_events().iter_event_ids() {
+                self.event_ledger.record_stage(
+                    EventStream::Secondary(format!("{index:?}")),
+                    id.inner(),
+                    stage.union(Stages::QUEUED),
+                );
+            }
+        }
+    }
+
     pub async fn collect_batch_from_op_id(
         &mut self,
         op_id: OperationId,
@@ -440,13 +491,23 @@ where
             ops.push(op);
         }
 
-        let mut op = BatchOperation::new(ops, info_wt);
+        self.record_ops_stage(&ops, Stages::COLLECTED);
+        let mut op = BatchOperation::new(ops, info_wt).with_event_ledger(self.event_ledger.clone());
         let invalid_for_this_batch_ops = op.validate(&self.last_events_ids, self.attempts).await?;
         if let Some(invalid_for_this_batch_ops) = invalid_for_this_batch_ops {
+            self.record_ops_stage(&invalid_for_this_batch_ops, Stages::REQUEUED);
             self.extend_from_iter(invalid_for_this_batch_ops.into_iter())?;
             let previous_primary = self.last_events_ids.primary_id;
             let last_ids = op.get_last_event_ids();
             let advanced = last_ids.primary_id > previous_primary;
+            self.event_ledger
+                .record_applied_upto(EventStream::Primary, last_ids.primary_id.inner());
+            if event_ledger::enabled() {
+                for (index, id) in &last_ids.secondary_ids {
+                    self.event_ledger
+                        .record_applied_upto(EventStream::Secondary(format!("{index:?}")), id.inner());
+                }
+            }
             self.last_events_ids.merge(last_ids);
             self.last_invalid_batch_size = 0;
             self.page_limit = MAX_PAGE_AMOUNT;
@@ -461,6 +522,7 @@ where
         } else {
             // can't collect batch for now
             let ops = op.ops();
+            self.record_ops_stage(&ops, Stages::REQUEUED);
             self.attempts += 1;
             self.no_progress += 1;
             if self.last_invalid_batch_size == ops.len() {
@@ -918,7 +980,7 @@ mod lifecycle_tests {
     #[tokio::test]
     async fn wake_landing_inside_the_pop_race_window_is_not_lost() {
         let lifecycle = Arc::new(PersistenceLifecycle::new());
-        let mut queue = Queue::<(), u64, TestEvents>::new(lifecycle.clone());
+        let mut queue = Queue::<(), u64, TestEvents>::new(lifecycle.clone(), "tests/queue");
         let gate = Arc::new(PopRaceWindowGate::new());
         queue.pop_race_window_gate = Some(gate.clone());
         let queue = Arc::new(queue);
@@ -1148,6 +1210,41 @@ impl PopRaceWindowGate {
     }
 }
 
+/// Primary index event ids lifted off an operation before it is moved into the
+/// queue, so they can be recorded once the push is known to have been accepted.
+///
+/// Primary only: `Queue` is generic over the secondary event type with no bound
+/// that could iterate it, and the primary stream is the one whose gap guard
+/// stalls the engine. The analyzer records secondary ids at collection time,
+/// where that bound does exist.
+///
+/// Cheap when bookkeeping is off: `EventLedger::event_ids` returns an empty
+/// `Vec`, which allocates nothing, and the rest is two `Copy` field reads.
+struct QueuedEventIds {
+    ids: Vec<u64>,
+    op_id: OperationId,
+    op_type: OperationType,
+}
+
+impl QueuedEventIds {
+    fn of<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>(
+        value: &Operation<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>,
+    ) -> Self {
+        Self {
+            ids: value
+                .primary_key_events()
+                .map(|evs| EventLedger::event_ids(evs.as_slice()))
+                .unwrap_or_default(),
+            op_id: value.operation_id(),
+            op_type: value.operation_type(),
+        }
+    }
+
+    fn record(&self, ledger: &EventLedger, site: &'static Location<'static>) {
+        ledger.record_queued(EventStream::Primary, &self.ids, self.op_id, self.op_type, site);
+    }
+}
+
 #[derive(Debug)]
 pub struct Queue<PrimaryKeyGenState, PrimaryKey, SecondaryKeys> {
     // Not `lockfree::queue::Queue`: its `Removable::empty` materializes the
@@ -1161,38 +1258,71 @@ pub struct Queue<PrimaryKeyGenState, PrimaryKey, SecondaryKeys> {
     // queue that still holds work.
     len: Arc<AtomicUsize>,
     lifecycle: Arc<PersistenceLifecycle>,
+    /// Producer-side half of the event-gap bookkeeping: every operation that
+    /// reaches persistence passes through this queue, so an event id the
+    /// engine is waiting for that never appears here was assigned by the index
+    /// and dropped before it was queued. Shared with the analyzer, which reads
+    /// it when the gap guard fires. Diagnostics only, and inert unless
+    /// [`event_ledger::enabled`].
+    event_ledger: Arc<EventLedger>,
     #[cfg(test)]
     pop_race_window_gate: Option<std::sync::Arc<PopRaceWindowGate>>,
 }
 
 impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys> Queue<PrimaryKeyGenState, PrimaryKey, SecondaryKeys> {
-    fn new(lifecycle: Arc<PersistenceLifecycle>) -> Self {
+    fn new(lifecycle: Arc<PersistenceLifecycle>, table_path: &str) -> Self {
         Self {
             queue: ParkingMutex::new(VecDeque::new()),
             notify: Notify::new(),
             len: Arc::new(AtomicUsize::new(0)),
             lifecycle,
+            event_ledger: Arc::new(EventLedger::new(table_path)),
             #[cfg(test)]
             pop_race_window_gate: None,
         }
     }
 
-    pub fn push(&self, value: Operation<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>) -> PersistenceResult {
-        self.push_message(PersistenceMessage::Operation(value))
+    /// The event bookkeeping this queue writes, for sharing with the analyzer.
+    pub fn event_ledger(&self) -> Arc<EventLedger> {
+        self.event_ledger.clone()
+    }
+
+    /// Enqueues one operation, naming the producer's call site.
+    ///
+    /// The site is passed explicitly rather than taken with `#[track_caller]`,
+    /// because that only reaches one frame up: a wrapper that wants its own
+    /// caller named in a gap report has to forward a location through here.
+    /// through here rather than call `push` and lose it.
+    pub fn push_at(
+        &self,
+        value: Operation<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>,
+        site: &'static Location<'static>,
+    ) -> PersistenceResult {
+        // The ids have to be lifted out before the operation is moved into the
+        // queue, but they are only recorded once the push is accepted: a
+        // refused push (the engine is closing or already failed) genuinely
+        // does not queue its events, and recording it as queued would hide
+        // exactly that leak mode.
+        let queued = QueuedEventIds::of(&value);
+        self.push_message(PersistenceMessage::Operation(value))?;
+        queued.record(&self.event_ledger, site);
+        Ok(())
     }
 
-    /// Enqueues a whole batch of operations under one lifecycle check, one
-    /// queue lock acquisition and one worker wake-up, so callers producing
-    /// many operations at once (`insert_many`) pay the intake overhead once
-    /// instead of per row. All-or-nothing: either every operation is accepted
-    /// or none is.
-    pub fn push_many(
+    /// Enqueues a whole batch under one lifecycle check, one queue lock and one
+    /// worker wake-up, so a caller producing many operations at once
+    /// (`insert_many`) pays the intake overhead once instead of per row.
+    /// All-or-nothing: either every operation is accepted or none is. Takes the
+    /// producer's call site for the same reason as [`Queue::push_at`].
+    pub fn push_many_at(
         &self,
         values: Vec<Operation<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>>,
+        site: &'static Location<'static>,
     ) -> PersistenceResult {
         if values.is_empty() {
             return Ok(());
         }
+        let queued = values.iter().map(QueuedEventIds::of).collect::<Vec<_>>();
         let state = self.lifecycle.state.lock();
         match &*state {
             PersistenceState::Running => {}
@@ -1205,6 +1335,11 @@ impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys> Queue<PrimaryKeyGenState, Pr
             .lock()
             .extend(values.into_iter().map(PersistenceMessage::Operation));
         self.notify.notify_one();
+        drop(state);
+        // Recorded after acceptance, for the reason given in `push_at`.
+        for queued in &queued {
+            queued.record(&self.event_ledger, site);
+        }
         Ok(())
     }
 
@@ -1309,13 +1444,21 @@ where
         primary_key_events: Vec<IndexChangeEvent<IndexPair<PrimaryKey, Link>>>,
         secondary_keys_events: SecondaryKeys,
     ) -> PersistenceResult {
-        self.push(Operation::Update(UpdateOperation {
-            id: OperationId::Single(uuid::Uuid::now_v7()),
-            primary_key_events,
-            secondary_keys_events,
-            bytes,
-            link: new_link,
-        }))
+        // `Location::caller()` without `#[track_caller]` resolves to this line
+        // rather than to vacuum's call site. That is deliberate: the trait
+        // declaration lives outside this module and cannot be annotated, and
+        // this line is already a unique producer label, because `apply_move`
+        // is only ever reached from a vacuum row move.
+        self.push_at(
+            Operation::Update(UpdateOperation {
+                id: OperationId::Single(uuid::Uuid::now_v7()),
+                primary_key_events,
+                secondary_keys_events,
+                bytes,
+                link: new_link,
+            }),
+            Location::caller(),
+        )
     }
 
     fn reclaim_pages(&self, page_ids: Vec<PageId>) -> PersistenceResult {
@@ -1391,17 +1534,21 @@ impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys, AvailableIndexes> Drop
 impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys, AvailableIndexes>
     PersistenceTask<PrimaryKeyGenState, PrimaryKey, SecondaryKeys, AvailableIndexes>
 {
+    /// `#[track_caller]` so an event-gap report names the producer that
+    /// pushed the operation rather than this forwarding line.
+    #[track_caller]
     pub fn apply_operation(&self, op: Operation<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>) -> PersistenceResult {
-        self.queue.push(op)
+        self.queue.push_at(op, Location::caller())
     }
 
     /// Enqueues a batch of operations atomically with a single worker
     /// wake-up. See [`Queue::push_many`].
+    #[track_caller]
     pub fn apply_operations(
         &self,
         ops: Vec<Operation<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>>,
     ) -> PersistenceResult {
-        self.queue.push_many(ops)
+        self.queue.push_many_at(ops, Location::caller())
     }
 
     pub fn ensure_running(&self) -> PersistenceResult {
@@ -1448,12 +1595,15 @@ impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys, AvailableIndexes>
     {
         let table_path = engine.config().table_path().to_owned();
         let lifecycle = Arc::new(PersistenceLifecycle::new());
-        let queue = Arc::new(Queue::new(lifecycle.clone()));
+        let queue = Arc::new(Queue::new(lifecycle.clone(), &table_path));
 
         let engine_queue = queue.clone();
         let engine_lifecycle = lifecycle.clone();
         let analyzer_inner_wt: Arc<QueueInnerWorkTable> = Default::default();
         let mut analyzer = QueueAnalyzer::new(analyzer_inner_wt.clone());
+        // Producer and consumer must share one ledger: the queue records who
+        // pushed an event, the analyzer's gap guard reads it back.
+        analyzer.attach_event_ledger(queue.event_ledger());
         let analyzer_in_progress = Arc::new(AtomicBool::new(true));
         let task_analyzer_in_progress = analyzer_in_progress.clone();
 

From f6447e9ac081d9ff64c6dbfdea46d7a93b460f2f Mon Sep 17 00:00:00 2001
From: meh <no-reply@pathscale.com>
Date: Tue, 8 Sep 2026 02:41:43 +0700
Subject: [PATCH 4/4] Acknowledge orphaned index events before propagating the
 error

The ledger from the previous commit points at three sites in generated
persisted query code where an event the index had already assigned an id to
was dropped instead of queued. A dropped event is a hole
`BatchOperation::validate` refuses forever, so the table stalls and the
message names a range rather than a cause.

All three now do what the rollback arms already did: build an `Acknowledge`
carrying the orphaned events and apply it before propagating. The
`IndexError::NotFound` arm of the update insert path acknowledges the events
its sibling `AlreadyExists` arm was already acknowledging, and the two bare
`res?` sites, in `gen_process_diffs_remove_on_index` and after
`delete_row_cdc`, became `if let Err(e) = res` blocks that acknowledge and
then return.

The events are moved into the acknowledge rather than cloned, and that is
load-bearing rather than a preference. Cloning does not compile: the events
type is still an inference variable at that point, pinned only by the
`op.extend_secondary_key_events` call further down, and method resolution for
`.clone()` needs the type known where the call is written. It fails as an
`E0282` reported against the `worktable!` invocation with no inner span, which
is expensive to diagnose. The comment in `docs/TODO.md` records that so nobody
pays for it twice.

The write failures are not forcible through the public API, so the wiring is
pinned on the emitted tokens the way the surrounding tests already do it. Both
assertions check the acknowledge is emitted *before* the return or the extend,
not merely that one appears somewhere in the output.

The in-memory generator has the same two `NotFound` arms and is deliberately
untouched: there is no persistence stream behind them to gap.
---
 .../src/generators/persist/queries/delete.rs  | 34 +++++++-
 .../src/generators/persist/queries/update.rs  | 84 ++++++++++++++++++-
 docs/TODO.md                                  | 34 ++++++--
 3 files changed, 144 insertions(+), 8 deletions(-)

diff --git a/codegen/src/generators/persist/queries/delete.rs b/codegen/src/generators/persist/queries/delete.rs
index c2efa53c..ebe4ea45 100644
--- a/codegen/src/generators/persist/queries/delete.rs
+++ b/codegen/src/generators/persist/queries/delete.rs
@@ -98,7 +98,23 @@ impl PersistGenerator {
                     row,
                     link,
                 );
-            res?;
+            // `delete_row_cdc` produces events whether or not it succeeds, and
+            // the index has already assigned their ids. Propagating the error
+            // without queueing them leaves a hole the persistence stream can
+            // never fill, exactly as the restore path below is careful not to.
+            if let core::result::Result::Err(e) = res {
+                let ack_op: Operation<
+                    <<#pk_ident as TablePrimaryKey>::Generator as PrimaryKeyGeneratorState>::State,
+                    #pk_ident,
+                    #secondary_events_ident
+                > = Operation::Acknowledge(AcknowledgeOperation {
+                    id: OperationId::Single(uuid::Uuid::now_v7()),
+                    primary_key_events: vec![],
+                    secondary_keys_events,
+                });
+                self.1.apply_operation(ack_op)?;
+                return core::result::Result::Err(e.into());
+            }
             let (_, primary_key_events) = self.0.primary_index.remove_cdc(pk.clone(), link);
             if let core::result::Result::Err(e) = self.0.data.delete(link) {
                 let mut secondary_keys_events = secondary_keys_events;
@@ -387,5 +403,21 @@ mod tests {
             emitted.contains("Operation :: Acknowledge"),
             "acknowledge op missing:\n{emitted}"
         );
+
+        // A failed secondary removal used to propagate through a bare `res?`,
+        // dropping the events `delete_row_cdc` had already produced. Their ids
+        // are assigned when the index produces them, so the persistence stream
+        // gapped permanently and the stall named a range rather than a cause.
+        let secondary = emitted.find("delete_row_cdc").expect("secondary removal emitted");
+        let tail = &emitted[secondary..];
+        let ack = tail
+            .find("Operation :: Acknowledge")
+            .expect("a failed secondary removal must acknowledge its events");
+        assert!(
+            ack < tail
+                .find("remove_cdc (pk . clone () , link)")
+                .expect("primary removal emitted"),
+            "the secondary removal propagates before acknowledging, which gaps the stream:\n{emitted}"
+        );
     }
 }
diff --git a/codegen/src/generators/persist/queries/update.rs b/codegen/src/generators/persist/queries/update.rs
index ffd41658..380df89d 100644
--- a/codegen/src/generators/persist/queries/update.rs
+++ b/codegen/src/generators/persist/queries/update.rs
@@ -501,6 +501,7 @@ impl PersistGenerator {
     fn gen_process_diffs_insert_on_index(&self, idents: &[Ident], idx_idents: Option<&Vec<Ident>>) -> TokenStream {
         let name_generator = WorktableNameGenerator::from_table_name(self.name.to_string());
         let avt_type_ident = name_generator.get_available_type_ident();
+        let pk_ident = name_generator.get_primary_key_type_ident();
         // `updated_bytes` is bound by gen_data_write_and_fetch, which captures
         // the real row bytes right after the data write.
         let diff_container = if idx_idents.is_some() {
@@ -567,7 +568,27 @@ impl PersistGenerator {
 
                                 Err(WorkTableError::AlreadyExists(at.to_string_value()))
                             }
-                            IndexError::NotFound => Err(WorkTableError::NotFound),
+                            IndexError::NotFound => {
+                                // The insert side produced events before it
+                                // failed, and the index has already assigned
+                                // their ids. Returning without queueing them
+                                // leaves a hole the persistence stream can
+                                // never fill, which is what the sibling arm
+                                // above avoids and what this arm used to
+                                // cause.
+                                let ack_op: Operation<
+                                    <<#pk_ident as TablePrimaryKey>::Generator as PrimaryKeyGeneratorState>::State,
+                                    #pk_ident,
+                                    #secondary_events_ident
+                                > = Operation::Acknowledge(AcknowledgeOperation {
+                                    id: OperationId::Single(uuid::Uuid::now_v7()),
+                                    primary_key_events: vec![],
+                                    secondary_keys_events: secondary_events.clone(),
+                                });
+                                self.1.apply_operation(ack_op)?;
+
+                                Err(WorkTableError::NotFound)
+                            }
                         };
                     }
                     let mut secondary_keys_events = secondary_events;
@@ -587,10 +608,29 @@ impl PersistGenerator {
     }
 
     fn gen_process_diffs_remove_on_index(&self, idx_idents: Option<&Vec<Ident>>) -> TokenStream {
+        let name_generator = WorktableNameGenerator::from_table_name(self.name.to_string());
+        let pk_ident = name_generator.get_primary_key_type_ident();
+        let secondary_events_ident = name_generator.get_space_secondary_index_events_ident();
         if idx_idents.is_some() {
             quote! {
                 let (secondary_keys_events_remove, res) = self.0.indexes.process_difference_remove_cdc(link, diffs);
-                res?;
+                // The removal produced events whether or not it succeeded, and
+                // their ids are already assigned. Propagating the error without
+                // queueing them gaps the stream permanently, so acknowledge
+                // them first and then propagate unchanged.
+                if let core::result::Result::Err(e) = res {
+                    let ack_op: Operation<
+                        <<#pk_ident as TablePrimaryKey>::Generator as PrimaryKeyGeneratorState>::State,
+                        #pk_ident,
+                        #secondary_events_ident
+                    > = Operation::Acknowledge(AcknowledgeOperation {
+                        id: OperationId::Single(uuid::Uuid::now_v7()),
+                        primary_key_events: vec![],
+                        secondary_keys_events: secondary_keys_events_remove,
+                    });
+                    self.1.apply_operation(ack_op)?;
+                    return core::result::Result::Err(e.into());
+                }
                 op.extend_secondary_key_events(secondary_keys_events_remove);
             }
         } else {
@@ -1143,5 +1183,45 @@ mod tests {
             .find("Operation :: Update (UpdateOperation")
             .expect("update op emitted");
         assert!(insert < write && write < op_build, "emission order broken:\n{emitted}");
+
+        // Every event the index produced must reach the persistence stream,
+        // including on the paths that fail. The index assigns an event id at
+        // the moment it produces the event, so a path that returns without
+        // queueing one leaves a hole `BatchOperation::validate` will refuse
+        // forever, and the stall it causes names a range rather than a cause.
+        //
+        // The `NotFound` arm used to be exactly that: its sibling
+        // `AlreadyExists` arm built an Acknowledge and it did not.
+        let not_found = emitted
+            .find("IndexError :: NotFound =>")
+            .expect("not-found arm emitted");
+        let tail = &emitted[not_found..];
+        let ack = tail
+            .find("Operation :: Acknowledge")
+            .expect("not-found arm must acknowledge its events");
+        let returns = tail
+            .find("Err (WorkTableError :: NotFound)")
+            .expect("not-found arm returns");
+        assert!(
+            ack < returns,
+            "the not-found arm returns before acknowledging, which gaps the stream:\n{emitted}"
+        );
+
+        // Same for the removal side, where the events were dropped by a bare
+        // `res?` before the extend that would have carried them.
+        let removal = emitted
+            .find("process_difference_remove_cdc (link , diffs)")
+            .expect("old-key removal emitted");
+        let tail = &emitted[removal..];
+        let ack = tail
+            .find("Operation :: Acknowledge")
+            .expect("a failed removal must acknowledge its events");
+        let extend = tail
+            .find("op . extend_secondary_key_events")
+            .expect("successful removal extends the operation");
+        assert!(
+            ack < extend,
+            "a failed removal propagates before acknowledging, which gaps the stream:\n{emitted}"
+        );
     }
 }
diff --git a/docs/TODO.md b/docs/TODO.md
index 8b6f79fb..6e8d0a85 100644
--- a/docs/TODO.md
+++ b/docs/TODO.md
@@ -140,7 +140,7 @@ not be created outside the thread and tree that uses it. Nothing in either
 repository does; every threaded test builds its guard inside the spawned
 closure.
 
-### Persistence event gap: three leak sites found, instrumented not yet fixed
+### Persistence event gap: three leak sites found, instrumented, and fixed
 
 Updated 2026-09-07. This section previously said the cause was unknown and that
 the next step was instrumentation rather than a repro hunt. The instrumentation
@@ -192,12 +192,36 @@ Always compiled, gated at run time on `debug_assertions` or `WT_EVENT_LEDGER`,
 which puts it on exactly where the bug appears, since the stall needs a full
 debug `--all-features` run.
 
-Not fixed, and not compiled. The three sites need the same treatment the
-rollback arms already use: build an `Acknowledge` carrying the orphaned events
-before propagating the error.
+**Fixed 2026-09-08.** All three sites now do what the rollback arms already did:
+build an `Acknowledge` carrying the orphaned events and apply it before
+propagating the error. The two `res?` sites became `if let Err(e) = res` so the
+events are moved into the acknowledge and the error is returned explicitly, and
+the `NotFound` arm acknowledges the events its sibling arm was already
+acknowledging.
+
+The events are **moved** into the acknowledge rather than cloned, and that is
+load-bearing rather than tidy. Cloning them fails to compile: the events type is
+still an inference variable at that point in the generated code, pinned only by
+the `op.extend_secondary_key_events` call further down, and method resolution for
+`.clone()` needs the type resolved where the call is written. The result is an
+`E0282` reported against the `worktable!` invocation with no inner span, which is
+an expensive thing to diagnose twice.
+
+Covered by emitted-token assertions in both generators
+(`indexed_update_write_failure_unwinds_and_acknowledges` and
+`delete_data_failure_restores_indexes_and_acknowledges`), which assert the
+acknowledge is emitted **before** the return or the extend rather than merely
+present somewhere in the output. The write failure itself is not forcible through
+the public API, so the wiring is pinned on the tokens, which is the same approach
+those tests already took.
+
+The in-memory generator has the same two `NotFound` arms
+(`codegen/src/generators/in_memory/queries/update.rs:528` and `552`) and they are
+correctly untouched: there is no persistence stream behind them to gap.
 ## Housekeeping
 
-- `CHANGELOG.md` stops at 0.4.1, long before the 1.0.0-beta line.
+- `CHANGELOG.md` was backfilled to 0.3.10 on 2026-09-08. It previously stopped at
+  beta.18.
 - `.github/workflows/rust.yml` has no `cargo fmt --check` job, so formatting
   drift accumulates unnoticed; `scripts/ci-local.sh` does check it, which makes
   the script stricter than CI rather than equal to it.