diff --git a/docs/superpowers/plans/2026-09-03-feed-debug-persist-batching.md b/docs/superpowers/plans/2026-09-03-feed-debug-persist-batching.md new file mode 100644 index 00000000..0b7855a2 --- /dev/null +++ b/docs/superpowers/plans/2026-09-03-feed-debug-persist-batching.md @@ -0,0 +1,71 @@ +# Feed debug: batch persistence appends by time + +Fixes #748. Refs #722, #103. + +## Problem + +`useFeedDebugPersist` runs in a `useEffect` keyed on the `runtimes` map, +which is replaced dozens of times per second while a turn streams. Every +replacement that finds new feed-debug entries sends them to main at once, +and the success path re-drains immediately, so the append cadence follows +the streaming cadence: 8,649–10,159 `debug:append-feed-log` invokes in a +77-minute run (20/s in bursts), each one a `writeFile(flag:'a')` plus a +retention-prune schedule on the main thread, with p95 352 ms and a 1.2 s +worst case on the Sep 1 run. + +## Design + +The hook keeps its two cursors (persisted / in-flight) and the one-append- +in-flight rule — those carry the retry and durability semantics — and only +changes WHEN a flush is started: + +- A pure policy, `decideFeedDebugFlush`, answers "flush now", "arm a timer + for N ms" or "nothing" from `{ pendingCount, pendingBytes, lastAttemptAt, + now, inFlight }`. The first batch after a quiet period flushes + immediately, so the entry that explains a bad paint is on disk within one + IPC round trip. Anything that arrives within `FEED_DEBUG_FLUSH_INTERVAL_MS` + (1.5 s) of the last attempt waits for one per-session timer. + `FEED_DEBUG_FLUSH_MAX_PENDING` (256 entries) or + `FEED_DEBUG_FLUSH_MAX_PENDING_BYTES` (1 MiB, from the ring's cached per- + entry estimate) forces an immediate flush: the ring is byte-capped at + 4 MiB and evicts from the head, and with the #722 shape (hundreds of KB + per entry) twenty entries already exceed it, so a count ceiling alone + would let unpersisted entries be evicted while waiting. +- On IPC success or failure the hook re-runs the policy, which arms the + timer unless a ceiling is crossed; a failed attempt counts as an attempt + so a rejecting main does not turn streaming into a retry storm. +- A session that leaves `runtimes` (replacement, pane close, tab kill, + reload) gets one final, unpaced flush of its trailing entries from the + last runtime snapshot — parked until an in-flight append resolves if + there is one — then its bookkeeping is dropped. Those entries (exit code, + kill reason) are the ones debug bundles read for closed panes. +- Timers live in a ref keyed by session and are cleared on unmount only — + the effect itself re-runs on every runtimes replacement and must not tear + them down. + +Trade-off accepted: up to 1.5 s of feed-debug entries can be lost when the +renderer goes away without unmounting cleanly (hard crash) or at window +close (an `invoke` cannot be awaited past teardown). The ring's own +durability note already accepts this. + +Not in scope: the main-side append path (`feedDebugLog.ts`) and the +per-frame `runtimes` replacement itself, which is a separate structural +item. Two pre-existing durability bugs found in review are tracked +separately: #770 (soft reload restarts ids below the persisted cursor) and +#771 (the fail-closed stat path resolves instead of rejecting). + +## Verification + +- `feedDebugFlushPolicy.test.ts`: immediate on first batch and after a + quiet period; timer for the remainder of the interval inside it; forced + flush at the count ceiling and at the byte ceiling; nothing while an + append is in flight. +- `useFeedDebugPersist.renderer.test.tsx` (fake timers, stubbed + `window.api.appendFeedDebugLog`): rapid runtimes replacements produce one + append immediately and one more after the interval carrying every entry + that arrived in between; entries arriving while an append is in flight + are paced on resolve, not drained; the byte ceiling forces a flush; a + failed append leaves the entries pending and retries after the interval; + a removed session's trailing entries are flushed at once, or once its + in-flight append resolves; unmount clears the timer. +- `npx tsc -b`. diff --git a/src/main/ipc/debug.ts b/src/main/ipc/debug.ts index 71c3d912..415072ff 100644 --- a/src/main/ipc/debug.ts +++ b/src/main/ipc/debug.ts @@ -16,9 +16,9 @@ import type { LifecycleIpcDiagnostics } from '@main/ipc/lifecycle.js' // Debug-panel IPC. // // Two endpoints: -// - debug:append-feed-log — streaming, fire-and-forget batches from -// the FeedDebugPanel flush timer. Serialized per-session in the -// storage module; this handler just validates shape and forwards. +// - debug:append-feed-log — paced batches from the renderer's +// persistence hook (useFeedDebugPersist, #748). Serialized per-session +// in the storage module; this handler just validates shape and forwards. // - debug:save-bundle — one-shot, user-triggered from the "Save // Debug Logs" command palette entry. Renderer assembles the bundle // (state + feed-debug + proxy semantic + html raw/clean) and we diff --git a/src/renderer/src/session-runtime/feedDebug.ts b/src/renderer/src/session-runtime/feedDebug.ts index ecfe9a9d..00fe770f 100644 --- a/src/renderer/src/session-runtime/feedDebug.ts +++ b/src/renderer/src/session-runtime/feedDebug.ts @@ -10,8 +10,10 @@ import { estimateJsonBytes } from '@renderer/session-runtime/liveEntryWindow' // (screen_update, process_state, submit, jsonl_entries, SEM, …) // appends one entry here, capped at FEED_DEBUG_LOG_CAP entries AND // FEED_DEBUG_LOG_MAX_BYTES. The FeedDebugPanel renders this in realtime; -// the same entries are shipped to main/storage/feedDebugLog.ts every tick -// to be written to disk (per-session JSONL under STATE_DIR/feed-debug/). +// the same entries are shipped to main/storage/feedDebugLog.ts in paced +// batches (useFeedDebugPersist, #748: at most one append per session per +// 1.5 s, sooner at a count/byte ceiling, immediately when a session is +// removed) to be written to disk (per-session JSONL under STATE_DIR/feed-debug/). // // Why cap the in-memory array: long-running sessions could // accumulate tens of thousands of entries, bloating the runtime map @@ -33,10 +35,12 @@ const FEED_DEBUG_LOG_CAP = 500 // entry window. 4 MiB is generous for small records (500 × a few hundred // bytes never gets near it) and only bites when payloads are large, which is // exactly the case that needs bounding. Disk persistence is best-effort (one -// in-flight append batch per session; entries evicted before it resolves are -// never written — see useFeedDebugPersist), and the byte budget narrows that -// loss window for large records. The ring exists for the live panel, not for -// durability; a durable buffer is a separate concern (see the header). +// in-flight append batch per session plus a ≤1.5 s pacing window; entries +// evicted before they are sent are never written — see useFeedDebugPersist, +// whose byte ceiling forces a flush at a quarter of this budget), and the +// byte budget narrows that loss window for large records. The ring exists +// for the live panel, not for durability; a durable buffer is a separate +// concern (see the header). export const FEED_DEBUG_LOG_MAX_BYTES = 4 * 1024 * 1024 // A real entry always serialises to at least its id/ts/summary envelope, so a diff --git a/src/renderer/src/workspace/hook/persistence/feedDebugFlushPolicy.test.ts b/src/renderer/src/workspace/hook/persistence/feedDebugFlushPolicy.test.ts new file mode 100644 index 00000000..3784ee3d --- /dev/null +++ b/src/renderer/src/workspace/hook/persistence/feedDebugFlushPolicy.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, it } from 'vitest' + +import { countPendingFeedDebug, decideFeedDebugFlush } from './feedDebugFlushPolicy' + +// #748: appends must be paced by time, not by runtimes replacements. + +describe('decideFeedDebugFlush', () => { + const base = { intervalMs: 1_000, maxPending: 10, maxPendingBytes: 4_096, pendingBytes: 100, inFlight: false } + + it('flushes the first batch immediately and again once the interval has elapsed', () => { + expect(decideFeedDebugFlush({ ...base, pendingCount: 1, lastAttemptAt: null, now: 5_000 })).toEqual({ kind: 'now' }) + expect(decideFeedDebugFlush({ ...base, pendingCount: 1, lastAttemptAt: 4_000, now: 5_000 })).toEqual({ kind: 'now' }) + }) + + it('waits out the remainder of the interval for entries that follow a flush', () => { + expect(decideFeedDebugFlush({ ...base, pendingCount: 3, lastAttemptAt: 4_700, now: 5_000 })).toEqual({ + kind: 'wait', + delayMs: 700, + }) + }) + + it('forces a flush at the pending ceiling even inside the interval', () => { + expect(decideFeedDebugFlush({ ...base, pendingCount: 10, lastAttemptAt: 4_900, now: 5_000 })).toEqual({ kind: 'now' }) + }) + + it('forces a flush when the pending bytes reach the byte ceiling before the count does', () => { + expect( + decideFeedDebugFlush({ ...base, pendingCount: 2, pendingBytes: 4_096, lastAttemptAt: 4_900, now: 5_000 }), + ).toEqual({ kind: 'now' }) + }) + + it('does nothing while an append is unresolved or when nothing is pending', () => { + expect(decideFeedDebugFlush({ ...base, pendingCount: 50, lastAttemptAt: null, now: 5_000, inFlight: true })).toEqual({ + kind: 'none', + }) + expect(decideFeedDebugFlush({ ...base, pendingCount: 0, lastAttemptAt: null, now: 5_000 })).toEqual({ kind: 'none' }) + }) +}) + +describe('countPendingFeedDebug', () => { + it('counts only the tail newer than the persisted cursor', () => { + const log = [{ id: 1 }, { id: 2 }, { id: 3 }, { id: 4 }] + expect(countPendingFeedDebug(log, 0)).toBe(4) + expect(countPendingFeedDebug(log, 2)).toBe(2) + expect(countPendingFeedDebug(log, 4)).toBe(0) + expect(countPendingFeedDebug([], 0)).toBe(0) + }) +}) diff --git a/src/renderer/src/workspace/hook/persistence/feedDebugFlushPolicy.ts b/src/renderer/src/workspace/hook/persistence/feedDebugFlushPolicy.ts new file mode 100644 index 00000000..47d3d236 --- /dev/null +++ b/src/renderer/src/workspace/hook/persistence/feedDebugFlushPolicy.ts @@ -0,0 +1,85 @@ +// When to ship pending feed-debug entries to main (#748). +// +// WHY a policy at all: `useFeedDebugPersist` runs on every `runtimes` +// replacement, which happens dozens of times per second while a turn +// streams. Flushing whenever there is something pending made the append +// cadence follow the streaming cadence — 8,649–10,159 `debug:append-feed-log` +// invokes in a 77-minute run, 20/s in bursts, each one a `writeFile(flag:'a')` +// and a retention-prune schedule on the main thread (p95 352 ms, max 1.2 s on +// the Sep 1 run). The entries are diagnostics; nothing reads them within a +// second of being written. Batching them by time turns that into ≤ 1 append +// per session per interval without touching the cursors that carry the +// retry/durability semantics. +// +// WHY the first batch after a quiet period goes out immediately: the entry +// that explains a bad paint is usually the first one after silence, and a +// "Save Debug Logs" right after seeing it should find it on disk. The timer +// only paces what FOLLOWS a flush. +// +// WHY two ceilings, count AND bytes: the ring is byte-capped at 4 MiB (#722) +// and evicts from the head, so an entry that waits on the timer can be +// evicted before it is ever persisted — a loss the immediate-flush version +// could only suffer during one in-flight append. Count alone does not bound +// that: the #722 shape is a few hundred KB per `visible_rows` entry, where +// twenty entries already exceed the ring. A byte ceiling at a quarter of the +// ring forces the flush long before eviction can reach unpersisted entries +// at any realistic rate, and keeps one append from carrying a +// multi-megabyte batch. +// +// WHY `lastAttemptAt` and not `lastSuccessAt`: a rejected append (main not +// ready, disk full) must not turn the streaming cadence back into a retry +// storm. Counting the failed attempt as a flush rate-limits retries to the +// same interval; the entries stay pending because the persisted cursor only +// advances on success. + +export const FEED_DEBUG_FLUSH_INTERVAL_MS = 1_500 +export const FEED_DEBUG_FLUSH_MAX_PENDING = 256 +export const FEED_DEBUG_FLUSH_MAX_PENDING_BYTES = 1024 * 1024 + +export type FeedDebugFlushDecision = + | { kind: 'now' } + | { kind: 'wait'; delayMs: number } + | { kind: 'none' } + +export type FeedDebugFlushInput = { + pendingCount: number + /** Estimated JSON bytes of the pending entries (the ring's own estimate). */ + pendingBytes: number + /** Epoch ms of the last append attempt for this session, or null. */ + lastAttemptAt: number | null + now: number + /** An append IPC is unresolved; the resolve path re-runs the policy. */ + inFlight: boolean + intervalMs?: number + maxPending?: number + maxPendingBytes?: number +} + +export function decideFeedDebugFlush(input: FeedDebugFlushInput): FeedDebugFlushDecision { + const intervalMs = input.intervalMs ?? FEED_DEBUG_FLUSH_INTERVAL_MS + const maxPending = input.maxPending ?? FEED_DEBUG_FLUSH_MAX_PENDING + const maxPendingBytes = input.maxPendingBytes ?? FEED_DEBUG_FLUSH_MAX_PENDING_BYTES + if (input.pendingCount <= 0) return { kind: 'none' } + if (input.inFlight) return { kind: 'none' } + if (input.pendingCount >= maxPending) return { kind: 'now' } + if (input.pendingBytes >= maxPendingBytes) return { kind: 'now' } + if (input.lastAttemptAt === null) return { kind: 'now' } + const elapsed = input.now - input.lastAttemptAt + if (elapsed >= intervalMs) return { kind: 'now' } + return { kind: 'wait', delayMs: intervalMs - elapsed } +} + +/** Entries with id > lastPersistedId. Ids are assigned in append order, so + * scanning from the tail stops at the first persisted entry — O(pending), + * not O(ring) — which matters because this runs per runtimes replacement. */ +export function countPendingFeedDebug( + log: ReadonlyArray<{ id: number }>, + lastPersistedId: number, +): number { + let count = 0 + for (let i = log.length - 1; i >= 0; i -= 1) { + if (log[i]!.id <= lastPersistedId) break + count += 1 + } + return count +} diff --git a/src/renderer/src/workspace/hook/persistence/useFeedDebugPersist.renderer.test.tsx b/src/renderer/src/workspace/hook/persistence/useFeedDebugPersist.renderer.test.tsx new file mode 100644 index 00000000..6fe0d367 --- /dev/null +++ b/src/renderer/src/workspace/hook/persistence/useFeedDebugPersist.renderer.test.tsx @@ -0,0 +1,322 @@ +import { act, renderHook } from '@testing-library/react' +import { StrictMode } from 'react' +import type { MutableRefObject } from 'react' +import { afterEach, describe, expect, it, vi } from 'vitest' + +import { appendFeedDebugLog } from '@renderer/session-runtime/feedDebug' +import { emptyRuntime, type SessionRuntime } from '@renderer/session-runtime/state' +import type { WorkspaceRefs } from '@renderer/workspace/hook/refs' +import type { SessionId } from '@renderer/workspace/types' + +import { + FEED_DEBUG_FLUSH_INTERVAL_MS, + FEED_DEBUG_FLUSH_MAX_PENDING, + FEED_DEBUG_FLUSH_MAX_PENDING_BYTES, +} from './feedDebugFlushPolicy' +import { useFeedDebugPersist } from './useFeedDebugPersist' + +// #748: the hook re-runs on every runtimes replacement (dozens per second +// while streaming); appends must be paced by the flush interval and carry +// everything that arrived in between, without weakening the cursor rules. + +const originalApiDescriptor = Object.getOwnPropertyDescriptor(window, 'api') + +afterEach(() => { + vi.useRealTimers() + vi.restoreAllMocks() + if (originalApiDescriptor) { + Object.defineProperty(window, 'api', originalApiDescriptor) + } else { + Reflect.deleteProperty(window, 'api') + } +}) + +function ref(current: T): MutableRefObject { + return { current } +} + +function withEntries(runtime: SessionRuntime, count: number, data?: unknown): SessionRuntime { + let next = runtime + for (let i = 0; i < count; i += 1) { + next = appendFeedDebugLog(next, { layer: 'RENDER', kind: 'visible_rows', summary: `rows ${i}`, data }) + } + return next +} + +function makeRefs(runtimes: Record) { + return { + latestRuntimesRef: ref(runtimes), + persistedFeedDebugIdRef: ref>({}), + inFlightFeedDebugIdRef: ref>({}), + } as unknown as WorkspaceRefs +} + +function install(appendFeedDebugLog: ReturnType) { + Object.defineProperty(window, 'api', { configurable: true, value: { appendFeedDebugLog } }) +} + +describe('useFeedDebugPersist pacing', () => { + it('sends the first batch at once, then one paced batch carrying every entry that arrived in between', async () => { + vi.useFakeTimers() + const append = vi.fn().mockResolvedValue(undefined) + install(append) + let runtimes: Record = { s1: withEntries(emptyRuntime(), 1) } + const refs = makeRefs(runtimes) + const { rerender } = renderHook(({ r }) => useFeedDebugPersist(r, refs), { initialProps: { r: runtimes } }) + expect(append).toHaveBeenCalledTimes(1) + await act(async () => { await vi.advanceTimersByTimeAsync(0) }) + expect(refs.persistedFeedDebugIdRef.current.s1).toBe(1) + + // Streaming: 20 replacements inside the interval, one entry each. + for (let i = 0; i < 20; i += 1) { + runtimes = { s1: withEntries(runtimes.s1!, 1) } + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + await act(async () => { await vi.advanceTimersByTimeAsync(10) }) + } + expect(append).toHaveBeenCalledTimes(1) + + await act(async () => { await vi.advanceTimersByTimeAsync(FEED_DEBUG_FLUSH_INTERVAL_MS) }) + expect(append).toHaveBeenCalledTimes(2) + const second = append.mock.calls[1]![0] as { entries: Array<{ id: number }> } + expect(second.entries.map(e => e.id)).toEqual(Array.from({ length: 20 }, (_, i) => i + 2)) + expect(refs.persistedFeedDebugIdRef.current.s1).toBe(21) + }) + + it('forces a flush inside the interval once pending entries reach the ceiling', async () => { + vi.useFakeTimers() + const append = vi.fn().mockResolvedValue(undefined) + install(append) + let runtimes: Record = { s1: withEntries(emptyRuntime(), 1) } + const refs = makeRefs(runtimes) + const { rerender } = renderHook(({ r }) => useFeedDebugPersist(r, refs), { initialProps: { r: runtimes } }) + await act(async () => { await vi.advanceTimersByTimeAsync(0) }) + expect(append).toHaveBeenCalledTimes(1) + + runtimes = { s1: withEntries(runtimes.s1!, FEED_DEBUG_FLUSH_MAX_PENDING) } + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + expect(append).toHaveBeenCalledTimes(2) + }) + + it('forces a flush inside the interval when a few large entries reach the byte ceiling', async () => { + // The #722 shape: hundreds of KB per entry. Waiting for the count + // ceiling would let the byte-capped ring evict them unpersisted. + vi.useFakeTimers() + const append = vi.fn().mockResolvedValue(undefined) + install(append) + let runtimes: Record = { s1: withEntries(emptyRuntime(), 1) } + const refs = makeRefs(runtimes) + const { rerender } = renderHook(({ r }) => useFeedDebugPersist(r, refs), { initialProps: { r: runtimes } }) + await act(async () => { await vi.advanceTimersByTimeAsync(0) }) + expect(append).toHaveBeenCalledTimes(1) + + const big = { rows: 'x'.repeat(FEED_DEBUG_FLUSH_MAX_PENDING_BYTES / 2) } + runtimes = { s1: withEntries(runtimes.s1!, 3, big) } + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + expect(append).toHaveBeenCalledTimes(2) + const forced = append.mock.calls[1]![0] as { entries: Array<{ id: number }> } + expect(forced.entries).toHaveLength(3) + }) + + it('keeps failed entries pending and retries no sooner than the interval', async () => { + vi.useFakeTimers() + vi.spyOn(console, 'warn').mockImplementation(() => undefined) + const append = vi.fn() + .mockRejectedValueOnce(new Error('main not ready')) + .mockResolvedValue(undefined) + install(append) + let runtimes: Record = { s1: withEntries(emptyRuntime(), 2) } + const refs = makeRefs(runtimes) + const { rerender } = renderHook(({ r }) => useFeedDebugPersist(r, refs), { initialProps: { r: runtimes } }) + await act(async () => { await vi.advanceTimersByTimeAsync(0) }) + expect(append).toHaveBeenCalledTimes(1) + expect(refs.persistedFeedDebugIdRef.current.s1).toBeUndefined() + + // A replacement right after the failure must not retry immediately. + runtimes = { s1: withEntries(runtimes.s1!, 1) } + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + expect(append).toHaveBeenCalledTimes(1) + + await act(async () => { await vi.advanceTimersByTimeAsync(FEED_DEBUG_FLUSH_INTERVAL_MS) }) + expect(append).toHaveBeenCalledTimes(2) + const retry = append.mock.calls[1]![0] as { entries: Array<{ id: number }> } + expect(retry.entries.map(e => e.id)).toEqual([1, 2, 3]) + expect(refs.persistedFeedDebugIdRef.current.s1).toBe(3) + }) + + it('paces entries that arrive while an append is in flight instead of draining on resolve', async () => { + vi.useFakeTimers() + let resolveFirst!: () => void + const append = vi.fn() + .mockImplementationOnce(() => new Promise(resolve => { resolveFirst = resolve })) + .mockResolvedValue(undefined) + install(append) + let runtimes: Record = { s1: withEntries(emptyRuntime(), 1) } + const refs = makeRefs(runtimes) + const { rerender } = renderHook(({ r }) => useFeedDebugPersist(r, refs), { initialProps: { r: runtimes } }) + expect(append).toHaveBeenCalledTimes(1) + + // Entries keep arriving while the first append is unresolved. + for (let i = 0; i < 3; i += 1) { + runtimes = { s1: withEntries(runtimes.s1!, 1) } + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + } + await act(async () => { + resolveFirst() + await vi.advanceTimersByTimeAsync(100) + }) + // Resolving must arm the timer, not drain immediately. + expect(append).toHaveBeenCalledTimes(1) + await act(async () => { await vi.advanceTimersByTimeAsync(FEED_DEBUG_FLUSH_INTERVAL_MS) }) + expect(append).toHaveBeenCalledTimes(2) + const paced = append.mock.calls[1]![0] as { entries: Array<{ id: number }> } + expect(paced.entries.map(e => e.id)).toEqual([2, 3, 4]) + }) + + it('flushes a removed session\'s trailing entries at once instead of losing them to the timer', async () => { + // Session replacement / pane close delete the runtime while the pacing + // timer is armed; the final entries (exit code, kill reason) must still + // reach disk. + vi.useFakeTimers() + const append = vi.fn().mockResolvedValue(undefined) + install(append) + let runtimes: Record = { s1: withEntries(emptyRuntime(), 1) } + const refs = makeRefs(runtimes) + const { rerender } = renderHook(({ r }) => useFeedDebugPersist(r, refs), { initialProps: { r: runtimes } }) + await act(async () => { await vi.advanceTimersByTimeAsync(0) }) + expect(append).toHaveBeenCalledTimes(1) + + runtimes = { s1: withEntries(runtimes.s1!, 2) } + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + expect(append).toHaveBeenCalledTimes(1) + + runtimes = {} + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + expect(append).toHaveBeenCalledTimes(2) + const final = append.mock.calls[1]![0] as { entries: Array<{ id: number }> } + expect(final.entries.map(e => e.id)).toEqual([2, 3]) + await act(async () => { await vi.advanceTimersByTimeAsync(FEED_DEBUG_FLUSH_INTERVAL_MS * 2) }) + expect(append).toHaveBeenCalledTimes(2) + }) + + it('flushes a removed session\'s trailing entries once its in-flight append resolves', async () => { + vi.useFakeTimers() + let resolveFirst!: () => void + const append = vi.fn() + .mockImplementationOnce(() => new Promise(resolve => { resolveFirst = resolve })) + .mockResolvedValue(undefined) + install(append) + let runtimes: Record = { s1: withEntries(emptyRuntime(), 1) } + const refs = makeRefs(runtimes) + const { rerender } = renderHook(({ r }) => useFeedDebugPersist(r, refs), { initialProps: { r: runtimes } }) + expect(append).toHaveBeenCalledTimes(1) + + runtimes = { s1: withEntries(runtimes.s1!, 1) } + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + runtimes = {} + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + expect(append).toHaveBeenCalledTimes(1) + + await act(async () => { + resolveFirst() + await vi.advanceTimersByTimeAsync(0) + }) + expect(append).toHaveBeenCalledTimes(2) + const final = append.mock.calls[1]![0] as { entries: Array<{ id: number }> } + expect(final.entries.map(e => e.id)).toEqual([2]) + }) + + it('clears the pacing timer on unmount', async () => { + vi.useFakeTimers() + const append = vi.fn().mockResolvedValue(undefined) + install(append) + let runtimes: Record = { s1: withEntries(emptyRuntime(), 1) } + const refs = makeRefs(runtimes) + const { rerender, unmount } = renderHook(({ r }) => useFeedDebugPersist(r, refs), { initialProps: { r: runtimes } }) + await act(async () => { await vi.advanceTimersByTimeAsync(0) }) + runtimes = { s1: withEntries(runtimes.s1!, 1) } + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + unmount() + await act(async () => { await vi.advanceTimersByTimeAsync(FEED_DEBUG_FLUSH_INTERVAL_MS * 2) }) + expect(append).toHaveBeenCalledTimes(1) + }) + + it('paces a removed session\'s final-flush retries instead of spinning on a persistent rejection', async () => { + // Review blocker: the rejection path re-invokes consider() with the + // session gone from latestRuntimesRef, which re-enters the final + // branch. Unpaced, a persistently failing append became one IPC round + // trip + one console.warn per microtask — the retry storm this PR + // exists to kill, resurrected on the removal path. + vi.useFakeTimers() + vi.spyOn(console, 'warn').mockImplementation(() => undefined) + const append = vi.fn() + .mockResolvedValueOnce(undefined) // initial live batch (id 1) + .mockRejectedValueOnce(new Error('disk full at close')) // final flush fails + .mockRejectedValueOnce(new Error('disk full at close')) // paced retry fails too + .mockResolvedValue(undefined) + install(append) + let runtimes: Record = { s1: withEntries(emptyRuntime(), 1) } + const refs = makeRefs(runtimes) + const { rerender } = renderHook(({ r }) => useFeedDebugPersist(r, refs), { initialProps: { r: runtimes } }) + await act(async () => { await vi.advanceTimersByTimeAsync(0) }) + expect(append).toHaveBeenCalledTimes(1) + + runtimes = { s1: withEntries(runtimes.s1!, 2) } + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + runtimes = {} + refs.latestRuntimesRef.current = runtimes + rerender({ r: runtimes }) + // The FIRST final flush is immediate: removal is one append, not a stream. + expect(append).toHaveBeenCalledTimes(2) + await act(async () => { await vi.advanceTimersByTimeAsync(0) }) + // Rejection landed. No retry before the interval, however many microtasks run. + expect(append).toHaveBeenCalledTimes(2) + + await act(async () => { await vi.advanceTimersByTimeAsync(FEED_DEBUG_FLUSH_INTERVAL_MS - 1) }) + expect(append).toHaveBeenCalledTimes(2) + await act(async () => { await vi.advanceTimersByTimeAsync(1) }) + expect(append).toHaveBeenCalledTimes(3) + + // The paced retry also rejects; exactly one more after another interval, + // then the success clears the cursors and stops the cycle. + await act(async () => { await vi.advanceTimersByTimeAsync(FEED_DEBUG_FLUSH_INTERVAL_MS - 1) }) + expect(append).toHaveBeenCalledTimes(3) + await act(async () => { await vi.advanceTimersByTimeAsync(1) }) + expect(append).toHaveBeenCalledTimes(4) + expect(refs.persistedFeedDebugIdRef.current.s1).toBe(3) + await act(async () => { await vi.advanceTimersByTimeAsync(FEED_DEBUG_FLUSH_INTERVAL_MS * 3) }) + expect(append).toHaveBeenCalledTimes(4) + }) + + it('recovers after a StrictMode-style double mount keeps the hook mounted', async () => { + // The unmount-only effect runs its cleanup on React 18's dev + // simulated unmount and MUST re-arm on the simulated remount, or + // every later .then/.catch bails and pacing only recovers on a lucky + // render. Render directly under StrictMode to exercise the same + // effect → cleanup → effect sequence on one hook instance. + vi.useFakeTimers() + const append = vi.fn().mockResolvedValue(undefined) + install(append) + const runtimes: Record = { s1: withEntries(emptyRuntime(), 2) } + const refs = makeRefs(runtimes) + const { rerender } = renderHook( + ({ r }) => useFeedDebugPersist(r, refs), + { initialProps: { r: runtimes }, wrapper: StrictMode }, + ) + await act(async () => { await vi.advanceTimersByTimeAsync(0) }) + expect(append).toHaveBeenCalledTimes(1) + // The resolve path after the simulated remount must still advance the + // durable cursor; with a stuck unmountedRef it never would. + expect(refs.persistedFeedDebugIdRef.current.s1).toBe(2) + }) +}) diff --git a/src/renderer/src/workspace/hook/persistence/useFeedDebugPersist.ts b/src/renderer/src/workspace/hook/persistence/useFeedDebugPersist.ts index 316fcd73..883e33d2 100644 --- a/src/renderer/src/workspace/hook/persistence/useFeedDebugPersist.ts +++ b/src/renderer/src/workspace/hook/persistence/useFeedDebugPersist.ts @@ -1,13 +1,17 @@ -import { useEffect } from 'react' +import { useEffect, useRef } from 'react' import type { SessionId } from '@renderer/workspace/types' import type { SessionRuntime } from '@renderer/session-runtime/state' import type { WorkspaceRefs } from '@renderer/workspace/hook/refs' -// Ship runtime feed-debug entries to the main process on every -// runtime update. The main-side queue writes them to -// STATE_DIR/feed-debug/.jsonl. +import { estimateFeedDebugLogBytes } from '@renderer/session-runtime/feedDebug' + +import { countPendingFeedDebug, decideFeedDebugFlush, FEED_DEBUG_FLUSH_INTERVAL_MS } from './feedDebugFlushPolicy' + +// Ship runtime feed-debug entries to the main process, batched by time +// (see feedDebugFlushPolicy.ts for the cadence and why). The main-side +// queue writes them to STATE_DIR/feed-debug/.jsonl. // // `persistedFeedDebugIdRef` tracks the largest feed-debug entry id // main has confirmed as written. `inFlightFeedDebugIdRef` tracks the @@ -42,15 +46,61 @@ export function useFeedDebugPersist( runtimes: Record, refs: WorkspaceRefs, ): void { + // Per-session pacing state. These are refs, not effect-local variables, + // because the effect below re-runs on every runtimes replacement and a + // timer armed in one pass must survive into the next; and they are not + // WorkspaceRefs because nothing outside this hook reads them. + const timersRef = useRef>>({}) + const lastAttemptAtRef = useRef>({}) + // WHY removed sessions get a final, unpaced flush (review of #750): a + // session leaves `runtimes` on replacement, pane close, tab kill and + // reload, and its LAST entries — the exit code, the kill reason — are the + // ones written in the final second. With pacing alone those entries sat + // on a timer that found no runtime when it fired and were never written; + // the pre-pacing hook shipped them from the same effect pass that + // appended them. Removal is one append per session, so it bypasses the + // cadence entirely. If an append for that session is still in flight the + // final runtime is parked here and flushed when it resolves. + const previousRuntimesRef = useRef>({}) + const finalRuntimesRef = useRef>({}) + // Stamp of the last FINAL-branch send per removed session. The live + // path's `lastAttemptAtRef` cannot answer "is this final flush a + // retry": the batch that preceded removal is typically only + // milliseconds old, and keying the final flush off it would delay a + // session's trailing entries (exit code, kill reason) by up to one + // interval — the loss this branch exists to prevent. Only attempts the + // final branch itself made count as retries. + const finalAttemptAtRef = useRef>({}) + const unmountedRef = useRef(false) + + // Unmount only: the pacing timers must NOT be torn down by the per- + // replacement effect's cleanup, or every streamed delta would cancel and + // re-arm them and the interval would never elapse under load. Anything + // still pending at unmount (window close) is lost — an `invoke` cannot be + // awaited past teardown — which is the one loss this hook accepts. + // The body re-arms the flag because React 18 StrictMode double-mounts in + // dev: effect → cleanup(true) → effect again. Without the reset every + // post-remount .then/.catch would bail forever and pacing would only + // recover on the next lucky render. Production never remounts, so this + // is dev-only correctness — but it is also free. + useEffect(() => { + unmountedRef.current = false + return () => { + unmountedRef.current = true + for (const timer of Object.values(timersRef.current)) clearTimeout(timer) + timersRef.current = {} + } + }, []) + useEffect(() => { - const flushSession = (sessionId: SessionId, runtime: SessionRuntime): void => { - if (runtime.feedDebugLog.length === 0) return + const send = (sessionId: SessionId, runtime: SessionRuntime): void => { const lastPersistedId = refs.persistedFeedDebugIdRef.current[sessionId] ?? 0 const lastInFlightId = refs.inFlightFeedDebugIdRef.current[sessionId] ?? 0 const batch = selectFeedDebugAppendBatch(runtime, lastPersistedId, lastInFlightId) if (!batch) return const { entries: pending, maxPendingId } = batch refs.inFlightFeedDebugIdRef.current[sessionId] = maxPendingId + lastAttemptAtRef.current[sessionId] = Date.now() // Advance the durable cursor ONLY after the IPC append actually // resolves. A previous version advanced optimistically before // the write, so a transient failure (disk full, IPC timeout, @@ -60,18 +110,15 @@ export function useFeedDebugPersist( // pending id range while the IPC is unresolved, then this `.then` // makes that reservation durable once main confirms the append. // - // Re-entrancy note: the effect fires on every runtimes object - // replacement, which can happen dozens of times per second while - // a semantic stream is active. We allow only ONE unresolved - // append per session, not just one append per id range. Sending - // a newer range while an older range is unresolved would re-open - // a subtle data-loss case: if the older disk write failed but - // the newer one succeeded, advancing `persisted` to the newer id - // would make the failed older entries look durable. Serializing - // at the renderer keeps retry semantics simple; the success path - // below immediately drains any entries that arrived while the IPC - // was in flight, so the one-at-a-time rule does not rely on a - // future React render to make progress. + // Re-entrancy note: we allow only ONE unresolved append per + // session, not just one append per id range. Sending a newer range + // while an older range is unresolved would re-open a subtle + // data-loss case: if the older disk write failed but the newer one + // succeeded, advancing `persisted` to the newer id would make the + // failed older entries look durable. Serializing at the renderer + // keeps retry semantics simple. The success path re-runs the pacing + // policy (which arms a timer rather than draining immediately), so + // progress does not depend on a future React render either. void window.api .appendFeedDebugLog({ sessionId, @@ -90,10 +137,8 @@ export function useFeedDebugPersist( if (refs.inFlightFeedDebugIdRef.current[sessionId] === maxPendingId) { delete refs.inFlightFeedDebugIdRef.current[sessionId] } - const latestRuntime = refs.latestRuntimesRef.current[sessionId] - if (latestRuntime) { - flushSession(sessionId, latestRuntime) - } + if (unmountedRef.current) return + consider(sessionId, refs.latestRuntimesRef.current[sessionId]) }) .catch(err => { if (refs.inFlightFeedDebugIdRef.current[sessionId] === maxPendingId) { @@ -101,11 +146,112 @@ export function useFeedDebugPersist( } // eslint-disable-next-line no-console console.warn(`[feed-debug ${sessionId.slice(0, 8)}] append failed`, err) + if (unmountedRef.current) return + // Re-run the policy so an idle session retries after the interval + // without waiting for another runtimes replacement; `lastAttemptAt` + // was stamped at send time, so this arms a timer, never a storm. + consider(sessionId, refs.latestRuntimesRef.current[sessionId]) }) } + const forget = (sessionId: SessionId): void => { + const timer = timersRef.current[sessionId] + if (timer !== undefined) { + clearTimeout(timer) + delete timersRef.current[sessionId] + } + delete lastAttemptAtRef.current[sessionId] + delete finalRuntimesRef.current[sessionId] + delete finalAttemptAtRef.current[sessionId] + } + + const consider = (sessionId: SessionId, runtime: SessionRuntime | undefined): void => { + const lastPersistedId = refs.persistedFeedDebugIdRef.current[sessionId] ?? 0 + const lastInFlightId = refs.inFlightFeedDebugIdRef.current[sessionId] ?? 0 + const inFlight = lastInFlightId > lastPersistedId + // A session that has left `runtimes`: flush whatever its final + // snapshot still holds, then drop its bookkeeping. + const final = runtime === undefined ? finalRuntimesRef.current[sessionId] : undefined + if (final !== undefined) { + if (inFlight) return + const remaining = countPendingFeedDebug(final.feedDebugLog, lastPersistedId) + if (remaining > 0) { + // WHY retries (but not the first flush) are paced here — review + // blocker on this PR: the rejection path below re-invokes + // consider() with the session already gone from + // latestRuntimesRef, which lands HERE again. Unpaced, a + // persistently failing append (disk full, EACCES at close time, + // main rejecting the shape) became a tight loop of one IPC round + // trip + one console.warn per microtask — the exact retry storm + // this PR exists to kill, resurrected on the removal path. The + // FIRST final flush stays immediate (removal is one append per + // session, not a stream); only attempts the final branch itself + // made inside the interval wait, via the same per-session timer + // the live path uses. That timer's callback reads + // latestRuntimesRef, which is undefined for a removed session + // and therefore routes back into this branch. + const finalAttemptAt = finalAttemptAtRef.current[sessionId] + if (finalAttemptAt !== undefined) { + const elapsed = Date.now() - finalAttemptAt + if (elapsed < FEED_DEBUG_FLUSH_INTERVAL_MS) { + if (timersRef.current[sessionId] !== undefined) return + timersRef.current[sessionId] = setTimeout(() => { + delete timersRef.current[sessionId] + consider(sessionId, refs.latestRuntimesRef.current[sessionId]) + }, FEED_DEBUG_FLUSH_INTERVAL_MS - elapsed) + return + } + } + finalAttemptAtRef.current[sessionId] = Date.now() + send(sessionId, final) + return + } + forget(sessionId) + return + } + if (!runtime || runtime.feedDebugLog.length === 0) return + const pendingCount = countPendingFeedDebug(runtime.feedDebugLog, lastPersistedId) + if (pendingCount === 0) return + const decision = decideFeedDebugFlush({ + pendingCount, + // The ring's per-entry byte estimate is cached per entry object, so + // this is a cache-hit walk over the pending tail, not a stringify. + pendingBytes: estimateFeedDebugLogBytes(runtime.feedDebugLog.slice(-pendingCount)), + lastAttemptAt: lastAttemptAtRef.current[sessionId] ?? null, + now: Date.now(), + inFlight, + }) + if (decision.kind === 'none') return + if (decision.kind === 'now') { + const timer = timersRef.current[sessionId] + if (timer !== undefined) { + clearTimeout(timer) + delete timersRef.current[sessionId] + } + send(sessionId, runtime) + return + } + // One timer per session; a later replacement inside the same interval + // rides the timer that is already armed. The callback reads the latest + // runtime from the ref, so it ships everything that arrived meanwhile. + if (timersRef.current[sessionId] !== undefined) return + timersRef.current[sessionId] = setTimeout(() => { + delete timersRef.current[sessionId] + consider(sessionId, refs.latestRuntimesRef.current[sessionId]) + }, decision.delayMs) + } + + // Sessions that left the map since the last pass get their final flush + // (see finalRuntimesRef); everything else goes through the policy. + const previous = previousRuntimesRef.current + previousRuntimesRef.current = runtimes + for (const [sessionId, runtime] of Object.entries(previous)) { + if (sessionId in runtimes) continue + finalRuntimesRef.current[sessionId] = runtime + consider(sessionId, undefined) + } for (const [sessionId, runtime] of Object.entries(runtimes)) { - flushSession(sessionId, runtime) + consider(sessionId, runtime) } }, [ refs.inFlightFeedDebugIdRef,