Skip to content

Replicate Calvin commit, barrier, and read state through the data group - #414

Merged
farhan-syah merged 19 commits into
mainfrom
fix/calvin-completion
Oct 11, 2026
Merged

farhan-syah merged 19 commits into
mainfrom
fix/calvin-completion

Conversation

@farhan-syah

@farhan-syah farhan-syah commented Oct 10, 2026 •

Copy link
Copy Markdown
Member

What changed

Calvin commit, recovery, and read validation now run through the vShard data group.

  • Commit: Scheduler stages, votes, and resolves only on the data-group leader. Other replicas install the stamped redo from the log (a4b497066).
  • Redo: a slice is applied through the stamped redo record. Redo larger than max_redo_entry_bytes travels as chunk entries, bounded by max_open_redo_bytes. nodedb/src/wal/redo/chunk.rs and append_redo_chunk.rs handle the WAL side.
  • Restart: a scheduler records its highest applied epoch and re-applies sequencer log entries up to it. The gateway opens after.
  • Restage: a leader whose stage fails after an earlier COMMIT vote restages with doubling backoff. New knobs are restage_attempts and restage_backoff_ms.
  • Dependent reads: passive read results live in a per-vShard barrier log (calvin_barrier_logs.rs, barrier_log.rs), replicated through the data group. The leader proposes ReplicatedWrite::CalvinReadTimeout after dependent_read_passive_timeout_ms, now 5000 ms by default. Every replica aborts the barrier alike.
  • Planner: TRANSFER_ITEM between collections on two vShards runs as one read-dependent Calvin transaction instead of returning an error.
  • Locks: the lock manager takes moded collection and UNIQUE keys. StageVote replaces the read_set_valid flag on stage responses.
  • Write gate: a replicated write passes DataProposeGate on the leader before it enters the log. A contended write routes to the sequencer (RouteToSequencer) or waits (AdmissionTimedOut). The transport no longer resends an unanswered request unless RaftRpc::resend_safe holds. It ends as ClusterError::Unanswered.
  • Read validation: WriteVersion (data-group epoch and log index) replaces the per-node LSN watermark. Any replica validates a read. Array slice and aggregate legs report observed versions.
  • Backup: the Calvin sequencer marker carries the consistent-cut barrier. Every group orders a transaction on the same side of it.
  • Statements: native, pgwire, and the HTTP routes share one path in statement_exec. INSERT..SELECT, MERGE, and UPDATE..FROM into a materialized-sum source run through the expanders in an implicit transaction.
  • Arrays: tile_prefix and cell_tile_prefix spread tile keys over the top bits.
  • Test support: fail points take a node scope (name@node<N>=action in NODEDB_FAILPOINTS, FailGuard::for_node). A Raft commit fail gate sits behind the failpoints feature.
  • Refactor: dispatch_remote_stream moves out of dispatch_remote. The bridge dispatcher's per-tenant accounting moves into TenantInflight.

BREAKING: MetaOp::CalvinFlush, the CalvinApplied WAL record, and the CDC position partition type are removed. The nodedb-lite repo must remove its MetaOp::CalvinFlush match arm in adapter/meta.rs.

Why

A cross-shard Calvin transaction lost its state on a leader move or restart. Dependent-read barriers lived in scheduler memory. Applied slices were tracked per node. This change keeps both in the replicated log, so a promoted follower or restarted node resumes from the log.

Closes #393. A barrier that times out parks the transaction with an abort vote (park_unstaged). A pre-stage dispatch failure does the same through reject_plan (AbortReason::PlanRejected). The tally completes, and every peer releases its locks at the abort verdict. Tests: a_rejected_plan_aborts_every_participant_through_the_verdict and calvin_dependent_read_timeout.

Known limits, left for the next PR:

  • Each Calvin barrier event costs one fsync of its durable row.
  • If removing a departed vShard's barrier rows fails, the rows stay until the next boot.
  • A remote execute to an unresponsive node waits the statement budget plus 2 s. A read can spend the whole budget before the retry on another node.
  • Arrays with unbounded i64 dimensions keep low-bit tile keys. Small-coordinate tiles route to one vShard. This affects load spread, not correctness.

How to check

Run each command from nodedb/, one at a time, with --retries 0.

cargo nextest run --workspace --exclude nodedb-cluster-tests --all-features --no-fail-fast --retries 0
cargo nextest run -p nodedb-cluster-tests --all-features --no-fail-fast --retries 0

Results: stage 1 passes 20640 of 20640. Stage 2 passes 497 of 497. Clippy is clean with all features and with none.

nodedb-cluster-tests/tests/common_suite/cases/calvin_dependent_read_timeout.rs covers the barrier timeout and needs --features failpoints.

A Calvin scheduler now records the highest epoch it applied and, after a
restart, re-applies sequencer log entries up to that epoch before the
node opens its client gateway. Startup waits on every local scheduler
and fails when the data-group recovery bound expires or a scheduler
halts. A newly elected sequencer leader seeds its epoch from the
installed sequencer snapshot instead of refusing to propose on a
compacted log.

Related changes:
- Completion routing, waiters, verdict handling and the owed-completion
  retry path track per-transaction history in the sequencer state
  machine and surface typed abort errors to the planner retry loop.
- Scheduler channel capacity, transaction deadline, passive-read
  timeout, verdict stall warning, in-flight backlog and catch-up window
  are configurable under `tuning.calvin`.
- The bridge dispatcher's per-tenant in-flight accounting moves into
  its own type.
- Raft leaders track voter contact from the election win, so a voter
  not yet heard this term counts from that point.
- The CI nextest profile runs every test, including the cluster test
  that was excluded.
… vote

The Calvin lock manager now takes moded requests. Row keys lock
exclusive or shared, and a collection key carries intent, shared, or
exclusive modes so a collection-wide writer or predicate reader orders
against row writers of the same collection. UNIQUE values a write claims
lock as their own keys, so transactions that claim one value serialize.
The planner derives these keys, and EngineKeySet and LockKeyWire gain
Collection and Unique variants.

A Calvin stage response now carries a typed StageVote (commit,
serialization conflict, prediction drift, participant error) in place of
the optional read_set_valid flag. A stage response with no readable vote
halts apply instead of counting as a commit.
A replicated write now passes the write gate on the data-group leader
before its entry enters the log, whether the proposal is local or
forwarded. The gate takes the write's lock keys on the vShard's Calvin
lock table and holds them until the leader starts the entry's apply, so
a Calvin transaction that needs those keys orders after the write.

- A contended write that a Calvin transaction can sequence is refused
  with RouteToSequencer, and the proposer submits it to the sequencer.
  Appends never route.
- Any other contended write waits, holding no key, until the keys are
  free or the proposer's deadline passes (AdmissionTimedOut). A
  forwarded entry carries what remains of that deadline.
- The gate itself moves into the host crate behind the DataProposeGate
  trait, split into admission keys, holds, and wait modules.

The transport no longer resends a request that was written and got no
answer unless RaftRpc::resend_safe holds. Any other such request ends
as the new ClusterError::Unanswered, an unknown outcome, and the
gateway reports a lost answer instead of running the request twice.
Proposals carry a deadline, and Calvin submits draw on a shared budget.
…data DDL state split

Calvin applied ledger and Data-Plane install
- The applied mirror becomes an applied ledger in the scheduler. The
  Data Plane installs a Calvin transaction's redo and records it applied
  through a dedicated install path, and the snapshot capture and install
  steps carry the ledger.

Chunked redo streams
- A committed redo larger than max_redo_entry_bytes travels in the
  data-group log as chunk entries followed by a final entry. The
  proposer can abandon a stream, and the apply loop reassembles chunks.
- A node bounds the bytes of open streams with max_open_redo_bytes, and
  a group leader refuses a new stream that would pass it.
- The WAL appends redo chunks and holds a floor on truncation while a
  stream is open, so crash recovery can resume or discard it. A snapshot
  owed record keeps a stream's floor across restarts.
- Raft limits the entry data of one AppendEntries request to
  MAX_APPEND_BATCH_BYTES. An entry larger than the cap ships alone.

Metadata DDL state split
- The pending DDL and metadata applier host state move out of the
  shared state fields into a metadata_ddl module.
Leader-driven commit
- Only the vShard's data-group leader stages, votes, resolves, and
  proposes a committed slice's stamped redo. Other replicas hold each
  granted txn as Following and install its slice from the log.
- Promotion stages every Following txn again in sequencer order.
  Demotion discards the leader's staged state and owed votes.
- A committed redo is proposed right after the resolve. The apply loop
  claims each position in the applied ledger, so a copy installs at most
  once. A slice that reads a whole collection at resolve waits for its
  turn behind lower txns.

Apply notifications
- The apply loop pushes one event per concluded redo into a bounded
  per-vShard scheduler inbox. The scheduler deposits the reply, owes the
  CompletionAck, releases locks, and completes. A refused or unapplied
  redo halts the scheduler.
- The separate flush turn, write-version record, redo window, and Data
  Plane fence are removed in favour of this path. A drop discards a txn's
  staged state on its core.

Transport
- An inbound RPC stream whose handler started but never answered is now
  reset instead of finished, so a sender can tell an unknown outcome
  from a refusal before the handler and never resends the former.

Tests cover leader change mid-transaction, write-version parity, and
the reworked two-phase apply cases.
A Calvin slice is now recorded as applied by the stamped redo record
that installs it, so the separate CalvinApplied WAL record and the
flush path that wrote it are removed. The scheduler restart path, the
reply handlers, and the planner and admission code follow the single
install path.

CDC positions drop the per-transaction base component: a replicated
event is positioned by its (epoch, index) alone, and the position
partition type is removed.
… copies

Redo sum targets now carry only the targets a source row folds on its
own core; balances that travel on an ApplyBalanceDelta task no longer
appear in the table, so the install skips them without a deferred list.

Count redo copies that install nothing because their position was
applied or another copy held the claim. Add a fail gate between a
durable install and its RedoApplied event so tests can exercise a
withheld event and a repeated proposal.

Statements in a transaction block that write a materialized-sum source
answer with the command tag of their own rows only. The derived balance
move no longer changes the verb or the count.

Add cluster, crash, wire, and unit tests for duplicate redo copies,
replica convergence, and installs cut by a crash.
Route INSERT..SELECT, MERGE and UPDATE..FROM join expanders through
materialized-sum resolution and admission so sums stay correct when the
expanded writes span shards. Add wire coverage for the cross-shard
expander cases.
Move statement execution into a shared statement_exec module so native,
pgwire and the HTTP routes (/v1/query, /v1/query/stream and WebSocket
RPC) run a statement through the same implicit transaction, implicit-edge
gate and Calvin commit. Statements the shared router does not take fall
back to each route's per-task loop.

A MERGE, UPDATE ... FROM or INSERT ... SELECT into a source that drives
a cross-shard materialized sum no longer needs the autocommit
orchestrators to refuse it: every protocol runs it through the expanders
in an implicit transaction. Remove the refusal and the native
sql_dispatch_task module it served.

Materialized-sum and period-lock writes in a transaction block now read
their pre-image as the transaction sees it, so a staged insert or
rewrite contributes the correct balance delta. Add wire cases for the
HTTP routes and for the in-transaction pre-image.
A leader under the gate counts no quorum, so cluster tests can keep an
entry uncommitted across a leadership change. The gate is compiled only
with the failpoints feature and is forwarded from the nodedb and
nodedb-cluster crates.
A leader whose own stage fails after an earlier leader's COMMIT vote
decided the transaction now stages it again with doubling backoff
instead of dropping it. The scheduler halts once the attempts run out.
Adds restage_attempts and restage_backoff_ms tuning knobs and restage
metrics.
Adds harness helpers to move and wait on data-group leadership, and
cluster cases for a split leader and an old-term redo copy.
The cut previously proposed the data-group barrier independently of the
Calvin sequencer marker, so a transaction could land before the barrier
in one group and after it in another. A backup or restore point could
then hold a transaction on one group only.

The marker now carries the barrier. Each group's leader proposes the
barrier once its schedulers pass the marker, and holds the redo of every
later transaction until it applies the barrier. Every slice of a
transaction sequenced before the marker sits before the barrier in each
group, and every slice of a later one sits after it.
Add tile_prefix and cell_tile_prefix, which scale each tile index over
its dimension's tile count before encoding the space-filling-curve key.
Tiles of one array now route to distinct buckets instead of clustering
in the low bits. Export tiles_per_dim for callers that need the count.
Replace the per-node LSN watermark on versioned reads with a
WriteVersion: the epoch and log index of the data-group entry that
wrote a row, or a local LSN for writes that apply no entry. Every
replica of a vShard holds the same versions, so any replica validates
a read, and the served_by node pin is no longer needed.

Array slice and aggregate legs report the versions they observed, and
the coordinator merges them into the session read-set. Snapshot
installs carry a version floor so versions stay comparable across a
leader move. Read validation, read splitting and pending shard reads
replace the homed, served and graph read trackers in the session
commit path.
Move passive read results of a read-dependent Calvin transaction out of
scheduler memory and into a per-vShard barrier log held in the system
catalog and replicated through the data group. Every replica of a vShard
folds the same barrier entries, so a follower promoted to leader, or a
node restarted mid-barrier, resumes the transaction from the log rather
than losing the reads.

A barrier that waits past dependent_read_passive_timeout_ms is resolved
by a timeout entry the data-group leader proposes, so every replica
aborts it alike. The default timeout rises to 5000 ms, and
max_inflight_backlog also bounds the barrier entries held in memory for
transactions the scheduler has not taken yet.

The planner builds read-dependent and predicted transaction classes
separately. TRANSFER_ITEM between collections on two vShards now runs
as one read-dependent Calvin transaction instead of returning an error.
An in-process cluster shares one fail-point registry across its nodes,
so a point could only target a node by encoding the node id in its
name. Fail points now carry an explicit scope (any node or one node),
evaluated against the scope of the core, producer, or raft group that
reaches them.

- Split the fail-point module in nodedb-types into action, scope,
  registry, guard, env, and macro files, with FailGuard::for_node to arm
  a point on one node.
- Read NODEDB_FAILPOINTS entries as `name=action` for every node or
  `name@node<N>=action` for node N only.
- Give each Data Plane core and its event producer the node's scope, and
  evaluate the Raft commit gate, WAL, funnel, and catalog gates in it.
- Re-export the test API from nodedb-test-support behind a failpoints
  feature, and move the cluster tests onto node-scoped guards.
- Add a cluster test that holds one replica's apply while fast-path and
  Calvin writes hit the same rows, then checks byte-equal convergence.
Split dispatch_remote_stream out of dispatch_remote to keep each file
focused on one dispatch mode.
Mark each time-based call in the read-result proposal and run loop as
liveness-only, with the reason the log order still decides the result.
@farhan-syah farhan-syah added the run-ci Opt this PR into the full test suite; re-add to force a re-run label Oct 11, 2026
@farhan-syah
farhan-syah merged commit 126bac7 into main Oct 11, 2026
13 of 14 checks passed
@farhan-syah
farhan-syah deleted the fix/calvin-completion branch October 11, 2026 00:12
farhan-syah added a commit that referenced this pull request Oct 12, 2026
Bound cluster waits by deadline and close the Calvin gaps from #414
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

run-ci Opt this PR into the full test suite; re-add to force a re-run

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Cluster/Calvin: barrier failure proposes no abort vote — peers hold locks until restart, client burns its deadline

1 participant