Execute a snapshot restore. Restores engine state from a snapshot in the object store, then writes a restore marker so the normal startup path replays WAL records from the snapshot LSN forward. `data_dir` is the server's data directory (where engine files live). `prefix` is the snapshot object-store prefix (e.g. `"snap-000001-lsn…"`). `snapshot_store` is the object store holding the snapshot obj
(
data_dir: &Path,
prefix: &str,
snapshot_store: &Arc<dyn ObjectStore>,
restore_store: &Arc<dyn ObjectStore>,
wal_records: &[nodedb_wal::WalRecord],
)
| 59 | /// |
| 60 | /// This is an **offline operation** — must not be called while serving traffic. |
| 61 | pub async fn execute_restore( |
| 62 | data_dir: &Path, |
| 63 | prefix: &str, |
| 64 | snapshot_store: &Arc<dyn ObjectStore>, |
| 65 | restore_store: &Arc<dyn ObjectStore>, |
| 66 | wal_records: &[nodedb_wal::WalRecord], |
| 67 | ) -> crate::Result<RestoreResult> { |
| 68 | let manifest = load_manifest(snapshot_store, prefix).await?; |
| 69 | let snapshot_lsn = manifest.meta.end_lsn.as_u64(); |
| 70 | |
| 71 | info!( |
| 72 | snapshot_id = manifest.meta.snapshot_id, |
| 73 | snapshot_lsn, |
| 74 | cores = manifest.num_cores, |
| 75 | "starting snapshot restore" |
| 76 | ); |
| 77 | |
| 78 | let mut total_docs = 0u64; |
| 79 | let mut total_vectors = 0u64; |
| 80 | |
| 81 | for core_id in 0..manifest.num_cores { |
| 82 | let core_snap = load_core_snapshot(snapshot_store, prefix, core_id, None).await?; |
| 83 | let (docs, vectors) = restore_core_state(data_dir, core_id, &core_snap)?; |
| 84 | total_docs += docs; |
| 85 | total_vectors += vectors; |
| 86 | |
| 87 | info!( |
| 88 | core_id, |
| 89 | documents = docs, |
| 90 | vectors, |
| 91 | watermark = core_snap.watermark, |
| 92 | "core state restored" |
| 93 | ); |
| 94 | } |
| 95 | |
| 96 | let wal_to_replay: Vec<_> = wal_records |
| 97 | .iter() |
| 98 | .filter(|r| r.header.lsn > snapshot_lsn) |
| 99 | .collect(); |
| 100 | |
| 101 | let wal_count = wal_to_replay.len() as u64; |
| 102 | if wal_count > 0 { |
| 103 | info!( |
| 104 | records = wal_count, |
| 105 | from_lsn = snapshot_lsn + 1, |
| 106 | "replaying WAL records after snapshot" |
| 107 | ); |
| 108 | write_restore_marker(restore_store, snapshot_lsn).await?; |
| 109 | } |
| 110 | |
| 111 | let result = RestoreResult { |
| 112 | snapshot_id: manifest.meta.snapshot_id, |
| 113 | restored_lsn: manifest.meta.end_lsn, |
| 114 | cores_restored: manifest.num_cores, |
| 115 | documents_restored: total_docs, |
| 116 | vectors_restored: total_vectors, |
| 117 | wal_records_replayed: wal_count, |
| 118 | }; |