cli: pre-publish endpoint call. (#3278)

# Description of Changes
PR contains:

* CLI changes for the `pre_publish` endpoint when publishing a module.
* The regular `--yes` flag will not bypasses the *break clients* warning
prompt — an extra confirmation is now required. For CI, a hidden flag
`--break-clients` is added.
* Added smoketest.
* Some trivial naming changes in `client-api-*` crates for consistency
reasons.
* `pre_publish` route to accept similar Body size limit as `publish`
route.



# API and ABI breaking changes
an additive API change, does not break anything.

# Expected complexity level and risk
2

# Testing
- Existing smoketests passing for backward compatibility.
- New smoketest for add columns

---------

Signed-off-by: Shubham Mishra <[email protected]>
Co-authored-by: Phoebe Goldman <[email protected]>
Co-authored-by: Zeke Foppa <[email protected]>

Author: 3 people

crates/bindings-macro/src/table.rs

@@ -703,7 +703,7 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R
                 spacetimedb::table::ColumnDefault {
                     col_id: #col_id,
                     value: #val.serialize(spacetimedb::sats::algebraic_value::ser::ValueSerializer).expect("default value serialization failed"),
-                }
+                },
             })
         } else {
             None

crates/cli/src/subcommands/publish.rs

@@ -2,13 +2,13 @@ use clap::Arg;
 use clap::ArgAction::{Set, SetTrue};
 use clap::ArgMatches;
 use reqwest::{StatusCode, Url};
-use spacetimedb_client_api_messages::name::PublishOp;
 use spacetimedb_client_api_messages::name::{is_identity, parse_database_name, PublishResult};
-use std::fs;
+use spacetimedb_client_api_messages::name::{PrePublishResult, PrettyPrintStyle, PublishOp};
 use std::path::PathBuf;
+use std::{env, fs};
 
 use crate::config::Config;
-use crate::util::{add_auth_header_opt, get_auth_header, ResponseExt};
+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::{build, common_args};
 
@@ -55,6 +55,12 @@ pub fn cli() -> clap::Command {
                 .hide(true)
                 .help("UNSTABLE: The number of replicas the database should have")
         )
+        .arg(
+            Arg::new("break_clients")
+                .long("break-clients")
+                .action(SetTrue)
+                .help("Allow breaking changes when publishing to an existing database identity. This will break existing clients.")
+        )
         .arg(
             common_args::anonymous()
         )
@@ -87,27 +93,14 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     let database_host = config.get_host_url(server)?;
     let build_options = args.get_one::<String>("build_options").unwrap();
     let num_replicas = args.get_one::<u8>("num_replicas");
+    let break_clients_flag = args.get_flag("break_clients");
 
     // If the user didn't specify an identity and we didn't specify an anonymous identity, then
     // we want to use the default identity
     // TODO(jdetter): We should maybe have some sort of user prompt here for them to be able to
     //  easily create a new identity with an email
     let auth_header = get_auth_header(&mut config, anon_identity, server, !force).await?;
 
-    let client = reqwest::Client::new();
-
-    // If a domain or identity was provided, we should locally make sure it looks correct and
-    let mut builder = if let Some(name_or_identity) = name_or_identity {
-        if !is_identity(name_or_identity) {
-            parse_database_name(name_or_identity)?;
-        }
-        let encode_set = const { &percent_encoding::NON_ALPHANUMERIC.remove(b'_').remove(b'-') };
-        let domain = percent_encoding::percent_encode(name_or_identity.as_bytes(), encode_set);
-        client.put(format!("{database_host}/v1/database/{domain}"))
-    } else {
-        client.post(format!("{database_host}/v1/database"))
-    };
-
     if !path_to_project.exists() {
         return Err(anyhow::anyhow!(
             "Project path does not exist: {}",
@@ -141,6 +134,35 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
         database_host
     );
 
+    let client = reqwest::Client::new();
+    // If a domain or identity was provided, we should locally make sure it looks correct and
+    let mut builder = if let Some(name_or_identity) = name_or_identity {
+        if !is_identity(name_or_identity) {
+            parse_database_name(name_or_identity)?;
+        }
+        let encode_set = const { &percent_encoding::NON_ALPHANUMERIC.remove(b'_').remove(b'-') };
+        let domain = percent_encoding::percent_encode(name_or_identity.as_bytes(), encode_set);
+
+        let mut builder = client.put(format!("{database_host}/v1/database/{domain}"));
+
+        if !clear_database {
+            builder = apply_pre_publish_if_needed(
+                builder,
+                &client,
+                &database_host,
+                &domain.to_string(),
+                &program_bytes,
+                &auth_header,
+                break_clients_flag,
+            )
+            .await?;
+        };
+
+        builder
+    } else {
+        client.post(format!("{database_host}/v1/database"))
+    };
+
     if clear_database {
         // Note: `name_or_identity` should be set, because it is `required` in the CLI arg config.
         println!(
@@ -219,3 +241,79 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
 
     Ok(())
 }
+
+/// Determine the pretty print style based on the NO_COLOR environment variable.
+///
+/// See: https://no-color.org
+pub fn pretty_print_style_from_env() -> PrettyPrintStyle {
+    match env::var("NO_COLOR") {
+        Ok(_) => PrettyPrintStyle::NoColor,
+        Err(_) => PrettyPrintStyle::AnsiColor,
+    }
+}
+
+/// Applies pre-publish logic: checking for migration plan, prompting user, and
+/// modifying the request builder accordingly.
+async fn apply_pre_publish_if_needed(
+    mut builder: reqwest::RequestBuilder,
+    client: &reqwest::Client,
+    base_url: &str,
+    domain: &String,
+    program_bytes: &[u8],
+    auth_header: &AuthHeader,
+    break_clients_flag: bool,
+) -> Result<reqwest::RequestBuilder, anyhow::Error> {
+    if let Some(pre) = call_pre_publish(client, base_url, &domain.to_string(), program_bytes, auth_header).await? {
+        println!("{}", pre.migrate_plan);
+
+        if pre.break_clients
+            && !y_or_n(
+                break_clients_flag,
+                "The above changes will BREAK existing clients. Do you want to proceed?",
+            )?
+        {
+            println!("Aborting");
+            // Early exit: return an error or a special signal. Here we bail out by returning Err.
+            anyhow::bail!("Publishing aborted by user");
+        }
+
+        builder = builder
+            .query(&[("token", pre.token)])
+            .query(&[("policy", "BreakClients")]);
+    }
+
+    Ok(builder)
+}
+
+async fn call_pre_publish(
+    client: &reqwest::Client,
+    database_host: &str,
+    domain: &String,
+    program_bytes: &[u8],
+    auth_header: &AuthHeader,
+) -> Result<Option<PrePublishResult>, anyhow::Error> {
+    let mut builder = client.post(format!("{database_host}/v1/database/{domain}/pre_publish"));
+    let style = pretty_print_style_from_env();
+    builder = builder.query(&[("pretty_print_style", style)]);
+
+    builder = add_auth_header_opt(builder, auth_header);
+
+    println!("Checking for breaking changes...");
+    let res = builder.body(program_bytes.to_vec()).send().await?;
+
+    if res.status() == StatusCode::NOT_FOUND {
+        // This is a new database, so there are no breaking changes
+        return Ok(None);
+    }
+
+    if !res.status().is_success() {
+        anyhow::bail!(
+            "Pre-publish check failed with status {}: {}",
+            res.status(),
+            res.text().await?
+        );
+    }
+
+    let pre_publish_result: PrePublishResult = res.json_or_error().await?;
+    Ok(Some(pre_publish_result))
+}

crates/client-api-messages/src/name.rs

@@ -121,7 +121,7 @@ pub enum PrettyPrintStyle {
 }
 
 #[derive(serde::Serialize, serde::Deserialize, Debug)]
-pub struct PrintPlanResult {
+pub struct PrePublishResult {
     pub migrate_plan: Box<str>,
     pub break_clients: bool,
     pub token: spacetimedb_lib::Hash,

crates/client-api/src/routes/database.rs

@@ -27,7 +27,7 @@ use spacetimedb::host::{MigratePlanResult, ReducerArgs};
 use spacetimedb::identity::Identity;
 use spacetimedb::messages::control_db::{Database, HostType};
 use spacetimedb_client_api_messages::name::{
-    self, DatabaseName, DomainName, MigrationPolicy, PrettyPrintStyle, PrintPlanResult, PublishOp, PublishResult,
+    self, DatabaseName, DomainName, MigrationPolicy, PrePublishResult, PrettyPrintStyle, PublishOp, PublishResult,
 };
 use spacetimedb_lib::db::raw_def::v9::RawModuleDefV9;
 use spacetimedb_lib::identity::AuthCtx;
@@ -694,7 +694,7 @@ pub async fn pre_publish<S: NodeDelegate + ControlStateDelegate>(
     Query(PrePublishQueryParams { style }): Query<PrePublishQueryParams>,
     Extension(auth): Extension<SpacetimeAuth>,
     body: Bytes,
-) -> axum::response::Result<axum::Json<PrintPlanResult>> {
+) -> axum::response::Result<axum::Json<PrePublishResult>> {
     // User should not be able to print migration plans for a database that they do not own
     let database_identity = resolve_and_authenticate(&ctx, &name_or_identity, &auth).await?;
     let style = match style {
@@ -729,7 +729,7 @@ pub async fn pre_publish<S: NodeDelegate + ControlStateDelegate>(
             }
             .hash();
 
-            Ok(PrintPlanResult {
+            Ok(PrePublishResult {
                 token,
                 migrate_plan: plan,
                 break_clients: breaks_client,
@@ -992,7 +992,7 @@ where
             .route("/logs", self.logs_get)
             .route("/sql", self.sql_post)
             .route("/unstable/timestamp", self.timestamp_get)
-            .route("/pre-publish", self.pre_publish);
+            .route("/pre_publish", self.pre_publish);
 
         axum::Router::new()
             .route("/", self.root_post)

crates/standalone/src/subcommands/start.rs

@@ -183,6 +183,7 @@ pub async fn exec(args: &ArgMatches, db_cores: JobCores) -> anyhow::Result<()> {
     let mut db_routes = DatabaseRoutes::default();
     db_routes.root_post = db_routes.root_post.layer(DefaultBodyLimit::disable());
     db_routes.db_put = db_routes.db_put.layer(DefaultBodyLimit::disable());
+    db_routes.pre_publish = db_routes.pre_publish.layer(DefaultBodyLimit::disable());
     let extra = axum::Router::new().nest("/health", spacetimedb_client_api::routes::health::router());
     let service = router(&ctx, db_routes, extra).with_state(ctx.clone());
 

docs/docs/cli-reference.md

@@ -89,6 +89,7 @@ Run `spacetime help publish` for more detailed information.
 
   Default value: `.`
 * `-b`, `--bin-path <WASM_FILE>` — The system path (absolute or relative) to the compiled wasm binary we should publish, instead of building the project.
+* `--break-clients` — Allow breaking changes when publishing to an existing database identity. This will break existing clients.
 * `--anonymous` — Perform this action with an anonymous identity
 * `-s`, `--server <SERVER>` — The nickname, domain name or URL of the server to host the database.
 * `-y`, `--yes` — Run non-interactively wherever possible. This will answer "yes" to almost all prompts, but will sometimes answer "no" to preserve non-interactivity (e.g. when prompting whether to log in with spacetimedb.com).

smoketests/__init__.py

@@ -216,7 +216,7 @@ def log_records(self, n):
         logs = self.spacetime("logs", "--format=json", "-n", str(n), "--", self.database_identity)
         return list(map(json.loads, logs.splitlines()))
 
-    def publish_module(self, domain=None, *, clear=True, capture_stderr=True, num_replicas=None):
+    def publish_module(self, domain=None, *, clear=True, capture_stderr=True, num_replicas=None, break_clients=False):
         print("publishing module", self.publish_module)
         publish_output = self.spacetime(
             "publish",
@@ -228,6 +228,7 @@ def publish_module(self, domain=None, *, clear=True, capture_stderr=True, num_re
             # and so the publish step prompts for confirmation.
             "--yes",
             *["--num-replicas", f"{num_replicas}"] if num_replicas is not None else [],
+            *["--break-clients"] if break_clients else [],
             capture_stderr=capture_stderr,
         )
         self.resolved_identity = re.search(r"identity: ([0-9a-fA-F]+)", publish_output)[1]

smoketests/tests/auto_migration.py

@@ -227,3 +227,131 @@ def test_reject_schema_changes(self):
             self.publish_module(self.database_identity, clear=False)
 
         logging.info("Rejected as expected.")
+
+class AddTableColumns(Smoketest):
+    MODULE_CODE = """
+use spacetimedb::{log, ReducerContext, Table};
+
+#[derive(Debug)]
+#[spacetimedb::table(name = person)]
+pub struct Person {
+    name: String,
+}
+
+#[spacetimedb::reducer]
+pub fn add_person(ctx: &ReducerContext, name: String) {
+    ctx.db.person().insert(Person { name });
+}
+
+#[spacetimedb::reducer]
+pub fn print_persons(ctx: &ReducerContext, prefix: String) {
+    for person in ctx.db.person().iter() {
+        log::info!("{}: {}", prefix, person.name);
+    }
+}
+"""
+
+    MODULE_UPDATED = """
+use spacetimedb::{log, ReducerContext, Table};
+
+#[derive(Debug)]
+#[spacetimedb::table(name = person)]
+pub struct Person {
+    name: String,
+    #[default(0)]
+    age: u16,
+    #[default(19)]
+    mass: u16,
+}
+
+#[spacetimedb::reducer]
+pub fn add_person(ctx: &ReducerContext, name: String) {
+    ctx.db.person().insert(Person { name, age: 70, mass: 180 });
+}
+
+#[spacetimedb::reducer]
+pub fn print_persons(ctx: &ReducerContext, prefix: String) {
+    for person in ctx.db.person().iter() {
+        log::info!("{}: {:?}", prefix, person);
+    }
+}
+
+#[spacetimedb::reducer(client_disconnected)]
+pub fn identity_disconnected(ctx: &ReducerContext) {
+    log::info!("FIRST_UPDATE: client disconnected");
+}
+"""
+
+    MODULE_UPDATED_AGAIN = """
+use spacetimedb::{log, ReducerContext, Table};
+
+#[derive(Debug)]
+#[spacetimedb::table(name = person)]
+pub struct Person {
+    name: String,
+    age: u16,
+    #[default(19)]
+    mass: u16,
+    #[default(160)]
+    height: u32,
+}
+
+#[spacetimedb::reducer]
+pub fn add_person(ctx: &ReducerContext, name: String) {
+    ctx.db.person().insert(Person { name, age: 70, mass: 180, height: 72 });
+}
+
+#[spacetimedb::reducer]
+pub fn print_persons(ctx: &ReducerContext, prefix: String) {
+    for person in ctx.db.person().iter() {
+        log::info!("{}: {:?}", prefix, person);
+    }
+}
+"""
+
+    def test_add_table_columns(self):
+        """Verify schema upgrades that add columns with defaults (twice)."""
+
+        # Subscribe to person table changes multiple times to simulate active clients
+        for _ in range(20):
+            self.subscribe("select * from person", n=5)
+
+        # Insert under initial schema
+        self.call("add_person", "Robert")
+
+        # First upgrade: add age & mass columns
+        self.write_module_code(self.MODULE_UPDATED)
+        self.publish_module(self.database_identity, clear=False, break_clients=True)
+        self.call("print_persons", "FIRST_UPDATE")
+
+        logs1 = self.logs(100)
+
+        # Validate disconnect + schema migration logs
+        self.assertIn("Disconnecting all users", logs1)
+        self.assertIn(
+            'FIRST_UPDATE: Person { name: "Robert", age: 0, mass: 19 }',
+            logs1,
+        )
+        disconnect_count = logs1.count("FIRST_UPDATE: client disconnected")
+        self.assertEqual(
+            disconnect_count,
+        # this is always +1, I believe its due to test setup's own connection
+            21,
+            msg=f"Unexpected disconnect counts: {disconnect_count}",
+        )
+
+        # Insert new data under upgraded schema
+        self.call("add_person", "Robert2")
+
+        # Second upgrade
+        self.write_module_code(self.MODULE_UPDATED_AGAIN)
+        self.publish_module(self.database_identity, clear=False, break_clients=True)
+        self.call("print_persons", "UPDATE_2")
+
+        logs2 = self.logs(100)
+
+        # Validate new schema with height
+        self.assertIn(
+            'UPDATE_2: Person { name: "Robert2", age: 70, mass: 180, height: 160 }',
+            logs2,
+        )