diff --git a/Cargo.lock b/Cargo.lock index 1a540dbad83..ca736dc72ea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8819,6 +8819,7 @@ name = "spacetimedb-sdk" version = "2.12.0" dependencies = [ "anymap3", + "base64 0.21.7", "brotli 3.5.0", "bytes", "flate2", @@ -8828,6 +8829,7 @@ dependencies = [ "gloo-console", "gloo-net", "gloo-storage", + "gloo-timers", "gloo-utils", "home", "http", @@ -8836,6 +8838,8 @@ dependencies = [ "native-tls", "once_cell", "prometheus", + "rand 0.9.2", + "serde_json", "shlex", "spacetimedb-client-api-messages", "spacetimedb-data-structures", diff --git a/crates/codegen/src/rust.rs b/crates/codegen/src/rust.rs index cea7ea87dd9..82c065adb6f 100644 --- a/crates/codegen/src/rust.rs +++ b/crates/codegen/src/rust.rs @@ -1795,6 +1795,10 @@ impl __sdk::DbContext for DbConnection {{ self.imp.is_active() }} + fn is_reconnecting(&self) -> bool {{ + self.imp.is_reconnecting() + }} + fn disconnect(&self) -> __sdk::Result<()> {{ self.imp.disconnect() }} @@ -2122,6 +2126,10 @@ impl __sdk::DbContext for {struct_and_trait_name} {{ self.imp.is_active() }} + fn is_reconnecting(&self) -> bool {{ + self.imp.is_reconnecting() + }} + fn disconnect(&self) -> __sdk::Result<()> {{ self.imp.disconnect() }} diff --git a/crates/codegen/tests/snapshots/codegen__codegen_rust.snap b/crates/codegen/tests/snapshots/codegen__codegen_rust.snap index af115ee46b5..6628d4f4095 100644 --- a/crates/codegen/tests/snapshots/codegen__codegen_rust.snap +++ b/crates/codegen/tests/snapshots/codegen__codegen_rust.snap @@ -1602,6 +1602,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -1821,6 +1825,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -1897,6 +1905,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -1969,6 +1981,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -2041,6 +2057,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -2117,6 +2137,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/docs/docs/00100-intro/00300-tutorials/00100-chat-app.md b/docs/docs/00100-intro/00300-tutorials/00100-chat-app.md index 85a00f88829..02d0cb95ad8 100644 --- a/docs/docs/00100-intro/00300-tutorials/00100-chat-app.md +++ b/docs/docs/00100-intro/00300-tutorials/00100-chat-app.md @@ -2452,13 +2452,13 @@ To `src/main.rs`, add: ```rust /// Our `on_connect_error` callback: print the error, then exit the process. -fn on_connect_error(_ctx: &ErrorContext, err: Error) { +fn on_connect_error(_ctx: &ErrorContext, err: Error, _next: Option) { eprintln!("Connection error: {:?}", err); std::process::exit(1); } /// Our `on_disconnect` callback: print a note, then exit the process. -fn on_disconnected(_ctx: &ErrorContext, err: Option) { +fn on_disconnected(_ctx: &ErrorContext, err: Option, _next: Option) { if let Some(err) = err { eprintln!("Disconnected: {}", err); std::process::exit(1); diff --git a/docs/docs/00200-core-concepts/00600-clients/00300-connection.md b/docs/docs/00200-core-concepts/00600-clients/00300-connection.md index 3101810230e..2bb448b1c6b 100644 --- a/docs/docs/00200-core-concepts/00600-clients/00300-connection.md +++ b/docs/docs/00200-core-concepts/00600-clients/00300-connection.md @@ -232,7 +232,7 @@ Failure to advance the connection means your client will not receive any updates ::: -In Rust and TypeScript, the connection processes messages automatically via the browser's event loop or Node.js's event loop. No manual polling is required. +TypeScript processes messages through the browser or Node.js event loop. Rust applications must advance their connection using `frame_tick()`, `run_threaded()`, `run_async()`, or the browser target's `run_background_task()`. ## Connection Lifecycle @@ -327,10 +327,10 @@ let conn = DbConnection::builder() println!("Connected! Saving token..."); // Save token for reconnection }) - .on_connect_error(|_ctx, error| { + .on_connect_error(|_ctx, error, _next| { eprintln!("Connection failed: {}", error); }) - .on_disconnect(|_ctx, error| { + .on_disconnect(|_ctx, error, _next| { if let Some(err) = error { eprintln!("Disconnected with error: {}", err); } else { @@ -450,7 +450,19 @@ Without automatic reconnect, create a new connection after a connection loss. Se -Automatic reconnect support is coming soon. For now, implement reconnection in your application: create a new connection and restore subscriptions after a connection loss. +Enable `.with_automatic_reconnect()` to recover after an established connection drops. The same connection, table handles, callbacks, and subscription handles survive. While `is_reconnecting()` is true, cache reads return the last known rows. **Keep advancing the connection during outages**, using `frame_tick()`, `run_threaded()`, or `run_async()`; do not stop ticking when `is_active()` becomes false. + +`on_connect` fires only for the initial connection. Register `on_automatic_reconnect` for successful automatic reconnects. Both receive the connection, identity, and token. Subscriptions replay together in one atomic batch, and their `on_applied` callbacks fire again; unchanged rows do not fire row callbacks. + +`on_disconnect(|ctx, error, next| ...)` and `on_connect_error(|ctx, error, next| ...)` receive `Option`, with the next attempt number and `Duration`, or `None` when no retry is scheduled. Initial connection failures are not retried. `disconnect()` cancels recovery. + +Use `.with_automatic_reconnect_options(AutomaticReconnectOptions { min_delay, max_delay })` to configure backoff. Defaults are one second and 30 seconds, with exponential growth and 50% jitter clamped to the bounds. The SDK raises `min_delay` to 500 ms if it is lower, then raises `max_delay` to the greater of one second and `min_delay` if necessary. + +For expiring credentials, combine `.with_token(Some(initial_token))` and `.with_token_provider(|| async { /* return Result */ })`. The provider refreshes near expiry or after rejection and must return a token for the same identity. Provider failures retry; rejection of a fresh token and identity changes are terminal. + +Function calls (i.e. reducers/procedures) made while reconnecting fail with a disconnected error. Calls interrupted by a drop may have executed: callback errors expose `is_unknown_result()`. Calls are never replayed. Subscription creation is queued during recovery, and cancellation removes a set from replay. + +Regenerate Rust bindings to expose `is_reconnecting()`. Lifecycle error callbacks gain the third argument; subscription `on_applied` closures must implement `FnMut`. `on_connect` remains `FnOnce`. Automatic reconnect requires a server supporting session replacement and batch subscriptions. diff --git a/docs/docs/00200-core-concepts/00600-clients/00500-rust-reference.md b/docs/docs/00200-core-concepts/00600-clients/00500-rust-reference.md index 83b03542645..c0fbd3959e5 100644 --- a/docs/docs/00200-core-concepts/00600-clients/00500-rust-reference.md +++ b/docs/docs/00200-core-concepts/00600-clients/00500-rust-reference.md @@ -142,7 +142,7 @@ This interface may change in an upcoming release as we rework SpacetimeDB's auth impl DbConnectionBuilder { fn on_connect_error( self, - callback: impl FnOnce(&ErrorContext, spacetimedb_sdk::Error), + callback: impl FnMut(&ErrorContext, spacetimedb_sdk::Error, Option), ) -> DbConnectionBuilder; } ``` @@ -155,13 +155,27 @@ Chain a call to `.on_connect_error(callback)` to your builder to register a call impl DbConnectionBuilder { fn on_disconnect( self, - callback: impl FnOnce(&ErrorContext, Option), + callback: impl FnMut(&ErrorContext, Option, Option), ) -> DbConnectionBuilder; } ``` Chain a call to `.on_disconnect(callback)` to your builder to register a callback to run when your established `DbConnection` disconnects from the remote database, either as a result of a call to [`disconnect`](#method-disconnect) or due to an error. +#### Automatic reconnection + +Enable `with_automatic_reconnect()` on the builder to retain the connection, cache, subscriptions, and callbacks across transient outages. `with_automatic_reconnect_options(AutomaticReconnectOptions { min_delay, max_delay })` changes the default one-second minimum and 30-second maximum delays. Retries use exponential backoff with 50% jitter clamped to these bounds. The SDK enforces a 500 ms minimum and a maximum of at least one second and `min_delay`. + +Register `on_automatic_reconnect(|conn, identity, token| { ... })` for each successful reconnect. It accepts `FnMut`; `on_connect` remains a one-shot initial-connection callback. The third parameter of `on_disconnect` and `on_connect_error` is `Some(NextReconnect { attempt, delay })` when a retry is scheduled, and `None` when the connection has ended. Initial connection failures do not retry. + +A reconnect reuses the retained token. Use `with_token_provider(|| async { /* return spacetimedb_sdk::Result */ })` to refresh expiring tokens. Refresh happens when remaining validity is at most 5% of token lifetime or 30 seconds, whichever is larger, when expiry cannot be read, or after the retained token is rejected. A provider failure retries. A freshly returned token that is rejected, a changed identity, and fatal protocol errors stop recovery. + +`is_reconnecting()` is true while waiting for or establishing a reconnect; `is_active()` is false. Keep advancing the connection during this interval. Cache reads remain available. Existing subscriptions replay in one batch and call `on_applied` again. Row callbacks report only net changes. Subscriptions created during an outage are queued; unsubscribed handles are excluded from replay. + +Reducer calls while reconnecting return `Error::Disconnected`; procedure callbacks receive a disconnected error. In-flight calls fail with an `InternalError` for which `is_unknown_result()` is true, because they may have executed before the connection dropped. The SDK never retries these calls. Synthetic reducer error events use `Timestamp::UNIX_EPOCH` because no server completion timestamp is known. Explicit `disconnect()` cancels all reconnect work. + +Regenerate client bindings when upgrading to expose `is_reconnecting()`. Update lifecycle error callbacks to accept their third parameter and make subscription `on_applied` callbacks repeatable. The server must support session replacement and batch subscriptions. + #### Method `with_token` ```rust @@ -361,7 +375,7 @@ Subscribe to queries by calling `ctx.subscription_builder()` and chaining config ```rust impl SubscriptionBuilder { - fn on_applied(self, callback: impl FnOnce(&SubscriptionEventContext)) -> Self; + fn on_applied(self, callback: impl FnMut(&SubscriptionEventContext)) -> Self; } ``` diff --git a/modules/sdk-test-procedure-cpp/client/src/main.rs b/modules/sdk-test-procedure-cpp/client/src/main.rs index 752f91d622a..65a3091ffd4 100644 --- a/modules/sdk-test-procedure-cpp/client/src/main.rs +++ b/modules/sdk-test-procedure-cpp/client/src/main.rs @@ -66,13 +66,13 @@ fn on_connected(_ctx: &DbConnection, identity: Identity, token: &str) { } /// Our `on_connect_error` callback: print the error, then exit the process. -fn on_connect_error(_ctx: &ErrorContext, err: Error) { +fn on_connect_error(_ctx: &ErrorContext, err: Error, _next: Option) { eprintln!("Connection error: {err}"); std::process::exit(1); } /// Our `on_disconnect` callback: print a note, then exit the process. -fn on_disconnected(_ctx: &ErrorContext, err: Option) { +fn on_disconnected(_ctx: &ErrorContext, err: Option, _next: Option) { if let Some(err) = err { eprintln!("Disconnected: {err}"); std::process::exit(1); diff --git a/modules/sdk-test-procedure-cpp/client/src/module_bindings/mod.rs b/modules/sdk-test-procedure-cpp/client/src/module_bindings/mod.rs index 53e6dd3f0a8..705170164c2 100644 --- a/modules/sdk-test-procedure-cpp/client/src/module_bindings/mod.rs +++ b/modules/sdk-test-procedure-cpp/client/src/module_bindings/mod.rs @@ -299,6 +299,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -528,6 +532,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -614,6 +622,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -696,6 +708,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -778,6 +794,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -864,6 +884,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/sdks/rust/Cargo.toml b/sdks/rust/Cargo.toml index 7f1f24f6ff7..dd3b47f33f9 100644 --- a/sdks/rust/Cargo.toml +++ b/sdks/rust/Cargo.toml @@ -17,6 +17,7 @@ browser = [ "dep:gloo-net", "dep:gloo-storage", "dep:gloo-utils", + "dep:gloo-timers", "dep:js-sys", "dep:tokio-tungstenite-wasm", "dep:wasm-bindgen", @@ -33,6 +34,10 @@ spacetimedb-metrics.workspace = true spacetimedb-query-builder.workspace = true thiserror.workspace = true +base64.workspace = true +serde_json.workspace = true +rand.workspace = true +gloo-timers = { version = "0.3", features = ["futures"], optional = true } anymap3 = "1.0.1" brotli.workspace = true bytes.workspace = true diff --git a/sdks/rust/src/callbacks.rs b/sdks/rust/src/callbacks.rs index a3fb083af1e..6da30881a1b 100644 --- a/sdks/rust/src/callbacks.rs +++ b/sdks/rust/src/callbacks.rs @@ -217,7 +217,7 @@ pub(crate) type ReducerCallback = Box< /// function. Users should not reference this struct directly. pub(crate) struct ReducerCallbacks { /// Maps WebSocket `request_id` to the arguments of that call and user-supplied callback. - callbacks: HashMap)>, + pub(crate) callbacks: HashMap)>, } impl Default for ReducerCallbacks { @@ -247,7 +247,7 @@ pub(crate) type ProcedureCallback = Box::ProcedureEventContext, Result) + Send + 'static>; pub struct ProcedureCallbacks { - request_id_to_callback: HashMap>, + pub(crate) request_id_to_callback: HashMap>, } impl Default for ProcedureCallbacks { diff --git a/sdks/rust/src/client_cache.rs b/sdks/rust/src/client_cache.rs index ab0a90980bd..8132323671e 100644 --- a/sdks/rust/src/client_cache.rs +++ b/sdks/rust/src/client_cache.rs @@ -341,6 +341,10 @@ pub struct ClientCache { /// /// The strings are table names, since we may have multiple tables with the same row type. tables: Map, + // One synthetic delete builder per table. Auto-reconnect combines these deletes + // with new subscription inserts and applies them through the existing DbUpdate code. + synthetic_delete_builders: + HashMap<&'static str, fn(&Self, &'static str) -> spacetimedb_client_api_messages::websocket::v2::TableUpdate>, /// Clone of the [`crate::db_connection::DbContextImpl::extra_logging`]. extra_logging: Option>, @@ -352,11 +356,19 @@ impl ClientCache { pub(crate) fn new(extra_logging: Option>) -> Self { Self { tables: Map::new(), + synthetic_delete_builders: HashMap::default(), extra_logging, _module: PhantomData, } } + pub(crate) fn build_synthetic_deletes(&self) -> Vec { + self.synthetic_delete_builders + .iter() + .map(|(&name, build_deletes)| build_deletes(self, name)) + .collect() + } + /// Get a handle on the [`TableCache`] which stores rows of type `Row` for the table `table_name`. pub(crate) fn get_table + Send + Sync + 'static>( &self, @@ -373,6 +385,29 @@ impl ClientCache { &mut self, table_name: &'static str, ) -> &mut TableCache { + self.synthetic_delete_builders + .entry(table_name) + .or_insert(|cache, name| { + use spacetimedb_client_api_messages::websocket::{common::RowSizeHint, v2}; + let mut bytes = Vec::new(); + let mut offsets = Vec::new(); + if let Some(table) = cache.get_table::(name) { + for (bsatn, entry) in &table.entries { + for _ in 0..entry.ref_count { + offsets.push(bytes.len() as u64); + bytes.extend_from_slice(bsatn); + } + } + } + v2::TableUpdate { + table_name: name.into(), + rows: vec![v2::TableUpdateRows::PersistentTable(v2::PersistentTableRows { + inserts: v2::BsatnRowList::default(), + deletes: v2::BsatnRowList::new(RowSizeHint::RowOffsets(offsets.into()), bytes.into()), + })] + .into(), + } + }); self.tables .entry::>>() .or_default() diff --git a/sdks/rust/src/db_connection.rs b/sdks/rust/src/db_connection.rs index c1810f7e5dd..72fbec501e6 100644 --- a/sdks/rust/src/db_connection.rs +++ b/sdks/rust/src/db_connection.rs @@ -28,10 +28,14 @@ use crate::{ spacetime_module::{AbstractEventContext, AppliedDiff, DbConnection, DbUpdate, InModule, SpacetimeModule}, subscription::{PendingUnsubscribeResult, SubscriptionHandleImpl, SubscriptionManager}, websocket::{WsConnection, WsParams}, - Event, ReducerEvent, Status, + AutomaticReconnectOptions, Event, NextReconnect, ReducerEvent, Status, TokenProvider, }; use bytes::Bytes; -use futures::StreamExt; +use futures::{ + future::{AbortHandle, Abortable}, + StreamExt, +}; + #[cfg(feature = "browser")] use futures::{pin_mut, FutureExt}; use futures_channel::mpsc; @@ -103,6 +107,7 @@ pub struct DbContextImpl { /// This may be none if we have not yet received the [`ws::v2::InitialConnection`] message. connection_id: SharedCell>, + reconnect: SharedCell, pub(crate) extra_logging: Option>, } @@ -123,6 +128,7 @@ impl Clone for DbContextImpl { identity: Arc::clone(&self.identity), connection_id: Arc::clone(&self.connection_id), extra_logging: Option::>::clone(&self.extra_logging), + reconnect: self.reconnect.clone(), } } } @@ -135,54 +141,115 @@ impl DbContextImpl { /// Process a parsed WebSocket message, /// applying its mutations to the client cache and invoking callbacks. fn process_message(&self, msg: ParsedMessage) -> crate::Result<()> { + let result = self.process_message_inner(msg); + let ended = self.inner.lock().unwrap().connection_lifecycle == ConnectionLifecycle::Ended; + match result { + Err(crate::Error::Disconnected) if !ended => self.connection_lost(None, false, false), + Err(error) if !ended => self.connection_lost(Some(error), true, false), + result => result, + } + } + + fn process_message_inner(&self, msg: ParsedMessage) -> crate::Result<()> { + // Discard queued messages after explicit disconnect or a terminal failure. + if self.inner.lock().unwrap().connection_lifecycle == ConnectionLifecycle::Ended { + return Err(crate::Error::Disconnected); + } self.debug_log(|out| writeln!(out, "`process_message`: {msg:?}")); match msg { // Error: route as a connection error if we never finished connecting, // otherwise treat it as an erroneous disconnect. - ParsedMessage::Error(e) => Err(self.end_connection(Some(e))), - - // Initial `IdentityToken` message: - // confirm that the received identity and connection ID are what we expect, - // store them, then invoke the on_connect callback. + ParsedMessage::Error(e) => self.connection_lost(Some(e), true, false), + ParsedMessage::TransportError(e) => { + let terminal = e.is_terminal(); + let auth = e.is_auth(); + // A normal close is EOF, not a failed connection. Keep other close + // codes as errors so their terminal/retry classification is retained. + let error = match e { + crate::websocket::WsError::Closed { code: 1000, .. } => None, + e => Some(InternalError::new(e.to_string()).with_cause(e).into()), + }; + self.connection_lost(error, terminal, auth) + } ParsedMessage::IdentityToken(identity, token, conn_id) => { - let on_connect = { - let mut inner = self.inner.lock().unwrap(); - match inner.connection_lifecycle { - ConnectionLifecycle::Connecting => { - inner.connection_lifecycle = ConnectionLifecycle::Connected; - inner.on_connect.take() - } - ConnectionLifecycle::Connected => None, - ConnectionLifecycle::Ended => return Ok(()), - } + if self.inner.lock().unwrap().connection_lifecycle == ConnectionLifecycle::Ended { + return Ok(()); + } + if self + .identity + .lock() + .unwrap() + .is_some_and(|previous| previous != identity) + { + return self.connection_lost( + Some(InternalError::new("Reconnect returned a different identity").into()), + true, + false, + ); + } + let reconnecting = { + let mut state = self.reconnect.lock().unwrap(); + let reconnecting = state.reconnecting; + state.reconnecting = false; + state.attempt = 0; + state.token = Some(token.to_string()); + reconnecting }; + *self.identity.lock().unwrap() = Some(identity); + *self.connection_id.lock().unwrap() = Some(conn_id); { - // Don't hold the `self.identity` lock while running callbacks. - // Callbacks can (will) call [`DbContext::identity`], which acquires that lock, - // so holding it while running a callback causes deadlocks. - let mut ident_store = self.identity.lock().unwrap(); - if let Some(prev_identity) = *ident_store { - assert_eq!(prev_identity, identity); + let mut inner = self.inner.lock().unwrap(); + if inner.connection_lifecycle == ConnectionLifecycle::Connected { + drop(inner); + return self.connection_lost( + Some(InternalError::new("Duplicate InitialConnection").into()), + true, + false, + ); } - *ident_store = Some(identity); + inner.connection_lifecycle = ConnectionLifecycle::Connected; } - { - // Don't hold the `self.connection_id` lock while running callbacks. - // Callbacks can (will) call [`DbContext::connection_id`], which acquires that lock, - // so holding it while running a callback causes deadlocks. - let mut conn_id_store = self.connection_id.lock().unwrap(); - // This would only happen if the client is using the unstable `set_connection_id` method. - if let Some(prev_conn_id) = *conn_id_store { - assert_eq!(prev_conn_id, conn_id); + let ctx = ::new(self.clone()); + if reconnecting { + let callback = self.inner.lock().unwrap().on_automatic_reconnect.take(); + if let Some(mut callback) = callback { + callback(&ctx, identity, &token); + self.inner.lock().unwrap().on_automatic_reconnect = Some(callback); + } + } else { + let callback = self.inner.lock().unwrap().on_connect.take(); + if let Some(callback) = callback { + callback(&ctx, identity, &token); } - *conn_id_store = Some(conn_id); } - if let Some(on_connect) = on_connect { - let ctx = ::new(self.clone()); - on_connect(&ctx, identity, &token); + if reconnecting { + // Include subscriptions queued from the reconnect callback in the batch. + self.reconnect.lock().unwrap().preparing_replay = true; + self.apply_pending_mutations()?; + if self.inner.lock().unwrap().connection_lifecycle != ConnectionLifecycle::Connected { + return Ok(()); + } + let mut inner = self.inner.lock().unwrap(); + let sets = inner.subscriptions.replay(&self.make_event_ctx(())); + let request_id = next_request_id(); + inner.replay = Some((request_id, sets.iter().map(|set| set.query_set_id).collect())); + self.reconnect.lock().unwrap().preparing_replay = false; + drop(inner); + if sets.is_empty() { + self.apply_replay(ws::v2::SubscribeBatchApplied { + request_id, + results: Box::new([]), + })?; + } else { + self.send_message(ws::v2::ClientMessage::SubscribeBatch(ws::v2::SubscribeBatch { + request_id, + sets, + }))?; + } } Ok(()) } + ParsedMessage::SubscribeBatchApplied(batch) => self.apply_replay(batch), // Transaction update: // apply the received diff to the client cache, @@ -312,49 +379,225 @@ impl DbContextImpl { applied_diff.invoke_row_callbacks(&row_event_ctx, &mut inner.db_callbacks); } - /// Mark the connection lifecycle as ended, route the terminal event to the - /// appropriate connection callback, and mark [`Self::is_active`] false. - /// - /// Returns the terminal error that should be returned from `advance_*` methods. - fn end_connection(&self, callback_error: Option) -> crate::Error { - let mut inner = self.inner.lock().unwrap(); - let return_error = callback_error.clone().unwrap_or(crate::Error::Disconnected); + fn send_message(&self, message: ws::v2::ClientMessage) -> crate::Result<()> { + self.send_chan + .lock() + .unwrap() + .as_ref() + .ok_or(crate::Error::Disconnected)? + .unbounded_send(message) + .map_err(|_| crate::Error::Disconnected) + } - let lifecycle = inner.connection_lifecycle; - if lifecycle == ConnectionLifecycle::Ended { - return return_error; + fn apply_replay(&self, batch: ws::v2::SubscribeBatchApplied) -> crate::Result<()> { + let expected = self.inner.lock().unwrap().replay.take(); + let ids: Vec<_> = batch.results.iter().map(|result| result.query_set_id).collect(); + if expected != Some((batch.request_id, ids)) { + return self.connection_lost( + Some(InternalError::new("Unexpected subscription replay response").into()), + true, + false, + ); } - inner.connection_lifecycle = ConnectionLifecycle::Ended; - - // Set `send_chan` to `None`, since `Self::is_active` checks that. - *self.send_chan.lock().unwrap() = None; - - match lifecycle { - ConnectionLifecycle::Connecting => { - let callback_error = callback_error.unwrap_or_else(|| crate::Error::FailedToConnect { - source: InternalError::new("Connection closed before receiving the initial connection message"), - }); - let ctx: M::ErrorContext = self.make_event_ctx(Some(callback_error.clone())); - if let Some(connect_error_callback) = inner.on_connect_error.take() { - connect_error_callback(&ctx, callback_error.clone()); + let mut tables = self.cache.lock().unwrap().build_synthetic_deletes(); + let mut outcomes = Vec::new(); + for result in batch.results { + let error = match result.outcome { + ws::v2::SubscribeSetOutcome::Applied(rows) => { + tables.extend(rows.tables.into_vec().into_iter().map(|table| { + ws::v2::TableUpdate { + table_name: table.table, + rows: vec![ws::v2::TableUpdateRows::PersistentTable(ws::v2::PersistentTableRows { + inserts: table.rows, + deletes: Default::default(), + })] + .into(), + } + })); + None } - callback_error - } - ConnectionLifecycle::Connected => { - let ctx: M::ErrorContext = self.make_event_ctx(callback_error.clone()); - if let Some(disconnect_callback) = inner.on_disconnect.take() { - disconnect_callback(&ctx, callback_error.clone()); + ws::v2::SubscribeSetOutcome::Error(error) => Some(error.to_string()), + }; + outcomes.push((result.query_set_id, error)); + } + let update = M::DbUpdate::parse_update(ws::v2::TransactionUpdate { + query_sets: vec![ws::v2::QuerySetUpdate { + query_set_id: QuerySetId { id: 0 }, + tables: tables.into(), + }] + .into(), + }); + let update = match update { + Ok(update) => update, + Err(error) => return self.connection_lost(Some(error), true, false), + }; + self.apply_update(update, |inner| { + for (id, error) in outcomes { + if let Some(error) = error { + inner.subscriptions.subscription_error( + &self.make_event_ctx(Some(crate::Error::SubscriptionError { error })), + id, + ); + } else { + inner.subscriptions.subscription_applied(&self.make_event_ctx(()), id); } + } + Event::SubscribeApplied + }); + Ok(()) + } - // Call the `on_disconnect` method for all subscriptions. - inner.subscriptions.on_disconnect(&ctx); + fn fail_pending_operations(&self, inner: &mut DbContextImplInner) { + let error = InternalError::new("Connection lost before the result was received") + .with_cause(crate::Error::UnknownResult); + for (_, (reducer, callback)) in std::mem::take(&mut inner.reducer_callbacks.callbacks) { + let ctx = self.make_event_ctx(ReducerEvent { + reducer, + timestamp: Timestamp::UNIX_EPOCH, + status: Status::Panic(error.clone()), + }); + callback(&ctx, Err(error.clone())); + } + for (_, callback) in std::mem::take(&mut inner.procedure_callbacks.request_id_to_callback) { + callback(&self.make_event_ctx(()), Err(error.clone())); + } + } - return_error + fn connection_lost(&self, error: Option, terminal: bool, auth: bool) -> crate::Result<()> { + let lifecycle = self.inner.lock().unwrap().connection_lifecycle; + if lifecycle == ConnectionLifecycle::Ended { + return Err(crate::Error::Disconnected); + } + *self.send_chan.lock().unwrap() = None; + let next = { + let mut state = self.reconnect.lock().unwrap(); + let can_retry = state.options.is_some() + && self.identity.lock().unwrap().is_some() + && !terminal + && !(auth && (state.provider.is_none() || state.used_fresh_token)); + if can_retry { + state.force_refresh |= auth; + state.reconnecting = true; + state.preparing_replay = true; + state.attempt = state.attempt.saturating_add(1); + Some(NextReconnect { + attempt: state.attempt, + delay: state.options.unwrap().delay(state.attempt, rand::random()), + }) + } else { + state.reconnecting = false; + state.preparing_replay = false; + state.generation += 1; + if let Some(abort) = state.abort.take() { + abort.abort(); + } + None + } + }; + if let Some(next) = next { + self.schedule_reconnect(next); + } + let mut inner = self.inner.lock().unwrap(); + inner.connection_lifecycle = if next.is_some() { + ConnectionLifecycle::Connecting + } else { + ConnectionLifecycle::Ended + }; + inner.replay = None; + self.fail_pending_operations(&mut inner); + if next.is_some() { + inner.subscriptions.suspend(&self.make_event_ctx(())); + } else { + inner.subscriptions.on_disconnect(&self.make_event_ctx(())); + } + let callback_error = if lifecycle == ConnectionLifecycle::Connecting { + Some(error.unwrap_or_else(|| InternalError::new("Connection closed before InitialConnection").into())) + } else { + error + }; + let ctx = self.make_event_ctx(callback_error.clone()); + if lifecycle == ConnectionLifecycle::Connected { + if let Some(callback) = inner.on_disconnect.as_mut() { + callback(&ctx, callback_error.clone(), next); } - ConnectionLifecycle::Ended => return_error, + } else if let Some(callback) = inner.on_connect_error.as_mut() { + callback(&ctx, callback_error.clone().unwrap(), next); + } + if next.is_some() { + Ok(()) + } else { + Err(callback_error.unwrap_or(crate::Error::Disconnected)) } } + fn schedule_reconnect(&self, next: NextReconnect) { + let (park_send, park_recv) = mpsc::unbounded(); + *get_lock_sync(&self.recv) = park_recv; + let mut state = self.reconnect.lock().unwrap(); + state.generation += 1; + let generation = state.generation; + let config = state.config.clone(); + let mut token = state.token.clone(); + let provider = state.provider.clone(); + let force_refresh = state.force_refresh; + let (abort, registration) = AbortHandle::new_pair(); + state.abort = Some(abort); + drop(state); + let pending = self.pending_mutations_send.clone(); + let extra_logging = self.extra_logging.clone(); + #[cfg(not(feature = "browser"))] + let runtime = self.runtime.clone(); + let task = async move { + #[cfg(not(feature = "browser"))] + tokio::time::sleep(next.delay).await; + #[cfg(feature = "browser")] + gloo_timers::future::TimeoutFuture::new(next.delay.as_millis().min(u32::MAX as u128) as u32).await; + let provider = provider.filter(|_| { + force_refresh || crate::reconnect::token_needs_refresh(token.as_deref(), crate::reconnect::now()) + }); + let fresh = provider.is_some(); + let result = async { + if let Some(provider) = provider { + token = Some(provider.token().await.map_err(ReconnectFailure::Provider)?); + if token.as_ref().is_none_or(|token| token.is_empty()) { + return Err(ReconnectFailure::Provider( + InternalError::new("Token provider returned an empty token").into(), + )); + } + } + let socket = + WsConnection::connect(config.uri, &config.database_name, token.as_deref(), None, config.params) + .await + .map_err(ReconnectFailure::Transport)?; + #[cfg(not(feature = "browser"))] + let (_, raw_recv, send) = socket.spawn_message_loop(&runtime, extra_logging.clone()); + #[cfg(feature = "browser")] + let (raw_recv, send) = socket.spawn_message_loop(); + #[cfg(not(feature = "browser"))] + let (_, recv) = spawn_parse_loop(raw_recv, &runtime, extra_logging); + #[cfg(feature = "browser")] + let recv = spawn_parse_loop(raw_recv, extra_logging); + Ok((send, recv)) + } + .await; + let _ = pending.unbounded_send(PendingMutation::ReconnectReady { + _park_send: park_send, + generation, + token, + fresh, + result, + }); + }; + #[cfg(not(feature = "browser"))] + self.runtime.spawn(async move { + let _ = Abortable::new(task, registration).await; + }); + #[cfg(feature = "browser")] + wasm_bindgen_futures::spawn_local(async move { + let _ = Abortable::new(task, registration).await; + }); + } + fn make_event_ctx>(&self, event: E) -> Ctx { let imp = self.clone(); Ctx::new(imp, event) @@ -371,28 +614,75 @@ impl DbContextImpl { /// Apply an individual [`PendingMutation`]. fn apply_mutation(&self, mutation: PendingMutation) -> crate::Result<()> { + let explicit_disconnect = matches!(&mutation, PendingMutation::Disconnect); + match self.apply_mutation_inner(mutation) { + Err(crate::Error::Disconnected) if !explicit_disconnect => self.connection_lost(None, false, false), + result => result, + } + } + + fn apply_mutation_inner(&self, mutation: PendingMutation) -> crate::Result<()> { self.debug_log(|out| writeln!(out, "`apply_mutation`: {mutation:?}")); match mutation { + PendingMutation::ReconnectReady { + _park_send, + generation, + token, + fresh, + result, + } => { + { + let mut state = self.reconnect.lock().unwrap(); + if generation != state.generation { + return Ok(()); + } + state.used_fresh_token = fresh; + if fresh && !matches!(&result, Err(ReconnectFailure::Provider(_))) { + state.token = token; + state.force_refresh = false; + } + } + match result { + Ok((send, recv)) => { + *self.send_chan.lock().unwrap() = Some(send); + *get_lock_sync(&self.recv) = recv; + } + Err(ReconnectFailure::Provider(error)) => return self.connection_lost(Some(error), false, false), + Err(ReconnectFailure::Transport(error)) => { + let terminal = error.is_terminal(); + let auth = error.is_auth(); + return self.connection_lost( + Some(InternalError::new(error.to_string()).with_cause(error).into()), + terminal, + auth, + ); + } + } + } + // Subscribe: register the subscription in the [`SubscriptionManager`] // and send the `Subscribe` WS message. PendingMutation::Subscribe { query_set_id, handle } => { let mut inner = self.inner.lock().unwrap(); // Register the subscription, so we can handle related messages from the server. inner.subscriptions.register_subscription(query_set_id, handle.clone()); + if self.reconnect.lock().unwrap().preparing_replay { + return Ok(()); + } if let Some(msg) = handle.start() { - self.send_chan - .lock() - .unwrap() - .as_mut() - .ok_or(crate::Error::Disconnected)? - .unbounded_send(ws::v2::ClientMessage::Subscribe(msg)) - .expect("Unable to send subscribe message: WS sender loop has dropped its recv channel"); + self.send_message(ws::v2::ClientMessage::Subscribe(msg))?; } // else, the handle was already cancelled. } PendingMutation::Unsubscribe { query_set_id } => { let mut inner = self.inner.lock().unwrap(); + if self.reconnect.lock().unwrap().preparing_replay { + inner + .subscriptions + .unsubscribe_applied(&self.make_event_ctx(()), query_set_id); + return Ok(()); + } match inner.subscriptions.handle_pending_unsubscribe(query_set_id) { PendingUnsubscribeResult::DoNothing => // The subscription was already unsubscribed, so we don't need to send an unsubscribe message. @@ -404,19 +694,25 @@ impl DbContextImpl { callback(&self.make_event_ctx(())); } PendingUnsubscribeResult::SendUnsubscribe(m) => { - self.send_chan - .lock() - .unwrap() - .as_mut() - .ok_or(crate::Error::Disconnected)? - .unbounded_send(ws::v2::ClientMessage::Unsubscribe(m)) - .expect("Unable to send unsubscribe message: WS sender loop has dropped its recv channel"); + self.send_message(ws::v2::ClientMessage::Unsubscribe(m))?; } } } // CallReducer: send the `CallReducer` WS message. PendingMutation::InvokeReducerWithCallback { reducer, callback } => { + if !self.is_active() { + let error = InternalError::new("Disconnected").with_cause(crate::Error::Disconnected); + callback( + &self.make_event_ctx(ReducerEvent { + reducer, + timestamp: Timestamp::UNIX_EPOCH, + status: Status::Panic(error.clone()), + }), + Err(error), + ); + return Ok(()); + } let request_id = next_request_id(); let reducer_name = reducer.reducer_name(); @@ -437,13 +733,7 @@ impl DbContextImpl { request_id, flags, }); - self.send_chan - .lock() - .unwrap() - .as_mut() - .ok_or(crate::Error::Disconnected)? - .unbounded_send(msg) - .expect("Unable to send reducer call message: WS sender loop has dropped its recv channel"); + self.send_message(msg)?; } // Invoke a procedure: stash its callback, then send the `CallProcedure` WS message. @@ -452,6 +742,13 @@ impl DbContextImpl { args, callback, } => { + if !self.is_active() { + callback( + &self.make_event_ctx(()), + Err(InternalError::new("Disconnected").with_cause(crate::Error::Disconnected)), + ); + return Ok(()); + } // We need to include a request_id in the message so that we can find the callback once it completes. let request_id = next_request_id(); self.inner @@ -466,31 +763,33 @@ impl DbContextImpl { request_id, flags: ws::v2::CallProcedureFlags::Default, }); - self.send_chan - .lock() - .unwrap() - .as_mut() - .ok_or(crate::Error::Disconnected)? - .unbounded_send(msg) - .expect("Unable to send procedure call message: WS sender loop has dropped its recv channel"); + self.send_message(msg)?; } // Disconnect: close the connection. PendingMutation::Disconnect => { { - let mut inner = self.inner.lock().unwrap(); - if inner.connection_lifecycle == ConnectionLifecycle::Connecting { - // If the user cancels before the initial connection finishes, - // don't report that as a connection error. - inner.connection_lifecycle = ConnectionLifecycle::Ended; + let mut state = self.reconnect.lock().unwrap(); + state.generation += 1; + state.reconnecting = false; + state.preparing_replay = false; + if let Some(abort) = state.abort.take() { + abort.abort(); } } - // Set `send_chan` to `None`, since `Self::is_active` checks that. - // This will close the WebSocket loop in websocket.rs, - // sending a close frame to the server, - // eventually resulting in disconnect callbacks being called - // if the initial connection had completed. *self.send_chan.lock().unwrap() = None; + let mut inner = self.inner.lock().unwrap(); + let notify_disconnect = + inner.connection_lifecycle != ConnectionLifecycle::Ended && self.identity.lock().unwrap().is_some(); + inner.connection_lifecycle = ConnectionLifecycle::Ended; + inner.replay = None; + self.fail_pending_operations(&mut inner); + let ctx = self.make_event_ctx(None); + inner.subscriptions.on_disconnect(&self.make_event_ctx(())); + if notify_disconnect && let Some(callback) = inner.on_disconnect.as_mut() { + callback(&ctx, None, None); + } + return Err(crate::Error::Disconnected); } // Callback stuff: these all do what you expect. @@ -578,8 +877,9 @@ impl DbContextImpl { // returns `Err(_)`. Similar behavior as `Iterator::next` and // `Stream::poll_next`. No comment on whether this is a good mental // model or not. - let res = match get_lock_sync(&self.recv).try_next() { - Ok(None) => Err(self.end_connection(None)), + let incoming = get_lock_sync(&self.recv).try_next(); + let res = match incoming { + Ok(None) => self.connection_lost(None, false, false).map(|_| true), Err(_) => Ok(false), Ok(Some(msg)) => self.process_message(msg).map(|_| true), }; @@ -634,7 +934,7 @@ impl DbContextImpl { pub fn advance_one_message_blocking(&self) -> crate::Result<()> { match self.runtime.block_on(self.get_message()) { Message::Local(pending) => self.apply_mutation(pending), - Message::Ws(None) => Err(self.end_connection(None)), + Message::Ws(None) => self.connection_lost(None, false, false), Message::Ws(Some(msg)) => self.process_message(msg), } } @@ -645,7 +945,7 @@ impl DbContextImpl { pub async fn advance_one_message_async(&self) -> crate::Result<()> { match self.get_message().await { Message::Local(pending) => self.apply_mutation(pending), - Message::Ws(None) => Err(self.end_connection(None)), + Message::Ws(None) => self.connection_lost(None, false, false), Message::Ws(Some(msg)) => self.process_message(msg), } } @@ -706,12 +1006,23 @@ impl DbContextImpl { /// Called by the autogenerated `DbConnection` method of the same name. pub fn is_active(&self) -> bool { - self.send_chan.lock().unwrap().is_some() + !self.is_reconnecting() + && self + .send_chan + .lock() + .unwrap() + .as_ref() + .is_some_and(|sender| !sender.is_closed()) + } + + /// True while waiting for or establishing an automatic reconnect. + pub fn is_reconnecting(&self) -> bool { + self.reconnect.lock().unwrap().reconnecting } /// Called by the autogenerated `DbConnection` method of the same name. pub fn disconnect(&self) -> crate::Result<()> { - if !self.is_active() { + if !self.is_active() && !self.is_reconnecting() { return Err(crate::Error::Disconnected); } self.pending_mutations_send @@ -754,6 +1065,9 @@ impl DbContextImpl { where ::Reducer: From, { + if !self.is_active() { + return Err(crate::Error::Disconnected); + } self.queue_mutation(PendingMutation::InvokeReducerWithCallback { reducer: reducer.into(), callback: Box::new(callback), @@ -788,6 +1102,13 @@ impl DbContextImpl { + Send + 'static, ) { + if !self.is_active() { + callback( + &self.make_event_ctx(()), + Err(InternalError::new("Disconnected").with_cause(crate::Error::Disconnected)), + ); + return; + } self.queue_mutation(PendingMutation::InvokeProcedureWithCallback { procedure: procedure_name, args: bsatn::to_vec(&args).expect("Failed to BSATN serialize procedure args"), @@ -806,10 +1127,14 @@ impl DbContextImpl { type OnConnectCallback = Box::DbConnection, Identity, &str) + Send + 'static>; -type OnConnectErrorCallback = Box::ErrorContext, crate::Error) + Send + 'static>; +type OnAutomaticReconnectCallback = + Box::DbConnection, Identity, &str) + Send + 'static>; + +type OnConnectErrorCallback = + Box::ErrorContext, crate::Error, Option) + Send + 'static>; type OnDisconnectCallback = - Box::ErrorContext, Option) + Send + 'static>; + Box::ErrorContext, Option, Option) + Send + 'static>; #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum ConnectionLifecycle { @@ -835,10 +1160,12 @@ pub(crate) struct DbContextImplInner { connection_lifecycle: ConnectionLifecycle, on_connect: Option>, + on_automatic_reconnect: Option>, on_connect_error: Option>, on_disconnect: Option>, procedure_callbacks: ProcedureCallbacks, + replay: Option<(u32, Vec)>, } /// A builder-pattern constructor for a `DbConnection` connection to the module `M`. @@ -855,12 +1182,15 @@ pub struct DbConnectionBuilder { token: Option, on_connect: Option>, + on_automatic_reconnect: Option>, on_connect_error: Option>, on_disconnect: Option>, additional_logging_path: Option, params: WsParams, + reconnect_options: Option, + token_provider: Option>, } /// This process's global connection ID, which will be attacked to all connections it makes. @@ -914,10 +1244,62 @@ impl DbConnectionBuilder { database_name: None, token: None, on_connect: None, + on_automatic_reconnect: None, on_connect_error: None, on_disconnect: None, additional_logging_path: None, params: <_>::default(), + reconnect_options: None, + token_provider: None, + } + } + + /// Enable unlimited automatic reconnect with the default backoff bounds. + pub fn with_automatic_reconnect(self) -> Self { + self.with_automatic_reconnect_options(AutomaticReconnectOptions::default()) + } + + /// Enable automatic reconnect using custom backoff bounds. + pub fn with_automatic_reconnect_options(mut self, options: AutomaticReconnectOptions) -> Self { + self.reconnect_options = Some(options.resolve()); + self + } + + pub fn with_token_provider(mut self, provider: impl TokenProvider) -> Self { + self.token_provider = Some(Arc::new(provider)); + self + } + + /// Called for each successful automatic reconnect, before subscription replay. + /// `on_connect` is reserved for the initial connection. + pub fn on_automatic_reconnect( + mut self, + callback: impl FnMut(&M::DbConnection, Identity, &str) + Send + 'static, + ) -> Self { + self.on_automatic_reconnect = Some(Box::new(callback)); + self + } + + fn reconnect_state(&mut self) -> ReconnectState { + if self.reconnect_options.is_some() { + self.params.session_id = Some(ConnectionId::from_u128(rand::random())); + } + ReconnectState { + config: ConnectionConfig { + uri: self.uri.clone().expect("with_uri is required"), + database_name: self.database_name.clone().expect("with_database_name is required"), + params: self.params, + }, + options: self.reconnect_options, + provider: self.token_provider.clone(), + token: self.token.clone(), + attempt: 0, + generation: 0, + reconnecting: false, + preparing_replay: false, + used_fresh_token: false, + force_refresh: false, + abort: None, } } @@ -962,7 +1344,8 @@ but you must call one of them, or else the connection will never progress. /// Open a WebSocket connection, build an empty client cache, &c, /// to construct a [`DbContextImpl`]. #[cfg(not(feature = "browser"))] - fn build_impl(self) -> crate::Result> { + fn build_impl(mut self) -> crate::Result> { + let reconnect = self.reconnect_state(); let extra_logging = self .additional_logging_path .map(|path| { @@ -978,7 +1361,7 @@ but you must call one of them, or else the connection will never progress. let connection_id_override = get_connection_id_override(); let ws_connection = tokio::task::block_in_place(|| { handle.block_on(WsConnection::connect( - self.uri.unwrap(), + self.uri.clone().unwrap(), self.database_name.as_ref().unwrap(), self.token.as_deref(), connection_id_override, @@ -998,7 +1381,13 @@ but you must call one of them, or else the connection will never progress. let (pending_mutations_send, pending_mutations_recv) = mpsc::unbounded(); let pending_mutations_recv = Arc::new(TokioMutex::new(pending_mutations_recv)); - let inner_ctx = build_db_ctx_inner(runtime, self.on_connect, self.on_connect_error, self.on_disconnect); + let inner_ctx = build_db_ctx_inner( + runtime, + self.on_connect, + self.on_automatic_reconnect, + self.on_connect_error, + self.on_disconnect, + ); Ok(build_db_ctx( handle, inner_ctx, @@ -1008,13 +1397,15 @@ but you must call one of them, or else the connection will never progress. pending_mutations_recv, connection_id_override, extra_logging, + reconnect, )) } /// Open a WebSocket connection, build an empty client cache, &c, /// to construct a [`DbContextImpl`]. #[cfg(feature = "browser")] - async fn build_impl(self) -> crate::Result> { + async fn build_impl(mut self) -> crate::Result> { + let reconnect = self.reconnect_state(); // The wasm/browser SDK target runs under `wasm32-unknown-unknown`, where we do not // have the native file APIs that back `with_debug_to_file`. Keeping the // shared `extra_logging` field as `None` lets the rest of the connection and @@ -1042,7 +1433,12 @@ but you must call one of them, or else the connection will never progress. let (pending_mutations_send, pending_mutations_recv) = mpsc::unbounded(); let pending_mutations_recv = Arc::new(StdMutex::new(pending_mutations_recv)); - let inner_ctx = build_db_ctx_inner(self.on_connect, self.on_connect_error, self.on_disconnect); + let inner_ctx = build_db_ctx_inner( + self.on_connect, + self.on_automatic_reconnect, + self.on_connect_error, + self.on_disconnect, + ); Ok(build_db_ctx( inner_ctx, raw_msg_send, @@ -1051,6 +1447,7 @@ but you must call one of them, or else the connection will never progress. pending_mutations_recv, connection_id_override, extra_logging, + reconnect, )) } @@ -1159,7 +1556,10 @@ Instead of registering multiple `on_connect` callbacks, register a single callba /// This callback is invoked only before the initial connection message is /// received from the host. Errors which prevent [`Self::build`] from creating /// a connection are returned by [`Self::build`] instead. - pub fn on_connect_error(mut self, callback: impl FnOnce(&M::ErrorContext, crate::Error) + Send + 'static) -> Self { + pub fn on_connect_error( + mut self, + callback: impl FnMut(&M::ErrorContext, crate::Error, Option) + Send + 'static, + ) -> Self { if self.on_connect_error.is_some() { panic!( "DbConnectionBuilder can only register a single `on_connect_error` callback. @@ -1179,7 +1579,7 @@ Instead of registering multiple `on_connect_error` callbacks, register a single /// [`Self::on_connect_error`] instead. pub fn on_disconnect( mut self, - callback: impl FnOnce(&M::ErrorContext, Option) + Send + 'static, + callback: impl FnMut(&M::ErrorContext, Option, Option) + Send + 'static, ) -> Self { if self.on_disconnect.is_some() { panic!( @@ -1198,6 +1598,7 @@ fn build_db_ctx_inner( #[cfg(not(feature = "browser"))] runtime: Option, on_connect_cb: Option>, + on_automatic_reconnect: Option>, on_connect_error_cb: Option>, on_disconnect_cb: Option>, ) -> Arc>> { @@ -1211,10 +1612,12 @@ fn build_db_ctx_inner( connection_lifecycle: ConnectionLifecycle::Connecting, on_connect: on_connect_cb, + on_automatic_reconnect, on_connect_error: on_connect_error_cb, on_disconnect: on_disconnect_cb, procedure_callbacks: ProcedureCallbacks::default(), + replay: None, })) } @@ -1230,6 +1633,7 @@ fn build_db_ctx( pending_mutations_recv: SharedAsyncCell>>, connection_id: Option, extra_logging: Option>, + reconnect: ReconnectState, ) -> DbContextImpl { let mut cache = ClientCache::new(extra_logging.clone()); M::register_tables(&mut cache); @@ -1247,6 +1651,7 @@ fn build_db_ctx( identity: Arc::new(StdMutex::new(None)), connection_id: Arc::new(StdMutex::new(connection_id)), extra_logging, + reconnect: Arc::new(StdMutex::new(reconnect)), } } @@ -1276,13 +1681,15 @@ fn enter_or_create_runtime() -> crate::Result<(Option, runtime::Handle) } } -/// Synchronous lock helper: native = blocking_lock, browser = lock().unwrap() +/// Synchronous receiver access must not overlap another connection advancement. #[cfg(not(feature = "browser"))] fn get_lock_sync(mutex: &TokioMutex) -> tokio::sync::MutexGuard<'_, T> { - mutex.blocking_lock() + mutex + .try_lock() + .expect("Concurrent connection advancement is unsupported") } -/// Synchronous lock helper: native = blocking_lock, browser = lock().unwrap() +/// Browser receiver access uses a synchronous mutex. #[cfg(feature = "browser")] fn get_lock_sync(mutex: &StdMutex) -> std::sync::MutexGuard<'_, T> { mutex.lock().unwrap() @@ -1302,7 +1709,7 @@ pub async fn get_lock_async(mutex: &StdMutex) -> std::sync::MutexGuard<'_, } #[derive(Debug)] -enum ParsedMessage { +pub(crate) enum ParsedMessage { TransactionUpdate(M::DbUpdate), IdentityToken(Identity, Box, ConnectionId), SubscribeApplied { @@ -1318,6 +1725,8 @@ enum ParsedMessage { error: String, }, Error(crate::Error), + TransportError(crate::websocket::WsError), + SubscribeBatchApplied(ws::v2::SubscribeBatchApplied), ReducerResult { request_id: u32, timestamp: Timestamp, @@ -1331,7 +1740,7 @@ enum ParsedMessage { #[cfg(not(feature = "browser"))] fn spawn_parse_loop( - raw_message_recv: mpsc::UnboundedReceiver, + raw_message_recv: mpsc::UnboundedReceiver>, handle: &runtime::Handle, extra_logging: Option>, ) -> (tokio::task::JoinHandle<()>, mpsc::UnboundedReceiver>) { @@ -1342,7 +1751,7 @@ fn spawn_parse_loop( #[cfg(feature = "browser")] fn spawn_parse_loop( - raw_message_recv: mpsc::UnboundedReceiver, + raw_message_recv: mpsc::UnboundedReceiver>, extra_logging: Option>, ) -> mpsc::UnboundedReceiver> { let (parsed_message_send, parsed_message_recv) = mpsc::unbounded(); @@ -1353,7 +1762,7 @@ fn spawn_parse_loop( /// A loop which reads raw WS messages from `recv`, parses them into domain types, /// and pushes the [`ParsedMessage`]s into `send`. async fn parse_loop( - mut recv: mpsc::UnboundedReceiver, + mut recv: mpsc::UnboundedReceiver>, send: mpsc::UnboundedSender>, extra_logging: Option>, ) { @@ -1361,6 +1770,13 @@ async fn parse_loop( debug_log(&extra_logging, |file| { writeln!(file, "`parse_loop`: Got raw message: {msg:?}") }); + let msg = match msg { + Ok(msg) => msg, + Err(error) => { + let _ = send.unbounded_send(ParsedMessage::TransportError(error)); + break; + } + }; let parsed = match msg { ws::v2::ServerMessage::TransactionUpdate(transaction_update) => { match M::DbUpdate::parse_update(transaction_update) { @@ -1387,10 +1803,12 @@ async fn parse_loop( ret_value, transaction_update, }) => { - assert!( - ret_value.is_empty(), - "Reducer return value should be unit, i.e. 0 bytes, but got {ret_value:?}" - ); + if !ret_value.is_empty() { + let _ = send.unbounded_send(ParsedMessage::Error( + InternalError::new("Non-unit reducer return value").into(), + )); + break; + } match M::DbUpdate::parse_update(transaction_update) { Ok(db_update) => ParsedMessage::ReducerResult { request_id, @@ -1437,7 +1855,7 @@ async fn parse_loop( connection_id, }) => ParsedMessage::IdentityToken(identity, token, connection_id), ws::v2::ServerMessage::OneOffQueryResult(_) => { - unreachable!("The Rust SDK does not implement one-off queries") + ParsedMessage::Error(InternalError::new("Unexpected one-off query response").into()) } ws::v2::ServerMessage::SubscribeApplied(subscribe_applied) => { let db_update = subscribe_applied.rows; @@ -1460,7 +1878,10 @@ async fn parse_loop( .. }) => { let Some(db_update) = db_update else { - unreachable!("The Rust SDK always requests rows to delete when unsubscribing") + let _ = send.unbounded_send(ParsedMessage::Error( + InternalError::new("Unsubscribe response omitted dropped rows").into(), + )); + break; }; match M::DbUpdate::parse_unsubscribe_rows(db_update) { Err(e) => ParsedMessage::Error( @@ -1478,11 +1899,7 @@ async fn parse_loop( query_set_id: e.query_set_id, error: e.error.to_string(), }, - // This SDK negotiates v2 and never sends `SubscribeBatch`, - // so the server should never send this response. - ws::v2::ServerMessage::SubscribeBatchApplied(_) => ParsedMessage::Error( - InternalError::new("Received SubscribeBatchApplied, which this client never requests").into(), - ), + ws::v2::ServerMessage::SubscribeBatchApplied(batch) => ParsedMessage::SubscribeBatchApplied(batch), ws::v2::ServerMessage::ProcedureResult(procedure_result) => ParsedMessage::ProcedureResult { request_id: procedure_result.request_id, result: match procedure_result.status { @@ -1494,13 +1911,29 @@ async fn parse_loop( debug_log(&extra_logging, |file| { writeln!(file, "`parse_loop`: Parsed as: {parsed:?}") }); - send.unbounded_send(parsed) - .expect("Failed to send ParsedMessage to main thread"); + if send.unbounded_send(parsed).is_err() { + break; + } } } /// Operations a user can make to a `DbContext` which must be postponed pub(crate) enum PendingMutation { + ReconnectReady { + // Keep the parked receive stream open until this mutation swaps it. + // Otherwise the async runner can observe EOF before ReconnectReady. + _park_send: mpsc::UnboundedSender>, + generation: u64, + token: Option, + fresh: bool, + result: Result< + ( + mpsc::UnboundedSender, + mpsc::UnboundedReceiver>, + ), + ReconnectFailure, + >, + }, Unsubscribe { query_set_id: QuerySetId, }, @@ -1551,6 +1984,9 @@ pub(crate) enum PendingMutation { impl std::fmt::Debug for PendingMutation { fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { match self { + PendingMutation::ReconnectReady { generation, .. } => { + f.debug_tuple("ReconnectReady").field(generation).finish() + } PendingMutation::Unsubscribe { query_set_id } => f .debug_struct("PendingMutation::Unsubscribe") .field("query_set_id", query_set_id) @@ -1627,3 +2063,36 @@ pub(crate) fn next_query_set_id() -> QuerySetId { id: NEXT_QUERY_SET_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed), } } + +#[derive(Clone)] +struct ConnectionConfig { + uri: Uri, + database_name: String, + params: WsParams, +} +struct ReconnectState { + config: ConnectionConfig, + options: Option, + provider: Option>, + token: Option, + attempt: u32, + // Incremented to invalidate results from older reconnect tasks. + // Only ReconnectReady messages matching this generation may be applied. + generation: u64, + reconnecting: bool, + preparing_replay: bool, + used_fresh_token: bool, + force_refresh: bool, + abort: Option, +} +impl Drop for ReconnectState { + fn drop(&mut self) { + if let Some(abort) = self.abort.take() { + abort.abort(); + } + } +} +pub(crate) enum ReconnectFailure { + Provider(crate::Error), + Transport(crate::websocket::WsError), +} diff --git a/sdks/rust/src/db_context.rs b/sdks/rust/src/db_context.rs index e9a4114a46c..cec09c245ee 100644 --- a/sdks/rust/src/db_context.rs +++ b/sdks/rust/src/db_context.rs @@ -40,6 +40,9 @@ pub trait DbContext { /// Returns `true` if the connection is active, i.e. has not yet disconnected. fn is_active(&self) -> bool; + /// True while the SDK is waiting for or establishing an automatic reconnect. + fn is_reconnecting(&self) -> bool; + /// Close the connection. /// /// Returns an error if we are already disconnected. diff --git a/sdks/rust/src/error.rs b/sdks/rust/src/error.rs index 6b33a41807f..bba9ab8eeb1 100644 --- a/sdks/rust/src/error.rs +++ b/sdks/rust/src/error.rs @@ -8,6 +8,9 @@ pub enum Error { #[error("Connection is already disconnected or has terminated normally")] Disconnected, + #[error("Connection lost before the result was received; the operation may have executed")] + UnknownResult, + #[error("Failed to connect: {source}")] FailedToConnect { #[source] @@ -17,6 +20,9 @@ pub enum Error { #[error("Host returned error when processing subscription query: {error}")] SubscriptionError { error: String }, + #[error("Token provider failed: {message}")] + TokenProvider { message: String }, + #[error("Subscription has already ended")] AlreadyEnded, @@ -34,6 +40,14 @@ pub struct InternalError { } impl InternalError { + /// Whether the operation may have executed before the transport was lost. + pub fn is_unknown_result(&self) -> bool { + self.cause + .as_ref() + .and_then(|e| e.downcast_ref::()) + .is_some_and(|e| matches!(e, Error::UnknownResult)) + } + pub(crate) fn new(message: impl Into) -> Self { Self { message: message.into(), @@ -97,7 +111,9 @@ impl std::error::Error for InternalError { &self.message } fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { - None + self.cause + .as_deref() + .map(|cause| cause as &(dyn std::error::Error + 'static)) } } diff --git a/sdks/rust/src/lib.rs b/sdks/rust/src/lib.rs index 09429973105..5f8456c1e55 100644 --- a/sdks/rust/src/lib.rs +++ b/sdks/rust/src/lib.rs @@ -15,6 +15,7 @@ mod client_cache; mod compression; mod db_connection; mod metrics; +mod reconnect; mod spacetime_module; mod subscription; mod websocket; @@ -29,6 +30,7 @@ pub use db_connection::DbConnectionBuilder; pub use db_context::DbContext; pub use error::{Error, Result}; pub use event::{Event, ReducerEvent, Status}; +pub use reconnect::{AutomaticReconnectOptions, NextReconnect, TokenFuture, TokenProvider}; pub use table::{EventTable, Table, TableAccessor, TableWithPrimaryKey}; pub use spacetime_module::SubscriptionHandle; diff --git a/sdks/rust/src/reconnect.rs b/sdks/rust/src/reconnect.rs new file mode 100644 index 00000000000..7966fc1522c --- /dev/null +++ b/sdks/rust/src/reconnect.rs @@ -0,0 +1,149 @@ +//! Automatic reconnect policy and credential refresh. +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; +use std::{ + future::Future, + pin::Pin, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +/// The upcoming automatic reconnect attempt (numbered starting at one). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct NextReconnect { + pub attempt: u32, + pub delay: Duration, +} + +/// Bounds for exponential backoff with 50% jitter. Attempts are unlimited. +#[derive(Clone, Copy, Debug)] +pub struct AutomaticReconnectOptions { + /// Defaults to one second; clamped to at least 500 milliseconds. + pub min_delay: Duration, + /// Defaults to 30 seconds; clamped to at least one second and `min_delay`. + pub max_delay: Duration, +} +impl Default for AutomaticReconnectOptions { + fn default() -> Self { + Self { + min_delay: Duration::from_secs(1), + max_delay: Duration::from_secs(30), + } + } +} +impl AutomaticReconnectOptions { + pub(crate) fn resolve(mut self) -> Self { + let min = self.min_delay.max(Duration::from_millis(500)); + let max = self.max_delay.max(Duration::from_secs(1)).max(min); + if min != self.min_delay || max != self.max_delay { + log::warn!("Reconnect delays raised to safety bounds: {min:?}..={max:?}"); + } + self.min_delay = min; + self.max_delay = max; + self + } + pub(crate) fn delay(self, attempt: u32, random: f64) -> Duration { + let base = (self.min_delay.as_secs_f64() * 2f64.powi(attempt.saturating_sub(1).min(30) as i32)) + .min(self.max_delay.as_secs_f64()); + Duration::try_from_secs_f64( + (base * (0.5 + random)).clamp(self.min_delay.as_secs_f64(), self.max_delay.as_secs_f64()), + ) + .unwrap_or(self.max_delay) + } +} + +/// Supplies a fresh token for the same identity. Invoked only during reconnect, +/// when the retained JWT approaches expiry, cannot be inspected, or is rejected. +/// Provider errors are retried using the normal backoff policy. +pub trait TokenProvider: Send + Sync + 'static { + fn token(&self) -> TokenFuture; +} + +/// An asynchronous credential result. Browser futures need not be `Send`. +#[cfg(not(feature = "browser"))] +pub type TokenFuture = Pin> + Send + 'static>>; +#[cfg(feature = "browser")] +pub type TokenFuture = Pin> + 'static>>; + +#[cfg(not(feature = "browser"))] +impl TokenProvider for F +where + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + fn token(&self) -> TokenFuture { + Box::pin(self()) + } +} +#[cfg(feature = "browser")] +impl TokenProvider for F +where + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future> + 'static, +{ + fn token(&self) -> TokenFuture { + Box::pin(self()) + } +} + +pub(crate) fn now() -> SystemTime { + #[cfg(not(feature = "browser"))] + { + SystemTime::now() + } + #[cfg(feature = "browser")] + { + UNIX_EPOCH + Duration::from_secs_f64(js_sys::Date::now() / 1000.0) + } +} + +pub(crate) fn token_needs_refresh(token: Option<&str>, now: SystemTime) -> bool { + let claims = (|| { + let payload = token?.split('.').nth(1)?; + let bytes = URL_SAFE_NO_PAD.decode(payload.trim_end_matches('=')).ok()?; + serde_json::from_slice::(&bytes).ok() + })(); + let Some(claims) = claims else { + return true; + }; + let Some(exp) = claims["exp"].as_f64().filter(|n| n.is_finite()) else { + return true; + }; + let margin = claims["iat"] + .as_f64() + .map(|iat| (exp - iat) * 0.05) + .unwrap_or(0.0) + .max(30.0); + exp - now.duration_since(UNIX_EPOCH).unwrap_or_default().as_secs_f64() <= margin +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn backoff_bounds_and_saturation() { + let policy = AutomaticReconnectOptions::default(); + for attempt in [1, 2, 3, 6, u32::MAX] { + for random in [0.0, 0.5, 1.0] { + let delay = policy.delay(attempt, random); + assert!(delay >= policy.min_delay && delay <= policy.max_delay); + } + } + assert_eq!(policy.delay(3, 0.5), Duration::from_secs(4)); + let policy = AutomaticReconnectOptions { + min_delay: Duration::ZERO, + max_delay: Duration::ZERO, + } + .resolve(); + assert_eq!(policy.min_delay, Duration::from_millis(500)); + assert_eq!(policy.max_delay, Duration::from_secs(1)); + } + #[test] + fn refresh_margin_and_unreadable_tokens() { + let now = UNIX_EPOCH + Duration::from_secs(1000); + let token = |claims: &str| format!("header.{}.signature", URL_SAFE_NO_PAD.encode(claims)); + assert!(token_needs_refresh(None, now)); + assert!(token_needs_refresh(Some("invalid"), now)); + assert!(token_needs_refresh(Some(&token(r#"{"exp":1030,"iat":900}"#)), now)); + assert!(!token_needs_refresh(Some(&token(r#"{"exp":1031,"iat":900}"#)), now)); + assert!(token_needs_refresh(Some(&token(r#"{"exp":1100,"iat":-1000}"#)), now)); + } +} diff --git a/sdks/rust/src/subscription.rs b/sdks/rust/src/subscription.rs index 2204985ea15..64f428ae259 100644 --- a/sdks/rust/src/subscription.rs +++ b/sdks/rust/src/subscription.rs @@ -30,7 +30,7 @@ impl Default for SubscriptionManager { } pub(crate) type OnAppliedCallback = - Box::SubscriptionEventContext) + Send + 'static>; + Box::SubscriptionEventContext) + Send + 'static>; pub(crate) type OnErrorCallback = Box::ErrorContext, crate::Error) + Send + 'static>; pub type OnEndedCallback = Box::SubscriptionEventContext) + Send + 'static>; @@ -46,24 +46,54 @@ pub(crate) enum PendingUnsubscribeResult { } impl SubscriptionManager { - pub(crate) fn on_disconnect(&mut self, _ctx: &M::ErrorContext) { - // We need to clear all the subscriptions. - // TODO: is this correct? We don't remove them from the client cache, - // we may want to resume them in the future if we impl reconnecting, - // and users can already register on-disconnect callbacks which will run in this case. - - // NOTE(cloutiertyler) - // This function previously invoke `on_error` for all subscriptions. - // However, this is inconsistent behavior given that `on_disconnect` for - // connections no longer always has an error argument and that the user - // can add an `on_ended` callback when unsubscribing. - // - // We propose instead that `on_ended` be added to the subscription - // builder so that it can be invoked when the subscription is ended - // because of a normal disconnect, but without the user calling - // `unsubscribe_then`. This can be done in a non-breaking way. - // - // For now, we will just do nothing when a subscription ends normally. + pub(crate) fn suspend(&mut self, ctx: &M::SubscriptionEventContext) { + let ids: Vec<_> = self.subscriptions.keys().copied().collect(); + for id in ids { + let sub = self.subscriptions[&id].clone(); + let cancelled = sub.inner.lock().unwrap().unsubscribe_called; + if cancelled { + self.unsubscribe_applied(ctx, id); + } else { + sub.inner.lock().unwrap().status = SubscriptionServerState::Pending; + } + } + } + + pub(crate) fn replay(&mut self, ctx: &M::SubscriptionEventContext) -> Box<[ws::v2::SubscribeSet]> { + self.suspend(ctx); + let old = std::mem::take(&mut self.subscriptions); + old.into_values() + .filter_map(|handle| { + let id = next_query_set_id(); + // Rebinding and starting must be atomic with unsubscribe_then: + // a concurrent cancellation must either end locally or target the new ID. + let mut state = handle.inner.lock().unwrap(); + if state.unsubscribe_called { + let on_ended = state.on_ended(); + drop(state); + if let Some(callback) = on_ended { + callback(ctx); + } + return None; + } + state.query_set_id = id; + let msg = state.start().expect("uncancelled replay subscription"); + drop(state); + self.subscriptions.insert(id, handle); + Some(ws::v2::SubscribeSet { + query_set_id: id, + query_strings: msg.query_strings, + }) + }) + .collect() + } + + pub(crate) fn on_disconnect(&mut self, ctx: &M::SubscriptionEventContext) { + for (_, mut sub) in std::mem::take(&mut self.subscriptions) { + if let Some(callback) = sub.on_ended() { + callback(ctx); + } + } } /// Register a new subscription. This does not send the subscription to the server. @@ -80,8 +110,10 @@ impl SubscriptionManager { // TODO: log or double check error handling. return; }; - if let Some(callback) = sub.on_applied() { - callback(ctx) + let callback = sub.on_applied(); + if let Some(mut callback) = callback { + callback(ctx); + sub.inner.lock().unwrap().on_applied = Some(callback); } } @@ -162,7 +194,7 @@ impl SubscriptionBuilder { } /// Register a callback to run when the subscription is applied. - pub fn on_applied(mut self, callback: impl FnOnce(&M::SubscriptionEventContext) + Send + 'static) -> Self { + pub fn on_applied(mut self, callback: impl FnMut(&M::SubscriptionEventContext) + Send + 'static) -> Self { self.on_applied = Some(Box::new(callback)); self } @@ -431,7 +463,7 @@ impl SubscriptionState { self.on_applied.take() } - pub fn on_ended(&mut self) -> Option> { + pub fn on_ended(&mut self) -> Option> { // TODO: Consider logging a warning if the state is wrong (like being in the Error state). if self.is_ended() { return None; diff --git a/sdks/rust/src/websocket.rs b/sdks/rust/src/websocket.rs index 235ef06138f..82421c504c0 100644 --- a/sdks/rust/src/websocket.rs +++ b/sdks/rust/src/websocket.rs @@ -79,6 +79,13 @@ pub enum WsError { source: Arc, }, + #[cfg(not(feature = "browser"))] + #[error("Token cannot be represented as an HTTP authorization header")] + InvalidToken, + + #[error("WebSocket closed with code {code}: {reason}")] + Closed { code: u16, reason: String }, + #[error("Received empty raw message, but valid messages always start with a one-byte compression flag")] EmptyMessage, @@ -104,6 +111,46 @@ pub enum WsError { TokenVerification(String), } +impl WsError { + pub(crate) fn is_auth(&self) -> bool { + #[cfg(not(feature = "browser"))] + if matches!(self, Self::InvalidToken) { + return true; + } + #[cfg(not(feature = "browser"))] + if let Self::Tungstenite { source, .. } = self + && let TokioTungsteniteError::Http(response) = &**source + { + return matches!(response.status().as_u16(), 400 | 401 | 403); + } + #[cfg(feature = "browser")] + if let Self::TokenVerification(message) = self { + return ["HTTP error: 400", "HTTP error: 401", "HTTP error: 403"] + .iter() + .any(|prefix| message.starts_with(prefix)); + } + false + } + pub(crate) fn is_terminal(&self) -> bool { + match self { + #[cfg(not(feature = "browser"))] + Self::InvalidToken => false, + Self::Closed { code, .. } => matches!(code, 1002 | 1003 | 1007 | 1008), + Self::UriError(_) + | Self::EmptyMessage + | Self::DeserializeMessage { .. } + | Self::Decompress { .. } + | Self::UnknownCompressionScheme { .. } => true, + #[cfg(not(feature = "browser"))] + Self::Tungstenite { source, .. } => { + matches!(&**source, TokioTungsteniteError::Protocol(error) if !matches!(error, tokio_tungstenite::tungstenite::error::ProtocolError::ResetWithoutClosingHandshake)) + } + #[cfg(feature = "browser")] + _ => false, + } + } +} + pub(crate) struct WsConnection { db_name: Box, #[cfg(not(feature = "browser"))] @@ -135,6 +182,7 @@ pub(crate) struct WsParams { /// `Some(false)` to disable them. /// `None` to not set the parameter and let the server choose. pub confirmed: Option, + pub session_id: Option, } #[cfg(not(feature = "browser"))] @@ -197,6 +245,11 @@ fn make_uri_impl( path.push_str(&cid.to_hex()); } + if let Some(session_id) = params.session_id { + path.push_str("&session_id="); + path.push_str(&session_id.to_hex()); + } + // Enable confirmed reads if requested. if let Some(confirmed) = params.confirmed { path.push_str("&confirmed="); @@ -239,7 +292,7 @@ fn make_request( source: Arc::new(source), })?; request_insert_protocol_header(&mut req); - request_insert_auth_header(&mut req, token); + request_insert_auth_header(&mut req, token)?; Ok(req) } @@ -252,11 +305,15 @@ fn request_insert_protocol_header(req: &mut http::Request<()>) { } #[cfg(not(feature = "browser"))] -fn request_insert_auth_header(req: &mut http::Request<()>, token: Option<&str>) { +fn request_insert_auth_header(req: &mut http::Request<()>, token: Option<&str>) -> Result<(), WsError> { if let Some(token) = token { - let auth = ["Bearer ", token].concat().try_into().unwrap(); + let auth = ["Bearer ", token] + .concat() + .try_into() + .map_err(|_| WsError::InvalidToken)?; req.headers_mut().insert(http::header::AUTHORIZATION, auth); } + Ok(()) } #[cfg(feature = "browser")] @@ -393,7 +450,7 @@ impl WsConnection { #[cfg(not(feature = "browser"))] async fn message_loop( mut self, - incoming_messages: mpsc::UnboundedSender, + incoming_messages: mpsc::UnboundedSender>, outgoing_messages: mpsc::UnboundedReceiver, extra_logging: Option>>, ) { @@ -448,27 +505,33 @@ impl WsConnection { }, Err(e) => { - maybe_log_error!( - &extra_logging, - "Error reading message from read WebSocket stream", - Result::<(), _>::Err(e) - ); + let error = WsError::Tungstenite { uri: http::Uri::default(), source: Arc::new(e) }; + let _ = incoming_messages.unbounded_send(Err(error)); + break; + }, + + Ok(Some(WebSocketMessage::Close(frame))) => { + if let Some(frame) = frame { + let _ = incoming_messages.unbounded_send(Err(WsError::Closed { + code: frame.code.into(), reason: frame.reason.to_string(), + })); + } break; }, + Ok(Some(WebSocketMessage::Binary(bytes))) => { idle = false; record_metrics(bytes.len()); match Self::parse_response(&bytes) { - Err(e) => maybe_log_error!( - &extra_logging, - "Error decoding WebSocketMessage::Binary payload", - Result::<(), _>::Err(e) - ), + Err(e) => { + let _ = incoming_messages.unbounded_send(Err(e)); + break; + }, Ok(msg) => maybe_log_error!( &extra_logging, "Error sending decoded message to incoming_messages queue", - incoming_messages.unbounded_send(msg) + incoming_messages.unbounded_send(Ok(msg)) ), } } @@ -529,7 +592,7 @@ impl WsConnection { } None => { maybe_log_error!(&extra_logging, "Error sending close frame", SinkExt::close(&mut self.sock).await); - outgoing_messages = None; + break; } }, } @@ -543,7 +606,7 @@ impl WsConnection { extra_logging: Option>>, ) -> ( JoinHandle<()>, - mpsc::UnboundedReceiver, + mpsc::UnboundedReceiver>, mpsc::UnboundedSender, ) { let (outgoing_send, outgoing_recv) = mpsc::unbounded(); @@ -556,7 +619,7 @@ impl WsConnection { pub(crate) fn spawn_message_loop( self, ) -> ( - mpsc::UnboundedReceiver, + mpsc::UnboundedReceiver>, mpsc::UnboundedSender, ) { let websocket_received = CLIENT_METRICS.websocket_received.with_label_values(&self.db_name); @@ -569,7 +632,7 @@ impl WsConnection { }; let (outgoing_tx, outgoing_rx) = mpsc::unbounded::(); - let (incoming_tx, incoming_rx) = mpsc::unbounded::(); + let (incoming_tx, incoming_rx) = mpsc::unbounded::>(); let (mut ws_writer, ws_reader) = self.sock.split(); @@ -590,20 +653,23 @@ impl WsConnection { record_metrics(bytes.len()); // parse + forward into `incoming_tx` match Self::parse_response(&bytes) { - Ok(msg) => if let Err(_e) = incoming_tx.unbounded_send(msg) { + Ok(msg) => if let Err(_e) = incoming_tx.unbounded_send(Ok(msg)) { gloo_console::warn!("Incoming receiver dropped."); break; }, Err(e) => { - gloo_console::warn!( - "Error decoding WebSocketMessage::Binay payload: ", - format!("{:?}", e) - ); + let _ = incoming_tx.unbounded_send(Err(e)); + break; }, } }, Some(Ok(WebSocketMessage::Close(r))) => { + if let Some(ref frame) = r { + let _ = incoming_tx.unbounded_send(Err(WsError::Closed { + code: frame.code.into(), reason: frame.reason.to_string(), + })); + } let reason: String = if let Some(r) = r { format!("{}:{:?}", r, r.code) } else {String::default()}; diff --git a/sdks/rust/tests/case-conversion-client/src/main.rs b/sdks/rust/tests/case-conversion-client/src/main.rs index 3bcc10dadee..52e4c60a058 100644 --- a/sdks/rust/tests/case-conversion-client/src/main.rs +++ b/sdks/rust/tests/case-conversion-client/src/main.rs @@ -86,7 +86,7 @@ fn connect_then( callback(ctx); connected_result(Ok(())); }) - .on_connect_error(|_ctx, error| panic!("Connect errored: {error:?}")) + .on_connect_error(|_ctx, error, _next| panic!("Connect errored: {error:?}")) .build() .unwrap(); conn.run_threaded(); @@ -104,7 +104,7 @@ fn exec_insert_player() { connect_then(&test_counter, move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.player_1().on_insert(move |_ctx, row| { let check = || { // Verify field names with digit boundaries are correctly case-converted @@ -123,7 +123,7 @@ fn exec_insert_player() { reducer_callback_assert_committed("create_player_1"), ) .unwrap(); - }) + })) // Query builder: subscribe to player_1 table (canonical: Player1Canonical) .add_query(|q| q.from.player_1().build()) .subscribe(); @@ -143,7 +143,7 @@ fn exec_insert_person() { connect_then(&test_counter, move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.person_2().on_insert(move |_ctx, person| { let check = || { assert_eq_or_bail!("Bob".to_string(), person.first_name); @@ -174,7 +174,7 @@ fn exec_insert_person() { Err(e) => panic!("create_player_1 panicked: {e:?}"), }) .unwrap(); - }) + })) // Query builder: subscribe to both tables .add_query(|q| q.from.player_1().build()) .add_query(|q| q.from.person_2().build()) @@ -194,7 +194,7 @@ fn exec_ban_player() { connect_then(&test_counter, move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.player_1().on_update(move |_ctx, _old, new| { let check = || { assert_eq_or_bail!(Player2Status::BannedUntil(9999), new.status_3_field); @@ -220,7 +220,7 @@ fn exec_ban_player() { Err(e) => panic!("create_player_1 panicked: {e:?}"), }) .unwrap(); - }) + })) // Query builder: subscribe to player_1 table .add_query(|q| q.from.player_1().build()) .subscribe(); @@ -239,7 +239,7 @@ fn exec_query_builder_filter() { connect_then(&test_counter, move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.player_1().on_insert(move |_ctx, row| { let check = || { // Only level-5 players should come through the filter @@ -267,7 +267,7 @@ fn exec_query_builder_filter() { reducer_callback_assert_committed("create_player_1"), ) .unwrap(); - }) + })) // Query builder: filter on digit-boundary column current_level_2 .add_query(|q| q.from.player_1().filter(|p| p.current_level_2.eq(5)).build()) .subscribe(); @@ -290,7 +290,7 @@ fn exec_query_builder_join() { connect_then(&test_counter, move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { // Listen for person_2 inserts that come through the join. // The join is: player_1 RIGHT SEMIJOIN person_2 ON player_1.player_1_id = person_2.player_ref // This means we see person_2 rows that have a matching player_1 row. @@ -335,7 +335,7 @@ fn exec_query_builder_join() { } }) .unwrap(); - }) + })) // Query builder: JOIN player_1 with person_2 on player_1_id = player_ref // player_1 RIGHT SEMIJOIN person_2 means: show person_2 rows that have a matching player_1 .add_query(|q| { @@ -362,7 +362,7 @@ fn exec_view() { connect_then(&test_counter, move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.person_at_level_2().on_insert(move |_ctx, row| { let check = || { assert_eq_or_bail!("ViewPerson".to_string(), row.first_name); @@ -392,7 +392,7 @@ fn exec_view() { reducer_callback_assert_committed("add_person_2"), ) .unwrap(); - }) + })) // Subscribe to the view which selects people at level 2 .add_query(|q| q.from.person_at_level_2().build()) .subscribe(); diff --git a/sdks/rust/tests/case-conversion-client/src/module_bindings/mod.rs b/sdks/rust/tests/case-conversion-client/src/module_bindings/mod.rs index de71b3ab94e..1255c31139e 100644 --- a/sdks/rust/tests/case-conversion-client/src/module_bindings/mod.rs +++ b/sdks/rust/tests/case-conversion-client/src/module_bindings/mod.rs @@ -310,6 +310,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -534,6 +538,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -610,6 +618,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -682,6 +694,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -754,6 +770,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -830,6 +850,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/sdks/rust/tests/connect_disconnect_client/src/module_bindings/mod.rs b/sdks/rust/tests/connect_disconnect_client/src/module_bindings/mod.rs index 13c455309e8..fc7c79a77f0 100644 --- a/sdks/rust/tests/connect_disconnect_client/src/module_bindings/mod.rs +++ b/sdks/rust/tests/connect_disconnect_client/src/module_bindings/mod.rs @@ -249,6 +249,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -473,6 +477,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -549,6 +557,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -621,6 +633,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -693,6 +709,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -769,6 +789,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/sdks/rust/tests/connect_disconnect_client/src/test_handlers.rs b/sdks/rust/tests/connect_disconnect_client/src/test_handlers.rs index a7795e8530b..a67b41e5066 100644 --- a/sdks/rust/tests/connect_disconnect_client/src/test_handlers.rs +++ b/sdks/rust/tests/connect_disconnect_client/src/test_handlers.rs @@ -15,14 +15,14 @@ pub async fn dispatch(db_name: &str) { let connection = DbConnection::builder() .with_database_name(db_name) .with_uri(server_url()) - .on_connect_error(|_ctx, error| panic!("on_connect_error: {error:?}")) + .on_connect_error(|_ctx, error, _next| panic!("on_connect_error: {error:?}")) .on_connect(move |ctx, _, _| { connected_result(Ok(())); ctx.subscription_builder() .on_error(|_ctx, error| { panic!("Subscription failed: {error:?}"); }) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { let check = || { anyhow::ensure!(ctx.db.connected().count() == 1); match ctx.db.connected().iter().next() { @@ -36,19 +36,21 @@ pub async fn dispatch(db_name: &str) { Ok(()) }; sub_applied_one_row_result(check()); - }) + })) .subscribe("SELECT * FROM connected"); }) - .on_disconnect(move |ctx, error| { - assert!( - !ctx.is_active(), - "on_disconnect callback, but `ctx.is_active()` is true" - ); - match error { - Some(err) => disconnect_result(Err(anyhow::anyhow!("{err:?}"))), - None => disconnect_result(Ok(())), - } - }); + .on_disconnect(test_counter::once3::( + move |ctx, error, _next| { + assert!( + !ctx.is_active(), + "on_disconnect callback, but `ctx.is_active()` is true" + ); + match error { + Some(err) => disconnect_result(Err(anyhow::anyhow!("{err:?}"))), + None => disconnect_result(Ok(())), + } + }, + )); let connection = build_connection(connection).await; #[cfg(not(target_arch = "wasm32"))] @@ -73,7 +75,7 @@ pub async fn dispatch(db_name: &str) { let sub_applied_one_row_result = reconnect_test_counter.add_test("disconnected_row"); let new_connection = DbConnection::builder() - .on_connect_error(|_ctx, error| panic!("on_connect_error: {error:?}")) + .on_connect_error(|_ctx, error, _next| panic!("on_connect_error: {error:?}")) .on_connect(move |_ctx, _, _| { reconnected_result(Ok(())); }) @@ -83,7 +85,7 @@ pub async fn dispatch(db_name: &str) { new_connection .subscription_builder() - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { let check = || { anyhow::ensure!(ctx.db.disconnected().count() == 1); match ctx.db.disconnected().iter().next() { @@ -97,7 +99,7 @@ pub async fn dispatch(db_name: &str) { Ok(()) }; sub_applied_one_row_result(check()); - }) + })) .on_error(|_ctx, error| panic!("subscription on_error: {error:?}")) .subscribe("SELECT * FROM disconnected"); diff --git a/sdks/rust/tests/event-table-client/src/module_bindings/mod.rs b/sdks/rust/tests/event-table-client/src/module_bindings/mod.rs index 1f83a93a8c3..e7f26dcc634 100644 --- a/sdks/rust/tests/event-table-client/src/module_bindings/mod.rs +++ b/sdks/rust/tests/event-table-client/src/module_bindings/mod.rs @@ -242,6 +242,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -466,6 +470,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -542,6 +550,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -614,6 +626,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -686,6 +702,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -762,6 +782,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/sdks/rust/tests/event-table-client/src/test_handlers.rs b/sdks/rust/tests/event-table-client/src/test_handlers.rs index 3e361125c66..1e4af648add 100644 --- a/sdks/rust/tests/event-table-client/src/test_handlers.rs +++ b/sdks/rust/tests/event-table-client/src/test_handlers.rs @@ -60,7 +60,7 @@ async fn connect_then( callback(ctx); connected_result(Ok(())); }) - .on_connect_error(|_ctx, error| panic!("Connect errored: {error:?}")); + .on_connect_error(|_ctx, error, _next| panic!("Connect errored: {error:?}")); build_and_run(conn).await } @@ -70,7 +70,7 @@ fn subscribe_these_then( callback: impl FnOnce(&SubscriptionEventContext) + Send + 'static, ) { ctx.subscription_builder() - .on_applied(callback) + .on_applied(test_counter::once::(callback)) .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) .subscribe(queries); } @@ -218,9 +218,11 @@ async fn exec_v1_rejects_event_table(db_name: &str) { let error_result = test_counter.add_test("v1-rejects-event-table"); ctx.subscription_builder() - .on_applied(move |_ctx: &SubscriptionEventContext| { - panic!("Subscription to event table should not succeed over v1"); - }) + .on_applied(test_counter::once::( + move |_ctx: &SubscriptionEventContext| { + panic!("Subscription to event table should not succeed over v1"); + }, + )) .on_error(move |_ctx, error| { let msg = format!("{error:?}"); if msg.contains("v2") || msg.contains("upgrade") || msg.contains("Upgrade") { diff --git a/sdks/rust/tests/procedural-view-pk-client/src/module_bindings/mod.rs b/sdks/rust/tests/procedural-view-pk-client/src/module_bindings/mod.rs index d0cd5e3b7e3..2196851e1e8 100644 --- a/sdks/rust/tests/procedural-view-pk-client/src/module_bindings/mod.rs +++ b/sdks/rust/tests/procedural-view-pk-client/src/module_bindings/mod.rs @@ -299,6 +299,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -523,6 +527,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -599,6 +607,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -671,6 +683,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -743,6 +759,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -819,6 +839,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/sdks/rust/tests/procedural-view-pk-client/src/test_handlers.rs b/sdks/rust/tests/procedural-view-pk-client/src/test_handlers.rs index 319faf66b8b..64a91be864c 100644 --- a/sdks/rust/tests/procedural-view-pk-client/src/test_handlers.rs +++ b/sdks/rust/tests/procedural-view-pk-client/src/test_handlers.rs @@ -55,7 +55,7 @@ async fn connect_then_named( callback(ctx); connected_result(Ok(())); }) - .on_connect_error(|_ctx, error| panic!("Connect errored: {error:?}")); + .on_connect_error(|_ctx, error, _next| panic!("Connect errored: {error:?}")); build_and_run(conn).await } @@ -70,7 +70,7 @@ async fn exec_sender_scoped_pk_view(db_name: &str) { let _sender_a = connect_then(db_name, &test_counter, move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.sender_left_view().on_update(move |_, old_row, new_row| { assert_eq!(old_row.id, 1); assert_eq!(old_row.filter, 10); @@ -85,7 +85,7 @@ async fn exec_sender_scoped_pk_view(db_name: &str) { ctx.reducers() .update_left_then(1, 11, reducer_callback_assert_committed("update_left")) .unwrap(); - }) + })) .add_query(|q| q.from.sender_left_view().build()) .subscribe(); }) @@ -94,7 +94,7 @@ async fn exec_sender_scoped_pk_view(db_name: &str) { let _sender_b = connect_then_named(db_name, &test_counter, "sender_b_on_connect", move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.sender_left_view().on_update(move |_, old_row, new_row| { assert_eq!(old_row.id, 2); assert_eq!(old_row.filter, 20); @@ -109,7 +109,7 @@ async fn exec_sender_scoped_pk_view(db_name: &str) { ctx.reducers() .update_left_then(2, 21, reducer_callback_assert_committed("update_left")) .unwrap(); - }) + })) .add_query(|q| q.from.sender_left_view().build()) .subscribe(); }) @@ -160,7 +160,7 @@ async fn exec_view_pk_left_semijoin(db_name: &str) { let mut joined_insert = Some(joined_insert); ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.sender_left_view().on_insert(move |ctx, row| { assert_eq!(ctx.db.sender_left_view().count(), 1); assert_eq!(row.id, 10); @@ -169,7 +169,7 @@ async fn exec_view_pk_left_semijoin(db_name: &str) { }); insert_semijoin_source_rows(ctx); - }) + })) .add_query(|q| { q.from .sender_right_view() @@ -188,7 +188,7 @@ async fn exec_view_pk_right_semijoin(db_name: &str) { let mut joined_insert = Some(joined_insert); ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.sender_right_view().on_insert(move |ctx, row| { assert_eq!(ctx.db.sender_right_view().count(), 1); assert_eq!(row.id, 10); @@ -197,7 +197,7 @@ async fn exec_view_pk_right_semijoin(db_name: &str) { }); insert_semijoin_source_rows(ctx); - }) + })) .add_query(|q| { q.from .sender_left_view() diff --git a/sdks/rust/tests/procedure-client/src/module_bindings/mod.rs b/sdks/rust/tests/procedure-client/src/module_bindings/mod.rs index 97afd723359..43490958f67 100644 --- a/sdks/rust/tests/procedure-client/src/module_bindings/mod.rs +++ b/sdks/rust/tests/procedure-client/src/module_bindings/mod.rs @@ -309,6 +309,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -533,6 +537,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -609,6 +617,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -681,6 +693,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -753,6 +769,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -829,6 +849,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/sdks/rust/tests/procedure-client/src/test_handlers.rs b/sdks/rust/tests/procedure-client/src/test_handlers.rs index fdfc417cd9b..7e4cb8ee89d 100644 --- a/sdks/rust/tests/procedure-client/src/test_handlers.rs +++ b/sdks/rust/tests/procedure-client/src/test_handlers.rs @@ -72,7 +72,7 @@ async fn connect_with_then( callback(ctx); connected_result(Ok(())); }) - .on_connect_error(|_ctx, error| panic!("Connect errored: {error:?}")); + .on_connect_error(|_ctx, error, _next| panic!("Connect errored: {error:?}")); build_and_run(with_builder(builder)).await } @@ -101,7 +101,7 @@ fn subscribe_these_then( callback: impl FnOnce(&SubscriptionEventContext) + Send + 'static, ) { ctx.subscription_builder() - .on_applied(callback) + .on_applied(test_counter::once::(callback)) .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) .subscribe(queries); } diff --git a/sdks/rust/tests/procedure-concurrency-client/src/module_bindings/mod.rs b/sdks/rust/tests/procedure-concurrency-client/src/module_bindings/mod.rs index 286a46580bb..21ce2cee225 100644 --- a/sdks/rust/tests/procedure-concurrency-client/src/module_bindings/mod.rs +++ b/sdks/rust/tests/procedure-concurrency-client/src/module_bindings/mod.rs @@ -307,6 +307,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -531,6 +535,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -607,6 +615,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -679,6 +691,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -751,6 +767,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -827,6 +847,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/sdks/rust/tests/procedure-concurrency-client/src/test_handlers.rs b/sdks/rust/tests/procedure-concurrency-client/src/test_handlers.rs index e059a8e2ee5..bb9cc0163db 100644 --- a/sdks/rust/tests/procedure-concurrency-client/src/test_handlers.rs +++ b/sdks/rust/tests/procedure-concurrency-client/src/test_handlers.rs @@ -66,7 +66,7 @@ async fn connect_with_then( callback(ctx); connected_result(Ok(())); }) - .on_connect_error(|_ctx, error| panic!("Connect errored: {error:?}")); + .on_connect_error(|_ctx, error, _next| panic!("Connect errored: {error:?}")); build_and_run(with_builder(builder)).await } @@ -82,7 +82,7 @@ const SUBSCRIBE_ALL: &[&str] = &["SELECT * FROM procedure_concurrency_row;"]; fn subscribe_all_then(ctx: &impl RemoteDbContext, callback: impl FnOnce(&SubscriptionEventContext) + Send + 'static) { ctx.subscription_builder() - .on_applied(callback) + .on_applied(test_counter::once::(callback)) .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) .subscribe(SUBSCRIBE_ALL); } diff --git a/sdks/rust/tests/reconnect.rs b/sdks/rust/tests/reconnect.rs new file mode 100644 index 00000000000..365cb642202 --- /dev/null +++ b/sdks/rust/tests/reconnect.rs @@ -0,0 +1,377 @@ +#![cfg(all(not(target_arch = "wasm32"), not(feature = "browser")))] + +#[path = "case-conversion-client/src/module_bindings/mod.rs"] +mod module_bindings; + +use module_bindings::*; +use spacetimedb_client_api_messages::websocket::v2 as ws; +use spacetimedb_lib::bsatn; +use spacetimedb_sdk::{AutomaticReconnectOptions, ConnectionId, DbContext, Identity}; +use std::{ + net::TcpListener, + sync::{mpsc, Arc, Mutex}, + time::{Duration, Instant}, +}; +use tokio_tungstenite::tungstenite::{ + self, + handshake::server::{Request, Response}, + Message, +}; + +fn accept(stream: std::net::TcpStream) -> tungstenite::WebSocket { + tungstenite::accept_hdr(stream, |_: &Request, mut response: Response| { + response + .headers_mut() + .insert("sec-websocket-protocol", "v2.bsatn.spacetimedb".parse().unwrap()); + Ok(response) + }) + .unwrap() +} + +fn send(socket: &mut tungstenite::WebSocket, message: ws::ServerMessage) { + let mut bytes = vec![0]; // uncompressed binary protocol + bytes.extend(bsatn::to_vec(&message).unwrap()); + socket.send(Message::Binary(bytes.into())).unwrap(); +} +fn tick_until(connection: &DbConnection, ready: impl Fn() -> bool) { + let deadline = Instant::now() + Duration::from_secs(8); + while !ready() { + let _ = connection.frame_tick(); + assert!(Instant::now() < deadline, "connection did not reach expected state"); + std::thread::sleep(Duration::from_millis(2)); + } +} +fn policy() -> AutomaticReconnectOptions { + AutomaticReconnectOptions { + min_delay: Duration::from_millis(500), + max_delay: Duration::from_secs(1), + } +} + +#[test] +fn reconnect_preserves_session_and_credentials_and_stops_on_identity_change() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let requests = Arc::new(Mutex::new(Vec::new())); + let captured = requests.clone(); + let (drop_tx, drop_rx) = mpsc::channel(); + let server = std::thread::spawn(move || { + for round in 0..3 { + let stream = listener.accept().unwrap().0; + let mut socket = tungstenite::accept_hdr(stream, |request: &Request, mut response: Response| { + captured.lock().unwrap().push(( + request.uri().to_string(), + request.headers().get("authorization").cloned(), + )); + response + .headers_mut() + .insert("sec-websocket-protocol", "v2.bsatn.spacetimedb".parse().unwrap()); + Ok(response) + }) + .unwrap(); + send( + &mut socket, + ws::ServerMessage::InitialConnection(ws::InitialConnection { + identity: Identity::from_byte_array([if round == 2 { 2 } else { 1 }; 32]), + token: "retained-token".into(), + connection_id: ConnectionId::from_u128(round + 1), + }), + ); + drop_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + } + }); + let events = Arc::new(Mutex::new(Vec::new())); + let connect = events.clone(); + let reconnect = events.clone(); + let disconnect = events.clone(); + let error = events.clone(); + let connection = DbConnection::builder() + .with_uri(format!("http://{address}")) + .with_database_name("test") + .with_automatic_reconnect_options(policy()) + .on_connect(move |_, _, _| connect.lock().unwrap().push("connect")) + .on_automatic_reconnect(move |_, _, _| reconnect.lock().unwrap().push("reconnect")) + .on_disconnect(move |_, _, next| { + assert_eq!(next.unwrap().attempt, 1); + disconnect.lock().unwrap().push("disconnect"); + }) + .on_connect_error(move |_, _, next| { + assert!(next.is_none()); + error.lock().unwrap().push("terminal"); + }) + .build() + .unwrap(); + tick_until(&connection, || events.lock().unwrap().len() == 1); + let identity = connection.identity(); + drop_tx.send(()).unwrap(); + tick_until(&connection, || events.lock().unwrap().len() == 3); + assert_eq!(connection.identity(), identity); + assert_eq!(connection.connection_id(), ConnectionId::from_u128(2)); + drop_tx.send(()).unwrap(); + tick_until(&connection, || events.lock().unwrap().len() == 5); + assert!(!connection.is_active() && !connection.is_reconnecting()); + assert_eq!( + *events.lock().unwrap(), + ["connect", "disconnect", "reconnect", "disconnect", "terminal"] + ); + let requests = requests.lock().unwrap(); + let session = |uri: &str| { + uri.split("session_id=") + .nth(1) + .unwrap() + .split('&') + .next() + .unwrap() + .to_owned() + }; + assert_eq!(session(&requests[0].0), session(&requests[1].0)); + assert!(requests[0].1.is_none()); + assert_eq!(requests[1].1.as_ref().unwrap(), "Bearer retained-token"); + drop_tx.send(()).unwrap(); + server.join().unwrap(); +} + +#[test] +fn initial_handshake_failure_never_retries() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = std::thread::spawn(move || { + let socket = accept(listener.accept().unwrap().0); + drop(socket); // No InitialConnection. + }); + let errors = Arc::new(Mutex::new(0)); + let called = errors.clone(); + let connection = DbConnection::builder() + .with_uri(format!("http://{address}")) + .with_database_name("test") + .with_automatic_reconnect() + .on_connect_error(move |_, _, next| { + assert!(next.is_none()); + *called.lock().unwrap() += 1; + }) + .build() + .unwrap(); + tick_until(&connection, || *errors.lock().unwrap() == 1); + assert!(!connection.is_reconnecting()); + server.join().unwrap(); +} + +#[test] +fn in_flight_call_is_unknown_and_disconnect_cancels_backoff() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = std::thread::spawn(move || { + let mut socket = accept(listener.accept().unwrap().0); + send( + &mut socket, + ws::ServerMessage::InitialConnection(ws::InitialConnection { + identity: Identity::from_byte_array([1; 32]), + token: "token".into(), + connection_id: ConnectionId::from_u128(1), + }), + ); + // Accept a reducer but deliberately lose its result. + let Message::Binary(bytes) = socket.read().unwrap() else { + panic!("expected reducer"); + }; + assert!(matches!( + bsatn::from_slice::(&bytes).unwrap(), + ws::ClientMessage::CallReducer(_) + )); + }); + let disconnects = Arc::new(Mutex::new(Vec::new())); + let events = disconnects.clone(); + let connection = DbConnection::builder() + .with_uri(format!("http://{address}")) + .with_database_name("test") + .with_automatic_reconnect_options(policy()) + .on_disconnect(move |_, error, next| { + events.lock().unwrap().push((error.is_none(), next)); + }) + .build() + .unwrap(); + tick_until(&connection, || connection.try_identity().is_some()); + let unknown = Arc::new(Mutex::new(false)); + let result = unknown.clone(); + connection + .reducers + .create_player_1_then("test".into(), 1, move |_, outcome| { + *result.lock().unwrap() = outcome.unwrap_err().is_unknown_result(); + }) + .unwrap(); + tick_until(&connection, || connection.is_reconnecting()); + assert!(*unknown.lock().unwrap()); + assert!(matches!( + connection.reducers.create_player_1("test".into(), 1), + Err(spacetimedb_sdk::Error::Disconnected) + )); + connection.disconnect().unwrap(); + assert!(connection.frame_tick().is_err()); + assert!(!connection.is_active() && !connection.is_reconnecting()); + assert_eq!(disconnects.lock().unwrap().len(), 2); + assert!(disconnects.lock().unwrap()[1].0); + assert!(disconnects.lock().unwrap()[1].1.is_none()); + server.join().unwrap(); +} + +#[test] +fn malformed_protocol_is_terminal() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let (go, wait) = mpsc::channel(); + let server = std::thread::spawn(move || { + let mut socket = accept(listener.accept().unwrap().0); + send( + &mut socket, + ws::ServerMessage::InitialConnection(ws::InitialConnection { + identity: Identity::from_byte_array([1; 32]), + token: "token".into(), + connection_id: ConnectionId::from_u128(1), + }), + ); + wait.recv().unwrap(); + socket.send(Message::Binary(vec![0, 255].into())).unwrap(); + }); + let terminal = Arc::new(Mutex::new(false)); + let event = terminal.clone(); + let connection = DbConnection::builder() + .with_uri(format!("http://{address}")) + .with_database_name("test") + .with_automatic_reconnect_options(policy()) + .on_disconnect(move |_, error, next| { + assert!(error.is_some()); + assert!(next.is_none()); + *event.lock().unwrap() = true; + }) + .build() + .unwrap(); + tick_until(&connection, || connection.try_identity().is_some()); + go.send(()).unwrap(); + tick_until(&connection, || *terminal.lock().unwrap()); + assert!(!connection.is_reconnecting()); + server.join().unwrap(); +} + +#[test] +fn rejected_retained_token_forces_refresh_even_after_provider_failure() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let (go, wait) = mpsc::channel(); + let server = std::thread::spawn(move || { + for round in 0..3 { + let stream = listener.accept().unwrap().0; + let socket = tungstenite::accept_hdr(stream, |request: &Request, mut response: Response| { + if round == 1 { + assert!(request.headers()["authorization"] + .to_str() + .unwrap() + .starts_with("Bearer header.")); + Err(tungstenite::http::Response::builder() + .status(401) + .body(Some("Rejected retained token".into())) + .unwrap()) + } else { + if round == 2 { + assert_eq!(request.headers()["authorization"], "Bearer refreshed"); + } + response + .headers_mut() + .insert("sec-websocket-protocol", "v2.bsatn.spacetimedb".parse().unwrap()); + Ok(response) + } + }); + if round == 1 { + assert!(socket.is_err()); + continue; + } + let mut socket = socket.unwrap(); + send( + &mut socket, + ws::ServerMessage::InitialConnection(ws::InitialConnection { + identity: Identity::from_byte_array([1; 32]), + // Valid far-future exp: refresh must be forced by the rejection. + token: "header.eyJleHAiOjQwMDAwMDAwMDAsImlhdCI6MTcwMDAwMDAwMH0.signature".into(), + connection_id: ConnectionId::from_u128(round + 1), + }), + ); + wait.recv_timeout(Duration::from_secs(10)).unwrap(); + } + }); + let calls = Arc::new(std::sync::atomic::AtomicU32::new(0)); + let called = calls.clone(); + let reconnects = Arc::new(Mutex::new(0)); + let reconnected = reconnects.clone(); + let connection = DbConnection::builder() + .with_uri(format!("http://{address}")) + .with_database_name("test") + .with_automatic_reconnect_options(policy()) + .with_token_provider(move || { + let attempt = called.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + async move { + if attempt == 0 { + Err(spacetimedb_sdk::Error::TokenProvider { + message: "refresh unavailable".into(), + }) + } else { + Ok("refreshed".into()) + } + } + }) + .on_automatic_reconnect(move |_, _, _| *reconnected.lock().unwrap() += 1) + .build() + .unwrap(); + tick_until(&connection, || connection.try_identity().is_some()); + go.send(()).unwrap(); + tick_until(&connection, || *reconnects.lock().unwrap() == 1); + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2); + connection.disconnect().unwrap(); + let _ = connection.frame_tick(); + go.send(()).unwrap(); + server.join().unwrap(); +} + +#[test] +fn async_runner_survives_a_drop_and_stops_from_reconnect_callback() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = std::thread::spawn(move || { + for round in 0..4 { + let mut socket = accept(listener.accept().unwrap().0); + send( + &mut socket, + ws::ServerMessage::InitialConnection(ws::InitialConnection { + identity: Identity::from_byte_array([1; 32]), + token: "retained".into(), + connection_id: ConnectionId::from_u128(round + 1), + }), + ); + if round == 3 { + let _ = socket.read(); + } + } + }); + let reconnected = Arc::new(Mutex::new(0)); + let called = reconnected.clone(); + let connection = DbConnection::builder() + .with_uri(format!("http://{address}")) + .with_database_name("test") + .with_automatic_reconnect_options(policy()) + .on_automatic_reconnect(move |ctx, _, _| { + let mut count = called.lock().unwrap(); + *count += 1; + if *count == 3 { + ctx.disconnect().unwrap(); + } + }) + .build() + .unwrap(); + let runtime = tokio::runtime::Runtime::new().unwrap(); + runtime.block_on(async { + tokio::time::timeout(Duration::from_secs(8), connection.run_async()) + .await + .unwrap() + .unwrap(); + }); + assert_eq!(*reconnected.lock().unwrap(), 3); + server.join().unwrap(); +} diff --git a/sdks/rust/tests/test-client/src/module_bindings/mod.rs b/sdks/rust/tests/test-client/src/module_bindings/mod.rs index 56cb87d3587..a4f40027af9 100644 --- a/sdks/rust/tests/test-client/src/module_bindings/mod.rs +++ b/sdks/rust/tests/test-client/src/module_bindings/mod.rs @@ -4352,6 +4352,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -4576,6 +4580,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -4652,6 +4660,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -4724,6 +4736,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -4796,6 +4812,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -4872,6 +4892,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/sdks/rust/tests/test-client/src/test_handlers.rs b/sdks/rust/tests/test-client/src/test_handlers.rs index c8b56c978fc..dcdd3c54bcb 100644 --- a/sdks/rust/tests/test-client/src/test_handlers.rs +++ b/sdks/rust/tests/test-client/src/test_handlers.rs @@ -374,7 +374,7 @@ async fn connect_with_then( callback(ctx); connected_result(Ok(())); }) - .on_connect_error(|_ctx, error| panic!("Connect errored: {error:?}")); + .on_connect_error(|_ctx, error, _next| panic!("Connect errored: {error:?}")); build_and_run(with_builder(builder)).await } @@ -420,7 +420,7 @@ fn subscribe_these_then( callback: impl FnOnce(&SubscriptionEventContext) + Send + 'static, ) { ctx.subscription_builder() - .on_applied(callback) + .on_applied(test_counter::once::(callback)) .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) .subscribe(queries); } @@ -446,9 +446,11 @@ async fn exec_subscribe_and_cancel(db_name: &str) { move |ctx| { let handle = ctx .subscription_builder() - .on_applied(move |_ctx: &SubscriptionEventContext| { - panic!("Subscription should never be applied"); - }) + .on_applied(test_counter::once::( + move |_ctx: &SubscriptionEventContext| { + panic!("Subscription should never be applied"); + }, + )) .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) .subscribe("SELECT * FROM one_u_8;"); assert!(!handle.is_active()); @@ -480,22 +482,24 @@ async fn exec_subscribe_and_unsubscribe(db_name: &str) { let hc_clone = handle_cell.clone(); let handle = ctx .subscription_builder() - .on_applied(move |ctx: &SubscriptionEventContext| { - let handle = { hc_clone.lock().unwrap().as_ref().unwrap().clone() }; - assert!(ctx.is_active()); - assert!(handle.is_active()); - assert!(!handle.is_ended()); - assert!(ctx.db.one_u_8().count() == 1); - let handle_clone = handle.clone(); - handle - .unsubscribe_then(Box::new(move |ectx| { - assert!(!handle_clone.is_active()); - assert!(handle_clone.is_ended()); - assert!(ectx.db.one_u_8().count() == 0); - cb(Ok(())); - })) - .unwrap(); - }) + .on_applied(test_counter::once::( + move |ctx: &SubscriptionEventContext| { + let handle = { hc_clone.lock().unwrap().as_ref().unwrap().clone() }; + assert!(ctx.is_active()); + assert!(handle.is_active()); + assert!(!handle.is_ended()); + assert!(ctx.db.one_u_8().count() == 1); + let handle_clone = handle.clone(); + handle + .unsubscribe_then(Box::new(move |ectx| { + assert!(!handle_clone.is_active()); + assert!(handle_clone.is_ended()); + assert!(ectx.db.one_u_8().count() == 0); + cb(Ok(())); + })) + .unwrap(); + }, + )) .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) .subscribe("SELECT * FROM one_u_8;"); handle_cell.lock().unwrap().replace(handle.clone()); @@ -514,9 +518,11 @@ async fn exec_subscription_error_smoke_test(db_name: &str) { move |ctx| { let handle = ctx .subscription_builder() - .on_applied(move |_ctx: &SubscriptionEventContext| { - panic!("Subscription should never be applied"); - }) + .on_applied(test_counter::once::( + move |_ctx: &SubscriptionEventContext| { + panic!("Subscription should never be applied"); + }, + )) .on_error(|_, _| cb(Ok(()))) .subscribe("SELEcCT * FROM one_u8;"); // intentional typo assert!(!handle.is_active()); @@ -1745,7 +1751,7 @@ async fn exec_reauth_part_1(db_name: &str) { .on_connect(move |_, _identity, token| { save_result(creds.save(token).map_err(Into::into)); }) - .on_connect_error(|_ctx, error| panic!("Connect failed: {error:?}")) + .on_connect_error(|_ctx, error, _next| panic!("Connect failed: {error:?}")) .with_database_name(name) .with_uri(server_url()) .build() @@ -1780,7 +1786,7 @@ async fn exec_reauth_part_2(db_name: &str) { creds_match_result(run_checks()); } }) - .on_connect_error(|_ctx, error| panic!("Connect failed: {error:?}")) + .on_connect_error(|_ctx, error, _next| panic!("Connect failed: {error:?}")) .with_database_name(name) .with_token(Some(token)) .with_uri(server_url()) @@ -1815,14 +1821,16 @@ async fn exec_reconnect_different_connection_id(db_name: &str) { DbConnection::builder() .with_database_name(db_name) .with_uri(server_url()) - .on_connect_error(|_ctx, error| panic!("on_connect_error: {error:?}")) + .on_connect_error(|_ctx, error, _next| panic!("on_connect_error: {error:?}")) .on_connect(move |_, _, _| { initial_connect_result(Ok(())); }) - .on_disconnect(|_, error| match error { - None => disconnect_result(Ok(())), - Some(err) => disconnect_result(Err(anyhow::anyhow!("{err:?}"))), - }), + .on_disconnect(test_counter::once3::( + |_, error, _next| match error { + None => disconnect_result(Ok(())), + Some(err) => disconnect_result(Err(anyhow::anyhow!("{err:?}"))), + }, + )), ) .await; @@ -1842,7 +1850,7 @@ async fn exec_reconnect_different_connection_id(db_name: &str) { DbConnection::builder() .with_database_name(db_name) .with_uri(server_url()) - .on_connect_error(|_ctx, error| panic!("on_connect_error: {error:?}")) + .on_connect_error(|_ctx, error, _next| panic!("on_connect_error: {error:?}")) .on_connect(move |ctx, _, _| { reconnect_result(Ok(())); let run_checks = || { @@ -1899,7 +1907,7 @@ async fn exec_subscribe_all_select_star(db_name: &str) { connection .subscription_builder() - .on_applied({ + .on_applied(test_counter::once::({ let test_counter = test_counter.clone(); move |ctx| { insert_one::(ctx, &test_counter, 0); @@ -1925,7 +1933,7 @@ async fn exec_subscribe_all_select_star(db_name: &str) { sub_applied_nothing_result(assert_all_tables_empty(ctx)); } - }) + })) .on_error(|_, e| panic!("Subscription error: {e:?}")) .subscribe_to_all_tables(); diff --git a/sdks/rust/tests/test-counter/src/lib.rs b/sdks/rust/tests/test-counter/src/lib.rs index 5ed52929ab2..0f9bfbcae80 100644 --- a/sdks/rust/tests/test-counter/src/lib.rs +++ b/sdks/rust/tests/test-counter/src/lib.rs @@ -215,3 +215,16 @@ impl TestCounter { } } } + +/// Adapt a one-shot test assertion to a repeatable subscription callback. +/// A second invocation fails the test, preserving the old tests' expectations. +pub fn once(callback: F) -> impl FnMut(&Ctx) { + let mut callback = Some(callback); + move |ctx| callback.take().expect("one-shot test callback fired twice")(ctx) +} + +/// The equivalent adapter for lifecycle callbacks with three arguments. +pub fn once3(callback: F) -> impl FnMut(&Ctx, A, B) { + let mut callback = Some(callback); + move |ctx, a, b| callback.take().expect("one-shot test callback fired twice")(ctx, a, b) +} diff --git a/sdks/rust/tests/view-client/src/module_bindings/mod.rs b/sdks/rust/tests/view-client/src/module_bindings/mod.rs index 2857b5d0af6..5b35de47148 100644 --- a/sdks/rust/tests/view-client/src/module_bindings/mod.rs +++ b/sdks/rust/tests/view-client/src/module_bindings/mod.rs @@ -342,6 +342,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -566,6 +570,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -642,6 +650,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -714,6 +726,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -786,6 +802,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -862,6 +882,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/sdks/rust/tests/view-client/src/test_handlers.rs b/sdks/rust/tests/view-client/src/test_handlers.rs index c8b0743166b..d35fdcd1f38 100644 --- a/sdks/rust/tests/view-client/src/test_handlers.rs +++ b/sdks/rust/tests/view-client/src/test_handlers.rs @@ -45,7 +45,7 @@ async fn build_connection( .on_connect(|ctx, _, _| { callback(ctx); }) - .on_connect_error(|_ctx, error| panic!("Connect errored: {error:?}")); + .on_connect_error(|_ctx, error, _next| panic!("Connect errored: {error:?}")); build_and_run(with_builder(builder)).await } @@ -78,7 +78,7 @@ fn subscribe_these_then( callback: impl FnOnce(&SubscriptionEventContext) + Send + 'static, ) { ctx.subscription_builder() - .on_applied(callback) + .on_applied(test_counter::once::(callback)) .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) .subscribe(queries); } @@ -95,20 +95,22 @@ async fn connect_my_player_client( build_connection( db_name, |builder| { - builder.on_disconnect(move |ctx, error| { - assert!( - !ctx.is_active(), - "on_disconnect callback, but `ctx.is_active()` is true" - ); - if let Some(disconnected_result) = disconnected_result { - match error { - Some(error) => disconnected_result(Err(anyhow::anyhow!("{error:?}"))), - None => disconnected_result(Ok(())), + builder.on_disconnect(test_counter::once3::( + move |ctx, error, _next| { + assert!( + !ctx.is_active(), + "on_disconnect callback, but `ctx.is_active()` is true" + ); + if let Some(disconnected_result) = disconnected_result { + match error { + Some(error) => disconnected_result(Err(anyhow::anyhow!("{error:?}"))), + None => disconnected_result(Ok(())), + } + } else if let Some(error) = error { + panic!("Disconnect errored: {error:?}"); } - } else if let Some(error) = error { - panic!("Disconnect errored: {error:?}"); - } - }) + }, + )) }, move |ctx| { subscribe_these_then(ctx, &["SELECT * FROM my_player"], { @@ -221,7 +223,7 @@ async fn exec_anonymous_subscribe_with_query_builder(db_name: &str) { connect_then(db_name, &test_counter, move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.player().on_insert(move |_, player| { if player.identity == Identity::from_byte_array([2; 32]) { return put_result(&mut insert_0, Ok(())); @@ -271,7 +273,7 @@ async fn exec_anonymous_subscribe_with_query_builder(db_name: &str) { reducer_callback_assert_committed("delete_player"), ) .unwrap(); - }) + })) .add_query(|ctx| { ctx.from .player_level() @@ -374,7 +376,7 @@ async fn exec_non_table_query_builder_return(db_name: &str) { connect_then(db_name, &test_counter, move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { let my_identity = ctx.identity(); ctx.db.my_player_and_level().on_insert(move |_, player| { assert_eq!(player.identity, my_identity); @@ -406,7 +408,7 @@ async fn exec_non_table_query_builder_return(db_name: &str) { ctx.reducers() .delete_player_then(my_identity, reducer_callback_assert_committed("delete_player")) .unwrap(); - }) + })) .add_query(|q_ctx| q_ctx.from.my_player_and_level().filter(|p| p.level.eq(1)).build()) .subscribe(); }) diff --git a/sdks/rust/tests/view-pk-client/src/module_bindings/mod.rs b/sdks/rust/tests/view-pk-client/src/module_bindings/mod.rs index 429aee0553b..6f9fe559f2f 100644 --- a/sdks/rust/tests/view-pk-client/src/module_bindings/mod.rs +++ b/sdks/rust/tests/view-pk-client/src/module_bindings/mod.rs @@ -366,6 +366,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -590,6 +594,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -666,6 +674,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -738,6 +750,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -810,6 +826,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -886,6 +906,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/sdks/rust/tests/view-pk-client/src/test_handlers.rs b/sdks/rust/tests/view-pk-client/src/test_handlers.rs index e19a5677449..2a5843022a9 100644 --- a/sdks/rust/tests/view-pk-client/src/test_handlers.rs +++ b/sdks/rust/tests/view-pk-client/src/test_handlers.rs @@ -58,7 +58,7 @@ async fn connect_then_named( callback(ctx); connected_result(Ok(())); }) - .on_connect_error(|_ctx, error| panic!("Connect errored: {error:?}")); + .on_connect_error(|_ctx, error, _next| panic!("Connect errored: {error:?}")); build_and_run(conn).await } @@ -68,7 +68,7 @@ fn subscribe_these_then( callback: impl FnOnce(&SubscriptionEventContext) + Send + 'static, ) { ctx.subscription_builder() - .on_applied(callback) + .on_applied(test_counter::once::(callback)) .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) .subscribe(queries); } @@ -150,7 +150,7 @@ async fn exec_view_pk_join_query_builder(db_name: &str) { connect_then(db_name, &test_counter, move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.all_view_pk_players().on_update(move |_, old_row, new_row| { assert_eq!(old_row.id, 1); assert_eq!(old_row.name, "before"); @@ -185,7 +185,7 @@ async fn exec_view_pk_join_query_builder(db_name: &str) { reducer_callback_assert_committed("update_view_pk_player"), ) .unwrap(); - }) + })) .add_query(|q| { q.from .view_pk_membership() @@ -229,7 +229,7 @@ async fn exec_view_pk_semijoin_two_sender_views_query_builder(db_name: &str) { connect_then(db_name, &test_counter, move |ctx| { ctx.subscription_builder() .on_error(|_ctx, error| panic!("Subscription errored: {error:?}")) - .on_applied(move |ctx| { + .on_applied(test_counter::once::(move |ctx| { ctx.db.sender_view_pk_players_b().on_update(move |_, old_row, new_row| { assert_eq!(old_row.id, 1); assert_eq!(old_row.name, "before"); @@ -273,7 +273,7 @@ async fn exec_view_pk_semijoin_two_sender_views_query_builder(db_name: &str) { reducer_callback_assert_committed("update_view_pk_player"), ) .unwrap(); - }) + })) .add_query(|q| { q.from .sender_view_pk_players_a() diff --git a/templates/basic-cpp/src/main.rs b/templates/basic-cpp/src/main.rs index c7d6b1b4f4a..ea39e8673f3 100644 --- a/templates/basic-cpp/src/main.rs +++ b/templates/basic-cpp/src/main.rs @@ -18,7 +18,7 @@ fn main() { .on_connect(|_, _, _| { println!("Connected to SpacetimeDB"); }) - .on_connect_error(|_ctx, e| { + .on_connect_error(|_ctx, e, _next| { eprintln!("Connection error: {:?}", e); std::process::exit(1); }) diff --git a/templates/basic-cpp/src/module_bindings/mod.rs b/templates/basic-cpp/src/module_bindings/mod.rs index 0bf2df1c622..30c372af919 100644 --- a/templates/basic-cpp/src/module_bindings/mod.rs +++ b/templates/basic-cpp/src/module_bindings/mod.rs @@ -224,6 +224,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -440,6 +444,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -516,6 +524,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -588,6 +600,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -660,6 +676,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -736,6 +756,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/templates/basic-rs/src/main.rs b/templates/basic-rs/src/main.rs index c7d6b1b4f4a..ea39e8673f3 100644 --- a/templates/basic-rs/src/main.rs +++ b/templates/basic-rs/src/main.rs @@ -18,7 +18,7 @@ fn main() { .on_connect(|_, _, _| { println!("Connected to SpacetimeDB"); }) - .on_connect_error(|_ctx, e| { + .on_connect_error(|_ctx, e, _next| { eprintln!("Connection error: {:?}", e); std::process::exit(1); }) diff --git a/templates/basic-rs/src/module_bindings/mod.rs b/templates/basic-rs/src/module_bindings/mod.rs index 0bf2df1c622..30c372af919 100644 --- a/templates/basic-rs/src/module_bindings/mod.rs +++ b/templates/basic-rs/src/module_bindings/mod.rs @@ -224,6 +224,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -440,6 +444,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -516,6 +524,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -588,6 +600,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -660,6 +676,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -736,6 +756,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } diff --git a/templates/chat-console-rs/src/main.rs b/templates/chat-console-rs/src/main.rs index 81ee204a456..877ab16fbff 100644 --- a/templates/chat-console-rs/src/main.rs +++ b/templates/chat-console-rs/src/main.rs @@ -71,13 +71,13 @@ fn on_connected(_ctx: &DbConnection, _identity: Identity, token: &str) { // ### Handle errors and disconnections /// Our `on_connect_error` callback: print the error, then exit the process. -fn on_connect_error(_ctx: &ErrorContext, err: Error) { +fn on_connect_error(_ctx: &ErrorContext, err: Error, _next: Option) { eprintln!("Connection error: {err}"); std::process::exit(1); } /// Our `on_disconnect` callback: print a note, then exit the process. -fn on_disconnected(_ctx: &ErrorContext, err: Option) { +fn on_disconnected(_ctx: &ErrorContext, err: Option, _next: Option) { if let Some(err) = err { eprintln!("Disconnected: {err}"); std::process::exit(1); diff --git a/templates/chat-console-rs/src/module_bindings/mod.rs b/templates/chat-console-rs/src/module_bindings/mod.rs index b40a5c09bef..73d7216244e 100644 --- a/templates/chat-console-rs/src/module_bindings/mod.rs +++ b/templates/chat-console-rs/src/module_bindings/mod.rs @@ -245,6 +245,10 @@ impl __sdk::DbContext for DbConnection { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -461,6 +465,10 @@ impl __sdk::DbContext for EventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -537,6 +545,10 @@ impl __sdk::DbContext for ReducerEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -609,6 +621,10 @@ impl __sdk::DbContext for ProcedureEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -681,6 +697,10 @@ impl __sdk::DbContext for SubscriptionEventContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() } @@ -757,6 +777,10 @@ impl __sdk::DbContext for ErrorContext { self.imp.is_active() } + fn is_reconnecting(&self) -> bool { + self.imp.is_reconnecting() + } + fn disconnect(&self) -> __sdk::Result<()> { self.imp.disconnect() }