Fix Calvin dispatch capacity, typed errors across planes, and backup/restart gaps

.config/nextest.toml

@@ -147,7 +147,7 @@ slow-timeout = { period = "30s", terminate-after = 8 }
 # hide the cause rather than fix it. The tail this costs is bounded — roughly a
 # dozen crash/shutdown tests at about ten seconds each.
 [[profile.default.overrides]]
-filter = 'binary(wal_direct_io) | binary(ilp_client_address) | binary(crash_recovery) | binary(crash_recovery_overlays) | binary(crash_recovery_analytics) | binary(crash_resp_kv_write) | binary(crash_metadata_applier_wedge) | binary(crash_dropped_collection_reclaim) | binary(crash_mid_replay) | binary(crash_checkpoint_corruption) | binary(crash_checkpoint_truncate_window) | binary(crash_refused_write_not_resurrected) | binary(crash_replay_fail_stop) | binary(crash_core_stall) | test(/^cases::startup_failure::/) | test(/^cases::shutdown_in_flight::/) | test(/^cases::shutdown_budget::/) | test(/^cases::shutdown_abort_offender::/) | test(/^cases::shutdown_idempotent::/)'
+filter = 'binary(wal_direct_io) | binary(ilp_client_address) | binary(crash_recovery) | binary(crash_recovery_overlays) | binary(crash_recovery_analytics) | binary(crash_resp_kv_write) | binary(crash_metadata_applier_wedge) | binary(crash_dropped_collection_reclaim) | binary(crash_purge_not_resurrected) | binary(crash_mid_replay) | binary(crash_checkpoint_corruption) | binary(crash_checkpoint_truncate_window) | binary(crash_refused_write_not_resurrected) | binary(crash_replay_fail_stop) | binary(crash_core_stall) | binary(crash_replay_stamp) | binary(crash_replay_stamp_calvin) | binary(calvin_hold_liveness) | binary(apply_pipeline_group_independence) | binary(crash_kv_atomic_autocommit) | test(/^cases::startup_failure::/) | test(/^cases::shutdown_in_flight::/) | test(/^cases::shutdown_budget::/) | test(/^cases::shutdown_abort_offender::/) | test(/^cases::shutdown_idempotent::/)'
 test-group = 'server-process'
 threads-required = 'num-test-threads'
 

docs/kv.md

@@ -155,10 +155,21 @@ SELECT KV_GETSET('session_token', 'player-123', 'new-token-xyz');
 
 **RESP (Redis) equivalents:** `INCR`, `DECR`, `INCRBY`, `DECRBY`, `INCRBYFLOAT`, `GETSET` — all work over the RESP protocol.
 
+**Value shapes:**
+
+- Raw value (a single `value` column, or RESP `SET`): the value is a byte string. `INCR`/`INCRBY`/`DECR`/`DECRBY` read it as decimal integer text and store the result as decimal text. `INCRBYFLOAT` and `KV_INCR_FLOAT` read decimal text (plain or exponent form, such as `5.0e3`), add exactly, and store the trimmed decimal text: `"0.1"` plus `0.2` stores `"0.3"`, `"3.0"` plus `0` stores `"3"`. Exact addition covers 28 significant digits below 7.9e28. A value outside that range adds in 64-bit float. `SET k 5` then `INCR k` leaves `"6"`.
+- Absent key: the counter starts at 0 and is stored as decimal text.
+- Typed row (several columns): the first numeric column in key order moves. Every other column stays.
+
 **Error handling:**
 
-- `TYPE_MISMATCH` (SQLSTATE 42846) — INCR on a non-numeric value
-- `OVERFLOW` (SQLSTATE 22003) — i64 overflow on INCR
+| Condition | RESP reply | SQLSTATE |
+|---|---|---|
+| Raw value is not a decimal integer in the i64 range | `ERR value is not an integer or out of range` | `22P02` |
+| Raw value is not a decimal float | `ERR value is not a valid float` | `22P02` |
+| Integer result leaves the i64 range | `ERR increment or decrement would overflow` | `22003` |
+| Float result is NaN or infinite | `ERR increment would produce NaN or Infinity` | `22003` |
+| Typed row has no numeric column | `WRONGTYPE ...` | `42846` |
 
 ## Sorted Indexes (Leaderboards)
 

nodedb-client/src/native/connection/response.rs

@@ -40,10 +40,16 @@ fn error_frame_to_typed(
     if payload.ndb_code == 0 {
         return NodeDbError::internal(payload.message.clone());
     }
-    NodeDbError::from_wire(
+    let error = NodeDbError::from_wire_with_details(
         nodedb_types::error::ErrorCode(payload.ndb_code),
         payload.message.clone(),
-    )
+        payload.details.clone(),
+    );
+    // The typed cause, when the server sent one, becomes the error's cause.
+    match &payload.cause {
+        Some(cause) => error.with_cause(cause.to_error()),
+        None => error,
+    }
 }
 
 pub(super) fn response_to_query_result(resp: NativeResponse) -> NodeDbResult<QueryResult> {

nodedb-cluster-tests/Cargo.toml

@@ -10,6 +10,10 @@ description = "3-node integration test suite for NodeDB. Heavy; runs in its own
 [lib]
 path = "src/lib.rs"
 
+[features]
+# Arms the in-process fail points the cluster tests park writes at.
+failpoints = ["nodedb/failpoints", "nodedb-types/failpoints"]
+
 [dependencies]
 
 [dev-dependencies]

nodedb-cluster-tests/tests/cluster_common/calvin_test_node.rs

@@ -27,6 +27,8 @@
 //!      for shutdown.
 //!    - `add_vshard_sender(vshard_id, sender)` — wire a per-vshard channel
 //!      into the state machine so tests can assert fan-out.
+//!    - `try_recv_txn(rx)` — read the next sequenced txn a fan-out channel
+//!      holds.
 
 #![allow(dead_code)] // Not every test file uses every helper.
 
@@ -130,24 +132,15 @@ impl CalvinTestNode {
 
     /// Register a per-vshard output sender so the state machine can
     /// fan out sequenced transactions to the receiving test code.
-    pub fn add_vshard_sender(&self, vshard_id: u32, sender: mpsc::Sender<SequencedTxn>) {
-        // The state machine now fans out `SchedulerInput`; these tests assert on
-        // the sequenced-txn stream, so adapt: forward only `Txn` payloads to the
-        // caller's `SequencedTxn` channel (reservation inputs don't occur here).
-        let (adapt_tx, mut adapt_rx) = mpsc::channel::<SchedulerInput>(512);
-        tokio::spawn(async move {
-            while let Some(input) = adapt_rx.recv().await {
-                if let SchedulerInput::Txn(txn) = input
-                    && sender.send(txn).await.is_err()
-                {
-                    break;
-                }
-            }
-        });
+    ///
+    /// The state machine sends into `sender` itself, inside `apply`, before
+    /// it advances `last_applied_epoch`. A test that waits for the epoch can
+    /// then read the fan-out with `try_recv_txn` and no further wait.
+    pub fn add_vshard_sender(&self, vshard_id: u32, sender: mpsc::Sender<SchedulerInput>) {
         self.state_machine
             .lock()
             .unwrap_or_else(|p| p.into_inner())
-            .set_vshard_sender(vshard_id, adapt_tx);
+            .set_vshard_sender(vshard_id, sender);
     }
 
     /// Start the sequencer service epoch-ticker task on this node.
@@ -417,3 +410,14 @@ pub async fn wait_for_sequencer_leader(
         tokio::time::sleep(step).await;
     }
 }
+
+/// The next sequenced txn `rx` holds, skipping any other scheduler input.
+/// `None` when the channel holds no txn.
+pub fn try_recv_txn(rx: &mut mpsc::Receiver<SchedulerInput>) -> Option<SequencedTxn> {
+    while let Ok(input) = rx.try_recv() {
+        if let SchedulerInput::Txn(txn) = input {
+            return Some(txn);
+        }
+    }
+    None
+}

nodedb-cluster-tests/tests/cluster_common/mod.rs

@@ -7,6 +7,6 @@ pub mod rebalancer;
 pub mod test_node;
 
 pub use calvin_test_node::{
-    CalvinApplier, CalvinTestNode, spawn_with_sequencer, wait_for_sequencer_leader,
+    CalvinApplier, CalvinTestNode, spawn_with_sequencer, try_recv_txn, wait_for_sequencer_leader,
 };
 pub use test_node::{NoopApplier, TestNode, test_transport, wait_for};

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

@@ -19,22 +19,21 @@ use std::time::Duration;
 
 use nodedb_cluster::calvin::{
     sequencer::{SequencerConfig, new_inbox},
-    types::{EngineKeySet, ReadWriteSet, SequencedTxn, SortedVec, TxClass, VersionedReadSet},
-};
-use nodedb_types::{
-    TenantId,
-    id::{DatabaseId, VShardId},
+    types::{EngineKeySet, ReadWriteSet, SchedulerInput, SortedVec, TxClass, VersionedReadSet},
 };
+use nodedb_types::{TenantId, id::DatabaseId};
 use tokio::sync::mpsc;
 
-use super::cluster_common::{spawn_with_sequencer, wait_for_sequencer_leader};
+use super::cluster_common::{spawn_with_sequencer, try_recv_txn, wait_for_sequencer_leader};
 
 /// Find two collection names that hash to distinct vshards.
 fn two_distinct_collections() -> (String, String) {
     let mut first: Option<(String, u32)> = None;
     for i in 0u32..512 {
         let name = format!("col_{i}");
-        let vshard = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &name).as_u32();
+        let vshard = nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, &name)
+            .vshard()
+            .as_u32();
         if let Some((ref fname, fv)) = first {
             if fv != vshard {
                 return (fname.clone(), name);
@@ -48,8 +47,12 @@ fn two_distinct_collections() -> (String, String) {
 
 fn make_multishard_txclass() -> (TxClass, u32, u32) {
     let (col_a, col_b) = two_distinct_collections();
-    let va = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &col_a).as_u32();
-    let vb = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &col_b).as_u32();
+    let va = nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, &col_a)
+        .vshard()
+        .as_u32();
+    let vb = nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, &col_b)
+        .vshard()
+        .as_u32();
     let write_set = ReadWriteSet::new(vec![
         EngineKeySet::Document {
             collection: col_a,
@@ -92,11 +95,15 @@ async fn sequencer_normal_path_commit_on_all_replicas() {
 
     // Wire per-vshard receivers on every node.
     let (tx_a, col_b_name) = two_distinct_collections();
-    let va = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &tx_a).as_u32();
-    let vb = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &col_b_name).as_u32();
-
-    let mut vshard_rxs_a: Vec<mpsc::Receiver<SequencedTxn>> = Vec::new();
-    let mut vshard_rxs_b: Vec<mpsc::Receiver<SequencedTxn>> = Vec::new();
+    let va = nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, &tx_a)
+        .vshard()
+        .as_u32();
+    let vb = nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, &col_b_name)
+        .vshard()
+        .as_u32();
+
+    let mut vshard_rxs_a: Vec<mpsc::Receiver<SchedulerInput>> = Vec::new();
+    let mut vshard_rxs_b: Vec<mpsc::Receiver<SchedulerInput>> = Vec::new();
     for node in &nodes {
         let (tx_a_ch, rx_a) = mpsc::channel(64);
         let (tx_b_ch, rx_b) = mpsc::channel(64);
@@ -149,8 +156,8 @@ async fn sequencer_normal_path_commit_on_all_replicas() {
         .zip(vshard_rxs_b.iter_mut())
         .enumerate()
     {
-        let got_a = rx_a.try_recv().is_ok();
-        let got_b = rx_b.try_recv().is_ok();
+        let got_a = try_recv_txn(rx_a).is_some();
+        let got_b = try_recv_txn(rx_b).is_some();
         assert!(
             got_a || got_b,
             "node {}: neither vshard receiver got the txn fan-out",

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

@@ -43,23 +43,22 @@ use std::time::Duration;
 
 use nodedb_cluster::calvin::{
     sequencer::{SequencerConfig, new_inbox},
-    types::{EngineKeySet, ReadWriteSet, SequencedTxn, SortedVec, TxClass, VersionedReadSet},
-};
-use nodedb_types::{
-    TenantId,
-    id::{DatabaseId, VShardId},
+    types::{EngineKeySet, ReadWriteSet, SchedulerInput, SortedVec, TxClass, VersionedReadSet},
 };
+use nodedb_types::{TenantId, id::DatabaseId};
 use tokio::sync::mpsc;
 
-use super::cluster_common::{spawn_with_sequencer, wait_for_sequencer_leader};
+use super::cluster_common::{spawn_with_sequencer, try_recv_txn, wait_for_sequencer_leader};
 
 // ── Helpers ──────────────────────────────────────────────────────────────────
 
 fn two_distinct_collections() -> (String, String) {
     let mut first: Option<(String, u32)> = None;
     for i in 0u32..512 {
         let name = format!("col_{i}");
-        let vshard = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &name).as_u32();
+        let vshard = nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, &name)
+            .vshard()
+            .as_u32();
         if let Some((ref fname, fv)) = first {
             if fv != vshard {
                 return (fname.clone(), name);
@@ -130,11 +129,15 @@ async fn scheduler_catchup_via_raft_log_replay() {
 
     // Wire per-vshard receivers on every node so we can verify fan-out.
     let (col_a, col_b) = two_distinct_collections();
-    let va = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &col_a).as_u32();
-    let vb = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &col_b).as_u32();
+    let va = nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, &col_a)
+        .vshard()
+        .as_u32();
+    let vb = nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, &col_b)
+        .vshard()
+        .as_u32();
 
-    let mut vshard_rxs_a: Vec<mpsc::Receiver<SequencedTxn>> = Vec::new();
-    let mut vshard_rxs_b: Vec<mpsc::Receiver<SequencedTxn>> = Vec::new();
+    let mut vshard_rxs_a: Vec<mpsc::Receiver<SchedulerInput>> = Vec::new();
+    let mut vshard_rxs_b: Vec<mpsc::Receiver<SchedulerInput>> = Vec::new();
     for node in &nodes {
         let (tx_a, rx_a) = mpsc::channel(128);
         let (tx_b, rx_b) = mpsc::channel(128);
@@ -246,7 +249,7 @@ async fn scheduler_catchup_via_raft_log_replay() {
     // Drain whatever arrived — we care that the routing worked, not the count.
     let mut total_received = 0usize;
     for rx in vshard_rxs_a.iter_mut().chain(vshard_rxs_b.iter_mut()) {
-        while rx.try_recv().is_ok() {
+        while try_recv_txn(rx).is_some() {
             total_received += 1;
         }
     }

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

@@ -37,15 +37,12 @@ use std::time::Duration;
 
 use nodedb_cluster::calvin::{
     sequencer::{SequencerConfig, new_inbox},
-    types::{EngineKeySet, ReadWriteSet, SequencedTxn, SortedVec, TxClass, VersionedReadSet},
-};
-use nodedb_types::{
-    TenantId,
-    id::{DatabaseId, VShardId},
+    types::{EngineKeySet, ReadWriteSet, SchedulerInput, SortedVec, TxClass, VersionedReadSet},
 };
+use nodedb_types::{TenantId, id::DatabaseId};
 use tokio::sync::mpsc;
 
-use super::cluster_common::{spawn_with_sequencer, wait_for_sequencer_leader};
+use super::cluster_common::{spawn_with_sequencer, try_recv_txn, wait_for_sequencer_leader};
 
 /// Find two collection names that hash to distinct vshards.
 ///
@@ -54,7 +51,9 @@ fn two_distinct_vshard_collections() -> (String, String) {
     let mut first: Option<(String, u32)> = None;
     for i in 0u32..512 {
         let name = format!("ollp_col_{i}");
-        let vshard = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &name).as_u32();
+        let vshard = nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, &name)
+            .vshard()
+            .as_u32();
         if let Some((ref fname, fv)) = first {
             if fv != vshard {
                 return (fname.clone(), name);
@@ -104,14 +103,17 @@ fn make_ollp_tx_class(
     .expect("valid multi-vshard OLLP TxClass")
 }
 
-/// Assert: one specific vshard channel received at least one SequencedTxn.
+/// Assert: one specific vshard channel received at least one sequenced txn.
+///
+/// The state machine sends into the channel inside `apply`, before it
+/// advances the epoch the caller waited for, so the txn is already there.
 fn assert_fan_out_received(
-    rx: &mut mpsc::Receiver<SequencedTxn>,
+    rx: &mut mpsc::Receiver<SchedulerInput>,
     vshard_id: u32,
     replica_idx: usize,
 ) {
     assert!(
-        rx.try_recv().is_ok(),
+        try_recv_txn(rx).is_some(),
         "replica {replica_idx}: vshard {vshard_id} fan-out channel received no txn"
     );
 }
@@ -134,13 +136,16 @@ async fn ollp_bulk_update_txclass_admitted_and_fanned_out() {
 
     // Find two collections that hash to distinct vshards.
     let (col_static, col_ollp) = two_distinct_vshard_collections();
-    let vs_static =
-        VShardId::from_collection_in_database(DatabaseId::DEFAULT, &col_static).as_u32();
-    let vs_ollp = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &col_ollp).as_u32();
+    let vs_static = nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, &col_static)
+        .vshard()
+        .as_u32();
+    let vs_ollp = nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, &col_ollp)
+        .vshard()
+        .as_u32();
 
     // Wire per-vshard fan-out receivers on every replica.
-    let mut rxs_static: Vec<mpsc::Receiver<SequencedTxn>> = Vec::new();
-    let mut rxs_ollp: Vec<mpsc::Receiver<SequencedTxn>> = Vec::new();
+    let mut rxs_static: Vec<mpsc::Receiver<SchedulerInput>> = Vec::new();
+    let mut rxs_ollp: Vec<mpsc::Receiver<SchedulerInput>> = Vec::new();
     for node in &nodes {
         let (tx_s, rx_s) = mpsc::channel(64);
         let (tx_o, rx_o) = mpsc::channel(64);
@@ -191,8 +196,8 @@ async fn ollp_bulk_update_txclass_admitted_and_fanned_out() {
 
     // Simulate an OLLP retry: concurrent insert added surrogate 4 to col_ollp.
     // Re-wire fresh fan-out receivers and re-submit with the corrected set.
-    let mut retry_rxs_static: Vec<mpsc::Receiver<SequencedTxn>> = Vec::new();
-    let mut retry_rxs_ollp: Vec<mpsc::Receiver<SequencedTxn>> = Vec::new();
+    let mut retry_rxs_static: Vec<mpsc::Receiver<SchedulerInput>> = Vec::new();
+    let mut retry_rxs_ollp: Vec<mpsc::Receiver<SchedulerInput>> = Vec::new();
     for node in &nodes {
         let (tx_s, rx_s) = mpsc::channel(64);
         let (tx_o, rx_o) = mpsc::channel(64);