smoketests: Smoketest enable replication for existing database

crates/cli/src/subcommands/subscribe.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 http::uri::{PathAndQuery, Scheme};
 use serde_json::Value;
 use spacetimedb_client_api_messages::websocket::{self as ws, JsonFormat};
 use spacetimedb_data_structures::map::HashMap;
@@ -65,6 +65,13 @@ pub fn cli() -> clap::Command {
                 .action(ArgAction::SetTrue)
                 .help("Print the initial update for the queries."),
         )
+        .arg(
+            Arg::new("confirmed")
+                .required(false)
+                .long("confirmed")
+                .action(ArgAction::SetTrue)
+                .help("Instruct the server to deliver only updates of confirmed transactions"),
+        )
         .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,6 +137,7 @@ 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);
@@ -146,6 +154,9 @@ pub async fn exec(config: Config, args: &ArgMatches) -> Result<(), anyhow::Error
             s
         }
     });
+    if confirmed {
+        append_query_param(&mut uri, ("confirmed", "true"));
+    }
 
     // Create the websocket request.
     let mut req = http::Uri::from_parts(uri)?.into_client_request()?;
@@ -334,3 +345,21 @@ fn format_output_json(msg: &ws::DatabaseUpdate<JsonFormat>, schema: &RawModuleDe
 
     Ok(output)
 }
+
+fn append_query_param(uri: &mut http::uri::Parts, (k, v): (&str, &str)) {
+    let (mut path, query) = uri
+        .path_and_query
+        .as_ref()
+        .map(|pq| (pq.path().to_owned(), pq.query()))
+        .unwrap_or_default();
+    path.push('?');
+    if let Some(query) = query {
+        path.push_str(query);
+        path.push('&');
+    }
+    path.push_str(k);
+    path.push('=');
+    path.push_str(v);
+
+    uri.path_and_query = Some(PathAndQuery::from_maybe_shared(path).unwrap());
+}

crates/client-api/src/lib.rs

@@ -67,14 +67,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 +116,27 @@ 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 {
+                durable_offset.wait_for(tx_offset).await.map_err(log_and_500)?;
+            }
+        }
+
         Ok(json)
     }
 

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

@@ -382,12 +382,17 @@ pub struct SqlParams {
 }
 
 #[derive(Deserialize)]
-pub struct SqlQueryParams {}
+pub struct SqlQueryParams {
+    /// If `true`, return the query result only after its transaction offset
+    /// is confirmed to be durable.
+    #[serde(default)]
+    confirmed: bool,
+}
 
 pub async fn sql<S>(
     State(worker_ctx): State<S>,
     Path(SqlParams { name_or_identity }): Path<SqlParams>,
-    Query(SqlQueryParams {}): Query<SqlQueryParams>,
+    Query(SqlQueryParams { confirmed }): Query<SqlQueryParams>,
     Extension(auth): Extension<SpacetimeAuth>,
     body: String,
 ) -> axum::response::Result<impl IntoResponse>
@@ -410,7 +415,7 @@ where
         .await
         .map_err(log_and_500)?
         .ok_or(StatusCode::NOT_FOUND)?;
-    let json = host.exec_sql(auth, database, body).await?;
+    let json = host.exec_sql(auth, database, confirmed, body).await?;
 
     let total_duration = json.iter().fold(0, |acc, x| acc + x.total_duration_micros);