feat(sid): migrate PrecomputedOutput + sinks to PolicyFingerprint

crates/asap_types/src/policy_fingerprint.rs

@@ -56,11 +56,25 @@ use crate::aggregation_config::AggregationConfig;
 /// with an `aggregation_id` — they're both u64-shaped but they index
 /// different things (content-addressed vs. controller-allocated).
 #[derive(
-    Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize,
+    Debug, Clone, Copy, Default, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize,
 )]
 #[serde(transparent)]
 pub struct PolicyFingerprint(pub u64);
 
+impl PolicyFingerprint {
+    /// Sentinel "unset / legacy" fingerprint produced by
+    /// `PolicyFingerprint::default()`. Callers that haven't yet been
+    /// migrated to compute the real fingerprint hand this through;
+    /// downstream consumers (sinks, registries) treat it as
+    /// "fall back to `aggregation_id` lookup". Removed in PR 5.
+    pub const UNSET: PolicyFingerprint = PolicyFingerprint(0);
+
+    /// True when this fingerprint is the [`Self::UNSET`] sentinel.
+    pub fn is_unset(self) -> bool {
+        self.0 == 0
+    }
+}
+
 impl PolicyFingerprint {
     /// Compute the fingerprint of an [`AggregationConfig`].
     ///

data_plane/src/precompute_engine/output_sink.rs

@@ -56,19 +56,53 @@ impl SketchStoreSink {
     /// inconsistencies — a SketchStore miss is recoverable in
     /// practice because the control plane will re-emit the agg_config
     /// on its next reconcile pass.
+    ///
+    /// PR 4 (sid identity chain): resolves the source `AggregationConfig`
+    /// via [`PolicyFingerprint`] first when the output carries a
+    /// non-sentinel `policy_fp`; falls back to the legacy
+    /// `aggregation_id` lookup when the field is the
+    /// `PolicyFingerprint::UNSET` sentinel (set by call sites still on
+    /// the legacy `PrecomputedOutput::new` constructor). Both paths
+    /// resolve to the same `AggregationConfig` while
+    /// `StreamingConfig::aggregation_configs` is the source of truth.
     fn append_to_index(
         &self,
         output: &PrecomputedOutput,
         accumulator: &dyn AggregateCore,
     ) -> bool {
         let cfg = self.hot_reload.snapshot();
-        let Some(agg_cfg) = cfg.get_aggregation_config(output.aggregation_id) else {
-            warn!(
-                agg_id = output.aggregation_id,
-                "SketchStoreSink: agg_config missing from streaming snapshot; skipping write"
-            );
-            return false;
-        };
+        let agg_cfg_owned;
+        let agg_cfg: &asap_types::aggregation_config::AggregationConfig =
+            if !output.policy_fp.is_unset() {
+                let registry = cfg.policy_registry();
+                match registry.get(output.policy_fp) {
+                    Some(c) => {
+                        // Clone out so the borrow on the snapshot
+                        // doesn't outlive this scope; the existing
+                        // legacy branch ALSO borrows from the snapshot,
+                        // so this is structurally equivalent.
+                        agg_cfg_owned = c.clone();
+                        &agg_cfg_owned
+                    }
+                    None => {
+                        warn!(
+                            policy_fp = %output.policy_fp,
+                            agg_id = output.aggregation_id,
+                            "SketchStoreSink: policy_fp missing from registry; skipping write"
+                        );
+                        return false;
+                    }
+                }
+            } else {
+                let Some(c) = cfg.get_aggregation_config(output.aggregation_id) else {
+                    warn!(
+                        agg_id = output.aggregation_id,
+                        "SketchStoreSink: agg_config missing from streaming snapshot; skipping write"
+                    );
+                    return false;
+                };
+                c
+            };
         let resolver = self.series_resolver.clone();
         self.sketch_index
             .ingest_precompute_for_agg_config(

data_plane/src/precompute_engine/worker.rs

@@ -8,6 +8,7 @@ use crate::precompute_engine::series_router::WorkerMessage;
 use crate::precompute_engine::window_manager::WindowManager;
 use crate::precompute_engine::operators::sum_accumulator::SumAccumulator;
 use asap_types::aggregation_config::AggregationConfig;
+use asap_types::PolicyFingerprint;
 use std::collections::{BTreeMap, HashMap};
 use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering};
 use std::sync::Arc;
@@ -368,11 +369,12 @@ impl Worker {
                         let mut updater = create_accumulator_updater(&state.config);
                         apply_sample(&mut *updater, series_key, *val, *ts, &state.config);
                         let key = build_group_key_label_values(group_key);
-                        let output = PrecomputedOutput::new(
+                        let output = PrecomputedOutput::new_with_policy_fp(
                             window_start as u64,
                             window_end as u64,
                             Some(key),
                             agg_id,
+                            PolicyFingerprint::from_config(&state.config),
                         );
                         emit_batch.push((output, updater.take_accumulator()));
                         debug!(
@@ -410,11 +412,12 @@ impl Worker {
             if let Some(accumulator) = merge_panes_for_window(&mut state.active_panes, &pane_starts)
             {
                 let key = build_group_key_label_values(group_key);
-                let output = PrecomputedOutput::new(
+                let output = PrecomputedOutput::new_with_policy_fp(
                     *window_start as u64,
                     window_end as u64,
                     Some(key),
                     agg_id,
+                    PolicyFingerprint::from_config(&state.config),
                 );
                 emit_batch.push((output, accumulator));
             }
@@ -501,11 +504,12 @@ impl Worker {
                     let window_start = pane_start;
                     let window_end = pane_start + state.window_manager.window_size_ms();
                     let key = build_group_key_label_values(group_key);
-                    let output = PrecomputedOutput::new(
+                    let output = PrecomputedOutput::new_with_policy_fp(
                         window_start as u64,
                         window_end as u64,
                         Some(key),
                         agg_id,
+                        PolicyFingerprint::from_config(&state.config),
                     );
                     emit_batch.push((output, incoming));
                     debug!(
@@ -552,11 +556,12 @@ impl Worker {
             if let Some(accumulator) = merge_panes_for_window(&mut state.active_panes, &pane_starts)
             {
                 let key = build_group_key_label_values(group_key);
-                let output = PrecomputedOutput::new(
+                let output = PrecomputedOutput::new_with_policy_fp(
                     *window_start as u64,
                     window_end as u64,
                     Some(key),
                     agg_id,
+                    PolicyFingerprint::from_config(&state.config),
                 );
                 emit_batch.push((output, accumulator));
             }
@@ -566,11 +571,12 @@ impl Worker {
                 merge_sketch_panes_for_window(&mut state.sketch_panes, &pane_starts)
             {
                 let key = build_group_key_label_values(group_key);
-                let output = PrecomputedOutput::new(
+                let output = PrecomputedOutput::new_with_policy_fp(
                     *window_start as u64,
                     window_end as u64,
                     Some(key),
                     agg_id,
+                    PolicyFingerprint::from_config(&state.config),
                 );
                 emit_batch.push((output, accumulator));
             }
@@ -603,6 +609,12 @@ impl Worker {
             Vec::with_capacity(samples.len());
 
         for (ts, val) in samples {
+            // Raw-mode path does not carry an `AggregationConfig` for
+            // the source aggregation (it's an aggregation-config-less
+            // pass-through with a synthetic agg_id), so we leave
+            // `policy_fp` as the `PolicyFingerprint::UNSET` sentinel.
+            // The sink falls back to `aggregation_id` lookup — the
+            // dual-keyed transition this PR is structured around.
             let output =
                 PrecomputedOutput::new(ts as u64, ts as u64, None, self.raw_mode_aggregation_id);
             let accumulator = SumAccumulator::with_sum(val);
@@ -736,11 +748,12 @@ impl Worker {
                     merge_panes_for_window(&mut state.active_panes, &pane_starts)
                 {
                     let key = build_group_key_label_values(group_key);
-                    let output = PrecomputedOutput::new(
+                    let output = PrecomputedOutput::new_with_policy_fp(
                         *window_start as u64,
                         window_end as u64,
                         Some(key),
                         *agg_id,
+                        PolicyFingerprint::from_config(&state.config),
                     );
                     emit_batch.push((output, accumulator));
                 }
@@ -749,11 +762,12 @@ impl Worker {
                     merge_sketch_panes_for_window(&mut state.sketch_panes, &pane_starts)
                 {
                     let key = build_group_key_label_values(group_key);
-                    let output = PrecomputedOutput::new(
+                    let output = PrecomputedOutput::new_with_policy_fp(
                         *window_start as u64,
                         window_end as u64,
                         Some(key),
                         *agg_id,
+                        PolicyFingerprint::from_config(&state.config),
                     );
                     emit_batch.push((output, accumulator));
                 }

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

@@ -72,6 +72,7 @@ use tracing::debug;
 use crate::storage_engines::types::{AggregateCore, HotReloadStreamingConfig, KeyByLabelValues};
 use crate::precompute_engine::worker::parse_labels_from_series_key;
 use asap_types::aggregation_config::AggregationConfig;
+use asap_types::PolicyFingerprint;
 
 use super::BackfillRegistry;
 use super::window_builder::build_backfilled_accumulator;
@@ -242,13 +243,15 @@ impl WindowProcessor for BackfillWindowProcessor {
             } else {
                 Some(build_group_key_label_values(&group_key))
             };
-            let output = crate::storage_engines::types::PrecomputedOutput::new_backfilled(
-                window_range.0,
-                window_range.1,
-                key,
-                agg_id,
-                self.job_id,
-            );
+            let output =
+                crate::storage_engines::types::PrecomputedOutput::new_backfilled_with_policy_fp(
+                    window_range.0,
+                    window_range.1,
+                    key,
+                    agg_id,
+                    self.job_id,
+                    PolicyFingerprint::from_config(&config),
+                );
             batch.push((output, accumulator));
         }
 

data_plane/src/storage_engines/types/precomputed_output.rs

@@ -1,3 +1,4 @@
+use asap_types::PolicyFingerprint;
 use serde::{Deserialize, Serialize};
 
 use crate::storage_engines::types::KeyByLabelValues;
@@ -40,11 +41,27 @@ pub struct PrecomputedOutput {
     /// forward-compat with older on-disk payloads.
     #[serde(default)]
     pub origin: Origin,
+    /// Content-addressed policy identity — the merged-sid-identity-chain
+    /// successor to [`Self::aggregation_id`]. `#[serde(default)]` on
+    /// read means records persisted before PR 4 (which adds this field)
+    /// deserialise as `PolicyFingerprint(0)`; consumers MUST tolerate
+    /// the sentinel and fall back to `aggregation_id` lookup until PR 5
+    /// retires the legacy field.
+    ///
+    /// Construction sites that have the source `AggregationConfig` in
+    /// hand should populate this via
+    /// [`PolicyFingerprint::from_config`]; sites that only have an
+    /// `aggregation_id` from legacy plumbing should use
+    /// [`Self::new`] (which leaves this as the sentinel).
+    #[serde(default)]
+    pub policy_fp: PolicyFingerprint,
 }
 
 impl PrecomputedOutput {
-    /// Construct a `Native` precompute — the default used by the
-    /// live ingest pipeline.
+    /// Construct a `Native` precompute with only an `aggregation_id` —
+    /// legacy path. `policy_fp` is left as the `PolicyFingerprint(0)`
+    /// sentinel; sinks fall back to `aggregation_id` lookup. New code
+    /// should prefer [`Self::new_with_policy_fp`].
     pub fn new(
         start_timestamp: u64,
         end_timestamp: u64,
@@ -57,13 +74,38 @@ impl PrecomputedOutput {
             key,
             aggregation_id,
             origin: Origin::Native,
+            policy_fp: PolicyFingerprint(0),
+        }
+    }
+
+    /// Construct a `Native` precompute carrying both the
+    /// `aggregation_id` (for transition compat) and the
+    /// `PolicyFingerprint`. Used by the precompute worker + OTLP sketch
+    /// ingest path now that they have the source `AggregationConfig`
+    /// in hand at emit time.
+    pub fn new_with_policy_fp(
+        start_timestamp: u64,
+        end_timestamp: u64,
+        key: Option<KeyByLabelValues>,
+        aggregation_id: u64,
+        policy_fp: PolicyFingerprint,
+    ) -> Self {
+        Self {
+            start_timestamp,
+            end_timestamp,
+            key,
+            aggregation_id,
+            origin: Origin::Native,
+            policy_fp,
         }
     }
 
     /// Construct a `Backfilled { job_id }` precompute. Called by
     /// [`crate::storage_engines::sketch_db::backfill::processor::BackfillWindowProcessor`]
     /// so each backfilled window carries its provenance back to the
-    /// originating `BackfillJob`.
+    /// originating `BackfillJob`. Legacy variant — leaves `policy_fp`
+    /// as the sentinel. New code should prefer
+    /// [`Self::new_backfilled_with_policy_fp`].
     pub fn new_backfilled(
         start_timestamp: u64,
         end_timestamp: u64,
@@ -77,6 +119,26 @@ impl PrecomputedOutput {
             key,
             aggregation_id,
             origin: Origin::Backfilled { job_id },
+            policy_fp: PolicyFingerprint(0),
+        }
+    }
+
+    /// `new_backfilled` with policy-fingerprint identity.
+    pub fn new_backfilled_with_policy_fp(
+        start_timestamp: u64,
+        end_timestamp: u64,
+        key: Option<KeyByLabelValues>,
+        aggregation_id: u64,
+        job_id: u64,
+        policy_fp: PolicyFingerprint,
+    ) -> Self {
+        Self {
+            start_timestamp,
+            end_timestamp,
+            key,
+            aggregation_id,
+            origin: Origin::Backfilled { job_id },
+            policy_fp,
         }
     }