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
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
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 15824525..6e8d0a85 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,46 +113,115 @@ 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, 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
+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.
+
+**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.
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();