From 3032534f56e329add6f2e1d93e2bccddd5c335df Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 04:08:04 +0000 Subject: [PATCH 1/2] perf(watch): stop watch-mode signature checks from stalling the review UI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Watch mode re-runs `watchSignature` on every debounced file event and every safety poll. The Git backend's implementation ran three `Bun.spawnSync` calls — a full `git diff`, a `rev-parse`, and `ls-files --others` — so during active editing, which is exactly when events fire most, the TUI froze once a second for as long as Git took on the repo. A render-loop proxy ticking at 10ms across five signature checks saw 5 ticks before and 12 after over the same wall time. Total time is unchanged: this does not make the check faster, it stops it freezing the terminal. Three changes, all Effect-independent findings from the migration spike: - Add async Git runners alongside the sync ones. Only the spawn differs; argument building, exit-code policy, and stderr translation stay in shared helpers so the two paths cannot drift. `watchSignature` widens to `string | Promise` — backward compatible, an existing synchronous implementation still satisfies it — and `ExtensionVcsLoadContext` gains an optional `signal`. - Funnel the controller's cancellation checks. `beginCheck` had four `isClosed()` guards and two identical catch blocks, so safety depended on remembering a guard at every new await site. One `runCheckStep` helper now answers both questions in one place, and closing aborts the signal handed to `getSignature`/`refresh` so in-flight work stops rather than running to completion for a result nobody reads. - Extract the named-deadline scheduler. Four deadlines collapsed into one chained timer moves to `watchDeadlines.ts` with its own tests, leaving the controller to talk about phases instead of timer handles. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01YLF3qLZdxVvT87YXBESEib --- .changeset/async-watch-signatures.md | 5 + docs/extensions.md | 5 +- src/core/loaders.test.ts | 4 +- src/core/loaders.ts | 6 +- src/core/vcs/git.ts | 310 ++++++++++++++---- src/core/vcs/types.ts | 4 +- src/core/watch.test.ts | 34 +- src/core/watch.ts | 15 +- src/core/watchController.test.ts | 117 +++++++ src/core/watchController.ts | 195 +++++------ src/core/watchDeadlines.test.ts | 195 +++++++++++ src/core/watchDeadlines.ts | 133 ++++++++ src/extension-api/types.ts | 20 +- src/extensions/default/vcs/git/index.test.ts | 8 +- src/extensions/default/vcs/git/index.ts | 40 +-- src/ui/hooks/useWatchedInput.ts | 67 ++-- .../content/docs/docs/extend/vcs-adapters.md | 2 +- 17 files changed, 923 insertions(+), 237 deletions(-) create mode 100644 .changeset/async-watch-signatures.md create mode 100644 src/core/watchDeadlines.test.ts create mode 100644 src/core/watchDeadlines.ts diff --git a/.changeset/async-watch-signatures.md b/.changeset/async-watch-signatures.md new file mode 100644 index 000000000..94356ed47 --- /dev/null +++ b/.changeset/async-watch-signatures.md @@ -0,0 +1,5 @@ +--- +"hunkdiff": minor +--- + +Watch mode no longer freezes the review UI while it checks for changes, and VCS extensions can now return a promise from `watchSignature`. diff --git a/docs/extensions.md b/docs/extensions.md index f0b27f48e..561b6d16f 100644 --- a/docs/extensions.md +++ b/docs/extensions.md @@ -324,7 +324,10 @@ checkout some other adapter finds. `--watch` works through extension adapters. Each operation may add: - `watchSignature(input, ctx)` — a cheap fingerprint of the reviewed state. - Hunk polls it and reloads when it changes. + Hunk polls it and reloads when it changes. It may return a promise, and + should when it shells out: this runs on every debounced file event and every + safety poll, so a blocking implementation stalls the review UI each time. + `ctx.signal` aborts when the watcher closes. - `watchPlan(input, ctx)` — the filesystem targets that cover that state, so Hunk reacts to events instead of polling on a timer. diff --git a/src/core/loaders.test.ts b/src/core/loaders.test.ts index 288249e9a..96841ba35 100644 --- a/src/core/loaders.test.ts +++ b/src/core/loaders.test.ts @@ -239,7 +239,7 @@ describe("loadAppBootstrap", () => { expect(bootstrap.reloadContext.cwd).toBe(dir); expect(bootstrap.reloadContext.initialWatchSignature).toBeDefined(); - expect(computeWatchSignature(bootstrap.input, bootstrap.reloadContext)).not.toBe( + expect(await computeWatchSignature(bootstrap.input, bootstrap.reloadContext)).not.toBe( bootstrap.reloadContext.initialWatchSignature, ); } finally { @@ -346,7 +346,7 @@ describe("loadAppBootstrap", () => { ); expect(bootstrap.changeset.files[0]?.path).toBe("example.ts"); expect(bootstrap.changeset.files[0]?.agent?.annotations).toHaveLength(1); - expect(computeWatchSignature(bootstrap.input, bootstrap.reloadContext)).toBe( + expect(await computeWatchSignature(bootstrap.input, bootstrap.reloadContext)).toBe( bootstrap.reloadContext.initialWatchSignature!, ); }); diff --git a/src/core/loaders.ts b/src/core/loaders.ts index df1f3c3dd..0c3d59690 100644 --- a/src/core/loaders.ts +++ b/src/core/loaders.ts @@ -455,7 +455,11 @@ export async function loadAppBootstrap( let initialWatchSignature: string | undefined; if (input.options.watch) { try { - initialWatchSignature = computeWatchSignature(input, { cwd, gitExecutable, vcsAdapters }); + initialWatchSignature = await computeWatchSignature(input, { + cwd, + gitExecutable, + vcsAdapters, + }); } catch { // A transient signature failure must not prevent an otherwise valid initial review. } diff --git a/src/core/vcs/git.ts b/src/core/vcs/git.ts index 25b97aa3a..3f91d01fb 100644 --- a/src/core/vcs/git.ts +++ b/src/core/vcs/git.ts @@ -31,6 +31,15 @@ export interface RunGitTextOptions { cwd?: string; gitExecutable?: string; preventOptionalLocks?: boolean; + /** + * Abandon the command when the caller loses interest. + * + * Only the async runners honor this: a synchronous spawn cannot be + * interrupted once it has started. Watch-mode polling passes the controller's + * signal so closing a review kills the `git diff` it started rather than + * waiting for it and discarding the result. + */ + signal?: AbortSignal; } interface RunGitCommandResult { @@ -431,44 +440,107 @@ function translateGitExitFailure(input: GitBackedInput, stderr: string) { return createGenericGitError(input, stderr); } -/** Spawn one Git command and accept only the exit codes the caller declared as non-errors. */ -function runGitCommand({ - input, - args, +/** + * Build the environment-shaped spawn options both runners share. + * + * The stdio literals stay at each call site so Bun can narrow the subprocess + * type to its piped form; only the parts that encode a policy decision live here. + */ +function gitSpawnEnvironment({ cwd = process.cwd(), - gitExecutable = "git", preventOptionalLocks = false, - acceptedExitCodes = [0], -}: RunGitCommandOptions): RunGitCommandResult { +}: Pick) { + return { + cwd, + env: preventOptionalLocks ? { ...process.env, GIT_OPTIONAL_LOCKS: "0" } : undefined, + }; +} + +/** + * Turn one finished Git invocation into a result or a user-facing failure. + * + * Both the sync and async runners funnel through here, so the two paths cannot + * disagree about which exit codes are acceptable or how stderr is translated. + */ +function interpretGitResult( + { input, args, gitExecutable = "git", acceptedExitCodes = [0] }: RunGitCommandOptions, + raw: { stdout?: Uint8Array | null; stderr?: Uint8Array | null; exitCode: number }, +): RunGitCommandResult { + const stdout = Buffer.from(raw.stdout ?? []).toString("utf8"); + const stderr = Buffer.from(raw.stderr ?? []).toString("utf8"); + + if (!acceptedExitCodes.includes(raw.exitCode)) { + throw translateGitExitFailure( + input, + stderr.trim() || `Command failed: ${gitExecutable} ${args.join(" ")}`, + ); + } + + return { stderr, stdout, exitCode: raw.exitCode }; +} + +/** Spawn one Git command and accept only the exit codes the caller declared as non-errors. */ +function runGitCommand(options: RunGitCommandOptions): RunGitCommandResult { + const { input, args, gitExecutable = "git" } = options; let proc: ReturnType; try { proc = Bun.spawnSync([gitExecutable, ...args], { - cwd, + ...gitSpawnEnvironment(options), stdin: "ignore", stdout: "pipe", stderr: "pipe", - env: preventOptionalLocks ? { ...process.env, GIT_OPTIONAL_LOCKS: "0" } : undefined, }); } catch (error) { throw translateGitSpawnFailure(input, error, gitExecutable); } - const stdout = Buffer.from(proc.stdout ?? []).toString("utf8"); - const stderr = Buffer.from(proc.stderr ?? []).toString("utf8"); + return interpretGitResult(options, proc); +} - if (!acceptedExitCodes.includes(proc.exitCode)) { - throw translateGitExitFailure( - input, - stderr.trim() || `Command failed: ${gitExecutable} ${args.join(" ")}`, - ); +/** Start one piped Git subprocess, translating a failed launch the way the sync runner does. */ +function spawnGitAsync(options: RunGitCommandOptions) { + const { input, args, gitExecutable = "git", signal } = options; + + try { + return Bun.spawn([gitExecutable, ...args], { + ...gitSpawnEnvironment(options), + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + signal, + }); + } catch (error) { + throw translateGitSpawnFailure(input, error, gitExecutable); } +} - return { - stderr, - stdout, - exitCode: proc.exitCode, - }; +/** + * Spawn one Git command without blocking the event loop. + * + * The synchronous runner stays the default for one-shot CLI work, where + * blocking is free and simpler. This variant exists for the interactive paths + * that run Git repeatedly while the TUI is drawing — watch-mode polling above + * all — where a synchronous `git diff` stalls rendering and input for as long + * as Git takes on the repo. + */ +async function runGitCommandAsync(options: RunGitCommandOptions): Promise { + const { signal } = options; + const proc = spawnGitAsync(options); + + // Drain both pipes concurrently with the exit wait: a large `git diff` fills + // the stdout pipe buffer, and a process blocked on a full pipe never exits. + const [stdout, stderr, exitCode] = await Promise.all([ + new Response(proc.stdout).bytes(), + new Response(proc.stderr).bytes(), + proc.exited, + ]); + + // An aborted command has no meaningful output; surfacing Git's exit status + // here would report a spurious failure for work the caller already dropped. + signal?.throwIfAborted(); + + return interpretGitResult(options, { stdout, stderr, exitCode }); } /** Run a git command and translate common failures into user-facing Hunk errors. */ @@ -476,6 +548,11 @@ export function runGitText(options: RunGitTextOptions) { return runGitCommand(options).stdout; } +/** Run a git command off the event loop, translating failures the same way `runGitText` does. */ +export async function runGitTextAsync(options: RunGitTextOptions) { + return (await runGitCommandAsync(options)).stdout; +} + const GIT_BOOLEAN_TRUE_VALUES = new Set(["true", "yes", "on", "1", "always"]); const GIT_BOOLEAN_FALSE_VALUES = new Set(["false", "no", "off", "0", "never"]); @@ -542,69 +619,113 @@ export function resolveGitColorMovedOptions( }; } +/** Memoized `rev-parse` answers, so repeated watch polls re-ask Git only for new ranges. */ +const workingTreeGitDiffInputCache = new Map(); + +type WorkingTreeGitDiffOptions = Pick< + RunGitTextOptions, + "cwd" | "gitExecutable" | "preventOptionalLocks" | "signal" +> & { repoRoot?: string }; + /** - * Return whether one `hunk diff` input still compares against the live working tree. + * Decide the parts of the working-tree question that need no subprocess. * - * Plain `hunk diff ` keeps the working tree on one side, so untracked files should still - * appear. Explicit revision-set expressions like `a..b`, `a...b`, or `rev^!` expand into positive - * and negative revisions and should stay commit-to-commit only. + * Returns a definite answer when the input alone settles it, or the arguments + * and cache key the caller needs to ask Git. Both the sync and async variants + * below share this so they cannot drift on which inputs count as working-tree + * reviews or on how the cache is keyed. */ -const workingTreeGitDiffInputCache = new Map(); - -function isWorkingTreeGitDiffInput( +function planWorkingTreeGitDiffCheck( input: ExtensionVcsDiffInput, - { - cwd = process.cwd(), - gitExecutable = "git", - repoRoot, - preventOptionalLocks = false, - }: Pick & { - repoRoot?: string; - } = {}, -) { + { cwd = process.cwd(), gitExecutable = "git", repoRoot }: WorkingTreeGitDiffOptions, +): { settled: boolean } | { settled?: undefined; cacheKey: string; args: string[] } { if (input.staged) { - return false; + return { settled: false }; } if (!input.range) { - return true; + return { settled: true }; } const cacheKey = `${gitExecutable}\0${repoRoot ?? cwd}\0${input.range}`; const cached = workingTreeGitDiffInputCache.get(cacheKey); if (cached !== undefined) { - return cached; + return { settled: cached }; } - const revs = runGitText({ - input, - args: ["rev-parse", "--revs-only", input.range], - cwd, - gitExecutable, - preventOptionalLocks, - }) + return { cacheKey, args: ["rev-parse", "--revs-only", input.range] }; +} + +/** Classify `rev-parse --revs-only` output: one positive revision and no negatives keeps the working tree. */ +function revsIncludeWorkingTree(revsText: string) { + const revs = revsText .split("\n") .map((line) => line.trim()) .filter(Boolean); const positiveRevs = revs.filter((line) => !line.startsWith("^")); const negativeRevs = revs.filter((line) => line.startsWith("^")); - const includesWorkingTree = positiveRevs.length === 1 && negativeRevs.length === 0; + return positiveRevs.length === 1 && negativeRevs.length === 0; +} + +/** + * Return whether one `hunk diff` input still compares against the live working tree. + * + * Plain `hunk diff ` keeps the working tree on one side, so untracked files should still + * appear. Explicit revision-set expressions like `a..b`, `a...b`, or `rev^!` expand into positive + * and negative revisions and should stay commit-to-commit only. + */ +function isWorkingTreeGitDiffInput( + input: ExtensionVcsDiffInput, + options: WorkingTreeGitDiffOptions = {}, +) { + const plan = planWorkingTreeGitDiffCheck(input, options); + if (plan.settled !== undefined) { + return plan.settled; + } + + const includesWorkingTree = revsIncludeWorkingTree( + runGitText({ input, args: plan.args, ...options }), + ); + workingTreeGitDiffInputCache.set(plan.cacheKey, includesWorkingTree); + return includesWorkingTree; +} - workingTreeGitDiffInputCache.set(cacheKey, includesWorkingTree); +async function isWorkingTreeGitDiffInputAsync( + input: ExtensionVcsDiffInput, + options: WorkingTreeGitDiffOptions = {}, +) { + const plan = planWorkingTreeGitDiffCheck(input, options); + if (plan.settled !== undefined) { + return plan.settled; + } + + const includesWorkingTree = revsIncludeWorkingTree( + await runGitTextAsync({ input, args: plan.args, ...options }), + ); + workingTreeGitDiffInputCache.set(plan.cacheKey, includesWorkingTree); return includesWorkingTree; } /** Return whether working-tree review should synthesize untracked files into the patch stream. */ function shouldIncludeUntrackedFiles( input: ExtensionVcsDiffInput, - options: Pick & { - repoRoot?: string; - } = {}, + options: WorkingTreeGitDiffOptions = {}, ) { return input.options.excludeUntracked !== true && isWorkingTreeGitDiffInput(input, options); } +/** Non-blocking `shouldIncludeUntrackedFiles`. */ +async function shouldIncludeUntrackedFilesAsync( + input: ExtensionVcsDiffInput, + options: WorkingTreeGitDiffOptions = {}, +) { + return ( + input.options.excludeUntracked !== true && + (await isWorkingTreeGitDiffInputAsync(input, options)) + ); +} + /** Parse porcelain status output down to repo-root-relative untracked file paths. */ function parseUntrackedFilePaths(statusText: string) { return statusText @@ -695,26 +816,62 @@ export function listGitUntrackedFiles( return []; } - const statusText = runGitText({ - input, - args: buildGitStatusArgs(input), - cwd, - gitExecutable, - preventOptionalLocks, - }); + const untrackedFiles = parseUntrackedFilePaths( + runGitText({ + input, + args: buildGitStatusArgs(input), + cwd, + gitExecutable, + preventOptionalLocks, + }), + ); + + if (untrackedFiles.length === 0) { + return []; + } + + return filterReviewableUntrackedPaths( + untrackedFiles, + repoRoot ?? resolveGitRepoRoot(input, { cwd, gitExecutable, preventOptionalLocks }), + ); +} + +/** Non-blocking `listGitUntrackedFiles`, for callers already running off the event loop. */ +export async function listGitUntrackedFilesAsync( + input: ExtensionVcsDiffInput, + { + cwd = process.cwd(), + repoRoot, + gitExecutable = "git", + preventOptionalLocks = false, + signal, + }: Omit & { repoRoot?: string } = {}, +) { + const gitOptions = { cwd, gitExecutable, preventOptionalLocks, signal }; + + if (!(await shouldIncludeUntrackedFilesAsync(input, gitOptions))) { + return []; + } + + const untrackedFiles = parseUntrackedFilePaths( + await runGitTextAsync({ input, args: buildGitStatusArgs(input), ...gitOptions }), + ); - const untrackedFiles = parseUntrackedFilePaths(statusText); if (untrackedFiles.length === 0) { return []; } - const normalizedRepoRoot = - repoRoot ?? resolveGitRepoRoot(input, { cwd, gitExecutable, preventOptionalLocks }); - return untrackedFiles.filter((filePath) => - isReviewableUntrackedPath(normalizedRepoRoot, filePath), + return filterReviewableUntrackedPaths( + untrackedFiles, + repoRoot ?? (await resolveGitRepoRootAsync(input, gitOptions)), ); } +/** Drop untracked entries Git reports that Hunk cannot synthesize a file patch for. */ +function filterReviewableUntrackedPaths(untrackedFiles: string[], repoRoot: string) { + return untrackedFiles.filter((filePath) => isReviewableUntrackedPath(repoRoot, filePath)); +} + /** Rewrite Git's quoted untracked-file headers into parser-friendly paths. */ export function normalizeUntrackedPatchHeaders(patchText: string, filePath: string) { const safePath = escapeUntrackedPatchPath(filePath); @@ -794,16 +951,29 @@ export function resolveGitMetadata( return { repoRoot, gitDir, commonDir }; } +const REPO_ROOT_ARGS = ["rev-parse", "--show-toplevel"]; + +/** Normalize `rev-parse --show-toplevel` output into a comparable absolute path. */ +function normalizeRepoRootOutput(output: string) { + return normalizePathForOS(output.trim()); +} + +/** Resolve the repository root for one Git-backed review input. */ export function resolveGitRepoRoot( input: GitBackedInput, options: Omit = {}, ) { - const repoRoot = runGitText({ - input, - args: ["rev-parse", "--show-toplevel"], - ...options, - }).trim(); - return normalizePathForOS(repoRoot); + return normalizeRepoRootOutput(runGitText({ input, args: REPO_ROOT_ARGS, ...options })); +} + +/** Non-blocking `resolveGitRepoRoot`, for callers already running off the event loop. */ +export async function resolveGitRepoRootAsync( + input: GitBackedInput, + options: Omit = {}, +) { + return normalizeRepoRootOutput( + await runGitTextAsync({ input, args: REPO_ROOT_ARGS, ...options }), + ); } /** Resolve one commit-ish ref to the exact commit object used for later blob reads. */ diff --git a/src/core/vcs/types.ts b/src/core/vcs/types.ts index 1e4cf32be..78a389b69 100644 --- a/src/core/vcs/types.ts +++ b/src/core/vcs/types.ts @@ -17,6 +17,8 @@ export interface VcsDetection { export interface VcsLoadContext { cwd: string; gitExecutable?: string; + /** Set when the caller can lose interest before the work finishes; see the public contract. */ + signal?: AbortSignal; } export type VcsReviewInput = VcsDiffCommandInput | VcsShowCommandInput | VcsStashShowCommandInput; @@ -30,7 +32,7 @@ export type VcsReviewOperationKind = VcsReviewOperation["kind"]; export interface VcsOperation { load(input: Input, context: VcsLoadContext): Promise; - watchSignature?: (input: Input, context: VcsLoadContext) => string; + watchSignature?: (input: Input, context: VcsLoadContext) => string | Promise; watchPlan?: (input: Input, context: VcsLoadContext) => WatchPlan; } diff --git a/src/core/watch.test.ts b/src/core/watch.test.ts index abbe1eee9..eedccd718 100644 --- a/src/core/watch.test.ts +++ b/src/core/watch.test.ts @@ -67,14 +67,14 @@ afterEach(() => { }); describe("computeWatchSignature", () => { - test("resolves direct files, patch files, and agent sidecars against the supplied cwd", () => { + test("resolves direct files, patch files, and agent sidecars against the supplied cwd", async () => { const dir = createTempRepo("hunk-watch-files-cwd-"); writeFileSync(join(dir, "left.ts"), "one\n"); writeFileSync(join(dir, "right.ts"), "two\n"); writeFileSync(join(dir, "review.patch"), "patch\n"); writeFileSync(join(dir, "agent.json"), "{}\n"); - const direct = computeWatchSignature( + const direct = await computeWatchSignature( { kind: "diff", left: "left.ts", @@ -83,7 +83,7 @@ describe("computeWatchSignature", () => { }, { cwd: dir }, ); - const patch = computeWatchSignature( + const patch = await computeWatchSignature( { kind: "patch", file: "review.patch", options: {} }, { cwd: dir }, ); @@ -94,7 +94,7 @@ describe("computeWatchSignature", () => { expect(patch).toContain(join(dir, "review.patch")); }); - test("does not embed full untracked file contents in git watch signatures", () => { + test("does not embed full untracked file contents in git watch signatures", async () => { const dir = createTempRepo("hunk-watch-untracked-"); writeFileSync(join(dir, "tracked.ts"), "export const tracked = 1;\n"); @@ -105,16 +105,16 @@ describe("computeWatchSignature", () => { const untrackedPath = join(dir, "large-untracked.txt"); writeFileSync(untrackedPath, largeMarker); - const initialSignature = computeWatchSignature(createGitInput(), { cwd: dir }); + const initialSignature = await computeWatchSignature(createGitInput(), { cwd: dir }); writeFileSync(untrackedPath, `${largeMarker}changed`); - const changedSignature = computeWatchSignature(createGitInput(), { cwd: dir }); + const changedSignature = await computeWatchSignature(createGitInput(), { cwd: dir }); expect(initialSignature).not.toContain(largeMarker); expect(changedSignature).not.toContain(largeMarker); expect(changedSignature).not.toEqual(initialSignature); }); - test("ignores untracked file changes when the git input excludes them", () => { + test("ignores untracked file changes when the git input excludes them", async () => { const dir = createTempRepo("hunk-watch-exclude-untracked-"); writeFileSync(join(dir, "tracked.ts"), "export const tracked = 1;\n"); @@ -124,12 +124,12 @@ describe("computeWatchSignature", () => { const untrackedPath = join(dir, "note.txt"); writeFileSync(untrackedPath, "first\n"); - const initialSignature = computeWatchSignature( + const initialSignature = await computeWatchSignature( createGitInput({ options: { excludeUntracked: true } }), { cwd: dir }, ); writeFileSync(untrackedPath, "second\n"); - const changedSignature = computeWatchSignature( + const changedSignature = await computeWatchSignature( createGitInput({ options: { excludeUntracked: true } }), { cwd: dir }, ); @@ -137,7 +137,7 @@ describe("computeWatchSignature", () => { expect(changedSignature).toEqual(initialSignature); }); - test("signs a review through an extension adapter threaded into the context", () => { + test("signs a review through an extension adapter threaded into the context", async () => { const adapter: VcsAdapter = { id: "hg", name: "Mercurial", @@ -160,13 +160,13 @@ describe("computeWatchSignature", () => { options: { mode: "auto", vcs: "hg" }, } satisfies CliInput; - expect(computeWatchSignature(input, { cwd: process.cwd(), vcsAdapters: [adapter] })).toBe( + expect(await computeWatchSignature(input, { cwd: process.cwd(), vcsAdapters: [adapter] })).toBe( "vcs\n---\nhg:working-copy", ); }); - test("rejects unsupported watch operations before invoking adapter signatures", () => { - expect(() => + test("rejects unsupported watch operations before invoking adapter signatures", async () => { + await expect( computeWatchSignature( { kind: "stash-show", @@ -174,10 +174,10 @@ describe("computeWatchSignature", () => { }, { cwd: process.cwd() }, ), - ).toThrow("`hunk stash show` requires Git VCS mode."); + ).rejects.toThrow("`hunk stash show` requires Git VCS mode."); }); - test("tracks untracked file changes when diff compares the working tree against one ref", () => { + test("tracks untracked file changes when diff compares the working tree against one ref", async () => { const dir = createTempRepo("hunk-watch-ref-untracked-"); writeFileSync(join(dir, "tracked.ts"), "export const tracked = 1;\n"); @@ -192,11 +192,11 @@ describe("computeWatchSignature", () => { const untrackedPath = join(dir, "note.txt"); writeFileSync(untrackedPath, "first\n"); - const initialSignature = computeWatchSignature(createGitInput({ range: "main" }), { + const initialSignature = await computeWatchSignature(createGitInput({ range: "main" }), { cwd: dir, }); writeFileSync(untrackedPath, "second\n"); - const changedSignature = computeWatchSignature(createGitInput({ range: "main" }), { + const changedSignature = await computeWatchSignature(createGitInput({ range: "main" }), { cwd: dir, }); diff --git a/src/core/watch.ts b/src/core/watch.ts index f970a91dd..bddca826d 100644 --- a/src/core/watch.ts +++ b/src/core/watch.ts @@ -38,10 +38,19 @@ export interface WatchSignatureContext { gitExecutable?: string; /** Extension-contributed adapters, so a watched review keeps its own backend. */ vcsAdapters?: readonly VcsAdapter[]; + /** Abandon the check when the watcher closes before it finishes. */ + signal?: AbortSignal; } -/** Compute a change-detection signature relative to the source's stable load context. */ -export function computeWatchSignature(input: CliInput, context: WatchSignatureContext) { +/** + * Compute a change-detection signature relative to the source's stable load context. + * + * Asynchronous because watch mode re-runs this on every debounced file event and + * every safety poll: an adapter that shells out — Git computes a whole patch — + * would otherwise stall the terminal UI each time. Adapters may still answer + * synchronously; this awaits either shape. + */ +export async function computeWatchSignature(input: CliInput, context: WatchSignatureContext) { const parts: string[] = [input.kind]; const resolveInputPath = (path: string) => resolve(context.cwd, path); @@ -49,7 +58,7 @@ export function computeWatchSignature(input: CliInput, context: WatchSignatureCo case "vcs": case "show": case "stash-show": - parts.push(vcsPatchSignature(input, context)); + parts.push(await vcsPatchSignature(input, context)); break; case "diff": case "difftool": diff --git a/src/core/watchController.test.ts b/src/core/watchController.test.ts index 572f94f0d..b5252a5b9 100644 --- a/src/core/watchController.test.ts +++ b/src/core/watchController.test.ts @@ -639,4 +639,121 @@ describe("createWatchController", () => { expect(newSource.closes).toBe(0); expect(errors).toEqual([]); }); + + test("aborts the signal it handed to an in-flight signature check on close", async () => { + const clock = new FakeWatchClock(); + const source = fakeSource(); + const pending = deferred(); + let observed: AbortSignal | undefined; + const controller = createWatchController({ + initialSignature: "same", + clock, + createEventSource: source.create, + getSignature: (signal) => { + observed = signal; + return pending.promise; + }, + refresh: () => {}, + }); + + source.event(); + clock.advance(200); + await settle(); + + expect(observed?.aborted).toBe(false); + controller.close(); + // Work started for a check nobody will read can now stop, instead of + // running to completion so its result can be discarded. + expect(observed?.aborted).toBe(true); + pending.resolve("changed"); + await settle(); + }); + + test("closing during a refresh neither applies the signature nor reports the abort", async () => { + const clock = new FakeWatchClock(); + const source = fakeSource(); + const pending = deferred(); + const errors: unknown[] = []; + let refreshSignal: AbortSignal | undefined; + const controller = createWatchController({ + initialSignature: "old", + clock, + createEventSource: source.create, + getSignature: () => "new", + refresh: (signal) => { + refreshSignal = signal; + return pending.promise; + }, + reportError: (error) => errors.push(error), + }); + + source.event(); + clock.advance(200); + await settle(); + expect(refreshSignal?.aborted).toBe(false); + + controller.close(); + pending.reject(Object.assign(new Error("aborted"), { name: "AbortError" })); + await settle(); + + // The refresh never completed, so the new signature must not be recorded, + // and our own abort is not a diagnostic worth showing the user. + expect(controller.getState().appliedSignature).toBe("old"); + expect(errors).toEqual([]); + }); + + test("an abort that is not the controller's own close is still reported", async () => { + const clock = new FakeWatchClock(); + const source = fakeSource(); + const errors: unknown[] = []; + createWatchController({ + initialSignature: "same", + clock, + createEventSource: source.create, + getSignature: () => { + throw Object.assign(new Error("upstream gave up"), { name: "AbortError" }); + }, + refresh: () => {}, + reportError: (error) => errors.push(error), + }); + + source.event(); + clock.advance(200); + await settle(); + + expect(errors).toHaveLength(1); + expect((errors[0] as Error).message).toBe("upstream gave up"); + }); + + test("an asynchronous signature check does not block the caller between polls", async () => { + const clock = new FakeWatchClock(); + const source = fakeSource(); + const gate = deferred(); + let refreshes = 0; + createWatchController({ + initialSignature: "old", + clock, + createEventSource: source.create, + getSignature: () => gate.promise, + refresh: () => { + refreshes++; + }, + }); + + source.event(); + clock.advance(200); + await settle(); + + // The check is parked on the pending signature; events arriving meanwhile + // are held as one trailing check rather than starting a second one. + expect(refreshes).toBe(0); + source.event(); + source.event(); + await settle(); + expect(refreshes).toBe(0); + + gate.resolve("new"); + await settle(); + expect(refreshes).toBe(1); + }); }); diff --git a/src/core/watchController.ts b/src/core/watchController.ts index cd69f3e6e..de12ae411 100644 --- a/src/core/watchController.ts +++ b/src/core/watchController.ts @@ -1,3 +1,9 @@ +import { + createDeadlineScheduler, + defaultDeadlineClock, + type DeadlineClock, +} from "./watchDeadlines"; + export type WatchControllerPhase = | "starting" | "idle" @@ -6,11 +12,13 @@ export type WatchControllerPhase = | "refreshing" | "closed"; -export interface WatchControllerClock { - now(): number; - setTimeout(callback: () => void, delayMs: number): unknown; - clearTimeout(handle: unknown): void; -} +/** + * Test seam for time. + * + * Re-exported from the deadline scheduler, which owns every timer the + * controller arms, so callers keep one clock type to inject. + */ +export type WatchControllerClock = DeadlineClock; export interface WatchEventSource { close(): void; @@ -27,8 +35,17 @@ export const WATCH_EVENT_SOURCE_STARTUP_TIMEOUT_CODE = "HUNK_WATCH_EVENT_SOURCE_ export interface WatchControllerOptions { initialSignature: string; - getSignature: () => string | Promise; - refresh: () => void | Promise; + /** + * Compute the current signature. + * + * Receives a signal that aborts when the controller closes, so an + * implementation that shells out can stop work nobody will read. Prefer an + * asynchronous implementation: this runs on every debounced event and every + * safety poll, and a blocking one stalls the terminal UI each time. + */ + getSignature: (signal: AbortSignal) => string | Promise; + /** Apply a refresh. Receives the same close signal as `getSignature`. */ + refresh: (signal: AbortSignal) => void | Promise; /** A source event arrived and a debounced signature check is now pending. */ onReloadPending?: () => void; clock?: WatchControllerClock; @@ -55,12 +72,6 @@ export interface WatchController { getState(): Readonly; } -const defaultClock: WatchControllerClock = { - now: () => Date.now(), - setTimeout: (callback, delayMs) => setTimeout(callback, delayMs), - clearTimeout: (handle) => clearTimeout(handle as ReturnType), -}; - /** Read an error code without relying on a particular watcher error class. */ function getErrorCode(error: unknown) { if (typeof error !== "object" || error === null || !("code" in error)) return undefined; @@ -75,6 +86,13 @@ function getErrorKey(error: unknown) { return String(error); } +/** Report whether a failure is just this controller's own abort landing on in-flight work. */ +function isAbortError(error: unknown) { + return ( + (error instanceof Error && error.name === "AbortError") || getErrorCode(error) === "ABORT_ERR" + ); +} + /** Build the stable diagnostic reported when an event source cannot establish readiness. */ function createEventSourceStartupTimeoutError(timeoutMs: number) { return Object.assign( @@ -86,9 +104,12 @@ function createEventSourceStartupTimeoutError(timeoutMs: number) { ); } +/** What one awaited step of a check produced, once closure and failure are accounted for. */ +type CheckStep = { done: true; value: T } | { done: false }; + /** Coordinate event hints and periodic checks without coupling to a watcher backend. */ export function createWatchController(options: WatchControllerOptions): WatchController { - const clock = options.clock ?? defaultClock; + const clock = options.clock ?? defaultDeadlineClock; const quietDelayMs = options.quietDelayMs ?? 200; const maximumDelayMs = options.maximumDelayMs ?? 1_000; const healthyCheckMs = options.healthyCheckMs ?? 10_000; @@ -105,14 +126,13 @@ export function createWatchController(options: WatchControllerOptions): WatchCon }; let eventSource: WatchEventSource | undefined; let sourceStatus: "none" | "starting" | "ready" | "closed" = "none"; - let timer: unknown; - let timerDeadline: number | undefined; - let startupDeadline: number | undefined; - let quietDeadline: number | undefined; - let maximumDeadline: number | undefined; - let safetyDeadline: number | undefined; + // Aborted on close, so work started for a check nobody will read can stop + // instead of running to completion and having its result discarded. + const lifetime = new AbortController(); const reportedAt = new Map(); + const deadlines = createDeadlineScheduler({ clock, onDue: () => onTimer() }); + /** Report an error at most once per configured interval for the same error key. */ const reportError = (error: unknown) => { const key = getErrorKey(error); @@ -125,27 +145,12 @@ export function createWatchController(options: WatchControllerOptions): WatchCon const safetyInterval = () => (state.degraded ? degradedCheckMs : healthyCheckMs); - /** Clear the one active chained timeout. */ - const clearTimer = () => { - if (timer !== undefined) clock.clearTimeout(timer); - timer = undefined; - timerDeadline = undefined; - }; - - /** Schedule only the earliest outstanding deadline. */ + /** Arm the next deadline unless a check owns the controller right now. */ const schedule = () => { if (state.phase === "closed" || state.phase === "checking" || state.phase === "refreshing") { return; } - const deadlines = [startupDeadline, quietDeadline, maximumDeadline, safetyDeadline].filter( - (deadline): deadline is number => deadline !== undefined, - ); - if (deadlines.length === 0) return; - const deadline = Math.min(...deadlines); - if (timerDeadline === deadline) return; - clearTimer(); - timerDeadline = deadline; - timer = clock.setTimeout(onTimer, Math.max(0, deadline - clock.now())); + deadlines.arm(); }; /** Test closure across async boundaries without relying on narrowed phase state. */ @@ -158,7 +163,7 @@ export function createWatchController(options: WatchControllerOptions): WatchCon const closeEventSource = () => { if (sourceStatus === "none" || sourceStatus === "closed") return; sourceStatus = "closed"; - startupDeadline = undefined; + deadlines.clear("startup"); const source = eventSource; eventSource = undefined; source?.close(); @@ -167,7 +172,7 @@ export function createWatchController(options: WatchControllerOptions): WatchCon /** Finish work and honor all in-flight hints as one trailing check. */ const finishCheck = () => { if (isClosed()) return; - safetyDeadline = clock.now() + safetyInterval(); + deadlines.set("safety", clock.now() + safetyInterval()); state.phase = "idle"; if (state.dirty) { state.dirty = false; @@ -177,43 +182,53 @@ export function createWatchController(options: WatchControllerOptions): WatchCon schedule(); }; + /** + * Await one step of a check, absorbing the two things that end it early. + * + * Every `await` inside a check has to answer the same two questions — did the + * controller close while we were parked, and did the work fail — and the + * answer is always the same. Funneling them here means adding a step to + * `beginCheck` cannot reintroduce the missing-guard bug that a hand-written + * `if (isClosed()) return` at each site invites. + */ + const runCheckStep = async (work: () => T | Promise): Promise> => { + try { + const value = await work(); + if (isClosed()) return { done: false }; + return { done: true, value }; + } catch (error) { + if (isClosed()) return { done: false }; + // An abort that is not our own close still deserves a report. + if (!isAbortError(error) || !lifetime.signal.aborted) { + reportError(error); + } + finishCheck(); + return { done: false }; + } + }; + /** Run one serialized signature check and refresh only when it changed. */ const beginCheck = async () => { if (state.phase === "closed" || state.phase === "checking" || state.phase === "refreshing") { return; } - clearTimer(); - quietDeadline = undefined; - maximumDeadline = undefined; - safetyDeadline = undefined; + deadlines.disarm(); + deadlines.clear("quiet", "maximum", "safety"); state.phase = "checking"; - let signature: string; - try { - signature = await options.getSignature(); - } catch (error) { - if (isClosed()) return; - reportError(error); - finishCheck(); - return; - } - if (isClosed()) return; - if (signature === state.appliedSignature) { + const signature = await runCheckStep(() => options.getSignature(lifetime.signal)); + if (!signature.done) return; + + if (signature.value === state.appliedSignature) { finishCheck(); return; } state.phase = "refreshing"; - try { - await options.refresh(); - } catch (error) { - if (isClosed()) return; - reportError(error); - finishCheck(); - return; - } - if (isClosed()) return; - state.appliedSignature = signature; + const refreshed = await runCheckStep(() => options.refresh(lifetime.signal)); + if (!refreshed.done) return; + + state.appliedSignature = signature.value; finishCheck(); }; @@ -223,32 +238,28 @@ export function createWatchController(options: WatchControllerOptions): WatchCon closeEventSource(); state.degraded = true; state.phase = "idle"; - quietDeadline = undefined; - maximumDeadline = undefined; - safetyDeadline = clock.now() + degradedCheckMs; + deadlines.clear("quiet", "maximum"); + deadlines.set("safety", clock.now() + degradedCheckMs); reportError(createEventSourceStartupTimeoutError(startupTimeoutMs)); schedule(); }; /** Consume a due startup, debounce, maximum-delay, or safety deadline as one check. */ function onTimer() { - timer = undefined; - timerDeadline = undefined; if (state.phase === "closed") return; - const now = clock.now(); - if (startupDeadline !== undefined && startupDeadline <= now) { + + const due = new Set(deadlines.due(clock.now())); + if (due.has("startup")) { degradeStalledSource(); return; } - const eventDue = - (quietDeadline !== undefined && quietDeadline <= now) || - (maximumDeadline !== undefined && maximumDeadline <= now); - const safetyDue = safetyDeadline !== undefined && safetyDeadline <= now; - if (eventDue || safetyDue) { + + if (due.has("quiet") || due.has("maximum") || due.has("safety")) { void beginCheck(); - } else { - schedule(); + return; } + + schedule(); } /** Treat an event as a hint and retain the first event's maximum deadline. */ @@ -265,8 +276,8 @@ export function createWatchController(options: WatchControllerOptions): WatchCon if (state.phase !== "debouncing") { options.onReloadPending?.(); } - quietDeadline = now + quietDelayMs; - maximumDeadline ??= now + maximumDelayMs; + deadlines.set("quiet", now + quietDelayMs); + deadlines.setIfUnset("maximum", now + maximumDelayMs); state.phase = "debouncing"; schedule(); }; @@ -275,7 +286,7 @@ export function createWatchController(options: WatchControllerOptions): WatchCon const onSourceReady = () => { if (state.phase === "closed" || sourceStatus !== "starting") return; sourceStatus = "ready"; - startupDeadline = undefined; + deadlines.clear("startup"); if (state.phase === "checking" || state.phase === "refreshing") { state.dirty = true; return; @@ -290,20 +301,17 @@ export function createWatchController(options: WatchControllerOptions): WatchCon if (code === "ENOSPC" || code === "EMFILE") { state.degraded = true; closeEventSource(); - safetyDeadline = Math.min( - safetyDeadline ?? Number.POSITIVE_INFINITY, - clock.now() + degradedCheckMs, - ); + deadlines.advance("safety", clock.now() + degradedCheckMs); schedule(); } reportError(error); }; state.phase = "idle"; - safetyDeadline = clock.now() + safetyInterval(); + deadlines.set("safety", clock.now() + safetyInterval()); if (options.createEventSource && !options.pollOnly) { sourceStatus = "starting"; - startupDeadline = clock.now() + startupTimeoutMs; + deadlines.set("startup", clock.now() + startupTimeoutMs); // This JS timer is a secondary guard: FSEvents lock contention can also delay timers, so // bounded native registration remains the primary macOS protection. schedule(); @@ -317,25 +325,24 @@ export function createWatchController(options: WatchControllerOptions): WatchCon else eventSource = createdSource; } catch (error) { sourceStatus = "closed"; - startupDeadline = undefined; + deadlines.clear("startup"); state.degraded = true; - safetyDeadline = clock.now() + degradedCheckMs; + deadlines.set("safety", clock.now() + degradedCheckMs); reportError(error); } } schedule(); return { - /** Stop observation and ignore any later asynchronous completion. */ + /** Stop observation and abandon any in-flight asynchronous work. */ close() { if (state.phase === "closed") return; state.phase = "closed"; state.dirty = false; - quietDeadline = undefined; - maximumDeadline = undefined; - safetyDeadline = undefined; - clearTimer(); + deadlines.clear(); + deadlines.disarm(); closeEventSource(); + lifetime.abort(); }, /** Expose a snapshot for diagnostics without allowing state mutation. */ getState() { diff --git a/src/core/watchDeadlines.test.ts b/src/core/watchDeadlines.test.ts new file mode 100644 index 000000000..083b55165 --- /dev/null +++ b/src/core/watchDeadlines.test.ts @@ -0,0 +1,195 @@ +import { describe, expect, test } from "bun:test"; + +import { createDeadlineScheduler, type DeadlineClock } from "./watchDeadlines"; + +class FakeDeadlineClock implements DeadlineClock { + nowMs = 0; + nextId = 1; + timers = new Map void }>(); + scheduledDelays: number[] = []; + + /** Return deterministic virtual time. */ + now() { + return this.nowMs; + } + + /** Record one virtual timeout. */ + setTimeout(callback: () => void, delayMs: number) { + const id = this.nextId++; + this.scheduledDelays.push(delayMs); + this.timers.set(id, { at: this.nowMs + delayMs, callback }); + return id; + } + + /** Cancel one virtual timeout. */ + clearTimeout(handle: unknown) { + this.timers.delete(handle as number); + } + + /** Advance through every timeout due in the requested interval. */ + advance(ms: number) { + const target = this.nowMs + ms; + for (;;) { + const due = [...this.timers.entries()] + .filter(([, timer]) => timer.at <= target) + .sort((left, right) => left[1].at - right[1].at || left[0] - right[0])[0]; + if (!due) break; + this.nowMs = due[1].at; + this.timers.delete(due[0]); + due[1].callback(); + } + this.nowMs = target; + } +} + +/** Build a scheduler plus a record of every time its timer fired. */ +function createHarness() { + const clock = new FakeDeadlineClock(); + const fires: number[] = []; + const scheduler = createDeadlineScheduler({ + clock, + onDue: () => fires.push(clock.nowMs), + }); + return { clock, fires, scheduler }; +} + +describe("createDeadlineScheduler", () => { + test("arms one timer for the earliest pending deadline", () => { + const { clock, fires, scheduler } = createHarness(); + + scheduler.set("safety", 10_000); + scheduler.set("quiet", 200); + scheduler.set("maximum", 1_000); + scheduler.arm(); + + // Three deadlines, one timer, aimed at the soonest. + expect(clock.timers.size).toBe(1); + expect(clock.scheduledDelays).toEqual([200]); + + clock.advance(200); + expect(fires).toEqual([200]); + }); + + test("re-arming for an unchanged earliest deadline does not churn the timer", () => { + const { clock, scheduler } = createHarness(); + + scheduler.set("quiet", 200); + scheduler.arm(); + scheduler.arm(); + scheduler.arm(); + + expect(clock.scheduledDelays).toEqual([200]); + }); + + test("re-arming after a deadline moves earlier retargets the timer", () => { + const { clock, fires, scheduler } = createHarness(); + + scheduler.set("safety", 10_000); + scheduler.arm(); + scheduler.set("quiet", 200); + scheduler.arm(); + + expect(clock.scheduledDelays).toEqual([10_000, 200]); + clock.advance(200); + expect(fires).toEqual([200]); + }); + + test("reports every deadline that has arrived, and no others", () => { + const { scheduler } = createHarness(); + + scheduler.set("quiet", 200); + scheduler.set("maximum", 1_000); + scheduler.set("safety", 10_000); + + expect(new Set(scheduler.due(1_000))).toEqual(new Set(["quiet", "maximum"])); + expect(scheduler.due(199)).toEqual([]); + }); + + test("setIfUnset keeps the first value so a burst cannot extend its own cap", () => { + const { scheduler } = createHarness(); + + scheduler.setIfUnset("maximum", 1_000); + scheduler.setIfUnset("maximum", 5_000); + + expect(scheduler.due(1_000)).toEqual(["maximum"]); + }); + + test("advance only ever moves a deadline earlier", () => { + const { scheduler } = createHarness(); + + scheduler.set("safety", 10_000); + scheduler.advance("safety", 2_000); + expect(scheduler.due(2_000)).toEqual(["safety"]); + + scheduler.advance("safety", 8_000); + expect(scheduler.due(2_000)).toEqual(["safety"]); + }); + + test("advance sets the deadline when nothing is pending", () => { + const { scheduler } = createHarness(); + + scheduler.advance("safety", 2_000); + expect(scheduler.has("safety")).toBe(true); + expect(scheduler.due(2_000)).toEqual(["safety"]); + }); + + test("clear drops named deadlines, and clears everything when given no names", () => { + const { scheduler } = createHarness(); + + scheduler.set("quiet", 200); + scheduler.set("maximum", 1_000); + scheduler.set("safety", 10_000); + + scheduler.clear("quiet", "maximum"); + expect(scheduler.has("quiet")).toBe(false); + expect(scheduler.has("safety")).toBe(true); + + scheduler.clear(); + expect(scheduler.has("safety")).toBe(false); + }); + + test("arming with nothing pending installs no timer", () => { + const { clock, scheduler } = createHarness(); + + scheduler.arm(); + expect(clock.timers.size).toBe(0); + }); + + test("disarm cancels the timer without discarding deadlines", () => { + const { clock, fires, scheduler } = createHarness(); + + scheduler.set("quiet", 200); + scheduler.arm(); + scheduler.disarm(); + + clock.advance(1_000); + expect(fires).toEqual([]); + // The deadline itself survived, so the owner can arm again later. + expect(scheduler.due(1_000)).toEqual(["quiet"]); + }); + + test("stays disarmed after firing so the owner decides when to wake next", () => { + const { clock, fires, scheduler } = createHarness(); + + scheduler.set("quiet", 200); + scheduler.arm(); + clock.advance(500); + expect(fires).toEqual([200]); + + // A deadline still in the past does not re-fire on its own. + clock.advance(5_000); + expect(fires).toEqual([200]); + }); + + test("a deadline already in the past arms with a non-negative delay", () => { + const { clock, fires, scheduler } = createHarness(); + + clock.advance(1_000); + scheduler.set("safety", 500); + scheduler.arm(); + + expect(clock.scheduledDelays).toEqual([0]); + clock.advance(0); + expect(fires).toEqual([1_000]); + }); +}); diff --git a/src/core/watchDeadlines.ts b/src/core/watchDeadlines.ts new file mode 100644 index 000000000..8cca4f544 --- /dev/null +++ b/src/core/watchDeadlines.ts @@ -0,0 +1,133 @@ +/** + * One timer standing in for several named deadlines. + * + * Watch mode tracks four independent "wake me at" times — event debounce, the + * debounce's hard cap, the periodic safety check, and the event source's + * startup budget — but only ever needs the earliest of them armed. Keeping that + * collapsing here leaves `watchController` free to talk about phases and checks + * instead of timer handles, and lets the arithmetic be tested on its own. + * + * The scheduler is deliberately passive: it never re-arms itself after firing. + * The controller decides what a due deadline means and arms again when it is + * ready for the next one, which is what keeps timers from firing during a check. + */ + +/** Test seam for time, so deadline arithmetic can be driven without real timers. */ +export interface DeadlineClock { + now(): number; + setTimeout(callback: () => void, delayMs: number): unknown; + clearTimeout(handle: unknown): void; +} + +export const defaultDeadlineClock: DeadlineClock = { + now: () => Date.now(), + setTimeout: (callback, delayMs) => setTimeout(callback, delayMs), + clearTimeout: (handle) => clearTimeout(handle as ReturnType), +}; + +/** The deadlines watch mode tracks. Each name may be pending at most once. */ +export type WatchDeadline = "startup" | "quiet" | "maximum" | "safety"; + +export interface DeadlineScheduler { + /** Set one deadline to an absolute time, replacing any pending value. */ + set(name: WatchDeadline, at: number): void; + /** Set one deadline only when it is not already pending, retaining the first value. */ + setIfUnset(name: WatchDeadline, at: number): void; + /** Move one deadline earlier, never later, setting it when nothing is pending. */ + advance(name: WatchDeadline, at: number): void; + /** Drop the named deadlines, or every deadline when called with no names. */ + clear(...names: WatchDeadline[]): void; + /** Report whether one deadline is currently pending. */ + has(name: WatchDeadline): boolean; + /** List the pending deadlines that have arrived by `now`. */ + due(now: number): WatchDeadline[]; + /** Arm the single timer for the earliest pending deadline. */ + arm(): void; + /** Cancel the timer, leaving pending deadlines in place. */ + disarm(): void; +} + +export interface DeadlineSchedulerOptions { + clock?: DeadlineClock; + /** Called once each time the armed timer fires; the scheduler stays disarmed until re-armed. */ + onDue: () => void; +} + +/** Coordinate several named deadlines behind one chained timeout. */ +export function createDeadlineScheduler({ + clock = defaultDeadlineClock, + onDue, +}: DeadlineSchedulerOptions): DeadlineScheduler { + const deadlines = new Map(); + let timer: unknown; + // The time the armed timer is currently aimed at, so re-arming for an + // unchanged earliest deadline does not churn the underlying timeout. + let armedFor: number | undefined; + + const disarm = () => { + if (timer !== undefined) { + clock.clearTimeout(timer); + } + timer = undefined; + armedFor = undefined; + }; + + const fire = () => { + timer = undefined; + armedFor = undefined; + onDue(); + }; + + return { + set(name, at) { + deadlines.set(name, at); + }, + + setIfUnset(name, at) { + if (!deadlines.has(name)) { + deadlines.set(name, at); + } + }, + + advance(name, at) { + const pending = deadlines.get(name); + deadlines.set(name, pending === undefined ? at : Math.min(pending, at)); + }, + + clear(...names) { + if (names.length === 0) { + deadlines.clear(); + return; + } + + for (const name of names) { + deadlines.delete(name); + } + }, + + has(name) { + return deadlines.has(name); + }, + + due(now) { + return [...deadlines].flatMap(([name, at]) => (at <= now ? [name] : [])); + }, + + arm() { + if (deadlines.size === 0) { + return; + } + + const earliest = Math.min(...deadlines.values()); + if (armedFor === earliest) { + return; + } + + disarm(); + armedFor = earliest; + timer = clock.setTimeout(fire, Math.max(0, earliest - clock.now())); + }, + + disarm, + }; +} diff --git a/src/extension-api/types.ts b/src/extension-api/types.ts index b76d73c63..a0a1f82ae 100644 --- a/src/extension-api/types.ts +++ b/src/extension-api/types.ts @@ -435,6 +435,15 @@ export interface ExtensionVcsDetection { export interface ExtensionVcsLoadContext { cwd: string; gitExecutable?: string; + /** + * Set when Hunk may stop caring about the result before it arrives. + * + * Watch-mode polling passes one: closing a review aborts the signature check + * it started. Honoring it is optional — pass it to `spawn` or check it + * between steps to avoid leaving a subprocess running for an answer nobody + * will read. + */ + signal?: AbortSignal; } /** @@ -665,8 +674,15 @@ export interface ExtensionVcsWatchPlan { /** One review operation an adapter implements. */ export interface ExtensionVcsOperation { load(input: Input, context: ExtensionVcsLoadContext): Promise; - /** Optional cheap fingerprint of the reviewed state, for `--watch`. */ - watchSignature?: (input: Input, context: ExtensionVcsLoadContext) => string; + /** + * Optional cheap fingerprint of the reviewed state, for `--watch`. + * + * Returning a promise is preferred: Hunk re-runs this on every debounced file + * event and every safety poll, so a synchronous implementation that shells + * out blocks the terminal UI from drawing for as long as the command takes. + * A synchronous implementation still satisfies the contract. + */ + watchSignature?: (input: Input, context: ExtensionVcsLoadContext) => string | Promise; /** * Optional filesystem targets `--watch` observes instead of polling. * diff --git a/src/extensions/default/vcs/git/index.test.ts b/src/extensions/default/vcs/git/index.test.ts index a7fc87183..c6005de32 100644 --- a/src/extensions/default/vcs/git/index.test.ts +++ b/src/extensions/default/vcs/git/index.test.ts @@ -283,7 +283,7 @@ describe("GitVcsAdapter", () => { ]); }); - test("computes watch signatures for each review operation", () => { + test("computes watch signatures for each review operation", async () => { const repo = createTempRepo("hunk-git-adapter-watch-"); writeFileSync(join(repo, "file.txt"), "one\n"); git(repo, "add", "file.txt"); @@ -293,14 +293,14 @@ describe("GitVcsAdapter", () => { // Measure the working-tree signature while the tree is actually dirty, so the assertion is // meaningful: it must carry the tracked diff and an untracked-file stat signature. - const diffSignature = GitVcsAdapter.operations["working-tree-diff"]!.watchSignature!( + const diffSignature = await GitVcsAdapter.operations["working-tree-diff"]!.watchSignature!( { kind: "vcs", staged: false, options: {} }, { cwd: repo }, ); expect(diffSignature).toContain("diff --git a/file.txt b/file.txt"); expect(diffSignature).toContain("untracked:"); - const showSignature = GitVcsAdapter.operations["revision-show"]!.watchSignature!( + const showSignature = await GitVcsAdapter.operations["revision-show"]!.watchSignature!( { kind: "show", ref: "HEAD", options: {} }, { cwd: repo }, ); @@ -308,7 +308,7 @@ describe("GitVcsAdapter", () => { // Stash the dirty state so a stash entry exists for the stash-show signature. git(repo, "stash", "push", "--include-untracked", "-m", "watch stash"); - const stashSignature = GitVcsAdapter.operations["stash-show"]!.watchSignature!( + const stashSignature = await GitVcsAdapter.operations["stash-show"]!.watchSignature!( { kind: "stash-show", options: {} }, { cwd: repo }, ); diff --git a/src/extensions/default/vcs/git/index.ts b/src/extensions/default/vcs/git/index.ts index 840af91c3..67ac61d10 100644 --- a/src/extensions/default/vcs/git/index.ts +++ b/src/extensions/default/vcs/git/index.ts @@ -7,6 +7,7 @@ import { buildGitStashShowArgs, listGitIgnoredDirectoryRoots, listGitUntrackedFiles, + listGitUntrackedFilesAsync, normalizeUntrackedPatchHeaders, parseGitNumstat, resolveGitColorMovedOptions, @@ -14,7 +15,9 @@ import { resolveGitDiffEndpoints, resolveGitMetadata, resolveGitRepoRoot, + resolveGitRepoRootAsync, runGitText, + runGitTextAsync, runGitUntrackedFileDiffText, shouldSkipLargeTrackedDiff, type GitBackedInput, @@ -323,25 +326,20 @@ export const GitVcsAdapter = { watchPlan(input, { cwd, gitExecutable = "git" }) { return buildGitWatchPlan(input, cwd, gitExecutable); }, - watchSignature(input, { cwd, gitExecutable = "git" }) { - const trackedPatch = runGitText({ + // Watch signatures run on every debounced file event and every safety + // poll while the TUI is drawing, so they use the non-blocking runners. + // The one-shot `load` above stays synchronous, where blocking is free. + async watchSignature(input, { cwd, gitExecutable = "git", signal }) { + const gitOptions = { cwd, gitExecutable, preventOptionalLocks: true, signal }; + const trackedPatch = await runGitTextAsync({ input, args: buildGitDiffArgs(input), - cwd, - gitExecutable, - preventOptionalLocks: true, + ...gitOptions, }); - const repoRoot = resolveGitRepoRoot(input, { - cwd, - gitExecutable, - preventOptionalLocks: true, - }); - const untrackedSignatures = listGitUntrackedFiles(input, { - cwd, - repoRoot, - gitExecutable, - preventOptionalLocks: true, - }).map((filePath) => `untracked:${statSignature(join(repoRoot, filePath))}`); + const repoRoot = await resolveGitRepoRootAsync(input, gitOptions); + const untrackedSignatures = ( + await listGitUntrackedFilesAsync(input, { repoRoot, ...gitOptions }) + ).map((filePath) => `untracked:${statSignature(join(repoRoot, filePath))}`); return [trackedPatch, ...untrackedSignatures].join("\n---\n"); }, }, @@ -374,13 +372,14 @@ export const GitVcsAdapter = { watchPlan(input, { cwd, gitExecutable = "git" }) { return buildGitWatchPlan(input, cwd, gitExecutable); }, - watchSignature(input, { cwd, gitExecutable = "git" }) { - return runGitText({ + watchSignature(input, { cwd, gitExecutable = "git", signal }) { + return runGitTextAsync({ input, args: buildGitShowArgs(input), cwd, gitExecutable, preventOptionalLocks: true, + signal, }); }, }, @@ -413,13 +412,14 @@ export const GitVcsAdapter = { watchPlan(input, { cwd, gitExecutable = "git" }) { return buildGitWatchPlan(input, cwd, gitExecutable); }, - watchSignature(input, { cwd, gitExecutable = "git" }) { - return runGitText({ + watchSignature(input, { cwd, gitExecutable = "git", signal }) { + return runGitTextAsync({ input, args: buildGitStashShowArgs(input), cwd, gitExecutable, preventOptionalLocks: true, + signal, }); }, }, diff --git a/src/ui/hooks/useWatchedInput.ts b/src/ui/hooks/useWatchedInput.ts index 44a391e7a..2ed8efd9c 100644 --- a/src/ui/hooks/useWatchedInput.ts +++ b/src/ui/hooks/useWatchedInput.ts @@ -1,6 +1,7 @@ import { useEffect, useRef } from "react"; import { createWatchController, + type WatchController, type WatchControllerClock, type WatchEventSourceCallbacks, } from "../../core/watchController"; @@ -11,7 +12,10 @@ import type { CliInput, ReloadContext } from "../../core/types"; export interface WatchedInputRuntime { clock?: WatchControllerClock; - getSignature?: (input: CliInput, context: ReloadContext) => string; + getSignature?: ( + input: CliInput, + context: ReloadContext & { signal?: AbortSignal }, + ) => string | Promise; resolvePlan?: (input: CliInput, context: ReloadContext) => WatchPlan | null; createEventSource?: (plan: WatchPlan, callbacks: WatchEventSourceCallbacks) => { close(): void }; } @@ -52,34 +56,55 @@ export function useWatchedInput({ const getSignature = runtime.getSignature ?? computeWatchSignature; let plan: WatchPlan | null; - let initialSignature: string; try { plan = (runtime.resolvePlan ?? resolveWatchPlan)(input, reloadContext); if (!plan) return; - initialSignature = - runtime.getSignature === undefined && reloadContext.initialWatchSignature !== undefined - ? reloadContext.initialWatchSignature - : getSignature(input, reloadContext); } catch (error) { console.error("Failed to initialize watch mode.", error); return; } - const eventSourceFactory = runtime.createEventSource - ? (callbacks: WatchEventSourceCallbacks) => runtime.createEventSource!(plan, callbacks) - : createWatchEventSource(plan); - const controller = createWatchController({ - clock: runtime.clock, - createEventSource: eventSourceFactory, - getSignature: () => getSignature(input, reloadContext), - healthyCheckMs: hasDirectFileContent(plan) ? DIRECT_FILE_WATCH_SAFETY_CHECK_MS : undefined, - initialSignature, - onReloadPending: () => pendingRef.current?.(), - pollOnly: plan.coverage === "poll-only", - refresh: () => refreshRef.current(), - reportError: (error) => console.error("Failed to auto-reload the current diff.", error), - }); + const watchedPlan = plan; + // The controller needs its baseline signature before it can start, and + // computing one may now shell out. Bootstrap normally supplies it, so this + // only awaits on the fallback path; either way the controller is created + // once and torn down by whichever of the two branches below runs last. + let controller: WatchController | undefined; + let cancelled = false; - return () => controller.close(); + const start = (initialSignature: string) => { + if (cancelled) return; + + controller = createWatchController({ + clock: runtime.clock, + createEventSource: runtime.createEventSource + ? (callbacks: WatchEventSourceCallbacks) => + runtime.createEventSource!(watchedPlan, callbacks) + : createWatchEventSource(watchedPlan), + getSignature: (signal) => getSignature(input, { ...reloadContext, signal }), + healthyCheckMs: hasDirectFileContent(watchedPlan) + ? DIRECT_FILE_WATCH_SAFETY_CHECK_MS + : undefined, + initialSignature, + onReloadPending: () => pendingRef.current?.(), + pollOnly: watchedPlan.coverage === "poll-only", + refresh: () => refreshRef.current(), + reportError: (error) => console.error("Failed to auto-reload the current diff.", error), + }); + }; + + if (runtime.getSignature === undefined && reloadContext.initialWatchSignature !== undefined) { + start(reloadContext.initialWatchSignature); + } else { + void Promise.resolve() + .then(() => getSignature(input, reloadContext)) + .then(start) + .catch((error) => console.error("Failed to initialize watch mode.", error)); + } + + return () => { + cancelled = true; + controller?.close(); + }; }, [enabled, input, reloadContext, runtime]); } diff --git a/website/src/content/docs/docs/extend/vcs-adapters.md b/website/src/content/docs/docs/extend/vcs-adapters.md index aff70f9fc..1896278f7 100644 --- a/website/src/content/docs/docs/extend/vcs-adapters.md +++ b/website/src/content/docs/docs/extend/vcs-adapters.md @@ -74,7 +74,7 @@ What detection never overrides is an explicit choice: a `vcs = ""` in Hunk c `--watch` works through extension adapters. Each operation may add: -- `watchSignature(input, ctx)` — a cheap fingerprint of the reviewed state. Hunk polls it and reloads when it changes. +- `watchSignature(input, ctx)` — a cheap fingerprint of the reviewed state. Hunk polls it and reloads when it changes. It may return a promise, and should when it shells out: this runs on every debounced file event and every safety poll, so a blocking implementation stalls the review UI each time. `ctx.signal` aborts when the watcher closes. - `watchPlan(input, ctx)` — the filesystem targets that cover that state, so Hunk reacts to events instead of polling on a timer. ```ts From c00f562da7270d836a6fbd2c05a05dc9b2e028af Mon Sep 17 00:00:00 2001 From: Ben Vinegar Date: Wed, 5 Aug 2026 12:41:17 -0400 Subject: [PATCH 2/2] fix(watch): make asynchronous reloads race-safe --- src/app/sessionBootstrap.test.ts | 32 ++++ src/app/sessionBootstrap.ts | 7 + src/core/loaders.ts | 21 ++- src/core/vcs/git.test.ts | 165 ++++++++++++++++- src/core/vcs/git.ts | 99 +++++++++- src/core/watchController.test.ts | 104 +++++++++++ src/core/watchController.ts | 29 ++- src/extensions/runExtension.test.ts | 39 ++++ src/extensions/runExtension.ts | 4 +- src/session/types.ts | 4 + src/ui/App.tsx | 15 +- src/ui/AppHost.tsx | 249 ++++++++++++++------------ src/ui/AppHost.watch.test.tsx | 134 ++++++++++++++ src/ui/hooks/useWatchedInput.test.tsx | 114 ++++++++++++ src/ui/hooks/useWatchedInput.ts | 12 +- src/ui/lib/extensionReload.test.ts | 92 ++++++++++ src/ui/lib/extensionReload.ts | 43 +++++ src/ui/lib/reloadCoordinator.test.ts | 97 ++++++++++ src/ui/lib/reloadCoordinator.ts | 78 ++++++++ 19 files changed, 1199 insertions(+), 139 deletions(-) create mode 100644 src/ui/hooks/useWatchedInput.test.tsx create mode 100644 src/ui/lib/extensionReload.test.ts create mode 100644 src/ui/lib/extensionReload.ts create mode 100644 src/ui/lib/reloadCoordinator.test.ts create mode 100644 src/ui/lib/reloadCoordinator.ts diff --git a/src/app/sessionBootstrap.test.ts b/src/app/sessionBootstrap.test.ts index 861b2c067..660b3d1af 100644 --- a/src/app/sessionBootstrap.test.ts +++ b/src/app/sessionBootstrap.test.ts @@ -20,6 +20,15 @@ function createTestConfig(input: CliInput): HunkConfigResolution { }; } +/** Create an externally resolved promise for reload-lifetime tests. */ +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((nextResolve) => { + resolve = nextResolve; + }); + return { promise, resolve }; +} + /** Build a loader result with a stable changeset for transform assertions. */ function createTestBootstrap(input: CliInput): AppBootstrap { return { @@ -54,4 +63,27 @@ describe("loadConfiguredSessionBootstrap", () => { expect(result.bootstrap.keybindings).toEqual({ "hunk.review.nextHunk": "]" }); expect(result.bootstrap.viewPreferencesConfigPath).toBe("/tmp/hunk-config.toml"); }); + + test("rejects a bootstrap whose reload signal was aborted while loading", async () => { + const input = createTestInput(); + const load = deferred(); + const controller = new AbortController(); + let loaderSignal: AbortSignal | undefined; + const pending = loadConfiguredSessionBootstrap({ + configured: createTestConfig(input), + cwd: process.cwd(), + signal: controller.signal, + loadAppBootstrapImpl: async (_resolvedInput, options) => { + loaderSignal = options?.signal; + return await load.promise; + }, + }); + + await Promise.resolve(); + expect(loaderSignal).toBe(controller.signal); + controller.abort(); + load.resolve(createTestBootstrap(input)); + + await expect(pending).rejects.toMatchObject({ name: "AbortError" }); + }); }); diff --git a/src/app/sessionBootstrap.ts b/src/app/sessionBootstrap.ts index 49c4694cb..28afa4a8f 100644 --- a/src/app/sessionBootstrap.ts +++ b/src/app/sessionBootstrap.ts @@ -18,6 +18,8 @@ export interface SessionBootstrapOptions { initialThemeMode?: AppBootstrap["initialThemeMode"]; /** Reloads can reopen another directory; initial launch relies on the loader's default cwd. */ loadAtCwd?: boolean; + /** Abort a retired reload before its bootstrap is returned to the host. */ + signal?: AbortSignal; loadAppBootstrapImpl?: typeof loadAppBootstrap; } @@ -42,8 +44,10 @@ export async function loadConfiguredSessionBootstrap({ extensions, initialThemeMode, loadAtCwd = false, + signal, loadAppBootstrapImpl = loadAppBootstrap, }: SessionBootstrapOptions): Promise { + signal?.throwIfAborted(); const sessionThemes = collectSessionCustomThemes( configured.customThemes, extensions?.registry.themes, @@ -69,8 +73,11 @@ export async function loadConfiguredSessionBootstrap({ ...(loadAtCwd ? { cwd } : {}), customThemes: sessionThemes.themes, vcsAdapters: applied.vcsAdapters, + signal, }); + signal?.throwIfAborted(); bootstrap.changeset = await applyExtensionChangesetTransforms(extensions, bootstrap.changeset); + signal?.throwIfAborted(); bootstrap.initialThemeMode = initialThemeMode ?? bootstrap.initialThemeMode; bootstrap.extensions = extensions; bootstrap.viewPreferencesConfigPath = configured.viewPreferencesConfigPath; diff --git a/src/core/loaders.ts b/src/core/loaders.ts index 0c3d59690..95ccc2fc0 100644 --- a/src/core/loaders.ts +++ b/src/core/loaders.ts @@ -41,6 +41,8 @@ interface LoadAppBootstrapOptions { /** Extension-contributed VCS backends this session may load reviews through. */ vcsAdapters?: readonly VcsAdapter[]; gitExecutable?: string; + /** Abort a retired reload before it can publish a stale bootstrap. */ + signal?: AbortSignal; } /** Return the final path segment for display-oriented labels. */ @@ -381,10 +383,12 @@ async function loadVcsChangeset( cwd = process.cwd(), gitExecutable = "git", extensionVcsAdapters: readonly VcsAdapter[] = [], + signal?: AbortSignal, ) { const adapter = getConfiguredVcsAdapter(input.options.vcs, extensionVcsAdapters); const operation = operationFromInput(input); - const result = await loadVcsReview(adapter, operation, { cwd, gitExecutable }); + const result = await loadVcsReview(adapter, operation, { cwd, gitExecutable, signal }); + signal?.throwIfAborted(); const parsedChangeset = normalizePatchChangeset( result.patchText, result.title, @@ -449,8 +453,10 @@ export async function loadAppBootstrap( customThemes, vcsAdapters, gitExecutable = "git", + signal, }: LoadAppBootstrapOptions = {}, ): Promise { + signal?.throwIfAborted(); // Capture before loading content so watch mode can detect mutations that race initial loading. let initialWatchSignature: string | undefined; if (input.options.watch) { @@ -459,13 +465,16 @@ export async function loadAppBootstrap( cwd, gitExecutable, vcsAdapters, + signal, }); } catch { + signal?.throwIfAborted(); // A transient signature failure must not prevent an otherwise valid initial review. } } const agentContext = await loadAgentContext(input.options.agentContext, { cwd }); + signal?.throwIfAborted(); let changeset: Changeset; let repoRoot: string | undefined; @@ -475,7 +484,14 @@ export async function loadAppBootstrap( case "show": case "stash-show": { - const result = await loadVcsChangeset(input, agentContext, cwd, gitExecutable, vcsAdapters); + const result = await loadVcsChangeset( + input, + agentContext, + cwd, + gitExecutable, + vcsAdapters, + signal, + ); changeset = result.changeset; repoRoot = result.repoRoot; } @@ -491,6 +507,7 @@ export async function loadAppBootstrap( break; } + signal?.throwIfAborted(); changeset = { ...changeset, files: orderDiffFiles(changeset.files, agentContext), diff --git a/src/core/vcs/git.test.ts b/src/core/vcs/git.test.ts index c0a4feb30..0c8827a4a 100644 --- a/src/core/vcs/git.test.ts +++ b/src/core/vcs/git.test.ts @@ -1,5 +1,14 @@ import { afterEach, describe, expect, test } from "bun:test"; -import { mkdirSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs"; +import { + chmodSync, + existsSync, + mkdirSync, + mkdtempSync, + readFileSync, + realpathSync, + rmSync, + writeFileSync, +} from "node:fs"; import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import { @@ -13,6 +22,7 @@ import { parseGitNumstat, resolveGitMetadata, runGitText, + runGitTextAsync, shouldSkipLargeTrackedDiff, } from "./git"; import type { VcsDiffCommandInput } from "../types"; @@ -56,6 +66,21 @@ function normalizeComparablePath(path: string) { return realpathSync.native(path).replace(/\\/g, "/"); } +/** Quote one path for the POSIX helper scripts used by Unix-only process tests. */ +function shellQuote(path: string) { + return `'${path.replaceAll("'", `'\\''`)}'`; +} + +/** Report whether a Unix process still exists without sending it a signal. */ +function processExists(pid: number) { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + function makeGitInput(overrides: Partial = {}): VcsDiffCommandInput { return { kind: "vcs", @@ -150,6 +175,144 @@ describe("git command helpers", () => { "Git is required for `hunk diff`, but `definitely-not-a-real-git-binary` was not found in PATH.", ); }); + + test.skipIf(process.platform === "win32")( + "aborting async Git kills descendants after the root process exits", + async () => { + const repoRoot = createTempDir("hunk-git-exited-root-abort-"); + const executablePath = join(repoRoot, "fake-git.sh"); + const pidPath = join(repoRoot, "child-pid"); + writeFileSync( + executablePath, + [ + "#!/bin/sh", + "trap '' HUP TERM", + "sleep 30 &", + "child=$!", + `printf '%s\\n' "$child" > ${shellQuote(pidPath)}`, + "exit 0", + "", + ].join("\n"), + ); + chmodSync(executablePath, 0o755); + + const controller = new AbortController(); + const pending = runGitTextAsync({ + input: makeGitInput(), + args: ["status"], + cwd: repoRoot, + gitExecutable: executablePath, + signal: controller.signal, + }); + const outcome = pending.then( + () => "resolved", + (error: Error) => error.name, + ); + let childPid: number | undefined; + + try { + for (let attempt = 0; attempt < 100 && !existsSync(pidPath); attempt++) { + await Bun.sleep(10); + } + expect(existsSync(pidPath)).toBe(true); + childPid = Number(readFileSync(pidPath, "utf8").trim()); + expect(processExists(childPid)).toBe(true); + // The wrapper has exited, leaving only its child holding stdout open. + await Bun.sleep(50); + + controller.abort(); + expect(await Promise.race([outcome, Bun.sleep(1_000).then(() => "timeout")])).toBe( + "AbortError", + ); + for (let attempt = 0; attempt < 50 && processExists(childPid); attempt++) { + await Bun.sleep(10); + } + expect(processExists(childPid)).toBe(false); + } finally { + controller.abort(); + if (childPid !== undefined) { + try { + process.kill(childPid, "SIGKILL"); + } catch { + // The expected path already terminated the descendant group. + } + } + } + }, + ); + + test.skipIf(process.platform === "win32")( + "aborting an async diff terminates textconv helpers and settles promptly", + async () => { + const repoRoot = createTempRepo("hunk-git-async-abort-"); + const helperPath = join(repoRoot, "slow-textconv.sh"); + const pidPath = join(repoRoot, "textconv-pids"); + writeFileSync( + helperPath, + [ + "#!/bin/sh", + // Reproduces helpers that ignore graceful termination; cancellation + // must still settle and reap the whole process group. + "trap '' TERM", + "sleep 30 &", + "child=$!", + `printf '%s\\n%s\\n' "$$" "$child" > ${shellQuote(pidPath)}`, + 'wait "$child"', + 'cat "$1"', + "", + ].join("\n"), + ); + chmodSync(helperPath, 0o755); + git(repoRoot, "config", "diff.slow.textconv", helperPath); + writeFileSync(join(repoRoot, ".gitattributes"), "*.slow diff=slow\n"); + writeFileSync(join(repoRoot, "example.slow"), "before\n"); + git(repoRoot, "add", ".gitattributes", "example.slow"); + git(repoRoot, "commit", "-m", "initial"); + writeFileSync(join(repoRoot, "example.slow"), "after\n"); + + const input = makeGitInput(); + const controller = new AbortController(); + const pending = runGitTextAsync({ + input, + args: buildGitDiffArgs(input), + cwd: repoRoot, + signal: controller.signal, + }); + const outcome = pending.then( + () => "resolved", + (error: Error) => error.name, + ); + let helperPids: number[] = []; + + try { + for (let attempt = 0; attempt < 100 && !existsSync(pidPath); attempt++) { + await Bun.sleep(10); + } + expect(existsSync(pidPath)).toBe(true); + helperPids = readFileSync(pidPath, "utf8").trim().split("\n").map(Number); + expect(helperPids).toHaveLength(2); + + controller.abort(); + expect(await Promise.race([outcome, Bun.sleep(1_000).then(() => "timeout")])).toBe( + "AbortError", + ); + + for (let attempt = 0; attempt < 50 && helperPids.some(processExists); attempt++) { + await Bun.sleep(10); + } + expect(helperPids.filter(processExists)).toEqual([]); + } finally { + controller.abort(); + for (const pid of helperPids) { + try { + process.kill(pid, "SIGKILL"); + } catch { + // The expected path already terminated the helper process group. + } + } + } + }, + ); }); describe("listGitIgnoredDirectoryRoots", () => { diff --git a/src/core/vcs/git.ts b/src/core/vcs/git.ts index 3f91d01fb..4bbfdf779 100644 --- a/src/core/vcs/git.ts +++ b/src/core/vcs/git.ts @@ -503,18 +503,88 @@ function spawnGitAsync(options: RunGitCommandOptions) { const { input, args, gitExecutable = "git", signal } = options; try { + signal?.throwIfAborted(); return Bun.spawn([gitExecutable, ...args], { ...gitSpawnEnvironment(options), stdin: "ignore", stdout: "pipe", stderr: "pipe", - signal, + // A separate POSIX process group lets cancellation terminate textconv + // and other helpers that inherit Git's output pipes. Windows falls back + // to Bun's direct-process kill below. + detached: process.platform !== "win32", }); } catch (error) { throw translateGitSpawnFailure(input, error, gitExecutable); } } +/** Terminate Git and, where process groups are available, every helper it started. */ +function terminateGitProcess(proc: ReturnType) { + if (process.platform === "win32") { + try { + // Bun's direct kill does not include descendants on Windows; taskkill's + // tree mode covers helpers before they can retain Git's output handles. + const killed = Bun.spawnSync(["taskkill", "/pid", String(proc.pid), "/t", "/f"], { + stdin: "ignore", + stdout: "ignore", + stderr: "ignore", + }); + if (killed.exitCode === 0) return; + } catch { + // Fall through to Bun's direct-process kill when taskkill is unavailable. + } + } else { + try { + // Signal the group even when Git itself has already exited: a textconv + // descendant can remain in that group while retaining Git's pipes. + process.kill(-proc.pid, "SIGKILL"); + return; + } catch { + // The group may have exited between the abort and this signal. + } + } + + if (proc.exitCode !== null) return; + try { + proc.kill(); + } catch { + // Cancellation is best effort once the subprocess has already exited. + } +} + +/** Drain one subprocess pipe, closing the read side immediately on cancellation. */ +async function readGitPipe(stream: ReadableStream, signal?: AbortSignal) { + const reader = stream.getReader(); + const chunks: Uint8Array[] = []; + let byteLength = 0; + const cancel = () => { + void reader.cancel(signal?.reason).catch(() => {}); + }; + + signal?.addEventListener("abort", cancel, { once: true }); + if (signal?.aborted) cancel(); + + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + chunks.push(value); + byteLength += value.byteLength; + } + } finally { + signal?.removeEventListener("abort", cancel); + } + + const output = new Uint8Array(byteLength); + let offset = 0; + for (const chunk of chunks) { + output.set(chunk, offset); + offset += chunk.byteLength; + } + return output; +} + /** * Spawn one Git command without blocking the event loop. * @@ -527,14 +597,27 @@ function spawnGitAsync(options: RunGitCommandOptions) { async function runGitCommandAsync(options: RunGitCommandOptions): Promise { const { signal } = options; const proc = spawnGitAsync(options); + const terminate = () => terminateGitProcess(proc); + signal?.addEventListener("abort", terminate, { once: true }); + if (signal?.aborted) terminate(); - // Drain both pipes concurrently with the exit wait: a large `git diff` fills - // the stdout pipe buffer, and a process blocked on a full pipe never exits. - const [stdout, stderr, exitCode] = await Promise.all([ - new Response(proc.stdout).bytes(), - new Response(proc.stderr).bytes(), - proc.exited, - ]); + let stdout: Uint8Array; + let stderr: Uint8Array; + let exitCode: number; + try { + // Drain both pipes concurrently with the exit wait: a large `git diff` + // fills stdout, and a process blocked on a full pipe never exits. + [stdout, stderr, exitCode] = await Promise.all([ + readGitPipe(proc.stdout, signal), + readGitPipe(proc.stderr, signal), + proc.exited, + ]); + } catch (error) { + signal?.throwIfAborted(); + throw error; + } finally { + signal?.removeEventListener("abort", terminate); + } // An aborted command has no meaningful output; surfacing Git's exit status // here would report a spurious failure for work the caller already dropped. diff --git a/src/core/watchController.test.ts b/src/core/watchController.test.ts index b5252a5b9..f21c0fe64 100644 --- a/src/core/watchController.test.ts +++ b/src/core/watchController.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test"; import { createWatchController, + WATCH_CHECK_CANCELLED_CODE, WATCH_EVENT_SOURCE_STARTUP_TIMEOUT_CODE, type WatchControllerClock, type WatchEventSourceCallbacks, @@ -380,6 +381,69 @@ describe("createWatchController", () => { expect(checks).toBe(2); }); + test("keeps the source startup timeout live during an asynchronous signature check", async () => { + const clock = new FakeWatchClock(); + const source = fakeSource(); + const signature = deferred(); + const errors: unknown[] = []; + const controller = createWatchController({ + initialSignature: "same", + clock, + createEventSource: source.create, + getSignature: () => signature.promise, + refresh: () => {}, + reportError: (error) => errors.push(error), + startupTimeoutMs: 500, + }); + + source.event(); + clock.advance(200); + await settle(); + expect(controller.getState().phase).toBe("checking"); + + // The event check is still parked, but it must not suspend the independent + // bound on source registration. + clock.advance(300); + expect(controller.getState()).toMatchObject({ phase: "checking", degraded: true }); + expect(source.closes).toBe(1); + expect(errors).toHaveLength(1); + expect(errors[0]).toMatchObject({ code: WATCH_EVENT_SOURCE_STARTUP_TIMEOUT_CODE }); + + signature.resolve("same"); + await settle(); + expect(controller.getState().phase).toBe("idle"); + }); + + test("keeps the source startup timeout live during an asynchronous refresh", async () => { + const clock = new FakeWatchClock(); + const source = fakeSource(); + const refresh = deferred(); + const errors: unknown[] = []; + const controller = createWatchController({ + initialSignature: "old", + clock, + createEventSource: source.create, + getSignature: () => "new", + refresh: () => refresh.promise, + reportError: (error) => errors.push(error), + startupTimeoutMs: 500, + }); + + source.event(); + clock.advance(200); + await settle(); + expect(controller.getState().phase).toBe("refreshing"); + + clock.advance(300); + expect(controller.getState()).toMatchObject({ phase: "refreshing", degraded: true }); + expect(source.closes).toBe(1); + expect(errors[0]).toMatchObject({ code: WATCH_EVENT_SOURCE_STARTUP_TIMEOUT_CODE }); + + refresh.resolve(); + await settle(); + expect(controller.getState()).toMatchObject({ phase: "idle", appliedSignature: "new" }); + }); + test("accepts readiness just before an injected startup deadline", async () => { const clock = new FakeWatchClock(); const source = fakeSource(); @@ -725,6 +789,46 @@ describe("createWatchController", () => { expect((errors[0] as Error).message).toBe("upstream gave up"); }); + test("an explicitly superseded check is silent and keeps its baseline retryable", async () => { + const clock = new FakeWatchClock(); + const source = fakeSource(); + const errors: unknown[] = []; + const signatures = [ + Object.assign(new Error("superseded"), { + name: "AbortError", + code: WATCH_CHECK_CANCELLED_CODE, + }), + "new", + ]; + let refreshes = 0; + const controller = createWatchController({ + initialSignature: "old", + clock, + createEventSource: source.create, + getSignature: () => { + const result = signatures.shift(); + if (result instanceof Error) throw result; + return result!; + }, + refresh: () => { + refreshes++; + }, + reportError: (error) => errors.push(error), + }); + + source.event(); + clock.advance(200); + await settle(); + expect(errors).toEqual([]); + expect(controller.getState().appliedSignature).toBe("old"); + + source.event(); + clock.advance(200); + await settle(); + expect(refreshes).toBe(1); + expect(controller.getState().appliedSignature).toBe("new"); + }); + test("an asynchronous signature check does not block the caller between polls", async () => { const clock = new FakeWatchClock(); const source = fakeSource(); diff --git a/src/core/watchController.ts b/src/core/watchController.ts index de12ae411..2be126e02 100644 --- a/src/core/watchController.ts +++ b/src/core/watchController.ts @@ -32,6 +32,8 @@ export interface WatchEventSourceCallbacks { export const DEFAULT_WATCH_EVENT_SOURCE_STARTUP_TIMEOUT_MS = 2_000; export const WATCH_EVENT_SOURCE_STARTUP_TIMEOUT_CODE = "HUNK_WATCH_EVENT_SOURCE_STARTUP_TIMEOUT"; +/** Intentional cancellation of one check without closing its controller. */ +export const WATCH_CHECK_CANCELLED_CODE = "HUNK_WATCH_CHECK_CANCELLED"; export interface WatchControllerOptions { initialSignature: string; @@ -145,11 +147,18 @@ export function createWatchController(options: WatchControllerOptions): WatchCon const safetyInterval = () => (state.degraded ? degradedCheckMs : healthyCheckMs); - /** Arm the next deadline unless a check owns the controller right now. */ + /** Arm the next relevant deadline, retaining source startup coverage during a check. */ const schedule = () => { - if (state.phase === "closed" || state.phase === "checking" || state.phase === "refreshing") { + if (state.phase === "closed") return; + + if (state.phase === "checking" || state.phase === "refreshing") { + // Debounce and safety deadlines pause while a check owns the controller, + // but source registration must remain bounded even when that check awaits. + if (deadlines.has("startup")) deadlines.arm(); + else deadlines.disarm(); return; } + deadlines.arm(); }; @@ -198,8 +207,11 @@ export function createWatchController(options: WatchControllerOptions): WatchCon return { done: true, value }; } catch (error) { if (isClosed()) return { done: false }; - // An abort that is not our own close still deserves a report. - if (!isAbortError(error) || !lifetime.signal.aborted) { + // An abort that is neither our close nor an explicitly superseded check + // still deserves a report. + const checkWasSuperseded = + isAbortError(error) && getErrorCode(error) === WATCH_CHECK_CANCELLED_CODE; + if (!checkWasSuperseded && (!isAbortError(error) || !lifetime.signal.aborted)) { reportError(error); } finishCheck(); @@ -215,6 +227,9 @@ export function createWatchController(options: WatchControllerOptions): WatchCon deadlines.disarm(); deadlines.clear("quiet", "maximum", "safety"); state.phase = "checking"; + // The source may still be registering. Keep that independent deadline live + // while an asynchronous signature check owns every other controller phase. + schedule(); const signature = await runCheckStep(() => options.getSignature(lifetime.signal)); if (!signature.done) return; @@ -235,9 +250,10 @@ export function createWatchController(options: WatchControllerOptions): WatchCon /** Degrade one source that failed to establish readiness before its deadline. */ const degradeStalledSource = () => { if (sourceStatus !== "starting") return; + const checkInFlight = state.phase === "checking" || state.phase === "refreshing"; closeEventSource(); state.degraded = true; - state.phase = "idle"; + if (!checkInFlight) state.phase = "idle"; deadlines.clear("quiet", "maximum"); deadlines.set("safety", clock.now() + degradedCheckMs); reportError(createEventSourceStartupTimeoutError(startupTimeoutMs)); @@ -288,6 +304,9 @@ export function createWatchController(options: WatchControllerOptions): WatchCon sourceStatus = "ready"; deadlines.clear("startup"); if (state.phase === "checking" || state.phase === "refreshing") { + // The startup timer can be the one deadline still armed during a check. + // Readiness retires it while preserving one trailing verification pass. + schedule(); state.dirty = true; return; } diff --git a/src/extensions/runExtension.test.ts b/src/extensions/runExtension.test.ts index 396707cf4..2c3b53d46 100644 --- a/src/extensions/runExtension.test.ts +++ b/src/extensions/runExtension.test.ts @@ -1,4 +1,6 @@ import { describe, expect, test } from "bun:test"; +import { HunkExtensionUserError } from "../extension-api/types"; +import { HunkUserError } from "../core/errors"; import { runExtensionFactory, toInternalVcsAdapter } from "./runExtension"; import { createEmptyExtensionRegistry, type ExtensionLoadIssue } from "./types"; @@ -361,6 +363,43 @@ describe("registerCommand", () => { }); }); +describe("toInternalVcsAdapter operation errors", () => { + test("normalizes an asynchronously rejected watch signature", async () => { + const adapter = toInternalVcsAdapter({ + id: "hg", + name: "Mercurial", + detect: () => null, + operations: { + "working-tree-diff": { + load: async () => ({ + repoRoot: "/repo", + sourceLabel: "/repo", + title: "hg", + patchText: "", + }), + watchSignature: async () => { + await Promise.resolve(); + throw new HunkExtensionUserError("watch failed", { + suggestions: ["Repair the working copy."], + }); + }, + }, + }, + }); + + const operation = adapter.operations["working-tree-diff"]!; + const error = await Promise.resolve( + operation.watchSignature!({ kind: "vcs", staged: false, options: {} }, { cwd: "/repo" }), + ).catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(HunkUserError); + expect(error).toMatchObject({ + message: "watch failed", + suggestions: ["Repair the working copy."], + }); + }); +}); + describe("toInternalVcsAdapter detection ids", () => { test("forces a mismatched detection id back to the registered adapter id", () => { const mismatches: string[] = []; diff --git a/src/extensions/runExtension.ts b/src/extensions/runExtension.ts index 9bac93904..a9c76efff 100644 --- a/src/extensions/runExtension.ts +++ b/src/extensions/runExtension.ts @@ -95,9 +95,9 @@ function toInternalVcsOperation( // Watch support stays optional inward as well as outward: an absent hook is // what tells planning to fall back to signature polling. ...(watchSignature && { - watchSignature(input, context) { + async watchSignature(input, context) { try { - return watchSignature(input, context); + return await watchSignature(input, context); } catch (error) { throw toUserFacingError(error); } diff --git a/src/session/types.ts b/src/session/types.ts index ef67677ab..b96d01190 100644 --- a/src/session/types.ts +++ b/src/session/types.ts @@ -179,6 +179,10 @@ export interface ClearedCommentsResult { export interface ReloadSessionOptions { /** False keeps the mounted App and its in-memory review state. */ resetApp?: boolean; + /** Abort a retired watch reload before it can replace newer session content. */ + signal?: AbortSignal; + /** Content generation that owns an internally triggered watch reload. */ + watchContentGeneration?: number; sourcePath?: string; /** What triggered the reload; forwarded to extension `session_reload` handlers. */ reason?: SessionReloadReason; diff --git a/src/ui/App.tsx b/src/ui/App.tsx index b737bf02b..632ce70ca 100644 --- a/src/ui/App.tsx +++ b/src/ui/App.tsx @@ -171,6 +171,7 @@ function withCurrentViewOptions( /** Orchestrate global app state, layout, navigation, and pane coordination. */ export function App({ bootstrap, + contentGeneration = 0, hostClient, noticeText, onQuit = () => process.exit(0), @@ -178,6 +179,8 @@ export function App({ watchRuntime, }: { bootstrap: AppBootstrap; + /** Host generation that owns this mounted review's watch reloads. */ + contentGeneration?: number; hostClient?: HunkSessionBrokerClient; noticeText?: string | null; onQuit?: () => void; @@ -1241,7 +1244,12 @@ export function App({ /** Rebuild the current diff source while preserving the active app view options. */ const refreshCurrentInput = useCallback( - async (options?: Pick) => { + async ( + options?: Pick< + ReloadSessionOptions, + "reason" | "reloadExtensions" | "signal" | "watchContentGeneration" + >, + ) => { if (!canRefreshCurrentInput) { return; } @@ -1290,8 +1298,9 @@ export function App({ /** Reload because the watcher saw the reviewed source change on disk. */ const refreshWatchedInput = useCallback( - () => refreshCurrentInput({ reason: "watch" }), - [refreshCurrentInput], + (signal: AbortSignal) => + refreshCurrentInput({ reason: "watch", signal, watchContentGeneration: contentGeneration }), + [contentGeneration, refreshCurrentInput], ); /** diff --git a/src/ui/AppHost.tsx b/src/ui/AppHost.tsx index c6f959960..ec905f99b 100644 --- a/src/ui/AppHost.tsx +++ b/src/ui/AppHost.tsx @@ -23,6 +23,8 @@ import type { HunkSessionBrokerClient, ReloadSessionOptions } from "../session/t import { App } from "./App"; import { useStartupNotices } from "./hooks/useStartupNotices"; import type { WatchedInputRuntime } from "./hooks/useWatchedInput"; +import { stageExtensionReload } from "./lib/extensionReload"; +import { createSessionReloadCoordinator } from "./lib/reloadCoordinator"; /** Keep one live Hunk app mounted while allowing daemon-driven session reloads. */ export function AppHost({ @@ -40,6 +42,8 @@ export function AppHost({ }) { const [activeBootstrap, setActiveBootstrap] = useState(bootstrap); const [appVersion, setAppVersion] = useState(0); + const [contentGeneration, setContentGeneration] = useState(0); + const [reloadCoordinator] = useState(() => createSessionReloadCoordinator()); // Extensions outlive App remounts, and a trust grant can replace the whole // load result mid-session, so the host owns them rather than the bootstrap. const extensionsRef = useRef(bootstrap.extensions); @@ -93,136 +97,152 @@ export function AppHost({ const reloadSession = useCallback( async (nextInput: CliInput, options?: ReloadSessionOptions) => { - // Re-run the same startup normalization pipeline used on first launch so reloads honor - // runtime defaults and config layering instead of assuming `nextInput` is already final. - // `sourcePath` matters for daemon-driven reloads that ask Hunk to reopen content from a - // different working directory than the process originally started in. - const runtimeInput = resolveRuntimeCliInput({ - ...nextInput, - options: { - ...nextInput.options, - experimental: launchExperimental, - extensions: launchExtensionsEnabled, - extensionPaths: launchExtensionPaths, - }, - }); - const { cwd } = validateSessionReloadWithinBounds(sessionFileBounds, runtimeInput, { - sourcePath: options?.sourcePath, - }); - const configured = resolveConfiguredCliInput(runtimeInput, { cwd }); - - // Extensions loaded before this pass; used below to tell newly loaded ones apart. - const previouslyLoadedIds = new Set( - (extensionsRef.current?.loaded ?? []).map((extension) => extension.id), - ); - let reloadedExtensions = false; - - if (options?.reloadExtensions || cwd !== extensionsCwdRef.current) { - // A reloaded extension set owns a fresh ephemeral bus. Detach the old - // registry first so delayed callbacks from a retired extension cannot - // keep publishing into listeners that no longer belong to this session. - if (extensionsRef.current) { - extensionsRef.current.registry.emitCustomEvent = undefined; - extensionsRef.current.registry.eventBusPhase = "closed"; - extensionsRef.current.registry.pendingCustomEvents.length = 0; + const reload = reloadCoordinator.begin(options); + + try { + reload.assertCurrent(); + // Re-run the same startup normalization pipeline used on first launch so reloads honor + // runtime defaults and config layering instead of assuming `nextInput` is already final. + // `sourcePath` matters for daemon-driven reloads that ask Hunk to reopen content from a + // different working directory than the process originally started in. + const runtimeInput = resolveRuntimeCliInput({ + ...nextInput, + options: { + ...nextInput.options, + experimental: launchExperimental, + extensions: launchExtensionsEnabled, + extensionPaths: launchExtensionPaths, + }, + }); + const { cwd } = validateSessionReloadWithinBounds(sessionFileBounds, runtimeInput, { + sourcePath: options?.sourcePath, + }); + const configured = resolveConfiguredCliInput(runtimeInput, { cwd }); + + // Extensions loaded before this pass; used below to tell newly loaded ones apart. + const previouslyLoadedIds = new Set( + (extensionsRef.current?.loaded ?? []).map((extension) => extension.id), + ); + let reloadedExtensions = false; + + if (options?.reloadExtensions || cwd !== extensionsCwdRef.current) { + const activeExtensions = extensionsRef.current; + // Keep the active bus open while loading. Only a replacement that + // still owns this reload generation may retire and replace it. + await stageExtensionReload({ + active: activeExtensions, + assertCurrent: reload.assertCurrent, + load: () => + loadStartupExtensions({ + extensions: configured.extensions, + cwd, + cliExtensionPaths: configured.input.options.extensionPaths, + // Reuse the session's notification hub so the mounted toast + // surface keeps receiving notifications across replacement. + notifications: activeExtensions?.notifications, + }), + publish: (replacement) => { + extensionsRef.current = replacement; + extensionsCwdRef.current = cwd; + reloadedExtensions = true; + }, + }); } - // Reuse the session's notification hub so the mounted toast surface keeps - // receiving `ctx.notify` from the extensions this pass loads. - extensionsRef.current = await loadStartupExtensions({ - extensions: configured.extensions, + + const extensions = extensionsRef.current; + const { + applied, + bootstrap: nextBootstrap, + input: reloadInput, + sessionVcs, + } = await loadConfiguredSessionBootstrap({ + configured, cwd, - cliExtensionPaths: configured.input.options.extensionPaths, - notifications: extensionsRef.current?.notifications, + extensions, + loadAtCwd: true, + signal: options?.signal, }); - extensionsCwdRef.current = cwd; - reloadedExtensions = true; - } + // Publish only while this reload still owns the coordinator generation. + // This closes the gap before React can clean up the retired watcher. + const { nextSnapshot, sessionId } = reload.publishContent((nextContentGeneration) => { + if (extensions) { + reportExtensionApplyIssues(applied.issues, extensions.context); + } + nextBootstrap.startupNotices = + sessionVcs.unknownVcsId !== undefined + ? [ + ...(configured.startupNotices ?? []), + // Names the backend the reload really used, detection override included. + createUnknownVcsNotice(sessionVcs.unknownVcsId, String(reloadInput.options.vcs)), + ] + : configured.startupNotices; + const nextSnapshot = createInitialSessionSnapshot(nextBootstrap); - const extensions = extensionsRef.current; - const { - applied, - bootstrap: nextBootstrap, - input: reloadInput, - sessionVcs, - } = await loadConfiguredSessionBootstrap({ - configured, - cwd, - extensions, - loadAtCwd: true, - }); - if (extensions) { - reportExtensionApplyIssues(applied.issues, extensions.context); - } - nextBootstrap.startupNotices = - sessionVcs.unknownVcsId !== undefined - ? [ - ...(configured.startupNotices ?? []), - // Names the backend the reload really used, detection override included. - createUnknownVcsNotice(sessionVcs.unknownVcsId, String(reloadInput.options.vcs)), - ] - : configured.startupNotices; - const nextSnapshot = createInitialSessionSnapshot(nextBootstrap); - - let sessionId = "local-session"; - if (hostClient) { - // Keep the daemon-facing session registration in sync with whatever the UI is about to - // show. Replacing both registration and snapshot here means external session commands see - // the new source, title, and selection baseline immediately after reload. - const nextRegistration = updateSessionRegistration( - hostClient.getRegistration(), - nextBootstrap, - ); - sessionId = nextRegistration.sessionId; - hostClient.replaceSession(nextRegistration, nextSnapshot); - } + let sessionId = "local-session"; + if (hostClient) { + // Keep daemon registration synchronized with the content React is about to show. + const nextRegistration = updateSessionRegistration( + hostClient.getRegistration(), + nextBootstrap, + ); + sessionId = nextRegistration.sessionId; + hostClient.replaceSession(nextRegistration, nextSnapshot); + } - setActiveBootstrap(nextBootstrap); - if (options?.resetApp !== false) { - // Bumping the key forces a full App remount. Callers that pass `resetApp: false` get a - // soft reload that preserves in-memory UI state like selection, filter text, and pane size. - setAppVersion((current) => current + 1); - } + setActiveBootstrap(nextBootstrap); + setContentGeneration(nextContentGeneration); + if (options?.resetApp !== false) { + // A new key resets the App; soft reloads preserve in-memory review state. + setAppVersion((current) => current + 1); + } - if (reloadedExtensions) { - // Extensions this pass loaded for the first time — after a trust grant, or - // after moving into another repository — never saw the mount emit, so they - // get `startup` now that the review UI is showing their changeset. Ordered - // before `session_reload` so an extension's own lifecycle stays in sequence. - const newlyLoadedIds = new Set( - (extensions?.loaded ?? []) - .map((extension) => extension.id) - .filter( - (id) => !previouslyLoadedIds.has(id) && !startedExtensionIdsRef.current.has(id), - ), - ); + return { nextSnapshot, sessionId }; + }); + + if (reloadedExtensions) { + // Extensions this pass loaded for the first time — after a trust grant, or + // after moving into another repository — never saw the mount emit, so they + // get `startup` now that the review UI is showing their changeset. Ordered + // before `session_reload` so an extension's own lifecycle stays in sequence. + const newlyLoadedIds = new Set( + (extensions?.loaded ?? []) + .map((extension) => extension.id) + .filter( + (id) => !previouslyLoadedIds.has(id) && !startedExtensionIdsRef.current.has(id), + ), + ); - for (const id of newlyLoadedIds) { - startedExtensionIdsRef.current.add(id); + for (const id of newlyLoadedIds) { + startedExtensionIdsRef.current.add(id); + } + + emitExtensionEventToExtensions(extensions, "startup", { cwd }, newlyLoadedIds); } - emitExtensionEventToExtensions(extensions, "startup", { cwd }, newlyLoadedIds); - } + emitExtensionEvent(extensions, "session_reload", { + changeset: nextBootstrap.changeset, + reason: options?.reason ?? "daemon", + }); - emitExtensionEvent(extensions, "session_reload", { - changeset: nextBootstrap.changeset, - reason: options?.reason ?? "daemon", - }); - - return { - sessionId, - inputKind: nextBootstrap.input.kind, - title: nextBootstrap.changeset.title, - sourceLabel: nextBootstrap.changeset.sourceLabel, - fileCount: nextBootstrap.changeset.files.length, - selectedFilePath: nextSnapshot.state.selectedFilePath, - selectedHunkIndex: nextSnapshot.state.selectedHunkIndex, - }; + return { + sessionId, + inputKind: nextBootstrap.input.kind, + title: nextBootstrap.changeset.title, + sourceLabel: nextBootstrap.changeset.sourceLabel, + fileCount: nextBootstrap.changeset.files.length, + selectedFilePath: nextSnapshot.state.selectedFilePath, + selectedHunkIndex: nextSnapshot.state.selectedHunkIndex, + }; + } finally { + reload.finish(); + } }, [ hostClient, launchExperimental, launchExtensionsEnabled, launchExtensionPaths, + reloadCoordinator, sessionFileBounds, ], ); @@ -236,6 +256,7 @@ export function AppHost({ () { + let resolve!: (value: T) => void; + const promise = new Promise((nextResolve) => { + resolve = nextResolve; + }); + return { promise, resolve }; +} + async function flush(setup: Awaited>) { await act(async () => { await Promise.resolve(); @@ -169,6 +181,128 @@ describe("watched input lifecycle", () => { } }); + test("superseded watch reloads and retries cannot replace newer daemon content", async () => { + const dir = mkdtempSync(join(process.cwd(), ".hunk-watch-race-ui-")); + const left = join(dir, "before.ts"); + const watchedRight = join(dir, "watched.ts"); + const daemonRight = join(dir, "daemon.ts"); + writeFileSync(left, "export const state = 'before';\n"); + writeFileSync(watchedRight, "export const state = 'initial';\n"); + writeFileSync(daemonRight, "export const state = 'daemon current';\n"); + const bootstrap = await loadAppBootstrap({ + kind: "diff", + left, + right: watchedRight, + options: { mode: "stack", watch: true }, + }); + const watchTransformStarted = deferred(); + const watchTransformFinished = deferred(); + const releaseWatchTransform = deferred(); + const daemonTransformStarted = deferred(); + const releaseDaemonTransform = deferred(); + const extensions = createEmptyExtensionLoadResult(dir); + let transformCalls = 0; + extensions.registry.changesetTransforms.push({ + extensionId: "slow-watch-transform", + async transform(changeset) { + transformCalls++; + if (transformCalls === 1) { + watchTransformStarted.resolve(); + await releaseWatchTransform.promise; + watchTransformFinished.resolve(); + return { ...changeset, title: "retired watch" }; + } + if (transformCalls === 2) { + daemonTransformStarted.resolve(); + await releaseDaemonTransform.promise; + return { ...changeset, title: "daemon current" }; + } + return changeset; + }, + }); + bootstrap.extensions = extensions; + + let registration = createSessionRegistration(bootstrap); + const publishedTitles: string[] = []; + let bridge: { dispatchCommand(message: HunkSessionServerMessage): Promise } | null = + null; + const hostClient = { + setBridge(nextBridge: typeof bridge) { + bridge = nextBridge; + }, + updateSnapshot() {}, + getRegistration() { + return registration; + }, + replaceSession(nextRegistration: typeof registration) { + registration = nextRegistration; + publishedTitles.push(nextRegistration.info.title); + }, + } as unknown as HunkSessionBrokerClient; + const watch = createWatchTestRuntime(); + const setup = await testRender( + , + { width: 120, height: 20 }, + ); + + try { + await flush(setup); + expect(bridge).not.toBeNull(); + writeFileSync(watchedRight, "export const state = 'retired watch';\n"); + watch.setSignature("signature:retired"); + watch.emit(); + await advanceWatch(setup, watch, 200); + await watchTransformStarted.promise; + + const activeBridge = bridge!; + const daemonReload = activeBridge.dispatchCommand({ + type: "command", + requestId: "reload-newer", + command: "reload_session", + input: { + sessionId: registration.sessionId, + nextInput: { + kind: "diff", + left, + right: daemonRight, + options: { mode: "stack" }, + }, + }, + }); + await daemonTransformStarted.promise; + + // Finish the old watch reload after the newer generation has started but + // before React can commit it and abort the old hook. Signal-only guards + // miss this interval; the host generation must reject publication. + releaseWatchTransform.resolve(); + await watchTransformFinished.promise; + await flush(setup); + expect(publishedTitles).not.toContain("retired watch"); + + // The superseded controller retains its old signature so it can retry. + // While the explicit replacement is still pending, that retry must not + // become a newer generation and cancel the replacement. + watch.emit(0); + await advanceWatch(setup, watch, 200); + expect(publishedTitles).not.toContain("retired watch"); + + releaseDaemonTransform.resolve(); + await act(async () => await daemonReload); + await flushUntil( + setup, + () => setup.captureCharFrame().includes("daemon current"), + "the newer daemon reload to render", + ); + expect(publishedTitles).toEqual(["daemon current"]); + expect(setup.captureCharFrame()).not.toContain("retired watch"); + } finally { + releaseWatchTransform.resolve(); + releaseDaemonTransform.resolve(); + await act(async () => setup.renderer.destroy()); + rmSync(dir, { force: true, recursive: true }); + } + }); + test("replacement and unmount dispose observers once while late events remain inert", async () => { const dir = mkdtempSync(join(process.cwd(), ".hunk-watch-dispose-ui-")); const left = join(dir, "before.ts"); diff --git a/src/ui/hooks/useWatchedInput.test.tsx b/src/ui/hooks/useWatchedInput.test.tsx new file mode 100644 index 000000000..1ce75b31f --- /dev/null +++ b/src/ui/hooks/useWatchedInput.test.tsx @@ -0,0 +1,114 @@ +import { describe, expect, test } from "bun:test"; +import { testRender } from "@opentui/react/test-utils"; +import { act } from "react"; +import { createWatchTestClock } from "../../../test/helpers/watchTest"; +import type { CliInput, ReloadContext } from "../../core/types"; +import type { WatchEventSourceCallbacks } from "../../core/watchController"; +import { useWatchedInput, type WatchedInputRuntime } from "./useWatchedInput"; + +const input = { + kind: "diff", + left: "before.ts", + right: "after.ts", + options: { watch: true }, +} satisfies CliInput; +const reloadContext: ReloadContext = { cwd: process.cwd() }; + +/** Create an externally resolved promise for hook-lifetime tests. */ +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((nextResolve) => { + resolve = nextResolve; + }); + return { promise, resolve }; +} + +/** Mount only the watch hook behind a minimal OpenTUI renderable. */ +function WatchHarness({ + refresh, + runtime, +}: { + refresh: (signal: AbortSignal) => void | Promise; + runtime: WatchedInputRuntime; +}) { + useWatchedInput({ enabled: true, input, reloadContext, refresh, runtime }); + return watch; +} + +/** Settle the effect and its bounded promise chain. */ +async function settle(setup: Awaited>) { + await act(async () => { + for (let attempt = 0; attempt < 8; attempt++) await Promise.resolve(); + await setup.renderOnce(); + }); +} + +describe("useWatchedInput cancellation", () => { + test("aborts fallback signature initialization when the hook unmounts", async () => { + let initializationSignal: AbortSignal | undefined; + const runtime: WatchedInputRuntime = { + resolvePlan: () => ({ coverage: "hybrid", targets: [] }), + getSignature: (_input, context) => { + initializationSignal = context.signal; + return new Promise((_resolve, reject) => { + context.signal?.addEventListener("abort", () => reject(context.signal?.reason), { + once: true, + }); + }); + }, + createEventSource: () => { + throw new Error("initialization should not create an event source"); + }, + }; + const setup = await testRender( {}} runtime={runtime} />, { + width: 20, + height: 4, + }); + + await settle(setup); + expect(initializationSignal?.aborted).toBe(false); + await act(async () => setup.renderer.destroy()); + expect(initializationSignal?.aborted).toBe(true); + }); + + test("forwards the controller signal to an in-flight refresh", async () => { + const clock = createWatchTestClock(); + const refreshGate = deferred(); + let signature = "old"; + let sourceCallbacks: WatchEventSourceCallbacks | undefined; + let refreshSignal: AbortSignal | undefined; + const runtime: WatchedInputRuntime = { + clock: clock.clock, + resolvePlan: () => ({ coverage: "hybrid", targets: [] }), + getSignature: () => signature, + createEventSource: (_plan, callbacks) => { + sourceCallbacks = callbacks; + return { close() {} }; + }, + }; + const setup = await testRender( + { + refreshSignal = signal; + return refreshGate.promise; + }} + />, + { width: 20, height: 4 }, + ); + + await settle(setup); + expect(sourceCallbacks).toBeDefined(); + signature = "new"; + await act(async () => { + sourceCallbacks?.onEvent(); + clock.advanceBy(200); + for (let attempt = 0; attempt < 8; attempt++) await Promise.resolve(); + }); + expect(refreshSignal?.aborted).toBe(false); + + await act(async () => setup.renderer.destroy()); + expect(refreshSignal?.aborted).toBe(true); + refreshGate.resolve(); + }); +}); diff --git a/src/ui/hooks/useWatchedInput.ts b/src/ui/hooks/useWatchedInput.ts index 2ed8efd9c..aff4e34a2 100644 --- a/src/ui/hooks/useWatchedInput.ts +++ b/src/ui/hooks/useWatchedInput.ts @@ -43,7 +43,7 @@ export function useWatchedInput({ input: CliInput; onReloadPending?: () => void; reloadContext: ReloadContext; - refresh: () => void | Promise; + refresh: (signal: AbortSignal) => void | Promise; runtime?: WatchedInputRuntime; }) { const refreshRef = useRef(refresh); @@ -70,6 +70,7 @@ export function useWatchedInput({ // only awaits on the fallback path; either way the controller is created // once and torn down by whichever of the two branches below runs last. let controller: WatchController | undefined; + const initialization = new AbortController(); let cancelled = false; const start = (initialSignature: string) => { @@ -88,7 +89,7 @@ export function useWatchedInput({ initialSignature, onReloadPending: () => pendingRef.current?.(), pollOnly: watchedPlan.coverage === "poll-only", - refresh: () => refreshRef.current(), + refresh: (signal) => refreshRef.current(signal), reportError: (error) => console.error("Failed to auto-reload the current diff.", error), }); }; @@ -97,13 +98,16 @@ export function useWatchedInput({ start(reloadContext.initialWatchSignature); } else { void Promise.resolve() - .then(() => getSignature(input, reloadContext)) + .then(() => getSignature(input, { ...reloadContext, signal: initialization.signal })) .then(start) - .catch((error) => console.error("Failed to initialize watch mode.", error)); + .catch((error) => { + if (!cancelled) console.error("Failed to initialize watch mode.", error); + }); } return () => { cancelled = true; + initialization.abort(); controller?.close(); }; }, [enabled, input, reloadContext, runtime]); diff --git a/src/ui/lib/extensionReload.test.ts b/src/ui/lib/extensionReload.test.ts new file mode 100644 index 000000000..e125df81b --- /dev/null +++ b/src/ui/lib/extensionReload.test.ts @@ -0,0 +1,92 @@ +import { describe, expect, test } from "bun:test"; +import { createEmptyExtensionLoadResult } from "../../extensions/types"; +import { stageExtensionReload } from "./extensionReload"; + +/** Create an externally resolved promise for reload-order tests. */ +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((nextResolve) => { + resolve = nextResolve; + }); + return { promise, resolve }; +} + +describe("stageExtensionReload", () => { + test("leaves the active registry open while a replacement is loading", async () => { + const active = createEmptyExtensionLoadResult(); + active.registry.eventBusPhase = "ready"; + active.registry.emitCustomEvent = () => {}; + const replacement = createEmptyExtensionLoadResult(); + const load = deferred(); + let current = true; + const pending = stageExtensionReload({ + active, + load: () => load.promise, + assertCurrent: () => { + if (!current) throw new Error("superseded"); + }, + publish: () => { + throw new Error("a stale replacement must not publish"); + }, + }); + + expect(active.registry.eventBusPhase).toBe("ready"); + expect(active.registry.emitCustomEvent).toBeFunction(); + current = false; + load.resolve(replacement); + + await expect(pending).rejects.toThrow("superseded"); + expect(active.registry.eventBusPhase).toBe("ready"); + expect(active.registry.emitCustomEvent).toBeFunction(); + expect(replacement.registry.eventBusPhase).toBe("closed"); + }); + + test("retires the active registry only after the replacement is current", async () => { + const active = createEmptyExtensionLoadResult(); + active.registry.eventBusPhase = "ready"; + active.registry.emitCustomEvent = () => {}; + const replacement = createEmptyExtensionLoadResult(); + let published = active; + + expect( + await stageExtensionReload({ + active, + load: async () => replacement, + assertCurrent: () => {}, + publish: (next) => { + published = next; + }, + }), + ).toBe(replacement); + expect(published).toBe(replacement); + expect(String(active.registry.eventBusPhase)).toBe("closed"); + expect(active.registry.emitCustomEvent).toBeUndefined(); + }); + + test("publishes before another load continuation can observe the retired registry", async () => { + const active = createEmptyExtensionLoadResult(); + active.registry.eventBusPhase = "ready"; + const replacement = createEmptyExtensionLoadResult(); + const load = deferred(); + let published = active; + const pending = stageExtensionReload({ + active, + load: () => load.promise, + assertCurrent: () => {}, + publish: (next) => { + published = next; + }, + }); + const observed = load.promise.then(() => ({ + published, + activePhase: active.registry.eventBusPhase, + })); + + load.resolve(replacement); + const duringPublication = await observed; + await pending; + + expect(duringPublication.published).toBe(replacement); + expect(duringPublication.activePhase).toBe("closed"); + }); +}); diff --git a/src/ui/lib/extensionReload.ts b/src/ui/lib/extensionReload.ts new file mode 100644 index 000000000..3c5cda9f5 --- /dev/null +++ b/src/ui/lib/extensionReload.ts @@ -0,0 +1,43 @@ +import type { ExtensionLoadResult } from "../../extensions/types"; + +/** Retire one extension registry so none of its delayed callbacks can publish. */ +function retireExtensionResult(result: ExtensionLoadResult | undefined) { + if (!result) return; + result.registry.emitCustomEvent = undefined; + result.registry.eventBusPhase = "closed"; + result.registry.pendingCustomEvents.length = 0; +} + +/** + * Load a replacement without closing the active registry until publication is safe. + * + * A stale replacement is retired instead, leaving the still-active registry + * untouched for whichever newer reload won the generation check. + */ +export async function stageExtensionReload({ + active, + assertCurrent, + load, + publish, +}: { + active: ExtensionLoadResult | undefined; + assertCurrent: () => void; + load: () => Promise; + /** Synchronously replace the host reference in the validated continuation. */ + publish: (replacement: ExtensionLoadResult) => void; +}) { + const replacement = await load(); + try { + assertCurrent(); + } catch (error) { + retireExtensionResult(replacement); + throw error; + } + + // Publish and retire in one synchronous continuation. Returning the result + // for caller-side assignment would open a microtask gap with a closed active + // registry still stored in the host reference. + publish(replacement); + retireExtensionResult(active); + return replacement; +} diff --git a/src/ui/lib/reloadCoordinator.test.ts b/src/ui/lib/reloadCoordinator.test.ts new file mode 100644 index 000000000..d83a0188d --- /dev/null +++ b/src/ui/lib/reloadCoordinator.test.ts @@ -0,0 +1,97 @@ +import { describe, expect, test } from "bun:test"; +import { WATCH_CHECK_CANCELLED_CODE } from "../../core/watchController"; +import { createSessionReloadCoordinator } from "./reloadCoordinator"; + +/** Assert that a reload operation was silently superseded. */ +function expectSuperseded(run: () => unknown) { + try { + run(); + throw new Error("Expected the reload to be superseded."); + } catch (error) { + expect(error).toMatchObject({ + name: "AbortError", + code: WATCH_CHECK_CANCELLED_CODE, + }); + } +} + +describe("createSessionReloadCoordinator", () => { + test("rejects watch events owned by retired content", () => { + const coordinator = createSessionReloadCoordinator(); + const initial = coordinator.begin({ reason: "manual" }); + initial.publishContent(() => {}); + initial.finish(); + + expectSuperseded(() => coordinator.begin({ reason: "watch", watchContentGeneration: 0 })); + coordinator.begin({ reason: "watch", watchContentGeneration: 1 }).finish(); + }); + + test("prevents watch retries from superseding an authoritative reload", () => { + const coordinator = createSessionReloadCoordinator(); + const watch = coordinator.begin({ reason: "watch", watchContentGeneration: 0 }); + const daemon = coordinator.begin({ reason: "daemon" }); + + expectSuperseded(() => watch.assertCurrent()); + expectSuperseded(() => coordinator.begin({ reason: "watch", watchContentGeneration: 0 })); + + daemon.finish(); + coordinator.begin({ reason: "watch", watchContentGeneration: 0 }).finish(); + }); + + test("finishing superseded work does not release newer reload priority", () => { + const coordinator = createSessionReloadCoordinator(); + const first = coordinator.begin({ reason: "manual" }); + const second = coordinator.begin({ reason: "daemon" }); + + first.finish(); + expectSuperseded(() => coordinator.begin({ reason: "watch", watchContentGeneration: 0 })); + + second.finish(); + coordinator.begin({ reason: "watch", watchContentGeneration: 0 }).finish(); + }); + + test("checks cancellation before publishing content", () => { + const coordinator = createSessionReloadCoordinator(); + const controller = new AbortController(); + const reload = coordinator.begin({ + reason: "watch", + signal: controller.signal, + watchContentGeneration: 0, + }); + controller.abort(); + + expect(() => reload.publishContent(() => {})).toThrow(); + reload.finish(); + }); + + test("allows each attempt to publish once and rejects use after finish", () => { + const coordinator = createSessionReloadCoordinator(); + const reload = coordinator.begin({ reason: "manual" }); + + reload.publishContent(() => {}); + expect(() => reload.publishContent(() => {})).toThrow("already published"); + reload.finish(); + expect(() => reload.assertCurrent()).toThrow("finished"); + expect(() => reload.publishContent(() => {})).toThrow("finished"); + reload.finish(); + }); + + test("advances content ownership only after publication succeeds", () => { + const coordinator = createSessionReloadCoordinator(); + const failed = coordinator.begin({ reason: "manual" }); + expect(() => + failed.publishContent(() => { + throw new Error("publication failed"); + }), + ).toThrow("publication failed"); + failed.finish(); + + const retry = coordinator.begin({ reason: "watch", watchContentGeneration: 0 }); + expect(retry.publishContent((generation) => generation)).toBe(1); + retry.finish(); + + const current = coordinator.begin({ reason: "watch", watchContentGeneration: 1 }); + expect(current.publishContent((generation) => generation)).toBe(2); + current.finish(); + }); +}); diff --git a/src/ui/lib/reloadCoordinator.ts b/src/ui/lib/reloadCoordinator.ts new file mode 100644 index 000000000..af9c4e083 --- /dev/null +++ b/src/ui/lib/reloadCoordinator.ts @@ -0,0 +1,78 @@ +import { WATCH_CHECK_CANCELLED_CODE } from "../../core/watchController"; +import type { ReloadSessionOptions } from "../../session/types"; + +/** Build the silent cancellation used when a newer reload supersedes older work. */ +function createSupersededReloadError() { + return Object.assign(new Error("A newer session reload superseded this result."), { + name: "AbortError", + code: WATCH_CHECK_CANCELLED_CODE, + }); +} + +export interface SessionReloadAttempt { + /** Reject work that was aborted or replaced by a newer reload. */ + assertCurrent(): void; + /** Publish one content generation without an async gap in the ownership check. */ + publishContent(publish: (contentGeneration: number) => T): T; + /** Release any authoritative-reload priority still owned by this attempt. */ + finish(): void; +} + +/** + * Coordinate overlapping watch, manual, and daemon reloads. + * + * Explicit reloads are authoritative: they supersede older work and prevent a + * retired watch controller from retrying until the replacement has settled. + */ +export function createSessionReloadCoordinator() { + let currentReloadGeneration = 0; + let currentContentGeneration = 0; + let authoritativeReloadGeneration: number | null = null; + + return { + /** Claim ownership for one reload or reject a stale watch event. */ + begin(options?: ReloadSessionOptions): SessionReloadAttempt { + const isWatchReload = options?.reason === "watch"; + if ( + isWatchReload && + (options.watchContentGeneration !== currentContentGeneration || + authoritativeReloadGeneration !== null) + ) { + throw createSupersededReloadError(); + } + + const reloadGeneration = ++currentReloadGeneration; + if (!isWatchReload) authoritativeReloadGeneration = reloadGeneration; + let published = false; + let finished = false; + + const assertCurrent = () => { + if (finished) throw new Error("Cannot use a finished session reload attempt."); + options?.signal?.throwIfAborted(); + if (reloadGeneration !== currentReloadGeneration) { + throw createSupersededReloadError(); + } + }; + + return { + assertCurrent, + publishContent(publish: (contentGeneration: number) => T) { + assertCurrent(); + if (published) throw new Error("Session reload content was already published."); + const nextContentGeneration = currentContentGeneration + 1; + const result = publish(nextContentGeneration); + currentContentGeneration = nextContentGeneration; + published = true; + return result; + }, + finish() { + if (finished) return; + finished = true; + if (authoritativeReloadGeneration === reloadGeneration) { + authoritativeReloadGeneration = null; + } + }, + }; + }, + }; +}