@@ -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"
 
@@ -638,6 +638,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
 
@@ -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");
 
@@ -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 {
 
@@ -19,15 +19,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,
@@ -405,3 +458,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());
+    }
+}
@@ -131,6 +131,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,
 
@@ -21,7 +21,7 @@ use crate::namespace::{
 use crate::run_periodic_checkpoint;
 use crate::schema::{has_pending_migration_task, setup_migration_table};
 
-use super::helpers::cleanup_primary;
+use super::helpers::{cleanup_primary, FreshDir};
 use super::{BaseNamespaceConfig, ConfigureNamespace, PrimaryConfig};
 
 pub struct PrimaryConfigurator {
@@ -129,35 +129,31 @@ impl ConfigureNamespace for PrimaryConfigurator {
     ) -> Pin<Box<dyn Future<Output = crate::Result<Namespace>> + Send + 'a>> {
         Box::pin(async move {
             let db_path: Arc<Path> = self.base.base_path.join("dbs").join(name.as_str()).into();
-            let fresh_namespace = !db_path.try_exists()?;
-            // FIXME: make that truly atomic. explore the idea of using temp directories, and it's implications
-            match self
+            // A brand-new directory is discarded again if setup fails or this future is dropped.
+            let fresh = FreshDir::begin(&self.base.base_path, &db_path)?;
+            let ns = self
                 .try_new_primary(
                     name.clone(),
                     meta_store_handle,
                     restore_option,
                     resolve_attach_path,
-                    db_path.clone(),
+                    db_path,
                     broadcaster,
                     self.base.encryption_config.clone(),
                 )
-                .await
-            {
-                Ok(this) => Ok(this),
-                Err(e) if fresh_namespace => {
-                    tracing::error!(
-                        "an error occured while deleting creating namespace, cleaning..."
-                    );
-                    if let Err(e) = tokio::fs::remove_dir_all(&db_path).await {
-                        tracing::error!("failed to remove dirty namespace directory: {e}")
-                    }
-                    Err(e)
-                }
-                Err(e) => Err(e),
+                .await?;
+            if let Some(fresh) = fresh {
+                fresh.keep();
             }
+            Ok(ns)
         })
     }
 
+    fn discard_incomplete(&self, namespace: &NamespaceName) -> crate::Result<()> {
+        crate::namespace::incomplete::discard_if_incomplete(&self.base.base_path, namespace)?;
+        Ok(())
+    }
+
     fn cleanup<'a>(
         &'a self,
         namespace: &'a NamespaceName,