Rebuild a drive's arena from its source stores when the snapshot is missing or quarantined (#3139)

* Rebuild a drive's arena from its source stores when the snapshot is missing or quarantined

Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com>

* Store rebuild runs off the request path and skips a partition a restore already filled

Every reader on a drive funnels through the restore gate, so the fill now
runs on its own task: the gate resolves as soon as it starts, roots stay in
progress until they land and listings serve the store meanwhile. A partition
rebuild_quiesced already filled resolves as restored without a second read.
The fill reads a rebuild-only select without the content join: 18.9 s for a
1.01M-record store in release, from 33.1 s.

Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com>

* A browse beneath a source being filled serves the store instead of clearing the subtree

is_under_indexing is the ancestor-aware check: a store fill or recursive
walk over a root covers everything beneath it, and a browse dispatched into
that tree ran clear_for_reindex over what the fill had already placed. The
acceptance test lists a subdirectory during the fill and the deepest one
after it. REBUILD_PAGE moved above the fill's doc block.

Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com>

* Scope the mid-fill store answer to roots being filled from a store

indexing_in_progress also holds walk roots, whose stores cannot promise an
answer beneath them, so the prefix gate reads a separate filling_from_store
set and falls through to the ordinary dispatch when the store has no row.

Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com>

---------

Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com>

core/src/ops/files/query/directory_listing.rs

@@ -304,6 +304,21 @@ impl DirectoryListingQuery {
 			});
 		}
 
+		// A source whose map is being filled from its store is in that state
+		// for the length of the fill, and its records are already true, so
+		// the store answers for anything beneath the root until the arena
+		// does; a browse dispatched into that tree would clear the subtree
+		// the fill has already placed.
+		if cache.is_filling_from_store(&local_path) {
+			if let Some(listing) = self.list_from_store(&context, &local_path).await {
+				tracing::debug!(
+					"Map of {} is being filled from its store; serving the store",
+					local_path.display()
+				);
+				return Ok(listing);
+			}
+		}
+
 		// No cached index or index doesn't cover this path
 		// Check if indexing is already in progress
 		if cache.is_indexing(&local_path) {

core/src/ops/indexing/snapshot.rs

@@ -223,6 +223,34 @@ pub(super) fn save_snapshot_impl(
 	Ok(())
 }
 
+/// Write an empty artifact in the current layout stamped with `version`, so
+/// a test can stand in for the file a previous build left behind.
+#[cfg(test)]
+pub(super) fn write_artifact_with_version(
+	snapshot_path: &Path,
+	version: u32,
+	source_id: Uuid,
+	root_path: &Path,
+) {
+	let stale = IndexSnapshot {
+		version,
+		source_id,
+		root_path: root_path.to_path_buf(),
+		created_at_secs: 0,
+		path_index: HashMap::new(),
+		entry_uuids: HashMap::new(),
+		content_kinds: HashMap::new(),
+		stats: IndexerStats::default(),
+		name_cache_strings: Vec::new(),
+		name_registry_map: Vec::new(),
+		arena_entries: Vec::new(),
+		stubs: Vec::new(),
+	};
+	let bytes = postcard::to_allocvec(&stale).expect("encode");
+	let compressed = zstd::encode_all(bytes.as_slice(), 6).expect("compress");
+	fs::write(snapshot_path, compressed).expect("write artifact");
+}
+
 /// Move an artifact that will not load to `<name>.corrupt-<unix seconds>`
 /// beside it.
 ///

core/src/ops/indexing/volume_index.rs

@@ -45,15 +45,20 @@ pub struct Partition {
 	index: Arc<TokioRwLock<Arena>>,
 	indexed_paths: RwLock<HashSet<PathBuf>>,
 	indexing_in_progress: RwLock<HashSet<PathBuf>>,
+	/// Source roots whose arena is being filled from their stores. A fill
+	/// covers everything beneath its root and the store answers for all of
+	/// it, which a walk in `indexing_in_progress` does not promise.
+	filling_from_store: RwLock<HashSet<PathBuf>>,
 	watched_paths: RwLock<HashSet<PathBuf>>,
 	/// A detached source's root is not present on disk; its index is served
 	/// read-only from a restored snapshot and must never trigger indexing.
 	detached: AtomicBool,
-	/// Set once a snapshot restore has populated the index this session.
+	/// Set once a snapshot restore or a store rebuild has populated the
+	/// index this session.
 	restored: AtomicBool,
 	/// Single restore attempt per session: concurrent callers await the same
 	/// load instead of racing three copies of a 100 MB deserialization.
-	restore_once: tokio::sync::OnceCell<bool>,
+	restore_once: tokio::sync::OnceCell<RestoreOutcome>,
 	/// Serializes snapshot saves for this drive.
 	save_lock: tokio::sync::Mutex<()>,
 	/// Entry count at the last completed save; identical partitions skip the
@@ -61,6 +66,21 @@ pub struct Partition {
 	last_saved_entries: std::sync::atomic::AtomicU64,
 }
 
+/// How a partition's session restore ended.
+///
+/// A snapshot carries the whole drive map; a store rebuild carries only the
+/// registered sources on it. The discovery pass walks the rest of a drive
+/// whose snapshot is gone, so the two are kept apart.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum RestoreOutcome {
+	/// Nothing populated the partition: no snapshot and no source store.
+	Nothing,
+	/// The drive snapshot loaded.
+	Snapshot,
+	/// At least one source's records were read back from its store.
+	Stores,
+}
+
 impl Partition {
 	fn new(volume: VolumeKey, root: Option<PathBuf>) -> std::io::Result<Arc<Self>> {
 		Ok(Arc::new(Self {
@@ -69,6 +89,7 @@ impl Partition {
 			index: Arc::new(TokioRwLock::new(Arena::new()?)),
 			indexed_paths: RwLock::new(HashSet::new()),
 			indexing_in_progress: RwLock::new(HashSet::new()),
+			filling_from_store: RwLock::new(HashSet::new()),
 			watched_paths: RwLock::new(HashSet::new()),
 			detached: AtomicBool::new(false),
 			restored: AtomicBool::new(false),
@@ -135,14 +156,20 @@ struct TrackedVolume {
 	is_mount: bool,
 }
 
+/// Rows read per arena lock during a store rebuild. At the measured insert
+/// cost a page holds the lock for a few milliseconds.
+const REBUILD_PAGE: usize = 2_000;
+
 /// Add every filesystem row of a store to an arena, rooted at `root`.
 ///
 /// Ancestors are synthesized by the arena itself and content kinds derive
 /// from extensions the way a fresh walk derives them, which is what makes a
 /// database browsable and searchable through the same paths an arena
-/// snapshot is. Returns how many entries were added.
+/// snapshot is. The arena is locked one store page at a time, so a
+/// multi-million-record rebuild never holds a listing on the same drive
+/// for longer than one page of inserts. Returns how many entries were added.
 pub(crate) async fn fill_arena_from_store(
-	index: &mut Arena,
+	index: &TokioRwLock<Arena>,
 	db: &sd_store::SourceDb,
 	root: &Path,
 ) -> anyhow::Result<usize> {
@@ -153,12 +180,14 @@ pub(crate) async fn fill_arena_from_store(
 	let mut added = 0usize;
 	let mut after_rowid = 0i64;
 	loop {
-		let (entries, last) = sd_store::read::all_entries_page(db.pool(), after_rowid, 2_000)
-			.await
-			.map_err(|e| anyhow::anyhow!("database page failed: {e}"))?;
-		let done = entries.len() < 2_000;
+		let (entries, last) =
+			sd_store::read::rebuild_entries_page(db.pool(), after_rowid, REBUILD_PAGE)
+				.await
+				.map_err(|e| anyhow::anyhow!("database page failed: {e}"))?;
+		let done = entries.len() < REBUILD_PAGE;
 		after_rowid = last;
 
+		let mut index = index.write().await;
 		for entry in entries {
 			let path = root.join(&entry.relative_path);
 			let from_ms = |ms: Option<i64>| {
@@ -186,6 +215,7 @@ pub(crate) async fn fill_arena_from_store(
 			index.add_entry(path, entry.uuid, metadata)?;
 			added += 1;
 		}
+		drop(index);
 
 		if done {
 			break;
@@ -194,6 +224,29 @@ pub(crate) async fn fill_arena_from_store(
 	Ok(added)
 }
 
+/// Fill a source's subtree of a partition from its store and, when the
+/// store held anything, mark the root indexed and the partition restored.
+/// Announcing the root is the caller's: it decides whether a watch should
+/// follow. `None` when the store could not be read.
+async fn fill_source_from_store(
+	slot: &Partition,
+	db: &sd_store::SourceDb,
+	root: &Path,
+) -> Option<usize> {
+	let loaded = match fill_arena_from_store(&slot.index, db, root).await {
+		Ok(loaded) => loaded,
+		Err(error) => {
+			tracing::warn!(root = %root.display(), %error, "could not rebuild the map from the store");
+			return None;
+		}
+	};
+	if loaded > 0 {
+		slot.indexed_paths.write().insert(root.to_path_buf());
+		slot.restored.store(true, Ordering::Release);
+	}
+	Some(loaded)
+}
+
 /// Whether a volume root is a service prefix rather than a filesystem path.
 ///
 /// A cloud volume mounts at `s3://bucket` or `gdrive://id`: nothing on this
@@ -1778,6 +1831,19 @@ impl VolumeIndex {
 			.contains(path)
 	}
 
+	/// Whether `path` sits under a source root being filled from its store.
+	/// The fill covers everything beneath the root and the store answers for
+	/// all of it, so a browse dispatched into that tree would only clear what
+	/// the fill has already placed. A walk in progress makes no such promise
+	/// and is not consulted here.
+	pub fn is_filling_from_store(&self, path: &Path) -> bool {
+		self.resolve(path)
+			.filling_from_store
+			.read()
+			.iter()
+			.any(|filling| path.starts_with(filling))
+	}
+
 	/// Restore a registered source's snapshot into its partition, if it has
 	/// one and hasn't been restored this session. Returns true when the
 	/// source's data is available afterwards (restored now or already live).
@@ -1799,16 +1865,137 @@ impl VolumeIndex {
 			VolumeKey::Id(uuid) => self.volume_mounted(*uuid).unwrap_or(true),
 			_ => true,
 		};
-		let restored = *slot
+		let outcome = *slot
 			.restore_once
-			.get_or_init(|| Self::attempt_restore(self.dirs.clone(), slot.clone(), mounted))
+			.get_or_init(|| async {
+				// A library restore fills the drive's fresh partition through
+				// `rebuild_quiesced` before anything asks here; reading the
+				// stores again would only repeat it.
+				if slot.restored.load(Ordering::Acquire) {
+					return RestoreOutcome::Stores;
+				}
+				if Self::attempt_restore(self.dirs.clone(), slot.clone(), mounted).await {
+					return RestoreOutcome::Snapshot;
+				}
+				// A missing, quarantined or outdated snapshot costs a walk of
+				// every source on the drive, and every format bump produces
+				// exactly that. The stores hold the same records with the same
+				// uuids, so they refill the arena instead, off this request:
+				// every reader on the drive funnels through this gate, so the
+				// fill runs on its own task and each source announces itself
+				// as it lands, which flips its routing and arms its watch. A
+				// detached drive keeps nothing to rebuild for.
+				if slot.is_detached() || !self.rebuild_sources_from_stores(&slot).await {
+					return RestoreOutcome::Nothing;
+				}
+				RestoreOutcome::Stores
+			})
 			.await;
 
-		if restored && !already_restored {
+		if outcome == RestoreOutcome::Snapshot && !already_restored {
 			self.announce_restored(&root);
 		}
 
-		restored || !slot.indexed_paths.read().is_empty()
+		outcome != RestoreOutcome::Nothing || !slot.indexed_paths.read().is_empty()
+	}
+
+	/// Whether the drive under `path` was restored from its snapshot this
+	/// session. A store rebuild does not count: it covers the registered
+	/// sources and nothing else on the drive.
+	pub fn restored_from_snapshot(&self, path: &Path) -> bool {
+		let Some(resolved) = self.locate(path) else {
+			return false;
+		};
+		let slots = self.slots.read();
+		slots
+			.get(&resolved.volume)
+			.and_then(|slot| slot.restore_once.get())
+			.is_some_and(|outcome| *outcome == RestoreOutcome::Snapshot)
+	}
+
+	/// Start rebuilding every attached source on a drive partition from its
+	/// store, on one background task, and say whether anything was started.
+	///
+	/// Only a source whose row records a count is rebuilt. The count is
+	/// written when a snapshot is saved, so it is the evidence that a map
+	/// existed and was lost; a source nothing has walked to completion keeps
+	/// answering from its store without an arena, as R6 routes it, and a
+	/// source with no readable store is left for the coverage heal to walk.
+	///
+	/// Each root is marked in progress until its fill lands, so the heal
+	/// does not dispatch a walk over it, a listing serves its store in the
+	/// meantime, and the status surface shows the work. The stores are
+	/// opened here, before the task starts, so the caller learns at once
+	/// whether the drive has anything to rebuild from.
+	async fn rebuild_sources_from_stores(&self, slot: &Arc<Partition>) -> bool {
+		let mut pending = Vec::new();
+		for (_, located) in self.all_sources() {
+			if located.volume != slot.volume
+				|| located.record.record_count.unwrap_or(0) == 0
+				|| !self.root_attached(&located.volume, &located.record.root)
+			{
+				continue;
+			}
+			let db = self.read_store(located.record.id).await;
+			let first_row = match &db {
+				Some(db) => sd_store::read::rebuild_entries_page(db.pool(), 0, 1)
+					.await
+					.map(|(rows, _)| !rows.is_empty())
+					.unwrap_or(false),
+				None => false,
+			};
+			let (Some(db), true) = (db, first_row) else {
+				tracing::warn!(
+					source = %located.record.id,
+					root = %located.record.root.display(),
+					"no usable snapshot and no store records; the source will be walked"
+				);
+				continue;
+			};
+			slot.indexing_in_progress
+				.write()
+				.insert(located.record.root.clone());
+			slot.filling_from_store
+				.write()
+				.insert(located.record.root.clone());
+			pending.push((located.record.id, located.record.root, db));
+		}
+		if pending.is_empty() {
+			return false;
+		}
+
+		let slot = slot.clone();
+		let announce = self.restored_roots.read().clone();
+		tokio::spawn(async move {
+			for (source_id, root, db) in pending {
+				let started = Instant::now();
+				let filled = fill_source_from_store(&slot, &db, &root).await;
+				slot.indexing_in_progress.write().remove(&root);
+				slot.filling_from_store.write().remove(&root);
+				match filled {
+					Some(loaded) if loaded > 0 => {
+						if let Some(sender) = &announce {
+							let _ = sender.send(root.clone());
+						}
+						tracing::info!(
+							source = %source_id,
+							root = %root.display(),
+							loaded,
+							took = ?started.elapsed(),
+							"no usable snapshot; map rebuilt from the source store"
+						);
+					}
+					_ => {
+						tracing::warn!(
+							source = %source_id,
+							root = %root.display(),
+							"no usable snapshot and no store records; the source will be walked"
+						);
+					}
+				}
+			}
+		});
+		true
 	}
 
 	/// Rebuild a source's map from its store, without walking the disk.
@@ -1819,24 +2006,17 @@ impl VolumeIndex {
 	/// the arena, the root is marked indexed and announced the way a snapshot
 	/// restore announces it, which is what arms the watch over it again.
 	/// Returns how many entries were loaded, or `None` when the source has
-	/// no readable store.
+	/// no readable store. A store with no records marks nothing: an empty
+	/// arena has no parent to file a change under, so the root is left for
+	/// a walk to claim.
 	pub async fn rebuild_from_store(&self, source_id: Uuid) -> Option<usize> {
 		let root = self.source_root(source_id)?;
 		let db = self.read_store(source_id).await?;
 		let slot = self.resolve(&root);
-		let loaded = {
-			let mut index = slot.index.write().await;
-			match fill_arena_from_store(&mut index, &db, &root).await {
-				Ok(loaded) => loaded,
-				Err(error) => {
-					tracing::warn!(source = %source_id, %error, "could not rebuild the map from the store");
-					return None;
-				}
-			}
-		};
-		slot.indexed_paths.write().insert(root.clone());
-		slot.restored.store(true, Ordering::Release);
-		self.announce_restored(&root);
+		let loaded = fill_source_from_store(&slot, &db, &root).await?;
+		if loaded > 0 {
+			self.announce_restored(&root);
+		}
 		Some(loaded)
 	}
 
@@ -2701,6 +2881,36 @@ mod tests {
 			}
 		}
 
+		/// A store rebuild lands on its own task; wait for the root to leave
+		/// the in-progress set.
+		async fn wait_for_rebuild(cache: &VolumeIndex, root: &Path) {
+			for _ in 0..600 {
+				if !cache.is_indexing(root) {
+					return;
+				}
+				tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+			}
+			panic!("the rebuild of {} did not finish", root.display());
+		}
+
+		/// Index `count` files under an already registered source and save
+		/// its snapshot, the way a completed walk leaves it. Store records
+		/// written before this call land on the registry row's count.
+		async fn walked_and_saved(cache: &VolumeIndex, root: &Path, count: u64) {
+			let index = cache.create_for_indexing(root.to_path_buf());
+			{
+				let mut index = index.write().await;
+				for i in 0..count {
+					let path = root.join(format!("file-{i}"));
+					index
+						.add_entry(path.clone(), Uuid::now_v7(), entry(&path))
+						.expect("add");
+				}
+			}
+			cache.mark_indexing_complete(root);
+			cache.save_snapshot(root).await.expect("save");
+		}
+
 		/// A source whose partition holds `count` files, indexed and saved the
 		/// way a completed walk leaves it.
 		async fn indexed_source(
@@ -3113,18 +3323,19 @@ mod tests {
 
 		/// R8 "Missing or invalid restart snapshot".
 		///
-		/// A snapshot that will not parse restores nothing, but the source it
-		/// belonged to stays registered and listed, its store still answers,
-		/// and the unreadable artifact stays on disk for diagnosis.
+		/// A snapshot that will not parse restores nothing from the file, but
+		/// the source it belonged to stays registered and listed, and its map
+		/// comes back from its store with the store's uuids, so no walk is
+		/// needed to list it again.
 		#[tokio::test]
-		async fn an_invalid_snapshot_leaves_the_source_visible_and_its_store_readable() {
+		async fn an_invalid_snapshot_leaves_the_source_visible_and_rebuilt_from_its_store() {
 			let data = tempfile::tempdir().unwrap();
 			let library = test_library(data.path()).await;
 			let root_dir = tempfile::tempdir().unwrap();
 			let root = root_dir.path().to_path_buf();
 			std::fs::write(root.join("kept.txt"), b"kept").unwrap();
 
-			let (id, snapshot_path) = {
+			let (id, snapshot_path, kept_uuid) = {
 				let cache =
 					VolumeIndex::with_sources_dir(Some(data.path().to_path_buf())).expect("cache");
 				cache
@@ -3132,17 +3343,25 @@ mod tests {
 					.await
 					.expect("attach");
 				let anchor = tracked_volume(&library, &root).await;
-				let id = indexed_source(&cache, &root, anchor, 4).await;
+				let id = cache
+					.register_source(&root, Some(anchor))
+					.await
+					.expect("register");
 				let store = cache
 					.store_for(&root.join("kept.txt"))
 					.await
 					.expect("store");
-				store
+				let kept_uuid = store
 					.identify_one(&entry(&root.join("kept.txt")), None)
 					.await
 					.expect("identified");
 				store.flush().await.expect("flush");
-				(id, cache.snapshot_path_for(&root).expect("snapshot path"))
+				walked_and_saved(&cache, &root, 4).await;
+				(
+					id,
+					cache.snapshot_path_for(&root).expect("snapshot path"),
+					kept_uuid,
+				)
 			};
 			assert!(snapshot_path.exists());
 			std::fs::write(&snapshot_path, b"this is not a snapshot").unwrap();
@@ -3154,22 +3373,279 @@ mod tests {
 				.await
 				.expect("attach");
 			assert!(
-				!cache.ensure_restored(&root).await,
-				"junk must not restore as an index"
+				cache.ensure_restored(&root).await,
+				"the store refills the map the junk could not"
+			);
+			wait_for_rebuild(&cache, &root).await;
+			assert!(
+				!cache.restored_from_snapshot(&root),
+				"nothing came from the snapshot file"
 			);
 			assert!(
 				cache.sources().iter().any(|s| s.id == id),
 				"the source stays registered without its cache"
 			);
 			let db = cache.read_store(id).await.expect("the store still opens");
-			assert!(
-				db.resolve_path("kept.txt").await.expect("query").is_some(),
-				"retained records answer without the arena"
+			assert_eq!(
+				db.resolve_path("kept.txt").await.expect("query"),
+				Some(kept_uuid),
+				"retained records answer from the store"
 			);
 			assert!(
-				!cache.arena_answers(&root),
-				"a query over this source routes to the store, not an empty arena"
+				cache.arena_answers(&root),
+				"a query over this source routes to the rebuilt arena"
 			);
+			let index = cache.get_for_search(&root).expect("rebuilt index");
+			let index = index.read().await;
+			assert_eq!(
+				index.get_entry_uuid(&root.join("kept.txt")),
+				Some(kept_uuid),
+				"the rebuilt map carries the store's identity"
+			);
+			assert!(
+				index.get_entry_uuid(&root.join("file-0")).is_none(),
+				"entries the snapshot alone held are gone with it"
+			);
+		}
+
+		/// A v3-shaped artifact after the v4 bump: the header decodes, the
+		/// version check refuses it, and the slot is quarantined like any
+		/// other unreadable snapshot. The map is rebuilt from the store
+		/// rather than walked, and the artifact is kept for diagnosis.
+		#[tokio::test]
+		async fn an_older_format_snapshot_is_quarantined_and_the_map_rebuilt_from_the_store() {
+			let data = tempfile::tempdir().unwrap();
+			let library = test_library(data.path()).await;
+			let root_dir = tempfile::tempdir().unwrap();
+			let root = root_dir.path().to_path_buf();
+			for name in ["a.txt", "b.txt", "sub/c.txt"] {
+				let path = root.join(name);
+				std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+				std::fs::write(&path, name).unwrap();
+			}
+
+			let (id, snapshot_path, volume_index_id, uuids) = {
+				let cache =
+					VolumeIndex::with_sources_dir(Some(data.path().to_path_buf())).expect("cache");
+				cache
+					.attach_library(LIBRARY, library.clone())
+					.await
+					.expect("attach");
+				let anchor = tracked_volume(&library, &root).await;
+				let volume_index_id = VolumeKey::Id(anchor.uuid).id();
+				let id = cache
+					.register_source(&root, Some(anchor))
+					.await
+					.expect("register");
+				let store = cache.store_for(&root.join("a.txt")).await.expect("store");
+				let mut uuids = Vec::new();
+				for name in ["a.txt", "b.txt", "sub/c.txt"] {
+					let path = root.join(name);
+					let uuid = store
+						.identify_one(&entry(&path), None)
+						.await
+						.expect("identified");
+					uuids.push((path, uuid));
+				}
+				store.flush().await.expect("flush");
+				walked_and_saved(&cache, &root, 0).await;
+				(
+					id,
+					cache.snapshot_path_for(&root).expect("snapshot path"),
+					volume_index_id,
+					uuids,
+				)
+			};
+
+			// The previous build's artifact: the same container and header
+			// layout, stamped with the format it was written in.
+			crate::ops::indexing::snapshot::write_artifact_with_version(
+				&snapshot_path,
+				3,
+				volume_index_id,
+				&root,
+			);
+
+			let cache =
+				VolumeIndex::with_sources_dir(Some(data.path().to_path_buf())).expect("cache");
+			cache
+				.attach_library(LIBRARY, library)
+				.await
+				.expect("attach");
+			assert!(
+				cache.ensure_restored(&root).await,
+				"the map is rebuilt without a walk"
+			);
+			assert!(
+				cache.is_indexing(&root) || cache.arena_answers(&root),
+				"the rebuild is in progress or landed"
+			);
+			wait_for_rebuild(&cache, &root).await;
+			assert!(!cache.restored_from_snapshot(&root));
+			assert!(cache.sources().iter().any(|s| s.id == id));
+			assert!(
+				!snapshot_path.exists(),
+				"the slot is cleared so the next save lands clean"
+			);
+			let name = snapshot_path.file_name().unwrap().to_string_lossy();
+			let retained = std::fs::read_dir(snapshot_path.parent().unwrap())
+				.unwrap()
+				.filter_map(|entry| entry.ok().map(|entry| entry.path()))
+				.filter(|path| {
+					path.file_name()
+						.map(|n| n.to_string_lossy().starts_with(&format!("{name}.corrupt-")))
+						.unwrap_or(false)
+				})
+				.count();
+			assert_eq!(retained, 1, "the v3 artifact is kept beside the slot");
+
+			let index = cache.get_for_search(&root).expect("rebuilt index");
+			let index = index.read().await;
+			for (path, uuid) in uuids {
+				assert_eq!(
+					index.get_entry_uuid(&path),
+					Some(uuid),
+					"{} keeps the store's uuid",
+					path.display()
+				);
+			}
+			assert!(
+				index
+					.list_directory(&root)
+					.is_some_and(|children| !children.is_empty()),
+				"the source root has coverage, so the heal has nothing to walk"
+			);
+		}
+
+		/// The rebuild time behind the missing-snapshot path: a store of
+		/// `SD_REBUILD_RECORDS` files (one million by default) in directories
+		/// of a hundred, the way a walk lays them out, read back into an
+		/// empty arena. Run with
+		/// `cargo test -p sd-core --release --lib
+		///   a_million_record_store_rebuilds -- --ignored --nocapture`.
+		#[tokio::test]
+		#[ignore = "measurement, not regression; run with --ignored --nocapture"]
+		async fn a_million_record_store_rebuilds_in_seconds() {
+			use sd_store::file::{FileKind, FileWrite, Ledger, Observation};
+
+			let records: usize = std::env::var("SD_REBUILD_RECORDS")
+				.ok()
+				.and_then(|v| v.parse().ok())
+				.unwrap_or(1_000_000);
+			let dir = tempfile::tempdir().unwrap();
+			let manager = sd_store::SourceManager::new(dir.path().to_path_buf());
+			manager
+				.create("source-1", &sd_store::filesystem_schema())
+				.await
+				.expect("create");
+			let db = manager.open("source-1").await.expect("open");
+			db.begin_sync().await.expect("epoch");
+			let mut ledger = Ledger::load(db.pool()).await.expect("ledger");
+			let observe = |external_id: String, kind: FileKind| {
+				let name = external_id.rsplit('/').next().unwrap().to_string();
+				Observation {
+					extension: (kind == FileKind::File).then(|| "dat".to_string()),
+					external_id,
+					kind,
+					name,
+					size: 1_000,
+					mtime: 1_700_000_000_000,
+					created: None,
+					accessed: None,
+					inode: None,
+					mode: Some(0o644),
+					uid: None,
+					gid: None,
+					link_target: None,
+					is_hidden: false,
+					identity: None,
+				}
+			};
+			let built = Instant::now();
+			let mut written = 0usize;
+			let mut top = 0usize;
+			while written < records {
+				let mut writes = Vec::with_capacity(10_100);
+				let top_obs = observe(format!("dir-{top}"), FileKind::Directory);
+				let top_res = ledger.resolve(&top_obs);
+				let top_uuid = top_res.uuid();
+				writes.push(FileWrite {
+					resolution: top_res,
+					parent_uuid: None,
+					observation: top_obs,
+				});
+				for sub in 0..100 {
+					let sub_obs = observe(format!("dir-{top}/sub-{sub}"), FileKind::Directory);
+					let sub_res = ledger.resolve(&sub_obs);
+					let sub_uuid = sub_res.uuid();
+					writes.push(FileWrite {
+						resolution: sub_res,
+						parent_uuid: Some(top_uuid),
+						observation: sub_obs,
+					});
+					for file in 0..100 {
+						if written >= records {
+							break;
+						}
+						let obs = observe(
+							format!("dir-{top}/sub-{sub}/file-{written}.dat"),
+							FileKind::File,
+						);
+						let resolution = ledger.resolve(&obs);
+						writes.push(FileWrite {
+							resolution,
+							parent_uuid: Some(sub_uuid),
+							observation: obs,
+						});
+						written += 1;
+					}
+				}
+				db.apply_files(&writes, &[], &[], None)
+					.await
+					.expect("apply");
+				top += 1;
+			}
+			let total: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM record")
+				.fetch_one(db.pool())
+				.await
+				.expect("count");
+			println!(
+				"built a store of {total} records ({records} files) in {:?}",
+				built.elapsed()
+			);
+
+			let read_only = Instant::now();
+			let mut rows = 0usize;
+			let mut after = 0i64;
+			loop {
+				let (entries, last) =
+					sd_store::read::rebuild_entries_page(db.pool(), after, REBUILD_PAGE)
+						.await
+						.expect("page");
+				rows += entries.len();
+				after = last;
+				if entries.len() < REBUILD_PAGE {
+					break;
+				}
+			}
+			println!(
+				"read {rows} rows in {:?} (store pages alone)",
+				read_only.elapsed()
+			);
+
+			let index = TokioRwLock::new(Arena::new().expect("arena"));
+			let root = PathBuf::from("/volume/source");
+			let started = Instant::now();
+			let loaded = fill_arena_from_store(&index, &db, &root)
+				.await
+				.expect("rebuild");
+			let elapsed = started.elapsed();
+			let stats = index.read().await.get_stats();
+			println!(
+				"rebuilt {loaded} records ({} live entries) in {elapsed:?}",
+				stats.total_entries
+			);
+			assert_eq!(loaded as i64, total);
 		}
 
 		/// R8 "Missing or invalid restart snapshot", the diagnosis half: an
@@ -3201,7 +3677,11 @@ mod tests {
 				.attach_library(LIBRARY, library)
 				.await
 				.expect("attach");
-			assert!(!cache.ensure_restored(&root).await);
+			cache.ensure_restored(&root).await;
+			assert!(
+				!cache.restored_from_snapshot(&root),
+				"junk must not restore as an index"
+			);
 			assert!(
 				!snapshot_path.exists(),
 				"the slot is cleared so the next save lands clean"

core/src/ops/volumes/index/map.rs

@@ -236,17 +236,22 @@ pub async fn map_attached_volumes(
 
 		let cache = context.volume_index();
 		cache.track_detected_volume(volume.id, volume.mount_point.clone(), volume.is_mounted);
-		let restored = cache.ensure_restored(&volume.mount_point).await;
+		cache.ensure_restored(&volume.mount_point).await;
 
 		// Runs whether or not the volume restored: a restored snapshot can
 		// faithfully persist a coverage hole, which is exactly the state
-		// this repairs.
+		// this repairs. A drive whose sources were rebuilt from their stores
+		// is covered and skipped here the same way.
 		heal_uncovered_sources(library, context, &volume).await;
 
 		if !whole_drives {
 			continue;
 		}
-		if restored {
+		// Only a snapshot carries the whole drive. Sources rebuilt from their
+		// stores already list; the walk below maps the rest of the drive
+		// around them and keeps their uuids, as any walk over a live arena
+		// does.
+		if cache.restored_from_snapshot(&volume.mount_point) {
 			debug!(
 				"{} restored from its snapshot; not walking it again",
 				volume.mount_point.display()

core/src/service/mounts/peer.rs

@@ -1230,9 +1230,9 @@ async fn arena_from_database(
 	db: &sd_store::SourceDb,
 	share_root: &std::path::Path,
 ) -> anyhow::Result<Arena> {
-	let mut index = Arena::new()?;
-	crate::ops::indexing::volume_index::fill_arena_from_store(&mut index, db, share_root).await?;
-	Ok(index)
+	let index = TokioRwLock::new(Arena::new()?);
+	crate::ops::indexing::volume_index::fill_arena_from_store(&index, db, share_root).await?;
+	Ok(index.into_inner())
 }
 
 /// Fetch one source's database into the device's replica directory, rebuild

core/tests/snapshot_rebuild_acceptance_test.rs

@@ -0,0 +1,347 @@
+//! Missing or unusable restart snapshot (R8 "Missing or invalid restart
+//! snapshot", `docs/core/acceptance/source-runtime.md`): a source whose
+//! drive snapshot is gone, corrupt, or from an older format lists from a
+//! map rebuilt out of its store, with the store's uuids, before any walk
+//! runs. The unreadable artifact stays beside the slot for diagnosis.
+//!
+//! One `Core` over temporary directories, restarted between the fixture and
+//! the check. No loop device, no network.
+
+mod helpers;
+
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Duration;
+
+use helpers::TestConfigBuilder;
+use sd_core::{
+	domain::addressing::SdPath,
+	infra::{
+		action::LibraryAction,
+		api::SessionContext,
+		query::{CoreQuery, LibraryQuery},
+	},
+	library::Library,
+	ops::{
+		core::index_status::query::{IndexStatusInput, IndexStatusQuery},
+		files::query::directory_listing::{
+			DirectoryListingInput, DirectoryListingQuery, DirectorySortBy,
+		},
+		sources::track::action::{TrackSourceAction, TrackSourceInput},
+	},
+	Core,
+};
+use tempfile::TempDir;
+use uuid::Uuid;
+
+/// A daemon over `data_dir`: watcher on, so the restart restores every
+/// registered source the way a real launch does; networking off.
+async fn boot(data_dir: &Path) -> Arc<Core> {
+	let mut config = TestConfigBuilder::new(data_dir.to_path_buf())
+		.build()
+		.expect("config");
+	config.services.fs_watcher_enabled = true;
+	config.save().expect("save config");
+	Arc::new(Core::new(data_dir.to_path_buf()).await.expect("core"))
+}
+
+fn session(library: &Library) -> SessionContext {
+	let mut session = SessionContext::device_session(
+		sd_core::device::get_current_device_id(),
+		sd_core::device::get_current_device_slug(),
+	);
+	session.current_library_id = Some(library.id());
+	session
+}
+
+/// The sorted names a directory listing returns. `File::name` is the stem;
+/// the extension travels separately.
+async fn list_dir(core: &Arc<Core>, library: &Arc<Library>, path: PathBuf) -> Vec<String> {
+	let listing = DirectoryListingQuery::from_input(DirectoryListingInput {
+		path: SdPath::Physical {
+			device_slug: sd_core::device::get_current_device_slug(),
+			path,
+		},
+		limit: None,
+		include_hidden: Some(false),
+		sort_by: DirectorySortBy::Name,
+		folders_first: Some(false),
+		overlay: None,
+	})
+	.expect("input")
+	.execute(core.context.clone(), session(library))
+	.await
+	.expect("listing");
+	let mut names: Vec<String> = listing.files.iter().map(|file| file.name.clone()).collect();
+	names.sort();
+	names
+}
+
+async fn job_count(library: &Library) -> usize {
+	library.jobs().list_jobs(None).await.expect("jobs").len()
+}
+
+/// A walked and hashed source whose daemon shut down cleanly, so its
+/// snapshot and store are both on disk.
+struct Walked {
+	_root_dir: TempDir,
+	root: PathBuf,
+	data_dir: TempDir,
+	library_path: PathBuf,
+	snapshot_path: PathBuf,
+	/// Every file's path and the uuid its store assigned.
+	identities: Vec<(PathBuf, Uuid)>,
+}
+
+const FILES: [&str; 6] = [
+	"a.txt",
+	"b.txt",
+	"photos/c.jpg",
+	"photos/d.jpg",
+	"photos/trip/e.jpg",
+	"notes/f.md",
+];
+
+async fn walked_source(name: &str) -> Walked {
+	let root_dir = tempfile::Builder::new()
+		.prefix(name)
+		.tempdir()
+		.expect("root dir");
+	let root = root_dir.path().to_path_buf();
+	for file in FILES {
+		let path = root.join(file);
+		std::fs::create_dir_all(path.parent().unwrap()).expect("mkdir");
+		std::fs::write(&path, file).expect("write");
+	}
+
+	let data_dir = tempfile::tempdir().expect("data dir");
+	let core = boot(data_dir.path()).await;
+	let library = core
+		.libraries
+		.list()
+		.await
+		.into_iter()
+		.next()
+		.expect("default library");
+	let library_path = library.path().to_path_buf();
+
+	let output = TrackSourceAction::from_input(TrackSourceInput {
+		path: root.clone(),
+		name: None,
+		overrides: Default::default(),
+	})
+	.expect("input")
+	.execute(library.clone(), core.context.clone())
+	.await
+	.expect("track");
+	let root = output.root.clone();
+	let job_id = output.job_id.expect("tracking dispatched a walk");
+	if let Some(walk) = library
+		.jobs()
+		.get_job(sd_core::infra::job::prelude::JobId(job_id))
+		.await
+	{
+		walk.wait().await.expect("walk");
+	}
+
+	let cache = core.context.volume_index();
+	let store = cache.store_for(&root).await.expect("store");
+	// The hashing pass the walk queued is part of the fixture; wait on its
+	// effect so the restart is not about job resumption.
+	for _ in 0..300 {
+		if store
+			.files_needing_content_count()
+			.await
+			.expect("pending count")
+			== 0
+		{
+			break;
+		}
+		tokio::time::sleep(Duration::from_millis(100)).await;
+	}
+	store.flush().await.expect("flush");
+	let db = store.db();
+	let mut identities = Vec::new();
+	for file in FILES {
+		let uuid = db
+			.resolve_path(file)
+			.await
+			.expect("query")
+			.unwrap_or_else(|| panic!("{file} is in the store"));
+		identities.push((root.join(file), uuid));
+	}
+	let snapshot_path = cache.snapshot_path_for(&root).expect("snapshot path");
+
+	drop(library);
+	core.shutdown().await.expect("shutdown");
+	drop(core);
+	// Spawned service tasks let go of the key store after shutdown returns.
+	tokio::time::sleep(Duration::from_secs(2)).await;
+
+	assert!(
+		snapshot_path.exists(),
+		"the fixture left a snapshot at {}",
+		snapshot_path.display()
+	);
+
+	Walked {
+		_root_dir: root_dir,
+		root,
+		data_dir,
+		library_path,
+		snapshot_path,
+		identities,
+	}
+}
+
+/// Restart over the fixture and check the source lists from a map the
+/// store rebuilt: no walk dispatched, the listing served, the store's uuids
+/// kept, the status surface reporting the source restored with nothing in
+/// progress.
+async fn restart_and_check(walked: &Walked, retained_prefix: Option<&str>) {
+	let core = boot(walked.data_dir.path()).await;
+	let library = core
+		.libraries
+		.list()
+		.await
+		.into_iter()
+		.find(|library| library.path() == walked.library_path)
+		.expect("the library reloads");
+	let jobs_before = job_count(&library).await;
+	let cache = core.context.volume_index();
+
+	let list_dir = |path: PathBuf| list_dir(&core, &library, path);
+
+	// The first listings land while the rebuild may still be running: they
+	// are served from the store in that case, and from the arena once the
+	// rebuild has landed. Either way they name the files and start no walk,
+	// a subdirectory included, since a browse dispatched beneath the fill
+	// would clear what it had already placed.
+	assert_eq!(
+		list_dir(walked.root.join("photos")).await,
+		vec!["c", "d", "trip"],
+		"a subdirectory lists while the map comes back"
+	);
+	assert_eq!(
+		list_dir(walked.root.clone()).await,
+		vec!["a", "b", "notes", "photos"],
+		"the listing is served while the map comes back"
+	);
+	assert_eq!(
+		job_count(&library).await,
+		jobs_before,
+		"the first listings dispatched no walk"
+	);
+	for _ in 0..600 {
+		if !cache.is_indexing(&walked.root) && cache.arena_answers(&walked.root) {
+			break;
+		}
+		tokio::time::sleep(Duration::from_millis(50)).await;
+	}
+	assert_eq!(
+		list_dir(walked.root.clone()).await,
+		vec!["a", "b", "notes", "photos"],
+		"the listing is served from the rebuilt map"
+	);
+	assert_eq!(
+		list_dir(walked.root.join("photos/trip")).await,
+		vec!["e"],
+		"the deepest subtree survived the fill"
+	);
+	assert!(
+		cache.arena_answers(&walked.root),
+		"the arena answers for the source"
+	);
+	assert!(
+		!cache.restored_from_snapshot(&walked.root),
+		"nothing came from the snapshot file"
+	);
+	assert!(
+		!cache.is_indexing(&walked.root),
+		"no walk is in progress over the source"
+	);
+	assert_eq!(
+		job_count(&library).await,
+		jobs_before,
+		"the listing dispatched no walk"
+	);
+
+	let index = cache.get_for_search(&walked.root).expect("rebuilt index");
+	let index = index.read().await;
+	for (path, uuid) in &walked.identities {
+		assert_eq!(
+			index.get_entry_uuid(path),
+			Some(*uuid),
+			"{} keeps the uuid its store holds",
+			path.display()
+		);
+	}
+	drop(index);
+
+	let status = IndexStatusQuery::from_input(IndexStatusInput::default())
+		.expect("input")
+		.execute(core.context.clone(), session(&library))
+		.await
+		.expect("index status");
+	let source = status
+		.sources
+		.iter()
+		.find(|source| source.root == walked.root)
+		.expect("index status lists the source");
+	assert!(source.attached, "the source is attached");
+	assert!(
+		source.restored,
+		"core.index_status reports the map restored"
+	);
+	assert_eq!(
+		status.indexing_in_progress_count, 0,
+		"core.index_status reports no walk in progress: {:?}",
+		status.paths_in_progress
+	);
+
+	assert!(
+		!walked.snapshot_path.exists(),
+		"the slot is clear for the next save"
+	);
+	if let Some(prefix) = retained_prefix {
+		let retained = std::fs::read_dir(walked.snapshot_path.parent().unwrap())
+			.expect("snapshot dir")
+			.filter_map(|entry| entry.ok().map(|entry| entry.path()))
+			.filter(|path| {
+				path.file_name()
+					.map(|name| name.to_string_lossy().starts_with(prefix))
+					.unwrap_or(false)
+			})
+			.count();
+		assert_eq!(retained, 1, "the unusable artifact is kept beside the slot");
+	}
+
+	drop(library);
+	core.shutdown().await.expect("shutdown");
+}
+
+/// The snapshot is gone: deleted by hand, lost with the cache directory, or
+/// never written. The store refills the map.
+#[tokio::test]
+async fn a_missing_snapshot_rebuilds_the_map_from_the_store() {
+	let _ = tracing_subscriber::fmt::try_init();
+	let walked = walked_source("sd-snapshot-missing").await;
+	std::fs::remove_file(&walked.snapshot_path).expect("delete snapshot");
+	restart_and_check(&walked, None).await;
+}
+
+/// The snapshot will not parse. It is quarantined for diagnosis and the
+/// store refills the map.
+#[tokio::test]
+async fn a_corrupt_snapshot_is_quarantined_and_the_map_rebuilt_from_the_store() {
+	let _ = tracing_subscriber::fmt::try_init();
+	let walked = walked_source("sd-snapshot-corrupt").await;
+	std::fs::write(&walked.snapshot_path, b"this is not a snapshot").expect("corrupt");
+	let name = walked
+		.snapshot_path
+		.file_name()
+		.unwrap()
+		.to_string_lossy()
+		.into_owned();
+	restart_and_check(&walked, Some(&format!("{name}.corrupt-"))).await;
+}

crates/store/src/read.rs

@@ -751,6 +751,87 @@ pub async fn all_entries_page(
 	Ok((rows.into_iter().filter_map(entry_from_row).collect(), last))
 }
 
+/// What an arena rebuild needs from a row: identity, kind, path and the
+/// facet metadata, without the content columns a listing also carries.
+#[derive(Debug, Clone)]
+pub struct RebuildEntry {
+	pub uuid: Uuid,
+	pub kind: FileKind,
+	pub relative_path: String,
+	pub size: Option<i64>,
+	pub mtime_ms: Option<i64>,
+	pub atime_ms: Option<i64>,
+	pub created_ms: Option<i64>,
+	pub is_hidden: bool,
+	pub link_target: Option<String>,
+	pub inode: Option<i64>,
+	pub mode: Option<i64>,
+	pub uid: Option<i64>,
+	pub gid: Option<i64>,
+}
+
+#[derive(FromRow)]
+struct RebuildRow {
+	rowid: i64,
+	uuid: Uuid,
+	kind: String,
+	rel_path: String,
+	size: Option<i64>,
+	mtime_ms: Option<i64>,
+	atime_ms: Option<i64>,
+	created_ms: Option<i64>,
+	is_hidden: i64,
+	link_target: Option<String>,
+	inode: Option<i64>,
+	mode: Option<i64>,
+	uid: Option<i64>,
+	gid: Option<i64>,
+}
+
+/// A page of entries for rebuilding an arena, in rowid order after
+/// `after_rowid`, with the last rowid read.
+///
+/// The same rows [`all_entries_page`] returns, minus the content join and
+/// its columns: a rebuild derives content kinds from extensions the way a
+/// walk does, and the join plus six decoded columns per row were a third of
+/// the time a million-record rebuild spent reading.
+pub async fn rebuild_entries_page(
+	pool: &SqlitePool,
+	after_rowid: i64,
+	limit: usize,
+) -> Result<(Vec<RebuildEntry>, i64)> {
+	let rows: Vec<RebuildRow> = sqlx::query_as(
+		"SELECT r.rowid AS rowid, r.uuid AS uuid, r.type AS kind, 		 COALESCE(own.path, parent.path || '/' || r.title, COALESCE(r.title, '')) AS rel_path, 		 f.size AS size, f.mtime AS mtime_ms, f.atime AS atime_ms, r.created_at AS created_ms, 		 COALESCE(f.is_hidden, 0) AS is_hidden, f.link_target AS link_target, 		 f.inode AS inode, f.mode AS mode, f.uid AS uid, f.gid AS gid 		 FROM record r 		 LEFT JOIN directory_path own ON own.record_uuid = r.uuid 		 LEFT JOIN directory_path parent ON parent.record_uuid = r.parent_uuid 		 LEFT JOIN facet_file f ON f.record_uuid = r.uuid 		 WHERE r.rowid > ? ORDER BY r.rowid LIMIT ?",
+	)
+	.bind(after_rowid)
+	.bind(limit as i64)
+	.fetch_all(pool)
+	.await?;
+	let last = rows.last().map(|row| row.rowid).unwrap_or(after_rowid);
+	let entries = rows
+		.into_iter()
+		.filter_map(|row| {
+			let kind = FileKind::parse(&row.kind)?;
+			Some(RebuildEntry {
+				uuid: row.uuid,
+				kind,
+				relative_path: row.rel_path,
+				size: row.size,
+				mtime_ms: row.mtime_ms,
+				atime_ms: row.atime_ms,
+				created_ms: row.created_ms,
+				is_hidden: row.is_hidden != 0,
+				link_target: row.link_target,
+				inode: row.inode,
+				mode: row.mode,
+				uid: row.uid,
+				gid: row.gid,
+			})
+		})
+		.collect();
+	Ok((entries, last))
+}
+
 /// Case-folded substring search over record titles.
 ///
 /// Folding happens in Rust so the semantics match the arena's registry

docs/core/acceptance/source-runtime.md

@@ -50,7 +50,7 @@ cargo test -p sd-core --lib repeated_clear_and_refill_keeps_allocation_bounded -
 | 7 | Old schema with origin offline | `crates/store/tests/files.rs` `an_index_that_predates_parent_addressing_is_refused_intact` (the compatibility-status branch: the old table and its assertions stay intact; no data-preserving migration exists yet) | passing |
 | 8 | Same volume, nested roots, reversed registration order | `core/src/ops/indexing/sources.rs` `a_nested_source_never_redefines_its_volume_root`, `reversed_registration_order_keeps_one_volume_boundary`; `core/src/ops/indexing/volume_index.rs` `a_nested_source_shares_the_drive_it_sits_on`, `nested_sources_registered_inner_first_share_one_map_and_identity` | passing |
 | 9 | Remount and APFS alias changes | Remount: `core/src/ops/indexing/sources.rs` `a_remount_keeps_the_source`, `folders_on_one_drive_are_distinct_sources`, `a_different_drive_at_the_same_mount_point_is_a_different_source`; `core/src/ops/indexing/volume_index.rs` `a_root_mismatched_snapshot_moves_aside_instead_of_deleting`. APFS alias: none | passing (remount) / not automatable (APFS: firmlink spellings such as `/System/Volumes/Data/Users/x` only exist on macOS; `VolumeManager::locate_path` resolves them through the live volume list, which a Linux runner cannot produce) |
-| 10 | Missing or invalid restart snapshot | `core/src/ops/indexing/volume_index.rs` `an_invalid_snapshot_leaves_the_source_visible_and_its_store_readable`, `test_snapshot_roundtrip_and_detached_restore` (missing); `an_invalid_snapshot_is_retained_for_diagnosis` | passing / fixed (see F2) |
+| 10 | Missing or invalid restart snapshot | `core/tests/snapshot_rebuild_acceptance_test.rs` `a_missing_snapshot_rebuilds_the_map_from_the_store`, `a_corrupt_snapshot_is_quarantined_and_the_map_rebuilt_from_the_store` (real restart); `core/src/ops/indexing/volume_index.rs` `an_invalid_snapshot_leaves_the_source_visible_and_rebuilt_from_its_store`, `an_older_format_snapshot_is_quarantined_and_the_map_rebuilt_from_the_store`, `an_invalid_snapshot_is_retained_for_diagnosis`, `test_snapshot_roundtrip_and_detached_restore` | passing / fixed (see F2) |
 | 11 | Failed watcher subscription | `core/tests/source_runtime_acceptance_test.rs` `a_refused_watch_is_not_reported_active` (refusal reported with its reason on `core.index_status`, then a retry flips the root to active) | fixed (see F3) |
 | 12 | Offline client restart | `core/src/service/mounts/peer.rs` `a_cold_restore_rebuilds_the_inventory_without_the_owner`, `published_facts_outlive_the_owners_connection`, `a_manifest_without_facts_still_restores` | passing |
 | 13 | One failing source among nine | `core/src/service/mounts/peer.rs` `one_failing_source_among_nine_stays_listed_as_unavailable` | passing |
@@ -112,6 +112,28 @@ the aside path. One retained copy is enough: when a `.corrupt-*` sibling
 already exists, a later unreadable artifact in the slot is removed instead,
 so a recurring failure cannot fill the disk.
 
+Follow-up: a partition whose snapshot is missing, quarantined, or from an
+older format no longer costs a walk. `VolumeIndex::ensure_restored` starts
+a background fill of every attached source on the drive from its store, the
+same fill a library restore uses, and resolves at once so no reader waits
+on it; each root is in progress until its fill lands, a listing serves the
+store meanwhile, and the landed source announces its root, which routes it
+to the arena and re-arms its watch. The coverage heal finds the source in
+progress or covered and dispatches nothing. The rebuild
+runs for a source whose registry row carries a record count, which a
+snapshot save writes, so it is the evidence a map existed and was lost; a
+source never walked to completion keeps answering from its store without
+an arena (rows 4 and 5 of this matrix), and a source with no store records
+is left for the heal to walk. A store rebuild covers
+the registered sources and not the rest of the drive, so the discovery pass
+(`map_attached_volumes` with defaults) still maps the drive around them in
+the background, as it does for any drive without a snapshot. The real
+restart is `core/tests/snapshot_rebuild_acceptance_test.rs`: the listing is
+served from the store while the fill runs and from the arena once it has
+landed, no job is dispatched, and `core.index_status` then reports the
+source restored with nothing in progress. The 1M-record rebuild time is
+recorded in `docs/core/design/ephemeral-snapshot-format.md`.
+
 ### F3. Row 11: a refused watch was reported active (fixed)
 
 `core/tests/source_runtime_acceptance_test.rs` `a_refused_watch_is_not_reported_active`.

docs/core/design/ephemeral-snapshot-format.md

@@ -21,7 +21,7 @@ Observed density on real data: a ~1.06 MB payload compresses to ~214 KB (≈4.9:
 Postcard: varint (LEB128) integers, length-prefixed strings and collections, **no field names or tags — struct field order is the schema.** Any reordering, insertion, or type change to `IndexSnapshot` is a format break. That fragility is owned by policy rather than avoided:
 
 - The `version` field is decoded and checked first.
-- A **version mismatch or any decode failure deletes the file** and reports "no snapshot." Recovery is a clean reindex, never a retry against a dead artifact. Bump `SNAPSHOT_VERSION` for *every* struct change; there is no migration path by design — snapshots are caches, not data.
+- A **version mismatch or any decode failure quarantines the file** (`<name>.corrupt-<unix seconds>`, one copy kept) and reports "no snapshot." Recovery is a rebuild from the source stores, or a walk for a source with no store records, never a retry against a dead artifact. Bump `SNAPSHOT_VERSION` for *every* struct change; there is no migration path by design — snapshots are caches, not data.
 
 > **Amended 2026-08-18** (`docs/plans/2026-08-18-storage-consolidation.md`, artifact classes): "caches, not data" holds only while the source is attached. A detached drive's snapshot is an availability-bearing replica — the only local copy of a promised capability — so for a detached source, version mismatch and decode failure **quarantine** the file (`.snapshot.bad`) instead of deleting it; a later decoder recovers or a reattach reindexes. Delete-and-reindex applies only when the root is present. Two hardenings follow from the positional fragility: a golden-bytes test pinning the serialized layout of a fixture snapshot (an unbumped field reorder fails CI, not user data — postcard decodes a swap of two same-typed fields *successfully and wrongly*), and zstd frame checksums explicitly enabled so damage surfaces as a decode failure rather than silent garbage.
 
@@ -50,12 +50,22 @@ The writer renumbers through the arena's live projection (`Arena::snapshot_proje
 
 Restore attaches the snapshot to its source's partition (never a shared index), marks `root_path` as indexed, and sets the slot detached when the root is absent — which is what makes an unplugged drive browsable read-only.
 
+### When the snapshot cannot be used
+
+A partition whose snapshot is missing, unreadable, or stamped with another format version does not re-walk its sources. `VolumeIndex::ensure_restored` runs one restore attempt per drive per session; when the snapshot load yields nothing and the drive is attached, it starts one background task that refills the arena from every registered source's store (`fill_arena_from_store`, the same fill a library restore uses through `rebuild_from_store`). The gate resolves as soon as the task is started, because every reader on the drive funnels through it; each source root is marked in progress until its fill lands, so the coverage heal dispatches no walk over it and a directory listing serves the source's store meanwhile. The fill reads `rebuild_entries_page` (the record, path and facet columns, no content join) in pages of 2,000 rows and takes the arena's write lock one page at a time. When a source lands it keeps the uuids its store holds, is marked restored and indexed, and announces its root, which flips its routing to the arena and arms its filesystem watch. The unusable artifact stays beside the slot as `<name>.corrupt-<unix seconds>` (see Integrity) and the slot is clear for the next save, which writes the rebuilt arena in the current format. A partition a library restore already refilled (`rebuild_quiesced`) resolves as restored without reading its stores again.
+
+The rebuild runs for a source whose registry row carries a record count. The count is written when a snapshot is saved, so it is the evidence that a map existed and was lost; a source never walked to completion has no count and keeps answering from its store without an arena, which is the R6 routing (`docs/plans/2026-09-15-source-runtime-reliability.md`).
+
+What a store rebuild does not restore: the rest of the drive map outside the registered sources, and directory stubs. The discovery pass still walks a drive whose snapshot is gone when it maps whole drives (`VolumeIndex::restored_from_snapshot` is its test), and that walk keeps every uuid the rebuilt arena already holds. A source whose store has no records is the only one the coverage heal walks.
+
+Measured on a 4 vCPU cloud machine (release build, `a_million_record_store_rebuilds_in_seconds` in `volume_index.rs`, run with `--ignored`): a store of 1,010,100 records (1M files in 10,100 directories) rebuilds in 18.9 s, of which 13.9 s is reading the pages and the rest is arena inserts; 101,010 records take 1.7 s. Reading through the listing select (`all_entries_page`, with the content join and its six columns) took 28.2 s of a 33.1 s rebuild, which is why the rebuild has its own select.
+
 ## Known redundancy (future v3)
 
-`path_index` stores every full absolute path even though arena parent chains already encode the tree. This mirrors the same redundancy in RAM; the zero-onboarding memory-diet work (`zero-onboarding-startup.md` § change 3) removes the stored path maps in favor of parent-chain resolution, and the format shrinks with it — likely to arena entries + name pool + the small maps, with paths derived on load. That change is a version bump like any other: old files self-delete, sources reindex.
+`path_index` stores every full absolute path even though arena parent chains already encode the tree. This mirrors the same redundancy in RAM; the zero-onboarding memory-diet work (`zero-onboarding-startup.md` § change 3) removes the stored path maps in favor of parent-chain resolution, and the format shrinks with it — likely to arena entries + name pool + the small maps, with paths derived on load. That change is a version bump like any other: old files are quarantined and the arena is rebuilt from the stores.
 
 ## Integrity
 
-No checksums beyond zstd's frame checksum, no journaling. Cells of this cache are recomputable by definition; corruption manifests as a decode failure, which triggers the delete-and-reindex path. Anything needing stronger guarantees belongs in the portable artifact format, not here.
+No checksums beyond zstd's frame checksum, no journaling. Cells of this cache are recomputable by definition; corruption manifests as a decode failure, which quarantines the artifact and rebuilds the arena from the source stores (see Restore). Anything needing stronger guarantees belongs in the portable artifact format, not here.
 
 Note that rollup data (`FileNode.subtree_bytes`) is deliberately **not serialized**: directory sizes are recomputed from live entries during restore, so the format carries no derived totals that could drift from their inputs.

xtask/src/test_core.rs

@@ -185,6 +185,13 @@ pub const CORE_TESTS: &[TestSuite] = &[
 		"Locked volumes acceptance test",
 		&["--test", "locked_volumes_acceptance_test"],
 	),
+	// A source whose drive snapshot is missing or unusable at restart lists
+	// from a map rebuilt out of its store before any walk runs.
+	acceptance(
+		"sd-core",
+		"Snapshot rebuild acceptance test",
+		&["--test", "snapshot_rebuild_acceptance_test"],
+	),
 	// A daemon with two libraries lists every library's sources, before and
 	// after a restart, whichever order the data directory lists them in.
 	acceptance(