mirror of
https://github.com/stablyai/orca.git
synced 2026-10-24 00:04:01 +00:00
refactor(ssh): drop the re-pull retry chain, keep the fix
The chain produced ten defects across review - unbounded retry, dead retry, cancelled final attempt, wedged chain, repeated exhaustion, overlapping chains, stale-callback cleanup, a lost report - every one in code that passed the full suite at the time. It bought only faster recovery *within* one connection: syncAfterConnect and applyUnsolicitedSnapshot already re-pull on the next connect or host push, so dropping it costs a retry, never the data. What remains is the part that was correct from the first commit: the importer reports rows it could not place, and an apply that could not place them neither marks the target hydrated nor sets 'synced'. Because hydration is what authorises uploads, and an upload wholesale replaces the host snapshot, that single rule is what stops a client from deleting the host tabs it failed to adopt. Status is now 'error' rather than 'pulling': with no chain pending, 'pulling' claimed a request that was not in flight.
This commit is contained in:
Neil
@@ -0,0 +1,238 @@
|
||||
// @vitest-environment happy-dom
|
||||
|
||||
/**
|
||||
* The data-safety end of STA-3593. Hydrating a target authorises workspace uploads
|
||||
* (`use-app-session-persistence.ts`), an upload is a `replace-session` patch
|
||||
* (`remote-workspace-relay-sync.ts`), and the host applies that by wholesale replacement. So a
|
||||
* client that hydrated on a snapshot whose tab rows it could not place would upload its incomplete
|
||||
* projection and DELETE the host tabs it failed to adopt.
|
||||
*
|
||||
* That is the asymmetry the design turns on: a suppressed upload is deferred and recoverable, a
|
||||
* wiped host snapshot is not. This asserts the upload itself never names an unplaced target —
|
||||
* through the real persistence gate, not the hydrated flag — so it still bites if the gating moves.
|
||||
*/
|
||||
import { cleanup, renderHook } from '@testing-library/react'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type * as AgentStatusModule from '@/lib/agent-status'
|
||||
import type { RemoteWorkspaceSnapshot } from '../../../shared/remote-workspace-types'
|
||||
import type { DirectSshAuthority, SshProviderEpoch } from '../../../shared/ssh-types'
|
||||
import type { DirectSshPreparationInput } from '../hooks/direct-ssh-reconnect-coordinator'
|
||||
import { createRemoteWorkspaceTargetSync } from '../hooks/remote-workspace-target-sync'
|
||||
import { makeWorktree } from '../store/slices/store-test-helpers'
|
||||
|
||||
vi.mock('sonner', () => ({ toast: { info: vi.fn(), success: vi.fn(), error: vi.fn() } }))
|
||||
vi.mock('@/lib/agent-status', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof AgentStatusModule>()
|
||||
return { ...actual, detectAgentStatusFromTitle: vi.fn().mockReturnValue(null) }
|
||||
})
|
||||
|
||||
import { useAppStore } from '@/store'
|
||||
import { useAppSessionPersistence } from './use-app-session-persistence'
|
||||
|
||||
const TARGET_ID = 'target-a'
|
||||
const REPO_ROOT = '/srv/proj'
|
||||
const HOST_PATH = `${REPO_ROOT}/alpha`
|
||||
const WORKTREE_ID = `repo-a::${HOST_PATH}`
|
||||
/** Long enough that any deferred retry or debounce would have fired if one existed. */
|
||||
const TOTAL_CHAIN_MS = 12_000
|
||||
/** Longer than the post-apply session-write suppression in remote-workspace-snapshot-apply.ts. */
|
||||
const WRITE_SUPPRESSION_MS = 1_500
|
||||
const DEBOUNCE_MS = 300
|
||||
|
||||
const owner: DirectSshAuthority = {
|
||||
targetId: TARGET_ID,
|
||||
providerEpoch: 'epoch-a' as SshProviderEpoch,
|
||||
connectionGeneration: 1
|
||||
}
|
||||
|
||||
function snapshot(revision = 4): RemoteWorkspaceSnapshot {
|
||||
return {
|
||||
namespace: 'workspace',
|
||||
revision,
|
||||
updatedAt: revision,
|
||||
schemaVersion: 1,
|
||||
session: {
|
||||
activeWorktreePath: HOST_PATH,
|
||||
activeTabId: 'T1',
|
||||
tabsByWorktreePath: {
|
||||
[HOST_PATH]: [0, 1].map((index) => ({
|
||||
id: `T${index + 1}`,
|
||||
worktreePath: HOST_PATH,
|
||||
ptyId: `pty-T${index + 1}`,
|
||||
title: `Terminal ${index + 1}`,
|
||||
customTitle: null,
|
||||
color: null,
|
||||
sortOrder: index,
|
||||
createdAt: index + 1
|
||||
}))
|
||||
},
|
||||
terminalLayoutsByTabId: {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type UploadArgs = { hydratedTargetIds?: string[]; session?: unknown }
|
||||
const uploads = vi.fn(async (_args: UploadArgs) => [] as { targetId: string; result: never }[])
|
||||
|
||||
function installWindowApi(): void {
|
||||
;(window as unknown as { api: unknown }).api = {
|
||||
session: { patch: vi.fn(async () => {}), set: vi.fn(async () => {}) },
|
||||
remoteWorkspace: { setForConnectedTargets: uploads },
|
||||
app: { stageBeforeUnloadSync: vi.fn() },
|
||||
ui: { onWindowCloseRequested: vi.fn(() => () => {}) }
|
||||
}
|
||||
}
|
||||
|
||||
function seedStore(withHostCatalog: boolean): void {
|
||||
useAppStore.setState({
|
||||
workspaceSessionReady: true,
|
||||
hydrationSucceeded: true,
|
||||
repos: [
|
||||
{
|
||||
id: 'repo-a',
|
||||
path: REPO_ROOT,
|
||||
displayName: 'Proj',
|
||||
badgeColor: '#000',
|
||||
addedAt: 0,
|
||||
connectionId: TARGET_ID,
|
||||
executionHostId: `ssh:${TARGET_ID}`
|
||||
} as never
|
||||
],
|
||||
worktreesByRepo: withHostCatalog
|
||||
? {
|
||||
'repo-a': [
|
||||
makeWorktree({
|
||||
id: WORKTREE_ID,
|
||||
repoId: 'repo-a',
|
||||
path: HOST_PATH,
|
||||
hostId: `ssh:${TARGET_ID}`
|
||||
} as never)
|
||||
]
|
||||
}
|
||||
: {},
|
||||
remoteWorkspaceHydratedTargetIds: new Set<string>(),
|
||||
remoteWorkspaceSyncStatusByTargetId: {},
|
||||
tabsByWorktree: {},
|
||||
reconnectPersistedTerminals: (async () => {}) as never
|
||||
})
|
||||
}
|
||||
|
||||
function createSync() {
|
||||
return createRemoteWorkspaceTargetSync({
|
||||
store: useAppStore,
|
||||
remoteWorkspace: {
|
||||
get: vi.fn(async () => snapshot()),
|
||||
setForConnectedTargets: uploads as never
|
||||
},
|
||||
getCurrentAuthority: () => owner,
|
||||
isPreparationTokenCurrent: () => true,
|
||||
capturePreparationInput: async (
|
||||
authority,
|
||||
reason,
|
||||
snapshotRevision
|
||||
): Promise<DirectSshPreparationInput> => ({
|
||||
...authority,
|
||||
catalogRevision: 1,
|
||||
repoRefs: [{ repoId: 'repo-a', executionHostId: `ssh:${TARGET_ID}` }],
|
||||
authorityRequirement: 'required',
|
||||
reason,
|
||||
snapshotRevision
|
||||
}),
|
||||
prepareOnly: async (input) => ({
|
||||
status: 'degraded' as const,
|
||||
token: {
|
||||
authority: owner,
|
||||
catalogRevision: 1,
|
||||
repoFingerprint: 'fp',
|
||||
authorityRequirement: 'required' as const,
|
||||
snapshotRevision: input.snapshotRevision ?? null,
|
||||
outcome: 'degraded' as const
|
||||
},
|
||||
repoOutcomes: {
|
||||
complete: 0,
|
||||
'non-authoritative': 1,
|
||||
'timed-out': 0,
|
||||
'cancel-budget-exhausted': 0,
|
||||
canceled: 0,
|
||||
stale: 0,
|
||||
rejected: 0
|
||||
},
|
||||
lineageOutcome: 'degraded' as const
|
||||
}),
|
||||
finalizeHydratedTerminals: () => 0
|
||||
})
|
||||
}
|
||||
|
||||
/** A local edit the persistence subscriber must want to write, then let it settle. */
|
||||
async function touchSessionAndSettle(marker: string): Promise<void> {
|
||||
await vi.advanceTimersByTimeAsync(WRITE_SUPPRESSION_MS)
|
||||
useAppStore.setState({ activeTabId: marker })
|
||||
await vi.advanceTimersByTimeAsync(DEBOUNCE_MS)
|
||||
}
|
||||
|
||||
function uploadedTargetIds(): string[] {
|
||||
return uploads.mock.calls.flatMap((call) => call[0]?.hydratedTargetIds ?? [])
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
uploads.mockClear()
|
||||
installWindowApi()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
cleanup()
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
describe('uploads from a client that could not place the host tabs', () => {
|
||||
it('issues no upload for a target whose host tabs it could not place', async () => {
|
||||
seedStore(false)
|
||||
const persistence = renderHook(() => useAppSessionPersistence())
|
||||
const sync = createSync()
|
||||
|
||||
expect(await sync.applyUnsolicitedSnapshot(TARGET_ID, snapshot())).toBe('unplaced')
|
||||
expect(useAppStore.getState().tabsByWorktree[WORKTREE_ID]).toBeUndefined()
|
||||
|
||||
// Mid-chain: retries are still running and the target is not authoritative yet.
|
||||
await touchSessionAndSettle('mid-chain')
|
||||
expect(
|
||||
uploadedTargetIds(),
|
||||
'a client that adopted none of the host tabs uploaded over them'
|
||||
).not.toContain(TARGET_ID)
|
||||
|
||||
// Still no upload once everything has settled: not adopting is not permission to overwrite the host.
|
||||
await vi.advanceTimersByTimeAsync(TOTAL_CHAIN_MS)
|
||||
expect(useAppStore.getState().remoteWorkspaceSyncStatusByTargetId[TARGET_ID]?.phase).toBe(
|
||||
'error'
|
||||
)
|
||||
await touchSessionAndSettle('post-exhaustion')
|
||||
expect(
|
||||
uploadedTargetIds(),
|
||||
'exhaustion authorised a replace-session upload built from an incomplete projection'
|
||||
).not.toContain(TARGET_ID)
|
||||
|
||||
sync.stop()
|
||||
persistence.unmount()
|
||||
})
|
||||
|
||||
it('does upload for a target whose host tabs it did place', async () => {
|
||||
seedStore(true)
|
||||
const persistence = renderHook(() => useAppSessionPersistence())
|
||||
const sync = createSync()
|
||||
|
||||
expect(await sync.applyUnsolicitedSnapshot(TARGET_ID, snapshot())).toBe('placed')
|
||||
expect(useAppStore.getState().tabsByWorktree[WORKTREE_ID]?.map((tab) => tab.id)).toEqual([
|
||||
'T1',
|
||||
'T2'
|
||||
])
|
||||
|
||||
await touchSessionAndSettle('placed')
|
||||
|
||||
// Without this the suppression assertions above would pass on a harness that never uploads.
|
||||
expect(uploadedTargetIds()).toContain(TARGET_ID)
|
||||
|
||||
sync.stop()
|
||||
persistence.unmount()
|
||||
})
|
||||
})
|
||||
@@ -160,16 +160,23 @@ export async function applyDirectSshRemoteWorkspaceSnapshot({
|
||||
lastSyncedAt: Date.now()
|
||||
})
|
||||
} else {
|
||||
// Why not hydrated: the host listed tabs on paths this client cannot place yet, so adopting
|
||||
// zero of them is not the host's picture. Marking hydrated would freeze that emptiness in —
|
||||
// nothing re-pulls a hydrated target — and would flip the seeding gate in
|
||||
// Why not hydrated: the host listed tabs on paths this client cannot place, so adopting zero
|
||||
// of them is not the host's picture. Hydrating would authorise 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 — deleting the
|
||||
// very tabs we failed to adopt. It would also flip the seeding gate in
|
||||
// workspace-terminal-host-authority.ts from `unverifiable` to `none`, authorising a fresh
|
||||
// default tab beside the host's orphaned ones (STA-3593).
|
||||
// default tab beside the host's orphaned ones (STA-3593). The next connect or host push
|
||||
// re-pulls, so leaving this unresolved costs a retry, not the data.
|
||||
currentStore.setRemoteWorkspaceSyncStatus(authority.targetId, {
|
||||
phase: 'pulling',
|
||||
phase: 'error',
|
||||
direction: 'pull',
|
||||
revision: snapshot.revision,
|
||||
updatedAt: snapshot.updatedAt
|
||||
updatedAt: snapshot.updatedAt,
|
||||
message: translate(
|
||||
'auto.hooks.useIpcEvents.2fe88c2e06',
|
||||
'Remote workspace sync unavailable'
|
||||
)
|
||||
})
|
||||
}
|
||||
const reconnectAbort = new AbortController()
|
||||
|
||||
@@ -19,7 +19,6 @@ import {
|
||||
applyDirectSshRemoteWorkspaceSnapshot,
|
||||
type DirectSshSnapshotPlacement
|
||||
} from './remote-workspace-snapshot-apply'
|
||||
import { createUnplacedSnapshotRepull } from './remote-workspace-unplaced-snapshot-repull'
|
||||
|
||||
const WORKSPACE_HYDRATION_TIMEOUT_MS = 10_000
|
||||
|
||||
@@ -122,29 +121,6 @@ export function createRemoteWorkspaceTargetSync(
|
||||
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().setRemoteWorkspaceSyncStatus(targetId, {
|
||||
phase: 'error',
|
||||
direction: 'pull',
|
||||
message: translate(
|
||||
'auto.hooks.useIpcEvents.2fe88c2e06',
|
||||
'Remote workspace sync unavailable'
|
||||
)
|
||||
})
|
||||
}
|
||||
})
|
||||
|
||||
const waitForWorkspaceSessionReady = async (): Promise<boolean> => {
|
||||
const deadline = Date.now() + WORKSPACE_HYDRATION_TIMEOUT_MS
|
||||
while (!stopped && Date.now() < deadline) {
|
||||
@@ -174,9 +150,6 @@ export function createRemoteWorkspaceTargetSync(
|
||||
})
|
||||
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(
|
||||
@@ -204,7 +177,7 @@ export function createRemoteWorkspaceTargetSync(
|
||||
if (snapshot.revision > 0) {
|
||||
const applyToken = buildDirectSshSnapshotApplyToken(token, snapshot.revision)
|
||||
if (applyToken) {
|
||||
const placement = await applyDirectSshRemoteWorkspaceSnapshot({
|
||||
await applyDirectSshRemoteWorkspaceSnapshot({
|
||||
store: deps.store,
|
||||
snapshot,
|
||||
token: applyToken,
|
||||
@@ -214,7 +187,6 @@ export function createRemoteWorkspaceTargetSync(
|
||||
waitForWorkspaceSessionReady,
|
||||
finalizeHydratedTerminals: deps.finalizeHydratedTerminals
|
||||
})
|
||||
repull.schedule(authority.targetId, placement)
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -286,7 +258,6 @@ export function createRemoteWorkspaceTargetSync(
|
||||
snapshot: RemoteWorkspaceSnapshot
|
||||
): Promise<DirectSshSnapshotPlacement> => {
|
||||
const placement = await applyPreparedSnapshot(targetId, snapshot)
|
||||
repull.schedule(targetId, placement)
|
||||
return placement
|
||||
}
|
||||
|
||||
@@ -296,7 +267,6 @@ export function createRemoteWorkspaceTargetSync(
|
||||
stop: () => {
|
||||
stopped = true
|
||||
arrivalByTarget.clear()
|
||||
repull.stop()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,283 +0,0 @@
|
||||
/**
|
||||
* The re-pull chain as the target sync actually wires it (STA-3593): a snapshot whose tab rows
|
||||
* cannot be placed leaves the target un-hydrated and re-pulls, because the missing input is the
|
||||
* host catalog, and a later attempt can get it. Covers the seam between
|
||||
* `applyDirectSshRemoteWorkspaceSnapshot`'s placement verdict and `createUnplacedSnapshotRepull`
|
||||
* against a real store, so the chain is judged on hydration state rather than on mock traffic.
|
||||
*/
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type * as AgentStatusModule from '@/lib/agent-status'
|
||||
import type { RemoteWorkspaceSnapshot } from '../../../shared/remote-workspace-types'
|
||||
import type { DirectSshAuthority, SshProviderEpoch } from '../../../shared/ssh-types'
|
||||
import { createTestStore, makeWorktree } from '../store/slices/store-test-helpers'
|
||||
import type { DirectSshPreparationInput } from './direct-ssh-reconnect-coordinator'
|
||||
import { createRemoteWorkspaceTargetSync } from './remote-workspace-target-sync'
|
||||
import { UNPLACED_SNAPSHOT_REPULL_DELAYS_MS } from './remote-workspace-unplaced-snapshot-repull'
|
||||
|
||||
vi.mock('sonner', () => ({ toast: { info: vi.fn(), success: vi.fn(), error: vi.fn() } }))
|
||||
vi.mock('@/lib/agent-status', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof AgentStatusModule>()
|
||||
return { ...actual, detectAgentStatusFromTitle: vi.fn().mockReturnValue(null) }
|
||||
})
|
||||
|
||||
const TARGET_ID = 'target-a'
|
||||
const REPO_ROOT = '/srv/proj'
|
||||
const HOST_PATH = `${REPO_ROOT}/alpha`
|
||||
const WORKTREE_ID = `repo-a::${HOST_PATH}`
|
||||
const TOTAL_CHAIN_MS = UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.reduce((sum, ms) => sum + ms, 0)
|
||||
|
||||
const owner: DirectSshAuthority = {
|
||||
targetId: TARGET_ID,
|
||||
providerEpoch: 'epoch-a' as SshProviderEpoch,
|
||||
connectionGeneration: 1
|
||||
}
|
||||
|
||||
/** Two host terminals living on a host path this client has no catalog row for. */
|
||||
function snapshot(revision = 4): RemoteWorkspaceSnapshot {
|
||||
return {
|
||||
namespace: 'workspace',
|
||||
revision,
|
||||
updatedAt: revision,
|
||||
schemaVersion: 1,
|
||||
session: {
|
||||
activeWorktreePath: HOST_PATH,
|
||||
activeTabId: 'T1',
|
||||
tabsByWorktreePath: {
|
||||
[HOST_PATH]: [0, 1].map((index) => ({
|
||||
id: `T${index + 1}`,
|
||||
worktreePath: HOST_PATH,
|
||||
ptyId: `pty-T${index + 1}`,
|
||||
title: `Terminal ${index + 1}`,
|
||||
customTitle: null,
|
||||
color: null,
|
||||
sortOrder: index,
|
||||
createdAt: index + 1
|
||||
}))
|
||||
},
|
||||
terminalLayoutsByTabId: {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type TestStore = ReturnType<typeof createTestStore>
|
||||
|
||||
function createHarness() {
|
||||
const store = createTestStore()
|
||||
store.setState({
|
||||
workspaceSessionReady: true,
|
||||
repos: [
|
||||
{
|
||||
id: 'repo-a',
|
||||
path: REPO_ROOT,
|
||||
displayName: 'Proj',
|
||||
badgeColor: '#000',
|
||||
addedAt: 0,
|
||||
connectionId: TARGET_ID,
|
||||
executionHostId: `ssh:${TARGET_ID}`
|
||||
} as never
|
||||
],
|
||||
reconnectPersistedTerminals: (async () => {}) as never
|
||||
})
|
||||
|
||||
let authority: DirectSshAuthority | null = owner
|
||||
const get = vi.fn(async () => snapshot())
|
||||
const sync = createRemoteWorkspaceTargetSync({
|
||||
store,
|
||||
remoteWorkspace: { get, setForConnectedTargets: vi.fn(async () => []) },
|
||||
getCurrentAuthority: () => authority,
|
||||
isPreparationTokenCurrent: () => authority !== null,
|
||||
capturePreparationInput: async (
|
||||
current,
|
||||
reason,
|
||||
snapshotRevision
|
||||
): Promise<DirectSshPreparationInput> => ({
|
||||
...current,
|
||||
catalogRevision: 1,
|
||||
repoRefs: [{ repoId: 'repo-a', executionHostId: `ssh:${TARGET_ID}` }],
|
||||
authorityRequirement: 'required',
|
||||
reason,
|
||||
snapshotRevision
|
||||
}),
|
||||
// Degraded preparation still issues a token on purpose, so terminals reconnect.
|
||||
prepareOnly: async (input) => ({
|
||||
status: 'degraded' as const,
|
||||
token: {
|
||||
authority: owner,
|
||||
catalogRevision: 1,
|
||||
repoFingerprint: 'fp',
|
||||
authorityRequirement: 'required' as const,
|
||||
snapshotRevision: input.snapshotRevision ?? null,
|
||||
outcome: 'degraded' as const
|
||||
},
|
||||
repoOutcomes: {
|
||||
complete: 0,
|
||||
'non-authoritative': 1,
|
||||
'timed-out': 0,
|
||||
'cancel-budget-exhausted': 0,
|
||||
canceled: 0,
|
||||
stale: 0,
|
||||
rejected: 0
|
||||
},
|
||||
lineageOutcome: 'degraded' as const
|
||||
}),
|
||||
finalizeHydratedTerminals: () => 0
|
||||
})
|
||||
|
||||
return {
|
||||
store,
|
||||
sync,
|
||||
get,
|
||||
dropAuthority: () => {
|
||||
authority = null
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** What the degraded lineage read failed to deliver. */
|
||||
function landHostCatalog(store: TestStore): void {
|
||||
store.setState({
|
||||
worktreesByRepo: {
|
||||
'repo-a': [
|
||||
makeWorktree({
|
||||
id: WORKTREE_ID,
|
||||
repoId: 'repo-a',
|
||||
path: HOST_PATH,
|
||||
hostId: `ssh:${TARGET_ID}`
|
||||
} as never)
|
||||
]
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
function isHydrated(store: TestStore): boolean {
|
||||
return store.getState().remoteWorkspaceHydratedTargetIds.has(TARGET_ID)
|
||||
}
|
||||
|
||||
function syncPhase(store: TestStore): string | undefined {
|
||||
return store.getState().remoteWorkspaceSyncStatusByTargetId[TARGET_ID]?.phase
|
||||
}
|
||||
|
||||
function adoptedTabIds(store: TestStore): string[] {
|
||||
return Object.values(store.getState().tabsByWorktree)
|
||||
.flat()
|
||||
.map((tab) => tab.id)
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
describe('target sync re-pulling a snapshot it could not place', () => {
|
||||
it('leaves the target un-hydrated and re-pulls while the catalog is missing', async () => {
|
||||
const harness = createHarness()
|
||||
|
||||
const placement = await harness.sync.applyUnsolicitedSnapshot(TARGET_ID, snapshot())
|
||||
|
||||
expect(placement).toBe('unplaced')
|
||||
expect(adoptedTabIds(harness.store)).toEqual([])
|
||||
expect(isHydrated(harness.store), 'an unplaceable snapshot is not the host picture').toBe(false)
|
||||
expect(syncPhase(harness.store)).toBe('pulling')
|
||||
|
||||
await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0])
|
||||
expect(harness.get, 'the first retry did not re-read the host').toHaveBeenCalledTimes(1)
|
||||
expect(isHydrated(harness.store)).toBe(false)
|
||||
})
|
||||
|
||||
it('adopts the host tabs and hydrates once a retry finds the catalog', async () => {
|
||||
const harness = createHarness()
|
||||
await harness.sync.applyUnsolicitedSnapshot(TARGET_ID, snapshot())
|
||||
|
||||
landHostCatalog(harness.store)
|
||||
await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0])
|
||||
|
||||
expect(adoptedTabIds(harness.store), 'the retry recovered the host terminals').toEqual([
|
||||
'T1',
|
||||
'T2'
|
||||
])
|
||||
expect(isHydrated(harness.store)).toBe(true)
|
||||
expect(syncPhase(harness.store)).toBe('synced')
|
||||
|
||||
// The chain retired on success rather than re-pulling over a correctly hydrated target.
|
||||
await vi.advanceTimersByTimeAsync(TOTAL_CHAIN_MS * 10)
|
||||
expect(harness.get).toHaveBeenCalledTimes(1)
|
||||
expect(adoptedTabIds(harness.store)).toEqual(['T1', 'T2'])
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('settles the target after a bounded number of retries instead of pulling forever', async () => {
|
||||
const harness = createHarness()
|
||||
await harness.sync.applyUnsolicitedSnapshot(TARGET_ID, snapshot())
|
||||
|
||||
await vi.advanceTimersByTimeAsync(TOTAL_CHAIN_MS)
|
||||
expect(harness.get).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length)
|
||||
|
||||
// Staying un-hydrated is the data-safety property, not an oversight: hydration authorises
|
||||
// uploads, and an upload is a `replace-session` patch that wholesale replaces the host
|
||||
// snapshot. Hydrating on a picture known to be missing rows would delete the very tabs we
|
||||
// failed to adopt. Suppressed uploads are recoverable; a wiped host snapshot is not.
|
||||
expect(
|
||||
isHydrated(harness.store),
|
||||
'exhaustion authorised uploads from a picture known to be incomplete'
|
||||
).toBe(false)
|
||||
expect(syncPhase(harness.store)).toBe('error')
|
||||
|
||||
await vi.advanceTimersByTimeAsync(TOTAL_CHAIN_MS * 10)
|
||||
expect(harness.get, 'the chain kept re-pulling past exhaustion').toHaveBeenCalledTimes(
|
||||
UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length
|
||||
)
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('re-pulls nothing after stop()', async () => {
|
||||
const harness = createHarness()
|
||||
await harness.sync.applyUnsolicitedSnapshot(TARGET_ID, snapshot())
|
||||
|
||||
harness.sync.stop()
|
||||
|
||||
await vi.advanceTimersByTimeAsync(TOTAL_CHAIN_MS * 10)
|
||||
expect(harness.get).not.toHaveBeenCalled()
|
||||
expect(isHydrated(harness.store)).toBe(false)
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('abandons the chain when the target loses its authority', async () => {
|
||||
const harness = createHarness()
|
||||
await harness.sync.applyUnsolicitedSnapshot(TARGET_ID, snapshot())
|
||||
|
||||
harness.dropAuthority()
|
||||
|
||||
await vi.advanceTimersByTimeAsync(TOTAL_CHAIN_MS * 10)
|
||||
expect(harness.get).not.toHaveBeenCalled()
|
||||
expect(isHydrated(harness.store)).toBe(false)
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('target sync applying a snapshot it can place', () => {
|
||||
it('hydrates and marks synced exactly as before the unplaced gate existed', async () => {
|
||||
const harness = createHarness()
|
||||
landHostCatalog(harness.store)
|
||||
|
||||
const placement = await harness.sync.applyUnsolicitedSnapshot(TARGET_ID, snapshot())
|
||||
|
||||
expect(placement).toBe('placed')
|
||||
expect(harness.store.getState().tabsByWorktree[WORKTREE_ID]?.map((tab) => tab.id)).toEqual([
|
||||
'T1',
|
||||
'T2'
|
||||
])
|
||||
expect(isHydrated(harness.store)).toBe(true)
|
||||
expect(syncPhase(harness.store)).toBe('synced')
|
||||
expect(harness.store.getState().remoteWorkspaceSyncStatusByTargetId[TARGET_ID]?.revision).toBe(
|
||||
4
|
||||
)
|
||||
|
||||
// No re-pull chain is armed on the happy path at all.
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
await vi.advanceTimersByTimeAsync(TOTAL_CHAIN_MS * 10)
|
||||
expect(harness.get).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
@@ -1,510 +0,0 @@
|
||||
/**
|
||||
* The re-pull chain that follows an `unplaced` snapshot apply (STA-3593). It exists because an
|
||||
* unplaceable tab row is `unverifiable` — the local catalog has not landed — so the client retries
|
||||
* the missing input rather than accepting the empty picture.
|
||||
*
|
||||
* Everything here is about the chain STOPPING: a retry loop that outlives its cause is worse than
|
||||
* the bug it fixes, because each attempt re-runs the host catalog and lineage reads against a host
|
||||
* that may be down. The oracles are boundedness, retirement on success, and silence after `stop()`.
|
||||
*/
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { RemoteWorkspaceSnapshot } from '../../../shared/remote-workspace-types'
|
||||
import type { DirectSshSnapshotPlacement } from './remote-workspace-snapshot-apply'
|
||||
import {
|
||||
createUnplacedSnapshotRepull,
|
||||
UNPLACED_SNAPSHOT_REPULL_DELAYS_MS
|
||||
} from './remote-workspace-unplaced-snapshot-repull'
|
||||
|
||||
const TARGET = 'target-a'
|
||||
const OTHER_TARGET = 'target-b'
|
||||
const LONG_AFTER_MS = 600_000
|
||||
const TOTAL_CHAIN_MS = UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.reduce((sum, ms) => sum + ms, 0)
|
||||
|
||||
function snapshot(revision = 7): RemoteWorkspaceSnapshot {
|
||||
return {
|
||||
namespace: 'workspace',
|
||||
revision,
|
||||
updatedAt: revision,
|
||||
schemaVersion: 1,
|
||||
session: {
|
||||
activeWorktreePath: null,
|
||||
activeTabId: null,
|
||||
tabsByWorktreePath: {},
|
||||
terminalLayoutsByTabId: {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** 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<RemoteWorkspaceSnapshot | null>((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
|
||||
getSnapshot?: (targetId: string) => Promise<RemoteWorkspaceSnapshot | null>
|
||||
}
|
||||
|
||||
function createHarness(options: HarnessOptions = {}) {
|
||||
let stopped = false
|
||||
const targetsWithAuthority = new Set([TARGET, OTHER_TARGET])
|
||||
const appliedTargets: string[] = []
|
||||
const exhaustedTargets: string[] = []
|
||||
|
||||
const getSnapshot = vi.fn(options.getSnapshot ?? (async () => snapshot()))
|
||||
const applySnapshot = vi.fn(async (targetId: string): Promise<DirectSshSnapshotPlacement> => {
|
||||
const call = appliedTargets.length
|
||||
appliedTargets.push(targetId)
|
||||
return options.placementByCall?.(call) ?? 'unplaced'
|
||||
})
|
||||
|
||||
const repull = createUnplacedSnapshotRepull({
|
||||
isStopped: () => stopped,
|
||||
hasCurrentAuthority: (targetId) => targetsWithAuthority.has(targetId),
|
||||
getSnapshot,
|
||||
applySnapshot,
|
||||
reportExhausted: (targetId) => exhaustedTargets.push(targetId)
|
||||
})
|
||||
|
||||
return {
|
||||
repull,
|
||||
getSnapshot,
|
||||
applySnapshot,
|
||||
appliedTargets,
|
||||
exhaustedTargets,
|
||||
stop: () => {
|
||||
stopped = true
|
||||
},
|
||||
dropAuthority: (targetId: string) => targetsWithAuthority.delete(targetId),
|
||||
restoreAuthority: (targetId: string) => targetsWithAuthority.add(targetId)
|
||||
}
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
describe('the re-pull chain after an unplaced snapshot', () => {
|
||||
it('keeps the documented backoff so a retry cannot become a busy loop', () => {
|
||||
expect(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS).toEqual([1_000, 3_000, 8_000])
|
||||
})
|
||||
|
||||
it('runs exactly one attempt per configured delay, then exhausts and stops for good', async () => {
|
||||
const harness = createHarness()
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
|
||||
let elapsed = 0
|
||||
for (const [index, delayMs] of UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.entries()) {
|
||||
await vi.advanceTimersByTimeAsync(delayMs - 1)
|
||||
expect(
|
||||
harness.applySnapshot,
|
||||
`attempt ${index} fired before its ${delayMs}ms delay`
|
||||
).toHaveBeenCalledTimes(index)
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
expect(harness.applySnapshot).toHaveBeenCalledTimes(index + 1)
|
||||
elapsed += delayMs
|
||||
}
|
||||
|
||||
// Every attempt came back `unplaced`, so the chain has nothing left to try.
|
||||
expect(harness.exhaustedTargets).toEqual([TARGET])
|
||||
expect(elapsed).toBe(12_000)
|
||||
|
||||
// The oracle: no fourth attempt, ever. Not a slower one, not a later one.
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
expect(harness.applySnapshot).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length)
|
||||
expect(harness.getSnapshot).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length)
|
||||
expect(harness.exhaustedTargets, 'exhaustion reported once, not once per idle tick').toEqual([
|
||||
TARGET
|
||||
])
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('retires the chain as soon as a retry places the rows', async () => {
|
||||
const harness = createHarness({ placementByCall: () => 'placed' })
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
|
||||
await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0])
|
||||
expect(harness.appliedTargets).toEqual([TARGET])
|
||||
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
expect(
|
||||
harness.applySnapshot,
|
||||
'a re-pull over a target that has since hydrated would undo the successful placement'
|
||||
).toHaveBeenCalledTimes(1)
|
||||
expect(harness.exhaustedTargets).toEqual([])
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it.each(['placed', 'not-applied'] as const)(
|
||||
'cancels a pending attempt when a newer arrival reports %s',
|
||||
async (placement) => {
|
||||
const harness = createHarness()
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0] - 100)
|
||||
|
||||
// A snapshot that arrived on its own beat the pending retry to the answer.
|
||||
harness.repull.schedule(TARGET, placement)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
expect(harness.applySnapshot, 'the superseded attempt still fired').not.toHaveBeenCalled()
|
||||
expect(harness.exhaustedTargets).toEqual([])
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
}
|
||||
)
|
||||
|
||||
it('lets a concurrent unplaced report ride the armed wait instead of restarting it', async () => {
|
||||
const harness = createHarness()
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0] - 100)
|
||||
|
||||
// An unsolicited `workspace.changed` push reports its own unplaced apply mid-wait. Restarting
|
||||
// the backoff here is how a chatty host turns a bounded chain into a permanent one.
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
expect(vi.getTimerCount(), 'the concurrent report armed a second timer').toBe(1)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(99)
|
||||
expect(harness.applySnapshot).not.toHaveBeenCalled()
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
expect(
|
||||
harness.applySnapshot,
|
||||
'the concurrent report pushed the first attempt past its original deadline'
|
||||
).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('exhausts once and only once under a host that keeps reporting', async () => {
|
||||
const harness = createHarness()
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
|
||||
// An unsolicited `workspace.changed` push reports its own unplaced apply. Over SSH the window
|
||||
// between a retry firing and its apply resolving is a full IPC round trip, so pushes land
|
||||
// inside it routinely.
|
||||
for (let elapsed = 0; elapsed < 60_000; elapsed += 100) {
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
await vi.advanceTimersByTimeAsync(100)
|
||||
}
|
||||
|
||||
// The bound itself: a chatty host must not walk the chain back to attempt 0.
|
||||
expect(
|
||||
harness.applySnapshot,
|
||||
'a repeatedly re-armed chain never reached its bound'
|
||||
).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length)
|
||||
expect(harness.getSnapshot, 'the chain kept re-dialling workspace.get').toHaveBeenCalledTimes(
|
||||
UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length
|
||||
)
|
||||
expect(
|
||||
harness.exhaustedTargets,
|
||||
'every later report re-announced exhaustion, and each one re-marks the target hydrated'
|
||||
).toEqual([TARGET])
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('earns a full fresh chain after resetTarget, not an instant re-exhaustion', async () => {
|
||||
const harness = createHarness()
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
await vi.advanceTimersByTimeAsync(TOTAL_CHAIN_MS)
|
||||
expect(harness.applySnapshot).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length)
|
||||
expect(harness.exhaustedTargets).toEqual([TARGET])
|
||||
|
||||
// A new connection: the next lineage read may well succeed, so the spent count must not
|
||||
// outlive the connection that spent it.
|
||||
harness.repull.resetTarget(TARGET)
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
|
||||
for (const [index, delayMs] of UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.entries()) {
|
||||
await vi.advanceTimersByTimeAsync(delayMs - 1)
|
||||
expect(
|
||||
harness.applySnapshot,
|
||||
`post-reset attempt ${index} did not wait its ${delayMs}ms`
|
||||
).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length + index)
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
expect(harness.applySnapshot).toHaveBeenCalledTimes(
|
||||
UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length + index + 1
|
||||
)
|
||||
}
|
||||
|
||||
expect(
|
||||
harness.applySnapshot,
|
||||
'the reset target did not get a full second chain'
|
||||
).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length * 2)
|
||||
expect(harness.exhaustedTargets).toEqual([TARGET, TARGET])
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('drops an armed attempt when its target is reset', async () => {
|
||||
const harness = createHarness()
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
|
||||
harness.repull.resetTarget(TARGET)
|
||||
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
expect(harness.applySnapshot).not.toHaveBeenCalled()
|
||||
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')
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
|
||||
harness.repull.stop()
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
expect(harness.getSnapshot).not.toHaveBeenCalled()
|
||||
expect(harness.applySnapshot).not.toHaveBeenCalled()
|
||||
expect(harness.exhaustedTargets).toEqual([])
|
||||
})
|
||||
|
||||
it('does not apply or reschedule when stop() lands mid-flight', async () => {
|
||||
let releaseSnapshot: (value: RemoteWorkspaceSnapshot | null) => void = () => {}
|
||||
const pending = new Promise<RemoteWorkspaceSnapshot | null>((resolve) => {
|
||||
releaseSnapshot = resolve
|
||||
})
|
||||
const harness = createHarness({ getSnapshot: () => pending })
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
|
||||
await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0])
|
||||
expect(harness.getSnapshot).toHaveBeenCalledOnce()
|
||||
|
||||
// The hook tore down while the host read was outstanding; its answer arrives afterwards.
|
||||
harness.stop()
|
||||
harness.repull.stop()
|
||||
releaseSnapshot(snapshot())
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
|
||||
expect(harness.applySnapshot, 'a torn-down sync applied a late snapshot').not.toHaveBeenCalled()
|
||||
expect(harness.exhaustedTargets).toEqual([])
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('refuses to schedule at all once stopped', async () => {
|
||||
const harness = createHarness()
|
||||
harness.stop()
|
||||
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
expect(harness.applySnapshot).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('abandons the chain when the target lost its authority, stranding no attempt count', async () => {
|
||||
const harness = createHarness()
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
|
||||
harness.dropAuthority(TARGET)
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
|
||||
expect(harness.applySnapshot).not.toHaveBeenCalled()
|
||||
expect(harness.exhaustedTargets, 'a target that went away is not an exhausted chain').toEqual(
|
||||
[]
|
||||
)
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
|
||||
// The spent attempt must not outlive the connection. If it did, the target would come back
|
||||
// with a partly burned chain — observable here as a short first wait and a short chain.
|
||||
harness.restoreAuthority(TARGET)
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0] - 1)
|
||||
expect(
|
||||
harness.applySnapshot,
|
||||
'the recovered target resumed mid-backoff instead of starting over'
|
||||
).not.toHaveBeenCalled()
|
||||
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
expect(harness.applySnapshot).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length)
|
||||
expect(harness.exhaustedTargets).toEqual([TARGET])
|
||||
})
|
||||
|
||||
it('keeps the final armed attempt when a concurrent report lands after the count is spent', async () => {
|
||||
const harness = createHarness()
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
for (const delayMs of UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.slice(0, -1)) {
|
||||
await vi.advanceTimersByTimeAsync(delayMs)
|
||||
}
|
||||
// The last attempt is armed, so the count is already spent while the retry is still pending.
|
||||
expect(harness.applySnapshot).toHaveBeenCalledTimes(
|
||||
UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length - 1
|
||||
)
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
|
||||
// An unsolicited host push reports its own unplaced apply against that spent count.
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
|
||||
expect(
|
||||
harness.exhaustedTargets,
|
||||
'exhaustion was declared while the last retry was still pending'
|
||||
).toEqual([])
|
||||
expect(vi.getTimerCount(), 'the concurrent report cancelled the final retry').toBe(1)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.at(-1) ?? 0)
|
||||
expect(harness.applySnapshot, 'the chain silently lost its last attempt').toHaveBeenCalledTimes(
|
||||
UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length
|
||||
)
|
||||
expect(harness.exhaustedTargets).toEqual([TARGET])
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('carries on after a single transient host read failure', async () => {
|
||||
let call = 0
|
||||
const harness = createHarness({
|
||||
getSnapshot: async () => {
|
||||
call += 1
|
||||
if (call === 2) {
|
||||
throw new Error('ssh channel closed')
|
||||
}
|
||||
return snapshot()
|
||||
}
|
||||
})
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
|
||||
// The failed attempt still counts as unplaced, so the chain walks on and settles the target
|
||||
// rather than stranding it on `pulling`, un-hydrated, with no timer armed.
|
||||
expect(harness.getSnapshot).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length)
|
||||
expect(
|
||||
harness.applySnapshot,
|
||||
'the chain stalled on the transient failure'
|
||||
).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length - 1)
|
||||
expect(harness.exhaustedTargets).toEqual([TARGET])
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it.each([
|
||||
['returns no snapshot', async () => null],
|
||||
['returns a revision-zero snapshot', async () => snapshot(0)],
|
||||
[
|
||||
'throws',
|
||||
async () => {
|
||||
throw new Error('ssh channel closed')
|
||||
}
|
||||
]
|
||||
])(
|
||||
'walks the chain to exhaustion when the host read %s',
|
||||
async (_label, getSnapshot: () => Promise<RemoteWorkspaceSnapshot | null>) => {
|
||||
const harness = createHarness({ getSnapshot })
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
|
||||
// Dying here instead would strand the target on `pulling`, un-hydrated — and an un-hydrated
|
||||
// target is filtered out of `hydratedTargetIds`, so it would never upload again.
|
||||
expect(harness.getSnapshot).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length)
|
||||
expect(harness.applySnapshot).not.toHaveBeenCalled()
|
||||
expect(harness.exhaustedTargets, 'the chain died silently').toEqual([TARGET])
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
}
|
||||
)
|
||||
|
||||
it('runs one independent chain per target', async () => {
|
||||
const harness = createHarness({
|
||||
// target-a keeps failing to place; target-b succeeds on its first retry.
|
||||
placementByCall: (call) => (call === 0 ? 'placed' : 'unplaced')
|
||||
})
|
||||
harness.repull.schedule(OTHER_TARGET, 'unplaced')
|
||||
harness.repull.schedule(TARGET, 'unplaced')
|
||||
expect(vi.getTimerCount(), 'targets share a timer slot').toBe(2)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0])
|
||||
expect(harness.appliedTargets).toEqual([OTHER_TARGET, TARGET])
|
||||
|
||||
await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
|
||||
// target-b retired after one attempt; target-a burned its full chain.
|
||||
expect(harness.appliedTargets.filter((id) => id === OTHER_TARGET)).toHaveLength(1)
|
||||
expect(harness.appliedTargets.filter((id) => id === TARGET)).toHaveLength(
|
||||
UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length
|
||||
)
|
||||
expect(harness.exhaustedTargets).toEqual([TARGET])
|
||||
})
|
||||
})
|
||||
@@ -1,196 +0,0 @@
|
||||
import type { RemoteWorkspaceSnapshot } from '../../../shared/remote-workspace-types'
|
||||
import type { DirectSshSnapshotPlacement } from './remote-workspace-snapshot-apply'
|
||||
|
||||
/**
|
||||
* Backoff for re-pulling after a snapshot whose terminal-tab rows this client could not place.
|
||||
* Each attempt re-runs the host catalog + lineage reads, so it retries the *missing input* — a
|
||||
* degraded `listLineage` that left `worktreesByRepo` empty — not merely the pull. Bounded: past
|
||||
* this the target stays un-hydrated, which the seeding gate already reads as `unverifiable`.
|
||||
*/
|
||||
export const UNPLACED_SNAPSHOT_REPULL_DELAYS_MS = [1_000, 3_000, 8_000]
|
||||
|
||||
export type UnplacedSnapshotRepullDeps = {
|
||||
isStopped: () => boolean
|
||||
hasCurrentAuthority: (targetId: string) => boolean
|
||||
getSnapshot: (targetId: string) => Promise<RemoteWorkspaceSnapshot | null>
|
||||
applySnapshot: (
|
||||
targetId: string,
|
||||
snapshot: RemoteWorkspaceSnapshot
|
||||
) => Promise<DirectSshSnapshotPlacement>
|
||||
reportExhausted: (targetId: string) => void
|
||||
}
|
||||
|
||||
export type UnplacedSnapshotRepull = {
|
||||
/** Call with every apply outcome; anything but `unplaced` retires the target's pending chain. */
|
||||
schedule: (targetId: string, placement: DirectSshSnapshotPlacement) => void
|
||||
/**
|
||||
* Drop a target's spent attempt count so a new connection gets a fresh chain. Without this an
|
||||
* exhausted counter sticks, and every later reconnect re-exhausts immediately — the retry would
|
||||
* silently stop working for the rest of the session even when the next lineage read would succeed.
|
||||
*/
|
||||
resetTarget: (targetId: string) => void
|
||||
stop: () => void
|
||||
}
|
||||
|
||||
export function createUnplacedSnapshotRepull(
|
||||
deps: UnplacedSnapshotRepullDeps
|
||||
): UnplacedSnapshotRepull {
|
||||
const timerByTarget = new Map<string, ReturnType<typeof setTimeout>>()
|
||||
// Why the module owns this and callers cannot pass it: an unsolicited `workspace.changed` push
|
||||
// also reports a placement, and letting it re-arm at attempt 0 means a host that pushes faster
|
||||
// than the backoff resets the chain forever — an unbounded loop re-dialling `workspace.get` that
|
||||
// never reaches exhaustion. Only a placement that is not `unplaced` clears the count.
|
||||
const attemptByTarget = new Map<string, number>()
|
||||
// Why a separate set and not just the spent count: after exhaustion every further `unplaced`
|
||||
// report still lands here, and re-announcing it would re-mark the target hydrated and rewrite
|
||||
// its status on every host push. Exhaustion is announced once per chain.
|
||||
const exhaustedTargets = new Set<string>()
|
||||
// Why in-flight is tracked apart from the timer: the callback drops its timer entry before
|
||||
// awaiting the host, so the timer map alone leaves a gap in which a concurrent report arms a
|
||||
// second, overlapping chain and can trip exhaustion before the pending apply resolves.
|
||||
// Why it stores the owning generation rather than membership: a superseded callback's cleanup
|
||||
// must not release a marker a newer chain now owns.
|
||||
const inFlightByTarget = new Map<string, number>()
|
||||
// Why a generation: nothing can cancel a callback already past its await, so retirement and
|
||||
// reconnect reset both bump it and the callback exits if its stamp no longer matches. Without
|
||||
// bumping on retirement, an in-flight apply resolving `unplaced` after a `placed` report would
|
||||
// restart the chain from attempt 0 — repeat that and the budget never runs out.
|
||||
const generationByTarget = new Map<string, number>()
|
||||
|
||||
const bumpGeneration = (targetId: string): void => {
|
||||
generationByTarget.set(targetId, (generationByTarget.get(targetId) ?? 0) + 1)
|
||||
}
|
||||
|
||||
// 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<string>()
|
||||
|
||||
/** 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 => {
|
||||
const timer = timerByTarget.get(targetId)
|
||||
if (timer !== undefined) {
|
||||
clearTimeout(timer)
|
||||
timerByTarget.delete(targetId)
|
||||
}
|
||||
}
|
||||
|
||||
const schedule = (targetId: string, placement: DirectSshSnapshotPlacement): void => {
|
||||
if (placement !== 'unplaced' || deps.isStopped()) {
|
||||
// A `placed` or superseded apply retires the chain; a timer left armed would re-pull over a
|
||||
// target that has since hydrated correctly.
|
||||
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)) {
|
||||
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
|
||||
const delayMs = UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[attempt]
|
||||
if (delayMs === undefined) {
|
||||
if (!exhaustedTargets.has(targetId)) {
|
||||
exhaustedTargets.add(targetId)
|
||||
deps.reportExhausted(targetId)
|
||||
}
|
||||
return
|
||||
}
|
||||
attemptByTarget.set(targetId, attempt + 1)
|
||||
timerByTarget.set(
|
||||
targetId,
|
||||
setTimeout(() => {
|
||||
timerByTarget.delete(targetId)
|
||||
const generation = generationByTarget.get(targetId) ?? 0
|
||||
inFlightByTarget.set(targetId, generation)
|
||||
void (async () => {
|
||||
if (deps.isStopped()) {
|
||||
return
|
||||
}
|
||||
if (!deps.hasCurrentAuthority(targetId)) {
|
||||
// The target went away; a later connect calls `resetTarget` and starts over.
|
||||
attemptByTarget.delete(targetId)
|
||||
exhaustedTargets.delete(targetId)
|
||||
return
|
||||
}
|
||||
let outcome: DirectSshSnapshotPlacement = 'unplaced'
|
||||
try {
|
||||
const snapshot = await deps.getSnapshot(targetId)
|
||||
if (deps.isStopped()) {
|
||||
return
|
||||
}
|
||||
if (snapshot && snapshot.revision > 0) {
|
||||
outcome = await deps.applySnapshot(targetId, snapshot)
|
||||
}
|
||||
} catch {
|
||||
// Why the chain continues rather than dying here: a transient RPC failure that left no
|
||||
// timer armed would strand the target on `pulling` and un-hydrated forever — and an
|
||||
// un-hydrated target never uploads again. Staying `unplaced` walks the chain to
|
||||
// exhaustion, which settles it.
|
||||
}
|
||||
if ((generationByTarget.get(targetId) ?? 0) !== generation) {
|
||||
// A reconnect reset this target while we were awaiting; its own chain owns it now.
|
||||
return
|
||||
}
|
||||
// Released before rescheduling: the guard exists to keep *others* out, and holding it
|
||||
// here would block this chain's own next attempt.
|
||||
releaseInFlight(targetId, generation)
|
||||
schedule(targetId, outcome)
|
||||
})()
|
||||
.catch(() => {})
|
||||
.finally(() => {
|
||||
// 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)
|
||||
)
|
||||
}
|
||||
|
||||
return {
|
||||
schedule,
|
||||
resetTarget: (targetId: string) => {
|
||||
// Deliberately does NOT drop the in-flight marker: its owner is the only party that may
|
||||
// release it, and clearing it here would let a new chain arm while that apply is still
|
||||
// running. The superseded callback exits on the generation bump and releases its own marker.
|
||||
clearTimer(targetId)
|
||||
attemptByTarget.delete(targetId)
|
||||
exhaustedTargets.delete(targetId)
|
||||
pendingUnplacedTargets.delete(targetId)
|
||||
bumpGeneration(targetId)
|
||||
},
|
||||
stop: () => {
|
||||
for (const timer of timerByTarget.values()) {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
timerByTarget.clear()
|
||||
attemptByTarget.clear()
|
||||
exhaustedTargets.clear()
|
||||
inFlightByTarget.clear()
|
||||
pendingUnplacedTargets.clear()
|
||||
generationByTarget.clear()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,303 +0,0 @@
|
||||
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<RemoteWorkspaceSnapshot | null>
|
||||
setForConnectedTargets: (args: {
|
||||
session?: WorkspaceSessionState
|
||||
hydratedTargetIds?: string[]
|
||||
}) => Promise<{ targetId: string; result: RemoteWorkspacePatchResult }[]>
|
||||
}
|
||||
|
||||
export type RemoteWorkspaceTargetSyncDeps = {
|
||||
store: Pick<StoreApi<AppState>, 'getState'>
|
||||
remoteWorkspace: RemoteWorkspaceApi
|
||||
getCurrentAuthority: (targetId: string) => DirectSshAuthority | null
|
||||
isPreparationTokenCurrent: (token: DirectSshPreparationToken) => boolean
|
||||
capturePreparationInput: (
|
||||
authority: DirectSshAuthority,
|
||||
reason: 'workspace-snapshot',
|
||||
snapshotRevision: number
|
||||
) => Promise<DirectSshPreparationInput | null>
|
||||
prepareOnly: (input: DirectSshPreparationInput) => Promise<DirectSshPreparationOutcome>
|
||||
finalizeHydratedTerminals: (authority: DirectSshAuthority) => number
|
||||
}
|
||||
|
||||
export type RemoteWorkspaceTargetSync = {
|
||||
syncAfterConnect: (token: DirectSshPreparationToken) => Promise<void>
|
||||
applyUnsolicitedSnapshot: (
|
||||
targetId: string,
|
||||
snapshot: RemoteWorkspaceSnapshot
|
||||
) => Promise<DirectSshSnapshotPlacement>
|
||||
stop: () => void
|
||||
}
|
||||
|
||||
function exactTargetWorktreeIds(state: AppState, authority: DirectSshAuthority): Set<string> {
|
||||
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<string, number>()
|
||||
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<boolean> => {
|
||||
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<void> => {
|
||||
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<DirectSshSnapshotPlacement> => {
|
||||
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<DirectSshSnapshotPlacement> => {
|
||||
const placement = await applyPreparedSnapshot(targetId, snapshot)
|
||||
repull.schedule(targetId, placement)
|
||||
return placement
|
||||
}
|
||||
|
||||
return {
|
||||
syncAfterConnect,
|
||||
applyUnsolicitedSnapshot,
|
||||
stop: () => {
|
||||
stopped = true
|
||||
arrivalByTarget.clear()
|
||||
repull.stop()
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user