diff --git a/crates/walrus-service/src/node.rs b/crates/walrus-service/src/node.rs index 27bbd7d468..be57acf86b 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(()) } @@ -9124,13 +9136,23 @@ mod tests { .get_sequentially_processed_event_count()? ); - // Unblock the epoch change start event, and expect that processed event count should - // make progress. Use `+2` instead of `+3` is because certify blob initiates a blob - // sync, and sync we don't upload the blob data, so it won't get processed. The point - // here is that the epoch change start event should be marked completed. + // Unblock the epoch change start event, and expect that the processed event count + // makes progress. All three new events are persisted: the certify event records the + // blob in the pending-recovery table instead of waiting for the blob sync (the blob + // data is never uploaded, so its sync cannot finish). unblock.notify_one(); - wait_until_events_processed_exact(&cluster.nodes[0], processed_event_count_initial + 2) + wait_until_events_processed_exact(&cluster.nodes[0], processed_event_count_initial + 3) .await?; + assert!( + cluster.nodes[0] + .storage_node + .inner + .storage + .scan_pending_recover_blobs()? + .iter() + .any(|(blob_id, _)| *blob_id == OTHER_BLOB_ID), + "the certified blob should have a pending-recovery record" + ); Ok(()) } @@ -10408,6 +10430,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..052656949e 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,241 @@ 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. + // TODO(WAL-1322): each pass rescans and sorts the whole table and + // re-runs the certification and stored-at-all-shards checks for every record not in flight; + // bound the per-pass work (chunked scanning with a resume cursor) so large backlogs do not + // amplify reads. + #[tracing::instrument(skip_all)] + async fn start_pending_recoveries( + &self, + in_flight: &mut FuturesUnordered>, + in_flight_blobs: &mut HashSet, + ) -> anyhow::Result<()> { + // TODO(WAL-1323): this scan blocks the async runtime thread + // and holds the whole table in memory for the duration of the pass; move it behind + // `spawn_blocking` and iterate in bounded chunks. + 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..55e5beebee 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, @@ -85,7 +83,7 @@ impl PendingRecoverBlobsTable { // Count with error propagation: an iterator that keeps yielding a read error would // otherwise be counted forever and hang the open. - // TODO(zhewu-create-issue-before-merging): this iterates the whole table (including + // TODO(WAL-1322): this iterates the whole table (including // RocksDB tombstones) and delays the storage open when the table is large; use a // cheaper way to initialize the count. let mut count: u64 = 0; @@ -142,10 +140,10 @@ impl PendingRecoverBlobsTable { } /// Returns all pending-recovery records. - // TODO(zhewu-create-issue-before-merging): this materializes the whole table in memory + // TODO(WAL-1322): this materializes the whole table in memory // (tens of MB per million records); replace with bounded chunked iteration with a resume // cursor so large backlogs are processed with capped memory. - // TODO(zhewu-create-issue-before-merging): the insert-then-delete churn of this table + // TODO(WAL-1324): the insert-then-delete churn of this table // leaves tombstones that slow scans until compaction; consider a periodic or post-drain // manual compaction of this column family. pub fn scan_all(&self) -> Result, TypedStoreError> { 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 @@