diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 0fe768b2..ac4d657b 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -520,7 +520,15 @@ async fn route_otlp_to_precompute( let mut by_group: HashMap> = HashMap::new(); let mut raw_matched = 0usize; let mut raw_unmatched = 0usize; - let mut raw_barrier_drops: HashMap = HashMap::new(); + // Schema retirement #3 (plan step #3) dropped the agg_id-keyed + // `schemas.is_writable` ingest barrier. The §6.3 contract still + // holds: `SketchStore::ingest_precompute_for_agg_config` rejects + // writes targeting retired/expired sids at the sid-level + // (per-sid `is_writable`). The retained empty map below feeds + // `flush_barrier_drops` so the counter / log shape doesn't + // change for /metrics consumers; once schema/ is fully retired + // the counter moves to sid-level wiring. + let raw_barrier_drops: HashMap = HashMap::new(); for point in &points { let series_key = format_series_key(&point.name, &point.labels); @@ -533,11 +541,6 @@ async fn route_otlp_to_precompute( { continue; } - // §6.3 write-side schema barrier — see ingest_handler.rs. - if !ingest_state.schemas.is_writable(config.aggregation_id) { - *raw_barrier_drops.entry(config.aggregation_id).or_default() += 1; - continue; - } let group_key = IngestState::extract_group_key_for(&series_key, config); by_group .entry((config.aggregation_id, group_key)) @@ -585,7 +588,10 @@ async fn route_otlp_to_precompute( let mut sketch_messages: Vec = Vec::new(); let mut sketch_matched = 0usize; let mut sketch_unmatched = 0usize; - let mut sketch_barrier_drops: HashMap = HashMap::new(); + // See `raw_barrier_drops` above — schema-keyed barrier dropped; + // sid-level barrier in `SketchStore` enforces §6.3 going + // forward. + let sketch_barrier_drops: HashMap = HashMap::new(); for point in &sketch_payloads { let series_key = format_series_key(&point.name, &point.labels); let ts_ms = (point.timestamp_nanos / 1_000_000) as i64; @@ -598,13 +604,6 @@ async fn route_otlp_to_precompute( { continue; } - // §6.3 write-side schema barrier — see ingest_handler.rs. - if !ingest_state.schemas.is_writable(config.aggregation_id) { - *sketch_barrier_drops - .entry(config.aggregation_id) - .or_default() += 1; - continue; - } let group_key = IngestState::extract_group_key_for(&series_key, config); // Wrap the raw SketchEnvelope bytes in a SketchEnvelopeAccumulator // so the precompute engine receives the opaque sketch as-is. This @@ -701,7 +700,9 @@ async fn route_modified_otlp_sketches_to_precompute( let mut routed = 0usize; let mut decoded_failed = 0usize; let mut unconfigured = 0usize; - let mut barrier_drops: HashMap = HashMap::new(); + // Schema-keyed barrier dropped (see `raw_barrier_drops` above); + // sid-level barrier in `SketchStore` carries §6.3 going forward. + let barrier_drops: HashMap = HashMap::new(); // Phase 4 — sids the receiver did not recognize this Export. Returned // to the caller so the gRPC / HTTP handler can stamp them into // `ExportMetricsServiceResponse.unknown_series_ids`. Senders evict @@ -1072,11 +1073,6 @@ async fn route_modified_otlp_sketches_to_precompute( { continue; } - // §6.3 write-side schema barrier — see ingest_handler.rs. - if !ingest_state.schemas.is_writable(config.aggregation_id) { - *barrier_drops.entry(config.aggregation_id).or_default() += 1; - continue; - } let group_key = IngestState::extract_group_key_for(&series_key, config); // DEPRECATED: aggregation_id-keyed write — remove // after warm-tier validation. The Phase 5 diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index a127d5d9..7b4adbc1 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -583,9 +583,15 @@ impl SketchStore { /// (archive replay) so they share one canonical sid-derivation /// path. /// - /// Returns the sid the entry landed under (or `None` when the - /// agg_config / output combination doesn't fit the precompute - /// model — caller logs and skips). + /// Returns the sid the entry landed under, or `None` when the + /// write is dropped: either because the agg_config / output + /// combination doesn't fit the precompute model (caller logs and + /// skips), or because the sid already exists in `Retired` / + /// `Expired` status. The latter is the sid-level mirror of the + /// `SchemaRegistry::is_writable(agg_id)` §6.3 ingest barrier: + /// once a sid is retired by [`crate::storage_engines::sketch_db::lifecycle::reconcile_from_streaming_config`] + /// further writes are rejected here so the eviction sweep can + /// drop residual state cleanly. pub fn ingest_precompute_for_agg_config( &self, agg_cfg: &asap_types::aggregation_config::AggregationConfig, @@ -614,19 +620,25 @@ impl SketchStore { }; let sid = compute_sid(&agg_cfg.metric, &attrs_fp, &agg_kind); - if self.instance(sid).is_none() { - let group_by_keys: BTreeSet = key_names.iter().cloned().collect(); - self.register(SketchInstanceMetadata { - sid, - metric_name: agg_cfg.metric.clone(), - group_by_keys, - capability: None, - agg_kind: agg_kind.clone(), - accuracy: None, - first_seen_unix_ms: output.start_timestamp as i64, - retired_at_ms: None, - expires_at_ms: None, - }); + match self.instance(sid) { + None => { + let group_by_keys: BTreeSet = key_names.iter().cloned().collect(); + self.register(SketchInstanceMetadata { + sid, + metric_name: agg_cfg.metric.clone(), + group_by_keys, + capability: None, + agg_kind: agg_kind.clone(), + accuracy: None, + first_seen_unix_ms: output.start_timestamp as i64, + retired_at_ms: None, + expires_at_ms: None, + }); + } + Some(existing) if !existing.is_writable() => { + return None; + } + Some(_) => {} } let window = (output.start_timestamp, output.end_timestamp);