feat: include segment manifest in shard snapshots (#9558)

Follow-up to #9530. When the `write_segment_manifest` feature flag is
enabled, a shard maintains a `segments/manifest.json` listing its
segments so out-of-process readers can discover them without scanning
the filesystem. That manifest was not included in shard snapshots.

Include the segment manifest in the snapshot when the shard maintains
one, capturing it from the live segment holder before proxying (proxies
preserve the wrapped segments' UUIDs, which are the directories written
into the snapshot, so the manifest matches the snapshot contents).

On restore, the snapshot's `segments/manifest.json` is skipped while
restoring segment directories in place; the manifest is regenerated from
the loaded segments when the segment holder is built. The partial
snapshot manifest loader also skips this file.

Co-authored-by: Cursor <cursoragent@cursor.com>

lib/collection/src/shards/local_shard/snapshot.rs

@@ -12,7 +12,7 @@ use segment::common::operation_error::{OperationError, OperationResult};
 use segment::data_types::manifest::SegmentManifest;
 use segment::entry::StorageSegmentEntry;
 use segment::types::{SegmentConfig, SnapshotFormat};
-use shard::files::{APPLIED_SEQ_FILE, SEGMENTS_PATH, WAL_PATH};
+use shard::files::{APPLIED_SEQ_FILE, SEGMENT_MANIFEST_FILE, SEGMENTS_PATH, WAL_PATH};
 use shard::locked_segment::LockedSegment;
 use shard::operations::OperationWithClockTag;
 use shard::payload_index_schema::PayloadIndexSchema;
@@ -266,6 +266,22 @@ pub fn snapshot_all_segments(
     // Snapshotting may take long-running read locks on segments blocking incoming writes, do
     // this through proxied segments to allow writes to continue.
 
+    // If the shard maintains a segment manifest (`segments/manifest.json`), include it in the
+    // snapshot so out-of-process readers can discover segments without scanning the filesystem.
+    // Captured before proxying: proxies preserve the wrapped segments' UUIDs, which are exactly the
+    // segment directories written into the snapshot, so the manifest matches the snapshot contents.
+    if let Some(segment_manifest) = segments.read().segment_manifest_for_snapshot() {
+        let segment_manifest_json = serde_json::to_vec(&segment_manifest).map_err(|err| {
+            OperationError::service_error(format!(
+                "failed to serialize segment manifest into JSON: {err}"
+            ))
+        })?;
+        tar.blocking_append_data(&segment_manifest_json, Path::new(SEGMENT_MANIFEST_FILE))
+            .map_err(|err| {
+                OperationError::service_error(format!("failed to archive segment manifest: {err}"))
+            })?;
+    }
+
     proxy_all_segments_and_apply(
         segments,
         segments_path,

lib/collection/src/shards/local_shard/snapshot_tests.rs

@@ -1,14 +1,19 @@
 use std::collections::HashSet;
+use std::io::Read as _;
 use std::sync::Arc;
 
+use common::flags::{FeatureFlags, init_feature_flags};
 use common::save_on_disk::SaveOnDisk;
 use common::tar_ext;
 use fs_err::File;
+use segment::entry::ReadSegmentEntry as _;
 use segment::types::SnapshotFormat;
+use shard::files::{SEGMENT_MANIFEST_FILE, SEGMENTS_PATH};
 use shard::fixtures::{build_segment_1, build_segment_2};
 use shard::payload_index_schema::PayloadIndexSchema;
 use shard::segment_holder::SegmentHolder;
 use shard::segment_holder::locked::LockedSegmentHolder;
+use shard::segment_manifest::{SegmentManifestState, SegmentsManifest};
 use tempfile::Builder;
 
 use crate::shards::local_shard::snapshot::snapshot_all_segments;
@@ -71,3 +76,90 @@ fn test_snapshot_all() {
     // one archive produced per concrete segment in the SegmentHolder
     assert_eq!(archive_count, 2);
 }
+
+/// With the `write_segment_manifest` flag enabled, a shard snapshot includes the segment manifest
+/// (`segments/manifest.json`) listing every snapshotted segment as `active`.
+#[test]
+#[allow(clippy::field_reassign_with_default)]
+fn test_snapshot_includes_segment_manifest() {
+    let mut flags = FeatureFlags::default();
+    flags.write_segment_manifest = true;
+    init_feature_flags(flags);
+
+    // Another test in this process may have initialized the feature flags first; the manifest is
+    // only written when the flag is actually enabled.
+    if !common::flags::feature_flags().write_segment_manifest {
+        return;
+    }
+
+    let dir = Builder::new().prefix("segment_dir").tempdir().unwrap();
+    let segment1 = build_segment_1(dir.path());
+    let segment2 = build_segment_2(dir.path());
+
+    let expected_uuids = [segment1.segment_uuid(), segment2.segment_uuid()]
+        .into_iter()
+        .collect::<HashSet<_>>();
+
+    // The manifest is written to `<shard>/segments/manifest.json`, so the directory must exist.
+    fs_err::create_dir_all(dir.path().join(SEGMENTS_PATH)).unwrap();
+
+    let mut holder = SegmentHolder::builder();
+    holder.add_new(segment1);
+    holder.add_new(segment2);
+    let holder = holder.build(dir.path()).unwrap();
+
+    let holder = LockedSegmentHolder::new(holder);
+
+    let segments_dir = Builder::new().prefix("segments_dir").tempdir().unwrap();
+    let temp_dir = Builder::new().prefix("temp_dir").tempdir().unwrap();
+    let snapshot_file = Builder::new().suffix(".snapshot.tar").tempfile().unwrap();
+    let tar = tar_ext::BuilderExt::new_seekable_owned(File::create(snapshot_file.path()).unwrap());
+
+    let payload_schema_file = dir.path().join("payload.schema");
+    let schema: Arc<SaveOnDisk<PayloadIndexSchema>> =
+        Arc::new(SaveOnDisk::load_or_init_default(payload_schema_file).unwrap());
+
+    snapshot_all_segments(
+        holder.clone(),
+        segments_dir.path(),
+        None,
+        schema,
+        None,
+        temp_dir.path(),
+        // Descend into `segments/`, mirroring how the local shard snapshot is produced.
+        &tar.descend(std::path::Path::new(SEGMENTS_PATH)).unwrap(),
+        SnapshotFormat::Regular,
+        None,
+    )
+    .unwrap();
+
+    let manifest_entry_path = format!("{SEGMENTS_PATH}/{SEGMENT_MANIFEST_FILE}");
+
+    let mut tar = tar::Archive::new(File::open(snapshot_file.path()).unwrap());
+    let mut manifest_bytes = None;
+    for entry in tar.entries_with_seek().unwrap() {
+        let mut entry = entry.unwrap();
+        let path = entry.path().unwrap().to_string_lossy().into_owned();
+        if path == manifest_entry_path {
+            let mut buf = Vec::new();
+            entry.read_to_end(&mut buf).unwrap();
+            manifest_bytes = Some(buf);
+        }
+    }
+
+    let manifest_bytes =
+        manifest_bytes.expect("snapshot must contain segments/manifest.json when flag is enabled");
+    let manifest: SegmentsManifest = serde_json::from_slice(&manifest_bytes).unwrap();
+
+    let manifest_uuids = manifest
+        .iter()
+        .map(|(uuid, _)| *uuid)
+        .collect::<HashSet<_>>();
+    assert_eq!(
+        manifest_uuids, expected_uuids,
+        "manifest must list exactly the snapshotted segments",
+    );
+    for (_uuid, state) in manifest.iter() {
+        assert_eq!(*state, SegmentManifestState::Active);
+    }
+}

lib/shard/src/segment_holder/mod.rs

@@ -243,6 +243,17 @@ impl SegmentHolder {
         SegmentsManifest::sync(self.segment_manifest.as_ref(), self, token.map(|t| t.id()))
     }
 
+    /// Build the segment manifest (`segments/manifest.json`) describing the current live segments,
+    /// for inclusion in a shard snapshot.
+    ///
+    /// Returns `None` when no manifest is attached (the `write_segment_manifest` feature flag is
+    /// off), so the snapshot omits the file exactly when the running shard would not have one.
+    pub fn segment_manifest_for_snapshot(&self) -> Option<SegmentsManifest> {
+        self.segment_manifest
+            .as_ref()
+            .map(|_| SegmentsManifest::from_segment_holder(self))
+    }
+
     pub fn len(&self) -> usize {
         self.appendable_segments.len() + self.non_appendable_segments.len()
     }

lib/shard/src/snapshots/snapshot_manifest.rs

@@ -8,7 +8,7 @@ use segment::data_types::manifest::SegmentManifest;
 use segment::segment::snapshot::SEGMENT_MANIFEST_FILE_NAME;
 use segment::types::SeqNumberType;
 
-use crate::files::segments_path;
+use crate::files::{SEGMENT_MANIFEST_FILE, segments_path};
 
 #[derive(Clone, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
 #[serde(transparent)]
@@ -31,7 +31,15 @@ impl SnapshotManifest {
         let mut snapshot_manifest = SnapshotManifest::default();
 
         for segment_entry in fs::read_dir(segments_path)? {
-            let segment_path = segment_entry?.path();
+            let segment_entry = segment_entry?;
+
+            // The segment manifest (`segments/manifest.json`) lives alongside the segment
+            // directories but is not a segment; skip it.
+            if segment_entry.file_name() == SEGMENT_MANIFEST_FILE {
+                continue;
+            }
+
+            let segment_path = segment_entry.path();
 
             if !segment_path.is_dir() {
                 log::warn!(

lib/shard/src/snapshots/snapshot_utils.rs

@@ -5,7 +5,7 @@ use fs_err as fs;
 use segment::common::operation_error::OperationResult;
 use segment::segment::Segment;
 
-use crate::files::{ShardDataFiles, get_shard_data_files, segments_path};
+use crate::files::{SEGMENT_MANIFEST_FILE, ShardDataFiles, get_shard_data_files, segments_path};
 use crate::snapshots::snapshot_manifest::SnapshotManifest;
 
 pub struct SnapshotUtils;
@@ -32,19 +32,20 @@ impl SnapshotUtils {
         // Read dir first as the directory contents would change during restore
         let entries = fs::read_dir(segments_path(snapshot_path))?.collect::<Result<Vec<_>, _>>()?;
 
-        // Filter out hidden entries
+        // Filter out hidden entries and the segment manifest (`segments/manifest.json`), which is
+        // not a segment directory. The manifest is regenerated from the loaded segments when the
+        // shard's segment holder is built, so it does not need to be restored in place here.
         let entries = entries.into_iter().filter(|entry| {
-            let is_hidden = entry
-                .file_name()
-                .to_str()
-                .is_some_and(|s| s.starts_with('.'));
+            let file_name = entry.file_name();
+            let is_hidden = file_name.to_str().is_some_and(|s| s.starts_with('.'));
             if is_hidden {
                 log::debug!(
                     "Ignoring hidden segment in local shard during snapshot recovery: {}",
                     entry.path().display(),
                 );
             }
-            !is_hidden
+            let is_segment_manifest = file_name == SEGMENT_MANIFEST_FILE;
+            !is_hidden && !is_segment_manifest
         });
 
         for entry in entries {