feat(precompute): migrate precompute mint to resolver; delete compute_sid (PR-4)

data_plane/src/main.rs

@@ -459,6 +459,7 @@ async fn main() -> Result<()> {
         let output_sink = Arc::new(SketchStoreSink::new(
             sketch_index.clone(),
             hot_reload_config.clone(),
+            series_resolver.clone(),
         ));
         let engine = PrecomputeEngine::new(
             precompute_config,
@@ -745,8 +746,11 @@ async fn main() -> Result<()> {
             data_plane::storage_engines::sketch_db::BackfillServiceConfig::default(),
         )
         // M2.3.6e — replayed batches land in SketchStore (the only
-        // destination after the M2.3.6g store retirement).
-        .with_sketch_index(sketch_index.clone());
+        // destination after the M2.3.6g store retirement). Resolver
+        // is the same shared mint authority as live ingest, so
+        // backfilled precompute sids share the OTel namespace.
+        .with_sketch_index(sketch_index.clone())
+        .with_series_resolver(series_resolver.clone());
         info!(
             "Spawning BackfillService drain loop (reader factory: default — Prometheus sources wired, S3/OtherSketch fail fast)"
         );

data_plane/src/precompute_engine/output_sink.rs

@@ -1,3 +1,4 @@
+use crate::drivers::ingest::series_resolver::SeriesIdResolver;
 use crate::storage_engines::sketch_db::index::SketchStore;
 use crate::storage_engines::types::hot_reload_config::HotReloadStreamingConfig;
 use crate::storage_engines::types::{AggregateCore, PrecomputedOutput};
@@ -28,13 +29,25 @@ pub trait OutputSink: Send + Sync {
 pub struct SketchStoreSink {
     sketch_index: Arc<SketchStore>,
     hot_reload: HotReloadStreamingConfig,
+    /// Single shared resolver across the ingest + precompute paths. Under
+    /// the registry-allocated sid model (PR-1..3), this is the canonical
+    /// mint authority — precompute sids share the same `next_sid` counter
+    /// as OTel-sketch sids, so the two paths can never collide on identity
+    /// even when the same `(metric, attrs)` carries both a sketch and an
+    /// exact precompute.
+    series_resolver: Arc<SeriesIdResolver>,
 }
 
 impl SketchStoreSink {
-    pub fn new(sketch_index: Arc<SketchStore>, hot_reload: HotReloadStreamingConfig) -> Self {
+    pub fn new(
+        sketch_index: Arc<SketchStore>,
+        hot_reload: HotReloadStreamingConfig,
+        series_resolver: Arc<SeriesIdResolver>,
+    ) -> Self {
         Self {
             sketch_index,
             hot_reload,
+            series_resolver,
         }
     }
 
@@ -56,8 +69,14 @@ impl SketchStoreSink {
             );
             return false;
         };
+        let resolver = self.series_resolver.clone();
         self.sketch_index
-            .ingest_precompute_for_agg_config(agg_cfg, output, accumulator)
+            .ingest_precompute_for_agg_config(
+                |metric, fp, ak| resolver.resolve(metric, fp, ak),
+                agg_cfg,
+                output,
+                accumulator,
+            )
             .is_some()
     }
 }
@@ -195,7 +214,11 @@ mod tests {
         let hot_reload = HotReloadStreamingConfig::new(streaming.clone());
 
         let sketch_index = Arc::new(SketchStore::new());
-        let sink = SketchStoreSink::new(sketch_index.clone(), hot_reload);
+        let sink = SketchStoreSink::new(
+            sketch_index.clone(),
+            hot_reload,
+            Arc::new(SeriesIdResolver::new()),
+        );
 
         let key = KeyByLabelValues::new_with_labels(vec!["z0".to_string()]);
         let output = PrecomputedOutput::new(1000, 2000, Some(key), agg_id);
@@ -234,7 +257,11 @@ mod tests {
         let streaming = StreamingConfig::new(HashMap::new());
         let hot_reload = HotReloadStreamingConfig::new(streaming.clone());
         let sketch_index = Arc::new(SketchStore::new());
-        let sink = SketchStoreSink::new(sketch_index.clone(), hot_reload);
+        let sink = SketchStoreSink::new(
+            sketch_index.clone(),
+            hot_reload,
+            Arc::new(SeriesIdResolver::new()),
+        );
 
         let output = PrecomputedOutput::new(1000, 2000, None, 99);
         let acc: Box<dyn AggregateCore> = Box::new(SumAccumulator::with_sum(1.0));

data_plane/src/storage_engines/sketch_db/backfill/processor.rs

@@ -124,6 +124,12 @@ pub struct BackfillWindowProcessor {
     /// effects can skip attaching one (the processor becomes a
     /// registry-only logger in that case).
     sketch_index: Option<Arc<crate::storage_engines::sketch_db::index::SketchStore>>,
+    /// Shared sid mint authority. Same `SeriesIdResolver` the OTel
+    /// ingest path uses, so backfilled precompute sids land in the
+    /// same unified namespace as live precompute / sketch sids. Only
+    /// consulted when `sketch_index` is also set (no sketch_index ⇒
+    /// no precompute write ⇒ no sid mint).
+    series_resolver: Option<Arc<crate::drivers::ingest::series_resolver::SeriesIdResolver>>,
     /// Registry where we record which `(agg_id, window_range)`
     /// tuples this job wrote. Phase 5f's coverage tracker reads
     /// this list.
@@ -142,6 +148,7 @@ impl BackfillWindowProcessor {
         Self {
             config,
             sketch_index: None,
+            series_resolver: None,
             registry,
             job_id,
         }
@@ -157,6 +164,19 @@ impl BackfillWindowProcessor {
         self
     }
 
+    /// Attach the shared `SeriesIdResolver` so precompute writes mint
+    /// sids via the same registry the OTel ingest path uses. Required
+    /// alongside [`Self::with_sketch_index`] — without a resolver the
+    /// processor still runs but skips the precompute write (registry
+    /// provenance still recorded, with a warn-log per missing call).
+    pub fn with_series_resolver(
+        mut self,
+        series_resolver: Arc<crate::drivers::ingest::series_resolver::SeriesIdResolver>,
+    ) -> Self {
+        self.series_resolver = Some(series_resolver);
+        self
+    }
+
     /// Look up the `AggregationConfig` for `agg_id` in the current
     /// `StreamingConfig` snapshot. Returns an error string if the
     /// agg has been removed from the config since the job was
@@ -236,10 +256,32 @@ impl WindowProcessor for BackfillWindowProcessor {
         // No legacy SketchStore write path remains. When no
         // sketch_index is attached (tests), the writes are simply
         // dropped — the registry still records the (agg_id, range)
-        // provenance below.
+        // provenance below. When sketch_index is attached but no
+        // resolver was provided, the precompute write is skipped
+        // with a warn — sid minting requires the shared resolver.
         if let Some(idx) = self.sketch_index.as_ref() {
-            for (output, accumulator) in &batch {
-                idx.ingest_precompute_for_agg_config(&config, output, accumulator.as_ref());
+            match self.series_resolver.as_ref() {
+                Some(resolver) => {
+                    for (output, accumulator) in &batch {
+                        let resolver = resolver.clone();
+                        idx.ingest_precompute_for_agg_config(
+                            |metric, fp, ak| resolver.resolve(metric, fp, ak),
+                            &config,
+                            output,
+                            accumulator.as_ref(),
+                        );
+                    }
+                }
+                None => {
+                    tracing::warn!(
+                        job_id = self.job_id,
+                        agg_id,
+                        batch_size = batch.len(),
+                        "BackfillWindowProcessor has sketch_index but no \
+                         series_resolver attached; skipping precompute writes \
+                         for this window",
+                    );
+                }
             }
         }
         // Hold `batch` alive until after the registry record below,

data_plane/src/storage_engines/sketch_db/backfill/service.rs

@@ -96,6 +96,9 @@ pub struct BackfillService {
     /// Phase 5 M2.3.6g — replayed batches land in `SketchStore` only;
     /// the legacy `Arc<dyn Store>` field is gone.
     sketch_index: Option<Arc<crate::storage_engines::sketch_db::index::SketchStore>>,
+    /// Shared sid mint authority — wired alongside `sketch_index` so
+    /// backfilled precompute sids share the namespace with live ingest.
+    series_resolver: Option<Arc<crate::drivers::ingest::series_resolver::SeriesIdResolver>>,
     config_source: HotReloadStreamingConfig,
     reader_factory: ReaderFactory,
     service_config: BackfillServiceConfig,
@@ -111,6 +114,7 @@ impl BackfillService {
         Self {
             registry,
             sketch_index: None,
+            series_resolver: None,
             config_source,
             reader_factory,
             service_config,
@@ -127,6 +131,19 @@ impl BackfillService {
         self
     }
 
+    /// Attach the shared `SeriesIdResolver` so the per-job
+    /// `BackfillWindowProcessor` mints sids via the same registry the
+    /// OTel ingest path uses. Builder-style; should be paired with
+    /// [`Self::with_sketch_index`] in production (without it,
+    /// processor writes are skipped with a warn).
+    pub fn with_series_resolver(
+        mut self,
+        series_resolver: Arc<crate::drivers::ingest::series_resolver::SeriesIdResolver>,
+    ) -> Self {
+        self.series_resolver = Some(series_resolver);
+        self
+    }
+
     /// Spawn the service as a tokio task. Returns a `BackfillServiceHandle`
     /// with a `shutdown` oneshot so `main.rs` can stop it cleanly on
     /// ctrl-c.
@@ -199,6 +216,9 @@ impl BackfillService {
             if let Some(idx) = self.sketch_index.as_ref() {
                 processor = processor.with_sketch_index(idx.clone());
             }
+            if let Some(resolver) = self.series_resolver.as_ref() {
+                processor = processor.with_series_resolver(resolver.clone());
+            }
             let worker = BackfillWorker::new(self.registry.clone());
 
             let filter = LabelFilter::for_metric(

data_plane/src/storage_engines/sketch_db/data/mod.rs

@@ -26,10 +26,10 @@
 //!   the read path (sid + label values + per-window samples).
 //! - [`AccuracyBound`] — `(epsilon, confidence)` derived from a
 //!   `SketchConfig`. Surfaces in HTTP response headers.
-//! - [`compute_sid`] — the canonical sid hash used today only by the
-//!   precompute ingest path (`SketchStore::ingest_precompute_for_agg_config`).
-//!   The OTel sketch ingest path moved to `SeriesIdResolver` (Option B
-//!   — registry-allocated sids); the precompute path follows in PR-4.
+//! - sid minting is no longer in this module — `SeriesIdResolver`
+//!   (in `drivers::ingest::series_resolver`) is the single mint
+//!   authority for every sid in the system. `AggKind::canonical_string()`
+//!   here produces the third element of the resolver's cache key.
 //! - [`canonical_parameters`] — helper that renders a parameters
 //!   `HashMap` into the canonical string form
 //!   `AggKind::Precompute::parameters_canonical` expects.
@@ -41,7 +41,9 @@
 //! - [`AggregationType`] — the agg-type enum that
 //!   `AggKind::Precompute` carries.
 
-use xxhash_rust::xxh64::xxh64;
+// `xxhash_rust::xxh64` import retired alongside `compute_sid` (PR-4).
+// Sid minting is now registry-allocated via `SeriesIdResolver` — no
+// content-addressed hash is computed at this layer.
 
 // ── Capability re-exports ────────────────────────────────────────────────────
 //
@@ -196,102 +198,13 @@ fn sketch_config_canonical(cfg: &SketchConfig) -> String {
 
 // ── sid hash ────────────────────────────────────────────────────────────────
 //
-// `compute_sketch_sid` was retired alongside the OTel ingest path's
-// migration to `SeriesIdResolver` (Option B — registry-allocated sids).
-// The remaining `compute_sid` function is still called by
-// `SketchStore::ingest_precompute_for_agg_config` for PRECOMPUTE
-// aggregations; that path migrates to the resolver in PR-4, at which
-// point `compute_sid` and its `sketch_kind_tag` / `encode_sketch_config`
-// helpers will go away too. Sketch sids today come exclusively from the
-// resolver — same authority as precompute sids will after PR-4.
-
-/// Generalized content-addressed sid hash. Same `(metric, attrs, agg_kind)`
-/// tuple always yields the same sid.
-///
-/// The two branches encode disjointly: a `Sketch` payload starts with
-/// `sketch_kind_tag` (1..=7), while a `Precompute` payload starts with
-/// the byte `b'P'` (ASCII 80), which no sketch tag will ever produce.
-/// So an attacker (or a colliding hash input) can't force a sketch sid
-/// to overlap a precompute sid at the encoding level.
-///
-/// Critically, the `Sketch` branch is bit-identical to the historical
-/// `compute_sketch_sid` byte layout — existing sketch sids the M2
-/// wire format introduced stay stable across this generalization.
-pub fn compute_sid(
-    metric_name: &str,
-    attrs_fingerprint: &str,
-    agg_kind: &AggKind,
-) -> u64 {
-    let mut buf: Vec<u8> =
-        Vec::with_capacity(metric_name.len() + attrs_fingerprint.len() + 32);
-    buf.extend_from_slice(metric_name.as_bytes());
-    buf.push(0);
-    buf.extend_from_slice(attrs_fingerprint.as_bytes());
-    buf.push(0);
-    match agg_kind {
-        AggKind::Sketch { kind, config } => {
-            buf.push(sketch_kind_tag(*kind));
-            buf.push(0);
-            encode_sketch_config(config, &mut buf);
-        }
-        AggKind::Precompute {
-            agg_type,
-            parameters_canonical,
-        } => {
-            buf.push(b'P');
-            buf.push(0);
-            buf.extend_from_slice(agg_type.as_str().as_bytes());
-            buf.push(0);
-            buf.extend_from_slice(parameters_canonical.as_bytes());
-        }
-    }
-    let h = xxh64(&buf, 0);
-    if h == 0 {
-        1
-    } else {
-        h
-    }
-}
-
-fn sketch_kind_tag(k: SketchKindHandle) -> u8 {
-    match k {
-        SketchKindHandle::DDSketch => 1,
-        SketchKindHandle::Kll => 2,
-        SketchKindHandle::Hll => 3,
-        SketchKindHandle::CountSketch => 4,
-        SketchKindHandle::CountMin => 5,
-        SketchKindHandle::CmsWithHeap => 6,
-        SketchKindHandle::CountSketchWithHeap => 7,
-        SketchKindHandle::Any => 0,
-    }
-}
-
-fn encode_sketch_config(cfg: &SketchConfig, buf: &mut Vec<u8>) {
-    match cfg {
-        SketchConfig::DDSketch { relative_accuracy } => {
-            buf.push(b'D');
-            buf.extend_from_slice(&relative_accuracy.to_le_bytes());
-        }
-        SketchConfig::Kll { k } => {
-            buf.push(b'K');
-            buf.extend_from_slice(&k.to_le_bytes());
-        }
-        SketchConfig::Hll { precision } => {
-            buf.push(b'H');
-            buf.extend_from_slice(&precision.to_le_bytes());
-        }
-        SketchConfig::CountSketch { rows, cols } => {
-            buf.push(b'S');
-            buf.extend_from_slice(&rows.to_le_bytes());
-            buf.extend_from_slice(&cols.to_le_bytes());
-        }
-        SketchConfig::CountMin { rows, cols } => {
-            buf.push(b'M');
-            buf.extend_from_slice(&rows.to_le_bytes());
-            buf.extend_from_slice(&cols.to_le_bytes());
-        }
-    }
-}
+// `compute_sid` was retired alongside `compute_sketch_sid` (PR-3) and
+// the precompute ingest migration (PR-4). All sid minting now flows
+// through `SeriesIdResolver` — one registry-allocated u64 per
+// `(metric, attrs_fingerprint, agg_kind_canonical)` triple, shared
+// across the OTel sketch ingest path and the precompute output path.
+// See `AggKind::canonical_string` for the agg-kind canonicalization
+// that replaces the byte-layout this hash used to produce.
 
 // ── Accuracy ────────────────────────────────────────────────────────────────