fix: Distinguish between restoring twice and name collisions on restore

And return them in the correct situations.

influxdb_iox/src/influxdb_ioxd/server_type/database/rpc/error.rs

@@ -69,8 +69,10 @@ pub fn default_server_error_handler(error: server::Error) -> tonic::Status {
         }
         e @ Error::StoreWriteErrors { .. } => tonic::Status::invalid_argument(e.to_string()),
         Error::CannotRestoreDatabase {
-            source: e @ server::database::InitError::DatabaseAlreadyActive { .. },
+            source: e @ server::database::InitError::AlreadyActive { .. },
         } => tonic::Status::already_exists(e.to_string()),
+        e @ Error::DatabaseAlreadyActive { .. } => tonic::Status::already_exists(e.to_string()),
+        e @ Error::DatabaseAlreadyExists { .. } => tonic::Status::already_exists(e.to_string()),
         Error::CouldNotGetDatabaseNameFromRules {
             source: DatabaseNameFromRulesError::DatabaseRulesNotFound { uuid, .. },
         } => tonic::Status::not_found(format!("Could not find a database with UUID `{}`", uuid)),
@@ -137,9 +139,6 @@ pub fn default_database_error_handler(error: server::database::Error) -> tonic::
         Error::CannotDeleteInactiveDatabase { .. } => {
             tonic::Status::failed_precondition(error.to_string())
         }
-        Error::CannotRestoreActiveDatabase { .. } => {
-            tonic::Status::failed_precondition(error.to_string())
-        }
     }
 }
 

influxdb_iox/src/influxdb_ioxd/server_type/database/rpc/management.rs

@@ -132,15 +132,7 @@ where
             .server
             .create_database(provided_rules)
             .await
-            .map_err(|e| match e {
-                Error::DatabaseAlreadyExists { db_name } => AlreadyExists {
-                    resource_type: "database".to_string(),
-                    resource_name: db_name,
-                    ..Default::default()
-                }
-                .into(),
-                _ => default_server_error_handler(e),
-            })?;
+            .map_err(default_server_error_handler)?;
 
         let uuid = database
             .uuid()

influxdb_iox/tests/end_to_end_cases/management_api.rs

@@ -423,6 +423,28 @@ async fn test_create_get_update_delete_restore_database() {
         err.to_string(),
         format!("Could not find a database with UUID `{}`", unknown_uuid)
     );
+
+    client
+        .delete_database(&db_name)
+        .await
+        .expect("delete database failed");
+
+    let newly_created_uuid = client
+        .create_database(rules.clone())
+        .await
+        .expect("create database failed");
+    let newly_created_uuid = newly_created_uuid.to_string();
+
+    assert_ne!(deleted_uuid, newly_created_uuid);
+
+    let err = client
+        .restore_database(&deleted_uuid)
+        .await
+        .expect_err("restore database should have failed but didn't");
+    assert_contains!(
+        err.to_string(),
+        format!("A database with the name `{}` already exists", db_name)
+    );
 }
 
 /// gets configuration both with and without defaults, and verifies

influxdb_iox/tests/end_to_end_cases/management_cli.rs

@@ -290,6 +290,22 @@ async fn delete_restore_database() {
         .success()
         .stdout(predicate::str::contains(db));
 
+    // Restoring the 1st database is an error because the new, currently active database has the
+    // same name
+    Command::cargo_bin("influxdb_iox")
+        .unwrap()
+        .arg("database")
+        .arg("restore")
+        .arg(db_uuid)
+        .arg("--host")
+        .arg(addr)
+        .assert()
+        .failure()
+        .stderr(predicate::str::contains(format!(
+            "A database with the name `{}` already exists",
+            db
+        )));
+
     // Delete the 2nd database
     Command::cargo_bin("influxdb_iox")
         .unwrap()
@@ -356,6 +372,7 @@ async fn delete_restore_database() {
 
     // Restoring a database with a valid but unknown UUID is an error
     let unknown_uuid = Uuid::new_v4();
+    dbg!(unknown_uuid);
     Command::cargo_bin("influxdb_iox")
         .unwrap()
         .arg("database")

server/src/database.rs

@@ -93,9 +93,6 @@ pub enum Error {
         db_name
     ))]
     CannotDeleteInactiveDatabase { db_name: String },
-
-    #[snafu(display("cannot restore database named {} that is already active", db_name))]
-    CannotRestoreActiveDatabase { db_name: String },
 }
 
 #[derive(Debug, Snafu)]
@@ -240,7 +237,6 @@ impl Database {
     /// Mark this database as deleted.
     pub async fn delete(&self) -> Result<Uuid, Error> {
         let db_name = &self.shared.config.name;
-        info!(%db_name, "marking database deleted");
 
         let handle = self.shared.state.read().freeze();
         let handle = handle.await;
@@ -253,6 +249,8 @@ impl Database {
             state.uuid().expect("Active databases have UUIDs")
         };
 
+        info!(%db_name, %uuid, "marking database deleted");
+
         // If there is an object store for this database, write out a tombstone file.
         // If there isn't an object store, something is wrong and we shouldn't switch the
         // state without being able to write the tombstone file.
@@ -299,7 +297,7 @@ impl Database {
         let iox_object_store = match iox_object_store_result {
             Ok(iox_os) => iox_os,
             Err(iox_object_store::IoxObjectStoreError::DatabaseAlreadyActive { .. }) => {
-                return DatabaseAlreadyActive {
+                return AlreadyActive {
                     name: db_name.to_string(),
                     uuid,
                 }
@@ -979,7 +977,7 @@ pub enum InitError {
     },
 
     #[snafu(display("The database with UUID `{}` named `{}` is already active", uuid, name))]
-    DatabaseAlreadyActive { name: String, uuid: Uuid },
+    AlreadyActive { name: String, uuid: Uuid },
 
     #[snafu(display("cannot create preserved catalog: {}", source))]
     CannotCreatePreservedCatalog { source: crate::db::load::Error },

server/src/lib.rs

@@ -165,9 +165,12 @@ pub enum Error {
     #[snafu(display("{}", source))]
     CannotRestoreDatabase { source: crate::database::InitError },
 
-    #[snafu(display("database already exists: {}", db_name))]
+    #[snafu(display("A database with the name `{}` already exists", db_name))]
     DatabaseAlreadyExists { db_name: String },
 
+    #[snafu(display("The database with UUID `{}` named `{}` is already active", uuid, name))]
+    DatabaseAlreadyActive { name: String, uuid: Uuid },
+
     #[snafu(display("Server error: {}", source))]
     ServerError { source: std::io::Error },
 
@@ -681,6 +684,8 @@ where
         let uuid = Uuid::new_v4();
         let db_name = rules.db_name().clone();
 
+        info!(%db_name, %uuid, "creating new database");
+
         // Wait for exclusive access to mutate server state
         let handle_fut = self.shared.state.read().freeze();
         let handle = handle_fut.await;
@@ -770,6 +775,7 @@ where
         let handle_fut = self.shared.state.read().freeze();
         let handle = handle_fut.await;
 
+        // Don't proceed without a server ID
         let server_id = {
             let state = self.shared.state.read();
             let initialized = state.initialized()?;
@@ -777,6 +783,7 @@ where
             initialized.server_id
         };
 
+        // Read the database's rules from object storage to get the database name
         let db_name = database_name_from_rules_file(
             Arc::clone(self.shared.application.object_store()),
             server_id,
@@ -785,6 +792,25 @@ where
         .await
         .context(CouldNotGetDatabaseNameFromRules)?;
 
+        info!(%db_name, %uuid, "start restoring database");
+
+        // Check that this name is unique among currently active databases
+        if let Ok(existing_db) = self.database(&db_name) {
+            if matches!(dbg!(existing_db.uuid()),
+                    Some(existing_uuid) if existing_uuid == dbg!(uuid))
+            {
+                return DatabaseAlreadyActive {
+                    name: db_name,
+                    uuid,
+                }
+                .fail();
+            } else {
+                return DatabaseAlreadyExists { db_name }.fail();
+            }
+        }
+
+        // Mark the database as restored in object storage and get its location for the server
+        // config file
         let location = Database::restore(
             Arc::clone(&self.shared.application),
             &db_name,
@@ -2500,12 +2526,7 @@ mod tests {
         // restoring again fails
         let err = server.restore_database(foo_uuid).await.unwrap_err();
         assert!(
-            matches!(
-                err,
-                Error::CannotRestoreDatabase {
-                    source: database::InitError::DatabaseAlreadyActive { .. }
-                }
-            ),
+            matches!(err, Error::DatabaseAlreadyActive { .. }),
             "got: {:?}",
             err
         );