feat: end-to-end truncation

AGENTS.md

@@ -30,6 +30,7 @@ go build -tags libsqlite3 ./go/bindings
 # Regenerate checked-in protobuf (required after .proto changes)
 mise run build:go-protobufs
 mise run build:rust-protobufs
+cargo fmt --all  # remove format-only churn from codegen
 
 # Run pgTAP SQL Tests
 mise run ci:sql-tap

crates/dekaf/src/read.rs

@@ -342,8 +342,9 @@ impl Read {
             };
             let (producer, clock, flags) = gazette::uuid::parse_str(uuid.as_str())?;
 
-            // Is this a non-content control document, such as a transaction ACK?
-            let is_control = flags.is_ack();
+            // Is this a non-content control document, such as a transaction ACK
+            // or a backfill control message?
+            let is_control = flags.is_ack() || flags.is_control();
 
             let should_skip = match (self.not_before, self.not_after) {
                 (Some(not_before), Some(not_after)) => clock < not_before || clock > not_after,

crates/dekaf/src/topology.rs

@@ -261,7 +261,7 @@ impl Collection {
 
         let key_schema = avro::key_to_avro(&key_ptr, collection_schema_shape);
 
-        let (not_before, not_after) = (
+        let (mut not_before, not_after) = (
             binding.not_before.map(|b| {
                 uuid::Clock::from_unix(b.seconds.try_into().unwrap(), b.nanos.try_into().unwrap())
             }),
@@ -270,6 +270,23 @@ impl Collection {
             }),
         );
 
+        // Honor truncated-at journal labels: if any partition carries a
+        // truncated-at label, use the max across all partitions and the
+        // binding's not_before as the effective not_before.
+        for partition in &partitions {
+            if let Some(truncated_at_str) = partition
+                .spec
+                .labels
+                .as_ref()
+                .and_then(|ls| ls.labels.iter().find(|l| l.name == ::labels::TRUNCATED_AT))
+                .map(|l| &l.value)
+            {
+                let truncated_clock =
+                    uuid::Clock::from_u64(::labels::parse_truncated_at(truncated_at_str)?);
+                not_before = Some(not_before.map_or(truncated_clock, |nb| nb.max(truncated_clock)));
+            }
+        }
+
         tracing::debug!(
             collection_name,
             partitions = partitions.len(),

crates/derive/src/extract_api.rs

@@ -135,8 +135,11 @@ impl cgo::Service for API {
                     }
                 })?;
 
-                if proto_gazette::message_flags::ACK_TXN & uuid.node != 0 {
-                    // Transaction acknowledgements aren't expected to validate.
+                if proto_gazette::message_flags::ACK_TXN & uuid.node != 0
+                    || proto_gazette::message_flags::CONTROL & uuid.node != 0
+                {
+                    // Transaction acknowledgements and application control
+                    // messages aren't expected to validate.
                 } else if let Some(validator) = &mut state.validator {
                     validator
                         .validate(&doc, |_| None)
@@ -167,7 +170,7 @@ impl cgo::Service for API {
 
 #[cfg(test)]
 mod test {
-    use super::{API, Code, extract_uuid_parts};
+    use super::{API, Code, Error, extract_uuid_parts};
     use cgo::Service;
     use proto_flow::flow;
     use serde_json::{Value, json};
@@ -282,4 +285,48 @@ mod test {
 
         insta::assert_debug_snapshot!((String::from_utf8_lossy(&arena), out));
     }
+
+    #[test]
+    fn ack_and_control_docs_skip_validation() {
+        let extract_validating = |uuid_doc: &str| -> Result<(), Error> {
+            let mut svc = API::create();
+            let (mut arena, mut out) = (Vec::new(), Vec::new());
+            svc.invoke_message(
+                Code::Configure as u32,
+                flow::extract_api::Config {
+                    uuid_ptr: "/0".to_string(),
+                    schema_json: bytes::Bytes::from_static(br#"{"type":"object"}"#),
+                    ..Default::default()
+                },
+                &mut arena,
+                &mut out,
+            )
+            .unwrap();
+            svc.invoke(
+                Code::Extract as u32,
+                uuid_doc.as_bytes(),
+                &mut arena,
+                &mut out,
+            )
+        };
+
+        // A plain (OUTSIDE_TXN, flags 0x0) doc is validated and fails the schema.
+        assert!(
+            matches!(
+                extract_validating(r#"["9f2952f3-c6a3-11ea-8800-080607050309", 0, 0]"#),
+                Err(Error::FailedValidation(_)),
+            ),
+            "a plain doc is validated and fails the object schema",
+        );
+        // ACK_TXN (0x2) and CONTROL (0x4) docs — e.g. backfill begin/complete —
+        // skip validation, so the same array-shaped doc is accepted.
+        assert!(
+            extract_validating(r#"["9f2952f3-c6a3-11ea-8802-080607050309", 0, 0]"#).is_ok(),
+            "an ACK_TXN doc skips validation",
+        );
+        assert!(
+            extract_validating(r#"["9f2952f3-c6a3-11ea-8804-080607050309", 0, 0]"#).is_ok(),
+            "a CONTROL doc skips validation",
+        );
+    }
 }

crates/doc/src/combine/memtable.rs

@@ -66,7 +66,7 @@ impl Entries {
     }
 
     fn compact(&mut self, alloc: &'static Bump) -> Result<(), Error> {
-        // `sort_ord` orders over (binding, key, !front):
+        // `sort_ord` orders over (binding, key, !prior_gen, !front):
         // For each (binding, key), we take front() entries first, and further
         // rely on sort preserving the order in which entries were added.
         // This maintains the left-to-right associative ordering of reductions.
@@ -80,6 +80,7 @@ impl Entries {
                     // Cold path: Meta prefix was equal, so compare the full key.
                     compare_root_keys(&self.spec.keys[l.meta.binding()], &l.root, &r.root)
                 })
+                .then_with(|| l.meta.prior_gen().cmp(&r.meta.prior_gen()).reverse())
                 .then_with(|| l.meta.front().cmp(&r.meta.front()).reverse())
         };
         let validators = &mut self.spec.validators;
@@ -234,6 +235,25 @@ impl MemTable {
 
     /// Add the document to the MemTable.
     pub fn add<'s>(&'s self, binding: u16, root: HeapNode<'s>, front: bool) -> Result<(), Error> {
+        self.add_with_flags(binding, root, front, false, false)
+    }
+
+    /// Add a document from a prior generation to the MemTable.
+    /// A prior-generation document is not reduced with current-generation
+    /// documents and is never drained: it transfers front() onto a following
+    /// current-generation document of its same key, and is otherwise discarded.
+    pub fn add_prior_gen<'s>(&'s self, binding: u16, root: HeapNode<'s>) -> Result<(), Error> {
+        self.add_with_flags(binding, root, true, true, true)
+    }
+
+    fn add_with_flags<'s>(
+        &'s self,
+        binding: u16,
+        root: HeapNode<'s>,
+        front: bool,
+        prior_gen: bool,
+        known_valid: bool,
+    ) -> Result<(), Error> {
         // Safety: mutable borrow does not escape this function.
         let entries = unsafe { &mut *self.entries.get() };
         let root = unsafe { std::mem::transmute::<HeapNode<'s>, HeapNode<'static>>(root) };
@@ -245,12 +265,7 @@ impl MemTable {
             &mut entries.scratch,
             None,
         );
-        let meta = Meta::new(
-            binding,
-            &entries.scratch,
-            front,
-            false, // `known_valid`
-        );
+        let meta = Meta::new(binding, &entries.scratch, front, known_valid, prior_gen);
 
         entries.queued.push(HeapEntry {
             meta,
@@ -303,7 +318,13 @@ impl MemTable {
         let raw = embedded.as_u64le_slice();
         let root = HeapRoot::Embedded(raw.as_ptr(), raw.len() as u32);
 
-        let meta = Meta::from_packed_prefix(binding, packed_key_prefix, front, known_valid);
+        let meta = Meta::from_packed_prefix(
+            binding,
+            packed_key_prefix,
+            front,
+            known_valid,
+            false, // prior generation documents are suppressed by the shuffle reader
+        );
         entries.queued.push(HeapEntry { meta, root });
 
         if entries.should_compact() {
@@ -432,6 +453,33 @@ impl MemDrainer {
         let Some(HeapEntry { mut meta, mut root }) = self.it.next() else {
             return Ok(None);
         };
+
+        // Prior-generation entries are never yielded: their content is
+        // superseded. One followed by its same-(binding, key)
+        // current-generation entry transfers front() onto it — the key was
+        // loaded, though its stale content is dropped — while an orphan
+        // (no such successor) is discarded outright.
+        while meta.prior_gen() {
+            let Some(next) = self.it.next() else {
+                return Ok(None); // Trailing orphan: drain is complete.
+            };
+
+            // Does `next` pair with the discarded prior-generation entry,
+            // as its current-generation sibling of the same (binding, key)?
+            // Consecutive prior-generation entries don't pair (and don't
+            // compare keys): each defers to the following iteration.
+            let paired = !next.meta.prior_gen()
+                && meta.0 == next.meta.0
+                && compare_root_keys(&self.spec.keys[meta.binding()], &root, &next.root).is_eq();
+
+            HeapEntry { meta, root } = next;
+            self.in_group = false;
+
+            if paired {
+                meta.set_front();
+            }
+        }
+
         let is_full = self.spec.is_full[meta.binding()];
         let keys = self.spec.keys[meta.binding()].as_ref();
         let name = &self.spec.names[meta.binding()];
@@ -645,6 +693,111 @@ mod test {
             .unwrap();
     }
 
+    #[test]
+    fn test_prior_gen_partition() {
+        // Prior-generation rows never reduce with, and are never drained
+        // alongside, current-generation rows: a paired row transfers front()
+        // onto its current-generation successor, while an orphan is discarded.
+        // Behavior must be identical whether drained in memory or through a
+        // spill file.
+        let make = || {
+            let schema = build_schema(
+                &url::Url::parse("http://example/schema").unwrap(),
+                &json!({
+                    "properties": { "v": { "type": "array", "reduce": { "strategy": "append" } } },
+                    "reduce": { "strategy": "merge" }
+                }),
+            )
+            .unwrap();
+            let memtable = MemTable::new(Spec::with_one_binding(
+                true,
+                vec![Extractor::with_default(
+                    "/key",
+                    &SerPolicy::noop(),
+                    json!("def"),
+                )],
+                "test",
+                Vec::new(),
+                Validator::new(schema).unwrap(),
+            ));
+
+            let add = |doc: Value, front: bool| {
+                memtable
+                    .add(0, HeapNode::from_node(&doc, memtable.alloc()), front)
+                    .unwrap()
+            };
+            let add_prior = |doc: Value| {
+                memtable
+                    .add_prior_gen(0, HeapNode::from_node(&doc, memtable.alloc()))
+                    .unwrap()
+            };
+
+            // Same key, a prior-generation run + its current generation:
+            // drained as one doc ["fresh"] with front() transferred (the run
+            // is held back from compaction into a single entry, so the drain
+            // skips through multiple prior-generation entries).
+            add(json!({"key": "k", "v": ["fresh"]}), false);
+            add_prior(json!({"key": "k", "v": ["stale"]}));
+            add_prior(json!({"key": "k", "v": ["stale2"]}));
+            // A schema-invalid orphan prior-generation row is discarded without
+            // FailedValidation: add_prior_gen marks it `known_valid` and it's
+            // dropped before drain-time validation.
+            add_prior(json!({"key": "bad", "v": 42}));
+
+            memtable
+        };
+
+        // Project a drained doc to (v, prior_gen, front). A `fn` (not a closure)
+        // so it can be reused across both drainers.
+        fn project(doc: crate::combine::DrainedDoc) -> (serde_json::Value, bool, bool) {
+            let root = serde_json::to_value(SerPolicy::noop().on_owned(&doc.root)).unwrap();
+            (root["v"].clone(), doc.meta.prior_gen(), doc.meta.front())
+        }
+        let expected = vec![
+            (json!(["fresh"]), false, true), // "k": stale content dropped, front() gained
+        ];
+
+        // In-memory drain.
+        let in_memory = make()
+            .try_into_drainer()
+            .unwrap()
+            .map_ok(project)
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+        assert_eq!(in_memory, expected, "in-memory drain");
+
+        // Spill drain must partition identically — including across segments:
+        // a second segment holds the current-generation sibling of "bad",
+        // pairing with the first segment's (invalid, never-validated) row.
+        let mut spill = SpillWriter::new(io::Cursor::new(Vec::new())).unwrap();
+        let spec = make().spill(&mut spill, CHUNK_TARGET_SIZE).unwrap();
+
+        let memtable = MemTable::new(spec);
+        memtable
+            .add(
+                0,
+                HeapNode::from_node(&json!({"key": "bad", "v": ["fixed"]}), memtable.alloc()),
+                false,
+            )
+            .unwrap();
+        let spec = memtable.spill(&mut spill, CHUNK_TARGET_SIZE).unwrap();
+
+        let (spill, ranges) = spill.into_parts();
+        let spilled = crate::combine::SpillDrainer::new(spec, spill, &ranges)
+            .unwrap()
+            .map_ok(project)
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+        assert_eq!(
+            spilled,
+            vec![
+                (json!(["fixed"]), false, true), // "bad": paired across segments
+                (json!(["fresh"]), false, true),
+            ],
+            "spill drain"
+        );
+    }
+
     #[test]
     fn test_memtable_combine_reduce_sequence() {
         let key = vec![Extractor::with_default(