From 0b5d36ec45d943a3789c6362ea09d1df74fceb23 Mon Sep 17 00:00:00 2001 From: Badcuban <108198679+badcuban@users.noreply.github.com> Date: Sun, 2 Aug 2026 01:51:35 -0400 Subject: [PATCH 1/2] Revert "Revert self-healing status subscriptions until CI-stable" This reverts commit c36a7cd6102f37474d3e32af63651832df419722. --- apps/web/src/components/ChatView.browser.tsx | 1 + .../components/KeybindingsToast.browser.tsx | 1 + .../SourceControlPanel.browser.tsx | 1 + .../source-control/SourceControlPanel.tsx | 5 + apps/web/src/lib/gitStatusState.test.ts | 266 ++++++++++++++++- apps/web/src/lib/gitStatusState.ts | 270 ++++++++++++++---- apps/web/src/observability/clientTracing.ts | 47 +++ .../src/observability/streamDiagnostics.ts | Bin 0 -> 1918 bytes apps/web/src/rpc/wsTransport.ts | 21 ++ 9 files changed, 560 insertions(+), 52 deletions(-) create mode 100644 apps/web/src/observability/streamDiagnostics.ts diff --git a/apps/web/src/components/ChatView.browser.tsx b/apps/web/src/components/ChatView.browser.tsx index 87e7dfdc0..a8ce4ce9c 100644 --- a/apps/web/src/components/ChatView.browser.tsx +++ b/apps/web/src/components/ChatView.browser.tsx @@ -84,6 +84,7 @@ vi.mock("../lib/gitStatusState", () => ({ GIT_STATUS_STALE_MESSAGE: "Source control status isn't updating.", useGitStatus: () => ({ data: null, error: null, cause: null, isPending: false }), useGitStatuses: () => new Map(), + rebuildGitStatusSubscription: () => undefined, refreshGitStatus: () => Promise.resolve(null), refreshLocalGitStatus: () => Promise.resolve(null), resetGitStatusStateForTests: () => undefined, diff --git a/apps/web/src/components/KeybindingsToast.browser.tsx b/apps/web/src/components/KeybindingsToast.browser.tsx index 584f11323..f1b0bbc7a 100644 --- a/apps/web/src/components/KeybindingsToast.browser.tsx +++ b/apps/web/src/components/KeybindingsToast.browser.tsx @@ -46,6 +46,7 @@ vi.mock("../lib/gitStatusState", () => ({ GIT_STATUS_STALE_MESSAGE: "Source control status isn't updating.", useGitStatus: () => ({ data: null, error: null, cause: null, isPending: false }), useGitStatuses: () => new Map(), + rebuildGitStatusSubscription: () => undefined, refreshGitStatus: () => Promise.resolve(null), refreshLocalGitStatus: () => Promise.resolve(null), resetGitStatusStateForTests: () => undefined, diff --git a/apps/web/src/components/source-control/SourceControlPanel.browser.tsx b/apps/web/src/components/source-control/SourceControlPanel.browser.tsx index 1d4e666cc..5702b9c40 100644 --- a/apps/web/src/components/source-control/SourceControlPanel.browser.tsx +++ b/apps/web/src/components/source-control/SourceControlPanel.browser.tsx @@ -62,6 +62,7 @@ vi.mock("~/lib/gitStatusState", () => ({ isPending: false, }), useGitStatuses: () => new Map(), + rebuildGitStatusSubscription: () => undefined, refreshGitStatus: gitStatusMock.refreshGitStatus, refreshLocalGitStatus: gitStatusMock.refreshLocalGitStatus, resetGitStatusStateForTests: () => { diff --git a/apps/web/src/components/source-control/SourceControlPanel.tsx b/apps/web/src/components/source-control/SourceControlPanel.tsx index 0cc806069..51dab15eb 100644 --- a/apps/web/src/components/source-control/SourceControlPanel.tsx +++ b/apps/web/src/components/source-control/SourceControlPanel.tsx @@ -98,6 +98,7 @@ import { } from "~/lib/gitReactQuery"; import { GIT_STATUS_STALE_MESSAGE, + rebuildGitStatusSubscription, refreshGitStatus, refreshLocalGitStatus, useGitStatus, @@ -2571,6 +2572,10 @@ export function SourceControlPanel({ return; } setIsManualRefreshPending(true); + // Refreshing repairs the data over the unary RPC; rebuilding repairs the + // push stream that stopped delivering it. Retry has to do both, or the + // panel goes stale again on the next change nobody hears about. + rebuildGitStatusSubscription({ environmentId, cwd }); void refreshGitStatus({ environmentId, cwd }, undefined, { force: true }) .then(() => queryClient.invalidateQueries({ diff --git a/apps/web/src/lib/gitStatusState.test.ts b/apps/web/src/lib/gitStatusState.test.ts index 1328293cc..ecf9f1541 100644 --- a/apps/web/src/lib/gitStatusState.test.ts +++ b/apps/web/src/lib/gitStatusState.test.ts @@ -4,7 +4,9 @@ import { afterEach, describe, expect, it, vi } from "vite-plus/test"; import type { WsRpcClient } from "../rpc/wsRpcClient"; import { resetAppAtomRegistryForTests } from "../rpc/atomRegistry"; import { + GIT_STATUS_STALE_MESSAGE, getGitStatusSnapshot, + rebuildGitStatusSubscription, resetGitStatusStateForTests, refreshLocalGitStatus, refreshGitStatus, @@ -73,6 +75,51 @@ function emitGitStatus(event: VcsStatusResult) { } } +interface StreamHooks { + onResubscribe?: () => void; + onRetry?: (error: unknown, attempt: number) => void; +} + +/** + * A client whose stream can be driven directly: emit snapshots, replay the + * transport's resubscribe/retry hooks, and count how many times the module + * opened a *new* subscription (the observable signal that a rebuild happened). + */ +function createControllableGitStatusClient() { + const listeners = new Set<(event: VcsStatusResult) => void>(); + let hooks: StreamHooks = {}; + const onStatus = vi.fn( + ( + _input: { cwd: string }, + listener: (event: VcsStatusResult) => void, + options?: StreamHooks, + ) => { + hooks = options ?? {}; + return registerListener(listeners, listener); + }, + ); + + return { + client: { + refreshStatus: vi.fn(async () => BASE_STATUS), + refreshLocalStatus: vi.fn(async (input: { cwd: string }) => ({ + ...BASE_STATUS, + refName: `${input.cwd}-local-refreshed`, + })), + onStatus, + }, + onStatus, + subscriberCount: () => listeners.size, + emit: (event: VcsStatusResult) => { + for (const listener of listeners) { + listener(event); + } + }, + resubscribe: () => hooks.onResubscribe?.(), + retry: (error: unknown, attempt: number) => hooks.onRetry?.(error, attempt), + }; +} + function createRegisteredGitStatusClient(environmentId: EnvironmentId) { const listeners = new Set<(event: VcsStatusResult) => void>(); const client = { @@ -267,13 +314,18 @@ describe("gitStatusState", () => { const release = watchGitStatus(TARGET, gitClient); emitGitStatus(BASE_STATUS); + const healthy = getGitStatusSnapshot(TARGET); const refreshed = await refreshLocalGitStatus(TARGET, gitClient); expect(gitClient.onStatus).toHaveBeenCalledOnce(); expect(gitClient.refreshLocalStatus).toHaveBeenCalledWith({ cwd: "/repo" }); expect(gitClient.refreshStatus).not.toHaveBeenCalled(); expect(refreshed).toEqual({ ...BASE_STATUS, refName: "/repo-local-refreshed" }); - expect(getGitStatusSnapshot(TARGET)).toEqual({ + // Reference equality, not just deep equality: a poll landing on a healthy + // atom must not write at all, or it races the live stream and re-renders + // every consumer every 5 seconds. + expect(getGitStatusSnapshot(TARGET)).toBe(healthy); + expect(healthy).toEqual({ data: BASE_STATUS, error: null, cause: null, @@ -407,7 +459,9 @@ describe("gitStatusState", () => { expect(getGitStatusSnapshot(TARGET).isPending).toBe(true); - await vi.advanceTimersByTimeAsync(20_000); + // Two silent windows are spent rebuilding the stream; the notice only + // appears after the third. + await vi.advanceTimersByTimeAsync(60_000); const snapshot = getGitStatusSnapshot(TARGET); expect(snapshot.isPending).toBe(false); @@ -468,8 +522,9 @@ describe("gitStatusState", () => { }; const release = watchGitStatus(TARGET, client); - // The stream never delivers; the watchdog marks the status broken. - await vi.advanceTimersByTimeAsync(20_000); + // The stream never delivers; the watchdog rebuilds twice, then marks + // the status broken. + await vi.advanceTimersByTimeAsync(60_000); expect(getGitStatusSnapshot(TARGET).error).not.toBeNull(); // Retry goes over the unary path, which still works when only the @@ -509,4 +564,207 @@ describe("gitStatusState", () => { void refreshGitStatus(TARGET, stalledClient, { force: true }).catch(() => undefined); expect(stalledClient.refreshStatus).toHaveBeenCalledTimes(2); }); + + it("rebuilds a silent subscription twice before showing the stale notice", async () => { + vi.useFakeTimers(); + try { + const harness = createControllableGitStatusClient(); + const release = watchGitStatus(TARGET, harness.client); + + expect(harness.onStatus).toHaveBeenCalledOnce(); + + // The reproduced failure is a lost opening snapshot on an otherwise + // working socket, so each silent window buys a fresh subscribe rather + // than a notice the user has to act on. + await vi.advanceTimersByTimeAsync(20_000); + expect(harness.onStatus).toHaveBeenCalledTimes(2); + expect(harness.subscriberCount()).toBe(1); + expect(getGitStatusSnapshot(TARGET).error).toBeNull(); + expect(getGitStatusSnapshot(TARGET).isPending).toBe(true); + + await vi.advanceTimersByTimeAsync(20_000); + expect(harness.onStatus).toHaveBeenCalledTimes(3); + expect(getGitStatusSnapshot(TARGET).error).toBeNull(); + + await vi.advanceTimersByTimeAsync(20_000); + expect(getGitStatusSnapshot(TARGET).error?.message).toBe(GIT_STATUS_STALE_MESSAGE); + + // Bounded: once the notice is up nothing keeps resubscribing, or an + // environment where subscribing can never succeed loops forever. + await vi.advanceTimersByTimeAsync(120_000); + expect(harness.onStatus).toHaveBeenCalledTimes(3); + + release(); + } finally { + vi.useRealTimers(); + } + }); + + it("converges when the poll lane heals while the stream keeps failing", async () => { + vi.useFakeTimers(); + try { + const harness = createControllableGitStatusClient(); + const release = watchGitStatus(TARGET, harness.client); + + // Stream never delivers; polls always succeed. This is the browser-test + // environment and the real dead-stream case at once: after the first + // poll heals the atom, the fighting lanes must go quiet — a stale mark + // that re-breaks a poll-fed atom alternates broken↔healthy forever. + harness.retry(new Error("boom"), 2); + expect(getGitStatusSnapshot(TARGET).error).not.toBeNull(); + + await refreshLocalGitStatus(TARGET, harness.client, { force: true }); + const healed = getGitStatusSnapshot(TARGET); + expect(healed.data).not.toBeNull(); + expect(healed.error).toBeNull(); + + for (let round = 0; round < 3; round += 1) { + harness.retry(new Error("boom"), 3 + round); + await vi.advanceTimersByTimeAsync(5_000); + await refreshLocalGitStatus(TARGET, harness.client, { force: true }); + } + expect(getGitStatusSnapshot(TARGET)).toBe(healed); + + release(); + } finally { + vi.useRealTimers(); + } + }); + + it("does not touch the atom while rebuilding a stream that never delivers", async () => { + vi.useFakeTimers(); + try { + const harness = createControllableGitStatusClient(); + const release = watchGitStatus(TARGET, harness.client); + + // One pending write at mount, one stale write when the budget runs out, + // and nothing in between: every rebuild-cycle write re-renders every + // consumer, which is enough churn to destabilize the app while a + // connection is down (this exact pattern broke CI's browser suite). + const pending = getGitStatusSnapshot(TARGET); + expect(pending.isPending).toBe(true); + + await vi.advanceTimersByTimeAsync(20_000); + expect(getGitStatusSnapshot(TARGET)).toBe(pending); + await vi.advanceTimersByTimeAsync(20_000); + expect(getGitStatusSnapshot(TARGET)).toBe(pending); + + await vi.advanceTimersByTimeAsync(20_000); + const stale = getGitStatusSnapshot(TARGET); + expect(stale.error?.message).toBe(GIT_STATUS_STALE_MESSAGE); + + await vi.advanceTimersByTimeAsync(120_000); + expect(getGitStatusSnapshot(TARGET)).toBe(stale); + + release(); + } finally { + vi.useRealTimers(); + } + }); + + it("gives a stream that delivered again a fresh rebuild budget", async () => { + vi.useFakeTimers(); + try { + const harness = createControllableGitStatusClient(); + const release = watchGitStatus(TARGET, harness.client); + + await vi.advanceTimersByTimeAsync(60_000); + expect(harness.onStatus).toHaveBeenCalledTimes(3); + expect(getGitStatusSnapshot(TARGET).error).not.toBeNull(); + + harness.emit(BASE_STATUS); + expect(getGitStatusSnapshot(TARGET)).toEqual({ + data: BASE_STATUS, + error: null, + cause: null, + isPending: false, + }); + + // The transport restarting the stream reopens the watchdog. With the + // budget reset by the delivered event, the next silence rebuilds again + // instead of jumping straight back to the notice. + harness.resubscribe(); + await vi.advanceTimersByTimeAsync(20_000); + expect(harness.onStatus).toHaveBeenCalledTimes(4); + expect(getGitStatusSnapshot(TARGET).error).toBeNull(); + + release(); + } finally { + vi.useRealTimers(); + } + }); + + it("rebuilds the subscription on demand for the panel's retry", () => { + const harness = createControllableGitStatusClient(); + const release = watchGitStatus(TARGET, harness.client); + + harness.emit(BASE_STATUS); + rebuildGitStatusSubscription(TARGET); + + expect(harness.onStatus).toHaveBeenCalledTimes(2); + expect(harness.subscriberCount()).toBe(1); + + release(); + expect(harness.subscriberCount()).toBe(0); + + // Nothing is watching this target any more, so retrying must be inert. + rebuildGitStatusSubscription(TARGET); + expect(harness.onStatus).toHaveBeenCalledTimes(2); + }); + + it("feeds the local poll response into a status atom the stream never filled", async () => { + vi.useFakeTimers(); + try { + const harness = createControllableGitStatusClient(); + const release = watchGitStatus(TARGET, harness.client); + + await vi.advanceTimersByTimeAsync(60_000); + expect(getGitStatusSnapshot(TARGET).data).toBeNull(); + expect(getGitStatusSnapshot(TARGET).error).not.toBeNull(); + + await refreshLocalGitStatus(TARGET, harness.client, { force: true }); + + const snapshot = getGitStatusSnapshot(TARGET); + expect(snapshot.data?.refName).toBe("/repo-local-refreshed"); + expect(snapshot.error).toBeNull(); + expect(snapshot.cause).toBeNull(); + expect(snapshot.isPending).toBe(false); + + release(); + } finally { + vi.useRealTimers(); + } + }); + + it("overlays the local poll response without dropping remote-derived fields", async () => { + vi.useFakeTimers(); + try { + const harness = createControllableGitStatusClient(); + const release = watchGitStatus(TARGET, harness.client); + + harness.emit({ ...BASE_STATUS, aheadCount: 3, behindCount: 1 }); + + // A healthy data-bearing atom can only go stale through a reconnect + // whose replacement stream stays silent past the whole rebuild budget — + // a lone retry no longer breaks data the poll lane keeps fresh. + harness.resubscribe(); + await vi.advanceTimersByTimeAsync(60_000); + expect(getGitStatusSnapshot(TARGET).error).not.toBeNull(); + expect(getGitStatusSnapshot(TARGET).data).not.toBeNull(); + + await refreshLocalGitStatus(TARGET, harness.client, { force: true }); + + const snapshot = getGitStatusSnapshot(TARGET); + expect(snapshot.error).toBeNull(); + expect(snapshot.data?.refName).toBe("/repo-local-refreshed"); + // Only the local half is fresh: ahead/behind come from the last snapshot + // the stream did deliver, and the local RPC knows nothing about them. + expect(snapshot.data?.aheadCount).toBe(3); + expect(snapshot.data?.behindCount).toBe(1); + + release(); + } finally { + vi.useRealTimers(); + } + }); }); diff --git a/apps/web/src/lib/gitStatusState.ts b/apps/web/src/lib/gitStatusState.ts index 009f2ba70..758f27843 100644 --- a/apps/web/src/lib/gitStatusState.ts +++ b/apps/web/src/lib/gitStatusState.ts @@ -5,6 +5,7 @@ import { type VcsStatusLocalResult, type VcsStatusResult, } from "@threadlines/contracts"; +import { applyGitStatusStreamEvent } from "@threadlines/shared/git"; import * as Cause from "effect/Cause"; import { Atom } from "effect/unstable/reactivity"; import { useEffect } from "react"; @@ -14,6 +15,10 @@ import { readEnvironmentConnection, subscribeEnvironmentConnections, } from "../environments/runtime"; +import { + recordStreamDiagnostic, + STREAM_DIAGNOSTIC_NAMES, +} from "../observability/streamDiagnostics"; import type { WsRpcClient } from "~/rpc/wsRpcClient"; /** @@ -37,9 +42,20 @@ interface ResolvedGitStatusClient { readonly client: GitStatusClient; } +/** + * A live subscription that can be torn down and rebuilt in place. `unsubscribe` + * is stable across rebuilds — it always releases whatever stream is current — + * so `watchedGitStatuses` never has to swap a handle mid-rebuild and + * refcounting stays correct while a rebuild is in flight. + */ +interface GitStatusSubscription { + readonly unsubscribe: () => void; + readonly rebuild: () => void; +} + interface WatchedGitStatus { refCount: number; - unsubscribe: () => void; + readonly subscription: GitStatusSubscription; } interface GitStatusTarget { @@ -81,9 +97,16 @@ const GIT_STATUS_LOCAL_REFRESH_TIMEOUT_MS = 20_000; // The full refresh can include a real `git fetch`, so it gets a much longer // budget than the local-only one. const GIT_STATUS_REFRESH_TIMEOUT_MS = 120_000; -// How long a (re)subscription may go without delivering a snapshot before the -// UI is told the live status is broken rather than still loading. +// How long a (re)subscription may go without delivering a snapshot before it +// is treated as broken. The first expiries rebuild the stream; only the last +// one tells the UI the live status is broken rather than still loading. const GIT_STATUS_FIRST_SNAPSHOT_TIMEOUT_MS = 20_000; +// The observed failure is a lost opening snapshot: the stream is accepted, the +// one-shot snapshot never arrives, and a quiet repo emits nothing afterwards. +// A fresh subscribe always heals it, so rebuild before showing a notice. The +// cap matters — in environments where subscribing can never succeed (browser +// tests, a server that is gone) this must settle instead of looping forever. +const GIT_STATUS_MAX_STREAM_REBUILDS = 2; // One failed stream attempt is normal churn (socket blip, server restart). // Only a retry that did not immediately recover is worth showing. const GIT_STATUS_RETRY_ERROR_ATTEMPT_THRESHOLD = 2; @@ -214,12 +237,26 @@ export function watchGitStatus(target: GitStatusTarget, client?: GitStatusClient watchedGitStatuses.set(targetKey, { refCount: 1, - unsubscribe: subscribeToGitStatusTarget(targetKey, target, client), + subscription: subscribeToGitStatusTarget(targetKey, target, client), }); return () => unwatchGitStatus(targetKey); } +/** + * Forces the teardown-and-recreate that heals a stream whose opening snapshot + * was lost. Exported for the panel's Retry: refreshing over the unary RPC + * repairs the *data*, but leaves the dead stream in place, so the next real + * change would go unnoticed again. + */ +export function rebuildGitStatusSubscription(target: GitStatusTarget): void { + const targetKey = getGitStatusTargetKey(target); + if (targetKey === null) { + return; + } + watchedGitStatuses.get(targetKey)?.subscription.rebuild(); +} + export function refreshGitStatus( target: GitStatusTarget, client?: GitStatusClient, @@ -280,6 +317,30 @@ function recoverGitStatusFromRefresh(targetKey: string, status: VcsStatusResult) }); } +/** + * Same contract as `recoverGitStatusFromRefresh`, for the panel's 5s local + * poll: while the stream is dead this response is the only fresh data the UI + * can get, and dropping it left the panel frozen behind the stale notice. + * + * `localUpdated` is exactly this situation in the shared stream reducer, so it + * does the merge: fresh local fields over the remote fields the last merged + * snapshot carried (ahead/behind, PR), or the empty remote part when there is + * no snapshot yet. + */ +function recoverGitStatusFromLocalRefresh(targetKey: string, local: VcsStatusLocalResult): void { + const atom = gitStatusStateAtom(targetKey); + const current = appAtomRegistry.get(atom); + if (current.error === null && current.data !== null) { + return; + } + appAtomRegistry.set(atom, { + data: applyGitStatusStreamEvent(current.data, { _tag: "localUpdated", local }), + error: null, + cause: null, + isPending: false, + }); +} + export function refreshLocalGitStatus( target: GitStatusTarget, client?: GitStatusClient, @@ -310,7 +371,10 @@ export function refreshLocalGitStatus( return trackGitStatusRefresh( gitStatusLocalRefreshInFlight, targetKey, - resolvedClient.refreshLocalStatus({ cwd: target.cwd }), + resolvedClient.refreshLocalStatus({ cwd: target.cwd }).then((local) => { + recoverGitStatusFromLocalRefresh(targetKey, local); + return local; + }), GIT_STATUS_LOCAL_REFRESH_TIMEOUT_MS, "Local git status refresh", ); @@ -318,7 +382,7 @@ export function refreshLocalGitStatus( export function resetGitStatusStateForTests(): void { for (const watched of watchedGitStatuses.values()) { - watched.unsubscribe(); + watched.subscription.unsubscribe(); } watchedGitStatuses.clear(); clearGitStatusRefreshTracking(); @@ -355,22 +419,27 @@ function unwatchGitStatus(targetKey: string): void { return; } - watched.unsubscribe(); + watched.subscription.unsubscribe(); watchedGitStatuses.delete(targetKey); } +const NOOP_GIT_STATUS_SUBSCRIPTION: GitStatusSubscription = { + unsubscribe: NOOP, + rebuild: NOOP, +}; + function subscribeToGitStatusTarget( targetKey: string, target: GitStatusTarget, providedClient?: GitStatusClient, -): () => void { +): GitStatusSubscription { if (target.cwd === null) { - return NOOP; + return NOOP_GIT_STATUS_SUBSCRIPTION; } const cwd = target.cwd; let currentClientIdentity: string | null = null; - let currentUnsubscribe = NOOP; + let current: GitStatusSubscription = NOOP_GIT_STATUS_SUBSCRIPTION; const syncClientSubscription = () => { const resolved = providedClient @@ -382,8 +451,8 @@ function subscribeToGitStatusTarget( if (!resolved) { if (currentClientIdentity !== null) { - currentUnsubscribe(); - currentUnsubscribe = NOOP; + current.unsubscribe(); + current = NOOP_GIT_STATUS_SUBSCRIPTION; currentClientIdentity = null; } markGitStatusPending(targetKey); @@ -394,9 +463,9 @@ function subscribeToGitStatusTarget( return; } - currentUnsubscribe(); + current.unsubscribe(); currentClientIdentity = resolved.clientIdentity; - currentUnsubscribe = subscribeToGitStatus(targetKey, cwd, resolved.client); + current = subscribeToGitStatus(targetKey, cwd, resolved.client); }; const unsubscribeRegistry = providedClient @@ -404,14 +473,25 @@ function subscribeToGitStatusTarget( : subscribeEnvironmentConnections(syncClientSubscription); syncClientSubscription(); - return () => { - unsubscribeRegistry(); - currentUnsubscribe(); + return { + unsubscribe: () => { + unsubscribeRegistry(); + current.unsubscribe(); + }, + rebuild: () => current.rebuild(), }; } -function subscribeToGitStatus(targetKey: string, cwd: string, client: GitStatusClient): () => void { +function subscribeToGitStatus( + targetKey: string, + cwd: string, + client: GitStatusClient, +): GitStatusSubscription { let firstSnapshotTimer: ReturnType | null = null; + let unsubscribeStream = NOOP; + let hasOpenedStream = false; + let rebuildCount = 0; + let disposed = false; const clearFirstSnapshotWatchdog = () => { if (firstSnapshotTimer !== null) { @@ -424,42 +504,127 @@ function subscribeToGitStatus(targetKey: string, cwd: string, client: GitStatusC // sits on `isPending` forever, which is what a half-dead socket produced. const startFirstSnapshotWatchdog = () => { clearFirstSnapshotWatchdog(); - firstSnapshotTimer = setTimeout(() => { - firstSnapshotTimer = null; + firstSnapshotTimer = setTimeout(onFirstSnapshotTimeout, GIT_STATUS_FIRST_SNAPSHOT_TIMEOUT_MS); + }; + + const onFirstSnapshotTimeout = () => { + firstSnapshotTimer = null; + if (disposed) { + return; + } + + // The bucket carries the attempt so each of the (at most three) expiries in + // a cycle is recorded: they are 20s apart, inside the 30s rate-limit window. + recordStreamDiagnostic( + STREAM_DIAGNOSTIC_NAMES.gitStatusWatchdogExpired, + `${targetKey}#${rebuildCount}`, + { + "git.status.target": targetKey, + "git.status.rebuild_count": rebuildCount, + "git.status.timeout_ms": GIT_STATUS_FIRST_SNAPSHOT_TIMEOUT_MS, + }, + ); + + if (rebuildCount >= GIT_STATUS_MAX_STREAM_REBUILDS) { + // Out of rebuilds: the notice is the last resort, not the first + // response. Nothing restarts the watchdog from here, so this settles. markGitStatusStale(targetKey, gitStatusStreamError()); - }, GIT_STATUS_FIRST_SNAPSHOT_TIMEOUT_MS); + return; + } + + rebuildCount += 1; + recordStreamDiagnostic( + STREAM_DIAGNOSTIC_NAMES.gitStatusRebuild, + `${targetKey}#${rebuildCount}`, + { + "git.status.target": targetKey, + "git.status.rebuild_attempt": rebuildCount, + "git.status.trigger": "watchdog", + }, + ); + openStream(); }; - markGitStatusPending(targetKey); - startFirstSnapshotWatchdog(); - const unsubscribe = client.onStatus( - { cwd }, - (status: VcsStatusResult) => { - clearFirstSnapshotWatchdog(); - appAtomRegistry.set(gitStatusStateAtom(targetKey), { - data: status, - error: null, - cause: null, - isPending: false, - }); - }, - { - onResubscribe: () => { - markGitStatusPending(targetKey); - startFirstSnapshotWatchdog(); + function openStream(): void { + if (disposed) { + return; + } + + // Release the old stream before opening the new one, and clear the handle + // first so a teardown racing this rebuild cannot double-release it. + const previousUnsubscribe = unsubscribeStream; + unsubscribeStream = NOOP; + previousUnsubscribe(); + + // Only the first open announces "loading". Rebuilds keep whatever is on + // screen (data or the stale notice) until an event actually arrives: + // flipping back to pending on every rebuild makes the atom oscillate + // pending↔stale in environments where subscribing never succeeds, and + // that churn re-renders every consumer for a minute after mount. + if (!hasOpenedStream) { + hasOpenedStream = true; + markGitStatusPending(targetKey); + } + startFirstSnapshotWatchdog(); + unsubscribeStream = client.onStatus( + { cwd }, + (status: VcsStatusResult) => { + clearFirstSnapshotWatchdog(); + // The stream is alive again, so the next silent stretch gets its own + // full rebuild budget. + rebuildCount = 0; + appAtomRegistry.set(gitStatusStateAtom(targetKey), { + data: status, + error: null, + cause: null, + isPending: false, + }); }, - onRetry: (error: unknown, attempt: number) => { - if (attempt < GIT_STATUS_RETRY_ERROR_ATTEMPT_THRESHOLD) { - return; - } - markGitStatusStale(targetKey, gitStatusStreamError(error)); + { + onResubscribe: () => { + rebuildCount = 0; + markGitStatusPending(targetKey); + startFirstSnapshotWatchdog(); + }, + onRetry: (error: unknown, attempt: number) => { + recordStreamDiagnostic(STREAM_DIAGNOSTIC_NAMES.subscriptionRetry, targetKey, { + "rpc.stream.tag": "vcs.subscribeVcsStatus", + "git.status.target": targetKey, + "rpc.stream.attempt": attempt, + "error.message": error instanceof Error ? error.message : String(error), + }); + if (attempt < GIT_STATUS_RETRY_ERROR_ATTEMPT_THRESHOLD) { + return; + } + markGitStatusStale(targetKey, gitStatusStreamError(error)); + }, }, - }, - ); + ); + } - return () => { - clearFirstSnapshotWatchdog(); - unsubscribe(); + openStream(); + + return { + unsubscribe: () => { + disposed = true; + clearFirstSnapshotWatchdog(); + const previousUnsubscribe = unsubscribeStream; + unsubscribeStream = NOOP; + previousUnsubscribe(); + }, + rebuild: () => { + if (disposed) { + return; + } + // A manual rebuild is a user action (Retry), so it opens a fresh cycle + // with a fresh budget. It cannot loop on its own. + rebuildCount = 0; + recordStreamDiagnostic(STREAM_DIAGNOSTIC_NAMES.gitStatusRebuild, targetKey, { + "git.status.target": targetKey, + "git.status.trigger": "manual", + }); + openStream(); + }, }; } @@ -477,6 +642,15 @@ function markGitStatusStale(targetKey: string, error: GitStatusError): void { if (!current.isPending && current.error?.message === error.message) { return; } + // An atom that holds healthy data stays healthy: the poll lane refreshes it + // every few seconds while the stream is broken, and the stale notice only + // renders when there is no data. Setting the error here buys no UI and + // makes the atom alternate broken↔healthy against the poll feed — the + // exact churn loop that has destabilized the app twice (poll heals, retry + // re-marks, forever). Dead-stream repair stays the rebuild machinery's job. + if (current.data !== null && current.error === null && !current.isPending) { + return; + } appAtomRegistry.set(atom, { data: current.data, error, diff --git a/apps/web/src/observability/clientTracing.ts b/apps/web/src/observability/clientTracing.ts index d5ddc2737..9812fb120 100644 --- a/apps/web/src/observability/clientTracing.ts +++ b/apps/web/src/observability/clientTracing.ts @@ -1,6 +1,8 @@ +import * as Context from "effect/Context"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as ManagedRuntime from "effect/ManagedRuntime"; +import * as Option from "effect/Option"; import * as Scope from "effect/Scope"; import * as Tracer from "effect/Tracer"; import { FetchHttpClient, HttpClient } from "effect/unstable/http"; @@ -46,6 +48,51 @@ export const ClientTracingLive = Layer.succeed( }), ); +/** + * Emits a zero-duration span straight onto the OTLP exporter, so plain + * (non-Effect) code can put a fact in the server's trace file. + * + * The renderer's telemetry pipeline is traces-only — `OtlpTracer` posting to + * `/api/observability/v1/traces`, which the server appends to + * `server.trace.ndjson`. There is no log exporter, so a span *is* the + * log-style event here; `OtlpTracer` exports on `end()`, hence start and end + * at the same instant. + * + * Returns false when tracing has not been configured yet (the span would go + * nowhere), so callers can decide whether a console fallback is worthwhile. + */ +export function recordClientTraceEvent( + name: string, + attributes: Readonly>, +): boolean { + const delegate = activeDelegate; + if (delegate === null) { + return false; + } + + try { + const startTime = BigInt(Date.now()) * 1_000_000n; + const span = delegate.span({ + name, + parent: Option.none(), + annotations: Context.empty(), + links: [], + startTime, + kind: "internal", + root: true, + sampled: true, + }); + for (const [key, value] of Object.entries(attributes)) { + span.attribute(key, value); + } + span.end(startTime, Exit.void); + return true; + } catch { + // Diagnostics must never take down the code path they are observing. + return false; + } +} + export function configureClientTracing(config: ClientTracingConfig = {}): Promise { if (config.exportIntervalMs === undefined && activeConfigKey !== null) { return pendingConfiguration; diff --git a/apps/web/src/observability/streamDiagnostics.ts b/apps/web/src/observability/streamDiagnostics.ts new file mode 100644 index 0000000000000000000000000000000000000000..bb1e0fd55d9a5e59f7e787ad710fb585136d423a GIT binary patch literal 1918 zcmaJ?+in{-5bd+RVk!iQY)ESZd6O@w6T5{UYy*-2KUx%bR?B*qT##HVf+3(E(J$Ox-vi>6JWd=ku?XVH;;&X})>E7j3BUuBLu=qW4e6F8gP_ z-pTQCkB(^?eXC>ZIj4nZm22PDi$I0*6icQ;nF^0oyPN}VR3wF?u`ec=Ma86w$jMZK z6-vi3DQenKQizCGcRou8?_esyv?}pG1rJU?bEIeHCXljW#U9|OwaTw0oK{s8hIq-7 zxzuPa>qxcIrLpYSUG&0ugkP{$ZD4W*nV8Y6M{DN5pMR!xd7H!L9HaM$T3V)gUm_{C&dbM?Z)b`B3T5`JfuUG zLT5JrGdMekCvb@E8l_u+q^0Mkk=CpeaI&iA3;|Ah{4I4JjD$$j+2rQx?&W}+} z>FoCF)BV-m&6Fc-UXHtxwt|g5=)n?n!R}%Fe+Vyo9M7|t`RX< z+!Q!;laORX1?DL0{Y(LmxA)f%k56}^g3qTCc9TiZ&7c+*Y9Rs6F4(!swV%PYE5 z%{j1wmmyhf0``}Kqwm||w?=u)w8)TXicw0R!*0vWedThFX$rAmx`>5UbK*t`p(d6WvG?#zVRnOYdT=I3H4QaJe!g!|)-lx|@5XV1 jfScZKPrhj{lK!{5;dkd}9AX%}ZOQwXb;ZgH?VH}ecUqW8 literal 0 HcmV?d00001 diff --git a/apps/web/src/rpc/wsTransport.ts b/apps/web/src/rpc/wsTransport.ts index e57238209..6ccea9d27 100644 --- a/apps/web/src/rpc/wsTransport.ts +++ b/apps/web/src/rpc/wsTransport.ts @@ -10,6 +10,10 @@ import * as Stream from "effect/Stream"; import { RpcClient } from "effect/unstable/rpc"; import { ClientTracingLive } from "../observability/clientTracing"; +import { + recordStreamDiagnostic, + STREAM_DIAGNOSTIC_NAMES, +} from "../observability/streamDiagnostics"; import { clearAllTrackedRpcRequests } from "./requestLatencyState"; import { createWsRpcProtocolLayer, @@ -216,6 +220,12 @@ export class WsTransport { throw new TransportRequestRetriesExhaustedError(label, elapsedMs, error); } if (session === this.session && !this.isHeartbeatFresh(REQUEST_RETRY_HEARTBEAT_FRESH_MS)) { + recordStreamDiagnostic(STREAM_DIAGNOSTIC_NAMES.transportReconnect, "zombie-socket", { + "rpc.reconnect.trigger": "zombie-socket", + "rpc.request.label": label, + "rpc.request.attempt": attempt, + "error.message": formatErrorMessage(error), + }); await this.reconnect().catch(() => undefined); } await sleep(retryDelayMs); @@ -322,6 +332,12 @@ export class WsTransport { } retryAttempt += 1; + const diagnosticTag = options?.tag ?? "unknown"; + recordStreamDiagnostic(STREAM_DIAGNOSTIC_NAMES.subscriptionRetry, diagnosticTag, { + "rpc.stream.tag": diagnosticTag, + "rpc.stream.attempt": retryAttempt, + "error.message": formatErrorMessage(error), + }); try { options?.onRetry?.(error, retryAttempt); } catch { @@ -375,6 +391,11 @@ export class WsTransport { throw new Error("Transport disposed"); } + recordStreamDiagnostic(STREAM_DIAGNOSTIC_NAMES.transportReconnect, "session-swap", { + "rpc.reconnect.trigger": "session-swap", + "rpc.heartbeat.fresh": this.isHeartbeatFresh(), + }); + const reconnectOperation = this.reconnectChain.then(async () => { if (this.disposed) { throw new Error("Transport disposed"); From baa30139d0ad7cd9f85a60d902a6855afaadb685 Mon Sep 17 00:00:00 2001 From: Badcuban <108198679+badcuban@users.noreply.github.com> Date: Sun, 2 Aug 2026 14:18:42 -0400 Subject: [PATCH 2/2] Treat delivered status, not reconnect attempts, as stream recovery The transport announces every reconnect attempt via onResubscribe before any data arrives. Treating those announcements as recovery reset the first-snapshot watchdog and refilled the rebuild budget, so attempts arriving faster than the watchdog postponed it forever and the bounded rebuild cycle never converged. Only a delivered status ends the waiting phase now; attempt starts during one are inert. Also hardens the palette browser test to await the committed Sources view and an attached action before clicking, which was the actual CI failure signature all along. --- apps/web/src/components/ChatView.browser.tsx | 9 +++- apps/web/src/lib/gitStatusState.test.ts | 43 ++++++++++++++++++++ apps/web/src/lib/gitStatusState.ts | 18 +++++++- 3 files changed, 67 insertions(+), 3 deletions(-) diff --git a/apps/web/src/components/ChatView.browser.tsx b/apps/web/src/components/ChatView.browser.tsx index a8ce4ce9c..3e96e1e69 100644 --- a/apps/web/src/components/ChatView.browser.tsx +++ b/apps/web/src/components/ChatView.browser.tsx @@ -7436,7 +7436,14 @@ describe("ChatView timeline estimator parity (full app)", () => { .element(palette.getByText("This device", { exact: true }).first()) .toBeInTheDocument(); await palette.getByText("Staging", { exact: true }).click(); - await palette.getByText("Local folder", { exact: true }).click(); + // The palette re-renders its list when the environment pick commits the + // Sources view. Await the committed view and an attached action before + // clicking, or the click can land on a mid-transition node that detaches + // under it on a slow runner. + await expect.element(palette.getByText("Sources", { exact: true })).toBeInTheDocument(); + const localFolderAction = palette.getByText("Local folder", { exact: true }); + await expect.element(localFolderAction).toBeInTheDocument(); + await localFolderAction.click(); const browseInput = await waitForCommandPaletteInput(ADD_PROJECT_SUBMENU_PLACEHOLDER); await expect.element(browseInput).toHaveValue("~/workspaces/"); diff --git a/apps/web/src/lib/gitStatusState.test.ts b/apps/web/src/lib/gitStatusState.test.ts index ecf9f1541..8ac1ea3e6 100644 --- a/apps/web/src/lib/gitStatusState.test.ts +++ b/apps/web/src/lib/gitStatusState.test.ts @@ -600,6 +600,49 @@ describe("gitStatusState", () => { } }); + it("converges when reconnect attempts arrive faster than the watchdog", async () => { + vi.useFakeTimers(); + try { + const harness = createControllableGitStatusClient(); + const release = watchGitStatus(TARGET, harness.client); + harness.emit(BASE_STATUS); + + // A dead-but-retrying transport announces an attempt start every few + // seconds without ever delivering. Each announcement must NOT reset the + // watchdog or refill the rebuild budget, or the bounded cycle never + // converges and the stale notice never appears. + harness.resubscribe(); + for (let elapsed = 0; elapsed < 60_000; elapsed += 5_000) { + await vi.advanceTimersByTimeAsync(5_000); + harness.resubscribe(); + } + + // Initial open + exactly two watchdog rebuilds, then the notice. + expect(harness.onStatus).toHaveBeenCalledTimes(3); + expect(getGitStatusSnapshot(TARGET).error?.message).toBe(GIT_STATUS_STALE_MESSAGE); + expect(getGitStatusSnapshot(TARGET).isPending).toBe(false); + + // Exhausted stays exhausted: more attempt starts open nothing new... + await vi.advanceTimersByTimeAsync(120_000); + harness.resubscribe(); + await vi.advanceTimersByTimeAsync(120_000); + expect(harness.onStatus).toHaveBeenCalledTimes(3); + + // ...but a delivered value still heals and re-arms the cycle. + harness.emit(BASE_STATUS); + expect(getGitStatusSnapshot(TARGET)).toEqual({ + data: BASE_STATUS, + error: null, + cause: null, + isPending: false, + }); + + release(); + } finally { + vi.useRealTimers(); + } + }); + it("converges when the poll lane heals while the stream keeps failing", async () => { vi.useFakeTimers(); try { diff --git a/apps/web/src/lib/gitStatusState.ts b/apps/web/src/lib/gitStatusState.ts index 758f27843..94430f6ef 100644 --- a/apps/web/src/lib/gitStatusState.ts +++ b/apps/web/src/lib/gitStatusState.ts @@ -490,6 +490,10 @@ function subscribeToGitStatus( let firstSnapshotTimer: ReturnType | null = null; let unsubscribeStream = NOOP; let hasOpenedStream = false; + // True from stream open (or a genuine post-delivery resubscribe) until a + // status value arrives. Guards the watchdog and rebuild budget against + // transport attempt-starts, which say nothing about data flowing. + let isAwaitingSnapshot = false; let rebuildCount = 0; let disposed = false; @@ -565,13 +569,15 @@ function subscribeToGitStatus( hasOpenedStream = true; markGitStatusPending(targetKey); } + isAwaitingSnapshot = true; startFirstSnapshotWatchdog(); unsubscribeStream = client.onStatus( { cwd }, (status: VcsStatusResult) => { clearFirstSnapshotWatchdog(); - // The stream is alive again, so the next silent stretch gets its own - // full rebuild budget. + // A delivered value is the only recovery signal: it ends the waiting + // phase and grants the next silent stretch its own rebuild budget. + isAwaitingSnapshot = false; rebuildCount = 0; appAtomRegistry.set(gitStatusStateAtom(targetKey), { data: status, @@ -582,6 +588,14 @@ function subscribeToGitStatus( }, { onResubscribe: () => { + // The transport announces every reconnect ATTEMPT, not deliveries. + // While a snapshot is already awaited, another attempt starting is + // not progress — resetting the watchdog and budget here let + // attempts arriving faster than the watchdog postpone it forever. + if (isAwaitingSnapshot) { + return; + } + isAwaitingSnapshot = true; rebuildCount = 0; markGitStatusPending(targetKey); startFirstSnapshotWatchdog();