Enforce ordering invariants with typestate

.github/workflows/ci.yml

@@ -283,7 +283,7 @@ jobs:
       - name: Coverage summary
         working-directory: walshadow
         env:
-          MAX_MISSED_LINES: 3130
+          MAX_MISSED_LINES: 2950
         run: |
           awk -F'[:,]' -v max="$MAX_MISSED_LINES" '
             /^DA:/ { f++; if ($3 > 0) h++ }

Cargo.toml

@@ -37,6 +37,7 @@ path = "src/lib.rs"
 default = ["lz4", "zstd"]
 lz4 = ["clickhouse-c-rs/lz4"]
 zstd = ["clickhouse-c-rs/zstd"]
+test-support = []
 
 [dependencies]
 wal-rus = "0.3.5"
@@ -101,6 +102,7 @@ name = "spool_read"
 harness = false
 
 [dev-dependencies]
+walshadow = { path = ".", features = ["test-support"] }
 # Pre-main tracing subscriber for tests
 ctor = "1.0.13"
 tempfile = "3"

benches/bootstrap_pump.rs

@@ -401,9 +401,7 @@ async fn bench_drain(shape: Shape) -> Report {
     .unwrap();
     let elapsed = started.elapsed().as_secs_f64();
     feeder.await.unwrap();
-    drop(msg_tx);
-    drop(ack);
-    tail.join().await;
+    tail.close(msg_tx, ack).await;
 
     Report {
         label: "drain".into(),

benches/xact_spill.rs

@@ -9,7 +9,7 @@ use clap::Parser;
 use walrus::pg::walparser::RelFileNode;
 use walshadow::heap_decoder::{ColumnValue, DecodedHeap, DecodedTuple, DescribedHeap, HeapOp};
 use walshadow::schema::{RelDescriptor, RelName, ReplIdent};
-use walshadow::xact_buffer::{XactBuffer, XactBufferConfig};
+use walshadow::xact_buffer::{StashResolved, XactBuffer, XactBufferConfig};
 
 #[global_allocator]
 static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
@@ -96,7 +96,13 @@ async fn run(args: Args) -> anyhow::Result<()> {
             let start = Instant::now();
             let commit_lsn = u64::from(xid + 1) * (args.rows as u64 + 1);
             let mut drain = buffer
-                .drain_committed(xid, 0, commit_lsn, &[], false)
+                .drain_committed(
+                    StashResolved::nothing_stashed(xid),
+                    0,
+                    commit_lsn,
+                    &[],
+                    false,
+                )
                 .await?;
             let mut rows = 0;
             while let Some(batch) = drain.next_batch(1024, 1 << 20, None).await? {

plans/coverage100.md

@@ -136,9 +136,9 @@ Check candidate gaps against fresh merged report before adding tests
   for `reopen_walk_spools` and ready-checkpoint resume, or hand-seed checkpoints
   as `backup_checkpoint_e2e` does
 - `copy_backfill` pending tables: hold an open xact on target table across a
-  base-backup opt-in, commit or abort after walk, for `record_pending` and
-  `settle_ended_pending`. `copy_chunk_blocks = 1` on a multi-block table for chunk
-  loop and progress ledger. Empty table fast path. Opt-out and CH schema change
+  base-backup opt-in, commit or abort after walk, for `hold_pending`,
+  `release_pending` and `settle_ended_pending`. `copy_chunk_blocks = 1` on a
+  multi-block table for chunk loop and progress ledger. Empty table fast path. Opt-out and CH schema change
   mid-pass for publish/swap edges. Two opt-ins inside coalesce window
 - `copy_backfill` units: `wire_kind` per OID, `decode_field` per type and invalid
   UTF-8, ledger version rejection. Ledger persist failures via fs faults

src/backfill/backfill_staging.rs

@@ -22,6 +22,7 @@ use anyhow::{Context, Result, bail};
 use clickhouse_c::{Block, Event};
 
 use crate::backfill::backfill_types::BackupRequest;
+use crate::backfill::copy_backfill::SwapPermit;
 use crate::ch::{ChConn, EmitterError, exec_drain, quote_ident, with_timeout};
 use crate::config::DestEmitter;
 use crate::mapping::{MappingHandle, TableMapping, TableTarget};
@@ -271,7 +272,8 @@ impl StagingSession {
     }
 
     /// Atomic publish; requires an Atomic/Replicated database engine.
-    pub async fn exchange(&mut self, rel: &StagingRel) -> Result<()> {
+    pub async fn exchange(&mut self, permit: SwapPermit<'_>) -> Result<()> {
+        let rel = permit.rel();
         self.exec_once(&format!(
             "EXCHANGE TABLES {} AND {}",
             rel.real_sql(),