libsql-server: make namespace creation all-or-nothing

docs/ADMIN_API.md

@@ -24,6 +24,31 @@ body:
 }
 ```
 
+`dump_url` initializes the new namespace from a SQLite SQL dump (`sqlite3 db .dump` or this
+server's `GET /dump` output). Supported schemes are `file:` (an absolute path on the server
+host) and `http(s):`. The dump must run inside a transaction and end with `COMMIT`; `ATTACH` is
+rejected.
+
+Creation is all-or-nothing. The namespace becomes visible, and its configuration is persisted,
+only after it has been fully set up (including the dump import). If the request fails, is
+cancelled by the client, or the server dies while it is in flight, the namespace either does not
+exist (its name can be created again right away: a failed attempt is fully undone before the
+error is returned) or is complete, never partially imported. Requests addressed to a namespace
+while it is being created wait for the outcome. Directories of creations the server died in the
+middle of are discarded at the next start (`<db-path>/.discarded/` holds them until they are
+deleted). For callers:
+
+- `2xx`: the namespace is complete;
+- any other outcome, including a lost connection: retry the same request; a
+  `400 Namespace already exists` on the retry means the earlier attempt did complete. (Once
+  the configuration is persisted, the creation finishes even if the client has gone away.)
+
+In single-namespace mode (`--disable-namespaces`) creating the default namespace is a
+configuration upsert and always succeeds, except with a `dump_url`: a dump can only be imported
+into a namespace that does not exist yet, so that request is rejected like any other.
+
+See `docs/ATOMIC_NAMESPACE_CREATE_DESIGN.md`.
+
 ```HTTP
 DELETE /v1/namespaces/:namespace
 ```

libsql-server/Cargo.toml

@@ -11,7 +11,7 @@ path = "src/main.rs"
 
 [dependencies]
 anyhow = "1.0.66"
-async-lock = "2.6.0"
+async-lock = "3"
 async-stream = "0.3.5"
 async-tempfile = "0.4.0"
 async-trait = "0.1.58"

libsql-server/src/lib.rs

@@ -637,6 +637,13 @@ where
         )
         .await?;
 
+        // Namespace creations the previous process died in the middle of leave marked
+        // directories (and tombstones of discarded ones) behind; clean them up before serving.
+        if let Err(e) = namespace::incomplete::sweep(&self.path, namespace_store.meta_store()).await
+        {
+            tracing::error!("failed to sweep unfinished namespace directories: {e}");
+        }
+
         self.spawn_monitoring_tasks(&mut task_manager, stats_receiver)?;
 
         // if namespaces are enabled, then bottomless must have set DB ID

libsql-server/src/metrics.rs

@@ -53,6 +53,14 @@ pub static NAMESPACE_EMPTIED_ENTRY_GAVE_UP: Lazy<Counter> = Lazy::new(|| {
     );
     register_counter!(NAME)
 });
+pub static NAMESPACE_LEFT_EMPTY_ENTRY_DROPPED: Lazy<Counter> = Lazy::new(|| {
+    const NAME: &str = "libsql_server_namespace_left_empty_entry_dropped";
+    describe_counter!(
+        NAME,
+        "number of empty namespace cache entries dropped so that the next lookup loads the namespace (left by a rejected creation, a failed reset or fork)"
+    );
+    register_counter!(NAME)
+});
 pub static NAMESPACE_LOAD_LATENCY: Lazy<Histogram> = Lazy::new(|| {
     const NAME: &str = "libsql_server_namespace_load_latency";
     describe_histogram!(NAME, "latency is us when loading a namespace");

libsql-server/src/namespace/configurator/fork.rs

@@ -130,9 +130,13 @@ impl ForkTask {
         let dest_namespace = self.to_namespace.clone();
         match self.try_fork().await {
             Err(e) => {
-                let _ =
-                    tokio::fs::remove_dir_all(base_path.join("dbs").join(dest_namespace.as_str()))
-                        .await;
+                // Only what this fork put there (the renamed, marked temp dir); a directory that
+                // was already at the destination is not ours to remove.
+                if let Err(e) =
+                    crate::namespace::incomplete::discard_if_incomplete(&base_path, &dest_namespace)
+                {
+                    tracing::error!(namespace = %dest_namespace, "failed to discard fork directory: {e}");
+                }
                 Err(e)
             }
             Ok(ns) => Ok(ns),
@@ -145,6 +149,9 @@ impl ForkTask {
         let temp_dir = BLOCKING_RT
             .spawn_blocking(move || tempfile::tempdir_in(base_path))
             .await??;
+        // Marked incomplete until the store has persisted the namespace's config, so that a
+        // fork that does not complete is discarded like any other unfinished creation.
+        crate::namespace::incomplete::mark_incomplete(temp_dir.path())?;
         let db_path = temp_dir.path().join("data");
 
         if let Some(restore) = self.restore_to {

libsql-server/src/namespace/configurator/helpers.rs

@@ -27,15 +27,68 @@ use crate::namespace::broadcasters::BroadcasterHandle;
 use crate::namespace::meta_store::MetaStoreHandle;
 use crate::namespace::replication_wal::{make_replication_wal_wrapper, ReplicationWalWrapper};
 use crate::namespace::{
-    NamespaceBottomlessDbId, NamespaceBottomlessDbIdInit, NamespaceName, ResolveNamespacePathFn,
-    RestoreOption,
+    incomplete, NamespaceBottomlessDbId, NamespaceBottomlessDbIdInit, NamespaceName,
+    ResolveNamespacePathFn, RestoreOption,
 };
 use crate::replication::ReplicationLogger;
 use crate::stats::Stats;
 use crate::{StatsSender, BLOCKING_RT, DB_CREATE_TIMEOUT, DEFAULT_AUTO_CHECKPOINT};
 
 use super::{BaseNamespaceConfig, PrimaryConfig};
 
+/// The directory of a namespace that does not exist yet, created marked incomplete.
+///
+/// If the guard is dropped before [`keep`](Self::keep) — setup failed, or the future creating
+/// the namespace was dropped — the directory is discarded (atomically renamed away and deleted
+/// in the background, see [`incomplete::discard_dir`]). The marker itself is removed by the
+/// namespace store once the config is persisted.
+pub(super) struct FreshDir {
+    base: Arc<Path>,
+    path: Option<Arc<Path>>,
+}
+
+impl FreshDir {
+    /// Returns `None` when a namespace directory already exists at `db_path`.
+    ///
+    /// Synchronous on purpose: the guard exists before anything is created, and there is no
+    /// await point between creating the directory and marking it.
+    pub(super) fn begin(base: &Arc<Path>, db_path: &Arc<Path>) -> crate::Result<Option<Self>> {
+        if db_path.try_exists()? {
+            return Ok(None);
+        }
+        let guard = Self {
+            base: base.clone(),
+            path: Some(db_path.clone()),
+        };
+        incomplete::create_marked_dir(db_path)?;
+        Ok(Some(guard))
+    }
+
+    /// The namespace was set up; the directory stays.
+    pub(super) fn keep(mut self) {
+        self.path.take();
+    }
+}
+
+impl Drop for FreshDir {
+    fn drop(&mut self) {
+        let Some(path) = self.path.take() else {
+            return;
+        };
+        tracing::info!(
+            path = %path.display(),
+            "namespace creation did not complete; discarding its directory"
+        );
+        if let Err(e) = incomplete::discard_dir(&self.base, &path) {
+            // still marked: discarded by the next creation of the name, or by the startup sweep
+            tracing::error!(
+                path = %path.display(),
+                "failed to discard unfinished namespace directory: {e}"
+            );
+        }
+    }
+}
+
 #[tracing::instrument(skip_all)]
 pub(super) async fn make_primary_connection_maker(
     primary_config: &PrimaryConfig,
@@ -509,3 +562,90 @@ pub(super) async fn cleanup_primary(
 
     Ok(())
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::namespace::incomplete::INCOMPLETE_MARKER;
+
+    async fn settle_background_deletes() {
+        for _ in 0..50 {
+            tokio::task::yield_now().await;
+            tokio::time::sleep(Duration::from_millis(2)).await;
+        }
+    }
+
+    #[tokio::test]
+    async fn fresh_dir_is_discarded_unless_kept() {
+        let tmp = tempfile::tempdir().unwrap();
+        let base: Arc<Path> = tmp.path().into();
+        let path: Arc<Path> = tmp.path().join("dbs").join("ns").into();
+
+        let fresh = FreshDir::begin(&base, &path).unwrap().expect("fresh");
+        assert!(path.join(INCOMPLETE_MARKER).exists());
+        drop(fresh);
+        // the path is free synchronously, the tombstone goes away in the background
+        assert!(!path.exists(), "dropped before keep: discarded");
+        settle_background_deletes().await;
+        assert!(std::fs::read_dir(tmp.path().join(".discarded"))
+            .unwrap()
+            .next()
+            .is_none());
+
+        let fresh = FreshDir::begin(&base, &path).unwrap().expect("fresh again");
+        fresh.keep();
+        assert!(
+            path.join(INCOMPLETE_MARKER).exists(),
+            "kept: stays, marker included"
+        );
+
+        assert!(
+            FreshDir::begin(&base, &path).unwrap().is_none(),
+            "an existing directory is not fresh"
+        );
+    }
+
+    #[tokio::test]
+    async fn fresh_dir_is_discarded_when_its_future_is_dropped() {
+        let tmp = tempfile::tempdir().unwrap();
+        let base: Arc<Path> = tmp.path().into();
+        let path: Arc<Path> = tmp.path().join("dbs").join("ns").into();
+        let (ready, is_ready) = tokio::sync::oneshot::channel();
+
+        let setup = {
+            let (base, path) = (base.clone(), path.clone());
+            async move {
+                let fresh = FreshDir::begin(&base, &path).unwrap();
+                ready.send(()).unwrap();
+                std::future::pending::<()>().await; // "the import"
+                fresh.map(FreshDir::keep);
+            }
+        };
+        let setup = tokio::spawn(setup);
+        is_ready.await.unwrap();
+        assert!(path.join(INCOMPLETE_MARKER).exists());
+        setup.abort();
+        let _ = setup.await;
+        assert!(!path.exists());
+    }
+
+    #[tokio::test]
+    async fn discard_if_incomplete_only_removes_marked_directories() {
+        let tmp = tempfile::tempdir().unwrap();
+        let marked = tmp.path().join("dbs").join("marked");
+        let unmarked = tmp.path().join("dbs").join("unmarked");
+        for dir in [&marked, &unmarked] {
+            std::fs::create_dir_all(dir).unwrap();
+            std::fs::write(dir.join("data"), b"junk").unwrap();
+        }
+        std::fs::write(marked.join(INCOMPLETE_MARKER), b"").unwrap();
+
+        let ns = |s: &str| NamespaceName::from_string(s.into()).unwrap();
+        assert!(incomplete::discard_if_incomplete(tmp.path(), &ns("marked")).unwrap());
+        assert!(!incomplete::discard_if_incomplete(tmp.path(), &ns("unmarked")).unwrap());
+        assert!(!incomplete::discard_if_incomplete(tmp.path(), &ns("absent")).unwrap());
+
+        assert!(!marked.exists());
+        assert!(unmarked.join("data").exists());
+    }
+}

libsql-server/src/namespace/configurator/mod.rs

@@ -129,6 +129,16 @@ pub trait ConfigureNamespace {
         bottomless_db_id_init: NamespaceBottomlessDbIdInit,
     ) -> Pin<Box<dyn Future<Output = crate::Result<()>> + Send + 'a>>;
 
+    /// Discard whatever a creation of `namespace` that did not complete left on disk, so that
+    /// the name starts from nothing. Only called for a name that is neither loaded nor known to
+    /// the metastore; must leave anything that does not carry the
+    /// [`INCOMPLETE_MARKER`](super::incomplete::INCOMPLETE_MARKER) alone. Synchronous so that
+    /// it can run while a creating future is being dropped (see `NamespaceStore`).
+    fn discard_incomplete(&self, namespace: &NamespaceName) -> crate::Result<()> {
+        let _ = namespace;
+        Ok(())
+    }
+
     fn fork<'a>(
         &'a self,
         from_ns: &'a Namespace,