From 10a4224e08a2e24c40145b29fa2466aaea3c9be3 Mon Sep 17 00:00:00 2001 From: qdrant-cloud-bot <111755117+qdrant-cloud-bot@users.noreply.github.com> Date: Wed, 24 Jun 2026 08:59:16 +0200 Subject: [PATCH] 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 --- .../src/shards/local_shard/snapshot.rs | 18 +++- .../src/shards/local_shard/snapshot_tests.rs | 92 +++++++++++++++++++ lib/shard/src/segment_holder/mod.rs | 11 +++ lib/shard/src/snapshots/snapshot_manifest.rs | 12 ++- lib/shard/src/snapshots/snapshot_utils.rs | 15 +-- 5 files changed, 138 insertions(+), 10 deletions(-) diff --git a/lib/collection/src/shards/local_shard/snapshot.rs b/lib/collection/src/shards/local_shard/snapshot.rs index cc237197df..dbf8515c93 100644 --- a/lib/collection/src/shards/local_shard/snapshot.rs +++ b/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, diff --git a/lib/collection/src/shards/local_shard/snapshot_tests.rs b/lib/collection/src/shards/local_shard/snapshot_tests.rs index 0fc167c7fe..4e4781d6b4 100644 --- a/lib/collection/src/shards/local_shard/snapshot_tests.rs +++ b/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::>(); + + // The manifest is written to `/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> = + 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::>(); + assert_eq!( + manifest_uuids, expected_uuids, + "manifest must list exactly the snapshotted segments", + ); + for (_uuid, state) in manifest.iter() { + assert_eq!(*state, SegmentManifestState::Active); + } +} diff --git a/lib/shard/src/segment_holder/mod.rs b/lib/shard/src/segment_holder/mod.rs index f04a7792d5..72fa95d25b 100644 --- a/lib/shard/src/segment_holder/mod.rs +++ b/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 { + 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() } diff --git a/lib/shard/src/snapshots/snapshot_manifest.rs b/lib/shard/src/snapshots/snapshot_manifest.rs index e4b775a056..cb4335fb20 100644 --- a/lib/shard/src/snapshots/snapshot_manifest.rs +++ b/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!( diff --git a/lib/shard/src/snapshots/snapshot_utils.rs b/lib/shard/src/snapshots/snapshot_utils.rs index 37f5f4971b..8d01931658 100644 --- a/lib/shard/src/snapshots/snapshot_utils.rs +++ b/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::, _>>()?; - // 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 {