Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

runtime-next: converge startup reconciliation via rescan Persists - #3218

#3218
Merged
jgraettinger merged 5 commits into
masterestuary/flow:masterfrom
johnny/materialization-frontier-persist-e55db6estuary/flow:johnny/materialization-frontier-persist-e55db6Copy head branch name to clipboard
Jul 22, 2026
Merged

runtime-next: converge startup reconciliation via rescan Persists#3218
jgraettinger merged 5 commits into
masterestuary/flow:masterfrom
johnny/materialization-frontier-persist-e55db6estuary/flow:johnny/materialization-frontier-persist-e55db6Copy head branch name to clipboard

Conversation

@jgraettinger

@jgraettinger jgraettinger commented Jul 21, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Summary

Reworks runtime-next leader startup reconciliation from a one-shot cleanup Persist into an iterative convergence loop, so a leader's startup state is a projection of the actually-scanned durable state rather than of an in-memory reconstruction — the leader can never diverge from disk.

The reconcile policy is factored into a pure, IO-free reconcile function (one each for derive/materialize) expressed as an ordered sequence of triggered, self-clearing steps:

  • convert a V1-written legacy checkpoint,
  • drop a V1 rollback,
  • fold a hinted commit the connector confirms,
  • adopt a connector checkpoint that committed ahead.

reconcile_loop repeatedly reconciles the scanned RocksDB Baseline against its authoritative checkpoints, applying each step as a rescan Persist and re-reading the freshly written durable state before the next iteration.

Bug fix: crash-loop in the remote-authoritative connector case

The old materialize reconciliation applied the hinted-commit fold purely in memory — it advanced the session's committed_close/committed Frontier but emitted no Persist, leaving RocksDB's committed_close at its pre-fold value. A session could fold a hinted commit (in memory), write the next hint via StartCommit (bumping on-disk hinted_close), then crash before that transaction committed. Recovery then saw a stale committed_close, a newer hinted_close, and an endpoint clock matching neither — hitting the reconcile_recovered mismatch bail! on every recovery, crash-looping the shard.

Routing the fold through reconcile_loop as a rescan Persist makes the advance durable, so a later next-hint crash recovers on the in-sync branch and never bails. Covered by the reconcile_fold_then_next_hint_matches_committed_close regression test.

Commits

  • flow.schema.json: regenerate and enforce up-to-date in CI — fix drifted schema; add ci:flow-schema-check gating future model changes.
  • runtime-next: RocksDB::scan takes binding-index-ordered state keys — scan builds its own sorted index internally; becomes a sync fn returning a lifetime-free Future.
  • runtime: add Persist.rescan field — when set, shard zero re-scans RocksDB after applying the WriteBatch and replies with a fresh Recover instead of a Persisted echo. Regenerates Go/Rust protobuf.
  • runtime-next: extract shared leader test fixtures — pure refactor pulling leader test builders into a leader::fixtures module.
  • runtime-next: converge startup reconciliation via rescan Persists — the reconcile loop and crash-loop fix above.

QA: local-stack soak

Ran a chaos soak on a local stack (runtime built from this branch) using the tests/soak/ harness — capture → accounts derivation → views (materialize-postgres, remote-authoritative, the fold target) + ledger (recovery-log-authoritative). Shards were force-recovered by repeatedly deleting their Etcd assignment keys under continuous load (graceful cancel → reassign → fresh recovery + startup reconciliation).

Reproduced the crash-loop scenario and confirmed the fix. With defaults the fold window is too narrow to sample (65 random kills → 0 folds, all in-sync recoveries clean). Capping maxTxnDuration to force frequent commits then made the leader log land in the fold window repeatedly:

leader.materialize … connector checkpoint matches hinted_close; folding the hinted delta as committed
    committed_close=…:07.909Z  hinted_close=…:07.980Z      # endpoint committed the hint, then crashed
leader … persisting reconciled startup baseline and re-scanning  iteration=1

That second line is the fix: the fold is now persisted durably via a rescan Persist rather than applied in memory.

Results across the run:

  • 22 fold events, 22 matching rescan-persists — every fold recovered cleanly; zero did not converge / committed_close-mismatch bails / crash-loops.
  • 309 views recoveries (plus ledger/accounts/source recoveries); all shards stayed live.
  • All 6 views at-rest SQL oracles = 0 (conservation std+delta, oracle-match std+delta, sentinels, std-vs-delta), full account population materialized.
  • ledger in-flight self-checks: no conservation/oracle/seq/sentinel violations (its only ERRORs were expected Stop or CloseNow … unexpected EOF, i.e. the kill severing the gRPC stream). accounts: only the expected schema-widening churn.

Unrelated finding (not this change): aggressive shard cancellation can panic crates/async-process/src/lib.rs:49 (handle.await.unwrap() → JoinError::Cancelled) reaping a connector subprocess — rare (1/135 kills), non-fatal (shard recovers). Pre-existing connector-lifecycle plumbing; being fixed separately.

Test plan

Covered by PR CI (passing):

  • cargo test -p runtime-next (includes the reconcile_* unit tests / reconcile_fold_then_next_hint_matches_committed_close regression)
  • cargo test -p proto-flow
  • CI ci:flow-schema-check
  • Local-stack chaos soak (above)

@jgraettinger
flow.schema.json: regenerate and enforce up-to-date in CI
f51c72c 
The checked-in schema had drifted from crates/models (a TriggerConfig
'interval' field was missing). Regenerate it, and add ci:flow-schema-check
(mirroring ci:graphql-schema-check) which rebuilds the schema and fails on
any diff, so future model changes affecting the JSON schema can't land
without regenerating it.

The platform-test pipeline (task and GitHub Actions workflow) now runs
ci:flow-schema-check in place of the bare build:flow-schema step; flowctl
is already built earlier in that job by ci:gnu-dev.
@jgraettinger
runtime-next: RocksDB::scan takes binding-index-ordered state keys
7ae4d05 
Change `RocksDB::scan` to accept an iterator yielding each binding's
`state_key` in binding-index order, rather than a pre-sorted
`Vec<(state_key, binding_index)>`. `scan` now builds its own sorted
`(state_key, binding)` index internally, so callers no longer duplicate
the enumerate-and-sort boilerplate.

The signature also becomes a synchronous function returning a Future:
the index is built eagerly so the returned Future owns it and is free of
the iterator's lifetime.

Updates the capture, derive, and materialize shard startup callers and
their tests accordingly.
@jgraettinger
runtime: add Persist.rescan field
bfb7344 
Add a `rescan` boolean to the `Persist` message. When set, shard zero
re-scans RocksDB after applying the Persist's WriteBatch and replies with
a fresh `Recover` (reflecting the reconciled on-disk state) instead of a
`Persisted` echo.

This supports the leader's startup reconciliation loop, which converges
recovered state against an authoritative checkpoint by repeatedly
persisting a reconciled baseline and re-scanning to observe the shard's
persist semantics.

Regenerates the checked-in Go and Rust protobuf bindings.
@jgraettinger
runtime-next: extract shared leader test fixtures
2fc3977 
Pull the leader test builders (producer IDs, ProducerFrontier /
JournalFrontier, and checkpoint Source / ProducerEntry) into a new
leader::fixtures module, and refactor the frontier_mapping tests onto
them, dropping their duplicated local copies.

Pure refactor: no behavior change.
@jgraettinger
runtime-next: converge startup reconciliation via rescan Persists
0cd55e7 
Rework leader startup reconciliation from a one-shot cleanup Persist into
an iterative convergence loop. `reconcile_loop` repeatedly reconciles the
scanned RocksDB `Baseline` against its authoritative checkpoints, applying
each self-clearing step as a `rescan` Persist and re-reading the freshly
written durable state before the next iteration. Session startup state is
then a projection of the actually-scanned durable state rather than of an
in-memory reconstruction, so the leader can never diverge from disk.

The reconcile policy is factored into a pure, IO-free `reconcile` function
(one per derive/materialize) expressed as an ordered sequence of triggered,
self-clearing steps: convert a V1-written legacy checkpoint, drop V1
rollback, fold a hinted commit the connector confirms, and adopt a
connector checkpoint that committed ahead. `frontier_mapping::adopt_checkpoint`
builds the Persist that adopts a checkpoint wholesale, and
`COMMITTED_CLOSE_FLOOR` seeds the committed close when an adopted checkpoint
carries no embedded Clock.

Shard actors gain `rescan` handling: a rescan Persist re-scans the DB and
replies with a fresh `Recover` (rather than a `Persisted` echo) so the
leader observes reconciled on-disk state.

Extract shared reconciliation test fixtures and make `uuid::Clock::from_u64`
const so the floor Clock is a constant.

Fix a crash-loop in the remote-authoritative connector case
===========================================================

The old materialize reconciliation applied the hinted-commit fold — the
case where the connector's checkpoint confirms a recovered hinted
transaction did in fact commit at the endpoint — purely in memory. It
advanced `committed_close` and reduced the committed Frontier for the
session, but emitted no Persist (it set neither `committed_frontier_rebuilt`
nor a cleanup Persist), so RocksDB's `committed_close` stayed at its
pre-fold value.

That left durable state inconsistent with the endpoint. A session could:

  1. start up and fold a hinted commit the endpoint had confirmed (in
     memory only — `committed_close` on disk unchanged),
  2. run a transaction that wrote the *next* hint via StartCommit (bumping
     the on-disk `hinted_close`), then
  3. crash before that transaction committed at the endpoint.

The subsequent recovery then saw a stale on-disk `committed_close`, a newer
`hinted_close`, and an endpoint clock equal to the folded value — which
matched neither — so `reconcile_recovered` hit its mismatch `bail!`. Because
the crash never changed durable state, every recovery re-derived the same
mismatch and the shard crash-looped.

Routing the fold through `reconcile_loop` as a `rescan` Persist makes the
advance durable: `committed_close` moves on disk during the fold recovery,
so a later next-hint crash recovers with the endpoint clock matching the
persisted `committed_close` (the in-sync branch) and never bails. Covered
by the `reconcile_fold_then_next_hint_matches_committed_close` regression
test.
@jgraettinger
jgraettinger requested a review from a team July 21, 2026 20:31
@williamhbaker williamhbaker linked an issue Jul 21, 2026 that may be closed by this pull request
runtime-v2: ahead-remote-authoritative recovery promotes committed_close in memory only #3176
Closed
williamhbaker
williamhbaker approved these changes Jul 21, 2026
View reviewed changes

@williamhbaker williamhbaker left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM

Something I realized when thinking about this: A remote authoritative connector that also uses the driver checkpoint mechanism for persisting state in the recovery log might lose its driver checkpoint state if it commits the runtime checkpoint in the endpoint and then crashes before the recovery log commit. Then upon recovery, it would resume from its own endpoint checkpoint. I think this is necessarily the way it has to be, and existed in runtime V1 as well.

@jgraettinger
jgraettinger merged commit f4cee81 into master Jul 22, 2026
11 checks passed
@jgraettinger
jgraettinger deleted the johnny/materialization-frontier-persist-e55db6 branch July 22, 2026 00:17
@github-actions github-actions Bot mentioned this pull request Jul 23, 2026
Merged
mdibaiee added a commit to estuary/homebrew-flowctl that referenced this pull request Jul 23, 2026
@github-actions @mdibaiee
Release 0.6.12 (#52)
74ec244 
## 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>
@github-actions github-actions Bot mentioned this pull request Jul 23, 2026
Merged
mdibaiee added a commit to estuary/homebrew-flowctl that referenced this pull request Jul 23, 2026
@github-actions @mdibaiee
Release 0.6.12 (#53)
1a0065c 
## 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>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@williamhbaker williamhbaker williamhbaker approved these changes

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

runtime-v2: ahead-remote-authoritative recovery promotes committed_close in memory only

2 participants

@jgraettinger @williamhbaker
Morty Proxy This is a proxified and sanitized view of the page, visit original site.