feat(sketch_db): sid-level §6.3 ingest barrier; drop schemas.is_writable

data_plane/src/drivers/ingest/otel.rs

@@ -520,7 +520,15 @@ async fn route_otlp_to_precompute(
     let mut by_group: HashMap<GroupKey, Vec<SampleTuple>> = HashMap::new();
     let mut raw_matched = 0usize;
     let mut raw_unmatched = 0usize;
-    let mut raw_barrier_drops: HashMap<u64, u64> = 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<u64, u64> = 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<WorkerMessage> = Vec::new();
     let mut sketch_matched = 0usize;
     let mut sketch_unmatched = 0usize;
-    let mut sketch_barrier_drops: HashMap<u64, u64> = 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<u64, u64> = 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<u64, u64> = 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<u64, u64> = 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

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<String> = 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<String> = 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);