diff --git a/apps/backend/internal/system/logbundle/service.go b/apps/backend/internal/system/logbundle/service.go index d55d74430a..c4bb221523 100644 --- a/apps/backend/internal/system/logbundle/service.go +++ b/apps/backend/internal/system/logbundle/service.go @@ -22,6 +22,7 @@ import ( const ( defaultCaptureWindow = 15 * time.Second + maxCaptureTimeout = 15 * time.Second defaultBuildLifetime = 5 * time.Minute defaultReadyLifetime = 15 * time.Minute maxActiveJobs = 8 @@ -568,10 +569,18 @@ func (s *Service) beginCollection(owner, id string, deadline time.Time) { notifier := s.notifier ctx := s.ctx s.mu.Unlock() + captureTimeout := deadline.Sub(s.config.Now().UTC()) + if captureTimeout < 0 { + captureTimeout = 0 + } + if captureTimeout > maxCaptureTimeout { + captureTimeout = maxCaptureTimeout + } if notifier != nil { message, err := ws.NewNotification("system.logs.capture_requested", map[string]any{ "bundle_id": id, "capture_deadline": deadline, - "max_chunk_bytes": 1024 * 1024, "max_browser_profiles": maxBrowserProfiles, + "capture_timeout_ms": captureTimeout.Milliseconds(), + "max_chunk_bytes": 1024 * 1024, "max_browser_profiles": maxBrowserProfiles, }) if err == nil { notifier.SendToIdentity(owner, message) diff --git a/apps/backend/internal/system/logbundle/service_test.go b/apps/backend/internal/system/logbundle/service_test.go index 98e979c06e..7331bdc891 100644 --- a/apps/backend/internal/system/logbundle/service_test.go +++ b/apps/backend/internal/system/logbundle/service_test.go @@ -563,14 +563,17 @@ func TestFrontendCreateNotifiesOnlyThroughIdentityNotifier(t *testing.T) { t.Fatalf("notification = %#v", notifier.message) } var payload struct { - BundleID string `json:"bundle_id"` - MaxChunkBytes int `json:"max_chunk_bytes"` - MaxBrowserProfiles int `json:"max_browser_profiles"` + BundleID string `json:"bundle_id"` + CaptureDeadline time.Time `json:"capture_deadline"` + CaptureTimeoutMS int64 `json:"capture_timeout_ms"` + MaxChunkBytes int `json:"max_chunk_bytes"` + MaxBrowserProfiles int `json:"max_browser_profiles"` } if err := json.Unmarshal(notifier.message.Payload, &payload); err != nil { t.Fatal(err) } - if payload.BundleID != job.ID || payload.MaxChunkBytes != 1024*1024 || + if payload.BundleID != job.ID || !payload.CaptureDeadline.Equal(*job.CaptureDeadline) || + payload.CaptureTimeoutMS != 15_000 || payload.MaxChunkBytes != 1024*1024 || payload.MaxBrowserProfiles != maxBrowserProfiles { t.Fatalf("capture payload = %#v", payload) } diff --git a/apps/web/lib/logger/buffer.ts b/apps/web/lib/logger/buffer.ts index 47f65c8cd2..124e0b3a9f 100644 --- a/apps/web/lib/logger/buffer.ts +++ b/apps/web/lib/logger/buffer.ts @@ -69,9 +69,13 @@ export class RingBuffer { } snapshot(identityScope?: string): LogEntry[] { + return this.snapshotPrepared(identityScope).map(({ entry }) => entry); + } + + snapshotPrepared(identityScope?: string): PreparedLogEntry[] { return this.entries .filter(({ entry }) => identityScope === undefined || entry.identity_scope === identityScope) - .map(({ entry }) => cloneEntry(entry)); + .map(({ entry, bytes }) => ({ entry: cloneEntry(entry), bytes })); } clear(): void { @@ -111,6 +115,10 @@ export function snapshotLogs(identityScope?: string): LogEntry[] { return getLogBuffer().snapshot(identityScope); } +export function snapshotPreparedLogs(identityScope?: string): PreparedLogEntry[] { + return getLogBuffer().snapshotPrepared(identityScope); +} + export function clearLogs(): void { getLogBuffer().clear(); } diff --git a/apps/web/lib/logger/capture.test.ts b/apps/web/lib/logger/capture.test.ts index 6ab8f72dcc..5290aa9f29 100644 --- a/apps/web/lib/logger/capture.test.ts +++ b/apps/web/lib/logger/capture.test.ts @@ -1,18 +1,316 @@ -import { describe, expect, it } from "vitest"; -import { chunkEntries } from "./capture"; -import type { LogEntry } from "./buffer"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { handleBrowserLogCapture } from "./capture"; +import { prepareLogEntry, type LogEntry } from "./buffer"; -describe("frontend log capture chunks", () => { - it("keeps sequential chunks below the requested target", () => { - const entries: LogEntry[] = Array.from({ length: 5 }, (_, index) => ({ - timestamp: String(index), - level: "info", - source: "console", - message: "x".repeat(40), - })); - const chunks = chunkEntries(entries, 160); - expect(chunks.length).toBeGreaterThan(1); - expect(chunks.flat()).toEqual(entries); - expect(chunks.every((chunk) => chunk.length > 0)).toBe(true); +const captureMocks = vi.hoisted(() => ({ + upload: vi.fn(), + begin: vi.fn(), + browserInstallationID: vi.fn(), + browserLogMetadata: vi.fn(), +})); + +vi.mock("@/lib/api/domains/system-api", () => ({ + uploadFrontendBundleChunk: captureMocks.upload, +})); + +vi.mock("./runtime", () => ({ + beginBrowserLogCapture: captureMocks.begin, + browserInstallationID: captureMocks.browserInstallationID, + browserLogMetadata: captureMocks.browserLogMetadata, +})); + +beforeEach(() => { + captureMocks.upload.mockReset().mockResolvedValue(undefined); + captureMocks.begin.mockReset(); + captureMocks.browserInstallationID.mockReset().mockReturnValue("browser"); + captureMocks.browserLogMetadata.mockReset().mockReturnValue({ + storage_mode: "indexeddb", + persistence_failures: 0, + }); +}); + +describe("frontend log capture", () => { + it("uploads each page before it reads the next page", async () => { + const entries: LogEntry[] = [ + { + timestamp: new Date(1).toISOString(), + level: "info", + source: "console", + message: "x".repeat(60 * 1024), + }, + { + timestamp: new Date(2).toISOString(), + level: "info", + source: "console", + message: "y".repeat(60 * 1024), + }, + ]; + const prepared = entries.map(prepareLogEntry); + const events: string[] = []; + let releaseSecondPage: (() => void) | undefined; + const secondPage = new Promise((resolve) => { + releaseSecondPage = resolve; + }); + let readCount = 0; + let resolveFirstUpload: (() => void) | undefined; + const firstUploadStarted = new Promise((resolve) => { + resolveFirstUpload = resolve; + }); + captureMocks.upload.mockImplementation(async (_id, chunk) => { + events.push(`upload-${chunk.chunk_index}`); + if (chunk.chunk_index === 0) resolveFirstUpload?.(); + }); + const readPage = vi.fn(async (maxBytes: number) => { + events.push(`read-${readCount}`); + readCount += 1; + if (readCount === 1) { + expect(maxBytes).toBe(128 * 1024); + return { + entries: prepared.slice(0, 1), + nextCursor: { timestamp_ms: 1, primary_key: 1 }, + done: false, + }; + } + await secondPage; + return { entries: prepared.slice(1), nextCursor: null, done: true }; + }); + captureMocks.begin.mockResolvedValue({ + storageMode: "indexeddb", + flushTimeout: false, + memoryEntries: prepared, + readPage, + }); + + const capture = handleBrowserLogCapture( + { + bundle_id: "bundle", + capture_deadline: new Date(Date.now() + 60_000).toISOString(), + capture_timeout_ms: 15_000, + max_chunk_bytes: 1024 * 1024, + max_browser_profiles: 4, + }, + "identity", + ); + + await firstUploadStarted; + expect(events).toEqual(["read-0", "upload-0"]); + expect(readPage).toHaveBeenCalledTimes(1); + const firstEntries = captureMocks.upload.mock.calls[0]?.[1].entries as LogEntry[]; + const firstBytes = firstEntries.reduce( + (total, entry) => total + new TextEncoder().encode(JSON.stringify(entry)).byteLength, + 0, + ); + expect(firstBytes).toBeLessThanOrEqual(128 * 1024); + + releaseSecondPage?.(); + await capture; + expect(captureMocks.upload).toHaveBeenCalledTimes(2); + expect(readPage.mock.calls[1]?.[0]).toBe(800 * 1024); + expect(events).toEqual(["read-0", "upload-0", "read-1", "upload-1"]); + }); +}); + +describe("frontend log capture deadlines and fallbacks", () => { + it("uses the relative budget despite wall-clock skew and stops later pages", async () => { + vi.useFakeTimers(); + const performanceNow = vi.spyOn(performance, "now"); + performanceNow.mockReturnValue(100); + const first = prepareLogEntry({ + timestamp: new Date(1).toISOString(), + level: "info", + source: "console", + message: "first", + }); + const readPage = vi.fn().mockResolvedValue({ + entries: [first], + nextCursor: { timestamp_ms: 1, primary_key: 1 }, + done: false, + }); + captureMocks.begin.mockResolvedValue({ + storageMode: "indexeddb", + flushTimeout: false, + memoryEntries: [first], + readPage, + }); + captureMocks.upload.mockImplementation(async () => { + performanceNow.mockReturnValue(200); + }); + + try { + const capture = handleBrowserLogCapture( + { + bundle_id: "bundle", + capture_deadline: new Date(0).toISOString(), + capture_timeout_ms: 50, + max_chunk_bytes: 1024 * 1024, + max_browser_profiles: 4, + }, + "identity", + ); + + await capture; + expect(captureMocks.upload).toHaveBeenCalledTimes(1); + expect(readPage).toHaveBeenCalledTimes(1); + } finally { + performanceNow.mockRestore(); + vi.useRealTimers(); + } + }); +}); + +describe("active frontend log capture deadlines", () => { + it("aborts an active upload when the monotonic budget expires", async () => { + vi.useFakeTimers(); + const performanceNow = vi.spyOn(performance, "now"); + performanceNow.mockReturnValue(100); + let signal: AbortSignal | undefined; + const entry = prepareLogEntry({ + timestamp: new Date(1).toISOString(), + level: "info", + source: "console", + message: "active", + }); + captureMocks.begin.mockResolvedValue({ + storageMode: "memory", + flushTimeout: false, + memoryEntries: [entry], + readPage: null, + }); + captureMocks.upload.mockImplementation((_id, _chunk, options) => { + signal = options?.init?.signal; + return new Promise((_resolve, reject) => { + signal?.addEventListener("abort", () => { + reject(new DOMException("aborted", "AbortError")); + }); + }); + }); + + try { + const capture = handleBrowserLogCapture( + { + bundle_id: "bundle", + capture_deadline: new Date(0).toISOString(), + capture_timeout_ms: 50, + max_chunk_bytes: 1024 * 1024, + max_browser_profiles: 4, + }, + "identity", + ); + await Promise.resolve(); + await vi.advanceTimersByTimeAsync(50); + await expect(capture).resolves.toBeUndefined(); + expect(signal?.aborted).toBe(true); + } finally { + performanceNow.mockRestore(); + vi.useRealTimers(); + } + }); +}); + +describe("frontend log capture read fallback", () => { + it("falls back to the receipt memory snapshot when a page read fails", async () => { + const entry = prepareLogEntry({ + timestamp: new Date(1).toISOString(), + level: "info", + source: "console", + message: "fallback", + }); + const readPage = vi.fn().mockRejectedValue(new Error("IndexedDB unavailable")); + captureMocks.begin.mockResolvedValue({ + storageMode: "indexeddb", + flushTimeout: false, + memoryEntries: [entry], + readPage, + }); + + await handleBrowserLogCapture( + { + bundle_id: "bundle", + capture_deadline: new Date(Date.now() + 60_000).toISOString(), + capture_timeout_ms: 15_000, + max_chunk_bytes: 1024 * 1024, + max_browser_profiles: 4, + }, + "identity", + ); + + expect(readPage).toHaveBeenCalledTimes(1); + expect(captureMocks.upload.mock.calls[0]?.[1]).toMatchObject({ + done: true, + storage_mode: "memory", + entries: [entry.entry], + }); + }); + + it("does not switch sources after the first page has uploaded", async () => { + const entry = prepareLogEntry({ + timestamp: new Date(1).toISOString(), + level: "info", + source: "console", + message: "first page", + }); + const readPage = vi + .fn() + .mockResolvedValueOnce({ + entries: [entry], + nextCursor: { timestamp_ms: 1, primary_key: 1 }, + done: false, + }) + .mockRejectedValueOnce(new Error("IndexedDB read failed")); + captureMocks.begin.mockResolvedValue({ + storageMode: "indexeddb", + flushTimeout: false, + memoryEntries: [entry], + readPage, + }); + + await expect( + handleBrowserLogCapture( + { + bundle_id: "bundle", + capture_deadline: new Date(Date.now() + 60_000).toISOString(), + capture_timeout_ms: 15_000, + max_chunk_bytes: 1024 * 1024, + max_browser_profiles: 4, + }, + "identity", + ), + ).rejects.toThrow("IndexedDB read failed"); + + expect(captureMocks.upload).toHaveBeenCalledTimes(1); + }); +}); + +describe("frontend log capture metadata", () => { + it("marks a timed-out persistence flush in final capture metadata", async () => { + const entry = prepareLogEntry({ + timestamp: new Date(1).toISOString(), + level: "info", + source: "console", + message: "memory", + }); + captureMocks.begin.mockResolvedValue({ + storageMode: "memory", + flushTimeout: true, + memoryEntries: [entry], + readPage: null, + }); + + await handleBrowserLogCapture( + { + bundle_id: "bundle", + capture_deadline: new Date(Date.now() + 60_000).toISOString(), + capture_timeout_ms: 15_000, + max_chunk_bytes: 1024 * 1024, + max_browser_profiles: 4, + }, + "identity", + ); + + expect(captureMocks.upload.mock.calls[0]?.[1]).toMatchObject({ + done: true, + storage_mode: "memory", + capture_metadata: { flush_timeout: true, storage_mode: "memory" }, + }); }); }); diff --git a/apps/web/lib/logger/capture.ts b/apps/web/lib/logger/capture.ts index 1ef38a7f26..714eb0ce7e 100644 --- a/apps/web/lib/logger/capture.ts +++ b/apps/web/lib/logger/capture.ts @@ -1,12 +1,21 @@ import { uploadFrontendBundleChunk } from "@/lib/api/domains/system-api"; -import type { LogEntry } from "./buffer"; -import { browserInstallationID, browserLogMetadata, snapshotBrowserLogs } from "./runtime"; +import type { PreparedLogEntry } from "./buffer"; +import type { LogPage, LogPageCursor } from "./indexeddb-store"; +import { + beginBrowserLogCapture, + browserInstallationID, + browserLogMetadata, + type BrowserLogCapture, +} from "./runtime"; +const FIRST_PAGE_BYTES = 128 * 1024; const TARGET_CHUNK_BYTES = 800 * 1024; +const CAPTURE_METADATA_BYTES = 8 * 1024; export type CaptureRequest = { bundle_id: string; capture_deadline: string; + capture_timeout_ms?: number; max_chunk_bytes: number; max_browser_profiles: number; }; @@ -15,45 +24,220 @@ export async function handleBrowserLogCapture( request: CaptureRequest, identityScope: string | null, ): Promise { - if (!identityScope || !request.bundle_id || Date.now() >= Date.parse(request.capture_deadline)) { + const deadline = localDeadline(request); + if (!identityScope || !request.bundle_id || deadline === null || performance.now() >= deadline) { return; } - const streamID = randomID(); - const entries = await snapshotBrowserLogs(identityScope); - const chunks = chunkEntries(entries, Math.min(TARGET_CHUNK_BYTES, request.max_chunk_bytes)); - const storageMode = browserLogMetadata().storage_mode === "memory" ? "memory" : "indexeddb"; - for (let index = 0; index < chunks.length; index++) { - const done = index === chunks.length - 1; - await uploadFrontendBundleChunk(request.bundle_id, { - browser_id: browserInstallationID(), - capture_stream_id: streamID, - chunk_index: index, - done, - storage_mode: storageMode, - capture_metadata: done ? boundedMetadata(browserLogMetadata()) : null, - entries: chunks[index], + const capture = await beginBrowserLogCapture(identityScope); + const pageLimit = effectivePageLimit(request.max_chunk_bytes); + if (pageLimit <= 0 || performance.now() >= deadline) return; + + await uploadCapturePages({ + request, + capture, + browserID: browserInstallationID(), + streamID: randomID(), + pageLimit, + deadline, + }); +} + +type CaptureContext = { + request: CaptureRequest; + capture: BrowserLogCapture; + browserID: string; + streamID: string; + pageLimit: number; + deadline: number; +}; + +async function uploadCapturePages({ + request, + capture, + browserID, + streamID, + pageLimit, + deadline, +}: CaptureContext): Promise { + let source = capture.storageMode; + let indexedCursor: LogPageCursor | null = null; + let memoryOffset = 0; + let firstPage = true; + let uploaded = false; + + for (let chunkIndex = 0; ; chunkIndex += 1) { + if (isExpired(deadline)) return; + const result = await readCapturePage({ + capture, + storageMode: source, + indexedCursor, + memoryOffset, + maxBytes: firstPage ? Math.min(FIRST_PAGE_BYTES, pageLimit) : pageLimit, + uploaded, }); + source = result.storageMode; + if (isExpired(deadline)) return; + + const completed = await uploadCapturePage({ + request, + capture, + page: result.page, + storageMode: source, + browserID, + streamID, + chunkIndex, + deadline, + }); + if (!completed) return; + uploaded = true; + if (result.page.done) return; + indexedCursor = result.page.nextCursor; + memoryOffset = result.page.memoryOffset ?? memoryOffset; + firstPage = false; } } -export function chunkEntries(entries: LogEntry[], maxBytes: number): LogEntry[][] { - const chunks: LogEntry[][] = [[]]; - let currentBytes = 0; - for (const entry of entries) { - const bytes = new TextEncoder().encode(JSON.stringify(entry)).byteLength + 1; - if (chunks.at(-1)!.length > 0 && currentBytes + bytes > maxBytes) { - chunks.push([]); - currentBytes = 0; - } - chunks.at(-1)!.push(entry); - currentBytes += bytes; +type CapturePage = LogPage & { memoryOffset?: number }; + +type PageReadResult = { + page: CapturePage; + storageMode: "indexeddb" | "memory"; +}; + +type PageReadContext = { + capture: BrowserLogCapture; + storageMode: "indexeddb" | "memory"; + indexedCursor: LogPageCursor | null; + memoryOffset: number; + maxBytes: number; + uploaded: boolean; +}; + +async function readCapturePage({ + capture, + storageMode, + indexedCursor, + memoryOffset, + maxBytes, + uploaded, +}: PageReadContext): Promise { + try { + const page = + storageMode === "indexeddb" + ? await capture.readPage!(maxBytes, indexedCursor) + : memoryPage(capture.memoryEntries, memoryOffset, maxBytes); + return { page, storageMode }; + } catch (error) { + if (uploaded) throw error; + return { + page: memoryPage(capture.memoryEntries, 0, maxBytes), + storageMode: "memory", + }; } - return chunks; +} + +type PageUploadContext = { + request: CaptureRequest; + capture: BrowserLogCapture; + page: CapturePage; + storageMode: "indexeddb" | "memory"; + browserID: string; + streamID: string; + chunkIndex: number; + deadline: number; +}; + +async function uploadCapturePage({ + request, + capture, + page, + storageMode, + browserID, + streamID, + chunkIndex, + deadline, +}: PageUploadContext): Promise { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), remainingTime(deadline)); + try { + await uploadFrontendBundleChunk( + request.bundle_id, + { + browser_id: browserID, + capture_stream_id: streamID, + chunk_index: chunkIndex, + done: page.done, + storage_mode: storageMode, + capture_metadata: page.done ? captureMetadata(storageMode, capture.flushTimeout) : null, + entries: page.entries.map(({ entry }) => entry), + }, + { init: { signal: controller.signal } }, + ); + } catch (error) { + if (controller.signal.aborted) return false; + throw error; + } finally { + clearTimeout(timeout); + } + return true; +} + +function isExpired(deadline: number): boolean { + return performance.now() >= deadline; +} + +function remainingTime(deadline: number): number { + return Math.max(0, deadline - performance.now()); +} + +function memoryPage( + entries: PreparedLogEntry[], + offset: number, + maxBytes: number, +): LogPage & { memoryOffset: number } { + const page: PreparedLogEntry[] = []; + let bytes = 0; + let index = offset; + while (index < entries.length) { + const next = entries[index]; + if (page.length > 0 && bytes + next.bytes > maxBytes) break; + page.push(next); + bytes += next.bytes; + index += 1; + } + return { entries: page, nextCursor: null, done: index >= entries.length, memoryOffset: index }; +} + +function effectivePageLimit(maxChunkBytes: number): number { + return Number.isFinite(maxChunkBytes) && maxChunkBytes > 0 + ? Math.min(TARGET_CHUNK_BYTES, maxChunkBytes) + : 0; +} + +function localDeadline(request: CaptureRequest): number | null { + const started = performance.now(); + if (request.capture_timeout_ms !== undefined) { + return Number.isFinite(request.capture_timeout_ms) && request.capture_timeout_ms >= 0 + ? started + request.capture_timeout_ms + : null; + } + const serverDeadline = Date.parse(request.capture_deadline); + const remaining = serverDeadline - Date.now(); + return Number.isFinite(remaining) && remaining >= 0 ? started + remaining : null; +} + +function captureMetadata( + storageMode: "indexeddb" | "memory", + flushTimeout: boolean, +): Record { + const metadata: Record = { ...browserLogMetadata(), storage_mode: storageMode }; + if (flushTimeout) metadata.flush_timeout = true; + return boundedMetadata(metadata); } function boundedMetadata(metadata: Record): Record { const serialized = JSON.stringify(metadata); - if (new TextEncoder().encode(serialized).byteLength <= 8 * 1024) return metadata; + if (new TextEncoder().encode(serialized).byteLength <= CAPTURE_METADATA_BYTES) return metadata; return { storage_mode: metadata.storage_mode, metadata_truncated: true }; } diff --git a/apps/web/lib/logger/indexeddb-store.test.ts b/apps/web/lib/logger/indexeddb-store.test.ts index b9ef20ec90..227a2a331c 100644 --- a/apps/web/lib/logger/indexeddb-store.test.ts +++ b/apps/web/lib/logger/indexeddb-store.test.ts @@ -11,6 +11,10 @@ const RETENTION_KEY = "retention"; const MAX_ENTRIES = 10_000; const MAX_BYTES = 20 * 1024 * 1024; const THREE_DAYS_MS = 3 * 24 * 60 * 60 * 1000; +const IDENTITY_A = "identity-a"; +const IDENTITY_B = "identity-b"; +const BEFORE_RECEIPT_MESSAGE = "before-receipt"; +const RECEIPT_PREFIX_MESSAGE = "receipt-prefix"; type PersistedTestEntry = { id?: number; @@ -99,7 +103,7 @@ describe("IndexedDB log retention planning and schema", () => { await new Promise((resolve) => setTimeout(resolve, 0)); await expect( store.append(prepareEntries([logEntry("a", Date.now(), "retried")])), - ).resolves.toBeUndefined(); + ).resolves.toEqual(expect.any(Array)); await expect(store.snapshot("a")).resolves.toEqual([ expect.objectContaining({ message: "retried" }), ]); @@ -163,13 +167,171 @@ describe("IndexedDB log retention writes", () => { const store = new IndexedDBLogStore(); await store.append( prepareEntries([ - logEntry("identity-a", Date.now(), "a"), - logEntry("identity-b", Date.now() + 1, "b"), + logEntry(IDENTITY_A, Date.now(), "a"), + logEntry(IDENTITY_B, Date.now() + 1, "b"), ]), ); - await expect(store.snapshot("identity-a")).resolves.toHaveLength(1); - await expect(store.snapshot("identity-b")).resolves.toHaveLength(1); + await expect(store.snapshot(IDENTITY_A)).resolves.toHaveLength(1); + await expect(store.snapshot(IDENTITY_B)).resolves.toHaveLength(1); + }); +}); + +describe("IndexedDB log paging", () => { + it("reads identity snapshots through the chronological timestamp index", async () => { + const now = Date.now(); + const store = new IndexedDBLogStore(); + const first = logEntry(IDENTITY_A, now, "first"); + const other = logEntry(IDENTITY_B, now, "other"); + const second = logEntry(IDENTITY_A, now, "second"); + await store.append(prepareEntries([first, other, second])); + + const getAll = vi.spyOn(IDBIndex.prototype, "getAll"); + const openCursor = vi.spyOn(IDBIndex.prototype, "openCursor"); + try { + await expect(store.snapshot(IDENTITY_A)).resolves.toEqual([first, second]); + expect(getAll).not.toHaveBeenCalled(); + expect(openCursor).toHaveBeenCalled(); + } finally { + getAll.mockRestore(); + openCursor.mockRestore(); + } + }); + + it("continues pages by timestamp and primary key without gaps or duplicates", async () => { + const timestamp = Date.now(); + const store = new IndexedDBLogStore(); + const entries = [ + logEntry(IDENTITY_A, timestamp, "first"), + logEntry(IDENTITY_B, timestamp, "other"), + logEntry(IDENTITY_A, timestamp, "second"), + logEntry(IDENTITY_A, timestamp, "third"), + ]; + await store.append(prepareEntries(entries)); + + const messages: string[] = []; + let cursor: import("./indexeddb-store").LogPageCursor | null = null; + let pageCount = 0; + let done = false; + while (!done) { + const page = await store.readPage(IDENTITY_A, 1, cursor); + messages.push(...page.entries.map(({ entry }) => entry.message)); + cursor = page.nextCursor; + done = page.done; + pageCount += 1; + } + + expect(messages).toEqual(["first", "second", "third"]); + expect(pageCount).toBe(3); + }); +}); + +describe("IndexedDB cursor continuation bounds", () => { + it("keeps equal-timestamp cursor visits linear across pages", async () => { + const timestamp = Date.now(); + const store = new IndexedDBLogStore(); + const entries = Array.from({ length: 64 }, (_, index) => + logEntry(IDENTITY_A, timestamp, `entry-${String(index).padStart(2, "0")}`), + ); + await store.append(prepareEntries(entries)); + + const cursorContinue = vi.spyOn(IDBCursor.prototype, "continue"); + const cursorContinuePrimaryKey = vi.spyOn(IDBCursor.prototype, "continuePrimaryKey"); + try { + let cursor: import("./indexeddb-store").LogPageCursor | null = null; + let done = false; + let pageCount = 0; + while (!done) { + const page = await store.readPage(IDENTITY_A, prepareLogEntry(entries[0]).bytes, cursor); + cursor = page.nextCursor; + done = page.done; + pageCount += 1; + } + + expect(pageCount).toBe(entries.length); + expect(cursorContinuePrimaryKey).toHaveBeenCalledTimes(entries.length - 2); + expect(cursorContinue.mock.calls.length).toBeLessThanOrEqual(entries.length * 3); + } finally { + cursorContinue.mockRestore(); + cursorContinuePrimaryKey.mockRestore(); + } + }); +}); + +describe("IndexedDB capture boundaries", () => { + it("excludes records persisted after the capture upper bound", async () => { + const timestamp = Date.now(); + const store = new IndexedDBLogStore(); + const first = logEntry(IDENTITY_A, timestamp, "first"); + const second = logEntry(IDENTITY_A, timestamp, "second"); + await store.append(prepareEntries([first, second])); + + const maxPrimaryKey = await store.beginCaptureBoundary(); + const firstPage = await store.readPage( + IDENTITY_A, + prepareLogEntry(first).bytes, + null, + maxPrimaryKey, + ); + expect(firstPage.entries.map(({ entry }) => entry.message)).toEqual(["first"]); + expect(firstPage.done).toBe(false); + + await store.append(prepareEntries([logEntry(IDENTITY_A, timestamp, "after-receipt")])); + + const secondPage = await store.readPage( + IDENTITY_A, + MAX_BYTES, + firstPage.nextCursor, + maxPrimaryKey, + ); + expect(secondPage.entries.map(({ entry }) => entry.message)).toEqual(["second"]); + expect(secondPage.done).toBe(true); + }); + + it("freezes the boundary before a later tab commits a row", async () => { + const firstStore = new IndexedDBLogStore(); + const secondStore = new IndexedDBLogStore(); + const timestamp = Date.now(); + await firstStore.append( + prepareEntries([logEntry(IDENTITY_A, timestamp, BEFORE_RECEIPT_MESSAGE)]), + ); + await secondStore.readPage(IDENTITY_A, MAX_BYTES); + + const boundaryPromise = firstStore.beginCaptureBoundary(); + const laterAppend = secondStore.append( + prepareEntries([logEntry(IDENTITY_A, timestamp, "after-receipt")]), + ); + const maxPrimaryKey = await boundaryPromise; + await laterAppend; + + const page = await firstStore.readPage(IDENTITY_A, MAX_BYTES, null, maxPrimaryKey); + expect(page.entries.map(({ entry }) => entry.message)).toEqual([BEFORE_RECEIPT_MESSAGE]); + expect(page.done).toBe(true); + }); + + it("excludes another tab write interleaved before the receipt prefix", async () => { + const firstStore = new IndexedDBLogStore(); + const secondStore = new IndexedDBLogStore(); + const timestamp = Date.now(); + await firstStore.append( + prepareEntries([logEntry(IDENTITY_A, timestamp, BEFORE_RECEIPT_MESSAGE)]), + ); + await secondStore.readPage(IDENTITY_A, MAX_BYTES); + + const boundary = await firstStore.beginCaptureBoundary(); + await secondStore.append( + prepareEntries([logEntry(IDENTITY_A, timestamp, "interleaved-write")]), + ); + const prefix = await firstStore.append( + prepareEntries([logEntry(IDENTITY_A, timestamp, RECEIPT_PREFIX_MESSAGE)]), + ); + + const page = await firstStore.readPage(IDENTITY_A, MAX_BYTES, null, boundary, new Set(prefix)); + expect(page.entries.map(({ entry }) => entry.message)).toEqual([ + BEFORE_RECEIPT_MESSAGE, + RECEIPT_PREFIX_MESSAGE, + ]); + expect(page.done).toBe(true); }); }); diff --git a/apps/web/lib/logger/indexeddb-store.ts b/apps/web/lib/logger/indexeddb-store.ts index 5936c4c35f..32875d2d3f 100644 --- a/apps/web/lib/logger/indexeddb-store.ts +++ b/apps/web/lib/logger/indexeddb-store.ts @@ -8,6 +8,10 @@ const RETENTION_METADATA_KEY = "retention"; const THREE_DAYS_MS = 3 * 24 * 60 * 60 * 1000; const MAX_ENTRIES = 10_000; const MAX_BYTES = 20 * 1024 * 1024; +// i18n-exempt: internal IndexedDB diagnostic, never rendered to a user. +const INDEXEDDB_CURSOR_ERROR = "IndexedDB cursor failed"; +// i18n-exempt: internal IndexedDB diagnostic, never rendered to a user. +const INDEXEDDB_BOUNDARY_ERROR = "IndexedDB capture boundary unavailable"; type PersistedEntry = { id?: number; @@ -21,12 +25,21 @@ type RetentionTotals = { count: number; bytes: number }; type RetentionMetadata = RetentionTotals & { key: string }; +export type LogPageCursor = { timestamp_ms: number; primary_key: number }; + +export type LogPage = { + entries: PreparedLogEntry[]; + nextCursor: LogPageCursor | null; + done: boolean; +}; + const EMPTY_TOTALS: RetentionTotals = { count: 0, bytes: 0 }; export class IndexedDBLogStore { private database: Promise | null = null; + private databaseHandle: IDBDatabase | null = null; - async append(entries: readonly PreparedLogEntry[]): Promise { + async append(entries: readonly PreparedLogEntry[]): Promise { const persistedEntries = entries.flatMap(({ entry, bytes }) => { const identity = entry.identity_scope; if (!identity) return []; @@ -41,12 +54,13 @@ export class IndexedDBLogStore { } satisfies PersistedEntry, ]; }); - if (persistedEntries.length === 0) return; + if (persistedEntries.length === 0) return []; const database = await this.open(); const transaction = database.transaction([STORE_NAME, METADATA_STORE_NAME], "readwrite"); const store = transaction.objectStore(STORE_NAME); const metadataStore = transaction.objectStore(METADATA_STORE_NAME); + const primaryKeys: number[] = []; const storedMetadata = await requestResult( metadataStore.get(RETENTION_METADATA_KEY), ); @@ -55,7 +69,11 @@ export class IndexedDBLogStore { : await scanTotals(store); for (const entry of persistedEntries) { - store.add(entry); + const request = store.add(entry); + request.onsuccess = () => { + const primaryKey = Number(request.result); + if (Number.isSafeInteger(primaryKey)) primaryKeys.push(primaryKey); + }; totals.count += 1; totals.bytes += entry.bytes; } @@ -65,21 +83,123 @@ export class IndexedDBLogStore { await deleteOldestUntilWithinBounds(store.index("timestamp_ms"), totals); metadataStore.put({ key: RETENTION_METADATA_KEY, ...totals } satisfies RetentionMetadata); await transactionDone(transaction); + return primaryKeys; } async snapshot(identityScope: string): Promise { + const entries: LogEntry[] = []; + let cursor: LogPageCursor | null = null; + let done = false; + while (!done) { + const page = await this.readPage(identityScope, MAX_BYTES, cursor); + entries.push(...page.entries.map(({ entry }) => entry)); + cursor = page.nextCursor; + done = page.done; + } + return entries; + } + + beginCaptureBoundary(): Promise { + const database = this.databaseHandle; + if (!database) return Promise.reject(new Error(INDEXEDDB_BOUNDARY_ERROR)); + + let transaction: IDBTransaction; + try { + transaction = database.transaction(STORE_NAME, "readwrite"); + } catch (error) { + return Promise.reject(error); + } + const store = transaction.objectStore(STORE_NAME); + let highWatermark: number | null = null; + + return new Promise((resolve, reject) => { + const request = store.openCursor(null, "prev"); + request.onerror = () => reject(request.error ?? new Error(INDEXEDDB_CURSOR_ERROR)); + request.onsuccess = () => { + const cursor = request.result; + highWatermark = cursor ? Number(cursor.primaryKey) : null; + }; + transaction.oncomplete = () => resolve(highWatermark); + transaction.onerror = () => + reject(transaction.error ?? new Error("IndexedDB transaction failed")); + transaction.onabort = () => + reject(transaction.error ?? new Error("IndexedDB transaction aborted")); + }); + } + + async readPage( + identityScope: string, + maxBytes: number, + after: LogPageCursor | null = null, + maxPrimaryKey?: number | null, + additionalPrimaryKeys?: ReadonlySet, + ): Promise { + if (maxPrimaryKey === null && (!additionalPrimaryKeys || additionalPrimaryKeys.size === 0)) { + return { entries: [], nextCursor: null, done: true }; + } + const database = await this.open(); const transaction = database.transaction(STORE_NAME, "readonly"); - const index = transaction.objectStore(STORE_NAME).index("identity_scope"); - const records = await requestResult( - index.getAll(IDBKeyRange.only(identityScope)), - ); - await transactionDone(transaction); + const index = transaction.objectStore(STORE_NAME).index("timestamp_ms"); const cutoff = Date.now() - THREE_DAYS_MS; - return records - .filter((record) => record.timestamp_ms >= cutoff) - .sort((left, right) => left.timestamp_ms - right.timestamp_ms) - .map((record) => record.entry); + const startTimestamp = Math.max(cutoff, after?.timestamp_ms ?? cutoff); + const pageLimit = Number.isFinite(maxBytes) && maxBytes > 0 ? maxBytes : MAX_BYTES; + const page: LogPage = { entries: [], nextCursor: null, done: false }; + let pageBytes = 0; + let afterApplied = after === null || startTimestamp > after.timestamp_ms; + + return new Promise((resolve, reject) => { + const request = index.openCursor(IDBKeyRange.lowerBound(startTimestamp)); + request.onerror = () => reject(request.error ?? new Error(INDEXEDDB_CURSOR_ERROR)); + request.onsuccess = () => { + const cursor = request.result; + if (!cursor) { + page.done = true; + return; + } + const primaryKey = Number(cursor.primaryKey); + if (!afterApplied) { + afterApplied = true; + if ( + after && + Number(cursor.key) === after.timestamp_ms && + primaryKey <= after.primary_key + ) { + if (primaryKey === after.primary_key) { + cursor.continue(); + } else { + cursor.continuePrimaryKey(after.timestamp_ms, after.primary_key + 1); + } + return; + } + } + const record = cursor.value as PersistedEntry; + if ( + maxPrimaryKey !== undefined && + maxPrimaryKey !== null && + primaryKey > maxPrimaryKey && + !additionalPrimaryKeys?.has(primaryKey) + ) { + cursor.continue(); + return; + } + if (record.identity_scope !== identityScope) { + cursor.continue(); + return; + } + if (page.entries.length > 0 && pageBytes + record.bytes > pageLimit) return; + + page.entries.push({ entry: record.entry, bytes: record.bytes }); + pageBytes += record.bytes; + page.nextCursor = { timestamp_ms: record.timestamp_ms, primary_key: primaryKey }; + cursor.continue(); + }; + transaction.oncomplete = () => resolve(page); + transaction.onerror = () => + reject(transaction.error ?? new Error("IndexedDB transaction failed")); + transaction.onabort = () => + reject(transaction.error ?? new Error("IndexedDB transaction aborted")); + }); } async clear(): Promise { @@ -129,7 +249,12 @@ export class IndexedDBLogStore { database.close(); return; } - database.onversionchange = () => database.close(); + database.onversionchange = () => { + if (this.databaseHandle === database) this.databaseHandle = null; + if (this.database === databasePromise) this.database = null; + database.close(); + }; + this.databaseHandle = database; resolve(database); }; request.onerror = () => fail(request.error ?? new Error("IndexedDB open failed")); @@ -180,7 +305,7 @@ function scanTotals(store: IDBObjectStore): Promise { return new Promise((resolve, reject) => { const totals = { ...EMPTY_TOTALS }; const request = store.openCursor(); - request.onerror = () => reject(request.error ?? new Error("IndexedDB cursor failed")); + request.onerror = () => reject(request.error ?? new Error(INDEXEDDB_CURSOR_ERROR)); request.onsuccess = () => { const cursor = request.result; if (!cursor) { @@ -202,7 +327,7 @@ function deleteExpiredPrefix( ): Promise { return new Promise((resolve, reject) => { const request = index.openCursor(IDBKeyRange.upperBound(cutoff, true)); - request.onerror = () => reject(request.error ?? new Error("IndexedDB cursor failed")); + request.onerror = () => reject(request.error ?? new Error(INDEXEDDB_CURSOR_ERROR)); request.onsuccess = () => { const cursor = request.result; if (!cursor) { @@ -222,7 +347,7 @@ function deleteOldestUntilWithinBounds(index: IDBIndex, totals: RetentionTotals) if (totals.count <= MAX_ENTRIES && totals.bytes <= MAX_BYTES) return Promise.resolve(); return new Promise((resolve, reject) => { const request = index.openCursor(); - request.onerror = () => reject(request.error ?? new Error("IndexedDB cursor failed")); + request.onerror = () => reject(request.error ?? new Error(INDEXEDDB_CURSOR_ERROR)); request.onsuccess = () => { const cursor = request.result; if (!cursor || (totals.count <= MAX_ENTRIES && totals.bytes <= MAX_BYTES)) { diff --git a/apps/web/lib/logger/runtime.test.ts b/apps/web/lib/logger/runtime.test.ts index 5aa5212d6f..d5975c638a 100644 --- a/apps/web/lib/logger/runtime.test.ts +++ b/apps/web/lib/logger/runtime.test.ts @@ -1,6 +1,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; -import { encodedBytes, _resetForTesting, MAX_ENTRY_BYTES } from "./buffer"; +import { encodedBytes, _resetForTesting, MAX_ENTRY_BYTES, type LogEntry } from "./buffer"; import { + beginBrowserLogCapture, _resetRuntimeForTesting, browserLogMetadata, snapshotBrowserLogs, @@ -9,15 +10,18 @@ import { const storeMocks = vi.hoisted(() => ({ append: vi.fn(), - snapshot: vi.fn(), + beginCaptureBoundary: vi.fn(), + readPage: vi.fn(), })); const TEST_SOURCE = "test"; const TEST_SCOPE = "default-user"; +const RECEIPT_MESSAGE = "receipt-entry"; vi.mock("./indexeddb-store", () => ({ IndexedDBLogStore: class { append = storeMocks.append; - snapshot = storeMocks.snapshot; + beginCaptureBoundary = storeMocks.beginCaptureBoundary; + readPage = storeMocks.readPage; }, })); @@ -25,7 +29,12 @@ beforeEach(() => { _resetForTesting(); _resetRuntimeForTesting(); storeMocks.append.mockReset(); - storeMocks.snapshot.mockReset().mockResolvedValue([]); + storeMocks.beginCaptureBoundary.mockReset().mockResolvedValue(0); + storeMocks.readPage.mockReset().mockResolvedValue({ + entries: [], + nextCursor: null, + done: true, + }); }); describe("browser logger scheduling", () => { @@ -296,7 +305,8 @@ describe("browser logger runtime", () => { releaseFirstAppend?.(); await snapshot; - expect(storeMocks.append).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(250); + await vi.waitFor(() => expect(storeMocks.append).toHaveBeenCalledTimes(2)); expect(maximumActiveAppends).toBe(1); } finally { vi.runOnlyPendingTimers(); @@ -304,3 +314,151 @@ describe("browser logger runtime", () => { } }); }); + +describe("browser logger capture", () => { + it("establishes the receipt boundary before flushing the receipt prefix", async () => { + const order: string[] = []; + storeMocks.beginCaptureBoundary.mockImplementation(async () => { + order.push("boundary"); + return 17; + }); + storeMocks.append.mockImplementation(async () => { + order.push("append"); + return [18]; + }); + + stageLogEntry({ + timestamp: new Date().toISOString(), + level: "info", + source: TEST_SOURCE, + message: RECEIPT_MESSAGE, + }); + + const capture = await beginBrowserLogCapture(TEST_SCOPE); + await capture.readPage?.(256, null); + + expect(storeMocks.beginCaptureBoundary).toHaveBeenCalledTimes(1); + expect(order).toEqual(["boundary", "append"]); + expect(storeMocks.readPage).toHaveBeenCalledWith(TEST_SCOPE, 256, null, 17, new Set([18])); + }); + + it("keeps the persisted capture boundary fixed for every page", async () => { + storeMocks.beginCaptureBoundary.mockResolvedValue(17); + + const capture = await beginBrowserLogCapture(TEST_SCOPE); + await capture.readPage?.(256, null); + + expect(storeMocks.beginCaptureBoundary).toHaveBeenCalledTimes(1); + expect(storeMocks.readPage).toHaveBeenCalledWith(TEST_SCOPE, 256, null, 17, new Set()); + }); + + it("uses the receipt memory snapshot when the boundary cannot be proven", async () => { + storeMocks.beginCaptureBoundary.mockRejectedValueOnce(new Error("boundary unavailable")); + stageLogEntry({ + timestamp: new Date().toISOString(), + level: "info", + source: TEST_SOURCE, + message: RECEIPT_MESSAGE, + }); + + const capture = await beginBrowserLogCapture(TEST_SCOPE); + + expect(capture.storageMode).toBe("memory"); + expect(capture.flushTimeout).toBe(false); + expect(capture.memoryEntries.map(({ entry }) => entry.message)).toEqual([RECEIPT_MESSAGE]); + expect(browserLogMetadata()).toMatchObject({ storage_mode: "indexeddb" }); + }); +}); + +describe("browser logger capture flush", () => { + it("keeps a capture flush at the receipt watermark", async () => { + vi.useFakeTimers(); + let releaseFirstAppend: (() => void) | undefined; + const firstAppend = new Promise((resolve) => { + releaseFirstAppend = resolve; + }); + let appendCalls = 0; + storeMocks.append.mockImplementation(async () => { + appendCalls += 1; + if (appendCalls === 1) await firstAppend; + }); + + try { + stageLogEntry({ + timestamp: new Date().toISOString(), + level: "info", + source: TEST_SOURCE, + message: "at-receipt", + }); + const snapshot = snapshotBrowserLogs(TEST_SCOPE); + await Promise.resolve(); + expect(storeMocks.append).toHaveBeenCalledTimes(1); + + stageLogEntry({ + timestamp: new Date().toISOString(), + level: "info", + source: TEST_SOURCE, + message: "after-receipt", + }); + releaseFirstAppend?.(); + + await snapshot; + expect(storeMocks.append).toHaveBeenCalledTimes(1); + expect(storeMocks.append.mock.calls[0]?.[0]).toEqual([ + expect.objectContaining({ + entry: expect.objectContaining({ message: "at-receipt" }), + }), + ]); + } finally { + releaseFirstAppend?.(); + vi.runOnlyPendingTimers(); + vi.useRealTimers(); + } + }); + + it("uses the receipt memory snapshot after the one-second flush timeout", async () => { + vi.useFakeTimers(); + let releaseFirstAppend: (() => void) | undefined; + const firstAppend = new Promise((resolve) => { + releaseFirstAppend = resolve; + }); + storeMocks.append.mockImplementation(() => firstAppend); + let result: LogEntry[] | undefined; + + try { + stageLogEntry({ + timestamp: new Date().toISOString(), + level: "info", + source: TEST_SOURCE, + message: RECEIPT_MESSAGE, + }); + const snapshot = snapshotBrowserLogs(TEST_SCOPE).then((entries) => { + result = entries; + }); + await Promise.resolve(); + expect(storeMocks.append).toHaveBeenCalledTimes(1); + + stageLogEntry({ + timestamp: new Date().toISOString(), + level: "info", + source: TEST_SOURCE, + message: "later-entry", + }); + await vi.advanceTimersByTimeAsync(999); + expect(result).toBeUndefined(); + + await vi.advanceTimersByTimeAsync(1); + await Promise.resolve(); + expect(result?.map((entry) => entry.message)).toEqual([RECEIPT_MESSAGE]); + expect(browserLogMetadata()).toMatchObject({ storage_mode: "indexeddb" }); + + releaseFirstAppend?.(); + await snapshot; + await vi.waitFor(() => expect(storeMocks.append).toHaveBeenCalledTimes(2)); + } finally { + releaseFirstAppend?.(); + vi.runOnlyPendingTimers(); + vi.useRealTimers(); + } + }); +}); diff --git a/apps/web/lib/logger/runtime.ts b/apps/web/lib/logger/runtime.ts index 41ff0f5725..aa92f85b1e 100644 --- a/apps/web/lib/logger/runtime.ts +++ b/apps/web/lib/logger/runtime.ts @@ -1,12 +1,12 @@ import { getLogBuffer, prepareLogEntry, - snapshotLogs, + snapshotPreparedLogs, type LogEntry, type LogLevel, type PreparedLogEntry, } from "./buffer"; -import { IndexedDBLogStore } from "./indexeddb-store"; +import { IndexedDBLogStore, type LogPage, type LogPageCursor } from "./indexeddb-store"; const DRAIN_ENTRY_LIMIT = 50; const DRAIN_BYTE_LIMIT = 256 * 1024; @@ -14,9 +14,20 @@ const STAGING_ENTRY_LIMIT = 500; const STAGING_BYTE_LIMIT = 2 * 1024 * 1024; const COLLECTION_WINDOW_MS = 250; const IDLE_DEADLINE_MS = 1_000; +const CAPTURE_FLUSH_TIMEOUT_MS = 1_000; const POST_COLLECTION_IDLE_TIMEOUT_MS = IDLE_DEADLINE_MS - COLLECTION_WINDOW_MS; +const SNAPSHOT_PAGE_BYTES = 20 * 1024 * 1024; -type Staged = PreparedLogEntry; +type Staged = PreparedLogEntry & { sequence: number }; +type DrainResult = { completed: boolean; primaryKeys: number[] }; +type CaptureBoundaryResult = { proven: boolean; maxPrimaryKey: number | null }; + +export type BrowserLogCapture = { + storageMode: "indexeddb" | "memory"; + flushTimeout: boolean; + memoryEntries: PreparedLogEntry[]; + readPage: ((maxBytes: number, after: LogPageCursor | null) => Promise) | null; +}; const store = new IndexedDBLogStore(); let identityScope: string | null = "default-user"; @@ -26,7 +37,8 @@ let collectionTimer: ReturnType | null = null; let idleCallbackHandle: number | null = null; let idleFallbackTimer: ReturnType | null = null; let idleGeneration = 0; -let drainPromise: Promise | null = null; +let drainPromise: Promise | null = null; +let nextSequence = 0; let storageMode: "indexeddb" | "memory" = "indexeddb"; let persistenceFailures = 0; let stagingDropped = 0; @@ -44,21 +56,59 @@ export function stageLogEntry(entry: Omit): void { stagingDropped += 1; return; } - staging.push(prepared); + staging.push({ ...prepared, sequence: ++nextSequence }); stagingBytes += prepared.bytes; scheduleDrain(); } -export async function snapshotBrowserLogs(scope: string): Promise { - await flushStaging(); - if (storageMode === "indexeddb") { - try { - return await store.snapshot(scope); - } catch { - degradePersistence(); - } +export async function beginBrowserLogCapture(scope: string): Promise { + const watermark = nextSequence; + const memoryEntries = snapshotPreparedLogs(scope); + const boundaryPromise = captureBoundary(); + let flushResult: DrainResult | undefined; + let boundaryResult: CaptureBoundaryResult | undefined; + const flushAndBoundary = Promise.all([ + flushStaging(watermark).then((result) => { + flushResult = result; + }), + boundaryPromise.then((result) => { + boundaryResult = result; + }), + ]); + const completed = await waitForCaptureFlush(flushAndBoundary); + const completedFlush = flushResult; + const completedBoundary = boundaryResult; + if ( + !completed || + storageMode === "memory" || + !completedFlush || + !completedFlush.completed || + !completedBoundary || + !completedBoundary.proven + ) { + return { storageMode: "memory", flushTimeout: !completed, memoryEntries, readPage: null }; + } + const receiptPrimaryKeys = new Set(completedFlush.primaryKeys); + return { + storageMode: "indexeddb", + flushTimeout: false, + memoryEntries, + readPage: (maxBytes, after) => + store.readPage(scope, maxBytes, after, completedBoundary.maxPrimaryKey, receiptPrimaryKeys), + }; +} + +export async function snapshotBrowserLogs(scope: string): Promise { + const capture = await beginBrowserLogCapture(scope); + if (capture.storageMode === "memory") { + return capture.memoryEntries.map(({ entry }) => entry); + } + try { + return await readCapturePages(capture); + } catch { + degradePersistence(); + return capture.memoryEntries.map(({ entry }) => entry); } - return snapshotLogs(scope); } export function browserLogMetadata(): Record { @@ -138,27 +188,39 @@ function schedulePostWindowDrain(): void { }, POST_COLLECTION_IDLE_TIMEOUT_MS); } -function requestDrain(): Promise { +function requestDrain(maxSequence = nextSequence): Promise { if (drainPromise) return drainPromise; - if (storageMode === "memory" || staging.length === 0) return Promise.resolve(); - drainPromise = drainLoop().finally(() => { + if (storageMode === "memory" || !hasStagedThrough(maxSequence)) { + return Promise.resolve(emptyDrainResult()); + } + drainPromise = drainLoop(maxSequence).finally(() => { drainPromise = null; if (storageMode === "indexeddb" && staging.length > 0) scheduleDrain(); }); return drainPromise; } -async function drainLoop(): Promise { - while (storageMode === "indexeddb" && staging.length > 0) { - if (!(await drainBatch())) return; +async function drainLoop(maxSequence: number): Promise { + let primaryKeys: number[] = []; + while (storageMode === "indexeddb" && hasStagedThrough(maxSequence)) { + const result = await drainBatch(maxSequence); + primaryKeys = primaryKeys.concat(result.primaryKeys); + if (!result.completed) return { completed: false, primaryKeys }; } + return { completed: true, primaryKeys }; } -async function drainBatch(): Promise { - if (storageMode === "memory" || staging.length === 0) return true; +async function drainBatch(maxSequence: number): Promise { + if (storageMode === "memory" || !hasStagedThrough(maxSequence)) { + return emptyDrainResult(); + } const batch: Staged[] = []; let bytes = 0; - while (staging.length > 0 && batch.length < DRAIN_ENTRY_LIMIT) { + while ( + staging.length > 0 && + staging[0].sequence <= maxSequence && + batch.length < DRAIN_ENTRY_LIMIT + ) { const next = staging[0]; if (batch.length > 0 && bytes + next.bytes > DRAIN_BYTE_LIMIT) break; staging.shift(); @@ -167,24 +229,102 @@ async function drainBatch(): Promise { bytes += next.bytes; } try { - await store.append(batch); + const persistedPrimaryKey = await store.append(batch); + return { + completed: true, + primaryKeys: normalizePrimaryKeys(persistedPrimaryKey), + }; } catch { for (const item of batch.reverse()) { staging.unshift(item); stagingBytes += item.bytes; } degradePersistence(); - return false; + return { completed: false, primaryKeys: [] }; } - return true; } -async function flushStaging(): Promise { +function hasStagedThrough(maxSequence: number): boolean { + return staging.some((item) => item.sequence <= maxSequence); +} + +async function flushStaging(maxSequence = nextSequence): Promise { cancelCollectionTimer(); cancelIdleCallback(); - while (drainPromise || (storageMode === "indexeddb" && staging.length > 0)) { - await requestDrain(); + let result = emptyDrainResult(); + while (drainPromise || (storageMode === "indexeddb" && hasStagedThrough(maxSequence))) { + const drainResult = await requestDrain(maxSequence); + result = mergeDrainResults(result, drainResult); + if (!drainResult.completed) break; } + return result; +} + +function waitForCaptureFlush(flush: Promise): Promise { + return new Promise((resolve) => { + let settled = false; + const timer = setTimeout(() => { + if (settled) return; + settled = true; + resolve(false); + }, CAPTURE_FLUSH_TIMEOUT_MS); + const finish = (completed: boolean) => { + if (settled) return; + settled = true; + clearTimeout(timer); + resolve(completed); + }; + flush.then( + () => finish(true), + () => finish(true), + ); + }); +} + +function captureBoundary(): Promise { + if (storageMode === "memory") return Promise.resolve(unprovenBoundary()); + try { + return store.beginCaptureBoundary().then( + (maxPrimaryKey) => ({ proven: true, maxPrimaryKey }), + () => unprovenBoundary(), + ); + } catch { + return Promise.resolve(unprovenBoundary()); + } +} + +async function readCapturePages(capture: BrowserLogCapture): Promise { + if (!capture.readPage) return capture.memoryEntries.map(({ entry }) => entry); + const entries: LogEntry[] = []; + let cursor: LogPageCursor | null = null; + let done = false; + while (!done) { + const page = await capture.readPage(SNAPSHOT_PAGE_BYTES, cursor); + entries.push(...page.entries.map(({ entry }) => entry)); + cursor = page.nextCursor; + done = page.done; + } + return entries; +} + +function emptyDrainResult(): DrainResult { + return { completed: true, primaryKeys: [] }; +} + +function mergeDrainResults(left: DrainResult, right: DrainResult): DrainResult { + return { + completed: left.completed && right.completed, + primaryKeys: left.primaryKeys.concat(right.primaryKeys), + }; +} + +function normalizePrimaryKeys(value: readonly number[] | undefined): number[] { + if (!Array.isArray(value)) return []; + return value.filter((primaryKey) => Number.isSafeInteger(primaryKey)); +} + +function unprovenBoundary(): CaptureBoundaryResult { + return { proven: false, maxPrimaryKey: null }; } function cancelCollectionTimer(): void { @@ -233,6 +373,7 @@ export function _resetRuntimeForTesting(): void { identityScope = "default-user"; staging = []; stagingBytes = 0; + nextSequence = 0; drainPromise = null; storageMode = "indexeddb"; persistenceFailures = 0; diff --git a/docs/plans/frontend-log-capture-latency/plan.md b/docs/plans/frontend-log-capture-latency/plan.md new file mode 100644 index 0000000000..a1ed2b1719 --- /dev/null +++ b/docs/plans/frontend-log-capture-latency/plan.md @@ -0,0 +1,262 @@ +--- +created: 2026-09-08 +updated: 2026-09-09 +status: blocked +requirements: + - REQ-PLATFORM-DIAGNOSTIC-LOGGING-001 + - REQ-PLATFORM-BROWSER-CONSOLE-RETENTION-001 +system_design: + - ../../specs/platform/system-design/diagnostic-logging-01.md + - ../../specs/platform/system-design/diagnostic-logging-02.md + - ../../specs/platform/system-design/browser-console-retention.md +legacy_specs: [] +--- + +# Implementation Plan: Reduce Frontend Log Capture Latency + +## Overview + +Review the available capture evidence before the implementation starts. The +required active-profile evidence is not available, so Task 01 remains blocked. +Task 02 is implemented as a bounded mitigation based on the available trace +and isolated benchmark. It does not claim root-cause confirmation. + +It then uploads one chronological page before it reads the next page. The +backend supplies a relative duration for a monotonic local deadline. + +## Scope + +### In scope + +- Phase timing for the frontend capture and backend upload handler. +- A fixed staging watermark and a one-second persistence-flush limit. +- Receipt-time memory fallback that does not stop normal persistence. +- Chronological paging through the existing IndexedDB timestamp index. +- A 128 KiB first page and the existing 800 KiB target for later pages. +- A relative capture duration and an abort signal for the active upload. +- Regression tests for ordering, bounds, clock skew, timeout, and fallback. + +### Out of scope + +- An IndexedDB schema change or a new write index. +- Changes to console levels, producer frequency, retention, or staging limits. +- Continuous frontend telemetry or background upload retries. +- A longer backend collection window or a larger request-body limit. +- A backend lock refactor without timing evidence from Task 01. +- User-facing UI or copy changes. + +## Evidence and attribution gap + +The backend sent both observed capture notifications immediately. The first +frontend request reached the server 11.6 seconds later. The second request +reached the server 14.8 seconds later. + +The server then spent 24.8 seconds and 16.2 seconds in those requests. The HTTP +timer includes request-body decoding. The handler decodes the body before it +enters the diagnostic bundle service lock. + +An isolated Chromium benchmark used 10,000 entries with approximately 20 MiB +of data. The current `getAll`, sort, and chunk preparation took 1.08 seconds on +the first read. Four later reads took 0.20 to 0.54 seconds. + +The eager snapshot is a bounded cost, but this benchmark did not reproduce the +reported delay. The current evidence does not separate these possible costs: + +- The unbounded staging flush. +- IndexedDB contention in the active browser profile. +- Request-body serialization and transfer. +- Backend lock wait and file writes. + +Task 01 remains blocked because the required phase tables, 90 percent timing +accounting, request sizes and counts, and backend gate results are missing. The +phrase "confirmed root cause" does not apply until this evidence is complete. + +## Requirement conformance + +This package amends the existing platform requirements. It does not create a +new capability owner. + +- Console calls remain synchronous and do not wait for persistence or network + work. +- A capture includes only entries that exist at notification receipt. +- Later console entries do not extend the capture flush. +- Identity isolation, chronological order, and all retention limits remain. +- The backend remains authoritative for job expiry. +- The relative duration removes dependence on the browser wall clock. +- A partial capture does not retry or write through the console interceptor. + +No new ADR is necessary. The privacy, ownership, persistence, and transport +boundaries remain the same. The added relative duration is backward compatible. + +## Technical approach + +### Phase measurement + +Task 01 adds temporary, non-console timing around each capture phase. Browser +timing uses `performance.now()`. Backend timing uses the existing structured +logger. The task removes this temporary code after it records the results. + +The frontend phase table includes these intervals: + +1. Notification receipt to flush start. +2. The current persistence flush. +3. The IndexedDB request. +4. Snapshot filtering and ordering. +5. Chunk preparation and request-body serialization. +6. Each upload request. + +The backend phase table includes request-body decoding, stream claim, service +lock wait, validation, file writes, and response completion. Task 01 compares +an active profile with an isolated maximum-size profile. + +The frontend intervals do not overlap. Their total covers notification receipt +through the fetch response. The backend intervals subdivide the server request +and do not add to the frontend total. + +The backend gate is 250 milliseconds and 10 percent of request time for lock +wait and file writes. Task 01 has not applied this gate because its backend +phase table is missing. No backend lock work order can be ruled out. + +### Fixed capture watermark + +In `apps/web/lib/logger/runtime.ts`, assign a local sequence number to each +staged entry. Each drain request owns a fixed maximum sequence number. Entries +that arrive later start a later drain. + +At notification receipt, save the current sequence number and a bounded memory +snapshot. Start a serialized IndexedDB boundary transaction before the +receipt-prefix drain. If the boundary cannot be proven, use the memory +snapshot. Join the active drain and persist only through that sequence number. +Wait at most one second. If the wait expires, use the saved memory snapshot and +let persistence continue. Report memory storage for this capture and include +`flush_timeout: true` in its final metadata. + +### Existing-index snapshot pages + +In `apps/web/lib/logger/indexeddb-store.ts`, keep database version 2. Start the +receipt boundary on the already-open database connection. Pass its upper key +and the exact keys returned by receipt-prefix append batches to every page. +Read the existing `timestamp_ms` index with a continuation pair of timestamp +and primary key. Use `continuePrimaryKey()` for equal-timestamp continuation. +Filter the requested identity while the cursor scans the globally bounded +store. + +Return one page and close its readonly transaction before the uploader asks for +the next page. Use stored prepared-byte counts for page limits. Do not use +`getAll` or sort the complete identity partition. + +Start the fixed boundary transaction at receipt, before the prefix flush. Pass +the combined boundary key to every page read so entries written after receipt +cannot enter the capture between page transactions. Use receipt memory when the +boundary transaction cannot be proven. + +### Upload budget and page sizes + +In `apps/backend/internal/system/logbundle/service.go`, add +`capture_timeout_ms` to `system.logs.capture_requested`. Calculate it from the +backend clock when the notification is created. Keep `capture_deadline` for +backend authority and compatibility. + +In `apps/web/lib/logger/capture.ts`, use `performance.now()` to create the local +deadline. Use the absolute deadline only when an older backend omits the +duration. Check the local deadline before each page and upload. + +Use 128 KiB of entry data for the first page. Use the existing 800 KiB target +for later pages. This change adds at most one request to a maximum-size capture. +Use an `AbortController` to cancel the active request when the local budget +expires. + +If no production caller remains, remove `chunkEntries`. Replace its direct test +with tests for the streaming capture behavior. + +## Tests + +| Acceptance criterion | Test evidence | +| --- | --- | +| `AC-PLATFORM-DIAGNOSTIC-LOGGING-001.10` | `capture.test.ts` holds the second page and proves that the first upload starts first. | +| `AC-PLATFORM-DIAGNOSTIC-LOGGING-001.11` | Backend notification tests cover both deadline fields. Frontend fake-time tests prove monotonic behavior with wall-clock skew. | +| `AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.1` | Existing store tests retain the three-day, 10,000-entry, and 20 MiB limits. | +| `AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.3` | Existing concurrent-store tests retain shared transaction totals. | +| `AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.5` | Runtime tests prove that persistence uses one active drain. | +| `AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.6` | Runtime tests prove that capture fallback does not change persistence mode or erase staging. | +| `AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.9` | Runtime tests freeze a sequence watermark and add later entries while the drain is held. | +| `AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.10` | Store tests cover timestamp ties, identity filtering, continuation, page bounds, and bounded cursor visits. | +| `AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.11` | Fake-time runtime tests prove the one-second memory fallback and continued persistence. | +| `AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.12` | Capture tests prove that budget expiry stops new pages and cancels an active upload. | + +## Work orders + +- [ ] [Task 01: Measure Capture Phases](task-01-measure-capture-phases.md) +- [ ] [Task 02: Bound and Stream Capture](task-02-bound-and-stream-capture.md) + +Task 02 has an implementation and focused verification, but its work order +remains blocked by the missing Task 01 evidence. The active-profile phase table +could not be reproduced because this workspace had no attached browser session. + +## Verification results + +- `pnpm install --frozen-lockfile` completed from `apps`. +- The focused frontend logger suite passed: 4 files and 45 tests. +- `pnpm run typecheck` passed from `apps/web`. +- The log-bundle package tests passed: 23 tests. +- The log-bundle package race tests passed: 23 tests. +- The focused frontend ESLint and Prettier checks passed. +- `gofmt -l` reported no changed Go files, and `git diff --check` passed. +- The repository-wide backend test target was run twice. Both runs failed in + unrelated process-probe, configuration-discovery, launcher, and Office + migration tests. The first run also used the workspace home configuration. +- The required active-profile phase tables, 90 percent accounting, request + sizes and counts, and backend 250 millisecond / 10 percent gate results + remain uncollected. + +## Execution TODO + +- [ ] Collect two active-profile captures and the isolated-profile phase tables. +- [ ] Account for at least 90 percent of frontend and backend timing. +- [ ] Apply the backend 250 millisecond and 10 percent gate. +- [x] Implement Task 02 bounded, paged capture with regression tests. +- [x] Run the exact task-defined checks and update work-order/plan status. +- [ ] Commit the current remediation with hooks and push the feature branch. +- [ ] Run PR fixup for the current remediation head. + +## Previous PR Fixup Result (superseded) + +- PR #3540: https://github.com/kdlbs/kandev/pull/3540 +- Fixup commit: `105549d3f79f3527719275669b5f0c4f178e12d4` +- Final exact-head CI: 54 passed, 0 failed, 0 pending; snapshot complete. +- Final review state: 0 unresolved threads, 0 hidden unresolved threads, and 0 actionable issue comments. All six original threads were explicitly resolved. +- Final mergeability: `MERGEABLE / CLEAN`. The `main` base advanced during CI; local `git merge-tree --write-tree` validation completed without conflicts. + +## Current remediation + +- Replaced the post-flush high-watermark read with a receipt-started serialized + boundary and receipt-prefix primary-key tracking. +- Replaced equal-timestamp rescans with `continuePrimaryKey()` and added a + cursor-visit bound regression. +- Reopened Task 01 and this plan. The missing phase evidence remains an + explicit blocker, and Task 02 is a bounded mitigation rather than a root + cause claim. +- Local remediation tests passed. Remote CI and review checks for the new head + are pending. + +## Risks + +- A timestamp cursor can skip or duplicate entries when timestamps are equal. +- An IndexedDB boundary can be unavailable when its connection is not open. +- A slow active write can continue after capture selects its memory fallback. +- A delayed notification can leave less server time than the relative client + budget. The backend can reject the late request. +- A 128 KiB first request can still exceed the budget on a very slow link. +- Request-body transfer can remain the largest cost after frontend paging. + +## Mobile parity + +This change does not alter layout, navigation, touch behavior, scrolling, or +viewport-specific controls. Desktop and mobile browsers use the same logger +runtime, capture protocol, and tests. Focused unit and protocol tests are +sufficient. No new mobile Playwright test is necessary. + +## Documentation impact + +No public documentation changes are necessary. The visible bundle workflow is +unchanged. The platform requirements and designs own the changed contract. diff --git a/docs/plans/frontend-log-capture-latency/task-01-measure-capture-phases.md b/docs/plans/frontend-log-capture-latency/task-01-measure-capture-phases.md new file mode 100644 index 0000000000..3727459ad8 --- /dev/null +++ b/docs/plans/frontend-log-capture-latency/task-01-measure-capture-phases.md @@ -0,0 +1,134 @@ +--- +id: "01-measure-capture-phases" +title: "Measure capture phases" +status: blocked +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-PLATFORM-DIAGNOSTIC-LOGGING-001 + - REQ-PLATFORM-BROWSER-CONSOLE-RETENTION-001 +acceptance_criteria: + - AC-PLATFORM-DIAGNOSTIC-LOGGING-001.10 + - AC-PLATFORM-DIAGNOSTIC-LOGGING-001.11 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.9 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.10 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.11 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.12 +system_design: + - ../../specs/platform/system-design/diagnostic-logging-01.md + - ../../specs/platform/system-design/diagnostic-logging-02.md + - ../../specs/platform/system-design/browser-console-retention.md +--- + +# Task 01: Measure Capture Phases + +## Summary + +Measure the frontend and backend phases that make one capture exceed 30 +seconds. This work order remains blocked until the required captures and phase +tables identify the dominant phases before the capture implementation changes. + +## In scope + +- Add temporary `performance.now()` timing around frontend capture phases. +- Do not emit timing through an intercepted console method. +- Add temporary structured backend timing around body decoding and service work. +- Measure two captures from the reported active browser profile. +- Measure two captures from an isolated 10,000-entry, approximately 20 MiB + profile. +- Record request size, entry count, and each phase duration in `## Results`. +- Compare frontend and backend timing with the HTTP middleware duration. +- Before this task is complete, remove all temporary instrumentation. +- If the evidence changes the implementation boundary, update Task 02. + +## Out of scope + +- A production behavior change. +- A permanent telemetry endpoint or browser global. +- Changes to page size, retention, IndexedDB schema, or backend locking. + +## Acceptance + +1. The frontend phases account for at least 90 percent of + notification-to-response time. The backend phases separately account for at + least 90 percent of server request time. +2. The results identify the dominant pre-request and in-request phases. +3. The task removes temporary code and leaves only its recorded results and any + necessary plan corrections. + +## Verification + +```bash +(cd apps && pnpm install --frozen-lockfile) +(cd apps/web && pnpm run typecheck) +make -C apps/backend test +git diff --check +``` + +## Files likely touched + +- `apps/web/lib/logger/capture.ts` (temporary instrumentation only) +- `apps/web/lib/logger/runtime.ts` (temporary instrumentation only) +- `apps/web/lib/logger/indexeddb-store.ts` (temporary instrumentation only) +- `apps/backend/internal/system/logbundle/handler.go` (temporary instrumentation only) +- `apps/backend/internal/system/logbundle/service.go` (temporary instrumentation only) +- `docs/plans/frontend-log-capture-latency/plan.md` +- `docs/plans/frontend-log-capture-latency/task-01-measure-capture-phases.md` +- `docs/plans/frontend-log-capture-latency/task-02-bound-and-stream-capture.md` + +## Dependencies + +None. + +## Risks + +- Console timing can recurse into the logger. Use the Performance API instead. +- A synthetic profile can differ from the active browser profile. +- Temporary instrumentation can change timing. Keep each marker constant-time. + +## Parallelism + +`sequential` + +## Inputs + +- The two observed capture timelines in `plan.md`. +- HTTP middleware timing in `apps/backend/internal/common/httpmw/logging.go`. +- The frontend capture, runtime, and IndexedDB code paths. +- The backend bundle upload handler and service. + +## Results + +- The available active-profile trace recorded 11.6 seconds from the first + notification to the first frontend request and 14.8 seconds to the second. + The corresponding HTTP request durations were 24.8 seconds and 16.2 seconds. +- The isolated Chromium benchmark used 10,000 entries and about 20 MiB of + data. The old eager read, sort, and chunk preparation took 1.08 seconds on + the first read and 0.20 to 0.54 seconds on later reads. +- The required two active-profile captures were not collected. The available + debug backend had no attached Chromium client. +- The required two isolated-profile captures were not collected. The old + Chromium benchmark is context only and does not provide the required phase + markers. +- The frontend phase table is missing notification-to-flush, persistence, + IndexedDB, filtering, serialization, and upload durations. +- The backend phase table is missing body decoding, stream claim, lock wait, + validation, file write, and response completion durations. +- The 90 percent accounting cannot be calculated. Request sizes and request + counts for both profiles are also missing. +- The 250 millisecond and 10 percent backend gate was not applied. These + results do not prove or rule out a backend lock bottleneck. +- Task 02 is a bounded mitigation based on the available trace and isolated + benchmark. It does not claim that Task 01 identified the root cause. +- The temporary measurement code was not retained. The task leaves only these + results and the bounded implementation in Task 02. +- Related verification passed: web typecheck, log-bundle tests, focused logger + tests, focused ESLint, `gofmt -l`, and `git diff --check`. + +## Blocker + +Run the temporary measurement procedure in an active browser profile and in an +isolated 10,000-entry profile. Record two captures for each profile, all phase +durations, request sizes and counts, 90 percent accounting, and the backend +250 millisecond / 10 percent gate result. Then update this task and Task 02. diff --git a/docs/plans/frontend-log-capture-latency/task-02-bound-and-stream-capture.md b/docs/plans/frontend-log-capture-latency/task-02-bound-and-stream-capture.md new file mode 100644 index 0000000000..dcb6a52c97 --- /dev/null +++ b/docs/plans/frontend-log-capture-latency/task-02-bound-and-stream-capture.md @@ -0,0 +1,174 @@ +--- +id: "02-bound-and-stream-capture" +title: "Bound and stream capture" +status: blocked +wave: 2 +depends_on: + - "01-measure-capture-phases" +plan: "plan.md" +requirements: + - REQ-PLATFORM-DIAGNOSTIC-LOGGING-001 + - REQ-PLATFORM-BROWSER-CONSOLE-RETENTION-001 +acceptance_criteria: + - AC-PLATFORM-DIAGNOSTIC-LOGGING-001.10 + - AC-PLATFORM-DIAGNOSTIC-LOGGING-001.11 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.1 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.3 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.5 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.6 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.9 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.10 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.11 + - AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.12 +system_design: + - ../../specs/platform/system-design/diagnostic-logging-01.md + - ../../specs/platform/system-design/diagnostic-logging-02.md + - ../../specs/platform/system-design/browser-console-retention.md +--- + +# Task 02: Bound and Stream Capture + +## Summary + +Bound the pre-capture persistence wait and stream chronological pages through +the existing IndexedDB schema. Use a monotonic relative budget and a small first +request so a slow link can return partial evidence. + +## Failing regression first + +Add these regressions before production changes: + +1. In `runtime.test.ts`, hold one append and add later entries. Prove that the + capture watermark does not move and that the one-second timeout selects the + receipt-time memory snapshot. +2. In `indexeddb-store.test.ts`, seed two identities and equal timestamps. Prove + exact chronological continuation without `getAll`, gaps, or duplicates. +3. In `capture.test.ts`, hold the second page. Prove that upload zero starts + first and contains no more than 128 KiB of entry data. +4. In `capture.test.ts`, skew `Date.now()` and control `performance.now()`. + Prove that the relative duration controls page admission and aborts an active + upload. +5. In `service_test.go`, inspect the capture notification. Prove that it + contains the absolute deadline and a nonnegative duration of at most 15,000 + milliseconds. + +Run these tests and record their expected failures before implementation. + +## In scope + +- Add a monotonic sequence number to frontend staging entries. +- Make each drain request stop at its fixed sequence watermark. +- Save a bounded memory snapshot when capture starts. +- Wait one second for persistence through the capture watermark. +- If that wait expires, use the saved memory snapshot. +- Keep the active persistence drain and IndexedDB storage mode unchanged. +- Report memory storage and `flush_timeout: true` for the fallback capture. +- Page the existing `timestamp_ms` index with timestamp and primary-key + continuation. +- Establish a persisted primary-key boundary at receipt before the capture + flush, and retain exact primary keys for the receipt-prefix rows. +- Filter the requested identity without a complete-partition sort. +- Reuse prepared byte counts for page bounds. +- Add `capture_timeout_ms` to the backend WS notification. +- Use a monotonic local deadline and cancel an active upload at expiry. +- Use 128 KiB for the first page and 800 KiB for later pages. +- Remove `chunkEntries` and replace its direct unit test. +- Preserve empty final chunks, final metadata, and sequential chunk indexes. + +## Out of scope + +- IndexedDB version 3 or another write index. +- A change to the process-wide backend service lock. +- More retention, staging, profile, request, or collection capacity. +- Continuous upload, retry, new UI, or new user-facing copy. + +## Acceptance + +1. New entries cannot extend capture flush beyond its receipt watermark. A slow + drain selects memory evidence within one second and continues persistence. +2. IndexedDB pages preserve identity and chronological order without a schema + upgrade, a full `getAll`, gaps, or duplicates. +3. The first page uploads before later enumeration. The monotonic budget stops + new work and cancels an active upload. + +## Verification + +```bash +(cd apps && pnpm install --frozen-lockfile) +(cd apps && pnpm --filter @kandev/web exec vitest run lib/logger/buffer.test.ts lib/logger/indexeddb-store.test.ts lib/logger/runtime.test.ts lib/logger/capture.test.ts) +(cd apps/web && pnpm exec eslint lib/logger/buffer.ts lib/logger/buffer.test.ts lib/logger/indexeddb-store.ts lib/logger/indexeddb-store.test.ts lib/logger/runtime.ts lib/logger/runtime.test.ts lib/logger/capture.ts lib/logger/capture.test.ts) +(cd apps/web && pnpm run typecheck) +make -C apps/backend test +make -C apps/backend lint +``` + +## Files likely touched + +- `apps/web/lib/logger/buffer.ts` +- `apps/web/lib/logger/buffer.test.ts` +- `apps/web/lib/logger/indexeddb-store.ts` +- `apps/web/lib/logger/indexeddb-store.test.ts` +- `apps/web/lib/logger/runtime.ts` +- `apps/web/lib/logger/runtime.test.ts` +- `apps/web/lib/logger/capture.ts` +- `apps/web/lib/logger/capture.test.ts` +- `apps/backend/internal/system/logbundle/service.go` +- `apps/backend/internal/system/logbundle/service_test.go` + +## Dependencies + +- Task 01 must record the complete phase table. +- Task 01 must show that backend locking does not require a separate repair. + +## Dependency revision + +Task 01 did not collect the required evidence, so this work order cannot claim +that the implementation addresses the confirmed root cause. The bounded +capture implementation remains in the branch as a mitigation based on the +available trace and isolated benchmark. Task 02 remains blocked until Task 01 +records the phase tables and applies the backend gate. + +## Risks + +- Timestamp and primary-key continuation can skip equal-timestamp entries. +- A read transaction can scan entries for other identities before it fills a + page. The global retention cap bounds this scan. +- The memory snapshot can contain less history than IndexedDB. +- The local budget can extend past backend expiry after transport delay. +- Abort support must pass through the shared API client without changing other + requests. + +## Parallelism + +`sequential` + +## Inputs + +- Task 01 results and any resulting plan correction. +- Platform diagnostic logging requirements and designs. +- Browser console retention requirements and design. +- Existing `fetchJson` abort-signal support. + +## Results + +- RED: the new tests failed before implementation. The eager capture prepared + an oversized first request, IndexedDB used `getAll`, the drain included later + entries, and the notification omitted the relative timeout. +- GREEN: capture now freezes the staging watermark, waits one second, and uses + a receipt-time memory snapshot when the wait expires. Persistence continues. +- GREEN: the store starts a serialized receipt boundary before the prefix flush, + passes the boundary key and exact prefix append keys to each page, and reads + the timestamp index with `continuePrimaryKey()` for equal-timestamp + continuation. It preserves identity filtering, equal-timestamp order, and + global retention bounds while excluding writes that arrive between page + transactions. +- GREEN: capture uploads a 128 KiB first page, waits for each upload before the + next read, uses 800 KiB later pages, and aborts at the monotonic deadline. +- GREEN: the backend notification includes the absolute deadline and a capped + relative duration. +- Verification passed: 4 frontend files and 45 focused tests, focused ESLint, + web typecheck, 23 log-bundle tests, 23 log-bundle race tests, `gofmt -l`, and + `git diff --check`. +- The repository-wide backend test target was run twice. Both runs failed in + unrelated process-probe, configuration-discovery, launcher, and Office + migration tests. The changed log-bundle package passed in both runs. diff --git a/docs/specs/platform/requirements/browser-console-retention.md b/docs/specs/platform/requirements/browser-console-retention.md index e5f16f3ff4..d29661fa58 100644 --- a/docs/specs/platform/requirements/browser-console-retention.md +++ b/docs/specs/platform/requirements/browser-console-retention.md @@ -2,7 +2,7 @@ status: active system: platform created: 2026-08-23 -updated: 2026-08-27 +updated: 2026-09-09 owners: - kandev --- @@ -55,13 +55,31 @@ retained history after every log batch. 250 ms collection window shall use the fewest committed batches allowed by the 50-entry and 256 KiB batch limits. Browser idle time shall not shorten that collection window. -- **AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.9:** An explicit diagnostic - snapshot shall bypass an outstanding collection window, flush all staged - entries, and wait for the existing serialized drain. +- **AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.9:** When an explicit diagnostic + snapshot starts, it shall bypass the collection window and freeze the current + staged entries as its capture watermark. Entries staged later shall not delay + that capture. +- **AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.10:** An explicit diagnostic + capture shall enumerate the requested identity partition in chronological + batches bounded by the upload target. Each batch shall become available + before the capture enumerates or serializes the remainder of the partition. + The frontend shall establish the persisted upper boundary at notification + receipt before it flushes the receipt prefix. Entries committed after + notification receipt shall not appear in any batch. If the frontend cannot + prove this boundary, it shall use the bounded memory snapshot for the + capture. +- **AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.11:** A capture shall wait no more + than one second for the serialized drain to persist its capture watermark. + When this wait expires, the capture shall use its bounded memory snapshot and + shall not stop normal persistence. +- **AC-PLATFORM-BROWSER-CONSOLE-RETENTION-001.12:** A browser wall-clock offset + shall not change the frontend capture budget. When this budget expires, the + frontend shall stop new snapshot reads and uploads. It shall cancel an active + upload. ## Compatibility and persistence -- The existing database name and retained entries survive the schema upgrade. +- The existing database name and schema version remain unchanged. - Entries remain partitioned by the authenticated Kandev identity. - Each entry remains limited to 64 KiB. - A diagnostic snapshot returns only the requested identity partition. diff --git a/docs/specs/platform/requirements/diagnostic-logging.md b/docs/specs/platform/requirements/diagnostic-logging.md index 151b1959d4..d1a59cfe7c 100644 --- a/docs/specs/platform/requirements/diagnostic-logging.md +++ b/docs/specs/platform/requirements/diagnostic-logging.md @@ -2,7 +2,7 @@ status: active system: platform created: 2026-07-30 -updated: 2026-08-27 +updated: 2026-09-09 owners: - tbd --- @@ -33,6 +33,13 @@ Users can see frontend failures that leave no backend evidence, and support cann that scans a growing application collection shall compute and emit at most one latest-state sample per 250 ms while that collection changes continuously. +- **AC-PLATFORM-DIAGNOSTIC-LOGGING-001.10:** A connected frontend shall begin + uploading retained browser evidence without first materializing the complete + retained history. +- **AC-PLATFORM-DIAGNOSTIC-LOGGING-001.11:** A frontend capture notification + shall include the absolute server deadline and a remaining duration of no more + than 15 seconds. A browser wall-clock offset shall not change this duration. + The backend shall remain authoritative for job expiry. ## System design diff --git a/docs/specs/platform/system-design/browser-console-retention.md b/docs/specs/platform/system-design/browser-console-retention.md index d7a50f6bc4..c5e602c86d 100644 --- a/docs/specs/platform/system-design/browser-console-retention.md +++ b/docs/specs/platform/system-design/browser-console-retention.md @@ -3,31 +3,33 @@ status: current system: platform requirements: - REQ-PLATFORM-BROWSER-CONSOLE-RETENTION-001 + - REQ-PLATFORM-DIAGNOSTIC-LOGGING-001 --- # Browser Console Retention System Design ## Purpose and boundaries -This design makes browser-log retention incremental. It preserves the existing -diagnostic limits, identity partitions, capture protocol, and memory fallback. -It changes IndexedDB bookkeeping and the per-tab drain coordinator. +This design makes browser-log retention and explicit bundle snapshots +incremental. It preserves the existing diagnostic limits, identity partitions, +and memory fallback. The capture notification adds a relative time budget. The +IndexedDB schema and normal write transaction do not change. ## Requirement mapping | Requirement | Design section | | -------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `REQ-PLATFORM-BROWSER-CONSOLE-RETENTION-001` | [IndexedDB model](#indexeddb-model), [Write transaction](#write-transaction), [Per-tab drain ownership](#per-tab-drain-ownership), [Migration and recovery](#migration-and-recovery) | +| `REQ-PLATFORM-DIAGNOSTIC-LOGGING-001` | [Incremental capture reads](#incremental-capture-reads) | ## IndexedDB model `apps/web/lib/logger/indexeddb-store.ts` keeps the database name `kandev-diagnostic-logs-v1` so existing profiles retain their history. The -IndexedDB schema version increases from 1 to 2. +schema version remains 2. -The existing `entries` store and its `identity_scope` and `timestamp_ms` -indexes remain. A new metadata object store owns one retention record with -these fields: +The `entries` store keeps its `identity_scope` and `timestamp_ms` indexes. The +metadata object store owns one retention record with these fields: ```text key: "retention" @@ -36,7 +38,7 @@ bytes: number ``` The totals cover all identity partitions. They use each entry's stored `bytes` -field, which is the existing serialized-size measure. +field. This field contains the serialized size of the entry. ## Write transaction @@ -61,17 +63,84 @@ evict. `clear()` uses both stores in one transaction and writes zero totals. ## Per-tab drain ownership -`apps/web/lib/logger/runtime.ts` owns one in-flight drain promise. Idle and -timeout callbacks request work from the same drain loop. They do not start a -second `store.append()` while the first call is pending. +`apps/web/lib/logger/runtime.ts` owns one active drain promise. Idle and timeout +callbacks request work from the same drain loop. They do not start a second +`store.append()` while the first call is pending. -The loop removes one bounded batch, waits for its append transaction, and then -continues if staging still contains entries. `flushStaging()` awaits that same -loop before a diagnostic snapshot. It does not create a parallel writer. +Each staged entry receives a local increasing sequence number. Each drain +request records the newest sequence number that it owns. The drain stops after +it processes that fixed prefix. Entries that arrive later start a new drain. + +At capture receipt, the runtime records the current sequence number. It also +takes a bounded prepared-entry snapshot from the memory buffer. The capture +then joins the active drain and requests persistence through the recorded +sequence number. Entries that arrive after receipt do not extend this wait. + +The capture waits at most one second for this fixed drain prefix. If the wait +expires, the capture uses the memory snapshot that it took at receipt. The +active persistence drain continues. This fallback does not change the normal +storage mode or erase staged entries. The upload reports `storage_mode: memory` +and records `flush_timeout: true` in its final capture metadata. The console interception path remains synchronous and bounded. It only adds a reference-free entry to staging and schedules the loop. +## Incremental capture reads + +At receipt, the runtime starts a `readwrite` boundary transaction on an already +open IndexedDB connection. The transaction is serialized with append +transactions and records the highest committed object-store primary key before +the receipt-prefix drain starts. If the runtime cannot start or complete this +boundary transaction, the capture uses its receipt memory snapshot. + +After the fixed-prefix drain completes, the runtime selects one snapshot source +for the capture. IndexedDB mode reads through the existing `timestamp_ms` +index. Memory mode reads the prepared entries from the receipt snapshot. The +drain returns the exact primary keys that it persisted through the receipt +watermark. The capture uses the boundary key for prior rows and the returned +keys for receipt-prefix rows. Thus, a row written by another tab between the +boundary transaction and the prefix drain cannot enter the capture. + +Every page excludes rows with a larger primary key, so entries written after +receipt cannot enter a later page even when their timestamps sort before +earlier rows. A capture with no persisted rows uses an empty IndexedDB source. + +Each IndexedDB page uses one readonly transaction. A continuation token contains +the timestamp index key and the object-store primary key. The index cursor uses +`continuePrimaryKey()` to seek past an equal-timestamp continuation pair. This +pair gives a stable order without rescanning the start of a timestamp group. + +The cursor scans forward from that token and includes only the requested +identity. The store holds at most 10,000 entries or 20 MiB across all +identities. Thus, one complete capture scans no more than the existing global +retention limit. A new write index is not necessary. + +The cursor stops before the next matching entry exceeds the requested page +size. The browser uploads the page before it opens the next transaction. Thus, +the browser does not clone or sort the complete identity partition. + +Persisted and memory records carry their prepared UTF-8 byte count. Page +accounting reuses that count. The request encoder performs the one required +payload serialization. + +The first page uses a 128 KiB entry-data target. Later pages use the existing +800 KiB target. Both targets use a lower value when the server advertises a +smaller maximum. This layout adds at most one request to a maximum-size capture +and gives slow links a smaller first transfer. + +The notification supplies `capture_timeout_ms`. The frontend converts this +duration to a local deadline with `performance.now()`. The frontend checks this +deadline before each page and upload. An `AbortController` cancels an active +upload when the local deadline expires. + +The backend still closes collection at its absolute `capture_deadline`. It can +reject a late request even when transport delay leaves time in the frontend +budget. The frontend does not retry this request. + +If an IndexedDB read fails before upload, the capture uses its receipt-time +memory snapshot. If a read fails after upload, the capture stops. It does not +switch sources and duplicate entries. + ### Burst collection and entry preparation The first staged entry opens one 250 ms collection window. Browser idle time @@ -92,10 +161,9 @@ change which log levels a diagnostic bundle contains. ## Migration and recovery -The version-2 upgrade transaction creates the metadata store and walks the -existing entries once to calculate count and bytes. It writes the retention -record before the upgrade commits. An interrupted upgrade rolls back as one -IndexedDB transaction. +The version-2 upgrade transaction creates the metadata store and walks existing +entries once. It calculates the count and byte totals before the transaction +commits. This capture change does not open a database upgrade. If a version-2 database lacks a valid retention record, the next write rebuilds the totals in one repair transaction before it accepts normal incremental @@ -114,11 +182,17 @@ scope. Vitest uses the dev-only `fake-indexeddb` package so these tests exercise real IndexedDB request and transaction behavior without a product dependency. Runtime tests hold the first append promise while more entries arrive. They -confirm that append concurrency stays at one and that a snapshot waits for the -same drain. Fake-time tests also prove that browser idle time cannot split one -collection window into one-entry transactions, and that a snapshot bypasses -the wait. Entry-preparation tests prove that the memory and IndexedDB paths use -the same encoded byte count. +show that append concurrency stays at one. They also prove that capture +drains only its fixed sequence prefix. Fake-time tests prove the one-second +memory fallback and continued background persistence. + +Capture tests populate more than one page across identities and timestamp ties. +They prove that continuation has no gaps or duplicates. They also prove that +the browser uploads the 128 KiB first page before it reads the next page. + +Protocol tests prove that the backend sends both deadline fields. Fake-time +frontend tests use a wall-clock skew and a monotonic clock. They prove that the +relative budget controls capture and cancels an active upload. ## Related decisions diff --git a/docs/specs/platform/system-design/diagnostic-logging-01.md b/docs/specs/platform/system-design/diagnostic-logging-01.md index 1fdaea377e..8577152504 100644 --- a/docs/specs/platform/system-design/diagnostic-logging-01.md +++ b/docs/specs/platform/system-design/diagnostic-logging-01.md @@ -165,6 +165,12 @@ paired [system design](../system-design/browser-console-retention.md). synchronously in the intercepted console call. A full staging queue drops lower-priority `debug`/`info` entries first and records loss metadata; it never delays the original console call. +- Frontend bundle capture freezes the entries that exist at notification + receipt. Entries that arrive later do not extend the capture flush. The + capture uses its receipt-time memory snapshot if the flush exceeds one second. +- The frontend reads and uploads one chronological browser page at a time. The + first page is smaller than later pages. A monotonic receipt-relative budget + stops page reads and uploads without using the browser wall clock. - A browser debug producer that scans a growing collection coalesces its own derived payload before it calls `console.debug`. The `messages:process` producer keeps the latest inputs and emits one trailing diff --git a/docs/specs/platform/system-design/diagnostic-logging-02.md b/docs/specs/platform/system-design/diagnostic-logging-02.md index f10e89f906..7c5069cde9 100644 --- a/docs/specs/platform/system-design/diagnostic-logging-02.md +++ b/docs/specs/platform/system-design/diagnostic-logging-02.md @@ -3,6 +3,7 @@ status: draft system: platform requirements: - REQ-PLATFORM-DIAGNOSTIC-LOGGING-001 + - REQ-PLATFORM-BROWSER-CONSOLE-RETENTION-001 created: 2026-07-30 owners: - tbd @@ -18,6 +19,7 @@ This design preserves the technical source detail for `REQ-PLATFORM-DIAGNOSTIC-L | Requirement | Design section | | --- | --- | | `REQ-PLATFORM-DIAGNOSTIC-LOGGING-001` | [Migrated source detail](#migrated-source-detail) | +| `REQ-PLATFORM-BROWSER-CONSOLE-RETENTION-001` | [Frontend capture paging](#frontend-capture-paging) | ## Migrated source detail @@ -187,11 +189,17 @@ the requesting identity: { "bundle_id": "01J...", "capture_deadline": "2026-07-30T12:35:15Z", + "capture_timeout_ms": 15000, "max_chunk_bytes": 1048576, "max_browser_profiles": 4 } ``` +`capture_deadline` is the absolute backend expiry time. The backend calculates +`capture_timeout_ms` from its clock when it creates the notification. The value +is not more than 15,000 milliseconds. A new frontend uses this duration with a +monotonic clock. An older frontend can continue to use `capture_deadline`. + Each frontend snapshots its local three-day store and uploads sequential chunks to `POST /api/v1/system/logs/bundles/:id/frontend`: @@ -223,6 +231,54 @@ to `POST /api/v1/system/logs/bundles/:id/frontend`: - Accepted chunks return `204 No Content`; invalid ordering or bounds return `400` or `413`; unknown/expired jobs return `404` or `410`. +#### Frontend capture paging + +The browser records a capture watermark when it receives the notification. It +also keeps a bounded memory snapshot for that watermark. Entries that arrive +later remain in normal staging and cannot extend this capture. + +At receipt, the browser starts a `readwrite` boundary transaction on its already +open IndexedDB connection. The transaction is serialized with append +transactions before the receipt-prefix drain starts. If the browser cannot +prove this boundary, it uses the receipt-time memory snapshot. + +The browser waits at most one second for the serialized persistence drain to +reach the watermark. If this wait expires, the browser uses the receipt-time +memory snapshot. The persistence drain continues and its storage mode does not +change. The upload uses `storage_mode: memory`. Its final metadata includes +`flush_timeout: true`. + +The fixed-prefix drain returns the exact persisted object-store primary keys. +IndexedDB capture uses the receipt boundary key for prior rows and those exact +prefix keys for later rows. A row written by another tab between the boundary +transaction and the prefix drain is not in that key set. Every page excludes +other rows written after receipt, including rows written between page +transactions. A capture with no persisted rows sends an empty IndexedDB page. + +IndexedDB capture reads use the existing `timestamp_ms` index. Each page resumes +with the prior timestamp and primary key. The index cursor uses +`continuePrimaryKey()` when it resumes inside an equal-timestamp group. The +cursor filters the requested identity while it scans the globally bounded +store. The browser does not use an unbounded `getAll` or sort the complete +partition in memory. + +The first page uses a 128 KiB entry-data target. Later pages use the existing +800 KiB target. A server value less than either target lowers that target. The +request envelope stays below the 1 MiB body limit. + +The browser uploads each page before it reads the next page. This sequence +keeps chunk indexes ordered and gives the renderer time between pages. The final +page carries `done: true`. An empty history sends one empty final page. + +The frontend creates a local deadline from `capture_timeout_ms` and +`performance.now()`. It checks this deadline before each page and upload. An +`AbortController` cancels an active upload at the local deadline. + +The backend still uses `capture_deadline` as the authoritative job boundary. +Transport delay can cause an upload to reach the backend after this boundary. +The backend rejects that request. The browser does not retry or log the error +through the console interceptor. + `GET /api/v1/system/logs/bundles/:id` returns the caller-owned job state: `collecting`, `building`, `ready`, `partial`, `failed`, or `expired`, plus source/capture counts, selected ACP session counts and aggregate availability,