Update Rust client SDK for V2 WebSocket format (#4257)

# Description of Changes

Update the Rust client SDK to use the new V2 WebSocket format, and
present the V2 user-facing API.

## Reducer events

### Remove on-reducer callbacks

It's no longer possible to observe reducers called by other clients by
registering callbacks with `ctx.reducers.on_{my_reducer}`. We no longer
code-generate those methods, or the associated
`ctx.reducers.remove_on_{my_reducer}`. Internal plumbing for storing and
invoking those callbacks is also removed.

### Add specific reducer invocation callbacks

In addition to the previous way to invoke reducers,
`ctx.reducers.{my_reducer}(args...)`, we add a method that registers a
callback to run after the reducer is finished. This method has the
suffix `_then`, as in `ctx.reducers.{my_reducer}_then(args...,
callback)`.

The callback will accept two arguments:
- `ctx: &ReducerEventContext`, the same context as was previously passed
to on-reducer callbacks.
- `status: Result<Result<(), String>, InternalError>`, denoting the
outcome of the reducer.
- `Ok(Ok(())` means the reducer committed. This corresponds to
`ReducerOutcome::Ok` or `ReducerOutcome::Okmpty` in the new WS format.
- `Ok(Err(message))` means the reducer returned an "expected" or "user"
error. This corresponds to `ReducerOutcome::Err` in the new WS format.
- `Err(internal_error)` means something went wrong with host execution.
This corresponds to `ReducerOutcome::InternalError` in the new WS
format.

Internally, the SDK stores the callbacks in its `ReducerCallbacks` map.
This is keyed on `request_id: u32`, a number that is generated for each
reducer call (from an `AtomicU32` that we increment each time), and
included in the `ClientMessage::CallReducer` request. The
`ServerMessage::ReducerResult` includes the same `request_id`, so the
SDK pops out of the `ReducerCallbacks` and invokes the appropriate
callback when processing that message.

These new callbacks are very similar to the existing procedure
callbacks.

### The `Event` exposed to row callbacks

Row callbacks caused by a reducer invoked by this client will see
`Event::Reducer`, the same as they would prior to this PR. These
callbacks will be the result of a `ServerMessage::ReducerResult` with
`ReducerOutcome::Ok`. In order to expose the reducer name and arguments
to this event, the client stores them in its `ReducerCallbacks` map,
alongside the callback for when the reducer is complete.

Row callbacks caused by any other reducer, or any non-reducer
transaction, are now indistinguishable to the client. These will see
`Event::Transaction`, which is renamed from the old
`Event::UnknownTransaction`.

### Less metadata in `ReducerEvent`

Some metadata is removed from `ReducerEvent`, as the V2 WebSocket format
no longer publishes it, even to the caller.

## `CallReducerFlags` are removed

All machinery for setting, storing and applying call reducer flags is
removed from the SDK, as the new WS format does not have any non-default
flags.

## Requesting rows in unsubscribe

When sending a `ClientMessage::Unsubscribe`, we always request that the
server include the matching rows in its response
`ServerMessage::UnsubscribeApplied`. This saves us having to update the
SDK to store query sets separately, at least for now. (We'll do that
later.)

## Handling rows

The new SDK does some additional parsing to wrangle rows in the new
WebSocket format into the same internal data structures as before,
rather than re-writing the client cache. (We'll do that later.)
Specifically, parsing of `DbUpdate` is changed so that:

- We parse raw `TransactionUpdate` into the generated `DbUpdate` type,
which requires an additional loop compared to the previous version, to
cope with the new WS format's dividing updates by query set. We define a
function `transaction_update_iter_table_updates` which encapsulates this
nested loop in an iterator.
- We have two new functions for parsing raw `QueryRows` into the
generated `DbUpdate` type, one for when they come from a
`SubscribeApplied`, and the other when they come from an
`UnsubscribeApplied`. `QueryRows` from `SubscribeApplied` translate to a
`DbUpdate` of all inserts, while one from `UnsubscribeApplied` will be
all deletes.

## Legacy subscriptions

"Legacy subscriptions" are removed. These were only used for
`subscribe_to_all_tables`, which as of now is stubbed. I will follow up
with a change to re-implement `subscribe_to_all_tables` by
code-generating a list of all known tables, and having it subscribe to
`select * from {table}` for every table in that list.

## `subscribe_to_all_tables` via a list

Previously, `subscribe_to_all_tables` worked by sending a legacy
subscription with the query `SELECT * FROM *`, which the host had
special handling to expand to subscribing to all tables. As legacy
subscriptions are no longer usable in V2 clients, this can't work.
Instead, we code-generate `SpacetimeModule::ALL_TABLE_NAMES`, a list of
all the known table names. `subscribe_to_all_tables` then maps across
this list to construct a list of queries in the form `SELECT * FROM
{table_name}`, and subscribes to all of those queries. This has the
upside that defining a new table in the module without regenerating
client bindings will no longer result in the client seeing rows of
tables it does not know about and cannot parse.

## Light mode removed

Light mode is no longer meaningful in the V2 WS format, so all code
related to it is removed.

## Internal changes

### Renamed WS messages

The SDK's internal code is updated to account for various renames:

- `QueryId` -> `QuerySetId`, `query_id` -> `query_set_id`.
- `SubscribeMulti` -> `Subscribe`, `UnsubscribeMulti` -> `Unsubscribe`.

## Incidental changes in this PR, not necessary for other client SDKs

### Don't filter out empty ranges in `RowSizeHint`

The Rust implementation of `RowSizeHint` in `BsatnRowList` got regressed
in the base branch to not work with zero-sized rows. This change fixes
that.

# API and ABI breaking changes

Boy howdy is it!

# Expected complexity level and risk

3? Changes ended up being less complicated than I feared, but we do have
some fiddly code here, and we have internal dependencies on the SDK.

# Testing

<!-- Describe any testing you've done, and any testing you'd like your
reviewers to do,
so that you're confident that all the changes work as expected! -->

- [x] Updated automated test suite.
  - Known failures:
- [ ] `subscribe_all_select_star`, which is currently broken because
it's trying to subscribe to rows from private tables. #4241 will fix
this.

---------

Co-authored-by: Jeffrey Dallatezza <jeffreydallatezza@gmail.com>
Co-authored-by: = <cloutiertyler@gmail.com>

Author: 3 people

crates/codegen/src/rust.rs

@@ -82,6 +82,9 @@ pub struct Test {
     /// not any of the aliases the SpacetimeDB CLI's `generate` command would accept.
     generate_language: String,
 
+    /// If true, pass `--include-private` to `spacetime generate` to include bindings for private items.
+    generate_include_private: bool,
+
     /// A relative path within the `client_project` to place the module bindings.
     ///
     /// Usually `src/module_bindings`.
@@ -129,6 +132,7 @@ impl Test {
             host_type,
             &self.client_project,
             &self.generate_subdir,
+            self.generate_include_private,
         );
 
         compile_client(&self.compile_command, &self.client_project);
@@ -283,6 +287,7 @@ fn generate_bindings(
     host_type: HostType,
     client_project: &str,
     generate_subdir: &str,
+    generate_include_private: bool,
 ) {
     // We need these to be owned `String`s so we can memoize on them.
     let client_project = client_project.to_owned();
@@ -304,6 +309,10 @@ fn generate_bindings(
             wasm_file,
         ];
 
+        if generate_include_private {
+            args.push("--include-private");
+        }
+
         let generate_dir: String;
 
         // `generate --lang unrealcpp` takes different arguments from non-Unreal languages
@@ -401,6 +410,7 @@ pub struct TestBuilder {
     module_name: Option<String>,
     client_project: Option<String>,
     generate_language: Option<String>,
+    generate_include_private: bool,
     generate_subdir: Option<String>,
     compile_command: Option<String>,
     run_command: Option<String>,
@@ -464,6 +474,13 @@ impl TestBuilder {
         }
     }
 
+    pub fn with_generate_private_items(self, include_private: bool) -> Self {
+        TestBuilder {
+            generate_include_private: include_private,
+            ..self
+        }
+    }
+
     pub fn build(self) -> Test {
         let generate_language = self
             .generate_language
@@ -487,6 +504,7 @@ impl TestBuilder {
                 .client_project
                 .expect("Supply a client project directory using TestBuilder::with_client"),
             generate_language,
+            generate_include_private: self.generate_include_private,
             generate_subdir,
             compile_command: self
                 .compile_command

crates/codegen/tests/snapshots/codegen__codegen_rust.snap

@@ -2406,8 +2406,6 @@ We need to handle several sorts of events:
 1. When a new user joins, we'll print a message introducing them.
 2. When a user is updated, we'll print their new name, or declare their new online status.
 3. When we receive a new message, we'll print it.
-4. If the server rejects our attempt to set our name, we'll print an error.
-5. If the server rejects a message we send, we'll print an error.
 
 To `src/main.rs`, add:
 
@@ -2422,12 +2420,6 @@ fn register_callbacks(ctx: &DbConnection) {
 
     // When a new message is received, print it.
     ctx.db.message().on_insert(on_message_inserted);
-
-    // When we fail to set our name, print a warning.
-    ctx.reducers.on_set_name(on_name_set);
-
-    // When we fail to send a message, print a warning.
-    ctx.reducers.on_send_message(on_message_sent);
 }
 ```
 
@@ -2532,39 +2524,6 @@ fn print_message(ctx: &impl RemoteDbContext, message: &Message) {
 }
 ```
 
-#### Handle reducer failures
-
-We can also register callbacks to run each time a reducer is invoked. We register these callbacks using the `on_reducer` method of the `Reducer` trait, which is automatically implemented for each reducer by `spacetime generate`.
-
-Each reducer callback first takes a `&ReducerEventContext` which contains metadata about the reducer call, including the identity of the caller and whether or not the reducer call suceeded.
-
-These callbacks will be invoked in one of two cases:
-
-1. If the reducer was successful and altered any of our subscribed rows.
-2. If we requested an invocation which failed.
-
-Note that a status of `Failed` or `OutOfEnergy` implies that the caller identity is our own identity.
-
-We already handle successful `set_name` invocations using our `ctx.db.user().on_update(..)` callback, but if the module rejects a user's chosen name, we'd like that user's client to let them know. We define a function `on_set_name` as a `conn.reducers.on_set_name(..)` callback which checks if the reducer failed, and if it did, prints a message including the rejected name and the error.
-
-To `src/main.rs`, add:
-
-```rust
-/// Our `on_set_name` callback: print a warning if the reducer failed.
-fn on_name_set(ctx: &ReducerEventContext, name: &String) {
-    if let Status::Failed(err) = &ctx.event.status {
-        eprintln!("Failed to change name to {:?}: {}", name, err);
-    }
-}
-
-/// Our `on_send_message` callback: print a warning if the reducer failed.
-fn on_message_sent(ctx: &ReducerEventContext, text: &String) {
-    if let Status::Failed(err) = &ctx.event.status {
-        eprintln!("Failed to send message {:?}: {}", text, err);
-    }
-}
-```
-
 ### Subscribe to queries
 
 SpacetimeDB is set up so that each client subscribes via SQL queries to some subset of the database, and is notified about changes only to that subset. For complex apps with large databases, judicious subscriptions can save each client significant network bandwidth, memory and computation. For example, in [BitCraft](https://bitcraftonline.com), each player's client subscribes only to the entities in the "chunk" of the world where that player currently resides, rather than the entire game world. Our app is much simpler than BitCraft, so we'll just subscribe to the whole database.
@@ -2628,6 +2587,8 @@ Our app should allow the user to interact by typing lines into their terminal. I
 
 For each reducer defined by our module, `ctx.reducers` has a method to request an invocation. In our case, we pass `set_name` and `send_message` a `String`, which gets sent to the server to execute the corresponding reducer.
 
+When we invoke either of these reducers, we'll register a callback to run when the client hears back about the result. If the database rejects our input, we'll print a message to the user. If there's a more serious error, we'll `panic!`.
+
 To `src/main.rs`, add:
 
 ```rust
@@ -2638,9 +2599,26 @@ fn user_input_loop(ctx: &DbConnection) {
             panic!("Failed to read from stdin.");
         };
         if let Some(name) = line.strip_prefix("/name ") {
-            ctx.reducers.set_name(name.to_string()).unwrap();
+            ctx.reducers
+                .set_name_then(name.to_string(), {
+                    let name = name.to_string();
+                    move |_ctx, result| match result {
+                        Err(e) => panic!("Internal error when setting name: {e}"),
+                        Ok(Err(e)) => eprintln!("Failed to set name to {name}: {e}"),
+                        Ok(Ok(())) => (),
+                    }
+                })
+                .unwrap();
         } else {
-            ctx.reducers.send_message(line).unwrap();
+            ctx.reducers
+                .send_message_then(line.clone(), {
+                    move |_ctx, result| match result {
+                        Err(e) => panic!("Internal error when sending message: {e}"),
+                        Ok(Err(e)) => eprintln!("Failed to send message {line:?}: {e}"),
+                        Ok(Ok(())) => (),
+                    }
+                })
+                .unwrap();
         }
     }
 }

crates/testing/src/sdk.rs

@@ -15,11 +15,8 @@
 //!
 //! This module is internal, and may incompatibly change without warning.
 
-use crate::{
-    client_cache::TableAppliedDiff,
-    error::InternalError,
-    spacetime_module::{AbstractEventContext, Reducer, SpacetimeModule},
-};
+use crate::{client_cache::TableAppliedDiff, error::InternalError, spacetime_module::SpacetimeModule};
+use bytes::Bytes;
 use spacetimedb_data_structures::map::HashMap;
 use std::{
     any::Any,
@@ -208,17 +205,19 @@ impl<M: SpacetimeModule> TableCallbacks<M> {
 /// Reducer arguments are passed to callbacks within the `EventContext`,
 /// and a wrapper inserted by the SDK will destructure the contained `Event`
 /// before invoking the user-supplied function.
-pub(crate) type ReducerCallback<M> = Box<dyn FnMut(&<M as SpacetimeModule>::ReducerEventContext) + Send + 'static>;
-
-type ReducerCallbackMap<M> = HashMap<CallbackId, ReducerCallback<M>>;
+pub(crate) type ReducerCallback<M> = Box<
+    dyn FnOnce(&<M as SpacetimeModule>::ReducerEventContext, Result<Result<(), String>, InternalError>)
+        + Send
+        + 'static,
+>;
 
 /// A collection of reducer callbacks.
 ///
 /// References to this struct are autogenerated in the `handle_event`
 /// function. Users should not reference this struct directly.
 pub(crate) struct ReducerCallbacks<M: SpacetimeModule> {
-    /// Maps reducer name to a set of callbacks.
-    callbacks: HashMap<&'static str, ReducerCallbackMap<M>>,
+    /// Maps WebSocket `request_id` to the arguments of that call and user-supplied callback.
+    callbacks: HashMap<u32, (M::Reducer, ReducerCallback<M>)>,
 }
 
 impl<M: SpacetimeModule> Default for ReducerCallbacks<M> {
@@ -230,45 +229,22 @@ impl<M: SpacetimeModule> Default for ReducerCallbacks<M> {
 }
 
 impl<M: SpacetimeModule> ReducerCallbacks<M> {
-    pub(crate) fn invoke_on_reducer(&mut self, ctx: &M::ReducerEventContext) {
-        let reducer = ctx.event();
-        let name = reducer.reducer.reducer_name();
-        if let Some(callbacks) = self.callbacks.get_mut(name) {
-            for callback in callbacks.values_mut() {
-                callback(ctx);
-            }
-        }
-    }
-
-    pub(crate) fn register_on_reducer(
-        &mut self,
-        reducer: &'static str,
-        callback_id: CallbackId,
-        callback: ReducerCallback<M>,
-    ) {
-        self.callbacks.entry(reducer).or_default().insert(callback_id, callback);
+    pub(crate) fn pop_call_info(&mut self, request_id: u32) -> Option<(M::Reducer, ReducerCallback<M>)> {
+        self.callbacks.remove(&request_id)
     }
 
-    pub(crate) fn remove_on_reducer(&mut self, reducer: &'static str, callback_id: CallbackId) {
-        // Ugly: `impl FnMut` is `must_use`.
-        // If we don't `.expect` this, no diagnostic,
-        // but we want to assert that we actually removed a callback,
-        // we just don't want to invoke it.
-        // So we have to `let _ =`.
-        let _ = self
-            .callbacks
-            .get_mut(reducer)
-            .expect("Attempt to remove a callback from a reducer which doesn't have any")
-            .remove(&callback_id)
-            .expect("Attempt to remove non-existent reducer callback");
+    pub(crate) fn store_call_info(&mut self, request_id: u32, args: M::Reducer, callback: ReducerCallback<M>) {
+        if self.callbacks.insert(request_id, (args, callback)).is_some() {
+            panic!("Re-used `request_id` {request_id} for multiple in-flight reducer requests.");
+        }
     }
 }
 
 /// A procedure callback for a procedure defined by the module `M`.
 ///
 /// Procedure return values are deserialized within this function by code injected by the SDK.
 pub(crate) type ProcedureCallback<M> =
-    Box<dyn FnOnce(&<M as SpacetimeModule>::ProcedureEventContext, Result<Box<[u8]>, InternalError>) + Send + 'static>;
+    Box<dyn FnOnce(&<M as SpacetimeModule>::ProcedureEventContext, Result<Bytes, InternalError>) + Send + 'static>;
 
 pub struct ProcedureCallbacks<M: SpacetimeModule> {
     request_id_to_callback: HashMap<u32, ProcedureCallback<M>>,
@@ -293,7 +269,7 @@ impl<M: SpacetimeModule> ProcedureCallbacks<M> {
         &mut self,
         ctx: &<M as SpacetimeModule>::ProcedureEventContext,
         request_id: u32,
-        result: Result<Box<[u8]>, InternalError>,
+        result: Result<Bytes, InternalError>,
     ) {
         let callback = self
             .request_id_to_callback

docs/docs/00100-intro/00300-tutorials/00100-chat-app.md

@@ -1,7 +1,5 @@
 use crate::websocket::WsError;
-use spacetimedb_client_api_messages::websocket::common as ws_common;
-use spacetimedb_client_api_messages::websocket::v1 as ws_v1;
-use spacetimedb_sats::bsatn;
+use spacetimedb_client_api_messages::websocket as ws;
 use std::borrow::Cow;
 use std::io::{self, Read as _};
 use std::sync::Arc;
@@ -18,22 +16,6 @@ fn gzip_decompress(bytes: &[u8]) -> Result<Vec<u8>, io::Error> {
     Ok(decompressed)
 }
 
-pub(crate) fn maybe_decompress_cqu(
-    cqu: ws_v1::CompressableQueryUpdate<ws_v1::BsatnFormat>,
-) -> ws_v1::QueryUpdate<ws_v1::BsatnFormat> {
-    match cqu {
-        ws_v1::CompressableQueryUpdate::Uncompressed(qu) => qu,
-        ws_v1::CompressableQueryUpdate::Brotli(bytes) => {
-            let bytes = brotli_decompress(&bytes).unwrap();
-            bsatn::from_slice(&bytes).unwrap()
-        }
-        ws_v1::CompressableQueryUpdate::Gzip(bytes) => {
-            let bytes = gzip_decompress(&bytes).unwrap();
-            bsatn::from_slice(&bytes).unwrap()
-        }
-    }
-}
-
 /// Decompresses a `ServerMessage` encoded in BSATN into the raw BSATN
 /// for further deserialization.
 pub(crate) fn decompress_server_message(raw: &[u8]) -> Result<Cow<'_, [u8]>, WsError> {
@@ -45,11 +27,11 @@ pub(crate) fn decompress_server_message(raw: &[u8]) -> Result<Cow<'_, [u8]>, WsE
     };
     match raw {
         [] => Err(WsError::EmptyMessage),
-        [ws_common::SERVER_MSG_COMPRESSION_TAG_NONE, bytes @ ..] => Ok(Cow::Borrowed(bytes)),
-        [ws_common::SERVER_MSG_COMPRESSION_TAG_BROTLI, bytes @ ..] => brotli_decompress(bytes)
+        [ws::common::SERVER_MSG_COMPRESSION_TAG_NONE, bytes @ ..] => Ok(Cow::Borrowed(bytes)),
+        [ws::common::SERVER_MSG_COMPRESSION_TAG_BROTLI, bytes @ ..] => brotli_decompress(bytes)
             .map(Cow::Owned)
             .map_err(err_decompress("brotli")),
-        [ws_common::SERVER_MSG_COMPRESSION_TAG_GZIP, bytes @ ..] => {
+        [ws::common::SERVER_MSG_COMPRESSION_TAG_GZIP, bytes @ ..] => {
             gzip_decompress(bytes).map(Cow::Owned).map_err(err_decompress("gzip"))
         }
         [c, ..] => Err(WsError::UnknownCompressionScheme { scheme: *c }),