perf: optimize storage - batch flush, prune audit data, disable indexing

- Replace per-write flush with batched flush every 5s (reduces sled blob accumulation)
- Disable audit logging and indexing in TrackedStorage (massive I/O reduction)
- Startup migration: prune audit_block/audit_key/idx/counter data from sled
- Export→reimport cycle to reclaim dead blob space (47GB → expected few GB)
- Add flush_if_dirty() for periodic batched flushing

Author: echobt

Cargo.lock

@@ -367,15 +367,21 @@ async fn main() -> Result<()> {
     let local_storage = LocalStorageBuilder::new(&validator_hotkey)
         .path(db_path.to_string_lossy().to_string())
         .build()?;
+    let local_storage = Arc::new(local_storage);
 
-    // Wrap with TrackedStorage for automatic compression, indexing, and audit
+    // Wrap with TrackedStorage for compression only (audit + indexing disabled to reduce disk I/O)
     let tracked_config = TrackedStorageConfig {
         validator_id: validator_hotkey.clone(),
+        enable_audit: false,
+        enable_indexing: false,
         ..Default::default()
     };
-    let tracked_storage = TrackedStorage::new(Arc::new(local_storage), tracked_config);
+    let tracked_storage = TrackedStorage::new(
+        Arc::clone(&local_storage) as Arc<dyn platform_distributed_storage::DistributedStore>,
+        tracked_config,
+    );
     let storage = Arc::new(tracked_storage);
-    info!("Distributed storage initialized with compression and tracking");
+    info!("Distributed storage initialized with compression (audit/indexing disabled)");
 
     // Determine listen address - p2p_port overrides listen_addr if specified
     let listen_addr = if let Some(port) = args.p2p_port {
@@ -982,6 +988,7 @@ async fn main() -> Result<()> {
     let mut last_sync_block: u64 = 0;
     let sync_block_interval: u64 = 1; // Call WASM sync every block, WASM decides frequency internally
     let mut storage_stats_interval = tokio::time::interval(Duration::from_secs(300));
+    let mut storage_flush_interval = tokio::time::interval(Duration::from_secs(5));
     // Track last synced block per challenge for delta sync
     let challenge_last_sync: Arc<
         RwLock<std::collections::HashMap<platform_core::ChallengeId, u64>>,
@@ -1410,6 +1417,12 @@ async fn main() -> Result<()> {
             }
 
             // Periodic state persistence
+            _ = storage_flush_interval.tick() => {
+                if let Err(e) = local_storage.flush_if_dirty().await {
+                    warn!("Periodic storage flush failed: {}", e);
+                }
+            }
+
             _ = state_persist_interval.tick() => {
                 if let Err(e) = persist_state_to_storage(&storage, &state_manager).await {
                     warn!("Failed to persist P2P state: {}", e);
@@ -4260,13 +4273,15 @@ async fn fetch_remote_weights() -> anyhow::Result<Vec<(u8, Vec<u16>, Vec<u16>)>>
     Ok(result)
 }
 
-/// Compact the sled distributed.db if blob files are wasting space.
+/// Compact the sled distributed.db: prune useless data, then reclaim disk via export→reimport.
 ///
-/// Sled accumulates dead data in blob files that are never reclaimed.
-/// This does a full export→delete→reimport cycle to reclaim disk space.
-/// A marker file `.compacted_v1` is written after success so it only runs once.
+/// Purges: audit logs (audit_block:*, audit_key:*), old index entries (idx:*, idx_meta:*),
+/// counter entries (counter:*). These are generated by TrackedStorage and no longer needed.
+/// Then does sled export→delete→reimport to reclaim dead blob space.
+///
+/// A marker file `.compacted_v2` prevents re-running on subsequent startups.
 fn compact_storage_if_needed(db_path: &std::path::Path) -> anyhow::Result<()> {
-    let marker = db_path.parent().unwrap_or(db_path).join(".compacted_v1");
+    let marker = db_path.parent().unwrap_or(db_path).join(".compacted_v2");
     if marker.exists() {
         return Ok(());
     }
@@ -4275,97 +4290,117 @@ fn compact_storage_if_needed(db_path: &std::path::Path) -> anyhow::Result<()> {
         return Ok(());
     }
 
-    // Check total size of blob files
+    // Calculate blob size
     let blobs_dir = db_path.join("blobs");
-    let total_blob_size = if blobs_dir.exists() {
-        std::fs::read_dir(&blobs_dir)
-            .map(|entries| {
-                entries
-                    .filter_map(|e| e.ok())
-                    .filter_map(|e| e.metadata().ok())
-                    .map(|m| m.len())
-                    .sum::<u64>()
-            })
-            .unwrap_or(0)
-    } else {
-        0
-    };
-
-    let threshold = 2 * 1024 * 1024 * 1024u64; // 2 GB
-    if total_blob_size < threshold {
-        info!(
-            blob_size_mb = total_blob_size / (1024 * 1024),
-            "Storage compaction not needed"
-        );
-        std::fs::write(&marker, b"skipped").ok();
-        return Ok(());
-    }
+    let total_blob_size = dir_size(&blobs_dir);
 
+    // Always run if v2 marker doesn't exist (prune audit data even if blobs are small)
     warn!(
         blob_size_gb = total_blob_size / (1024 * 1024 * 1024),
-        "Large blob files detected, compacting distributed.db (this may take a few minutes)"
+        "Compacting distributed.db: pruning audit/index data and reclaiming disk space"
     );
 
-    // Phase 1: Open and export all data
+    // Phase 1: Open DB and prune useless namespaces from the data tree
     let db = sled::open(db_path)?;
-    let export = db.export();
-    let db_size_before = total_blob_size;
+    let data_tree = db.open_tree("data")?;
+    let index_tree = db.open_tree("index")?;
+
+    let prune_prefixes: &[&[u8]] = &[
+        b"audit_block:",
+        b"audit_key:",
+        b"idx:",
+        b"idx_meta:",
+        b"counter:",
+    ];
+
+    let mut pruned_count = 0u64;
+    let mut pruned_bytes = 0u64;
+
+    for prefix in prune_prefixes {
+        for entry in data_tree.scan_prefix(prefix) {
+            if let Ok((key, value)) = entry {
+                pruned_bytes += key.len() as u64 + value.len() as u64;
+                data_tree.remove(&key)?;
+                // Also remove from index tree
+                if let Some(ns_end) = key.iter().position(|&b| b == b':') {
+                    let ns = &key[..ns_end];
+                    let key_part = &key[ns_end + 1..];
+                    let index_key =
+                        format!("{}:{}", String::from_utf8_lossy(ns), hex::encode(key_part));
+                    index_tree.remove(index_key.as_bytes())?;
+                }
+                pruned_count += 1;
+            }
+        }
+    }
+
+    info!(
+        pruned_entries = pruned_count,
+        pruned_mb = pruned_bytes / (1024 * 1024),
+        "Pruned audit/index/counter data"
+    );
 
+    db.flush()?;
+    drop(data_tree);
+    drop(index_tree);
+
+    // Phase 2: Export → reimport to reclaim sled blob space
+    let export = db.export();
     let tree_count = export.len();
     info!(trees = tree_count, "Exported data from distributed.db");
     drop(db);
 
-    // Phase 2: Move old DB to temp, create fresh one
     let tmp_path = db_path.with_extension("db.old");
     if tmp_path.exists() {
         std::fs::remove_dir_all(&tmp_path)?;
     }
     std::fs::rename(db_path, &tmp_path)?;
 
-    // Phase 3: Import into fresh DB
     let new_db = sled::open(db_path)?;
     new_db.import(export);
     new_db.flush()?;
     drop(new_db);
 
-    // Phase 4: Measure new size and cleanup
-    let new_blob_size = if db_path.join("blobs").exists() {
-        std::fs::read_dir(db_path.join("blobs"))
-            .map(|entries| {
-                entries
-                    .filter_map(|e| e.ok())
-                    .filter_map(|e| e.metadata().ok())
-                    .map(|m| m.len())
-                    .sum::<u64>()
-            })
-            .unwrap_or(0)
-    } else {
-        0
-    };
+    // Phase 3: Measure and cleanup
+    let new_blob_size = dir_size(&db_path.join("blobs"));
 
     info!(
-        before_gb = db_size_before / (1024 * 1024 * 1024),
+        before_gb = total_blob_size / (1024 * 1024 * 1024),
         after_gb = new_blob_size / (1024 * 1024 * 1024),
-        saved_gb = (db_size_before.saturating_sub(new_blob_size)) / (1024 * 1024 * 1024),
+        saved_gb = total_blob_size.saturating_sub(new_blob_size) / (1024 * 1024 * 1024),
         "Storage compaction complete"
     );
 
-    // Remove old DB
     if let Err(e) = std::fs::remove_dir_all(&tmp_path) {
-        warn!(error = %e, "Failed to remove old distributed.db.old (can be deleted manually)");
+        warn!(error = %e, "Failed to remove old distributed.db.old");
     }
 
-    // Write marker so we don't compact again
     std::fs::write(
         &marker,
         format!(
-            "compacted: before={}GB after={}GB",
-            db_size_before / (1024 * 1024 * 1024),
+            "compacted_v2: pruned={} entries, before={}GB after={}GB",
+            pruned_count,
+            total_blob_size / (1024 * 1024 * 1024),
             new_blob_size / (1024 * 1024 * 1024),
         ),
     )?;
 
     Ok(())
 }
 
+fn dir_size(path: &std::path::Path) -> u64 {
+    if !path.exists() {
+        return 0;
+    }
+    std::fs::read_dir(path)
+        .map(|entries| {
+            entries
+                .filter_map(|e| e.ok())
+                .filter_map(|e| e.metadata().ok())
+                .map(|m| m.len())
+                .sum()
+        })
+        .unwrap_or(0)
+}
+
 // Build trigger: 1771754356

bins/validator-node/src/main.rs

@@ -90,6 +90,8 @@ pub struct LocalStorage {
     node_id: String,
     /// Cache for namespace entry counts
     namespace_counts: RwLock<HashMap<String, u64>>,
+    /// Dirty flag for batched flushing
+    dirty: std::sync::atomic::AtomicBool,
 }
 
 impl LocalStorage {
@@ -107,6 +109,7 @@ impl LocalStorage {
             block_index_tree,
             node_id,
             namespace_counts: RwLock::new(HashMap::new()),
+            dirty: std::sync::atomic::AtomicBool::new(false),
         };
 
         // Initialize namespace counts
@@ -129,6 +132,7 @@ impl LocalStorage {
             block_index_tree,
             node_id,
             namespace_counts: RwLock::new(HashMap::new()),
+            dirty: std::sync::atomic::AtomicBool::new(false),
         })
     }
 
@@ -273,7 +277,7 @@ impl LocalStorage {
         if let Some(mut entry) = self.get_entry(key)? {
             entry.replication.mark_replicated(node_id);
             self.put_entry(key, &entry)?;
-            self.flush_async().await?;
+            self.mark_dirty();
         }
         Ok(())
     }
@@ -317,22 +321,39 @@ impl LocalStorage {
         Ok(self.get_entry(key)?.map(|e| e.replication))
     }
 
+    /// Mark storage as having unflushed writes
+    fn mark_dirty(&self) {
+        self.dirty.store(true, std::sync::atomic::Ordering::Relaxed);
+    }
+
     /// Flush all changes to disk synchronously (blocking)
     pub fn flush(&self) -> StorageResult<()> {
+        self.dirty
+            .store(false, std::sync::atomic::Ordering::Relaxed);
         self.db.flush()?;
         Ok(())
     }
 
     /// Flush all changes to disk asynchronously using spawn_blocking
     pub async fn flush_async(&self) -> StorageResult<()> {
         let db = Arc::clone(&self.db);
+        self.dirty
+            .store(false, std::sync::atomic::Ordering::Relaxed);
         tokio::task::spawn_blocking(move || db.flush())
             .await
             .map_err(|e| StorageError::Database(format!("flush task panicked: {}", e)))?
             .map_err(StorageError::from)?;
         Ok(())
     }
 
+    /// Flush only if there are pending writes
+    pub async fn flush_if_dirty(&self) -> StorageResult<()> {
+        if self.dirty.load(std::sync::atomic::Ordering::Relaxed) {
+            self.flush_async().await?;
+        }
+        Ok(())
+    }
+
     /// Get the underlying database (for advanced operations)
     pub fn db(&self) -> &Db {
         &self.db
@@ -637,7 +658,7 @@ impl DistributedStore for LocalStorage {
         };
 
         self.put_entry(&key, &entry)?;
-        self.flush_async().await?;
+        self.mark_dirty();
 
         debug!(
             "LocalStorage::put completed key={} version={}",
@@ -650,7 +671,7 @@ impl DistributedStore for LocalStorage {
     async fn delete(&self, key: &StorageKey) -> StorageResult<bool> {
         trace!("LocalStorage::delete key={}", key);
         let deleted = self.delete_entry(key)?;
-        self.flush_async().await?;
+        self.mark_dirty();
         Ok(deleted)
     }
 
@@ -945,7 +966,7 @@ impl DistributedStore for LocalStorage {
         options: PutOptions,
     ) -> StorageResult<ValueMetadata> {
         let metadata = self.put_entry_with_block_internal(&key, value, block_id, &options)?;
-        self.flush_async().await?;
+        self.mark_dirty();
         Ok(metadata)
     }
 }