diff --git a/codex-plugin/plugins/spacetimedb/skills/csharp-client/SKILL.md b/codex-plugin/plugins/spacetimedb/skills/csharp-client/SKILL.md index 4c1283e815b..0c3564b7581 100644 --- a/codex-plugin/plugins/spacetimedb/skills/csharp-client/SKILL.md +++ b/codex-plugin/plugins/spacetimedb/skills/csharp-client/SKILL.md @@ -48,6 +48,18 @@ var conn = DbConnection.Builder() Compression options are `Compression.Brotli`, `Compression.Gzip`, and `Compression.None`. The SDK uses Brotli when `WithCompression` is omitted. +## Automatic Reconnect and Token Refresh + +Automatic reconnect is opt-in: add `.WithAutomaticReconnect()` to the builder. Initial connection failures do not retry. After an established connection is lost, the SDK retries with exponential backoff and jitter, capped at 30 seconds. `Disconnect()` permanently stops recovery. + +Keep calling `FrameTick()` while `IsActive` is false. `IsReconnecting` is true while recovering before the next handshake. Use `.OnDisconnect((conn, error, next) => ...)` and `.OnConnectError((error, next) => ...)` to inspect `NextReconnect?`: a non-null value provides `Attempt` and `Delay`; null means no retry is scheduled. Existing callback overloads still work. + +`OnConnect` runs on every successful reconnect, so create the subscription in the example above only on the first connection, or move it after `Build()`. Register row callbacks once as well. The same connection, identity, table handles, and subscriptions survive; each attempt gets a fresh `ConnectionId`. Cached rows stay readable but stale during outages. The SDK replays subscriptions in one batch and emits only net row changes. Subscription `OnApplied` runs again after replay; use it to mark data ready, not for repeated one-time setup. + +For expiring credentials, combine `.WithToken(initialToken)` with `.WithTokenProvider(() => RefreshTokenAsync())`, where your provider returns `Task` for the same identity. The provider is not called for the initial connection. Before retries, it is called when remaining validity is at most 30 seconds or 5% of the token lifetime, whichever is greater, when expiry cannot be read, or after a reused token is rejected. Provider failures retry; rejection of a freshly provided token is terminal. No periodic refresh runs while connected. Disconnecting ignores a pending provider result but does not cancel the provider's own work. + +Calls made while disconnected fail immediately. Pending reducer calls receive `Status.UnknownResult`; pending procedures and one-off queries fail with `UnknownResultException`. These calls may already have executed and are never replayed. Regenerate bindings so unhandled unknown reducer outcomes reach `OnUnhandledReducerError`. + ## Event Loop (Critical) **`FrameTick()` must be called in your main loop.** The SDK queues all network messages and only processes them when you call `FrameTick()`. Without it, no callbacks fire. @@ -153,6 +165,8 @@ conn.Reducers.OnSendMessage += (ReducerEventContext ctx, string text) => Console.WriteLine($"Message sent: {text}"); else if (ctx.Event.Status is Status.Failed(var reason)) Console.Error.WriteLine($"Send failed: {reason}"); + else if (ctx.Event.Status is Status.UnknownResult) + Console.Error.WriteLine("Connection lost before the result arrived; the message may have been sent."); }; ``` diff --git a/codex-plugin/plugins/spacetimedb/skills/unity/SKILL.md b/codex-plugin/plugins/spacetimedb/skills/unity/SKILL.md index 61d523f1e97..928712328e1 100644 --- a/codex-plugin/plugins/spacetimedb/skills/unity/SKILL.md +++ b/codex-plugin/plugins/spacetimedb/skills/unity/SKILL.md @@ -120,6 +120,16 @@ public class SpacetimeManager : MonoBehaviour **Thread safety**: `FrameTick()` processes messages on the calling thread (the main thread in Unity). Do NOT call it from a background thread. Do NOT access `conn.Db` from background threads. +## Automatic Reconnect + +Add `.WithAutomaticReconnect()` to the builder to recover an established connection after an outage. In the singleton above, create subscriptions only on the first `OnConnected`, or move them after `Build()`: `OnConnected` runs again after each successful reconnect. Keep calling `FrameTick()` during outages, or use `SpacetimeDBNetworkManager`, which also ticks reconnecting connections. Do not gate ticking on `IsActive`. + +The SDK retains the identity, table handles, callbacks, and subscriptions, and replays subscriptions in one batch. Cached rows stay readable during an outage; row callbacks report net changes after recovery. `OnApplied` runs again, so separate one-time object setup from marking data ready. Each reconnect attempt has a fresh `ConnectionId`. + +`IsReconnecting` reports recovery before the next successful handshake. The `OnDisconnect((conn, error, next) => ...)` and `OnConnectError((error, next) => ...)` overloads expose `NextReconnect?`, with the upcoming `Attempt` and `Delay`, or null for a terminal failure. Initial connection failures do not retry. Retries use exponential backoff and jitter capped at 30 seconds; `Disconnect()` stops them. + +Calls made while disconnected fail immediately. Pending reducer calls may report `Status.UnknownResult`, meaning the server may have executed them; they are not replayed. Pending procedures and one-off queries fail with `UnknownResultException`. Regenerate bindings when upgrading to include unknown-outcome handling. + --- ## Row Callbacks for Game State @@ -143,7 +153,7 @@ void RegisterCallbacks() } ``` -Register these in `OnSubscriptionApplied` (after initial data is loaded) or in `Start()` before connecting. +Register these once after building the connection, before the first `FrameTick()`. Do not register them repeatedly in `OnConnected` or `OnSubscriptionApplied` when automatic reconnect is enabled. --- @@ -172,6 +182,8 @@ SpacetimeManager.Instance.Connection.Reducers.OnSendMessage += (ReducerEventCont Debug.Log($"Message sent: {text}"); else if (ctx.Event.Status is Status.Failed(var reason)) Debug.LogError($"Send failed: {reason}"); + else if (ctx.Event.Status is Status.UnknownResult) + Debug.LogWarning("Connection lost before the result arrived; the message may have been sent."); }; ``` @@ -220,3 +232,6 @@ The SpacetimeDB SDK uses code generation. If you encounter issues with IL2CPP bu ### Token Persistence Token save/load via `PlayerPrefs` is demonstrated in the SpacetimeManager singleton above. Persisting the server-issued token and passing it back on reconnect keeps the same identity; without a saved token the server issues a new identity in the `OnConnect` callback. This token does not expire and a lost one can't be recovered, so self-issued identities are for development. For production, authenticate with an OIDC provider such as SpacetimeAuth, which handles token lifecycle. +With automatic reconnect enabled, the SDK retains the authentication token and passes it to `OnConnect`, including on WebGL. Without automatic reconnect, a WebGL connection using a saved token may return a short-lived WebSocket token instead; in that case, keep the original saved token rather than overwriting it in the callback above. + +For expiring credentials, configure `.WithToken(initialToken)`, `.WithAutomaticReconnect()`, and `.WithTokenProvider(() => RefreshTokenAsync())`. The provider returns `Task` for the same identity. It is used before reconnect attempts when expiry is unreadable, remaining validity is at most 30 seconds or 5% of the original lifetime, or a reused token is rejected. It is not called for the initial connection or periodically while connected. Provider failures retry; rejection of a freshly provided token stops recovery. `Disconnect()` ignores a pending provider result without canceling the provider's own asynchronous work. diff --git a/crates/codegen/src/csharp.rs b/crates/codegen/src/csharp.rs index edc372a14e0..2f9605826de 100644 --- a/crates/codegen/src/csharp.rs +++ b/crates/codegen/src/csharp.rs @@ -873,6 +873,7 @@ impl Lang for Csharp<'_> { indented_block(output, |output| { writeln!(output, "case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break;"); writeln!(output, "case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception(\"out of energy\")); break;"); + writeln!(output, "case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break;"); }); }); writeln!(output, "return false;"); diff --git a/crates/codegen/tests/snapshots/codegen__codegen_csharp.snap b/crates/codegen/tests/snapshots/codegen__codegen_csharp.snap index 21fe014ee8d..f036b5dd380 100644 --- a/crates/codegen/tests/snapshots/codegen__codegen_csharp.snap +++ b/crates/codegen/tests/snapshots/codegen__codegen_csharp.snap @@ -388,6 +388,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -464,6 +465,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -533,6 +535,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -602,6 +605,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -657,6 +661,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -725,6 +730,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -794,6 +800,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -870,6 +877,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -938,6 +946,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -993,6 +1002,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -1048,6 +1058,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -1103,6 +1114,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; @@ -1191,6 +1203,7 @@ namespace SpacetimeDB { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/demo/Blackholio/client-godot/module_bindings/Reducers/EnterGame.g.cs b/demo/Blackholio/client-godot/module_bindings/Reducers/EnterGame.g.cs index 136542a8e57..d039a05cd0f 100644 --- a/demo/Blackholio/client-godot/module_bindings/Reducers/EnterGame.g.cs +++ b/demo/Blackholio/client-godot/module_bindings/Reducers/EnterGame.g.cs @@ -30,6 +30,7 @@ public bool InvokeEnterGame(ReducerEventContext ctx, Reducer.EnterGame args) { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/demo/Blackholio/client-godot/module_bindings/Reducers/PlayerSplit.g.cs b/demo/Blackholio/client-godot/module_bindings/Reducers/PlayerSplit.g.cs index 85b0502b6d6..1755a18ee6f 100644 --- a/demo/Blackholio/client-godot/module_bindings/Reducers/PlayerSplit.g.cs +++ b/demo/Blackholio/client-godot/module_bindings/Reducers/PlayerSplit.g.cs @@ -30,6 +30,7 @@ public bool InvokePlayerSplit(ReducerEventContext ctx, Reducer.PlayerSplit args) { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/demo/Blackholio/client-godot/module_bindings/Reducers/Respawn.g.cs b/demo/Blackholio/client-godot/module_bindings/Reducers/Respawn.g.cs index 44273b7c198..dfb13b2eb09 100644 --- a/demo/Blackholio/client-godot/module_bindings/Reducers/Respawn.g.cs +++ b/demo/Blackholio/client-godot/module_bindings/Reducers/Respawn.g.cs @@ -30,6 +30,7 @@ public bool InvokeRespawn(ReducerEventContext ctx, Reducer.Respawn args) { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/demo/Blackholio/client-godot/module_bindings/Reducers/Suicide.g.cs b/demo/Blackholio/client-godot/module_bindings/Reducers/Suicide.g.cs index 428d87e5d34..0925aaae549 100644 --- a/demo/Blackholio/client-godot/module_bindings/Reducers/Suicide.g.cs +++ b/demo/Blackholio/client-godot/module_bindings/Reducers/Suicide.g.cs @@ -30,6 +30,7 @@ public bool InvokeSuicide(ReducerEventContext ctx, Reducer.Suicide args) { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/demo/Blackholio/client-godot/module_bindings/Reducers/UpdatePlayerInput.g.cs b/demo/Blackholio/client-godot/module_bindings/Reducers/UpdatePlayerInput.g.cs index 612397e3e9c..bb80bf8e949 100644 --- a/demo/Blackholio/client-godot/module_bindings/Reducers/UpdatePlayerInput.g.cs +++ b/demo/Blackholio/client-godot/module_bindings/Reducers/UpdatePlayerInput.g.cs @@ -30,6 +30,7 @@ public bool InvokeUpdatePlayerInput(ReducerEventContext ctx, Reducer.UpdatePlaye { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/demo/Blackholio/client-godot/module_bindings/SpacetimeDBClient.g.cs b/demo/Blackholio/client-godot/module_bindings/SpacetimeDBClient.g.cs index 44743e2dfab..f61670e7dac 100644 --- a/demo/Blackholio/client-godot/module_bindings/SpacetimeDBClient.g.cs +++ b/demo/Blackholio/client-godot/module_bindings/SpacetimeDBClient.g.cs @@ -1,7 +1,7 @@ // THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE // WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. -// This was generated using spacetimedb cli version 2.1.0 (commit 6981f48b4bc1a71c8dd9bdfe5a2c343f6370243d). +// This was generated using spacetimedb cli version 2.10.1 (commit 81ea9891ec5eca8312d652635899a604b082a7a4). #nullable enable diff --git a/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/EnterGame.g.cs b/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/EnterGame.g.cs index 136542a8e57..d039a05cd0f 100644 --- a/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/EnterGame.g.cs +++ b/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/EnterGame.g.cs @@ -30,6 +30,7 @@ public bool InvokeEnterGame(ReducerEventContext ctx, Reducer.EnterGame args) { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/PlayerSplit.g.cs b/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/PlayerSplit.g.cs index 85b0502b6d6..1755a18ee6f 100644 --- a/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/PlayerSplit.g.cs +++ b/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/PlayerSplit.g.cs @@ -30,6 +30,7 @@ public bool InvokePlayerSplit(ReducerEventContext ctx, Reducer.PlayerSplit args) { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/Respawn.g.cs b/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/Respawn.g.cs index 44273b7c198..dfb13b2eb09 100644 --- a/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/Respawn.g.cs +++ b/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/Respawn.g.cs @@ -30,6 +30,7 @@ public bool InvokeRespawn(ReducerEventContext ctx, Reducer.Respawn args) { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/Suicide.g.cs b/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/Suicide.g.cs index 428d87e5d34..0925aaae549 100644 --- a/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/Suicide.g.cs +++ b/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/Suicide.g.cs @@ -30,6 +30,7 @@ public bool InvokeSuicide(ReducerEventContext ctx, Reducer.Suicide args) { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/UpdatePlayerInput.g.cs b/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/UpdatePlayerInput.g.cs index 612397e3e9c..bb80bf8e949 100644 --- a/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/UpdatePlayerInput.g.cs +++ b/demo/Blackholio/client-unity/Assets/Scripts/autogen/Reducers/UpdatePlayerInput.g.cs @@ -30,6 +30,7 @@ public bool InvokeUpdatePlayerInput(ReducerEventContext ctx, Reducer.UpdatePlaye { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/demo/Blackholio/client-unity/Assets/Scripts/autogen/SpacetimeDBClient.g.cs b/demo/Blackholio/client-unity/Assets/Scripts/autogen/SpacetimeDBClient.g.cs index e7febb6fa13..f61670e7dac 100644 --- a/demo/Blackholio/client-unity/Assets/Scripts/autogen/SpacetimeDBClient.g.cs +++ b/demo/Blackholio/client-unity/Assets/Scripts/autogen/SpacetimeDBClient.g.cs @@ -1,7 +1,7 @@ // THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE // WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. -// This was generated using spacetimedb cli version 2.1.0 (commit 6cae7a4ca81a3c90d01d3f3303d46fa7bf7b3d41). +// This was generated using spacetimedb cli version 2.10.1 (commit 81ea9891ec5eca8312d652635899a604b082a7a4). #nullable enable diff --git a/docs/docs/00100-intro/00200-quickstarts/00600-c-sharp.md b/docs/docs/00100-intro/00200-quickstarts/00600-c-sharp.md index 5403013d6fc..b3952542c51 100644 --- a/docs/docs/00100-intro/00200-quickstarts/00600-c-sharp.md +++ b/docs/docs/00100-intro/00200-quickstarts/00600-c-sharp.md @@ -240,5 +240,6 @@ spacetime logs ## Next steps +- Enable [automatic reconnect](../../00200-core-concepts/00600-clients/00600-csharp-reference.md#method-withautomaticreconnect) with `.WithAutomaticReconnect()` on your client's connection builder. Keep calling `FrameTick()` during outages and register subscriptions only once; the SDK replays them after recovery. For expiring authentication tokens, also use [`.WithTokenProvider(() => RefreshTokenAsync())`](../../00200-core-concepts/00600-clients/00600-csharp-reference.md#method-withtokenprovider). - See the [Chat App Tutorial](../00300-tutorials/00100-chat-app.md) for a complete example - Read the [C# SDK Reference](../../00200-core-concepts/00600-clients/00600-csharp-reference.md) for detailed API docs 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 c58dd6784b2..33564db0701 100644 --- a/docs/docs/00100-intro/00300-tutorials/00100-chat-app.md +++ b/docs/docs/00100-intro/00300-tutorials/00100-chat-app.md @@ -1856,6 +1856,12 @@ DbConnection ConnectToDB() } ``` +#### Automatic reconnect + +The C# connection above uses the default behavior and will not reconnect after an outage. Add `.WithAutomaticReconnect()` to the builder to recover an established connection automatically. Keep calling `FrameTick()` while disconnected, register row callbacks and subscriptions only once, and use the `OnDisconnect((conn, error, next) => ...)` and `OnConnectError((error, next) => ...)` overloads to inspect the next retry. `OnConnect` and subscription `OnApplied` run again after recovery; guard any one-time setup or commands in those callbacks. Initial connection failures do not retry, and `Disconnect()` stops recovery. + +For expiring credentials, combine `WithToken(initialToken)` with `.WithTokenProvider(() => RefreshTokenAsync())`. The provider is called before reconnect attempts when the retained token needs refreshing and must return a token for the same identity. Calls made during an outage fail immediately; pending reducer calls may report `Status.UnknownResult` and are not replayed. See the [C# reconnect reference](/clients/c-sharp#method-withautomaticreconnect) for details. + #### Save credentials SpacetimeDB will accept any [OpenID Connect](https://openid.net/developers/how-connect-works/) compliant [JSON Web Token](https://jwt.io/) and use it to compute an `Identity` for the user. More complex applications will generally authenticate their user somehow, generate or retrieve a token, and attach it to their connection via `WithToken`. In our case, though, we'll connect anonymously the first time, let SpacetimeDB generate a fresh `Identity` and corresponding JWT for us, and save that token locally to re-use the next time we connect. diff --git a/docs/docs/00100-intro/00300-tutorials/00300-unity-tutorial/00300-part-2.md b/docs/docs/00100-intro/00300-tutorials/00300-unity-tutorial/00300-part-2.md index 3d9f4347e5e..1ee9502aff8 100644 --- a/docs/docs/00100-intro/00300-tutorials/00300-unity-tutorial/00300-part-2.md +++ b/docs/docs/00100-intro/00300-tutorials/00300-unity-tutorial/00300-part-2.md @@ -801,7 +801,9 @@ public class GameManager : MonoBehaviour } ``` -> Unity WebGL needs one extra precaution here. Browser WebSocket APIs cannot set an `Authorization` header, so reconnecting with a saved server-issued token may yield a short-lived WebSocket token in `HandleConnect`. The `#if UNITY_WEBGL` guard keeps the original saved token instead of overwriting it during reconnect. +> This example uses the default connection behavior, without automatic reconnect. On Unity WebGL, a new connection using a saved server-issued token may yield a short-lived WebSocket token in `HandleConnect`. The `#if UNITY_WEBGL` guard keeps the original saved token instead of overwriting it. With `.WithAutomaticReconnect()` enabled, `HandleConnect` receives the retained or refreshed authentication token, which can be saved directly. + +To add [automatic reconnect](/clients/c-sharp#method-withautomaticreconnect), call `.WithAutomaticReconnect()` on the builder and create subscriptions only on the first connection: `HandleConnect` runs again after recovery, and the SDK replays existing subscriptions. Keep `SpacetimeDBNetworkManager` active, or call `FrameTick()` every frame even during an outage. The callback overloads with `NextReconnect?` distinguish scheduled retries from a terminal disconnect. For expiring credentials, also configure [`.WithTokenProvider(() => RefreshTokenAsync())`](/clients/c-sharp#method-withtokenprovider). Subscription `OnApplied` runs after every successful replay, so separate initial game setup from data-readiness handling. Here we configure the connection to the database, by passing it some callbacks in addition to providing the `SERVER_URL` and `DATABASE_NAME` to the connection. When the client connects, the SpacetimeDB SDK will call the `HandleConnect` method, allowing us to start up the game. diff --git a/docs/docs/00100-intro/00300-tutorials/00300-unity-tutorial/00400-part-3.md b/docs/docs/00100-intro/00300-tutorials/00300-unity-tutorial/00400-part-3.md index 612042afcb3..27621bfb33f 100644 --- a/docs/docs/00100-intro/00300-tutorials/00300-unity-tutorial/00400-part-3.md +++ b/docs/docs/00100-intro/00300-tutorials/00300-unity-tutorial/00400-part-3.md @@ -1384,7 +1384,9 @@ void HandleConnect(DbConnection conn, Identity identity, string token) } ``` -Keep the same WebGL guard from Part 2 here as well. On Unity WebGL, a reconnect can surface a short-lived WebSocket token in `HandleConnect`, so you should not overwrite an already-saved long-lived server-issued token. +Keep the same WebGL guard from Part 2 while using the default connection behavior. A new connection using a saved token can return a short-lived WebSocket token, which should not overwrite the saved authentication token. With [automatic reconnect](/clients/c-sharp#method-withautomaticreconnect) enabled, `HandleConnect` receives the retained or refreshed authentication token instead. + +If you enable automatic reconnect, run the row-callback registrations, initial `OnConnected` setup, and subscription creation above only once. `HandleConnect` runs again on each successful reconnect, while existing callbacks and subscriptions survive. The SDK updates the cache and emits net row changes after replay; use subscription `OnApplied` to mark the data ready again. Next add the following implementations for those callbacks to the `GameManager` class. diff --git a/docs/docs/00100-intro/00300-tutorials/00500-godot-tutorial/00300-part-2.md b/docs/docs/00100-intro/00300-tutorials/00500-godot-tutorial/00300-part-2.md index 845219c5e26..cce24b64201 100644 --- a/docs/docs/00100-intro/00300-tutorials/00500-godot-tutorial/00300-part-2.md +++ b/docs/docs/00100-intro/00300-tutorials/00500-godot-tutorial/00300-part-2.md @@ -794,6 +794,8 @@ Here we configure the connection to the database, by passing it some callbacks i When the client connects, the SpacetimeDB SDK will call the `HandleConnect` method, allowing us to start up the game. +This example uses the default connection behavior. To enable [automatic reconnect](/clients/c-sharp#method-withautomaticreconnect), add `.WithAutomaticReconnect()` to the builder. Keep the connection registered with `STDBUpdateManager` during outages so `FrameTick()` continues driving retries. Create subscriptions only on the first connection: `HandleConnect` runs again after recovery, while the SDK retains and replays existing subscriptions. Use the callback overloads with `NextReconnect?` to distinguish retries from a terminal disconnect, and keep one-time game setup separate from subscription `OnApplied`, which runs after every successful replay. For expiring credentials, supply the initial token with `WithToken` and add [`.WithTokenProvider(() => RefreshTokenAsync())`](/clients/c-sharp#method-withtokenprovider). + In our `HandleConnect` callback we build a subscription and call `Subscribe`, subscribing to all data in the database. This causes SpacetimeDB to synchronize the state of all your tables with your Godot client's SDK client cache. --- diff --git a/docs/docs/00100-intro/00300-tutorials/00500-godot-tutorial/00400-part-3.md b/docs/docs/00100-intro/00300-tutorials/00500-godot-tutorial/00400-part-3.md index 46291fe7202..f706c9eac6a 100644 --- a/docs/docs/00100-intro/00300-tutorials/00500-godot-tutorial/00400-part-3.md +++ b/docs/docs/00100-intro/00300-tutorials/00500-godot-tutorial/00400-part-3.md @@ -1511,6 +1511,8 @@ private void HandleConnect(DbConnection conn, Identity identity, string token) } ``` +If you enable [automatic reconnect](/clients/c-sharp#method-withautomaticreconnect), create the `Instantiator`, run initial `OnConnected` setup, and create subscriptions only once. `HandleConnect` runs again on each successful reconnect. Recreating these objects would duplicate row handlers and subscriptions; the existing `Instantiator` can process the net cache changes from subscription replay. Use subscription `OnApplied` to mark the data ready after each recovery. + ### Camera Controller One of the last steps is to create a camera controller to make sure the camera follows the local player around. Create a new script called `CameraController.cs`. Replace the contents of the file with this: diff --git a/docs/docs/00200-core-concepts/00500-authentication.md b/docs/docs/00200-core-concepts/00500-authentication.md index bbbdba76be3..2dbe206920f 100644 --- a/docs/docs/00200-core-concepts/00500-authentication.md +++ b/docs/docs/00200-core-concepts/00500-authentication.md @@ -33,6 +33,20 @@ This matters when you persist tokens on the client: - Expect this distinction on browser-style transports where WebSocket headers are unavailable, such as Unity WebGL builds. +For C# clients with `.WithAutomaticReconnect()` enabled, the SDK retains the +authentication token and reuses it during recovery. `OnConnect` receives that +retained or refreshed token, including on Unity WebGL, so it can be persisted +for future application sessions without saving the transport's short-lived token. +The WebGL precaution above still applies to C# connections without automatic reconnect. + +For expiring credentials, also configure `.WithTokenProvider(() => RefreshTokenAsync())`. +Supply the initial token through `.WithToken(initialToken)`; the provider is only +used during automatic reconnect. It returns a token for the same identity when +the retained token is near expiry, its expiry cannot be read, or a reused token +has been rejected. The SDK does not refresh tokens periodically while connected. +See the [C# token-provider reference](./00600-clients/00600-csharp-reference.md#method-withtokenprovider) +for refresh timing, rejection handling, and cancellation behavior. + ## SpacetimeAuth To make it easier to get started with authentication, SpacetimeDB offers 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 009053f2aaf..bc28af5aa29 100644 --- a/docs/docs/00200-core-concepts/00600-clients/00300-connection.md +++ b/docs/docs/00200-core-concepts/00600-clients/00300-connection.md @@ -399,7 +399,13 @@ Conn->Disconnect(); :::note[Reconnection behavior] -Lower-level `DbConnection` objects do not reconnect themselves. If you create a `DbConnection` directly and the connection is interrupted, create a new `DbConnection` to re-establish connectivity. We recommend implementing reconnection logic in your application if reliable connectivity is critical. +C# connections support opt-in automatic reconnect through `.WithAutomaticReconnect()` on the builder. After an established connection is interrupted, the SDK retries with exponential backoff and jitter, up to a 30-second delay. It retains the connection object, identity, cached rows, callbacks, and subscription handles, then replays subscriptions in one batch. `OnConnect` fires again and subscription `OnApplied` callbacks report when the cache has been reconciled. Register subscriptions and row callbacks only once. + +Keep calling `FrameTick()` during outages, including while `IsActive` is false. `IsReconnecting` reports recovery before the next successful handshake. Unity's `SpacetimeDBNetworkManager` continues ticking reconnecting connections. The `OnDisconnect((conn, error, next) => ...)` and `OnConnectError((error, next) => ...)` overloads provide a nullable `NextReconnect` with the upcoming `Attempt` and `Delay`; `null` means no retry is scheduled. Initial connection failures do not retry, and `Disconnect()` permanently stops recovery. + +For expiring C# credentials, combine `.WithToken(initialToken)` with `.WithTokenProvider(() => RefreshTokenAsync())`. The provider is used before reconnect attempts when the retained token is near expiry or cannot be read, and after a reused token is rejected. It must return a token for the same identity. See the [C# reconnect and token-provider reference](./00600-csharp-reference.md#method-withautomaticreconnect) for the full lifecycle, terminal errors, and pending-call behavior. + +Without `.WithAutomaticReconnect()`, C# applications must create a new `DbConnection` after a lost connection. Direct TypeScript, Rust, and Unreal connections also require application-managed reconnection. The TypeScript React, Solid, and Svelte providers manage their connections through the SDK's shared connection manager. While a provider is mounted, that manager automatically rebuilds unexpectedly closed connections with exponential backoff and re-checks connection liveness when the page becomes visible, regains focus, returns online, or is restored from the back-forward cache. diff --git a/docs/docs/00200-core-concepts/00600-clients/00600-csharp-reference.md b/docs/docs/00200-core-concepts/00600-clients/00600-csharp-reference.md index d49069285e6..20d4d950961 100644 --- a/docs/docs/00200-core-concepts/00600-clients/00600-csharp-reference.md +++ b/docs/docs/00200-core-concepts/00600-clients/00600-csharp-reference.md @@ -111,6 +111,8 @@ Construct a `DbConnection` by calling `DbConnection.Builder()`, chaining configu | [OnConnectError callback](#callback-onconnecterror) | Register a callback to run if the connection is rejected or the host is unreachable. | | [OnDisconnect callback](#callback-ondisconnect) | Register a callback to run when the connection ends. | | [WithToken method](#method-withtoken) | Supply a token to authenticate with the remote database. | +| [WithAutomaticReconnect method](#method-withautomaticreconnect) | Reconnect automatically after an established connection is lost. | +| [WithTokenProvider method](#method-withtokenprovider) | Obtain refreshed credentials before a reconnect attempt. | | [Build method](#method-build) | Finalize configuration and open the connection. | #### Method `WithUri` @@ -174,28 +176,40 @@ class DbConnectionBuilder Chain a call to `.OnConnect(callback)` to your builder to register a callback to run when your new `DbConnection` successfully initiates its connection to the remote database. The callback accepts three arguments: a reference to the `DbConnection`, the `Identity` by which SpacetimeDB identifies this connection, and a private access token which can be saved and later passed to [`WithToken`](#method-withtoken) to authenticate the same user in future connections. +With automatic reconnect enabled, this callback also runs after each successful reconnect, before subscriptions are replayed. Register row callbacks and subscriptions only once. Use subscription `OnApplied` callbacks to know when the cache is ready. The token argument is the retained or refreshed authentication token, including on Unity WebGL, rather than the transport's short-lived WebSocket token. + #### Callback `OnConnectError` ```csharp class DbConnectionBuilder { - public DbConnectionBuilder OnConnectError(Action callback); + public delegate void ConnectErrorCallback(Exception error); + public delegate void ConnectErrorWithReconnectCallback(Exception error, NextReconnect? nextReconnect); + + public DbConnectionBuilder OnConnectError(ConnectErrorCallback callback); + public DbConnectionBuilder OnConnectError(ConnectErrorWithReconnectCallback callback); } ``` -Chain a call to `.OnConnectError(callback)` to your builder to register a callback to run when your connection fails. +Register a callback for an initial connection failure or a failed reconnect attempt. The overload accepting [`NextReconnect?`](#type-nextreconnect) reports the next attempt and delay, or `null` when no retry will occur. Initial connection failures never retry automatically. Both overloads remain supported. #### Callback `OnDisconnect` ```csharp class DbConnectionBuilder { - public DbConnectionBuilder OnDisconnect(Action callback); + public delegate void DisconnectCallback(DbConnection conn, Exception? error); + public delegate void DisconnectWithReconnectCallback(DbConnection conn, Exception? error, NextReconnect? nextReconnect); + + public DbConnectionBuilder OnDisconnect(DisconnectCallback callback); + public DbConnectionBuilder OnDisconnect(DisconnectWithReconnectCallback callback); } ``` Chain a call to `.OnDisconnect(callback)` to your builder to register a callback to run when your `DbConnection` disconnects from the remote database, either as a result of a call to [`Disconnect`](#method-disconnect) or due to an error. +With automatic reconnect enabled, losing an established connection reports the scheduled retry through [`NextReconnect?`](#type-nextreconnect). Keep the connection and its update loop running when this value is non-null. A null value means the connection has ended permanently. An explicit `Disconnect()` reports a null error and no retry, including when called during recovery. + #### Method `WithToken` ```csharp @@ -207,6 +221,87 @@ class DbConnectionBuilder Chain a call to `.WithToken(token)` to your builder to provide an OpenID Connect compliant JSON Web Token to authenticate with, or to explicitly select an anonymous connection. If this method is not called or `null` is passed, SpacetimeDB will generate a new `Identity` and sign a new private access token for the connection. +#### Method `WithAutomaticReconnect` + +```csharp +class DbConnectionBuilder +{ + public DbConnectionBuilder WithAutomaticReconnect(); +} +``` + +Enable automatic recovery after a connection has succeeded at least once. This is opt-in; without it, a lost connection must be replaced by the application. Initial connection failures do not retry. + +Retries use exponential backoff starting at one second, with jitter and a maximum delay of 30 seconds. There is no attempt limit, and a successful connection resets the retry counter. Explicit `Disconnect()`, a changed identity, or a terminal protocol or authentication error stops recovery. See [token refresh](#method-withtokenprovider) for authentication rejection handling. + +The SDK keeps the same `DbConnection`, `Identity`, table handles, subscriptions, and callbacks. Each reconnect attempt uses a fresh `ConnectionId`. Keep calling `FrameTick()` during outages; it drives retries as well as callbacks. + +```csharp +var conn = DbConnection.Builder() + .WithUri("http://localhost:3000") + .WithDatabaseName("my-database") + .WithAutomaticReconnect() + .OnConnect((connection, identity, token) => Console.WriteLine($"Connected as {identity}")) + .OnDisconnect((connection, error, next) => + { + if (next is { } retry) + Console.WriteLine($"Retry {retry.Attempt} in {retry.Delay}."); + }) + .OnConnectError((error, next) => Console.WriteLine(error.Message)) + .Build(); + +conn.SubscriptionBuilder() + .OnApplied(ctx => Console.WriteLine("Subscription ready")) + .SubscribeToAllTables(); +``` + +Arrange for your application's update loop to call `conn.FrameTick()`. With automatic reconnect enabled, subscriptions may be created immediately after `Build()` or during an outage; they are sent once the connection is ready. + +During an outage, cached rows remain readable but may be stale. After `OnConnect`, the SDK replays the retained subscriptions in one batch and reconciles the resulting snapshot with the cache. All tables are updated before row callbacks run; unchanged rows do not fire callbacks, and changed rows produce net insert, update, or delete events. Each successful subscription's `OnApplied` runs again. A rejected subscription gets `OnError` after the successful subscriptions are reconciled. + +Reducer, procedure, and one-off query calls made while disconnected fail immediately and are not queued. Calls awaiting a result when the connection is lost are not replayed: reducer callbacks receive [`Status.UnknownResult`](#variant-unknownresult), while procedures and one-off queries fail with `UnknownResultException`. The server may already have executed those calls; do not automatically repeat non-idempotent operations. + +#### Method `WithTokenProvider` + +```csharp +class DbConnectionBuilder +{ + public DbConnectionBuilder WithTokenProvider(Func> provider); +} +``` + +Register an optional asynchronous provider of authentication tokens for automatic reconnect. Use it together with `WithAutomaticReconnect()`. The provider is not called for the initial connection; supply the initial token through `WithToken(initialToken)`. + +```csharp +var conn = DbConnection.Builder() + .WithUri("http://localhost:3000") + .WithDatabaseName("my-database") + .WithToken(initialToken) + .WithAutomaticReconnect() + .WithTokenProvider(() => RefreshTokenAsync()) + .Build(); +``` + +`initialToken` and `RefreshTokenAsync` come from your authentication integration. The provider must return a non-empty token for the same identity. To switch users, disconnect and build a new connection. + +Before each reconnect attempt, the SDK checks the retained token's JWT `exp` and `iat` claims. It calls the provider when the remaining lifetime is at most 30 seconds or 5% of the token's original lifetime, whichever is greater. If expiry cannot be read, it calls the provider on every attempt. Without a provider, the SDK reuses the retained token. This does not schedule background token refresh while the connection is healthy. + +If the server rejects a reused token, the next attempt forces a provider call regardless of expiry. Rejection of a freshly provided token, or rejection with no provider available, is terminal. If the provider throws, its task faults, or it returns an empty token, `OnConnectError` reports the failure and schedules another attempt. + +The provider is invoked from `FrameTick`; return a task instead of blocking that thread. `Disconnect()` prevents a pending provider result from opening another connection, but does not cancel the provider's own asynchronous work. + +#### Type `NextReconnect` + +```csharp +public readonly struct NextReconnect +{ + public int Attempt { get; } + public TimeSpan Delay { get; } +} +``` + +Retry information supplied to the connection-error and disconnect callbacks. `Attempt` starts at 1 and `Delay` is the scheduled wait before that attempt. A nullable `NextReconnect` value of `null` means no retry is scheduled. A server response indicating that the previous session is still closing retains the current attempt number and uses the initial backoff delay. + #### Method `Build` ```csharp @@ -234,9 +329,11 @@ class DbConnection { } ``` -`FrameTick` will advance the connection until no work remains or until it is disconnected, then return rather than blocking. Games might arrange for this message to be called every frame. +`FrameTick` processes pending updates and advances reconnect timers and token-provider tasks, then returns rather than blocking. Games might call this method every frame. Keep calling it even when `IsActive` is false so automatic reconnect can progress. + +In Unity projects, a `SpacetimeDBNetworkManager` component can call `FrameTick` for active and reconnecting connections automatically. Use either the manager or your own update loop; without one of them, callbacks will not be invoked. -In Unity projects, a `SpacetimeDBNetworkManager` component can call `FrameTick` for active connections automatically. Use either the manager or your own update loop; without one of them, callbacks will not be invoked. +In Godot projects, register the connection with `STDBUpdateManager.Add(conn)` and keep it registered during recovery, or call `FrameTick()` from your own `_Process` method. Remove and disconnect it when the application is done with the connection. It is not advised to run `FrameTick` on a background thread, since it modifies [`dbConnection.Db`](#property-db). If main thread code is also accessing the `Db`, it may observe data races when `FrameTick` runs on another thread. @@ -344,7 +441,7 @@ interface IRemoteDbContext } ``` -Gracefully close the `DbConnection`. Throws an error if the connection is already closed. +Permanently close the `DbConnection`, cancel scheduled reconnects, and ignore any pending token-provider result. This also works during an outage. Repeated calls have no effect. To connect again after an explicit disconnect or terminal failure, build a new connection. ### Subscribe to queries @@ -387,6 +484,8 @@ class SubscriptionBuilder Register a callback to run when the subscription is applied and the matching rows are inserted into the client cache. +With automatic reconnect enabled, this callback runs again after each successful replay. Use it to mark data ready, and keep one-time setup separate so it is not repeated on every reconnect. + ##### Callback `OnError` ```csharp @@ -613,6 +712,8 @@ Terminate this subscription, causing matching rows to be removed from the client Unsubscribing is an asynchronous operation. Matching rows are not removed from the client cache immediately. Use [`UnsubscribeThen`](#method-unsubscribethen) to run a callback once the unsubscribe operation is completed. +With automatic reconnect enabled, you can also unsubscribe while offline or while the subscription is pending. An offline unsubscribe ends the handle locally and excludes it from replay; its cached rows remain until the next snapshot reconciliation. In that case, `UnsubscribeThen` runs when the handle ends, before stale rows are removed. + Returns an error if the subscription has already ended, either due to a previous call to `Unsubscribe` or [`UnsubscribeThen`](#method-unsubscribethen), or due to an error. ##### Method `UnsubscribeThen` @@ -626,6 +727,8 @@ class SubscriptionHandle Terminate this subscription, and run the `onEnded` callback when the subscription is ended and its matching rows are removed from the client cache. Any rows removed from the client cache this way will have [`OnDelete` callbacks](#callback-ondelete) run for them. +During an automatic-reconnect outage, the callback instead runs when the handle ends locally. Stale rows are removed on the next snapshot reconciliation, as described under [`Unsubscribe`](#method-unsubscribe). + Returns an error if the subscription has already ended, either due to a previous call to [`Unsubscribe`](#method-unsubscribe) or `UnsubscribeThen`, or due to an error. ### Read connection metadata @@ -652,6 +755,8 @@ interface IDbContext Get the [`ConnectionId`](#type-connectionid) with which SpacetimeDB identifies the connection. +Automatic reconnect preserves `Identity` but assigns a fresh `ConnectionId` for each attempt. + #### Property `IsActive` ```csharp @@ -661,7 +766,18 @@ interface IDbContext } ``` -`true` if the connection has not yet disconnected. Note that a connection `IsActive` when it is constructed, before its [`OnConnect` callback](#callback-onconnect) is invoked. +`true` when the transport is connected and the connection has not closed. With automatic reconnect enabled, it also requires the initial handshake to have completed; it is false before the first `OnConnect` and during outages. It becomes true before subscription replay completes, so use subscription `OnApplied` to determine when cached data is current. + +#### Property `IsReconnecting` + +```csharp +class DbConnection +{ + public bool IsReconnecting { get; } +} +``` + +`true` while an automatically reconnecting connection that previously succeeded is waiting for a retry, obtaining a token, or connecting again. It is false during the initial connection, after a successful reconnect handshake, and after a terminal failure or explicit `Disconnect()`. Continue ticking the connection while this property is true. ## Type `EventContext` @@ -806,7 +922,8 @@ A `ReducerEvent` contains metadata about a reducer run. record Status : TaggedEnum<( Unit Committed, string Failed, - Unit OutOfEnergy + Unit OutOfEnergy, + Unit UnknownResult )>; ``` @@ -817,6 +934,7 @@ record Status : TaggedEnum<( | [`Committed` variant](#variant-committed) | The reducer ran successfully. | | [`Failed` variant](#variant-failed) | The reducer errored. | | [`OutOfEnergy` variant](#variant-outofenergy) | The reducer was aborted due to insufficient energy. | +| [`UnknownResult` variant](#variant-unknownresult) | The connection was lost before the reducer's result arrived. | #### Variant `Committed` @@ -830,6 +948,12 @@ The reducer returned an error, panicked, or threw an exception. The record paylo The reducer was aborted due to insufficient energy balance of the module owner. +#### Variant `UnknownResult` + +With automatic reconnect enabled, the SDK reports this status for reducer calls whose results were still pending when the connection was lost. The reducer may have committed; this status does not mean that it failed. The SDK does not retry the call. Check the reconciled state or use an application-level idempotency key before deciding to repeat it. + +Regenerate your C# bindings when upgrading. Generated dispatch code forwards this status as an `UnknownResultException` to `OnUnhandledReducerError` when no callback is registered for that reducer. Include `Status.UnknownResult` in your own reducer-result handling as well. + ### Record `Reducer` The module bindings contains an record `Reducer` with a variant for each reducer defined by the module. Each variant has a payload containing the arguments to the reducer. @@ -1107,7 +1231,7 @@ All [`IDbContext`](#interface-idbcontext) implementors, including [`DbConnection For a reducer named `send_message`, generated C# bindings use PascalCase names: - An invoke method, like `SendMessage(...)`. This requests that the module run the reducer. -- A result event, like `OnSendMessage`. This event fires on the calling connection when SpacetimeDB reports that reducer call's result, including committed, failed, and out-of-energy statuses. +- A result event, like `OnSendMessage`. This event fires on the calling connection for committed, failed, and out-of-energy results, or with `Status.UnknownResult` if automatic reconnect is enabled and the connection is lost before the result arrives. Subscribe to reducer result events with `+=` and unsubscribe with `-=`, as with any C# event. diff --git a/sdks/csharp/README.dotnet.md b/sdks/csharp/README.dotnet.md index 749a371c540..0390429888a 100644 --- a/sdks/csharp/README.dotnet.md +++ b/sdks/csharp/README.dotnet.md @@ -6,4 +6,37 @@ This repository contains the [C#](https://learn.microsoft.com/en-us/dotnet/cshar ## Documentation -The C# SDK has a [Quick Start](https://spacetimedb.com/docs/sdks/c-sharp/quickstart) guide and a [Reference](https://spacetimedb.com/docs/sdks/c-sharp). +The C# SDK has a [Quick Start](https://spacetimedb.com/docs/quickstarts/c-sharp) guide and a [Reference](https://spacetimedb.com/docs/clients/c-sharp). + +## Automatic reconnect + +Enable reconnect on the builder and keep calling `FrameTick` during outages: + +```csharp +var conn = DbConnection.Builder() + .WithUri("http://localhost:3000") + .WithDatabaseName("my-database") + .WithToken(savedToken) + .WithAutomaticReconnect() + .WithTokenProvider(() => RefreshAccessTokenAsync()) + .OnConnect((conn, identity, token) => SaveToken(token)) + .OnDisconnect((conn, error, next) => + { + if (next is { } retry) + Console.WriteLine($"Retry {retry.Attempt} in {retry.Delay}."); + }) + .OnConnectError((error, next) => Console.WriteLine(error.Message)) + .Build(); +``` + +Automatic reconnect is opt-in. Retries use exponential backoff with jitter and a 30-second cap, with no attempt limit. Initial connection failures do not retry. `Disconnect()` cancels scheduled attempts and ignores pending token-provider results; it does not cancel the provider's own asynchronous work. Terminal authentication or protocol errors and identity changes also stop retries. + +`WithTokenProvider(Func>)` is optional and is used only during automatic reconnect. Supply the initial token through `WithToken`; the provider is not called for the first connection. Before each retry, the SDK calls the provider when the token has at most 30 seconds or 5% of its original lifetime remaining, whichever is greater. If expiry cannot be read, it calls the provider every attempt. It also forces refresh after a reused token is rejected. Provider failures retry; rejection of a freshly provided token, or rejection without a provider, is terminal. The provider must return a non-empty token for the same identity. No periodic refresh runs while connected. + +`IsActive` is false while waiting for a retry, token provider, or reconnect handshake, and `IsReconnecting` is true. After the handshake, `IsActive` becomes true before subscriptions finish replaying; use `OnApplied` to know when data is ready. Cached rows remain readable during outages. Existing subscription handles and row callbacks survive; subscriptions replay in one batch, `OnApplied` fires again, and row callbacks report only net changes. Register subscriptions once, since `OnConnect` also fires on each reconnect. Create subscriptions in the first `OnConnect`, or immediately after `Build()` when automatic reconnect is enabled. The identity stays the same, while each attempt gets a fresh `ConnectionId`. + +Reducer, procedure, and query calls made while disconnected fail immediately. In-flight queries and procedures fail with `UnknownResultException`; reducer callbacks receive `Status.UnknownResult`, since the server may have executed the call. Regenerate bindings to route this status to `OnUnhandledReducerError` when a reducer has no registered handler. + +Existing callback overloads remain available. The overloads with `NextReconnect?` report the upcoming attempt and delay, or `null` when no retry is scheduled. `SpacetimeDBNetworkManager` keeps ticking reconnecting Unity connections. + +See the [reconnect test application](examples~/reconnect/README.md) for executable outage, batch subscription, and JWT refresh scenarios. diff --git a/sdks/csharp/README.md b/sdks/csharp/README.md index b697aa12753..607423c53d8 100644 --- a/sdks/csharp/README.md +++ b/sdks/csharp/README.md @@ -20,3 +20,11 @@ There is also a comprehensive Godot tutorial/demo available: ## Internal developer documentation See [`DEVELOP.md`](./DEVELOP.md). + +## Automatic reconnect + +C#, Unity, and Godot clients can opt in with `.WithAutomaticReconnect()` on the connection builder. The SDK recovers lost connections, retains cached rows and subscription handles, and replays subscriptions in one batch. Keep calling `FrameTick()` during outages; Unity's `SpacetimeDBNetworkManager` does this automatically. Register subscriptions and row callbacks once, because `OnConnect` and subscription `OnApplied` run again after recovery. + +For expiring credentials, supply the initial token with `.WithToken(initialToken)` and add `.WithTokenProvider(() => RefreshTokenAsync())`. The provider obtains a token for the same identity before a reconnect when needed. Initial connection failures do not retry, and `Disconnect()` stops recovery. Handle `Status.UnknownResult` for pending reducer calls whose outcome was lost; the SDK does not repeat them. + +See the [C# SDK Reference](https://spacetimedb.com/docs/clients/c-sharp#method-withautomaticreconnect) for retry callbacks, token refresh, and subscription behavior, or the [reconnect test application](examples~/reconnect/README.md) for runnable scenarios. diff --git a/sdks/csharp/src/Event.cs b/sdks/csharp/src/Event.cs index 767265b4224..de656244cdc 100644 --- a/sdks/csharp/src/Event.cs +++ b/sdks/csharp/src/Event.cs @@ -108,7 +108,8 @@ public interface IProcedureArgs : BSATN.IStructuralReadWrite public partial record Status : TaggedEnum<( Unit Committed, string Failed, - Unit OutOfEnergy + Unit OutOfEnergy, + Unit UnknownResult )>; public record ReducerEvent( @@ -144,6 +145,7 @@ public record UnknownTransaction : Event; public interface ISubscriptionHandle { void OnApplied(ISubscriptionEventContext ctx); + void RebindQuerySetId(QuerySetId id); void OnError(IErrorContext ctx); void OnEnded(ISubscriptionEventContext ctx); } @@ -202,6 +204,12 @@ public bool IsActive } } + void ISubscriptionHandle.RebindQuerySetId(QuerySetId id) + { + queryId = id; + state = new SubscriptionState.Pending(new()); + } + void ISubscriptionHandle.OnApplied(ISubscriptionEventContext ctx) { state = new SubscriptionState.Active(queryId ?? throw new InvalidOperationException("Subscription query id is missing.")); @@ -257,7 +265,7 @@ public void Unsubscribe() /// public void UnsubscribeThen(Action? onEnded) { - if (state is not SubscriptionState.Active) + if (state is SubscriptionState.Ended || (state is not SubscriptionState.Active && !conn.AutomaticReconnectEnabled)) { throw new Exception("Cannot unsubscribe from inactive subscription."); } diff --git a/sdks/csharp/src/Plugins/WebSocket.jslib b/sdks/csharp/src/Plugins/WebSocket.jslib index 820d4a17125..9384aa59fa2 100644 --- a/sdks/csharp/src/Plugins/WebSocket.jslib +++ b/sdks/csharp/src/Plugins/WebSocket.jslib @@ -9,7 +9,7 @@ mergeInto(LibraryManager.library, { WebSocket_Init__deps: ['$WebSocketDynCall'], WebSocket_Init: function(openCallback, messageCallback, closeCallback, errorCallback) { - this._webSocketManager = { + this._webSocketManager = this._webSocketManager || { instances: {}, nextId: 1, callbacks: { @@ -27,7 +27,7 @@ mergeInto(LibraryManager.library, { manager.callbacks.error = errorCallback; }, - WebSocket_Connect: async function(baseUriPtr, uriPtr, protocolPtr, authTokenPtr, callbackPtr) { + WebSocket_Connect: async function(baseUriPtr, uriPtr, protocolPtr, authTokenPtr, requestId, callbackPtr) { try { var manager = this._webSocketManager; var host = UTF8ToString(baseUriPtr); @@ -51,7 +51,9 @@ mergeInto(LibraryManager.library, { uri += `&token=${token}`; } } else { - throw new Error(`Failed to verify token: ${response.statusText}`); + var error = new Error(`Failed to verify token: ${response.statusText}`); + error.status = response.status; + throw error; } } @@ -93,10 +95,10 @@ mergeInto(LibraryManager.library, { } }; - WebSocketDynCall('vi', callbackPtr, [socketId]); + WebSocketDynCall('vii', callbackPtr, [requestId, socketId]); } catch (e) { console.error("WebSocket connection error:", e); - WebSocketDynCall('vi', callbackPtr, [-1]); + WebSocketDynCall('vii', callbackPtr, [requestId, e.status ? -e.status : -1]); } }, diff --git a/sdks/csharp/src/ProcedureCallbacks.cs b/sdks/csharp/src/ProcedureCallbacks.cs index b542f2c4b7b..9ac84f6c498 100644 --- a/sdks/csharp/src/ProcedureCallbacks.cs +++ b/sdks/csharp/src/ProcedureCallbacks.cs @@ -64,7 +64,8 @@ public void FailAll(IProcedureEventContext ctx, Exception error) foreach (var wrapper in wrappers) { - wrapper.InvokeFailure(ctx, error); + try { wrapper.InvokeFailure(ctx, error); } + catch (Exception e) { Log.Exception(e); } } } diff --git a/sdks/csharp/src/Reconnect.cs b/sdks/csharp/src/Reconnect.cs new file mode 100644 index 00000000000..b7df55e9fa8 --- /dev/null +++ b/sdks/csharp/src/Reconnect.cs @@ -0,0 +1,396 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.IO; +using System.Linq; +using System.Runtime.Serialization; +using System.Runtime.Serialization.Json; +using System.Threading.Tasks; +using SpacetimeDB.ClientApi; + +namespace SpacetimeDB +{ + public readonly struct NextReconnect + { + public int Attempt { get; } + public TimeSpan Delay { get; } + + internal NextReconnect(int attempt, TimeSpan delay) + { + Attempt = attempt; + Delay = delay; + } + } + + public class UnknownResultException : SpacetimeDBException + { + public UnknownResultException() : base("Connection lost before the result was received. The operation may have executed.") { } + } + + internal class ConnectionProtocolException : SpacetimeDBException + { + internal ConnectionProtocolException(string message, Exception? inner = null) : base(message, inner) { } + } + + internal static class ReconnectPolicy + { + internal static TimeSpan Delay(int attempt, double random) => TimeSpan.FromMilliseconds( + Math.Min(30000, Math.Min(30000, 1000 * Math.Pow(2, Math.Min(30, Math.Max(0, attempt - 1)))) * (0.5 + random))); + + [DataContract] +#if UNITY_5_3_OR_NEWER + [UnityEngine.Scripting.Preserve] +#endif + private class Claims + { + [DataMember(Name = "exp")] + public double? Exp { get; set; } + [DataMember(Name = "iat")] + public double? Iat { get; set; } + } + + internal static bool TokenNeedsRefresh(string? token, DateTimeOffset now) + { + try + { + var payload = token!.Split('.')[1].Replace('-', '+').Replace('_', '/'); + payload = payload.PadRight((payload.Length + 3) / 4 * 4, '='); + using var stream = new MemoryStream(Convert.FromBase64String(payload)); + var claims = (Claims)new DataContractJsonSerializer(typeof(Claims)).ReadObject(stream)!; + if (claims.Exp is not double exp || double.IsNaN(exp) || double.IsInfinity(exp)) return true; + var margin = Math.Max(30, claims.Iat is double iat ? (exp - iat) * 0.05 : 0); + return exp - now.ToUnixTimeMilliseconds() / 1000.0 <= margin; + } + catch + { + return true; + } + } + } + + public abstract partial class DbConnectionBase + { + private bool automaticReconnect; + private Func>? tokenProvider; + private string? retainedToken; + private readonly ConnectionId sessionId = ConnectionId.Random(); + private (string Uri, string Database, Compression Compression, bool Light, bool? Confirmed) connectionOptions; + private bool hasEverConnected; + private bool preparingReplay; + private bool forceTokenRefresh; + private bool usedFreshToken; + private int reconnectAttempt; + private double? reconnectAt; + private Task? tokenTask; + private volatile int socketGeneration; + private uint? replayRequestId; + private HashSet? replayQueryIds; + private readonly Random reconnectRandom = new(); + private readonly Dictionary subscriptionQueries = new(); + private readonly HashSet unsubscribeRequested = new(); + private event Action? onConnectError; + private event Action? onDisconnect; + + internal Func ReconnectClock = () => Stopwatch.GetTimestamp() / (double)Stopwatch.Frequency; + internal Func SocketFactory = () => new WebSocket(new WebSocket.ConnectOptions { Protocol = "v2.bsatn.spacetimedb" }); + + bool IDbConnection.AutomaticReconnectEnabled => automaticReconnect; + + public bool IsReconnecting => automaticReconnect && hasEverConnected && !onConnectInvoked && !isClosing; + + void IDbConnection.ConfigureReconnect(bool enabled, Func>? provider) + { + automaticReconnect = enabled; + tokenProvider = provider; + } + + private WebSocket CreateWebSocket() + { + var socket = SocketFactory(); + var generation = socketGeneration; + socket.OnMessage += (bytes, timestamp) => EnqueueMessage(bytes, timestamp, generation); + socket.OnClose += error => EnqueueSocketAction(() => HandleSocketFailure(error), generation); + socket.OnConnectError += error => EnqueueSocketAction(() => HandleSocketFailure(error), generation); + socket.OnSendError += error => EnqueueSocketAction(() => + { + onSendError?.Invoke(error); + HandleSocketFailure(error); + }, generation); + return socket; + } + + private void EnqueueMessage(byte[] bytes, DateTime timestamp, int generation) + { + if (isClosing || generation != socketGeneration) return; + _parseQueue.Add(new UnparsedMessage + { + bytes = bytes, + timestamp = timestamp, + generation = generation, + parseQueueTrackerId = stats.ParseMessageQueueTracker.StartTrackingRequest() + }); + } + + private void EnqueueSocketAction(Action action, int generation) + { + if (!isClosing && generation == socketGeneration) + _parseQueue.Add(new UnparsedMessage { action = action, generation = generation }); + } + + private void StartSocket() + { + if (isClosing) return; + connectionClosed = false; + onConnectInvoked = false; + initialConnectionId = null; + if (hasEverConnected) ConnectionId = ConnectionId.Random(); + socketGeneration++; + webSocket.Abort(); + webSocket = CreateWebSocket(); + Log.Info($"SpacetimeDBClient: Connecting to {connectionOptions.Uri} {connectionOptions.Database}"); + if (IsTesting) return; + var socket = webSocket; + var connectionId = ConnectionId; + var generation = socketGeneration; + async Task ConnectSocket() + { + try + { + await socket.Connect(retainedToken, connectionOptions.Uri, connectionOptions.Database, + connectionId, connectionOptions.Compression, connectionOptions.Light, connectionOptions.Confirmed, + automaticReconnect ? sessionId : null); + } + catch (Exception error) + { + EnqueueSocketAction(() => HandleSocketFailure(error), generation); + } + } +#if UNITY_WEBGL && !UNITY_EDITOR + _ = ConnectSocket(); +#else + _ = Task.Run(ConnectSocket); +#endif + } + + private void HandleInitialConnection(InitialConnection initial) + { + if (automaticReconnect && (onConnectInvoked || initial.ConnectionId != ConnectionId || + (Identity is Identity identity && identity != initial.Identity))) + { + HandleSocketFailure(new ConnectionProtocolException("Unexpected identity or connection ID in InitialConnection.")); + return; + } + if (!automaticReconnect && ((Identity.HasValue && Identity.Value != initial.Identity) || + (initialConnectionId.HasValue && initialConnectionId.Value != initial.ConnectionId))) + { + Log.Error("Received InitialConnection with an unexpected identity or connection ID."); + return; + } + if (onConnectInvoked) return; + var reconnect = hasEverConnected; + Identity = initial.Identity; + initialConnectionId = initial.ConnectionId; + ConnectionId = initial.ConnectionId; + if (string.IsNullOrEmpty(retainedToken)) retainedToken = initial.Token; + hasEverConnected = true; + onConnectInvoked = true; + reconnectAttempt = 0; + reconnectAt = null; + try + { + onConnect?.Invoke(initial.Identity, automaticReconnect ? retainedToken! : initial.Token); + } + catch (Exception error) + { + Log.Exception(error); + } + finally + { + if (!automaticReconnect) onConnect = null; + if (!isClosing && automaticReconnect) + { + if (reconnect) ReplaySubscriptions(); + else + { + preparingReplay = false; + foreach (var id in subscriptions.Keys.ToArray()) SendSubscription(id); + } + } + } + } + + private void HandleSocketFailure(Exception? error) + { + if (isClosing || connectionClosed) return; + var established = onConnectInvoked; + connectionClosed = true; + onConnectInvoked = false; + socketGeneration++; + webSocket.Abort(); + preparingReplay = automaticReconnect; + replayRequestId = null; + replayQueryIds = null; + tokenTask = null; + var authError = error is WebSocket.ConnectException { StatusCode: 400 or 401 or 403 }; + var terminal = error is ConnectionProtocolException || + (authError && (tokenProvider == null || usedFreshToken)); + NextReconnect? next = null; + if (automaticReconnect && hasEverConnected && !terminal) + { + if (authError) forceTokenRefresh = true; + var sessionBusy = !established && error is WebSocket.CloseException { Code: 4000 }; + if (!sessionBusy) reconnectAttempt = reconnectAttempt == int.MaxValue ? int.MaxValue : reconnectAttempt + 1; + var delay = ReconnectPolicy.Delay(sessionBusy ? 1 : reconnectAttempt, reconnectRandom.NextDouble()); + reconnectAt = ReconnectClock() + delay.TotalSeconds; + next = new NextReconnect(reconnectAttempt, delay); + } + else + { + EndConnection(); + } + FailPendingOperations(automaticReconnect ? new UnknownResultException() : new OperationCanceledException("Connection closed.")); + foreach (var id in unsubscribeRequested.ToArray()) EndSubscription(id); + if (isClosing && next != null) return; + if (established) onDisconnect?.Invoke(error, next); + else onConnectError?.Invoke(error ?? new SpacetimeDBException("Connection closed before InitialConnection."), next); + } + + private void TickReconnect() + { + if (isClosing) return; + if (tokenTask != null) + { + if (!tokenTask.IsCompleted) return; + var completed = tokenTask; + tokenTask = null; + try + { + retainedToken = completed.GetAwaiter().GetResult(); + if (string.IsNullOrEmpty(retainedToken)) throw new InvalidOperationException("Token provider returned an empty token."); + forceTokenRefresh = false; + usedFreshToken = true; + StartSocket(); + } + catch (Exception error) + { + connectionClosed = false; + HandleSocketFailure(error); + } + return; + } + if (reconnectAt is not double due || ReconnectClock() < due) return; + reconnectAt = null; + usedFreshToken = false; + if (tokenProvider != null && (forceTokenRefresh || ReconnectPolicy.TokenNeedsRefresh(retainedToken, DateTimeOffset.UtcNow))) + { + try + { + tokenTask = tokenProvider() ?? Task.FromException(new InvalidOperationException("Token provider returned no task.")); + } + catch (Exception error) + { + connectionClosed = false; + HandleSocketFailure(error); + } + } + else StartSocket(); + } + + private void EndConnection() + { + isClosing = true; + connectionClosed = true; + onConnectInvoked = false; + reconnectAt = null; + tokenTask = null; + socketGeneration++; + webSocket.Abort(); + _parseCancellationTokenSource.Cancel(); + while (_parseQueue.TryTake(out _)) { } + while (_applyQueue.TryTake(out _)) { } + FailPendingOperations(automaticReconnect ? new UnknownResultException() : new OperationCanceledException("Connection closed.")); +#if UNITY_5_3_OR_NEWER + SpacetimeDBNetworkManager._instance?.RemoveConnection(this); +#endif + } + + private void SendSubscription(uint id) + { + webSocket.Send(new ClientMessage.Subscribe(new Subscribe( + stats.SubscriptionRequestTracker.StartTrackingRequest(), new QuerySetId(id), subscriptionQueries[id].ToList()))); + } + + private void RemoveSubscription(uint id) + { + subscriptions.Remove(id); + subscriptionQueries.Remove(id); + unsubscribeRequested.Remove(id); + } + + private void EndSubscription(uint id) + { + if (!subscriptions.TryGetValue(id, out var handle)) return; + RemoveSubscription(id); + try { handle.OnEnded(MakeSubscriptionEventContext()); } + catch (Exception error) { Log.Exception(error); } + } + + private void ReplaySubscriptions() + { + var entries = subscriptions.Select(entry => (entry.Value, subscriptionQueries[entry.Key])).ToArray(); + subscriptions.Clear(); + subscriptionQueries.Clear(); + var sets = new List(); + foreach (var (handle, queries) in entries) + { + var id = querySetIdAllocator.Next(); + handle.RebindQuerySetId(new QuerySetId(id)); + subscriptions[id] = handle; + subscriptionQueries[id] = queries; + sets.Add(new SubscribeSet(new QuerySetId(id), queries.ToList())); + } + preparingReplay = false; + replayRequestId = stats.SubscriptionRequestTracker.StartTrackingRequest(); + replayQueryIds = new HashSet(subscriptions.Keys); + if (sets.Count == 0) + { + stats.SubscriptionRequestTracker.FinishTrackingRequest(replayRequestId.Value); + ApplyReplayBatch(new SubscribeBatchApplied(replayRequestId.Value, new()), ParsedDatabaseUpdate.New()); + return; + } + webSocket.Send(new ClientMessage.SubscribeBatch(new SubscribeBatch(replayRequestId.Value, sets))); + } + + private void ApplyReplayBatch(SubscribeBatchApplied batch, ParsedDatabaseUpdate update) + { + var ids = new HashSet(batch.Results.Select(result => result.QuerySetId.Id)); + if (batch.RequestId != replayRequestId || replayQueryIds == null || + ids.Count != batch.Results.Count || !ids.SetEquals(replayQueryIds)) + { + HandleSocketFailure(new ConnectionProtocolException("Unexpected subscription replay response.")); + return; + } + replayRequestId = null; + replayQueryIds = null; + foreach (var table in Db.AllTables) table.AddSnapshotDeletes(update); + ApplyUpdate(ToEventContext(new Event.SubscribeApplied()), update); + foreach (var result in batch.Results) + { + if (!subscriptions.TryGetValue(result.QuerySetId.Id, out var handle)) continue; + try + { + if (result.Outcome is SubscribeSetOutcome.Error(var message)) + { + RemoveSubscription(result.QuerySetId.Id); + handle.OnError(ToErrorContext(new SpacetimeDBException(message))); + } + else handle.OnApplied(MakeSubscriptionEventContext()); + } + catch (Exception error) + { + Log.Exception(error); + } + } + } + } +} diff --git a/sdks/csharp/src/Reconnect.cs.meta b/sdks/csharp/src/Reconnect.cs.meta new file mode 100644 index 00000000000..35656a51df9 --- /dev/null +++ b/sdks/csharp/src/Reconnect.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: ad344a08673046929d4622c9984f4211 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/RemoteTablesBase.cs b/sdks/csharp/src/RemoteTablesBase.cs index a1a6e2b64c4..f500d5bac9e 100644 --- a/sdks/csharp/src/RemoteTablesBase.cs +++ b/sdks/csharp/src/RemoteTablesBase.cs @@ -11,6 +11,8 @@ protected void AddTable(IRemoteTableHandle table) tables.Add(table.RemoteTableName, table); } + internal IEnumerable AllTables => tables.Values; + internal IRemoteTableHandle? GetTable(string name) { if (tables.TryGetValue(name, out var table)) diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/ClientMessage.g.cs b/sdks/csharp/src/SpacetimeDB/ClientApi/ClientMessage.g.cs index 49fa3db43be..8cb70ea672c 100644 --- a/sdks/csharp/src/SpacetimeDB/ClientApi/ClientMessage.g.cs +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/ClientMessage.g.cs @@ -13,6 +13,7 @@ public partial record ClientMessage : SpacetimeDB.TaggedEnum<( Unsubscribe Unsubscribe, OneOffQuery OneOffQuery, CallReducer CallReducer, - CallProcedure CallProcedure + CallProcedure CallProcedure, + SubscribeBatch SubscribeBatch )>; } diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/ServerMessage.g.cs b/sdks/csharp/src/SpacetimeDB/ClientApi/ServerMessage.g.cs index 6082a13cd44..ef7975c9371 100644 --- a/sdks/csharp/src/SpacetimeDB/ClientApi/ServerMessage.g.cs +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/ServerMessage.g.cs @@ -16,6 +16,7 @@ public partial record ServerMessage : SpacetimeDB.TaggedEnum<( TransactionUpdate TransactionUpdate, OneOffQueryResult OneOffQueryResult, ReducerResult ReducerResult, - ProcedureResult ProcedureResult + ProcedureResult ProcedureResult, + SubscribeBatchApplied SubscribeBatchApplied )>; } diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatch.g.cs b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatch.g.cs new file mode 100644 index 00000000000..1729315dabe --- /dev/null +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatch.g.cs @@ -0,0 +1,35 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#nullable enable + +using System; +using System.Collections.Generic; +using System.Runtime.Serialization; + +namespace SpacetimeDB.ClientApi +{ + [SpacetimeDB.Type] + [DataContract] + public sealed partial class SubscribeBatch + { + [DataMember(Name = "request_id")] + public uint RequestId; + [DataMember(Name = "sets")] + public System.Collections.Generic.List Sets; + + public SubscribeBatch( + uint RequestId, + System.Collections.Generic.List Sets + ) + { + this.RequestId = RequestId; + this.Sets = Sets; + } + + public SubscribeBatch() + { + this.Sets = new(); + } + } +} diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatch.g.cs.meta b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatch.g.cs.meta new file mode 100644 index 00000000000..223adb62dca --- /dev/null +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatch.g.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: ef069d8f605f474291057cf0a59557bb +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatchApplied.g.cs b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatchApplied.g.cs new file mode 100644 index 00000000000..ff439dfaa96 --- /dev/null +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatchApplied.g.cs @@ -0,0 +1,35 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#nullable enable + +using System; +using System.Collections.Generic; +using System.Runtime.Serialization; + +namespace SpacetimeDB.ClientApi +{ + [SpacetimeDB.Type] + [DataContract] + public sealed partial class SubscribeBatchApplied + { + [DataMember(Name = "request_id")] + public uint RequestId; + [DataMember(Name = "results")] + public System.Collections.Generic.List Results; + + public SubscribeBatchApplied( + uint RequestId, + System.Collections.Generic.List Results + ) + { + this.RequestId = RequestId; + this.Results = Results; + } + + public SubscribeBatchApplied() + { + this.Results = new(); + } + } +} diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatchApplied.g.cs.meta b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatchApplied.g.cs.meta new file mode 100644 index 00000000000..8207019c899 --- /dev/null +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeBatchApplied.g.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: b10d4dc07acd4c6c8e57d665627d5144 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSet.g.cs b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSet.g.cs new file mode 100644 index 00000000000..1e5cfbe19c1 --- /dev/null +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSet.g.cs @@ -0,0 +1,36 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#nullable enable + +using System; +using System.Collections.Generic; +using System.Runtime.Serialization; + +namespace SpacetimeDB.ClientApi +{ + [SpacetimeDB.Type] + [DataContract] + public sealed partial class SubscribeSet + { + [DataMember(Name = "query_set_id")] + public QuerySetId QuerySetId; + [DataMember(Name = "query_strings")] + public System.Collections.Generic.List QueryStrings; + + public SubscribeSet( + QuerySetId QuerySetId, + System.Collections.Generic.List QueryStrings + ) + { + this.QuerySetId = QuerySetId; + this.QueryStrings = QueryStrings; + } + + public SubscribeSet() + { + this.QuerySetId = new(); + this.QueryStrings = new(); + } + } +} diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSet.g.cs.meta b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSet.g.cs.meta new file mode 100644 index 00000000000..a4c971f694b --- /dev/null +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSet.g.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 12694437c5a9427d93caac103c9b727b +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetOutcome.g.cs b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetOutcome.g.cs new file mode 100644 index 00000000000..8c3c940c930 --- /dev/null +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetOutcome.g.cs @@ -0,0 +1,15 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#nullable enable + +using System; + +namespace SpacetimeDB.ClientApi +{ + [SpacetimeDB.Type] + public partial record SubscribeSetOutcome : SpacetimeDB.TaggedEnum<( + QueryRows Applied, + string Error + )>; +} diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetOutcome.g.cs.meta b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetOutcome.g.cs.meta new file mode 100644 index 00000000000..eeb2a5831f6 --- /dev/null +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetOutcome.g.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: cbfe02e8526449db8f523aad5ee89f33 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetResult.g.cs b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetResult.g.cs new file mode 100644 index 00000000000..9ee8fe25497 --- /dev/null +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetResult.g.cs @@ -0,0 +1,36 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#nullable enable + +using System; +using System.Collections.Generic; +using System.Runtime.Serialization; + +namespace SpacetimeDB.ClientApi +{ + [SpacetimeDB.Type] + [DataContract] + public sealed partial class SubscribeSetResult + { + [DataMember(Name = "query_set_id")] + public QuerySetId QuerySetId; + [DataMember(Name = "outcome")] + public SubscribeSetOutcome Outcome; + + public SubscribeSetResult( + QuerySetId QuerySetId, + SubscribeSetOutcome Outcome + ) + { + this.QuerySetId = QuerySetId; + this.Outcome = Outcome; + } + + public SubscribeSetResult() + { + this.QuerySetId = new(); + this.Outcome = null!; + } + } +} diff --git a/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetResult.g.cs.meta b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetResult.g.cs.meta new file mode 100644 index 00000000000..6dccdc10c6b --- /dev/null +++ b/sdks/csharp/src/SpacetimeDB/ClientApi/SubscribeSetResult.g.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: c025a1bc88d94e34b22286a2c3d5587d +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/SpacetimeDBClient.cs b/sdks/csharp/src/SpacetimeDBClient.cs index 5e8e0b49057..c41242a1a13 100644 --- a/sdks/csharp/src/SpacetimeDBClient.cs +++ b/sdks/csharp/src/SpacetimeDBClient.cs @@ -26,6 +26,8 @@ public sealed class DbConnectionBuilder Compression? compression; bool light; bool? confirmedReads; + bool automaticReconnect; + Func>? tokenProvider; public DbConnection Build() { @@ -37,6 +39,7 @@ public DbConnection Build() { throw new InvalidOperationException("Building DbConnection with a null nameOrAddress. Call WithDatabaseName() first."); } + conn.ConfigureReconnect(automaticReconnect, tokenProvider); conn.Connect(token, uri, nameOrAddress, compression ?? Compression.Brotli, light, confirmedReads); #if UNITY_5_3_OR_NEWER if (SpacetimeDBNetworkManager._instance != null) @@ -65,6 +68,18 @@ public DbConnectionBuilder WithToken(string? token) return this; } + public DbConnectionBuilder WithAutomaticReconnect() + { + automaticReconnect = true; + return this; + } + + public DbConnectionBuilder WithTokenProvider(Func> provider) + { + tokenProvider = provider ?? throw new ArgumentNullException(nameof(provider)); + return this; + } + public DbConnectionBuilder WithCompression(Compression compression) { this.compression = compression; @@ -92,18 +107,26 @@ public DbConnectionBuilder OnConnect(ConnectCallback cb) } public delegate void ConnectErrorCallback(Exception e); + public delegate void ConnectErrorWithReconnectCallback(Exception e, NextReconnect? nextReconnect); - public DbConnectionBuilder OnConnectError(ConnectErrorCallback cb) + public DbConnectionBuilder OnConnectError(ConnectErrorCallback cb) => + OnConnectError((e, _) => cb(e)); + + public DbConnectionBuilder OnConnectError(ConnectErrorWithReconnectCallback cb) { - conn.AddOnConnectError(e => cb(e)); + conn.AddOnConnectError((e, next) => cb(e, next)); return this; } public delegate void DisconnectCallback(DbConnection conn, Exception? e); + public delegate void DisconnectWithReconnectCallback(DbConnection conn, Exception? e, NextReconnect? nextReconnect); + + public DbConnectionBuilder OnDisconnect(DisconnectCallback cb) => + OnDisconnect((conn, e, _) => cb(conn, e)); - public DbConnectionBuilder OnDisconnect(DisconnectCallback cb) + public DbConnectionBuilder OnDisconnect(DisconnectWithReconnectCallback cb) { - conn.AddOnDisconnect(e => cb(conn, e)); + conn.AddOnDisconnect((e, next) => cb(conn, e, next)); return this; } } @@ -113,8 +136,11 @@ public interface IDbConnection internal void Connect(string? token, string uri, string addressOrName, Compression compression, bool light, bool? confirmedReads); internal void AddOnConnect(Action cb); - internal void AddOnConnectError(WebSocket.ConnectErrorEventHandler cb); - internal void AddOnDisconnect(WebSocket.CloseEventHandler cb); + internal void AddOnConnectError(Action cb); + internal void AddOnDisconnect(Action cb); + internal void ConfigureReconnect(bool enabled, Func>? provider); + bool IsReconnecting { get; } + internal bool AutomaticReconnectEnabled { get; } internal QuerySetId? Subscribe(ISubscriptionHandle handle, string[] querySqls); internal void Unsubscribe(QuerySetId queryId); @@ -132,7 +158,7 @@ void InternalCallProcedure( where TReturn : IStructuralReadWrite, new(); } - public abstract class DbConnectionBase : IDbConnection + public abstract partial class DbConnectionBase : IDbConnection where DbConnection : DbConnectionBase, new() where Tables : RemoteTablesBase { @@ -164,7 +190,7 @@ public abstract class DbConnectionBase : IDbConne /// private UintAllocator querySetIdAllocator; - public readonly ConnectionId ConnectionId = ConnectionId.Random(); + public ConnectionId ConnectionId { get; private set; } = ConnectionId.Random(); public Identity? Identity { get; private set; } private ConnectionId? initialConnectionId; private bool onConnectInvoked; @@ -202,7 +228,17 @@ private void FailPendingOperations(Exception error) } } - pendingReducerCalls.Clear(); + foreach (var entry in pendingReducerCalls.ToArray()) + { + if (!pendingReducerCalls.TryRemove(entry.Key, out var pending) || !automaticReconnect) continue; + try + { + var reducerEvent = new ReducerEvent(default, new Status.UnknownResult(default), + Identity ?? default, ConnectionId, null, pending.Reducer); + Dispatch(ToReducerEventContext(reducerEvent), pending.Reducer); + } + catch (Exception e) { Log.Exception(e); } + } try { @@ -224,33 +260,19 @@ private void FailPendingOperations(Exception error) } } - private bool isClosing; + private volatile bool isClosing; +#if !(UNITY_WEBGL && !UNITY_EDITOR) private readonly Thread networkMessageParseThread; +#endif public readonly Stats stats = new(); protected DbConnectionBase() { - var options = new WebSocket.ConnectOptions - { - Protocol = "v2.bsatn.spacetimedb" - }; - webSocket = new WebSocket(options); - webSocket.OnMessage += OnMessageReceived; - webSocket.OnSendError += a => onSendError?.Invoke(a); -#if UNITY_5_3_OR_NEWER - webSocket.OnClose += (e) => - { - if (SpacetimeDBNetworkManager._instance != null) - { - SpacetimeDBNetworkManager._instance.RemoveConnection(this); - } - }; - + webSocket = CreateWebSocket(); #if UNITY_WEBGL && !UNITY_EDITOR if (SpacetimeDBNetworkManager._instance != null) SpacetimeDBNetworkManager._instance.StartCoroutine(ParseMessages()); #endif -#endif #if !(UNITY_WEBGL && !UNITY_EDITOR) // For targets other than webgl we start a thread to parse messages @@ -266,6 +288,8 @@ internal struct UnparsedMessage /// The bytes of the message. /// public byte[] bytes; + public int generation; + public Action? action; /// /// The timestamp the message came off the wire. @@ -281,11 +305,13 @@ internal struct UnparsedMessage internal struct ParsedMessage { public ServerMessage message; + public int generation; + public Action? action; + public Exception? error; + public Status? reducerStatus; public ParsedDatabaseUpdate dbOps; public DateTime receiveTimestamp; public uint applyQueueTrackerId; - public ReducerEvent? reducerEvent; - public ProcedureEvent? procedureEvent; } private readonly BlockingCollection _parseQueue = @@ -336,9 +362,9 @@ internal void ParseMessages() } } - ParsedDatabaseUpdate ParseSubscribeRows(QueryRows queryRows) + ParsedDatabaseUpdate ParseSubscribeRows(QueryRows queryRows, ParsedDatabaseUpdate? target = null) { - var dbOps = ParsedDatabaseUpdate.New(); + var dbOps = target ?? ParsedDatabaseUpdate.New(); var empty = EmptyRowList(); foreach (var tableRows in queryRows.Tables) { @@ -426,17 +452,6 @@ string DecodeReducerError(List bytes) } } - void ParseOneOffQuery(OneOffQueryResult resp) - { - if (!waitingOneOffQueries.TryRemove(resp.RequestId, out var resultSource)) - { - Log.Error($"Response to unknown one-off-query request_id: {resp.RequestId}"); - return; - } - - resultSource.TrySetResult(resp); - } - while (!isClosing) { @@ -447,7 +462,18 @@ void ParseOneOffQuery(OneOffQueryResult resp) try { var message = _parseQueue.Take(_parseCancellationToken); - var parsedMessage = ParseMessage(message); + if (message.generation != socketGeneration) continue; + ParsedMessage parsedMessage; + try + { + parsedMessage = message.action != null + ? new ParsedMessage { action = message.action, generation = message.generation } + : ParseMessage(message); + } + catch (Exception e) + { + parsedMessage = new ParsedMessage { error = e, generation = message.generation }; + } _applyQueue.Add(parsedMessage, _parseCancellationToken); } catch (OperationCanceledException) @@ -461,8 +487,7 @@ void ParseOneOffQuery(OneOffQueryResult resp) catch (Exception e) { Log.Exception(e); - FailPendingOperations(new OperationCanceledException("Message parsing failed; connection closed.", e)); - Disconnect(); + _applyQueue.Add(new ParsedMessage { error = e, generation = socketGeneration }); #if UNITY_WEBGL && !UNITY_EDITOR break; #else @@ -480,8 +505,7 @@ ParsedMessage ParseMessage(UnparsedMessage unparsed) stats.ParseMessageQueueTracker.FinishTrackingRequest(unparsed.parseQueueTrackerId, trackerMetadata); var parseStart = DateTime.UtcNow; - ReducerEvent? reducerEvent = default; - ProcedureEvent? procedureEvent = default; + Status? reducerStatus = null; switch (message) { @@ -491,6 +515,14 @@ ParsedMessage ParseMessage(UnparsedMessage unparsed) stats.SubscriptionRequestTracker.FinishTrackingRequest(subscribeApplied.RequestId, unparsed.timestamp); dbOps = ParseSubscribeRows(subscribeApplied.Rows); break; + case ServerMessage.SubscribeBatchApplied(var batch): + stats.SubscriptionRequestTracker.FinishTrackingRequest(batch.RequestId, unparsed.timestamp); + foreach (var result in batch.Results) + { + if (result.Outcome is SubscribeSetOutcome.Applied(var rows)) + ParseSubscribeRows(rows, dbOps); + } + break; case ServerMessage.UnsubscribeApplied(var unsubscribeApplied): stats.SubscriptionRequestTracker.FinishTrackingRequest(unsubscribeApplied.RequestId, unparsed.timestamp); if (unsubscribeApplied.Rows != null) @@ -508,7 +540,6 @@ ParsedMessage ParseMessage(UnparsedMessage unparsed) dbOps = ParseTransactionUpdate(transactionUpdate); break; case ServerMessage.OneOffQueryResult(var resp): - ParseOneOffQuery(resp); break; case ServerMessage.ReducerResult(var reducerResult): if (!stats.ReducerRequestTracker.FinishTrackingRequest(reducerResult.RequestId, unparsed.timestamp)) @@ -516,7 +547,7 @@ ParsedMessage ParseMessage(UnparsedMessage unparsed) Log.Warn($"Failed to finish tracking reducer request: {reducerResult.RequestId}"); } - var reducerStatus = reducerResult.Result switch + reducerStatus = reducerResult.Result switch { ReducerOutcome.Ok => Committed, ReducerOutcome.OkEmpty => Committed, @@ -530,40 +561,8 @@ ParsedMessage ParseMessage(UnparsedMessage unparsed) dbOps = ParseTransactionUpdate(ok.TransactionUpdate); } - if (pendingReducerCalls.TryRemove(reducerResult.RequestId, out var pendingReducer)) - { - try - { - reducerEvent = new( - (DateTimeOffset)reducerResult.Timestamp, - reducerStatus, - Identity ?? throw new InvalidOperationException("Identity not set"), - ConnectionId, - null, - pendingReducer.Reducer); - } - catch (Exception) - { - // The local reducer request still completed; failure here should not block update apply. - } - } - else - { - throw new InvalidOperationException( - $"Reducer result for unknown request_id {reducerResult.RequestId}" - ); - } break; case ServerMessage.ProcedureResult(var procedureResult): - procedureEvent = new ProcedureEvent( - procedureResult.Timestamp, - procedureResult.Status, - Identity ?? throw new InvalidOperationException("Identity not set"), - ConnectionId, - procedureResult.TotalHostExecutionDuration, - procedureResult.RequestId - ); - if (!stats.ProcedureRequestTracker.FinishTrackingRequest(procedureResult.RequestId, unparsed.timestamp)) { Log.Warn($"Failed to finish tracking procedure request: {procedureResult.RequestId}"); @@ -577,31 +576,14 @@ ParsedMessage ParseMessage(UnparsedMessage unparsed) stats.ParseMessageTracker.InsertRequest(parseStart, trackerMetadata); var applyTracker = stats.ApplyMessageQueueTracker.StartTrackingRequest(trackerMetadata); - return new ParsedMessage { message = message, dbOps = dbOps, receiveTimestamp = unparsed.timestamp, applyQueueTrackerId = applyTracker, reducerEvent = reducerEvent, procedureEvent = procedureEvent }; + return new ParsedMessage { generation = unparsed.generation, reducerStatus = reducerStatus, message = message, dbOps = dbOps, receiveTimestamp = unparsed.timestamp, applyQueueTrackerId = applyTracker }; } } public void Disconnect() { - isClosing = true; - connectionClosed = true; - FailPendingOperations(new OperationCanceledException("Connection closed.")); - - // Only try to close if the connection is active - if (webSocket.IsConnected) - { - webSocket.Close(); - } -#if UNITY_WEBGL && !UNITY_EDITOR - else if (webSocket.IsConnecting) -#else - else if (webSocket.IsConnecting || webSocket.IsNoneState) -#endif - { - webSocket.Abort(); // forceful during connecting - } - - _parseCancellationTokenSource.Cancel(); + EndConnection(); + onDisconnect?.Invoke(null, null); } /// @@ -626,14 +608,7 @@ public void Disconnect() /// void IDbConnection.Connect(string? token, string uri, string addressOrName, Compression compression, bool light, bool? confirmedReads) { - isClosing = false; - connectionClosed = false; - Identity = null; - initialConnectionId = null; - onConnectInvoked = false; - while (_parseQueue.TryTake(out _)) { } - while (_applyQueue.TryTake(out _)) { } - + retainedToken = token; uri = uri.Replace("http://", "ws://"); uri = uri.Replace("https://", "wss://"); if (!uri.StartsWith("ws://") && !uri.StartsWith("wss://")) @@ -644,36 +619,9 @@ void IDbConnection.Connect(string? token, string uri, string addressOrName, Comp // like `/foo` and then end up with `//` in the URI. uri = uri.TrimEnd('/'); - Log.Info($"SpacetimeDBClient: Connecting to {uri} {addressOrName}"); - if (!IsTesting) - { -#if UNITY_WEBGL && !UNITY_EDITOR - async Task Function() -#else - Task.Run(async () => -#endif - { - try - { - await webSocket.Connect(token, uri, addressOrName, ConnectionId, compression, light, confirmedReads); - } - catch (Exception e) - { - if (connectionClosed) - { - Log.Info("Connection closed gracefully."); - return; - } - - Log.Exception(e); - } -#if UNITY_WEBGL && !UNITY_EDITOR - } - _ = Function(); -#else - }); -#endif - } + connectionOptions = (uri, addressOrName, compression, light, confirmedReads); + preparingReplay = automaticReconnect; + StartSocket(); } @@ -700,6 +648,22 @@ private void ApplyUpdate(IEventContext eventContext, ParsedDatabaseUpdate dbOps) private void ApplyMessage(ParsedMessage parsed) { + if (parsed.generation != socketGeneration || isClosing) return; + if (parsed.action != null) + { + parsed.action(); + return; + } + if (parsed.error != null) + { + HandleSocketFailure(new ConnectionProtocolException("Could not parse server message.", parsed.error)); + return; + } + if (automaticReconnect && !onConnectInvoked && parsed.message is not ServerMessage.InitialConnection) + { + HandleSocketFailure(new ConnectionProtocolException("Expected InitialConnection.")); + return; + } var message = parsed.message; var dbOps = parsed.dbOps; var timestamp = parsed.receiveTimestamp; @@ -709,6 +673,9 @@ private void ApplyMessage(ParsedMessage parsed) switch (message) { + case ServerMessage.SubscribeBatchApplied(var batch): + ApplyReplayBatch(batch, dbOps); + break; case ServerMessage.SubscribeApplied(var subscribeApplied): { var eventContext = MakeSubscriptionEventContext(); @@ -753,7 +720,7 @@ private void ApplyMessage(ParsedMessage parsed) Log.Exception(e); } - subscriptions.Remove(subscriptionError.QuerySetId.Id); + RemoveSubscription(subscriptionError.QuerySetId.Id); } else { @@ -780,7 +747,7 @@ private void ApplyMessage(ParsedMessage parsed) } } - subscriptions.Remove(unsubscribeApplied.QuerySetId.Id); + RemoveSubscription(unsubscribeApplied.QuerySetId.Id); } break; @@ -792,8 +759,11 @@ private void ApplyMessage(ParsedMessage parsed) } case ServerMessage.ReducerResult(var reducerResult): { - if (parsed.reducerEvent is { } reducerEvent) + if (pendingReducerCalls.TryRemove(reducerResult.RequestId, out var pending)) { + var reducerEvent = new ReducerEvent( + (DateTimeOffset)reducerResult.Timestamp, parsed.reducerStatus!, + Identity ?? default, ConnectionId, null, pending.Reducer); var legacyEventContext = ToEventContext(new Event.Reducer(reducerEvent)); ApplyUpdate(legacyEventContext, dbOps); var eventContext = ToReducerEventContext(reducerEvent); @@ -801,49 +771,22 @@ private void ApplyMessage(ParsedMessage parsed) } else { - var legacyEventContext = ToEventContext(new Event.UnknownTransaction()); - ApplyUpdate(legacyEventContext, dbOps); + HandleSocketFailure(new ConnectionProtocolException($"Reducer result for unknown request_id {reducerResult.RequestId}.")); } break; } case ServerMessage.InitialConnection(var initialConnection): - try - { - if (Identity is Identity identity && identity != initialConnection.Identity) - { - throw new InvalidOperationException( - $"Received InitialConnection with unexpected identity. Previous={identity}, New={initialConnection.Identity}" - ); - } - - if (initialConnectionId is ConnectionId connectionId - && connectionId != initialConnection.ConnectionId) - { - throw new InvalidOperationException( - $"Received InitialConnection with unexpected connection_id. Previous={connectionId}, New={initialConnection.ConnectionId}" - ); - } - - Identity = initialConnection.Identity; - initialConnectionId = initialConnection.ConnectionId; - if (!onConnectInvoked) - { - onConnectInvoked = true; - onConnect?.Invoke(initialConnection.Identity, initialConnection.Token); - onConnect = null; - } - } - catch (Exception e) - { - Log.Exception(e); - } + HandleInitialConnection(initialConnection); break; - case ServerMessage.OneOffQueryResult: - /* OneOffQuery is async and handles its own responses */ + case ServerMessage.OneOffQueryResult(var result): + if (waitingOneOffQueries.TryRemove(result.RequestId, out var completion)) + completion.TrySetResult(result); break; case ServerMessage.ProcedureResult(var procedureResult): - var procedureEventContext = ToProcedureEventContext(parsed.procedureEvent!); + var procedureEventContext = ToProcedureEventContext(new ProcedureEvent( + procedureResult.Timestamp, procedureResult.Status, Identity ?? default, + ConnectionId, procedureResult.TotalHostExecutionDuration, procedureResult.RequestId)); if (!procedureCallbacks.TryResolveCallback(procedureEventContext, procedureResult.RequestId, procedureResult)) { Log.Warn($"Received ProcedureResult for unknown request ID: {procedureResult.RequestId}"); @@ -859,11 +802,12 @@ private void ApplyMessage(ParsedMessage parsed) // Note: this method is called from unit tests. internal void OnMessageReceived(byte[] bytes, DateTime timestamp) { - _parseQueue.Add(new UnparsedMessage { bytes = bytes, timestamp = timestamp, parseQueueTrackerId = stats.ParseMessageQueueTracker.StartTrackingRequest() }); + EnqueueMessage(bytes, timestamp, socketGeneration); } void IDbConnection.InternalCallReducer(T args) { + if (automaticReconnect && !IsActive) throw new InvalidOperationException("Not connected to server."); if (!webSocket.IsConnected) { Log.Error("Cannot call reducer, not connected to server!"); @@ -897,6 +841,7 @@ void IDbConnection.InternalCallProcedure( TArgs args, ProcedureCallback callback) { + if (automaticReconnect && !IsActive) throw new InvalidOperationException("Not connected to server."); if (!webSocket.IsConnected) { Log.Error("Cannot call procedure, not connected to server!"); @@ -916,25 +861,17 @@ void IDbConnection.InternalCallProcedure( QuerySetId? IDbConnection.Subscribe(ISubscriptionHandle handle, string[] querySqls) { - if (!webSocket.IsConnected) + if (!automaticReconnect && !webSocket.IsConnected) { Log.Error("Cannot subscribe, not connected to server!"); return null; } - - var id = stats.SubscriptionRequestTracker.StartTrackingRequest(); - // We use a distinct ID from the request ID as a sanity check that we're not - // casting request IDs to query IDs anywhere in the new code path. + if (isClosing) throw new InvalidOperationException("Connection closed."); var querySetId = querySetIdAllocator.Next(); subscriptions[querySetId] = handle; - webSocket.Send(new ClientMessage.Subscribe( - new Subscribe - { - RequestId = id, - QuerySetId = new QuerySetId(querySetId), - QueryStrings = querySqls.ToList(), - } - )); + subscriptionQueries[querySetId] = (string[])querySqls.Clone(); + if (!automaticReconnect || (IsActive && !preparingReplay)) + SendSubscription(querySetId); return new QuerySetId(querySetId); } @@ -945,7 +882,7 @@ void IDbConnection.InternalCallProcedure( async Task IDbConnection.RemoteQuery(string query) { - if (!webSocket.IsConnected) + if (!IsActive) { var error = "Cannot run one-off query, not connected to server!"; Log.Error(error); @@ -1018,7 +955,7 @@ T[] LogAndThrow(string error) return output; } - public bool IsActive => webSocket.IsConnected; + public bool IsActive => !connectionClosed && webSocket.IsConnected && (!automaticReconnect || onConnectInvoked); public void FrameTick() { @@ -1027,6 +964,7 @@ public void FrameTick() { ApplyMessage(parsedMessage); } + TickReconnect(); } void IDbConnection.Unsubscribe(QuerySetId queryId) @@ -1036,6 +974,12 @@ void IDbConnection.Unsubscribe(QuerySetId queryId) Log.Warn($"Unsubscribing from a subscription that the DbConnection does not know about, with QuerySetId {queryId.Id}"); } + unsubscribeRequested.Add(queryId.Id); + if (automaticReconnect && (!IsActive || preparingReplay)) + { + EndSubscription(queryId.Id); + return; + } var requestId = stats.SubscriptionRequestTracker.StartTrackingRequest(); webSocket.Send(new ClientMessage.Unsubscribe(new() @@ -1049,9 +993,9 @@ void IDbConnection.Unsubscribe(QuerySetId queryId) void IDbConnection.AddOnConnect(Action cb) => onConnect += cb; - void IDbConnection.AddOnConnectError(WebSocket.ConnectErrorEventHandler cb) => webSocket.OnConnectError += cb; + void IDbConnection.AddOnConnectError(Action cb) => onConnectError += cb; - void IDbConnection.AddOnDisconnect(WebSocket.CloseEventHandler cb) => webSocket.OnClose += cb; + void IDbConnection.AddOnDisconnect(Action cb) => onDisconnect += cb; } /// diff --git a/sdks/csharp/src/Table.cs b/sdks/csharp/src/Table.cs index 680dfc11b52..054b2707561 100644 --- a/sdks/csharp/src/Table.cs +++ b/sdks/csharp/src/Table.cs @@ -37,6 +37,7 @@ public interface IRemoteTableHandle /// /// An representing the parsed update. internal IParsedTableUpdate MakeParsedTableUpdate(); + internal void AddSnapshotDeletes(ParsedDatabaseUpdate update); /// /// Parses an insert-only table update and applies the results to the specified parsed database update. @@ -294,6 +295,17 @@ IParsedTableUpdate IRemoteTableHandle.MakeParsedTableUpdate() return new ParsedTableUpdate(); } + void IRemoteTableHandle.AddSnapshotDeletes(ParsedDatabaseUpdate update) + { + if (IsEventTable) return; + var delta = ((ParsedTableUpdate)update.UpdateForTable(this)).Delta; + foreach (var entry in Entries.Entries) + { + for (var i = 0u; i < Entries.Multiplicity(entry.Key); i++) + delta.Remove(entry.Key, entry.Value); + } + } + /// /// Parses an insert-only table update and applies the results to the specified parsed database update. /// diff --git a/sdks/csharp/src/WebSocket.cs b/sdks/csharp/src/WebSocket.cs index 97703c6b716..ff8cc73e760 100644 --- a/sdks/csharp/src/WebSocket.cs +++ b/sdks/csharp/src/WebSocket.cs @@ -3,6 +3,7 @@ using System; using System.Collections.Concurrent; +using System.Collections.Generic; using System.Net.Sockets; using System.Net.WebSockets; using System.Runtime.InteropServices; @@ -23,6 +24,35 @@ internal class WebSocket public delegate void ConnectErrorEventHandler(Exception e); public delegate void SendErrorEventHandler(Exception e); + internal class ConnectException : SpacetimeDBException + { + internal int StatusCode { get; } + internal ConnectException(int statusCode, Exception? inner = null) + : base($"WebSocket handshake failed with HTTP status {statusCode}.", inner) => StatusCode = statusCode; + } + + private static Exception ClassifyConnectError(WebSocketException error) + { + // .NET Standard does not expose the handshake response status directly. + var match = System.Text.RegularExpressions.Regex.Match(error.Message, @"status code '(\d{3})'"); + if (match.Success) return new ConnectException(int.Parse(match.Groups[1].Value), error); + return error.WebSocketErrorCode is WebSocketError.UnsupportedProtocol or WebSocketError.NotAWebSocket or WebSocketError.HeaderError + ? new ConnectionProtocolException("Invalid WebSocket handshake.", error) : error; + } + + internal class CloseException : SpacetimeDBException + { + internal int Code { get; } + internal CloseException(int code, string? reason) : base($"WebSocket closed ({code}): {reason}") => Code = code; + } + + internal static Exception? CloseError(int code, string? reason) => code switch + { + 1000 => null, + 1002 or 1003 or 1007 or 1008 => new ConnectionProtocolException($"WebSocket closed ({code}): {reason}"), + _ => new CloseException(code, reason) + }; + public struct ConnectOptions { public string Protocol; @@ -62,10 +92,10 @@ public WebSocket(ConnectOptions options) private bool _isConnected = false; private bool _isConnecting = false; private bool _cancelConnectRequested = false; - public bool IsConnected => _isConnected; + public virtual bool IsConnected => _isConnected; public bool IsConnecting => _isConnecting; #else - public bool IsConnected { get { return Ws != null && Ws.State == WebSocketState.Open; } } + public virtual bool IsConnected { get { return Ws != null && Ws.State == WebSocketState.Open; } } public bool IsConnecting { get { return Ws != null && Ws.State == WebSocketState.Connecting; } } public bool IsNoneState { get { return Ws != null && Ws.State == WebSocketState.None; } } #endif @@ -80,7 +110,7 @@ IntPtr errorCallback ); [DllImport("__Internal")] - private static extern int WebSocket_Connect(string host, string uri, string protocol, string authToken, IntPtr callbackPtr); + private static extern int WebSocket_Connect(string host, string uri, string protocol, string? authToken, int requestId, IntPtr callbackPtr); [DllImport("__Internal")] private static extern int WebSocket_Send(int socketId, byte[] data, int length); @@ -91,7 +121,7 @@ IntPtr errorCallback [AOT.MonoPInvokeCallback(typeof(Action))] private static void WebGLOnOpen(int socketId) { - Instance?.HandleWebGLOpen(socketId); + if (webglSockets.TryGetValue(socketId, out var socket)) socket.HandleWebGLOpen(socketId); } [AOT.MonoPInvokeCallback(typeof(Action))] @@ -100,7 +130,7 @@ private static void WebGLOnMessage(int socketId, IntPtr dataPtr, int length) try { byte[] data = new byte[length]; Marshal.Copy(dataPtr, data, 0, length); - Instance?.HandleWebGLMessage(socketId, data); + if (webglSockets.TryGetValue(socketId, out var socket)) socket.HandleWebGLMessage(socketId, data); } catch (Exception e) { UnityEngine.Debug.LogError($"Error handling message: {e}"); } @@ -111,7 +141,8 @@ private static void WebGLOnClose(int socketId, int code, IntPtr reasonPtr) { try { string reason = Marshal.PtrToStringUTF8(reasonPtr); - Instance?.HandleWebGLClose(socketId, code, reason); + if (webglSockets.TryGetValue(socketId, out var socket)) socket.HandleWebGLClose(socketId, code, reason); + webglSockets.Remove(socketId); } catch (Exception e) { UnityEngine.Debug.LogError($"Error handling close: {e}"); } @@ -120,22 +151,30 @@ private static void WebGLOnClose(int socketId, int code, IntPtr reasonPtr) [AOT.MonoPInvokeCallback(typeof(Action))] private static void WebGLOnError(int socketId) { - Instance?.HandleWebGLError(socketId); + if (webglSockets.TryGetValue(socketId, out var socket)) socket.HandleWebGLError(socketId); } - [AOT.MonoPInvokeCallback(typeof(Action))] - private static void OnSocketIdReceived(int socketId) + [AOT.MonoPInvokeCallback(typeof(Action))] + private static void OnSocketIdReceived(int requestId, int socketId) { - Instance?._socketId.TrySetResult(socketId); + if (!webglRequests.TryGetValue(requestId, out var socket)) return; + webglRequests.Remove(requestId); + if (socketId >= 0) + { + socket._webglSocketId = socketId; + webglSockets[socketId] = socket; + } + socket._socketId.TrySetResult(socketId); } - private static WebSocket Instance; + private static readonly Dictionary webglSockets = new(); + private static readonly Dictionary webglRequests = new(); + private static int nextWebglRequest; private int _webglSocketId = -1; - private TaskCompletionSource _socketId; + private TaskCompletionSource _socketId = null!; private void InitializeWebGL() { - Instance = this; // Convert callbacks to function pointers var openPtr = Marshal.GetFunctionPointerForDelegate((Action)WebGLOnOpen); var messagePtr = Marshal.GetFunctionPointerForDelegate((Action)WebGLOnMessage); @@ -146,7 +185,7 @@ private void InitializeWebGL() } #endif - public async Task Connect(string? auth, string host, string nameOrAddress, ConnectionId connectionId, Compression compression, bool light, bool? confirmedReads) + public virtual async Task Connect(string? auth, string host, string nameOrAddress, ConnectionId connectionId, Compression compression, bool light, bool? confirmedReads, ConnectionId? sessionId = null) { #if UNITY_WEBGL && !UNITY_EDITOR if (_isConnecting || _isConnected) return; @@ -156,6 +195,7 @@ public async Task Connect(string? auth, string host, string nameOrAddress, Conne try { var uri = $"{host}/v1/database/{nameOrAddress}/subscribe?connection_id={connectionId}&compression={compression}"; + if (sessionId.HasValue) uri += $"&session_id={sessionId.Value}"; if (light) uri += "&light=true"; if (confirmedReads.HasValue) { @@ -165,13 +205,17 @@ public async Task Connect(string? auth, string host, string nameOrAddress, Conne } _socketId = new TaskCompletionSource(); - var callbackPtr = Marshal.GetFunctionPointerForDelegate((Action)OnSocketIdReceived); - WebSocket_Connect(host, uri, _options.Protocol, auth, callbackPtr); + var callbackPtr = Marshal.GetFunctionPointerForDelegate((Action)OnSocketIdReceived); + var requestId = ++nextWebglRequest; + webglRequests[requestId] = this; + WebSocket_Connect(host, uri, _options.Protocol, auth, requestId, callbackPtr); _webglSocketId = await _socketId.Task; - if (_webglSocketId == -1) + if (_webglSocketId < 0) { - dispatchQueue.Enqueue(() => OnConnectError?.Invoke( - new Exception("Failed to connect WebSocket"))); + var statusCode = -_webglSocketId; + dispatchQueue.Enqueue(() => OnConnectError?.Invoke(statusCode == 1 + ? new Exception("Failed to connect WebSocket") : new ConnectException(statusCode))); + _webglSocketId = -1; } else if (_cancelConnectRequested) { @@ -190,6 +234,7 @@ public async Task Connect(string? auth, string host, string nameOrAddress, Conne // Events will be handled via UnitySendMessage callbacks #else var uri = $"{host}/v1/database/{nameOrAddress}/subscribe?connection_id={connectionId}&compression={compression}"; + if (sessionId.HasValue) uri += $"&session_id={sessionId.Value}"; if (light) { uri += "&light=true"; @@ -233,49 +278,9 @@ public async Task Connect(string? auth, string host, string nameOrAddress, Conne return; } } - catch (WebSocketException ex) when (ex.WebSocketErrorCode == WebSocketError.Success) - { - // How can we get here: - // - When you go to connect and the server isn't running (port closed) - target machine actively refused - // - 404 - No module with at that module address instead of 101 upgrade - // - 401? - When the identity received by SpacetimeDB wasn't signed by its signing key - // - 400 - When the auth is malformed - if (OnConnectError != null) - { - // .net 6,7,8 has support for Ws.HttpStatusCode as long as you set - // ClientWebSocketOptions.CollectHttpResponseDetails = true - var message = "A WebSocketException occurred, even though the WebSocketErrorCode is \"Success\".\n" - + "This indicates that there was no native error information for the exception.\n" - + "Due to limitations in the .NET core version we do not have access to the HTTP status code returned by the request which would provide more info on the nature of the error.\n\n" - + "This error could arise for a number of reasons:\n" - + "1. The target machine actively refused the connection.\n" - + "2. The module you are trying to connect to does not exist (404 NOT FOUND).\n" - + "3. The auth token you sent to SpacetimeDB was not signed by the correct signing key (400 BAD REQUEST).\n" - + "4. The auth token is malformed (400 BAD REQUEST).\n" - + "5. You are not authorized (401 UNAUTHORIZED).\n\n" - + "Did you forget to start the server or publish your module?\n\n" - + "Here are some values that might help you debug:\n" - + $"Message: {ex.Message}\n" - + $"WebSocketErrorCode: {ex.WebSocketErrorCode}\n" - + $"ErrorCode: {ex.ErrorCode}\n" - + $"NativeErrorCode: {ex.NativeErrorCode}\n" - + $"InnerException Message: {ex.InnerException?.Message}\n" - + $"WebSocket CloseStatus: {Ws.CloseStatus}\n" - + $"WebSocket State: {Ws.State}\n" - + $"InnerException: {ex.InnerException}\n" - + $"Exception: {ex}" - ; - dispatchQueue.Enqueue(() => OnConnectError(new Exception(message))); - } - } catch (WebSocketException ex) { - if (OnConnectError != null) - { - var message = $"WebSocket connection failed: {ex.WebSocketErrorCode}\n" - + $"Exception message: {ex.Message}\n"; - dispatchQueue.Enqueue(() => OnConnectError(new Exception(message))); - } + dispatchQueue.Enqueue(() => OnConnectError?.Invoke(ClassifyConnectError(ex))); } catch (SocketException ex) { @@ -308,42 +313,8 @@ await Ws.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, } if (OnClose != null) { - switch (receiveResult.CloseStatus) - { - case WebSocketCloseStatus.NormalClosure: - dispatchQueue.Enqueue(() => OnClose(null)); - break; - case WebSocketCloseStatus.EndpointUnavailable: - dispatchQueue.Enqueue(() => OnClose(new Exception("(1000) The connection has closed after the request was fulfilled."))); - break; - case WebSocketCloseStatus.ProtocolError: - dispatchQueue.Enqueue(() => OnClose(new Exception("(1002) The client or server is terminating the connection because of a protocol error."))); - break; - case WebSocketCloseStatus.InvalidMessageType: - dispatchQueue.Enqueue(() => OnClose(new Exception("(1003) The client or server is terminating the connection because it cannot accept the data type it received."))); - break; - case WebSocketCloseStatus.Empty: - dispatchQueue.Enqueue(() => OnClose(new Exception("(1005) No error specified."))); - break; - case WebSocketCloseStatus.InvalidPayloadData: - dispatchQueue.Enqueue(() => OnClose(new Exception("(1007) The client or server is terminating the connection because it has received data inconsistent with the message type."))); - break; - case WebSocketCloseStatus.PolicyViolation: - dispatchQueue.Enqueue(() => OnClose(new Exception("(1008) The connection will be closed because an endpoint has received a message that violates its policy."))); - break; - case WebSocketCloseStatus.MessageTooBig: - dispatchQueue.Enqueue(() => OnClose(new Exception("(1009) Message too big"))); - break; - case WebSocketCloseStatus.MandatoryExtension: - dispatchQueue.Enqueue(() => OnClose(new Exception("(1010) The client is terminating the connection because it expected the server to negotiate an extension."))); - break; - case WebSocketCloseStatus.InternalServerError: - dispatchQueue.Enqueue(() => OnClose(new Exception("(1011) The connection will be closed by the server because of an error on the server."))); - break; - default: - dispatchQueue.Enqueue(() => OnClose(new Exception("Unknown error"))); - break; - } + var error = CloseError((int?)receiveResult.CloseStatus ?? 1006, receiveResult.CloseStatusDescription); + dispatchQueue.Enqueue(() => OnClose(error)); } return; } @@ -360,7 +331,7 @@ await Ws.CloseAsync(WebSocketCloseStatus.MessageTooBig, closeMessage, CancellationToken.None); if (OnClose != null) { - dispatchQueue.Enqueue(() => OnClose(new Exception("(1009) Message too big"))); + dispatchQueue.Enqueue(() => OnClose(new ConnectionProtocolException("(1009) Message too big"))); } return; } @@ -380,7 +351,7 @@ await Ws.CloseAsync(WebSocketCloseStatus.MessageTooBig, closeMessage, OnMessage(message, startReceive); } } - catch (WebSocketException ex) + catch (Exception ex) { if (OnClose != null) dispatchQueue.Enqueue(() => OnClose(ex)); return; @@ -403,7 +374,7 @@ public void CancelConnect() #endif } - public Task Close(WebSocketCloseStatus code = WebSocketCloseStatus.NormalClosure) + public virtual Task Close(WebSocketCloseStatus code = WebSocketCloseStatus.NormalClosure) { #if UNITY_WEBGL && !UNITY_EDITOR if (_webglSocketId >= 0) @@ -448,7 +419,7 @@ private void EnsureReceiveCapacity(int minimumCapacity) /// Forcefully abort the WebSocket connection. This terminates any in-flight connect/receive/send /// and ensures the server-side socket is torn down promptly. Prefer Close() for graceful shutdowns. /// - public void Abort() + public virtual void Abort() { #if UNITY_WEBGL && !UNITY_EDITOR if (_webglSocketId >= 0) @@ -464,7 +435,10 @@ public void Abort() #else try { + CancelConnect(); Ws?.Abort(); + Ws?.Dispose(); + while (messageSendQueue.TryDequeue(out _)) { } } catch { @@ -482,14 +456,15 @@ public void Abort() /// before we start another one. This function is also thread safe, just in case. /// /// The message to send - public void Send(ClientMessage message) + public virtual void Send(ClientMessage message) { #if UNITY_WEBGL && !UNITY_EDITOR try { var messageBSATN = new ClientMessage.BSATN(); var encodedMessage = IStructuralReadWrite.ToBytes(messageBSATN, message); - WebSocket_Send(_webglSocketId, encodedMessage, encodedMessage.Length); + if (WebSocket_Send(_webglSocketId, encodedMessage, encodedMessage.Length) != 0) + throw new InvalidOperationException("WebSocket send failed."); } catch (Exception e) { @@ -577,11 +552,7 @@ public void HandleWebGLClose(int socketId, int code, string reason) _isConnecting = false; _webglSocketId = -1; _cancelConnectRequested = false; - if (ReferenceEquals(Instance, this)) - { - Instance = null; - } - var ex = code != (int)WebSocketCloseStatus.NormalClosure ? new Exception($"WebSocket closed with code {code}: {reason}") : null; + var ex = CloseError(code, reason); dispatchQueue.Enqueue(() => OnClose?.Invoke(ex)); } } @@ -592,7 +563,6 @@ public void HandleWebGLError(int socketId) if (socketId == _webglSocketId && OnConnectError != null) { _isConnecting = false; - _webglSocketId = -1; dispatchQueue.Enqueue(() => OnConnectError(new Exception($"Socket {socketId} error."))); } } diff --git a/sdks/csharp/tests~/README.md b/sdks/csharp/tests~/README.md index 95f9a400393..43d2b466bca 100644 --- a/sdks/csharp/tests~/README.md +++ b/sdks/csharp/tests~/README.md @@ -18,3 +18,13 @@ $ ( cd "${SPACETIMEDB_REPO_PATH}"/crates/bindings-csharp/BSATN.Runtime && dotnet $ dotnet nuget locals all --clear $ dotnet test ``` + +# Reconnect coverage + +`ReconnectTests` uses the existing xUnit framework, serialized BSATN messages, a fake transport, and a controllable retry clock. Run it with: + +```sh +dotnet test --filter FullyQualifiedName~ReconnectTests +``` + +For live server coverage, including a fault-injection proxy and short-lived JWT rotation, use the [reconnect test application](../examples~/reconnect/README.md). diff --git a/sdks/csharp/tests~/ReconnectTests.cs b/sdks/csharp/tests~/ReconnectTests.cs new file mode 100644 index 00000000000..1aff5bff7ee --- /dev/null +++ b/sdks/csharp/tests~/ReconnectTests.cs @@ -0,0 +1,741 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Reflection; +using System.Text; +using SpacetimeDB; +using SpacetimeDB.BSATN; +using SpacetimeDB.ClientApi; +using Xunit; + +namespace SpacetimeDB.Tests; + +[CollectionDefinition("SDK connections", DisableParallelization = true)] +public class ConnectionTestCollection { } + +[Collection("SDK connections")] +public partial class ReconnectTests +{ + [SpacetimeDB.Type] + public partial class Row + { + public uint Id; + public string Value = ""; + } + + [SpacetimeDB.Type] + public partial class Args : IReducerArgs, IProcedureArgs + { + string IReducerArgs.ReducerName => "test"; + string IProcedureArgs.ProcedureName => "test"; + } + + public sealed class Context : IEventContext, ISubscriptionEventContext, IErrorContext, IReducerEventContext, IProcedureEventContext + { + public Exception Event { get; init; } = new Exception(); + public ReducerEvent? ReducerEvent; + } + + public sealed class TestTable : RemoteTableHandle + { + private readonly string name; + private readonly bool primaryKey; + public override string RemoteTableName => name; + protected override object? GetPrimaryKey(Row row) => primaryKey ? row.Id : null; + public TestTable(IDbConnection conn, string name, bool primaryKey) : base(conn) + { + this.name = name; + this.primaryKey = primaryKey; + } + } + + public sealed class Tables : RemoteTablesBase + { + public readonly TestTable Keyed; + public readonly TestTable Unkeyed; + public Tables(IDbConnection conn) + { + AddTable(Keyed = new(conn, "keyed", true)); + AddTable(Unkeyed = new(conn, "unkeyed", false)); + } + } + + public sealed class Connection : DbConnectionBase, IDisposable + { + public override Tables Db { get; } + internal readonly List Sockets = new(); + internal readonly List<(string Kind, Exception? Error, NextReconnect? Next)> Events = new(); + internal readonly List ReducerResults = new(); + internal double Now; + internal FakeSocket Socket => Sockets[^1]; + internal int Connects; + internal int CallbackThread; + + public Connection() + { + Db = new(this); + SocketFactory = () => + { + var socket = new FakeSocket(); + Sockets.Add(socket); + return socket; + }; + ReconnectClock = () => Now; + ((IDbConnection)this).AddOnConnect((_, _) => { Connects++; CallbackThread = Environment.CurrentManagedThreadId; }); + ((IDbConnection)this).AddOnDisconnect((error, next) => Events.Add(("disconnect", error, next))); + ((IDbConnection)this).AddOnConnectError((error, next) => Events.Add(("error", error, next))); + } + + protected override IEventContext ToEventContext(Event e) => new Context(); + protected override IReducerEventContext ToReducerEventContext(ReducerEvent e) => new Context { ReducerEvent = e }; + protected override ISubscriptionEventContext MakeSubscriptionEventContext() => new Context(); + protected override IErrorContext ToErrorContext(Exception e) => new Context { Event = e }; + protected override IProcedureEventContext ToProcedureEventContext(ProcedureEvent e) => new Context(); + protected override bool Dispatch(IReducerEventContext context, Args reducer) + { + ReducerResults.Add(((Context)context).ReducerEvent!.Status); + return true; + } + public void Dispose() => Disconnect(); + } + + internal sealed class FakeSocket : SpacetimeDB.WebSocket + { + internal bool Open; + internal readonly ConcurrentQueue Sent = new(); + internal string? Token; + internal ConnectionId? Session; + internal ConnectionId Id; + internal volatile bool Started; + public override bool IsConnected => Open; + internal FakeSocket() : base(new ConnectOptions { Protocol = "v2.bsatn.spacetimedb" }) { } + public override Task Connect(string? auth, string host, string database, ConnectionId id, Compression compression, + bool light, bool? confirmed, ConnectionId? sessionId = null) + { + Token = auth; + Session = sessionId; + Id = id; + Started = true; + return Task.CompletedTask; + } + public override void Send(ClientMessage message) => Sent.Enqueue(message); + public override void Abort() => Open = false; + private T Handler(string name) where T : Delegate => + (T)typeof(SpacetimeDB.WebSocket).GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(this)!; + internal void Receive(ServerMessage message) + { + var bytes = IStructuralReadWrite.ToBytes(new ServerMessage.BSATN(), message); + Handler("OnMessage")([0, .. bytes], DateTime.UtcNow); + } + internal void Lose(Exception? error = null) + { + Open = false; + Handler("OnClose")(error); + } + internal void Fail(Exception error) => Handler("OnConnectError")(error); + } + + private sealed class Handle : SubscriptionHandleBase + { + internal Handle(Connection conn, string query = "SELECT * FROM keyed") : base(conn, null, null, [query]) { } + } + + private sealed class TrackingHandle : ISubscriptionHandle + { + internal int Applied; + internal int Errors; + internal int Ended; + internal QuerySetId Id = new(); + public void RebindQuerySetId(QuerySetId id) => Id = id; + public void OnApplied(ISubscriptionEventContext ctx) => Applied++; + public void OnError(IErrorContext ctx) => Errors++; + public void OnEnded(ISubscriptionEventContext ctx) => Ended++; + } + + private static Connection Create(bool automatic = true, string? token = null, Func>? provider = null) + { + var builder = Connection.Builder().WithUri("ws://localhost").WithDatabaseName("test").WithToken(token); + if (automatic) builder.WithAutomaticReconnect(); + if (provider != null) builder.WithTokenProvider(provider); + var conn = builder.Build(); + Pump(conn, () => conn.Socket.Started); + return conn; + } + + private static void Pump(Connection conn, Func done) + { + var timer = Stopwatch.StartNew(); + do + { + conn.FrameTick(); + if (done()) return; + Thread.Sleep(1); + } while (timer.Elapsed < TimeSpan.FromSeconds(5)); + Assert.Fail("Timed out waiting for SDK work."); + } + + private static readonly Identity identity = Identity.From(new byte[32]); + private static void Establish(Connection conn, string token = "issued-token", Identity? asIdentity = null) + { + var count = conn.Connects; + var errors = conn.Events.Count; + conn.Socket.Open = true; + conn.Socket.Receive(new ServerMessage.InitialConnection(new(asIdentity ?? identity, conn.ConnectionId, token))); + Pump(conn, () => conn.Connects > count || conn.Events.Count > errors && conn.Events[^1].Next == null); + } + + private static void Drop(Connection conn) + { + var count = conn.Events.Count; + conn.Socket.Lose(new IOException("dropped")); + Pump(conn, () => conn.Events.Count > count); + } + + private static void Retry(Connection conn) + { + var count = conn.Sockets.Count; + conn.Now += 31; + Pump(conn, () => conn.Sockets.Count > count && conn.Socket.Started); + } + + private static TrackingHandle Subscribe(Connection conn, string query = "SELECT * FROM keyed") + { + var handle = new TrackingHandle(); + handle.Id = ((IDbConnection)conn).Subscribe(handle, [query])!; + return handle; + } + + private static BsatnRowList Rows(params Row[] rows) + { + var bytes = new List(); + var offsets = new List(); + foreach (var row in rows) + { + offsets.Add((ulong)bytes.Count); + bytes.AddRange(IStructuralReadWrite.ToBytes(row)); + } + return new(new RowSizeHint.RowOffsets(offsets), bytes); + } + + private static QueryRows Query(string table, params Row[] rows) => new(new() { new(table, Rows(rows)) }); + + private static void Apply(Connection conn, TrackingHandle handle, QueryRows rows) + { + var count = handle.Applied; + var sent = conn.Socket.Sent.OfType().Last().Subscribe_; + conn.Socket.Receive(new ServerMessage.SubscribeApplied(new(sent.RequestId, handle.Id, rows))); + Pump(conn, () => handle.Applied > count); + } + + private static SubscribeBatch Batch(Connection conn) => conn.Socket.Sent.OfType().Single().SubscribeBatch_; + [Fact] + public void ReusesIdentitySessionAndHandlesWithNewConnectionIds() + { + using var conn = Create(); + Establish(conn); + var socket = conn.Socket; + var table = conn.Db.Keyed; + var first = conn.ConnectionId; + var handle = Subscribe(conn); + Apply(conn, handle, Query("keyed", new Row { Id = 1, Value = "unchanged" })); + var oldId = handle.Id; + var inserts = 0; + var updates = 0; + var deletes = 0; + table.OnInsert += (_, _) => inserts++; + table.OnUpdate += (_, _, _) => updates++; + table.OnDelete += (_, _) => deletes++; + Drop(conn); + Assert.True(conn.IsReconnecting); + Assert.False(conn.IsActive); + Assert.Equal(1, table.Count); + Assert.Equal(1, conn.Events[^1].Next!.Value.Attempt); + Retry(conn); + Assert.Equal("issued-token", conn.Socket.Token); + Assert.Equal(socket.Session, conn.Socket.Session); + Assert.NotEqual(first, conn.ConnectionId); + Establish(conn); + Assert.NotEqual(oldId.Id, handle.Id.Id); + conn.Socket.Receive(new ServerMessage.SubscribeBatchApplied(new(Batch(conn).RequestId, + new() { new(handle.Id, new SubscribeSetOutcome.Applied(Query("keyed", new Row { Id = 1, Value = "unchanged" }))) }))); + Pump(conn, () => handle.Applied == 2); + Assert.Same(table, conn.Db.Keyed); + Assert.Equal(0, inserts + updates + deletes); + Assert.Equal(Environment.CurrentManagedThreadId, conn.CallbackThread); + Drop(conn); + Assert.Equal(1, conn.Events[^1].Next!.Value.Attempt); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void InitialFailureNeverRetries(bool automatic) + { + using var conn = Create(automatic); + conn.Socket.Fail(new IOException("unreachable")); + Pump(conn, () => conn.Events.Count == 1); + Assert.Equal("error", conn.Events[0].Kind); + Assert.Null(conn.Events[0].Next); + conn.Now += 100; + conn.FrameTick(); + Assert.Single(conn.Sockets); + Assert.False(conn.IsReconnecting); + } + + [Fact] + public void OptOutNeverReplaysOrSendsSessionId() + { + using var conn = Create(false); + Establish(conn); + Assert.Null(conn.Socket.Session); + Drop(conn); + Assert.Null(conn.Events[^1].Next); + Assert.False(conn.IsReconnecting); + } + + [Fact] + public void DuplicateLossAndStaleMessagesCannotAffectRetry() + { + using var conn = Create(); + Establish(conn); + var old = conn.Socket; + old.Lose(); + old.Fail(new IOException()); + Pump(conn, () => conn.Events.Count > 0); + Retry(conn); + old.Receive(new ServerMessage.InitialConnection(new(identity, old.Id, "wrong"))); + old.Lose(); + Establish(conn); + Assert.Equal(2, conn.Connects); + Assert.Single(conn.Events); + Assert.Equal("issued-token", conn.Socket.Token); + } + + [Fact] + public void RetriesBackOffAndCancelFromCallback() + { + using var conn = Create(); + Establish(conn); + Drop(conn); + Retry(conn); + conn.Socket.Fail(new IOException("offline")); + Pump(conn, () => conn.Events.Count == 2); + Assert.Equal("error", conn.Events[^1].Kind); + Assert.Equal(2, conn.Events[^1].Next!.Value.Attempt); + ((IDbConnection)conn).AddOnConnectError((_, _) => conn.Disconnect()); + Retry(conn); + conn.Socket.Fail(new IOException("offline")); + Pump(conn, () => conn.Events.Count == 4); + Assert.Equal("disconnect", conn.Events[^1].Kind); + Assert.Null(conn.Events[^1].Error); + Assert.Null(conn.Events[^1].Next); + Assert.False(conn.IsReconnecting); + } + + [Fact] + public void IdentityChangeAndProtocolErrorsAreTerminal() + { + using var conn = Create(); + Establish(conn); + Drop(conn); + Retry(conn); + var bytes = new byte[32]; + bytes[0] = 1; + Establish(conn, asIdentity: Identity.From(bytes)); + Assert.Equal(1, conn.Connects); + Assert.Equal("error", conn.Events[^1].Kind); + Assert.IsType(conn.Events[^1].Error); + Assert.Null(conn.Events[^1].Next); + Assert.Equal(identity, conn.Identity); + } + + [Fact] + public void InFlightCallsHaveUnknownResultsAndOfflineCallsFailFast() + { + using var conn = Create(); + Establish(conn); + Exception? procedureError = null; + ((IDbConnection)conn).InternalCallReducer(new Args()); + ((IDbConnection)conn).InternalCallProcedure(new(), (_, result) => procedureError = result.Error); + var query = ((IDbConnection)conn).RemoteQuery("SELECT * FROM keyed"); + Drop(conn); + Assert.IsType(Assert.Single(conn.ReducerResults)); + Assert.IsType(procedureError); + Assert.IsType(query.Exception!.InnerException); + Assert.Throws(() => ((IDbConnection)conn).InternalCallReducer(new Args())); + Assert.Throws(() => ((IDbConnection)conn).InternalCallProcedure(new(), (_, _) => { })); + Assert.IsType(((IDbConnection)conn).RemoteQuery("SELECT * FROM keyed").Exception!.InnerException); + } + + [Fact] + public void ReconciliationCombinesOverlapsNetChangesAndFailures() + { + using var conn = Create(); + Establish(conn); + var first = Subscribe(conn); + Apply(conn, first, Query("keyed", new Row { Id = 1, Value = "same" }, new Row { Id = 2, Value = "old" }, new Row { Id = 3 })); + var second = Subscribe(conn); + Apply(conn, second, Query("keyed", new Row { Id = 1, Value = "same" })); + var failed = Subscribe(conn); + Apply(conn, failed, Query("keyed", new Row { Id = 4 })); + var changes = new List(); + var callbackCounts = new List(); + conn.Db.Keyed.OnInsert += (_, row) => { callbackCounts.Add(conn.Db.Keyed.Count); changes.Add($"insert:{row.Id}"); }; + conn.Db.Keyed.OnUpdate += (_, old, row) => changes.Add($"update:{old.Value}:{row.Value}"); + conn.Db.Keyed.OnDelete += (_, row) => changes.Add($"delete:{row.Id}"); + Drop(conn); + Retry(conn); + Establish(conn); + var batch = Batch(conn); + Assert.Equal(3, batch.Sets.Count); + conn.Socket.Receive(new ServerMessage.SubscribeBatchApplied(new(batch.RequestId, new() + { + new(first.Id, new SubscribeSetOutcome.Applied(Query("keyed", new Row { Id = 1, Value = "same" }, new Row { Id = 2, Value = "new" }, new Row { Id = 5 }))), + new(second.Id, new SubscribeSetOutcome.Applied(Query("keyed", new Row { Id = 1, Value = "same" }))), + new(failed.Id, new SubscribeSetOutcome.Error("invalid query")) + }))); + Pump(conn, () => failed.Errors == 1); + Assert.Equal(new[] { "delete:3", "delete:4", "insert:5", "update:old:new" }, changes.OrderBy(x => x)); + Assert.All(callbackCounts, count => Assert.Equal(3, count)); + Assert.Equal(2, first.Applied); + Assert.Equal(2, second.Applied); + ((IDbConnection)conn).Unsubscribe(second.Id); + conn.Socket.Receive(new ServerMessage.UnsubscribeApplied(new(0, second.Id, Query("keyed", new Row { Id = 1, Value = "same" })))) ; + Pump(conn, () => second.Ended == 1); + Assert.Equal(3, conn.Db.Keyed.Count); + } + + [Fact] + public void QueuedAndCancelledSubscriptionsAreReflectedInReplay() + { + using var conn = Create(); + Establish(conn); + var old = Subscribe(conn); + Apply(conn, old, Query("keyed", new Row { Id = 1 })); + Drop(conn); + ((IDbConnection)conn).Unsubscribe(old.Id); + Assert.Equal(1, old.Ended); + var added = Subscribe(conn, "SELECT * FROM keyed WHERE id = 2"); + var cancelled = new Handle(conn); + cancelled.Unsubscribe(); + Assert.True(cancelled.IsEnded); + Retry(conn); + Establish(conn); + Assert.Single(Batch(conn).Sets); + conn.Socket.Receive(new ServerMessage.SubscribeBatchApplied(new(Batch(conn).RequestId, new() + { + new(added.Id, new SubscribeSetOutcome.Applied(Query("keyed", new Row { Id = 2 }))) + }))); + Pump(conn, () => added.Applied == 1); + Assert.Equal(2u, conn.Db.Keyed.Iter().Single().Id); + } + + [Fact] + public void EmptyReplayClearsCacheAndBadBatchIsTerminal() + { + using var conn = Create(); + Establish(conn); + var handle = Subscribe(conn); + Apply(conn, handle, Query("keyed", new Row { Id = 1 })); + Drop(conn); + ((IDbConnection)conn).Unsubscribe(handle.Id); + Retry(conn); + Establish(conn); + Assert.Equal(0, conn.Db.Keyed.Count); + Assert.Empty(conn.Socket.Sent); + conn.Socket.Receive(new ServerMessage.SubscribeBatchApplied(new(123, new()))); + Pump(conn, () => conn.Events.Count == 2); + Assert.Null(conn.Events[^1].Next); + Assert.IsType(conn.Events[^1].Error); + } + + [Fact] + public void TokenProviderRefreshesOnlyWhenNeededAndCanBeCancelled() + { + var providerCalls = 0; + var source = new TaskCompletionSource(); + using var conn = Create(provider: () => { providerCalls++; return source.Task; }); + Assert.Equal(0, providerCalls); + Establish(conn); + Drop(conn); + conn.Now += 31; + conn.FrameTick(); + Assert.Equal(1, providerCalls); + Assert.True(conn.IsReconnecting); + conn.Disconnect(); + source.SetResult("fresh-token"); + conn.FrameTick(); + Assert.Single(conn.Sockets); + } + + [Fact] + public void ProviderFailureRetriesAndFreshTokenRejectionStops() + { + var calls = 0; + using var conn = Create(provider: () => ++calls == 1 + ? Task.FromException(new IOException("provider unavailable")) : Task.FromResult("fresh-token")); + Establish(conn); + Drop(conn); + conn.Now += 31; + Pump(conn, () => conn.Events.Count == 2); + Assert.Equal(2, conn.Events[^1].Next!.Value.Attempt); + Retry(conn); + Assert.Equal("fresh-token", conn.Socket.Token); + conn.Socket.Fail(new SpacetimeDB.WebSocket.ConnectException(401)); + Pump(conn, () => conn.Events.Count == 3); + Assert.Null(conn.Events[^1].Next); + } + + [Fact] + public void RejectedRetainedTokenForcesRefreshDespiteDistantExpiry() + { + var calls = 0; + var token = Jwt(DateTimeOffset.UtcNow.ToUnixTimeSeconds() + 3600, DateTimeOffset.UtcNow.ToUnixTimeSeconds()); + using var conn = Create(token: token, provider: () => { calls++; return Task.FromResult("fresh-token"); }); + Establish(conn); + Drop(conn); + Retry(conn); + Assert.Equal(0, calls); + Assert.Equal(token, conn.Socket.Token); + conn.Socket.Fail(new SpacetimeDB.WebSocket.ConnectException(401)); + Pump(conn, () => conn.Events.Count == 2); + Retry(conn); + Assert.Equal(1, calls); + Assert.Equal("fresh-token", conn.Socket.Token); + Establish(conn); + Assert.Equal(2, conn.Connects); + } + + [Fact] + public void NoPrimaryKeyChangesProduceDeleteInsertAfterAtomicApply() + { + using var conn = Create(); + Establish(conn); + var keyed = Subscribe(conn); + Apply(conn, keyed, Query("keyed", new Row { Id = 1, Value = "old" })); + var unkeyed = Subscribe(conn, "SELECT * FROM unkeyed"); + Apply(conn, unkeyed, Query("unkeyed", new Row { Id = 1, Value = "old" })); + var changes = new List(); + string? observedOtherTable = null; + conn.Db.Keyed.OnUpdate += (_, _, _) => observedOtherTable = conn.Db.Unkeyed.Iter().Single().Value; + conn.Db.Unkeyed.OnDelete += (_, row) => changes.Add($"delete:{row.Value}"); + conn.Db.Unkeyed.OnInsert += (_, row) => changes.Add($"insert:{row.Value}"); + Drop(conn); + Retry(conn); + Establish(conn); + conn.Socket.Receive(new ServerMessage.SubscribeBatchApplied(new(Batch(conn).RequestId, new() + { + new(keyed.Id, new SubscribeSetOutcome.Applied(Query("keyed", new Row { Id = 1, Value = "new" }))), + new(unkeyed.Id, new SubscribeSetOutcome.Applied(Query("unkeyed", new Row { Id = 1, Value = "new" }))) + }))); + Pump(conn, () => unkeyed.Applied == 2); + Assert.Equal("new", observedOtherTable); + Assert.Equal(new[] { "delete:old", "insert:new" }, changes.OrderBy(x => x)); + } + + [Theory] + [InlineData("missing")] + [InlineData("duplicate")] + [InlineData("unknown")] + [InlineData("request")] + public void MalformedReplayNeverChangesTheCache(string kind) + { + using var conn = Create(); + Establish(conn); + var handle = Subscribe(conn); + Apply(conn, handle, Query("keyed", new Row { Id = 1 })); + Drop(conn); + Retry(conn); + Establish(conn); + var result = new SubscribeSetResult(handle.Id, new SubscribeSetOutcome.Applied(Query("keyed", new Row { Id = 2 }))); + var results = new List { result }; + if (kind == "missing") results.Clear(); + if (kind == "duplicate") results.Add(result); + if (kind == "unknown") result.QuerySetId = new(9999); + conn.Socket.Receive(new ServerMessage.SubscribeBatchApplied(new(Batch(conn).RequestId + (kind == "request" ? 1u : 0u), results))); + Pump(conn, () => conn.Events.Count == 2); + Assert.Null(conn.Events[^1].Next); + Assert.Equal(1u, conn.Db.Keyed.Iter().Single().Id); + Assert.Equal(1, handle.Applied); + } + + [Fact] + public void SessionBusyKeepsAttemptNumberAndFirstDelay() + { + using var conn = Create(); + Establish(conn); + Drop(conn); + Retry(conn); + conn.Socket.Lose(new SpacetimeDB.WebSocket.CloseException(4000, "Session busy")); + Pump(conn, () => conn.Events.Count == 2); + Assert.Equal(1, conn.Events[^1].Next!.Value.Attempt); + Assert.InRange(conn.Events[^1].Next!.Value.Delay.TotalMilliseconds, 500, 1500); + } + + [Fact] + public void DisconnectWhileWaitingCancelsTheTimer() + { + using var conn = Create(); + Establish(conn); + Drop(conn); + conn.Disconnect(); + conn.Now += 100; + conn.FrameTick(); + Assert.Single(conn.Sockets); + Assert.False(conn.IsReconnecting); + Assert.Null(conn.Events[^1].Next); + Assert.Null(conn.Events[^1].Error); + } + + [Fact] + public void ReconnectCallbackCanChangeSubscriptionsOrDisconnect() + { + using var conn = Create(); + Establish(conn); + var before = Subscribe(conn); + Apply(conn, before, Query("keyed", new Row { Id = 1 })); + Drop(conn); + ((IDbConnection)conn).AddOnConnect((_, _) => + { + ((IDbConnection)conn).Unsubscribe(before.Id); + Subscribe(conn, "SELECT * FROM keyed WHERE id = 2"); + }); + Retry(conn); + Establish(conn); + Assert.Equal(1, before.Ended); + Assert.Equal("SELECT * FROM keyed WHERE id = 2", Assert.Single(Batch(conn).Sets).QueryStrings.Single()); + Drop(conn); + ((IDbConnection)conn).AddOnConnect((_, _) => conn.Disconnect()); + Retry(conn); + Establish(conn); + Assert.Empty(conn.Socket.Sent); + Assert.False(conn.IsReconnecting); + } + + [Fact] + public void LostPendingUnsubscribeIsNotReplayed() + { + using var conn = Create(); + Establish(conn); + var handle = Subscribe(conn); + Apply(conn, handle, Query("keyed", new Row { Id = 1 })); + ((IDbConnection)conn).Unsubscribe(handle.Id); + Drop(conn); + Assert.Equal(1, handle.Ended); + Retry(conn); + Establish(conn); + Assert.Empty(conn.Socket.Sent); + Assert.Empty(conn.Db.Keyed.Iter()); + } + + [Fact] + public void InitialSubscriptionsWaitForTheHandshake() + { + using var conn = Create(); + var handle = Subscribe(conn); + Assert.Empty(conn.Socket.Sent); + Establish(conn); + Assert.Single(conn.Socket.Sent.OfType()); + Apply(conn, handle, Query("keyed", new Row { Id = 1 })); + } + + [Theory] + [InlineData(400)] + [InlineData(401)] + [InlineData(403)] + public void AuthenticationRejectionWithoutProviderIsTerminal(int status) + { + using var conn = Create(); + Establish(conn); + Drop(conn); + Retry(conn); + conn.Socket.Fail(new SpacetimeDB.WebSocket.ConnectException(status)); + Pump(conn, () => conn.Events.Count == 2); + Assert.Null(conn.Events[^1].Next); + } + + [Fact] + public void BatchWireDiscriminantsMatchTheSharedProtocol() + { + ClientMessage client = new ClientMessage.SubscribeBatch(new(0x12345678, new())); + ServerMessage server = new ServerMessage.SubscribeBatchApplied(new(0x12345678, new())); + Assert.Equal(new byte[] { 5, 0x78, 0x56, 0x34, 0x12, 0, 0, 0, 0 }, IStructuralReadWrite.ToBytes(new ClientMessage.BSATN(), client)); + Assert.Equal(new byte[] { 8, 0x78, 0x56, 0x34, 0x12, 0, 0, 0, 0 }, IStructuralReadWrite.ToBytes(new ServerMessage.BSATN(), server)); + } + + [Fact] + public void SubscriptionHandleRebindsItsUnsubscribeId() + { + using var conn = Create(); + Establish(conn); + var handle = new Handle(conn); + var initial = conn.Socket.Sent.OfType().Single().Subscribe_; + conn.Socket.Receive(new ServerMessage.SubscribeApplied(new(initial.RequestId, initial.QuerySetId, new()))); + Pump(conn, () => handle.IsActive); + Drop(conn); + Retry(conn); + Establish(conn); + Assert.False(handle.IsActive); + var replay = Batch(conn); + var newId = replay.Sets.Single().QuerySetId; + Assert.NotEqual(initial.QuerySetId.Id, newId.Id); + conn.Socket.Receive(new ServerMessage.SubscribeBatchApplied(new(replay.RequestId, + new() { new(newId, new SubscribeSetOutcome.Applied(new QueryRows())) }))); + Pump(conn, () => handle.IsActive); + handle.Unsubscribe(); + Assert.Equal(newId, conn.Socket.Sent.OfType().Single().Unsubscribe_.QuerySetId); + } + + [Fact] + public void DropDuringReplayKeepsTheOldCacheUntilNextBatch() + { + using var conn = Create(); + Establish(conn); + var handle = Subscribe(conn); + Apply(conn, handle, Query("keyed", new Row { Id = 1 })); + Drop(conn); + Retry(conn); + Establish(conn); + var failed = conn.Socket; + var batch = Batch(conn); + var id = handle.Id; + Drop(conn); + Assert.Equal(1, handle.Applied); + Assert.Equal(1u, conn.Db.Keyed.Iter().Single().Id); + Retry(conn); + Establish(conn); + failed.Receive(new ServerMessage.SubscribeBatchApplied(new(batch.RequestId, + new() { new(id, new SubscribeSetOutcome.Applied(Query("keyed", new Row { Id = 999 }))) }))); + conn.Socket.Receive(new ServerMessage.SubscribeBatchApplied(new(Batch(conn).RequestId, + new() { new(handle.Id, new SubscribeSetOutcome.Applied(Query("keyed", new Row { Id = 2 }))) }))); + Pump(conn, () => handle.Applied == 2); + Assert.Equal(2u, conn.Db.Keyed.Iter().Single().Id); + } + + private static string Jwt(double exp, double iat) => "header." + Convert.ToBase64String(Encoding.UTF8.GetBytes( + System.Text.Json.JsonSerializer.Serialize(new { exp, iat }))).TrimEnd('=').Replace('+', '-').Replace('/', '_') + ".signature"; + + [Theory] + [InlineData(1000, 900, 969, false)] + [InlineData(1000, 900, 970, true)] + [InlineData(10000, 0, 9499, false)] + [InlineData(10000, 0, 9500, true)] + public void TokenRefreshUsesLifetimeMargin(double exp, double iat, long now, bool refresh) + { + Assert.Equal(refresh, ReconnectPolicy.TokenNeedsRefresh(Jwt(exp, iat), DateTimeOffset.FromUnixTimeSeconds(now))); + } + + [Theory] + [InlineData(null)] + [InlineData("garbage")] + [InlineData("header.e30.signature")] + public void UnreadableTokenNeedsRefresh(string? token) => Assert.True(ReconnectPolicy.TokenNeedsRefresh(token, DateTimeOffset.UtcNow)); + + [Theory] + [InlineData(1, 0, 500)] + [InlineData(1, 0.5, 1000)] + [InlineData(2, 0.5, 2000)] + [InlineData(6, 0, 15000)] + [InlineData(int.MaxValue, 1, 30000)] + public void BackoffHasJitterAndCap(int attempt, double random, double expected) => + Assert.Equal(expected, ReconnectPolicy.Delay(attempt, random).TotalMilliseconds); +} diff --git a/sdks/csharp/tests~/SnapshotTests.VerifySampleDump_dumpName=LegacySubscribeAll.verified.txt b/sdks/csharp/tests~/SnapshotTests.VerifySampleDump_dumpName=LegacySubscribeAll.verified.txt index 5ae3339cfaa..f961e0c49e4 100644 --- a/sdks/csharp/tests~/SnapshotTests.VerifySampleDump_dumpName=LegacySubscribeAll.verified.txt +++ b/sdks/csharp/tests~/SnapshotTests.VerifySampleDump_dumpName=LegacySubscribeAll.verified.txt @@ -106,10 +106,12 @@ }, Db: { Message: { + RemoteTableName: message, Count: 4 }, User: { Identity: {}, + RemoteTableName: user, Count: 3 } }, @@ -138,10 +140,12 @@ }, Db: { Message: { + RemoteTableName: message, Count: 4 }, User: { Identity: {}, + RemoteTableName: user, Count: 3 } }, @@ -170,10 +174,12 @@ }, Db: { Message: { + RemoteTableName: message, Count: 4 }, User: { Identity: {}, + RemoteTableName: user, Count: 3 } }, diff --git a/sdks/csharp/tests~/SnapshotTests.VerifySampleDump_dumpName=SubscribeApplied.verified.txt b/sdks/csharp/tests~/SnapshotTests.VerifySampleDump_dumpName=SubscribeApplied.verified.txt index 7dd87750886..1cefa360546 100644 --- a/sdks/csharp/tests~/SnapshotTests.VerifySampleDump_dumpName=SubscribeApplied.verified.txt +++ b/sdks/csharp/tests~/SnapshotTests.VerifySampleDump_dumpName=SubscribeApplied.verified.txt @@ -106,10 +106,12 @@ }, Db: { Message: { + RemoteTableName: message, Count: 4 }, User: { Identity: {}, + RemoteTableName: user, Count: 3 } }, @@ -138,10 +140,12 @@ }, Db: { Message: { + RemoteTableName: message, Count: 4 }, User: { Identity: {}, + RemoteTableName: user, Count: 3 } }, diff --git a/sdks/csharp/tests~/SnapshotTests.cs b/sdks/csharp/tests~/SnapshotTests.cs index 95a188c8553..e4f73f74246 100644 --- a/sdks/csharp/tests~/SnapshotTests.cs +++ b/sdks/csharp/tests~/SnapshotTests.cs @@ -10,11 +10,22 @@ namespace SpacetimeDB.Tests; using U128 = SpacetimeDB.U128; -public class SnapshotTests +[Collection("SDK connections")] +public class SnapshotTests : IDisposable { + private readonly ISpacetimeDBLogger previousLogger = Log.Current; + private readonly bool previousTesting = DbConnection.IsTesting; + + public void Dispose() + { + Log.Current = previousLogger; + DbConnection.IsTesting = previousTesting; + } + sealed class TestSubscriptionHandle : ISubscriptionHandle { public void OnApplied(ISubscriptionEventContext ctx) { } + public void RebindQuerySetId(QuerySetId id) { } public void OnError(IErrorContext ctx) { } public void OnEnded(ISubscriptionEventContext ctx) { } } diff --git a/skills/csharp-client/SKILL.md b/skills/csharp-client/SKILL.md index 4c1283e815b..0c3564b7581 100644 --- a/skills/csharp-client/SKILL.md +++ b/skills/csharp-client/SKILL.md @@ -48,6 +48,18 @@ var conn = DbConnection.Builder() Compression options are `Compression.Brotli`, `Compression.Gzip`, and `Compression.None`. The SDK uses Brotli when `WithCompression` is omitted. +## Automatic Reconnect and Token Refresh + +Automatic reconnect is opt-in: add `.WithAutomaticReconnect()` to the builder. Initial connection failures do not retry. After an established connection is lost, the SDK retries with exponential backoff and jitter, capped at 30 seconds. `Disconnect()` permanently stops recovery. + +Keep calling `FrameTick()` while `IsActive` is false. `IsReconnecting` is true while recovering before the next handshake. Use `.OnDisconnect((conn, error, next) => ...)` and `.OnConnectError((error, next) => ...)` to inspect `NextReconnect?`: a non-null value provides `Attempt` and `Delay`; null means no retry is scheduled. Existing callback overloads still work. + +`OnConnect` runs on every successful reconnect, so create the subscription in the example above only on the first connection, or move it after `Build()`. Register row callbacks once as well. The same connection, identity, table handles, and subscriptions survive; each attempt gets a fresh `ConnectionId`. Cached rows stay readable but stale during outages. The SDK replays subscriptions in one batch and emits only net row changes. Subscription `OnApplied` runs again after replay; use it to mark data ready, not for repeated one-time setup. + +For expiring credentials, combine `.WithToken(initialToken)` with `.WithTokenProvider(() => RefreshTokenAsync())`, where your provider returns `Task` for the same identity. The provider is not called for the initial connection. Before retries, it is called when remaining validity is at most 30 seconds or 5% of the token lifetime, whichever is greater, when expiry cannot be read, or after a reused token is rejected. Provider failures retry; rejection of a freshly provided token is terminal. No periodic refresh runs while connected. Disconnecting ignores a pending provider result but does not cancel the provider's own work. + +Calls made while disconnected fail immediately. Pending reducer calls receive `Status.UnknownResult`; pending procedures and one-off queries fail with `UnknownResultException`. These calls may already have executed and are never replayed. Regenerate bindings so unhandled unknown reducer outcomes reach `OnUnhandledReducerError`. + ## Event Loop (Critical) **`FrameTick()` must be called in your main loop.** The SDK queues all network messages and only processes them when you call `FrameTick()`. Without it, no callbacks fire. @@ -153,6 +165,8 @@ conn.Reducers.OnSendMessage += (ReducerEventContext ctx, string text) => Console.WriteLine($"Message sent: {text}"); else if (ctx.Event.Status is Status.Failed(var reason)) Console.Error.WriteLine($"Send failed: {reason}"); + else if (ctx.Event.Status is Status.UnknownResult) + Console.Error.WriteLine("Connection lost before the result arrived; the message may have been sent."); }; ``` diff --git a/skills/unity/SKILL.md b/skills/unity/SKILL.md index 61d523f1e97..928712328e1 100644 --- a/skills/unity/SKILL.md +++ b/skills/unity/SKILL.md @@ -120,6 +120,16 @@ public class SpacetimeManager : MonoBehaviour **Thread safety**: `FrameTick()` processes messages on the calling thread (the main thread in Unity). Do NOT call it from a background thread. Do NOT access `conn.Db` from background threads. +## Automatic Reconnect + +Add `.WithAutomaticReconnect()` to the builder to recover an established connection after an outage. In the singleton above, create subscriptions only on the first `OnConnected`, or move them after `Build()`: `OnConnected` runs again after each successful reconnect. Keep calling `FrameTick()` during outages, or use `SpacetimeDBNetworkManager`, which also ticks reconnecting connections. Do not gate ticking on `IsActive`. + +The SDK retains the identity, table handles, callbacks, and subscriptions, and replays subscriptions in one batch. Cached rows stay readable during an outage; row callbacks report net changes after recovery. `OnApplied` runs again, so separate one-time object setup from marking data ready. Each reconnect attempt has a fresh `ConnectionId`. + +`IsReconnecting` reports recovery before the next successful handshake. The `OnDisconnect((conn, error, next) => ...)` and `OnConnectError((error, next) => ...)` overloads expose `NextReconnect?`, with the upcoming `Attempt` and `Delay`, or null for a terminal failure. Initial connection failures do not retry. Retries use exponential backoff and jitter capped at 30 seconds; `Disconnect()` stops them. + +Calls made while disconnected fail immediately. Pending reducer calls may report `Status.UnknownResult`, meaning the server may have executed them; they are not replayed. Pending procedures and one-off queries fail with `UnknownResultException`. Regenerate bindings when upgrading to include unknown-outcome handling. + --- ## Row Callbacks for Game State @@ -143,7 +153,7 @@ void RegisterCallbacks() } ``` -Register these in `OnSubscriptionApplied` (after initial data is loaded) or in `Start()` before connecting. +Register these once after building the connection, before the first `FrameTick()`. Do not register them repeatedly in `OnConnected` or `OnSubscriptionApplied` when automatic reconnect is enabled. --- @@ -172,6 +182,8 @@ SpacetimeManager.Instance.Connection.Reducers.OnSendMessage += (ReducerEventCont Debug.Log($"Message sent: {text}"); else if (ctx.Event.Status is Status.Failed(var reason)) Debug.LogError($"Send failed: {reason}"); + else if (ctx.Event.Status is Status.UnknownResult) + Debug.LogWarning("Connection lost before the result arrived; the message may have been sent."); }; ``` @@ -220,3 +232,6 @@ The SpacetimeDB SDK uses code generation. If you encounter issues with IL2CPP bu ### Token Persistence Token save/load via `PlayerPrefs` is demonstrated in the SpacetimeManager singleton above. Persisting the server-issued token and passing it back on reconnect keeps the same identity; without a saved token the server issues a new identity in the `OnConnect` callback. This token does not expire and a lost one can't be recovered, so self-issued identities are for development. For production, authenticate with an OIDC provider such as SpacetimeAuth, which handles token lifecycle. +With automatic reconnect enabled, the SDK retains the authentication token and passes it to `OnConnect`, including on WebGL. Without automatic reconnect, a WebGL connection using a saved token may return a short-lived WebSocket token instead; in that case, keep the original saved token rather than overwriting it in the callback above. + +For expiring credentials, configure `.WithToken(initialToken)`, `.WithAutomaticReconnect()`, and `.WithTokenProvider(() => RefreshTokenAsync())`. The provider returns `Task` for the same identity. It is used before reconnect attempts when expiry is unreadable, remaining validity is at most 30 seconds or 5% of the original lifetime, or a reused token is rejected. It is not called for the initial connection or periodically while connected. Provider failures retry; rejection of a freshly provided token stops recovery. `Disconnect()` ignores a pending provider result without canceling the provider's own asynchronous work. diff --git a/templates/chat-console-cs/module_bindings/Reducers/SendMessage.g.cs b/templates/chat-console-cs/module_bindings/Reducers/SendMessage.g.cs index caa1bf37165..a42e600ffb0 100644 --- a/templates/chat-console-cs/module_bindings/Reducers/SendMessage.g.cs +++ b/templates/chat-console-cs/module_bindings/Reducers/SendMessage.g.cs @@ -30,6 +30,7 @@ public bool InvokeSendMessage(ReducerEventContext ctx, Reducer.SendMessage args) { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false; diff --git a/templates/chat-console-cs/module_bindings/Reducers/SetName.g.cs b/templates/chat-console-cs/module_bindings/Reducers/SetName.g.cs index fc905ecc50c..65039203e75 100644 --- a/templates/chat-console-cs/module_bindings/Reducers/SetName.g.cs +++ b/templates/chat-console-cs/module_bindings/Reducers/SetName.g.cs @@ -30,6 +30,7 @@ public bool InvokeSetName(ReducerEventContext ctx, Reducer.SetName args) { case Status.Failed(var reason): InternalOnUnhandledReducerError(ctx, new Exception(reason)); break; case Status.OutOfEnergy(var _): InternalOnUnhandledReducerError(ctx, new Exception("out of energy")); break; + case Status.UnknownResult(var _): InternalOnUnhandledReducerError(ctx, new UnknownResultException()); break; } } return false;