feat: blob sliver data existence check

.pre-commit-config.yaml

@@ -76,6 +76,7 @@ repos:
         language: rust
         files: (^crates/|Cargo\.(toml|lock)$|nextest\.toml$)
         pass_filenames: false
+        verbose: true
       - id: cargo-doctests
         name: cargo-doctests
         entry: cargo test --doc
@@ -122,5 +123,5 @@ repos:
         name: Build docs if mdBook is installed
         language: system
         entry: bash -c 'which mdbook && mdbook build || ! which mdbook >> /dev/null'
-        files: 'book.toml|docs/.*'
+        files: "book.toml|docs/.*"
         pass_filenames: false

crates/walrus-service/node_config_example.yaml

@@ -148,3 +148,7 @@ balance_check:
 thread_pool:
   max_concurrent_tasks: null
   max_blocking_io_threads: 1024
+consistency_check:
+  enable_consistency_check: true
+  enable_sliver_data_existence_check: false
+  sliver_data_existence_check_sample_rate_percentage: 100

crates/walrus-service/src/node.rs

@@ -16,6 +16,7 @@ use std::{
 use anyhow::{Context, anyhow, bail};
 use blob_retirement_notifier::BlobRetirementNotifier;
 use committee::{BeginCommitteeChangeError, EndCommitteeChangeError};
+use consistency_check::StorageNodeConsistencyCheckConfig;
 use epoch_change_driver::EpochChangeDriver;
 use errors::{ListSymbolsError, Unavailable};
 use events::{CheckpointEventPosition, event_blob_writer::EventBlobWriter};
@@ -166,6 +167,7 @@ use crate::{
 
 pub mod committee;
 pub mod config;
+pub mod consistency_check;
 pub mod contract_service;
 pub mod dbtool;
 pub mod events;
@@ -176,7 +178,6 @@ pub(crate) mod metrics;
 
 mod blob_retirement_notifier;
 mod blob_sync;
-mod consistency_check;
 mod epoch_change_driver;
 mod node_recovery;
 mod recovery_symbol_service;
@@ -517,6 +518,7 @@ pub struct StorageNodeInner {
     thread_pool: BoundedThreadPool,
     registry: Registry,
     latest_event_epoch: AtomicU32, // The epoch of the latest event processed by the node.
+    consistency_check_config: StorageNodeConsistencyCheckConfig,
 }
 
 /// Parameters for configuring and initializing a node.
@@ -629,6 +631,7 @@ impl StorageNode {
             encoding_config,
             registry: registry.clone(),
             latest_event_epoch: AtomicU32::new(0),
+            consistency_check_config: config.consistency_check.clone(),
         });
 
         blocklist.start_refresh_task();
@@ -1329,17 +1332,20 @@ impl StorageNode {
             .wait_until_previous_task_done()
             .await;
 
-        if let Err(err) = consistency_check::schedule_background_consistency_check(
-            self.inner.clone(),
-            event.epoch,
-        )
-        .await
-        {
-            tracing::warn!(
-                ?err,
-                epoch = %event.epoch,
-                "failed to schedule background blob info consistency check"
-            );
+        if self.inner.consistency_check_config.enable_consistency_check {
+            if let Err(err) = consistency_check::schedule_background_consistency_check(
+                self.inner.clone(),
+                self.blob_sync_handler.clone(),
+                event.epoch,
+            )
+            .await
+            {
+                tracing::warn!(
+                    ?err,
+                    epoch = %event.epoch,
+                    "failed to schedule background blob info consistency check"
+                );
+            }
         }
 
         // During epoch change, we need to lock the read access to shard map until all the new
@@ -1874,8 +1880,16 @@ impl StorageNodeInner {
             None => self.owned_shards_at_latest_epoch(),
         };
 
+        self.is_stored_at_specific_shards(blob_id, &shards).await
+    }
+
+    async fn is_stored_at_specific_shards(
+        &self,
+        blob_id: &BlobId,
+        shards: &[ShardIndex],
+    ) -> anyhow::Result<bool> {
         for shard in shards {
-            match self.storage.is_stored_at_shard(blob_id, shard).await {
+            match self.storage.is_stored_at_shard(blob_id, *shard).await {
                 Ok(false) => return Ok(false),
                 Ok(true) => continue,
                 Err(error) => {
@@ -1907,6 +1921,27 @@ impl StorageNodeInner {
             .await
     }
 
+    /// Returns true if the blob is stored at all shards that are in Active state.
+    pub(crate) async fn is_stored_at_all_active_shards(
+        &self,
+        blob_id: &BlobId,
+    ) -> anyhow::Result<bool> {
+        let shards = self
+            .storage
+            .existing_shard_storages()
+            .await
+            .iter()
+            .filter_map(|shard_storage| {
+                shard_storage
+                    .status()
+                    .is_ok_and(|status| status == ShardStatus::Active)
+                    .then_some(shard_storage.id())
+            })
+            .collect::<Vec<_>>();
+
+        self.is_stored_at_specific_shards(blob_id, &shards).await
+    }
+
     pub(crate) fn storage(&self) -> &Storage {
         &self.storage
     }

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

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use std::{
-    collections::{HashMap, hash_map::Entry},
+    collections::{HashMap, HashSet, hash_map::Entry},
     ops::Not,
     sync::{Arc, Mutex},
     time::Duration,
@@ -378,6 +378,16 @@ impl BlobSyncHandler {
 
         Ok(count)
     }
+
+    /// Returns the list of blob ids that are currently being synced.
+    pub fn blob_sync_in_progress(&self) -> HashSet<BlobId> {
+        self.blob_syncs_in_progress
+            .lock()
+            .expect("should be able to acquire lock")
+            .keys()
+            .cloned()
+            .collect()
+    }
 }
 
 type SyncJoinHandle = JoinHandle<Option<EventHandle>>;

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

@@ -40,7 +40,7 @@ use walrus_sui::types::{
     move_structs::{NodeMetadata, VotingParams},
 };
 
-use super::storage::DatabaseConfig;
+use super::{consistency_check::StorageNodeConsistencyCheckConfig, storage::DatabaseConfig};
 use crate::{
     common::{config::SuiConfig, utils},
     node::events::EventProcessorConfig,
@@ -171,6 +171,9 @@ pub struct StorageNodeConfig {
     /// Configuration for the blocking thread pool.
     #[serde(default, skip_serializing_if = "defaults::is_default")]
     pub thread_pool: ThreadPoolConfig,
+    /// Configuration for the consistency check.
+    #[serde(default, skip_serializing_if = "defaults::is_default")]
+    pub consistency_check: StorageNodeConsistencyCheckConfig,
 }
 
 impl Default for StorageNodeConfig {
@@ -209,6 +212,7 @@ impl Default for StorageNodeConfig {
             num_uncertified_blob_threshold: None,
             balance_check: Default::default(),
             thread_pool: Default::default(),
+            consistency_check: Default::default(),
         }
     }
 }