Add some tests of procedure concurrency

Cargo.toml

@@ -48,6 +48,7 @@ members = [
   "modules/sdk-test",
   "modules/sdk-test-connect-disconnect",
   "modules/sdk-test-procedure",
+  "modules/sdk-test-procedure-concurrency",
   "modules/sdk-test-view",
   "modules/sdk-test-case-conversion",
   "modules/sdk-test-view-pk",
@@ -57,6 +58,7 @@ members = [
   "sdks/rust/tests/test-counter",
   "sdks/rust/tests/connect_disconnect_client",
   "sdks/rust/tests/procedure-client",
+  "sdks/rust/tests/procedure-concurrency-client",
   "sdks/rust/tests/view-client",
   "sdks/rust/tests/case-conversion-client",
   "sdks/rust/tests/view-pk-client",

modules/sdk-test-procedure-concurrency/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "sdk-test-procedure-concurrency-module"
+version = "0.1.0"
+edition.workspace = true
+license-file = "LICENSE"
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+log.workspace = true
+anyhow.workspace = true
+paste.workspace = true
+
+[dependencies.spacetimedb]
+workspace = true
+features = ["unstable"]

modules/sdk-test-procedure-concurrency/README.md

@@ -0,0 +1,8 @@
+# `sdk-test-procedure-concurrency` *Rust* test
+
+This module isolates procedure concurrency behavior that currently only has
+Rust module coverage.
+
+It is separate from [`sdk-test-procedure`](../sdk-test-procedure) so the shared
+procedure test suite can continue targeting other module languages without also
+requiring `ctx.sleep_until` support.

modules/sdk-test-procedure-concurrency/src/lib.rs

@@ -0,0 +1,89 @@
+use spacetimedb::{
+    procedure, reducer, table, DbContext, ProcedureContext, ReducerContext, ScheduleAt, Table, TxContext,
+};
+use std::time::Duration;
+
+#[table(public, accessor = procedure_concurrency_row)]
+struct ProcedureConcurrencyRow {
+    #[auto_inc]
+    insertion_order: u32,
+    insertion_context: String,
+}
+
+fn insert_procedure_concurrency_row(ctx: &TxContext, insertion_context: &str) {
+    ctx.db.procedure_concurrency_row().insert(ProcedureConcurrencyRow {
+        insertion_order: 0,
+        insertion_context: insertion_context.into(),
+    });
+}
+
+#[reducer]
+fn insert_reducer_row(ctx: &ReducerContext) {
+    ctx.db().procedure_concurrency_row().insert(ProcedureConcurrencyRow {
+        insertion_order: 0,
+        insertion_context: "reducer".into(),
+    });
+}
+
+#[procedure]
+fn procedure_sleep_between_inserts(ctx: &mut ProcedureContext) {
+    ctx.with_tx(|ctx| insert_procedure_concurrency_row(ctx, "procedure_before"));
+    ctx.sleep_until(ctx.timestamp + Duration::from_secs(10));
+    ctx.with_tx(|ctx| insert_procedure_concurrency_row(ctx, "procedure_after"));
+}
+
+#[table(accessor = scheduled_reducer_row, scheduled(insert_scheduled_reducer))]
+struct ScheduledReducerRow {
+    #[primary_key]
+    #[auto_inc]
+    scheduled_id: u64,
+    scheduled_at: ScheduleAt,
+}
+
+#[reducer]
+fn insert_scheduled_reducer(ctx: &ReducerContext, _schedule: ScheduledReducerRow) {
+    ctx.db().procedure_concurrency_row().insert(ProcedureConcurrencyRow {
+        insertion_order: 0,
+        insertion_context: "scheduled_reducer".into(),
+    });
+}
+
+#[procedure]
+fn procedure_schedule_reducer_between_inserts(ctx: &mut ProcedureContext) {
+    ctx.with_tx(|ctx| {
+        insert_procedure_concurrency_row(ctx, "procedure_before");
+        ctx.db.scheduled_reducer_row().insert(ScheduledReducerRow {
+            scheduled_id: 0,
+            scheduled_at: ctx.timestamp.into(),
+        });
+    });
+    ctx.sleep_until(ctx.timestamp + Duration::from_secs(10));
+    ctx.with_tx(|ctx| insert_procedure_concurrency_row(ctx, "procedure_after"));
+}
+
+#[table(accessor = scheduled_procedure_row, scheduled(scheduled_procedure_sleep_between_inserts))]
+struct ScheduledProcedureRow {
+    #[primary_key]
+    #[auto_inc]
+    scheduled_id: u64,
+    scheduled_at: ScheduleAt,
+}
+
+#[procedure]
+fn scheduled_procedure_sleep_between_inserts(ctx: &mut ProcedureContext, _schedule: ScheduledProcedureRow) {
+    ctx.with_tx(|ctx| insert_procedure_concurrency_row(ctx, "scheduled_procedure_before"));
+    ctx.sleep_until(ctx.timestamp + Duration::from_secs(10));
+    ctx.with_tx(|ctx| insert_procedure_concurrency_row(ctx, "scheduled_procedure_after"));
+}
+
+#[reducer]
+fn schedule_procedure_then_reducer(ctx: &ReducerContext) {
+    ctx.db().scheduled_procedure_row().insert(ScheduledProcedureRow {
+        scheduled_id: 0,
+        scheduled_at: ctx.timestamp.into(),
+    });
+    ctx.db().scheduled_reducer_row().insert(ScheduledReducerRow {
+        scheduled_id: 0,
+        scheduled_at: (ctx.timestamp + Duration::from_secs(2)).into(),
+    });
+}

sdks/rust/tests/procedure-concurrency-client/Cargo.toml

@@ -0,0 +1,44 @@
+[package]
+name = "procedure-concurrency-client"
+version.workspace = true
+edition.workspace = true
+license-file = "LICENSE"
+
+[lib]
+crate-type = ["cdylib", "rlib"]
+
+[features]
+default = ["native"]
+
+native = [
+    "dep:env_logger",
+    "dep:tokio",
+]
+
+browser = [
+    "spacetimedb-sdk/browser",
+    "dep:wasm-bindgen",
+    "dep:wasm-bindgen-futures",
+    "dep:console_error_panic_hook",
+    "dep:futures",
+]
+
+[[bin]]
+name = "procedure-concurrency-client"
+path = "src/main.rs"
+required-features = ["native"]
+
+[dependencies]
+spacetimedb-sdk = { path = "../.." }
+test-counter = { path = "../test-counter" }
+anyhow.workspace = true
+env_logger = { workspace = true, optional = true }
+tokio = { workspace = true, optional = true }
+futures = { workspace = true, optional = true }
+
+wasm-bindgen = { version = "0.2.100", optional = true }
+wasm-bindgen-futures = { version = "0.4.45", optional = true }
+console_error_panic_hook = { version = "0.1.7", optional = true }
+
+[lints]
+workspace = true

sdks/rust/tests/procedure-concurrency-client/README.md

@@ -0,0 +1,6 @@
+This test client is used with the module:
+
+- [`sdk-test-procedure-concurrency`](/modules/sdk-test-procedure-concurrency)
+
+The goal of the test is to exercise the current reducer/procedure concurrency
+behavior of the Rust module ABI and the Rust SDK.

sdks/rust/tests/procedure-concurrency-client/src/lib.rs

@@ -0,0 +1,14 @@
+#![allow(clippy::disallowed_macros)]
+
+mod module_bindings;
+pub mod test_handlers;
+
+#[cfg(all(target_arch = "wasm32", feature = "browser"))]
+use wasm_bindgen::prelude::wasm_bindgen;
+
+#[cfg(all(target_arch = "wasm32", feature = "browser"))]
+#[wasm_bindgen]
+pub async fn run(test_name: String, db_name: String) {
+    console_error_panic_hook::set_once();
+    test_handlers::dispatch(&test_name, &db_name).await;
+}

sdks/rust/tests/procedure-concurrency-client/src/main.rs

@@ -0,0 +1,23 @@
+use procedure_concurrency_client::test_handlers;
+
+fn exit_on_panic() {
+    let default_hook = std::panic::take_hook();
+    std::panic::set_hook(Box::new(move |panic_info| {
+        default_hook(panic_info);
+        std::process::exit(1);
+    }));
+}
+
+fn main() {
+    env_logger::init();
+    exit_on_panic();
+
+    let test = std::env::args()
+        .nth(1)
+        .expect("Pass a test name as a command-line argument to the test client");
+    let db_name = std::env::var("SPACETIME_SDK_TEST_DB_NAME").expect("Failed to read db name from env");
+
+    tokio::runtime::Runtime::new()
+        .unwrap()
+        .block_on(test_handlers::dispatch(&test, &db_name));
+}