From d1689a4bea4d8d9e23a6b2ff41e4708d3dc5a142 Mon Sep 17 00:00:00 2001 From: Yuren Ju Date: Mon, 13 Jul 2026 11:49:32 +0900 Subject: [PATCH 1/2] feat(drive): add --debug NDJSON sync log to drive watch --- src/handwritten/commands/drive/debug-log.ts | 45 ++++ src/handwritten/commands/drive/sync.ts | 48 +++- src/handwritten/commands/drive/watch.ts | 38 ++- test/handwritten/drive/debug-log.test.ts | 254 ++++++++++++++++++++ 4 files changed, 376 insertions(+), 9 deletions(-) create mode 100644 src/handwritten/commands/drive/debug-log.ts create mode 100644 test/handwritten/drive/debug-log.test.ts diff --git a/src/handwritten/commands/drive/debug-log.ts b/src/handwritten/commands/drive/debug-log.ts new file mode 100644 index 0000000..90fc416 --- /dev/null +++ b/src/handwritten/commands/drive/debug-log.ts @@ -0,0 +1,45 @@ +import { appendFileSync, mkdirSync, renameSync, statSync } from "node:fs" +import { join } from "node:path" +import { driveIsoTimestamp, systemDriveClock, type DriveClock } from "./clock.js" +import { DRIVE_DIR } from "./state.js" + +export interface DriveDebugLogger { + log(event: string, fields?: Record): void +} + +export const noopDriveDebugLogger: DriveDebugLogger = { log() {} } + +export const DEBUG_LOG_FILE = "debug.log" +const MAX_DEBUG_LOG_BYTES = 10 * 1024 * 1024 + +export function createDriveDebugLogger( + root: string, + options: { clock?: DriveClock; maxBytes?: number } = {}, +): DriveDebugLogger { + const clock = options.clock ?? systemDriveClock + const maxBytes = options.maxBytes ?? MAX_DEBUG_LOG_BYTES + const dir = join(root, DRIVE_DIR) + const file = join(dir, DEBUG_LOG_FILE) + let warned = false + return { + log(event, fields = {}) { + // Debug logging must never break sync; failures are swallowed after a + // single stderr warning. + try { + mkdirSync(dir, { recursive: true }) + try { + if (statSync(file).size > maxBytes) renameSync(file, `${file}.old`) + } catch { + // File missing or unreadable; append below decides the outcome. + } + appendFileSync(file, `${JSON.stringify({ ts: driveIsoTimestamp(clock), event, ...fields })}\n`) + } catch (error) { + if (!warned) { + warned = true + const message = error instanceof Error ? error.message : String(error) + process.stderr.write(`wspc drive: debug log write failed: ${message}\n`) + } + } + }, + } +} diff --git a/src/handwritten/commands/drive/sync.ts b/src/handwritten/commands/drive/sync.ts index 7f35e4c..7d7b355 100644 --- a/src/handwritten/commands/drive/sync.ts +++ b/src/handwritten/commands/drive/sync.ts @@ -3,6 +3,7 @@ import { mkdir, readFile, rm, writeFile } from "node:fs/promises" import { basename, dirname, join, posix as pathPosix, resolve } from "node:path" import { createHash, randomUUID } from "node:crypto" import { createDriveApi } from "./api.js" +import { noopDriveDebugLogger, type DriveDebugLogger } from "./debug-log.js" import { driveConflictTimestamp, driveIsoTimestamp, systemDriveClock, type DriveClock } from "./clock.js" import { decideDriveAction, type DriveAction } from "./decision.js" import { normalizeRemoteManifest } from "./manifest.js" @@ -89,18 +90,23 @@ export async function runDriveSyncOnce( api?: DriveSyncApi, clock: DriveClock = systemDriveClock, onProgress?: DriveSyncProgress, + debug: DriveDebugLogger = noopDriveDebugLogger, ): Promise { return withDriveLock(root, async () => { let state = await readDriveState(root) const syncApi = api ?? (await createDriveApi()) const summary = emptySummary() const blockedPaths = new Set() + const scanStartedMs = Date.now() const localFiles = await scanDriveFiles(root, { onPathError: async (path, error) => { await recordPathError(summary, blockedPaths, path, error) }, }) + const scanMs = Date.now() - scanStartedMs + const manifestStartedMs = Date.now() const remoteFiles = await fetchRemoteManifest(root, state, syncApi, summary, blockedPaths) + const manifestMs = Date.now() - manifestStartedMs const paths = Array.from( new Set([...Object.keys(localFiles), ...Object.keys(remoteFiles), ...Object.keys(state.entries)]), ) @@ -115,11 +121,27 @@ export async function runDriveSyncOnce( let processed = 0 onProgress?.(processed, total) + const processStartedMs = Date.now() for (const path of paths) { const remote = remoteFiles[path] - const action = decideDriveAction(state.entries[path], localFiles[path], remote) - const result = await processPath({ root, state, api: syncApi, path, action, remote, local: localFiles[path], summary, clock }) + const local = localFiles[path] + const action = decideDriveAction(state.entries[path], local, remote) + if (isActionableAction(action)) { + debug.log("decision", decisionFields(path, action, state.entries[path], local, remote)) + } + const previousConflict = state.conflicts[path] + const result = await processPath({ root, state, api: syncApi, path, action, remote, local, summary, clock }) state = result.state + const recordedConflict = state.conflicts[path] + if (recordedConflict !== undefined && recordedConflict !== previousConflict) { + debug.log("conflict", { + path, + reason: recordedConflict.reason, + ...(recordedConflict.type === undefined ? {} : { type: recordedConflict.type }), + ...(recordedConflict.strategy === undefined ? {} : { strategy: recordedConflict.strategy }), + ...(recordedConflict.conflict_paths === undefined ? {} : { conflict_paths: recordedConflict.conflict_paths }), + }) + } if (isActionableAction(action)) { processed += 1 onProgress?.(processed, total) @@ -128,10 +150,32 @@ export async function runDriveSyncOnce( } recordUnresolvedConflicts(summary, state) + debug.log("sync_phases", { scan_ms: scanMs, manifest_ms: manifestMs, process_ms: Date.now() - processStartedMs }) return summary }) } +function decisionFields( + path: string, + action: DriveAction, + entry: DriveStateEntry | undefined, + local: { sha256: string; size_bytes: number } | undefined, + remote: RemoteEntry | undefined, +): Record { + return { + path, + action: action.type, + ...("reason" in action && action.reason !== undefined ? { reason: action.reason } : {}), + ...(entry === undefined + ? {} + : { base_version_id: entry.current_version_id, base_entry_version: entry.entry_version, base_sha256: entry.content_sha256 }), + ...(local === undefined ? {} : { local_sha256: local.sha256, local_size_bytes: local.size_bytes }), + ...(remote === undefined + ? {} + : { remote_version_id: remote.current_version_id, remote_entry_version: remote.entry_version, remote_sha256: remote.content_sha256 }), + } +} + export function driveSyncCommand(api?: DriveSyncApi): Command { const sync = new Command("sync").description("Drive sync commands") sync diff --git a/src/handwritten/commands/drive/watch.ts b/src/handwritten/commands/drive/watch.ts index 80c6afd..5cdb4dc 100644 --- a/src/handwritten/commands/drive/watch.ts +++ b/src/handwritten/commands/drive/watch.ts @@ -4,6 +4,7 @@ import { watch as fsWatch } from "node:fs" import { relative, resolve } from "node:path" import { loadRealtimeAuthHeaders } from "../../auth/load-sdk-client.js" import { render, shouldOutputJson } from "../../output/render.js" +import { createDriveDebugLogger, noopDriveDebugLogger, type DriveDebugLogger } from "./debug-log.js" import { createDriveRealtimeSource } from "./realtime.js" import { DRIVE_DIR, ensureDriveRealtimeState, readDriveState, writeDriveRealtimeState } from "./state.js" import { runDriveSyncOnce, type DriveSyncProgress, type DriveSyncSummary } from "./sync.js" @@ -39,12 +40,15 @@ export interface DriveWatchOptions { debounceMs?: number remoteDebounceMs?: number onEvent?: (event: unknown) => void + debug?: boolean + debugLogger?: DriveDebugLogger } export async function runDriveWatch(root: string, options: DriveWatchOptions = {}): Promise { + const dbg = options.debugLogger ?? (options.debug ? createDriveDebugLogger(root) : noopDriveDebugLogger) const runSync = options.runSync ?? - ((syncRoot: string, onProgress?: DriveSyncProgress) => runDriveSyncOnce(syncRoot, undefined, undefined, onProgress)) + ((syncRoot: string, onProgress?: DriveSyncProgress) => runDriveSyncOnce(syncRoot, undefined, undefined, onProgress, dbg)) const debounceMs = options.debounceMs ?? 500 const remoteDebounceMs = options.remoteDebounceMs ?? 2000 // Watch is a long-lived event stream: json mode must emit NDJSON (one event @@ -59,6 +63,7 @@ export async function runDriveWatch(root: string, options: DriveWatchOptions = { } render({ kind: "drive_watch", display: { shape: "object" } }, event) }) + let nextTrigger = "initial" let debounceTimer: NodeJS.Timeout | undefined let debounceDeadlineMs: number | undefined let retryTimer: NodeJS.Timeout | undefined @@ -85,14 +90,28 @@ export async function runDriveWatch(root: string, options: DriveWatchOptions = { // A large first upload can run for minutes; without this the app // shows a stale badge and users assume sync never started. emit({ kind: "drive_sync_start" }) + dbg.log("sync_start", { trigger: nextTrigger }) + const startedAtMs = Date.now() const summary = await runSync(root, (processed, total) => { emit({ kind: "drive_sync_progress", processed, total }) }) emit({ kind: "drive_sync_once", ...summary }) + dbg.log("sync_end", { + duration_ms: Date.now() - startedAtMs, + uploaded: summary.uploaded, + downloaded: summary.downloaded, + deleted: summary.deleted, + merged: summary.merged, + unchanged: summary.unchanged, + conflicts: summary.conflicts, + errors: summary.errors, + }) backoffMs = 1000 } catch (error) { if (isAuthError(error) || isFatalWatchError(error) || !isRetryableWatchError(error)) throw error emit({ kind: "drive_watch_retry", delay_ms: backoffMs, error: errorMessage(error) }) + dbg.log("retry", { delay_ms: backoffMs, error: errorMessage(error) }) + nextTrigger = "retry" if (stopped) return await waitForManagedTimer(backoffMs) if (stopped) return @@ -112,7 +131,8 @@ export async function runDriveWatch(root: string, options: DriveWatchOptions = { debounceDeadlineMs = undefined } - function scheduleSync(delayMs: number): void { + function scheduleSync(delayMs: number, trigger: string): void { + nextTrigger = trigger if (running) { rerunRequested = true return @@ -187,7 +207,8 @@ export async function runDriveWatch(root: string, options: DriveWatchOptions = { source = options.source ?? createDefaultWatchSource(root) source.onChange((path) => { if (isDriveInternalPath(root, path)) return - scheduleSync(debounceMs) + dbg.log("fs_event", { path: relative(root, path) }) + scheduleSync(debounceMs, "local") }) emit({ kind: "drive_watch_started", root, library_id: state.library_id }) @@ -198,11 +219,12 @@ export async function runDriveWatch(root: string, options: DriveWatchOptions = { }, onEvent(event) { emit(realtimeEvent(event)) + dbg.log("realtime_event", { ...event }) if (event.immediate) { - scheduleSync(0) + scheduleSync(0, "remote") return } - scheduleSync(event.debounce_ms ?? remoteDebounceMs) + scheduleSync(event.debounce_ms ?? remoteDebounceMs, "remote") }, onReconnect(delayMs, error) { emit({ kind: "drive_realtime_reconnecting", delay_ms: delayMs, error }) @@ -236,8 +258,10 @@ export function driveWatchCommand(options: DriveWatchOptions = {}): Command { return new Command("watch") .description("Watch a bound Drive folder and sync local changes") .argument("[path]", "local folder path", ".") - .action(async (path: string) => { - await runDriveWatch(resolve(path), options) + .option("--debug", "append debug NDJSON events to /.wspc-drive/debug.log") + .action(async (path: string, flags: { debug?: boolean }) => { + const debug = options.debug ?? (flags.debug === true || process.env.WSPC_DRIVE_DEBUG === "1") + await runDriveWatch(resolve(path), { ...options, debug }) }) } diff --git a/test/handwritten/drive/debug-log.test.ts b/test/handwritten/drive/debug-log.test.ts new file mode 100644 index 0000000..0f526a1 --- /dev/null +++ b/test/handwritten/drive/debug-log.test.ts @@ -0,0 +1,254 @@ +import { mkdtemp, readFile, stat, writeFile } from "node:fs/promises" +import { join } from "node:path" +import { tmpdir } from "node:os" +import { createHash } from "node:crypto" +import { DateTime } from "luxon" +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest" +import { createDriveDebugLogger } from "../../../src/handwritten/commands/drive/debug-log.js" +import { runDriveWatch, type DriveWatchSource } from "../../../src/handwritten/commands/drive/watch.js" +import { runDriveSyncOnce, type DriveSyncApi } from "../../../src/handwritten/commands/drive/sync.js" +import { initDriveState } from "../../../src/handwritten/commands/drive/state.js" +import type { DriveClock } from "../../../src/handwritten/commands/drive/clock.js" + +const fixedClock: DriveClock = { + now: () => DateTime.fromISO("2026-07-13T10:00:00+08:00", { setZone: true }), +} + +async function readDebugLines(root: string): Promise>> { + const raw = await readFile(join(root, ".wspc-drive", "debug.log"), "utf8") + return raw + .split("\n") + .filter((line) => line !== "") + .map((line) => JSON.parse(line) as Record) +} + +describe("drive debug logger", () => { + it("appends NDJSON lines with ts and event fields", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-debug-log-")) + const logger = createDriveDebugLogger(root, { clock: fixedClock }) + + logger.log("fs_event", { path: "notes.txt" }) + logger.log("retry", { delay_ms: 1000, error: "network" }) + + const lines = await readDebugLines(root) + expect(lines).toEqual([ + { ts: "2026-07-13T10:00:00.000+08:00", event: "fs_event", path: "notes.txt" }, + { ts: "2026-07-13T10:00:00.000+08:00", event: "retry", delay_ms: 1000, error: "network" }, + ]) + }) + + it("rotates debug.log to debug.log.old past the size threshold", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-debug-rotate-")) + const logger = createDriveDebugLogger(root, { clock: fixedClock, maxBytes: 10 }) + + logger.log("fs_event", { path: "a".repeat(50) }) + logger.log("fs_event", { path: "second" }) + + const oldStat = await stat(join(root, ".wspc-drive", "debug.log.old")) + expect(oldStat.size).toBeGreaterThan(10) + const lines = await readDebugLines(root) + expect(lines).toHaveLength(1) + expect(lines[0]).toMatchObject({ event: "fs_event", path: "second" }) + }) + + it("swallows write failures without throwing", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-debug-fail-")) + // Make /.wspc-drive a file so mkdir/append must fail. + await writeFile(join(root, ".wspc-drive"), "not a dir") + const logger = createDriveDebugLogger(root, { clock: fixedClock }) + + expect(() => logger.log("fs_event", { path: "x" })).not.toThrow() + }) +}) + +function syncSummary() { + return { + uploaded: 1, + downloaded: 0, + deleted: 0, + unchanged: 0, + merged: 0, + conflicts: 0, + errors: 0, + conflict_paths: [], + paths: [], + } +} + +function fakeSource(): DriveWatchSource & { emit(path: string): void } { + const handlers: Array<(path: string) => void> = [] + return { + onChange(handler) { + handlers.push(handler) + }, + async close() {}, + emit(path: string) { + for (const handler of handlers) handler(path) + }, + } +} + +describe("drive watch --debug", () => { + it("logs fs_event, sync_start, and sync_end", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-watch-debug-")) + const source = fakeSource() + const runSync = vi.fn(async () => { + // Emit only once: emitting on every call would schedule a rerun after + // each sync and the watch loop would never drain. + if (runSync.mock.calls.length === 1) source.emit(join(root, "notes.txt")) + return syncSummary() + }) + const readState = async () => ({ library_id: "lib_1" }) as never + + await runDriveWatch(root, { source, runSync, readState, once: true, debug: true }) + + const lines = await readDebugLines(root) + const events = lines.map((line) => line.event) + expect(events).toContain("fs_event") + expect(events).toContain("sync_start") + expect(events).toContain("sync_end") + const start = lines.find((line) => line.event === "sync_start") + expect(start).toMatchObject({ trigger: "initial" }) + const end = lines.find((line) => line.event === "sync_end") + expect(end).toMatchObject({ uploaded: 1, conflicts: 0, errors: 0 }) + expect(typeof end?.duration_ms).toBe("number") + }) + + it("logs retry events with backoff delay", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-watch-debug-retry-")) + let calls = 0 + const runSync = vi.fn(async () => { + calls += 1 + if (calls === 1) throw new Error("network flake") + return syncSummary() + }) + const readState = async () => ({ library_id: "lib_1" }) as never + + // Real timers: the first backoff is only 1s. + await runDriveWatch(root, { source: fakeSource(), runSync, readState, once: true, debug: true }) + expect(calls).toBe(2) + + const lines = await readDebugLines(root) + const retry = lines.find((line) => line.event === "retry") + expect(retry).toMatchObject({ delay_ms: 1000, error: "network flake" }) + }) + + it("does not create debug.log when debug is off", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-watch-nodebug-")) + const runSync = vi.fn(async () => syncSummary()) + const readState = async () => ({ library_id: "lib_1" }) as never + + await runDriveWatch(root, { source: fakeSource(), runSync, readState, once: true }) + + await expect(stat(join(root, ".wspc-drive", "debug.log"))).rejects.toThrow() + }) +}) + +function sha256(content: string): string { + return createHash("sha256").update(content).digest("hex") +} + +type ManifestEntry = { + id: string + path: string + kind: "file" + entry_version: number + current_version_id?: string + content_sha256?: string + size_bytes: number + updated_at: string +} + +function remoteEntry(path: string, content: string, version = 1): ManifestEntry { + return { + id: `ent_${path.replace(/[^a-z0-9]/gi, "_")}_${version}`, + path, + kind: "file", + entry_version: version, + current_version_id: `ver_${version}`, + content_sha256: sha256(content), + size_bytes: Buffer.byteLength(content), + updated_at: "2026-07-13T00:00:00.000Z", + } +} + +function mkApi(entries: ManifestEntry[], downloads: Record = {}): DriveSyncApi { + return { + async getManifest() { + return { + library: { + id: "lib_1", + org_id: "org_1", + name: "Docs", + version: 1, + file_count: entries.length, + storage_bytes: 0, + created_by_user_id: "usr_1", + created_at: 1, + updated_at: 2, + }, + entries, + next_cursor: null, + } as never + }, + async uploadFile(_id, path, body, digest, expectedEntryVersion) { + const content = typeof body === "string" ? body : Buffer.from(await new Response(body).arrayBuffer()).toString("utf8") + return { + entry: remoteEntry(path, content, (expectedEntryVersion ?? 0) + 1), + result: expectedEntryVersion === 0 ? "created" : "updated", + } as never + }, + async downloadFile(_id, path, versionId) { + const key = versionId === undefined ? path : `${path}@${versionId}` + const content = downloads[key] + if (content === undefined) throw new Error(`missing test download: ${key}`) + return new Response(content) + }, + async deleteFile() { + throw new Error("unexpected delete") + }, + } +} + +describe("drive sync debug events", () => { + it("logs decision, conflict, and sync_phases events", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-sync-debug-")) + await initDriveState(root, "lib_1") + await writeFile(join(root, "new.txt"), "hello") + // create_create conflict: local and remote both exist with no base state. + await writeFile(join(root, "clash.bin"), "localbytes") + const remote = remoteEntry("clash.bin", "remote bytes", 1) + const api = mkApi([remote], { "clash.bin@ver_1": "remote bytes" }) + const logger = createDriveDebugLogger(root, { clock: fixedClock }) + + await runDriveSyncOnce(root, api, undefined, undefined, logger) + + const lines = await readDebugLines(root) + const decisions = lines.filter((line) => line.event === "decision") + expect(decisions).toContainEqual( + expect.objectContaining({ path: "new.txt", action: "upload_create", local_sha256: sha256("hello") }), + ) + expect(decisions).toContainEqual( + expect.objectContaining({ + path: "clash.bin", + action: "conflict", + reason: "local_and_remote_without_base", + remote_version_id: "ver_1", + }), + ) + // Unchanged paths must not be logged. + expect(decisions.every((line) => line.action !== "unchanged")).toBe(true) + + const conflict = lines.find((line) => line.event === "conflict") + expect(conflict).toMatchObject({ + path: "clash.bin", + type: "create_create", + strategy: "conflict_copy", + }) + + const phases = lines.find((line) => line.event === "sync_phases") + expect(typeof phases?.scan_ms).toBe("number") + expect(typeof phases?.manifest_ms).toBe("number") + expect(typeof phases?.process_ms).toBe("number") + }) +}) From 63c6bf031f4c59dd99f42c09dce019669c56a19f Mon Sep 17 00:00:00 2001 From: Yuren Ju Date: Mon, 13 Jul 2026 12:07:18 +0900 Subject: [PATCH 2/2] feat(drive): identify sync client with x-wspc-client header --- src/handwritten/commands/drive/api.ts | 6 ++++++ test/handwritten/drive/api.test.ts | 19 +++++++++++++++++++ 2 files changed, 25 insertions(+) diff --git a/src/handwritten/commands/drive/api.ts b/src/handwritten/commands/drive/api.ts index c31c047..8e7cc13 100644 --- a/src/handwritten/commands/drive/api.ts +++ b/src/handwritten/commands/drive/api.ts @@ -29,6 +29,10 @@ async function expectJsonResult(result: JsonResult): Promise { return result.data } +// Lets the server attribute drive file operations to the sync loop instead +// of generic API traffic (drive_file_operation actor: "sync" vs "api"). +const SYNC_CLIENT_HEADERS = { "x-wspc-client": "drive-sync" } as const + function driveContentUrl(baseUrl: string, id: string): URL { const baseWithTrailingSlash = baseUrl.endsWith("/") ? baseUrl : `${baseUrl}/` return new URL(`drive/libraries/${encodeURIComponent(id)}/files/content`, baseWithTrailingSlash) @@ -58,6 +62,7 @@ export async function createDriveApi(opts: DriveApiOptions = {}) { const result = await driveFileDelete({ client: rawClient, path: { id }, + headers: SYNC_CLIENT_HEADERS, body: { path, expected_entry_version: expectedEntryVersion, @@ -83,6 +88,7 @@ export async function createDriveApi(opts: DriveApiOptions = {}) { headers: { "content-type": "application/octet-stream", "x-drive-content-sha256": sha256, + ...SYNC_CLIENT_HEADERS, }, body, }) diff --git a/test/handwritten/drive/api.test.ts b/test/handwritten/drive/api.test.ts index f45fec1..c906e01 100644 --- a/test/handwritten/drive/api.test.ts +++ b/test/handwritten/drive/api.test.ts @@ -146,6 +146,25 @@ describe("createDriveApi", () => { expect(fetchImpl).toHaveBeenCalledOnce() }) + it("uploadFile and deleteFile identify the sync client via x-wspc-client", async () => { + const seenHeaders: Array = [] + const fetchImpl = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const req = mkReq(input, init) + seenHeaders.push(req.headers.get("x-wspc-client")) + const payload = req.method === "PUT" ? DRIVE_UPLOAD : DRIVE_DELETE + return new Response(JSON.stringify(payload), { + status: 200, + headers: { "content-type": "application/json" }, + }) + }) as typeof fetch + + const api = await mkDriveApi(fetchImpl) + await api.uploadFile("lib_1", "notes/hello.txt", new TextEncoder().encode("hello"), "3a6eb7", 2) + await api.deleteFile("lib_1", "notes/hello.txt", 2) + + expect(seenHeaders).toEqual(["drive-sync", "drive-sync"]) + }) + it("throws useful errors for failed JSON SDK calls", async () => { const fetchImpl = vi.fn(async () => new Response(JSON.stringify({ error: { code: "SERVER_ERROR", message: "boom" } }), {