namespace: add bottomless checks in exists()

Even if the namespace doesn't exist locally, it might have a functional
backup.

bottomless/src/replicator.rs

@@ -376,6 +376,33 @@ impl Replicator {
         })
     }
 
+    /// Checks if there exists any backup of given database
+    pub async fn has_backup_of(db_name: impl AsRef<str>, options: &Options) -> Result<bool> {
+        let prefix = match &options.db_id {
+            Some(db_id) => format!("{db_id}-"),
+            None => format!("ns-:{}-", db_name.as_ref()),
+        };
+        let config = options.client_config().await?;
+        let client = Client::from_conf(config);
+        let bucket = options.bucket_name.clone();
+
+        match client.head_bucket().bucket(&bucket).send().await {
+            Ok(_) => tracing::trace!("Bucket {bucket} exists and is accessible"),
+            Err(e) => {
+                tracing::trace!("Bucket checking error: {e}");
+                return Err(e.into());
+            }
+        }
+
+        let mut last_frame = 0;
+        let list_objects = client.list_objects().bucket(&bucket).prefix(&prefix);
+        let response = list_objects.send().await?;
+        let _ = Self::try_get_last_frame_no(response, &mut last_frame);
+        tracing::trace!("Last frame of {prefix}: {last_frame}");
+
+        Ok(last_frame > 0)
+    }
+
     pub async fn shutdown_gracefully(&mut self) -> Result<()> {
         tracing::info!("bottomless replicator: shutting down...");
         // 1. wait for all committed WAL frames to be committed locally

libsql-server/src/namespace/mod.rs

@@ -74,6 +74,12 @@ impl Default for NamespaceName {
     }
 }
 
+impl AsRef<str> for NamespaceName {
+    fn as_ref(&self) -> &str {
+        self.as_str()
+    }
+}
+
 impl NamespaceName {
     pub fn from_string(s: String) -> crate::Result<Self> {
         Self::validate(&s)?;
@@ -170,7 +176,7 @@ pub trait MakeNamespace: Sync + Send + 'static {
         meta_store: &MetaStore,
     ) -> crate::Result<Namespace<Self::Database>>;
 
-    fn exists(&self, namespace: &NamespaceName) -> bool;
+    async fn exists(&self, namespace: &NamespaceName) -> bool;
 }
 
 /// Creates new primary `Namespace`
@@ -287,9 +293,32 @@ impl MakeNamespace for PrimaryNamespaceMaker {
         Ok(ns)
     }
 
-    fn exists(&self, namespace: &NamespaceName) -> bool {
+    async fn exists(&self, namespace: &NamespaceName) -> bool {
         let ns_path = self.config.base_path.join("dbs").join(namespace.as_str());
-        ns_path.try_exists().unwrap_or(false)
+        if let Ok(true) = ns_path.try_exists() {
+            return true;
+        }
+
+        if let Some(replication_options) = self.config.bottomless_replication.as_ref() {
+            tracing::info!("Bottomless: {:?}", replication_options);
+            match bottomless::replicator::Replicator::has_backup_of(namespace, replication_options)
+                .await
+            {
+                Ok(true) => {
+                    tracing::debug!("Bottomless: Backup found");
+                    return true;
+                }
+                Ok(false) => {
+                    tracing::debug!("Bottomless: No backup found");
+                }
+                Err(err) => {
+                    tracing::debug!("Bottomless: Error checking backup: {}", err);
+                }
+            }
+        } else {
+            tracing::debug!("Bottomless: No backup configured");
+        }
+        false
     }
 }
 
@@ -347,7 +376,7 @@ impl MakeNamespace for ReplicaNamespaceMaker {
         return Err(ForkError::ForkReplica.into());
     }
 
-    fn exists(&self, namespace: &NamespaceName) -> bool {
+    async fn exists(&self, namespace: &NamespaceName) -> bool {
         let ns_path = self.config.base_path.join("dbs").join(namespace.as_str());
         ns_path.try_exists().unwrap_or(false)
     }
@@ -541,7 +570,7 @@ impl<M: MakeNamespace> NamespaceStore<M> {
         }
 
         // check that the source namespace exists
-        if !self.inner.make_namespace.exists(&from) {
+        if !self.inner.make_namespace.exists(&from).await {
             return Err(crate::error::Error::NamespaceDoesntExist(from.to_string()));
         }
 
@@ -608,7 +637,7 @@ impl<M: MakeNamespace> NamespaceStore<M> {
             let namespace = namespace.clone();
             async move {
                 if namespace != NamespaceName::default()
-                    && !self.inner.make_namespace.exists(&namespace)
+                    && !self.inner.make_namespace.exists(&namespace).await
                     && !self.inner.allow_lazy_creation
                 {
                     return Err(Error::NamespaceDoesntExist(namespace.to_string()));
@@ -678,7 +707,7 @@ impl<M: MakeNamespace> NamespaceStore<M> {
         // otherwise it's an error.
         if self.inner.allow_lazy_creation || namespace == NamespaceName::default() {
             tracing::trace!("auto-creating the namespace");
-        } else if self.inner.make_namespace.exists(&namespace) {
+        } else if self.inner.make_namespace.exists(&namespace).await {
             return Err(Error::NamespaceAlreadyExist(namespace.to_string()));
         }