diff --git a/config/scripts/mobile-log-revisions-benchmark.mjs b/config/scripts/mobile-log-revisions-benchmark.mjs new file mode 100644 index 00000000000..aa42094fee6 --- /dev/null +++ b/config/scripts/mobile-log-revisions-benchmark.mjs @@ -0,0 +1,85 @@ +import assert from 'node:assert/strict' +import { execFileSync } from 'node:child_process' +import { readFileSync } from 'node:fs' +import { dirname, resolve } from 'node:path' +import { performance } from 'node:perf_hooks' +import { build } from 'esbuild' +import { buildCounterbalancedSchedule } from './counterbalanced-benchmark-schedule.mjs' + +const baseline = process.argv[2] ?? '20ab9950654' +const file = 'mobile/src/transport/connection-log-buffer.ts' +async function load(contents) { + const result = await build({ + stdin: { contents, loader: 'ts', resolveDir: dirname(resolve(file)) }, + bundle: true, + platform: 'node', + format: 'esm', + write: false, + tsconfigRaw: {} + }) + return import( + `data:text/javascript;base64,${Buffer.from(result.outputFiles[0].text).toString('base64')}` + ) +} +const before = await load( + execFileSync('git', ['show', `${baseline}:${file}`], { encoding: 'utf8' }) +) +const after = await load(readFileSync(file, 'utf8')) +const drain = () => new Promise((resolve) => setImmediate(resolve)) +async function run(module, count, startup) { + let calls = 0 + let bytes = 0 + let stored = '' + const store = module.createConnectionLogStore(200, { + load: async () => [], + save: async (_host, snapshot) => { + stored = JSON.stringify(snapshot) + calls++ + bytes += Buffer.byteLength(stored) + } + }) + if (!startup) { + await store.hydrate('a') + await drain() + calls = 0 + bytes = 0 + } + const start = performance.now() + for (let i = 0; i < count; i++) { + store.append('a', { id: `${i}`, ts: i, level: 'info', message: `connection event ${i}` }) + } + await drain() + return { ms: performance.now() - start, calls, bytes, stored } +} +const results = [] +for (const count of [1, 25, 200, 1000]) { + for (const startup of [false, true]) { + const arms = { before, after } + const initialBefore = await run(before, count, startup) + const initialAfter = await run(after, count, startup) + assert.equal(initialAfter.stored, initialBefore.stored) + const samples = { before: [], after: [] } + for (const pair of buildCounterbalancedSchedule(10, 'before', 'after')) { + for (const arm of pair) { + samples[arm].push((await run(arms[arm], count, startup)).ms) + } + } + const median = (values) => { + const sorted = [...values].sort((a, b) => a - b) + return (sorted[4] + sorted[5]) / 2 + } + results.push({ + count, + startup, + before: { + calls: initialBefore.calls, + bytes: initialBefore.bytes, + ms: median(samples.before) + }, + after: { calls: initialAfter.calls, bytes: initialAfter.bytes, ms: median(samples.after) } + }) + } +} +console.log( + JSON.stringify({ baseline, node: process.version, platform: process.platform, results }, null, 2) +) diff --git a/mobile/src/transport/connection-log-buffer.ts b/mobile/src/transport/connection-log-buffer.ts index 941f7971498..5af67051cf8 100644 --- a/mobile/src/transport/connection-log-buffer.ts +++ b/mobile/src/transport/connection-log-buffer.ts @@ -32,6 +32,10 @@ export function createConnectionLogStore( const hydrationFailedHosts = new Set() const hydrationByHost = new Map>() const saveByHost = new Map>() + const persistenceRevisionByHost = new Map< + string, + { snapshot: readonly ConnectionLogEntry[]; saved: boolean } + >() // Why: useSyncExternalStore compares snapshots by reference — getSnapshot // must return the SAME array until the data actually changes, or React // loops re-rendering. Cache per host; invalidate on append. @@ -46,6 +50,7 @@ export function createConnectionLogStore( const notify = (hostId: string): void => { snapshotByHost.delete(hostId) + persistenceRevisionByHost.delete(hostId) const listeners = listenersByHost.get(hostId) if (listeners) { for (const listener of listeners) { @@ -58,16 +63,29 @@ export function createConnectionLogStore( if (!persistence || !hydratedHosts.has(hostId)) { return } - const snapshot = [...(entriesByHost.get(hostId) ?? [])] + let revision = persistenceRevisionByHost.get(hostId) + if (!revision) { + revision = { snapshot: [...(entriesByHost.get(hostId) ?? [])], saved: false } + persistenceRevisionByHost.set(hostId, revision) + } + const currentRevision = revision + if (currentRevision.saved) { + return + } const previous = saveByHost.get(hostId) ?? Promise.resolve() const pending = previous .catch(() => {}) .then(async () => { - try { - await persistence.save(hostId, snapshot) - } catch { - await persistence.save(hostId, snapshot) + // Duplicate requests retain retry opportunities until this revision is durable. + if (currentRevision.saved) { + return } + try { + await persistence.save(hostId, currentRevision.snapshot) + } catch { + await persistence.save(hostId, currentRevision.snapshot) + } + currentRevision.saved = true }) .catch(() => {}) saveByHost.set(hostId, pending) diff --git a/mobile/src/transport/connection-log-persistence-revisions.test.ts b/mobile/src/transport/connection-log-persistence-revisions.test.ts new file mode 100644 index 00000000000..e85567d9eaf --- /dev/null +++ b/mobile/src/transport/connection-log-persistence-revisions.test.ts @@ -0,0 +1,119 @@ +import { describe, expect, it, vi } from 'vitest' +import { createConnectionLogStore, type ConnectionLogPersistence } from './connection-log-buffer' +import type { ConnectionLogEntry } from './types' + +const entry = (id: number): ConnectionLogEntry => ({ + id: `${id}`, + ts: id, + level: 'info', + message: `event ${id}` +}) +const drain = () => new Promise((resolve) => setTimeout(resolve, 0)) + +describe('connection log persistence revisions', () => { + it('writes a synchronous burst once per host after hydration', async () => { + const save = vi.fn(async () => {}) + const store = createConnectionLogStore(200, { load: async () => [], save }) + await Promise.all(['a', 'b'].map((host) => store.hydrate(host))) + await drain() + save.mockClear() + for (let i = 0; i < 1000; i++) { + store.append('a', entry(i)) + store.append('b', entry(i + 2000)) + } + await drain() + expect(save).toHaveBeenCalledTimes(2) + expect(save).toHaveBeenCalledWith('a', store.get('a')) + expect(save).toHaveBeenCalledWith('b', store.get('b')) + }) + + it('shares one snapshot across startup appends waiting on hydration', async () => { + let loaded!: (entries: ConnectionLogEntry[]) => void + const save = vi.fn(async () => {}) + const store = createConnectionLogStore(200, { + load: () => + new Promise((resolve) => { + loaded = resolve + }), + save + }) + for (let i = 1; i <= 100; i++) { + store.append('a', entry(i)) + } + loaded([entry(0)]) + await store.hydrate('a') + await drain() + expect(save).toHaveBeenCalledTimes(1) + expect(save).toHaveBeenLastCalledWith( + 'a', + Array.from({ length: 101 }, (_, i) => entry(i)) + ) + }) + + it('preserves distinct queued revisions while a save is delayed', async () => { + const saved: string[][] = [] + let release!: () => void + let delay = false + const store = createConnectionLogStore(2, { + load: async () => [], + save: async (_host, entries) => { + saved.push(entries.map((item) => item.id)) + if (delay) { + delay = false + await new Promise((resolve) => { + release = resolve + }) + } + } + }) + await store.hydrate('a') + await drain() + saved.length = 0 + delay = true + store.append('a', entry(1)) + await drain() + store.append('a', entry(2)) + await drain() + store.append('a', entry(3)) + await drain() + expect(saved).toEqual([['1']]) + release() + await drain() + expect(saved).toEqual([['1'], ['1', '2'], ['2', '3']]) + }) + + it('retains retries and later attempts when both initial save attempts fail', async () => { + const save = vi.fn(async () => {}) + const store = createConnectionLogStore(200, { load: async () => [], save }) + await store.hydrate('a') + await drain() + save.mockClear() + save.mockRejectedValueOnce(new Error('first')).mockRejectedValueOnce(new Error('retry')) + store.append('a', entry(1)) + store.append('a', entry(2)) + store.append('a', entry(3)) + await drain() + expect(save).toHaveBeenCalledTimes(3) + expect(save.mock.calls[2]?.[1]).toBe(save.mock.calls[0]?.[1]) + for (const [, snapshot] of save.mock.calls) { + expect(snapshot).toEqual([entry(1), entry(2), entry(3)]) + } + }) + it('keeps every queued retry opportunity during a sustained failure', async () => { + const save = vi.fn(async () => {}) + const store = createConnectionLogStore(200, { load: async () => [], save }) + await store.hydrate('a') + await drain() + save.mockReset().mockRejectedValue(new Error('unavailable')) + for (let i = 0; i < 3; i++) { + store.append('a', entry(i)) + } + await drain() + expect(save).toHaveBeenCalledTimes(6) + save.mockClear().mockResolvedValue(undefined) + store.append('a', entry(3)) + await drain() + expect(save).toHaveBeenCalledTimes(1) + expect(save).toHaveBeenLastCalledWith('a', store.get('a')) + }) +})