Online DDL - stage 1 (#80)

Author: losfair

mvclient/src/lib.rs

@@ -41,6 +41,8 @@ pub struct MultiVersionClientConfig {
     pub ns_key: String,
 
     pub ns_key_hashproof: Option<String>,
+
+    pub lock_owner: Option<String>,
 }
 
 impl MultiVersionClientConfig {
@@ -97,6 +99,8 @@ pub struct CommitGlobalInit<'a> {
     pub allow_skip_idempotency_check: bool,
 
     pub num_namespaces: usize,
+
+    pub lock_owner: Option<&'a str>,
 }
 
 #[derive(Serialize)]
@@ -183,6 +187,10 @@ impl MultiVersionClient {
                 .append_pair("from_version", from_version);
         }
 
+        if let Some(lock_owner) = &self.config.lock_owner {
+            url.query_pairs_mut().append_pair("lock_owner", lock_owner);
+        }
+
         let mut boff = RandomizedExponentialBackoff::default();
         let stat_res: StatResponse = loop {
             let resp = request_and_check(self.client.get(url.clone()).decorate(self)).await?;
@@ -259,6 +267,7 @@ impl MultiVersionClient {
                 idempotency_key: &idempotency_key[..],
                 allow_skip_idempotency_check,
                 num_namespaces: intents.len(),
+                lock_owner: self.config.lock_owner.as_deref(),
             };
             allow_skip_idempotency_check = false;
             let mut raw_request: Vec<u8> = Vec::new();
@@ -682,10 +691,15 @@ impl Transaction {
     }
 }
 
+#[derive(Error, Debug)]
+#[error("status {0}")]
+pub struct StatusCodeError(pub StatusCode);
+
 async fn request_and_check(r: RequestBuilder) -> Result<Option<(HeaderMap, Bytes)>> {
     request_and_check_returning_status(r)
         .await
-        .map_err(|e| anyhow::anyhow!("status {}", e))
+        .map_err(StatusCodeError)
+        .map_err(anyhow::Error::from)
 }
 
 async fn request_and_check_returning_status(

mvfs/src/vfs.rs

@@ -12,7 +12,8 @@ use std::{
 };
 
 use mvclient::{
-    CommitOutput, MultiVersionClient, MultiVersionClientConfig, TimeToVersionResponse, Transaction,
+    CommitOutput, MultiVersionClient, MultiVersionClientConfig, StatusCodeError,
+    TimeToVersionResponse, Transaction,
 };
 
 use crate::types::LockKind;
@@ -30,6 +31,7 @@ pub struct MultiVersionVfs {
     pub sector_size: usize,
     pub http_client: reqwest::Client,
     pub db_name_map: Arc<HashMap<String, String>>,
+    pub lock_owner: Option<String>,
 }
 
 impl MultiVersionVfs {
@@ -93,6 +95,7 @@ impl MultiVersionVfs {
                     .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?,
                 ns_key: ns_key.to_string(),
                 ns_key_hashproof,
+                lock_owner: self.lock_owner.clone(),
             },
             self.http_client.clone(),
         )?;
@@ -604,6 +607,18 @@ impl Connection {
                 Ok(x) => x,
                 Err(e) => {
                     tracing::error!(ns_key = self.client.config().ns_key, error = %e, "transaction initialization failed");
+
+                    if let Some(sc_err) =
+                        e.chain().find_map(|x| x.downcast_ref::<StatusCodeError>())
+                    {
+                        if sc_err.0.as_u16() == 410 {
+                            tracing::error!(
+                                "this client can no longer start transaction on this database"
+                            );
+                            return Err(std::io::Error::new(std::io::ErrorKind::Other, "error"));
+                        }
+                    }
+
                     return Ok(false);
                 }
             };

mvsqlite-fuse/src/main.rs

@@ -96,6 +96,7 @@ async fn main() -> Result<()> {
         sector_size: opt.sector_size,
         http_client: reqwest::Client::new(),
         db_name_map: Arc::new(Default::default()),
+        lock_owner: None,
     };
     let fuse_fs = FuseFs {
         namespaces,

mvsqlite/src/lib.rs

@@ -101,6 +101,14 @@ fn init_with_options_impl(opts: InitOptions) {
     }
     let db_name_map = Arc::new(db_name_map);
 
+    let mut lock_owner: Option<String> = None;
+    if let Ok(s) = std::env::var("MVSQLITE_LOCK_OWNER") {
+        if !s.is_empty() {
+            tracing::debug!(lock_owner = s, "configuring lock owner");
+            lock_owner = Some(s);
+        }
+    }
+
     let mut builder = reqwest::ClientBuilder::new();
     builder = builder.timeout(Duration::from_secs(timeout_secs));
     if force_http2 {
@@ -118,6 +126,7 @@ fn init_with_options_impl(opts: InitOptions) {
             sector_size,
             http_client: http_client.clone(),
             db_name_map: db_name_map.clone(),
+            lock_owner: lock_owner.clone(),
         },
     };
 
@@ -131,6 +140,7 @@ fn init_with_options_impl(opts: InitOptions) {
                 sector_size,
                 http_client: http_client.clone(),
                 db_name_map: db_name_map.clone(),
+                lock_owner: lock_owner.clone(),
             },
         };
         sqlite_vfs::register(&format!("{}-{}", VFS_NAME, sector_size), vfs, false)

mvstore-stress/src/main.rs

@@ -79,6 +79,7 @@ async fn main() -> Result<()> {
             data_plane: vec![opt.data_plane.parse()?],
             ns_key: opt.ns_key.clone(),
             ns_key_hashproof: None,
+            lock_owner: None,
         },
         reqwest::Client::new(),
     )?;

mvstore/src/commit.rs

@@ -16,7 +16,7 @@ use rand::RngCore;
 use crate::{
     delta::reader::DeltaReader,
     server::Server,
-    util::{decode_version, generate_suffix_versionstamp_atomic_op},
+    util::{decode_version, generate_suffix_versionstamp_atomic_op, GoneError},
     util::{get_txn_read_version_as_versionstamp, ContentIndex},
     write::{WriteApplier, WriteApplierContext, WriteRequest},
 };
@@ -39,6 +39,7 @@ pub struct CommitContext<'a> {
     pub idempotency_key: [u8; 16],
     pub allow_skip_idempotency_check: bool,
     pub namespaces: &'a [CommitNamespaceContext<'a>],
+    pub lock_owner: Option<&'a str>,
 }
 
 pub struct CommitNamespaceContext<'a> {
@@ -175,9 +176,31 @@ impl Server {
         // Phase 2 - content index insertion
 
         for ns in ctx.namespaces {
-            if let Some(md) = &ns.metadata {
-                let metadata_key = self.key_codec.construct_nsmd_key(ns.ns_id);
-                txn.set(&metadata_key, md.as_bytes());
+            let metadata = self
+                .ns_metadata_cache
+                .get(&txn, &self.key_codec, ns.ns_id)
+                .await?;
+
+            // Does the client own the lock, if any?
+            if let Some(lock) = &metadata.lock {
+                match ctx.lock_owner {
+                    Some(lock_owner) => {
+                        // Validate lock ownership and state
+                        if lock.owner.as_str() != lock_owner {
+                            return Err(GoneError("you no longer own the lock").into());
+                        }
+                        if lock.rolling_back {
+                            return Err(GoneError("rolling back").into());
+                        }
+                    }
+                    None => {
+                        return Ok(CommitResult::Conflict);
+                    }
+                }
+            } else {
+                if ctx.lock_owner.is_some() {
+                    return Err(GoneError("you do not own the lock").into());
+                }
             }
 
             let mut written_pages: BTreeSet<u32> = BTreeSet::new();

mvstore/src/delta/reader.rs

@@ -34,14 +34,6 @@ impl<'a> DeltaReader<'a> {
             Some(x) => decode_version(x)?,
             None => [0xffu8; 10],
         };
-        if let Some(rm) = self.replica_manager {
-            let current_rv = rm.replica_version(self.txn).await?;
-            let requested_rv = i64::from_be_bytes(page_version[0..8].try_into().unwrap());
-            if current_rv < requested_rv {
-                anyhow::bail!("this replica does not have the requested read version");
-            }
-            tracing::debug!(current_rv, requested_rv, "read_page_hash replica read");
-        }
         let scan_end = self
             .key_codec
             .construct_page_key(self.ns_id, page_index, page_version);

mvstore/src/gc.rs

@@ -8,14 +8,18 @@ use std::{
 
 use anyhow::{Context, Result};
 use bloom::{BloomFilter, ASMS};
-use foundationdb::{
-    future::FdbKeyValue,
-    options::{ConflictRangeType, StreamingMode},
-    RangeOption,
+use foundationdb::{future::FdbKeyValue, options::StreamingMode, RangeOption};
+
+use crate::{
+    fixed::FixedKeyVec,
+    lock::DistributedLock,
+    server::Server,
+    util::{
+        add_single_key_read_conflict_range, decode_version, extract_10_byte_suffix,
+        get_txn_read_version_as_versionstamp, truncate_10_byte_suffix, ContentIndex,
+    },
 };
 
-use crate::{fixed::FixedKeyVec, lock::DistributedLock, server::Server, util::ContentIndex};
-
 pub static GC_SCAN_BATCH_SIZE: AtomicUsize = AtomicUsize::new(5000);
 pub static GC_FRESH_PAGE_TTL_SECS: AtomicU64 = AtomicU64::new(3600);
 
@@ -24,9 +28,32 @@ impl Server {
         self: Arc<Self>,
         dry_run: bool,
         ns_id: [u8; 10],
-        before_version: [u8; 10],
+        mut before_version: [u8; 10],
         mut progress_callback: impl FnMut(Option<u64>),
     ) -> Result<()> {
+        // Fix up `before_version` to be the minimum of the three:
+        // - The supplied value
+        // - The cluster's current read version as seen by this snapshot
+        // - The NS lock version in the same snapshot
+        {
+            let txn = self.db.create_trx()?;
+
+            before_version = before_version.min(get_txn_read_version_as_versionstamp(&txn).await?);
+
+            let metadata = self
+                .ns_metadata_cache
+                .get(&txn, &self.key_codec, ns_id)
+                .await?;
+            if let Some(lock) = &metadata.lock {
+                before_version = before_version.min(decode_version(&lock.snapshot_version)?);
+            }
+        }
+
+        tracing::info!(
+            ns = hex::encode(&ns_id),
+            before_version = hex::encode(&before_version),
+            "starting version truncation"
+        );
         let scan_start = self.key_codec.construct_page_key(ns_id, 0, [0u8; 10]);
         let scan_end = self
             .key_codec
@@ -409,15 +436,7 @@ impl Server {
                         self.key_codec.construct_delta_referrer_key(ns_id, *hash);
 
                     // 3e. Add the CAM index of the remaining pages to the conflict set.
-                    txn.add_conflict_range(
-                        &ci_key,
-                        &ci_key
-                            .iter()
-                            .copied()
-                            .chain(std::iter::once(0u8))
-                            .collect::<Vec<u8>>(),
-                        ConflictRangeType::Read,
-                    )?;
+                    add_single_key_read_conflict_range(&txn, &ci_key)?;
 
                     // 3f. Delete the remaining pages from the CAM.
                     txn.clear(&ci_key);
@@ -452,13 +471,3 @@ impl Server {
         Ok(())
     }
 }
-
-fn truncate_10_byte_suffix(data: &[u8]) -> &[u8] {
-    assert!(data.len() >= 10);
-    &data[..data.len() - 10]
-}
-
-fn extract_10_byte_suffix(data: &[u8]) -> [u8; 10] {
-    assert!(data.len() >= 10);
-    <[u8; 10]>::try_from(&data[data.len() - 10..]).unwrap()
-}

mvstore/src/keys.rs

@@ -15,6 +15,13 @@ impl KeyCodec {
         key
     }
 
+    pub fn construct_nsrollbackcursor_key(&self, ns_id: [u8; 10]) -> Vec<u8> {
+        let mut key = pack(&(self.metadata_prefix.as_str(), "nsrollbackcursor"));
+        key.push(0x32);
+        key.extend_from_slice(&ns_id);
+        key
+    }
+
     pub fn construct_nstask_key(&self, ns_id: [u8; 10], task: &str) -> Vec<u8> {
         let mut key = pack(&(self.metadata_prefix.as_str(), "nstask"));
         key.push(0x32);

mvstore/src/main.rs

@@ -4,6 +4,8 @@ mod fixed;
 mod gc;
 mod keys;
 mod lock;
+mod metadata;
+mod nslock;
 mod page;
 mod replica;
 mod server;