feat: return clear error on storage-node disk full instead of hanging

crates/walrus-service/node_config_example.yaml

@@ -102,6 +102,9 @@ live_upload_deferral:
     defer_secs: 30
   max_total_defer_secs: 120
   max_checkpoint_lag: 1500
+storage_write:
+  write_timeout_secs: 60
+  min_available_disk_space: 1073741824
 protocol_key_pair:
   path: /opt/walrus/config/protocol.key
 next_protocol_key_pair: null

crates/walrus-service/src/node.rs

@@ -158,6 +158,7 @@ use self::{
         RetrieveSymbolError,
         SetRecoveryDeferralError,
         ShardNotAssigned,
+        StorageWriteFailure,
         StoreMetadataError,
         StoreSliverError,
         SyncNodeConfigError,
@@ -195,7 +196,12 @@ use crate::{
     },
     node::{
         blob_event_processor::pending_events::PendingEventCounter,
-        config::{EpochStateConsistencyConfig, LiveUploadDeferralConfig, PriceCurrency},
+        config::{
+            EpochStateConsistencyConfig,
+            LiveUploadDeferralConfig,
+            PriceCurrency,
+            StorageWriteConfig,
+        },
         event_blob_writer::EventBlobWriter,
         garbage_collector::GarbageCollector,
         wal_price_monitor::WalPriceMonitor,
@@ -703,6 +709,7 @@ pub struct StorageNodeInner {
     recovery_deferral_notify: Arc<Notify>,
     recovery_deferral_cleanup_token: CancellationToken,
     live_upload_deferral_config: LiveUploadDeferralConfig,
+    storage_write_config: StorageWriteConfig,
     sliver_ref_cache: Cache<SliverRefCacheKey, Arc<RwLock<Weak<Sliver>>>>,
     #[cfg_attr(any(test, msim), allow(dead_code))]
     epoch_state_consistency_config: EpochStateConsistencyConfig,
@@ -910,6 +917,7 @@ impl StorageNode {
             recovery_deferral_notify: Arc::new(Notify::new()),
             recovery_deferral_cleanup_token: CancellationToken::new(),
             live_upload_deferral_config: config.live_upload_deferral.clone(),
+            storage_write_config: config.storage_write.clone(),
             sliver_ref_cache: Cache::builder()
                 .name("sliver-refs")
                 .eviction_policy(EvictionPolicy::lru())
@@ -3478,6 +3486,53 @@ impl StorageNodeInner {
         thread_pool::unwrap_or_resume_panic(result)
     }
 
+    /// Runs a database write future under the configured write timeout, rejecting it up-front if
+    /// the disk is already full.
+    ///
+    /// Ensures that a write which fails or stalls (for example because the disk is full and RocksDB
+    /// can no longer make progress) surfaces as an error to the caller instead of hanging
+    /// indefinitely. When the disk is (nearly) full, the failure is classified as an out-of-space
+    /// condition so that the client receives a clear signal that the node is out of space.
+    async fn run_storage_write<T>(
+        &self,
+        write: impl Future<Output = Result<T, TypedStoreError>>,
+    ) -> Result<T, StorageWriteFailure> {
+        let write_timeout = self.storage_write_config.write_timeout;
+        let result =
+            classify_storage_write(write_timeout, write, || self.disk_appears_full()).await;
+
+        match &result {
+            Err(StorageWriteFailure::OutOfSpace) => {
+                tracing::warn!("storage write rejected or failed: the disk appears full");
+            }
+            Err(StorageWriteFailure::TimedOut) => {
+                tracing::warn!(
+                    ?write_timeout,
+                    "storage write did not complete within the timeout"
+                );
+            }
+            // Database errors are propagated and reported by the caller, as before this
+            // classification existed.
+            Err(StorageWriteFailure::Database(_)) | Ok(_) => {}
+        }
+
+        result
+    }
+
+    /// Returns `true` if the database disk has less free space than the configured minimum.
+    fn disk_appears_full(&self) -> bool {
+        match self.storage.available_disk_space() {
+            Ok(available) => available < self.storage_write_config.min_available_disk_space,
+            Err(error) => {
+                tracing::warn!(
+                    ?error,
+                    "failed to query available disk space; assuming the disk is not full"
+                );
+                false
+            }
+        }
+    }
+
     async fn prepare_sliver_for_storage(
         &self,
         metadata: Arc<VerifiedBlobMetadataWithId>,
@@ -3534,10 +3589,9 @@ impl StorageNodeInner {
 
         let sliver_type = verified_sliver.r#type();
 
-        shard_storage
-            .put_sliver(*metadata.blob_id(), verified_sliver)
+        self.run_storage_write(shard_storage.put_sliver(*metadata.blob_id(), verified_sliver))
             .await
-            .context("unable to store sliver")?;
+            .map_err(StoreSliverError::from)?;
 
         walrus_utils::with_label!(self.metrics.slivers_stored_total, sliver_type).inc();
 
@@ -3561,11 +3615,9 @@ impl StorageNodeInner {
             return Ok(false);
         }
 
-        self.storage
-            .put_verified_metadata(&verified)
+        self.run_storage_write(self.storage.put_verified_metadata(&verified))
             .await
-            .context("unable to store metadata")
-            .map_err(StoreMetadataError::Internal)?;
+            .map_err(StoreMetadataError::from)?;
 
         self.pending_metadata_cache.remove(blob_id).await;
 
@@ -3626,11 +3678,9 @@ impl StorageNodeInner {
         }
 
         if let Some(metadata) = self.pending_metadata_cache.remove(blob_id).await {
-            self.storage
-                .put_verified_metadata(&metadata)
+            self.run_storage_write(self.storage.put_verified_metadata(&metadata))
                 .await
-                .context("unable to persist pending metadata")
-                .map_err(StoreMetadataError::Internal)?;
+                .map_err(StoreMetadataError::from)?;
 
             self.metrics
                 .uploaded_metadata_unencoded_blob_bytes
@@ -5109,6 +5159,35 @@ enum PendingCacheError {
     Sliver(StoreSliverError),
 }
 
+/// Classifies the outcome of a database write, bounding it by `write_timeout`.
+///
+/// Fails fast with [`StorageWriteFailure::OutOfSpace`] when the disk is already full, without
+/// dispatching the write: a stalled RocksDB write cannot be cancelled and would occupy a
+/// blocking-pool thread until it completes, so refusing up-front avoids both waiting out the
+/// timeout and piling up blocked threads while the disk remains full. Failures and timeouts of
+/// dispatched writes are likewise reported as out-of-space when the disk is full.
+async fn classify_storage_write<T>(
+    write_timeout: Duration,
+    write: impl Future<Output = Result<T, TypedStoreError>>,
+    disk_appears_full: impl Fn() -> bool,
+) -> Result<T, StorageWriteFailure> {
+    if disk_appears_full() {
+        return Err(StorageWriteFailure::OutOfSpace);
+    }
+
+    let failure = match tokio::time::timeout(write_timeout, write).await {
+        Ok(Ok(value)) => return Ok(value),
+        Ok(Err(error)) => StorageWriteFailure::Database(error),
+        Err(_elapsed) => StorageWriteFailure::TimedOut,
+    };
+
+    if disk_appears_full() {
+        Err(StorageWriteFailure::OutOfSpace)
+    } else {
+        Err(failure)
+    }
+}
+
 fn map_sliver_error_to_metadata(error: StoreSliverError) -> StoreMetadataError {
     match error {
         StoreSliverError::Internal(inner) => StoreMetadataError::Internal(inner),
@@ -5125,6 +5204,8 @@ fn map_sliver_error_to_metadata(error: StoreSliverError) -> StoreMetadataError {
             StoreMetadataError::Internal(anyhow!("sliver cache flush failed: {error:?}"))
         }
         StoreSliverError::ShardNotAssigned(inner) => StoreMetadataError::Internal(inner.into()),
+        StoreSliverError::OutOfSpace(inner) => StoreMetadataError::OutOfSpace(inner),
+        StoreSliverError::WriteTimeout(inner) => StoreMetadataError::WriteTimeout(inner),
     }
 }
 
@@ -5140,6 +5221,8 @@ fn map_metadata_error_to_sliver(error: StoreMetadataError) -> StoreSliverError {
             "metadata associated with event {event:?} was invalid"
         )),
         StoreMetadataError::Internal(inner) => StoreSliverError::Internal(inner),
+        StoreMetadataError::OutOfSpace(inner) => StoreSliverError::OutOfSpace(inner),
+        StoreMetadataError::WriteTimeout(inner) => StoreSliverError::WriteTimeout(inner),
     }
 }
 
@@ -5257,6 +5340,70 @@ mod tests {
             .expect("storage node creation in setup should not fail")
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn classify_storage_write_times_out_stalled_write() {
+        let result = classify_storage_write(
+            Duration::from_secs(1),
+            std::future::pending::<Result<(), TypedStoreError>>(),
+            || false,
+        )
+        .await;
+
+        assert!(matches!(result, Err(StorageWriteFailure::TimedOut)));
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn classify_storage_write_rejects_up_front_when_disk_full() {
+        // The write never resolves; the up-front disk check must reject it without waiting for
+        // the timeout to elapse.
+        let result = tokio::time::timeout(
+            Duration::from_millis(1),
+            classify_storage_write(
+                Duration::from_secs(3600),
+                std::future::pending::<Result<(), TypedStoreError>>(),
+                || true,
+            ),
+        )
+        .await
+        .expect("the write must be rejected before its own timeout can elapse");
+
+        assert!(matches!(result, Err(StorageWriteFailure::OutOfSpace)));
+    }
+
+    #[tokio::test]
+    async fn classify_storage_write_passes_through_success_and_database_errors() {
+        let success =
+            classify_storage_write(Duration::from_secs(1), async { Ok(7) }, || false).await;
+        assert_eq!(success.expect("write should succeed"), 7);
+
+        let failure = classify_storage_write(
+            Duration::from_secs(1),
+            async { Err::<(), _>(TypedStoreError::RocksDBError("boom".to_owned())) },
+            || false,
+        )
+        .await;
+        assert!(matches!(failure, Err(StorageWriteFailure::Database(_))));
+    }
+
+    #[tokio::test]
+    async fn classify_storage_write_reports_out_of_space_when_write_fails_on_full_disk() {
+        // The disk fills up while the write is in flight: the up-front check passes, the write
+        // fails, and the failure must be classified as out-of-space.
+        let probe_calls = std::cell::Cell::new(0);
+        let result = classify_storage_write(
+            Duration::from_secs(1),
+            async { Err::<(), _>(TypedStoreError::RocksDBError("no space".to_owned())) },
+            || {
+                let calls = probe_calls.get();
+                probe_calls.set(calls + 1);
+                calls > 0
+            },
+        )
+        .await;
+
+        assert!(matches!(result, Err(StorageWriteFailure::OutOfSpace)));
+    }
+
     #[tokio::test]
     async fn wait_until_recovery_deferral_expires_returns_false_without_deferral() -> TestResult {
         let _lock = global_test_lock().lock().await;

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

@@ -229,6 +229,9 @@ pub struct StorageNodeConfig {
     /// Configuration for deferring recovery while uploads are in progress.
     #[serde(default, skip_serializing_if = "defaults::is_default")]
     pub live_upload_deferral: LiveUploadDeferralConfig,
+    /// Configuration governing storage write timeouts and disk-full handling.
+    #[serde(default, skip_serializing_if = "defaults::is_default")]
+    pub storage_write: StorageWriteConfig,
     /// Key pair used in Walrus protocol messages.
     // Important: this name should be in-sync with the name used in `rotate_protocol_key_pair()`
     #[serde_as(as = "PathOrInPlace<Base64>")]
@@ -530,6 +533,7 @@ impl Default for StorageNodeConfig {
                 max_total_defer: Duration::from_secs(120),
                 max_checkpoint_lag: 1500,
             },
+            storage_write: Default::default(),
             disable_event_blob_writer: Default::default(),
             commission_rate: defaults::commission_rate(),
             voting_params: VotingParamsConfig {
@@ -1304,6 +1308,36 @@ pub struct SizeDeferralEntry {
     pub defer: Duration,
 }
 
+/// Configuration governing storage writes, ensuring a full disk surfaces as a clear client error
+/// instead of an indefinite hang.
+#[serde_as]
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
+#[serde(default)]
+pub struct StorageWriteConfig {
+    /// Maximum time to wait for a metadata or sliver write to the database to complete.
+    ///
+    /// If the write does not complete within this duration (for example because RocksDB has stalled
+    /// on a full disk), the request returns an error instead of blocking indefinitely.
+    #[serde_as(as = "DurationSeconds<u64>")]
+    #[serde(rename = "write_timeout_secs")]
+    pub write_timeout: Duration,
+    /// Minimum free disk space, in bytes, below which the node considers itself out of space.
+    ///
+    /// When a write fails or times out and the available disk space is below this threshold, the
+    /// node returns a dedicated out-of-space error so clients get a clear signal that the node is
+    /// full.
+    pub min_available_disk_space: u64,
+}
+
+impl Default for StorageWriteConfig {
+    fn default() -> Self {
+        Self {
+            write_timeout: Duration::from_secs(60),
+            min_available_disk_space: 1 << 30, // 1 GiB
+        }
+    }
+}
+
 /// Configuration that controls deferring recovery when a live client upload is likely.
 #[serde_as]
 #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]