feat(txn): serializable snapshot isolation for transaction blocks

.github/codeql/suppressions.json

@@ -165,6 +165,12 @@
       "path": "nodedb/src/control/security/keystore/wal_key.rs",
       "sink": "^\\s*(let\\ expected\\ =\\ nodedb_wal::crypto::WalEncryptionKey::with_epoch\\(\\&\\[0x24u8;\\ 32\\],\\ \\[0u8;\\ 4\\]\\);)\\s*$",
       "reason": "Test-only: inside #[cfg(test)] mod tests, a fixed key and epoch build the expected key for a round-trip check against a fixture key file. No production key or nonce comes from it."
+    },
+    {
+      "rule": "rust/hard-coded-cryptographic-value",
+      "path": "nodedb/src/control/security/credential/store/core.rs",
+      "sink": "^\\s*let (password_hash = hash_password_argon2\\(\"\", &Argon2Config::default\\(\\)\\)|scram_salted_password = compute_scram_salted_password\\(\"\", &salt\\);)\\s*$",
+      "reason": "Test-only: inside #[cfg(test)] mod tests, an empty password seeds a stored user so the test can check that CredentialStore::open refuses empty derived credentials. No production credential comes from it."
     }
   ]
 }

docs/architecture.md

@@ -144,7 +144,7 @@ Write transactions that touch multiple vShards go through the **Calvin sequencer
 
 **Interactive transactions are atomic across shards.** A `BEGIN ... COMMIT` block whose statements span multiple vShards (or nodes) flushes at COMMIT as one Calvin transaction bound by a durable **Vote/Verdict barrier**: each participant replicates a commit/abort vote through the sequencer log; when the tally completes, a replicated verdict commits or drops every participant's staged writes together. Expected-participant counts are seeded from the replicated epoch batch, so the barrier survives sequencer failover. Read-only participants count too — a transaction that wrote shard X but read shard Y validates and votes on both.
 
-**Serializable OCC.** Reads recorded during the transaction (point reads, predicate scans, index equality/range probes, both sides of gathered cross-node JOINs) are validated at COMMIT against per-key, per-collection, and per-index-value write LSNs. A stale read aborts the transaction with SQLSTATE `40001` (retryable) instead of committing silently.
+**Serializable snapshot transactions.** A transaction block reads one snapshot, taken at its first statement. Every read sees that snapshot plus the block's own writes. At COMMIT, at the moment the writes apply, NodeDB checks that nothing the transaction read or overwrote changed after its snapshot. The checks cover point reads, predicate scans, index equality/range probes, and both sides of gathered cross-node JOINs. A changed read aborts the transaction with SQLSTATE `40001`, and the client retries the whole block. A read-only block commits without validation and never fails with `40001`. A snapshot older than `[tuning.network] txn_snapshot_max_age_secs` (default 300) fails the next statement with `40001`. A write that cannot be buffered inside a block fails with SQLSTATE `25001`. Autocommit statements read the current state at a per-statement cut.
 
 **Durability.** A committed transaction's writes are persisted as a single atomically-replayable `TransactionRedo` WAL record appended before the commit is acknowledged. WAL-only recovery replays it in LSN order, rebuilding even in-memory secondary indexes (vector HNSW, FTS postings) that base storage cannot reconstruct alone.
 

docs/query-language.md

@@ -1114,7 +1114,15 @@ RELEASE SAVEPOINT sp1;       -- also: RELEASE sp1
 COMMIT;
 ```
 
-Isolation level: **Snapshot Isolation (SI)**. Reads see a consistent snapshot from `BEGIN` time. Write conflicts detected at `COMMIT` and surfaced as SQLSTATE `40001` (`could not serialize access due to concurrent update`).
+Isolation level: **serializable snapshot**. Every transaction block runs serializable, whatever level `BEGIN`, `START TRANSACTION`, `SET TRANSACTION` or `SET SESSION CHARACTERISTICS` names. `SERIALIZABLE`, `REPEATABLE READ`, `READ COMMITTED` and `READ UNCOMMITTED` are all accepted, and `SHOW transaction_isolation` reports `serializable`. `DEFERRABLE` is refused with SQLSTATE `0A000`.
+
+- **Snapshot.** A block takes one snapshot at its first statement, not at `BEGIN`. Every read in the block sees that snapshot plus the block's own writes.
+- **Validation.** `COMMIT` checks, at the moment the writes apply, that nothing the block read or overwrote changed after its snapshot. If something changed, `COMMIT` fails with SQLSTATE `40001` (`could not serialize access`). The client must retry the whole block.
+- **Read-only blocks.** A block that wrote nothing commits without validation and never fails with `40001`. A reader can page through a large result set inside one read-only block, and every page reads the same snapshot.
+- **Age limit.** A snapshot older than `[tuning.network] txn_snapshot_max_age_secs` (default `300`) fails the next statement with `40001`. Retry the block.
+- **Unbufferable writes.** A write that cannot be buffered inside a block is refused with SQLSTATE `25001`.
+- **Autocommit.** An autocommit statement reads the current state at a per-statement cut.
+- **Time travel.** `AS OF SYSTEM TIME` works on every collection.
 
 ### Cross-Shard Transactions
 

nodedb-array/src/codec/tag.rs

@@ -6,7 +6,7 @@
 // Any byte not recognized as a valid CodecTag is treated as corruption and
 // must be surfaced as a SegmentCorruption error by the caller.
 
-/// One-byte codec tag at the front of a new-format (v4+) tile payload.
+/// One-byte codec tag at the front of every tile payload.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 #[repr(u8)]
 pub enum CodecTag {
@@ -33,10 +33,10 @@ impl CodecTag {
 
 /// Peek at the first byte of a tile payload and return the codec tag.
 ///
-/// Returns `Some(tag)` for a recognized new-format tag byte.
-/// Returns `None` for an empty payload or any unrecognized byte (including
-/// former msgpack map-start bytes 0x80–0x8f, 0xde, 0xdf) — callers must
-/// surface a `SegmentCorruption` error.
+/// Returns `Some(tag)` for a recognized tag byte.
+/// Returns `None` for an empty payload or any unrecognized byte, msgpack
+/// map-start bytes 0x80–0x8f, 0xde and 0xdf included. Callers must surface a
+/// `SegmentCorruption` error.
 pub fn peek_tag(payload: &[u8]) -> Option<CodecTag> {
     let first = *payload.first()?;
     CodecTag::from_byte(first)
@@ -65,9 +65,8 @@ mod tests {
     }
 
     #[test]
-    fn peek_tag_former_msgpack_bytes_return_none() {
-        // Bytes 0x80..=0x8f, 0xde, 0xdf were formerly treated as legacy msgpack
-        // map-start markers. They are now unknown tag bytes and must return None.
+    fn peek_tag_msgpack_map_bytes_return_none() {
+        // Msgpack map-start bytes 0x80..=0x8f, 0xde, 0xdf are no tag bytes.
         for b in 0x80u8..=0x8fu8 {
             let payload = [b, 0x00];
             assert_eq!(peek_tag(&payload), None, "byte {b:#04x} should be None");

nodedb-array/src/codec/tile_decode.rs

@@ -3,9 +3,8 @@
 // Structural sparse-tile decoder — symmetric to tile_encode.
 //
 // Validates the version byte, dispatches on CodecTag, and reconstructs
-// a SparseTile from the structural payload. Legacy (v3) msgpack tiles are
-// NOT handled here; the segment reader peeks the tag and dispatches to
-// zerompk for those.
+// a SparseTile from the structural payload. A payload whose first byte is
+// no CodecTag is corruption.
 
 use crate::codec::attr_col::decode_attr_col;
 use crate::codec::column_codec::{decode_row_kinds, decode_surrogates, decode_timestamps_col};
@@ -73,15 +72,11 @@ fn decode_sparse_tile_inner(
         });
     }
 
-    let tag_result = peek_tag(payload).ok_or_else(|| {
-        // peek_tag returns None for both legacy msgpack and unknown bytes.
-        // The reader should never call us with a legacy payload.
-        ArrayError::SegmentCorruption {
-            detail: format!(
-                "decode_sparse_tile called with legacy or unknown tag byte: {:#04x}",
-                payload[0]
-            ),
-        }
+    let tag_result = peek_tag(payload).ok_or_else(|| ArrayError::SegmentCorruption {
+        detail: format!(
+            "sparse tile payload has unknown tag byte: {:#04x}",
+            payload[0]
+        ),
     })?;
 
     let version = payload[1];

nodedb-array/src/query/retention.rs

@@ -349,7 +349,7 @@ mod tests {
     fn build_segment(pairs: Vec<(TileId, SparseTile)>) -> Vec<u8> {
         let mut w = SegmentWriter::new(0xBEEF);
         for (id, tile) in pairs {
-            w.append_sparse(id, &tile).unwrap();
+            w.append_sparse(id, 0, &tile).unwrap();
         }
         w.finish(None).unwrap()
     }

nodedb-array/src/segment/format/footer.rs

@@ -109,6 +109,7 @@ mod tests {
             0xABCDEF,
             vec![TileEntry::new(
                 TileId::snapshot(7),
+                3,
                 TileKind::Sparse,
                 32,
                 64,

nodedb-array/src/segment/format/header.rs

@@ -21,7 +21,7 @@ use crate::error::{ArrayError, ArrayResult};
 /// `b"NDAS\0\0\0\1"` — NodeDB Array Segment, version slot in last byte.
 pub const HEADER_MAGIC: [u8; 8] = *b"NDAS\0\0\0\x01";
 
-/// On-disk format version. Bump on layout-incompatible changes.
+/// On-disk format version. A segment of any other version is refused.
 pub const FORMAT_VERSION: u16 = 1;
 
 pub const HEADER_SIZE: usize = 24;
@@ -135,7 +135,7 @@ mod tests {
     }
 
     #[test]
-    fn header_rejects_v2_segment() {
+    fn header_rejects_a_segment_of_another_version() {
         let mut buf = Vec::new();
         buf.extend_from_slice(&HEADER_MAGIC);
         buf.extend_from_slice(&2u16.to_le_bytes());

nodedb-array/src/segment/format/tile_entry.rs

@@ -46,16 +46,28 @@ pub enum TileKind {
 )]
 pub struct TileEntry {
     pub tile_id: TileId,
+    /// The placement the writer stored the tile version under. A store that
+    /// hosts several shards tags each version with the shard whose write
+    /// stored it. A store with one placement writes `0`.
+    pub placement: u32,
     pub kind: TileKind,
     pub offset: u64,
     pub length: u32,
     pub mbr: TileMBR,
 }
 
 impl TileEntry {
-    pub fn new(tile_id: TileId, kind: TileKind, offset: u64, length: u32, mbr: TileMBR) -> Self {
+    pub fn new(
+        tile_id: TileId,
+        placement: u32,
+        kind: TileKind,
+        offset: u64,
+        length: u32,
+        mbr: TileMBR,
+    ) -> Self {
         Self {
             tile_id,
+            placement,
             kind,
             offset,
             length,
@@ -71,7 +83,7 @@ mod tests {
     #[test]
     fn tile_entry_round_trip_msgpack() {
         let mbr = TileMBR::new(0, 0);
-        let e = TileEntry::new(TileId::snapshot(42), TileKind::Sparse, 100, 256, mbr);
+        let e = TileEntry::new(TileId::snapshot(42), 7, TileKind::Sparse, 100, 256, mbr);
         let bytes = zerompk::to_msgpack_vec(&e).unwrap();
         let d: TileEntry = zerompk::from_msgpack(&bytes).unwrap();
         assert_eq!(d, e);

nodedb-array/src/segment/mbr_index/build.rs

@@ -146,7 +146,7 @@ mod tests {
             nnz: 1,
             attr_stats: vec![AttrStats::AllNull { null_count: 0 }],
         };
-        TileEntry::new(TileId::snapshot(tid), TileKind::Sparse, 0, 0, mbr)
+        TileEntry::new(TileId::snapshot(tid), 0, TileKind::Sparse, 0, 0, mbr)
     }
 
     fn pred(lo: i64, hi: i64) -> MbrQueryPredicate {

nodedb-array/src/segment/reader.rs

@@ -449,9 +449,9 @@ mod tests {
     fn reader_round_trips_sparse_tiles() {
         let s = schema();
         let mut w = SegmentWriter::new(0xCAFE);
-        w.append_sparse(TileId::snapshot(1), &make_sparse(&s, 1))
+        w.append_sparse(TileId::snapshot(1), 0, &make_sparse(&s, 1))
             .unwrap();
-        w.append_sparse(TileId::snapshot(2), &make_sparse(&s, 2))
+        w.append_sparse(TileId::snapshot(2), 0, &make_sparse(&s, 2))
             .unwrap();
         let bytes = w.finish(None).unwrap();
         let r = SegmentReader::open(&bytes).unwrap();
@@ -467,7 +467,7 @@ mod tests {
     fn reader_round_trips_dense_tile() {
         let s = schema();
         let mut w = SegmentWriter::new(0xBEEF);
-        w.append_dense(TileId::snapshot(1), &DenseTile::empty(&s))
+        w.append_dense(TileId::snapshot(1), 0, &DenseTile::empty(&s))
             .unwrap();
         let bytes = w.finish(None).unwrap();
         let r = SegmentReader::open(&bytes).unwrap();
@@ -486,7 +486,7 @@ mod tests {
         // CRC path which guards the same invariant.)
         let s = schema();
         let mut w = SegmentWriter::new(0x1);
-        w.append_sparse(TileId::snapshot(1), &make_sparse(&s, 1))
+        w.append_sparse(TileId::snapshot(1), 0, &make_sparse(&s, 1))
             .unwrap();
         let mut bytes = w.finish(None).unwrap();
         bytes[12] ^= 0xFF; // corrupt header schema_hash
@@ -497,7 +497,7 @@ mod tests {
     fn reader_rejects_out_of_range_tile() {
         let s = schema();
         let mut w = SegmentWriter::new(0x1);
-        w.append_sparse(TileId::snapshot(1), &make_sparse(&s, 1))
+        w.append_sparse(TileId::snapshot(1), 0, &make_sparse(&s, 1))
             .unwrap();
         let bytes = w.finish(None).unwrap();
         let r = SegmentReader::open(&bytes).unwrap();
@@ -509,11 +509,11 @@ mod tests {
         let s = schema();
         let mut w = SegmentWriter::new(0xCAFE);
         // Three versions of prefix=1 at system_from_ms 100, 200, 300.
-        w.append_sparse(TileId::new(1, 100), &make_sparse(&s, 1))
+        w.append_sparse(TileId::new(1, 100), 0, &make_sparse(&s, 1))
             .unwrap();
-        w.append_sparse(TileId::new(1, 200), &make_sparse(&s, 2))
+        w.append_sparse(TileId::new(1, 200), 0, &make_sparse(&s, 2))
             .unwrap();
-        w.append_sparse(TileId::new(1, 300), &make_sparse(&s, 3))
+        w.append_sparse(TileId::new(1, 300), 0, &make_sparse(&s, 3))
             .unwrap();
         let bytes = w.finish(None).unwrap();
         let r = SegmentReader::open(&bytes).unwrap();
@@ -534,7 +534,7 @@ mod tests {
     fn read_tile_as_of_returns_none_below_first_version() {
         let s = schema();
         let mut w = SegmentWriter::new(0xBEEF);
-        w.append_sparse(TileId::new(1, 100), &make_sparse(&s, 1))
+        w.append_sparse(TileId::new(1, 100), 0, &make_sparse(&s, 1))
             .unwrap();
         let bytes = w.finish(None).unwrap();
         let r = SegmentReader::open(&bytes).unwrap();
@@ -587,11 +587,11 @@ mod tests {
     fn iter_tile_versions_newest_first() {
         let s = schema();
         let mut w = SegmentWriter::new(0xCAFE);
-        w.append_sparse(TileId::new(1, 100), &make_sparse(&s, 1))
+        w.append_sparse(TileId::new(1, 100), 0, &make_sparse(&s, 1))
             .unwrap();
-        w.append_sparse(TileId::new(1, 200), &make_sparse(&s, 2))
+        w.append_sparse(TileId::new(1, 200), 0, &make_sparse(&s, 2))
             .unwrap();
-        w.append_sparse(TileId::new(1, 300), &make_sparse(&s, 3))
+        w.append_sparse(TileId::new(1, 300), 0, &make_sparse(&s, 3))
             .unwrap();
         let bytes = w.finish(None).unwrap();
         let r = SegmentReader::open(&bytes).unwrap();
@@ -607,11 +607,11 @@ mod tests {
     fn iter_tile_versions_respects_system_as_of() {
         let s = schema();
         let mut w = SegmentWriter::new(0xBEEF);
-        w.append_sparse(TileId::new(1, 100), &make_sparse(&s, 1))
+        w.append_sparse(TileId::new(1, 100), 0, &make_sparse(&s, 1))
             .unwrap();
-        w.append_sparse(TileId::new(1, 200), &make_sparse(&s, 2))
+        w.append_sparse(TileId::new(1, 200), 0, &make_sparse(&s, 2))
             .unwrap();
-        w.append_sparse(TileId::new(1, 300), &make_sparse(&s, 3))
+        w.append_sparse(TileId::new(1, 300), 0, &make_sparse(&s, 3))
             .unwrap();
         let bytes = w.finish(None).unwrap();
         let r = SegmentReader::open(&bytes).unwrap();
@@ -715,11 +715,11 @@ mod tests {
     fn read_tile_as_of_finds_exact_match() {
         let s = schema();
         let mut w = SegmentWriter::new(0xDEAD);
-        w.append_sparse(TileId::new(1, 100), &make_sparse(&s, 1))
+        w.append_sparse(TileId::new(1, 100), 0, &make_sparse(&s, 1))
             .unwrap();
-        w.append_sparse(TileId::new(1, 200), &make_sparse(&s, 2))
+        w.append_sparse(TileId::new(1, 200), 0, &make_sparse(&s, 2))
             .unwrap();
-        w.append_sparse(TileId::new(1, 300), &make_sparse(&s, 3))
+        w.append_sparse(TileId::new(1, 300), 0, &make_sparse(&s, 3))
             .unwrap();
         let bytes = w.finish(None).unwrap();
         let r = SegmentReader::open(&bytes).unwrap();
@@ -740,14 +740,14 @@ mod tests {
 
     fn write_plain(s: &crate::schema::ArraySchema, id: TileId) -> Vec<u8> {
         let mut w = SegmentWriter::new(0xCAFE);
-        w.append_sparse(id, &make_sparse(s, 1)).unwrap();
+        w.append_sparse(id, 0, &make_sparse(s, 1)).unwrap();
         w.finish(None).unwrap()
     }
 
     fn write_encrypted(s: &crate::schema::ArraySchema, id: TileId) -> Vec<u8> {
         let kek = test_kek();
         let mut w = SegmentWriter::new(0xCAFE);
-        w.append_sparse(id, &make_sparse(s, 1)).unwrap();
+        w.append_sparse(id, 0, &make_sparse(s, 1)).unwrap();
         w.finish(Some(&kek)).unwrap()
     }
 
@@ -801,9 +801,9 @@ mod tests {
         let s = schema();
         let kek = test_kek();
         let mut w = SegmentWriter::new(0x1234);
-        w.append_sparse(TileId::new(1, 100), &make_sparse(&s, 1))
+        w.append_sparse(TileId::new(1, 100), 0, &make_sparse(&s, 1))
             .unwrap();
-        w.append_sparse(TileId::new(2, 200), &make_sparse(&s, 2))
+        w.append_sparse(TileId::new(2, 200), 0, &make_sparse(&s, 2))
             .unwrap();
         let encrypted = w.finish(Some(&kek)).unwrap();
         let owned = OwnedSegmentReader::open_with_kek(&encrypted, Some(&kek)).unwrap();