fix(storage,raft): fix snapshot-logindex synchronization issues (#265)

* feat(raft): implement snapshot creation and restore with logindex sync

This PR implements raft snapshot creation and installation with full
logindex synchronization support:

Snapshot creation:
- Checkpoint-based snapshot creation (tar packaging of checkpoint + metadata)
- Includes LogIndex collector state and cf_tracker state in snapshot metadata
- Version-aware snapshot metadata with forward compatibility (< instead of !=)

Snapshot installation:
- ArcSwap-based Storage hot-swapping with pause/resume coordination
- State updates (cf_tracker init, collector restore) happen BEFORE resume
  to prevent applied_state() from returning stale values
- Refresh collector/cf_tracker from new storage after swap (not orphaned refs)
- Proper error cleanup via pause_controller resume

Logindex tracking:
- Event listener (DB-level, not CF-level) for tracking (log_index, seqno)
- LogIndexOfColumnFamilies tracks applied/flushed state per CF
- Mappings persisted to SST table properties via custom collector
- State sync during snapshot build/install for follower bootstrap
- watchdog uses std::thread::spawn (no tokio runtime in RocksDB callback threads)

CI fixes:
- License header truncation in snapshot_logindex_test.rs
- collapsible_match clippy warnings in hscan/sscan/zscan/message.rs
- is_retryable for LogIndex errors: distinguish transient vs structural
- CF metadata compile-time assertions to prevent drift
- Import consolidation, pub use re-exports, doc additions, test improvements

Co-Authored-By: github-actions

* chore: ensure .claude path is ignored

Cover both file and directory cases in .gitignore so personal Claude Code
configuration cannot be committed.

* docs: restore CLAUDE.md project documentation

Replace local-path content with the canonical project documentation
imported from main.

* refactor(raft): drop unused KiwiSnapshotBuilder logindex fields

The builder's collector and cf_tracker Arc fields were never read — the
build_snapshot path always re-fetches the collector from the live
storage. Carrying them is misleading: after install_snapshot swaps
storage, the builder fields would point at orphaned instances anyway.
Remove them so the builder only carries what it actually uses.

* fix(raft): export logindex collector state per Storage instance

Snapshot meta previously only captured instance 0's collector via
`get_logindex_collector(0)`, so when `db_instance_num > 1` the
(log_index, seqno) mappings written to non-zero instances were lost on
snapshot install and the follower's collector started empty for those
shards.

`RaftSnapshotMeta` now stores a `Vec<Vec<String>>` (outer index = Storage
instance id), and `KiwiSnapshotBuilder`/`install_snapshot` iterate every
instance's collector. The state machine no longer caches stale collector
references either — it always looks them up through `storage_swap` so
they remain valid after a snapshot-install hot swap.

* test(storage): keep TempDir alive in storage_basic_test

`tempfile::tempdir().unwrap().path().to_path_buf()` drops the TempDir
guard before `Storage::open()` runs, so the directory was deleted from
underneath the test. Bind the guard so it lives until the end of the test.

* refactor(storage): extract commit-and-track-logindex helper

Move the (commit, latest_sequence_number, collector.update) sequence out
of `Storage::on_binlog_write` and into a new
`Redis::commit_batch_and_track_logindex` method. This gives us one place
to reason about the seqno-after-commit pattern and to harden later
(e.g. a per-instance write mutex or a future rust-rocksdb API exposing
the batch's own sequence number) without touching every caller. No
behavior change.

* fix(storage): wire flush_trigger to RocksDB and stop logging snapshot stub as info

`flush_trigger` was a `log::info!(...) + TODO`, so when the LogIndex
collector exceeded its bound the listener "asked" for a flush that
nothing performed — entries piled up. The trigger now resolves the CF
handle through a shared `OnceCell<Arc<DB>>` (populated post-open from
the same DB used by compaction filters) and calls `db.flush_cf`.

`snapshot_callback` still cannot be wired without a storage→raft
back-channel that does not exist today. Demote it to `log::debug!` and
spell out in the comment that it is intentionally a no-op while the
manual snapshot path remains the only trigger, instead of the
misleading "will be connected" comment.

* style: cargo fmt

* docs(storage): document single-writer invariant on on_binlog_write

Reviewer asked the call site to make the single-writer assumption explicit so
future changes don't accidentally introduce a second writer that would inflate
the captured seqno via latest_sequence_number().

* ci: skip snapshot_logindex_test under leak sanitizer

RocksDB integration tests are excluded from LSan/TSan upstream because
RocksDB internals trigger spurious leak reports. The new snapshot logindex
tests exercise the same Storage::open path and need the same exclusion.

* fix(raft): drop old Storage before restoring checkpoint

The previous block-scoped placeholder pattern relied on the comment 'old
Arc dropped at the end of this block', but `current_storage` is owned by
the outer scope, so it lived through restore_checkpoint_layout(). With
the old RocksDB handle still alive, restore could race with an open lock
on db_path. Explicitly drop current_storage right after the swap so the
restore runs with no live handle.

Reported by AlexStocks in PR #265 review (P0).

---------

Co-authored-by: lupengfan1 <lupengfan1@xiaomi.com>

Author: ruojieranyishen

.github/workflows/ci.yml

@@ -297,7 +297,10 @@ jobs:
               cargo +nightly test -Z build-std --target x86_64-unknown-linux-gnu \
                 --workspace --exclude storage \
                 -- --skip cursor_snapshot_roundtrip \
-                --skip install_snapshot_with_existing_data
+                --skip install_snapshot_with_existing_data \
+                --skip test_snapshot_with_logindex_state \
+                --skip test_on_binlog_write_updates_collector \
+                --skip test_collector_state_export_restore
               cargo +nightly test -Z build-std --target x86_64-unknown-linux-gnu \
                 -p storage --lib
             } 2>&1 | tee sanitizer.log

.gitignore

@@ -46,6 +46,7 @@ test_debug_*
 STORAGE_COMPARISON.md
 
 # Claude Code configuration
+.claude
 .claude/
 
 # Claude Spec Workflow configuration

src/common/runtime/error.rs

@@ -182,6 +182,16 @@ impl DualRuntimeError {
             StorageError::Unknown { .. } => false,
             StorageError::OptionNone { .. } => false,
             StorageError::RedisErr { .. } => true, // Redis protocol errors are typically recoverable
+            StorageError::LogIndex { message, .. } => {
+                let msg = message.to_lowercase();
+                // Structural/programming errors are not retryable
+                if msg.contains("not found") || msg.contains("invalid") || msg.contains("unknown") {
+                    false
+                } else {
+                    // RocksDB-originating errors may be transient
+                    true
+                }
+            }
         }
     }
 

src/engine/src/engine.rs

@@ -127,4 +127,10 @@ pub trait Engine: Send + Sync {
 
     /// Latest sequence number after writes.
     fn latest_sequence_number(&self) -> u64;
+
+    /// Get properties of all SST files for a column family.
+    fn get_properties_of_all_tables_cf(
+        &self,
+        cf: &ColumnFamilyRef<'_>,
+    ) -> Result<rocksdb::table_properties::TablePropertiesCollection>;
 }

src/engine/src/rocksdb_engine.rs

@@ -207,6 +207,13 @@ impl Engine for RocksdbEngine {
     fn latest_sequence_number(&self) -> u64 {
         self.db.latest_sequence_number()
     }
+
+    fn get_properties_of_all_tables_cf(
+        &self,
+        cf: &ColumnFamilyRef<'_>,
+    ) -> Result<rocksdb::table_properties::TablePropertiesCollection> {
+        self.db.get_properties_of_all_tables_cf(cf)
+    }
 }
 
 impl Clone for RocksdbEngine {

src/raft/src/db_access.rs

@@ -15,19 +15,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use rocksdb::table_properties::TablePropertiesCollection;
+//! Shim module that re-exports from storage::logindex for backward compatibility.
+//! Tests that use `raft::db_access::DbCfAccess` can continue to work.
 
-/// Error type (reuse rocksdb::Error or custom)
-pub type Result<T> = std::result::Result<T, rocksdb::Error>;
-
-/// Thin wrapper interface: provides ability to get TableProperties by CF
-///
-/// Equivalent to C++ Redis's GetDB() + GetColumnFamilyHandles()[cf_id],
-/// used for LogIndexOfColumnFamilies::Init to iterate CFs and call GetPropertiesOfAllTables.
-pub trait DbCfAccess {
-    /// Get TableProperties of all SSTs for specified CF
-    ///
-    /// # Arguments
-    /// * `cf_id` - ColumnFamily index, range [0, COLUMN_FAMILY_COUNT)
-    fn get_properties_of_all_tables_cf(&self, cf_id: usize) -> Result<TablePropertiesCollection>;
-}
+pub use storage::logindex::DbCfAccess;

src/raft/src/lib.rs

@@ -15,11 +15,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub mod cf_tracker;
-pub mod collector;
+//! Raft module for Kiwi
+//!
+//! Re-exports logindex types from storage::logindex to avoid code duplication.
+
 pub mod conversion;
-pub mod db_access;
-pub mod event_listener;
+pub mod db_access; // Shim for backward compatibility with tests
 pub mod grpc;
 pub mod leader_gate;
 pub mod log_store;
@@ -28,24 +29,25 @@ pub mod network;
 pub mod node;
 pub mod snapshot_archive;
 pub mod state_machine;
+
 pub mod raft_proto {
     // 使用版本化的 proto 包名 kiwi.raft.v1
     tonic::include_proto!("kiwi.raft.v1");
     pub const FILE_DESCRIPTOR_SET: &[u8] =
         tonic::include_file_descriptor_set!("kiwi.raft.v1_descriptor");
 }
 
-pub mod table_properties;
-pub mod types;
-
-pub use crate::cf_tracker::{LogIndexOfColumnFamilies, SmallestIndexRes};
-pub use crate::collector::LogIndexAndSequenceCollector;
-pub use crate::event_listener::LogIndexAndSequenceCollectorPurger;
-pub use table_properties::{
-    LogIndexTablePropertiesCollectorFactory, PROPERTY_KEY, get_largest_log_index_from_collection,
+// Re-export logindex types from storage::logindex (no duplicate implementations)
+pub use storage::logindex::{
+    DbCfAccess, FlushTrigger, LogIndex, LogIndexAndSequenceCollector,
+    LogIndexAndSequenceCollectorPurger, LogIndexAndSequencePair, LogIndexOfColumnFamilies,
+    LogIndexSeqnoPair, LogIndexTablePropertiesCollectorFactory, PROPERTY_KEY, SequenceNumber,
+    SmallestIndexRes, SnapshotCallback, cf_name_to_index, get_largest_log_index_from_collection,
     read_stats_from_table_props,
 };
-pub use types::{LogIndex, LogIndexAndSequencePair, LogIndexSeqnoPair, SequenceNumber};
+
+// Re-export error types for tests implementing DbCfAccess trait
+pub use storage::logindex::types::LogIndexError;
 
 /// Number of column families, consistent with storage::ColumnFamilyIndex::COUNT
 pub const COLUMN_FAMILY_COUNT: usize = storage::ColumnFamilyIndex::COUNT;
@@ -65,11 +67,6 @@ const _: () = assert!(
     "CF_NAMES length must match storage::ColumnFamilyIndex::COUNT"
 );
 
-/// Convert CF name to index
-pub fn cf_name_to_index(name: &[u8]) -> Option<usize> {
-    CF_NAMES.iter().position(|n| n.as_bytes() == name)
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

src/raft/src/node.rs

@@ -145,6 +145,21 @@ impl Default for RaftConfig {
 }
 
 fn build_raft_config(config: &RaftConfig) -> Result<Arc<Config>, anyhow::Error> {
+    // Validate snapshot configuration parameters
+    if config.snapshot_logs_threshold == 0 {
+        return Err(anyhow::anyhow!("snapshot_logs_threshold must be > 0"));
+    }
+    if config.snapshot_max_chunk_size == 0 {
+        return Err(anyhow::anyhow!("snapshot_max_chunk_size must be > 0"));
+    }
+    if config.install_snapshot_timeout == 0 {
+        return Err(anyhow::anyhow!("install_snapshot_timeout must be > 0"));
+    }
+    if config.replication_lag_threshold == 0 {
+        return Err(anyhow::anyhow!("replication_lag_threshold must be > 0"));
+    }
+    // max_in_snapshot_log_to_keep: 0 is intentionally allowed (keep no in-snapshot logs)
+
     let raft_config = Config {
         heartbeat_interval: config.heartbeat_interval,
         election_timeout_min: config.election_timeout_min,
@@ -168,6 +183,9 @@ pub async fn create_raft_node(
     let snapshot_work_dir = config.data_dir.join("snapshots");
     fs::create_dir_all(&snapshot_work_dir)?;
 
+    // Per-instance LogIndex collectors / cf_trackers live in the Storage; the state
+    // machine looks them up through storage_swap so it sees the right ones after a
+    // snapshot install hot-swaps Storage.
     let mut state_machine = KiwiStateMachine::new(
         config.node_id,
         storage_swap.clone(),

src/raft/src/snapshot_archive.rs

@@ -15,6 +15,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//! Tar packing/unpacking for Raft snapshot checkpoints.
+//!
+//! TODO: stream snapshot bytes (read/write without holding the full tar in memory) for
+//! build and install paths; align with OpenRaft snapshot APIs when switching off in-memory buffers.
+
 use std::io::{self, Cursor};
 use std::path::Path;
 
@@ -56,6 +61,12 @@ fn append_dir_all_skip_lock<W: std::io::Write>(
     Ok(())
 }
 
+/// Unpack a tar archive (from `build_snapshot` / OpenRaft `SnapshotData`) into `dst`.
+///
+/// Security: validates paths to prevent path traversal attacks.
+/// - Rejects if dst exists but is not a directory
+/// - Rejects tar entries containing ".." components
+/// - Rejects paths that escape the destination directory
 pub fn unpack_tar_to_dir(bytes: &[u8], dst: &Path) -> io::Result<()> {
     if dst.exists() && !dst.is_dir() {
         return Err(io::Error::new(
@@ -92,6 +103,7 @@ pub fn unpack_tar_to_dir(bytes: &[u8], dst: &Path) -> io::Result<()> {
     Ok(())
 }
 
+/// Directory inside `unpack_tar_to_dir` output that contains `0/`, `1/`, … and `__raft_snapshot_meta`.
 pub fn unpacked_checkpoint_root(unpack_root: &Path) -> std::path::PathBuf {
     unpack_root.join("snap")
 }

src/raft/src/state_machine.rs

@@ -77,7 +77,6 @@ fn persist_current_snapshot(
     // Use temporary files + atomic rename to prevent TOCTOU race conditions.
     let data_tmp = work_dir.join(format!(".{}.tmp", CURRENT_SNAPSHOT_DATA));
     let meta_tmp = work_dir.join(format!(".{}.tmp", CURRENT_SNAPSHOT_META));
-
     std::fs::write(&data_tmp, bytes).map_err(|e| {
         StorageError::from_io_error(ErrorSubject::Snapshot(None), ErrorVerb::Write, e)
     })?;
@@ -190,6 +189,10 @@ pub struct KiwiStateMachine {
 
 impl KiwiStateMachine {
     /// Create a new state machine.
+    ///
+    /// Per-instance LogIndex collectors and cf_trackers are owned by the underlying
+    /// Storage; snapshot build/install paths look them up through `storage_swap` so
+    /// they remain valid after a hot swap.
     pub fn new(
         node_id: u64,
         storage_swap: Arc<ArcSwap<Storage>>,
@@ -208,6 +211,13 @@ impl KiwiStateMachine {
         }
     }
 
+    /// Initialize cf_tracker from restored SST properties after snapshot install
+    /// or from existing DB on startup
+    pub fn init_cf_tracker(&self) -> Result<(), io::Error> {
+        let storage = self.storage_swap.load_full();
+        storage.init_cf_trackers().map_err(io::Error::other)
+    }
+
     /// Set pause controller for coordinating with StorageServer.
     pub fn set_pause_controller(&mut self, controller: Arc<dyn PauseController>) {
         self.pause_controller = Some(controller);
@@ -349,31 +359,17 @@ impl RaftStateMachine<KiwiTypeConfig> for KiwiStateMachine {
         let db_id = current_storage.db_id;
 
         // Close old Storage to release RocksDB lock before restoring checkpoint.
-        // ArcSwap guarantees we have exclusive access during pause.
-        {
-            // Note: During normal operation, there may be other Arc references from
-            // pending requests or Raft apply operations. We need to close via ArcSwap.
-            // Use a different approach: create a temporary "closed" placeholder.
-            //
-            // Actually, we can call close() on the Storage directly since RocksDB
-            // close doesn't require exclusive ownership - it's idempotent.
-            // But to ensure proper cleanup, we create a new empty Storage as placeholder,
-            // swap it in, then drop the old one.
-
-            // Create placeholder Storage (not opened)
-            let placeholder = Arc::new(Storage::new(db_instance_num, db_id));
-
-            // Swap placeholder in - this releases ArcSwap's reference to old Storage
-            self.storage_swap.swap(placeholder);
-
-            // Now current_storage (the old Arc) is the only reference left
-            // It will be dropped at the end of this block, releasing RocksDB lock
-        }
+        // pause_controller has already drained pending requests, so swapping the
+        // placeholder in and dropping `current_storage` here is the only remaining
+        // reference — `restore_checkpoint_layout` below must run with no live
+        // RocksDB handle on `db_path`.
+        let placeholder = Arc::new(Storage::new(db_instance_num, db_id));
+        self.storage_swap.swap(placeholder);
+        drop(current_storage);
 
         log::info!("Old Storage dropped, RocksDB lock released");
 
         // ========== Phase 4: Restore checkpoint (atomic operation) ==========
-        // Now we can safely restore checkpoint since placeholder Storage holds no lock.
         restore_checkpoint_layout(&checkpoint_root, &self.db_path, db_instance_num).map_err(
             |e| {
                 cleanup_on_error();
@@ -394,16 +390,45 @@ impl RaftStateMachine<KiwiTypeConfig> for KiwiStateMachine {
         self.storage_swap.swap(Arc::new(new_storage));
         log::info!("Storage swapped to new instance after snapshot installation");
 
-        // ========== Phase 6: Resume StorageServer ==========
-        if let Some(ctrl) = &self.pause_controller {
-            ctrl.resume();
+        // ========== Phase 7: Update state and persist ==========
+        // Initialize cf_tracker from restored SST properties
+        self.init_cf_tracker().map_err(|e| {
+            cleanup_on_error();
+            StorageError::from_io_error(ErrorSubject::Snapshot(None), ErrorVerb::Write, e)
+        })?;
+
+        // Restore each instance's collector state from snapshot metadata. The new
+        // storage created its own collector/tracker instances during open(), so we
+        // must look them up through storage_swap rather than reusing pre-swap refs.
+        let storage_after_swap = self.storage_swap.load();
+        let collectors: Vec<_> = (0..db_instance_num)
+            .filter_map(|i| storage_after_swap.get_logindex_collector(i))
+            .collect();
+        file_meta.restore_collector_states(&collectors);
+
+        // Purge collector entries for indices compacted into the snapshot.
+        // This immediately compacts restored pairs to a single boundary entry at
+        // last_included_index, which is acceptable since the follower will receive
+        // new entries via replication after this snapshot is installed.
+        let purge_idx = file_meta.last_included_index as storage::logindex::LogIndex;
+        for c in &collectors {
+            c.purge(purge_idx);
         }
+        drop(storage_after_swap);
 
-        // ========== Phase 7: Update state and persist ==========
         self.last_applied = meta.last_log_id;
         self.last_membership = meta.last_membership.clone();
 
-        persist_current_snapshot(&self.snapshot_work_dir, meta, &bytes)?;
+        persist_current_snapshot(&self.snapshot_work_dir, meta, &bytes).inspect_err(|_| {
+            cleanup_on_error();
+        })?;
+
+        // ========== Phase 8: Resume StorageServer ==========
+        // Resume only after all state updates are complete, so applied_state()
+        // returns correct values if queries arrive immediately after resume.
+        if let Some(ctrl) = &self.pause_controller {
+            ctrl.resume();
+        }
 
         drop(unpack_root);
         log::info!("Snapshot installation complete");
@@ -420,6 +445,18 @@ impl RaftStateMachine<KiwiTypeConfig> for KiwiStateMachine {
         &mut self,
     ) -> Result<(Option<LogId<u64>>, StoredMembership<u64, KiwiNode>), openraft::StorageError<u64>>
     {
+        // On first access, lazily load from persisted snapshot to recover last_applied
+        // after restart (otherwise openraft would scan from index 0 and fail if logs were purged).
+        if self.last_applied.is_none() {
+            if let Some(snap) = load_current_snapshot(&self.snapshot_work_dir)? {
+                self.last_applied = snap.meta.last_log_id;
+                self.last_membership = snap.meta.last_membership.clone();
+                log::info!(
+                    "Recovered last_applied={:?} from persisted snapshot",
+                    self.last_applied
+                );
+            }
+        }
         Ok((self.last_applied, self.last_membership.clone()))
     }
 }
@@ -446,9 +483,14 @@ impl RaftSnapshotBuilder<KiwiTypeConfig> for KiwiSnapshotBuilder {
         } else {
             (0, 0)
         };
-        let raft_meta = RaftSnapshotMeta::new(last_idx, last_term);
 
+        // Snapshot meta carries each instance's collector state so the receiver can
+        // rebuild every (log_index, seqno) mapping, not just instance 0's.
         let storage = self._storage.load_full();
+        let collectors: Vec<_> = (0..storage.db_instance_num)
+            .filter_map(|i| storage.get_logindex_collector(i))
+            .collect();
+        let raft_meta = RaftSnapshotMeta::with_collector_states(last_idx, last_term, &collectors);
         storage
             .create_checkpoint(&dir, &raft_meta)
             .map_err(storage_err_to_raft)?;
@@ -470,6 +512,13 @@ impl RaftSnapshotBuilder<KiwiTypeConfig> for KiwiSnapshotBuilder {
 
         persist_current_snapshot(&self.snapshot_work_dir, &meta, &bytes)?;
 
+        // Purge collector entries that are now covered by the snapshot.
+        // This prevents unbounded memory growth as the leader continues accepting writes.
+        let purge_idx = raft_meta.last_included_index as storage::logindex::LogIndex;
+        for c in &collectors {
+            c.purge(purge_idx);
+        }
+
         Ok(Snapshot {
             meta,
             snapshot: Box::new(Cursor::new(bytes)),