feat: multi-part AOF persistence + fast RDB loader

src/command/persistence.rs

@@ -212,6 +212,21 @@ pub fn bgrewriteaof_start(aof_tx: &channel::MpscSender<AofMessage>, db: SharedDa
     }
 }
 
+/// Start BGREWRITEAOF in sharded mode using ShardDatabases.
+pub fn bgrewriteaof_start_sharded(
+    aof_tx: &channel::MpscSender<AofMessage>,
+    shard_databases: std::sync::Arc<crate::shard::shared_databases::ShardDatabases>,
+) -> Frame {
+    match aof_tx.try_send(AofMessage::RewriteSharded(shard_databases)) {
+        Ok(()) => Frame::SimpleString(Bytes::from_static(
+            b"Background append only file rewriting started",
+        )),
+        Err(_) => Frame::Error(Bytes::from_static(
+            b"ERR Background AOF rewrite failed to start",
+        )),
+    }
+}
+
 /// SAVE command: synchronous save to disk. Blocks until complete.
 ///
 /// Clones all entries under read locks (same as BGSAVE), then serializes

src/main.rs

@@ -240,6 +240,43 @@ fn main() -> anyhow::Result<()> {
         })
         .collect();
 
+    // Multi-part AOF replay layered on top of v2/v3 recovery.
+    // Priority: if appendonlydir/ manifest exists → load multi-part (skip legacy v2 fallback).
+    // Otherwise v2 already handled legacy appendonly.aof during restore_from_persistence.
+    if config.appendonly == "yes" {
+        if let Some(ref dir) = persistence_dir {
+            use moon::persistence::aof_manifest::AofManifest;
+            use moon::persistence::replay::DispatchReplayEngine;
+            let base_dir = std::path::PathBuf::from(dir);
+            if let Some(manifest) = AofManifest::load(&base_dir) {
+                if num_shards == 1 {
+                    match moon::persistence::aof_manifest::replay_multi_part(
+                        &mut shards[0].databases,
+                        &manifest,
+                        &DispatchReplayEngine,
+                    ) {
+                        Ok(n) => info!(
+                            "AOF multi-part loaded (seq {}): {} entries",
+                            manifest.seq, n
+                        ),
+                        Err(e) => tracing::error!("Multi-part AOF load failed: {}", e),
+                    }
+                } else {
+                    tracing::warn!(
+                        "Multi-part AOF skipped in multi-shard mode (not yet supported)"
+                    );
+                }
+            }
+            // Initialize manifest for writer thread (safe — recovery is complete).
+            // On fresh/legacy upgrade, writer is blocked waiting for this file.
+            if AofManifest::load(&base_dir).is_none() {
+                if let Err(e) = AofManifest::initialize(&base_dir) {
+                    tracing::error!("Failed to initialize AOF manifest: {}", e);
+                }
+            }
+        }
+    }
+
     // Extract databases from all shards and wrap in ShardDatabases
     let all_dbs: Vec<Vec<moon::storage::Database>> = shards
         .iter_mut()