libsql-server: do not create a replica namespace the primary's fence refuses

A replica server creating a namespace lazily for a name the primary's fence
refuses (a quarantined or aborted target, a read-fenced source, a fence state
the primary cannot establish) now fails the request at once with the primary's
code (Error::NamespaceFence: 423 and the stable code) instead of retrying the
handshake, and leaves no local namespace behind: the namespace directory the
setup created is removed (a directory that already existed is kept), and
NamespaceStore::with forgets the metastore entry handle() added and the
controller the attempt created, when neither holds anything durable or is in
use by another attempt. A later request, once the primary admits the name,
creates it normally.

Co-authored-by: Tomasz Szymczyszyn <tomasz.szymczyszyn@shopify.com>

Author: shopify-river

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

@@ -67,6 +67,9 @@ impl ConfigureNamespace for ReplicaConfigurator {
         Box::pin(async move {
             tracing::debug!("creating replica namespace");
             let db_path = self.base.base_path.join("dbs").join(name.as_str());
+            // Whether this server already had a copy of the namespace: a directory this setup
+            // creates is removed again if the primary's fence refuses the namespace.
+            let had_copy = db_path.try_exists()?;
             let channel = self.channel.clone();
             let uri = self.uri.clone();
 
@@ -112,7 +115,28 @@ impl ConfigureNamespace for ReplicaConfigurator {
                         )
                         .await;
                 }
-                Err(e) => Err(e)?,
+                Err(e) => {
+                    if let Some(refusal) = PrimaryFenceRefusal::of(&e) {
+                        // The primary's fence refuses the namespace (a quarantined target, a
+                        // read-fenced source, a fence state it cannot establish; section 13.3):
+                        // fail at once with its code rather than retrying the handshake, and
+                        // leave no local copy behind that this setup created.
+                        let denial = refusal.0.clone();
+                        drop(replicator);
+                        if !had_copy {
+                            if let Err(e) = tokio::fs::remove_dir_all(&db_path).await {
+                                if e.kind() != std::io::ErrorKind::NotFound {
+                                    tracing::warn!(
+                                        "failed to remove {} after the primary refused {name}: {e}",
+                                        db_path.display()
+                                    );
+                                }
+                            }
+                        }
+                        return Err(crate::Error::NamespaceFence(denial));
+                    }
+                    Err(e)?
+                }
                 Ok(_) => (),
             }
 

libsql-server/src/namespace/fence/registry.rs

@@ -66,6 +66,29 @@ impl FenceRegistry {
         self.controllers.lock().remove(namespace)
     }
 
+    /// Forget `namespace`'s controller if it holds nothing worth keeping: no fence record and
+    /// no in-memory gate (only what a replica learned of its primary's fence, which the next
+    /// answered `hello` would replace), and nothing but the registry refers to it. For a name
+    /// whose setup failed before it was ever served, such as a replica's lazy creation that
+    /// the primary's fence refused. Returns whether it was forgotten.
+    pub fn forget_idle(&self, namespace: &NamespaceName) -> bool {
+        let mut controllers = self.controllers.lock();
+        let idle = controllers.get(namespace).is_some_and(|controller| {
+            let gate = controller.gate();
+            // Under the registry lock nobody can take another reference to it.
+            Arc::strong_count(controller) == 1
+                && matches!(gate.fence, StoredFence::None { .. })
+                && gate.indeterminate.is_none()
+                && gate.installing.is_none()
+                && gate.closing_reads.is_none()
+                && gate.creating_target.is_none()
+        });
+        if idle {
+            controllers.remove(namespace);
+        }
+        idle
+    }
+
     /// Refuse a namespace whose fence state is `UNKNOWN_UNAVAILABLE`, or that is being created
     /// as a quarantined target, before any work is done to serve it.
     pub fn check_available(&self, namespace: &NamespaceName) -> Result<(), FenceError> {
@@ -245,4 +268,41 @@ mod tests {
         assert!(registry.remove(&"ns".into()).is_some());
         assert!(!Arc::ptr_eq(&registry.controller(&"ns".into()), &a));
     }
+
+    /// A replica's lazy creation that the primary refused leaves nothing behind in the registry,
+    /// unless the controller holds fence state or somebody else still refers to it.
+    #[test]
+    fn forget_idle_only_unreferenced_plain_controllers() {
+        let registry = FenceRegistry::default();
+        assert!(!registry.forget_idle(&"missing".into()));
+
+        // What a refused replication taught it does not keep it.
+        let refused = registry.controller(&"refused".into());
+        refused.observe_primary(Some(FenceError::new(
+            FenceOutcome::MigrationTargetQuarantined,
+            "quarantined on the primary",
+        )));
+        // Still referenced: kept.
+        assert!(!registry.forget_idle(&"refused".into()));
+        drop(refused);
+        assert!(registry.forget_idle(&"refused".into()));
+        assert!(registry.get(&"refused".into()).is_none());
+        // The next use starts from a fresh UNFENCED controller.
+        assert!(registry
+            .controller(&"refused".into())
+            .permits(OperationClass::NormalRead)
+            .is_ok());
+
+        // A controller with fence state is never forgotten.
+        let registry = FenceRegistry::seeded([(
+            "lost".into(),
+            StoredFence::Unavailable {
+                detail: FenceDetail::CorruptRecord,
+                reason: "test".into(),
+                marker: None,
+            },
+        )]);
+        assert!(!registry.forget_idle(&"lost".into()));
+        assert!(registry.get(&"lost".into()).is_some());
+    }
 }

libsql-server/src/namespace/meta_store.rs

@@ -15,7 +15,7 @@ use libsql_sys::wal::{
 };
 use parking_lot::Mutex;
 use prost::Message;
-use rusqlite::TransactionBehavior;
+use rusqlite::{OptionalExtension, TransactionBehavior};
 use tokio::sync::oneshot;
 use tokio::sync::{
     mpsc,
@@ -1303,6 +1303,41 @@ impl MetaStore {
         r
     }
 
+    /// Take out the in-memory entry that [`handle`](Self::handle) put in the map for a
+    /// namespace whose creation then failed, so that [`exists`](Self::exists) and
+    /// [`lookup`](Self::lookup) do not report a namespace that was never created
+    /// (`docs/NAMESPACE_FENCE.md` section 13.3, replica lazy creation). Only an entry that no
+    /// handle is subscribed to any more and that has no stored config row is removed: a config
+    /// that was persisted, or a creation of the same name still in progress, keeps its entry.
+    /// Returns whether the entry was removed.
+    pub async fn forget_unstored(&self, namespace: NamespaceName) -> Result<bool> {
+        let inner = self.inner.clone();
+        tokio::task::spawn_blocking(move || -> std::result::Result<bool, FenceStoreError> {
+            // The connection lock first, as everywhere else that takes both: a config being
+            // persisted concurrently is either already in its row here, or finds no entry when
+            // it publishes and inserts its own.
+            let conn = inner.conn.blocking_lock();
+            let stored = conn
+                .query_row(
+                    "SELECT 1 FROM namespace_configs WHERE namespace = ?1",
+                    [namespace.as_str()],
+                    |_| Ok(()),
+                )
+                .optional()?
+                .is_some();
+            let mut configs = inner.configs.blocking_lock();
+            match configs.get(&namespace) {
+                Some(sender) if !stored && sender.receiver_count() == 0 => {
+                    configs.remove(&namespace);
+                    Ok(true)
+                }
+                _ => Ok(false),
+            }
+        })
+        .await?
+        .map_err(fence_store_error)
+    }
+
     // TODO: we need to either make sure that the metastore is restored
     // before we start accepting connections or we need to contact bottomless
     // here to check if a namespace exists. Preferably the former.
@@ -1739,6 +1774,38 @@ mod fence_tests {
         }
     }
 
+    /// The in-memory entry a failed creation left is forgotten, so `exists()` and `lookup()` do
+    /// not report the name; a stored config, or a handle still held, keeps the entry.
+    #[tokio::test]
+    async fn forget_unstored_only_unused_unstored_entries() {
+        let tmp = tempdir().unwrap();
+        let meta = open(tmp.path(), true).await;
+        let ns = NamespaceName::from("lazy");
+
+        assert!(!meta.forget_unstored(ns.clone()).await.unwrap());
+
+        let handle = meta.handle(ns.clone()).await.unwrap();
+        assert!(meta.exists(&ns).await);
+        // A handle is still held (a creation in progress): kept.
+        assert!(!meta.forget_unstored(ns.clone()).await.unwrap());
+        assert!(meta.exists(&ns).await);
+        drop(handle);
+        assert!(meta.forget_unstored(ns.clone()).await.unwrap());
+        assert!(!meta.exists(&ns).await);
+        assert!(meta.lookup(&ns).await.unwrap().is_none());
+
+        // A stored config is never forgotten.
+        let stored = NamespaceName::from("stored");
+        meta.handle(stored.clone())
+            .await
+            .unwrap()
+            .store(DatabaseConfig::default())
+            .await
+            .unwrap();
+        assert!(!meta.forget_unstored(stored.clone()).await.unwrap());
+        assert!(meta.lookup(&stored).await.unwrap().is_some());
+    }
+
     fn request(
         ns: &'static str,
         op: Uuid,

libsql-server/src/namespace/store.rs

@@ -400,17 +400,46 @@ impl NamespaceStore {
 
         // A lookup that cannot create: only the default namespace and lazy creation create a
         // namespace here, and those refuse a name whose fence state is not established.
-        let handle = match self.inner.metadata.lookup(&namespace).await? {
-            Some(handle) => handle,
+        let (handle, created) = match self.inner.metadata.lookup(&namespace).await? {
+            Some(handle) => (handle, false),
             None if namespace == NamespaceName::default() || self.inner.allow_lazy_creation => {
-                self.inner.metadata.handle(namespace.clone()).await?
+                (self.inner.metadata.handle(namespace.clone()).await?, true)
             }
             None => return Err(Error::NamespaceDoesntExist(namespace.to_string())),
         };
-        f(self
+        let entry = match self
             .load_namespace(&namespace, handle, RestoreOption::Latest)
-            .await?)
-        .await
+            .await
+        {
+            Ok(entry) => entry,
+            Err(e) => {
+                if created && e.fence_error().is_some() {
+                    self.forget_refused_creation(&namespace).await;
+                }
+                return Err(e);
+            }
+        };
+        f(entry).await
+    }
+
+    /// Undo what a lazy creation that a fence refused left in memory (on a replica server, the
+    /// primary refused to replicate the name; `docs/NAMESPACE_FENCE.md` section 13.3): the
+    /// metastore entry [`MetaStore::handle`] added, which would otherwise make the name look
+    /// like an existing namespace, and the controller the attempt created. Both are kept when
+    /// they hold anything durable or are in use by another attempt.
+    async fn forget_refused_creation(&self, namespace: &NamespaceName) {
+        match self.inner.metadata.forget_unstored(namespace.clone()).await {
+            Ok(forgotten) => {
+                let controller = self.inner.fences.forget_idle(namespace);
+                tracing::debug!(
+                    "refused creation of {namespace}: metastore entry forgotten: {forgotten}, \
+                     controller forgotten: {controller}"
+                );
+            }
+            Err(e) => {
+                tracing::warn!("failed to forget the refused creation of {namespace}: {e}")
+            }
+        }
     }
 
     fn resolve_attach_fn(&self) -> ResolveNamespacePathFn {

libsql-server/tests/fence/protocol.rs

@@ -1335,3 +1335,128 @@ fn replica_resumes_after_clear_read_fence() {
     });
     sim.run().unwrap();
 }
+
+/// Walks the quarantined target `ns` of `op` to `TARGET_WRITE_FENCED` (readable): seal the
+/// (empty) import, record a successful validation, publish. Returns the revision.
+async fn publish_target(admin: &Admin, ns: &str, op: Uuid) -> anyhow::Result<u64> {
+    let (_, body) = admin.inspect(ns).await?;
+    let (state, mut rev) = state_of(&body);
+    assert_eq!(state, "TARGET_QUARANTINED", "{body}");
+    for (n, route, from, extra, to) in [
+        (
+            2,
+            "target/seal-import",
+            "TARGET_QUARANTINED",
+            json!({}),
+            "TARGET_VALIDATING",
+        ),
+        (
+            3,
+            "target/validation-receipt",
+            "TARGET_VALIDATING",
+            json!({ "result": "ok", "summary": "empty" }),
+            "TARGET_VALIDATING",
+        ),
+        (
+            4,
+            "target/publish-readable",
+            "TARGET_VALIDATING",
+            json!({}),
+            "TARGET_WRITE_FENCED",
+        ),
+    ] {
+        let (status, body) = admin
+            .command(
+                ns,
+                route,
+                command_body(op, uuid(op.as_u128() + n), from, rev, extra),
+            )
+            .await?;
+        assert_eq!(status, StatusCode::OK, "{route}: {body}");
+        assert_eq!(state_of(&body).0, to, "{route}: {body}");
+        rev = state_of(&body).1;
+    }
+    Ok(rev)
+}
+
+/// A replica server creating a namespace lazily, for a name the primary's fence refuses to
+/// replicate (here a quarantined target), fails the request at once with the primary's code
+/// instead of retrying the handshake, and leaves no local copy: no namespace directory it
+/// created (a directory that was already there stays) and no metastore entry. Once the target
+/// is published the same name is created and served normally (section 13.3).
+#[test]
+fn replica_lazy_creation_refused_by_fence() {
+    let mut sim = sim();
+    let primary = tempdir().unwrap();
+    let replica = tempdir().unwrap();
+    // A directory that was on the replica before: a refused creation must not delete it.
+    let kept = replica.path().join("dbs").join("kept");
+    std::fs::create_dir_all(&kept).unwrap();
+    std::fs::write(kept.join("sentinel"), b"x").unwrap();
+    let dbs = replica.path().join("dbs");
+    make_primary(&mut sim, primary.path().to_path_buf(), Primary::default());
+    make_replica(&mut sim, replica.path().to_path_buf());
+    sim.client("client", async move {
+        let admin = Admin::new(Some(ADMIN_KEY));
+        let user = User::on("replica0");
+        let op = uuid(0x100);
+        quarantined_target(&admin, "tgt", op).await?;
+        quarantined_target(&admin, "kept", uuid(0x200)).await?;
+
+        for attempt in 0..2 {
+            let started = tokio::time::Instant::now();
+            assert_locked(
+                &format!("replica read of a quarantined target, attempt {attempt}"),
+                &user.legacy("tgt", &["select 1"]).await?,
+                QUARANTINED,
+            );
+            // Refused at the first handshake, not after a second of retries per attempt.
+            let took = started.elapsed();
+            assert!(
+                took < std::time::Duration::from_millis(500),
+                "took {took:?}"
+            );
+            assert!(
+                !dbs.join("tgt").exists(),
+                "the refused creation left dbs/tgt behind"
+            );
+        }
+        assert_locked(
+            "replica read of a quarantined target with a directory",
+            &user.legacy("kept", &["select 1"]).await?,
+            QUARANTINED,
+        );
+        assert!(
+            kept.join("sentinel").exists(),
+            "a pre-existing directory was removed"
+        );
+
+        let rev = publish_target(&admin, "tgt", op).await?;
+        let (status, body) = user.legacy("tgt", &["select 1"]).await?;
+        assert_eq!(status, StatusCode::OK, "after publication: {body}");
+        assert!(
+            dbs.join("tgt").exists(),
+            "published target not created on the replica"
+        );
+
+        // Writes through the replica reach the primary once writes are enabled.
+        let (status, body) = admin
+            .command(
+                "tgt",
+                "target/enable-writes",
+                command_body(
+                    op,
+                    uuid(op.as_u128() + 5),
+                    "TARGET_WRITE_FENCED",
+                    rev,
+                    json!({}),
+                ),
+            )
+            .await?;
+        assert_eq!(status, StatusCode::OK, "{body}");
+        let (status, body) = user.legacy("tgt", &["create table t (x)"]).await?;
+        assert_eq!(status, StatusCode::OK, "write through the replica: {body}");
+        Ok(())
+    });
+    sim.run().unwrap();
+}