feat(node): add a durable table of blobs pending recovery

crates/walrus-service/node_config_example.yaml

@@ -79,6 +79,7 @@ db_config:
   per_object_pooled_blob_info: null
   storage_pool_info: null
   event_cursor: null
+  pending_recover_blobs: null
   shard: null
   shard_status: null
   shard_sync_progress: null
@@ -150,6 +151,8 @@ blob_recovery:
   node_connect_timeout_secs: 1
   experimental_sliver_recovery_additional_symbols: 0
   monitor_interval_secs: 60
+  max_concurrent_pending_recoveries: 100
+  pending_recovery_drain_interval_secs: 60
 tls:
   disable_tls: false
   certificate_path: null

crates/walrus-service/src/node/config.rs

@@ -1038,6 +1038,13 @@ pub struct BlobRecoveryConfig {
     #[serde_as(as = "DurationSeconds")]
     #[serde(rename = "monitor_interval_secs")]
     pub monitor_interval: Duration,
+    /// The maximum number of concurrent blob syncs started by the pending-recovery executor.
+    pub max_concurrent_pending_recoveries: usize,
+    /// The fallback interval at which the pending-recovery executor re-scans the
+    /// pending-recovery table (it is additionally woken whenever a record is inserted).
+    #[serde_as(as = "DurationSeconds")]
+    #[serde(rename = "pending_recovery_drain_interval_secs")]
+    pub pending_recovery_drain_interval: Duration,
 }
 
 impl Default for BlobRecoveryConfig {
@@ -1048,16 +1055,19 @@ impl Default for BlobRecoveryConfig {
             max_proof_cache_elements: 7_500,
             committee_service_config: CommitteeServiceConfig::default(),
             monitor_interval: Duration::from_mins(1),
+            max_concurrent_pending_recoveries: 100,
+            pending_recovery_drain_interval: Duration::from_mins(1),
         }
     }
 }
 
 impl BlobRecoveryConfig {
-    /// Returns a default configuration with a shorter monitor interval for testing.
+    /// Returns a default configuration with shorter intervals for testing.
     #[cfg(any(test, feature = "test-utils"))]
     pub fn default_for_test() -> Self {
         Self {
             monitor_interval: Duration::from_secs(5),
+            pending_recovery_drain_interval: Duration::from_secs(1),
             ..Default::default()
         }
     }

crates/walrus-service/src/node/dbtool.rs

@@ -56,6 +56,7 @@ use crate::{
             pending_cf_name,
         },
         storage::{
+            PendingRecoverBlob,
             PrimarySliverData,
             SecondarySliverData,
             blob_info::{
@@ -74,6 +75,7 @@ use crate::{
                 garbage_collector_table_cf_name,
                 metadata_cf_name,
                 node_status_cf_name,
+                pending_recover_blobs_cf_name,
                 per_object_blob_info_cf_name,
                 per_object_pooled_blob_info_cf_name,
                 primary_slivers_column_family_name,
@@ -208,6 +210,20 @@ pub enum DbToolCommands {
         count: usize,
     },
 
+    /// Read pending-recovery records from the RocksDB database.
+    ReadPendingRecoverBlobs {
+        /// Path to the RocksDB database directory.
+        #[arg(long)]
+        db_path: PathBuf,
+        /// Start blob ID in URL-safe base64 format (no padding).
+        #[arg(long)]
+        #[serde_as(as = "Option<DisplayFromStr>")]
+        start_blob_id: Option<BlobId>,
+        /// Number of entries to scan.
+        #[arg(long, default_value = "1")]
+        count: usize,
+    },
+
     /// Count the number of certified blobs in the RocksDB database.
     CountCertifiedBlobs {
         /// Path to the RocksDB database directory.
@@ -420,6 +436,11 @@ impl DbToolCommands {
                 start_object_id,
                 count,
             } => read_object_blob_info(db_path, start_object_id, count),
+            Self::ReadPendingRecoverBlobs {
+                db_path,
+                start_blob_id,
+                count,
+            } => read_pending_recover_blobs(db_path, start_blob_id, count),
             Self::CountCertifiedBlobs { db_path, epoch } => count_certified_blobs(db_path, epoch),
             Self::DropColumnFamilies {
                 db_path,
@@ -750,6 +771,53 @@ fn read_blob_info(db_path: PathBuf, start_blob_id: Option<BlobId>, count: usize)
     Ok(())
 }
 
+fn read_pending_recover_blobs(
+    db_path: PathBuf,
+    start_blob_id: Option<BlobId>,
+    count: usize,
+) -> Result<()> {
+    let pending_recover_blobs_options =
+        DatabaseTableOptionsFactory::new(DatabaseConfig::default(), false).pending_recover_blobs();
+    let db = DB::open_cf_with_opts_for_read_only(
+        &RocksdbOptions::default(),
+        db_path,
+        [(
+            pending_recover_blobs_cf_name(),
+            pending_recover_blobs_options,
+        )],
+        false,
+    )?;
+
+    let cf = db
+        .cf_handle(pending_recover_blobs_cf_name())
+        .expect("pending recover blobs column family should exist");
+
+    let iter = if let Some(blob_id) = start_blob_id {
+        db.iterator_cf(
+            &cf,
+            rocksdb::IteratorMode::From(&be_fix_int_ser(&blob_id)?, rocksdb::Direction::Forward),
+        )
+    } else {
+        db.iterator_cf(&cf, rocksdb::IteratorMode::Start)
+    };
+
+    for result in iter.take(count) {
+        match result {
+            Ok((key, value)) => {
+                let blob_id: BlobId = bcs::from_bytes(&key)?;
+                let record: PendingRecoverBlob = bcs::from_bytes(&value)?;
+                println!("Blob ID: {blob_id}, PendingRecoverBlob: {record:?}");
+            }
+            Err(e) => {
+                println!("Error: {e:?}");
+                return Err(e.into());
+            }
+        }
+    }
+
+    Ok(())
+}
+
 fn read_object_blob_info(
     db_path: PathBuf,
     start_object_id: Option<ObjectID>,
@@ -1272,6 +1340,7 @@ fn report_storage_probe(db: &DB, column_families: &[String], exact_counts: bool)
         node_status_cf_name(),
         event_cursor_cf_name(),
         event_index_cf_name(),
+        pending_recover_blobs_cf_name(),
         garbage_collector_table_cf_name(),
     ] {
         if column_families.iter().any(|name| name == cf_name) {
@@ -1517,6 +1586,7 @@ fn cf_options_for_name(
                 name if name == metadata_cf_name() => factory.metadata(),
                 name if name == node_status_cf_name() => factory.node_status(),
                 name if name == event_cursor_cf_name() => event_cursor_cf_options(factory),
+                name if name == pending_recover_blobs_cf_name() => factory.pending_recover_blobs(),
                 name if name == garbage_collector_table_cf_name() => factory.garbage_collector(),
                 _ => RocksdbOptions::default(),
             }

crates/walrus-service/src/node/metrics.rs

@@ -132,6 +132,16 @@ walrus_utils::metrics::define_metric_set! {
         #[help = "The number of blob recoveries currently pending"]
         recover_blob_backlog: IntGaugeVec["state"],
 
+        #[help = "The number of records in the pending-recovery table"]
+        pending_recover_blob_count: U64Gauge[],
+
+        #[help = "The event index of the oldest record in the pending-recovery table, or 0 if \
+        the table is empty"]
+        pending_recover_blob_oldest_event_index: U64Gauge[],
+
+        #[help = "The number of pending-recovery records handled by the executor, by outcome"]
+        pending_recovery_executor_total: IntCounterVec["outcome"],
+
         #[help = "Time (in seconds) spent processing events"]
         event_process_duration_seconds: HistogramVec["event_type"],
 

crates/walrus-service/src/node/storage.rs

@@ -80,6 +80,10 @@ pub(crate) use event_cursor_table::event_cursor_cf_options;
 
 mod event_sequencer;
 mod metrics;
+mod pending_recover_blobs;
+pub(crate) use pending_recover_blobs::PendingRecoverBlob;
+use pending_recover_blobs::PendingRecoverBlobsTable;
+
 mod shard;
 
 pub(crate) use shard::{PrimarySliverData, SecondarySliverData, ShardStatus, ShardStorage};
@@ -342,6 +346,7 @@ pub struct Storage {
     metadata: DBMap<BlobId, BlobMetadata>,
     blob_info: BlobInfoTable,
     event_cursor: EventCursorTable,
+    pending_recover_blobs: PendingRecoverBlobsTable,
     garbage_collector_table: DBMap<String, Epoch>,
     shards: Arc<RwLock<HashMap<ShardIndex, Arc<ShardStorage>>>>,
     db_table_opts_factory: DatabaseTableOptionsFactory,
@@ -440,6 +445,8 @@ impl Storage {
         let blob_info_column_families = BlobInfoTable::options(&db_table_opts_factory);
         let (event_cursor_cf_name, event_cursor_options) =
             EventCursorTable::options(&db_table_opts_factory);
+        let (pending_recover_blobs_cf_name, pending_recover_blobs_options) =
+            PendingRecoverBlobsTable::options(&db_table_opts_factory);
         let garbage_collector_table_cf_name = garbage_collector_table_cf_name();
         let garbage_collector_table_options = db_table_opts_factory.garbage_collector();
 
@@ -450,6 +457,7 @@ impl Storage {
                 (node_status_cf_name, node_status_options),
                 (metadata_cf_name, metadata_options),
                 (event_cursor_cf_name, event_cursor_options),
+                (pending_recover_blobs_cf_name, pending_recover_blobs_options),
                 (
                     garbage_collector_table_cf_name,
                     garbage_collector_table_options,
@@ -515,6 +523,7 @@ impl Storage {
         )?;
 
         let event_cursor = EventCursorTable::reopen(&database)?;
+        let pending_recover_blobs = PendingRecoverBlobsTable::reopen(&database)?;
         let blob_info = BlobInfoTable::reopen(&database)?;
         let shards = Arc::new(RwLock::new(
             existing_shards_ids
@@ -543,6 +552,7 @@ impl Storage {
             metadata,
             blob_info,
             event_cursor,
+            pending_recover_blobs,
             garbage_collector_table,
             shards,
             db_table_opts_factory,
@@ -1508,6 +1518,48 @@ impl Storage {
         self.blob_info.get_latest_handled_event_index()
     }
 
+    /// Durably records that the blob needs to be recovered, overwriting any existing record, and
+    /// returns the number of pending-recovery records.
+    ///
+    /// 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,
+        event_index: u64,
+        certified_epoch: Epoch,
+    ) -> Result<u64, TypedStoreError> {
+        self.pending_recover_blobs.insert(
+            blob_id,
+            &PendingRecoverBlob::new(event_index, certified_epoch),
+        )
+    }
+
+    /// 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,
+    ) -> Result<u64, TypedStoreError> {
+        self.pending_recover_blobs.delete(blob_id)
+    }
+
+    /// 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<Vec<(BlobId, PendingRecoverBlob)>, TypedStoreError> {
+        self.pending_recover_blobs.scan_all()
+    }
+
+    /// 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()
+    }
+
     /// Clears the metadata in the storage for testing purposes.
     #[cfg(test)]
     pub fn clear_metadata_in_test(&self) -> Result<(), TypedStoreError> {

crates/walrus-service/src/node/storage/constants.rs

@@ -16,6 +16,7 @@ const EVENT_CURSOR_KEY: [u8; 6] = *b"cursor";
 const GARBAGE_COLLECTOR_TABLE_COLUMN_FAMILY_NAME: &str = "garbage_collector_last_completed_epoch";
 const GARBAGE_COLLECTOR_LAST_STARTED_EPOCH_KEY: &str = "started";
 const GARBAGE_COLLECTOR_LAST_COMPLETED_EPOCH_KEY: &str = "completed";
+const PENDING_RECOVER_BLOBS_COLUMN_FAMILY_NAME: &str = "pending_recover_blobs";
 
 // Base name for shard-related column families
 const SHARD_BASE_COLUMN_FAMILY_NAME: &str = "shard";
@@ -87,6 +88,11 @@ pub fn garbage_collector_last_completed_epoch_key() -> String {
     GARBAGE_COLLECTOR_LAST_COMPLETED_EPOCH_KEY.to_string()
 }
 
+/// Returns the name of the pending recover blobs column family.
+pub fn pending_recover_blobs_cf_name() -> &'static str {
+    PENDING_RECOVER_BLOBS_COLUMN_FAMILY_NAME
+}
+
 /// Returns the column family name for primary slivers of a shard.
 pub fn primary_slivers_column_family_name(id: ShardIndex) -> String {
     format!(
@@ -144,6 +150,7 @@ mod tests {
         assert_eq!(per_object_blob_info_cf_name(), "per_object_blob_info");
         assert_eq!(node_status_cf_name(), "node_status");
         assert_eq!(event_index_cf_name(), "latest_handled_event_index");
+        assert_eq!(pending_recover_blobs_cf_name(), "pending_recover_blobs");
 
         let shard = ShardIndex(900);
         assert_eq!(base_column_family_name(shard), "shard-900");

crates/walrus-service/src/node/storage/database_config.rs

@@ -348,6 +348,8 @@ pub struct DatabaseConfig {
     pub(super) storage_pool_info: Option<DatabaseTableOptions>,
     /// Event cursor database options.
     pub(super) event_cursor: Option<DatabaseTableOptions>,
+    /// Pending recover blobs database options.
+    pub(super) pending_recover_blobs: Option<DatabaseTableOptions>,
     /// Shard database options.
     pub(super) shard: Option<DatabaseTableOptions>,
     /// Shard status database options.
@@ -468,6 +470,11 @@ impl DatabaseConfig {
         Self::inherit_from_or_use_template(&self.event_cursor, self.standard())
     }
 
+    /// Returns the pending recover blobs database option.
+    pub fn pending_recover_blobs(&self) -> DatabaseTableOptions {
+        Self::inherit_from_or_use_template(&self.pending_recover_blobs, self.standard())
+    }
+
     /// Returns the shard database option.
     pub fn shard(&self) -> DatabaseTableOptions {
         Self::inherit_from_or_use_template(&self.shard, self.optimized_for_blobs())
@@ -549,6 +556,7 @@ impl Default for DatabaseConfig {
             per_object_pooled_blob_info: None,
             storage_pool_info: None,
             event_cursor: None,
+            pending_recover_blobs: None,
             shard: None,
             shard_status: None,
             shard_sync_progress: None,
@@ -771,6 +779,11 @@ impl DatabaseTableOptionsFactory {
         self.to_options(&self.config.event_cursor(), false)
     }
 
+    /// Returns the pending recover blobs database option.
+    pub fn pending_recover_blobs(&self) -> Options {
+        self.to_options(&self.config.pending_recover_blobs(), false)
+    }
+
     // Below 4 options are for shard storage column families.
 
     /// Returns the shard database option with shared cache.