Implement server-side support for the v2 websocket protocol (#4213)

# Description of Changes

This adds the v2 websocket protocol and adds support on the server side.
For context on many of the changes/decisions, you can look at the
discussion on #4023.

To restate some of the key changes:
- The reducer event information is no longer sent with transaction
updates (because we don't want to broadcast reducer call information
anymore).
- If a client calls a reducer, they are sent a `ReducerResult` which
includes the outcome of the reducer call and and related row updates for
queries that the client is subscribed to.
- We no longer dedupe queries that appear in multiple query sets for the
same client. This is because we are moving toward per-query storage.
- Related to that, Unsubscribe requests have an option to send the
related rows. We need this for now, since clients don't have per-query
storage implemented yet.
 - We don't have the json format in v2.

Notes for reviewers:
- This moves around the messages in
`crates/client-api-messages/src/websocket` (into `common`, `v1`, and
`v2`), and this renaming of existing messages adds a lot of noise to the
PR.
- In many places, I chose to duplicate a lot of code to have a v1
version and a v2 version. I went with this to make it easier to remove
the v1 version in the future (hopefully we can just fully delete most of
the v1 functions).
- `module_subscription_manager.rs` has probably has the biggest changes,
since we now track queries by query_set_id, and we get to remove some
complexity of v1's FormatSwitch.

<!-- Please describe your change, mention any related tickets, and so on
here. -->

# API and ABI breaking changes

The v1 protocol still works, though we won't send the reducer event info
for v10 modules.

# Expected complexity level and risk

4. This touches a lot of places.

# Testing

Unit testing is pretty minimal for the new code paths. I've done some
manual e2e testing with the typescript quickstart, and this has been
tested with a different branch implementing the v2 rust client.

---------

Co-authored-by: Phoebe Goldman <phoebe@goldman-tribe.org>
Co-authored-by: Jeffrey Dallatezza <jeffreydallatezza@gmail.com>

Author: 3 people

.github/workflows/ci.yml

@@ -958,14 +958,15 @@ jobs:
               exit 1
           }
 
-      - name: Check client-api bindings are up to date
-        working-directory: sdks/csharp
-        run: |
-          bash tools~/gen-client-api.sh
-          "${GITHUB_WORKSPACE}"/tools/check-diff.sh src/SpacetimeDB/ClientApi || {
-              echo 'Error: Client API bindings are dirty. Please run `sdks/csharp/tools~/gen-client-api.sh`.'
-              exit 1
-          }
+      # TODO: Re-enable this once csharp is using the v2 ws api.
+      # - name: Check client-api bindings are up to date
+      #   working-directory: sdks/csharp
+      #   run: |
+      #     bash tools~/gen-client-api.sh
+      #     "${GITHUB_WORKSPACE}"/tools/check-diff.sh src/SpacetimeDB/ClientApi || {
+      #         echo 'Error: Client API bindings are dirty. Please run `sdks/csharp/tools~/gen-client-api.sh`.'
+      #         exit 1
+      #     }
 
       - name: Start SpacetimeDB
         run: |

Cargo.lock

@@ -38,6 +38,7 @@ bench = false
 
 [dependencies]
 spacetimedb-client-api = { path = "../client-api" }
+spacetimedb-client-api-messages = { path = "../client-api-messages" }
 spacetimedb-core = { path = "../core", features = ["test"] }
 spacetimedb-data-structures.workspace = true
 spacetimedb-datastore.workspace = true

crates/bench/Cargo.toml

@@ -1,18 +1,18 @@
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 use spacetimedb::client::consume_each_list::ConsumeEachBuffer;
+use spacetimedb::db::relational_db::RelationalDB;
 use spacetimedb::error::DBError;
 use spacetimedb::host::module_host::DatabaseTableUpdate;
 use spacetimedb::identity::AuthCtx;
-use spacetimedb::messages::websocket::BsatnFormat;
 use spacetimedb::sql::ast::SchemaViewer;
 use spacetimedb::subscription::query::compile_read_only_queryset;
 use spacetimedb::subscription::row_list_builder_pool::BsatnRowListBuilderPool;
 use spacetimedb::subscription::subscription::ExecutionSet;
 use spacetimedb::subscription::tx::DeltaTx;
 use spacetimedb::subscription::{collect_table_update, TableUpdateType};
-use spacetimedb::{db::relational_db::RelationalDB, messages::websocket::Compression};
 use spacetimedb_bench::database::BenchDatabase as _;
 use spacetimedb_bench::spacetime_raw::SpacetimeRaw;
+use spacetimedb_client_api_messages::websocket::v1::{BsatnFormat, Compression};
 use spacetimedb_datastore::execution_context::Workload;
 use spacetimedb_execution::pipelined::PipelinedProject;
 use spacetimedb_primitives::{col_list, TableId};

crates/bench/benches/subscription.rs

@@ -4,7 +4,7 @@ use futures::{Sink, SinkExt, TryStream, TryStreamExt};
 use http::header;
 use reqwest::Url;
 use serde_json::Value;
-use spacetimedb_client_api_messages::websocket::{self as ws, JsonFormat};
+use spacetimedb_client_api_messages::websocket::v1 as ws_v1;
 use spacetimedb_data_structures::map::HashMap;
 use spacetimedb_lib::db::raw_def::v9::RawModuleDefV9;
 use spacetimedb_lib::de::serde::{DeserializeWrapper, SeedWrapper};
@@ -71,16 +71,16 @@ pub fn cli() -> clap::Command {
         .arg(common_args::server().help("The nickname, host name or URL of the server hosting the database"))
 }
 
-fn parse_msg_json(msg: &WsMessage) -> Option<ws::ServerMessage<JsonFormat>> {
+fn parse_msg_json(msg: &WsMessage) -> Option<ws_v1::ServerMessage<ws_v1::JsonFormat>> {
     let WsMessage::Text(msg) = msg else { return None };
-    serde_json::from_str::<DeserializeWrapper<ws::ServerMessage<JsonFormat>>>(msg)
+    serde_json::from_str::<DeserializeWrapper<ws_v1::ServerMessage<ws_v1::JsonFormat>>>(msg)
         .inspect_err(|e| eprintln!("couldn't parse message from server: {e}"))
         .map(|wrapper| wrapper.0)
         .ok()
 }
 
 fn reformat_update<'a>(
-    msg: &'a ws::DatabaseUpdate<JsonFormat>,
+    msg: &'a ws_v1::DatabaseUpdate<ws_v1::JsonFormat>,
     schema: &RawModuleDefV9,
 ) -> anyhow::Result<HashMap<&'a str, SubscriptionTable>> {
     msg.tables
@@ -152,7 +152,7 @@ pub async fn exec(config: Config, args: &ArgMatches) -> Result<(), anyhow::Error
     let mut req = url.into_client_request()?;
     req.headers_mut().insert(
         header::SEC_WEBSOCKET_PROTOCOL,
-        http::HeaderValue::from_static(ws::TEXT_PROTOCOL),
+        http::HeaderValue::from_static(ws_v1::TEXT_PROTOCOL),
     );
     //  Add the authorization header, if any.
     if let Some(auth_header) = api.con.auth_header.to_header() {
@@ -241,8 +241,8 @@ async fn subscribe<S>(ws: &mut S, query_strings: Box<[Box<str>]>) -> Result<(),
 where
     S: Sink<WsMessage, Error = WsError> + Unpin,
 {
-    let msg = serde_json::to_string(&SerializeWrapper::new(ws::ClientMessage::<()>::Subscribe(
-        ws::Subscribe {
+    let msg = serde_json::to_string(&SerializeWrapper::new(ws_v1::ClientMessage::<()>::Subscribe(
+        ws_v1::Subscribe {
             query_strings,
             request_id: 0,
         },
@@ -262,22 +262,22 @@ where
     while let Some(msg) = ws.try_next().await.map_err(|source| Error::Websocket { source })? {
         let Some(msg) = parse_msg_json(&msg) else { continue };
         match msg {
-            ws::ServerMessage::InitialSubscription(sub) => {
+            ws_v1::ServerMessage::InitialSubscription(sub) => {
                 if let Some(module_def) = module_def {
                     let output = format_output_json(&sub.database_update, module_def)?;
                     tokio::io::stdout().write_all(output.as_bytes()).await?
                 }
                 break;
             }
-            ws::ServerMessage::TransactionUpdate(ws::TransactionUpdate { status, .. }) => {
+            ws_v1::ServerMessage::TransactionUpdate(ws_v1::TransactionUpdate { status, .. }) => {
                 return Err(match status {
-                    ws::UpdateStatus::Failed(msg) => Error::TransactionFailure { reason: msg },
+                    ws_v1::UpdateStatus::Failed(msg) => Error::TransactionFailure { reason: msg },
                     _ => Error::Protocol {
                         details: RECV_TX_UPDATE,
                     },
                 })
             }
-            ws::ServerMessage::TransactionUpdateLight(ws::TransactionUpdateLight { .. }) => {
+            ws_v1::ServerMessage::TransactionUpdateLight(ws_v1::TransactionUpdateLight { .. }) => {
                 return Err(Error::Protocol {
                     details: RECV_TX_UPDATE,
                 })
@@ -310,14 +310,14 @@ where
 
         let Some(msg) = parse_msg_json(&msg) else { continue };
         match msg {
-            ws::ServerMessage::InitialSubscription(_) => {
+            ws_v1::ServerMessage::InitialSubscription(_) => {
                 return Err(Error::Protocol {
                     details: "received a second initial subscription update",
                 })
             }
-            ws::ServerMessage::TransactionUpdateLight(ws::TransactionUpdateLight { update, .. })
-            | ws::ServerMessage::TransactionUpdate(ws::TransactionUpdate {
-                status: ws::UpdateStatus::Committed(update),
+            ws_v1::ServerMessage::TransactionUpdateLight(ws_v1::TransactionUpdateLight { update, .. })
+            | ws_v1::ServerMessage::TransactionUpdate(ws_v1::TransactionUpdate {
+                status: ws_v1::UpdateStatus::Committed(update),
                 ..
             }) => {
                 let output = format_output_json(&update, module_def)?;
@@ -329,7 +329,10 @@ where
     }
 }
 
-fn format_output_json(msg: &ws::DatabaseUpdate<JsonFormat>, schema: &RawModuleDefV9) -> Result<String, Error> {
+fn format_output_json(
+    msg: &ws_v1::DatabaseUpdate<ws_v1::JsonFormat>,
+    schema: &RawModuleDefV9,
+) -> Result<String, Error> {
     let formatted = reformat_update(msg, schema).map_err(|source| Error::Reformat { source })?;
     let output = serde_json::to_string(&formatted)? + "\n";
 

crates/cli/src/subcommands/subscribe.rs

@@ -1,5 +1,5 @@
 use bytes::Bytes;
-use spacetimedb_client_api_messages::websocket::{BsatnFormat, ClientMessage, ServerMessage};
+use spacetimedb_client_api_messages::websocket::v1::{BsatnFormat, ClientMessage, ServerMessage};
 use spacetimedb_lib::ser::serde::SerializeWrapper;
 use spacetimedb_lib::{RawModuleDef, RawModuleDefV8};