Merge pull request #15 from pbudzik/feat/rollup-input-provenance

Rollup → raw segment provenance (closes spec §19.10)

Author: pbudzik

src/api/http_server.rs

@@ -526,22 +526,35 @@ async fn handle_explain(
     let corrections = execute_plan(&state, &plan_corr).await;
 
     // Segment provenance from the manifest. Filtering by bucket here
-    // matches the executor's pruning — the same segments the query would
-    // open are the ones we report.
-    let (watermark_ms, rollup_segments, raw_segments) = {
+    // matches the executor's pruning. For each overlapping rollup
+    // segment we also surface its `input_segment_ids` so an operator
+    // can name every raw segment that contributed to a rollup line —
+    // spec §19.10 (invoice snapshots reference a watermark + source
+    // segment set).
+    let (watermark_ms, rollup_segments, rollup_inputs, raw_segments) = {
         let manifest = state.manifest.read().await;
         let bucket_count = manifest.bucket_count.max(1);
         let target_bucket = bucket_for_account(&AccountId(account_id.clone()), bucket_count);
-        let rollups: Vec<String> = manifest
-            .rollup_segments
-            .iter()
-            .filter(|s| {
-                s.bucket == target_bucket
-                    && s.min_timestamp_ms < to_ms
-                    && s.max_timestamp_ms >= from_ms
-            })
-            .map(|s| s.segment_id.clone())
-            .collect();
+        let mut rollup_ids = Vec::new();
+        let mut inputs_map = serde_json::Map::new();
+        for s in &manifest.rollup_segments {
+            if s.bucket != target_bucket
+                || s.min_timestamp_ms >= to_ms
+                || s.max_timestamp_ms < from_ms
+            {
+                continue;
+            }
+            rollup_ids.push(s.segment_id.clone());
+            inputs_map.insert(
+                s.segment_id.clone(),
+                serde_json::Value::Array(
+                    s.input_segment_ids
+                        .iter()
+                        .map(|id| serde_json::Value::String(id.clone()))
+                        .collect(),
+                ),
+            );
+        }
         let raws: Vec<String> = manifest
             .raw_segments
             .iter()
@@ -552,7 +565,12 @@ async fn handle_explain(
             })
             .map(|s| s.segment_id.clone())
             .collect();
-        (manifest.watermarks.hourly_rollup_ms, rollups, raws)
+        (
+            manifest.watermarks.hourly_rollup_ms,
+            rollup_ids,
+            inputs_map,
+            raws,
+        )
     };
 
     Ok(Json(serde_json::json!({
@@ -562,6 +580,7 @@ async fn handle_explain(
         "watermark_ms": watermark_ms,
         "lines": lines,
         "rollup_segments": rollup_segments,
+        "rollup_inputs": rollup_inputs,
         "raw_segments": raw_segments,
         "corrections": corrections,
     })))

src/ingest/flusher.rs

@@ -239,5 +239,9 @@ pub fn build_segment_meta(segment_id: &str, batch: &[UsageEvent], bucket: u32, c
         model_ids,
         quantity_sum: Some(quantity_sum),
         checksum,
+        // Raw segments have no inputs — they're the ground truth.
+        // Compacted segments also leave this empty; their provenance is
+        // tracked via Manifest.compacted_replacements.
+        input_segment_ids: Vec::new(),
     }
 }

src/rollup/worker.rs

@@ -202,6 +202,13 @@ impl RollupWorker {
             let hour_end = hour + HOUR_MS;
 
             let mut by_bucket: HashMap<u32, RollupBuilder> = HashMap::new();
+            // Track per-bucket provenance: which raw segment IDs
+            // contributed events to each output rollup segment. Used to
+            // populate `SegmentMeta.input_segment_ids` so invoice
+            // lineage is auditable (spec §19.10).
+            let mut inputs_by_bucket: HashMap<u32, std::collections::BTreeSet<String>> =
+                HashMap::new();
+
             for seg in &raw_segments {
                 if seg.max_timestamp_ms < hour || seg.min_timestamp_ms >= hour_end {
                     continue;
@@ -225,6 +232,10 @@ impl RollupWorker {
                         continue;
                     }
                     by_bucket.entry(bucket).or_default().process_event(&event);
+                    inputs_by_bucket
+                        .entry(bucket)
+                        .or_default()
+                        .insert(seg.segment_id.clone());
                 }
             }
 
@@ -233,7 +244,12 @@ impl RollupWorker {
                 if records.is_empty() {
                     continue;
                 }
-                let (meta, path) = self.write_rollup_segment(records, bucket, hour, hour_end)?;
+                let inputs: Vec<String> = inputs_by_bucket
+                    .remove(&bucket)
+                    .map(|set| set.into_iter().collect())
+                    .unwrap_or_default();
+                let (meta, path) =
+                    self.write_rollup_segment(records, bucket, hour, hour_end, inputs)?;
                 new_segments.push((meta, path));
             }
 
@@ -331,6 +347,7 @@ impl RollupWorker {
         bucket: u32,
         hour_start: i64,
         hour_end: i64,
+        input_segment_ids: Vec<String>,
     ) -> anyhow::Result<(SegmentMeta, std::path::PathBuf)> {
         let segment_id = format!("rollup_{}", uuid::Uuid::new_v4().simple());
         let path = self
@@ -353,6 +370,7 @@ impl RollupWorker {
             hour_start,
             hour_end - 1,
             row_count,
+            input_segment_ids,
         );
         Ok((meta, path))
     }
@@ -366,6 +384,7 @@ fn build_rollup_segment_meta(
     min_ts: i64,
     max_ts: i64,
     row_count: u64,
+    input_segment_ids: Vec<String>,
 ) -> SegmentMeta {
     let mut product_ids = std::collections::HashSet::new();
     let mut meter_ids = std::collections::HashSet::new();
@@ -408,6 +427,7 @@ fn build_rollup_segment_meta(
         model_ids,
         quantity_sum: Some(quantity_sum),
         checksum,
+        input_segment_ids,
     }
 }
 

src/storage/manifest.rs

@@ -26,6 +26,17 @@ pub struct SegmentMeta {
     pub model_ids: HashSet<ModelId>,
     pub quantity_sum: Option<i128>,
     pub checksum: u64,
+    /// For rollup segments: the raw segment IDs whose events were
+    /// aggregated to produce this rollup. Empty for raw and compacted
+    /// segments (compacted segments' provenance lives in
+    /// `Manifest.compacted_replacements`).
+    ///
+    /// Spec §19.10: invoice snapshots must reference a watermark + the
+    /// source segment set. This is the per-rollup half of that — given
+    /// a rollup segment, you can name every raw segment that
+    /// contributed to it, so an invoice line's lineage is auditable.
+    #[serde(default)]
+    pub input_segment_ids: Vec<String>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]