Refactor smoketest publish and subscribe helpers to builder pattern

crates/smoketests/tests/smoketests/add_remove_index.rs

@@ -19,25 +19,25 @@ fn test_add_then_remove_index() {
 
     // Publish and attempt a subscribing to a join query.
     // There are no indices, resulting in an unsupported unindexed join.
-    test.publish_module_named(&name, false).unwrap();
-    let result = test.subscribe(&[JOIN_QUERY], 0);
+    test.publish().name(&name).run().unwrap();
+    let result = test.subscribe(&[JOIN_QUERY]).expect_rows(0).run();
     assert!(result.is_err(), "Expected subscription to fail without indices");
 
     // Publish the indexed version.
     // Now we have indices, so the query should be accepted.
     test.use_precompiled_module("add-remove-index-indexed");
-    test.publish_module_named(&name, false).unwrap();
+    test.publish().name(&name).run().unwrap();
 
     // Subscribe and hold across the call, then collect results
-    let sub = test.subscribe_background(&[JOIN_QUERY], 1).unwrap();
+    let sub = test.subscribe(&[JOIN_QUERY]).expect_rows(1).background().unwrap();
     test.call_anon("add", &[]).unwrap();
     let results = sub.collect().unwrap();
     assert_eq!(results.len(), 1, "Expected 1 update from subscription");
 
     // Publish the unindexed version again, removing the index.
     // The initial subscription should be rejected again.
     test.use_precompiled_module("add-remove-index");
-    test.publish_module_named(&name, false).unwrap();
-    let result = test.subscribe(&[JOIN_QUERY], 0);
+    test.publish().name(&name).run().unwrap();
+    let result = test.subscribe(&[JOIN_QUERY]).expect_rows(0).run();
     assert!(result.is_err(), "Expected subscription to fail after removing indices");
 }

crates/smoketests/tests/smoketests/auto_migration.rs

@@ -50,7 +50,7 @@ fn test_reject_schema_changes() {
 
     // Try to update with incompatible schema (adding column without default)
     test.write_module_code(MODULE_CODE_UPDATED_INCOMPATIBLE).unwrap();
-    let result = test.publish_module_clear(false);
+    let result = test.publish().current_database().unwrap().run();
 
     assert!(
         result.is_err(),
@@ -201,7 +201,11 @@ pub fn print_books(ctx: &ReducerContext, prefix: String) {
 fn test_add_table_auto_migration() {
     let mut test = Smoketest::builder().module_code(MODULE_CODE_INIT).build();
 
-    let sub = test.subscribe_background(&["select * from person"], 4).unwrap();
+    let sub = test
+        .subscribe(&["select * from person"])
+        .expect_rows(4)
+        .background()
+        .unwrap();
 
     // Add initial data
     test.call("add_person", &["Robert", "Student"]).unwrap();
@@ -228,7 +232,7 @@ fn test_add_table_auto_migration() {
 
     // Update module without clearing database
     test.write_module_code(MODULE_CODE_UPDATED).unwrap();
-    test.publish_module_clear(false).unwrap();
+    test.publish().current_database().unwrap().run().unwrap();
 
     // Add new data with updated schema
     test.call("add_person", &["Husserl", "Student"]).unwrap();
@@ -344,10 +348,7 @@ fn test_add_table_columns() {
     let mut subs = Vec::with_capacity(NUM_SUBSCRIBERS);
     for _ in 0..NUM_SUBSCRIBERS {
         // The migration below should disconnect all existing subscribers.
-        subs.push(
-            test.subscribe_background_until_closed(&["select * from person"])
-                .unwrap(),
-        );
+        subs.push(test.subscribe(&["select * from person"]).background().unwrap());
     }
 
     // Insert under initial schema
@@ -356,7 +357,7 @@ fn test_add_table_columns() {
     // First upgrade: add age & mass columns
     test.write_module_code(MODULE_CODE_ADD_TABLE_COLUMNS_UPDATED).unwrap();
     let identity = test.database_identity.clone().unwrap();
-    test.publish_module_with_options(&identity, false, true).unwrap();
+    test.publish().name(&identity).break_clients(true).run().unwrap();
     test.call("print_persons", &["FIRST_UPDATE"]).unwrap();
 
     let logs1 = test.logs(100).unwrap();
@@ -394,7 +395,7 @@ fn test_add_table_columns() {
     // Second upgrade
     test.write_module_code(MODULE_CODE_ADD_TABLE_COLUMNS_UPDATED_AGAIN)
         .unwrap();
-    test.publish_module_with_options(&identity, false, true).unwrap();
+    test.publish().name(&identity).break_clients(true).run().unwrap();
     test.call("print_persons", &["UPDATE_2"]).unwrap();
 
     let logs2 = test.logs(100).unwrap();
@@ -477,12 +478,18 @@ fn test_remove_primary_key_issue_3934() {
 
     // Step 2: Remove primary key. Should succeed.
     test.write_module_code(MODULE_CODE_WITHOUT_PK).unwrap();
-    test.publish_module_with_options(&identity, false, true)
+    test.publish()
+        .name(&identity)
+        .break_clients(true)
+        .run()
         .expect("Removing primary key should succeed");
 
     // Step 3: Trivial change (add a reducer). This is where #3934 crashes.
     test.write_module_code(MODULE_CODE_WITHOUT_PK_V2).unwrap();
-    test.publish_module_with_options(&identity, false, true)
+    test.publish()
+        .name(&identity)
+        .break_clients(true)
+        .run()
         .expect("Publish after PK removal should succeed (issue #3934)");
 }
 
@@ -532,12 +539,18 @@ fn automigrate_reschema_event_table_arbitrarily() {
 
     // Step 2: Reschema event table. Should work fine, even though we'd reject this change for a non-event table.
     test.write_module_code(MODULE_CODE_WITH_EVENT_TABLE_AFTER).unwrap();
-    test.publish_module_with_options(&identity, false, true)
+    test.publish()
+        .name(&identity)
+        .break_clients(true)
+        .run()
         .expect("Changing schema of event table should succeed");
 
     // Step 3: Reschema event table right back. Should still work fine.
     test.write_module_code(MODULE_CODE_WITH_EVENT_TABLE_BEFORE).unwrap();
-    test.publish_module_with_options(&identity, false, true)
+    test.publish()
+        .name(&identity)
+        .break_clients(true)
+        .run()
         .expect("Changing schema of event table should succeed");
 }
 
@@ -608,7 +621,10 @@ fn automigrate_drop_event_table_replays_after_restart() {
 
     // Drop the event table.
     test.write_module_code(MODULE_CODE_DROP_EVENT_TABLE_AFTER).unwrap();
-    test.publish_module_with_options(&identity, false, true)
+    test.publish()
+        .name(&identity)
+        .break_clients(true)
+        .run()
         .expect("Dropping the event table should succeed");
 
     // Wait until data written after the drop is durable,

crates/smoketests/tests/smoketests/change_host_type.rs

@@ -1,5 +1,5 @@
 use spacetimedb::messages::control_db::HostType;
-use spacetimedb_smoketests::{require_local_server, require_pnpm, Smoketest};
+use spacetimedb_smoketests::{require_local_server, require_pnpm, ModuleLanguage, Smoketest};
 
 const TS_MODULE_BASIC: &str = r#"import { schema, t, table } from "spacetimedb/server";
 
@@ -35,11 +35,14 @@ fn test_update_with_different_host_type() {
         .autopublish(false)
         .build();
 
-    let database_identity = test.publish_module().unwrap();
+    let database_identity = test.publish().run().unwrap();
     add_person(&test, PERSON_A, "initial");
 
     // Publish a TS module.
-    test.publish_typescript_module_source_clear("modules-basic-ts", &database_identity, TS_MODULE_BASIC, false)
+    test.publish()
+        .name(&database_identity)
+        .source(ModuleLanguage::TypeScript, "modules-basic-ts", TS_MODULE_BASIC)
+        .run()
         .unwrap();
     add_person(&test, PERSON_B, "post module update");
 
@@ -48,7 +51,7 @@ fn test_update_with_different_host_type() {
     assert_has_rows(&test, &[PERSON_A, PERSON_B], "post restart");
 
     // Change back to original module and assert that the data is still there.
-    test.publish_module_clear(false).unwrap();
+    test.publish().current_database().unwrap().run().unwrap();
     add_person(&test, PERSON_C, "post revert");
 
     // Restart once more and assert that the data is still there.
@@ -99,7 +102,10 @@ fn test_repair_host_type() {
 
     let mut test = Smoketest::builder().autopublish(false).build();
 
-    test.publish_typescript_module_source("modules-basic-ts", "basic-ts-change-host-type", TS_MODULE_BASIC)
+    test.publish()
+        .name("basic-ts-change-host-type")
+        .source(ModuleLanguage::TypeScript, "modules-basic-ts", TS_MODULE_BASIC)
+        .run()
         .unwrap();
     assert_host_type(&test, HostType::Js);
     // Set the program kind to the wrong value.

crates/smoketests/tests/smoketests/cli/list.rs

@@ -28,7 +28,7 @@ fn cli_list_shows_database_names_and_identities() {
     let primary_name = format!("list-db-{}", std::process::id());
     let alias_name = format!("{primary_name}-alias");
     let second_alias_name = format!("{primary_name}-alt");
-    let identity = test.publish_module_named(&primary_name, false).unwrap();
+    let identity = test.publish().name(&primary_name).run().unwrap();
 
     let json_body = format!(r#"["{}","{}"]"#, alias_name, second_alias_name);
     let response = test

crates/smoketests/tests/smoketests/client_connection_errors.rs

@@ -8,7 +8,7 @@ fn test_client_connected_error_rejects_connection() {
         .build();
 
     // Subscribe should fail because client_connected returns an error
-    let result = test.subscribe(&["SELECT * FROM all_u8s"], 0);
+    let result = test.subscribe(&["SELECT * FROM all_u8s"]).expect_rows(0).run();
     assert!(
         result.is_err(),
         "Expected subscribe to fail when client_connected returns error"
@@ -35,7 +35,7 @@ fn test_client_disconnected_error_still_deletes_st_client() {
         .build();
 
     // Subscribe should succeed (client_connected returns Ok)
-    let result = test.subscribe(&["SELECT * FROM all_u8s"], 0);
+    let result = test.subscribe(&["SELECT * FROM all_u8s"]).expect_rows(0).run();
     assert!(result.is_ok(), "Expected subscribe to succeed");
 
     let logs = test.logs(100).unwrap();

crates/smoketests/tests/smoketests/confirmed_reads.rs

@@ -11,7 +11,10 @@ fn test_confirmed_reads_receive_updates() {
 
     // Start subscription in background with confirmed flag
     let sub = test
-        .subscribe_background_confirmed(&["SELECT * FROM person"], 2)
+        .subscribe(&["SELECT * FROM person"])
+        .expect_rows(2)
+        .confirmed(true)
+        .background()
         .unwrap();
 
     // Insert via reducer

crates/smoketests/tests/smoketests/delete_database.rs

@@ -21,7 +21,7 @@ fn test_delete_database_aborts_without_confirmation() {
         .build();
 
     let name = format!("delete-db-abort-{}", std::process::id());
-    test.publish_module_named(&name, false).unwrap();
+    test.publish().name(&name).run().unwrap();
 
     let output = test
         .spacetime(&["delete", "--server", &test.server_url, &name])
@@ -43,12 +43,10 @@ fn test_delete_database_with_confirmation() {
         .build();
 
     let name = format!("delete-database-{}", std::process::id());
-    test.publish_module_named(&name, false).unwrap();
+    test.publish().name(&name).run().unwrap();
 
     // Start subscription in background to collect updates until deleting the database closes it.
-    let sub = test
-        .subscribe_background_until_closed(&["SELECT * FROM counter"])
-        .unwrap();
+    let sub = test.subscribe(&["SELECT * FROM counter"]).background().unwrap();
 
     // Let the scheduled reducer run for a bit
     thread::sleep(Duration::from_secs(2));
@@ -79,7 +77,7 @@ fn test_delete_database_yes_skips_confirmation() {
         .build();
 
     let name = format!("delete-db-yes-{}", std::process::id());
-    test.publish_module_named(&name, false).unwrap();
+    test.publish().name(&name).run().unwrap();
 
     let output = test
         .spacetime(&["delete", "--server", &test.server_url, "--yes", &name])

crates/smoketests/tests/smoketests/dml.rs

@@ -9,7 +9,11 @@ fn test_subscribe() {
     let test = Smoketest::builder().precompiled_module("dml").build();
 
     // Start subscription FIRST (in background), matching Python semantics
-    let sub = test.subscribe_background(&["SELECT * FROM t"], 2).unwrap();
+    let sub = test
+        .subscribe(&["SELECT * FROM t"])
+        .expect_rows(2)
+        .background()
+        .unwrap();
 
     // Small delay to ensure subscription is connected before inserts
     thread::sleep(Duration::from_millis(500));

crates/smoketests/tests/smoketests/domains.rs

@@ -6,7 +6,7 @@ fn test_set_name() {
     let mut test = Smoketest::builder().autopublish(false).build();
 
     let orig_name = format!("domains-set-name-{}", std::process::id());
-    test.publish_module_named(&orig_name, false).unwrap();
+    test.publish().name(&orig_name).run().unwrap();
 
     let rand_name = format!("domains-set-name-{}-renamed", std::process::id());
 
@@ -34,16 +34,16 @@ fn test_subdomain_behavior() {
     let mut test = Smoketest::builder().autopublish(false).build();
 
     let root_name = format!("domains-subdomain-behavior-{}", std::process::id());
-    test.publish_module_named(&root_name, false).unwrap();
+    test.publish().name(&root_name).run().unwrap();
 
     // Double slash should fail
     let double_slash_name = format!("{}//test", root_name);
-    let result = test.publish_module_named(&double_slash_name, false);
+    let result = test.publish().name(&double_slash_name).run();
     assert!(result.is_err(), "Expected publish to fail with double slash in name");
 
     // Trailing slash should fail
     let trailing_slash_name = format!("{}/test/", root_name);
-    let result = test.publish_module_named(&trailing_slash_name, false);
+    let result = test.publish().name(&trailing_slash_name).run();
     assert!(result.is_err(), "Expected publish to fail with trailing slash in name");
 }
 
@@ -53,12 +53,12 @@ fn test_set_to_existing_name() {
     let mut test = Smoketest::builder().autopublish(false).build();
 
     // Publish first database (no name)
-    test.publish_module().unwrap();
+    test.publish().run().unwrap();
     let id_to_rename = test.database_identity.clone().unwrap();
 
     // Publish second database with a name
     let rename_to = format!("domains-set-existing-target-{}", std::process::id());
-    test.publish_module_named(&rename_to, false).unwrap();
+    test.publish().name(&rename_to).run().unwrap();
 
     // Try to rename first db to the name of the second - should fail
     let result = test.spacetime(&[
@@ -83,7 +83,7 @@ fn test_replace_names() {
     let orig_name = format!("domains-replace-names-{}", std::process::id());
     let alt_name1 = format!("domains-replace-names-{}-alt1", std::process::id());
     let alt_name2 = format!("domains-replace-names-{}-alt2", std::process::id());
-    test.publish_module_named(&orig_name, false).unwrap();
+    test.publish().name(&orig_name).run().unwrap();
 
     // Use the API to replace names
     let json_body = format!(r#"["{}","{}"]"#, alt_name1, alt_name2);

crates/smoketests/tests/smoketests/fail_initial_publish.rs

@@ -27,7 +27,7 @@ fn test_fail_initial_publish() {
     let name = format!("fail-initial-publish-{}", std::process::id());
 
     // First publish should fail due to broken module
-    let result = test.publish_module_named(&name, false);
+    let result = test.publish().name(&name).run();
     assert!(result.is_err(), "Expected publish to fail with broken module");
 
     // Describe should fail because database doesn't exist
@@ -47,7 +47,7 @@ fn test_fail_initial_publish() {
     // This used to be broken; the failed initial publish would leave
     // the control database in a bad state.
     test.use_precompiled_module("fail-initial-publish-fixed");
-    test.publish_module_named(&name, false).unwrap();
+    test.publish().name(&name).run().unwrap();
 
     let describe_output = test
         .spacetime(&["describe", "--server", &test.server_url, "--json", &name])
@@ -61,7 +61,7 @@ fn test_fail_initial_publish() {
     // Publishing the broken code again fails, but the database still exists afterwards,
     // with the previous version of the module code.
     test.write_module_code(MODULE_CODE_BROKEN).unwrap();
-    let result = test.publish_module_named(&name, false);
+    let result = test.publish().name(&name).run();
     assert!(result.is_err(), "Expected publish to fail with broken module");
 
     // Database should still exist with the fixed code