From 617f124240773556bee8f6d45093b76e7ad6a48f Mon Sep 17 00:00:00 2001 From: Jeffrey Rooks Date: Mon, 3 Aug 2026 09:59:24 -0400 Subject: [PATCH 1/4] add query builder as trait capability --- crates/bindings/src/lib.rs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/crates/bindings/src/lib.rs b/crates/bindings/src/lib.rs index 2c7faa78c6b..94f4d06d08d 100644 --- a/crates/bindings/src/lib.rs +++ b/crates/bindings/src/lib.rs @@ -1585,6 +1585,23 @@ impl CtxDbRead for AnonymousViewContext { } } +/// Contexts which provide access to the view query builder. +pub trait CtxWithQueryBuilder { + fn query_builder(&self) -> &QueryBuilder; +} + +impl CtxWithQueryBuilder for ViewContext { + fn query_builder(&self) -> &QueryBuilder { + &self.from + } +} + +impl CtxWithQueryBuilder for AnonymousViewContext { + fn query_builder(&self) -> &QueryBuilder { + &self.from + } +} + /// Contexts which provide read-write access to the database. /// /// This trait is useful for writing reusable logic which is generic over the context type, From d8a54378d9b9aa603e67a898c9ea1074079319a4 Mon Sep 17 00:00:00 2001 From: Jeffrey Rooks Date: Fri, 21 Aug 2026 22:28:42 -0400 Subject: [PATCH 2/4] disconnect with intent --- sdks/rust/src/db_connection.rs | 115 +++++++++++++++++++++++++++++++-- sdks/rust/src/lib.rs | 2 +- 2 files changed, 109 insertions(+), 8 deletions(-) diff --git a/sdks/rust/src/db_connection.rs b/sdks/rust/src/db_connection.rs index 332aac1b322..b3164e5d640 100644 --- a/sdks/rust/src/db_connection.rs +++ b/sdks/rust/src/db_connection.rs @@ -45,7 +45,10 @@ use std::{ fs::File, io::Write, path::PathBuf, - sync::{atomic::AtomicU32, Arc, Mutex as StdMutex, OnceLock}, + sync::{ + atomic::{AtomicBool, AtomicU32, Ordering}, + Arc, Mutex as StdMutex, OnceLock, + }, }; #[cfg(not(feature = "browser"))] use tokio::{ @@ -55,6 +58,15 @@ use tokio::{ pub(crate) type SharedCell = Arc>; +/// Describes the intent associated with an established connection closing without an SDK error. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DisconnectIntent { + /// The client explicitly requested the connection to close. + Requested, + /// The connection closed without a local request or SDK error. + Lost, +} + #[cfg(not(feature = "browser"))] type SharedAsyncCell = Arc>; #[cfg(feature = "browser")] @@ -75,6 +87,9 @@ pub struct DbContextImpl { /// None if we have disconnected. pub(crate) send_chan: SharedCell>>, + /// Whether this connection was explicitly asked to disconnect. + disconnect_requested: Arc, + /// The client cache, which stores subscribed rows. cache: SharedCell>, @@ -116,6 +131,7 @@ impl Clone for DbContextImpl { // and we need it to be fast. inner: Arc::clone(&self.inner), send_chan: Arc::clone(&self.send_chan), + disconnect_requested: Arc::clone(&self.disconnect_requested), cache: Arc::clone(&self.cache), recv: Arc::clone(&self.recv), pending_mutations_send: self.pending_mutations_send.clone(), @@ -328,6 +344,7 @@ impl DbContextImpl { // Set `send_chan` to `None`, since `Self::is_active` checks that. *self.send_chan.lock().unwrap() = None; + let disconnect_requested = self.disconnect_requested.swap(false, Ordering::AcqRel); match lifecycle { ConnectionLifecycle::Connecting => { @@ -342,7 +359,9 @@ impl DbContextImpl { } ConnectionLifecycle::Connected => { let ctx: M::ErrorContext = self.make_event_ctx(callback_error.clone()); - if let Some(disconnect_callback) = inner.on_disconnect.take() { + if let Some(disconnect_callback) = inner.on_disconnect_with_intent.take() { + disconnect_callback(&ctx, disconnect_result(disconnect_requested, callback_error.clone())); + } else if let Some(disconnect_callback) = inner.on_disconnect.take() { disconnect_callback(&ctx, callback_error.clone()); } @@ -714,6 +733,7 @@ impl DbContextImpl { if !self.is_active() { return Err(crate::Error::Disconnected); } + self.disconnect_requested.store(true, Ordering::Release); self.pending_mutations_send .unbounded_send(PendingMutation::Disconnect) .unwrap(); @@ -811,6 +831,9 @@ type OnConnectErrorCallback = Box::ErrorCo type OnDisconnectCallback = Box::ErrorContext, Option) + Send + 'static>; +type OnDisconnectWithIntentCallback = + Box::ErrorContext, Result) + Send + 'static>; + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum ConnectionLifecycle { /// Waiting for the server's initial connection message. @@ -837,6 +860,7 @@ pub(crate) struct DbContextImplInner { on_connect: Option>, on_connect_error: Option>, on_disconnect: Option>, + on_disconnect_with_intent: Option>, procedure_callbacks: ProcedureCallbacks, } @@ -857,6 +881,7 @@ pub struct DbConnectionBuilder { on_connect: Option>, on_connect_error: Option>, on_disconnect: Option>, + on_disconnect_with_intent: Option>, additional_logging_path: Option, @@ -916,6 +941,7 @@ impl DbConnectionBuilder { on_connect: None, on_connect_error: None, on_disconnect: None, + on_disconnect_with_intent: None, additional_logging_path: None, params: <_>::default(), } @@ -998,7 +1024,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_connect_error, + self.on_disconnect, + self.on_disconnect_with_intent, + ); Ok(build_db_ctx( handle, inner_ctx, @@ -1042,7 +1074,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_connect_error, + self.on_disconnect, + self.on_disconnect_with_intent, + ); Ok(build_db_ctx( inner_ctx, raw_msg_send, @@ -1177,20 +1214,45 @@ Instead of registering multiple `on_connect_error` callbacks, register a single /// The connection is established after the initial connection message is /// received from the host. Connection failures before that point invoke /// [`Self::on_connect_error`] instead. + // TODO(v3): Replace this callback with the detailed disconnect-intent API. pub fn on_disconnect( mut self, callback: impl FnOnce(&M::ErrorContext, Option) + Send + 'static, ) -> Self { - if self.on_disconnect.is_some() { + if self.on_disconnect.is_some() || self.on_disconnect_with_intent.is_some() { panic!( - "DbConnectionBuilder can only register a single `on_disconnect` callback. + "DbConnectionBuilder can only register a single disconnect callback. -Instead of registering multiple `on_disconnect` callbacks, register a single callback which does multiple operations." +Use either `on_disconnect` or `on_disconnect_with_intent`, not both." ); } self.on_disconnect = Some(Box::new(callback)); self } + + /// Register a callback with the intent or error associated with an established connection closing. + /// + /// The callback receives `Ok(DisconnectIntent::Requested)` when the client explicitly called + /// [`DbContext::disconnect`], `Ok(DisconnectIntent::Lost)` when the connection closed without + /// an SDK error, and `Err` when the SDK reported a disconnect error. + /// Connection failures before the initial connection message invoke + /// [`Self::on_connect_error`] instead. + /// + /// This is mutually exclusive with [`Self::on_disconnect`]. + pub fn on_disconnect_with_intent( + mut self, + callback: impl FnOnce(&M::ErrorContext, Result) + Send + 'static, + ) -> Self { + if self.on_disconnect.is_some() || self.on_disconnect_with_intent.is_some() { + panic!( + "DbConnectionBuilder can only register a single disconnect callback. + +Use either `on_disconnect` or `on_disconnect_with_intent`, not both." + ); + } + self.on_disconnect_with_intent = Some(Box::new(callback)); + self + } } /// Create a [`DbContextImplInner`] wrapped in `Arc>`. @@ -1200,6 +1262,7 @@ fn build_db_ctx_inner( on_connect_cb: Option>, on_connect_error_cb: Option>, on_disconnect_cb: Option>, + on_disconnect_with_intent_cb: Option>, ) -> Arc>> { Arc::new(StdMutex::new(DbContextImplInner { #[cfg(not(feature = "browser"))] @@ -1213,6 +1276,7 @@ fn build_db_ctx_inner( on_connect: on_connect_cb, on_connect_error: on_connect_error_cb, on_disconnect: on_disconnect_cb, + on_disconnect_with_intent: on_disconnect_with_intent_cb, procedure_callbacks: ProcedureCallbacks::default(), })) @@ -1240,6 +1304,7 @@ fn build_db_ctx( runtime: runtime_handle, inner: inner_ctx, send_chan: Arc::new(StdMutex::new(Some(raw_msg_send))), + disconnect_requested: Arc::new(AtomicBool::new(false)), cache, recv: parsed_msg_recv, pending_mutations_send, @@ -1603,10 +1668,46 @@ enum Message { Local(PendingMutation), } +fn disconnect_result( + disconnect_requested: bool, + callback_error: Option, +) -> Result { + if disconnect_requested { + Ok(DisconnectIntent::Requested) + } else if let Some(error) = callback_error { + Err(error) + } else { + Ok(DisconnectIntent::Lost) + } +} + fn error_is_normal_disconnect(e: &crate::Error) -> bool { matches!(e, crate::Error::Disconnected) } +#[cfg(test)] +mod tests { + use super::{disconnect_result, DisconnectIntent}; + + #[test] + fn requested_disconnect_is_reported_as_requested() { + assert!(matches!(disconnect_result(true, None), Ok(DisconnectIntent::Requested))); + } + + #[test] + fn clean_disconnect_is_reported_as_lost() { + assert!(matches!(disconnect_result(false, None), Ok(DisconnectIntent::Lost))); + } + + #[test] + fn errored_disconnect_preserves_the_sdk_error() { + assert!(matches!( + disconnect_result(false, Some(crate::Error::Disconnected)), + Err(crate::Error::Disconnected) + )); + } +} + static NEXT_REQUEST_ID: AtomicU32 = AtomicU32::new(1); // Get the next request ID to use for a WebSocket message. diff --git a/sdks/rust/src/lib.rs b/sdks/rust/src/lib.rs index e7efd286514..d0183fe8b25 100644 --- a/sdks/rust/src/lib.rs +++ b/sdks/rust/src/lib.rs @@ -25,7 +25,7 @@ pub mod error; pub mod event; pub mod table; -pub use db_connection::DbConnectionBuilder; +pub use db_connection::{DbConnectionBuilder, DisconnectIntent}; pub use db_context::DbContext; pub use error::{Error, Result}; pub use event::{Event, ReducerEvent, Status}; From 2b00e98035fb807a859456473eca8eb9e63c3681 Mon Sep 17 00:00:00 2001 From: Jeffrey Rooks Date: Fri, 21 Aug 2026 22:35:46 -0400 Subject: [PATCH 3/4] reset file --- crates/bindings/src/lib.rs | 17 ----------------- 1 file changed, 17 deletions(-) diff --git a/crates/bindings/src/lib.rs b/crates/bindings/src/lib.rs index 94f4d06d08d..2c7faa78c6b 100644 --- a/crates/bindings/src/lib.rs +++ b/crates/bindings/src/lib.rs @@ -1585,23 +1585,6 @@ impl CtxDbRead for AnonymousViewContext { } } -/// Contexts which provide access to the view query builder. -pub trait CtxWithQueryBuilder { - fn query_builder(&self) -> &QueryBuilder; -} - -impl CtxWithQueryBuilder for ViewContext { - fn query_builder(&self) -> &QueryBuilder { - &self.from - } -} - -impl CtxWithQueryBuilder for AnonymousViewContext { - fn query_builder(&self) -> &QueryBuilder { - &self.from - } -} - /// Contexts which provide read-write access to the database. /// /// This trait is useful for writing reusable logic which is generic over the context type, From 5bdb357b612f4c8db5980a4978b16a1228807428 Mon Sep 17 00:00:00 2001 From: Jeffrey Rooks Date: Fri, 21 Aug 2026 22:39:32 -0400 Subject: [PATCH 4/4] comment cleanup --- sdks/rust/src/db_connection.rs | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/sdks/rust/src/db_connection.rs b/sdks/rust/src/db_connection.rs index b3164e5d640..49cef308e10 100644 --- a/sdks/rust/src/db_connection.rs +++ b/sdks/rust/src/db_connection.rs @@ -1230,14 +1230,18 @@ Use either `on_disconnect` or `on_disconnect_with_intent`, not both." self } - /// Register a callback with the intent or error associated with an established connection closing. + /// Register a callback to run when an established connection is closed, + /// with its disconnect intent. /// - /// The callback receives `Ok(DisconnectIntent::Requested)` when the client explicitly called - /// [`DbContext::disconnect`], `Ok(DisconnectIntent::Lost)` when the connection closed without - /// an SDK error, and `Err` when the SDK reported a disconnect error. - /// Connection failures before the initial connection message invoke + /// The connection is established after the initial connection message is + /// received from the host. Connection failures before that point invoke /// [`Self::on_connect_error`] instead. /// + /// The callback receives `Ok(DisconnectIntent::Requested)` when the client + /// explicitly called [`DbContext::disconnect`], `Ok(DisconnectIntent::Lost)` + /// when the connection closed without an SDK error, and `Err` when the SDK + /// reports a disconnect error. + /// /// This is mutually exclusive with [`Self::on_disconnect`]. pub fn on_disconnect_with_intent( mut self,