[teams 1/5] Reset database (#3611)

So far, the `--clear-database` option to `publish` has simply dropped
and then re-created the database (if it did exist).

This will no longer work when databases can have "children": because
dropping and re-creating is not atomic, children would either become
orphans, or be dropped as well.

To solve this, `reset_database` is introduced as a separate action that:

- shuts down all replicas
- if a `program_bytes` is supplied, replaces the database's initial
  program
- if a `host_type` is supplied, replaces the database's host type
- starts `num_replicas` or the previous number of replicas, which
  initialize themselves as normal

As this could be its own CLI command, the action is provided as its own
API endpoint (undocumented).

However, since `publish` has no way of knowing whether the database it
operates on actually exists, the `publish_database` handler will just
invoke the `reset_database` handler if `clear=true` and the database
exists, and return its result. This is to avoid starting the transfer of
the program in the request body, only to receive a redirect.

Some refactoring was necessary to dissect and understand the flow.


# API and ABI breaking changes

Introduces a new, undocumented API endpoint.
We may want to nest it under `/unstable`.

# Expected complexity level and risk

2

# Testing

From the outside, the observed behavior should be as before,
so smoketests should cover it.

Author: kim

Cargo.lock

@@ -8,8 +8,8 @@ use std::path::PathBuf;
 use std::{env, fs};
 
 use crate::config::Config;
-use crate::util::{add_auth_header_opt, get_auth_header, AuthHeader, ResponseExt};
-use crate::util::{decode_identity, unauth_error_context, y_or_n};
+use crate::util::{add_auth_header_opt, get_auth_header, unauth_error_context, AuthHeader, ResponseExt};
+use crate::util::{decode_identity, y_or_n};
 use crate::{build, common_args};
 
 pub fn cli() -> clap::Command {

crates/cli/src/subcommands/publish.rs

@@ -53,6 +53,7 @@ scopeguard.workspace = true
 serde_with.workspace = true
 async-stream.workspace = true
 humantime.workspace = true
+thiserror.workspace = true
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 jemalloc_pprof.workspace = true

crates/client-api/Cargo.toml

@@ -3,6 +3,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use axum::response::ErrorResponse;
+use bytes::Bytes;
 use http::StatusCode;
 
 use spacetimedb::client::ClientActorIndex;
@@ -155,7 +156,7 @@ pub struct DatabaseDef {
     /// The [`Identity`] the database shall have.
     pub database_identity: Identity,
     /// The compiled program of the database module.
-    pub program_bytes: Vec<u8>,
+    pub program_bytes: Bytes,
     /// The desired number of replicas the database shall have.
     ///
     /// If `None`, the edition default is used.
@@ -164,6 +165,14 @@ pub struct DatabaseDef {
     pub host_type: HostType,
 }
 
+/// Parameters for resetting a database via [`ControlStateDelegate::reset_database`].
+pub struct DatabaseResetDef {
+    pub database_identity: Identity,
+    pub program_bytes: Option<Bytes>,
+    pub num_replicas: Option<NonZeroU8>,
+    pub host_type: Option<HostType>,
+}
+
 /// API of the SpacetimeDB control plane.
 ///
 /// The trait is the composition of [`ControlStateReadAccess`] and
@@ -233,6 +242,10 @@ pub trait ControlStateWriteAccess: Send + Sync {
 
     async fn delete_database(&self, caller_identity: &Identity, database_identity: &Identity) -> anyhow::Result<()>;
 
+    /// Remove all data from a database, and reset it according to the
+    /// given [DatabaseResetDef].
+    async fn reset_database(&self, caller_identity: &Identity, spec: DatabaseResetDef) -> anyhow::Result<()>;
+
     // Energy
     async fn add_energy(&self, identity: &Identity, amount: EnergyQuanta) -> anyhow::Result<()>;
     async fn withdraw_energy(&self, identity: &Identity, amount: EnergyQuanta) -> anyhow::Result<()>;
@@ -332,6 +345,10 @@ impl<T: ControlStateWriteAccess + ?Sized> ControlStateWriteAccess for Arc<T> {
         (**self).delete_database(caller_identity, database_identity).await
     }
 
+    async fn reset_database(&self, caller_identity: &Identity, spec: DatabaseResetDef) -> anyhow::Result<()> {
+        (**self).reset_database(caller_identity, spec).await
+    }
+
     async fn add_energy(&self, identity: &Identity, amount: EnergyQuanta) -> anyhow::Result<()> {
         (**self).add_energy(identity, amount).await
     }