Serializable Snapshot Isolation (SSI)

.config/nextest.toml

@@ -0,0 +1,2 @@
+[profile.default]
+slow-timeout = "1m"

.github/workflows/test.yml

@@ -65,8 +65,12 @@ jobs:
         run: cargo test --features __internal_whitebox -- whitebox_ --test-threads=1
       - name: Run tests
         run: cargo nextest run --features lz4,miniz,single_writer_tx,bloom
+      - name: Run SSI tests
+        run: cargo nextest run --no-default-features --features ssi_tx tx_ssi_
       - name: Run doc tests
         run: cargo test --doc
+      - name: Run SSI doc tests
+        run: cargo test --no-default-features --features ssi_tx --doc
       - name: Build & test examples
         run: node compile_examples.mjs
   cross:

Cargo.toml

@@ -22,6 +22,7 @@ lz4 = ["lsm-tree/lz4"]
 miniz = ["lsm-tree/miniz"]
 bloom = ["lsm-tree/bloom"]
 single_writer_tx = []
+ssi_tx = []
 __internal_whitebox = []
 
 [dependencies]
@@ -42,6 +43,7 @@ rand = "0.8.5"
 
 [package.metadata.cargo-all-features]
 denylist = ["__internal_whitebox"]
+skip_feature_sets = [["ssi_tx", "single_writer_tx"]]
 
 [[bench]]
 name = "lsmt"

README.md

@@ -20,7 +20,7 @@ Fjall is an LSM-based embeddable key-value storage engine written in Rust. It fe
 - Automatic background maintenance
 - Partitions (a.k.a. column families) with cross-partition atomic semantics
 - Built-in compression (default = LZ4)
-- Single-writer, multi-reader transactions (optional)
+- Serializable transactions (optional)
 - Key-value separation for large blob use cases (optional)
 
 Each `Keyspace` is a single logical database and is split into `partitions` (a.k.a. column families) - you should probably only use a single keyspace for your application. Each partition is physically a single LSM-tree and its own logical collection (a persistent, sorted map); however, write operations across partitions are atomic as they are persisted in a single keyspace-level journal, which will be recovered on restart.
@@ -117,6 +117,13 @@ Allows opening a transactional Keyspace for single-writer (serialized) transacti
 
 *Enabled by default.*
 
+### ssi_tx
+
+Allows opening a transactional Keyspace for multi-writer, serializable transactions, allowing RYOW (read-your-own-write), fetch-and-update and other atomic operations.
+Conflict checking is done using optimistic concurrency control.
+
+*Disabled by default.*
+
 ## Stable disk format
 
 The disk format is stable as of 1.0.0.

examples/tx-ssi-cc/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "tx-ssi-cc"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+fjall = { path = "../../", default-features = false, features = [
+  "bloom",
+  "lz4",
+  "ssi_tx",
+] }

examples/tx-ssi-cc/README.md

@@ -0,0 +1,3 @@
+# tx-ssi-cc
+
+This example demonstrates concurrent transactions using SSI (serializable snapshot isolation).

examples/tx-ssi-cc/src/main.rs

@@ -0,0 +1,53 @@
+use std::time::{Duration, Instant};
+
+fn main() -> fjall::Result<()> {
+    let keyspace = fjall::Config::default()
+        .temporary(true)
+        .open_transactional()?;
+    let items = keyspace.open_partition("items", Default::default())?;
+
+    let start = Instant::now();
+
+    let t1 = {
+        let keyspace = keyspace.clone();
+        let items = items.clone();
+
+        std::thread::spawn(move || {
+            let mut wtx = keyspace.write_tx().unwrap();
+            println!("Started tx1");
+            std::thread::sleep(Duration::from_secs(3));
+            wtx.insert(&items, "a", "a");
+            wtx.commit()
+        })
+    };
+
+    let t2 = {
+        let keyspace = keyspace.clone();
+        let items = items.clone();
+
+        std::thread::spawn(move || {
+            let mut wtx = keyspace.write_tx().unwrap();
+            println!("Started tx2");
+            std::thread::sleep(Duration::from_secs(3));
+            wtx.insert(&items, "b", "b");
+            wtx.commit()
+        })
+    };
+
+    t1.join()
+        .expect("should join")?
+        .expect("tx should not fail");
+
+    t2.join()
+        .expect("should join")?
+        .expect("tx should not fail");
+
+    // NOTE: We would expect a single writer tx implementation to finish in
+    // ~6 seconds
+    println!("Done in {:?}, items.len={}", start.elapsed(), {
+        let rtx = keyspace.read_tx();
+        rtx.len(&items)?
+    });
+
+    Ok(())
+}

src/config.rs

@@ -233,7 +233,7 @@ impl Config {
     /// # Errors
     ///
     /// Will return `Err` if an IO error occurs.
-    #[cfg(feature = "single_writer_tx")]
+    #[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
     pub fn open_transactional(self) -> crate::Result<crate::TxKeyspace> {
         crate::TxKeyspace::open(self)
     }

src/error.rs

@@ -26,7 +26,7 @@ pub enum Error {
     /// Invalid or unparsable data format version
     InvalidVersion(Option<Version>),
 
-    /// A previous flush operation failed, indicating a hardware-related failure
+    /// A previous flush / commit operation failed, indicating a hardware-related failure
     ///
     /// Future writes will not be accepted as consistency cannot be guaranteed.
     ///

src/lib.rs

@@ -103,7 +103,7 @@ mod snapshot_nonce;
 mod snapshot_tracker;
 mod tracked_snapshot;
 
-#[cfg(feature = "single_writer_tx")]
+#[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
 mod tx;
 
 mod version;
@@ -127,7 +127,7 @@ pub use {
     version::Version,
 };
 
-#[cfg(feature = "single_writer_tx")]
+#[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
 pub use tx::{
     keyspace::{TransactionalKeyspace, TxKeyspace},
     partition::TransactionalPartitionHandle,
@@ -142,15 +142,15 @@ pub type WriteBatch = Batch;
 pub type Partition = PartitionHandle;
 
 /// Alias for [`TransactionalPartitionHandle`]
-#[cfg(feature = "single_writer_tx")]
+#[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
 pub type TxPartition = TransactionalPartitionHandle;
 
 /// Alias for [`TransactionalPartitionHandle`]
-#[cfg(feature = "single_writer_tx")]
+#[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
 pub type TxPartitionHandle = TransactionalPartitionHandle;
 
 /// Alias for [`TransactionalPartitionHandle`]
-#[cfg(feature = "single_writer_tx")]
+#[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
 pub type TransactionalPartition = TransactionalPartitionHandle;
 
 /// A snapshot moment

src/snapshot_tracker.rs

@@ -29,7 +29,7 @@ impl Default for SnapshotTrackerInner {
     fn default() -> Self {
         Self {
             data: DashMap::default(),
-            safety_gap: 100,
+            safety_gap: 50,
             lowest_freed_instant: RwLock::default(),
         }
     }

src/tx/conflict_manager.rs

@@ -0,0 +1,183 @@
+use crate::batch::PartitionKey;
+use core::ops::Bound;
+use lsm_tree::Slice;
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    ops::RangeBounds,
+};
+
+#[derive(Clone, Debug)]
+enum Read {
+    Single(Slice),
+    Range {
+        start: Bound<Slice>,
+        end: Bound<Slice>,
+    },
+    All,
+}
+
+#[derive(Default, Debug)]
+pub struct ConflictManager {
+    reads: BTreeMap<PartitionKey, Vec<Read>>,
+    conflict_keys: BTreeMap<PartitionKey, BTreeSet<Slice>>,
+}
+
+impl ConflictManager {
+    fn push_read(&mut self, partition: &PartitionKey, read: Read) {
+        if let Some(tbl) = self.reads.get_mut(partition) {
+            tbl.push(read);
+        } else {
+            self.reads.entry(partition.clone()).or_default().push(read);
+        }
+    }
+
+    pub fn mark_read(&mut self, partition: &PartitionKey, key: &Slice) {
+        self.push_read(partition, Read::Single(key.clone()));
+    }
+
+    pub fn mark_conflict(&mut self, partition: &PartitionKey, key: &[u8]) {
+        if let Some(tbl) = self.conflict_keys.get_mut(partition) {
+            tbl.insert(key.into());
+        } else {
+            self.conflict_keys
+                .entry(partition.clone())
+                .or_default()
+                .insert(key.into());
+        }
+    }
+
+    pub fn mark_range(&mut self, partition: &PartitionKey, range: impl RangeBounds<Slice>) {
+        let start = match range.start_bound() {
+            Bound::Included(k) => Bound::Included(k.clone()),
+            Bound::Excluded(k) => Bound::Excluded(k.clone()),
+            Bound::Unbounded => Bound::Unbounded,
+        };
+
+        let end = match range.end_bound() {
+            Bound::Included(k) => Bound::Included(k.clone()),
+            Bound::Excluded(k) => Bound::Excluded(k.clone()),
+            Bound::Unbounded => Bound::Unbounded,
+        };
+
+        let read = if start == Bound::Unbounded && end == Bound::Unbounded {
+            Read::All
+        } else {
+            Read::Range { start, end }
+        };
+
+        self.push_read(partition, read);
+    }
+
+    #[allow(clippy::too_many_lines)]
+    pub fn has_conflict(&self, other: &Self) -> bool {
+        if self.reads.is_empty() {
+            return false;
+        }
+
+        for (partition, keys) in &self.reads {
+            if let Some(other_conflict_keys) = other.conflict_keys.get(partition) {
+                for ro in keys {
+                    match ro {
+                        Read::Single(k) => {
+                            if other_conflict_keys.contains(k) {
+                                return true;
+                            }
+                        }
+                        Read::Range { start, end } => match (start, end) {
+                            (Bound::Included(start), Bound::Included(end)) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((
+                                        Bound::Included(start),
+                                        Bound::Included(end),
+                                    ))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Included(start), Bound::Excluded(end)) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((
+                                        Bound::Included(start),
+                                        Bound::Excluded(end),
+                                    ))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Included(start), Bound::Unbounded) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((Bound::Included(start), Bound::Unbounded))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Excluded(start), Bound::Included(end)) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((
+                                        Bound::Excluded(start),
+                                        Bound::Included(end),
+                                    ))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Excluded(start), Bound::Excluded(end)) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((
+                                        Bound::Excluded(start),
+                                        Bound::Excluded(end),
+                                    ))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Excluded(start), Bound::Unbounded) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((Bound::Excluded(start), Bound::Unbounded))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Unbounded, Bound::Included(end)) => {
+                                let range = ..=end;
+                                for write in other_conflict_keys {
+                                    if range.contains(&write) {
+                                        return true;
+                                    }
+                                }
+                            }
+                            (Bound::Unbounded, Bound::Excluded(end)) => {
+                                let range = ..end;
+                                for write in other_conflict_keys {
+                                    if range.contains(&write) {
+                                        return true;
+                                    }
+                                }
+                            }
+                            (Bound::Unbounded, Bound::Unbounded) => unreachable!(),
+                        },
+                        Read::All => {
+                            if !other_conflict_keys.is_empty() {
+                                return true;
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        false
+    }
+}