perf(mobile): skip successful duplicate connection log saves

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)
+)

mobile/src/transport/connection-log-buffer.ts

@@ -32,6 +32,10 @@ export function createConnectionLogStore(
   const hydrationFailedHosts = new Set<string>()
   const hydrationByHost = new Map<string, Promise<void>>()
   const saveByHost = new Map<string, Promise<void>>()
+  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)

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<void>((resolve) => setTimeout(resolve, 0))
+
+describe('connection log persistence revisions', () => {
+  it('writes a synchronous burst once per host after hydration', async () => {
+    const save = vi.fn<ConnectionLogPersistence['save']>(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<ConnectionLogPersistence['save']>(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<void>((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<ConnectionLogPersistence['save']>(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<ConnectionLogPersistence['save']>(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'))
+  })
+})