fix(storage): bound states GC backlog

core/src/block_strider/state_applier.rs

@@ -364,7 +364,8 @@ impl<S> Inner<S> {
                 .await?;
             blocks_gc_handles.extend_from_slice(&queue_deps);
 
-            block_handles.set_skip_states_gc(&block_handle);
+            self.prepare_persistent_shard_state(&block_handle, mc_block_id.seqno)
+                .await?;
             state_gc_handles.push(block_handle);
         }
 
@@ -373,7 +374,8 @@ impl<S> Inner<S> {
             .await?;
         blocks_gc_handles.extend_from_slice(&queue_deps);
 
-        block_handles.set_skip_states_gc(&mc_block_handle);
+        self.prepare_persistent_shard_state(&mc_block_handle, mc_block_id.seqno)
+            .await?;
         state_gc_handles.push(mc_block_handle);
 
         Ok(StorePersistentStateDeps {
@@ -382,13 +384,57 @@ impl<S> Inner<S> {
         })
     }
 
+    async fn prepare_persistent_shard_state(
+        &self,
+        handle: &BlockHandle,
+        mc_seqno: u32,
+    ) -> Result<()> {
+        let state_storage = self.storage.shard_state_storage();
+
+        // Ensure skipped state is stored in DB before saving persistent state.
+        if !state_storage.protect_state_from_gc(handle).await? {
+            let state = state_storage
+                .load_state(mc_seqno, handle.id())
+                .await
+                .with_context(|| {
+                    format!(
+                        "failed to reconstruct shard state for persistent save: {}",
+                        handle.id()
+                    )
+                })?;
+
+            state_storage
+                .store_state_ignore_cache(handle, &state, StoreStateHint {
+                    is_top_block: Some(true),
+                    ..Default::default()
+                })
+                .await
+                .with_context(|| {
+                    format!(
+                        "failed to store reconstructed shard state for persistent save: {}",
+                        handle.id()
+                    )
+                })?;
+        }
+
+        anyhow::ensure!(
+            state_storage
+                .load_state_root_hash_opt(handle.id())?
+                .is_some(),
+            "shard state row is missing after persistent-state preparation: {}",
+            handle.id(),
+        );
+        Ok(())
+    }
+
     async fn save_persistent_states(
         &self,
         mc_block: BlockStuff,
         deps: StorePersistentStateDeps,
     ) -> Result<()> {
         let node_state = self.storage.node_state();
         let block_handles = self.storage.block_handle_storage();
+        let state_storage = self.storage.shard_state_storage();
 
         let Some(mc_block_handle) = block_handles.load_handle(mc_block.id()) else {
             bail!("masterchain block handle not found: {}", mc_block.id());
@@ -411,7 +457,7 @@ impl<S> Inner<S> {
         tracing::debug!("saved persistent state for {}", mc_block_handle.id());
 
         for handle in deps.state_gc_handles {
-            block_handles.set_skip_states_gc_finished(&handle);
+            state_storage.set_skip_states_gc_finished(&handle).await;
         }
         for handle in deps.blocks_gc_handles {
             block_handles.set_skip_blocks_gc_finished(&handle);
@@ -428,7 +474,6 @@ impl<S> Inner<S> {
     ) -> Result<()> {
         let block_handles = self.storage.block_handle_storage();
         let persistent_states = self.storage.persistent_state_storage();
-        let state_storage = self.storage.shard_state_storage();
 
         let mc_seqno = mc_block_handle.id().seqno;
 
@@ -438,22 +483,6 @@ impl<S> Inner<S> {
                 anyhow::bail!("top shard block handle not found: {block_id}");
             };
 
-            // Ensure skipped state is stored in DB before saving persistent state.
-            if !block_handle.has_state() {
-                let state = state_storage
-                    .load_state(mc_seqno, &block_id)
-                    .await
-                    .context("failed to load skipped shard state for persistent save")?;
-
-                state_storage
-                    .store_state_ignore_cache(&block_handle, &state, StoreStateHint {
-                        is_top_block: Some(true),
-                        ..Default::default()
-                    })
-                    .await
-                    .context("failed to store skipped shard state for persistent save")?;
-            }
-
             // NOTE: We could have also called the `set_block_persistent` here, but we
             //       only do this in the first part of the `save_persistent_queue_states`.
 

core/src/storage/config.rs

@@ -143,6 +143,24 @@ pub struct StatesGcConfig {
     /// Default: 1s
     #[serde(with = "serde_helpers::humantime")]
     pub interval: Duration,
+
+    /// Above this value, a GC pass holds the state-write lock while scanning and sweeping.
+    pub high_watermark: usize,
+
+    /// An exclusive GC pass stops sweeping when it reaches this value.
+    pub low_watermark: usize,
+}
+
+impl StatesGcConfig {
+    pub(super) fn validate(&self) -> anyhow::Result<()> {
+        anyhow::ensure!(
+            self.low_watermark <= self.high_watermark,
+            "states GC low watermark ({}) exceeds high watermark ({})",
+            self.low_watermark,
+            self.high_watermark,
+        );
+        Ok(())
+    }
 }
 
 impl Default for StatesGcConfig {
@@ -153,6 +171,8 @@ impl Default for StatesGcConfig {
         Self {
             random_offset: false,
             interval: Duration::from_secs(1),
+            high_watermark: 100,
+            low_watermark: 50,
         }
     }
 }
@@ -243,3 +263,33 @@ impl Default for BlobDbConfig {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn states_gc_config_defaults_and_validation() {
+        let defaults = StatesGcConfig::default();
+        assert_eq!(defaults.high_watermark, 100);
+        assert_eq!(defaults.low_watermark, 50);
+        defaults.validate().unwrap();
+
+        StatesGcConfig {
+            low_watermark: 0,
+            ..Default::default()
+        }
+        .validate()
+        .unwrap();
+
+        let invalid = StatesGcConfig {
+            high_watermark: 7,
+            low_watermark: 8,
+            ..Default::default()
+        };
+        assert_eq!(
+            invalid.validate().unwrap_err().to_string(),
+            "states GC low watermark (8) exceeds high watermark (7)"
+        );
+    }
+}

core/src/storage/mod.rs

@@ -99,6 +99,10 @@ impl CoreStorage {
     }
 
     pub async fn open(ctx: StorageContext, config: CoreStorageConfig) -> Result<Self> {
+        if let Some(states_gc) = &config.states_gc {
+            states_gc.validate()?;
+        }
+
         let db: CoreDb = ctx.open_preconfigured(CORE_DB_SUBDIR)?;
         db.normalize_version()?;
         db.apply_migrations().await?;