fix: storage bootstrap sync + memory optimization

- Add pending_writes cache for read-your-own-writes during WASM sync cycles
  (propose_write populates cache, get() checks cache first, cleared after sync)
- Clear pending writes before get_weights (reads consensus-confirmed data only)
- validated_storage: remove proposal from HashMap after commit (was never cleaned)
- validated_storage: bound committed HashMap to last 1000 entries
- validated_storage: cleanup_expired removes all expired proposals unconditionally
- state_consensus: prune completed results to last 100 blocks
- Sled cache limited to 256MB (was 1GB default)
- Explicit sled flush on shutdown after state persistence
- Add clear_pending_writes() to StorageBackend trait (default no-op)

Author: echobt

bins/validator-node/src/challenge_storage.rs

@@ -5,18 +5,26 @@ use platform_distributed_storage::{
 };
 use platform_p2p_consensus::{P2PCommand, P2PMessage, StorageProposal, StorageProposalMessage};
 use sha2::{Digest, Sha256};
+use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::mpsc;
 use wasm_runtime_interface::storage::{StorageBackend, StorageHostError};
 
 /// Channel for local storage proposals (proposer adds to own state)
 pub type LocalProposalSender = mpsc::Sender<StorageProposal>;
 
+/// Pending write value: Some(data) = write, None = delete
+type PendingValue = Option<Vec<u8>>;
+
 pub struct ChallengeStorageBackend {
     storage: Arc<dyn DistributedStore>,
     p2p_tx: Option<mpsc::Sender<P2PCommand>>,
     local_proposal_tx: Option<LocalProposalSender>,
     keypair: Option<Keypair>,
+    /// Write-through cache for read-your-own-writes during a sync cycle.
+    /// Key: (challenge_id, hex-encoded key). Value: pending data (None = delete).
+    /// Cleared via `clear_pending_writes()` after each sync cycle completes.
+    pending_writes: parking_lot::RwLock<HashMap<(String, String), PendingValue>>,
 }
 
 impl ChallengeStorageBackend {
@@ -27,6 +35,7 @@ impl ChallengeStorageBackend {
             p2p_tx: None,
             local_proposal_tx: None,
             keypair: None,
+            pending_writes: parking_lot::RwLock::new(HashMap::new()),
         }
     }
 
@@ -41,8 +50,15 @@ impl ChallengeStorageBackend {
             p2p_tx: Some(p2p_tx),
             local_proposal_tx: Some(local_proposal_tx),
             keypair: Some(keypair),
+            pending_writes: parking_lot::RwLock::new(HashMap::new()),
         }
     }
+
+    /// Clear the pending writes cache. Call after each sync cycle completes
+    /// so that subsequent reads go through consensus-confirmed storage.
+    pub fn clear_pending_writes(&self) {
+        self.pending_writes.write().clear();
+    }
 }
 
 /// Build a standardized storage key for challenge data.
@@ -54,6 +70,12 @@ fn build_challenge_storage_key(challenge_id: &str, key: &[u8]) -> DStorageKey {
 
 impl StorageBackend for ChallengeStorageBackend {
     fn get(&self, challenge_id: &str, key: &[u8]) -> Result<Option<Vec<u8>>, StorageHostError> {
+        // Check pending writes cache first (read-your-own-writes during sync)
+        let cache_key = (challenge_id.to_string(), hex::encode(key));
+        if let Some(pending) = self.pending_writes.read().get(&cache_key) {
+            return Ok(pending.clone());
+        }
+
         let storage_key = build_challenge_storage_key(challenge_id, key);
         let result = tokio::task::block_in_place(|| {
             tokio::runtime::Handle::current()
@@ -76,9 +98,19 @@ impl StorageBackend for ChallengeStorageBackend {
         hasher.update(value);
         let proposal_id: [u8; 32] = hasher.finalize().into();
 
-        // DO NOT write locally here - data is only written after P2P consensus is reached.
-        // All validators (including the proposer) write in the StorageVote handler when
-        // 2f+1 votes approve. This ensures consistency across all nodes.
+        // Cache the write locally so WASM can read-its-own-writes during the
+        // current sync cycle. This cache is NOT persisted to storage and does NOT
+        // affect other validators. Actual storage write happens after P2P consensus.
+        // The cache is cleared after each sync cycle via clear_pending_writes().
+        {
+            let cache_key = (challenge_id.to_string(), hex::encode(key));
+            let cache_value = if value.is_empty() {
+                None // delete
+            } else {
+                Some(value.to_vec())
+            };
+            self.pending_writes.write().insert(cache_key, cache_value);
+        }
 
         // Broadcast via P2P for consensus
         if let (Some(tx), Some(kp)) = (&self.p2p_tx, &self.keypair) {

bins/validator-node/src/main.rs

@@ -1850,6 +1850,13 @@ async fn main() -> Result<()> {
                     info!("P2P state persisted on shutdown");
                 }
 
+                // Flush storage to disk to ensure all writes are durable
+                if let Err(e) = local_storage.flush_if_dirty().await {
+                    error!("Failed to flush storage on shutdown: {}", e);
+                } else {
+                    info!("Storage flushed to disk on shutdown");
+                }
+
                 // Create checkpoint
                 if let Some(handler) = shutdown_handler.as_mut() {
                     if let Err(e) = handler.create_checkpoint() {

bins/validator-node/src/wasm_executor.rs

@@ -1047,6 +1047,8 @@ impl WasmChallengeExecutor {
         block_height: u64,
         epoch: u64,
     ) -> Result<Vec<platform_challenge_sdk::WeightAssignment>> {
+        // Ensure get_weights reads only consensus-confirmed data, not pending sync writes
+        self.config.storage_backend.clear_pending_writes();
         let start = Instant::now();
 
         let module = self
@@ -1221,6 +1223,10 @@ impl WasmChallengeExecutor {
             "WASM sync completed"
         );
 
+        // Clear the pending writes cache so subsequent reads (e.g., get_weights)
+        // only see consensus-confirmed data, not transient sync writes.
+        self.config.storage_backend.clear_pending_writes();
+
         Ok(sync_result)
     }
 

crates/distributed-storage/src/local.rs

@@ -95,9 +95,13 @@ pub struct LocalStorage {
 }
 
 impl LocalStorage {
-    /// Open or create a local storage at the given path
+    /// Open or create a local storage at the given path.
+    /// Sled cache is limited to 256MB to bound memory usage.
     pub fn open<P: AsRef<Path>>(path: P, node_id: String) -> StorageResult<Self> {
-        let db = sled::open(path)?;
+        let db = sled::Config::new()
+            .path(path)
+            .cache_capacity(256 * 1024 * 1024) // 256MB cache limit
+            .open()?;
         let data_tree = db.open_tree("data")?;
         let index_tree = db.open_tree("index")?;
         let block_index_tree = db.open_tree("block_index")?;

crates/distributed-storage/src/state_consensus.rs

@@ -834,9 +834,16 @@ impl StateRootConsensus {
     }
 
     /// Store a completed consensus result.
+    /// Prunes old results to bound memory (keeps last 100 blocks).
     pub fn store_completed(&mut self, result: ConsensusResult) {
         let block = result.block_number;
         self.completed.insert(block, result);
+
+        // Prune old completed results to prevent unbounded growth
+        if self.completed.len() > 100 {
+            let cutoff = block.saturating_sub(100);
+            self.completed.retain(|b, _| *b > cutoff);
+        }
     }
 }
 

crates/distributed-storage/src/validated_storage.rs

@@ -684,13 +684,23 @@ impl<S: DistributedStore + 'static> ValidatedStorage<S> {
         {
             let mut committed = self.committed.write().await;
             committed.insert(*proposal_id, result.clone());
+            // Evict old committed results to bound memory (keep last 1000)
+            if committed.len() > 1000 {
+                let oldest: Vec<[u8; 32]> = committed
+                    .iter()
+                    .take(committed.len() - 1000)
+                    .map(|(id, _)| *id)
+                    .collect();
+                for id in oldest {
+                    committed.remove(&id);
+                }
+            }
         }
 
+        // Remove committed proposal from pending to free memory
         {
             let mut proposals = self.proposals.write().await;
-            if let Some(state) = proposals.get_mut(proposal_id) {
-                state.consensus_result = Some(result.clone());
-            }
+            proposals.remove(proposal_id);
         }
 
         Ok(result)
@@ -746,9 +756,9 @@ impl<S: DistributedStore + 'static> ValidatedStorage<S> {
         let timeout = self.config.proposal_timeout_ms;
         let before = proposals.len();
 
-        proposals.retain(|_, state| {
-            !state.proposal.is_expired(timeout) || state.consensus_result.is_some()
-        });
+        // Remove expired proposals regardless of consensus state.
+        // Committed proposals are already removed in commit_write().
+        proposals.retain(|_, state| !state.proposal.is_expired(timeout));
 
         let removed = before - proposals.len();
         if removed > 0 {

crates/wasm-runtime-interface/src/storage.rs

@@ -315,6 +315,10 @@ pub trait StorageBackend: Send + Sync {
         let _ = (challenge_id, prefix);
         Ok(0)
     }
+
+    /// Clear any pending write cache. Called after sync cycles complete
+    /// so subsequent reads reflect only consensus-confirmed data.
+    fn clear_pending_writes(&self) {}
 }
 
 pub struct NoopStorageBackend;