[bfops/bump-version]: Merge remote-tracking branch 'origin/master' into bfops/bump-version

Author: bfops

Cargo.lock

@@ -268,7 +268,7 @@ tokio = { version = "1.37", features = ["full"] }
 tokio_metrics = { version = "0.4.0" }
 tokio-postgres = { version = "0.7.8", features = ["with-chrono-0_4"] }
 tokio-stream = "0.1.17"
-tokio-tungstenite = { version = "0.27.0", features = ["native-tls"] }
+tokio-tungstenite = { version = "0.27.0", features = ["native-tls", "url"] }
 tokio-util = { version = "0.7.4", features = ["time"] }
 toml = "0.8"
 toml_edit = "0.22.22"

Cargo.toml

@@ -29,3 +29,11 @@ pub fn yes() -> Arg {
         .action(SetTrue)
         .help("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).")
 }
+
+pub fn confirmed() -> Arg {
+    Arg::new("confirmed")
+        .required(false)
+        .long("confirmed")
+        .action(SetTrue)
+        .help("Instruct the server to deliver only updates of confirmed transactions")
+}

crates/cli/src/common_args.rs

@@ -3,7 +3,7 @@ use std::process::ExitCode;
 use clap::{Arg, Command};
 use spacetimedb_cli::*;
 use spacetimedb_paths::cli::CliTomlPath;
-use spacetimedb_paths::{RootDir, SpacetimePaths};
+use spacetimedb_paths::RootDir;
 
 // Note that the standalone server is invoked through standaline/src/main.rs, so you will
 // also want to set the allocator there.
@@ -24,6 +24,8 @@ static GLOBAL: MiMalloc = MiMalloc;
 #[cfg(not(feature = "markdown-docs"))]
 #[tokio::main]
 async fn main() -> anyhow::Result<ExitCode> {
+    use spacetimedb_paths::SpacetimePaths;
+
     // Compute matches before loading the config, because `Config` has an observable `drop` method
     // (which deletes a lockfile),
     // and Clap calls `exit` on parse failure rather than panicking, so destructors never run.

crates/cli/src/main.rs

@@ -34,6 +34,7 @@ pub fn cli() -> clap::Command {
                 .conflicts_with("query")
                 .help("Instead of using a query, run an interactive command prompt for `SQL` expressions"),
         )
+        .arg(common_args::confirmed())
         .arg(common_args::anonymous())
         .arg(common_args::server().help("The nickname, host name or URL of the server hosting the database"))
         .arg(common_args::yes())
@@ -178,11 +179,15 @@ pub async fn exec(config: Config, args: &ArgMatches) -> Result<(), anyhow::Error
         crate::repl::exec(con).await?;
     } else {
         let query = args.get_one::<String>("query").unwrap();
+        let confirmed = args.get_flag("confirmed");
 
         let con = parse_req(config, args).await?;
-        let api = ClientApi::new(con);
+        let mut api = ClientApi::new(con).sql();
+        if confirmed {
+            api = api.query(&[("confirmed", "true")]);
+        }
 
-        run_sql(api.sql(), query, false).await?;
+        run_sql(api, query, false).await?;
     }
     Ok(())
 }

crates/cli/src/subcommands/sql.rs

@@ -2,7 +2,7 @@ use anyhow::Context;
 use clap::{value_parser, Arg, ArgAction, ArgMatches};
 use futures::{Sink, SinkExt, TryStream, TryStreamExt};
 use http::header;
-use http::uri::Scheme;
+use reqwest::Url;
 use serde_json::Value;
 use spacetimedb_client_api_messages::websocket::{self as ws, JsonFormat};
 use spacetimedb_data_structures::map::HashMap;
@@ -65,6 +65,7 @@ pub fn cli() -> clap::Command {
                 .action(ArgAction::SetTrue)
                 .help("Print the initial update for the queries."),
         )
+        .arg(common_args::confirmed())
         .arg(common_args::anonymous())
         .arg(common_args::yes())
         .arg(common_args::server().help("The nickname, host name or URL of the server hosting the database"))
@@ -130,25 +131,26 @@ pub async fn exec(config: Config, args: &ArgMatches) -> Result<(), anyhow::Error
     let num = args.get_one::<u32>("num-updates").copied();
     let timeout = args.get_one::<u32>("timeout").copied();
     let print_initial_update = args.get_flag("print_initial_update");
+    let confirmed = args.get_flag("confirmed");
 
     let conn = parse_req(config, args).await?;
     let api = ClientApi::new(conn);
     let module_def = api.module_def().await?;
 
+    let mut url = Url::parse(&api.con.db_uri("subscribe"))?;
     // Change the URI scheme from `http(s)` to `ws(s)`.
-    let mut uri = http::Uri::try_from(api.con.db_uri("subscribe"))?.into_parts();
-    uri.scheme = uri.scheme.map(|s| {
-        if s == Scheme::HTTP {
-            "ws".parse().unwrap()
-        } else if s == Scheme::HTTPS {
-            "wss".parse().unwrap()
-        } else {
-            s
-        }
-    });
+    url.set_scheme(match url.scheme() {
+        "http" => "ws",
+        "https" => "wss",
+        unknown => unreachable!("Invalid URL scheme in `Connection::db_uri`: {unknown}"),
+    })
+    .unwrap();
+    if confirmed {
+        url.query_pairs_mut().append_pair("confirmed", "true");
+    }
 
     // Create the websocket request.
-    let mut req = http::Uri::from_parts(uri)?.into_client_request()?;
+    let mut req = url.into_client_request()?;
     req.headers_mut().insert(
         header::SEC_WEBSOCKET_PROTOCOL,
         http::HeaderValue::from_static(ws::TEXT_PROTOCOL),

crates/cli/src/subcommands/subscribe.rs

@@ -106,6 +106,27 @@ pub enum PublishResult {
     PermissionDenied { name: DatabaseName },
 }
 
+#[derive(serde::Serialize, serde::Deserialize, Debug, Default)]
+pub enum MigrationPolicy {
+    #[default]
+    Compatible,
+    BreakClients,
+}
+
+#[derive(serde::Serialize, serde::Deserialize, Debug, Default)]
+pub enum PrettyPrintStyle {
+    #[default]
+    AnsiColor,
+    NoColor,
+}
+
+#[derive(serde::Serialize, serde::Deserialize, Debug)]
+pub struct PrintPlanResult {
+    pub migrate_plan: Box<str>,
+    pub break_clients: bool,
+    pub token: spacetimedb_lib::Hash,
+}
+
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum DnsLookupResponse {
     /// The lookup was successful and the domain and identity are returned.

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

@@ -7,14 +7,15 @@ use http::StatusCode;
 
 use spacetimedb::client::ClientActorIndex;
 use spacetimedb::energy::{EnergyBalance, EnergyQuanta};
-use spacetimedb::host::{HostController, ModuleHost, NoSuchModule, UpdateDatabaseResult};
+use spacetimedb::host::{HostController, MigratePlanResult, ModuleHost, NoSuchModule, UpdateDatabaseResult};
 use spacetimedb::identity::{AuthCtx, Identity};
 use spacetimedb::messages::control_db::{Database, HostType, Node, Replica};
 use spacetimedb::sql;
 use spacetimedb_client_api_messages::http::{SqlStmtResult, SqlStmtStats};
 use spacetimedb_client_api_messages::name::{DomainName, InsertDomainResult, RegisterTldResult, SetDomainsResult, Tld};
 use spacetimedb_lib::{ProductTypeElement, ProductValue};
 use spacetimedb_paths::server::ModuleLogsDir;
+use spacetimedb_schema::auto_migrate::{MigrationPolicy, PrettyPrintStyle};
 use tokio::sync::watch;
 
 pub mod auth;
@@ -67,14 +68,15 @@ impl Host {
         &self,
         auth: AuthCtx,
         database: Database,
+        confirmed_read: bool,
         body: String,
     ) -> axum::response::Result<Vec<SqlStmtResult<ProductValue>>> {
         let module_host = self
             .module()
             .await
             .map_err(|_| (StatusCode::NOT_FOUND, "module not found".to_string()))?;
 
-        let json = self
+        let (tx_offset, durable_offset, json) = self
             .host_controller
             .using_database(
                 database,
@@ -115,17 +117,28 @@ impl Host {
                         .map(|(col_name, col_type)| ProductTypeElement::new(col_type, Some(col_name)))
                         .collect();
 
-                    Ok(vec![SqlStmtResult {
-                        schema,
-                        rows: result.rows,
-                        total_duration_micros: total_duration.as_micros() as u64,
-                        stats: SqlStmtStats::from_metrics(&result.metrics),
-                    }])
+                    Ok((
+                        result.tx_offset,
+                        db.durable_tx_offset(),
+                        vec![SqlStmtResult {
+                            schema,
+                            rows: result.rows,
+                            total_duration_micros: total_duration.as_micros() as u64,
+                            stats: SqlStmtStats::from_metrics(&result.metrics),
+                        }],
+                    ))
                 },
             )
             .await
             .map_err(log_and_500)??;
 
+        if confirmed_read {
+            if let Some(mut durable_offset) = durable_offset {
+                let tx_offset = tx_offset.await.map_err(|_| log_and_500("transaction aborted"))?;
+                durable_offset.wait_for(tx_offset).await.map_err(log_and_500)?;
+            }
+        }
+
         Ok(json)
     }
 
@@ -134,9 +147,10 @@ impl Host {
         database: Database,
         host_type: HostType,
         program_bytes: Box<[u8]>,
+        policy: MigrationPolicy,
     ) -> anyhow::Result<UpdateDatabaseResult> {
         self.host_controller
-            .update_module_host(database, host_type, self.replica_id, program_bytes)
+            .update_module_host(database, host_type, self.replica_id, program_bytes, policy)
             .await
     }
 }
@@ -219,8 +233,11 @@ pub trait ControlStateWriteAccess: Send + Sync {
         &self,
         publisher: &Identity,
         spec: DatabaseDef,
+        policy: MigrationPolicy,
     ) -> anyhow::Result<Option<UpdateDatabaseResult>>;
 
+    async fn migrate_plan(&self, spec: DatabaseDef, style: PrettyPrintStyle) -> anyhow::Result<MigratePlanResult>;
+
     async fn delete_database(&self, caller_identity: &Identity, database_identity: &Identity) -> anyhow::Result<()>;
 
     // Energy
@@ -309,8 +326,13 @@ impl<T: ControlStateWriteAccess + ?Sized> ControlStateWriteAccess for Arc<T> {
         &self,
         identity: &Identity,
         spec: DatabaseDef,
+        policy: MigrationPolicy,
     ) -> anyhow::Result<Option<UpdateDatabaseResult>> {
-        (**self).publish_database(identity, spec).await
+        (**self).publish_database(identity, spec, policy).await
+    }
+
+    async fn migrate_plan(&self, spec: DatabaseDef, style: PrettyPrintStyle) -> anyhow::Result<MigratePlanResult> {
+        (**self).migrate_plan(spec, style).await
     }
 
     async fn delete_database(&self, caller_identity: &Identity, database_identity: &Identity) -> anyhow::Result<()> {