diff --git a/app/src/components/Actions/Actions.tsx b/app/src/components/Actions/Actions.tsx index b50c8b6c..d01f4981 100644 --- a/app/src/components/Actions/Actions.tsx +++ b/app/src/components/Actions/Actions.tsx @@ -262,6 +262,13 @@ function syncIndicatorPresentation(state: NotebookSyncState | null): { clickable: boolean } { switch (state?.status) { + case 'not-downloaded': + return { + label: 'Notebook has not been downloaded to this browser. Click to download now.', + className: 'border border-nb-text-faint bg-transparent', + clickable: true, + } + case 'synced': return { label: 'Notebook is synced', diff --git a/app/src/components/DriveQueueMonitor.test.tsx b/app/src/components/DriveQueueMonitor.test.tsx new file mode 100644 index 00000000..6badd1c5 --- /dev/null +++ b/app/src/components/DriveQueueMonitor.test.tsx @@ -0,0 +1,72 @@ +import '@testing-library/jest-dom/vitest' +import { act, render, screen } from '@testing-library/react' +import { afterEach, expect, it, vi } from 'vitest' + +import { SyncWorkQueue } from '../storage/syncWorkQueue' +import { DriveQueueCharts, DriveQueueMonitor } from './DriveQueueMonitor' + +afterEach(() => vi.useRealTimers()) + +it('shows backlog separately from completed dequeue observations', () => { + const queue = new SyncWorkQueue() + queue.add('waiting', async () => {}, 60_000) + try { + render() + expect( + screen.getByText('1 waiting · 0 eligible · 1 delayed · 0 active') + ).toBeInTheDocument() + expect( + screen.getByRole('img', { + name: 'Waiting queue depth, peak per ten seconds', + }) + ).toBeInTheDocument() + expect( + screen.getByRole('img', { + name: 'Eligible-to-dequeue wait histogram, 0 attempts', + }) + ).toBeInTheDocument() + expect(screen.getByText(/No dequeue attempts yet/)).toBeInTheDocument() + } finally { + queue.close() + } +}) + +it('polls owner metrics without overlapping requests and stops on unmount', async () => { + vi.useFakeTimers() + const queue = new SyncWorkQueue() + let resolve!: (value: ReturnType) => void + const store = { + getDriveQueueMetrics: vi.fn( + () => + new Promise>((r) => { + resolve = r + }) + ), + } + const view = render() + await act(() => vi.advanceTimersByTimeAsync(15_000)) + expect(store.getDriveQueueMetrics).toHaveBeenCalledTimes(1) + await act(async () => resolve(queue.getMetrics())) + expect(screen.getByText('Sync queue')).toBeInTheDocument() + await act(() => vi.advanceTimersByTimeAsync(5_000)) + expect(store.getDriveQueueMetrics).toHaveBeenCalledTimes(2) + view.unmount() + await act(async () => resolve(queue.getMetrics())) + await act(() => vi.advanceTimersByTimeAsync(10_000)) + expect(store.getDriveQueueMetrics).toHaveBeenCalledTimes(2) + queue.close() +}) + +it('reports unavailable diagnostics rather than displaying a healthy zero', async () => { + const store = { + getDriveQueueMetrics: vi + .fn() + .mockRejectedValue(new Error('Worker unavailable')), + } + render() + expect( + await screen.findByText( + 'Queue monitoring unavailable: Error: Worker unavailable' + ) + ).toBeInTheDocument() +}) diff --git a/app/src/components/DriveQueueMonitor.tsx b/app/src/components/DriveQueueMonitor.tsx new file mode 100644 index 00000000..37577b9e --- /dev/null +++ b/app/src/components/DriveQueueMonitor.tsx @@ -0,0 +1,221 @@ +import { useEffect, useState } from 'react' + +import type LocalNotebooks from '../storage/local' +import type { DriveQueueMetrics } from '../storage/syncQueueMetrics' + +/** Keep duration labels comparable across the summary and histogram axes. */ +function duration(ms: number): string { + if (ms < 1_000) return `${Math.round(ms)} ms` + if (ms < 60_000) return `${(ms / 1_000).toFixed(1)} s` + return `${(ms / 60_000).toFixed(1)} min` +} + +/** Charts use owner snapshots, so closing this view does not lose queue history. */ +export function DriveQueueCharts({ metrics }: { metrics: DriveQueueMetrics }) { + const maxDepth = Math.max(1, ...metrics.history.map((point) => point.depth)) + const start = metrics.history[0]?.at ?? metrics.capturedAt + const span = Math.max(10_000, metrics.capturedAt - start) + const points = metrics.history.map( + (point) => + `${44 + ((point.at - start) / span) * 480},${150 - (point.depth / maxDepth) * 120}` + ) + const maxCount = Math.max( + 1, + ...metrics.waitHistogram.map((bucket) => bucket.count) + ) + const attempts = metrics.waitHistogram.reduce( + (sum, bucket) => sum + bucket.count, + 0 + ) + const time = (value: number) => new Date(value).toLocaleTimeString() + const labels = metrics.waitHistogram.map((bucket, index) => { + const lower = index ? metrics.waitHistogram[index - 1].upperBoundMs! : 0 + return bucket.upperBoundMs === null + ? `≥ ${duration(lower)}` + : `${duration(lower)}–<${duration(bucket.upperBoundMs)}` + }) + + return ( +
+

Sync queue

+

+ {metrics.depth} waiting · {metrics.eligible} eligible ·{' '} + {metrics.delayed} delayed · {metrics.active} active +

+

+ Oldest eligible wait: {duration(metrics.oldestEligibleWaitMs)}. Active + attempt: {duration(metrics.activeForMs)}. +

+
+
+
+ Queue depth over time +
+ + + + + {maxDepth} + + + 0 + + + {time(start)} + + + {time(metrics.capturedAt)} + + + {metrics.history.map((point, index) => ( + + + {time(point.at)}: {point.depth} waiting + + + ))} + +

+ Peak waiting keys per 10 seconds, up to one hour. Includes delayed + retries; excludes the active attempt. +

+
+
+
+ Eligible-to-dequeue wait +
+ + + + + {maxCount} + + + 0 + + {metrics.waitHistogram.map((bucket, index) => { + const x = 52 + index * 67 + const height = (bucket.count / maxCount) * 120 + return ( + + + + {labels[index]}: {bucket.count} attempts + + + + {bucket.count} + + + {labels[index]} + + + ) + })} + +

+ {attempts + ? `${attempts} dequeue attempts` + : 'No dequeue attempts yet'} + . Excludes scheduled debounce/backoff, processing time and locks + acquired after dequeue. Retries count separately. +

+
+
+

+ Last updated {new Date(metrics.capturedAt).toLocaleTimeString()}. Shared + across tabs on this origin. History starts{' '} + {new Date(metrics.startedAt).toLocaleString()} and resets when the + storage worker restarts. Refreshes every 5 seconds. +

+
+ ) +} + +/** Poll one lightweight owner RPC at a time; stale/unmounted requests cannot update React. */ +export function DriveQueueMonitor({ + store, +}: { + store: Pick | null +}) { + const [metrics, setMetrics] = useState(null) + const [error, setError] = useState(null) + useEffect(() => { + let cancelled = false + let timer: ReturnType + setMetrics(null) + setError(null) + if (!store) return + const refresh = async () => { + try { + const snapshot = await store.getDriveQueueMetrics() + if (!cancelled) { + setMetrics(snapshot) + setError(null) + } + } catch (error) { + if (!cancelled) setError(String(error)) + } finally { + if (!cancelled) timer = setTimeout(() => void refresh(), 5_000) + } + } + void refresh() + return () => { + cancelled = true + clearTimeout(timer) + } + }, [store]) + if (error) + return ( +

+ Queue monitoring unavailable: {error} +

+ ) + if (!metrics) + return ( +

+ Loading sync queue monitoring… +

+ ) + return +} diff --git a/app/src/components/DriveSyncStatusTab.test.tsx b/app/src/components/DriveSyncStatusTab.test.tsx index 66cfb596..c4d9557a 100644 --- a/app/src/components/DriveSyncStatusTab.test.tsx +++ b/app/src/components/DriveSyncStatusTab.test.tsx @@ -11,6 +11,7 @@ import type { ButtonHTMLAttributes, ElementType, HTMLAttributes } from 'react' import { beforeEach, describe, expect, it, vi } from 'vitest' import type { NotebookSyncStatusRow } from '../storage/local' +import { SyncWorkQueue } from '../storage/syncWorkQueue' import type { GoogleDriveCredentialStatus } from '../contexts/GoogleAuthContext' let isDriveSyncing = false @@ -33,7 +34,9 @@ const openNotebookMock = vi.fn(async (uri: string) => ({ const setCurrentDocMock = vi.fn() const showDocumentMock = vi.fn() const clearLinkedResourceCacheMock = vi.fn(async () => 1536) +const queueMetrics = new SyncWorkQueue().getMetrics() const storeMock = { + getDriveQueueMetrics: vi.fn(async () => queueMetrics), listFileSyncStatuses: listFileSyncStatusesMock, sync: syncMock, } @@ -130,6 +133,29 @@ async function waitForStatusLoad(): Promise { } describe('DriveSyncStatusTab', () => { + it('excludes untouched Drive placeholders from bulk sync and allows filtering them', async () => { + listFileSyncStatusesMock.mockResolvedValue([ + ...rows, + { ...rows[0], localUri: 'local://file/unopened', title: 'Unopened', syncStatus: 'not-downloaded' }, + ]) + render() + await waitForStatusLoad() + fireEvent.click(screen.getByRole('button', { name: 'Sync Required (1)' })) + await waitFor(() => expect(syncMock).toHaveBeenCalledWith('local://file/beta')) + expect(syncMock).not.toHaveBeenCalledWith('local://file/unopened') + fireEvent.click(screen.getByRole('button', { name: 'Filter Sync Status: All statuses' })) + fireEvent.click(screen.getByRole('checkbox', { name: 'Filter Sync Status: not-downloaded' })) + expect(screen.getByText('Unopened')).toBeTruthy() + expect(screen.queryByText('Beta Notebook')).toBeNull() + }) + + it('includes owner queue monitoring above the file status table', async () => { + render() + await waitForStatusLoad() + expect(await screen.findByRole('img', { name: 'Waiting queue depth, peak per ten seconds' })).toBeTruthy() + expect(screen.getByRole('img', { name: 'Eligible-to-dequeue wait histogram, 0 attempts' })).toBeTruthy() + }) + beforeEach(() => { window.localStorage.clear() isDriveSyncing = true diff --git a/app/src/components/DriveSyncStatusTab.tsx b/app/src/components/DriveSyncStatusTab.tsx index f923377c..8378fc0f 100644 --- a/app/src/components/DriveSyncStatusTab.tsx +++ b/app/src/components/DriveSyncStatusTab.tsx @@ -22,6 +22,8 @@ import type { NotebookSyncStatusRow, } from '../storage/local' +import { DriveQueueMonitor } from './DriveQueueMonitor' + type SortDirection = 'asc' | 'desc' type SortKey = keyof Pick< @@ -68,6 +70,7 @@ const stringColumns: Array<{ ] const syncStatusOptions: NotebookSyncStatus[] = [ + 'not-downloaded', 'local-only', 'synced', 'pending', @@ -89,7 +92,7 @@ const columnDescriptions: Record = { lastSynced: 'Time of the last successful local-to-upstream or upstream-to-local sync.', syncStatus: - 'Computed local sync state, such as synced, pending, syncing, conflicted, error, or local-only.', + 'Computed local sync state. Not-downloaded files are untouched Drive placeholders and are excluded from Sync Required.', } const refreshDescription = @@ -185,6 +188,7 @@ function statusClassName(status: NotebookSyncStatus): string { return 'bg-orange-50 text-orange-700' case 'error': return 'bg-red-50 text-red-700' + case 'not-downloaded': case 'local-only': return 'bg-slate-100 text-slate-700' default: @@ -631,10 +635,10 @@ export function DriveSyncStatusTab() { return (
-
+
@@ -689,6 +693,8 @@ export function DriveSyncStatusTab() {
+ +
) : ( -
+
diff --git a/app/src/storage/local.test.ts b/app/src/storage/local.test.ts index b8084a91..27003b52 100644 --- a/app/src/storage/local.test.ts +++ b/app/src/storage/local.test.ts @@ -8823,6 +8823,102 @@ it('serializes source, export, and creation across controllers sharing an origin }) describe('SharedWorker metadata discovery', () => { + it('does not enqueue a large folder of untouched Drive placeholders', async () => { + const logs = new MemoryOperationLogStorage() + const read = vi.spyOn(logs, 'read') + const store = createTestStore({}, { operationLogStorage: logs }) + ;(store as any).runtime = { owner: true } + for (let i = 0; i < 1_000; i++) { + await store.addFile( + `https://drive.google.com/file/d/listed-${i}/view`, + `listed-${i}.runme` + ) + } + expect(await store.reconcileDriveBackedFiles()).toEqual([]) + expect(await store.reconcileDriveBackedFiles()).toEqual([]) + expect((await store.getDriveQueueMetrics()).depth).toBe(0) + const statuses = await store.listFileSyncStatuses() + expect(statuses).toHaveLength(1_000) + expect(statuses.every((row) => row.syncStatus === 'not-downloaded')).toBe(true) + expect(read).not.toHaveBeenCalled() + store.stopSyncQueue() + }) + + it('retains materialized, dirty, failed and pending-create records without scanning OPFS', async () => { + const logs = new MemoryOperationLogStorage() + const read = vi.spyOn(logs, 'read') + const store = createTestStore({}, { operationLogStorage: logs }) + ;(store as any).runtime = { owner: true } + const base = { + name: 'notebook.runme', + remoteId: 'https://drive.google.com/file/d/a/view', + doc: '', + md5Checksum: '', + lastRemoteChecksum: '', + lastSynced: '', + } + const records: LocalFileRecord[] = [ + { + ...base, + id: 'local://file/opfs', + operationLogRef: { + storage: 'opfs', + path: 'missing-must-report-on-processing', + }, + }, + { + ...base, + id: 'local://file/json', + name: 'legacy.json', + doc: '{"cells":[]}', + }, + { + ...base, + id: 'local://file/ipynb', + name: 'legacy.ipynb', + doc: '{"cells":[]}', + }, + { + ...base, + id: 'local://file/error', + lastSyncError: 'Failed first download', + }, + { + ...base, + id: 'local://file/create', + remoteId: '', + parentRemoteIdWhenCreated: + 'https://drive.google.com/drive/folders/parent', + }, + { + ...base, + id: 'local://file/dirty', + md5Checksum: 'new', + lastRemoteChecksum: 'old', + }, + ] + for (const record of records) await store.files.put(record) + await store.files.put({ + ...base, + id: 'local://file/clean', + md5Checksum: 'same', + lastRemoteChecksum: 'same', + }) + await store.files.put({ + ...records[0], + id: 'local://file/conflict', + conflict: { + detectedAt: '2026-09-23T00:00:00Z', + upstreamChecksum: 'remote', + localChecksumAtDetection: 'local', + }, + }) + expect(await store.listDriveBackedFilesNeedingSync()).toEqual( + records.map((record) => record.id) + ) + expect(read).not.toHaveBeenCalled() + }) + it('selects unknown checksums without reading OPFS or marking the mirror clean', async () => { const logs = new MemoryOperationLogStorage() const read = vi.spyOn(logs, 'read') @@ -9079,4 +9175,4 @@ describe('LocalNotebooks local-first open', () => { await expect(store.load(uri)).rejects.toThrow('offline first download') } ) -}) +}) \ No newline at end of file diff --git a/app/src/storage/local.ts b/app/src/storage/local.ts index 91760549..72170204 100644 --- a/app/src/storage/local.ts +++ b/app/src/storage/local.ts @@ -270,6 +270,7 @@ export interface NotebookConflictSummary { } export type NotebookSyncStatus = + | 'not-downloaded' | 'local-only' | 'synced' | 'pending' @@ -584,6 +585,11 @@ export class LocalNotebooks extends Dexie { this.workQueue = undefined } + /** Read bounded owner-side diagnostics without scanning files or touching OPFS. */ + async getDriveQueueMetrics() { + return this.getWorkQueue().getMetrics() + } + private getWorkQueue(): SyncWorkQueue { return (this.workQueue ??= new SyncWorkQueue({ onChange: (key) => this.notifySync(key.slice(key.indexOf(':') + 1)), @@ -1566,6 +1572,17 @@ export class LocalNotebooks extends Dexie { return syncStateForRecord(record, 'error') } + // Folder discovery is not a local edit. Keep the status table's bulk-sync + // action from recreating the placeholder backlog excluded by reconciliation. + if ( + isDriveUri(record.remoteId) && + !record.md5Checksum && + !record.operationLogRef && + !record.doc + ) { + return syncStateForRecord(record, 'not-downloaded') + } + const localChecksum = this.runtime?.owner ? record.md5Checksum : await this.getOrBackfillLocalChecksum(localUri, record) @@ -4517,7 +4534,12 @@ export class LocalNotebooks extends Dexie { if (!isDriveUri(record.remoteId)) return false if (record.lastSyncError) return true try { - if (this.runtime?.owner && !record.md5Checksum) return true + if (this.runtime?.owner && !record.md5Checksum) { + // A folder listing creates metadata-only placeholders with empty hashes. + // Only materialized content can have an invalidated local checksum. + // Do not read OPFS here: the owner verifies it when processing the item. + return Boolean(record.operationLogRef || record.doc) + } const local = this.runtime?.owner ? record.md5Checksum : await this.getOrBackfillLocalChecksum(record.id, record) diff --git a/app/src/storage/storageOwner.test.ts b/app/src/storage/storageOwner.test.ts index 9b825331..7aad22a1 100644 --- a/app/src/storage/storageOwner.test.ts +++ b/app/src/storage/storageOwner.test.ts @@ -7,10 +7,12 @@ import { FilesystemEntryAlreadyExistsError } from './fs' import type LocalNotebooks from './local' import { StorageOwnerClient } from './storageOwnerClient' import { StorageOwnerHost } from './storageOwnerHost' +import { SyncWorkQueue } from './syncWorkQueue' const clients: StorageOwnerClient[] = [] const ports: MessagePort[] = [] afterEach(() => { + vi.useRealTimers() for (const client of clients.splice(0)) client.close() for (const port of ports.splice(0)) port.close() }) @@ -45,6 +47,42 @@ function setup() { } describe('SharedWorker message boundary', () => { + it('times out diagnostics promptly without a mutation-outcome warning', async () => { + vi.useFakeTimers() + const channel = new MessageChannel() + ports.push(channel.port1 as unknown as MessagePort) + const client = new StorageOwnerClient( + channel.port2 as unknown as MessagePort, + async () => 'token' + ) + clients.push(client) + const result = expect( + client.request('getDriveQueueMetrics') + ).rejects.toThrow('Storage worker did not respond to queue diagnostics.') + await vi.advanceTimersByTimeAsync(10_000) + await result + }) + + it('serves the same queue diagnostics to both tabs through the RPC allowlist', async () => { + const { host, store } = setup() + const queue = new SyncWorkQueue() + Object.assign(store, { + getDriveQueueMetrics: async () => queue.getMetrics(), + }) + const first = connect(host), + second = connect(host) + queue.add('source:pending', async () => {}, 60_000) + try { + const a = (await first.request('getDriveQueueMetrics')) as any + const b = (await second.request('getDriveQueueMetrics')) as any + expect(a).toMatchObject({ depth: 1, delayed: 1, active: 0 }) + expect(b.startedAt).toBe(a.startedAt) + expect(b.waitHistogram).toEqual(a.waitHistogram) + } finally { + queue.close() + } + }) + it('two tabs use the same owner while local edits continue during a blocked sync', async () => { const { host, store } = setup() const first = connect(host), @@ -116,7 +154,9 @@ describe('SharedWorker message boundary', () => { const { host, store } = setup() const client = connect(host) store.save.mockRejectedValueOnce(original) - const error = await client.request('save', ['a', {}]).catch(error => error) + const error = await client + .request('save', ['a', {}]) + .catch((error) => error) expect(error).toBeInstanceOf(original.constructor) expect(error.message).toBe(original.message) if (original instanceof FilesystemEntryAlreadyExistsError) diff --git a/app/src/storage/storageOwnerClient.ts b/app/src/storage/storageOwnerClient.ts index 0519f5c0..01508735 100644 --- a/app/src/storage/storageOwnerClient.ts +++ b/app/src/storage/storageOwnerClient.ts @@ -49,11 +49,15 @@ export class StorageOwnerClient { this.pending.delete(id) reject( new Error( - 'Storage worker did not acknowledge the request. Its outcome is uncertain; do not repeat creation blindly.' + method === 'getDriveQueueMetrics' + ? 'Storage worker did not respond to queue diagnostics.' + : 'Storage worker did not acknowledge the request. Its outcome is uncertain; do not repeat creation blindly.' ) ) }, - method === 'hello' ? 10_000 : 300_000 + method === 'hello' || method === 'getDriveQueueMetrics' + ? 10_000 + : 300_000 ) this.pending.set(id, { resolve, reject, timer }) try { @@ -128,7 +132,10 @@ export class StorageOwnerClient { if (error.name === 'DriveCreateNotCommittedError') Object.setPrototypeOf(error, DriveCreateNotCommittedError.prototype) if (error.name === 'FilesystemEntryAlreadyExistsError') - Object.setPrototypeOf(error, FilesystemEntryAlreadyExistsError.prototype) + Object.setPrototypeOf( + error, + FilesystemEntryAlreadyExistsError.prototype + ) pending.reject(error) } else pending.resolve(message.value) } diff --git a/app/src/storage/storageOwnerProtocol.ts b/app/src/storage/storageOwnerProtocol.ts index 6676acd4..a51e685a 100644 --- a/app/src/storage/storageOwnerProtocol.ts +++ b/app/src/storage/storageOwnerProtocol.ts @@ -1,7 +1,7 @@ import type LocalNotebooks from './local' /** Bump whenever the wire contract or storage ownership assumptions change. */ -export const STORAGE_OWNER_VERSION = 2 +export const STORAGE_OWNER_VERSION = 3 export const STORAGE_OWNER_NAME = 'runme-storage-owner' /** Explicit RPC boundary: never expose arbitrary Dexie/prototype methods. */ @@ -16,6 +16,7 @@ export const STORAGE_METHODS = [ 'updateFolder', 'sync', 'getSyncState', + 'getDriveQueueMetrics', 'listFileSyncStatuses', 'getMetadata', 'save', diff --git a/app/src/storage/syncQueueMetrics.ts b/app/src/storage/syncQueueMetrics.ts new file mode 100644 index 00000000..a21aa6b3 --- /dev/null +++ b/app/src/storage/syncQueueMetrics.ts @@ -0,0 +1,82 @@ +/** Bounded, in-memory diagnostics owned by the sync worker, never notebook data. */ +export const QUEUE_HISTORY_INTERVAL_MS = 10_000 +export const QUEUE_HISTORY_BUCKETS = 360 +const WAIT_UPPER_BOUNDS_MS = [100, 1_000, 5_000, 30_000, 120_000, 600_000] + +export type QueueDepthPoint = { at: number; depth: number } +export type QueueWaitBucket = { upperBoundMs: number | null; count: number } +export type DriveQueueMetrics = { + startedAt: number + capturedAt: number + depth: number + eligible: number + delayed: number + active: number + activeForMs: number + oldestEligibleWaitMs: number + history: QueueDepthPoint[] + waitHistogram: QueueWaitBucket[] +} + +/** + * Keep peak depth in each 10-second bucket for up to an hour. Fill quiet periods + * using the previous depth on the next event/read; no sampling timer is needed. + * Histogram buckets count dequeue attempts for this worker's lifetime. + */ +export class SyncQueueMetrics { + private readonly startedAt = Date.now() + private depth = 0 + private history: QueueDepthPoint[] = [] + private readonly histogram: QueueWaitBucket[] = [ + ...WAIT_UPPER_BOUNDS_MS.map((upperBoundMs) => ({ upperBoundMs, count: 0 })), + { upperBoundMs: null, count: 0 }, + ] + + /** Advance the history without allocating for every enqueue or reconciliation. */ + private advance(now: number): void { + const bucket = + Math.floor(now / QUEUE_HISTORY_INTERVAL_MS) * QUEUE_HISTORY_INTERVAL_MS + const earliest = + bucket - (QUEUE_HISTORY_BUCKETS - 1) * QUEUE_HISTORY_INTERVAL_MS + this.history = this.history.filter((point) => point.at >= earliest) + let next = this.history.length + ? this.history[this.history.length - 1].at + QUEUE_HISTORY_INTERVAL_MS + : Math.max( + earliest, + Math.floor(this.startedAt / QUEUE_HISTORY_INTERVAL_MS) * + QUEUE_HISTORY_INTERVAL_MS + ) + while (next <= bucket) { + this.history.push({ at: next, depth: this.depth }) + next += QUEUE_HISTORY_INTERVAL_MS + } + } + + /** Repeated additions of the same key do not inflate depth. */ + depthChanged(depth: number): void { + this.advance(Date.now()) + this.depth = depth + const last = this.history[this.history.length - 1] + if (last) last.depth = Math.max(last.depth, depth) + } + + /** Delay is measured from eligibility, excluding scheduled debounce/backoff. */ + dequeued(waitMs: number): void { + const bucket = this.histogram.find( + (bucket) => bucket.upperBoundMs === null || waitMs < bucket.upperBoundMs + )! + bucket.count += 1 + } + + /** Detached snapshots can safely cross MessagePorts and be retained by React. */ + snapshot() { + const capturedAt = Date.now() + this.advance(capturedAt) + return { + startedAt: this.startedAt, + capturedAt, + history: this.history.map((point) => ({ ...point })), + waitHistogram: this.histogram.map((bucket) => ({ ...bucket })), + } + } +} diff --git a/app/src/storage/syncWorkQueue.test.ts b/app/src/storage/syncWorkQueue.test.ts index 14a2d16f..f5625bda 100644 --- a/app/src/storage/syncWorkQueue.test.ts +++ b/app/src/storage/syncWorkQueue.test.ts @@ -94,3 +94,79 @@ describe('delaying sync work queue', () => { expect(secondRun).toHaveBeenCalledTimes(1) }) }) + +describe('owner queue diagnostics', () => { + it('counts deduplicated waiting keys and measures only eligible wait', async () => { + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-09-23T00:00:00Z')) + const q = queue() + let release!: () => void + q.add( + 'active', + () => + new Promise((resolve) => { + release = resolve + }) + ) + await vi.advanceTimersByTimeAsync(0) + q.add('delayed', async () => {}, 20_000) + q.add('delayed', async () => {}, 20_000) + expect(q.getMetrics()).toMatchObject({ + depth: 1, + eligible: 0, + delayed: 1, + active: 1, + }) + await vi.advanceTimersByTimeAsync(25_000) + expect(q.getMetrics()).toMatchObject({ + oldestEligibleWaitMs: 5_000, + activeForMs: 25_000, + }) + // Neither a repeated credential wake nor explicit sync resets time already waiting. + q.wake() + const done = q.run('delayed', async () => {}) + expect(q.getMetrics().oldestEligibleWaitMs).toBe(5_000) + release() + await vi.advanceTimersByTimeAsync(0) + await done + const metrics = q.getMetrics() + expect(metrics).toMatchObject({ depth: 0, active: 0 }) + expect(metrics.waitHistogram.map((bucket) => bucket.count)).toEqual([ + 1, 0, 0, 1, 0, 0, 0, + ]) + expect(metrics.history.some((point) => point.depth === 1)).toBe(true) + }) + + it('counts retries separately without including their backoff', async () => { + vi.useFakeTimers() + const q = queue() + const run = vi + .fn() + .mockRejectedValueOnce(new SyncDeferred(120_000)) + .mockResolvedValue(undefined) + q.add('a', run) + await vi.advanceTimersByTimeAsync(0) + expect(q.getMetrics()).toMatchObject({ depth: 1, eligible: 0, delayed: 1 }) + await vi.advanceTimersByTimeAsync(120_000) + expect(q.getMetrics().waitHistogram.map((bucket) => bucket.count)).toEqual([ + 2, 0, 0, 0, 0, 0, 0, + ]) + }) + + it('keeps bounded history while the view is closed and returns detached snapshots', async () => { + vi.useFakeTimers() + const q = queue() + q.add('future', async () => {}, 10 * 60 * 60_000) + await vi.advanceTimersByTimeAsync(2 * 60 * 60_000) + const metrics = q.getMetrics() + expect(metrics.history).toHaveLength(360) + expect(metrics.history.every((point) => point.depth === 1)).toBe(true) + metrics.history[0].depth = 900 + metrics.waitHistogram[0].count = 900 + expect(q.getMetrics().history[0].depth).toBe(1) + expect(q.getMetrics().waitHistogram[0].count).toBe(0) + q.close() + expect(q.getMetrics().depth).toBe(0) + expect(queue().getMetrics().waitHistogram[0].count).toBe(0) + }) +}) diff --git a/app/src/storage/syncWorkQueue.ts b/app/src/storage/syncWorkQueue.ts index 68a0b760..577ff11c 100644 --- a/app/src/storage/syncWorkQueue.ts +++ b/app/src/storage/syncWorkQueue.ts @@ -1,3 +1,5 @@ +import { SyncQueueMetrics } from './syncQueueMetrics' + /** A retry can request a delay without counting an unavailable dependency as failure. */ export class SyncDeferred extends Error { constructor(readonly delayMs: number) { @@ -27,6 +29,7 @@ export class SyncWorkQueue { private processing?: string private timer?: ReturnType private stopped = false + private readonly metrics = new SyncQueueMetrics() constructor( private readonly options: { @@ -57,7 +60,8 @@ export class SyncWorkQueue { /** Wake delayed items after a credential/connectivity change. */ wake(): void { - for (const item of this.items.values()) item.readyAt = Date.now() + for (const item of this.items.values()) + item.readyAt = Math.min(item.readyAt, Date.now()) this.schedule() } @@ -78,6 +82,39 @@ export class SyncWorkQueue { item.waiters = [] } this.items.clear() + this.recordDepth() + } + + /** Waiting keys exclude the active attempt, including while it awaits a lock. */ + private recordDepth(): void { + this.metrics.depthChanged( + this.items.size - + (this.processing && this.items.has(this.processing) ? 1 : 0) + ) + } + + /** Snapshot data belongs to the queue owner, so all tabs observe the same history. */ + getMetrics() { + const now = Date.now() + const waiting = [...this.items].filter(([key]) => key !== this.processing) + const eligible = waiting.filter(([, item]) => item.readyAt <= now) + return { + ...this.metrics.snapshot(), + depth: waiting.length, + eligible: eligible.length, + delayed: waiting.length - eligible.length, + active: this.processing ? 1 : 0, + activeForMs: this.processing + ? Math.max( + 0, + now - (this.items.get(this.processing)?.lastStarted ?? now) + ) + : 0, + oldestEligibleWaitMs: eligible.reduce( + (max, [, item]) => Math.max(max, now - item.readyAt), + 0 + ), + } } private put( @@ -103,13 +140,14 @@ export class SyncWorkQueue { } else { item.run = run item.dirty = true - if (force) item.readyAt = Date.now() + if (force) item.readyAt = Math.min(item.readyAt, Date.now()) } this.schedule() return item } private schedule(): void { + this.recordDepth() clearTimeout(this.timer) if (this.stopped || this.processing || !this.items.size) return let next = Infinity @@ -134,6 +172,8 @@ export class SyncWorkQueue { this.processing = key item.dirty = false item.lastStarted = Date.now() + this.metrics.dequeued(Math.max(0, item.lastStarted - item.readyAt)) + this.recordDepth() this.lastStarts.set(key, item.lastStarted) for (const [oldKey, started] of this.lastStarts) { if (started + (this.options.minimumIntervalMs ?? 120_000) < Date.now()) diff --git a/docs-dev/CUJs/drive-sync-recovery.md b/docs-dev/CUJs/drive-sync-recovery.md index 61b07e61..1745e8af 100644 --- a/docs-dev/CUJs/drive-sync-recovery.md +++ b/docs-dev/CUJs/drive-sync-recovery.md @@ -134,3 +134,40 @@ The `createView` worker response now carries that snapshot, so the storage owner protocol is version 2. Mixed-version clients must fail with the existing reload message. After deployment, close/reopen all Runme tabs on that origin if an old worker remains alive; do not clear browser storage. + +## Lazy Drive discovery and queue monitoring + +Mirroring a large Drive folder must not enqueue metadata-only files that have +never been downloaded. In worker mode, an empty local checksum indicates source +work only with an OPFS operation-log reference or a cached legacy model. Pending +creation and failed first-download recovery still retry. A missing referenced log +must report its read error rather than being skipped or replaced. + +The Drive status page shows owner-side queue depth (peak waiting keys per ten +seconds, up to one hour) and eligible-to-dequeue wait (histogram of attempts since +owner startup). Waiting includes delayed work but excludes the active attempt; +wait excludes debounce, retry delay and post-dequeue processing/locks. Retries count +separately. Current eligible/delayed/active counts, oldest eligible wait and active +attempt duration distinguish stalled processing from slow throughput. Metrics are +shared across tabs, collected with the view closed, and reset on owner restart. +The diagnostics RPC uses storage owner protocol version 3. + +Automated coverage: `local.test.ts` mirrors 1,000 placeholders without enqueueing +or OPFS reads and preserves legitimate pending cases; `syncWorkQueue.test.ts` +checks depth, eligible wait/backoff, deduplication, bounded history and reset; +`storageOwner.test.ts` reads metrics from two MessagePorts; +`DriveQueueMonitor.test.tsx` covers both charts, polling and unavailable state. + +Manual acceptance: mount a large disposable folder without opening its notebooks; +confirm no source backlog from untouched entries. Edit a few cached notebooks and +create one offline. Open Drive status, then close/reopen only that view: the charts +retain owner history. Restore connectivity and observe the waiting count drain. +Check that the histogram excludes scheduled delays and the current oldest-wait +summary grows behind a deliberately stalled attempt. Close all same-origin tabs +and reopen: history restarts without deleting pending notebook work. + +Untouched placeholders are labeled `not-downloaded` and can be filtered in the +status table. “Sync Required” must exclude them, even when another row needs sync. +Explicit opening still downloads them. If the worker stops responding, diagnostics +show an unavailable message after ten seconds; the charts identify the time of +the last received snapshot instead of implying current health. diff --git a/docs-dev/design/20260923_drive_queue_over_enqueuing.md b/docs-dev/design/20260923_drive_queue_over_enqueuing.md new file mode 100644 index 00000000..09734f64 --- /dev/null +++ b/docs-dev/design/20260923_drive_queue_over_enqueuing.md @@ -0,0 +1,103 @@ +# Drive reconciliation over-enqueues untouched files + +## Bug and evidence + +After the SharedWorker storage-owner rollout (#391), opening a mirrored Drive +folder can enqueue notebooks that were never downloaded or edited locally. A +reported session repeatedly logged `queuedCount: 927` while the user edits roughly +ten documents per day. New notebooks remained at “waiting for upstream creation,” +and clicking sync appeared ineffective. The exact composition of those 927 live +records has not been inspected; the placeholder-selection bug is confirmed in code. + +Folder discovery calls `addFile`, which persists metadata-only records with empty +`doc`, `md5Checksum`, and `lastRemoteChecksum`, and no `operationLogRef`. The worker's +`needsDriveSourceSync` treated every missing checksum as dirty. That sentinel is +also deliberately used after local OPFS writes, where hashing is deferred until +reconciliation. These two states must not be conflated. + +This is a recurring selection bug, not a migration. A pass runs on startup/auth +recovery, connectivity recovery, and every two minutes. “Pass finished” means the +scan/enqueue completed. Its count includes eligible records already in the keyed +queue, not newly added entries or successful syncs. Repeated passes coalesce by +work type and local URI. Source, create, Markdown and IPYNB jobs share the serial +queue; manual sync removes delay but does not prioritize a key ahead of the backlog. + +## Fix + +In worker mode, an empty checksum is pending source work only if the record has an +`operationLogRef` or a nonempty legacy `doc`. Folder-listing placeholders stay lazy +until explicitly opened. This preserves offline-first opening and avoids reading +OPFS for every record on every scan. The processing path reads the referenced log; +a missing/corrupt log remains a recoverable error, never a clean placeholder. + +Keep these existing cases ahead of that check: pending upstream creation and saved +sync errors (including failed first downloads) still retry; conflicts are excluded. +Known hashes still compare against the correct format-specific upstream baseline. +JSON and IPYNB models remain eligible through their cached `doc`. + +This PR does not change queue priority. Fixing selection removes spurious work; +manual priority and explicit queued feedback can be addressed separately. + +## Drive status monitoring + +Collect metrics in `SyncWorkQueue` in the storage owner and expose a read-only, +allowlisted `getDriveQueueMetrics` RPC. The status view polls it every five seconds, +with one request outstanding. Opening the view does not enumerate files or read +OPFS for these metrics. Every tab on the origin sees the same queue history. + +- **Queue depth:** distinct waiting work keys, including scheduled/backoff work, + excluding the active attempt. A same-key follow-up during an active attempt enters + the waiting count after that attempt finishes. Keep peak depth per ten-second + bucket for at most one hour. Quiet buckets are filled lazily from the previous + depth, so metrics are collected even with the status view closed, without a timer. +- **Wait histogram:** `dequeuedAt - readyAt`, recorded once per dequeue attempt. + Bounds are 100 ms, 1 s, 5 s, 30 s, 2 min and 10 min, plus an overflow bucket. + Scheduled debounce/backoff is excluded; retries count separately. Credential + wake-ups and repeated manual requests cannot reset an already-eligible timestamp. + Processing time and locks acquired after dequeue are excluded. +- **Current state:** waiting, eligible, delayed and active counts, oldest eligible + wait, and active-attempt duration. These show a stuck backlog even before any + additional item is dequeued, when a histogram alone would appear healthy. + +History and histogram counts are bounded in-memory diagnostics, reset when the +owner restarts. Histogram counts cover that owner lifetime; depth covers at most +one hour. No notebook identifiers/content or credentials are stored in metrics. +The UI states scope, units, bucket semantics and reset time; RPC errors display as +unavailable rather than zero. The protocol version advances to 3 for the added +method. Existing tabs may need reopening together after deployment; do not clear +browser storage. + +## Validation + +Regression coverage mirrors 1,000 untouched files and reconciles twice, asserting +zero queued source jobs and zero OPFS reads. Separate cases preserve unknown-hash +OPFS logs, cached JSON/IPYNB, dirty known hashes, failed downloads, pending creation, +and conflict exclusion. Existing missing-log coverage keeps recovery visible. + +Queue tests cover deduplication, blocked active work, eligibility timing, repeated +wake/manual requests, retry backoff, bounded history, detached snapshots and reset. +MessagePort tests verify both tabs read the same owner's metrics. UI tests cover +both charts, empty observations, polling without overlap, cleanup and errors. + +Validation completed on this branch: `runme run build test` passed; the full app +suite passed 1,769 tests with two workers after an existing Actions lifecycle test +hit its five-second timeout in the default parallel run. The focused review suite +also passed 299 tests. Both charts were visually inspected using the real +component with synthetic queue data in a local browser preview. The standalone +app typecheck reports the same 125 errors as the base commit, with no additional +file/error-code diagnostics. + +Tracking issue: https://github.com/runmedev/web/issues/393. + +## Review corrections + +The status table must share the lazy-discovery semantics. Untouched placeholders +now report `not-downloaded`, appear in that filter, and are excluded from the +status page's bulk “Sync Required” action. Explicit opening/downloading still works; +failed downloads and pending creations keep their retryable states. + +Queue diagnostics use a ten-second read timeout and a diagnostic-specific error. +They must not leave stale charts looking current for the mutation RPC's five-minute +timeout or warn that notebook creation may have committed. The chart displays its +last snapshot time. The file table retains bounded vertical scrolling below the +monitoring section.