@@ -28,7 +28,7 @@ Both dump importers run inside this sequence; the problem is independent of PR #
 - No lock-step changes to clients, replication, the metastore schema or the `DatabaseConfig` protobuf.
 - Small: reuse the pattern `fork` already uses; one mechanism for `create` and `fork`.
 
-Non-goals: a server-side import timeout; idempotency keys / async create jobs; bottomless remote cleanup for aborted imports (remote objects under a never-published db_id are inert junk, as today); the moka eviction hazard shared with `reset`/`fork` (§8).
+Non-goals: a server-side import timeout; idempotency keys / async create jobs; bottomless (not used in our deployment; untested with this change). The moka eviction hazard shared with `reset`/`fork` turned out to need fixing (§8).
 
 ## 3. Design in one paragraph
 
@@ -291,7 +291,9 @@ Reviewers (correctness, architecture, testing, operations) plus a manual pass fo
 
 LOW items closed in the follow-up commit: `publish` keeps the guard inside the reservation across `install`'s awaits, so a cancellation there undoes the creation instead of leaking the pin; `remove_incomplete_marker` stats before taking the blocking-pool hop (every cold load passes through it); a failure to remove the marker after publish is logged at `error`; `FreshDir`'s log is `info` (every failed create, including client errors, passes through it); `acquire` logs when it has to wait; store-level `aborted_creation_is_undone` test (`JoinHandle::abort` mid-import); the restart test drops the first server before starting the second; the unmarked-directory test asserts the outcome.
 
-Not changed: bottomless objects uploaded under a never-published db_id are inert junk (as today); `fsync` of the marker (power loss between creating and syncing it could leave an unmarked orphan — accepted); the moka `get_with(default)` leftover in `checkpoint` is healed lazily rather than fixed at the source; after an *abort* the (empty) cache entry stays until a lookup heals it or the cache evicts it — it counts toward capacity but is never served; a metastore rebuilt from the filesystem (`destroy_on_error` + `allow_recover_from_fs`) in the milliseconds between a namespace's flush and its unmark would skip that namespace.
+Also closed: the marker and its directory entry are fsynced on creation (`mark_incomplete`), so the marker is durable before anything else in the directory can be; `checkpoint` no longer mints an empty cache entry for an unloaded namespace (the pre-existing source of the same wedge); `ForkTask::fork` only removes a failed fork's destination if it carries the marker.
+
+Not changed: after an *abort* the (empty) cache entry stays until a lookup heals it or the cache evicts it — it counts toward capacity but is never served; a metastore rebuilt from the filesystem (`destroy_on_error` + `allow_recover_from_fs`) in the milliseconds between a namespace's flush and its unmark would skip that namespace. Bottomless is not used in our deployment and this change is untested with it.
 
 **Upgrade / downgrade.** Ghost rows and unmarked orphan directories left by failed creates on *older* binaries are untouched (everything here is marker-gated); for them the old remedy stays: `DELETE /v1/namespaces/:ns`, then create again. Downgrading the binary is safe: the metastore schema and the config protobuf are unchanged; an old binary ignores `.incomplete` and `.discarded/` (it may adopt a marked orphan via `maybe_recover_from_fs` on an empty metastore, as it always could with any leftover directory).
 
 
@@ -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),
@@ -147,12 +151,7 @@ impl ForkTask {
             .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.
-        tokio::fs::File::create(
-            temp_dir
-                .path()
-                .join(crate::namespace::incomplete::INCOMPLETE_MARKER),
-        )
-        .await?;
+        crate::namespace::incomplete::mark_incomplete(temp_dir.path())?;
         let db_path = temp_dir.path().join("data");
 
         if let Some(restore) = self.restore_to {
 
@@ -38,12 +38,19 @@ pub(crate) fn namespace_dir(base: &Path, name: &NamespaceName) -> PathBuf {
 
 /// Create the directory of a namespace that does not exist yet, marked incomplete.
 ///
-/// Synchronous (two small filesystem operations) so that there is no await point between
-/// creating the directory and marking it.
+/// Synchronous (a few small filesystem operations) so that there is no await point between
+/// creating the directory and marking it. The marker and the directory entry are fsynced: the
+/// marker must be durable before anything else in the directory can be, or a power loss could
+/// leave a directory with content but no marker, which nothing would dare discard.
 pub(crate) fn create_marked_dir(db_path: &Path) -> std::io::Result<()> {
     std::fs::create_dir_all(db_path)?;
-    std::fs::File::create(db_path.join(INCOMPLETE_MARKER))?;
-    Ok(())
+    mark_incomplete(db_path)
+}
+
+/// Write the marker into an existing directory and make it durable.
+pub(crate) fn mark_incomplete(db_path: &Path) -> std::io::Result<()> {
+    std::fs::File::create(db_path.join(INCOMPLETE_MARKER))?.sync_all()?;
+    std::fs::File::open(db_path)?.sync_all()
 }
 
 /// Remove `db_path` (a namespace directory under `base`): rename it into the discarded area
 
@@ -369,11 +369,11 @@ impl NamespaceStore {
     }
 
     pub async fn checkpoint(&self, namespace: NamespaceName) -> crate::Result<()> {
-        let entry = self
-            .inner
-            .store
-            .get_with(namespace.clone(), async { Default::default() })
-            .await;
+        // A namespace that is not loaded has nothing to checkpoint. (`get_with(Default)` here
+        // used to leave an empty entry behind that lookups then mistook for the namespace.)
+        let Some(entry) = self.inner.store.get(&namespace).await else {
+            return Ok(());
+        };
         let lock = entry.read().await;
         if let Some(ns) = &*lock {
             ns.checkpoint().await?;
 
@@ -46,6 +46,11 @@ pub(super) fn file_url(path: &Path) -> String {
 pub(super) fn sim() -> Sim<'static> {
     Builder::new()
         .simulation_duration(Duration::from_secs(1000))
+        // Dumps are served in many small chunks that the server only starts reading once the
+        // namespace is set up (fsyncs, file creation — real time, during which turmoil keeps
+        // advancing simulated time and the dump store keeps writing). Real TCP would apply
+        // back-pressure; turmoil panics once its socket buffer (64 segments by default) is full.
+        .tcp_capacity(4096)
         .build()
 }