diff --git a/crates/bindings-typescript/README.md b/crates/bindings-typescript/README.md index 48e7cfd1535..6ebfd1d0e36 100644 --- a/crates/bindings-typescript/README.md +++ b/crates/bindings-typescript/README.md @@ -26,22 +26,28 @@ import { DbConnection, tables } from './module_bindings'; const connection = DbConnection.builder() .withUri('ws://localhost:3000') .withDatabaseName('MODULE_NAME') - .onDisconnect(() => { - console.log('disconnected'); + .withAutomaticReconnect() + .onConnect((_connection, identity) => { + console.log('Connected:', identity.toHexString()); }) - .onConnectError(() => { - console.log('client_error'); - }) - .onConnect((connection, identity, _token) => { + .onDisconnect((_ctx, error, attempt, delayMs) => { console.log( - 'Connected to SpacetimeDB with identity:', - identity.toHexString() + attempt === undefined + ? 'Disconnected' + : `Retry ${attempt} in ${delayMs} ms`, + error + ); + }) + .onConnectError((_ctx, error, attempt) => { + console.error( + attempt === undefined ? 'Connection failed' : 'Retry failed', + error ); - - connection.subscriptionBuilder().subscribe(tables.player); }) .withToken('TOKEN') .build(); + +connection.subscriptionBuilder().subscribe(tables.player); ``` If you need to disconnect the client: @@ -50,6 +56,10 @@ If you need to disconnect the client: connection.disconnect(); ``` +Automatic reconnection preserves the connection, cache, handles, and callbacks. Register subscriptions and row callbacks once, outside `onConnect`, which runs again after every reconnect. Cache reads remain available during outages. Initial connection failures are not retried by the core SDK. + +For expiring credentials, pass the initial token with `withToken` and add `withTokenProvider(() => auth.getAccessToken())`. The SDK asks for a fresh token before reconnecting when the retained token is near expiry. The provider must return a token for the same identity. + Typically, you will use the SDK with types generated from SpacetimeDB module. For example, given a table named `Player` you can subscribe to player updates like this: ```ts diff --git a/crates/bindings-typescript/src/lib/errors.ts b/crates/bindings-typescript/src/lib/errors.ts index c8ec99133c8..93890421b4d 100644 --- a/crates/bindings-typescript/src/lib/errors.ts +++ b/crates/bindings-typescript/src/lib/errors.ts @@ -24,3 +24,37 @@ export class InternalError extends Error { return 'InternalError'; } } + +/** The call was not sent because the connection was not established. */ +export class DisconnectedError extends Error { + constructor(message: string = 'Not connected to SpacetimeDB') { + super(message); + } + get name(): string { + return 'DisconnectedError'; + } +} + +/** The connection dropped before acknowledgement; the call may have run. */ +export class UnknownCallResultError extends Error { + constructor( + message: string = 'Connection lost before the call was acknowledged; it may or may not have run' + ) { + super(message); + } + get name(): string { + return 'UnknownCallResultError'; + } +} + +/** The reconnect returned a different identity, ending automatic reconnection. */ +export class IdentityChangedError extends Error { + constructor( + message: string = 'Reconnected with a different identity; the token was revoked or replaced' + ) { + super(message); + } + get name(): string { + return 'IdentityChangedError'; + } +} diff --git a/crates/bindings-typescript/src/sdk/client_api/types.ts b/crates/bindings-typescript/src/sdk/client_api/types.ts index 709a114da28..ce4f4ed0680 100644 --- a/crates/bindings-typescript/src/sdk/client_api/types.ts +++ b/crates/bindings-typescript/src/sdk/client_api/types.ts @@ -51,6 +51,9 @@ export const ClientMessage = __t.enum('ClientMessage', { get CallProcedure() { return CallProcedure; }, + get SubscribeBatch() { + return SubscribeBatch; + }, }); export type ClientMessage = __Infer; @@ -192,6 +195,9 @@ export const ServerMessage = __t.enum('ServerMessage', { get ProcedureResult() { return ProcedureResult; }, + get SubscribeBatchApplied() { + return SubscribeBatchApplied; + }, }); export type ServerMessage = __Infer; @@ -223,6 +229,48 @@ export const SubscribeApplied = __t.object('SubscribeApplied', { }); export type SubscribeApplied = __Infer; +export const SubscribeBatch = __t.object('SubscribeBatch', { + requestId: __t.u32(), + get sets() { + return __t.array(SubscribeSet); + }, +}); +export type SubscribeBatch = __Infer; + +export const SubscribeBatchApplied = __t.object('SubscribeBatchApplied', { + requestId: __t.u32(), + get results() { + return __t.array(SubscribeSetResult); + }, +}); +export type SubscribeBatchApplied = __Infer; + +export const SubscribeSet = __t.object('SubscribeSet', { + get querySetId() { + return QuerySetId; + }, + queryStrings: __t.array(__t.string()), +}); +export type SubscribeSet = __Infer; + +export const SubscribeSetOutcome = __t.enum('SubscribeSetOutcome', { + get Applied() { + return QueryRows; + }, + Error: __t.string(), +}); +export type SubscribeSetOutcome = __Infer; + +export const SubscribeSetResult = __t.object('SubscribeSetResult', { + get querySetId() { + return QuerySetId; + }, + get outcome() { + return SubscribeSetOutcome; + }, +}); +export type SubscribeSetResult = __Infer; + export const SubscriptionError = __t.object('SubscriptionError', { requestId: __t.option(__t.u32()), get querySetId() { diff --git a/crates/bindings-typescript/src/sdk/connection_manager.ts b/crates/bindings-typescript/src/sdk/connection_manager.ts index 2cf34ea2c36..052fa73e392 100644 --- a/crates/bindings-typescript/src/sdk/connection_manager.ts +++ b/crates/bindings-typescript/src/sdk/connection_manager.ts @@ -25,6 +25,16 @@ * Result: Single WebSocket survives ✓ * ``` * + * ## Reconnection: + * + * The manager forces {@link DbConnectionBuilder.withAutomaticReconnect} on + * every builder it builds from, so a connection lost mid-session reconnects + * *inside* the `DbConnection`: the object, its table handles and its callbacks + * all survive, and the manager merely mirrors the lifecycle events into its + * state snapshots. The manager itself rebuilds a connection only when the SDK + * reports it will not retry (a failed initial connection, or another terminal + * failure), preserving the frameworks' historical keep-trying behavior. + * * @module connection_manager */ import type { @@ -50,9 +60,13 @@ export const CONNECTION_MANAGER_RECONNECT_BASE_DELAY_MS = 1000; export const CONNECTION_MANAGER_RECONNECT_MAX_DELAY_MS = 30_000; /** - * Computes the reconnect delay for the given attempt (0-based) using + * Computes the rebuild delay for the given attempt (0-based) using * exponential backoff: the base delay doubles with each consecutive failed * attempt, capped at the maximum delay. + * + * This paces only the manager's own rebuild loop for failures the SDK will + * not retry; reconnects after a mid-session drop are paced by the SDK's own + * policy (see `computeReconnectDelayMs` in `db_connection_impl`). */ export function connectionManagerReconnectDelayMs(attempt: number): number { return Math.min( @@ -71,8 +85,18 @@ type ManagedConnection = { reconnectTimer: ReturnType | null; reconnectAttempt: number; onConnect?: (conn: DbConnectionImpl) => void; - onDisconnect?: (ctx: ErrorContextInterface, error?: Error) => void; - onConnectError?: (ctx: ErrorContextInterface, error: Error) => void; + onDisconnect?: ( + ctx: ErrorContextInterface, + error?: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number + ) => void; + onConnectError?: ( + ctx: ErrorContextInterface, + error: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number + ) => void; }; function defaultState(): ConnectionState { @@ -92,98 +116,6 @@ function defaultState(): ConnectionState { class ConnectionManagerImpl { #connections = new Map(); - constructor() { - // Auto-reconnect otherwise relies entirely on the browser firing - // `onclose` plus a `setTimeout` backoff. Both are unreliable across a - // backgrounded/frozen tab: the close event may never be delivered (the - // socket dies while the event loop is suspended), and background timers - // are heavily throttled or paused, so a scheduled reconnect can stall - // indefinitely and never resume when the window is refocused. - // - // These listeners make the manager proactively re-check liveness when the - // page comes back to the foreground / the network returns, bringing any - // stalled reconnect forward and rebuilding sockets that died silently. - if ( - typeof document !== 'undefined' && - typeof document.addEventListener === 'function' - ) { - document.addEventListener('visibilitychange', () => { - if (document.visibilityState === 'visible') { - this.#handleResume(); - } - }); - } - if ( - typeof window !== 'undefined' && - typeof window.addEventListener === 'function' - ) { - window.addEventListener('focus', this.#handleResume); - window.addEventListener('online', this.#handleResume); - // `pageshow` fires on bfcache restores, where `visibilitychange` may not. - window.addEventListener('pageshow', this.#handleResume); - } - } - - /** - * Called when the page is likely resuming from a background/frozen state: - * the tab became visible, the window regained focus, the network came back, - * or a bfcache page was restored. For each retained connection this brings a - * stalled reconnect forward immediately (resetting backoff) and rebuilds any - * socket that died silently while we were hidden. - */ - #handleResume = (): void => { - for (const managed of this.#connections.values()) { - if (managed.refCount <= 0 || managed.pendingRelease) { - continue; - } - - // A reconnect was scheduled but its timer is stuck behind background - // timer throttling / page freezing. Fire it now and reset backoff so we - // reconnect promptly instead of waiting out a (capped 30s, possibly - // paused) delay. - if (managed.reconnectTimer && !managed.connection) { - clearTimeout(managed.reconnectTimer); - managed.reconnectTimer = null; - managed.reconnectAttempt = 0; - if (managed.builder) { - this.#buildManagedConnection(managed, managed.builder); - } - continue; - } - - // We believe we're connected, but the socket may have died silently. - this.#reviveIfZombie(managed); - } - }; - - /** - * If `managed` holds a connection whose socket has entered CLOSING/CLOSED - * without a clean `onclose` (see {@link DbConnectionImpl.isSocketClosed}), - * for example because it was torn down while the tab was frozen, tear it down - * and build a fresh one immediately, resetting backoff. - */ - #reviveIfZombie(managed: ManagedConnection): void { - const connection = managed.connection; - if ( - !connection || - connection.isDisconnectRequested || - !connection.isSocketClosed - ) { - return; - } - - this.#detachCallbacks(managed, connection); - managed.connection = undefined; - // Close the dead socket in case it is only CLOSING; callbacks are already - // detached, so this won't trigger a duplicate reconnect. - connection.disconnect(); - this.#updateState(managed, { isActive: false }); - managed.reconnectAttempt = 0; - if (managed.builder) { - this.#buildManagedConnection(managed, managed.builder); - } - } - /** Generates a unique key for a connection based on URI and module name. */ static getKey(uri: string, moduleName: string): string { return `${uri}::${moduleName}`; @@ -246,7 +178,12 @@ class ConnectionManagerImpl { }); }; - managed.onDisconnect = (ctx, error) => { + // With automatic reconnection forced on, a non-undefined + // `nextReconnectAttempt` means the SDK is retrying inside the same + // connection object: mirror the event into state and leave the connection + // alone. Only when the SDK reports it will not retry does the manager + // rebuild a replacement (see #scheduleRebuild). + managed.onDisconnect = (ctx, error, nextReconnectAttempt) => { if (ctx !== managed.connection) { return; } @@ -254,10 +191,12 @@ class ConnectionManagerImpl { isActive: false, connectionError: error ?? undefined, }); - this.#scheduleReconnect(managed); + if (nextReconnectAttempt === undefined) { + this.#scheduleRebuild(managed); + } }; - managed.onConnectError = (ctx, error) => { + managed.onConnectError = (ctx, error, nextReconnectAttempt) => { if (ctx !== managed.connection) { return; } @@ -265,7 +204,9 @@ class ConnectionManagerImpl { isActive: false, connectionError: error, }); - this.#scheduleReconnect(managed); + if (nextReconnectAttempt === undefined) { + this.#scheduleRebuild(managed); + } }; } @@ -284,44 +225,24 @@ class ConnectionManagerImpl { connection: DbConnectionImpl ): void { if (managed.onConnect) { - connection.removeOnConnect(managed.onConnect as any); + connection.removeOnConnect(managed.onConnect); } if (managed.onDisconnect) { - connection.removeOnDisconnect(managed.onDisconnect as any); + connection.removeOnDisconnect(managed.onDisconnect); } if (managed.onConnectError) { - connection.removeOnConnectError(managed.onConnectError as any); + connection.removeOnConnectError(managed.onConnectError); } } - /** - * Builds a connection for `managed` from `builder`, adopting it as the - * entry's retained builder. - * - * `resumeSession` (the default) re-applies the session's current token to the - * builder first. This matters because the builder is a *long-lived template*: - * the application hands it over once, and every automatic rebuild — scheduled - * reconnect, resume-from-background, zombie-socket revival — reuses that same - * object. Its token, though, is a snapshot taken when the application built - * it, typically read out of storage at module load, before any session - * existed. Rebuilding from it verbatim would reconnect *anonymously* for any - * user whose token was issued during this page's lifetime, and the server - * would answer by minting a brand-new identity: a silent account switch, with - * no error raised on either side, curable only by a page reload. - * - * `state.token` is the token of the most recent connection (set below at - * build time, and again by `onConnect` when the server issues one), so - * re-applying it keeps every automatic rebuild on the same principal. - * - * Pass `resumeSession: false` when the caller is deliberately changing - * identity — see {@link rebuild} — so the builder's own token wins. - */ + /** Reuse the latest session token when rebuilding from the retained builder. */ #buildManagedConnection>( managed: ManagedConnection, builder: DbConnectionBuilder, { resumeSession = true }: { resumeSession?: boolean } = {} ): T { managed.builder = builder; + builder.withAutomaticReconnect(); if (resumeSession && managed.state.token) { builder.withToken(managed.state.token); } @@ -340,7 +261,8 @@ class ConnectionManagerImpl { return connection as T; } - #scheduleReconnect(managed: ManagedConnection): void { + /** Preserve framework retries for failures the core connection will not retry. */ + #scheduleRebuild(managed: ManagedConnection): void { if ( managed.refCount <= 0 || managed.pendingRelease || @@ -425,8 +347,8 @@ class ConnectionManagerImpl { * …) re-bind to the new connection automatically. * * The old connection's callbacks are detached before it is closed, so its - * disconnect event never leaks into pool state, and any pending auto-reconnect - * is cancelled (the caller is driving the reconnect explicitly). Returns the + * disconnect event never leaks into pool state, and any pending rebuild is + * cancelled (the caller is driving the replacement explicitly). Returns the * newly-built connection, or `null` if the key has no retained entry. * * @param key - Unique identifier for the connection (use getKey to generate) @@ -442,8 +364,8 @@ class ConnectionManagerImpl { } // The caller is taking over the connection lifecycle explicitly; cancel a - // deferred release or a pending auto-reconnect so neither races the fresh - // connection, and reset the backoff so the next unexpected drop starts over. + // deferred release or a pending rebuild so neither races the fresh + // connection, and reset the backoff so the next terminal failure starts over. if (managed.pendingRelease) { clearTimeout(managed.pendingRelease); managed.pendingRelease = null; diff --git a/crates/bindings-typescript/src/sdk/db_connection_builder.ts b/crates/bindings-typescript/src/sdk/db_connection_builder.ts index 282cab02d37..9af4fad1020 100644 --- a/crates/bindings-typescript/src/sdk/db_connection_builder.ts +++ b/crates/bindings-typescript/src/sdk/db_connection_builder.ts @@ -1,4 +1,8 @@ -import { DbConnectionImpl, type ConnectionEvent } from './db_connection_impl'; +import { + DbConnectionImpl, + type ConnectionEvent, + type TokenProvider, +} from './db_connection_impl'; import { EventEmitter } from './event_emitter'; import type { DbConnectionConfig, @@ -27,6 +31,8 @@ export class DbConnectionBuilder> { #compression: 'gzip' | 'brotli' | 'none' = 'gzip'; #lightMode: boolean = false; #confirmedReads?: boolean; + #automaticReconnect: boolean = false; + #tokenProvider?: TokenProvider; #createWSFn: WebSocketFactory; /** @@ -145,6 +151,26 @@ export class DbConnectionBuilder> { return this; } + /** + * Reconnect after an established connection drops, preserving handles and callbacks. + * Retries use exponential backoff until disconnect() or a terminal failure. + * Initial connection failures are not retried. Lifecycle callbacks report + * the next attempt and delay, or undefined when no retry is scheduled. + */ + withAutomaticReconnect(): this { + this.#automaticReconnect = true; + return this; + } + + /** + * Refresh the retained token before reconnecting when it is near expiry or + * rejected. The provider must return a token for the same identity. + */ + withTokenProvider(provider: TokenProvider): this { + this.#tokenProvider = provider; + return this; + } + /** * Register a callback to be invoked upon authentication with the database. * @@ -190,11 +216,19 @@ export class DbConnectionBuilder> { * console.log("Error connecting to SpacetimeDB:", error); * }); * ``` + * + * With {@link DbConnectionBuilder.withAutomaticReconnect} enabled, this + * callback also reports each failed reconnect attempt: + * `nextReconnectAttempt` is the number of the upcoming attempt and + * `nextReconnectDelayMs` the wait before it. Both are `undefined` when the + * SDK will not retry, as for a failed initial connection. */ onConnectError( callback: ( ctx: ErrorContextInterface>, - error: Error + error: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number ) => void ): this { this.#emitter.on('connectError', callback); @@ -225,13 +259,22 @@ export class DbConnectionBuilder> { * This is a concession to ergonomics; there's no clean place to return a `CallbackId` from this method * or from `build`. * + * With {@link DbConnectionBuilder.withAutomaticReconnect} enabled, this + * callback also reports connections lost mid-session, and the SDK keeps + * reconnecting afterwards: `nextReconnectAttempt` is the number of the + * upcoming attempt and `nextReconnectDelayMs` the wait before it. Both are + * `undefined` when the SDK will not retry, which is always the case without + * automatic reconnection. + * * @param {function(error?: Error): void} callback - The callback to invoke upon disconnection. * @throws {Error} Throws an error if called multiple times on the same `DbConnectionBuilder`. */ onDisconnect( callback: ( ctx: ErrorContextInterface>, - error?: Error | undefined + error?: Error | undefined, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number ) => void ): this { this.#emitter.on('disconnect', callback); @@ -285,6 +328,8 @@ export class DbConnectionBuilder> { confirmedReads: this.#confirmedReads, createWSFn: this.#createWSFn, remoteModule: this.remoteModule, + automaticReconnect: this.#automaticReconnect, + tokenProvider: this.#tokenProvider, }); } } diff --git a/crates/bindings-typescript/src/sdk/db_connection_impl.ts b/crates/bindings-typescript/src/sdk/db_connection_impl.ts index cf73bc270bb..efbbea89df2 100644 --- a/crates/bindings-typescript/src/sdk/db_connection_impl.ts +++ b/crates/bindings-typescript/src/sdk/db_connection_impl.ts @@ -10,6 +10,8 @@ import { ServerMessage, TableUpdateRows, UnsubscribeFlags, + type SubscribeBatchApplied, + type SubscribeBatch, } from './client_api/types'; import { ClientCache } from './client_cache.ts'; import { DbConnectionBuilder } from './db_connection_builder.ts'; @@ -61,8 +63,19 @@ import type { UntypedSchemaDef } from '../lib/schema'; import type { ProceduresView } from './procedures.ts'; import type { Values } from '../lib/type_util.ts'; import type { TransactionUpdate } from './client_api/types.ts'; -import { InternalError, SenderError } from '../lib/errors.ts'; -import type { WebSocketAdapter, WebSocketFactory } from './ws.ts'; +import type { SubscriptionEntry } from './subscription_builder_impl'; +import { + DisconnectedError, + IdentityChangedError, + InternalError, + SenderError, + UnknownCallResultError, +} from '../lib/errors.ts'; +import { + WebSocketTokenError, + type WebSocketAdapter, + type WebSocketFactory, +} from './ws.ts'; import { normalizeWsProtocol, PREFERRED_WS_PROTOCOLS, @@ -96,7 +109,21 @@ export type { ReducerEvent, }; -export type ConnectionEvent = 'connect' | 'disconnect' | 'connectError'; +export type ConnectionEventArgs = { + connect: [identity: Identity, token: string]; + disconnect: [ + error?: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number, + ]; + connectError: [ + error: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number, + ]; +}; + +export type ConnectionEvent = keyof ConnectionEventArgs; export type DbConnectionConfig = { uri: URL; @@ -109,14 +136,152 @@ export type DbConnectionConfig = { lightMode: boolean; confirmedReads?: boolean; remoteModule: RemoteModule; + /** + * Whether the connection reconnects on its own after losing its socket. + * Set by {@link DbConnectionBuilder.withAutomaticReconnect}. + */ + automaticReconnect?: boolean; + /** + * Supplies a fresh token for a reconnect attempt. + * Set by {@link DbConnectionBuilder.withTokenProvider}. + */ + tokenProvider?: TokenProvider; }; +/** + * Supplies an authentication token, called before a reconnect attempt whose + * retained token is close to expiring. + */ +export type TokenProvider = () => Promise; + +/** The delay before the first reconnect attempt. */ +export const RECONNECT_INITIAL_DELAY_MS = 1_000; +/** The upper bound on the delay between reconnect attempts. */ +export const RECONNECT_MAX_DELAY_MS = 30_000; +/** The random spread applied to each reconnect delay. */ +export const RECONNECT_JITTER = 0.5; +/** + * A token is refreshed when its remaining validity falls below this fraction + * of its lifetime, or below {@link TOKEN_REFRESH_MIN_MARGIN_MS}, whichever is + * larger. + */ +const TOKEN_REFRESH_MARGIN_FRACTION = 0.05; +const TOKEN_REFRESH_MIN_MARGIN_MS = 30_000; + +/** Exponential backoff with jitter; attempt numbers start at one. */ +export function computeReconnectDelayMs( + attempt: number, + random: () => number = Math.random +): number { + const base = Math.min( + RECONNECT_INITIAL_DELAY_MS * Math.pow(2, Math.max(0, attempt - 1)), + RECONNECT_MAX_DELAY_MS + ); + const jittered = base * (1 + RECONNECT_JITTER * (2 * random() - 1)); + return Math.max(0, Math.min(jittered, RECONNECT_MAX_DELAY_MS)); +} + +/** Refresh unreadable tokens, or tokens within 5% of expiry (at least 30 seconds). */ +export function tokenNeedsRefresh( + token: string | undefined, + nowMs: number = Date.now() +): boolean { + if (!token) { + return true; + } + const claims = decodeJwtClaims(token); + if (claims === undefined || claims.exp === undefined) { + return true; + } + const expiryMs = claims.exp * 1000; + const lifetimeMs = + claims.iat !== undefined ? expiryMs - claims.iat * 1000 : undefined; + const margin = Math.max( + TOKEN_REFRESH_MIN_MARGIN_MS, + lifetimeMs !== undefined ? lifetimeMs * TOKEN_REFRESH_MARGIN_FRACTION : 0 + ); + return expiryMs - nowMs <= margin; +} + +/** Normalize an unknown thrown value or error event into an `Error`. */ +function errorFromEvent(value: unknown, fallbackMessage: string): Error { + if (value instanceof Error) { + return value; + } + if (typeof value === 'object' && value !== null) { + const message = 'message' in value ? value.message : undefined; + const error = 'error' in value ? value.error : undefined; + if (error instanceof Error) { + return error; + } + if (typeof message === 'string' && message.length > 0) { + return new Error(message); + } + } + return new Error(fallbackMessage); +} + +/** + * Whether a failure to connect is one that retrying cannot fix, so that the + * SDK stops rather than looping. + */ +function isTerminalConnectError(error: Error): boolean { + return ( + error instanceof IdentityChangedError || + error instanceof WebSocketProtocolError + ); +} + +/** Whether a failure looks like the server rejecting our credentials. */ +function isAuthError(error: Error): boolean { + return ( + error instanceof WebSocketTokenError && + (error.status === 401 || error.status === 403) + ); +} + +class WebSocketProtocolError extends Error {} + +const SESSION_BUSY_CLOSE_CODE = 4000; + +type JwtClaims = { exp?: number; iat?: number }; + +function decodeJwtClaims(token: string): JwtClaims | undefined { + const parts = token.split('.'); + if (parts.length < 2) { + return undefined; + } + try { + const payload = parts[1].replace(/-/g, '+').replace(/_/g, '/'); + const padded = payload.padEnd( + payload.length + ((4 - (payload.length % 4)) % 4), + '=' + ); + const json = + typeof atob === 'function' + ? atob(padded) + : Buffer.from(padded, 'base64').toString('binary'); + const claims: unknown = JSON.parse(json); + if (typeof claims !== 'object' || claims === null) { + return undefined; + } + const exp = 'exp' in claims ? claims.exp : undefined; + const iat = 'iat' in claims ? claims.iat : undefined; + return { + exp: typeof exp === 'number' && Number.isFinite(exp) ? exp : undefined, + iat: typeof iat === 'number' && Number.isFinite(iat) ? iat : undefined, + }; + } catch { + return undefined; + } +} + type ProcedureCallback = (result: ProcedureResultMessage['result']) => void; type Deferred = { promise: Promise; resolve: (value: T | PromiseLike) => void; - reject: (reason?: unknown) => void; + reject: (reason: Error | string) => void; }; const TEXT_ENCODER = new TextEncoder(); @@ -177,8 +342,9 @@ export class DbConnectionImpl * Whether the underlying websocket has entered `CLOSING` (2) or `CLOSED` * (3). This becomes true even when the browser never delivered an * `onclose` event, for example if the socket was torn down while the tab was - * frozen or the machine was asleep. The `ConnectionManager` uses this to detect - * such "zombie" connections when the page resumes and to force a reconnect. + * frozen or the machine was asleep. The liveness listeners (see + * {@link DbConnectionImpl.#installLivenessListeners}) use this to detect + * such "zombie" sockets when the page resumes and to force a reconnect. * * Returns false while the socket is still `CONNECTING`/`OPEN`, or before * the socket has been created. @@ -231,6 +397,29 @@ export class DbConnectionImpl connectionId: ConnectionId = ConnectionId.random(); #connectionIdHex = this.connectionId.toHexString(); + // Stable across sockets; each attempt still gets a fresh ConnectionId. + #sessionIdHex = ConnectionId.random().toHexString(); + #automaticReconnect: boolean; + #tokenProvider?: TokenProvider; + #wsBaseUrl: URL; + #nameOrAddress: string; + #createWSFn: WebSocketFactory; + #compression: 'gzip' | 'brotli' | 'none'; + #lightMode: boolean; + #confirmedReads?: boolean; + // Invalidates events and asynchronous work from discarded sockets. + #socketGeneration = 0; + #hasEverConnected = false; + #connectionEnded = false; + #reconnectAttempt = 0; + #reconnectTimer?: ReturnType; + #forceTokenRefresh = false; + #usedFreshToken = false; + #livenessCleanup?: () => void; + #pendingReplay?: { requestId: number; querySetIds: Set }; + #preparingReplay = false; + #socketEstablished = false; + // These fields are meant to be strictly private. #queryId = 0; #requestId = 0; @@ -250,6 +439,10 @@ export class DbConnectionImpl >(); #reducerCallInfo = new Map(); #procedureCallbacks = new Map(); + /** + * Reject pending reducer and procedure calls when the socket drops. + */ + #pendingCallRejecters = new Map void>(); #rowDeserializers: Record>; #rowIdMetadata: Record< string, @@ -291,6 +484,8 @@ export class DbConnectionImpl compression, lightMode, confirmedReads, + automaticReconnect, + tokenProvider, }: DbConnectionConfig) { stdbLogger('info', 'Connecting to SpacetimeDB WS...'); @@ -357,43 +552,323 @@ export class DbConnectionImpl ); } - url.searchParams.set('connection_id', this.#connectionIdHex); - this.clientCache = new ClientCache(); this.db = this.#makeDbView(); this.reducers = this.#makeReducers(remoteModule); this.procedures = this.#makeProcedures(remoteModule); - this.wsPromise = createWSFn({ - url, - nameOrAddress, - wsProtocol: [...PREFERRED_WS_PROTOCOLS], - authToken: token, - compression: compression, - lightMode: lightMode, - confirmedReads: confirmedReads, - }) - .then(v => { - this.ws = v; - - this.ws.onclose = () => { - this.isActive = false; - this.#emitter.emit('disconnect', this); - }; - this.ws.onerror = (e: ErrorEvent) => { - this.isActive = false; - this.#emitter.emit('connectError', this, e); - }; - this.ws.onopen = this.#handleOnOpen.bind(this); - this.ws.onmessage = this.#handleOnMessage.bind(this); - return v; - }) - .catch(e => { - stdbLogger('error', 'Error connecting to SpacetimeDB WS'); - this.#emitter.emit('connectError', this, e); + this.#automaticReconnect = automaticReconnect ?? false; + this.#preparingReplay = this.#automaticReconnect; + this.#tokenProvider = tokenProvider; + this.#wsBaseUrl = url; + this.#nameOrAddress = nameOrAddress; + this.#createWSFn = createWSFn; + this.#compression = compression; + this.#lightMode = lightMode; + this.#confirmedReads = confirmedReads; + + this.wsPromise = this.#openSocket(); + } + + async #openSocket(): Promise { + const generation = ++this.#socketGeneration; + if (this.#hasEverConnected) { + this.#setConnectionId(ConnectionId.random()); + } + this.#usedFreshToken = false; + const url = new URL(this.#wsBaseUrl.toString()); + url.searchParams.set('connection_id', this.#connectionIdHex); + try { + const authToken = await this.#tokenForAttempt(); + if (generation !== this.#socketGeneration || this.#connectionEnded) { return undefined; + } + this.token = authToken; + const ws = await this.#createWSFn({ + url, + nameOrAddress: this.#nameOrAddress, + wsProtocol: [...PREFERRED_WS_PROTOCOLS], + authToken, + compression: this.#compression, + lightMode: this.#lightMode, + confirmedReads: this.#confirmedReads, + connectionId: this.#connectionIdHex, + sessionId: this.#automaticReconnect ? this.#sessionIdHex : undefined, }); + + if (generation !== this.#socketGeneration || this.#connectionEnded) { + // A newer attempt superseded this socket while it was opening, or the + // application disconnected in the meantime. + ws.close(); + return undefined; + } + + this.ws = ws; + this.#socketEstablished = false; + const isCurrent = (): boolean => + generation === this.#socketGeneration && !this.#connectionEnded; + const handleLoss = (error: Error, isErrorEvent: boolean): void => { + if (!isCurrent()) return; + this.isActive = false; + if (!this.#automaticReconnect) { + this.#emitter.emit( + isErrorEvent ? 'connectError' : 'disconnect', + this, + isErrorEvent ? error : undefined + ); + } else if (this.#socketEstablished) { + this.#handleConnectionLoss(error); + } else { + this.#handleAttemptFailure(error); + } + }; + + ws.onclose = event => { + if (!isCurrent()) return; + if ( + this.#automaticReconnect && + this.#hasEverConnected && + !this.#socketEstablished && + event.code === SESSION_BUSY_CLOSE_CODE + ) { + // The server is tearing down the previous session holder. + this.#discardSocket(); + const delayMs = computeReconnectDelayMs(1); + this.#emitter.emit( + 'connectError', + this, + new Error('Session busy'), + this.#reconnectAttempt, + delayMs + ); + this.#scheduleReconnect(this.#reconnectAttempt, delayMs); + return; + } + const message = `WebSocket closed (code ${event.code}${event.reason ? `: ${event.reason}` : ''})`; + const error = [1002, 1003, 1007, 1008].includes(event.code) + ? new WebSocketProtocolError(message) + : new Error(message); + handleLoss(error, false); + }; + ws.onerror = event => + handleLoss(errorFromEvent(event, 'WebSocket error'), true); + ws.onopen = () => { + if (isCurrent()) this.#handleOnOpen(); + }; + ws.onmessage = message => { + if (isCurrent()) this.#handleOnMessage(message); + }; + return ws; + } catch (e) { + if (generation !== this.#socketGeneration || this.#connectionEnded) { + return undefined; + } + stdbLogger('error', 'Error connecting to SpacetimeDB WS'); + this.isActive = false; + this.#handleAttemptFailure(errorFromEvent(e, 'Failed to connect')); + return undefined; + } + } + + async #tokenForAttempt(): Promise { + // The initial connection uses the token the application configured. + if (!this.#tokenProvider || !this.#hasEverConnected) { + return this.token; + } + if (!this.#forceTokenRefresh && !tokenNeedsRefresh(this.token)) { + return this.token; + } + const token = await this.#tokenProvider(); + this.#forceTokenRefresh = false; + this.#usedFreshToken = true; + return token; + } + + #handleConnectionLoss(error: Error): void { + if (this.#connectionEnded) { + return; + } + this.#discardSocket(); + this.#failInFlightCalls(new UnknownCallResultError()); + + const willReconnect = + this.#automaticReconnect && + !this.isDisconnectRequested && + !(error instanceof WebSocketProtocolError); + + if (!willReconnect) { + this.#endConnection(error); + return; + } + + const attempt = this.#reconnectAttempt + 1; + const delayMs = computeReconnectDelayMs(attempt); + this.#emitter.emit('disconnect', this, error, attempt, delayMs); + this.#scheduleReconnect(attempt, delayMs); + } + + #handleAttemptFailure(error: Error): void { + if (this.#connectionEnded) { + return; + } + this.#discardSocket(); + this.#failInFlightCalls(new UnknownCallResultError()); + + // A failed *initial* connection is not retried: the cause is usually a + // misconfigured URI or database name that no retry will fix. + const willReconnect = + this.#automaticReconnect && + this.#hasEverConnected && + !this.isDisconnectRequested && + !isTerminalConnectError(error) && + !(isAuthError(error) && (!this.#tokenProvider || this.#usedFreshToken)); + + if (!willReconnect) { + this.#endConnection(undefined, { alreadyReported: true }); + this.#emitter.emit('connectError', this, error); + return; + } + + // A rejected token is worth one forced refresh: the retained token may + // have been revoked, or the clock may be skewed. + if (this.#tokenProvider && isAuthError(error)) { + this.#forceTokenRefresh = true; + } + + const attempt = this.#reconnectAttempt + 1; + const delayMs = computeReconnectDelayMs(attempt); + this.#emitter.emit('connectError', this, error, attempt, delayMs); + this.#scheduleReconnect(attempt, delayMs); + } + + #scheduleReconnect(attempt: number, delayMs: number): void { + if (this.#connectionEnded || this.isDisconnectRequested) return; + this.#reconnectAttempt = attempt; + this.#clearReconnectTimer(); + this.#reconnectTimer = setTimeout(() => { + this.#reconnectTimer = undefined; + if (this.#connectionEnded || this.isDisconnectRequested) { + return; + } + this.wsPromise = this.#openSocket(); + }, delayMs); + } + + #clearReconnectTimer(): void { + if (this.#reconnectTimer !== undefined) { + clearTimeout(this.#reconnectTimer); + this.#reconnectTimer = undefined; + } + } + + #endConnection( + disconnectError: Error | undefined, + options?: { alreadyReported?: boolean } + ): void { + if (this.#connectionEnded) { + return; + } + this.#connectionEnded = true; + this.isActive = false; + this.#clearReconnectTimer(); + this.#discardSocket(); + this.#removeLivenessListeners(); + this.#failInFlightCalls(new UnknownCallResultError()); + if (!options?.alreadyReported) { + this.#emitter.emit('disconnect', this, disconnectError); + } + } + + #failInFlightCalls(error: Error): void { + const rejecters = [...this.#pendingCallRejecters.values()]; + this.#pendingCallRejecters.clear(); + this.#reducerCallbacks.clear(); + this.#reducerCallInfo.clear(); + this.#procedureCallbacks.clear(); + + for (const reject of rejecters) { + reject(error); + } + } + + /** True while the SDK is between a lost connection and a completed reconnect. */ + get isReconnecting(): boolean { + return ( + this.#automaticReconnect && + this.#hasEverConnected && + !this.isActive && + !this.#connectionEnded + ); + } + + #discardSocket(): void { + this.#socketGeneration += 1; + this.isActive = false; + this.#socketEstablished = false; + this.#pendingReplay = undefined; + this.#preparingReplay = this.#hasEverConnected; + this.#outboundQueue.length = 0; + this.#inboundQueue.length = 0; + const ws = this.ws; + this.ws = undefined; + ws?.close(); + for (const [id, entry] of this.#subscriptionManager.subscriptions) { + if (entry.unsubscribeRequested) this.#endSubscription(id); + } + } + + // Resume events recover sockets that closed silently while the page was frozen. + #installLivenessListeners(): void { + if (!this.#automaticReconnect || this.#livenessCleanup) { + return; + } + const doc = typeof document !== 'undefined' ? document : undefined; + const win = typeof window !== 'undefined' ? window : undefined; + if (!doc && !win) { + return; + } + + const onResume = (): void => this.#handleLivenessResume(); + const onVisibilityChange = (): void => { + if (doc?.visibilityState === 'visible') { + onResume(); + } + }; + + doc?.addEventListener('visibilitychange', onVisibilityChange); + win?.addEventListener('focus', onResume); + win?.addEventListener('online', onResume); + win?.addEventListener('pageshow', onResume); + + this.#livenessCleanup = () => { + doc?.removeEventListener('visibilitychange', onVisibilityChange); + win?.removeEventListener('focus', onResume); + win?.removeEventListener('online', onResume); + win?.removeEventListener('pageshow', onResume); + this.#livenessCleanup = undefined; + }; + } + + #removeLivenessListeners(): void { + this.#livenessCleanup?.(); + } + + #handleLivenessResume(): void { + if (this.#connectionEnded || this.isDisconnectRequested) { + return; + } + if (this.isSocketClosed) { + const error = new Error('WebSocket closed while suspended'); + if (this.#socketEstablished) this.#handleConnectionLoss(error); + else this.#handleAttemptFailure(error); + return; + } + if (this.#reconnectTimer !== undefined) { + // Retry now rather than waiting out a backoff computed before the pause. + this.#clearReconnectTimer(); + this.wsPromise = this.#openSocket(); + } } #getNextQueryId = () => { @@ -531,7 +1006,14 @@ export class DbConnectionImpl this.#subscriptionManager.subscriptions.set(querySetId, { handle, emitter: handleEmitter, + // Retained so the subscription can be replayed after a reconnect. + querySql: [...querySql], }); + if (!this.#preparingReplay) this.#sendSubscription(querySetId, querySql); + return querySetId; + } + + #sendSubscription(querySetId: number, querySql: string[]): void { const requestId = this.#getNextRequestId(); this.#sendMessage( ClientMessage.Subscribe({ @@ -540,10 +1022,145 @@ export class DbConnectionImpl requestId, }) ); - return querySetId; + } + + #replaySubscriptions(): void { + const entries = [...this.#subscriptionManager.subscriptions.entries()]; + const sets: SubscribeBatch['sets'] = []; + const replayed = new Map>(); + for (const [, entry] of entries) { + const querySetId = this.#getNextQueryId(); + entry.handle.rebindQuerySetId(querySetId); + replayed.set(querySetId, entry); + sets.push({ + querySetId: { id: querySetId }, + queryStrings: entry.querySql, + }); + } + this.#subscriptionManager.subscriptions = replayed; + + const requestId = this.#getNextRequestId(); + this.#pendingReplay = { + requestId, + querySetIds: new Set(replayed.keys()), + }; + this.#preparingReplay = false; + if (sets.length === 0) { + this.#applyReplayBatch({ requestId, results: [] }); + return; + } + this.#sendMessage( + ClientMessage.SubscribeBatch({ + requestId, + sets, + }) + ); + } + + #applyReplayBatch(applied: SubscribeBatchApplied): void { + const pending = this.#pendingReplay; + const resultIds = new Set( + applied.results.map(result => result.querySetId.id) + ); + if ( + !pending || + pending.requestId !== applied.requestId || + resultIds.size !== applied.results.length || + resultIds.size !== pending.querySetIds.size || + [...resultIds].some(id => !pending.querySetIds.has(id)) + ) { + this.#handleProtocolError( + new Error('Unexpected subscription replay response') + ); + return; + } + this.#pendingReplay = undefined; + + const event: Event = { + id: this.#nextEventId(), + tag: 'SubscribeApplied', + }; + const eventContext = this.#makeEventContext(event); + + // The removal half: every row the cache holds from the old connection. + // Only tables which have been populated exist in the cache. + const tableUpdates: CacheTableUpdate[] = []; + for (const [tableName, table] of this.clientCache.tables) { + const operations = table.snapshotDeleteOperations(); + if (operations.length > 0) { + tableUpdates.push({ tableName, operations }); + } + } + + // The addition half: the rows of every set which applied. + const failures: { + entry: SubscriptionEntry; + error: string; + }[] = []; + for (const result of applied.results) { + const entry = this.#subscriptionManager.subscriptions.get( + result.querySetId.id + ); + if (!entry) { + continue; + } + if (result.outcome.tag === 'Error') { + // The set is not registered, so drop it and report it below. + this.#subscriptionManager.subscriptions.delete(result.querySetId.id); + failures.push({ entry, error: result.outcome.value }); + continue; + } + tableUpdates.push( + ...this.#queryRowsToTableUpdates(result.outcome.value, 'insert') + ); + } + + const merged = this.#mergeTableUpdates(tableUpdates); + const callbacks = this.#applyTableUpdates(merged, eventContext, { + // A row which is unchanged across the outage appears as a + // delete/insert pair and must produce no callback. + skipIdenticalUpdates: true, + }); + const { event: _, ...subscriptionEventContext } = eventContext; + for (const [querySetId, entry] of this.#subscriptionManager.subscriptions) { + if (pending.querySetIds.has(querySetId)) { + entry.emitter.emit('applied', subscriptionEventContext); + } + } + for (const { entry, error: message } of failures) { + const error = Error(message); + const errorEventContext = this.#makeEventContext({ + id: this.#nextEventId(), + tag: 'Error', + value: error, + }); + entry.emitter.emit( + 'error', + { ...errorEventContext, event: error }, + error + ); + } + this.#dispatchPendingCallbacks(callbacks); + } + + #endSubscription(querySetId: number): void { + const entry = this.#subscriptionManager.subscriptions.get(querySetId); + this.#subscriptionManager.subscriptions.delete(querySetId); + const { event: _, ...ctx } = this.#makeEventContext({ + id: this.#nextEventId(), + tag: 'UnsubscribeApplied', + }); + entry?.emitter.emit('end', ctx); } unregisterSubscription(querySetId: number): void { + const entry = this.#subscriptionManager.subscriptions.get(querySetId); + if (!entry) return; + entry.unsubscribeRequested = true; + if (this.#automaticReconnect && (!this.isActive || this.#preparingReplay)) { + this.#endSubscription(querySetId); + return; + } const requestId = this.#getNextRequestId(); this.#sendMessage( ClientMessage.Unsubscribe({ @@ -772,6 +1389,12 @@ export class DbConnectionImpl } } + #rejectCallIfDisconnected(): DisconnectedError | undefined { + return this.#automaticReconnect && !this.isActive + ? new DisconnectedError() + : undefined; + } + #sendMessage(message: ClientMessage): void { const writer = this.#clientMessageEncoder; writer.clear(); @@ -840,15 +1463,17 @@ export class DbConnectionImpl if (this.ws) { this.#negotiatedWsProtocol = normalizeWsProtocol(this.ws.protocol); } - this.isActive = true; - if (this.ws) { + this.isActive = !this.#automaticReconnect; + this.#installLivenessListeners(); + if (this.ws && this.isActive) { this.#flushOutboundQueue(this.ws); } } #applyTableUpdates( tableUpdates: CacheTableUpdate[], - eventContext: EventContextInterface + eventContext: EventContextInterface, + options?: { skipIdenticalUpdates?: boolean } ): PendingCallback[] { const pendingCallbacks: PendingCallback[] = []; for (const tableUpdate of tableUpdates) { @@ -860,7 +1485,8 @@ export class DbConnectionImpl tableUpdate.operations as Operation< RowType> >[], - eventContext + eventContext, + options ); for (const callback of newCallbacks) { pendingCallbacks.push(callback); @@ -902,14 +1528,54 @@ export class DbConnectionImpl 'trace', () => `Processing server message: ${stringify(serverMessage)}` ); + if ( + this.#automaticReconnect && + (serverMessage.tag === 'InitialConnection') === this.#socketEstablished + ) { + this.#handleProtocolError( + new Error('Unexpected message during connection handshake') + ); + return; + } switch (serverMessage.tag) { case 'InitialConnection': { + const isReconnect = this.#hasEverConnected; + if ( + isReconnect && + this.identity && + !this.identity.isEqual(serverMessage.value.identity) + ) { + // Retrying cannot recover the old identity, so stop here rather + // than serving the application someone else's data. + const error = new IdentityChangedError(); + this.#endConnection(undefined, { alreadyReported: true }); + this.#emitter.emit('connectError', this, error); + break; + } + this.identity = serverMessage.value.identity; + // The server issues a token on the first connection; retain it so + // reconnects present the same identity. if (!this.token && serverMessage.value.token) { this.token = serverMessage.value.token; } this.#setConnectionId(serverMessage.value.connectionId); + this.isActive = true; + this.#hasEverConnected = true; + this.#socketEstablished = true; + // A connection was established, so the backoff schedule starts over. + this.#reconnectAttempt = 0; this.#emitter.emit('connect', this, this.identity, this.token); + if (this.#connectionEnded) break; + if (isReconnect) { + this.#replaySubscriptions(); + } else if (this.#preparingReplay) { + this.#preparingReplay = false; + for (const [id, entry] of this.#subscriptionManager.subscriptions) { + this.#sendSubscription(id, entry.querySql); + } + } + if (this.ws) this.#flushOutboundQueue(this.ws); break; } case 'SubscribeApplied': { @@ -1073,13 +1739,24 @@ export class DbConnectionImpl ); break; } + case 'SubscribeBatchApplied': { + this.#applyReplayBatch(serverMessage.value); + break; + } } } #processV2Message(data: Uint8Array): void { const reader = this.#messageReader; reader.reset(data); - this.#processServerMessage(ServerMessage.deserialize(reader)); + let message: ServerMessage; + try { + message = ServerMessage.deserialize(reader); + } catch (cause) { + this.#handleProtocolError(cause); + return; + } + this.#processServerMessage(message); } #processMessage(data: Uint8Array): void { @@ -1088,17 +1765,28 @@ export class DbConnectionImpl return; } - const messageCount = forEachServerMessageV3( - this.#messageReader, - data, - serverMessage => { + let dispatching = false; + const generation = this.#socketGeneration; + try { + forEachServerMessageV3(this.#messageReader, data, serverMessage => { + if (generation !== this.#socketGeneration) return; + dispatching = true; this.#processServerMessage(serverMessage); - } - ); - stdbLogger( - 'trace', - () => `Processing server v3 payload with ${messageCount} message(s)` - ); + dispatching = false; + }); + } catch (cause) { + if (dispatching) throw cause; + this.#handleProtocolError(cause); + } + } + + #handleProtocolError(cause: unknown): void { + if (!this.#automaticReconnect) throw cause; + const error = new WebSocketProtocolError('Invalid server message', { + cause, + }); + if (this.#socketEstablished) this.#handleConnectionLoss(error); + else this.#handleAttemptFailure(error); } /** @@ -1166,6 +1854,10 @@ export class DbConnectionImpl argsBuffer: Uint8Array, reducerArgs?: object ): Promise { + const rejected = this.#rejectCallIfDisconnected(); + if (rejected) { + return Promise.reject(rejected); + } const { promise, resolve, reject } = createDeferred(); const requestId = this.#getNextRequestId(); this.#sendCallReducerMessage(requestId, encodedReducerName, argsBuffer); @@ -1175,7 +1867,9 @@ export class DbConnectionImpl args: reducerArgs, }); } + this.#pendingCallRejecters.set(requestId, reject); this.#reducerCallbacks.set(requestId, result => { + this.#pendingCallRejecters.delete(requestId); if (result.tag === 'Ok' || result.tag === 'OkEmpty') { resolve(); } else { @@ -1201,6 +1895,10 @@ export class DbConnectionImpl argsBuffer: Uint8Array, reducerArgs?: object ): Promise { + const rejected = this.#rejectCallIfDisconnected(); + if (rejected) { + return Promise.reject(rejected); + } const { promise, resolve, reject } = createDeferred(); const requestId = this.#getNextRequestId(); const message = ClientMessage.CallReducer({ @@ -1216,7 +1914,9 @@ export class DbConnectionImpl args: reducerArgs, }); } + this.#pendingCallRejecters.set(requestId, reject); this.#reducerCallbacks.set(requestId, result => { + this.#pendingCallRejecters.delete(requestId); if (result.tag === 'Ok' || result.tag === 'OkEmpty') { resolve(); } else { @@ -1282,10 +1982,16 @@ export class DbConnectionImpl encodedProcedureName: Uint8Array, argsBuffer: Uint8Array ): Promise { + const rejected = this.#rejectCallIfDisconnected(); + if (rejected) { + return Promise.reject(rejected); + } const { promise, resolve, reject } = createDeferred(); const requestId = this.#getNextRequestId(); this.#sendCallProcedureMessage(requestId, encodedProcedureName, argsBuffer); + this.#pendingCallRejecters.set(requestId, reject); this.#procedureCallbacks.set(requestId, result => { + this.#pendingCallRejecters.delete(requestId); if (result.tag === 'Ok') { resolve(result.value); } else { @@ -1299,6 +2005,10 @@ export class DbConnectionImpl procedureName: string, argsBuffer: Uint8Array ): Promise { + const rejected = this.#rejectCallIfDisconnected(); + if (rejected) { + return Promise.reject(rejected); + } const { promise, resolve, reject } = createDeferred(); const requestId = this.#getNextRequestId(); const message = ClientMessage.CallProcedure({ @@ -1309,7 +2019,9 @@ export class DbConnectionImpl flags: 0, }); this.#sendMessage(message); + this.#pendingCallRejecters.set(requestId, reject); this.#procedureCallbacks.set(requestId, result => { + this.#pendingCallRejecters.delete(requestId); if (result.tag === 'Ok') { resolve(result.value); } else { @@ -1355,55 +2067,87 @@ export class DbConnectionImpl */ disconnect(): void { this.isDisconnectRequested = true; - this.wsPromise.then(ws => ws?.close()); + if (this.#automaticReconnect) { + if (this.#connectionEnded) { + this.#emitter.emit('disconnect', this); + } else { + this.#endConnection(undefined); + } + } else { + this.wsPromise.then(ws => ws?.close()); + } } - private on( - eventName: ConnectionEvent, - callback: (ctx: DbConnectionImpl, ...args: any[]) => void + private on( + eventName: E, + callback: ( + ctx: DbConnectionImpl, + ...args: ConnectionEventArgs[E] + ) => void ): void { this.#emitter.on(eventName, callback); } - private off( - eventName: ConnectionEvent, - callback: (ctx: DbConnectionImpl, ...args: any[]) => void + private off( + eventName: E, + callback: ( + ctx: DbConnectionImpl, + ...args: ConnectionEventArgs[E] + ) => void ): void { this.#emitter.off(eventName, callback); } private onConnect( - callback: (ctx: DbConnectionImpl, ...args: any[]) => void + callback: ( + ctx: DbConnectionImpl, + ...args: ConnectionEventArgs['connect'] + ) => void ): void { this.#emitter.on('connect', callback); } private onDisconnect( - callback: (ctx: DbConnectionImpl, ...args: any[]) => void + callback: ( + ctx: DbConnectionImpl, + ...args: ConnectionEventArgs['disconnect'] + ) => void ): void { this.#emitter.on('disconnect', callback); } private onConnectError( - callback: (ctx: DbConnectionImpl, ...args: any[]) => void + callback: ( + ctx: DbConnectionImpl, + ...args: ConnectionEventArgs['connectError'] + ) => void ): void { this.#emitter.on('connectError', callback); } removeOnConnect( - callback: (ctx: DbConnectionImpl, ...args: any[]) => void + callback: ( + ctx: DbConnectionImpl, + ...args: ConnectionEventArgs['connect'] + ) => void ): void { this.#emitter.off('connect', callback); } removeOnDisconnect( - callback: (ctx: DbConnectionImpl, ...args: any[]) => void + callback: ( + ctx: DbConnectionImpl, + ...args: ConnectionEventArgs['disconnect'] + ) => void ): void { this.#emitter.off('disconnect', callback); } removeOnConnectError( - callback: (ctx: DbConnectionImpl, ...args: any[]) => void + callback: ( + ctx: DbConnectionImpl, + ...args: ConnectionEventArgs['connectError'] + ) => void ): void { this.#emitter.off('connectError', callback); } diff --git a/crates/bindings-typescript/src/sdk/subscription_builder_impl.ts b/crates/bindings-typescript/src/sdk/subscription_builder_impl.ts index f7e271b4e85..cf72dad84b2 100644 --- a/crates/bindings-typescript/src/sdk/subscription_builder_impl.ts +++ b/crates/bindings-typescript/src/sdk/subscription_builder_impl.ts @@ -160,14 +160,15 @@ export class SubscriptionBuilderImpl { export type SubscribeEvent = 'applied' | 'error' | 'end'; +export type SubscriptionEntry = { + handle: SubscriptionHandleImpl; + emitter: EventEmitter; + querySql: string[]; + unsubscribeRequested?: boolean; +}; + export class SubscriptionManager { - subscriptions: Map< - number, - { - handle: SubscriptionHandleImpl; - emitter: EventEmitter; - } - > = new Map(); + subscriptions: Map> = new Map(); } export class SubscriptionHandleImpl { @@ -210,6 +211,11 @@ export class SubscriptionHandleImpl { ); } + /** @internal Rebind the retained handle to its replayed query set. */ + rebindQuerySetId(querySetId: number): void { + this.#querySetId = querySetId; + } + /** * Consumes self and issues an `Unsubscribe` message, * removing this query from the client's set of subscribed queries. @@ -220,7 +226,6 @@ export class SubscriptionHandleImpl { throw new Error('Unsubscribe has already been called'); } this.#unsubscribeCalled = true; - this.db.unregisterSubscription(this.#querySetId); this.#emitter.on( 'end', (_ctx: SubscriptionEventContextInterface) => { @@ -228,6 +233,7 @@ export class SubscriptionHandleImpl { this.#activeState = false; } ); + this.db.unregisterSubscription(this.#querySetId); } /** @@ -250,7 +256,6 @@ export class SubscriptionHandleImpl { throw new Error('Unsubscribe has already been called'); } this.#unsubscribeCalled = true; - this.db.unregisterSubscription(this.#querySetId); this.#emitter.on( 'end', (ctx: SubscriptionEventContextInterface) => { @@ -259,6 +264,7 @@ export class SubscriptionHandleImpl { onEnd(ctx); } ); + this.db.unregisterSubscription(this.#querySetId); } /** diff --git a/crates/bindings-typescript/src/sdk/table_cache.ts b/crates/bindings-typescript/src/sdk/table_cache.ts index 543d3fb78ed..32184e73426 100644 --- a/crates/bindings-typescript/src/sdk/table_cache.ts +++ b/crates/bindings-typescript/src/sdk/table_cache.ts @@ -262,11 +262,30 @@ export class TableCacheImpl< return this.iter(); } + /** Delete every cached reference, including overlaps between subscriptions. */ + snapshotDeleteOperations = (): Operation< + RowType> + >[] => { + const operations: Operation< + RowType> + >[] = []; + for (const [rowId, [row, refCount]] of this.rows) { + for (let i = 0; i < refCount; i++) { + operations.push({ type: 'delete', rowId, row }); + } + } + return operations; + }; + applyOperations = ( operations: Operation< RowType> >[], - ctx: EventContextInterface + ctx: EventContextInterface, + options?: { + /** Suppress unchanged rows during reconnect reconciliation. */ + skipIdenticalUpdates?: boolean; + } ): PendingCallback[] => { const pendingCallbacks: PendingCallback[] = []; @@ -286,7 +305,7 @@ export class TableCacheImpl< return pendingCallbacks; } - if (this.hasPrimaryKey) { + if (this.hasPrimaryKey || options?.skipIdenticalUpdates) { const insertMap = new Map< ComparablePrimitive, [ @@ -322,7 +341,8 @@ export class TableCacheImpl< ctx, primaryKey, insertOp.row, - refCountDelta + refCountDelta, + options?.skipIdenticalUpdates ); if (maybeCb) { pendingCallbacks.push(maybeCb); @@ -363,7 +383,8 @@ export class TableCacheImpl< ctx: EventContextInterface, rowId: ComparablePrimitive, newRow: RowType>, - refCountDelta: number = 0 + refCountDelta: number = 0, + skipIfIdentical: boolean = false ): PendingCallback | undefined => { const existingEntry = this.rows.get(rowId); if (!existingEntry) { @@ -384,6 +405,11 @@ export class TableCacheImpl< return undefined; } this.rows.set(rowId, [newRow, refCount]); + if (skipIfIdentical && deepEqual(oldRow, newRow)) { + // The row is unchanged; the reference count was adjusted above but no + // callback fires. + return undefined; + } // This indicates something is wrong, so we could arguably crash here. if (previousCount === 0) { stdbLogger( diff --git a/crates/bindings-typescript/src/sdk/websocket_test_adapter.ts b/crates/bindings-typescript/src/sdk/websocket_test_adapter.ts index 257f9cb2806..07ca43b3a89 100644 --- a/crates/bindings-typescript/src/sdk/websocket_test_adapter.ts +++ b/crates/bindings-typescript/src/sdk/websocket_test_adapter.ts @@ -1,7 +1,7 @@ import BinaryReader from '../lib/binary_reader.ts'; import BinaryWriter from '../lib/binary_writer.ts'; import { ClientMessage, ServerMessage } from './client_api/types'; -import type { WebSocketAdapter, WebSocketFactory } from './ws'; +import type { WebSocketAdapter, WebSocketArgs, WebSocketFactory } from './ws'; import { PREFERRED_WS_PROTOCOLS, V3_WS_PROTOCOL } from './websocket_protocols'; import { decodeClientMessagesV3, @@ -11,6 +11,9 @@ import { class WebsocketTestAdapter implements WebSocketAdapter { protocol: string = ''; + /** The arguments the connection passed to `openWebSocket`, for assertions. */ + connectArgs?: WebSocketArgs; + // WebSocket.CLOSED (3) / WebSocket.OPEN (1). Uses literals rather than the // `WebSocket` global, which is not defined when these tests run under Node. get readyState(): number { @@ -51,7 +54,15 @@ class WebsocketTestAdapter implements WebSocketAdapter { } error(error: Error): void { - this.#onerror(error as unknown as ErrorEvent); + this.#onerror( + Object.assign(new Event('error'), { + error, + message: error.message, + filename: '', + lineno: 0, + colno: 0, + }) + ); } send(message: Uint8Array): void { @@ -70,12 +81,31 @@ class WebsocketTestAdapter implements WebSocketAdapter { } close(): void { + this.serverClose(1000, 'normal closure', true); + } + + /** + * Simulate a close initiated by the server or the network, with an + * arbitrary close code (e.g. an abnormal closure or an + * application-specific code such as session-expired). + */ + serverClose( + code: number, + reason: string = '', + wasClean: boolean = false + ): void { + this.closed = true; + this.#onclose( + Object.assign(new Event('close'), { code, reason, wasClean }) + ); + } + + /** + * Mark the socket as closed without delivering any event, simulating a + * socket that died while the page was suspended (a "zombie" socket). + */ + dieSilently(): void { this.closed = true; - this.#onclose({ - code: 1000, - reason: 'normal closure', - wasClean: true, - } as CloseEvent); } acceptConnection(): void { @@ -98,16 +128,52 @@ class WebsocketTestAdapter implements WebSocketAdapter { this.#onmessage({ data: outboundData }); } - openWebSocket: WebSocketFactory = async ({ wsProtocol }) => { - const negotiatedProtocol = wsProtocol.find(protocol => + openWebSocket: WebSocketFactory = async args => { + const negotiatedProtocol = args.wsProtocol.find(protocol => this.supportedProtocols.includes(protocol) ); if (!negotiatedProtocol) { throw new Error('No compatible websocket protocol'); } this.protocol = negotiatedProtocol; + this.connectArgs = args; return this; }; } +/** + * A websocket factory that hands out a fresh {@link WebsocketTestAdapter} per + * connection attempt and records them all. Used to test automatic + * reconnection, where each attempt opens a new socket. + */ +export class WebsocketTestAdapterFactory { + /** Every adapter created so far, in creation order. */ + sockets: WebsocketTestAdapter[] = []; + /** + * When set, the next `openWebSocket` calls reject with this error instead + * of producing a socket (simulating an unreachable server or a failed + * token exchange). + */ + connectError?: Error; + + /** The most recently created adapter. */ + get current(): WebsocketTestAdapter { + const socket = this.sockets[this.sockets.length - 1]; + if (!socket) { + throw new Error('No websocket has been opened yet'); + } + return socket; + } + + openWebSocket: WebSocketFactory = async args => { + if (this.connectError) { + throw this.connectError; + } + const adapter = new WebsocketTestAdapter(); + await adapter.openWebSocket(args); + this.sockets.push(adapter); + return adapter; + }; +} + export default WebsocketTestAdapter; diff --git a/crates/bindings-typescript/src/sdk/ws.ts b/crates/bindings-typescript/src/sdk/ws.ts index 99d1b688ece..96eb7e5fe51 100644 --- a/crates/bindings-typescript/src/sdk/ws.ts +++ b/crates/bindings-typescript/src/sdk/ws.ts @@ -49,6 +49,16 @@ export interface WebSocketAdapter { set onerror(handler: (msg: ErrorEvent) => void); } +export class WebSocketTokenError extends Error { + constructor( + readonly status: number, + statusText: string + ) { + super(`Failed to verify token: ${status} ${statusText}`); + this.name = 'WebSocketTokenError'; + } +} + export interface WebSocketArgs { url: URL; wsProtocol: string[]; @@ -57,6 +67,10 @@ export interface WebSocketArgs { compression: 'gzip' | 'brotli' | 'none'; lightMode: boolean; confirmedReads?: boolean; + /** Hex-encoded id for this socket. */ + connectionId?: string; + /** Stable session id, sent only when automatic reconnection is enabled. */ + sessionId?: string; } export type WebSocketFactory = ( args: WebSocketArgs @@ -74,6 +88,8 @@ export async function openWebSocket({ compression, lightMode, confirmedReads, + connectionId, + sessionId, }: WebSocketArgs): Promise { const headers = new Headers(); @@ -92,7 +108,7 @@ export async function openWebSocket({ const { token } = await response.json(); temporaryAuthToken = token; } else { - throw new Error(`Failed to verify token: ${response.statusText}`); + throw new WebSocketTokenError(response.status, response.statusText); } } @@ -110,6 +126,14 @@ export async function openWebSocket({ if (confirmedReads !== undefined) { databaseUrl.searchParams.set('confirmed', confirmedReads.toString()); } + // Note that `url`'s own query parameters are not carried over by the `URL` + // constructor above, so these must be set here. + if (connectionId) { + databaseUrl.searchParams.set('connection_id', connectionId); + } + if (sessionId) { + databaseUrl.searchParams.set('session_id', sessionId); + } const ws = new WS(databaseUrl.toString(), wsProtocol); ws.binaryType = 'arraybuffer'; diff --git a/crates/bindings-typescript/tests/connection_manager_liveness.test.ts b/crates/bindings-typescript/tests/connection_manager_liveness.test.ts deleted file mode 100644 index 5c2e1e75c98..00000000000 --- a/crates/bindings-typescript/tests/connection_manager_liveness.test.ts +++ /dev/null @@ -1,319 +0,0 @@ -import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; -import { ConnectionId } from '../src'; -import { connectionManagerReconnectDelayMs } from '../src/sdk/connection_manager.ts'; - -// These tests exercise the page-resume + zombie-socket liveness recovery in the -// ConnectionManager. That logic wires itself to `document`/`window` events in -// its constructor, so each test installs minimal DOM stubs and re-imports the -// module to get a fresh singleton bound to those stubs. - -type ErrorContextInterface = { isActive: boolean }; - -class MockConnection { - isActive = false; - identity = undefined; - // A real DbConnectionImpl is constructed with the builder's token and keeps - // it in this field, so the mock takes it the same way. - token: string | undefined; - connectionId = ConnectionId.random(); - isDisconnectRequested = false; - disconnected = false; - // Controls the `isSocketClosed` signal the manager reads to detect a socket - // that died silently (CLOSING/CLOSED without a clean `onclose`). - socketClosed = false; - - #onConnect = new Set<(conn: MockConnection) => void>(); - #onDisconnect = new Set< - (ctx: ErrorContextInterface, error?: Error) => void - >(); - #onConnectError = new Set< - (ctx: ErrorContextInterface, error: Error) => void - >(); - - get isSocketClosed(): boolean { - return this.socketClosed; - } - - disconnect(): void { - this.isDisconnectRequested = true; - this.disconnected = true; - this.isActive = false; - } - - removeOnConnect(cb: (conn: MockConnection) => void): void { - this.#onConnect.delete(cb); - } - removeOnDisconnect( - cb: (ctx: ErrorContextInterface, error?: Error) => void - ): void { - this.#onDisconnect.delete(cb); - } - removeOnConnectError( - cb: (ctx: ErrorContextInterface, error: Error) => void - ): void { - this.#onConnectError.delete(cb); - } - - register( - type: 'connect' | 'disconnect' | 'connectError', - cb: (...args: any[]) => void - ): void { - if (type === 'connect') this.#onConnect.add(cb); - else if (type === 'disconnect') this.#onDisconnect.add(cb); - else this.#onConnectError.add(cb); - } - - /** - * @param issuedToken - the token the server hands back on connect, emulating - * a client being issued credentials it did not have when the builder was - * constructed. - */ - simulateConnect(issuedToken?: string): void { - this.isActive = true; - if (issuedToken !== undefined) this.token = issuedToken; - for (const cb of this.#onConnect) cb(this); - } - simulateDisconnect(error?: Error): void { - this.isActive = false; - for (const cb of this.#onDisconnect) - cb(this as unknown as ErrorContextInterface, error); - } -} - -class MockBuilder { - buildCount = 0; - connections: MockConnection[] = []; - /** The token each `build()` will stamp onto its connection. */ - token: string | undefined; - - #onConnect = new Set<(conn: MockConnection) => void>(); - #onDisconnect = new Set< - (ctx: ErrorContextInterface, error?: Error) => void - >(); - #onConnectError = new Set< - (ctx: ErrorContextInterface, error: Error) => void - >(); - - withToken(token?: string): MockBuilder { - this.token = token; - return this; - } - - build(): MockConnection { - const connection = new MockConnection(); - connection.token = this.token; - this.buildCount += 1; - this.connections.push(connection); - for (const cb of this.#onConnect) connection.register('connect', cb); - for (const cb of this.#onDisconnect) connection.register('disconnect', cb); - for (const cb of this.#onConnectError) - connection.register('connectError', cb); - return connection; - } - - onConnect(cb: (conn: MockConnection) => void): MockBuilder { - this.#onConnect.add(cb); - for (const c of this.connections) c.register('connect', cb); - return this; - } - onDisconnect( - cb: (ctx: ErrorContextInterface, error?: Error) => void - ): MockBuilder { - this.#onDisconnect.add(cb); - for (const c of this.connections) c.register('disconnect', cb); - return this; - } - onConnectError( - cb: (ctx: ErrorContextInterface, error: Error) => void - ): MockBuilder { - this.#onConnectError.add(cb); - for (const c of this.connections) c.register('connectError', cb); - return this; - } -} - -let keyCounter = 0; -function nextKey(): string { - keyCounter += 1; - return `connection-manager-liveness-${keyCounter}`; -} - -type DocStub = { - visibilityState: 'visible' | 'hidden'; - addEventListener: (ev: string, h: () => void) => void; -}; - -let ConnectionManager: typeof import('../src/sdk/connection_manager.ts').ConnectionManager; -let doc: DocStub; -let listeners: Record void>>; - -function retain(key: string, builder: MockBuilder): MockConnection { - return ConnectionManager.retain( - key, - builder as any - ) as unknown as MockConnection; -} - -function fire(name: string): void { - for (const h of listeners[name] ?? []) h(); -} - -async function loadManager(): Promise { - listeners = {}; - doc = { - visibilityState: 'visible', - addEventListener: (ev, h) => { - (listeners[`doc:${ev}`] ??= []).push(h); - }, - }; - const win = { - addEventListener: (ev: string, h: () => void) => { - (listeners[`win:${ev}`] ??= []).push(h); - }, - }; - (globalThis as any).document = doc; - (globalThis as any).window = win; - vi.resetModules(); - ({ ConnectionManager } = await import('../src/sdk/connection_manager.ts')); -} - -describe('ConnectionManager liveness recovery', () => { - beforeEach(async () => { - // Fake timers let us drive the reconnect backoff deterministically. - vi.useFakeTimers(); - await loadManager(); - }); - - afterEach(() => { - vi.runOnlyPendingTimers(); - vi.useRealTimers(); - delete (globalThis as any).document; - delete (globalThis as any).window; - }); - - test('registers resume + network listeners on construction', () => { - expect(listeners['doc:visibilitychange']?.length).toBe(1); - expect(listeners['win:focus']?.length).toBe(1); - expect(listeners['win:online']?.length).toBe(1); - expect(listeners['win:pageshow']?.length).toBe(1); - }); - - test('revives a silently-dead socket when the network returns', () => { - const key = nextKey(); - const builder = new MockBuilder(); - const first = retain(key, builder); - first.simulateConnect(); - expect(ConnectionManager.getSnapshot(key)?.isActive).toBe(true); - - // Socket dies while backgrounded: no disconnect event ever fires, but the - // underlying readyState is now CLOSED. - first.socketClosed = true; - expect(ConnectionManager.getConnection(key)).toBe(first); - - fire('win:online'); - - expect(builder.buildCount).toBe(2); - expect(ConnectionManager.getConnection(key)).toBe(builder.connections[1]); - ConnectionManager.release(key); - }); - - test('does not rebuild a healthy connection on resume', () => { - const key = nextKey(); - const builder = new MockBuilder(); - const first = retain(key, builder); - first.simulateConnect(); - - fire('win:focus'); - fire('doc:visibilitychange'); - - expect(builder.buildCount).toBe(1); - expect(ConnectionManager.getConnection(key)).toBe(first); - ConnectionManager.release(key); - }); - - test('does not revive a connection that was intentionally disconnected', () => { - const key = nextKey(); - const builder = new MockBuilder(); - const first = retain(key, builder); - first.simulateConnect(); - first.isDisconnectRequested = true; - first.socketClosed = true; - - fire('win:online'); - - expect(builder.buildCount).toBe(1); - ConnectionManager.release(key); - }); - - test('brings a stalled reconnect forward on resume and resets backoff', () => { - const key = nextKey(); - const builder = new MockBuilder(); - const first = retain(key, builder); - first.simulateDisconnect(); - - // The reconnect timer is scheduled but has not fired yet (simulating a - // background tab whose timers are throttled/frozen). - vi.advanceTimersByTime(connectionManagerReconnectDelayMs(0) - 1); - expect(builder.buildCount).toBe(1); - - // Regaining focus rebuilds immediately instead of waiting out the delay. - fire('doc:visibilitychange'); - expect(builder.buildCount).toBe(2); - - // Backoff was reset: the next failure reconnects after the base delay. - builder.connections[1].simulateDisconnect(); - vi.advanceTimersByTime(connectionManagerReconnectDelayMs(0)); - expect(builder.buildCount).toBe(3); - - ConnectionManager.release(key); - }); - - test('visibilitychange while still hidden does not reconnect', () => { - const key = nextKey(); - const builder = new MockBuilder(); - const first = retain(key, builder); - first.simulateDisconnect(); - - doc.visibilityState = 'hidden'; - fire('doc:visibilitychange'); - - expect(builder.buildCount).toBe(1); - ConnectionManager.release(key); - }); - - // The resume paths rebuild from the retained builder, whose token is a - // snapshot from before the session existed. Reconnecting anonymously here - // makes the server mint a new identity, so a user who merely switched tabs - // comes back as a stranger. - test('reviving a dead socket on resume keeps the session identity', () => { - const key = nextKey(); - // A first-time visitor: no stored credentials when the builder was made. - const builder = new MockBuilder(); - const first = retain(key, builder); - first.simulateConnect('session-token'); - - // Tab is backgrounded and the socket dies silently. - first.socketClosed = true; - fire('doc:visibilitychange'); - - expect(builder.buildCount).toBe(2); - expect(builder.connections[1].token).toBe('session-token'); - ConnectionManager.release(key); - }); - - test('a stalled reconnect brought forward on resume keeps the session identity', () => { - const key = nextKey(); - const builder = new MockBuilder(); - const first = retain(key, builder); - first.simulateConnect('session-token'); - first.simulateDisconnect(); - - // Timer still pending behind background throttling; focus fires it early. - vi.advanceTimersByTime(connectionManagerReconnectDelayMs(0) - 1); - fire('win:focus'); - - expect(builder.buildCount).toBe(2); - expect(builder.connections[1].token).toBe('session-token'); - ConnectionManager.release(key); - }); -}); diff --git a/crates/bindings-typescript/tests/connection_manager_reconnect.test.ts b/crates/bindings-typescript/tests/connection_manager_reconnect.test.ts index ee981ccf2f9..2d57d1ef657 100644 --- a/crates/bindings-typescript/tests/connection_manager_reconnect.test.ts +++ b/crates/bindings-typescript/tests/connection_manager_reconnect.test.ts @@ -6,6 +6,13 @@ import { ConnectionManager, } from '../src/sdk/connection_manager.ts'; +// Reconnection after a mid-session drop lives in the SDK: the manager forces +// `withAutomaticReconnect()` on every builder, and a `nextReconnectAttempt` +// on a disconnect/connect-error report means the SDK is retrying inside the +// same connection object, so the manager must leave it alone. The manager +// rebuilds only when the SDK reports it will not retry (no attempt number): +// a failed initial connection, or another terminal failure. + type ErrorContextInterface = { isActive: boolean; }; @@ -26,10 +33,20 @@ class MockConnection { #onConnectCallbacks = new Set<(conn: MockConnection) => void>(); #onDisconnectCallbacks = new Set< - (ctx: ErrorContextInterface, error?: Error) => void + ( + ctx: ErrorContextInterface, + error?: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number + ) => void >(); #onConnectErrorCallbacks = new Set< - (ctx: ErrorContextInterface, error: Error) => void + ( + ctx: ErrorContextInterface, + error: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number + ) => void >(); disconnect(): void { @@ -40,7 +57,7 @@ class MockConnection { this.disconnected = true; this.isActive = false; for (const cb of this.#onDisconnectCallbacks) { - cb(this as unknown as ErrorContextInterface); + cb(this); } } @@ -87,17 +104,29 @@ class MockConnection { } } - simulateDisconnect(error?: Error): void { + /** + * Passing `nextReconnectAttempt` emulates the SDK announcing it will retry + * internally; omitting it emulates a terminal report. + */ + simulateDisconnect( + error?: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number + ): void { this.isActive = false; for (const cb of this.#onDisconnectCallbacks) { - cb(this as unknown as ErrorContextInterface, error); + cb(this, error, nextReconnectAttempt, nextReconnectDelayMs); } } - simulateConnectError(error: Error): void { + simulateConnectError( + error: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number + ): void { this.isActive = false; for (const cb of this.#onConnectErrorCallbacks) { - cb(this as unknown as ErrorContextInterface, error); + cb(this, error, nextReconnectAttempt, nextReconnectDelayMs); } } @@ -106,13 +135,23 @@ class MockConnection { } registerOnDisconnect( - cb: (ctx: ErrorContextInterface, error?: Error) => void + cb: ( + ctx: ErrorContextInterface, + error?: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number + ) => void ): void { this.#onDisconnectCallbacks.add(cb); } registerOnConnectError( - cb: (ctx: ErrorContextInterface, error: Error) => void + cb: ( + ctx: ErrorContextInterface, + error: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number + ) => void ): void { this.#onConnectErrorCallbacks.add(cb); } @@ -125,6 +164,7 @@ class MockBuilder { token: string | undefined; /** Every token this builder was asked to carry, oldest first. */ tokenHistory: (string | undefined)[] = []; + automaticReconnect = false; constructor(token?: string) { this.token = token; @@ -144,6 +184,11 @@ class MockBuilder { return this; } + withAutomaticReconnect(): MockBuilder { + this.automaticReconnect = true; + return this; + } + build(): MockConnection { const connection = new MockConnection(this.token); this.buildCount += 1; @@ -205,7 +250,7 @@ function retainMock(key: string, builder: MockBuilder): MockConnection { ) as unknown as MockConnection; } -describe('ConnectionManager retained reconnect behavior', () => { +describe('ConnectionManager forces SDK automatic reconnection', () => { beforeEach(() => { vi.useFakeTimers(); }); @@ -215,7 +260,134 @@ describe('ConnectionManager retained reconnect behavior', () => { vi.useRealTimers(); }); - test('rebuilds a retained connection after disconnect', () => { + test('retain enables automatic reconnection on the builder', () => { + const key = nextKey(); + const builder = new MockBuilder(); + expect(builder.automaticReconnect).toBe(false); + + retainMock(key, builder); + + expect(builder.automaticReconnect).toBe(true); + + ConnectionManager.release(key); + }); + + test('rebuild enables automatic reconnection on the replacement builder', () => { + const key = nextKey(); + retainMock(key, new MockBuilder()); + + const replacement = new MockBuilder(); + ConnectionManager.rebuild(key, replacement as any); + + expect(replacement.automaticReconnect).toBe(true); + + ConnectionManager.release(key); + }); +}); + +describe('ConnectionManager during SDK-managed reconnection', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.runOnlyPendingTimers(); + vi.useRealTimers(); + }); + + test('a drop the SDK will retry does not rebuild the connection', () => { + const key = nextKey(); + const builder = new MockBuilder(); + const error = new Error('connection lost'); + + const first = retainMock(key, builder); + first.simulateConnect(); + + first.simulateDisconnect(error, 1, 1000); + + // The connection object stays managed and untouched: the SDK reconnects + // inside it. + expect(ConnectionManager.getConnection(key)).toBe(first); + expect(ConnectionManager.getSnapshot(key)?.isActive).toBe(false); + expect(ConnectionManager.getSnapshot(key)?.connectionError).toBe(error); + + vi.advanceTimersByTime(CONNECTION_MANAGER_RECONNECT_MAX_DELAY_MS); + expect(builder.buildCount).toBe(1); + + ConnectionManager.release(key); + }); + + test('failed attempts the SDK will retry do not rebuild the connection', () => { + const key = nextKey(); + const builder = new MockBuilder(); + + const first = retainMock(key, builder); + first.simulateConnect(); + first.simulateDisconnect(new Error('connection lost'), 1, 1000); + + const attemptError = new Error('still down'); + first.simulateConnectError(attemptError, 2, 2000); + + expect(ConnectionManager.getConnection(key)).toBe(first); + expect(ConnectionManager.getSnapshot(key)?.connectionError).toBe( + attemptError + ); + + vi.advanceTimersByTime(CONNECTION_MANAGER_RECONNECT_MAX_DELAY_MS); + expect(builder.buildCount).toBe(1); + + ConnectionManager.release(key); + }); + + test('manager callbacks stay attached while the SDK retries', () => { + const key = nextKey(); + const builder = new MockBuilder(); + + const first = retainMock(key, builder); + first.simulateConnect(); + first.simulateDisconnect(new Error('connection lost'), 1, 1000); + + expect(first.callbackCounts()).toEqual({ + connect: 1, + disconnect: 1, + connectError: 1, + }); + + ConnectionManager.release(key); + }); + + test('a successful SDK reconnect restores the state snapshot', () => { + const key = nextKey(); + const builder = new MockBuilder(); + + const first = retainMock(key, builder); + first.simulateConnect('session-token'); + first.simulateDisconnect(new Error('connection lost'), 1, 1000); + expect(ConnectionManager.getSnapshot(key)?.isActive).toBe(false); + + // The SDK reconnects inside the same object and fires onConnect again. + first.simulateConnect('session-token'); + + expect(ConnectionManager.getSnapshot(key)?.isActive).toBe(true); + expect(ConnectionManager.getSnapshot(key)?.connectionError).toBeUndefined(); + expect(ConnectionManager.getConnection(key)).toBe(first); + expect(builder.buildCount).toBe(1); + + ConnectionManager.release(key); + }); +}); + +describe('ConnectionManager rebuild on terminal failures', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.runOnlyPendingTimers(); + vi.useRealTimers(); + }); + + test('rebuilds a retained connection after a terminal disconnect', () => { const key = nextKey(); const builder = new MockBuilder(); @@ -242,9 +414,11 @@ describe('ConnectionManager retained reconnect behavior', () => { ConnectionManager.release(key); }); - test('rebuilds a retained connection after connectError', () => { + test('rebuilds a retained connection after a terminal connectError', () => { const key = nextKey(); const builder = new MockBuilder(); + // A failed *initial* connection is the SDK's main terminal case: it + // reports it through onConnectError with no next attempt. const error = new Error('network unavailable'); const first = retainMock(key, builder); @@ -263,7 +437,7 @@ describe('ConnectionManager retained reconnect behavior', () => { ConnectionManager.release(key); }); - test('same-key retain after disconnect returns a fresh connection immediately', () => { + test('same-key retain after a terminal failure returns a fresh connection immediately', () => { const key = nextKey(); const builder = new MockBuilder(); @@ -283,7 +457,7 @@ describe('ConnectionManager retained reconnect behavior', () => { ConnectionManager.release(key); }); - test('reconnect uses callbacks from a replacement same-key builder', () => { + test('rebuild uses callbacks from a replacement same-key builder', () => { const key = nextKey(); const firstBuilder = new MockBuilder(); const secondBuilder = new MockBuilder(); @@ -315,7 +489,7 @@ describe('ConnectionManager retained reconnect behavior', () => { ConnectionManager.release(key); }); - test('disconnect removes manager callbacks from the old connection before pending reconnect', () => { + test('a terminal failure removes manager callbacks from the old connection', () => { const key = nextKey(); const builder = new MockBuilder(); @@ -337,7 +511,7 @@ describe('ConnectionManager retained reconnect behavior', () => { ConnectionManager.release(key); }); - test('release cancels a pending reconnect', () => { + test('release cancels a pending rebuild', () => { const key = nextKey(); const builder = new MockBuilder(); @@ -351,7 +525,7 @@ describe('ConnectionManager retained reconnect behavior', () => { expect(ConnectionManager.getConnection(key)).toBeNull(); }); - test('manual disconnect does not trigger a reconnect', () => { + test('manual disconnect does not trigger a rebuild', () => { const key = nextKey(); const builder = new MockBuilder(); @@ -387,14 +561,14 @@ describe('ConnectionManager retained reconnect behavior', () => { ConnectionManager.release(key); }); - test('reconnect delay backs off exponentially across consecutive failures', () => { + test('rebuild delay backs off exponentially across consecutive failures', () => { const key = nextKey(); const builder = new MockBuilder(); const first = retainMock(key, builder); - first.simulateDisconnect(); + first.simulateConnectError(new Error('server unreachable')); - // First reconnect fires after the base delay. + // First rebuild fires after the base delay. vi.advanceTimersByTime(connectionManagerReconnectDelayMs(0)); expect(builder.buildCount).toBe(2); @@ -415,12 +589,12 @@ describe('ConnectionManager retained reconnect behavior', () => { ConnectionManager.release(key); }); - test('successful connect resets the reconnect backoff', () => { + test('successful connect resets the rebuild backoff', () => { const key = nextKey(); const builder = new MockBuilder(); const first = retainMock(key, builder); - first.simulateDisconnect(); + first.simulateConnectError(new Error('server unreachable')); vi.advanceTimersByTime(connectionManagerReconnectDelayMs(0)); builder.connections[1].simulateConnectError(new Error('still down')); @@ -437,7 +611,7 @@ describe('ConnectionManager retained reconnect behavior', () => { ConnectionManager.release(key); }); - test('reconnect delay is capped at the maximum delay', () => { + test('rebuild delay is capped at the maximum delay', () => { expect(connectionManagerReconnectDelayMs(0)).toBeLessThan( CONNECTION_MANAGER_RECONNECT_MAX_DELAY_MS ); @@ -532,18 +706,19 @@ describe('ConnectionManager.rebuild', () => { ConnectionManager.release(key); }); - test('cancels a pending auto-reconnect and resets the backoff', () => { + test('cancels a pending terminal-failure rebuild and resets the backoff', () => { const key = nextKey(); const builder = new MockBuilder(); const first = retainMock(key, builder); - // Two consecutive failures so the backoff has advanced past the base delay. - first.simulateDisconnect(); + // Two consecutive terminal failures so the backoff has advanced past the + // base delay. + first.simulateConnectError(new Error('server unreachable')); vi.advanceTimersByTime(connectionManagerReconnectDelayMs(0)); builder.connections[1].simulateConnectError(new Error('still down')); expect(builder.buildCount).toBe(2); - // rebuild() takes over: the scheduled reconnect must not also fire. + // rebuild() takes over: the scheduled rebuild must not also fire. const replacement = new MockBuilder(); ConnectionManager.rebuild(key, replacement as any); expect(replacement.buildCount).toBe(1); @@ -553,7 +728,8 @@ describe('ConnectionManager.rebuild', () => { expect(builder.buildCount).toBe(2); expect(replacement.buildCount).toBe(1); - // ...and the backoff was reset: a fresh drop reconnects after the base delay. + // ...and the backoff was reset: a fresh terminal failure rebuilds after + // the base delay. replacement.connections[0].simulateConnect(); replacement.connections[0].simulateDisconnect(); vi.advanceTimersByTime(connectionManagerReconnectDelayMs(0)); @@ -576,6 +752,9 @@ describe('ConnectionManager.rebuild', () => { build() { throw buildError; }, + withAutomaticReconnect() { + return this; + }, onConnect() { return this; }, @@ -629,7 +808,7 @@ describe('ConnectionManager session continuity across rebuilds', () => { vi.useRealTimers(); }); - test('auto-reconnect reuses the token issued after the builder was built', () => { + test('a terminal-failure rebuild reuses the token issued after the builder was built', () => { const key = nextKey(); // A first-time visitor: nothing in storage, so the builder carries no token. const builder = new MockBuilder(); @@ -642,8 +821,9 @@ describe('ConnectionManager session continuity across rebuilds', () => { first.simulateConnect('session-token'); expect(ConnectionManager.getSnapshot(key)?.token).toBe('session-token'); - // The socket drops and the manager auto-reconnects from the retained - // builder — which still holds the empty token it was constructed with. + // The connection ends terminally and the manager rebuilds from the + // retained builder — which still holds the empty token it was constructed + // with. first.simulateDisconnect(); vi.advanceTimersByTime(connectionManagerReconnectDelayMs(0)); @@ -654,7 +834,7 @@ describe('ConnectionManager session continuity across rebuilds', () => { expect(second.token).toBe('session-token'); }); - test('resumed session survives repeated reconnects', () => { + test('resumed session survives repeated rebuilds', () => { const key = nextKey(); const builder = new MockBuilder(); @@ -708,7 +888,7 @@ describe('ConnectionManager session continuity across rebuilds', () => { ConnectionManager.release(key); }); - test('retain after a drop resumes the session rather than the stale builder', () => { + test('retain after a terminal failure resumes the session rather than the stale builder', () => { const key = nextKey(); const builder = new MockBuilder(); @@ -716,7 +896,7 @@ describe('ConnectionManager session continuity across rebuilds', () => { first.simulateConnect('session-token'); first.simulateDisconnect(); - // A provider remount rebuilds through retain(), not the reconnect timer. + // A provider remount rebuilds through retain(), not the rebuild timer. const second = retainMock(key, builder); expect(second.token).toBe('session-token'); @@ -745,15 +925,16 @@ describe('ConnectionManager session continuity across rebuilds', () => { ConnectionManager.release(key); }); - test('auto-reconnect with a replacement builder keeps the session identity', () => { + test('a terminal-failure rebuild with a replacement builder keeps the session identity', () => { const key = nextKey(); const anonymous = new MockBuilder(); const first = retainMock(key, anonymous); first.simulateConnect('anonymous-token'); - // Swap the builder while the connection is live, then drop: the reconnect - // uses the replacement's callbacks but must not adopt its token. + // Swap the builder while the connection is live, then end it terminally: + // the rebuild uses the replacement's callbacks but must not adopt its + // token. ConnectionManager.release(key); const signedIn = new MockBuilder('signed-in-token'); retainMock(key, signedIn); @@ -787,7 +968,7 @@ describe('ConnectionManager session continuity across rebuilds', () => { ConnectionManager.release(key); }); - test('auto-reconnect after rebuild() keeps the new identity', () => { + test('a terminal-failure rebuild after rebuild() keeps the new identity', () => { const key = nextKey(); const anonymous = new MockBuilder(); @@ -799,8 +980,8 @@ describe('ConnectionManager session continuity across rebuilds', () => { signedIn as any ) as unknown as MockConnection; - // Drop *before* the new connection completes its handshake: the manager - // must not fall back to the identity rebuild() just replaced. + // Fail terminally *before* the new connection completes its handshake: + // the manager must not fall back to the identity rebuild() just replaced. second.simulateDisconnect(); vi.advanceTimersByTime(connectionManagerReconnectDelayMs(0)); diff --git a/crates/bindings-typescript/tests/db_connection_liveness.test.ts b/crates/bindings-typescript/tests/db_connection_liveness.test.ts new file mode 100644 index 00000000000..41e36479f85 --- /dev/null +++ b/crates/bindings-typescript/tests/db_connection_liveness.test.ts @@ -0,0 +1,267 @@ +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; +import { Identity } from '../src'; +import { ServerMessage } from '../src/sdk/client_api/types'; +import { WebsocketTestAdapterFactory } from '../src/sdk/websocket_test_adapter'; +import { ConnectionId } from '../src'; +import { DbConnection } from '../test-app/src/module_bindings'; +import { anIdentity } from './utils'; + +// These tests exercise the page-resume liveness recovery in DbConnectionImpl: +// with automatic reconnection enabled, the connection listens for the page +// coming back to the foreground (visibilitychange/focus/online/pageshow) and +// uses the moment to notice sockets that died silently while the tab was +// frozen, and to bring a backoff-stalled reconnect forward. +// +// The listeners bind to `document`/`window` when the socket opens, so each +// test installs minimal DOM stubs first. + +type ReconnectReport = { + error?: Error; + nextReconnectAttempt?: number; + nextReconnectDelayMs?: number; +}; + +type Harness = { + connection: DbConnection; + factory: WebsocketTestAdapterFactory; + connects: { identity: Identity; token: string }[]; + disconnects: ReconnectReport[]; + connectErrors: ReconnectReport[]; +}; + +let listeners: Record void>>; +let visibilityState: 'visible' | 'hidden'; + +function installDomStubs(): void { + listeners = {}; + visibilityState = 'visible'; + const record = + (scope: string) => + (ev: string, h: () => void): void => { + (listeners[`${scope}:${ev}`] ??= []).push(h); + }; + const remove = + (scope: string) => + (ev: string, h: () => void): void => { + const bucket = listeners[`${scope}:${ev}`]; + if (bucket) { + const at = bucket.indexOf(h); + if (at >= 0) bucket.splice(at, 1); + } + }; + vi.stubGlobal('document', { + get visibilityState() { + return visibilityState; + }, + addEventListener: record('doc'), + removeEventListener: remove('doc'), + }); + vi.stubGlobal('window', { + addEventListener: record('win'), + removeEventListener: remove('win'), + }); +} + +function removeDomStubs(): void { + vi.unstubAllGlobals(); +} + +function fire(name: string): void { + for (const h of [...(listeners[name] ?? [])]) h(); +} + +function listenerCounts(): Record { + return Object.fromEntries( + Object.entries(listeners).map(([name, hs]) => [name, hs.length]) + ); +} + +function build(options?: { automaticReconnect?: boolean }): Harness { + const factory = new WebsocketTestAdapterFactory(); + const connects: { identity: Identity; token: string }[] = []; + const disconnects: ReconnectReport[] = []; + const connectErrors: ReconnectReport[] = []; + + let builder = DbConnection.builder() + .withUri('ws://127.0.0.1:1234') + .withDatabaseName('db') + .withWSFn(factory.openWebSocket) + .onConnect((_conn, identity, token) => connects.push({ identity, token })) + .onDisconnect((_ctx, error, nextReconnectAttempt, nextReconnectDelayMs) => + disconnects.push({ error, nextReconnectAttempt, nextReconnectDelayMs }) + ) + .onConnectError((_ctx, error, nextReconnectAttempt, nextReconnectDelayMs) => + connectErrors.push({ error, nextReconnectAttempt, nextReconnectDelayMs }) + ); + if (options?.automaticReconnect ?? true) { + builder = builder.withAutomaticReconnect(); + } + + return { + connection: builder.build(), + factory, + connects, + disconnects, + connectErrors, + }; +} + +/** Let the connection's pending socket promise settle. */ +async function settle(harness: Harness): Promise { + await harness.connection['wsPromise']; + await Promise.resolve(); +} + +/** Bring a connection up to an established state on its current socket. */ +async function establish(harness: Harness): Promise { + await settle(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient( + ServerMessage.InitialConnection({ + identity: anIdentity, + connectionId: ConnectionId.random(), + token: 'issued-token', + }) + ); + await Promise.resolve(); +} + +beforeEach(() => { + vi.useFakeTimers(); + installDomStubs(); +}); + +afterEach(() => { + vi.useRealTimers(); + removeDomStubs(); +}); + +describe('liveness listeners', () => { + test('are installed when the socket opens with automatic reconnect enabled', async () => { + const harness = build(); + await establish(harness); + + expect(listenerCounts()).toEqual({ + 'doc:visibilitychange': 1, + 'win:focus': 1, + 'win:online': 1, + 'win:pageshow': 1, + }); + }); + + test('are not installed without automatic reconnect', async () => { + const harness = build({ automaticReconnect: false }); + await establish(harness); + + expect(listeners).toEqual({}); + }); + + test('are removed when the connection ends', async () => { + const harness = build(); + await establish(harness); + + harness.connection.disconnect(); + await settle(harness); + + expect(listenerCounts()).toEqual({ + 'doc:visibilitychange': 0, + 'win:focus': 0, + 'win:online': 0, + 'win:pageshow': 0, + }); + }); +}); + +describe('liveness recovery on page resume', () => { + test('treats a silently-dead socket as a lost connection when the network returns', async () => { + const harness = build(); + await establish(harness); + const firstSocket = harness.factory.current; + + // Socket dies while backgrounded: no close event is ever delivered, but + // the underlying readyState is now CLOSED. + firstSocket.dieSilently(); + expect(harness.disconnects).toHaveLength(0); + + fire('win:online'); + + // The loss is reported like any mid-session drop, announcing a retry... + expect(harness.disconnects).toHaveLength(1); + expect(harness.disconnects[0].nextReconnectAttempt).toBe(1); + + // ...and the scheduled attempt builds a fresh socket the connection can + // re-establish on. + await vi.runOnlyPendingTimersAsync(); + await settle(harness); + expect(harness.factory.current).not.toBe(firstSocket); + await establish(harness); + expect(harness.connects).toHaveLength(2); + expect(harness.connection.isActive).toBe(true); + }); + + test('does not disturb a healthy connection on resume', async () => { + const harness = build(); + await establish(harness); + const firstSocket = harness.factory.current; + + fire('win:focus'); + fire('doc:visibilitychange'); + await settle(harness); + + expect(harness.disconnects).toHaveLength(0); + expect(harness.factory.current).toBe(firstSocket); + expect(harness.connection.isActive).toBe(true); + }); + + test('does not revive a connection after an explicit disconnect', async () => { + const harness = build(); + await establish(harness); + + harness.connection.disconnect(); + await settle(harness); + const disconnectsBefore = harness.disconnects.length; + + harness.factory.current.dieSilently(); + fire('win:online'); + await vi.runOnlyPendingTimersAsync(); + + expect(harness.disconnects).toHaveLength(disconnectsBefore); + expect(harness.connection.isActive).toBe(false); + }); + + test('brings a backoff-stalled reconnect forward on resume', async () => { + const harness = build(); + await establish(harness); + const firstSocket = harness.factory.current; + + // Drop the connection; a reconnect is now waiting out its backoff delay + // (simulating a background tab whose timers are throttled/frozen). + firstSocket.serverClose(1006); + expect(harness.disconnects).toHaveLength(1); + const delay = harness.disconnects[0].nextReconnectDelayMs!; + vi.advanceTimersByTime(delay - 1); + expect(harness.factory.current).toBe(firstSocket); + + // Regaining visibility retries immediately instead of waiting out the + // remaining delay. + fire('doc:visibilitychange'); + await settle(harness); + expect(harness.factory.current).not.toBe(firstSocket); + + await establish(harness); + expect(harness.connects).toHaveLength(2); + }); + + test('visibilitychange while still hidden does nothing', async () => { + const harness = build(); + await establish(harness); + const firstSocket = harness.factory.current; + + firstSocket.serverClose(1006); + visibilityState = 'hidden'; + fire('doc:visibilitychange'); + await settle(harness); + + expect(harness.factory.current).toBe(firstSocket); + }); +}); diff --git a/crates/bindings-typescript/tests/db_connection_reconnect.test.ts b/crates/bindings-typescript/tests/db_connection_reconnect.test.ts new file mode 100644 index 00000000000..f255a9098f5 --- /dev/null +++ b/crates/bindings-typescript/tests/db_connection_reconnect.test.ts @@ -0,0 +1,1201 @@ +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; +import { ConnectionId, Identity } from '../src'; +import { + DisconnectedError, + IdentityChangedError, + UnknownCallResultError, +} from '../src/lib/errors'; +import { + computeReconnectDelayMs, + tokenNeedsRefresh, + RECONNECT_INITIAL_DELAY_MS, + RECONNECT_MAX_DELAY_MS, +} from '../src/sdk/db_connection_impl'; +import { + ServerMessage, + type SubscribeBatch, +} from '../src/sdk/client_api/types'; +import { WebSocketTokenError } from '../src/sdk/ws'; +import { WebsocketTestAdapterFactory } from '../src/sdk/websocket_test_adapter'; +import { DbConnection } from '../test-app/src/module_bindings'; +import { anIdentity, bobIdentity, encodeUser } from './utils'; + +/** The disconnect/connect-error reports an application sees. */ +type ReconnectReport = { + error?: Error; + nextReconnectAttempt?: number; + nextReconnectDelayMs?: number; +}; + +type Harness = { + connection: DbConnection; + factory: WebsocketTestAdapterFactory; + connects: { identity: Identity; token: string }[]; + disconnects: ReconnectReport[]; + connectErrors: ReconnectReport[]; +}; + +const TOKEN = 'issued-token'; + +function build(options?: { + automaticReconnect?: boolean; + token?: string; + tokenProvider?: () => Promise; +}): Harness { + const factory = new WebsocketTestAdapterFactory(); + const connects: { identity: Identity; token: string }[] = []; + const disconnects: ReconnectReport[] = []; + const connectErrors: ReconnectReport[] = []; + + let builder = DbConnection.builder() + .withUri('ws://127.0.0.1:1234') + .withDatabaseName('db') + .withWSFn(factory.openWebSocket) + .onConnect((_conn, identity, token) => connects.push({ identity, token })) + .onDisconnect((_ctx, error, nextReconnectAttempt, nextReconnectDelayMs) => + disconnects.push({ error, nextReconnectAttempt, nextReconnectDelayMs }) + ) + .onConnectError((_ctx, error, nextReconnectAttempt, nextReconnectDelayMs) => + connectErrors.push({ error, nextReconnectAttempt, nextReconnectDelayMs }) + ); + if (options?.token) { + builder = builder.withToken(options.token); + } + if (options?.automaticReconnect ?? true) { + builder = builder.withAutomaticReconnect(); + } + if (options?.tokenProvider) { + builder = builder.withTokenProvider(options.tokenProvider); + } + + return { + connection: builder.build(), + factory, + connects, + disconnects, + connectErrors, + }; +} + +/** Let the connection's pending socket promise settle. */ +async function settle(harness: Harness): Promise { + await harness.connection['wsPromise']; + await Promise.resolve(); +} + +function initialConnection( + identity: Identity = anIdentity, + connectionId: ConnectionId = ConnectionId.random() +): ServerMessage { + return ServerMessage.InitialConnection({ + identity, + connectionId, + token: TOKEN, + }); +} + +/** Bring a connection up to an established state on its current socket. */ +async function establish( + harness: Harness, + identity: Identity = anIdentity +): Promise { + await settle(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient(initialConnection(identity)); + await Promise.resolve(); +} + +/** Run the scheduled reconnect timer and let its socket be created. */ +async function runReconnectTimer(harness: Harness): Promise { + await vi.runOnlyPendingTimersAsync(); + await settle(harness); +} + +beforeEach(() => { + vi.useFakeTimers(); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('reconnect policy', () => { + test('delays grow exponentially from the initial delay', () => { + const noJitter = () => 0.5; + expect(computeReconnectDelayMs(1, noJitter)).toBe( + RECONNECT_INITIAL_DELAY_MS + ); + expect(computeReconnectDelayMs(2, noJitter)).toBe( + RECONNECT_INITIAL_DELAY_MS * 2 + ); + expect(computeReconnectDelayMs(3, noJitter)).toBe( + RECONNECT_INITIAL_DELAY_MS * 4 + ); + }); + + test('delays are capped', () => { + const noJitter = () => 0.5; + expect(computeReconnectDelayMs(20, noJitter)).toBe(RECONNECT_MAX_DELAY_MS); + }); + + test('jitter spreads the delay around the base but never exceeds the cap', () => { + expect(computeReconnectDelayMs(3, () => 0)).toBeLessThan( + computeReconnectDelayMs(3, () => 1) + ); + expect(computeReconnectDelayMs(30, () => 1)).toBeLessThanOrEqual( + RECONNECT_MAX_DELAY_MS + ); + expect(computeReconnectDelayMs(1, () => 0)).toBeGreaterThanOrEqual(0); + }); +}); + +describe('token refresh', () => { + const nowSeconds = 1_000_000; + const nowMs = nowSeconds * 1000; + + function jwt(claims: object): string { + const payload = Buffer.from(JSON.stringify(claims)).toString('base64url'); + return `header.${payload}.signature`; + } + + test('a token with plenty of life left is not refreshed', () => { + const token = jwt({ iat: nowSeconds - 60, exp: nowSeconds + 3600 }); + expect(tokenNeedsRefresh(token, nowMs)).toBe(false); + }); + + test('a token close to expiring is refreshed', () => { + const token = jwt({ iat: nowSeconds - 3590, exp: nowSeconds + 10 }); + expect(tokenNeedsRefresh(token, nowMs)).toBe(true); + }); + + test('a short-lived token is refreshed on the 30 second floor', () => { + // 5% of a 60 second lifetime is only 3 seconds, so the floor applies. + const token = jwt({ iat: nowSeconds - 40, exp: nowSeconds + 20 }); + expect(tokenNeedsRefresh(token, nowMs)).toBe(true); + }); + + test('an expired token is refreshed', () => { + const token = jwt({ iat: nowSeconds - 3600, exp: nowSeconds - 1 }); + expect(tokenNeedsRefresh(token, nowMs)).toBe(true); + }); + + test('a token whose expiry cannot be read is always refreshed', () => { + expect(tokenNeedsRefresh('not-a-jwt', nowMs)).toBe(true); + expect(tokenNeedsRefresh(jwt({ sub: 'no-exp' }), nowMs)).toBe(true); + expect(tokenNeedsRefresh(undefined, nowMs)).toBe(true); + }); +}); + +describe('losing an established connection', () => { + test('onDisconnect announces the first reconnect attempt', async () => { + const harness = build(); + await establish(harness); + + harness.factory.current.close(); + await Promise.resolve(); + + expect(harness.disconnects).toHaveLength(1); + expect(harness.disconnects[0].nextReconnectAttempt).toBe(1); + expect(harness.disconnects[0].nextReconnectDelayMs).toBeGreaterThan(0); + expect(harness.disconnects[0].error).toBeInstanceOf(Error); + }); + + test('a reconnect opens a new socket and fires onConnect again', async () => { + const harness = build(); + await establish(harness); + expect(harness.factory.sockets).toHaveLength(1); + + harness.factory.current.close(); + await runReconnectTimer(harness); + expect(harness.factory.sockets).toHaveLength(2); + + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient(initialConnection()); + await Promise.resolve(); + + expect(harness.connects).toHaveLength(2); + expect(harness.connection.isActive).toBe(true); + }); + + test('the connection object and its cache survive a reconnect', async () => { + const harness = build(); + await establish(harness); + const cacheBefore = harness.connection.db; + + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient(initialConnection()); + await Promise.resolve(); + + expect(harness.connection.db).toBe(cacheBefore); + }); + + test('the retained token is used to reconnect, keeping the identity stable', async () => { + // Connect anonymously; the server issues a token. + const harness = build(); + await establish(harness); + expect(harness.connects[0].token).toBe(TOKEN); + + harness.factory.current.close(); + await runReconnectTimer(harness); + + expect(harness.factory.current.connectArgs?.authToken).toBe(TOKEN); + }); + + test('a session id is sent so the server can supersede the old connection', async () => { + const harness = build(); + await establish(harness); + const sessionId = harness.factory.sockets[0].connectArgs?.sessionId; + expect(sessionId).toBeTruthy(); + + harness.factory.current.close(); + await runReconnectTimer(harness); + + // The same session id identifies both connections as one client session. + expect(harness.factory.current.connectArgs?.sessionId).toBe(sessionId); + // Each connection still has its own connection id. + expect(harness.factory.current.connectArgs?.connectionId).toBeTruthy(); + }); + + test('reconnection is off unless requested', async () => { + const harness = build({ automaticReconnect: false }); + await establish(harness); + + harness.factory.current.close(); + await vi.runOnlyPendingTimersAsync(); + + expect(harness.disconnects).toHaveLength(1); + expect(harness.disconnects[0].nextReconnectAttempt).toBeUndefined(); + expect(harness.factory.sockets).toHaveLength(1); + }); +}); + +describe('failed reconnect attempts', () => { + test('onConnectError announces the next attempt', async () => { + const harness = build(); + await establish(harness); + + harness.factory.current.close(); + await runReconnectTimer(harness); + // The attempt's socket dies before completing its handshake. + harness.factory.current.close(); + await Promise.resolve(); + + expect(harness.connectErrors).toHaveLength(1); + expect(harness.connectErrors[0].nextReconnectAttempt).toBe(2); + // A failed attempt is not a lost connection, so no second onDisconnect. + expect(harness.disconnects).toHaveLength(1); + }); + + test('the attempt number grows across consecutive failures', async () => { + const harness = build(); + await establish(harness); + + harness.factory.current.close(); + for (let expected = 2; expected <= 4; expected++) { + await runReconnectTimer(harness); + harness.factory.current.close(); + await Promise.resolve(); + expect( + harness.connectErrors[harness.connectErrors.length - 1] + .nextReconnectAttempt + ).toBe(expected); + } + }); + + test('the delay grows across consecutive failures', async () => { + const harness = build(); + await establish(harness); + + harness.factory.current.close(); + const delays: number[] = [harness.disconnects[0].nextReconnectDelayMs!]; + for (let i = 0; i < 3; i++) { + await runReconnectTimer(harness); + harness.factory.current.close(); + await Promise.resolve(); + delays.push( + harness.connectErrors[harness.connectErrors.length - 1] + .nextReconnectDelayMs! + ); + } + + // Jitter makes individual steps noisy, so compare the ends of the run. + expect(delays[delays.length - 1]).toBeGreaterThan(delays[0]); + }); + + test('a failure to open the socket at all counts as a failed attempt', async () => { + const harness = build(); + await establish(harness); + + harness.factory.current.close(); + harness.factory.connectError = new Error('server unreachable'); + await runReconnectTimer(harness); + + expect(harness.connectErrors).toHaveLength(1); + expect(harness.connectErrors[0].error?.message).toBe('server unreachable'); + expect(harness.connectErrors[0].nextReconnectAttempt).toBe(2); + }); + + test('the attempt counter resets once a connection is established', async () => { + const harness = build(); + await establish(harness); + + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.close(); + await Promise.resolve(); + expect( + harness.connectErrors[harness.connectErrors.length - 1] + .nextReconnectAttempt + ).toBe(2); + + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient(initialConnection()); + await Promise.resolve(); + + // A later drop starts again at attempt 1. + harness.factory.current.close(); + await Promise.resolve(); + expect( + harness.disconnects[harness.disconnects.length - 1].nextReconnectAttempt + ).toBe(1); + }); + + test('an initial connection failure is not retried', async () => { + const harness = build(); + await settle(harness); + + // The socket dies before ever completing a handshake. + harness.factory.current.close(); + await vi.runOnlyPendingTimersAsync(); + + expect(harness.connectErrors).toHaveLength(1); + expect(harness.connectErrors[0].nextReconnectAttempt).toBeUndefined(); + expect(harness.factory.sockets).toHaveLength(1); + }); +}); + +describe('terminal failures', () => { + test('a reconnect under a different identity stops the SDK', async () => { + const harness = build(); + await establish(harness, anIdentity); + + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + // The server hands us a different identity: the token was replaced. + harness.factory.current.sendToClient(initialConnection(bobIdentity)); + await Promise.resolve(); + + const lastError = + harness.connectErrors[harness.connectErrors.length - 1].error; + expect(lastError).toBeInstanceOf(IdentityChangedError); + expect( + harness.connectErrors[harness.connectErrors.length - 1] + .nextReconnectAttempt + ).toBeUndefined(); + + // No further attempts are scheduled. + const socketsBefore = harness.factory.sockets.length; + await vi.runOnlyPendingTimersAsync(); + expect(harness.factory.sockets).toHaveLength(socketsBefore); + }); +}); + +describe('explicit disconnect()', () => { + test('fires onDisconnect and stops reconnecting', async () => { + const harness = build(); + await establish(harness); + + harness.connection.disconnect(); + harness.factory.current.close(); + await vi.runOnlyPendingTimersAsync(); + + expect(harness.disconnects).toHaveLength(1); + expect(harness.disconnects[0].nextReconnectAttempt).toBeUndefined(); + expect(harness.factory.sockets).toHaveLength(1); + }); + + test('fires onDisconnect when called while reconnecting', async () => { + const harness = build(); + await establish(harness); + + harness.factory.current.close(); + await Promise.resolve(); + expect(harness.disconnects[0].nextReconnectAttempt).toBe(1); + + // Called between attempts, when there is no live socket whose close event + // would otherwise end the connection. + harness.connection.disconnect(); + await vi.runOnlyPendingTimersAsync(); + + expect(harness.disconnects).toHaveLength(2); + expect(harness.disconnects[1].nextReconnectAttempt).toBeUndefined(); + // The scheduled attempt was cancelled. + expect(harness.factory.sockets).toHaveLength(1); + }); + + test('cancels a scheduled attempt even after several failures', async () => { + const harness = build(); + await establish(harness); + + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.close(); + await Promise.resolve(); + + harness.connection.disconnect(); + const socketsBefore = harness.factory.sockets.length; + await vi.runOnlyPendingTimersAsync(); + + expect(harness.factory.sockets).toHaveLength(socketsBefore); + expect( + harness.disconnects[harness.disconnects.length - 1].nextReconnectAttempt + ).toBeUndefined(); + }); +}); + +describe('calls while reconnecting', () => { + test('a reducer call fails immediately rather than queueing', async () => { + const harness = build(); + await establish(harness); + harness.factory.current.close(); + await Promise.resolve(); + + await expect( + harness.connection.reducers.createPlayer({ + name: 'Alice', + location: { x: 1, y: 2 }, + }) + ).rejects.toBeInstanceOf(DisconnectedError); + }); + + test('an in-flight call settles with an unknown-result error', async () => { + const harness = build(); + await establish(harness); + + const pending = harness.connection.reducers.createPlayer({ + name: 'Alice', + location: { x: 1, y: 2 }, + }); + // The connection drops before the server acknowledges the call. + harness.factory.current.close(); + await Promise.resolve(); + + await expect(pending).rejects.toBeInstanceOf(UnknownCallResultError); + }); + + test('calls are not rejected without automatic reconnection', async () => { + const harness = build({ automaticReconnect: false }); + await establish(harness); + harness.factory.current.close(); + await Promise.resolve(); + + // Legacy behavior: the call queues on the dead socket rather than failing. + let settled = false; + void harness.connection.reducers + .createPlayer({ name: 'Alice', location: { x: 1, y: 2 } }) + .then( + () => (settled = true), + () => (settled = true) + ); + await Promise.resolve(); + expect(settled).toBe(false); + }); +}); + +describe('replaying subscriptions', () => { + /** The last batch-subscribe message the connection sent, if any. */ + function lastSubscribeBatch(harness: Harness): SubscribeBatch | undefined { + const messages = harness.factory.current.outgoingMessages; + for (let i = messages.length - 1; i >= 0; i--) { + const message = messages[i]; + if (message.tag === 'SubscribeBatch') { + return message.value; + } + } + return undefined; + } + + async function establishWithSubscription(): Promise { + const harness = build(); + await establish(harness); + harness.connection.subscriptionBuilder().subscribe(['SELECT * FROM user']); + await Promise.resolve(); + return harness; + } + + test('a reconnect replays live subscriptions in one batch', async () => { + const harness = await establishWithSubscription(); + + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient(initialConnection()); + await Promise.resolve(); + + const batch = lastSubscribeBatch(harness); + expect(batch).toBeDefined(); + expect(batch!.sets).toHaveLength(1); + }); + + test('replayed sets are registered under fresh query set ids', async () => { + const harness = await establishWithSubscription(); + const originalSubscribe = harness.factory.current.outgoingMessages.find( + message => message.tag === 'Subscribe' + ); + const originalQuerySetId = originalSubscribe!.value.querySetId.id; + + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient(initialConnection()); + await Promise.resolve(); + + const batch = lastSubscribeBatch(harness); + expect(batch!.sets[0].querySetId.id).not.toBe(originalQuerySetId); + }); + + test('nothing is replayed when there are no subscriptions', async () => { + const harness = build(); + await establish(harness); + + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient(initialConnection()); + await Promise.resolve(); + + expect(lastSubscribeBatch(harness)).toBeUndefined(); + }); + + test('rows unchanged across the outage produce no callbacks', async () => { + const harness = await establishWithSubscription(); + const querySetId = harness.factory.current.outgoingMessages.find( + message => message.tag === 'Subscribe' + )!.value.querySetId.id; + + // The server delivers one row for the subscription. + harness.factory.current.sendToClient( + ServerMessage.SubscribeApplied({ + requestId: 1, + querySetId: { id: querySetId }, + rows: { + tables: [ + { + table: 'user', + rows: { + sizeHint: { tag: 'RowOffsets', value: [0n] }, + rowsData: encodeUser({ + identity: anIdentity, + username: 'Alice', + }), + }, + }, + ], + }, + }) + ); + await Promise.resolve(); + + const inserts: string[] = []; + const updates: string[] = []; + const deletes: string[] = []; + harness.connection.db.user.onInsert((_ctx, row) => + inserts.push(row.username) + ); + harness.connection.db.user.onUpdate((_ctx, _old, row) => + updates.push(row.username) + ); + harness.connection.db.user.onDelete((_ctx, row) => + deletes.push(row.username) + ); + + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient(initialConnection()); + await Promise.resolve(); + + // The replay returns the same row it had before. + const batch = lastSubscribeBatch(harness)!; + harness.factory.current.sendToClient( + ServerMessage.SubscribeBatchApplied({ + requestId: batch.requestId, + results: [ + { + querySetId: batch.sets[0].querySetId, + outcome: { + tag: 'Applied', + value: { + tables: [ + { + table: 'user', + rows: { + sizeHint: { tag: 'RowOffsets', value: [0n] }, + rowsData: encodeUser({ + identity: anIdentity, + username: 'Alice', + }), + }, + }, + ], + }, + }, + }, + ], + }) + ); + await Promise.resolve(); + + expect(inserts).toEqual([]); + expect(updates).toEqual([]); + expect(deletes).toEqual([]); + // The row is still readable from the cache. + expect(harness.connection.db.user.count()).toBe(1n); + }); + + test('a row which changed during the outage produces one update callback', async () => { + const harness = await establishWithSubscription(); + const querySetId = harness.factory.current.outgoingMessages.find( + message => message.tag === 'Subscribe' + )!.value.querySetId.id; + + harness.factory.current.sendToClient( + ServerMessage.SubscribeApplied({ + requestId: 1, + querySetId: { id: querySetId }, + rows: { + tables: [ + { + table: 'user', + rows: { + sizeHint: { tag: 'RowOffsets', value: [0n] }, + rowsData: encodeUser({ + identity: anIdentity, + username: 'Alice', + }), + }, + }, + ], + }, + }) + ); + await Promise.resolve(); + + const updates: { from: string; to: string }[] = []; + harness.connection.db.user.onUpdate((_ctx, oldRow, newRow) => + updates.push({ from: oldRow.username, to: newRow.username }) + ); + + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient(initialConnection()); + await Promise.resolve(); + + const batch = lastSubscribeBatch(harness)!; + harness.factory.current.sendToClient( + ServerMessage.SubscribeBatchApplied({ + requestId: batch.requestId, + results: [ + { + querySetId: batch.sets[0].querySetId, + outcome: { + tag: 'Applied', + value: { + tables: [ + { + table: 'user', + rows: { + sizeHint: { tag: 'RowOffsets', value: [0n] }, + // The same identity, renamed while we were away. + rowsData: encodeUser({ + identity: anIdentity, + username: 'Alicia', + }), + }, + }, + ], + }, + }, + }, + ], + }) + ); + await Promise.resolve(); + + expect(updates).toEqual([{ from: 'Alice', to: 'Alicia' }]); + }); + + test('a row deleted during the outage produces a delete callback', async () => { + const harness = await establishWithSubscription(); + const querySetId = harness.factory.current.outgoingMessages.find( + message => message.tag === 'Subscribe' + )!.value.querySetId.id; + + harness.factory.current.sendToClient( + ServerMessage.SubscribeApplied({ + requestId: 1, + querySetId: { id: querySetId }, + rows: { + tables: [ + { + table: 'user', + rows: { + sizeHint: { tag: 'RowOffsets', value: [0n] }, + rowsData: encodeUser({ + identity: anIdentity, + username: 'Alice', + }), + }, + }, + ], + }, + }) + ); + await Promise.resolve(); + + const deletes: string[] = []; + harness.connection.db.user.onDelete((_ctx, row) => + deletes.push(row.username) + ); + + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient(initialConnection()); + await Promise.resolve(); + + // The replay returns no rows: the row is gone. + const batch = lastSubscribeBatch(harness)!; + harness.factory.current.sendToClient( + ServerMessage.SubscribeBatchApplied({ + requestId: batch.requestId, + results: [ + { + querySetId: batch.sets[0].querySetId, + outcome: { tag: 'Applied', value: { tables: [] } }, + }, + ], + }) + ); + await Promise.resolve(); + + expect(deletes).toEqual(['Alice']); + expect(harness.connection.db.user.count()).toBe(0n); + }); + + test('a rejected replayed query reports its error while the rest apply', async () => { + const harness = build(); + await establish(harness); + + const errors: string[] = []; + const applied: number[] = []; + harness.connection + .subscriptionBuilder() + .onApplied(() => applied.push(1)) + .onError(ctx => errors.push(ctx.event!.message)) + .subscribe(['SELECT * FROM user']); + harness.connection + .subscriptionBuilder() + .onApplied(() => applied.push(2)) + .onError(ctx => errors.push(ctx.event!.message)) + .subscribe(['SELECT * FROM no_such_table']); + await Promise.resolve(); + + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.sendToClient(initialConnection()); + await Promise.resolve(); + + const batch = lastSubscribeBatch(harness)!; + expect(batch.sets).toHaveLength(2); + harness.factory.current.sendToClient( + ServerMessage.SubscribeBatchApplied({ + requestId: batch.requestId, + results: [ + { + querySetId: batch.sets[0].querySetId, + outcome: { tag: 'Applied', value: { tables: [] } }, + }, + { + querySetId: batch.sets[1].querySetId, + outcome: { tag: 'Error', value: 'no such table: no_such_table' }, + }, + ], + }) + ); + await Promise.resolve(); + + expect(errors).toEqual(['no such table: no_such_table']); + // The healthy set applied, firing its onApplied again on the new connection. + expect(applied).toContain(1); + }); +}); + +describe('token provider', () => { + test('is not called while the retained token has life left', async () => { + const nowSeconds = Math.floor(Date.now() / 1000); + const longLived = `header.${Buffer.from( + JSON.stringify({ iat: nowSeconds, exp: nowSeconds + 3600 }) + ).toString('base64url')}.sig`; + + const provider = vi.fn(async () => 'fresh-token'); + const harness = build({ token: longLived, tokenProvider: provider }); + await establish(harness); + + harness.factory.current.close(); + await runReconnectTimer(harness); + + expect(provider).not.toHaveBeenCalled(); + expect(harness.factory.current.connectArgs?.authToken).toBe(longLived); + }); + + test('supplies a fresh token when the retained one is close to expiring', async () => { + const nowSeconds = Math.floor(Date.now() / 1000); + const expiring = `header.${Buffer.from( + JSON.stringify({ iat: nowSeconds - 3595, exp: nowSeconds + 5 }) + ).toString('base64url')}.sig`; + + const provider = vi.fn(async () => 'fresh-token'); + const harness = build({ token: expiring, tokenProvider: provider }); + await establish(harness); + + harness.factory.current.close(); + await runReconnectTimer(harness); + + expect(provider).toHaveBeenCalled(); + expect(harness.factory.current.connectArgs?.authToken).toBe('fresh-token'); + }); + + test('a provider failure counts as a failed attempt, not a terminal error', async () => { + const provider = vi.fn(async () => { + throw new Error('token endpoint down'); + }); + const harness = build({ token: 'opaque', tokenProvider: provider }); + await establish(harness); + + harness.factory.current.close(); + await runReconnectTimer(harness); + + expect(harness.connectErrors).toHaveLength(1); + expect(harness.connectErrors[0].error?.message).toBe('token endpoint down'); + // The SDK keeps trying. + expect(harness.connectErrors[0].nextReconnectAttempt).toBe(2); + }); +}); + +describe('reconnect regressions', () => { + test('uses a fresh connection id for every attempt and retains the session id', async () => { + const harness = build(); + await establish(harness); + const first = harness.factory.current.connectArgs!; + const establishedId = harness.connection.connectionId.toHexString(); + harness.factory.current.close(); + await runReconnectTimer(harness); + const second = harness.factory.current.connectArgs!; + expect(second.connectionId).not.toBe(establishedId); + expect(second.connectionId).not.toBe(first.connectionId); + expect(second.sessionId).toBe(first.sessionId); + harness.factory.current.close(); + await runReconnectTimer(harness); + expect(harness.factory.current.connectArgs!.connectionId).not.toBe( + second.connectionId + ); + }); + + test('session-busy responses retry without advancing the backoff', async () => { + const harness = build(); + await establish(harness); + harness.factory.current.close(); + for (let i = 0; i < 3; i++) { + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + harness.factory.current.serverClose(4000, 'session busy'); + expect(harness.connectErrors.at(-1)?.nextReconnectAttempt).toBe(1); + expect( + harness.connectErrors.at(-1)?.nextReconnectDelayMs + ).toBeLessThanOrEqual(1500); + } + }); + + test('ignores all late events from a discarded socket', async () => { + const harness = build(); + await establish(harness); + const old = harness.factory.current; + old.error(new Error('lost')); + expect(old.closed).toBe(true); + old.acceptConnection(); + old.sendToClient(initialConnection(bobIdentity)); + old.close(); + expect(harness.connection.isActive).toBe(false); + expect(harness.disconnects).toHaveLength(1); + expect(harness.connects).toHaveLength(1); + await runReconnectTimer(harness); + await establish(harness); + old.sendToClient(initialConnection(bobIdentity)); + expect(harness.connection.identity).toEqual(anIdentity); + expect(harness.connects).toHaveLength(2); + }); + + test('waits for InitialConnection before allowing calls on a reconnect', async () => { + const harness = build(); + await establish(harness); + harness.factory.current.close(); + await runReconnectTimer(harness); + harness.factory.current.acceptConnection(); + expect(harness.connection.isActive).toBe(false); + expect(harness.connection.isReconnecting).toBe(true); + await expect( + harness.connection.callReducer('test', new Uint8Array()) + ).rejects.toBeInstanceOf(DisconnectedError); + expect(harness.factory.current.outgoingMessages).toEqual([]); + }); + + test('does not send a queued call on the next socket after rejecting it', async () => { + const harness = build(); + await establish(harness); + const pending = harness.connection.callReducer('test', new Uint8Array()); + const rejected = expect(pending).rejects.toBeInstanceOf( + UnknownCallResultError + ); + harness.factory.current.close(); + await rejected; + await runReconnectTimer(harness); + await establish(harness); + expect(harness.factory.current.outgoingMessages).toEqual([]); + }); + + test('explicit disconnect reports no error and settles in-flight procedures', async () => { + const harness = build(); + await establish(harness); + const rejected = expect( + harness.connection.callProcedure('test', new Uint8Array()) + ).rejects.toBeInstanceOf(UnknownCallResultError); + harness.connection.disconnect(); + await rejected; + expect(harness.disconnects).toEqual([ + { + error: undefined, + nextReconnectAttempt: undefined, + nextReconnectDelayMs: undefined, + }, + ]); + }); + + test('disconnect from onDisconnect cancels retries immediately', async () => { + const harness = build(); + await establish(harness); + harness.connection['onDisconnect'](() => { + if (!harness.connection.isDisconnectRequested) + harness.connection.disconnect(); + }); + harness.factory.current.close(); + expect(vi.getTimerCount()).toBe(0); + expect(harness.disconnects).toHaveLength(2); + }); + + test('disconnect during token refresh prevents opening another socket', async () => { + let resolveToken!: (token: string) => void; + const provider = vi.fn( + () => + new Promise(resolve => { + resolveToken = resolve; + }) + ); + const harness = build({ tokenProvider: provider }); + await establish(harness); + harness.factory.current.close(); + await vi.runOnlyPendingTimersAsync(); + expect(provider).toHaveBeenCalledOnce(); + harness.connection.disconnect(); + resolveToken('fresh-token'); + await settle(harness); + expect(harness.factory.sockets).toHaveLength(1); + expect(harness.connection.isReconnecting).toBe(false); + }); + + test('subscribes during the outage and onConnect are sent only in the replay batch', async () => { + const harness = build(); + await establish(harness); + harness.connection.subscriptionBuilder().subscribe('SELECT * FROM user'); + harness.factory.current.close(); + harness.connection.subscriptionBuilder().subscribe('SELECT * FROM player'); + harness.connection['onConnect'](conn => + conn.subscriptionBuilder().subscribe('SELECT * FROM user') + ); + await runReconnectTimer(harness); + await establish(harness); + const messages = harness.factory.current.outgoingMessages; + expect(messages.map(m => m.tag)).toEqual(['SubscribeBatch']); + const batch = messages[0]; + if (batch.tag !== 'SubscribeBatch') throw new Error('Expected replay'); + expect(batch.value.sets).toHaveLength(3); + }); + + test.each(['before drop', 'during outage'] as const)( + 'unsubscribe %s ends locally and removes stale rows after reconnect', + async timing => { + const harness = build(); + await establish(harness); + const handle = harness.connection + .subscriptionBuilder() + .subscribe('SELECT * FROM user'); + await Promise.resolve(); + const subscribe = harness.factory.current.outgoingMessages[0]; + if (subscribe.tag !== 'Subscribe') + throw new Error('Expected subscription'); + harness.factory.current.sendToClient( + ServerMessage.SubscribeApplied({ + ...subscribe.value, + rows: { + tables: [ + { + table: 'user', + rows: { + sizeHint: { tag: 'RowOffsets', value: [0n] }, + rowsData: encodeUser({ + identity: anIdentity, + username: 'Alice', + }), + }, + }, + ], + }, + }) + ); + const onEnd = vi.fn(); + if (timing === 'before drop') handle.unsubscribeThen(onEnd); + harness.factory.current.close(); + if (timing === 'during outage') handle.unsubscribeThen(onEnd); + expect(handle.isEnded()).toBe(true); + expect(handle.isActive()).toBe(false); + expect(onEnd).toHaveBeenCalledOnce(); + expect(harness.connection.db.user.count()).toBe(1n); + await runReconnectTimer(harness); + await establish(harness); + expect(harness.factory.current.outgoingMessages).toEqual([]); + expect(harness.connection.db.user.count()).toBe(0n); + } + ); + + test('legacy sockets still emit both error and close events', async () => { + const harness = build({ automaticReconnect: false }); + await establish(harness); + harness.factory.current.error(new Error('network error')); + harness.factory.current.close(); + expect(harness.connectErrors).toHaveLength(1); + expect(harness.disconnects).toHaveLength(1); + }); + + test.each([1002, 1003, 1007, 1008])( + 'protocol/policy close code %i is terminal', + async code => { + const harness = build(); + await establish(harness); + harness.factory.current.serverClose(code); + expect(harness.disconnects[0].nextReconnectAttempt).toBeUndefined(); + expect(harness.connection.isReconnecting).toBe(false); + expect(vi.getTimerCount()).toBe(0); + } + ); +}); + +describe('token rejection classification', () => { + test.each([401, 403])( + 'status %i without a provider is terminal', + async status => { + const harness = build(); + await establish(harness); + harness.factory.current.close(); + harness.factory.connectError = new WebSocketTokenError( + status, + 'Rejected' + ); + await runReconnectTimer(harness); + expect(harness.connectErrors[0].nextReconnectAttempt).toBeUndefined(); + expect(harness.connection.isReconnecting).toBe(false); + } + ); + + test('refreshes a rejected retained token once, then stops if the fresh token is rejected', async () => { + const now = Date.now() / 1000; + const token = `header.${Buffer.from(JSON.stringify({ iat: now, exp: now + 3600 })).toString('base64url')}.sig`; + const provider = vi.fn(async () => 'replacement'); + const harness = build({ token, tokenProvider: provider }); + await establish(harness); + harness.factory.current.close(); + harness.factory.connectError = new WebSocketTokenError(401, 'Unauthorized'); + await runReconnectTimer(harness); + expect(provider).not.toHaveBeenCalled(); + expect(harness.connectErrors[0].nextReconnectAttempt).toBe(2); + await runReconnectTimer(harness); + expect(provider).toHaveBeenCalledOnce(); + expect(harness.connectErrors[1].nextReconnectAttempt).toBeUndefined(); + }); + + test('a token exchange server error retries without classifying it as bad credentials', async () => { + const harness = build(); + await establish(harness); + harness.factory.current.close(); + harness.factory.connectError = new WebSocketTokenError(503, 'Unavailable'); + await runReconnectTimer(harness); + expect(harness.connectErrors[0].nextReconnectAttempt).toBe(2); + }); +}); + +describe('handshake and replay boundaries', () => { + test('subscriptions cancelled before the initial handshake are never sent', async () => { + const harness = build(); + const cancelled = harness.connection + .subscriptionBuilder() + .subscribe('SELECT * FROM user'); + cancelled.unsubscribe(); + harness.connection.subscriptionBuilder().subscribe('SELECT * FROM player'); + await establish(harness); + expect(cancelled.isEnded()).toBe(true); + const messages = harness.factory.current.outgoingMessages; + expect(messages).toHaveLength(1); + expect(messages[0]).toMatchObject({ + tag: 'Subscribe', + value: { queryStrings: ['SELECT * FROM player'] }, + }); + }); + + test('an incomplete replay response is terminal', async () => { + const harness = build(); + await establish(harness); + harness.connection.subscriptionBuilder().subscribe('SELECT * FROM user'); + harness.factory.current.close(); + await runReconnectTimer(harness); + await establish(harness); + const message = harness.factory.current.outgoingMessages[0]; + if (message.tag !== 'SubscribeBatch') throw new Error('Expected replay'); + harness.factory.current.sendToClient( + ServerMessage.SubscribeBatchApplied({ + requestId: message.value.requestId, + results: [], + }) + ); + expect(harness.disconnects.at(-1)?.nextReconnectAttempt).toBeUndefined(); + expect(harness.connection.isReconnecting).toBe(false); + expect(harness.factory.current.closed).toBe(true); + }); + + test('disconnect in onConnect prevents subscription replay', async () => { + const harness = build(); + await establish(harness); + harness.connection.subscriptionBuilder().subscribe('SELECT * FROM user'); + harness.factory.current.close(); + harness.connection['onConnect'](() => harness.connection.disconnect()); + await runReconnectTimer(harness); + await establish(harness); + expect(harness.factory.current.closed).toBe(true); + expect(harness.factory.current.outgoingMessages).toEqual([]); + expect(vi.getTimerCount()).toBe(0); + }); +}); diff --git a/crates/bindings-typescript/tests/table_cache_reconnect.test.ts b/crates/bindings-typescript/tests/table_cache_reconnect.test.ts new file mode 100644 index 00000000000..657e434ab67 --- /dev/null +++ b/crates/bindings-typescript/tests/table_cache_reconnect.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, test } from 'vitest'; +import { ModuleContext, tablesToSchema } from '../src/lib/schema'; +import { table } from '../src/lib/table'; +import { t } from '../src/lib/type_builders'; +import { DbConnectionImpl } from '../src/sdk/db_connection_impl'; +import { EventEmitter } from '../src/sdk/event_emitter'; +import type { EventContextInterface } from '../src/sdk/event_context'; +import { TableCacheImpl, type Operation } from '../src/sdk/table_cache'; +import { WebsocketTestAdapterFactory } from '../src/sdk/websocket_test_adapter'; + +const schema = tablesToSchema(new ModuleContext(), { + item: table({ name: 'item' }, { value: t.string() }), +}); +const remoteModule = { + ...schema, + reducers: [], + procedures: [], + versionInfo: { cliVersion: '2.8.3' }, +}; + +describe('reconciliation without primary keys', () => { + test('preserves unchanged rows and adjusts overlapping subscription references', () => { + const connection = new DbConnectionImpl({ + uri: new URL('ws://localhost'), + nameOrAddress: 'test', + emitter: new EventEmitter(), + remoteModule, + createWSFn: new WebsocketTestAdapterFactory().openWebSocket, + compression: 'none', + lightMode: false, + }); + const cache = new TableCacheImpl( + schema.tables.item + ); + const ctx: EventContextInterface = { + db: connection.db, + reducers: connection.reducers, + isActive: true, + subscriptionBuilder: () => connection.subscriptionBuilder(), + disconnect: () => connection.disconnect(), + event: { id: 'test', tag: 'SubscribeApplied' }, + }; + const insert: Operation<{ value: string }> = { + type: 'insert', + rowId: 'row-bytes', + row: { value: 'same' }, + }; + cache.applyOperations([insert, insert], ctx); + const callbacks = cache.applyOperations( + [...cache.snapshotDeleteOperations(), insert], + ctx, + { skipIdenticalUpdates: true } + ); + expect(callbacks).toEqual([]); + expect(cache.count()).toBe(1n); + expect(cache.snapshotDeleteOperations()).toHaveLength(1); + const deletes = cache.applyOperations([{ ...insert, type: 'delete' }], ctx); + expect(deletes.map(callback => callback.type)).toEqual(['delete']); + expect(cache.count()).toBe(0n); + connection.disconnect(); + }); +}); diff --git a/docs/docs/00100-intro/00200-quickstarts/00300-nodejs.md b/docs/docs/00100-intro/00200-quickstarts/00300-nodejs.md index 0c75b7d8905..65081a59555 100644 --- a/docs/docs/00100-intro/00200-quickstarts/00300-nodejs.md +++ b/docs/docs/00100-intro/00200-quickstarts/00300-nodejs.md @@ -134,35 +134,32 @@ spacetime sql "SELECT * FROM person" - Open `src/main.ts` to see the Node.js client. It uses `DbConnection.builder()` to connect to SpacetimeDB, subscribes to tables, and registers callbacks for insert/delete events. Unlike browser apps, Node.js stores the authentication token in a file instead of localStorage. + Open `src/main.ts` to see the Node.js client. It uses `DbConnection.builder()` to connect to SpacetimeDB, subscribes to tables, and registers callbacks for insert/delete events. Unlike browser apps, Node.js stores the authentication token in a file instead of localStorage. Enable `withAutomaticReconnect()` to recover after connection loss. Register subscriptions and row callbacks once, outside `onConnect`, which fires again on reconnect. ```typescript import { DbConnection } from './module_bindings/index.js'; -DbConnection.builder() +const conn = DbConnection.builder() .withUri(HOST) .withDatabaseName(DB_NAME) - .withToken(loadToken()) // Load saved token from file - .onConnect((conn, identity, token) => { + .withToken(loadToken()) + .withAutomaticReconnect() + .onConnect((_conn, identity, token) => { console.log('Connected! Identity:', identity.toHexString()); - saveToken(token); // Save token for future connections - - // Subscribe to all tables - conn.subscriptionBuilder() - .onApplied((ctx) => { - // Show current people - const people = [...ctx.db.person.iter()]; - console.log('Current people:', people.length); - }) - .subscribeToAllTables(); - - // Listen for table changes - conn.db.person.onInsert((ctx, person) => { - console.log(`[Added] ${person.name}`); - }); + saveToken(token); }) .build(); + +conn.subscriptionBuilder() + .onApplied(ctx => { + console.log('Current people:', [...ctx.db.person.iter()].length); + }) + .subscribeToAllTables(); + +conn.db.person.onInsert((_ctx, person) => { + console.log(`[Added] ${person.name}`); +}); ```` diff --git a/docs/docs/00100-intro/00200-quickstarts/00400-typescript.md b/docs/docs/00100-intro/00200-quickstarts/00400-typescript.md index 3c83bde433b..5b2be5a7766 100644 --- a/docs/docs/00100-intro/00200-quickstarts/00400-typescript.md +++ b/docs/docs/00100-intro/00200-quickstarts/00400-typescript.md @@ -120,6 +120,12 @@ spacetime logs +## Reconnect after connection loss + +For browser or Node.js clients, enable `.withAutomaticReconnect()` on your generated `DbConnection` builder. Subscriptions and row callbacks survive reconnects; register them once rather than inside `onConnect`. If your auth tokens expire, also provide an initial token with `.withToken(initialToken)` and a refresh callback with `.withTokenProvider(() => auth.getAccessToken())`. + +See [automatic reconnection](../../00200-core-concepts/00600-clients/00700-typescript-reference.md#method-withautomaticreconnect) for lifecycle callbacks and framework behavior. + ## Next steps - See the [Chat App Tutorial](../00300-tutorials/00100-chat-app.md) for a complete example 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..32be97adb1e 100644 --- a/docs/docs/00200-core-concepts/00600-clients/00300-connection.md +++ b/docs/docs/00200-core-concepts/00600-clients/00300-connection.md @@ -29,6 +29,7 @@ import { DbConnection } from './module_bindings'; const conn = DbConnection.builder() .withUri("https://maincloud.spacetimedb.com") .withDatabaseName("my_database") + .withAutomaticReconnect() .build(); ``` @@ -84,6 +85,7 @@ To connect to a database hosted on MainCloud: const conn = DbConnection.builder() .withUri("https://maincloud.spacetimedb.com") .withDatabaseName("my_database") + .withAutomaticReconnect() .build(); ``` @@ -245,17 +247,25 @@ const TOKEN_KEY = `${HOST}/${DB_NAME}/auth_token`; const conn = DbConnection.builder() .withUri(HOST) .withDatabaseName(DB_NAME) + .withToken(localStorage.getItem(TOKEN_KEY) ?? undefined) + .withAutomaticReconnect() .onConnect((conn, identity, token) => { console.log(`Connected! Identity: ${identity.toHexString()}`); // Save token for reconnection — keyed per server/database localStorage.setItem(TOKEN_KEY, token); }) - .onConnectError((_ctx, error) => { - console.error(`Connection failed:`, error); + .onConnectError((_ctx, error, attempt, delayMs) => { + console.error('Connection failed:', error); + if (attempt !== undefined) console.log(`Retry ${attempt} in ${delayMs} ms`); }) - .onDisconnect(() => { - console.log('Disconnected from SpacetimeDB'); - }); + .onDisconnect((_ctx, error, attempt, delayMs) => { + if (attempt !== undefined) { + console.log(`Connection lost; retry ${attempt} in ${delayMs} ms`, error); + } else { + console.log('Connection ended', error); + } + }) + .build(); ``` @@ -397,13 +407,15 @@ Conn->Disconnect(); ### Reconnection Behavior -:::note[Reconnection behavior] +For TypeScript, add `.withAutomaticReconnect()` to the builder to recover after an established connection drops. The connection object, cache, table handles, and callbacks remain usable. Cache reads serve the last known data while `isReconnecting` is `true`; subscriptions are replayed and reconciled after reconnecting. Register subscriptions and row callbacks once, outside `onConnect`, because `onConnect` fires after every successful reconnect. -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. +`onDisconnect` and `onConnectError` receive optional `nextReconnectAttempt` and `nextReconnectDelayMs` parameters. Both are `undefined` when the core SDK will not retry. Initial connection failures are not retried, and `disconnect()` cancels recovery. Reducer and procedure calls made during an outage fail immediately; in-flight calls fail with an unknown-result error because they may have executed. -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. +Use `.withTokenProvider(() => auth.getAccessToken())` alongside `.withToken(initialToken)` for expiring credentials. The provider runs before reconnect attempts when the retained token needs refreshing, not periodically while connected. -::: +The TypeScript React, Solid, and Svelte providers enable automatic reconnection through their shared connection manager. Vue and Angular require `.withAutomaticReconnect()` on the provider's builder. See the [TypeScript reference](./00700-typescript-reference.md#method-withautomaticreconnect) for retry policy, token refresh, and framework behavior. This feature requires a server that supports session IDs and batch subscriptions. + +For TypeScript connections without this option, and for other SDKs described on this page, create a new connection if you need to recover after a connection loss. ## Connection Identity diff --git a/docs/docs/00200-core-concepts/00600-clients/00700-typescript-reference.md b/docs/docs/00200-core-concepts/00600-clients/00700-typescript-reference.md index 4980dc1b9f5..b6a04328131 100644 --- a/docs/docs/00200-core-concepts/00600-clients/00700-typescript-reference.md +++ b/docs/docs/00200-core-concepts/00600-clients/00700-typescript-reference.md @@ -129,6 +129,8 @@ Construct a `DbConnection` by calling `DbConnection.builder()` and chaining conf | [`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) | Keep the connection and subscriptions usable across connection loss. | +| [`withTokenProvider` method](#method-withtokenprovider) | Refresh credentials before reconnecting. | | [`build` method](#method-build) | Finalize configuration and connect. | #### Method `withUri` @@ -177,29 +179,41 @@ 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. +`onConnect` fires again after each successful automatic reconnect. Register row callbacks and subscriptions once, outside this callback, to avoid accumulating duplicate listeners or subscriptions. Use `onConnect` for work needed on every connection, such as saving the token. + #### Callback `onConnectError` ```typescript class DbConnectionBuilder { public onConnectError( - callback: (ctx: ErrorContext, error: Error) => void + callback: ( + ctx: ErrorContext, + error: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number + ) => void ): DbConnectionBuilder; } ``` -Chain a call to `.onConnectError(callback)` to your builder to register a callback to run when your connection fails. +Called when the initial connection or a reconnect attempt fails before `onConnect`. When another attempt is scheduled, `nextReconnectAttempt` is its one-based number and `nextReconnectDelayMs` is the delay in milliseconds. Both are `undefined` when the SDK will not retry. Initial connection failures are never retried by the core SDK. #### Callback `onDisconnect` ```typescript class DbConnectionBuilder { public onDisconnect( - callback: (ctx: ErrorContext, error: Error | null) => void + callback: ( + ctx: ErrorContext, + error?: Error, + nextReconnectAttempt?: number, + nextReconnectDelayMs?: number + ) => void ): DbConnectionBuilder; } ``` -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. +Called when an established connection is lost, or when the application calls [`disconnect`](#method-disconnect). With automatic reconnection enabled, the trailing parameters describe the next attempt and delay in milliseconds. Both are `undefined` when no retry is scheduled. An explicit `disconnect()` also passes `undefined` for `error`, including when called between reconnect attempts. #### Method `withToken` @@ -209,7 +223,91 @@ 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. +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 `undefined` is passed, SpacetimeDB will generate a new `Identity` and sign a new private access token for the connection. + +#### Method `withAutomaticReconnect` + +```typescript +class DbConnectionBuilder { + public withAutomaticReconnect(): this; +} +``` + +Enable automatic reconnection after an established connection drops. This is opt-in for plain TypeScript connections and requires a server with session IDs and batch subscription support. + +The SDK keeps the same connection object, table handles, subscription handles, and registered callbacks. Each new connection has a fresh `ConnectionId`, while the SDK retains the authentication token to preserve the client's `Identity`, including for anonymous clients. + +Retries continue until `disconnect()` or a recognized terminal failure, such as a changed identity, rejected credentials with no remaining refresh attempt, or a fatal protocol error. The base delays are 1, 2, 4, 8, 16, and 30 seconds, with ±50% jitter and a final 30-second cap. A successful connection resets the backoff. Initial connection failures do not retry. Browser resume events also check for silently closed sockets and bring scheduled retries forward. + +```typescript +import { DbConnection, tables } from './module_bindings'; + +const conn = DbConnection.builder() + .withUri('http://localhost:3000') + .withDatabaseName('my-database') + .withAutomaticReconnect() + .onConnect((_conn, identity) => { + console.log('Connected:', identity.toHexString()); + }) + .onDisconnect((_ctx, error, attempt, delayMs) => { + if (attempt !== undefined) { + console.log(`Disconnected; retry ${attempt} in ${delayMs} ms`, error); + } else { + console.log('Connection ended', error); + } + }) + .onConnectError((_ctx, error, attempt, delayMs) => { + if (attempt !== undefined) { + console.log(`Connect failed; retry ${attempt} in ${delayMs} ms`, error); + } else { + console.error('Connection failed; no retry scheduled', error); + } + }) + .build(); + +conn.db.player.onInsert((_ctx, player) => console.log(player)); +conn.subscriptionBuilder() + .onApplied(() => console.log('Subscription applied')) + .subscribe(tables.player); +``` + +While reconnecting: + +- `conn.isActive` is `false` and `conn.isReconnecting` is `true`. +- Cache reads still return the last known rows, which may be stale. +- Reducer and procedure calls fail immediately with `DisconnectedError` and are not sent. Calls already in flight fail with `UnknownCallResultError`: they may have executed, so retrying them could repeat an effect. +- New subscriptions are retained for the next connection. Unsubscribing ends the handle locally and removes it from replay. + +On reconnect, the SDK replays subscriptions in one batch and reconciles the cache before notifying callbacks. Unchanged rows produce no row callbacks; changed rows produce the usual insert, update, or delete callbacks. Each replayed subscription fires `onApplied` again, or `onError` if rejected. `onConnect` precedes subscription replay, so wait for `onApplied` when you need refreshed data. + +#### Method `withTokenProvider` + +```typescript +class DbConnectionBuilder { + public withTokenProvider(provider: () => Promise): this; +} +``` + +Supply fresh credentials for reconnect attempts. This method does not enable automatic reconnection by itself, and the provider is not called for the initial connection. Obtain the initial token first and pass it to `withToken`: + +```typescript +const initialToken = await auth.getAccessToken(); +const conn = DbConnection.builder() + .withUri('https://maincloud.spacetimedb.com') + .withDatabaseName('my-database') + .withToken(initialToken) + .withAutomaticReconnect() + .withTokenProvider(() => auth.getAccessToken()) + .build(); +``` + +Here `auth` is your application's authentication client. Its method should return a usable token, refreshing it through your identity provider when necessary. The returned token must identify the same user; use a new connection for sign-in or account changes. + +Before each reconnect attempt, the SDK reads the retained JWT's `exp` and `iat` claims. It calls the provider when the remaining validity is at most 5% of the token's lifetime, with a minimum margin of 30 seconds. Without `iat`, it uses the 30-second margin. If expiry cannot be read, it calls the provider on every attempt. + +A recognized rejection of the retained token forces a refresh on the next attempt. Rejection of a freshly supplied token is terminal. If the provider throws or rejects, the attempt fails and the SDK retries with backoff. A token exchange service outage is also retryable. + +Refresh happens before reconnecting, not on a periodic timer while connected. The SDK retains tokens in memory; persistence across page reloads remains the application's responsibility. #### Method `build` @@ -292,7 +390,7 @@ interface DbContext { } ``` -Gracefully close the `DbConnection`. Throws an error if the connection is already disconnected. +Close the `DbConnection`. With automatic reconnection enabled, this cancels any scheduled or pending reconnect and fires `onDisconnect` with no error or retry parameters. The connection cannot be restarted; build a new one to connect again. ### Subscribe to queries @@ -563,7 +661,17 @@ interface DbContext { } ``` -`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. +Whether the connection is currently active. With automatic reconnection enabled, this remains `false` until the server's initial connection message arrives, and becomes `false` again during an outage. It does not indicate whether subscriptions have finished applying. + +#### Field `isReconnecting` + +```typescript +class DbConnection { + readonly isReconnecting: boolean; +} +``` + +`true` after losing an established connection while the SDK is waiting for or attempting a reconnect. It is `false` during the initial connection, after a successful reconnect, and after the connection ends. This field is on `DbConnection`, not the general `DbContext` interface. ## Type `EventContext` @@ -1034,7 +1142,11 @@ The SpacetimeDB TypeScript SDK includes React bindings under the `spacetimedb/re The React integration is fully compatible with React StrictMode and correctly handles the double-mount behavior (only one WebSocket connection is created). -While a `SpacetimeDBProvider` is mounted, the shared connection manager also replaces the managed `DbConnection` if the underlying WebSocket closes or reports a connection error. Reconnect attempts use exponential backoff, starting at 1 second and doubling after each consecutive failure up to a 30 second maximum; the backoff resets after a successful connection. In browser environments, the manager also re-checks connection liveness when the page becomes visible, regains focus, returns online, or is restored from the back-forward cache, so a stalled reconnect or silently closed socket can be rebuilt promptly after a suspended tab resumes. Hooks such as `useTable` observe the provider state, receive the fresh connection, and establish their subscriptions again; while the replacement connection is being established, `useTable` reports `isReady` as `false` until its subscription is applied on the new connection. This provider-level recovery does not change the lower-level `DbConnection` contract: applications that create a `DbConnection` directly are still responsible for creating a new connection if they need reconnection behavior. +The React provider enables core automatic reconnection. After an established connection drops, it retains the same `DbConnection` and reflects its lifecycle events in provider state. `useSpacetimeDB().isActive` is `false` during the outage; `useTable` reports `isReady` as `false` until its subscription applies again. Cached rows can remain visible while stale, so use these flags for a connection-status indicator. + +The shared connection manager still creates a replacement connection when the core SDK reports that it will not retry, including initial connection failures. Those replacements use the manager's existing exponential backoff. Calling `disconnect()` explicitly prevents this recovery. + +Pass `withTokenProvider` on the provider's builder when your credentials expire. Keep the builder stable across renders, as in the example below. | Name | Description | | ----------------------------------------------------------- | --------------------------------------------------------- | @@ -1175,7 +1287,20 @@ An opaque identifier for a client connection to a database, intended to differen ## Framework Integrations -The SpacetimeDB TypeScript SDK includes built-in integrations for React, SolidJS, Vue, and Svelte. These provide reactive hooks that automatically subscribe to queries and re-render when data changes. +The SpacetimeDB TypeScript SDK includes built-in integrations for React, SolidJS, Vue, Svelte, and Angular. + +React, Solid, and Svelte use the shared connection manager, which enables automatic reconnection on their builders. Vue and Angular build connections directly: add `.withAutomaticReconnect()` to the builder passed to their provider. All integrations accept `.withTokenProvider(...)` on that builder. These settings belong on the builder, not on individual table hooks. + +For example, configure a Vue or Angular connection builder with: + +```typescript +const connectionBuilder = DbConnection.builder() + .withUri('http://localhost:3000') + .withDatabaseName('my-database') + .withAutomaticReconnect(); +``` + +For expiring credentials, obtain the initial token with your auth client, then add `.withToken(initialToken).withTokenProvider(() => auth.getAccessToken())`. See [token refresh](#method-withtokenprovider) for when the provider runs. ### React