feat(sketch_db): retire schema/ module — sid-only lifecycle

data_plane/src/drivers/ingest/otel.rs

@@ -465,21 +465,19 @@ fn process_otlp_request(request: &ExportMetricsServiceRequest, transport: &str)
 /// config are dropped with a debug log — the precompute engine only
 /// maintains state for configured metrics.
 ///
-/// Flush a per-driver `HashMap<agg_id, count>` of §6.3 write-barrier
-/// drops into `IngestState::record_barrier_drop`, emitting a single
-/// debug log summarising the batch. Called from every OTLP routing
-/// function after its inner loop finishes, so a query against the
-/// `/metrics` endpoint sees a unified `samples_blocked_by_schema_barrier`
-/// counter regardless of which OTLP variant the DataCollector is
-/// shipping.
-fn flush_barrier_drops(state: &IngestState, drops: &HashMap<u64, u64>, driver_tag: &'static str) {
+/// Log a per-driver `HashMap<agg_id, count>` of §6.3 write-barrier
+/// drops. Post-schema-retirement the agg_id-keyed
+/// `IngestState::record_barrier_drop` counter is gone — the
+/// sid-level barrier inside `SketchStore::ingest_precompute_for_agg_config`
+/// silently rejects retired-sid writes without crossing this
+/// observer. The function is kept (callers still hand it an empty
+/// map) so the call shape doesn't churn; if the map is non-empty
+/// it emits a single debug log for forensic visibility.
+fn flush_barrier_drops(_state: &IngestState, drops: &HashMap<u64, u64>, driver_tag: &'static str) {
     if drops.is_empty() {
         return;
     }
     let total: u64 = drops.values().sum();
-    for (agg_id, count) in drops {
-        state.record_barrier_drop(*agg_id, *count);
-    }
     debug!(
         driver = driver_tag,
         total_dropped = total,
@@ -499,19 +497,16 @@ async fn route_otlp_to_precompute(
     // new aggregations are visible without restart.
     let snap = ingest_state.config_snapshot();
     let agg_configs = snap.get_all_aggregation_configs();
-    // Reconcile schema registry against the snapshot — Phase 2a of
-    // the sketch DB design (`docs/design-sketch-db.md` §6). Kept
-    // alongside the new sid-level reconcile below until the schema
-    // module is fully retired (schema retirement #5): both registries
-    // run in parallel so the §6.3 ingest barrier on
-    // `ingest_state.schemas.is_writable(agg_id)` below still sees
-    // accurate `Active/Retired/Expired` transitions while the sid
-    // catalog gets the same transitions in its own lifecycle fields.
-    let _ = ingest_state.schemas.reconcile(&snap);
+    // Schema retirement #5 — the agg_id-keyed `SchemaRegistry` is
+    // gone; sid-level lifecycle now lives on `SketchStore`. Reconcile
+    // against the current streaming config so newly-added /
+    // newly-retired sids transition immediately. The §6.3 ingest
+    // barrier is enforced at the sid level inside
+    // `SketchStore::ingest_precompute_for_agg_config`.
     let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_from_streaming_config(
         ingest_state.sketch_index.as_ref(),
         &snap,
-        ingest_state.schemas.retirement_retention(),
+        crate::storage_engines::sketch_db::DEFAULT_RETIREMENT_RETENTION,
     );
 
     // Build (agg_id, group_key) → Vec<(series_key, ts_ms, value)> for raw points.
@@ -686,15 +681,13 @@ async fn route_modified_otlp_sketches_to_precompute(
     let ingest_received_at = Instant::now();
     let snap = ingest_state.config_snapshot();
     let agg_configs = snap.get_all_aggregation_configs();
-    // Reconcile schema registry against the snapshot — Phase 2a of
-    // the sketch DB design (`docs/design-sketch-db.md` §6). Schema
-    // retirement #4 adds the sid-level reconcile alongside; see the
-    // raw-OTLP path for the rationale on running both until #5.
-    let _ = ingest_state.schemas.reconcile(&snap);
+    // Schema retirement #5 — agg_id-keyed registry retired; sid-level
+    // reconcile is the only path going forward. See the raw-OTLP
+    // routine above for the full rationale.
     let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_from_streaming_config(
         ingest_state.sketch_index.as_ref(),
         &snap,
-        ingest_state.schemas.retirement_retention(),
+        crate::storage_engines::sketch_db::DEFAULT_RETIREMENT_RETENTION,
     );
     let mut messages: Vec<WorkerMessage> = Vec::new();
     let mut routed = 0usize;
@@ -1914,7 +1907,6 @@ mod sid_resolution_tests {
     use crate::storage_engines::types::{HotReloadStreamingConfig, StreamingConfig};
     use crate::drivers::ingest::series_resolver::SeriesIdResolver;
     use crate::precompute_engine::series_router::SeriesRouter;
-    use crate::storage_engines::sketch_db::SchemaRegistry;
     use crate::storage_engines::sketch_db::index::SketchStore;
     use asap_otel_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest;
     use asap_otel_proto::tonic::common::v1::{any_value::Value as AnyVal, AnyValue, KeyValue};
@@ -1930,13 +1922,11 @@ mod sid_resolution_tests {
         let router = SeriesRouter::new(vec![tx]);
         let streaming = StreamingConfig::new(std::collections::HashMap::new());
         let hot_reload = HotReloadStreamingConfig::new(streaming.clone());
-        let schemas = Arc::new(SchemaRegistry::from_streaming_config(&streaming));
         let state = Arc::new(IngestState {
             router,
             samples_ingested: std::sync::atomic::AtomicU64::new(0),
             samples_blocked_by_schema_barrier: std::sync::atomic::AtomicU64::new(0),
             hot_reload_config: hot_reload,
-            schemas,
             pass_raw_samples: false,
             sketch_snapshots: dashmap::DashMap::new(),
             series_resolver: Arc::new(SeriesIdResolver::new()),