From d6ab7a357f930b3beb5c731bf02206857e75de76 Mon Sep 17 00:00:00 2001 From: Anant Vindal Date: Tue, 4 Aug 2026 10:57:19 +0530 Subject: [PATCH] fix: manifest duplication issue occurred due to improper updating during writing manifests. Each standalone node now inserts in correct file without duplication --- src/catalog/mod.rs | 86 ++++---- src/handlers/http/modal/mod.rs | 93 ++++++++- src/handlers/http/modal/server.rs | 22 +++ src/migration/mod.rs | 22 ++- src/migration/stream_metadata_migration.rs | 217 +++++++++++++++++++++ src/storage/object_storage.rs | 12 ++ 6 files changed, 400 insertions(+), 52 deletions(-) diff --git a/src/catalog/mod.rs b/src/catalog/mod.rs index c01681c7a..819023c4f 100644 --- a/src/catalog/mod.rs +++ b/src/catalog/mod.rs @@ -262,8 +262,17 @@ async fn process_single_partition( storage_size: u64, tenant_id: &Option, ) -> Result, ObjectStorageError> { + // The entry we may update in place must be both time-overlapping *and* owned by this + // writer. Matching on the time range alone returns the first overlapping entry, which + // may belong to a different writer (e.g. after a restart changed the hostname). The + // ownership test then fails, a new entry is appended, and the next sync repeats the + // same lookup with the same result - appending one duplicate entry per sync cycle + // until the date rolls over. + let manifest_file_name = manifest_path("").to_string(); let pos = meta.snapshot.manifest_list.iter().position(|item| { - item.time_lower_bound <= partition_lower && partition_lower < item.time_upper_bound + item.time_lower_bound <= partition_lower + && partition_lower < item.time_upper_bound + && item.manifest_path.contains(&manifest_file_name) }); if let Some(pos) = pos { @@ -311,65 +320,48 @@ async fn handle_existing_partition( ) -> Result, ObjectStorageError> { let manifests = &mut meta.snapshot.manifest_list; - let manifest_file_name = manifest_path("").to_string(); - let should_update = manifests[pos].manifest_path.contains(&manifest_file_name); - - if should_update { - if let Some(mut manifest) = PARSEABLE + // `pos` is only ever produced for an entry this writer owns, so the manifest object it + // points at is ours to update in place. + if let Some(mut manifest) = PARSEABLE + .metastore + .get_manifest( + stream_name, + manifests[pos].time_lower_bound, + manifests[pos].time_upper_bound, + Some(manifests[pos].manifest_path.clone()), + tenant_id, + ) + .await + .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))? + { + // Update existing manifest + for change in partition_changes { + manifest.apply_change(change); + } + PARSEABLE .metastore - .get_manifest( + .put_manifest( + &manifest, stream_name, manifests[pos].time_lower_bound, manifests[pos].time_upper_bound, - Some(manifests[pos].manifest_path.clone()), tenant_id, ) .await - .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))? - { - // Update existing manifest - for change in partition_changes { - manifest.apply_change(change); - } - PARSEABLE - .metastore - .put_manifest( - &manifest, - stream_name, - manifests[pos].time_lower_bound, - manifests[pos].time_upper_bound, - tenant_id, - ) - .await - .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?; - - manifests[pos].events_ingested = events_ingested; - manifests[pos].ingestion_size = ingestion_size; - manifests[pos].storage_size = storage_size; - Ok(None) - } else { - // Manifest not found, create new one - create_manifest( - partition_lower, - partition_changes, - stream_name, - false, - meta.clone(), - events_ingested, - ingestion_size, - storage_size, - tenant_id, - ) - .await - } + .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?; + + manifests[pos].events_ingested = events_ingested; + manifests[pos].ingestion_size = ingestion_size; + manifests[pos].storage_size = storage_size; + Ok(None) } else { - // Create new manifest for different partition + // Manifest not found, create new one create_manifest( partition_lower, partition_changes, stream_name, false, - ObjectStoreFormat::default(), + meta.clone(), events_ingested, ingestion_size, storage_size, diff --git a/src/handlers/http/modal/mod.rs b/src/handlers/http/modal/mod.rs index e15dd675b..2c886728a 100644 --- a/src/handlers/http/modal/mod.rs +++ b/src/handlers/http/modal/mod.rs @@ -436,12 +436,18 @@ impl NodeMetadata { continue; } - let bytes = std::fs::read(&path).expect("File should be present"); + let bytes = match std::fs::read(&path) { + Ok(bytes) => bytes, + Err(e) => { + error!("Couldn't read {}: {}", path.display(), e); + continue; + } + }; match Self::from_bytes(&bytes, options.flight_port) { Ok(meta) => return Some(meta), Err(e) => { error!("Failed to extract {} metadata: {}", node_type_str, e); - return None; + continue; } } } @@ -449,11 +455,23 @@ impl NodeMetadata { None } - /// Check if a file is a valid metadata file for the given node type + /// Check if a path is this node type's metadata file, i.e. a *file* named + /// `{node_type}.{id}.json`. + /// + /// The shape has to be matched exactly rather than by substring. The staging root holds one + /// directory per stream alongside the node metadata, so a substring test lets a stream name + /// that happens to contain the node type - `calls` and `install_logs` both contain `all` - + /// masquerade as metadata. fn is_valid_metadata_file(path: &Path, node_type_str: &str) -> bool { + if !path.is_file() { + return false; + } + path.file_name() .and_then(|s| s.to_str()) - .is_some_and(|s| s.contains(node_type_str)) + .and_then(|s| s.strip_prefix(node_type_str)) + .and_then(|rest| rest.strip_suffix(".json")) + .is_some_and(|id| id.starts_with('.') && id.len() > 1) } /// Update metadata fields if they differ from the current configuration @@ -631,6 +649,7 @@ pub type IngestorMetadata = NodeMetadata; pub type IndexerMetadata = NodeMetadata; pub type QuerierMetadata = NodeMetadata; pub type PrismMetadata = NodeMetadata; +pub type StandaloneMetadata = NodeMetadata; /// Initialize hot tier metadata files for streams that have hot tier configuration /// in their stream metadata but don't have local hot tier metadata files yet. @@ -697,6 +716,72 @@ mod test { use crate::handlers::http::modal::NodeType; use super::IngestorMetadata; + use super::NodeMetadata; + + #[rstest] + fn valid_metadata_file_matches_exact_shape() { + let dir = temp_dir::TempDir::new().unwrap(); + let path = dir.path().join("all.01K9ZQ.json"); + std::fs::write(&path, b"{}").unwrap(); + + assert!(NodeMetadata::is_valid_metadata_file(&path, "all")); + // The node type has to be the whole prefix, not just present somewhere. + assert!(!NodeMetadata::is_valid_metadata_file(&path, "ingestor")); + } + + #[rstest] + fn stream_directory_containing_node_type_is_not_metadata() { + let dir = temp_dir::TempDir::new().unwrap(); + + // Staging holds one directory per stream next to the node metadata. A stream whose name + // contains the node type must not be mistaken for it - reading a directory would fail. + for stream in ["calls", "install_logs", "all"] { + let path = dir.path().join(stream); + std::fs::create_dir(&path).unwrap(); + assert!(!NodeMetadata::is_valid_metadata_file(&path, "all")); + } + } + + #[rstest] + fn bare_node_type_file_without_id_is_rejected() { + let dir = temp_dir::TempDir::new().unwrap(); + for name in ["all.json", "all..json", "allsomething.json", "all"] { + let path = dir.path().join(name); + std::fs::write(&path, b"{}").unwrap(); + assert!( + !NodeMetadata::is_valid_metadata_file(&path, "all"), + "{name} should not be treated as metadata" + ); + } + } + + #[rstest] + fn unparseable_candidate_does_not_abort_the_scan() { + let dir = temp_dir::TempDir::new().unwrap(); + let options = crate::cli::Options::default(); + + // A junk candidate sorts before the real one on most filesystems; either way the scan must + // keep going rather than give up and mint a fresh identity. + std::fs::write(dir.path().join("all.aaaa.json"), b"not json").unwrap(); + let good = NodeMetadata::new( + "8000".to_string(), + "http://0.0.0.0:8000".to_string(), + "bucket".to_string(), + "admin", + "admin", + "stable-id".to_owned(), + "8002".to_string(), + NodeType::All, + ); + std::fs::write( + dir.path().join("all.zzzz.json"), + serde_json::to_vec(&good).unwrap(), + ) + .unwrap(); + + let found = NodeMetadata::load_from_staging(dir.path(), "all", &options); + assert_eq!(found.map(|m| m.node_id), Some("stable-id".to_string())); + } #[rstest] fn test_deserialize_resource() { diff --git a/src/handlers/http/modal/server.rs b/src/handlers/http/modal/server.rs index 3b50dfd53..4afdd0835 100644 --- a/src/handlers/http/modal/server.rs +++ b/src/handlers/http/modal/server.rs @@ -57,6 +57,8 @@ use actix_web_prometheus::PrometheusMetrics; use actix_web_static_files::ResourceFiles; use async_trait::async_trait; use bytes::Bytes; +use std::sync::Arc; +use tokio::sync::OnceCell; use tokio::sync::mpsc; use tokio::sync::oneshot; @@ -71,12 +73,22 @@ use crate::{ }; // use super::generate; +use super::NodeType; use super::ParseableServer; +use super::StandaloneMetadata; use super::generate; use super::load_on_init; pub struct Server; +/// Identity of this standalone node. +/// +/// Standalone owns a single shared `.stream.json`, so the only thing distinguishing its manifests +/// from those of a previous incarnation is the writer name embedded in the manifest path. Deriving +/// that from the hostname makes a restarted pod look like a different writer; resolving a persisted +/// node id here keeps the identity stable across restarts. See [`super::NodeMetadata::load_node_metadata`]. +pub static NODE_META: OnceCell> = OnceCell::const_new(); + #[async_trait] impl ParseableServer for Server { fn configure_routes(config: &mut web::ServiceConfig) { @@ -136,6 +148,16 @@ impl ParseableServer for Server { prometheus: &PrometheusMetrics, shutdown_rx: oneshot::Receiver<()>, ) -> anyhow::Result<()> { + // Resolve node identity before anything can write a manifest: `manifest_path` reads this + // to name the manifests this node owns. + NODE_META + .get_or_init(|| async { + StandaloneMetadata::load_node_metadata(NodeType::All, &None) + .await + .expect("Node Metadata should be set in standalone mode") + }) + .await; + migration::run_migration(&PARSEABLE).await?; // load on init diff --git a/src/migration/mod.rs b/src/migration/mod.rs index 989cdbbac..d42a47d34 100644 --- a/src/migration/mod.rs +++ b/src/migration/mod.rs @@ -341,6 +341,17 @@ pub async fn migrate_stream_metadata( schema: &Bytes, tenant_id: &Option, ) -> anyhow::Result { + // Repair, not a version step: snapshots written before the manifest ownership fix can + // hold duplicate entries regardless of their metadata version, so this runs on every + // load and is a no-op once there is nothing left to collapse. + let duplicates_removed = + stream_metadata_migration::dedup_manifest_list(&mut stream_metadata_value); + if duplicates_removed > 0 { + warn!( + "Removed {duplicates_removed} duplicate manifest entries from snapshot of stream {stream}" + ); + } + let version = stream_metadata_value .as_object() .and_then(|meta| meta.get("version")) @@ -432,7 +443,16 @@ pub async fn migrate_stream_metadata( .await?; } _ => { - // If the version is not recognized, we assume it's already in the latest format + // If the version is not recognized, we assume it's already in the latest format. + // The snapshot repair above still needs persisting when it changed anything. + if duplicates_removed > 0 { + let stream_json: ObjectStoreFormat = + serde_json::from_value(stream_metadata_value.clone())?; + PARSEABLE + .metastore + .put_stream_json(&stream_json, stream, tenant_id) + .await?; + } return Ok(stream_metadata_value); } } diff --git a/src/migration/stream_metadata_migration.rs b/src/migration/stream_metadata_migration.rs index 13be5bec8..58a3750d3 100644 --- a/src/migration/stream_metadata_migration.rs +++ b/src/migration/stream_metadata_migration.rs @@ -23,6 +23,7 @@ use crate::{ storage, }; use serde_json::{Value, json}; +use std::collections::HashMap; pub fn v1_v4(mut stream_metadata: Value) -> Value { let stream_metadata_map = stream_metadata.as_object_mut().unwrap(); @@ -295,6 +296,57 @@ fn v1_v2_snapshot_migration(mut snapshot: Value) -> Value { snapshot } +/// Collapses duplicate entries in `snapshot.manifest_list`, returning the number removed. +/// +/// https://github.com/parseablehq/parseable/issues/1739 +pub fn dedup_manifest_list(stream_metadata: &mut Value) -> usize { + let Some(manifest_list) = stream_metadata + .get_mut("snapshot") + .and_then(|snapshot| snapshot.get_mut("manifest_list")) + .and_then(|list| list.as_array_mut()) + else { + return 0; + }; + + let original_len = manifest_list.len(); + let mut kept: Vec = Vec::with_capacity(original_len); + // manifest_path -> index into `kept` + let mut seen: HashMap = HashMap::new(); + + for entry in manifest_list.drain(..) { + let Some(path) = entry + .get("manifest_path") + .and_then(|path| path.as_str()) + .map(str::to_owned) + else { + // Keep anything we cannot key; dropping it would lose data. + kept.push(entry); + continue; + }; + + match seen.get(&path) { + Some(&pos) => { + if entry_rank(&entry) > entry_rank(&kept[pos]) { + kept[pos] = entry; + } + } + None => { + seen.insert(path, kept.len()); + kept.push(entry); + } + } + } + + *manifest_list = kept; + original_len - manifest_list.len() +} + +/// Orders duplicate entries for the same manifest path by how complete their statistics are. +fn entry_rank(entry: &Value) -> (u64, u64) { + let field = |name: &str| entry.get(name).and_then(|v| v.as_u64()).unwrap_or(0); + (field("events_ingested"), field("storage_size")) +} + #[cfg(test)] mod tests { #[test] @@ -400,4 +452,169 @@ mod tests { let updated_stream_metadata = super::v6_v7(stream_metadata.clone()); assert_eq!(updated_stream_metadata, expected); } + + fn entry(path: &str, date: &str, events: u64, storage: u64) -> serde_json::Value { + serde_json::json!({ + "manifest_path": path, + "time_lower_bound": format!("{date}T00:00:00Z"), + "time_upper_bound": format!("{date}T23:59:59.999999999Z"), + "events_ingested": events, + "ingestion_size": events * 10, + "storage_size": storage + }) + } + + fn snapshot_with(entries: Vec) -> serde_json::Value { + serde_json::json!({ + "version": "v7", + "snapshot": { "version": "v2", "manifest_list": entries } + }) + } + + fn manifest_list(metadata: &serde_json::Value) -> &Vec { + metadata["snapshot"]["manifest_list"].as_array().unwrap() + } + + #[test] + fn dedup_collapses_repeated_paths_keeping_the_largest_stats() { + let path = "test/date=2025-03-10/pod-a.manifest.json"; + let mut metadata = snapshot_with(vec![ + entry(path, "2025-03-10", 10, 100), + entry(path, "2025-03-10", 25, 250), + entry(path, "2025-03-10", 17, 170), + ]); + + assert_eq!(super::dedup_manifest_list(&mut metadata), 2); + assert_eq!( + manifest_list(&metadata), + &vec![entry(path, "2025-03-10", 25, 250)] + ); + } + + #[test] + fn dedup_keeps_one_entry_per_writer_for_the_same_date() { + let pod_a = "test/date=2025-03-10/pod-a.manifest.json"; + let pod_b = "test/date=2025-03-10/pod-b.manifest.json"; + let mut metadata = snapshot_with(vec![ + entry(pod_a, "2025-03-10", 10, 100), + entry(pod_b, "2025-03-10", 5, 50), + entry(pod_b, "2025-03-10", 8, 80), + entry(pod_a, "2025-03-10", 12, 120), + ]); + + assert_eq!(super::dedup_manifest_list(&mut metadata), 2); + // First-appearance order is preserved, so pod-a stays ahead of pod-b. + assert_eq!( + manifest_list(&metadata), + &vec![ + entry(pod_a, "2025-03-10", 12, 120), + entry(pod_b, "2025-03-10", 8, 80), + ] + ); + } + + #[test] + fn dedup_is_a_noop_on_a_clean_snapshot() { + let mut metadata = snapshot_with(vec![ + entry( + "test/date=2025-03-10/pod-a.manifest.json", + "2025-03-10", + 3, + 30, + ), + entry( + "test/date=2025-03-11/pod-a.manifest.json", + "2025-03-11", + 4, + 40, + ), + ]); + let before = metadata.clone(); + + assert_eq!(super::dedup_manifest_list(&mut metadata), 0); + assert_eq!(metadata, before); + } + + #[test] + fn dedup_is_idempotent() { + let path = "test/date=2025-03-10/pod-a.manifest.json"; + let mut metadata = snapshot_with(vec![ + entry(path, "2025-03-10", 10, 100), + entry(path, "2025-03-10", 25, 250), + ]); + + assert_eq!(super::dedup_manifest_list(&mut metadata), 1); + let after_first_pass = metadata.clone(); + assert_eq!(super::dedup_manifest_list(&mut metadata), 0); + assert_eq!(metadata, after_first_pass); + } + + #[test] + fn dedup_never_collapses_across_a_writer_rename() { + // Upgrading a standalone node switches its manifests from being named after the hostname + // to being named after its persisted node id. Both objects exist and both hold real data, + // so entries for the same date must survive as separate entries. + let by_hostname = "test/date=2025-03-10/node-a.manifest.json"; + let by_node_id = "test/date=2025-03-10/01K9ZQ.manifest.json"; + let mut metadata = snapshot_with(vec![ + entry(by_hostname, "2025-03-10", 40, 400), + entry(by_node_id, "2025-03-10", 12, 120), + ]); + let before = metadata.clone(); + + assert_eq!(super::dedup_manifest_list(&mut metadata), 0); + assert_eq!(metadata, before); + } + + #[test] + fn dedup_tolerates_a_missing_snapshot() { + let mut metadata = serde_json::json!({ "version": "v7" }); + assert_eq!(super::dedup_manifest_list(&mut metadata), 0); + } + + #[test] + fn dedup_tolerates_v1_metadata_with_no_snapshot_key() { + // v1/v2 metadata predates the snapshot entirely - `v1_v4`/`v2_v4` insert an empty one. + // The repair runs before the version chain, so it has to cope with the key being absent. + let mut metadata = serde_json::json!({ + "version": "v1", + "stats": {"events": 3, "ingestion": 70, "storage": 1969} + }); + let before = metadata.clone(); + + assert_eq!(super::dedup_manifest_list(&mut metadata), 0); + assert_eq!(metadata, before); + } + + #[test] + fn dedup_handles_v1_format_snapshot_entries_without_stats() { + // A v3 document can still carry a v1-format snapshot, whose entries have no + // events_ingested/ingestion_size/storage_size. Ranking must not trip over the missing + // fields, and the surviving entry must still migrate through v3_v4 afterwards. + let path = "test/date=2025-03-10/node-a.manifest.json"; + let v1_entry = serde_json::json!({ + "manifest_path": path, + "time_lower_bound": "2025-03-10T00:00:00Z", + "time_upper_bound": "2025-03-10T23:59:59.999999999Z" + }); + let mut metadata = serde_json::json!({ + "version": "v3", + "stats": {"events": 3, "ingestion": 70, "storage": 1969}, + "snapshot": { + "version": "v1", + "manifest_list": [v1_entry.clone(), v1_entry.clone(), v1_entry] + } + }); + + assert_eq!(super::dedup_manifest_list(&mut metadata), 2); + assert_eq!(manifest_list(&metadata).len(), 1); + + // The v1 -> v2 snapshot migration still runs cleanly over the collapsed list. + let migrated = super::v3_v4(metadata); + let entries = migrated["snapshot"]["manifest_list"].as_array().unwrap(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0]["manifest_path"], path); + assert_eq!(entries[0]["events_ingested"], 0); + assert_eq!(migrated["snapshot"]["version"], "v2"); + } } diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 3fcce6798..212ddd01b 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -23,6 +23,7 @@ use crate::handlers::DatasetTag; use crate::handlers::http::fetch_schema; use crate::handlers::http::modal::ingest_server::INGESTOR_EXPECT; use crate::handlers::http::modal::ingest_server::INGESTOR_META; +use crate::handlers::http::modal::server::NODE_META; use crate::handlers::http::users::{FILTER_DIR, USERS_ROOT_DIR}; use crate::metrics::increment_parquets_stored_by_date; use crate::metrics::increment_parquets_stored_size_by_date; @@ -1541,6 +1542,17 @@ pub fn manifest_path(prefix: &str) -> RelativePathBuf { let manifest_file_name = format!("ingestor.{hostname}.{id}.{MANIFEST_FILE}"); RelativePathBuf::from_iter([prefix, &manifest_file_name]) + } else if PARSEABLE.options.mode == Mode::All + && let Some(meta) = NODE_META.get() + { + // Standalone shares one `.stream.json` across restarts, so naming manifests after the + // hostname makes a restarted node look like a foreign writer. The persisted node id is + // stable across restarts, which keeps this node writing to its own manifest. + // + // Falls through to the hostname below if identity has not been resolved yet, which is the + // pre-existing behaviour rather than a new failure mode. + let manifest_file_name = format!("{}.{MANIFEST_FILE}", meta.get_node_id()); + RelativePathBuf::from_iter([prefix, &manifest_file_name]) } else { let manifest_file_name = format!("{hostname}.{MANIFEST_FILE}"); RelativePathBuf::from_iter([prefix, &manifest_file_name])