From 49f14d95231fba03f194b2b2cb10988cb223dd18 Mon Sep 17 00:00:00 2001 From: Zhe Wu Date: Wed, 19 Aug 2026 01:01:45 -0700 Subject: [PATCH 1/4] feat(node): mark certify events persisted immediately and recover blobs from the pending-recovery table A certify event that needs a blob sync no longer holds its event handle inside the sync task, which pinned the persisted event cursor on one slow or stuck recovery and forced a full event replay after a restart. Instead, the event durably records the blob in the pending-recovery table, is marked complete immediately, and a background executor on BlobSyncHandler drains the table: records are synced in event order, bounded by a config knob, and deleted only when the sync succeeds or the blob is retired. A sync that cannot finish occupies one concurrency slot without blocking other records. Retirement events delete the record before cancelling the sync so the executor cannot resurrect data. The persisted cursor now tracks the event tip, restarts resume recovery from the table instead of replaying events, and the pending-record count is exposed through metrics and the health endpoint. --- crates/walrus-service/src/node.rs | 186 ++++++++++-- .../src/node/blob_event_processor.rs | 38 ++- crates/walrus-service/src/node/blob_sync.rs | 265 +++++++++++++++++- crates/walrus-service/src/node/metrics.rs | 2 + crates/walrus-service/src/node/server.rs | 1 + crates/walrus-service/src/node/storage.rs | 4 - .../src/node/storage/pending_recover_blobs.rs | 2 - crates/walrus-service/storage_openapi.html | 2 +- crates/walrus-service/storage_openapi.yaml | 14 + .../walrus-simtest/tests/simtest_failure.rs | 149 ++++++++++ crates/walrus-storage-node-client/src/api.rs | 3 + 11 files changed, 619 insertions(+), 47 deletions(-) diff --git a/crates/walrus-service/src/node.rs b/crates/walrus-service/src/node.rs index 27bbd7d468..b219a74d01 100644 --- a/crates/walrus-service/src/node.rs +++ b/crates/walrus-service/src/node.rs @@ -909,12 +909,8 @@ impl StorageNode { inner.init_gauges()?; inner.start_recovery_deferral_cleanup_task(); - let blob_sync_handler = Arc::new(BlobSyncHandler::new( - inner.clone(), - config.blob_recovery.max_concurrent_blob_syncs, - config.blob_recovery.max_concurrent_sliver_syncs, - config.blob_recovery.monitor_interval, - )); + let blob_sync_handler = + Arc::new(BlobSyncHandler::new(inner.clone(), &config.blob_recovery)); let shard_sync_handler = Arc::new(ShardSyncHandler::new( inner.clone(), @@ -998,6 +994,7 @@ impl StorageNode { .await; node_recovery_handler .spawn_background_recovery(inner.subscribe_to_epoch_change_sync_and_recovery_info()); + BlobSyncHandler::spawn_pending_recovery_executor(blob_sync_handler.clone()); Ok(StorageNode { inner, @@ -1061,6 +1058,9 @@ impl StorageNode { // syncs, so that neither service starts new sync work afterwards. self.node_recovery_handler.shut_down().await; self.shard_sync_handler.shut_down().await; + self.blob_sync_handler + .shut_down_pending_recovery_executor() + .await; self.blob_sync_handler.cancel_all().await?; self.garbage_collector.abort().await; }, @@ -4203,6 +4203,7 @@ impl ServiceState for StorageNodeInner { shard_detail, shard_summary, latest_checkpoint_sequence_number, + pending_recover_blob_count: Some(self.storage.pending_recover_blob_count()), } } @@ -6570,7 +6571,7 @@ mod tests { } #[tokio::test] - async fn does_not_advance_cursor_past_incomplete_blobs() -> TestResult { + async fn advances_cursor_past_incomplete_blobs_with_pending_record() -> TestResult { walrus_test_utils::init_tracing(); let shards: &[&[u16]] = &[&[1, 6], &[0, 2, 3, 4, 5]]; @@ -6599,9 +6600,9 @@ mod tests { let blob2_registered_event = BlobRegistered::for_testing_with_random_object_id(*blob2_details.blob_id()); events.send(blob2_registered_event.clone().into())?; - let blob2_registered_event_id = blob2_registered_event.event_id; - // The node should not be able to advance past the following event. + // Blob2's data is stored nowhere, so its recovery cannot complete; the node still + // advances past this event, recording the blob in the pending-recovery table. events.send( blob2_registered_event .into_corresponding_certified_event_for_testing() @@ -6614,11 +6615,10 @@ mod tests { BlobRegistered::for_testing_with_random_object_id(*blob3_details.blob_id()); events.send(blob3_registered_event.clone().into())?; store_at_shards(&blob3_details, &cluster, store_at_other_node_fn).await?; - events.send( - blob3_registered_event - .into_corresponding_certified_event_for_testing() - .into(), - )?; + let blob3_certified_event = + blob3_registered_event.into_corresponding_certified_event_for_testing(); + let blob3_certified_event_id = blob3_certified_event.event_id; + events.send(blob3_certified_event.into())?; // All shards for blobs 1 and 3 should be synced by the node. for blob_details in [blob1_details, blob3_details] { @@ -6639,15 +6639,27 @@ mod tests { } } - // The cursor should not have moved beyond that of blob2 registration, since blob2 is yet - // to be synced. - let latest_cursor = cluster.nodes[0] - .storage_node - .inner - .storage - .get_event_cursor_and_next_index()? - .map(|e| e.event_id()); - assert_eq!(latest_cursor, Some(blob2_registered_event_id)); + // The cursor advances to the tip (past blob2's certify event) even though blob2 cannot + // be synced; blob2's recovery obligation remains in the pending-recovery table. + let storage = &cluster.nodes[0].storage_node.inner.storage; + retry_until_success_or_timeout(TIMEOUT, || async { + let latest_cursor = storage + .get_event_cursor_and_next_index()? + .map(|e| e.event_id()); + if latest_cursor == Some(blob3_certified_event_id) { + Ok(()) + } else { + bail!("event cursor has not reached the tip yet: {latest_cursor:?}") + } + }) + .await?; + assert!( + storage + .scan_pending_recover_blobs()? + .iter() + .any(|(blob_id, _)| blob_id == blob2_details.blob_id()), + "blob2 should have a pending-recovery record" + ); Ok(()) } @@ -10408,6 +10420,134 @@ mod tests { Ok(()) } + // Tests that a certify event requiring blob recovery is persisted immediately, with the + // recovery obligation recorded in the pending-recovery table, and that invalidating the blob + // deletes the record and cancels the sync. + #[tokio::test] + async fn certified_event_persisted_while_blob_recovery_pending() -> TestResult { + let shards: &[&[u16]] = &[&[1], &[0, 2, 3, 4, 5, 6]]; + + let (cluster, events) = + cluster_at_epoch1_without_blobs_waiting_for_active_nodes(shards, None).await?; + let node = &cluster.nodes[0]; + let storage = &node.storage_node.inner.storage; + + let baseline_event_count = storage.get_sequentially_processed_event_count()?; + + // Register and certify a blob whose data is stored nowhere, so that its recovery cannot + // complete. + let blob_id = random_blob_id(); + events.send(BlobRegistered::for_testing(blob_id).into())?; + events.send(BlobCertified::for_testing(blob_id).into())?; + + // The certify event is persisted even though the blob recovery cannot complete, and no + // completed events are parked behind it. + wait_until_events_processed(node, baseline_event_count + 2).await?; + assert_eq!(storage.get_event_cursor_progress()?.pending, 0); + + // The recovery obligation is durably recorded and the executor starts the sync. + assert_eq!(storage.pending_recover_blob_count(), 1); + retry_until_success_or_timeout(Duration::from_secs(10), || async { + if node + .storage_node + .blob_sync_handler + .blob_sync_in_progress() + .contains(&blob_id) + { + Ok(()) + } else { + bail!("blob sync has not started yet") + } + }) + .await?; + + // Invalidating the blob deletes the record and cancels the sync. + events.send(InvalidBlobId::for_testing(blob_id).into())?; + retry_until_success_or_timeout(Duration::from_secs(10), || async { + if storage.pending_recover_blob_count() == 0 + && node + .storage_node + .blob_sync_handler + .blob_sync_in_progress() + .is_empty() + { + Ok(()) + } else { + bail!("pending-recovery record or blob sync still present") + } + }) + .await?; + + Ok(()) + } + + // Tests that deleting a blob whose recovery is pending deletes the pending-recovery record + // and cancels the sync. + #[tokio::test] + async fn blob_deleted_event_deletes_pending_recovery_record() -> TestResult { + let shards: &[&[u16]] = &[&[1], &[0, 2, 3, 4, 5, 6]]; + + let (cluster, events) = + cluster_at_epoch1_without_blobs_waiting_for_active_nodes(shards, None).await?; + let node = &cluster.nodes[0]; + let storage = &node.storage_node.inner.storage; + + // Register and certify a deletable blob whose data is stored nowhere, so that its + // recovery cannot complete. + let blob_id = random_blob_id(); + let object_id = ObjectID::random(); + events.send( + BlobRegistered { + deletable: true, + object_id, + ..BlobRegistered::for_testing(blob_id) + } + .into(), + )?; + events.send( + BlobCertified { + deletable: true, + object_id, + ..BlobCertified::for_testing(blob_id) + } + .into(), + )?; + + retry_until_success_or_timeout(Duration::from_secs(10), || async { + if storage.pending_recover_blob_count() == 1 { + Ok(()) + } else { + bail!("pending-recovery record not inserted yet") + } + }) + .await?; + + // Deleting the blob (its only certification) deletes the record and cancels the sync. + events.send( + BlobDeleted { + object_id, + ..BlobDeleted::for_testing(blob_id) + } + .into(), + )?; + retry_until_success_or_timeout(Duration::from_secs(10), || async { + if storage.pending_recover_blob_count() == 0 + && node + .storage_node + .blob_sync_handler + .blob_sync_in_progress() + .is_empty() + { + Ok(()) + } else { + bail!("pending-recovery record or blob sync still present") + } + }) + .await?; + + Ok(()) + } + // Tests that `retrieve_multiple_decoding_symbols` correctly fetches decoding symbols // from storage nodes and successfully recovers the original sliver. // diff --git a/crates/walrus-service/src/node/blob_event_processor.rs b/crates/walrus-service/src/node/blob_event_processor.rs index 582baff8d1..1ffb0019bd 100644 --- a/crates/walrus-service/src/node/blob_event_processor.rs +++ b/crates/walrus-service/src/node/blob_event_processor.rs @@ -249,10 +249,20 @@ impl BackgroundEventProcessor { ) .await; - // Slivers and (possibly) metadata are not stored, so initiate blob sync. - self.blob_sync_handler - .start_sync(blob_id, epoch, Some(event_handle)) - .await?; + // Slivers and (possibly) metadata are not stored, so record that this blob needs + // recovery and let the pending-recovery executor sync it. The record must be written + // before the event is marked as complete: if the node crashes between the two writes, + // the event is replayed and the record is written again. + let pending_count = + self.node + .storage + .insert_pending_recover_blob(&blob_id, event_handle.index(), epoch)?; + self.node + .metrics + .pending_recover_blob_count + .set(pending_count); + event_handle.mark_as_complete(); + self.blob_sync_handler.notify_pending_recovery(); walrus_utils::with_label!(histogram_set, metrics::STATUS_QUEUED) .observe(start.elapsed().as_secs_f64()); @@ -282,6 +292,15 @@ impl BackgroundEventProcessor { }; if let Some(blob_info) = blob_info { if !blob_info.is_certified(current_committee_epoch) { + // Delete the pending-recovery record before cancelling the sync, so that the + // pending-recovery executor cannot restart the sync afterwards. Events for the + // same blob are processed in order, so no new record can appear while this + // event is being handled. + let pending_count = self.node.storage.delete_pending_recover_blob(&blob_id)?; + self.node + .metrics + .pending_recover_blob_count + .set(pending_count); self.node .blob_retirement_notifier .notify_blob_retirement(&blob_id); @@ -339,6 +358,17 @@ impl BackgroundEventProcessor { event_handle: EventHandle, event: InvalidBlobId, ) -> anyhow::Result<()> { + // Delete the pending-recovery record before cancelling the sync, so that the + // pending-recovery executor cannot restart the sync and write the invalid blob's data + // back after it is deleted. + let pending_count = self + .node + .storage + .delete_pending_recover_blob(&event.blob_id)?; + self.node + .metrics + .pending_recover_blob_count + .set(pending_count); self.node .blob_retirement_notifier .notify_blob_retirement(&event.blob_id); diff --git a/crates/walrus-service/src/node/blob_sync.rs b/crates/walrus-service/src/node/blob_sync.rs index 1528440447..5707f5c33f 100644 --- a/crates/walrus-service/src/node/blob_sync.rs +++ b/crates/walrus-service/src/node/blob_sync.rs @@ -11,13 +11,14 @@ use futures::{ FutureExt as _, StreamExt, TryFutureExt, - future::{self, try_join_all}, + future::{self, BoxFuture, try_join_all}, stream, + stream::FuturesUnordered, }; use mysten_metrics::{GaugeGuard, InflightGuardFutureExt as _}; use rayon::prelude::*; use tokio::{ - sync::{Semaphore, watch}, + sync::{Notify, Semaphore, watch}, task::{JoinHandle, JoinSet}, time::Instant, }; @@ -44,15 +45,20 @@ use super::{ LIVE_UPLOAD_DEFERRAL_OUTCOME_AVOIDED_RECOVERY, LIVE_UPLOAD_DEFERRAL_OUTCOME_RECOVERY_NEEDED, NodeMetricSet, + STATUS_ALREADY_STORED, + STATUS_CANCELLED, STATUS_IN_PROGRESS, STATUS_QUEUED, + STATUS_RETIRED, + STATUS_SKIPPED, + STATUS_SUCCESS, }, - storage::Storage, + storage::{PendingRecoverBlob, Storage}, system_events::{CompletableHandle, EventHandle}, }; use crate::{ common::utils::{self, FutureHelpers as _}, - node::NodeStatus, + node::{NodeStatus, config::BlobRecoveryConfig}, }; #[derive(Debug, Clone)] @@ -92,24 +98,29 @@ pub(crate) struct BlobSyncHandler { permits: Permits, task_monitors: TaskMonitorFamily<&'static str>, monitor_interval: Duration, + max_concurrent_pending_recoveries: usize, + pending_recovery_drain_interval: Duration, + // Wakes the pending-recovery executor when a record is inserted. + pending_recovery_notify: Arc, + pending_recovery_task: Arc>>>, } impl BlobSyncHandler { - pub fn new( - node: Arc, - max_concurrent_blob_syncs: usize, - max_concurrent_sliver_syncs: usize, - monitor_interval: Duration, - ) -> Self { + pub fn new(node: Arc, config: &BlobRecoveryConfig) -> Self { Self { blob_syncs_in_progress: Arc::default(), task_monitors: TaskMonitorFamily::new(node.registry.clone()), permits: Permits { - blob: Arc::new(Semaphore::new(max_concurrent_blob_syncs)), - sliver_pairs: Arc::new(Semaphore::new(max_concurrent_sliver_syncs)), + blob: Arc::new(Semaphore::new(config.max_concurrent_blob_syncs)), + sliver_pairs: Arc::new(Semaphore::new(config.max_concurrent_sliver_syncs)), }, node, - monitor_interval, + monitor_interval: config.monitor_interval, + // A zero bound would make the drain loop busy-wait; treat it as one. + max_concurrent_pending_recoveries: config.max_concurrent_pending_recoveries.max(1), + pending_recovery_drain_interval: config.pending_recovery_drain_interval, + pending_recovery_notify: Arc::default(), + pending_recovery_task: Arc::default(), } } @@ -508,6 +519,234 @@ impl BlobSyncHandler { .cloned() .collect() } + + /// Wakes the pending-recovery executor to drain newly inserted pending-recovery records. + pub fn notify_pending_recovery(&self) { + self.pending_recovery_notify.notify_one(); + } + + /// Spawns the background executor that drains the pending-recovery table by syncing the + /// recorded blobs. Called once at startup. + pub fn spawn_pending_recovery_executor(this: Arc) { + let mut task = this + .pending_recovery_task + .lock() + .expect("should be able to acquire lock"); + assert!( + task.is_none(), + "the pending-recovery executor is already running" + ); + let handler = this.clone(); + *task = Some(tokio::spawn(async move { + handler.run_pending_recovery_executor().await; + })); + } + + /// Shuts down the pending-recovery executor, waiting for it to exit so that it cannot start + /// new blob syncs afterwards. + pub async fn shut_down_pending_recovery_executor(&self) { + let task = self + .pending_recovery_task + .lock() + .expect("should be able to acquire lock") + .take(); + if let Some(task) = task { + task.abort(); + let _ = task.await; + } + } + + /// The pending-recovery executor loop. + /// + /// Each round scans the table and starts syncs for the recorded blobs, then waits until a + /// record is inserted or the fallback interval elapses before scanning again. Syncs run + /// independently of the rounds: a sync that cannot finish only occupies one concurrency + /// slot and never blocks other records from being processed. The fallback interval also + /// retries records whose syncs were cancelled or failed. + async fn run_pending_recovery_executor(&self) { + // Blob syncs need the current event epoch to determine which shards to recover, so + // wait until it is known. + if self.node.current_event_epoch().await.is_err() { + tracing::info!("event epoch channel closed; stopping the pending-recovery executor"); + return; + } + + // Syncs started by the executor that have not finished yet. Each future resolves to + // its blob ID; the blob IDs are mirrored in `in_flight_blobs` so that scans can skip + // blobs that are already being synced. + let mut in_flight: FuturesUnordered> = FuturesUnordered::new(); + let mut in_flight_blobs: HashSet = HashSet::new(); + + loop { + if let Err(error) = self + .start_pending_recoveries(&mut in_flight, &mut in_flight_blobs) + .await + { + tracing::warn!(?error, "pending-recovery pass failed"); + } + + // Wait for a reason to scan again, keeping the in-flight bookkeeping up to date as + // syncs finish in the meantime. + loop { + tokio::select! { + Some(blob_id) = in_flight.next(), if !in_flight.is_empty() => { + in_flight_blobs.remove(&blob_id); + } + _ = self.pending_recovery_notify.notified() => break, + _ = tokio::time::sleep(self.pending_recovery_drain_interval) => break, + } + } + } + } + + /// Runs one pass over the pending-recovery table, starting syncs for the recorded blobs. + /// + /// Records are processed in event order; records whose blobs are already being synced are + /// skipped. A record is deleted right away if its blob is no longer certified or is already + /// stored at all owned shards. The remaining blobs are synced, with a bound on how many + /// syncs run at once. A record is deleted only when its sync succeeds, so cancelled or + /// failed syncs are retried on a later pass. + #[tracing::instrument(skip_all)] + async fn start_pending_recoveries( + &self, + in_flight: &mut FuturesUnordered>, + in_flight_blobs: &mut HashSet, + ) -> anyhow::Result<()> { + let mut records = self.node.storage.scan_pending_recover_blobs()?; + self.record_pending_recovery_metrics(&records); + + if records.is_empty() { + return Ok(()); + } + + // While catching up, the node does not yet know its final shard assignment, so + // recovery would target the wrong shards. The records stay in the table and are + // drained after catch-up completes. + if self.node.storage.node_status()?.is_catching_up() { + tracing::debug!("node is catching up; skipping the pending-recovery pass"); + return Ok(()); + } + + tracing::info!( + count = records.len(), + "processing the pending-recovery records" + ); + + // Recover in event order: recovery symbols are then requested roughly in the order the + // blobs were written, improving read locality on the serving nodes. + records.sort_unstable_by_key(|(_, record)| record.event_index()); + + for (blob_id, record) in records { + if in_flight_blobs.contains(&blob_id) { + continue; + } + + while in_flight.len() >= self.max_concurrent_pending_recoveries { + if let Some(completed_blob_id) = in_flight.next().await { + in_flight_blobs.remove(&completed_blob_id); + } + } + + if self.node.is_blob_not_certified(&blob_id) { + delete_pending_recover_blob_record(&self.node, &blob_id, STATUS_RETIRED); + continue; + } + + let current_event_epoch = self.node.current_event_epoch().await?; + if self + .node + .is_stored_at_all_shards_at_epoch(&blob_id, current_event_epoch) + .await? + { + delete_pending_recover_blob_record(&self.node, &blob_id, STATUS_ALREADY_STORED); + continue; + } + + let mut receiver = self + .start_sync(blob_id, record.certified_epoch(), None) + .await?; + let node = self.node.clone(); + in_flight_blobs.insert(blob_id); + in_flight.push( + async move { + let outcome = match receiver + .wait_for(|status| matches!(status, SyncStatus::Done(_))) + .await + { + Ok(status) => match &*status { + SyncStatus::Done(outcome) => *outcome, + SyncStatus::Pending => { + unreachable!("wait_for only returns done statuses") + } + }, + Err(_) => { + // The sync task dropped the sender without publishing an outcome; + // leave the record for the next pass. + return blob_id; + } + }; + match outcome { + SyncOutcome::Success => { + delete_pending_recover_blob_record(&node, &blob_id, STATUS_SUCCESS); + } + SyncOutcome::Cancelled => walrus_utils::with_label!( + node.metrics.pending_recovery_executor_total, + STATUS_CANCELLED + ) + .inc(), + SyncOutcome::Skipped => walrus_utils::with_label!( + node.metrics.pending_recovery_executor_total, + STATUS_SKIPPED + ) + .inc(), + } + blob_id + } + .boxed(), + ); + } + + Ok(()) + } + + fn record_pending_recovery_metrics(&self, records: &[(BlobId, PendingRecoverBlob)]) { + self.node + .metrics + .pending_recover_blob_count + .set(self.node.storage.pending_recover_blob_count()); + self.node + .metrics + .pending_recover_blob_oldest_event_index + .set( + records + .iter() + .map(|(_, record)| record.event_index()) + .min() + .unwrap_or(0), + ); + } +} + +/// Deletes the pending-recovery record for the blob and records the outcome in metrics. +/// +/// A deletion failure is logged but not propagated: the record is then re-evaluated (and the +/// deletion retried) on the executor's next drain pass. +fn delete_pending_recover_blob_record( + node: &StorageNodeInner, + blob_id: &BlobId, + outcome: &'static str, +) { + match node.storage.delete_pending_recover_blob(blob_id) { + Ok(remaining) => { + node.metrics.pending_recover_blob_count.set(remaining); + walrus_utils::with_label!(node.metrics.pending_recovery_executor_total, outcome).inc(); + } + Err(error) => tracing::error!( + ?error, + walrus.blob_id = %blob_id, + "failed to delete the pending-recovery record" + ), + } } type SyncJoinHandle = JoinHandle>; diff --git a/crates/walrus-service/src/node/metrics.rs b/crates/walrus-service/src/node/metrics.rs index 1604c0e781..125c507a13 100644 --- a/crates/walrus-service/src/node/metrics.rs +++ b/crates/walrus-service/src/node/metrics.rs @@ -45,6 +45,8 @@ pub(crate) const STATUS_BLOB_INFO_CLEANUP_COMPLETED: &str = "blob_info_cleanup_c pub(crate) const STATUS_DATA_DELETION_STARTED: &str = "data_deletion_started"; pub(crate) const STATUS_COMPLETED: &str = "completed"; pub(crate) const STATUS_HIGHEST_FINISHED: &str = "highest_finished"; +pub(crate) const STATUS_RETIRED: &str = "retired"; +pub(crate) const STATUS_ALREADY_STORED: &str = "already_stored"; pub(crate) const LIVE_UPLOAD_DEFERRAL_OUTCOME_AVOIDED_RECOVERY: &str = "avoided_recovery"; pub(crate) const LIVE_UPLOAD_DEFERRAL_OUTCOME_RECOVERY_NEEDED: &str = "recovery_needed"; diff --git a/crates/walrus-service/src/node/server.rs b/crates/walrus-service/src/node/server.rs index 75d2a5bfef..1cbea5624e 100644 --- a/crates/walrus-service/src/node/server.rs +++ b/crates/walrus-service/src/node/server.rs @@ -823,6 +823,7 @@ mod tests { shard_detail: None, shard_summary: ShardStatusSummary::default(), latest_checkpoint_sequence_number: None, + pending_recover_blob_count: None, } } diff --git a/crates/walrus-service/src/node/storage.rs b/crates/walrus-service/src/node/storage.rs index 27ad97f46f..d98d0d079c 100644 --- a/crates/walrus-service/src/node/storage.rs +++ b/crates/walrus-service/src/node/storage.rs @@ -1523,7 +1523,6 @@ impl Storage { /// /// Must be called before the certify event is marked as complete: if the node crashes in /// between, the event is replayed and the record is written again. - #[allow(dead_code)] // The callers land in a follow-up change. pub(crate) fn insert_pending_recover_blob( &self, blob_id: &BlobId, @@ -1538,7 +1537,6 @@ impl Storage { /// Deletes the pending-recovery record for the blob, if any, and returns the number of /// remaining pending-recovery records. - #[allow(dead_code)] // The callers land in a follow-up change. pub(crate) fn delete_pending_recover_blob( &self, blob_id: &BlobId, @@ -1547,7 +1545,6 @@ impl Storage { } /// Returns all pending-recovery records. - #[allow(dead_code)] // The callers land in a follow-up change. pub(crate) fn scan_pending_recover_blobs( &self, ) -> Result, TypedStoreError> { @@ -1555,7 +1552,6 @@ impl Storage { } /// Returns the number of pending-recovery records. - #[allow(dead_code)] // The callers land in a follow-up change. pub(crate) fn pending_recover_blob_count(&self) -> u64 { self.pending_recover_blobs.count() } diff --git a/crates/walrus-service/src/node/storage/pending_recover_blobs.rs b/crates/walrus-service/src/node/storage/pending_recover_blobs.rs index 0d293eb24d..606a63876c 100644 --- a/crates/walrus-service/src/node/storage/pending_recover_blobs.rs +++ b/crates/walrus-service/src/node/storage/pending_recover_blobs.rs @@ -42,7 +42,6 @@ impl PendingRecoverBlob { } /// The index of the `BlobCertified` event that required the recovery. - #[allow(dead_code)] // The callers land in a follow-up change. pub fn event_index(&self) -> u64 { match self { PendingRecoverBlob::V1(v1) => v1.event_index, @@ -51,7 +50,6 @@ impl PendingRecoverBlob { /// The epoch in which the blob was certified, used to route recovery requests to the correct /// committee. - #[allow(dead_code)] // The callers land in a follow-up change. pub fn certified_epoch(&self) -> Epoch { match self { PendingRecoverBlob::V1(v1) => v1.certified_epoch, diff --git a/crates/walrus-service/storage_openapi.html b/crates/walrus-service/storage_openapi.html index 7683afb5d4..c59fb40615 100644 --- a/crates/walrus-service/storage_openapi.html +++ b/crates/walrus-service/storage_openapi.html @@ -22,7 +22,7 @@