Add a mode to the Rust SDK with additional logging to a file

In the Rust client SDK, with this PR,
doing `.with_debug_to_file("path.txt")` on the connection builder
will result in the SDK logging additional verbose info to `path.txt`.

Author: gefjon

crates/codegen/src/rust.rs

@@ -1308,7 +1308,7 @@ impl __sdk::InModule for Reducer {{
 }
 
 fn print_db_update_defn(module: &ModuleDef, visibility: CodegenVisibility, out: &mut Indenter) {
-    writeln!(out, "#[derive(Default)]");
+    writeln!(out, "#[derive(Default, Debug)]");
     writeln!(out, "#[allow(non_snake_case)]");
     writeln!(out, "#[doc(hidden)]");
     out.delimited_block(
@@ -1606,6 +1606,7 @@ fn print_const_db_context_types(out: &mut Indenter) {
         out,
         "
 #[doc(hidden)]
+#[derive(Debug)]
 pub struct RemoteModule;
 
 impl __sdk::InModule for RemoteModule {{

sdks/rust/src/db_connection.rs

@@ -37,7 +37,12 @@ use http::Uri;
 use spacetimedb_client_api_messages::websocket::{self as ws, common::QuerySetId};
 use spacetimedb_lib::{bsatn, ser::Serialize, ConnectionId, Identity, Timestamp};
 use spacetimedb_sats::Deserialize;
-use std::sync::{atomic::AtomicU32, Arc, Mutex as StdMutex, OnceLock};
+use std::{
+    fs::{File, OpenOptions},
+    io::Write,
+    path::PathBuf,
+    sync::{atomic::AtomicU32, Arc, Mutex as StdMutex, OnceLock},
+};
 use tokio::{
     runtime::{self, Runtime},
     sync::Mutex as TokioMutex,
@@ -86,6 +91,8 @@ pub struct DbContextImpl<M: SpacetimeModule> {
     ///
     /// This may be none if we have not yet received the [`ws::v2::InitialConnection`] message.
     connection_id: SharedCell<Option<ConnectionId>>,
+
+    pub(crate) extra_logging: Option<SharedCell<File>>,
 }
 
 impl<M: SpacetimeModule> Clone for DbContextImpl<M> {
@@ -103,14 +110,20 @@ impl<M: SpacetimeModule> Clone for DbContextImpl<M> {
             pending_mutations_recv: Arc::clone(&self.pending_mutations_recv),
             identity: Arc::clone(&self.identity),
             connection_id: Arc::clone(&self.connection_id),
+            extra_logging: Option::<Arc<_>>::clone(&self.extra_logging),
         }
     }
 }
 
 impl<M: SpacetimeModule> DbContextImpl<M> {
+    pub(crate) fn debug_log(&self, body: impl FnOnce(&mut File) -> std::result::Result<(), std::io::Error>) {
+        debug_log(&self.extra_logging, body);
+    }
+
     /// Process a parsed WebSocket message,
     /// applying its mutations to the client cache and invoking callbacks.
     fn process_message(&self, msg: ParsedMessage<M>) -> crate::Result<()> {
+        self.debug_log(|out| writeln!(out, "`process_message`: {msg:?}"));
         match msg {
             // Error: treat this as an erroneous disconnect.
             ParsedMessage::Error(e) => {
@@ -315,6 +328,7 @@ impl<M: SpacetimeModule> DbContextImpl<M> {
 
     /// Apply an individual [`PendingMutation`].
     fn apply_mutation(&self, mutation: PendingMutation<M>) -> crate::Result<()> {
+        self.debug_log(|out| writeln!(out, "`apply_mutation`: {mutation:?}"));
         match mutation {
             // Subscribe: register the subscription in the [`SubscriptionManager`]
             // and send the `Subscribe` WS message.
@@ -763,6 +777,8 @@ pub struct DbConnectionBuilder<M: SpacetimeModule> {
     on_connect_error: Option<OnConnectErrorCallback<M>>,
     on_disconnect: Option<OnDisconnectCallback<M>>,
 
+    additional_logging_path: Option<PathBuf>,
+
     params: WsParams,
 }
 
@@ -798,6 +814,15 @@ pub fn set_connection_id(id: ConnectionId) -> crate::Result<()> {
     Ok(())
 }
 
+pub(crate) fn debug_log(
+    extra_logging: &Option<SharedCell<File>>,
+    body: impl FnOnce(&mut File) -> std::result::Result<(), std::io::Error>,
+) {
+    if let Some(file) = extra_logging {
+        body(&mut *file.lock().expect("`extra_logging` file Mutex is poisoned")).expect("Writing debug log failed")
+    }
+}
+
 impl<M: SpacetimeModule> DbConnectionBuilder<M> {
     /// Implementation of the generated `DbConnection::builder` method.
     /// Call that method instead.
@@ -810,6 +835,7 @@ impl<M: SpacetimeModule> DbConnectionBuilder<M> {
             on_connect: None,
             on_connect_error: None,
             on_disconnect: None,
+            additional_logging_path: None,
             params: <_>::default(),
         }
     }
@@ -847,6 +873,16 @@ 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`].
     fn build_impl(self) -> crate::Result<DbContextImpl<M>> {
+        let extra_logging = self
+            .additional_logging_path
+            .map(|path| {
+                OpenOptions::new().append(true).create(true).open(&path).map_err(|e| {
+                    InternalError::new(format!("Failed to open file '{path:?}' for additional logging")).with_cause(e)
+                })
+            })
+            .transpose()?
+            .map(|file| Arc::new(StdMutex::new(file)));
+
         let (runtime, handle) = enter_or_create_runtime()?;
         let db_callbacks = DbCallbacks::default();
         let reducer_callbacks = ReducerCallbacks::default();
@@ -866,8 +902,10 @@ but you must call one of them, or else the connection will never progress.
             source: InternalError::new("Failed to initiate WebSocket connection").with_cause(source),
         })?;
 
-        let (_websocket_loop_handle, raw_msg_recv, raw_msg_send) = ws_connection.spawn_message_loop(&handle);
-        let (_parse_loop_handle, parsed_recv_chan) = spawn_parse_loop::<M>(raw_msg_recv, &handle);
+        let (_websocket_loop_handle, raw_msg_recv, raw_msg_send) =
+            ws_connection.spawn_message_loop(&handle, extra_logging.clone());
+        let (_parse_loop_handle, parsed_recv_chan) =
+            spawn_parse_loop::<M>(raw_msg_recv, &handle, extra_logging.clone());
 
         let inner = Arc::new(StdMutex::new(DbContextImplInner {
             runtime,
@@ -898,6 +936,7 @@ but you must call one of them, or else the connection will never progress.
             pending_mutations_recv: Arc::new(TokioMutex::new(pending_mutations_recv)),
             identity: Arc::new(StdMutex::new(None)),
             connection_id: Arc::new(StdMutex::new(connection_id_override)),
+            extra_logging,
         };
 
         Ok(ctx_imp)
@@ -963,6 +1002,23 @@ but you must call one of them, or else the connection will never progress.
         self
     }
 
+    /// Set `path` as a path for additional debug logging related to SDK internals.
+    ///
+    /// When enabled, the SDK will create or open `path` for write-append and write logs to it.
+    /// This is useful for diagnosing bugs in the SDK,
+    /// but will generate a large volume of text logs and may have performance overhead,
+    /// so it should not be used in production.
+    ///
+    /// When running multiple connections in parallel,
+    /// either within the same process or from separate processes,
+    /// prefer giving each its own unique path here;
+    /// multiple `DbConnection`s writing to the same debug file concurrently
+    /// may interleave or corrupt the output.
+    pub fn with_debug_to_file(mut self, path: impl Into<PathBuf>) -> Self {
+        self.additional_logging_path = Some(path.into());
+        self
+    }
+
     /// Register a callback to run when the connection is successfully initiated.
     ///
     /// The callback will receive three arguments:
@@ -1043,6 +1099,7 @@ fn enter_or_create_runtime() -> crate::Result<(Option<Runtime>, runtime::Handle)
     }
 }
 
+#[derive(Debug)]
 enum ParsedMessage<M: SpacetimeModule> {
     TransactionUpdate(M::DbUpdate),
     IdentityToken(Identity, Box<str>, ConnectionId),
@@ -1073,9 +1130,10 @@ enum ParsedMessage<M: SpacetimeModule> {
 fn spawn_parse_loop<M: SpacetimeModule>(
     raw_message_recv: mpsc::UnboundedReceiver<ws::v2::ServerMessage>,
     handle: &runtime::Handle,
+    extra_logging: Option<SharedCell<File>>,
 ) -> (tokio::task::JoinHandle<()>, mpsc::UnboundedReceiver<ParsedMessage<M>>) {
     let (parsed_message_send, parsed_message_recv) = mpsc::unbounded();
-    let handle = handle.spawn(parse_loop(raw_message_recv, parsed_message_send));
+    let handle = handle.spawn(parse_loop(raw_message_recv, parsed_message_send, extra_logging));
     (handle, parsed_message_recv)
 }
 
@@ -1084,9 +1142,13 @@ fn spawn_parse_loop<M: SpacetimeModule>(
 async fn parse_loop<M: SpacetimeModule>(
     mut recv: mpsc::UnboundedReceiver<ws::v2::ServerMessage>,
     send: mpsc::UnboundedSender<ParsedMessage<M>>,
+    extra_logging: Option<SharedCell<File>>,
 ) {
     while let Some(msg) = recv.next().await {
-        send.unbounded_send(match msg {
+        debug_log(&extra_logging, |file| {
+            writeln!(file, "`parse_loop`: Got raw message: {msg:?}")
+        });
+        let parsed = match msg {
             ws::v2::ServerMessage::TransactionUpdate(transaction_update) => {
                 match M::DbUpdate::parse_update(transaction_update) {
                     Err(e) => ParsedMessage::Error(
@@ -1210,8 +1272,12 @@ async fn parse_loop<M: SpacetimeModule>(
                     ws::v2::ProcedureStatus::Returned(val) => Ok(val),
                 },
             },
-        })
-        .expect("Failed to send ParsedMessage to main thread");
+        };
+        debug_log(&extra_logging, |file| {
+            writeln!(file, "`parse_loop`: Parsed as: {parsed:?}")
+        });
+        send.unbounded_send(parsed)
+            .expect("Failed to send ParsedMessage to main thread");
     }
 }
 
@@ -1263,6 +1329,62 @@ pub(crate) enum PendingMutation<M: SpacetimeModule> {
     },
 }
 
+// Hand-written `Debug` impl, 'cause `SubscriptionHandleImpl` and callbacks aren't printable.
+impl<M: SpacetimeModule> std::fmt::Debug for PendingMutation<M> {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        match self {
+            PendingMutation::Unsubscribe { query_set_id } => f
+                .debug_struct("PendingMutation::Unsubscribe")
+                .field("query_set_id", query_set_id)
+                .finish(),
+            PendingMutation::Subscribe { query_set_id, .. } => f
+                .debug_struct("PendingMutation::Subscribe")
+                .field("query_set_id", query_set_id)
+                .finish_non_exhaustive(),
+            PendingMutation::AddInsertCallback { table, callback_id, .. } => f
+                .debug_struct("PendingMutation::AddInsertCallback")
+                .field("table", table)
+                .field("callback_id", callback_id)
+                .finish_non_exhaustive(),
+            PendingMutation::RemoveInsertCallback { table, callback_id } => f
+                .debug_struct("PendingMutation::RemoveInsertCallback")
+                .field("table", table)
+                .field("callback_id", callback_id)
+                .finish(),
+            PendingMutation::AddDeleteCallback { table, callback_id, .. } => f
+                .debug_struct("PendingMutation::AddDeleteCallback")
+                .field("table", table)
+                .field("callback_id", callback_id)
+                .finish_non_exhaustive(),
+            PendingMutation::RemoveDeleteCallback { table, callback_id } => f
+                .debug_struct("PendingMutation::RemoveDeleteCallback")
+                .field("table", table)
+                .field("callback_id", callback_id)
+                .finish(),
+            PendingMutation::AddUpdateCallback { table, callback_id, .. } => f
+                .debug_struct("PendingMutation::AddUpdateCallback")
+                .field("table", table)
+                .field("callback_id", callback_id)
+                .finish_non_exhaustive(),
+            PendingMutation::RemoveUpdateCallback { table, callback_id } => f
+                .debug_struct("PendingMutation::RemoveUpdateCallback")
+                .field("table", table)
+                .field("callback_id", callback_id)
+                .finish(),
+            PendingMutation::Disconnect => write!(f, "PendingMutation::Disconnect"),
+            PendingMutation::InvokeReducerWithCallback { reducer, .. } => f
+                .debug_struct("PendingMutation::InvokeReducerWithCallback")
+                .field("reducer", reducer)
+                .finish_non_exhaustive(),
+            PendingMutation::InvokeProcedureWithCallback { procedure, args, .. } => f
+                .debug_struct("PendingMutation::InvokeProcedureWithCallback")
+                .field("procedure", procedure)
+                .field("args", args)
+                .finish_non_exhaustive(),
+        }
+    }
+}
+
 enum Message<M: SpacetimeModule> {
     Ws(Option<ParsedMessage<M>>),
     Local(PendingMutation<M>),

sdks/rust/src/spacetime_module.rs

@@ -32,7 +32,7 @@ pub struct QueryTableAccessor;
 
 /// Each module's codegen will define a unit struct which implements this trait,
 /// with associated type links to various other generated types.
-pub trait SpacetimeModule: Send + Sync + 'static {
+pub trait SpacetimeModule: Debug + Send + Sync + 'static {
     /// [`crate::DbContext`] implementor which exists in the global scope.
     type DbConnection: DbConnection<Module = Self>;
 
@@ -83,7 +83,7 @@ pub trait SpacetimeModule: Send + Sync + 'static {
 /// Implemented by the autogenerated `DbUpdate` type,
 /// which is a parsed and typed analogue of [`crate::ws::v2::TransactionUpdate`].
 pub trait DbUpdate:
-    TryFrom<ws::v2::TransactionUpdate, Error = crate::Error> + Default + InModule + Send + 'static
+    TryFrom<ws::v2::TransactionUpdate, Error = crate::Error> + Default + Debug + InModule + Send + 'static
 where
     Self::Module: SpacetimeModule<DbUpdate = Self>,
 {
@@ -217,11 +217,13 @@ where
     fn unsubscribe(self) -> crate::Result<()>;
 }
 
+#[derive(Debug)]
 pub struct WithBsatn<Row> {
     pub bsatn: Bytes,
     pub row: Row,
 }
 
+#[derive(Debug)]
 pub struct TableUpdate<Row> {
     pub inserts: Vec<WithBsatn<Row>>,
     pub deletes: Vec<WithBsatn<Row>>,