diff --git a/crates/walrus-service/node_config_example.yaml b/crates/walrus-service/node_config_example.yaml index a93abc8e90..3869e5ea21 100644 --- a/crates/walrus-service/node_config_example.yaml +++ b/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 diff --git a/crates/walrus-service/src/node/config.rs b/crates/walrus-service/src/node/config.rs index 616bffbe6b..d5ff67dc5b 100644 --- a/crates/walrus-service/src/node/config.rs +++ b/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() } } diff --git a/crates/walrus-service/src/node/dbtool.rs b/crates/walrus-service/src/node/dbtool.rs index cf5299f99e..29a57c39b5 100644 --- a/crates/walrus-service/src/node/dbtool.rs +++ b/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")] + start_blob_id: Option, + /// 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, count: usize) Ok(()) } +fn read_pending_recover_blobs( + db_path: PathBuf, + start_blob_id: Option, + 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, @@ -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(), } diff --git a/crates/walrus-service/src/node/metrics.rs b/crates/walrus-service/src/node/metrics.rs index bcb6a3db1f..1604c0e781 100644 --- a/crates/walrus-service/src/node/metrics.rs +++ b/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"], diff --git a/crates/walrus-service/src/node/storage.rs b/crates/walrus-service/src/node/storage.rs index 9bacdaf25d..27ad97f46f 100644 --- a/crates/walrus-service/src/node/storage.rs +++ b/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, blob_info: BlobInfoTable, event_cursor: EventCursorTable, + pending_recover_blobs: PendingRecoverBlobsTable, garbage_collector_table: DBMap, shards: Arc>>>, 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 { + 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 { + 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, 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> { diff --git a/crates/walrus-service/src/node/storage/constants.rs b/crates/walrus-service/src/node/storage/constants.rs index b8cd6b2068..79898fcef3 100644 --- a/crates/walrus-service/src/node/storage/constants.rs +++ b/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"); diff --git a/crates/walrus-service/src/node/storage/database_config.rs b/crates/walrus-service/src/node/storage/database_config.rs index 40b6fe2d4e..8e9c25233e 100644 --- a/crates/walrus-service/src/node/storage/database_config.rs +++ b/crates/walrus-service/src/node/storage/database_config.rs @@ -348,6 +348,8 @@ pub struct DatabaseConfig { pub(super) storage_pool_info: Option, /// Event cursor database options. pub(super) event_cursor: Option, + /// Pending recover blobs database options. + pub(super) pending_recover_blobs: Option, /// Shard database options. pub(super) shard: Option, /// 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. diff --git a/crates/walrus-service/src/node/storage/pending_recover_blobs.rs b/crates/walrus-service/src/node/storage/pending_recover_blobs.rs new file mode 100644 index 0000000000..0d293eb24d --- /dev/null +++ b/crates/walrus-service/src/node/storage/pending_recover_blobs.rs @@ -0,0 +1,202 @@ +// Copyright (c) Walrus Foundation +// SPDX-License-Identifier: Apache-2.0 + +//! Durable table of blobs whose recovery is pending. +//! +//! A record is inserted when a certify event needs a blob sync, before the event is marked as +//! complete. This way, a restart cannot lose the fact that the blob still needs recovery, and +//! the event cursor does not have to wait for the recovery to finish. Records are deleted when +//! the blob is recovered or retired (deleted, invalidated, or expired). + +use std::sync::{ + Arc, + Mutex, + atomic::{AtomicU64, Ordering}, +}; + +use rocksdb::Options; +use serde::{Deserialize, Serialize}; +use typed_store::{ + Map, + TypedStoreError, + rocks::{DBMap, ReadWriteOptions, RocksDB}, +}; +use walrus_core::{BlobId, Epoch}; + +use super::{DatabaseTableOptionsFactory, constants::pending_recover_blobs_cf_name}; + +/// A record of a blob whose recovery is pending. +// Important: this enum is committed to database. Only extend it with new variants. +#[derive(Eq, PartialEq, Debug, Clone, Deserialize, Serialize)] +pub(crate) enum PendingRecoverBlob { + V1(PendingRecoverBlobV1), +} + +impl PendingRecoverBlob { + /// Creates a `V1` pending-recovery record. + pub fn new(event_index: u64, certified_epoch: Epoch) -> Self { + Self::V1(PendingRecoverBlobV1 { + event_index, + certified_epoch, + }) + } + + /// 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, + } + } + + /// 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, + } + } +} + +#[derive(Eq, PartialEq, Debug, Clone, Deserialize, Serialize)] +pub(crate) struct PendingRecoverBlobV1 { + event_index: u64, + certified_epoch: Epoch, +} + +#[derive(Debug, Clone)] +pub(super) struct PendingRecoverBlobsTable { + inner: DBMap, + // Serializes inserts and deletes so that the cached count stays exact. + mutation_lock: Arc>, + // Cached number of records for cheap metric and health reads. + count: Arc, +} + +impl PendingRecoverBlobsTable { + pub fn reopen(database: &Arc) -> Result { + let inner: DBMap = DBMap::reopen( + database, + Some(pending_recover_blobs_cf_name()), + &ReadWriteOptions::default(), + false, + )?; + + // 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 + // 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; + for entry in inner.safe_iter()? { + entry?; + count += 1; + } + + Ok(Self { + inner, + mutation_lock: Arc::default(), + count: Arc::new(AtomicU64::new(count)), + }) + } + + pub fn options(db_table_opts_factory: &DatabaseTableOptionsFactory) -> (&'static str, Options) { + ( + pending_recover_blobs_cf_name(), + db_table_opts_factory.pending_recover_blobs(), + ) + } + + /// Inserts or overwrites the pending-recovery record for `blob_id` and returns the number of + /// records in the table. + pub fn insert( + &self, + blob_id: &BlobId, + record: &PendingRecoverBlob, + ) -> Result { + let _guard = self + .mutation_lock + .lock() + .expect("mutex should not be poisoned"); + let existed = self.inner.contains_key(blob_id)?; + self.inner.insert(blob_id, record)?; + if !existed { + self.count.fetch_add(1, Ordering::SeqCst); + } + Ok(self.count.load(Ordering::SeqCst)) + } + + /// Deletes the pending-recovery record for `blob_id`, if any, and returns the number of + /// records remaining in the table. + pub fn delete(&self, blob_id: &BlobId) -> Result { + let _guard = self + .mutation_lock + .lock() + .expect("mutex should not be poisoned"); + if self.inner.contains_key(blob_id)? { + self.inner.remove(blob_id)?; + self.count.fetch_sub(1, Ordering::SeqCst); + } + Ok(self.count.load(Ordering::SeqCst)) + } + + /// Returns all pending-recovery records. + // TODO(zhewu-create-issue-before-merging): 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 + // 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> { + self.inner.safe_iter()?.collect() + } + + /// Returns the number of pending-recovery records. + pub fn count(&self) -> u64 { + self.count.load(Ordering::SeqCst) + } +} + +#[cfg(test)] +mod tests { + use walrus_core::{ShardIndex, test_utils::random_blob_id}; + use walrus_test_utils::Result as TestResult; + + use super::*; + use crate::test_utils::empty_storage_with_shards; + + #[tokio::test] + async fn insert_delete_and_count() -> TestResult { + let storage = empty_storage_with_shards(&[ShardIndex(0)]).await; + let storage = storage.as_ref(); + + let blob_id_1 = random_blob_id(); + let blob_id_2 = random_blob_id(); + + assert_eq!(storage.pending_recover_blob_count(), 0); + assert_eq!(storage.insert_pending_recover_blob(&blob_id_1, 7, 2)?, 1); + // Overwriting an existing record does not change the count. + assert_eq!(storage.insert_pending_recover_blob(&blob_id_1, 9, 2)?, 1); + assert_eq!(storage.insert_pending_recover_blob(&blob_id_2, 11, 3)?, 2); + + let mut records = storage.scan_pending_recover_blobs()?; + records.sort_by_key(|(_, record)| record.event_index()); + assert_eq!( + records, + vec![ + (blob_id_1, PendingRecoverBlob::new(9, 2)), + (blob_id_2, PendingRecoverBlob::new(11, 3)), + ] + ); + + assert_eq!(storage.delete_pending_recover_blob(&blob_id_1)?, 1); + // Deleting a non-existent record is a no-op. + assert_eq!(storage.delete_pending_recover_blob(&blob_id_1)?, 1); + assert_eq!(storage.delete_pending_recover_blob(&blob_id_2)?, 0); + assert!(storage.scan_pending_recover_blobs()?.is_empty()); + assert_eq!(storage.pending_recover_blob_count(), 0); + + Ok(()) + } +}