feat(raft): implement snapshot creation and restore (#255)

* feat(raft): implement snapshot creation and restore

Implement core snapshot functionality to establish checkpoint → tar → restore

* fix

* feat(raft): implement snapshot install hot-swapping with ArcSwap

Solves P0/P1 issues from PR #255 review:

- P0: Active DB handle problem - use placeholder Storage swap to release
  old Storage before restoring checkpoint, avoiding RocksDB lock conflict
- P0: Metadata validation - validate checkpoint metadata against SnapshotMeta
- P1: Windows compatibility - atomic file replacement with cfg conditionals

Key changes:
- Add ArcSwap pattern: GlobalStorage wraps ArcSwap<Storage> for hot-swapping
- StorageServerPauseController for coordinating pause during snapshot install
- KiwiStateMachine.swap() atomically switches to new Storage after install
- PauseController trait integration between StorageServer and RaftNode

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fmt

* fix(storage): share ArcSwap across all components and deduplicate drop/close logic

Address PR review comments:

1. GlobalStorage now uses Arc<ArcSwap<Storage>> so all clones share the same
   ArcSwap instance. Previously, clone() created a new ArcSwap, causing
   StorageServer and Raft to use separate containers - swap() in StateMachine
   was invisible to StorageServer.

2. Extract release_resources() private method to deduplicate cleanup logic
   between Drop::drop() and close(). Both now call this shared method,
   eliminating code redundancy.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: ruojieranyishen

Cargo.lock

@@ -74,6 +74,7 @@ rand = "0.8"
 uuid = { version = "1.0", features = ["v4", "serde"] }
 openraft = { version = "0.9", features = ["serde", "storage-v2"] }
 proptest = "1.5"
+arc-swap = "1.7"
 
 ## workspaces members
 engine = { path = "src/engine" }

Cargo.toml

@@ -21,6 +21,7 @@ storage = { workspace = true }
 bytes = { workspace = true }
 snafu = { workspace = true }
 chrono = { version = "0.4", features = ["serde"] }
+arc-swap = { workspace = true }
 
 [dev-dependencies]
 tokio-test = "0.4"

src/common/runtime/Cargo.toml

@@ -0,0 +1,90 @@
+// Copyright (c) 2024-present, arana-db Community.  All rights reserved.
+//
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Global Storage wrapper using ArcSwap for hot-swapping Storage during snapshot installation.
+//!
+//! This module provides `GlobalStorage` which wraps `ArcSwap<Storage>`. All components
+//! that need access to Storage should use `GlobalStorage::load()` to get the current
+//! instance. During snapshot installation, `swap()` atomically switches to the new Storage.
+
+use std::sync::Arc;
+
+use arc_swap::ArcSwap;
+use storage::storage::Storage;
+
+/// Global storage wrapper that enables hot-swapping Storage during snapshot installation.
+///
+/// Uses `Arc<ArcSwap<Storage>>` internally so that cloning shares the same ArcSwap instance.
+/// All holders get the current Storage instance via `load()`, which is a cheap atomic read (~1-2ns).
+/// During snapshot installation, `swap()` atomically switches to the new Storage, and all
+/// clones see the update immediately.
+pub struct GlobalStorage {
+    inner: Arc<ArcSwap<Storage>>,
+}
+
+impl GlobalStorage {
+    /// Create a new GlobalStorage with the initial Storage instance.
+    pub fn new(storage: Storage) -> Self {
+        Self {
+            inner: Arc::new(ArcSwap::from(Arc::new(storage))),
+        }
+    }
+
+    /// Create from an existing Arc<Storage>.
+    pub fn from_arc(storage: Arc<Storage>) -> Self {
+        Self {
+            inner: Arc::new(ArcSwap::from(storage)),
+        }
+    }
+
+    /// Get the current Storage instance.
+    /// This is a cheap atomic read (~1-2ns overhead).
+    pub fn load(&self) -> Arc<Storage> {
+        self.inner.load_full()
+    }
+
+    /// Swap to a new Storage instance atomically.
+    /// Used during snapshot installation to switch to restored data.
+    /// All clones of this GlobalStorage will see the new Storage immediately.
+    pub fn swap(&self, new_storage: Arc<Storage>) {
+        self.inner.swap(new_storage);
+    }
+
+    /// Get the underlying Arc<ArcSwap<Storage>> for passing to Raft.
+    /// Raft needs Arc<ArcSwap<Storage>> for KiwiStateMachine.
+    pub fn arc_swap(&self) -> Arc<ArcSwap<Storage>> {
+        Arc::clone(&self.inner)
+    }
+
+    /// Get db_instance_num from current Storage.
+    pub fn db_instance_num(&self) -> usize {
+        self.load().db_instance_num
+    }
+
+    /// Get db_id from current Storage.
+    pub fn db_id(&self) -> usize {
+        self.load().db_id
+    }
+}
+
+impl Clone for GlobalStorage {
+    fn clone(&self) -> Self {
+        Self {
+            inner: Arc::clone(&self.inner), // Share the same ArcSwap!
+        }
+    }
+}

src/common/runtime/global_storage.rs

@@ -23,6 +23,7 @@
 pub mod config;
 pub mod error;
 pub mod error_logging;
+pub mod global_storage;
 pub mod manager;
 pub mod message;
 pub mod metrics;
@@ -46,6 +47,7 @@ pub use error_logging::{
     CorrelationId, ErrorCategory, ErrorEvent, ErrorLogger, ErrorLoggingConfig, ErrorMetrics,
     ErrorRates, RuntimeContext, get_global_error_logger, init_global_error_logger,
 };
+pub use global_storage::GlobalStorage;
 pub use manager::{RuntimeHealth as ManagerRuntimeHealth, RuntimeManager, RuntimeStats};
 pub use message::{
     BackpressureConfig, ChannelStats, CircuitBreaker, MessageChannel, QueueStats, QueuedRequest,
@@ -63,5 +65,5 @@ pub use metrics::{
 };
 pub use storage_server::{
     BackgroundTaskConfig, BackgroundTaskManager, BackgroundTaskStats, BatchConfig, BatchProcessor,
-    BatchStats, RocksDbStats, StorageServer, StorageServerConfig,
+    BatchStats, RocksDbStats, StorageServer, StorageServerConfig, StorageServerPauseController,
 };