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

.config/nextest.toml

@@ -212,18 +212,8 @@ server-process-serial = { max-threads = 1 }
 retries = { backoff = "fixed", count = 1, delay = "2s" }
 fail-fast = false
 
-# One cluster test is excluded from the CI profile ONLY. It still runs under the
-# default profile, which is what a local `cargo nextest run` uses.
-#
-# `learner_caught_up_via_real_install_snapshot` fails on CI runners about once in
-# six runs. A newly elected sequencer leader whose log was compacted refuses to
-# propose ("sequencer log was truncated below every retained epoch"), so the
-# sequencer stalls. The joining node then never catches up on the sequencer
-# group, and its authorization lease is withheld. Re-enable it once a leader can
-# seed its epoch from the installed sequencer snapshot. NEVER widen a timeout.
-default-filter = """
-not test(learner_caught_up_via_real_install_snapshot)
-"""
+# CI runs every test, including the two binaries the default profile excludes.
+default-filter = "all()"
 
 # CI-only cluster headroom. On CI runners (~2× slower per-core than a dev
 # workstation) the cluster tests' in-test `wait_for` budgets are proportionally

CONTRIBUTING.md

@@ -153,7 +153,7 @@ NodeDB routes all DML through `EngineRules` in `nodedb-sql`. This is the single
 
 **Integration tests** — place in `nodedb/tests/` or the relevant crate's `tests/` directory. Test public API only. Use `tests/common/mod.rs` for shared helpers (not `tests/common.rs`).
 
-**Failpoint tests** — for fault injection and panic recovery, use the `failpoints` feature flag and `FailGuard` from `nodedb::fail_point`. Compile with `--features failpoints`.
+**Failpoint tests** — for fault injection and panic recovery, use the `failpoints` feature flag and `FailGuard` from `nodedb::fail_point`. Compile with `--features failpoints`. An in-process cluster shares one fail-point registry across its nodes. Arm a point on one node with `FailGuard::for_node(node_id, name, action)`, never by putting the node in the name. A spawned server takes `NODEDB_FAILPOINTS='name=action'` for every node, or `name@node<N>=action` for node `N` only.
 
 **Cluster tests** — 3-node integration tests live in `nodedb/tests/` and are annotated with the `cluster` nextest group. They serialize automatically via `.config/nextest.toml` — don't run them with `cargo test`.
 

docs/getting-started.md

@@ -589,6 +589,48 @@ a crashed node's descriptor leases block DDL until they expire.
 All three must be positive. `scope_expiry_interval_secs` has a floor of `10`.
 Below that the sweep costs more than the resolution it buys.
 
+**Calvin scheduler settings** (config file only):
+
+| Config field                                     | Default |
+| ------------------------------------------------ | ------- |
+| `tuning.calvin.channel_capacity`                 | `512`   |
+| `tuning.calvin.txn_deadline_multiplier`          | `3`     |
+| `tuning.calvin.dependent_read_passive_timeout_ms` | `5000`  |
+| `tuning.calvin.verdict_stall_warn_ms`            | `5000`  |
+| `tuning.calvin.max_inflight_backlog`             | `1024`  |
+| `tuning.calvin.catch_up_window`                  | `512`   |
+| `tuning.calvin.restage_attempts`                 | `5`     |
+| `tuning.calvin.restage_backoff_ms`               | `100`   |
+
+Each hosted vShard runs one Calvin scheduler with these settings.
+
+- `channel_capacity` sizes each scheduler's bounded input and completion
+  channels.
+- `txn_deadline_multiplier` sets a transaction deadline in sequencer epochs.
+- `dependent_read_passive_timeout_ms` bounds the wait of a dependent-read
+  barrier for passive read results. Past it, the vShard's data-group leader
+  proposes a timeout entry, and every replica aborts the barrier alike.
+- `verdict_stall_warn_ms` spaces the stall warnings of a transaction waiting
+  on its global verdict. The transaction keeps waiting and never aborts.
+- `max_inflight_backlog` is the in-flight backlog at which a scheduler stops
+  taking new input. It also bounds the dependent-read barrier entries a
+  vShard holds in memory for transactions its scheduler has not taken yet.
+  Entries past it wait in the system catalog alone.
+- `catch_up_window` is the most sequencer log entries one catch-up pass
+  replays.
+- `restage_attempts` is the most times a data-group leader stages a
+  committed transaction again after its own stage failed. The scheduler
+  halts once they run out.
+- `restage_backoff_ms` is the wait before the first restage. Each later
+  restage waits twice as long, at most 60 seconds.
+
+Every value must be positive. `verdict_stall_warn_ms` has a floor of `4`.
+
+After a restart, a scheduler re-applies epochs up to the highest epoch it
+applied before. The client gateway opens only after every scheduler on the
+node reaches that epoch. The wait shares the data-group recovery bound, and
+startup fails when the bound expires.
+
 **Observability settings:**
 
 | Config field                                      | Environment variable              | Default        |

nodedb-array/src/lib.rs

@@ -36,7 +36,8 @@ pub use sync::{
     SnapshotHeader, SnapshotSink, TileSnapshot,
 };
 pub use tile::{
-    AttrStats, DenseTile, SparseTile, TileMBR, tile_id_for_cell, tile_indices_for_cell,
+    AttrStats, DenseTile, SparseTile, TileMBR, cell_tile_prefix, tile_id_for_cell,
+    tile_indices_for_cell, tile_prefix, tiles_per_dim,
 };
 pub use types::{ArrayId, CellValue, Coord, Domain, TileId};
 pub use wal::ArrayWalRecord;

nodedb-array/src/tile/layout.rs

@@ -8,10 +8,10 @@
 //! same tile share a [`TileId::hilbert_prefix`].
 
 use crate::coord::encode::encode_tile_prefix_with_order;
-use crate::coord::normalize::bits_per_dim;
+use crate::coord::normalize::{bits_per_dim, normalize_coord};
 use crate::error::{ArrayError, ArrayResult};
 use crate::schema::ArraySchema;
-use crate::schema::dim_spec::DimType;
+use crate::schema::dim_spec::{DimSpec, DimType};
 use crate::types::TileId;
 use crate::types::coord::value::CoordValue;
 use crate::types::domain::DomainBound;
@@ -76,11 +76,93 @@ pub fn tile_id_for_cell(
     system_from_ms: i64,
 ) -> ArrayResult<TileId> {
     let tile_indices = tile_indices_for_cell(schema, coord)?;
-    let bits = bits_per_dim(schema.arity());
-    let prefix = encode_tile_prefix_with_order(schema, &tile_indices, bits)?;
+    let prefix = tile_prefix(schema, &tile_indices)?;
     Ok(TileId::new(prefix, system_from_ms))
 }
 
+/// The space-filling-curve key of the tile at `tile_indices`.
+///
+/// The memtable, the segment index, the transaction overlay and retention
+/// group cells by this key, and a cluster routes a tile by its top bits.
+/// Every cell inside a group stays addressed by its full coordinate: a
+/// lookup matches the coordinate within the group
+/// (`TileBuffer::get_cell_bytes`, `extract_cell_bytes`), and a tile's MBR
+/// is built from the cells it holds. Two tiles that share a key therefore
+/// share one group and lose no cell.
+///
+/// A dimension whose tile count fits its share of the key bits scales each
+/// index over that count to the full bit width. Its tiles keep distinct,
+/// ordered keys that spread over the top bits. A dimension with more tiles
+/// than its bits can number, such as one with the default full-range
+/// domain, keeps the low bits of the index instead. Nearby tiles then keep
+/// distinct keys, and tiles `2^bits` apart share one.
+pub fn tile_prefix(schema: &ArraySchema, tile_indices: &[u64]) -> ArrayResult<u64> {
+    if tile_indices.len() != schema.arity() {
+        return Err(ArrayError::CoordArityMismatch {
+            array: schema.name.clone(),
+            expected: schema.arity(),
+            got: tile_indices.len(),
+        });
+    }
+    let bits = bits_per_dim(schema.arity());
+    let scaled: Vec<u64> = tile_indices
+        .iter()
+        .zip(schema.dims.iter().zip(schema.tile_extents.iter()))
+        .map(|(index, (dim, extent))| scale_tile_index(*index, tiles_per_dim(dim, *extent), bits))
+        .collect();
+    encode_tile_prefix_with_order(schema, &scaled, bits)
+}
+
+/// The key of the tile that holds the cell at `coord`: the key every
+/// cluster path routes the cell by. Fails when `coord` lies outside the
+/// schema's domain.
+pub fn cell_tile_prefix(schema: &ArraySchema, coord: &[CoordValue]) -> ArrayResult<u64> {
+    normalize_coord(schema, coord, bits_per_dim(schema.arity()))?;
+    tile_prefix(schema, &tile_indices_for_cell(schema, coord)?)
+}
+
+/// The number of tiles along `dim` for tile extent `extent`. Mirrors how
+/// [`tile_indices_for_cell`] numbers them: a cell in the domain has a tile
+/// index below this count.
+pub fn tiles_per_dim(dim: &DimSpec, extent: u64) -> u128 {
+    let extent = u128::from(extent.max(1));
+    match (dim.dtype, &dim.domain.lo, &dim.domain.hi) {
+        (DimType::Int64, DomainBound::Int64(lo), DomainBound::Int64(hi))
+        | (DimType::TimestampMs, DomainBound::TimestampMs(lo), DomainBound::TimestampMs(hi)) => {
+            let span = ((*hi as i128) - (*lo as i128)).max(0) as u128;
+            span / extent + 1
+        }
+        (DimType::Float64, DomainBound::Float64(lo), DomainBound::Float64(hi)) => {
+            if hi.is_finite() && lo.is_finite() && hi > lo {
+                u128::from((hi - lo) as u64) / extent + 1
+            } else {
+                1
+            }
+        }
+        (DimType::String, DomainBound::String(_), DomainBound::String(_)) => extent,
+        _ => 1,
+    }
+}
+
+/// Map tile index `index` of `tiles` tiles to `[0, 2^bits)`.
+///
+/// With `tiles <= 2^bits` the index is scaled over the tile count, so
+/// distinct indices keep distinct, ordered values. An index past the last
+/// tile, a cell above the domain, takes the top value. With more tiles the
+/// index keeps its low `bits` bits.
+fn scale_tile_index(index: u64, tiles: u128, bits: u32) -> u64 {
+    let limit: u128 = 1u128 << bits.min(64);
+    let max = limit - 1;
+    if tiles > limit {
+        return u64::try_from(u128::from(index) & max).unwrap_or(u64::MAX);
+    }
+    if tiles <= 1 {
+        return if index == 0 { 0 } else { max as u64 };
+    }
+    let scaled = (u128::from(index) * max / (tiles - 1)).min(max);
+    u64::try_from(scaled).unwrap_or(u64::MAX)
+}
+
 fn out_of_domain<T>(schema: &ArraySchema, dim: &str, detail: &str) -> ArrayResult<T> {
     Err(ArrayError::CoordOutOfDomain {
         array: schema.name.clone(),
@@ -138,6 +220,99 @@ mod tests {
         assert_ne!(t1.hilbert_prefix, t2.hilbert_prefix);
     }
 
+    #[test]
+    fn tile_keys_spread_over_the_top_bits() {
+        let s = schema();
+        let tops: std::collections::BTreeSet<u64> = (0..10)
+            .flat_map(|x| (0..10).map(move |y| (x * 10, y * 10)))
+            .map(|(x, y)| {
+                tile_id_for_cell(&s, &[CoordValue::Int64(x), CoordValue::Int64(y)], 0)
+                    .unwrap()
+                    .hilbert_prefix
+                    >> 56
+            })
+            .collect();
+        assert!(
+            tops.len() > 1,
+            "the tiles of one array route to more than one bucket: {tops:?}"
+        );
+    }
+
+    #[test]
+    fn every_tile_keeps_a_distinct_key() {
+        let s = schema();
+        let keys: std::collections::BTreeSet<u64> = (0..10)
+            .flat_map(|x| (0..10).map(move |y| (x * 10, y * 10)))
+            .map(|(x, y)| {
+                tile_id_for_cell(&s, &[CoordValue::Int64(x), CoordValue::Int64(y)], 0)
+                    .unwrap()
+                    .hilbert_prefix
+            })
+            .collect();
+        assert_eq!(keys.len(), 100);
+    }
+
+    #[test]
+    fn a_cell_routes_by_its_tile_key() {
+        let s = schema();
+        let coord = [CoordValue::Int64(37), CoordValue::Int64(81)];
+        assert_eq!(
+            cell_tile_prefix(&s, &coord).unwrap(),
+            tile_id_for_cell(&s, &coord, 0).unwrap().hilbert_prefix
+        );
+        assert_eq!(
+            cell_tile_prefix(&s, &[CoordValue::Int64(30), CoordValue::Int64(89)]).unwrap(),
+            cell_tile_prefix(&s, &coord).unwrap(),
+            "cells of one tile route together"
+        );
+    }
+
+    #[test]
+    fn a_cell_outside_the_domain_has_no_route() {
+        let s = schema();
+        assert!(cell_tile_prefix(&s, &[CoordValue::Int64(100), CoordValue::Int64(0)]).is_err());
+    }
+
+    #[test]
+    fn a_cell_above_the_domain_takes_the_last_key() {
+        let s = schema();
+        let above = tile_id_for_cell(&s, &[CoordValue::Int64(150), CoordValue::Int64(0)], 0)
+            .unwrap()
+            .hilbert_prefix;
+        let last = tile_id_for_cell(&s, &[CoordValue::Int64(99), CoordValue::Int64(0)], 0)
+            .unwrap()
+            .hilbert_prefix;
+        assert_eq!(above, last);
+    }
+
+    #[test]
+    fn nearby_tiles_of_a_full_range_dimension_keep_distinct_keys() {
+        let s = ArraySchemaBuilder::new("open")
+            .dim(DimSpec::new(
+                "x",
+                DimType::Int64,
+                Domain::new(DomainBound::Int64(i64::MIN), DomainBound::Int64(i64::MAX)),
+            ))
+            .dim(DimSpec::new(
+                "y",
+                DimType::Int64,
+                Domain::new(DomainBound::Int64(i64::MIN), DomainBound::Int64(i64::MAX)),
+            ))
+            .attr(AttrSpec::new("v", AttrType::Int64, false))
+            .tile_extents(vec![10, 10])
+            .build()
+            .unwrap();
+        let keys: std::collections::BTreeSet<u64> = (0..10)
+            .flat_map(|x| (0..10).map(move |y| (x * 10, y * 10)))
+            .map(|(x, y)| {
+                tile_id_for_cell(&s, &[CoordValue::Int64(x), CoordValue::Int64(y)], 0)
+                    .unwrap()
+                    .hilbert_prefix
+            })
+            .collect();
+        assert_eq!(keys.len(), 100, "a schema with any domain builds");
+    }
+
     #[test]
     fn system_from_ms_propagates() {
         let s = schema();

nodedb-array/src/tile/mod.rs

@@ -11,6 +11,8 @@ pub use cell_payload::{
     is_cell_gdpr_erasure, is_cell_sentinel, is_cell_tombstone,
 };
 pub use dense_tile::DenseTile;
-pub use layout::{tile_id_for_cell, tile_indices_for_cell};
+pub use layout::{
+    cell_tile_prefix, tile_id_for_cell, tile_indices_for_cell, tile_prefix, tiles_per_dim,
+};
 pub use mbr::{AttrStats, TileMBR};
 pub use sparse_tile::SparseTile;

nodedb-cluster-tests/Cargo.toml

@@ -12,7 +12,11 @@ path = "src/lib.rs"
 
 [features]
 # Arms the in-process fail points the cluster tests park writes at.
-failpoints = ["nodedb/failpoints", "nodedb-types/failpoints"]
+failpoints = [
+  "nodedb/failpoints",
+  "nodedb-types/failpoints",
+  "nodedb-test-support/failpoints",
+]
 
 [dependencies]
 

nodedb-cluster-tests/tests/cluster_suite/cases/calvin_e2e_ollp.rs

@@ -13,7 +13,7 @@
 //!    surrogates currently matching the predicate ("predicted surrogates").
 //! 2. The TxClass is built with the predicted surrogates for the OLLP
 //!    collection and static surrogates for any additional static-key writes
-//!    in the same transaction (via `build_dependent_tx_class`), then submitted
+//!    in the same transaction (via `build_predicted_tx_class`), then submitted
 //!    to the sequencer inbox.
 //! 3. The sequencer admits the txn and fans it out to all participant vshards.
 //! 4. The Data Plane executor verifies the predicted set against the actual
@@ -71,7 +71,7 @@ fn two_distinct_vshard_collections() -> (String, String) {
 /// - An OLLP BulkUpdate on `col_ollp` with `predicted_surrogates` as the
 ///   write set (since the exact set is not known statically).
 ///
-/// This mirrors what `build_dependent_tx_class` produces on the Control Plane
+/// This mirrors what `build_predicted_tx_class` produces on the Control Plane
 /// for a cross-shard transaction that contains a value-dependent predicate write.
 fn make_ollp_tx_class(
     col_static: &str,
@@ -94,7 +94,7 @@ fn make_ollp_tx_class(
         write_set,
         // Empty plan bytes — this test exercises the sequencer admission and
         // fan-out path only. The actual BulkUpdate plan bytes are embedded by
-        // `build_dependent_tx_class` in production.
+        // `build_predicted_tx_class` in production.
         vec![],
         TenantId::new(1),
         None,

nodedb-cluster-tests/tests/common_suite/cases/backup_schedule_failover/lagging_catalog.rs

@@ -4,8 +4,8 @@
 //! instead of rerunning the due minute.
 
 use nodedb::control::backup::schedule::envelope_name;
-use nodedb::control::cluster::metadata_applier::backup_mark_fail_point;
-use nodedb_types::fail_point::FailGuard;
+use nodedb::control::cluster::metadata_applier::BACKUP_MARK_FAIL_POINT;
+use nodedb_test_support::fail_point::{FailAction, FailGuard};
 
 use super::fixture::{DATABASE, Fixture};
 
@@ -26,9 +26,10 @@ async fn a_new_leader_whose_catalog_lags_never_reruns_the_due_minute() {
         .iter()
         .filter(|node| node.node_id != leader)
         .map(|node| {
-            FailGuard::fail(
-                &backup_mark_fail_point(node.node_id),
-                "held back by the test",
+            FailGuard::for_node(
+                node.node_id,
+                BACKUP_MARK_FAIL_POINT,
+                FailAction::Fail("held back by the test".to_string()),
             )
         })
         .collect();