From 2139fa0d28880e921fe2af6b00b018ab6796a917 Mon Sep 17 00:00:00 2001 From: Michael <55204436+michaelkaplan13@users.noreply.github.com> Date: Tue, 11 Aug 2026 14:45:22 -0400 Subject: [PATCH 1/3] recover QMDB writers from suffix checkpoints --- Cargo.lock | 12 +-- Cargo.toml | 14 ++-- crates/indexer/src/publisher/qmdb.rs | 119 ++++++++++++++++++++++----- 3 files changed, 110 insertions(+), 35 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 090f9c00..db05c9b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2935,7 +2935,7 @@ dependencies = [ [[package]] name = "exoware-qmdb" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=8150d0f2a9037875c06f5d833a9617169739bfe3#8150d0f2a9037875c06f5d833a9617169739bfe3" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" dependencies = [ "axum", "buffa", @@ -2965,7 +2965,7 @@ dependencies = [ [[package]] name = "exoware-sdk" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=8150d0f2a9037875c06f5d833a9617169739bfe3#8150d0f2a9037875c06f5d833a9617169739bfe3" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" dependencies = [ "anyhow", "base64", @@ -2994,7 +2994,7 @@ dependencies = [ [[package]] name = "exoware-server" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=8150d0f2a9037875c06f5d833a9617169739bfe3#8150d0f2a9037875c06f5d833a9617169739bfe3" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" dependencies = [ "buffa", "buffa-types", @@ -3011,7 +3011,7 @@ dependencies = [ [[package]] name = "exoware-simplex" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=8150d0f2a9037875c06f5d833a9617169739bfe3#8150d0f2a9037875c06f5d833a9617169739bfe3" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" dependencies = [ "bytes", "clap", @@ -3036,7 +3036,7 @@ dependencies = [ [[package]] name = "exoware-simulator" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=8150d0f2a9037875c06f5d833a9617169739bfe3#8150d0f2a9037875c06f5d833a9617169739bfe3" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" dependencies = [ "axum", "buffa", @@ -3060,7 +3060,7 @@ dependencies = [ [[package]] name = "exoware-sql" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=8150d0f2a9037875c06f5d833a9617169739bfe3#8150d0f2a9037875c06f5d833a9617169739bfe3" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" dependencies = [ "async-trait", "axum", diff --git a/Cargo.toml b/Cargo.toml index 09472d62..0bb4f35a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -100,13 +100,13 @@ prometheus-client = { version = "0.24.0", default-features = false } tracing-subscriber = "0.3" [patch.crates-io] -# All Exoware libraries must share one SDK crate instance because SDK types cross crate boundaries. -exoware-qmdb = { git = "https://github.com/exowarexyz/monorepo.git", rev = "8150d0f2a9037875c06f5d833a9617169739bfe3" } -exoware-sdk = { git = "https://github.com/exowarexyz/monorepo.git", rev = "8150d0f2a9037875c06f5d833a9617169739bfe3" } -exoware-server = { git = "https://github.com/exowarexyz/monorepo.git", rev = "8150d0f2a9037875c06f5d833a9617169739bfe3" } -exoware-simplex = { git = "https://github.com/exowarexyz/monorepo.git", rev = "8150d0f2a9037875c06f5d833a9617169739bfe3" } -exoware-simulator = { git = "https://github.com/exowarexyz/monorepo.git", rev = "8150d0f2a9037875c06f5d833a9617169739bfe3" } -exoware-sql = { git = "https://github.com/exowarexyz/monorepo.git", rev = "8150d0f2a9037875c06f5d833a9617169739bfe3" } +# Keep all Exoware crates on one revision because SDK types cross crate boundaries. +exoware-qmdb = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } +exoware-sdk = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } +exoware-server = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } +exoware-simplex = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } +exoware-simulator = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } +exoware-sql = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } [profile.samply] inherits = "release" diff --git a/crates/indexer/src/publisher/qmdb.rs b/crates/indexer/src/publisher/qmdb.rs index b88592dd..6ee140c8 100644 --- a/crates/indexer/src/publisher/qmdb.rs +++ b/crates/indexer/src/publisher/qmdb.rs @@ -220,8 +220,6 @@ pub enum PublishError { Sql(#[from] datafusion::error::DataFusionError), #[error("failed to encode SQL metadata row: {0}")] SqlRow(String), - #[error("cannot initialize QMDB writer from {locations} operation locations")] - CheckpointTooLarge { locations: u64 }, #[error("QMDB Store is empty but finalized block height {height} needs historical backfill")] StoreEmptyPastGenesis { height: u64 }, #[error( @@ -1269,11 +1267,11 @@ where UnorderedClient::::new(client, ()); recover_writer_state::( reader.writer_location_watermark().await?, - |watermark, max| { + |watermark, start, max| { let reader = reader.clone(); async move { reader - .operation_range_checkpoint(watermark, Location::new(0), max) + .operation_range_checkpoint(watermark, start, max) .await } }, @@ -1291,11 +1289,11 @@ where let reader = KeylessClient::>::new(client, ()); recover_writer_state::( reader.writer_location_watermark().await?, - |watermark, max| { + |watermark, start, max| { let reader = reader.clone(); async move { reader - .operation_range_checkpoint(watermark, Location::new(0), max) + .operation_range_checkpoint(watermark, start, max) .await } }, @@ -1309,7 +1307,7 @@ async fn recover_writer_state( ) -> Result, PublishError> where H: Hasher, - Fetch: FnOnce(Location, u32) -> Fut, + Fetch: FnOnce(Location, Location, u32) -> Fut, Fut: std::future::Future< Output = Result< exoware_qmdb::OperationRangeCheckpoint, @@ -1320,15 +1318,14 @@ where let Some(watermark) = watermark else { return Ok(WriterState::empty()); }; - let locations = watermark - .as_u64() - .checked_add(1) - .ok_or(PublishError::CheckpointTooLarge { - locations: u64::MAX, - })?; - let max = - u32::try_from(locations).map_err(|_| PublishError::CheckpointTooLarge { locations })?; - let checkpoint = fetch(watermark, max).await?; + + // Writer state only needs the MMR peaks at the watermark, and a checkpoint + // proof over any operation range ending at the watermark exposes every + // peak digest. Fetching just the final operation keeps recovery cost + // logarithmic in history size. Requesting the full prefix instead would + // transfer the entire operation log and fail outright once a namespace + // outgrows the u32 location count accepted by the checkpoint API. + let checkpoint = fetch(watermark, watermark, 1).await?; Ok(WriterState::from_checkpoint::(&checkpoint)?) } @@ -1992,6 +1989,69 @@ mod tests { }); } + #[test] + fn publisher_recovers_writer_state_on_reconnect() { + commonware_runtime::tokio::Runner::default().start(|context| async move { + let (handle, url) = exoware_simulator::open_temp() + .await + .expect("spawn simulator"); + + let publisher = Publisher::::connect( + context.child("first_publisher"), + &url, + 2, + crate::publisher::StoreCommitMetrics::new(&context.child("first_metrics")), + ) + .await + .expect("publisher connects to empty store"); + let completion = publisher + .enqueue_queued_finalized(test_queued_upload()) + .await + .expect("first upload accepted"); + assert!(completion.wait().await); + let locations = publisher.next_locations().await; + publisher.shutdown().await; + + // Reconnecting rebuilds writer state from a bounded checkpoint + // ending at the published watermark rather than the full + // operation history. + let publisher = Publisher::::connect( + context.child("second_publisher"), + &url, + 2, + crate::publisher::StoreCommitMetrics::new(&context.child("second_metrics")), + ) + .await + .expect("publisher reconnects to populated store"); + assert_eq!(publisher.next_locations().await, locations); + + // The recovered peaks must support further uploads end to end. + let (state_start, transaction_start) = locations; + let completion = publisher + .enqueue_queued_finalized(test_queued_upload_at(2, state_start, transaction_start)) + .await + .expect("follow-up upload accepted"); + assert!(completion.wait().await); + let locations = publisher.next_locations().await; + publisher.shutdown().await; + + // A second reconnect recovers from a checkpoint range that starts + // past location zero. + let publisher = Publisher::::connect( + context.child("third_publisher"), + &url, + 2, + crate::publisher::StoreCommitMetrics::new(&context.child("third_metrics")), + ) + .await + .expect("publisher reconnects after follow-up upload"); + assert_eq!(publisher.next_locations().await, locations); + + publisher.shutdown().await; + handle.abort(); + }); + } + #[test] fn queued_upload_roots_match_application_roots() { commonware_runtime::tokio::Runner::default().start(|context| async move { @@ -2423,7 +2483,19 @@ mod tests { } fn test_queued_upload() -> QueuedFinalizedUpload { + test_queued_upload_at(1, 0, 0) + } + + fn test_queued_upload_at( + height: u64, + state_start: u64, + transaction_start: u64, + ) -> QueuedFinalizedUpload { let leader = ed25519::PrivateKey::from_seed(7).public_key(); + + // An empty block body appends one transaction commit operation, plus + // the genesis commit when the store starts empty. + let transaction_ops = if transaction_start == 0 { 2 } else { 1 }; let header = Header { context: SimplexContext { round: Round::zero(), @@ -2431,12 +2503,15 @@ mod tests { parent: (View::zero(), Commitment::EMPTY), }, parent: Sha256Digest::EMPTY, - height: 1, + height, timestamp: 0, state_root: Sha256Digest::EMPTY, - state_range: non_empty_range!(0, 2), + state_range: non_empty_range!(state_start, state_start + 2), transactions_root: Sha256Digest::EMPTY, - transactions_range: non_empty_range!(0, 2), + transactions_range: non_empty_range!( + transaction_start, + transaction_start + transaction_ops + ), }; let block = Block::new(header, Vec::>::new()) .seal(&mut Sha256::default()); @@ -2445,7 +2520,7 @@ mod tests { StateOperation::Update(UnorderedUpdate( account_key, encode_account(Account { - balance: 1, + balance: height, nonce: Nonce::default(), }), )), @@ -2455,8 +2530,8 @@ mod tests { QueuedFinalizedUpload { block: Arc::new(block), finalized_ts_micros: 1_000, - state_start: 0, - transaction_start: 0, + state_start, + transaction_start, state_delta: Arc::new(state_delta), } } From 812146e332c38fe282a91cb8f8fd9f8128284753 Mon Sep 17 00:00:00 2001 From: Michael <55204436+michaelkaplan13@users.noreply.github.com> Date: Tue, 11 Aug 2026 14:45:36 -0400 Subject: [PATCH 2/3] skip capture for redelivered uploaded blocks --- bin/validator/src/run.rs | 11 +++++- crates/indexer/src/publisher/qmdb.rs | 51 +++++++++++++++++++++++++--- 2 files changed, 57 insertions(+), 5 deletions(-) diff --git a/bin/validator/src/run.rs b/bin/validator/src/run.rs index 900ef7c6..5cb6f10f 100644 --- a/bin/validator/src/run.rs +++ b/bin/validator/src/run.rs @@ -314,7 +314,16 @@ impl FinalizedUploadProducer { ) .await { - Ok(upload) => upload, + Ok(Some(upload)) => upload, + Ok(None) => { + info!( + height = block.header.height, + state_next = cursor.state_next, + transaction_next = cursor.transaction_next, + "finalized block already uploaded, skipping index capture" + ); + return; + } Err(PublishError::StoreEmptyPastGenesis { .. }) if cursor.state_next == 0 => { let publisher = self.publisher.publisher().await; let (state_next, transaction_next) = publisher.next_locations().await; diff --git a/crates/indexer/src/publisher/qmdb.rs b/crates/indexer/src/publisher/qmdb.rs index 6ee140c8..32da2753 100644 --- a/crates/indexer/src/publisher/qmdb.rs +++ b/crates/indexer/src/publisher/qmdb.rs @@ -428,6 +428,13 @@ where /// Capture the finalized-block upload material that must survive local pruning. /// + /// Returns `None` when both writer cursors sit at or past the block's + /// operation ranges. Crash recovery redelivers finalized blocks whose + /// uploads already committed, and each block's operations live at + /// consensus-assigned locations, so a fully covered block has nothing left + /// to capture. Appending its operations again would place duplicates at + /// locations owned by later blocks. + /// /// This deliberately stops at the durable local payload boundary. Remote /// Store staging and upload are handled later by the queue consumer: /// @@ -441,13 +448,18 @@ where transaction_writer_next: u64, block: &EngineBlock, databases: &Databases, - ) -> Result, PublishError> + ) -> Result>, PublishError> where Cx: Spawner, E: BufferPooler + Storage + Clock + Metrics + Send + Sync + 'static, S: Strategy + Send + Sync + 'static, { let state_end = block.header.state_range.end(); + let transaction_end = block.header.transactions_range.end(); + if state_writer_next >= state_end && transaction_writer_next >= transaction_end { + return Ok(None); + } + validate_writer_range(state_writer_next, state_end, block.header.height)?; transaction_upload_end(transaction_writer_next, block)?; let block = Arc::new(block.clone()); @@ -462,13 +474,13 @@ where .await .expect("QMDB state queue task exited")?; - Ok(QueuedFinalizedUpload { + Ok(Some(QueuedFinalizedUpload { block, finalized_ts_micros: current_time_micros(), state_start: state_writer_next, transaction_start: transaction_writer_next, state_delta: Arc::new(state_delta), - }) + })) } /// Queue a previously durable finalized-block payload for remote upload. @@ -2136,6 +2148,36 @@ mod tests { .await; assert_transaction_append_locations_match_block(&client, &second).await; + // Crash recovery redelivers finalized blocks whose uploads already + // committed. A fully covered block must skip capture instead of + // failing writer validation. + let (state_next, transaction_next) = publisher.next_locations().await; + let redelivered = Publisher::build_queued_finalized_upload_with_context( + context.child("redelivered"), + state_next, + transaction_next, + &second, + &databases, + ) + .await + .expect("redelivered block builds"); + assert!( + redelivered.is_none(), + "fully uploaded block must skip capture" + ); + + // Cursors covering only one namespace indicate real divergence and + // must still fail loudly. + let partial = Publisher::build_queued_finalized_upload_with_context( + context.child("partial"), + state_next, + second.header.transactions_range.end() - 1, + &second, + &databases, + ) + .await; + assert!(matches!(partial, Err(PublishError::WriterOutOfSync { .. }))); + publisher.shutdown().await; handle.abort(); }); @@ -2305,7 +2347,8 @@ mod tests { databases, ) .await - .expect("queued upload builds"); + .expect("queued upload builds") + .expect("block is not yet uploaded"); let state_start = upload.state_start(); let transaction_start = upload.transaction_start(); let completion = publisher From 5454961c27a4419b28f5255113fd60bbd4427e56 Mon Sep 17 00:00:00 2001 From: Michael <55204436+michaelkaplan13@users.noreply.github.com> Date: Thu, 13 Aug 2026 14:48:07 -0400 Subject: [PATCH 3/3] Use exoware writer recovery helper --- Cargo.lock | 12 +++---- Cargo.toml | 12 +++---- crates/indexer/src/publisher/qmdb.rs | 54 ++-------------------------- 3 files changed, 14 insertions(+), 64 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index db05c9b5..f86fad86 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2935,7 +2935,7 @@ dependencies = [ [[package]] name = "exoware-qmdb" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=563fc50f98b08a120450a7437f0224d590f1e194#563fc50f98b08a120450a7437f0224d590f1e194" dependencies = [ "axum", "buffa", @@ -2965,7 +2965,7 @@ dependencies = [ [[package]] name = "exoware-sdk" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=563fc50f98b08a120450a7437f0224d590f1e194#563fc50f98b08a120450a7437f0224d590f1e194" dependencies = [ "anyhow", "base64", @@ -2994,7 +2994,7 @@ dependencies = [ [[package]] name = "exoware-server" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=563fc50f98b08a120450a7437f0224d590f1e194#563fc50f98b08a120450a7437f0224d590f1e194" dependencies = [ "buffa", "buffa-types", @@ -3011,7 +3011,7 @@ dependencies = [ [[package]] name = "exoware-simplex" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=563fc50f98b08a120450a7437f0224d590f1e194#563fc50f98b08a120450a7437f0224d590f1e194" dependencies = [ "bytes", "clap", @@ -3036,7 +3036,7 @@ dependencies = [ [[package]] name = "exoware-simulator" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=563fc50f98b08a120450a7437f0224d590f1e194#563fc50f98b08a120450a7437f0224d590f1e194" dependencies = [ "axum", "buffa", @@ -3060,7 +3060,7 @@ dependencies = [ [[package]] name = "exoware-sql" version = "2026.7.1" -source = "git+https://github.com/exowarexyz/monorepo.git?rev=5826cf0402594cf5d5fd16a92590a46f3de759ef#5826cf0402594cf5d5fd16a92590a46f3de759ef" +source = "git+https://github.com/exowarexyz/monorepo.git?rev=563fc50f98b08a120450a7437f0224d590f1e194#563fc50f98b08a120450a7437f0224d590f1e194" dependencies = [ "async-trait", "axum", diff --git a/Cargo.toml b/Cargo.toml index 0bb4f35a..ee72d75e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -101,12 +101,12 @@ tracing-subscriber = "0.3" [patch.crates-io] # Keep all Exoware crates on one revision because SDK types cross crate boundaries. -exoware-qmdb = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } -exoware-sdk = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } -exoware-server = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } -exoware-simplex = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } -exoware-simulator = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } -exoware-sql = { git = "https://github.com/exowarexyz/monorepo.git", rev = "5826cf0402594cf5d5fd16a92590a46f3de759ef" } +exoware-qmdb = { git = "https://github.com/exowarexyz/monorepo.git", rev = "563fc50f98b08a120450a7437f0224d590f1e194" } +exoware-sdk = { git = "https://github.com/exowarexyz/monorepo.git", rev = "563fc50f98b08a120450a7437f0224d590f1e194" } +exoware-server = { git = "https://github.com/exowarexyz/monorepo.git", rev = "563fc50f98b08a120450a7437f0224d590f1e194" } +exoware-simplex = { git = "https://github.com/exowarexyz/monorepo.git", rev = "563fc50f98b08a120450a7437f0224d590f1e194" } +exoware-simulator = { git = "https://github.com/exowarexyz/monorepo.git", rev = "563fc50f98b08a120450a7437f0224d590f1e194" } +exoware-sql = { git = "https://github.com/exowarexyz/monorepo.git", rev = "563fc50f98b08a120450a7437f0224d590f1e194" } [profile.samply] inherits = "release" diff --git a/crates/indexer/src/publisher/qmdb.rs b/crates/indexer/src/publisher/qmdb.rs index 32da2753..a7930d10 100644 --- a/crates/indexer/src/publisher/qmdb.rs +++ b/crates/indexer/src/publisher/qmdb.rs @@ -1277,18 +1277,7 @@ where { let reader = UnorderedClient::::new(client, ()); - recover_writer_state::( - reader.writer_location_watermark().await?, - |watermark, start, max| { - let reader = reader.clone(); - async move { - reader - .operation_range_checkpoint(watermark, start, max) - .await - } - }, - ) - .await + Ok(reader.recover_writer_state().await?) } async fn recover_transaction_writer_state( @@ -1299,46 +1288,7 @@ where H::Digest: Codec + Send + Sync, { let reader = KeylessClient::>::new(client, ()); - recover_writer_state::( - reader.writer_location_watermark().await?, - |watermark, start, max| { - let reader = reader.clone(); - async move { - reader - .operation_range_checkpoint(watermark, start, max) - .await - } - }, - ) - .await -} - -async fn recover_writer_state( - watermark: Option>, - fetch: Fetch, -) -> Result, PublishError> -where - H: Hasher, - Fetch: FnOnce(Location, Location, u32) -> Fut, - Fut: std::future::Future< - Output = Result< - exoware_qmdb::OperationRangeCheckpoint, - QmdbError, - >, - >, -{ - let Some(watermark) = watermark else { - return Ok(WriterState::empty()); - }; - - // Writer state only needs the MMR peaks at the watermark, and a checkpoint - // proof over any operation range ending at the watermark exposes every - // peak digest. Fetching just the final operation keeps recovery cost - // logarithmic in history size. Requesting the full prefix instead would - // transfer the entire operation log and fail outright once a namespace - // outgrows the u32 location count accepted by the checkpoint API. - let checkpoint = fetch(watermark, watermark, 1).await?; - Ok(WriterState::from_checkpoint::(&checkpoint)?) + Ok(reader.recover_writer_state().await?) } struct PendingTransactionUpload