From 7eaa8f59b062f5fac3cb6fc8b36f53b8feca5efe Mon Sep 17 00:00:00 2001
From: Dom Dwyer <dom@itsallbroken.com>
Date: Thu, 26 Jan 2023 17:26:41 +0100
Subject: [PATCH] fix: explicit namespace creation w/ existing ns

Prior to this commit, namespaces that had been created on one router
could not be used on another router until the latter was restarted.
Effectively, newly created namespaces couldn't be used.

After this commit, the catalog is also checked when a cache miss occurs,
ensuring the router discovers new, not-yet-cached namespaces.
---
 router/src/namespace_resolver/mock.rs         | 10 +--
 .../src/namespace_resolver/ns_autocreation.rs | 65 +++++++++++++++++--
 2 files changed, 64 insertions(+), 11 deletions(-)

diff --git a/router/src/namespace_resolver/mock.rs b/router/src/namespace_resolver/mock.rs
index aa264819df..6334a3cc0b 100644
--- a/router/src/namespace_resolver/mock.rs
+++ b/router/src/namespace_resolver/mock.rs
@@ -36,10 +36,10 @@ impl NamespaceResolver for MockNamespaceResolver {
         &self,
         namespace: &NamespaceName<'static>,
     ) -> Result<NamespaceId, super::Error> {
-        Ok(*self
-            .map
-            .lock()
-            .get(namespace)
-            .expect("mock namespace resolver does not have ID"))
+        Ok(*self.map.lock().get(namespace).ok_or(super::Error::Lookup(
+            iox_catalog::interface::Error::NamespaceNotFoundByName {
+                name: namespace.to_string(),
+            },
+        ))?)
     }
 }
diff --git a/router/src/namespace_resolver/ns_autocreation.rs b/router/src/namespace_resolver/ns_autocreation.rs
index 004538c3dc..4ebd751cc3 100644
--- a/router/src/namespace_resolver/ns_autocreation.rs
+++ b/router/src/namespace_resolver/ns_autocreation.rs
@@ -91,15 +91,27 @@ where
         if self.cache.get_schema(namespace).is_none() {
             trace!(%namespace, "namespace not found in cache");
 
-            let mut repos = self.catalog.repositories().await;
-
             match self.action {
                 MissingNamespaceAction::Reject => {
-                    debug!(%namespace, "namespace not in catalog and autocreation disabled");
-                    return Err(NamespaceCreationError::Reject(namespace.into()).into());
+                    // The namespace is not cached, but may exist in the
+                    // catalog. Delegate discovery down to the inner handler,
+                    // and map the lookup error to a reject error.
+                    match self.inner.get_namespace_id(namespace).await {
+                        Ok(v) => return Ok(v),
+                        Err(super::Error::Lookup(
+                            iox_catalog::interface::Error::NamespaceNotFoundByName { .. },
+                        )) => {
+                            warn!(%namespace, "namespace not in catalog and auto-creation disabled");
+                            return Err(NamespaceCreationError::Reject(namespace.into()).into());
+                        }
+                        Err(e) => return Err(e),
+                    }
                 }
                 MissingNamespaceAction::AutoCreate(retention_period_ns) => {
-                    match repos
+                    match self
+                        .catalog
+                        .repositories()
+                        .await
                         .namespaces()
                         .create(
                             namespace.as_str(),
@@ -256,7 +268,7 @@ mod tests {
         let catalog: Arc<dyn Catalog> = Arc::new(MemCatalog::new(metrics));
 
         let creator = NamespaceAutocreation::new(
-            MockNamespaceResolver::default().with_mapping(ns.clone(), NamespaceId::new(1)),
+            MockNamespaceResolver::default(),
             cache,
             Arc::clone(&catalog),
             TopicId::new(42),
@@ -276,4 +288,45 @@ mod tests {
         let mut repos = catalog.repositories().await;
         assert_matches!(repos.namespaces().get_by_name(ns.as_str()).await, Ok(None));
     }
+
+    #[tokio::test]
+    async fn test_reject_exists_in_catalog() {
+        let ns = NamespaceName::try_from("bananas").unwrap();
+
+        let cache = Arc::new(MemoryNamespaceCache::default());
+        let metrics = Arc::new(metric::Registry::new());
+        let catalog: Arc<dyn Catalog> = Arc::new(MemCatalog::new(metrics));
+
+        // First drive the population of the catalog
+        let creator = NamespaceAutocreation::new(
+            MockNamespaceResolver::default().with_mapping(ns.clone(), NamespaceId::new(1)),
+            Arc::clone(&cache),
+            Arc::clone(&catalog),
+            TopicId::new(42),
+            QueryPoolId::new(42),
+            MissingNamespaceAction::AutoCreate(TEST_RETENTION_PERIOD_NS),
+        );
+
+        let created_id = creator
+            .get_namespace_id(&ns)
+            .await
+            .expect("handler should succeed");
+
+        // Now try in "reject" mode.
+        let creator = NamespaceAutocreation::new(
+            MockNamespaceResolver::default().with_mapping(ns.clone(), NamespaceId::new(1)),
+            cache,
+            Arc::clone(&catalog),
+            TopicId::new(42),
+            QueryPoolId::new(42),
+            MissingNamespaceAction::Reject,
+        );
+
+        // It should not autocreate because we specified "rejection" behaviour, above
+        let id = creator
+            .get_namespace_id(&ns)
+            .await
+            .expect("should allow existing namespace from catalog");
+        assert_eq!(created_id, id);
+    }
 }