Bound cluster waits by deadline and close the Calvin gaps from #414

.config/nextest.toml

@@ -157,7 +157,7 @@ threads-required = 'num-test-threads'
 # Every test in the `wire` target spawns its own real `nodedb` server process.
 # At full fan-out the box runs one server per core, and each server carries a
 # Data Plane plus Raft groups — enough CPU starvation that a metadata propose
-# misses its 5s deadline and the statement fails. Capping concurrent servers
+# misses its statement deadline and the statement fails. Capping concurrent servers
 # fixes the starvation; widening that deadline would only hide it, and the
 # deadline is real.
 #

nodedb-array/src/tile/layout.rs

@@ -83,19 +83,21 @@ pub fn tile_id_for_cell(
 /// 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
+/// group cells by this key. A cluster routes a tile by the top bits of
+/// `fmix64(key)` (`nodedb_types::array_route::prefix_bucket`), not by the
+/// key's own 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.
+/// share one group and one route, 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.
+/// ordered keys. 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. Such keys can share their top bits, and the routing mix
+/// still spreads them over buckets.
 pub fn tile_prefix(schema: &ArraySchema, tile_indices: &[u64]) -> ArrayResult<u64> {
     if tile_indices.len() != schema.arity() {
         return Err(ArrayError::CoordArityMismatch {
@@ -178,6 +180,7 @@ mod tests {
     use crate::schema::attr_spec::{AttrSpec, AttrType};
     use crate::schema::dim_spec::DimSpec;
     use crate::types::domain::Domain;
+    use nodedb_types::array_route::prefix_bucket;
 
     fn schema() -> ArraySchema {
         ArraySchemaBuilder::new("g")
@@ -221,20 +224,63 @@ mod tests {
     }
 
     #[test]
-    fn tile_keys_spread_over_the_top_bits() {
+    fn tile_keys_spread_over_routing_buckets() {
         let s = schema();
-        let tops: std::collections::BTreeSet<u64> = (0..10)
+        let buckets: std::collections::BTreeSet<u32> = (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)
+                let key = tile_id_for_cell(&s, &[CoordValue::Int64(x), CoordValue::Int64(y)], 0)
                     .unwrap()
-                    .hilbert_prefix
-                    >> 56
+                    .hilbert_prefix;
+                prefix_bucket(key, 8)
+            })
+            .collect();
+        // 100 tiles over 256 buckets: well over half of them distinct.
+        assert!(
+            buckets.len() >= 60,
+            "the tiles of one array route to {} buckets",
+            buckets.len()
+        );
+    }
+
+    fn full_range_2d() -> ArraySchema {
+        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![1, 1])
+            .build()
+            .unwrap()
+    }
+
+    /// The tiles of coordinates 0..64 x 0..64 in a full-range 2-D array
+    /// share their key's top bits, yet spread over the routing buckets.
+    #[test]
+    fn full_range_tiles_near_zero_spread_over_routing_buckets() {
+        let s = full_range_2d();
+        let keys: Vec<u64> = (0..64)
+            .flat_map(|x| (0..64).map(move |y| (x, y)))
+            .map(|(x, y)| {
+                cell_tile_prefix(&s, &[CoordValue::Int64(x), CoordValue::Int64(y)]).unwrap()
             })
             .collect();
+        let distinct: std::collections::BTreeSet<u64> = keys.iter().copied().collect();
+        assert_eq!(distinct.len(), 4096, "every tile keeps a distinct key");
+        let buckets: std::collections::BTreeSet<u32> =
+            keys.iter().map(|&k| prefix_bucket(k, 8)).collect();
+        // 4096 distinct keys over 256 buckets reach at least 7/8 of them.
         assert!(
-            tops.len() > 1,
-            "the tiles of one array route to more than one bucket: {tops:?}"
+            buckets.len() >= 224,
+            "4096 tiles route to only {} of 256 buckets",
+            buckets.len()
         );
     }
 

nodedb-cluster-tests/tests/common_suite/cases/calvin_backup_cut_snapshot_install.rs

@@ -0,0 +1,164 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! A replica that catches up by `InstallSnapshot` past a backup cut's
+//! barrier takes that cut at once.
+//!
+//! 1. A 3-node cluster with a low log-compaction threshold holds one
+//!    collection in data group `G`. The replica `F` neither leads `G` by
+//!    preference nor leads the metadata group.
+//! 2. `F` stops. A live node takes a backup cut at watermark `W`: every data
+//!    group applies the cut's barrier on the live nodes.
+//! 3. Writes compact `G`'s log past the barrier. `F` restarts and can catch
+//!    up only by a snapshot install, so it never applies the barrier entry.
+//! 4. `F` takes the same cut at `W`, as a remote source node of a backup
+//!    does. The snapshot carried the barrier, so the cut finishes at once.
+//!    Without it, `F` waits for a barrier no leader places again and the cut
+//!    fails with "Retry the backup".
+
+use std::time::{Duration, Instant};
+
+use crate::common::cluster_harness::{TestCluster, TestClusterNode, wait_for};
+
+/// Low enough that a few dozen single-row inserts compact `G`'s log.
+const COMPACTION_THRESHOLD: u64 = 4;
+
+const COLLECTION: &str = "cut_install_docs";
+
+/// Most rows the test writes to compact `G` past the barrier.
+const MAX_ROWS: usize = 200;
+
+const STEP_DEADLINE: Duration = Duration::from_secs(30);
+const STEP_POLL: Duration = Duration::from_millis(50);
+
+/// Well inside the cut's 30 s statement deadline: a cut that waits for a
+/// barrier no leader places runs to that deadline.
+const CUT_BUDGET: Duration = Duration::from_secs(20);
+
+fn node_by_id(cluster: &TestCluster, node_id: u64) -> &TestClusterNode {
+    cluster
+        .nodes
+        .iter()
+        .find(|node| node.node_id == node_id)
+        .unwrap_or_else(|| panic!("node {node_id} is live"))
+}
+
+/// The preferred leader of `group_id`, from `node`'s routing table.
+fn preferred_leader(node: &TestClusterNode, group_id: u64) -> u64 {
+    let routing = node
+        .shared
+        .cluster_routing
+        .as_ref()
+        .expect("cluster routing")
+        .read()
+        .unwrap_or_else(|p| p.into_inner());
+    nodedb_cluster::rebalancer::preferred_leaders(&routing)
+        .get(&group_id)
+        .copied()
+        .unwrap_or_else(|| panic!("group {group_id} has a preferred leader"))
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn a_snapshot_installed_replica_takes_a_cut_whose_barrier_it_skipped() {
+    let mut cluster = TestCluster::spawn_three_with_compaction_threshold(COMPACTION_THRESHOLD)
+        .await
+        .expect("3-node cluster with a low compaction threshold");
+    cluster
+        .exec_ddl_on_any_leader(&format!(
+            "CREATE COLLECTION {COLLECTION} (id TEXT PRIMARY KEY, payload TEXT) \
+             WITH (engine='document_strict')"
+        ))
+        .await
+        .expect("create the collection");
+    cluster.wait_for_preferred_leaders(STEP_DEADLINE).await;
+    cluster.wait_for_full_apply_convergence(STEP_DEADLINE).await;
+
+    let probe = &cluster.nodes[0];
+    let group_id = probe
+        .group_id_for_collection(COLLECTION)
+        .expect("the collection maps to a data group");
+    let group_leader = preferred_leader(probe, group_id);
+    let metadata_leader = probe.metadata_group_leader();
+    let lagging_index = cluster
+        .nodes
+        .iter()
+        .position(|node| node.node_id != group_leader && node.node_id != metadata_leader)
+        .expect("a third node leads neither group");
+    let lagging_id = cluster.nodes[lagging_index].node_id;
+    assert!(
+        cluster.nodes[lagging_index].hosts_data_group(group_id),
+        "node {lagging_id} replicates group {group_id}"
+    );
+
+    let stopped = cluster
+        .stop_member(lagging_index)
+        .await
+        .expect("stop the lagging replica");
+
+    // A live node takes the cut. Every group applies its barrier there.
+    let cutter = node_by_id(&cluster, group_leader);
+    let watermark = cutter.shared.hlc_clock.now().wall_ns;
+    nodedb::control::backup::cut::cut_at(&cutter.shared, watermark)
+        .await
+        .unwrap_or_else(|e| panic!("the cut on node {group_leader}: {e}"));
+    let barrier_floor = cutter.shared.applied_index_watcher(group_id).current();
+
+    // Compact the group's log past the barrier.
+    let mut written = 0;
+    while cutter.group_snapshot_index(group_id) <= barrier_floor {
+        assert!(
+            written < MAX_ROWS,
+            "group {group_id} did not compact past index {barrier_floor} after {written} \
+             writes: {}",
+            cutter.group_status_line(group_id)
+        );
+        cutter
+            .client
+            .simple_query(&format!(
+                "INSERT INTO {COLLECTION} (id, payload) VALUES ('r{written}', 'v')"
+            ))
+            .await
+            .unwrap_or_else(|e| panic!("insert r{written}: {e}"));
+        written += 1;
+        if written % 10 == 0 {
+            tokio::time::sleep(STEP_POLL).await;
+        }
+    }
+
+    cluster
+        .restart_member(stopped)
+        .await
+        .expect("restart the lagging replica");
+    let lagging = node_by_id(&cluster, lagging_id);
+    wait_for(
+        "the lagging replica installs a snapshot past the barrier",
+        STEP_DEADLINE,
+        STEP_POLL,
+        || lagging.local_snapshot_index_for_group(group_id) > barrier_floor,
+    )
+    .await;
+    cluster.wait_for_preferred_leaders(STEP_DEADLINE).await;
+    assert!(
+        !lagging.leads_group(group_id),
+        "node {lagging_id} leads group {group_id}: its own driver could place the barrier"
+    );
+
+    let started = Instant::now();
+    let cut = tokio::time::timeout(
+        CUT_BUDGET,
+        nodedb::control::backup::cut::cut_at(&lagging.shared, watermark),
+    )
+    .await
+    .unwrap_or_else(|_| {
+        panic!(
+            "node {lagging_id}: the cut at {watermark} still waits after {CUT_BUDGET:?}: {}",
+            lagging.group_status_line(group_id)
+        )
+    });
+    cut.unwrap_or_else(|e| panic!("node {lagging_id}: the cut at {watermark}: {e}"));
+    assert!(
+        !lagging.leads_group(group_id),
+        "node {lagging_id} led group {group_id} during the cut after {:?}",
+        started.elapsed()
+    );
+    cluster.shutdown().await;
+}

nodedb-cluster-tests/tests/common_suite/cases/calvin_multi_shard_redo_restart.rs

@@ -44,6 +44,18 @@ fn sequencer_leader(node: &TestClusterNode) -> u64 {
         .unwrap_or(0)
 }
 
+/// Whether `vshard_id`'s applied ledger on `node` holds an applied position.
+fn ledger_holds_a_position(node: &TestClusterNode, vshard_id: u32) -> bool {
+    node.shared
+        .calvin
+        .applied
+        .get(vshard_id)
+        .is_some_and(|ledger| {
+            ledger.max_applied_epoch()
+                != nodedb::control::cluster::calvin::scheduler::NOT_YET_APPLIED_EPOCH
+        })
+}
+
 /// Count of transactions the single-node sequencer has admitted to an epoch,
 /// or `0` if the sequencer metrics handle is not installed yet.
 fn admitted_total(node: &TestClusterNode) -> u64 {
@@ -206,6 +218,19 @@ async fn calvin_multi_shard_write_in_explicit_block_commits_and_survives_restart
         tokio::time::sleep(Duration::from_millis(25)).await;
     }
 
+    // The txn applied on the KV collection's vShard, so this node holds
+    // stored Calvin state for it.
+    let kv_vshard = nodedb_types::CollectionKey::from_bare(nodedb::types::DatabaseId::DEFAULT, &kv)
+        .vshard()
+        .as_u32();
+    wait_for(
+        "the KV vShard's applied ledger holds the txn",
+        Duration::from_secs(10),
+        Duration::from_millis(25),
+        || ledger_holds_a_position(&node, kv_vshard),
+    )
+    .await;
+
     // WAL-only restart: shut down cleanly (flushing the WAL, releasing every
     // redb handle) and reopen against the SAME directory — no vector
     // checkpoint, so the HNSW must be rebuilt purely from the committed
@@ -264,5 +289,34 @@ async fn calvin_multi_shard_write_in_explicit_block_commits_and_survives_restart
         tokio::time::sleep(Duration::from_millis(25)).await;
     }
 
+    // Boot queued the stored Calvin state of every vShard for a forget
+    // check. Once the data groups replayed, the reconcile pass keeps the
+    // state of each vShard this node still replicates.
+    wait_for(
+        "boot recovery replays the data groups",
+        Duration::from_secs(30),
+        Duration::from_millis(50),
+        || node.shared.data_groups_replayed.load(Ordering::Acquire),
+    )
+    .await;
+    wait_for(
+        "the reconcile pass settles every vShard queued at boot",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            node.shared
+                .calvin
+                .forget_pending
+                .lock()
+                .unwrap_or_else(|p| p.into_inner())
+                .is_empty()
+        },
+    )
+    .await;
+    assert!(
+        ledger_holds_a_position(&node, kv_vshard),
+        "a vShard this node replicates keeps its applied ledger across a restart"
+    );
+
     node.shutdown().await;
 }