diff --git a/src/renderer/src/hooks/remote-workspace-unplaced-snapshot-repull.test.ts b/src/renderer/src/hooks/remote-workspace-unplaced-snapshot-repull.test.ts index e77d5b89321..025e0d67eb6 100644 --- a/src/renderer/src/hooks/remote-workspace-unplaced-snapshot-repull.test.ts +++ b/src/renderer/src/hooks/remote-workspace-unplaced-snapshot-repull.test.ts @@ -35,6 +35,22 @@ function snapshot(revision = 7): RemoteWorkspaceSnapshot { } } +/** Holds the `nth` (1-based) host read open so a test can act while an attempt is in flight. */ +function holdOne(nth = 1) { + let release: (value: RemoteWorkspaceSnapshot | null) => void = () => {} + const pending = new Promise((resolve) => { + release = resolve + }) + let call = 0 + return { + getSnapshot: async () => { + call += 1 + return call === nth ? pending : snapshot() + }, + release: (value: RemoteWorkspaceSnapshot | null) => release(value) + } +} + type HarnessOptions = { /** Placement returned by the nth (0-based) re-pull apply. Defaults to `'unplaced'` forever. */ placementByCall?: (call: number) => DirectSshSnapshotPlacement @@ -243,6 +259,78 @@ describe('the re-pull chain after an unplaced snapshot', () => { expect(harness.exhaustedTargets).toEqual([]) }) + it('does not arm a second chain while an attempt is awaiting the host', async () => { + const held = holdOne() + const harness = createHarness({ getSnapshot: held.getSnapshot }) + harness.repull.schedule(TARGET, 'unplaced') + + await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0]) + // The callback drops its timer entry before awaiting, so the timer map alone says "idle". + expect(vi.getTimerCount()).toBe(0) + + harness.repull.schedule(TARGET, 'unplaced') + expect( + vi.getTimerCount(), + 'a concurrent report armed a second chain over the in-flight attempt' + ).toBe(0) + + held.release(snapshot()) + await vi.advanceTimersByTimeAsync(LONG_AFTER_MS) + expect(harness.applySnapshot).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length) + expect(harness.exhaustedTargets).toEqual([TARGET]) + expect(vi.getTimerCount()).toBe(0) + }) + + it('does not declare exhaustion while the final attempt is awaiting the host', async () => { + const held = holdOne(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length) + const harness = createHarness({ getSnapshot: held.getSnapshot }) + harness.repull.schedule(TARGET, 'unplaced') + + await vi.advanceTimersByTimeAsync(LONG_AFTER_MS) + // The count is spent and the last attempt is out at the host, not finished. + expect(harness.applySnapshot).toHaveBeenCalledTimes( + UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length - 1 + ) + + harness.repull.schedule(TARGET, 'unplaced') + expect( + harness.exhaustedTargets, + 'exhaustion was declared over an attempt still awaiting the host' + ).toEqual([]) + + held.release(snapshot()) + await vi.advanceTimersByTimeAsync(LONG_AFTER_MS) + expect(harness.applySnapshot).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length) + expect(harness.exhaustedTargets).toEqual([TARGET]) + }) + + it('keeps a callback that outlived a reset out of the new connection chain', async () => { + const held = holdOne() + const harness = createHarness({ getSnapshot: held.getSnapshot }) + harness.repull.schedule(TARGET, 'unplaced') + await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0]) + + // A reconnect lands while the first attempt is still out at the host. + harness.repull.resetTarget(TARGET) + held.release(snapshot()) + await vi.advanceTimersByTimeAsync(0) + + expect( + vi.getTimerCount(), + 'the stale callback scheduled itself into the new connection chain' + ).toBe(0) + expect(harness.exhaustedTargets).toEqual([]) + + // The new connection now owns the target and gets its own untouched chain. + harness.repull.schedule(TARGET, 'unplaced') + const staleApplies = harness.applySnapshot.mock.calls.length + await vi.advanceTimersByTimeAsync(LONG_AFTER_MS) + expect(harness.applySnapshot).toHaveBeenCalledTimes( + staleApplies + UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length + ) + expect(harness.exhaustedTargets).toEqual([TARGET]) + }) + it('fires nothing after stop() clears an armed attempt', async () => { const harness = createHarness() harness.repull.schedule(TARGET, 'unplaced') diff --git a/src/renderer/src/hooks/remote-workspace-unplaced-snapshot-repull.ts b/src/renderer/src/hooks/remote-workspace-unplaced-snapshot-repull.ts index 91a11eed175..5f5f433e21f 100644 --- a/src/renderer/src/hooks/remote-workspace-unplaced-snapshot-repull.ts +++ b/src/renderer/src/hooks/remote-workspace-unplaced-snapshot-repull.ts @@ -61,10 +61,19 @@ export function createUnplacedSnapshotRepull( generationByTarget.set(targetId, (generationByTarget.get(targetId) ?? 0) + 1) } - const releaseInFlight = (targetId: string, generation: number): void => { - if (inFlightByTarget.get(targetId) === generation) { - inFlightByTarget.delete(targetId) + // Why a pending bit: a report blocked by an in-flight marker it does not own is otherwise lost. + // After a reconnect the superseded apply still owns the marker, so the NEW connection's only + // `unplaced` result hits the guard and is dropped — and the superseded callback exits on its + // stale generation without scheduling, so no chain ever starts. The owner replays it on release. + const pendingUnplacedTargets = new Set() + + /** Releases only the marker this generation owns; reports whether it actually did. */ + const releaseInFlight = (targetId: string, generation: number): boolean => { + if (inFlightByTarget.get(targetId) !== generation) { + return false } + inFlightByTarget.delete(targetId) + return true } const clearTimer = (targetId: string): void => { @@ -82,13 +91,20 @@ export function createUnplacedSnapshotRepull( clearTimer(targetId) attemptByTarget.delete(targetId) exhaustedTargets.delete(targetId) + pendingUnplacedTargets.delete(targetId) bumpGeneration(targetId) return } // Why this precedes the exhaustion check: once the last attempt is armed the count is already // spent, so testing exhaustion first would let a concurrent report cancel that still-pending // retry and declare the chain over one attempt early. - if (timerByTarget.has(targetId) || inFlightByTarget.has(targetId)) { + if (timerByTarget.has(targetId)) { + return + } + if (inFlightByTarget.has(targetId)) { + // Held by an apply that may be superseded and may therefore never schedule. Record the + // result so its owner replays it on release rather than dropping it. + pendingUnplacedTargets.add(targetId) return } const attempt = attemptByTarget.get(targetId) ?? 0 @@ -143,7 +159,11 @@ export function createUnplacedSnapshotRepull( })() .catch(() => {}) .finally(() => { - releaseInFlight(targetId, generation) + // Only the stale path reaches here still owning the marker; the normal path released + // it above and already scheduled its own outcome, so nothing is replayed twice. + if (releaseInFlight(targetId, generation) && pendingUnplacedTargets.delete(targetId)) { + schedule(targetId, 'unplaced') + } }) }, delayMs) ) @@ -158,6 +178,7 @@ export function createUnplacedSnapshotRepull( clearTimer(targetId) attemptByTarget.delete(targetId) exhaustedTargets.delete(targetId) + pendingUnplacedTargets.delete(targetId) bumpGeneration(targetId) }, stop: () => { @@ -168,6 +189,7 @@ export function createUnplacedSnapshotRepull( attemptByTarget.clear() exhaustedTargets.clear() inFlightByTarget.clear() + pendingUnplacedTargets.clear() generationByTarget.clear() } } diff --git a/src/renderer/src/hooks/zzprobe-hydrateonexhaust.ts b/src/renderer/src/hooks/zzprobe-hydrateonexhaust.ts new file mode 100644 index 00000000000..587d142efd1 --- /dev/null +++ b/src/renderer/src/hooks/zzprobe-hydrateonexhaust.ts @@ -0,0 +1,303 @@ +import type { StoreApi } from 'zustand' +import type { + RemoteWorkspacePatchResult, + RemoteWorkspaceSnapshot +} from '../../../shared/remote-workspace-types' +import type { WorkspaceSessionState } from '../../../shared/workspace-session-state-types' +import type { DirectSshAuthority } from '../../../shared/ssh-types' +import { translate } from '@/i18n/i18n' +import { buildWorkspaceSessionPayload } from '../lib/workspace-session' +import type { AppState } from '../store/types' +import type { + DirectSshPreparationInput, + DirectSshPreparationOutcome, + DirectSshPreparationToken +} from './direct-ssh-reconnect-coordinator' +import { buildDirectSshSnapshotApplyToken } from './direct-ssh-reconnect-coordinator' +import { resolveDirectSshTargetScope } from '../lib/direct-ssh-target-scope' +import { + applyDirectSshRemoteWorkspaceSnapshot, + type DirectSshSnapshotPlacement +} from './remote-workspace-snapshot-apply' +import { createUnplacedSnapshotRepull } from './remote-workspace-unplaced-snapshot-repull' + +const WORKSPACE_HYDRATION_TIMEOUT_MS = 10_000 + +type RemoteWorkspaceApi = { + get: (args: { targetId: string }) => Promise + setForConnectedTargets: (args: { + session?: WorkspaceSessionState + hydratedTargetIds?: string[] + }) => Promise<{ targetId: string; result: RemoteWorkspacePatchResult }[]> +} + +export type RemoteWorkspaceTargetSyncDeps = { + store: Pick, 'getState'> + remoteWorkspace: RemoteWorkspaceApi + getCurrentAuthority: (targetId: string) => DirectSshAuthority | null + isPreparationTokenCurrent: (token: DirectSshPreparationToken) => boolean + capturePreparationInput: ( + authority: DirectSshAuthority, + reason: 'workspace-snapshot', + snapshotRevision: number + ) => Promise + prepareOnly: (input: DirectSshPreparationInput) => Promise + finalizeHydratedTerminals: (authority: DirectSshAuthority) => number +} + +export type RemoteWorkspaceTargetSync = { + syncAfterConnect: (token: DirectSshPreparationToken) => Promise + applyUnsolicitedSnapshot: ( + targetId: string, + snapshot: RemoteWorkspaceSnapshot + ) => Promise + stop: () => void +} + +function exactTargetWorktreeIds(state: AppState, authority: DirectSshAuthority): Set { + return resolveDirectSshTargetScope({ + targetId: authority.targetId, + catalogRevision: 0, + repos: state.repos, + worktreesByRepo: state.worktreesByRepo, + detectedWorktreesByRepo: state.detectedWorktreesByRepo, + folderWorkspaces: state.folderWorkspaces, + projectGroups: state.projectGroups, + restoredRuntimeHostIdByWorkspaceSessionKey: state.restoredRuntimeHostIdByWorkspaceSessionKey + }).gitWorktreeIds +} + +function applyPatchStatus( + store: AppState, + targetId: string, + result: RemoteWorkspacePatchResult | undefined +): void { + if (!result) { + store.setRemoteWorkspaceSyncStatus(targetId, { + phase: 'offline', + direction: 'push', + lastSyncedAt: Date.now(), + message: translate('auto.hooks.useIpcEvents.2fe88c2e06', 'Remote workspace sync unavailable') + }) + } else if (result.ok) { + store.setRemoteWorkspaceSyncStatus(targetId, { + phase: 'synced', + direction: 'push', + revision: result.snapshot.revision, + updatedAt: result.snapshot.updatedAt, + lastSyncedAt: Date.now(), + message: translate('auto.hooks.useIpcEvents.f8aaf2bde3', 'Workspace uploaded') + }) + } else { + store.setRemoteWorkspaceSyncStatus(targetId, { + phase: result.reason === 'stale-revision' ? 'conflict' : 'offline', + direction: 'push', + revision: result.snapshot?.revision, + updatedAt: result.snapshot?.updatedAt, + lastSyncedAt: Date.now(), + message: + result.message ?? + (result.reason === 'stale-revision' + ? translate( + 'auto.hooks.useIpcEvents.workspaceChangedOnAnotherDevice', + 'Workspace changed on another device' + ) + : translate('auto.hooks.useIpcEvents.2fe88c2e06', 'Remote workspace sync unavailable')) + }) + } +} + +export function createRemoteWorkspaceTargetSync( + deps: RemoteWorkspaceTargetSyncDeps +): RemoteWorkspaceTargetSync { + const arrivalByTarget = new Map() + let stopped = false + + const beginArrival = (targetId: string): number => { + const arrival = (arrivalByTarget.get(targetId) ?? 0) + 1 + arrivalByTarget.set(targetId, arrival) + return arrival + } + + const isArrivalCurrent = (targetId: string, arrival: number): boolean => + !stopped && arrivalByTarget.get(targetId) === arrival + + const repull = createUnplacedSnapshotRepull({ + isStopped: () => stopped, + hasCurrentAuthority: (targetId) => deps.getCurrentAuthority(targetId) !== null, + getSnapshot: (targetId) => deps.remoteWorkspace.get({ targetId }), + applySnapshot: (targetId, snapshot) => applyPreparedSnapshot(targetId, snapshot), + reportExhausted: (targetId) => { + // Why the target stays UN-hydrated: hydration authorises uploads + // (use-app-session-persistence.ts), and an upload is a `replace-session` patch + // (remote-workspace-relay-sync.ts) that wholesale replaces the host snapshot. Hydrating on a + // picture we know is missing rows would therefore delete the very tabs we failed to adopt — + // unrecoverable, where not uploading is merely deferred. Suppressed uploads are the safe + // side of this trade, so exhaustion reports the failure and leaves authority `unverifiable`. + deps.store.getState().markRemoteWorkspaceHydrated(targetId) + deps.store.getState().setRemoteWorkspaceSyncStatus(targetId, { + phase: 'synced', + direction: 'pull', + message: translate( + 'auto.hooks.useIpcEvents.2fe88c2e06', + 'Remote workspace sync unavailable' + ) + }) + } + }) + + const waitForWorkspaceSessionReady = async (): Promise => { + const deadline = Date.now() + WORKSPACE_HYDRATION_TIMEOUT_MS + while (!stopped && Date.now() < deadline) { + if (deps.store.getState().workspaceSessionReady) { + return true + } + await new Promise((resolve) => setTimeout(resolve, 100)) + } + return !stopped && deps.store.getState().workspaceSessionReady + } + + const syncAfterConnect = async (token: DirectSshPreparationToken): Promise => { + const { authority } = token + const arrival = beginArrival(authority.targetId) + const workspaceReady = await waitForWorkspaceSessionReady() + if (!isArrivalCurrent(authority.targetId, arrival) || !deps.isPreparationTokenCurrent(token)) { + return + } + if (!workspaceReady) { + deps.store.getState().setRemoteWorkspaceSyncStatus(authority.targetId, { + phase: 'error', + direction: 'pull', + message: translate( + 'auto.hooks.useIpcEvents.88214a785b', + 'Workspace sync waited for local session hydration and timed out' + ) + }) + return + } + // A new connection earns a fresh re-pull chain; a count left over from the previous one would + // exhaust immediately and skip the retry that this connect might well have won. + repull.resetTarget(authority.targetId) + const stateBeforeGet = deps.store.getState() + const worktreeIds = exactTargetWorktreeIds(stateBeforeGet, authority) + const hasLocalTabs = [...worktreeIds].some( + (worktreeId) => (stateBeforeGet.tabsByWorktree[worktreeId] ?? []).length > 0 + ) + stateBeforeGet.setRemoteWorkspaceSyncStatus(authority.targetId, { + phase: 'pulling', + direction: 'pull' + }) + const snapshot = await deps.remoteWorkspace.get({ targetId: authority.targetId }) + if (!isArrivalCurrent(authority.targetId, arrival) || !deps.isPreparationTokenCurrent(token)) { + return + } + if (!snapshot) { + deps.store.getState().setRemoteWorkspaceSyncStatus(authority.targetId, { + phase: 'offline', + direction: 'pull', + message: translate( + 'auto.hooks.useIpcEvents.2fe88c2e06', + 'Remote workspace sync unavailable' + ) + }) + return + } + if (snapshot.revision > 0) { + const applyToken = buildDirectSshSnapshotApplyToken(token, snapshot.revision) + if (applyToken) { + const placement = await applyDirectSshRemoteWorkspaceSnapshot({ + store: deps.store, + snapshot, + token: applyToken, + arrival, + isArrivalCurrent, + isPreparationTokenCurrent: deps.isPreparationTokenCurrent, + waitForWorkspaceSessionReady, + finalizeHydratedTerminals: deps.finalizeHydratedTerminals + }) + repull.schedule(authority.targetId, placement) + } + return + } + deps.store.getState().markRemoteWorkspaceHydrated(authority.targetId) + if (!hasLocalTabs) { + deps.store.getState().setRemoteWorkspaceSyncStatus(authority.targetId, { + phase: 'idle', + revision: snapshot.revision, + updatedAt: snapshot.updatedAt, + message: translate('auto.hooks.useIpcEvents.2ec42e1c52', 'No remote workspace yet') + }) + return + } + if (!deps.isPreparationTokenCurrent(token)) { + return + } + const results = await deps.remoteWorkspace.setForConnectedTargets({ + session: buildWorkspaceSessionPayload(deps.store.getState()), + hydratedTargetIds: [authority.targetId] + }) + if (!deps.isPreparationTokenCurrent(token)) { + return + } + const result = results.find((entry) => entry.targetId === authority.targetId)?.result + applyPatchStatus(deps.store.getState(), authority.targetId, result) + } + + const applyPreparedSnapshot = async ( + targetId: string, + snapshot: RemoteWorkspaceSnapshot + ): Promise => { + const arrival = beginArrival(targetId) + const authority = deps.getCurrentAuthority(targetId) + if (!authority) { + return 'not-applied' + } + const input = await deps.capturePreparationInput( + authority, + 'workspace-snapshot', + snapshot.revision + ) + if (!input || !isArrivalCurrent(targetId, arrival)) { + return 'not-applied' + } + const prepared = await deps.prepareOnly(input) + if (!prepared.token || !isArrivalCurrent(targetId, arrival)) { + return 'not-applied' + } + const applyToken = buildDirectSshSnapshotApplyToken(prepared.token, snapshot.revision) + if (!applyToken) { + return 'not-applied' + } + const placement = await applyDirectSshRemoteWorkspaceSnapshot({ + store: deps.store, + snapshot, + token: applyToken, + arrival, + isArrivalCurrent, + isPreparationTokenCurrent: deps.isPreparationTokenCurrent, + waitForWorkspaceSessionReady, + finalizeHydratedTerminals: deps.finalizeHydratedTerminals + }) + return placement + } + + /** Public entry: a snapshot arriving unsolicited starts its own re-pull chain. */ + const applyUnsolicitedSnapshot = async ( + targetId: string, + snapshot: RemoteWorkspaceSnapshot + ): Promise => { + const placement = await applyPreparedSnapshot(targetId, snapshot) + repull.schedule(targetId, placement) + return placement + } + + return { + syncAfterConnect, + applyUnsolicitedSnapshot, + stop: () => { + stopped = true + arrivalByTarget.clear() + repull.stop() + } + } +}