fix: critical bugs found during storage audit

- Fix list/query methods returning compressed data without decompressing
- Fix AuditLog race condition: add write_lock mutex
- Fix AtomicCounter underflow: clamp to 0 on negative values
- Add integration tests for compression roundtrip
- Add backward compatibility test for pre-compression data
- Fix clippy warnings in platform-rpc and platform-storage

Author: echobt

crates/distributed-storage/src/audit.rs

@@ -4,6 +4,7 @@
 
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
+use tokio::sync::Mutex;
 
 use crate::error::{StorageError, StorageResult};
 use crate::store::{DistributedStore, GetOptions, PutOptions, StorageKey};
@@ -113,6 +114,8 @@ pub struct AuditLog {
     storage: Arc<dyn DistributedStore>,
     /// Maximum entries per block (to limit storage size)
     max_entries_per_block: usize,
+    /// Write lock to prevent race conditions on read-modify-write
+    write_lock: Mutex<()>,
 }
 
 impl AuditLog {
@@ -121,11 +124,14 @@ impl AuditLog {
         Self {
             storage,
             max_entries_per_block: 10_000,
+            write_lock: Mutex::new(()),
         }
     }
 
     /// Append an audit entry
     pub async fn append(&self, entry: AuditEntry) -> StorageResult<()> {
+        let _guard = self.write_lock.lock().await;
+
         // Store by block number
         let block_key = StorageKey::new("audit_block", format!("{:016x}", entry.block_number));
 

crates/distributed-storage/src/index.rs

@@ -281,7 +281,10 @@ impl IndexManager {
     }
 }
 
-/// Atomic counter for tracking counts
+/// Storage-backed counter.
+///
+/// NOTE: Not truly atomic under concurrent access -- uses read-modify-write
+/// without CAS. Safe for single-writer scenarios only.
 pub struct AtomicCounter {
     storage: Arc<dyn DistributedStore>,
 }
@@ -306,12 +309,13 @@ impl AtomicCounter {
         Ok(0)
     }
 
-    /// Increment a counter and return the new value
+    /// Increment a counter and return the new value.
+    /// Clamps to 0 on the low end to avoid underflow.
     pub async fn increment(&self, namespace: &str, name: &str, delta: i64) -> StorageResult<i64> {
         let key = StorageKey::new("counter", format!("{}:{}", namespace, name));
 
         let current = self.get(namespace, name).await? as i64;
-        let new_value = current.saturating_add(delta);
+        let new_value = current.saturating_add(delta).max(0);
 
         let data = (new_value as u64).to_le_bytes().to_vec();
         self.storage.put(key, data, PutOptions::default()).await?;

crates/distributed-storage/src/tracked.rs

@@ -193,6 +193,22 @@ impl TrackedStorage {
         hasher.finalize().into()
     }
 
+    /// Decompress all items in a list result
+    fn decompress_list_result(&self, mut result: ListResult) -> StorageResult<ListResult> {
+        for (_key, value) in &mut result.items {
+            value.data = self.decompress(&value.data)?;
+        }
+        Ok(result)
+    }
+
+    /// Decompress all items in a query result
+    fn decompress_query_result(&self, mut result: QueryResult) -> StorageResult<QueryResult> {
+        for (_key, value) in &mut result.items {
+            value.data = self.decompress(&value.data)?;
+        }
+        Ok(result)
+    }
+
     /// Create an index for a namespace
     pub async fn create_index(
         &self,
@@ -377,9 +393,11 @@ impl DistributedStore for TrackedStorage {
         limit: usize,
         continuation_token: Option<&[u8]>,
     ) -> StorageResult<ListResult> {
-        self.inner
+        let result = self
+            .inner
             .list_prefix(namespace, prefix, limit, continuation_token)
-            .await
+            .await?;
+        self.decompress_list_result(result)
     }
 
     async fn stats(&self) -> StorageResult<StorageStats> {
@@ -392,9 +410,11 @@ impl DistributedStore for TrackedStorage {
         block_id: u64,
         limit: usize,
     ) -> StorageResult<QueryResult> {
-        self.inner
+        let result = self
+            .inner
             .list_before_block(namespace, block_id, limit)
-            .await
+            .await?;
+        self.decompress_query_result(result)
     }
 
     async fn list_after_block(
@@ -403,9 +423,11 @@ impl DistributedStore for TrackedStorage {
         block_id: u64,
         limit: usize,
     ) -> StorageResult<QueryResult> {
-        self.inner
+        let result = self
+            .inner
             .list_after_block(namespace, block_id, limit)
-            .await
+            .await?;
+        self.decompress_query_result(result)
     }
 
     async fn list_range(
@@ -415,17 +437,20 @@ impl DistributedStore for TrackedStorage {
         end_block: u64,
         limit: usize,
     ) -> StorageResult<QueryResult> {
-        self.inner
+        let result = self
+            .inner
             .list_range(namespace, start_block, end_block, limit)
-            .await
+            .await?;
+        self.decompress_query_result(result)
     }
 
     async fn count_by_namespace(&self, namespace: &str) -> StorageResult<u64> {
         self.inner.count_by_namespace(namespace).await
     }
 
     async fn query(&self, query: QueryBuilder) -> StorageResult<QueryResult> {
-        self.inner.query(query).await
+        let result = self.inner.query(query).await?;
+        self.decompress_query_result(result)
     }
 
     async fn put_with_block(
@@ -481,4 +506,99 @@ mod tests {
         let hash2 = TrackedStorage::hash_data(data);
         assert_eq!(hash, hash2);
     }
+
+    #[tokio::test]
+    async fn test_put_get_roundtrip_with_compression() {
+        let config = TrackedStorageConfig {
+            compression: CompressionMode::Lz4,
+            compression_threshold: 10,
+            enable_audit: true,
+            enable_indexing: true,
+            ..Default::default()
+        };
+
+        let storage = crate::local::LocalStorage::in_memory("test".to_string()).unwrap();
+        let tracked = TrackedStorage::new(Arc::new(storage), config);
+        tracked.set_block(100);
+
+        // Write large data (will be compressed)
+        let key = StorageKey::new("test", "large-value");
+        let data = vec![42u8; 500]; // 500 bytes of 42s
+        tracked
+            .put(key.clone(), data.clone(), PutOptions::default())
+            .await
+            .unwrap();
+
+        // Read back - should be decompressed
+        let result = tracked
+            .get(&key, GetOptions::default())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(result.data, data);
+
+        // Verify stats
+        let stats = tracked.tracked_stats().await;
+        assert_eq!(stats.total_writes, 1);
+        assert_eq!(stats.total_reads, 1);
+        assert!(stats.compression_ratio < 1.0); // Data was compressed
+    }
+
+    #[tokio::test]
+    async fn test_list_prefix_decompresses() {
+        let config = TrackedStorageConfig {
+            compression: CompressionMode::Lz4,
+            compression_threshold: 10,
+            enable_audit: false,
+            enable_indexing: false,
+            ..Default::default()
+        };
+
+        let storage = crate::local::LocalStorage::in_memory("test".to_string()).unwrap();
+        let tracked = TrackedStorage::new(Arc::new(storage), config);
+
+        // Write multiple values
+        for i in 0..3 {
+            let key = StorageKey::new("ns", format!("key-{}", i));
+            let data = vec![i as u8; 500]; // Will be compressed
+            tracked.put(key, data, PutOptions::default()).await.unwrap();
+        }
+
+        // List prefix should return decompressed data
+        let result = tracked.list_prefix("ns", None, 10, None).await.unwrap();
+        assert_eq!(result.items.len(), 3);
+        for (i, (_key, value)) in result.items.iter().enumerate() {
+            assert_eq!(value.data.len(), 500);
+            assert_eq!(value.data[0], i as u8);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_backward_compatibility_uncompressed() {
+        let inner = crate::local::LocalStorage::in_memory("test".to_string()).unwrap();
+        let inner_arc = Arc::new(inner);
+
+        // Write directly to inner (uncompressed, simulating old data)
+        let key = StorageKey::new("test", "old-data");
+        let data = b"this is old uncompressed data that was stored before compression".to_vec();
+        inner_arc
+            .put(key.clone(), data.clone(), PutOptions::default())
+            .await
+            .unwrap();
+
+        // Now wrap with TrackedStorage and read
+        let config = TrackedStorageConfig {
+            compression: CompressionMode::Lz4,
+            compression_threshold: 10,
+            ..Default::default()
+        };
+        let tracked = TrackedStorage::new(inner_arc, config);
+
+        let result = tracked
+            .get(&key, GetOptions::default())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(result.data, data); // Should read old data unchanged
+    }
 }