From a3abcaad3c3492c56bb14cd558b98e7f9d725a3e Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 13 May 2026 14:24:48 -0600 Subject: [PATCH] feat(sid): migrate PrecomputedOutput + sinks to PolicyFingerprint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR 4 of the merged-sid-identity chain. Threads `PolicyFingerprint` through the precompute output → sink path while keeping `aggregation_id` as a transitional fallback. ## What - `PrecomputedOutput` gains a `policy_fp: PolicyFingerprint` field. `#[serde(default)]` on read means records persisted before this PR deserialise as `PolicyFingerprint::UNSET` (the all-zero sentinel introduced here). - New constructors `PrecomputedOutput::new_with_policy_fp` and `new_backfilled_with_policy_fp` take both the legacy `aggregation_id` and the content-addressed `policy_fp`. The pre-existing `new` / `new_backfilled` continue to compile and leave `policy_fp` as the sentinel — used by the raw-mode fast-path that doesn't carry a source `AggregationConfig`. - `SketchStoreSink::append_to_index` now resolves the source config via `PolicyRegistry::get(policy_fp)` when the output carries a non-sentinel fp; falls back to the legacy `StreamingConfig::get_aggregation_config(aggregation_id)` lookup otherwise. - Precompute worker (`worker.rs`) — 7 group-state-driven emit sites migrated to `new_with_policy_fp`. The fp is computed once from `state.config` and threaded through each emit. - Backfill processor (`processor.rs`) — backfilled-window emit migrated to `new_backfilled_with_policy_fp`. - Raw-mode fast-path (`process_samples_raw`) stays on the legacy constructor with a code comment explaining the sentinel. The raw mode synthesises an `aggregation_id` without an associated `AggregationConfig`, so the fp can't be derived at emit time; the sink's fallback handles it. ## Dual-keyed invariant `StreamingConfig.aggregation_configs: HashMap` remains the source of truth. `PolicyRegistry` is a derived view (introduced in PR 3). The two lookup paths can never disagree — when the fp path returns a config, the agg_id path returns the same one (modulo the agg_id key itself). ## Test plan - [x] `cargo check --workspace` clean - [x] `cargo test --workspace --lib --bins` — all green at the time of commit (one pre-existing HashMap-iteration-order flake in `asap_types::capability_matching::tests::avg_finds_sum_and_count` is unrelated; documented for follow-up). ## Next - PR 5: delete `aggregation_id` from `AggregationConfig` and the YAML schema; control plane stops minting u64 ids; `policy_fp` becomes the only handle on the wire. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/asap_types/src/policy_fingerprint.rs | 16 ++++- .../src/precompute_engine/output_sink.rs | 48 +++++++++++-- data_plane/src/precompute_engine/worker.rs | 28 ++++++-- .../sketch_db/backfill/processor.rs | 17 +++-- .../types/precomputed_output.rs | 68 ++++++++++++++++++- 5 files changed, 152 insertions(+), 25 deletions(-) diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index 2df36a81..7cebe311 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/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`]. /// diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index 95a62930..3fcc82b0 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/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( diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 2ba2fb61..f908a558 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/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)); } diff --git a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs index 56cdc244..a7b0d422 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs +++ b/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)); } diff --git a/data_plane/src/storage_engines/types/precomputed_output.rs b/data_plane/src/storage_engines/types/precomputed_output.rs index 66a5addf..55c417cc 100644 --- a/data_plane/src/storage_engines/types/precomputed_output.rs +++ b/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, + 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, + 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, } }