feat(sqlite-storage): address US-060, US-061, US-062 review gaps

Author: NathanFlurry

engine/CLAUDE.md

@@ -55,7 +55,6 @@ Use `test-snapshot-gen` to generate and load RocksDB snapshots of the full UDB K
 - `sqlite-storage` `get_pages(...)` should keep META, cold PIDX loads, and DELTA/SHARD blob fetches inside one `db.run(...)` transaction, then decode each unique blob once and evict stale cached PIDX rows that now need SHARD fallback.
 - `sqlite-storage` fast-path commits should update an already-cached PIDX in memory after the store write, but must not load PIDX from store just to mutate it or the one-RTT path is gone.
 - `sqlite-storage` fast-path cutoffs should use raw dirty-page bytes, and slow-path finalize must accept larger encoded DELTA blobs because UniversalDB chunks logical values internally.
-- `sqlite-storage` staged commits should scan a stage-specific prefix like `stage_chunk_prefix(stage_id)` and delete the staged chunk keys in the same `atomic_write` that promotes DELTA, PIDX, and META.
 - `sqlite-storage` compaction should choose shard passes from the live PIDX scan, then delete DELTA blobs by comparing all existing delta keys against the remaining global PIDX references so multi-shard and overwritten deltas only disappear when every page ref is gone.
 - `sqlite-storage` compaction must re-read META inside its write transaction and fence on `generation` plus `head_txid` before updating `materialized_txid` or quota fields, so takeover and commits cannot rewind the head.
 - `sqlite-storage` metrics should record compaction pass duration and totals in `compaction/worker.rs`, while shard outcome metrics such as folded pages, deleted deltas, delta gauge updates, and lag stay in `compaction/shard.rs` to avoid double counting.

engine/packages/pegboard-envoy/src/sqlite_runtime.rs

@@ -21,9 +21,13 @@ pub async fn shared_engine(ctx: &StandaloneCtx) -> Result<Arc<SqliteEngine>> {
 			tracing::info!("initializing shared sqlite dispatch runtime");
 
 			let (engine, compaction_rx) = SqliteEngine::new(Arc::clone(&db), subspace.clone());
-			tokio::spawn(CompactionCoordinator::run(compaction_rx, db, subspace));
+			let engine = Arc::new(engine);
+			tokio::spawn(CompactionCoordinator::run(
+				compaction_rx,
+				Arc::clone(&engine),
+			));
 
-			Ok(Arc::new(engine))
+			Ok(engine)
 		})
 		.await
 		.cloned()

engine/packages/pegboard-envoy/src/ws_to_tunnel_task.rs

@@ -848,10 +848,7 @@ async fn handle_sqlite_commit_stage(
 	{
 		Ok(result) => Ok(protocol::SqliteCommitStageResponse::SqliteCommitStageOk(
 			protocol::SqliteCommitStageOk {
-				chunk_idx_committed: result
-					.chunk_idx_committed
-					.try_into()
-					.context("sqlite stage chunk index exceeded u16")?,
+				chunk_idx_committed: result.chunk_idx_committed,
 			},
 		)),
 		Err(err) => match sqlite_storage_error(&err) {

engine/packages/pegboard-outbound/src/lib.rs

@@ -37,9 +37,13 @@ async fn shared_sqlite_engine(ctx: &StandaloneCtx) -> Result<Arc<SqliteEngine>>
 	SQLITE_ENGINE
 		.get_or_try_init(|| async move {
 			let (engine, compaction_rx) = SqliteEngine::new(Arc::clone(&db), subspace.clone());
-			tokio::spawn(CompactionCoordinator::run(compaction_rx, db, subspace));
+			let engine = Arc::new(engine);
+			tokio::spawn(CompactionCoordinator::run(
+				compaction_rx,
+				Arc::clone(&engine),
+			));
 
-			Ok(Arc::new(engine))
+			Ok(engine)
 		})
 		.await
 		.cloned()