diff --git a/docs/connection_reconnect.md b/docs/connection_reconnect.md new file mode 100644 index 0000000000..c9ec7be092 --- /dev/null +++ b/docs/connection_reconnect.md @@ -0,0 +1,262 @@ +# Client connection and reconnection scheme + +## Overview + +The client talks to the server over a single WebSocket. All requests (findAll, tx, loadModel, +ping) go through it. State lives in `foundations/core/packages/client-resources/src/connection.ts`. + +## Connection lifecycle + +``` + [ new Connection() ] + | + v + scheduleOpen(force=false) ------+ + | | + v | + openConnection() | + | | + |-- create WebSocket | + |-- dialTimer = 30s -------+ | + v | | + wsocket.onopen | | + |-- send HelloRequest(-1) | | + v | | + wsocket.onmessage | | + |-- HelloResponse | | + | |-- helloReceived=true | | + | |-- clearTimeout dial -+ | + | |-- account, lastHash | + | |-- for req in requests: | + | | req.reconnect() ---+---> setTimeout 50ms -> sendData() + | |-- onConnect(event) | | + | | event = | +-- STORM of simultaneous + | | Connected | sends + | | | Reconnected | + | | | Maintenance | + | |-- schedulePing() | + | | + |-- (regular responses) | + | | + wsocket.onclose | + |-- scheduleOpen(force=true) -+ + | + wsocket.onerror + |-- delay += 1 (max 3s) +``` + +## Key timings + +| Constant | Value | Purpose | +|-----------|----------|-----------| +| `pingTimeout` | 10s | Ping send interval | +| `hangTimeout` | **5 min** | "Hung" socket threshold → force close | +| `dialTimeout` | 30s | Hello timeout. No hello → reconnect | +| `reconnect delay` | 50ms | Delay before retrying each pending request | + +## Connection state + +- `requests: Map` — all pending requests. +- `onConnectHandlers: OnConnectHandler[]` — waiting on `waitOpenConnection`. +- `websocket` — current socket (recreated on reconnect). +- `sockets` — counter guarding against a race between two parallel connects. +- `pingResponse` — timestamp of the last pong. +- `helloReceived` — hello completed → requests may be sent. +- `slowDownTimer` — adaptive delay under rate limiting. + +## What happens on reconnect + +### 1. Disconnect detection + +Three sources: +1. `wsocket.onclose` → `scheduleOpen(force=true)`. +2. Ping check `pingResponse > hangTimeout` → close(1000). +3. `dialTimer` did not fire within 30s → `onDialTimeout` + force. + +**Mobile problem**: when backgrounded, iOS/Android browsers freeze +timers and the WS. The socket is formally `OPEN`, but no data flows. The hangTimeout=5min +detector wakes up after 5 minutes — **the UI hangs for that entire time**. + +### 2. Repeated openConnection + +`openConnection()` creates a new WebSocket and sends Hello. On the response: + +```ts +for (const [, v] of this.requests.entries()) { + v.reconnect?.() +} +``` + +Each pending request resends `sendData()` after 50ms. + +### 3. onConnect callback → refresh UI + +In `plugins/workbench-resources/src/connect.ts:363`: + +| Event | Action | +|-------|---------| +| `Connected` + `_clientSet` | `refreshClient(tokenChanged)` + `refreshCommunicationClient` | +| `Reconnected` | **only** `refreshCommunicationClient` | +| `Refresh` | `refreshClient(true)` | +| `Upgraded` | `window.location.reload()` | +| `Maintenance` | Show banner | + +**Important**: `refreshClient` is **NOT** called on `Reconnected`. LiveQueries keep +their old results; updates arrive via the tx stream. + +### 4. refreshClient → LiveQueries.refreshConnect + +`foundations/core/packages/query/src/index.ts:112` `refreshConnect(clean)`: + +- Iterates over all queries (`this.queries` + `this.queue`). +- With `clean=true` it clears results and invokes callbacks with an empty array. +- Calls `this.refresh(q)` for each one → **another findAll to the server**. + +For N active queries this means N parallel findAll calls — extra load +on top of the already resent pending requests. + +## What accumulates while the app sleeps + +While the tab is in the background, the WebSocket may be frozen. During that time: + +1. **Ping requests** (10s interval). The `once:true` guard (connection.ts:693) prevents + duplicates, but one pending ping stays hanging. +2. **UI requests** started before backgrounding — e.g. a link navigation + managed to create a findAll that remained in `requests`. +3. **The first request on resume**: the UI becomes active before the dead socket is + detected → Svelte components fire findAll → they land in `requests` and + wait on `waitOpenConnection` until hangTimeout expires (up to 5 min). +4. **LiveQuery subscription requests** — the primary findAll for each subscription. + +When the reconnect finally happens: +- All those N requests get `reconnect()` → after 50ms **all of them fly to the + server at once**. +- The server responds, but the client is busy: Svelte reactivity + JSON parse + + tx-stream processing → the event loop is saturated. +- From the log: `time=1800ms`, `serverTime=200ms`, `toReceive~0` → **the delay is on + the client**, not the network. + +## Why data looks stale after `Reconnected` + +`refreshConnect` is NOT called on `Reconnected`. LiveQueries show +the last known result. Updates arrive only via the tx stream from the +server (broadcasted events). If the server accumulated txes for this session +while it slept, they will arrive — but only after hello and after the queue of +resent requests is worked through. + +## Server-side queue (the `queue` field) + +`service.requests.size` — the number of pending requests in the server session. +Visible in the log: `queue=13` — i.e. the client sent 13 requests at once. +It is not a limiting queue, just a metric. + +## What visibilitychange does + +The client **does** have a `document.visibilitychange` handler (`installVisibilityHandler`). +When the tab becomes visible again, the client checks the socket state and forces +a reconnect if needed: + +- If the socket is `null` / `readyState !== OPEN` / hello not received → immediate + `scheduleOpen(force=true)`. +- Otherwise send a short ping (`once=true`); if no pong arrives within + `visibilityProbeTimeout` (1s) → force reconnect. + +This does not imply a separate `refreshConnect` on `Reconnected`: LiveQueries +still keep the last known result, and updates arrive +via the tx stream after hello and after the resent request queue is processed. + +--- + +## Logging points (storm diagnostics) + +### 1. Connection — track request accumulation + +`foundations/core/packages/client-resources/src/connection.ts`: + +- In `sendRequest` when adding to `this.requests`: log `size, method, id`. +- In `handleMsg` on removal: log `size, age = Date.now() - startTime, method`. +- In the hello-response reconnect branch (line 384): log `reconnectingCount, oldest + pending age`. +- In the `schedulePing` hangTimeout branch: log `timeSinceLastPong, pendingCount`. + +### 2. Visibility + +Add to `Connection`: + +```ts +if (typeof document !== 'undefined') { + document.addEventListener('visibilitychange', () => { + const visible = document.visibilityState === 'visible' + console.log('[conn] visibility', { + visible, + readyState: this.websocket?.readyState, + pendingCount: this.requests.size, + sinceLastPong: Date.now() - this.pingResponse, + oldestRequest: this.getOldestRequestAge() + }) + // Force ping probe on resume + if (visible && this.websocket?.readyState === ClientSocketReadyState.OPEN) { + // liveness probe with a short timeout + } + }) +} +``` + +### 3. LiveQuery — track refresh storms + +`foundations/core/packages/query/src/index.ts` `refreshConnect`: + +```ts +console.log('[lq] refreshConnect', { + clean, + queriesCount: [...this.queries.values()].reduce((a, v) => a + v.size, 0), + queueCount: this.queue.size +}) +const t0 = Date.now() +// ... after loop +console.log('[lq] refreshConnect done', { ms: Date.now() - t0 }) +``` + +### 4. measure findAll — exists already, but extend + +`connection.ts:778-792` — add `pendingCount, sinceReconnect` to the log. + +### 5. Server side (optional) + +`foundations/server/packages/server/src/sessionManager.ts` — when handling a +request, log `sessionId, queueSize, sinceReconnect`. Helps determine whether +the queue is really on the server or on the client. + +### 6. Reconnect storm + +In the `handleMsg` hello branch (line 384), wrap in `setTimeout` with jitter: + +```ts +const reqs = [...this.requests.values()] +reqs.forEach((v, idx) => { + setTimeout(() => v.reconnect?.(), Math.random() * 200) +}) +``` + +For diagnostics, start by simply logging the length of reqs and request lifetimes. + +## Root-cause hypothesis + +A combination of: +1. iOS/Android freezes the WebSocket and setInterval. +2. After resume the socket is formally OPEN, but no data flows. +3. The UI is already active and fires 10-20 findAll calls. +4. Dead-socket detection waits up to 5 minutes (hangTimeout). +5. When finally detected: reconnect → a storm after 50ms across all pending requests. +6. Svelte reactivity + parsing saturate the event loop → findAll time of 1800ms + with serverTime of 200ms. +7. `Reconnected` does not restart queries → UI data stays stale until txes arrive. + +## Recommendations (for discussion, not implementation) + +1. `visibilitychange` → probe ping with 2s timeout → force reconnect on failure. +2. Lower `hangTimeout` to 30-60s or make it adaptive on mobile. +3. Jitter of 50-500ms instead of the shared 50ms in the reconnect branch. +4. On `Reconnected`, call `refreshClient(false)` (without clean) to + refresh critical queries. +5. Cap resend parallelism with a semaphore (e.g. 5 concurrent). diff --git a/foundations/core/packages/client-resources/src/connection.ts b/foundations/core/packages/client-resources/src/connection.ts index 0b3b44c772..6ed03048e9 100644 --- a/foundations/core/packages/client-resources/src/connection.ts +++ b/foundations/core/packages/client-resources/src/connection.ts @@ -38,6 +38,7 @@ import core, { FindOptions, FindResult, generateId, + platformNow, LoadModelResponse, type MeasureContext, MeasureMetricsContext, @@ -63,10 +64,25 @@ const SECOND = 1000 const pingTimeout = 10 * SECOND const hangTimeout = 5 * 60 * SECOND const dialTimeout = 30 * SECOND +// After visibilitychange -> visible we probe with short timeout. +const visibilityProbeTimeout = 1 * SECOND +// Jitter window for reconnect resend storm. +const reconnectJitterMs = 300 +// Throttle window for noisy diagnostics logs in hot paths. +const diagLogThrottleMs = 5 * SECOND + +interface MeasureData { + time: number + serverTime: number + queue: number + result: any + compressedSize: number + uncompressedSize: number +} class RequestPromise { - startTime: number = Date.now() - handleTime?: (diff: number, result: any, serverTime: number, queue: number, toRecieve: number) => void + startTime: number = platformNow() + handleTime?: (data: MeasureData) => void readonly promise: Promise resolve!: (value?: any) => void reject!: (reason?: any) => void @@ -114,12 +130,16 @@ class Connection implements ClientConnection { private upgrading: boolean = false - private pingResponse: number = Date.now() + private pingResponse: number = platformNow() private helloReceived: boolean = false private account: Account | undefined + private visibilityHandler?: () => void + private lastReconnectAt: number = 0 + private lastVisibilityLogAt: number = 0 + onConnect?: (event: ClientConnectEvent, lastTx: string | undefined, data: any) => Promise rpcHandler: RPCHandler @@ -158,9 +178,64 @@ class Connection implements ClientConnection { this.pushHandler(handler) this.onConnect = opt?.onConnect + this.installVisibilityHandler() + this.scheduleOpen(this.ctx, false) } + private installVisibilityHandler (): void { + if (typeof document === 'undefined') return + this.visibilityHandler = () => { + if (this.closed) return + if (document.visibilityState !== 'visible') return + const sinceLastPong = platformNow() - this.pingResponse + const pendingCount = this.requests.size + const readyState = this.websocket?.readyState + const now = platformNow() + if (now - this.lastVisibilityLogAt > diagLogThrottleMs) { + this.lastVisibilityLogAt = now + console.log('[conn] visibility=visible', { + readyState, + pendingCount, + sinceLastPong, + helloReceived: this.helloReceived, + workspace: this.workspace + }) + } + // If socket dead / not ready - force reconnect immediately. + if (this.websocket === null || readyState !== ClientSocketReadyState.OPEN || !this.helloReceived) { + console.log('[conn] visibility -> force reconnect (socket not ready)') + this.scheduleOpen(this.ctx, true) + return + } + // Probe with short timeout. If pong not arrive -> force reconnect. + const probeStart = platformNow() + const probeSocket = this.websocket + void this.sendRequest({ + method: pingConst, + params: [], + once: true, + handleResult: async () => { + if (this.websocket === probeSocket) { + this.pingResponse = platformNow() + } + } + }).catch(() => {}) + setTimeout(() => { + if (this.closed) return + if (this.websocket !== probeSocket) return + if (this.pingResponse < probeStart) { + console.log('[conn] visibility probe timeout -> force reconnect', { + probeTimeout: visibilityProbeTimeout, + pendingCount: this.requests.size + }) + this.scheduleOpen(this.ctx, true) + } + }, visibilityProbeTimeout) + } + document.addEventListener('visibilitychange', this.visibilityHandler) + } + pushHandler (handler: TxHandler): void { this.handlers.push(handler) } @@ -171,7 +246,7 @@ class Connection implements ClientConnection { } private schedulePing (socketId: number): void { - this.pingResponse = Date.now() + this.pingResponse = platformNow() const wsocket = this.websocket clearInterval(this.interval) @@ -180,11 +255,17 @@ class Connection implements ClientConnection { clearInterval(this.interval) return } - if (!this.upgrading && this.pingResponse !== 0 && Date.now() - this.pingResponse > hangTimeout) { + if (!this.upgrading && this.pingResponse !== 0 && platformNow() - this.pingResponse > hangTimeout) { // No ping response from server. if (this.websocket !== null) { - console.log('no ping response from server. Closing socket.', socketId, this.workspace, this.user) + console.log('no ping response from server. Closing socket.', { + socketId, + workspace: this.workspace, + user: this.user, + sinceLastPong: platformNow() - this.pingResponse, + pendingCount: this.requests.size + }) clearInterval(this.interval) this.websocket.close(1000) return @@ -192,6 +273,14 @@ class Connection implements ClientConnection { } if (!this.closed) { + const sinceLastPong = platformNow() - this.pingResponse + if (sinceLastPong > pingTimeout * 2 && this.requests.size > 0) { + console.log('[conn] ping tick - no pong but pending requests', { + sinceLastPong, + pendingCount: this.requests.size, + socketId + }) + } // eslint-disable-next-line @typescript-eslint/no-floating-promises void this.sendRequest({ method: pingConst, @@ -199,7 +288,7 @@ class Connection implements ClientConnection { once: true, handleResult: async (result) => { if (this.websocket === wsocket) { - this.pingResponse = Date.now() + this.pingResponse = platformNow() } } }).catch((err) => { @@ -216,6 +305,10 @@ class Connection implements ClientConnection { clearTimeout(this.openAction) clearTimeout(this.dialTimer) clearInterval(this.interval) + if (this.visibilityHandler !== undefined && typeof document !== 'undefined') { + document.removeEventListener('visibilitychange', this.visibilityHandler) + this.visibilityHandler = undefined + } for (const handler of this.onConnectHandlers) { handler.reject(new Error('Connection closed')) } @@ -286,7 +379,11 @@ class Connection implements ClientConnection { currentRateLimit: RateLimitInfo | undefined slowDownTimer = 0 - handleMsg (socketId: number, resp: Response): void { + handleMsg ( + socketId: number, + resp: Response, + sizes: { compressedSize: number, uncompressedSize: number } = { compressedSize: 0, uncompressedSize: 0 } + ): void { if (this.closed) { return } @@ -381,8 +478,26 @@ class Connection implements ClientConnection { h.resolve() } - for (const [, v] of this.requests.entries()) { - v.reconnect?.() + this.lastReconnectAt = platformNow() + const pendingReqs = [...this.requests.values()] + if (pendingReqs.length > 0) { + const oldest = pendingReqs.reduce((m, r) => Math.max(m, platformNow() - r.startTime), 0) + console.log('[conn] hello reconnect resend', { + count: pendingReqs.length, + oldestAgeMs: oldest, + workspace: this.workspace + }) + } + // Jitter resend to avoid burst on recovered slow mobile networks. + for (const v of pendingReqs) { + const delay = Math.floor(Math.random() * reconnectJitterMs) + if (delay === 0) { + v.reconnect?.() + } else { + setTimeout(() => { + if (!this.closed) v.reconnect?.() + }, delay) + } } void this.onConnect?.( @@ -449,13 +564,14 @@ class Connection implements ClientConnection { } const request = this.requests.get(resp.id) - promise.handleTime?.( - Date.now() - promise.startTime, - resp.result, - resp.time ?? 0, - resp.queue ?? 0, - Date.now() - (resp.bfst ?? 0) - ) + promise.handleTime?.({ + time: platformNow() - promise.startTime, + result: resp.result, + serverTime: resp.time ?? 0, + queue: resp.queue ?? 0, + compressedSize: sizes.compressedSize, + uncompressedSize: sizes.uncompressedSize + }) this.requests.delete(resp.id) if (resp.error !== undefined) { console.log( @@ -512,7 +628,7 @@ class Connection implements ClientConnection { return true } if (text === pongConst) { - this.pingResponse = Date.now() + this.pingResponse = platformNow() return true } } @@ -564,7 +680,7 @@ class Connection implements ClientConnection { return } if (event.data === pongConst) { - this.pingResponse = Date.now() + this.pingResponse = platformNow() return } if (event.data === pingConst) { @@ -584,6 +700,7 @@ class Connection implements ClientConnection { // Support ping/pong return } + const compressedSize = data.byteLength if (this.compressionMode && this.helloReceived) { try { data = uncompress(data) @@ -592,9 +709,10 @@ class Connection implements ClientConnection { console.error(err) } } + const uncompressedSize = data.byteLength try { const resp = this.rpcHandler.readResponse(data, this.binaryMode) - this.handleMsg(socketId, resp) + this.handleMsg(socketId, resp, { compressedSize, uncompressedSize }) } catch (err: any) { if (!this.helloReceived) { // Just error and ignore for now. @@ -609,6 +727,7 @@ class Connection implements ClientConnection { }) } else { let data = event.data + const compressedSize = typeof data === 'string' ? data.length : (data?.byteLength ?? 0) if (this.compressionMode && this.helloReceived) { try { data = uncompress(data) @@ -617,9 +736,10 @@ class Connection implements ClientConnection { console.error(err) } } + const uncompressedSize = typeof data === 'string' ? data.length : (data?.byteLength ?? 0) try { const resp = this.rpcHandler.readResponse(data, this.binaryMode) - this.handleMsg(socketId, resp) + this.handleMsg(socketId, resp, { compressedSize, uncompressedSize }) } catch (err: any) { if (!this.helloReceived) { // Just error and ignore for now. @@ -635,6 +755,12 @@ class Connection implements ClientConnection { wsocket.close() return } + console.log('[conn] wsocket.onclose', { + code: ev.code, + reason: ev.reason, + pendingCount: this.requests.size, + workspace: this.workspace + }) this.scheduleOpen(this.ctx, true) } wsocket.onopen = () => { @@ -673,7 +799,7 @@ class Connection implements ClientConnection { retry?: () => Promise handleResult?: (result: any) => Promise once?: boolean // Require handleResult to retrieve result - measure?: (time: number, result: any, serverTime: number, queue: number, toRecieve: number) => void + measure?: (data: MeasureData) => void allowReconnect?: boolean overrideId?: number }): Promise { @@ -714,7 +840,7 @@ class Connection implements ClientConnection { } promise.sendData = (): void => { if (this.websocket?.readyState === ClientSocketReadyState.OPEN) { - promise.startTime = Date.now() + promise.startTime = platformNow() if (data.method !== pingConst) { const dta = this.rpcHandler.serialize( @@ -775,20 +901,24 @@ class Connection implements ClientConnection { const result = await this.sendRequest({ method: 'findAll', params: [_class, query, options], - measure: (time, result, serverTime, queue, toReceive) => { + measure: ({ time, serverTime, queue, result, compressedSize, uncompressedSize }) => { if (typeof window !== 'undefined' && (time > 1000 || serverTime > 500)) { - console.warn( - 'measure slow findAll', + console.error('measure slow findAll', { time, serverTime, - toReceive, queue, _class, query, options, - result, - JSON.stringify(result).length - ) + reason: time > 1000 ? 'time' : 'serverTime', + compressedSize, + uncompressedSize, + ratio: compressedSize > 0 && uncompressedSize > 0 ? (uncompressedSize / compressedSize).toFixed(2) : 'n/a', + count: result.length, + lookupDocs: result.lookupMap !== undefined ? Object.keys(result.lookupMap).length : 0, + pendingCount: this.requests.size, + sinceReconnectMs: this.lastReconnectAt === 0 ? -1 : platformNow() - this.lastReconnectAt + }) } } }) diff --git a/foundations/core/packages/query/src/__tests__/livequery-coverage.test.ts b/foundations/core/packages/query/src/__tests__/livequery-coverage.test.ts index 0d71440532..92fe59594b 100644 --- a/foundations/core/packages/query/src/__tests__/livequery-coverage.test.ts +++ b/foundations/core/packages/query/src/__tests__/livequery-coverage.test.ts @@ -363,6 +363,85 @@ describe('LiveQuery Coverage Tests', () => { await close() }) + it('refreshConnect should drop idle queries when reconnect gap exceeds IDLE_DROP_GAP_MS', async () => { + const { liveQuery, close } = await getClient() + + const cb = jest.fn() + const unsubscribe = liveQuery.query(core.class.Space, {}, cb) + await new Promise((resolve) => setTimeout(resolve, 50)) + unsubscribe() + await new Promise((resolve) => setTimeout(resolve, 20)) + + // Idle entry sits in queue. Simulate gap above threshold. + const queueBefore = (liveQuery as any).queue.size + expect(queueBefore).toBeGreaterThan(0) + + await liveQuery.refreshConnect(false, LiveQuery.IDLE_DROP_GAP_MS + 1000) + + expect((liveQuery as any).queue.size).toBe(0) + await close() + }) + + it('refreshConnect should keep idle queries when gap is small', async () => { + const { liveQuery, close } = await getClient() + + const cb = jest.fn() + const unsubscribe = liveQuery.query(core.class.Space, {}, cb) + await new Promise((resolve) => setTimeout(resolve, 50)) + unsubscribe() + await new Promise((resolve) => setTimeout(resolve, 20)) + + const queueBefore = (liveQuery as any).queue.size + expect(queueBefore).toBeGreaterThan(0) + + await liveQuery.refreshConnect(false, 100) + + expect((liveQuery as any).queue.size).toBe(queueBefore) + await close() + }) + + it('refreshConnect should refresh active queries ordered by size (smallest first)', async () => { + const { liveQuery, factory, close } = await getClient() + + // Create several spaces so the unfiltered query returns more docs than the filtered one. + for (let i = 0; i < 5; i++) { + await factory.createDoc(core.class.Space, core.space.Model, { + name: `s-${i}`, + description: '', + private: false, + members: [], + archived: false + }) + } + + const bigCb = jest.fn() + const smallCb = jest.fn() + liveQuery.query(core.class.Space, {}, bigCb) + liveQuery.query(core.class.Space, { name: 'no-such-space-xyz' }, smallCb) + await new Promise((resolve) => setTimeout(resolve, 100)) + + const order: number[] = [] + const origRefresh = (liveQuery as any).refresh.bind(liveQuery) + ;(liveQuery as any).refresh = async (q: any) => { + order.push(q.id) + await origRefresh(q) + } + + await liveQuery.refreshConnect(false, 0) + + // Find query ids: smaller result must be refreshed first. + const queries: any[] = [] + for (const v of (liveQuery as any).queries.values()) { + for (const q of v.values()) queries.push(q) + } + const small = queries.find((q) => q.query.name === 'no-such-space-xyz') + const big = queries.find((q) => q.query.name === undefined) + expect(small).toBeDefined() + expect(big).toBeDefined() + expect(order.indexOf(small.id)).toBeLessThan(order.indexOf(big.id)) + await close() + }) + it('should handle associations option', async () => { const { liveQuery, close } = await getClient() diff --git a/foundations/core/packages/query/src/index.ts b/foundations/core/packages/query/src/index.ts index 3babe737be..b611830bfb 100644 --- a/foundations/core/packages/query/src/index.ts +++ b/foundations/core/packages/query/src/index.ts @@ -57,6 +57,7 @@ import core, { getObjectValue, matchQuery, platformNow, + RateLimiter, reduceCalls, shouldShowArchived, toFindResult, @@ -73,6 +74,22 @@ import { Callback, Query, type QueryId } from './types' const CACHE_SIZE = 125 +/** + * @public + * Stats returned from {@link LiveQuery.refreshConnect} so callers can log + * what was dropped vs refreshed after a reconnect. + */ +export interface RefreshConnectStats { + gapMs: number + dropIdle: boolean + droppedQueries: number + droppedDocs: number + activeQueries: number + activeDocs: number + droppedByClass: Map + activeByClass: Map +} + /** * @public */ @@ -108,47 +125,88 @@ export class LiveQuery implements WithTx, Client { return this.client.getModel() } + // Drop idle queries when reconnect gap exceeds this window. + // Idle = sits in queue with no active callbacks. Holding stale results across + // long disconnects (sleep/network drop) is wasteful — drop and let next + // subscription refetch fresh. + static readonly IDLE_DROP_GAP_MS = 5000 + + // Concurrency cap for parallel refresh after reconnect — avoid request storm. + static readonly REFRESH_CONCURRENCY = 4 + // Perform refresh of content since connection established. - async refreshConnect (clean: boolean): Promise { + // `lastReconnectGapMs` — ms since previous successful connection. When the + // gap is large, drop idle queries instead of refreshing them. + // Returns stats so the caller can log/track it. + async refreshConnect (clean: boolean, lastReconnectGapMs: number = 0): Promise { + const dropIdle = lastReconnectGapMs > LiveQuery.IDLE_DROP_GAP_MS + const sizeOf = (q: Query): number => (q.result instanceof Promise ? 0 : q.result.length) + + const stats: RefreshConnectStats = { + gapMs: lastReconnectGapMs, + dropIdle, + droppedQueries: 0, + droppedDocs: 0, + activeQueries: 0, + activeDocs: 0, + droppedByClass: new Map(), + activeByClass: new Map() + } for (const q of [...this.queue.values()]) { - if (!this.removeFromQueue(q)) { - try { - if (clean) { - this.cleanQuery(q) - } - // No need to refresh, since it will be on next for - } catch (err: any) { - if (err instanceof PlatformError) { - if (err.message === 'connection closed') { - continue - } - } - Analytics.handleError(err) - console.error(err) + if (q.callbacks.size === 0) { + // No active callbacks. On long gap drop the cached result; otherwise + // leave it — TTL eviction will clean it. + if (dropIdle) { + const docs = sizeOf(q) + stats.droppedQueries++ + stats.droppedDocs += docs + const key = q._class as string + const prev = stats.droppedByClass.get(key) ?? { count: 0, docs: 0 } + prev.count++ + prev.docs += docs + stats.droppedByClass.set(key, prev) + this.removeQueue(q) } - } else { - // No callbacks, let's remove it on conenct - this.removeQueue(q) + continue } + // Has callbacks but somehow ended up in queue — keep going to refresh below. } + // Collect active subscriptions, refresh smallest first so quick queries + // unblock UI before heavy ones. + const active: Query[] = [] for (const v of this.queries.values()) { for (const q of v.values()) { + if (q.callbacks.size === 0) continue + active.push(q) + const docs = sizeOf(q) + stats.activeQueries++ + stats.activeDocs += docs + const key = q._class as string + const prev = stats.activeByClass.get(key) ?? { count: 0, docs: 0 } + prev.count++ + prev.docs += docs + stats.activeByClass.set(key, prev) + } + } + active.sort((a, b) => sizeOf(a) - sizeOf(b)) + + const limiter = new RateLimiter(LiveQuery.REFRESH_CONCURRENCY) + for (const q of active) { + if (clean) { + this.cleanQuery(q) + } + void limiter.add(async () => { try { - if (clean) { - this.cleanQuery(q) - } - void this.refresh(q) + await this.refresh(q) } catch (err: any) { - if (err instanceof PlatformError) { - if (err.message === 'connection closed') { - continue - } - } + if (err instanceof PlatformError && err.message === 'connection closed') return Analytics.handleError(err) console.error(err) } - } + }) } + await limiter.waitProcessing() + return stats } private cleanQuery (q: Query): void { diff --git a/foundations/core/packages/rpc/src/rpc.ts b/foundations/core/packages/rpc/src/rpc.ts index d1aa70f22d..cab4593d11 100644 --- a/foundations/core/packages/rpc/src/rpc.ts +++ b/foundations/core/packages/rpc/src/rpc.ts @@ -135,7 +135,6 @@ export interface Response { final: boolean } time?: number // Server time to perform operation - bfst?: number // Server time to perform operation queue?: number } diff --git a/foundations/server/packages/server/src/__tests__/sessionManager.test.ts b/foundations/server/packages/server/src/__tests__/sessionManager.test.ts index 9f8c9c9bb4..0614a3c9f1 100644 --- a/foundations/server/packages/server/src/__tests__/sessionManager.test.ts +++ b/foundations/server/packages/server/src/__tests__/sessionManager.test.ts @@ -761,16 +761,6 @@ describe('TSessionManager', () => { expect(sessionManager.ticks).toBe(initialTicks + 1) }) - - it('should update now timestamp', () => { - const beforeNow = sessionManager.now - - // Wait a bit to ensure time passes - setTimeout(() => { - sessionManager.handleTick() - expect(sessionManager.now).toBeGreaterThanOrEqual(beforeNow) - }, 10) - }) }) describe('createSession', () => { diff --git a/foundations/server/packages/server/src/sessionManager.ts b/foundations/server/packages/server/src/sessionManager.ts index 55dfa351f8..58322b0bb6 100644 --- a/foundations/server/packages/server/src/sessionManager.ts +++ b/foundations/server/packages/server/src/sessionManager.ts @@ -126,8 +126,6 @@ export class TSessionManager implements SessionManager { usersProducer: PlatformQueueProducer workspaceConsumer: ConsumerHandle - now: number = Date.now() - ticksContext: MeasureContext hungSessionsWarnPercent = parseInt(process.env.HUNG_SESSIONS_WARN_PERCENT ?? '25') @@ -1111,7 +1109,7 @@ export class TSessionManager implements SessionManager { user: sessionRef.session.getSocialIds().find((it) => it.type !== SocialIdType.HULY)?.value, binary: sessionRef.session.binaryMode, compression: sessionRef.session.useCompression, - totalTime: this.now - sessionRef.session.createTime, + totalTime: Date.now() - sessionRef.session.createTime, workspaceUsers: workspace?.sessions?.size, totalUsers: this.sessions.size }) @@ -1318,7 +1316,6 @@ export class TSessionManager implements SessionManager { id: reqId, result: msg, time: platformNowDiff(st), - bfst: this.now, queue: service.requests.size, rateLimit }), @@ -1333,7 +1330,6 @@ export class TSessionManager implements SessionManager { error, time: platformNowDiff(st), rateLimit, - bfst: this.now, queue: service.requests.size }) } diff --git a/packages/presentation/src/utils.ts b/packages/presentation/src/utils.ts index 8771da2ab4..6aa5303845 100644 --- a/packages/presentation/src/utils.ts +++ b/packages/presentation/src/utils.ts @@ -32,6 +32,7 @@ import core, { type FindOptions, type FindResult, getCurrentAccount, + platformNow, hasAccountRole, type Hierarchy, MeasureMetricsContext, @@ -414,9 +415,29 @@ export async function setClient (_client: Client): Promise { /** * @public */ -export async function refreshClient (clean: boolean): Promise { +export async function refreshClient (clean: boolean, lastReconnectGapMs: number = 0): Promise { if (!(liveQuery?.isClosed() ?? true)) { - await liveQuery?.refreshConnect(clean) + const startedAt = platformNow() + const stats = await liveQuery?.refreshConnect(clean, lastReconnectGapMs) + const elapsedMs = platformNow() - startedAt + if (stats !== undefined && elapsedMs > 1000) { + const fmt = (m: Map): string[] => + [...m.entries()] + .sort((a, b) => b[1].docs - a[1].docs) + .slice(0, 10) + .map(([cls, v]) => `${cls}=${v.count}q/${v.docs}d`) + console.error('[refresh] slow liveQuery refreshConnect', { + elapsedMs, + gapMs: stats.gapMs, + dropIdle: stats.dropIdle, + droppedQueries: stats.droppedQueries, + droppedDocs: stats.droppedDocs, + activeQueries: stats.activeQueries, + activeDocs: stats.activeDocs, + topDropped: fmt(stats.droppedByClass), + topActive: fmt(stats.activeByClass) + }) + } for (const q of globalQueries) { q.refreshClient() } diff --git a/plugins/guest-resources/src/connect.ts b/plugins/guest-resources/src/connect.ts index 313521b7ad..9621156cf9 100644 --- a/plugins/guest-resources/src/connect.ts +++ b/plugins/guest-resources/src/connect.ts @@ -48,13 +48,19 @@ export async function connect (title: string): Promise { const selectWorkspace = await getResource(login.function.SelectWorkspace) let workspaceLoginInfo: WorkspaceLoginInfo | undefined + let connectAttempt = 0 while (true) { const selectResult = await selectWorkspace(wsUrl, exchangedToken) if (!selectResult[2]) { - // Connection error happen, wait and retry - await new Promise((resolve) => setTimeout(resolve, 1000)) + // Connection error: exponential backoff capped at 10s to avoid request storm when server is down + connectAttempt++ + const delay = Math.min(1000 * 2 ** Math.min(connectAttempt - 1, 4), 10000) + await new Promise((resolve) => setTimeout(resolve, delay)) + // Abort if user navigated away to another workspace while we were waiting + if (wsUrl !== getCurrentLocation().path[1]) return continue } + connectAttempt = 0 workspaceLoginInfo = selectResult[1] if (workspaceLoginInfo == null) { const err = `Error selecting workspace ${wsUrl}. There might be something wrong with the token. Please try to log in again.` diff --git a/plugins/workbench-resources/src/connect.ts b/plugins/workbench-resources/src/connect.ts index 0b6b13a49b..b8e6055f28 100644 --- a/plugins/workbench-resources/src/connect.ts +++ b/plugins/workbench-resources/src/connect.ts @@ -12,6 +12,7 @@ import core, { isWorkspaceCreating, type MeasureMetricsContext, metricsToString, + platformNow, pickPrimarySocialId, setCurrentAccount, type SocialId, @@ -105,14 +106,20 @@ export async function connect (title: string): Promise { const selectWorkspace = await getResource(login.function.SelectWorkspace) let workspaceLoginInfo: WorkspaceLoginInfo | undefined + let connectAttempt = 0 while (true) { const selectResult = await ctx.with('select-workspace', {}, async () => await selectWorkspace(wsUrl, null)) workspaceLoginInfo = selectResult[1] ?? undefined if (!selectResult[2]) { - // Connection error happen, wait and retry - await new Promise((resolve) => setTimeout(resolve, 1000)) + // Connection error: exponential backoff capped at 10s to avoid request storm when server is down + connectAttempt++ + const delay = Math.min(500 * 2 ** Math.min(connectAttempt - 1, 5), 10000) + await new Promise((resolve) => setTimeout(resolve, delay)) + // Abort if user navigated away to another workspace while we were waiting + if (wsUrl !== getCurrentLocation().path[1]) return continue } + connectAttempt = 0 // OK but unauthorized - we need to login if (workspaceLoginInfo == null) { @@ -211,6 +218,7 @@ export async function connect (title: string): Promise { } let tokenChanged = false + let lastOnConnectAt = 0 if (_token !== token && _client !== undefined) { // We need to flush all data from memory @@ -362,13 +370,17 @@ export async function connect (title: string): Promise { if (event === ClientConnectEvent.Connected || event === ClientConnectEvent.Reconnected) { setMetadata(presentation.metadata.SessionId, data) } + const gapMs = lastOnConnectAt === 0 ? 0 : platformNow() - lastOnConnectAt + lastOnConnectAt = platformNow() if ((_clientSet && event === ClientConnectEvent.Connected) || event === ClientConnectEvent.Refresh) { + console.log('[workbench] refreshClient triggered', { event, tokenChanged, gapMs }) void ctx.with('refresh client', {}, async () => { - await refreshClient(tokenChanged) + await refreshClient(tokenChanged, gapMs) await refreshCommunicationClient() }) tokenChanged = false } else if (event === ClientConnectEvent.Reconnected) { + console.log('[workbench] reconnected no-refresh (lastTx unchanged)', { gapMs }) await refreshCommunicationClient() }