diff --git a/crates/bindings-macro/src/table.rs b/crates/bindings-macro/src/table.rs index 4f569c6754f..5b1dd4246bf 100644 --- a/crates/bindings-macro/src/table.rs +++ b/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 diff --git a/crates/cli/src/subcommands/publish.rs b/crates/cli/src/subcommands/publish.rs index df63923eb59..4524f8c8081 100644 --- a/crates/cli/src/subcommands/publish.rs +++ b/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,6 +93,7 @@ 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::("build_options").unwrap(); let num_replicas = args.get_one::("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 @@ -94,20 +101,6 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E // 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 { + 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, 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)) +} diff --git a/crates/client-api-messages/src/name.rs b/crates/client-api-messages/src/name.rs index 49cee23a156..dcf39551496 100644 --- a/crates/client-api-messages/src/name.rs +++ b/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, pub break_clients: bool, pub token: spacetimedb_lib::Hash, diff --git a/crates/client-api/src/routes/database.rs b/crates/client-api/src/routes/database.rs index 5e01f43a6fc..ccab05d1280 100644 --- a/crates/client-api/src/routes/database.rs +++ b/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( Query(PrePublishQueryParams { style }): Query, Extension(auth): Extension, body: Bytes, -) -> axum::response::Result> { +) -> axum::response::Result> { // 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( } .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) diff --git a/crates/standalone/src/subcommands/start.rs b/crates/standalone/src/subcommands/start.rs index cac5fc3cab0..3d9cc44af4f 100644 --- a/crates/standalone/src/subcommands/start.rs +++ b/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()); diff --git a/docs/docs/cli-reference.md b/docs/docs/cli-reference.md index c09b1174ea1..66be4ef3aef 100644 --- a/docs/docs/cli-reference.md +++ b/docs/docs/cli-reference.md @@ -89,6 +89,7 @@ Run `spacetime help publish` for more detailed information. Default value: `.` * `-b`, `--bin-path ` — 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 ` — 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). diff --git a/smoketests/__init__.py b/smoketests/__init__.py index b853ae83c27..63486c09b08 100644 --- a/smoketests/__init__.py +++ b/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] diff --git a/smoketests/tests/auto_migration.py b/smoketests/tests/auto_migration.py index f338e718c22..ba2da98fafe 100644 --- a/smoketests/tests/auto_migration.py +++ b/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, + ) diff --git a/smoketests/tests/modules.py b/smoketests/tests/modules.py index da40510700b..1b45ce793cb 100644 --- a/smoketests/tests/modules.py +++ b/smoketests/tests/modules.py @@ -88,7 +88,7 @@ def test_module_update(self): self.write_module_code(self.MODULE_CODE_B) with self.assertRaises(CalledProcessError) as cm: self.publish_module(name, clear=False) - self.assertIn("Error: Database update rejected", cm.exception.stderr) + self.assertIn("Error: Pre-publish check failed", cm.exception.stderr) # Check that the old module is still running by calling say_hello self.call("say_hello")