Rotate retired physical series on catalog reactivation

data_plane/src/drivers/ingest/otel.rs

@@ -604,7 +604,8 @@ fn resolve_bucket_sid_for_agg_config(
     ingest_state: &Arc<IngestState>,
     config: &asap_types::aggregation_config::AggregationConfig,
     point_labels: &HashMap<String, String>,
-) -> (u64, asap_types::PolicyFingerprint) {
+    captured_generation: Option<&asap_types::sds::CatalogGeneration>,
+) -> Result<(u64, asap_types::PolicyFingerprint), String> {
     let grouping_pairs: Vec<(&str, &str)> = config
         .grouping_labels
         .labels
@@ -617,11 +618,28 @@ fn resolve_bucket_sid_for_agg_config(
     let fp = crate::drivers::ingest::canonical_attrs_fingerprint(&grouping_pairs);
     let agg_kind_canonical =
         crate::storage_engines::sketch_db::data::materialization_kind_for_config(config);
-    let sid = ingest_state
-        .series_resolver
-        .resolve(&config.metric, &fp, &agg_kind_canonical);
+    let sid = ingest_state.series_resolver.resolve_with_reactivation(
+        &config.metric,
+        &fp,
+        &agg_kind_canonical,
+        |sid| {
+            ingest_state
+                .sketch_index
+                .validate_routed_catalog_generation(captured_generation)?;
+            let activation = ingest_state
+                .sketch_index
+                .authorize_series_reactivation(sid, config.policy_fingerprint().into())?;
+            if activation
+                .as_deref()
+                .is_some_and(|generation| Some(generation) != captured_generation)
+            {
+                return Err("stale OTLP generation cannot reactivate series".into());
+            }
+            Ok(activation)
+        },
+    )?;
     let policy_fp = asap_types::PolicyFingerprint(config.policy_fp_u64());
-    (sid, policy_fp)
+    Ok((sid, policy_fp))
 }
 
 async fn route_otlp_to_precompute(
@@ -633,7 +651,15 @@ async fn route_otlp_to_precompute(
 
     // Snapshot the latest agg_configs from the hot-reload handle so
     // new aggregations are visible without restart.
-    let snap = ingest_state.config_snapshot();
+    let physical_plan_snapshot = ingest_state.physical_plan_snapshot();
+    let catalog_generation = physical_plan_snapshot
+        .as_ref()
+        .and_then(|plan| plan.precompute_plan.summary_catalog.clone())
+        .map(Arc::new);
+    let snap = physical_plan_snapshot
+        .as_ref()
+        .map(|plan| plan.runtime_config.clone())
+        .unwrap_or_else(|| ingest_state.config_snapshot());
     let agg_configs = snap.get_all_aggregation_configs();
     // Schema retirement #5 — the agg_id-keyed `SchemaRegistry` is
     // gone; sid-level lifecycle now lives on `SketchStore`. Reconcile
@@ -691,8 +717,18 @@ async fn route_otlp_to_precompute(
                 continue;
             }
             let group_key = IngestState::extract_group_key_for(&series_key, config);
-            let (sid, policy_fp) =
-                resolve_bucket_sid_for_agg_config(ingest_state, config, &point.labels);
+            let (sid, policy_fp) = match resolve_bucket_sid_for_agg_config(
+                ingest_state,
+                config,
+                &point.labels,
+                catalog_generation.as_deref(),
+            ) {
+                Ok(binding) => binding,
+                Err(error) => {
+                    warn!(%error, "configured ingest series reactivation rejected");
+                    continue;
+                }
+            };
             by_bucket
                 .entry(sid)
                 .or_insert_with(|| ((sid, policy_fp, group_key.clone()), Vec::new()))
@@ -724,7 +760,7 @@ async fn route_otlp_to_precompute(
     if !raw_messages.is_empty() {
         if let Err(e) = ingest_state
             .router
-            .route_group_batch(raw_messages, ingest_received_at)
+            .route_group_batch(raw_messages, ingest_received_at, catalog_generation.clone())
             .await
         {
             warn!("OTLP raw-sample routing error: {}", e);
@@ -782,8 +818,18 @@ async fn route_otlp_to_precompute(
             // `reconcile_from_streaming_config` derives from the same
             // config (otherwise the bucket would be reachable but never
             // reconciled).
-            let (sid, policy_fp) =
-                resolve_bucket_sid_for_agg_config(ingest_state, config, &point.labels);
+            let (sid, policy_fp) = match resolve_bucket_sid_for_agg_config(
+                ingest_state,
+                config,
+                &point.labels,
+                catalog_generation.as_deref(),
+            ) {
+                Ok(binding) => binding,
+                Err(error) => {
+                    warn!(%error, "configured ingest series reactivation rejected");
+                    continue;
+                }
+            };
             sketch_messages.push(WorkerMessage::AccumulatorInput {
                 sid,
                 policy_fp,
@@ -812,7 +858,11 @@ async fn route_otlp_to_precompute(
     if !sketch_messages.is_empty() {
         if let Err(e) = ingest_state
             .router
-            .route_group_batch(sketch_messages, ingest_received_at)
+            .route_group_batch(
+                sketch_messages,
+                ingest_received_at,
+                catalog_generation.clone(),
+            )
             .await
         {
             warn!("OTLP sketch routing error: {}", e);
@@ -875,6 +925,10 @@ async fn route_modified_otlp_sketches_to_precompute(
         .as_ref()
         .map(|plan| plan.runtime_config.clone())
         .unwrap_or_else(|| ingest_state.config_snapshot());
+    let catalog_generation = physical_plan_snapshot
+        .as_ref()
+        .and_then(|plan| plan.precompute_plan.summary_catalog.clone())
+        .map(Arc::new);
     let active_physical_plan = physical_plan_snapshot.filter(|plan| plan.plan_id() != 0);
     let lineage_batch_guard = active_physical_plan
         .as_ref()
@@ -1219,11 +1273,42 @@ async fn route_modified_otlp_sketches_to_precompute(
                             spatial_filter_canonical: String::new(),
                         };
                         let agg_kind_canonical = agg_kind.canonical_string();
-                        let assigned = ingest_state.series_resolver.resolve(
+                        let definition = frame_identity
+                            .as_ref()
+                            .map(|frame| frame.materialization)
+                            .unwrap_or_else(|| asap_types::PolicyFingerprint(0).into());
+                        let assigned = match ingest_state.series_resolver.resolve_with_reactivation(
                             &canonical_name,
                             &fp,
                             &agg_kind_canonical,
-                        );
+                            |sid| {
+                                ingest_state
+                                    .sketch_index
+                                    .validate_routed_catalog_generation(
+                                        catalog_generation.as_deref(),
+                                    )?;
+                                let activation = ingest_state
+                                    .sketch_index
+                                    .authorize_series_reactivation(sid, definition)?;
+                                if activation.as_deref().is_some_and(|generation| {
+                                    Some(generation) != catalog_generation.as_deref()
+                                }) {
+                                    return Err(
+                                        "stale OTLP generation cannot reactivate series".into()
+                                    );
+                                }
+                                Ok(activation)
+                            },
+                        ) {
+                            Ok(sid) => sid,
+                            Err(error) => {
+                                if dp.series_id != 0 {
+                                    unknown_sids.push(dp.series_id);
+                                }
+                                warn!(%error, "modified OTLP series reactivation rejected");
+                                continue;
+                            }
+                        };
                         if dp.series_id != 0 && dp.series_id != assigned {
                             // Sender's cached sid disagrees with the
                             // resolver's binding — sender's cache is
@@ -1751,8 +1836,18 @@ async fn route_modified_otlp_sketches_to_precompute(
                             // PERF-3 — `dp.attrs` is already a
                             // `HashMap<String, String>`; pass it directly
                             // instead of rebuilding `attrs_map` per config.
-                            let (bucket_sid, policy_fp) =
-                                resolve_bucket_sid_for_agg_config(ingest_state, config, &dp.attrs);
+                            let (bucket_sid, policy_fp) = match resolve_bucket_sid_for_agg_config(
+                                ingest_state,
+                                config,
+                                &dp.attrs,
+                                catalog_generation.as_deref(),
+                            ) {
+                                Ok(binding) => binding,
+                                Err(error) => {
+                                    warn!(%error, "configured ingest series reactivation rejected");
+                                    continue;
+                                }
+                            };
                             let acc_for_msg = if i + 1 == n {
                                 // Last (or only) match — move the owned
                                 // accumulator out, no clone.
@@ -1803,7 +1898,7 @@ async fn route_modified_otlp_sketches_to_precompute(
     if !messages.is_empty() {
         if let Err(e) = ingest_state
             .router
-            .route_group_batch(messages, ingest_received_at)
+            .route_group_batch(messages, ingest_received_at, catalog_generation.clone())
             .await
         {
             warn!("OTLP modified-proto sketch routing error: {}", e);

data_plane/src/drivers/ingest/prometheus_remote_write.rs

@@ -142,6 +142,8 @@ pub enum RemoteWriteError {
     DedupCapacity(usize),
     #[error("Remote Write requires an active PrometheusRemoteWriteV1 PhysicalPlan")]
     InactivePhysicalPlan,
+    #[error("series identity admission failed: {0}")]
+    SeriesIdentity(String),
     #[error(transparent)]
     Backpressure(#[from] TryRouteError),
 }
@@ -331,7 +333,7 @@ impl PrometheusRemoteWriteReceiver {
             return Err(RemoteWriteError::DedupCapacity(config.max_dedup_entries));
         }
 
-        let messages = route_messages(&new_samples, &self.inner.ingest, &physical_plan);
+        let messages = route_messages(&new_samples, &self.inner.ingest, &physical_plan)?;
         let generation = Arc::new(
             physical_plan
                 .precompute_plan
@@ -618,7 +620,7 @@ fn route_messages(
     samples: &[CanonicalSample],
     ingest: &Arc<IngestState>,
     physical_plan: &crate::storage_engines::types::ActivePhysicalPlan,
-) -> Vec<WorkerMessage> {
+) -> Result<Vec<WorkerMessage>, RemoteWriteError> {
     type Bucket = (
         u64,
         asap_types::PolicyFingerprint,
@@ -706,10 +708,25 @@ fn route_messages(
             // value-weighted Top-K). Keep those states on distinct SIDs.
             let materialization_kind =
                 crate::storage_engines::sketch_db::data::materialization_kind_for_config(config);
-            let sid =
-                ingest
-                    .series_resolver
-                    .resolve(&config.metric, attrs_fp, &materialization_kind);
+            let sid = ingest
+                .series_resolver
+                .resolve_with_reactivation(&config.metric, attrs_fp, &materialization_kind, |sid| {
+                    ingest.sketch_index.validate_routed_catalog_generation(
+                        physical_plan.precompute_plan.summary_catalog.as_ref(),
+                    )?;
+                    let activation = ingest
+                        .sketch_index
+                        .authorize_series_reactivation(sid, policy_fp.into())?;
+                    if let Some(generation) = &activation {
+                        if physical_plan.precompute_plan.summary_catalog.as_ref()
+                            != Some(generation.as_ref())
+                        {
+                            return Err("stale routed generation cannot reactivate series".into());
+                        }
+                    }
+                    Ok(activation)
+                })
+                .map_err(RemoteWriteError::SeriesIdentity)?;
             buckets
                 .entry(sid)
                 .or_insert_with(|| ((sid, policy_fp, group_key), Vec::new()))
@@ -718,7 +735,7 @@ fn route_messages(
         }
     }
     let received_at = Instant::now();
-    buckets
+    Ok(buckets
         .into_values()
         .map(
             |((sid, policy_fp, group_key), samples)| WorkerMessage::GroupSamples {
@@ -729,7 +746,7 @@ fn route_messages(
                 ingest_received_at: received_at,
             },
         )
-        .collect()
+        .collect())
 }
 
 #[derive(Debug)]
@@ -1082,6 +1099,10 @@ mod tests {
             sketch_index: Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()),
             observability: IngestObservability::default(),
         });
+        ingest
+            .sketch_index
+            .install_summary_catalog(physical_plan.summary_catalog.as_ref().unwrap().clone())
+            .unwrap();
         let request = WriteRequest {
             timeseries: ["api", "order", "payment", "user", "webapp"]
                 .into_iter()
@@ -1113,7 +1134,7 @@ mod tests {
         };
         let samples = canonicalize_request(&request, &PrometheusRemoteWriteConfig::default())
             .expect("canonical samples");
-        let messages = route_messages(&samples, &ingest, &physical_plan);
+        let messages = route_messages(&samples, &ingest, &physical_plan).unwrap();
         let mut cms_buckets = 0;
         let mut counter_buckets = 0;
         let mut cms_samples = 0;
@@ -1204,7 +1225,7 @@ mod tests {
         let drain = tokio::spawn(async move { handle.drain().await });
         assert!(matches!(
             worker.recv().await.unwrap(),
-            WorkerMessage::Admitted { .. }
+            WorkerMessage::BoundInput { .. }
         ));
         let WorkerMessage::Drain(reply) = worker.recv().await.unwrap() else {
             panic!("expected barrier")
@@ -1506,11 +1527,15 @@ mod tests {
         let (receiver, mut worker) = configured_receiver();
         receiver.accept(&one_sample(4.0)).unwrap();
         let message = worker.recv().await.expect("routed worker message");
-        let WorkerMessage::Admitted { input, revision } = message else {
+        let WorkerMessage::BoundInput {
+            input, revision, ..
+        } = message
+        else {
             panic!("missing admission receipt")
         };
-        assert!(revision.revision > 0);
+        assert!(revision.as_ref().unwrap().revision > 0);
         let message = *input;
+        let revision = revision.expect("remote-write input carries admission receipt");
         let WorkerMessage::GroupSamples {
             group_key, samples, ..
         } = message