feat: end-to-end truncation by williamhbaker · Pull Request #3132 · estuary/flow

williamhbaker · 2026-07-07T21:53:45Z

Description:

When a capture re-backfills a binding it re-snapshots the source, so the previously-captured documents in the collection become stale. This adds an end-to-end truncation boundary so downstream consumers skip or reset that superseded data instead of carrying it forward.

A capture connector signals BackfillBegin / BackfillComplete per binding. The capture publishes each as a CONTROL document into the collection's journals and stamps an estuary.dev/truncated-at label (the begin clock) on them. Every reader (V1 and V2 alike) treats CONTROL docs as metadata rather than content, and a fresh reader skips the stale pre-boundary prefix via the label. Shuffle carries the begin/complete clocks through the checkpoint frontier, and the runtime persists them so the boundary survives transaction rotation and restart. Materializations use the boundary to drop stale source documents, reset stale stored rows, and notify the connector so it can delete superseded destination rows.

Scoped to single-shard, full-key-range captures; multi-shard synchronized backfills are future work.

Tested on a local stack through various scenarios with a single shard capture and materialization processing sequences of BackfillBegin / BackfillComplete messages.

Closes #2821

Workflow steps:

(How does one use this feature, and how has it changed)

Documentation links affected:

(list any documentation links that you created, or existing ones that you've identified as needing updates, along with a brief description)

Notes for reviewers:

There's a constant stream of merge conflicting things landing on master. I've rebased and fixed these more times than I care to count at this point. I'll do one final pass just before this is ready to merge (and the requisite re-approvals).

jgraettinger

Great! We talked over VC about re-structuring the journal carrier (folding into ACKs with causal hints). A few other comments below.

williamhbaker · 2026-07-16T15:13:42Z

Addressed the comments so far, and also rebased & pushed a couple other substantial changes:

Having the backfill signals as part of ACKs, rather than a separate message. Overall a net simplification, and a better model I think. That got me thinking a lot about causal hints and the reader side, which led to...
A fix for the pre-backfill document filtering logic. I'd had a scan filter based on the backfillBegin threaded through on the Frontier, but that was incomplete since it didn't account for the prior frontiers that may have already been received and scanned unfiltered. The only tractable way I could come up with for getting this fully correct was to double down on the filtering in the combiner, and actually add clock metadata for every document, and a filter when draining. This is a pretty significant change, and there are performance implications, but it seems like an unavoidable cost.
Less of a code change, and more of a behavior change: Materializations will be required to produce a _meta/uuid field with their Loaded responses. As a compatibility mechanism for pre-existing "no flow document" cases that might not have this field, this is only enforced when the transaction contains an active backfillBegin-based filter. Practically I think this will shake out as "no flow document" materializations needing to require some representation of the timestamp in their field selection.

Ran a fresh claude review and manual E2E tests on these pushed changes as well.

williamhbaker · 2026-07-21T18:25:59Z

Ran the combiner benchmarks before and after this change.

For larger docs, like the github/citi benchmark, there is no measurable change.
The worst case is tiny docs with small keys and zero reduction, which when synthesized and run in the combiner perf test shows a 7-8% overhead from the extra 8 bytes added to Meta.

jgraettinger

A few minor comments below (plus our Slack thread re: the combiner, which is my only real concern). Everything else makes sense!

jgraettinger · 2026-07-22T02:15:22Z

+    // propagation of flush and progress reporting.
+    let is_append = is_append
+        && *clock >= binding.not_before
+        && *clock >= read_state.truncated_at


confirming my understanding: binding.not_before is (only) the built spec's not-before, and does not updated with the journals' truncation label (carried only in read_state.truncated_at). Thus we must check both.

Yep that's right. They are separate, with the not_before being a property of the binding and the truncated_at being a proper of the journal.

jgraettinger · 2026-07-22T02:18:42Z

 stall timeout: it fires only when no progress at all occurs between
 two consecutive ticks.

+A peek also carries `latest_backfill_begin` eagerly (cloned from


nice, makes sense

jgraettinger · 2026-07-22T02:27:06Z

+            // caller.
+            for cas_retries in 0..10 {
+                let listing = retry_transient("list partitions", || {
+                    client.list(broker::ListRequest {


It'd be more idiomatic to use a list_watch or list_watch_with here, which can be polled again to retry, already handles back-off internally, and will return a restatement of the journals after every Ok change. You'd use its items to drive convergence retries.

Switched to using list_watch, much better.

jgraettinger · 2026-07-22T02:30:18Z

 anyhow = { workspace = true }
 base64 = { workspace = true }
 bytes = { workspace = true }
+chrono = { workspace = true }


nit: I'm not actually seeing chrono in use in publisher changes ?

Left over from some prior iterations...removed.

jgraettinger · 2026-07-22T02:36:20Z

-    persist_fut: Option<BoxFuture<'static, anyhow::Result<(crate::shard::RocksDB, Vec<String>)>>>,
+    labels_apply_fut: Option<BoxFuture<'static, (P, BTreeMap<u32, u64>, tonic::Result<()>)>>,
+    persist_fut: Option<
+        BoxFuture<


nit: factor out a type alias here and below?

jgraettinger · 2026-07-22T02:41:12Z

+                // sub-range shard that inherited `active_backfills` (e.g. a split
+                // mid-backfill) must not re-apply labels it can never clear — it
+                // never receives the BackfillComplete that removes them.
+                if !self.is_single_shard || !self.labels_dirty || self.active_backfills.is_empty() {


I'm wondering if we should make the rule that shard zero, alone, manages truncation signals. that would allow a CDC capture to scale out by having shard zero manage the WAL, while other shards perform backfill fetches.

Seems reasonable, I could see that working out. Updated.

jgraettinger · 2026-07-22T02:52:02Z

                })?;
-            memtable.add(binding_index as u16, doc, true)?;
+
+            // Classify by the loaded document's UUID clock. As a fall-back for


Should we make this fail faster? A concern is that materialization tasks will appear to work fine, and then blow up on a first source truncation, right?

Yeah, we probably should, I was thinking not right away though. With it like this, pre-existing "no flow document" materializations that don't load a _meta/uuid will keep working until captures start to emit truncations. I think the implementation on the connector side will need to do at least materializations first, with the "no flow document" cases always emitting a clock, and then we can harden this.

williamhbaker · 2026-07-22T04:37:33Z

Pushed a commit that includes a different strategy for the combiner changes, instead of adding document clock to the meta. It adds a truncate method which handles memtable documents directly, and fences off already spilled segments. It's a bit more machinery but avoids the very real performance hit of the extra 8 bytes to the meta. PTAL.

williamhbaker · 2026-07-22T15:21:02Z

Also pushed an additional commit to incorporate the review feedback. I'll aim to do one final rebase prior to merging that will amend the two fixup! commits into the cohesive history.

jgraettinger

Changes look great! All of my comments are resolved. Ready to stamp after you rebase.

jgraettinger · 2026-07-22T18:54:46Z

+            } else if !entry.meta.front() {
+                return false;
+            }
+            entry.meta.set_stale();


Nice, I hadn't fully considered the implications of a connector that streams Loaded responses on demand, which you caught here and have handled 👍

I'm expecting this means we need to still surface front() documents of cut-off segments for purposes of the Exists flag. (confirmed)

Backfill truncation lets a capture re-backfill a binding and mark the previously-captured data as stale. It needs new protocol surface: capture BackfillBegin / BackfillComplete responses, materialize Flush backfill notifications, runtime Recover.active_backfills and a Persist active-backfill change, and shuffle Frontier latest_backfill_{begin, complete} clocks keyed by binding index.

A reader must skip the stale pre-backfill prefix of a journal once its binding is re-backfilled. Record the boundary as an `estuary.dev/truncated-at` label: a fixed-width hex Gazette clock that sorts lexically by value and only ever advances. dekaf honors it by raising its effective not_before to the max truncated-at of its journals.

A materialization that loads a stale row — one published before its binding's backfill boundary — must not reduce it into the fresh, current document for the same key, or the superseded value would leak forward. Expose an explicit Accumulator::truncate(binding): the live MemTable drops the binding's pre-boundary sources and flags its Loaded fronts stale, while already-spilled segments are fenced by a per-binding ordinal cutoff. Staleness is a single Meta flag bit, keeping HeapEntry at 32 bytes and the spill entry header at 24.

Fold the begin/complete clocks of backfill markers into the checkpoint frontier, keyed by binding index. The markers ride the bodies of immediately-committed ACK documents, so they never participate in causal-hint sequencing; their clocks are instead max-reduced per binding across a checkpoint delta and surface on a resolved NextCheckpoint.

…d-at labels A backfill marker rides the body of a transaction's ACK documents, broadcast to every partition journal of the binding under a single shared clock, so even a constrained reader observes it. Un-upgraded readers skip ACKs, so the marker is invisible to them by construction and commits atomically with the connector checkpoint. Also apply the `estuary.dev/truncated-at` label to a binding's journals.

Backfill state must survive transaction rotation and leader restart. Add key families, split by task type: `AB:` holds a capture's in-progress active backfills (the clocks that drive the truncated-at journal label), while `BB:`/`BC:` (and hinted `HB:`/`HC:`) hold a materialization's cumulative begin/complete boundary keyed by state_key.

A capture connector may signal BackfillBegin / BackfillComplete for a binding. Each signal must stand alone in its connector checkpoint, so isolate it in its own transaction, publish the marker on that transaction's ACK documents — the begin's assigned clock is the authoritative truncated_at — and stage the active-backfill change atomically with the connector checkpoint.

Carry the per-binding truncation boundary from the frontier into the shard. The scan drops source documents older than the boundary, and a loaded row older than it is classified prior-generation so the combiner discards it instead of reducing stale state forward — its current-generation sibling is then stored as an update. The leader broadcasts each begin/complete notification to every shard's connector.

Regenerating protobufs reformats the checked-in bindings; run `cargo fmt --all` afterward so the diff carries only real changes.

jgraettinger · 2026-07-23T00:34:53Z

re: the agents.md change, fyi I added cargo fmt to the build:rust-protobufs task a bit ago

@jwhartley

## What's Changed * docs: surface MCP and agent skills for coding agents by @jwhartley in estuary/flow#3093 * dekaf: serve the document containing a mid-document fetch offset by @jshearer in estuary/flow#3150 * runtime: let docker atomically assign published connector host ports by @dgreer-dev in estuary/flow#3162 * runtime-next: discard hinted frontier when committed frontier is rebuilt by @jgraettinger in estuary/flow#3165 * docs/source-postgres: Sync statement_timeout change by @willdonnelly in estuary/flow#3168 * Provide capture/materializations created_at date by @dgreer-dev in estuary/flow#3160 * docs: add Processing order section for materialization binding priority by @jwhartley in estuary/flow#3171 * docs: reorganize agent skills page by plugin, add derivations and schema by @jwhartley in estuary/flow#3170 * go.mod: bump gazette to latest by @williamhbaker in estuary/flow#3180 * Service Accounts by @GregorShear in estuary/flow#3058 * Bound too-large error messages over RPC by @dgreer-dev in estuary/flow#3182 * control-plane: guarded, auto-expiring temporary support access to restricted tenants by @skord in estuary/flow#3115 * private links: model links as rows with controller-observed status by @jshearer in estuary/flow#3063 * Adding support for an stripe web hook endpoint. by @bbartman in estuary/flow#3174 * docs: Dekaf consumer behaviors + Spark Structured Streaming guidance by @jwhartley in estuary/flow#3094 * shuffle: log canonical journal name in slice read events by @jgraettinger in estuary/flow#3190 * runtime: persist the max-key fail-safe sentinel at adoption by @jgraettinger in estuary/flow#3191 * local: make flow-plane-link robust to control-plane agent start-up by @jgraettinger in estuary/flow#3158 * gazette: surface fragment store diagnostic on FRAGMENT_STORE_UNHEALTHY by @jgraettinger in estuary/flow#3159 * docs: add warning about source-stripe-native's events API version by @nicolaslazo in estuary/flow#3166 * agent-api: Add `STRIPE_WEBHOOK_SECRET` to cloud run config by @jshearer in estuary/flow#3196 * shuffle, gazette: guard against wedged reads by @jgraettinger in estuary/flow#3193 * runtime-next: gate task-log stream by configured log level by @williamhbaker in estuary/flow#3198 * flowctl: preview-next appends a final drain session by @mdibaiee in estuary/flow#3195 * Allow localhost:3000 origin for the local control-plane agent by @GregorShear in estuary/flow#3206 * control-plane-api: don't lock injected ops collections during publications by @skord in estuary/flow#3164 * runtime-next: force txn close at any usage ceiling, bypassing min duration by @jgraettinger in estuary/flow#3203 * shuffle: replay gapped producers on restart by @jgraettinger in estuary/flow#3202 * docs: source-zuora by @Alex-Bair in estuary/flow#3204 * dekaf: only read back when a fetch offset lands inside a document by @mdibaiee in estuary/flow#3205 * mise: dev-VM zone configurability (create fallback, zone-agnostic SSH, vm:move-gcp) by @skord in estuary/flow#3208 * docs/postgres: Discovery Filters by @willdonnelly in estuary/flow#3215 * docs: document materialize-s3-iceberg nanosecond_timestamps option by @jacobmarble in estuary/flow#3199 * runtime v2 flag for new captures and derivations by @williamhbaker in estuary/flow#3219 * runtime-next: converge startup reconciliation via rescan Persists by @jgraettinger in estuary/flow#3218 * build(deps): bump fast-uri from 3.1.0 to 3.1.4 in /site by @dependabot[bot] in estuary/flow#3223 * build(deps): bump svgo from 3.3.3 to 3.3.4 in /site by @dependabot[bot] in estuary/flow#3224 * build(deps): bump dompurify from 3.3.3 to 3.4.12 in /site by @dependabot[bot] in estuary/flow#3222 * build(deps): bump shell-quote from 1.8.3 to 1.10.0 in /site by @dependabot[bot] in estuary/flow#3213 * build(deps): bump body-parser from 1.20.4 to 1.20.6 in /site by @dependabot[bot] in estuary/flow#3212 * build(deps): bump webpack-dev-server from 5.2.2 to 5.2.6 in /site by @dependabot[bot] in estuary/flow#3211 * build(deps): bump websocket-driver from 0.7.4 to 0.7.5 in /site by @dependabot[bot] in estuary/flow#3183 * Docs: MySQL discovery filters by @aeluce in estuary/flow#3217 * docs: source-smartsheet by @nicolaslazo in estuary/flow#3228 * control-plane-api: surface per-store health diagnostics in storage-mapping mutations by @GregorShear in estuary/flow#3181 * async-process: handle reaping-task JoinError in Child::wait() without panicking by @jgraettinger in estuary/flow#3225 * feat: end-to-end truncation by @williamhbaker in estuary/flow#3132 * docs: source-zuora uses the AQuA API now by @Alex-Bair in estuary/flow#3237 * Improve storage mapping GraphQL results by @GregorShear in estuary/flow#3200 * control-plane: add `closed` flag to data planes by @GregorShear in estuary/flow#3175 **Full Changelog**: estuary/flow@v0.6.11...v0.6.12 Co-authored-by: mdibaiee <mdibaiee@users.noreply.github.com>

@jwhartley

## What's Changed * docs: surface MCP and agent skills for coding agents by @jwhartley in estuary/flow#3093 * dekaf: serve the document containing a mid-document fetch offset by @jshearer in estuary/flow#3150 * runtime: let docker atomically assign published connector host ports by @dgreer-dev in estuary/flow#3162 * runtime-next: discard hinted frontier when committed frontier is rebuilt by @jgraettinger in estuary/flow#3165 * docs/source-postgres: Sync statement_timeout change by @willdonnelly in estuary/flow#3168 * Provide capture/materializations created_at date by @dgreer-dev in estuary/flow#3160 * docs: add Processing order section for materialization binding priority by @jwhartley in estuary/flow#3171 * docs: reorganize agent skills page by plugin, add derivations and schema by @jwhartley in estuary/flow#3170 * go.mod: bump gazette to latest by @williamhbaker in estuary/flow#3180 * Service Accounts by @GregorShear in estuary/flow#3058 * Bound too-large error messages over RPC by @dgreer-dev in estuary/flow#3182 * control-plane: guarded, auto-expiring temporary support access to restricted tenants by @skord in estuary/flow#3115 * private links: model links as rows with controller-observed status by @jshearer in estuary/flow#3063 * Adding support for an stripe web hook endpoint. by @bbartman in estuary/flow#3174 * docs: Dekaf consumer behaviors + Spark Structured Streaming guidance by @jwhartley in estuary/flow#3094 * shuffle: log canonical journal name in slice read events by @jgraettinger in estuary/flow#3190 * runtime: persist the max-key fail-safe sentinel at adoption by @jgraettinger in estuary/flow#3191 * local: make flow-plane-link robust to control-plane agent start-up by @jgraettinger in estuary/flow#3158 * gazette: surface fragment store diagnostic on FRAGMENT_STORE_UNHEALTHY by @jgraettinger in estuary/flow#3159 * docs: add warning about source-stripe-native's events API version by @nicolaslazo in estuary/flow#3166 * agent-api: Add `STRIPE_WEBHOOK_SECRET` to cloud run config by @jshearer in estuary/flow#3196 * shuffle, gazette: guard against wedged reads by @jgraettinger in estuary/flow#3193 * runtime-next: gate task-log stream by configured log level by @williamhbaker in estuary/flow#3198 * flowctl: preview-next appends a final drain session by @mdibaiee in estuary/flow#3195 * Allow localhost:3000 origin for the local control-plane agent by @GregorShear in estuary/flow#3206 * control-plane-api: don't lock injected ops collections during publications by @skord in estuary/flow#3164 * runtime-next: force txn close at any usage ceiling, bypassing min duration by @jgraettinger in estuary/flow#3203 * shuffle: replay gapped producers on restart by @jgraettinger in estuary/flow#3202 * docs: source-zuora by @Alex-Bair in estuary/flow#3204 * dekaf: only read back when a fetch offset lands inside a document by @mdibaiee in estuary/flow#3205 * mise: dev-VM zone configurability (create fallback, zone-agnostic SSH, vm:move-gcp) by @skord in estuary/flow#3208 * docs/postgres: Discovery Filters by @willdonnelly in estuary/flow#3215 * docs: document materialize-s3-iceberg nanosecond_timestamps option by @jacobmarble in estuary/flow#3199 * runtime v2 flag for new captures and derivations by @williamhbaker in estuary/flow#3219 * runtime-next: converge startup reconciliation via rescan Persists by @jgraettinger in estuary/flow#3218 * build(deps): bump fast-uri from 3.1.0 to 3.1.4 in /site by @dependabot[bot] in estuary/flow#3223 * build(deps): bump svgo from 3.3.3 to 3.3.4 in /site by @dependabot[bot] in estuary/flow#3224 * build(deps): bump dompurify from 3.3.3 to 3.4.12 in /site by @dependabot[bot] in estuary/flow#3222 * build(deps): bump shell-quote from 1.8.3 to 1.10.0 in /site by @dependabot[bot] in estuary/flow#3213 * build(deps): bump body-parser from 1.20.4 to 1.20.6 in /site by @dependabot[bot] in estuary/flow#3212 * build(deps): bump webpack-dev-server from 5.2.2 to 5.2.6 in /site by @dependabot[bot] in estuary/flow#3211 * build(deps): bump websocket-driver from 0.7.4 to 0.7.5 in /site by @dependabot[bot] in estuary/flow#3183 * Docs: MySQL discovery filters by @aeluce in estuary/flow#3217 * docs: source-smartsheet by @nicolaslazo in estuary/flow#3228 * control-plane-api: surface per-store health diagnostics in storage-mapping mutations by @GregorShear in estuary/flow#3181 * async-process: handle reaping-task JoinError in Child::wait() without panicking by @jgraettinger in estuary/flow#3225 * feat: end-to-end truncation by @williamhbaker in estuary/flow#3132 * docs: source-zuora uses the AQuA API now by @Alex-Bair in estuary/flow#3237 * Improve storage mapping GraphQL results by @GregorShear in estuary/flow#3200 * control-plane: add `closed` flag to data planes by @GregorShear in estuary/flow#3175 **Full Changelog**: estuary/flow@v0.6.11...v0.6.12 Co-authored-by: mdibaiee <mdibaiee@users.noreply.github.com>

williamhbaker force-pushed the wb/truncation-v2 branch from 91c351f to 1be6962 Compare July 8, 2026 19:56

williamhbaker requested a review from a team July 8, 2026 20:37

williamhbaker marked this pull request as ready for review July 8, 2026 20:38

jgraettinger reviewed Jul 9, 2026

View reviewed changes

Comment thread go/protocols/shuffle/shuffle.proto Outdated

Comment thread crates/runtime-next/src/shard/materialize/actor.rs Outdated

Comment thread crates/doc/src/combine/memtable.rs Outdated

Comment thread crates/runtime-next/src/shard/capture/actor.rs Outdated

williamhbaker force-pushed the wb/truncation-v2 branch from 1be6962 to 23983ca Compare July 16, 2026 13:57

jgraettinger reviewed Jul 22, 2026

View reviewed changes

Alex-Bair mentioned this pull request Jul 22, 2026

source-smartsheet: general implementation improvements estuary/connectors#4869

Merged

1 task

jgraettinger previously approved these changes Jul 22, 2026

View reviewed changes

williamhbaker added 9 commits July 22, 2026 20:12

AGENTS.md: note cargo fmt after protobuf codegen

b652d77

Regenerating protobufs reformats the checked-in bindings; run `cargo fmt --all` afterward so the diff carries only real changes.

williamhbaker dismissed jgraettinger’s stale review via b652d77 July 22, 2026 23:20

williamhbaker force-pushed the wb/truncation-v2 branch from 07dcc7a to b652d77 Compare July 22, 2026 23:20

jgraettinger approved these changes Jul 23, 2026

View reviewed changes

williamhbaker merged commit 7e6769e into master Jul 23, 2026
11 of 12 checks passed

williamhbaker deleted the wb/truncation-v2 branch July 23, 2026 04:51

github-actions Bot mentioned this pull request Jul 23, 2026

🚀 Release 0.6.12 estuary/homebrew-flowctl#52

Merged

github-actions Bot mentioned this pull request Jul 23, 2026

🚀 Release 0.6.12 estuary/homebrew-flowctl#53

Merged

Uh oh!

Conversation

williamhbaker commented Jul 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jgraettinger left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

williamhbaker commented Jul 16, 2026

Uh oh!

williamhbaker commented Jul 21, 2026

Uh oh!

jgraettinger left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

williamhbaker commented Jul 22, 2026

Uh oh!

williamhbaker commented Jul 22, 2026

Uh oh!

jgraettinger left a comment

Choose a reason for hiding this comment

Uh oh!

jgraettinger Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jgraettinger commented Jul 23, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

williamhbaker commented Jul 7, 2026 •

edited

Loading

jgraettinger Jul 22, 2026 •

edited

Loading