fix: manifest duplication

src/catalog/mod.rs

@@ -262,8 +262,17 @@ async fn process_single_partition(
     storage_size: u64,
     tenant_id: &Option<String>,
 ) -> Result<Option<snapshot::ManifestItem>, 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<Option<snapshot::ManifestItem>, 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,

src/handlers/http/modal/mod.rs

@@ -436,24 +436,42 @@ 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;
                 }
             }
         }
 
         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() {

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<Arc<StandaloneMetadata>> = 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

src/migration/mod.rs

@@ -341,6 +341,17 @@ pub async fn migrate_stream_metadata(
     schema: &Bytes,
     tenant_id: &Option<String>,
 ) -> anyhow::Result<Value> {
+    // 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);
         }
     }