mirror of
https://github.com/stablyai/orca.git
synced 2026-10-31 00:57:45 +00:00
feat(sidebar): one lifecycle line per workspace in main, with in-progress hints (switch-gated) (#31553)
* fix(sidebar): land flip-a's switch-on fixes on main, still dark (no flip) The fixes found while the sidebar ran from main's feed, without turning that on: ORCA_SIDEBAR_INPUTS_IS_MAIN stays false, and none of flip-a's stop-gaps come along. - Main stops counting a closed tab's terminal once it was asked to stop and no pane shows it; a failed or refused stop takes the request back, and a host out of contact stays unverifiable. - Main fetches a paired server's rows with the transport's normal bound, so a server slower than 5 s still lists. - A deferred unread raise flips the folder row it was decided for, and a read while the owner check is out cancels it. - Phone: no worktree.ps flash before the feed; a host's last installed feed stands in while a fresh read is out; it highlights its own last-opened workspace and a fold never scrolls; projects file under their project group with the desktop's indent; the collapse spinner shows only after 500 ms; "Reopen to update" keeps the rows on screen. - e2e helpers and journeys for the switch-on suite. * fix(sidebar): keep an owed relay stop marked, clear a thrown stop, and prune pending unread raises - A stop-and-wait that throws takes its stop request back, like one that fails. - A final stop the legacy relay still owes stays marked, since the relay retries it. - Pending unread raises are tracked only while one waits, so the map no longer grows. - The paired-server fallback poll test gives its third fetch a reply and asserts live. * feat(sidebar): main publishes cached snapshots and announces its own stores; detected, view and live-content domains (inert) * fix(sidebar): keep links confirmed only by a new git run, no rescan on metadata writes, no phone wait on detected checkouts * fix(sidebar): rescan the workspace rows when the Windows runtime default changes * feat(sidebar): one lifecycle queue per workspace in main, with in-progress hints (switch-gated) Create, delete, terminal.closeAll, terminal.sleep and worktree.sleep now run through one first-in, first-out queue per host-qualified workspace: strict order, no merging, delete cancels what has not started and waits for what has, an unconfirmed outcome pauses until the host's own answer. Hints publish in a new inProgress feed domain (no reader yet); the creations domain is its projection. A finished create or delete keeps its hint until a scan started after it applies, in the same revision. Adds the paired-server metadata serializer (saving domain, no caller yet). Replaces the creation registry. * fix(sidebar): a create never queues behind the previous occupant's delete * fix(sidebar): resume a paused workspace by re-observing, never replaying - A command issued while paused waits behind the pause; when the owner answers again main re-reads terminal liveness and the listing, then carries on. A forced delete runs past it. - Commands keep their arrival order across a workspace lookup; deletes by name key by the resolved id, so they wait for that workspace's running sleep. - A second delete with different options is refused as already in progress. - Close and delete read lost contact as sleep does. - A create always frees its own queue entry, even after a retry took over its id. - Drops the unused opId, and the paired-server write serializer and its feed section, which land with their first caller. Co-Authored-By: Claude <noreply@anthropic.com> * fix: restore main's final P2 test fixes a merge reverted --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -87,10 +87,10 @@ vi.mock('./pty/provider/state-cleanup', async () =>
|
||||
)
|
||||
|
||||
function creationsOf(stub: WorktreeRuntimeStub): Record<string, unknown> {
|
||||
if (!stub.workspaceCreations) {
|
||||
if (!stub.workspaceLifecycle) {
|
||||
throw new Error('the switch is mocked on')
|
||||
}
|
||||
return stub.workspaceCreations.snapshot().creations
|
||||
return stub.workspaceLifecycle.creationsSnapshot().creations
|
||||
}
|
||||
|
||||
describe('worktree create registry', () => {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { _resetPendingWorktreeRemovalsForTests } from '../worktree-background-removal'
|
||||
import { vi } from 'vitest'
|
||||
import { WorkspaceCreationRegistry } from '../runtime/sidebar-inputs/workspace-creation-registry'
|
||||
import { WorkspaceLifecycleQueue } from '../runtime/workspaces/workspace-lifecycle-queue'
|
||||
import { ORCA_SIDEBAR_INPUTS_IS_MAIN } from '../../shared/sidebar-inputs-switch'
|
||||
|
||||
export type WorktreeRuntimeStub = {
|
||||
@@ -23,7 +23,7 @@ export type WorktreeRuntimeStub = {
|
||||
acquireFileWatcherRemoval: ReturnType<typeof vi.fn>
|
||||
hydrateInferredWorktreeLineage: ReturnType<typeof vi.fn>
|
||||
publishWorktreeRemovalChange: ReturnType<typeof vi.fn>
|
||||
workspaceCreations: WorkspaceCreationRegistry | null
|
||||
workspaceLifecycle: WorkspaceLifecycleQueue | null
|
||||
}
|
||||
|
||||
/** Why: create-flow tests need a minimal runtime; full fetchRemoteWithCache behavior lives in fetch-remote-cache.test.ts. */
|
||||
@@ -34,7 +34,7 @@ export function createWorktreeRuntimeStub(mainWindow?: {
|
||||
_resetPendingWorktreeRemovalsForTests()
|
||||
const runtimeStub: WorktreeRuntimeStub = {
|
||||
// Mirrors the runtime field: listed only when the sidebar-inputs switch is on.
|
||||
workspaceCreations: ORCA_SIDEBAR_INPUTS_IS_MAIN ? new WorkspaceCreationRegistry() : null,
|
||||
workspaceLifecycle: ORCA_SIDEBAR_INPUTS_IS_MAIN ? new WorkspaceLifecycleQueue() : null,
|
||||
resolveRemoteTrackingBase: vi.fn().mockResolvedValue(null),
|
||||
hasRemoteTrackingRef: vi.fn().mockResolvedValue(false),
|
||||
getOrStartRemoteTrackingBaseRefresh: vi.fn().mockResolvedValue({ ok: true }),
|
||||
|
||||
@@ -28,6 +28,8 @@ import type { CreateWorktreeArgsWithSystemProvenance } from '../ipc-context-sche
|
||||
import { createFolderWorkspace } from './folder-workspace-creation'
|
||||
import { findExactRepoOwner, isCapturedRepoCurrent } from '../listing/worktree-host-ownership'
|
||||
import { requireWorktreeCreateRoute } from '../../../worktree-create-execution-host-route'
|
||||
import { getRepoExecutionHostId } from '../../../../shared/execution-host'
|
||||
import { workspaceLifecycleKey } from '../../../runtime/workspaces/workspace-lifecycle-commands'
|
||||
import type { WorktreeIpcContext } from '../worktree-ipc-context'
|
||||
import { beginWorkspaceCreateTelemetry } from '../../../workspace-create-telemetry'
|
||||
import { assertWslMoveRepoWritable } from '../../../persistence/moving-wsl-projects/wsl-move-scope-guard'
|
||||
@@ -67,9 +69,10 @@ export function registerWorktreeCreateHandlers(context: WorktreeIpcContext): voi
|
||||
fromExistingBranch: typeof args.baseBranch === 'string' && args.baseBranch.length > 0,
|
||||
isFolder: isFolderRepo(repo)
|
||||
})
|
||||
const creation = runtime.workspaceCreations?.open({
|
||||
const creation = runtime.workspaceLifecycle?.openCreate({
|
||||
creationId: args.creationId,
|
||||
repoId: repo.id
|
||||
repoId: repo.id,
|
||||
keyOf: (worktreeId) => workspaceLifecycleKey(getRepoExecutionHostId(repo), worktreeId)
|
||||
})
|
||||
try {
|
||||
// Why: wrap only the helpers; the pre-validation throws above are IPC-shape bugs, not the git/filesystem failures the funnel tracks.
|
||||
|
||||
@@ -23,6 +23,12 @@ import {
|
||||
} from '../../../worktree-background-removal'
|
||||
import { runSerializedWorktreeRemovalAcceptance } from '../../../worktree-removal-acceptance-queue'
|
||||
import { assertWslMoveRepoWritable } from '../../../persistence/moving-wsl-projects/wsl-move-scope-guard'
|
||||
import {
|
||||
deleteLostContact,
|
||||
observeTerminals,
|
||||
runWorkspaceLifecycleCommand,
|
||||
workspaceLifecycleKey
|
||||
} from '../../../runtime/workspaces/workspace-lifecycle-commands'
|
||||
|
||||
export function registerWorktreeRemovalHandlers(context: WorktreeIpcContext): void {
|
||||
const { store, options, worktreeRemovalsInFlight } = context
|
||||
@@ -129,5 +135,29 @@ export function registerWorktreeRemovalHandlers(context: WorktreeIpcContext): vo
|
||||
}
|
||||
}
|
||||
}
|
||||
ipcMain.handle('worktrees:remove', (_event, args: RemoveWorktreeArgs) => remove(args))
|
||||
ipcMain.handle('worktrees:remove', (_event, args: RemoveWorktreeArgs) => {
|
||||
const repo = resolveRepoForExecutionHost(
|
||||
store,
|
||||
parseWorktreeId(args.worktreeId).repoId,
|
||||
args.hostId
|
||||
)
|
||||
// Without its repo the removal refuses as before; there is no workspace to queue under.
|
||||
return repo
|
||||
? runWorkspaceLifecycleCommand(
|
||||
context.runtime.workspaceLifecycle,
|
||||
workspaceLifecycleKey(getRepoExecutionHostId(repo), args.worktreeId),
|
||||
{
|
||||
kind: 'delete',
|
||||
run: () => remove(args),
|
||||
lostContact: deleteLostContact,
|
||||
observe: observeTerminals(context.runtime, `id:${args.worktreeId}`),
|
||||
delete: {
|
||||
key: getWorktreeRemovalOptionsKey(args),
|
||||
force: args.allowUnverifiedPtyStop === true,
|
||||
label: args.worktreeId
|
||||
}
|
||||
}
|
||||
)
|
||||
: remove(args)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -95,10 +95,10 @@ function makeRuntime(repo: Record<string, unknown> = gitRepo): OrcaRuntimeServic
|
||||
type LocalCreateArgs = { onWorktreeMetadataPersisted: (created: { id: string }) => unknown }
|
||||
|
||||
function creationsOf(runtime: OrcaRuntimeService): Record<string, unknown> {
|
||||
if (!runtime.workspaceCreations) {
|
||||
if (!runtime.workspaceLifecycle) {
|
||||
throw new Error('the switch is mocked on')
|
||||
}
|
||||
return runtime.workspaceCreations.snapshot().creations
|
||||
return runtime.workspaceLifecycle.creationsSnapshot().creations
|
||||
}
|
||||
|
||||
describe('runtime creation registry', () => {
|
||||
|
||||
@@ -99,6 +99,7 @@ class OrcaRuntimeService extends OrcaRuntimeSidebarBase {
|
||||
scan: () => this.scanResolvedWorktreeRepos(),
|
||||
join: (scans) => this.joinResolvedWorktreeScans(scans),
|
||||
visible: (snapshot) => this.visibleResolvedWorktreeSnapshot(snapshot),
|
||||
lifecycle: this.workspaceLifecycle,
|
||||
onClientEvent: (listener, options) => this.onClientEvent(listener, options)
|
||||
})
|
||||
: null
|
||||
@@ -109,7 +110,7 @@ class OrcaRuntimeService extends OrcaRuntimeSidebarBase {
|
||||
sidebarInputsFeedDomains({
|
||||
facts: requireSwitchedOn(this.sidebarFacts),
|
||||
pairedServerRows: requireSwitchedOn(this.pairedServerRows),
|
||||
workspaceCreations: requireSwitchedOn(this.workspaceCreations),
|
||||
workspaceLifecycle: requireSwitchedOn(this.workspaceLifecycle),
|
||||
workspaceRows: requireSwitchedOn(this.workspaceRows),
|
||||
repoIcons: requireSwitchedOn(this.repoIconDataUrls),
|
||||
store: () => this.store,
|
||||
|
||||
@@ -109,7 +109,7 @@ export const SIDEBAR_INPUT_METHODS = [
|
||||
permission: 'workspace',
|
||||
params: null,
|
||||
handler: async (_params, { runtime }) =>
|
||||
requireSwitchedOn(runtime.workspaceCreations).snapshot()
|
||||
requireSwitchedOn(runtime.workspaceLifecycle).creationsSnapshot()
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'sidebarInputs.reportView',
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { terminalImageViewerSupported } from '../../../terminal-image-viewer-env'
|
||||
import { TERMINAL_WORKSPACE_STOP_METHODS } from './terminal-workspace-stop-methods'
|
||||
import { ORCA_SIDEBAR_INPUTS_IS_MAIN } from '../../../../../shared/sidebar-inputs-switch'
|
||||
import { defineMethod } from '../../core'
|
||||
import {
|
||||
@@ -9,15 +10,11 @@ import { withTerminalCloseAttribution } from '../../terminal-close-attribution'
|
||||
import {
|
||||
AgentTeamsPrepareLaunch,
|
||||
AgentTeamsTmuxCompat,
|
||||
TerminalCloseAll,
|
||||
TerminalCreateParams,
|
||||
TerminalFocus,
|
||||
TerminalHandle,
|
||||
TerminalReveal,
|
||||
TerminalSleep,
|
||||
TerminalSplit,
|
||||
TerminalStop,
|
||||
TerminalStopExact,
|
||||
TerminalWait
|
||||
} from '../../../../../shared/rpc-contract/terminal-unary-params'
|
||||
import { TerminalResizeForClient } from '../../../../../shared/rpc-contract/terminal-stream-params'
|
||||
@@ -161,34 +158,7 @@ export const TERMINAL_LIFECYCLE_METHODS = [
|
||||
})
|
||||
})
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'terminal.stop',
|
||||
permission: 'workspace',
|
||||
params: TerminalStop,
|
||||
handler: async (params, { runtime }) => runtime.stopTerminalsForWorktree(params.worktree)
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'terminal.closeAll',
|
||||
permission: 'workspace',
|
||||
params: TerminalCloseAll,
|
||||
handler: async (params, { runtime }) => runtime.closeTerminalsForWorktree(params.worktree)
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'terminal.sleep',
|
||||
permission: 'workspace',
|
||||
params: TerminalSleep,
|
||||
handler: async (params, { runtime }) => runtime.sleepTerminalsForWorktree(params.worktree)
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'terminal.stopExact',
|
||||
permission: 'workspace',
|
||||
params: TerminalStopExact,
|
||||
handler: async (params, { runtime }) =>
|
||||
runtime.stopExactTerminalsForWorktree(params.worktree, params.expectedPtyIds, {
|
||||
keepHistory: params.keepHistory,
|
||||
targetOnly: params.targetOnly
|
||||
})
|
||||
}),
|
||||
...TERMINAL_WORKSPACE_STOP_METHODS,
|
||||
defineMethod({
|
||||
name: 'terminal.resizeForClient',
|
||||
permission: 'workspace',
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
import {
|
||||
closeOutcome,
|
||||
observeTerminals,
|
||||
queueTerminalWorkspaceCommand,
|
||||
sleepOutcome,
|
||||
terminalLostContact
|
||||
} from '../../../workspaces/workspace-lifecycle-commands'
|
||||
import { defineMethod } from '../../core'
|
||||
import {
|
||||
TerminalCloseAll,
|
||||
TerminalSleep,
|
||||
TerminalStop,
|
||||
TerminalStopExact
|
||||
} from '../../../../../shared/rpc-contract/terminal-unary-params'
|
||||
|
||||
/** Stopping every terminal of a workspace; closing and sleeping them wait in its lifecycle queue. */
|
||||
export const TERMINAL_WORKSPACE_STOP_METHODS = [
|
||||
defineMethod({
|
||||
name: 'terminal.stop',
|
||||
permission: 'workspace',
|
||||
params: TerminalStop,
|
||||
handler: async (params, { runtime }) => runtime.stopTerminalsForWorktree(params.worktree)
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'terminal.closeAll',
|
||||
permission: 'workspace',
|
||||
params: TerminalCloseAll,
|
||||
handler: async (params, { runtime }) =>
|
||||
queueTerminalWorkspaceCommand(runtime, params.worktree, {
|
||||
kind: 'close',
|
||||
run: () => runtime.closeTerminalsForWorktree(params.worktree),
|
||||
outcome: closeOutcome,
|
||||
lostContact: terminalLostContact,
|
||||
observe: observeTerminals(runtime, params.worktree)
|
||||
})
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'terminal.sleep',
|
||||
permission: 'workspace',
|
||||
params: TerminalSleep,
|
||||
handler: async (params, { runtime }) =>
|
||||
queueTerminalWorkspaceCommand(runtime, params.worktree, {
|
||||
kind: 'sleep',
|
||||
run: () => runtime.sleepTerminalsForWorktree(params.worktree),
|
||||
outcome: sleepOutcome,
|
||||
lostContact: terminalLostContact,
|
||||
observe: observeTerminals(runtime, params.worktree)
|
||||
})
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'terminal.stopExact',
|
||||
permission: 'workspace',
|
||||
params: TerminalStopExact,
|
||||
handler: async (params, { runtime }) =>
|
||||
runtime.stopExactTerminalsForWorktree(params.worktree, params.expectedPtyIds, {
|
||||
keepHistory: params.keepHistory,
|
||||
targetOnly: params.targetOnly
|
||||
})
|
||||
})
|
||||
] as const
|
||||
@@ -0,0 +1,10 @@
|
||||
import { WorktreeSelector } from '../../../../shared/rpc-contract/worktree-params'
|
||||
import { queueWorktreeSleep } from '../../workspaces/workspace-lifecycle-commands'
|
||||
import { defineMethod } from '../core'
|
||||
|
||||
export const WORKTREE_SLEEP_METHOD = defineMethod({
|
||||
name: 'worktree.sleep',
|
||||
permission: 'workspace',
|
||||
params: WorktreeSelector,
|
||||
handler: async (params, { runtime }) => queueWorktreeSleep(runtime, params.worktree)
|
||||
})
|
||||
@@ -33,6 +33,8 @@ import {
|
||||
import { WORKTREE_CATALOG_METHODS } from './worktree-catalog-methods'
|
||||
import { WORKTREE_PROVISIONED_ROOT_METHOD } from './worktree-provisioned-root-method'
|
||||
import { readsWorktreeRemovalMarker } from '../worktree-removal-marker-projection'
|
||||
import { queueWorkspaceDelete } from '../../workspaces/workspace-lifecycle-commands'
|
||||
import { WORKTREE_SLEEP_METHOD } from './worktree-sleep-method'
|
||||
|
||||
export const WORKTREE_METHODS = [
|
||||
...WORKTREE_CATALOG_METHODS,
|
||||
@@ -64,12 +66,7 @@ export const WORKTREE_METHODS = [
|
||||
worktree: await runtime.showManagedWorktree(params.worktree)
|
||||
})
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'worktree.sleep',
|
||||
permission: 'workspace',
|
||||
params: WorktreeSelector,
|
||||
handler: async (params, { runtime }) => runtime.sleepManagedWorktree(params.worktree)
|
||||
}),
|
||||
WORKTREE_SLEEP_METHOD,
|
||||
defineMethod({
|
||||
name: 'worktree.activate',
|
||||
permission: 'workspace',
|
||||
@@ -280,16 +277,19 @@ export const WORKTREE_METHODS = [
|
||||
// resolution costs a scan and throws for an id two hosts share. Other selectors stay unstamped.
|
||||
const explicitWorktreeId = getExplicitWorktreeIdSelector(params.worktree)
|
||||
const repoId = explicitWorktreeId ? splitWorktreeId(explicitWorktreeId)?.repoId : undefined
|
||||
const result = await runtime.removeManagedWorktree(params.worktree, {
|
||||
force: params.force === true,
|
||||
runHooks: params.runHooks === true,
|
||||
allowUnverifiedPtyStop: params.allowUnverifiedPtyStop === true,
|
||||
allowFailedArchiveHook: params.allowFailedArchiveHook === true,
|
||||
...(resolvedHostId ? { hostId: resolvedHostId } : {}),
|
||||
// Why: only a client that shows the `removing` marker can wait out Git's delete; older
|
||||
// clients get the acceptance and never see the row again.
|
||||
...(readsWorktreeRemovalMarker(context) ? { waitForBackgroundRemoval: true } : {})
|
||||
})
|
||||
const result = await queueWorkspaceDelete(
|
||||
runtime,
|
||||
params.worktree,
|
||||
resolvedHostId,
|
||||
params,
|
||||
(removal) =>
|
||||
runtime.removeManagedWorktree(params.worktree, {
|
||||
...removal,
|
||||
// Why: only a client that shows the `removing` marker can wait out Git's delete; older
|
||||
// clients get the acceptance and never see the row again.
|
||||
...(readsWorktreeRemovalMarker(context) ? { waitForBackgroundRemoval: true } : {})
|
||||
})
|
||||
)
|
||||
return {
|
||||
removed: true,
|
||||
...result,
|
||||
|
||||
@@ -85,4 +85,36 @@ describe('CachedScan', () => {
|
||||
expect(cache.current).toBe(1)
|
||||
vi.restoreAllMocks()
|
||||
})
|
||||
|
||||
it('calls a fresh waiter right after a scan started after it applies, not an older one', async () => {
|
||||
const { cache, pending, applied, finish } = scanner()
|
||||
cache.request()
|
||||
const seen: number[][] = []
|
||||
// A scan already in flight started before the wait, so it cannot satisfy it.
|
||||
cache.afterFresh(() => seen.push([...applied]))
|
||||
await finish()
|
||||
expect(applied).toEqual([1])
|
||||
expect(seen).toEqual([])
|
||||
expect(pending).toHaveLength(1)
|
||||
await finish()
|
||||
// Called synchronously after the apply, so its change publishes in the same revision.
|
||||
expect(seen).toEqual([[1, 2]])
|
||||
})
|
||||
|
||||
it('calls a fresh waiter at once with no demand, and after a failed scan', async () => {
|
||||
const idle = scanner(false)
|
||||
const calls: string[] = []
|
||||
idle.cache.afterFresh(() => calls.push('idle'))
|
||||
expect(calls).toEqual(['idle'])
|
||||
vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const failing = new CachedScan<number>({
|
||||
label: 'test',
|
||||
demand: () => true,
|
||||
applied: () => {},
|
||||
scan: () => Promise.reject(new Error('git failed'))
|
||||
})
|
||||
failing.afterFresh(() => calls.push('failed'))
|
||||
await vi.waitFor(() => expect(calls).toEqual(['idle', 'failed']))
|
||||
vi.restoreAllMocks()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -11,6 +11,7 @@ export class CachedScan<T> {
|
||||
private scanning: Promise<void> | null = null
|
||||
private rescanQueued = false
|
||||
private stale = false
|
||||
private readonly freshWaiters: { after: number; done: () => void }[] = []
|
||||
|
||||
constructor(
|
||||
private readonly deps: {
|
||||
@@ -41,8 +42,17 @@ export class CachedScan<T> {
|
||||
this.stale = false
|
||||
this.scanning = this.deps
|
||||
.scan()
|
||||
.then((value) => this.apply(version, value))
|
||||
.catch((error: unknown) => console.warn(`[${this.deps.label}] scan failed`, error))
|
||||
.then((value) => {
|
||||
try {
|
||||
this.apply(version, value)
|
||||
} finally {
|
||||
this.settleFresh(version)
|
||||
}
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
console.warn(`[${this.deps.label}] scan failed`, error)
|
||||
this.settleFresh(version)
|
||||
})
|
||||
.finally(() => {
|
||||
this.scanning = null
|
||||
if (this.rescanQueued) {
|
||||
@@ -62,6 +72,29 @@ export class CachedScan<T> {
|
||||
}
|
||||
}
|
||||
|
||||
/** Calls `done` once a scan started after this call has finished; at once with no demand. */
|
||||
afterFresh(done: () => void): void {
|
||||
if (!this.deps.demand()) {
|
||||
done()
|
||||
return
|
||||
}
|
||||
this.freshWaiters.push({ after: this.version, done })
|
||||
this.request(true)
|
||||
}
|
||||
|
||||
// Why synchronous after apply: a waiter's change lands in the same revision as the result.
|
||||
private settleFresh(version: number): void {
|
||||
const due = this.freshWaiters.filter((waiter) => waiter.after < version)
|
||||
if (due.length === 0) {
|
||||
return
|
||||
}
|
||||
const waiting = this.freshWaiters.filter((waiter) => waiter.after >= version)
|
||||
this.freshWaiters.splice(0, this.freshWaiters.length, ...waiting)
|
||||
for (const waiter of due) {
|
||||
waiter.done()
|
||||
}
|
||||
}
|
||||
|
||||
/** Settles once every scan requested so far has applied; for one-shot readers. */
|
||||
async settled(): Promise<void> {
|
||||
await this.ready()
|
||||
|
||||
@@ -28,7 +28,7 @@ describe('the runtime sidebar inputs feed', () => {
|
||||
await phone.settle()
|
||||
expect(phone.view().temporal).toEqual({ sessionId: null, retention: null })
|
||||
|
||||
const creation = requireSwitchedOn(runtime.workspaceCreations).open({
|
||||
const creation = requireSwitchedOn(runtime.workspaceLifecycle).openCreate({
|
||||
creationId: 'c1',
|
||||
repoId: 'repo-1'
|
||||
})
|
||||
|
||||
@@ -85,10 +85,11 @@ async function host() {
|
||||
catalogs: () => []
|
||||
} as unknown as SidebarInputsFeedSources['pairedServerRows'],
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: as above.
|
||||
workspaceCreations: {
|
||||
workspaceLifecycle: {
|
||||
revision: revisions.creations,
|
||||
snapshot: () => ({ creations: {} })
|
||||
} as unknown as SidebarInputsFeedSources['workspaceCreations'],
|
||||
creationsSnapshot: () => ({ creations: {} }),
|
||||
hints: () => ({})
|
||||
} as unknown as SidebarInputsFeedSources['workspaceLifecycle'],
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: as above; one cached worktree, already scanned.
|
||||
workspaceRows: {
|
||||
revision: revisions.rows,
|
||||
@@ -118,7 +119,8 @@ async function host() {
|
||||
store.patchWorkspaceSession({ activeTabId: 'tab-1' }, toRuntimeExecutionHostId('env-1'))
|
||||
},
|
||||
temporal: { 'retention owner': () => temporal.forEach((listener) => listener()) },
|
||||
creations: { 'creation registry': () => revisions.creations.bump() },
|
||||
creations: { 'lifecycle queue': () => revisions.creations.bump() },
|
||||
inProgress: { 'lifecycle queue': () => revisions.creations.bump() },
|
||||
workspaces: { 'workspace rows producer': () => revisions.rows.bump() },
|
||||
detected: { 'detected checkouts producer': () => revisions.detected.bump() },
|
||||
liveContent: { 'live content producer': () => revisions.liveContent.bump() },
|
||||
@@ -215,7 +217,8 @@ describe('sidebar inputs feed announcements', () => {
|
||||
detach?.()
|
||||
return heard.some((change) => change?.urgent === true)
|
||||
}
|
||||
expect(urgency('creations', 'creation registry')).toBe(true)
|
||||
expect(urgency('creations', 'lifecycle queue')).toBe(true)
|
||||
expect(urgency('inProgress', 'lifecycle queue')).toBe(true)
|
||||
expect(urgency('desktop', 'pin write')).toBe(true)
|
||||
expect(urgency('view', 'ui write')).toBe(true)
|
||||
})
|
||||
|
||||
@@ -10,7 +10,7 @@ import { WORKTREE_METADATA_DOMAINS } from '../../persistence/loading-store/workt
|
||||
import type { PersistedUIState } from '../../../shared/persisted-ui-state-types'
|
||||
import type { OrcaRuntimeWithSidebarTemporal } from './orca-runtime-sidebar-temporal'
|
||||
import type { PairedServerRowsProducer } from './paired-server-rows-producer'
|
||||
import type { WorkspaceCreationRegistry } from './workspace-creation-registry'
|
||||
import type { WorkspaceLifecycleQueue } from '../workspaces/workspace-lifecycle-queue'
|
||||
import type { WorkspaceRowsProducer } from './workspace-rows-producer'
|
||||
import type { RepoIconDataUrls } from './repo-icon-data-urls'
|
||||
import type { SidebarInputsFeedDomain } from './sidebar-inputs-feed'
|
||||
@@ -68,7 +68,7 @@ const DESKTOP_WRITE_DOMAINS = [...WORKTREE_METADATA_DOMAINS, 'settings', 'ui']
|
||||
export type SidebarInputsFeedSources = {
|
||||
facts: OrcaRuntimeWithSidebarTemporal
|
||||
pairedServerRows: PairedServerRowsProducer
|
||||
workspaceCreations: WorkspaceCreationRegistry
|
||||
workspaceLifecycle: WorkspaceLifecycleQueue
|
||||
workspaceRows: WorkspaceRowsProducer
|
||||
repoIcons: Pick<RepoIconDataUrls, 'idsByRepo'>
|
||||
store: () => RuntimeStore | null
|
||||
@@ -82,7 +82,7 @@ export type SidebarInputsFeedSources = {
|
||||
export function sidebarInputsFeedDomains(
|
||||
sources: SidebarInputsFeedSources
|
||||
): SidebarInputsFeedDomain[] {
|
||||
const { facts, pairedServerRows, workspaceCreations, workspaceRows } = sources
|
||||
const { facts, pairedServerRows, workspaceLifecycle, workspaceRows } = sources
|
||||
const urgent = { urgent: true }
|
||||
const watchCatalog = (changed: () => void): (() => void) =>
|
||||
sources.onClientEvent(
|
||||
@@ -142,10 +142,17 @@ export function sidebarInputsFeedDomains(
|
||||
watch: (changed) => facts.onSidebarTemporalChange(changed)
|
||||
},
|
||||
{
|
||||
// The queue's creates, as the creations readers read them.
|
||||
name: 'creations',
|
||||
read: () => workspaceCreations.snapshot().creations,
|
||||
read: () => workspaceLifecycle.creationsSnapshot().creations,
|
||||
// An accepted command shows at once.
|
||||
watch: (changed) => workspaceCreations.revision.onChange(() => changed(urgent))
|
||||
watch: (changed) => workspaceLifecycle.revision.onChange(() => changed(urgent))
|
||||
},
|
||||
{
|
||||
// Each workspace's running, waiting or paused lifecycle command, keyed host-qualified.
|
||||
name: 'inProgress',
|
||||
read: () => workspaceLifecycle.hints(),
|
||||
watch: (changed) => workspaceLifecycle.revision.onChange(() => changed(urgent))
|
||||
},
|
||||
{
|
||||
name: 'workspaces',
|
||||
|
||||
@@ -10,7 +10,7 @@ describe('sidebar inputs with the switch off', () => {
|
||||
expect(runtime.sidebarFacts).toBeNull()
|
||||
expect(runtime.pairedServerRows).toBeNull()
|
||||
expect(runtime.repoIconDataUrls).toBeNull()
|
||||
expect(runtime.workspaceCreations).toBeNull()
|
||||
expect(runtime.workspaceLifecycle).toBeNull()
|
||||
expect(runtime.collapseWriteOrder).toBeNull()
|
||||
const gated = new Set<string>(SIDEBAR_INPUT_METHODS.map((method) => method.name))
|
||||
expect(ALL_RPC_METHODS.filter((method) => gated.has(method.name))).toEqual([])
|
||||
|
||||
@@ -1,52 +0,0 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { WorkspaceCreationRegistry } from './workspace-creation-registry'
|
||||
|
||||
describe('WorkspaceCreationRegistry', () => {
|
||||
it('lists a creation once opened, names its workspace, and drops it on close', () => {
|
||||
const registry = new WorkspaceCreationRegistry()
|
||||
const creation = registry.open({ creationId: 'c1', repoId: 'repo-1' })
|
||||
expect(registry.snapshot().creations).toEqual({ c1: { repoId: 'repo-1' } })
|
||||
const opened = registry.snapshot().revision
|
||||
|
||||
creation.named('repo-1::/wt')
|
||||
expect(registry.snapshot().creations).toEqual({
|
||||
c1: { repoId: 'repo-1', worktreeId: 'repo-1::/wt' }
|
||||
})
|
||||
expect(registry.snapshot().revision).toBeGreaterThan(opened)
|
||||
|
||||
creation.close()
|
||||
expect(registry.snapshot().creations).toEqual({})
|
||||
})
|
||||
|
||||
it('mints an id for a caller that sends none', () => {
|
||||
const registry = new WorkspaceCreationRegistry()
|
||||
const creation = registry.open({ repoId: 'repo-1' })
|
||||
const ids = Object.keys(registry.snapshot().creations)
|
||||
expect(ids).toHaveLength(1)
|
||||
expect(ids[0]).not.toBe('')
|
||||
creation.close()
|
||||
expect(registry.snapshot().creations).toEqual({})
|
||||
})
|
||||
|
||||
it('leaves a concurrent retry under the same id listed when the first create closes', () => {
|
||||
const registry = new WorkspaceCreationRegistry()
|
||||
const first = registry.open({ creationId: 'c1', repoId: 'repo-1' })
|
||||
const second = registry.open({ creationId: 'c1', repoId: 'repo-1' })
|
||||
|
||||
first.named('repo-1::/stale')
|
||||
first.close()
|
||||
expect(registry.snapshot().creations).toEqual({ c1: { repoId: 'repo-1' } })
|
||||
|
||||
second.close()
|
||||
expect(registry.snapshot().creations).toEqual({})
|
||||
})
|
||||
|
||||
it('closes once', () => {
|
||||
const registry = new WorkspaceCreationRegistry()
|
||||
const creation = registry.open({ creationId: 'c1', repoId: 'repo-1' })
|
||||
creation.close()
|
||||
const closed = registry.snapshot().revision
|
||||
creation.close()
|
||||
expect(registry.snapshot().revision).toBe(closed)
|
||||
})
|
||||
})
|
||||
@@ -1,51 +0,0 @@
|
||||
import { ProducerRevision } from './producer-revision'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type {
|
||||
WorkspaceCreation,
|
||||
WorkspaceCreationsSnapshot
|
||||
} from '../../../shared/sidebar-list/workspace-creations'
|
||||
|
||||
/** One listed creation, owned by the create that opened it. */
|
||||
export type WorkspaceCreationHandle = {
|
||||
/** The workspace is persisted; called before the host announces it. */
|
||||
named: (worktreeId: string) => void
|
||||
close: () => void
|
||||
}
|
||||
|
||||
/** Every creation this host is running, whoever asked: its own window, a phone, the CLI. */
|
||||
export class WorkspaceCreationRegistry {
|
||||
private readonly creations = new Map<string, WorkspaceCreation>()
|
||||
readonly revision = new ProducerRevision()
|
||||
|
||||
/** Opened once the project is known. A caller without an id gets one, so a CLI create is listed. */
|
||||
open(start: { creationId?: string; repoId: string }): WorkspaceCreationHandle {
|
||||
const creationId = start.creationId ?? randomUUID()
|
||||
const entry: WorkspaceCreation = { repoId: start.repoId }
|
||||
this.creations.set(creationId, entry)
|
||||
this.revision.bump()
|
||||
// Identity, not id: a concurrent retry under the same id owns the entry from then on.
|
||||
const owns = (): boolean => this.creations.get(creationId) === entry
|
||||
return {
|
||||
named: (worktreeId) => {
|
||||
if (owns() && entry.worktreeId !== worktreeId) {
|
||||
entry.worktreeId = worktreeId
|
||||
this.revision.bump()
|
||||
}
|
||||
},
|
||||
close: () => {
|
||||
if (owns()) {
|
||||
this.creations.delete(creationId)
|
||||
this.revision.bump()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
snapshot(): WorkspaceCreationsSnapshot {
|
||||
const creations: WorkspaceCreationsSnapshot['creations'] = {}
|
||||
for (const [creationId, entry] of this.creations) {
|
||||
creations[creationId] = { ...entry }
|
||||
}
|
||||
return { revision: this.revision.value, creations }
|
||||
}
|
||||
}
|
||||
@@ -17,6 +17,7 @@ import { WORKTREE_METADATA_DOMAINS } from '../../persistence/loading-store/workt
|
||||
import { ProducerRevision } from './producer-revision'
|
||||
import { WorkspaceGitIdentity } from './workspace-git-identity'
|
||||
import { CachedScan } from './cached-scan'
|
||||
import type { WorkspaceLifecycleQueue } from '../workspaces/workspace-lifecycle-queue'
|
||||
|
||||
// The settings the visibility filter reads: the defaults, and the layouts that classify a checkout.
|
||||
const VISIBILITY_SETTINGS = [
|
||||
@@ -51,6 +52,8 @@ export type WorkspaceRowsSources = {
|
||||
join: (scans: ResolvedWorktreeRepoScans) => ResolvedWorktreeScanJoin
|
||||
/** worktree.ps's visibility filters over a listing. */
|
||||
visible: (snapshot: ResolvedWorktreeSnapshot) => ResolvedWorktreeSnapshot
|
||||
/** Its finished creates and deletes keep their hints until a fresh scan lists them. */
|
||||
lifecycle?: Pick<WorkspaceLifecycleQueue, 'setMembership'> | null
|
||||
onClientEvent: (
|
||||
listener: (event: RuntimeClientEvent) => void,
|
||||
options: { consumesTerminalSideEffects: false }
|
||||
@@ -89,6 +92,7 @@ export class WorkspaceRowsProducer {
|
||||
demand: () => this.revision.hasListeners(),
|
||||
applied: () => this.applied()
|
||||
})
|
||||
sources.lifecycle?.setMembership((done) => this.scans.afterFresh(done))
|
||||
sources.onClientEvent(
|
||||
(event) => {
|
||||
if (event.type === 'worktreesChanged' || event.type === 'reposChanged') {
|
||||
|
||||
@@ -23,10 +23,10 @@ import type { RuntimeWorkspaceCreateEvents } from '../../workspace-create-teleme
|
||||
import { createdWorkspaceActivationCallbacks } from './workspace-create-activation'
|
||||
import { assertWslMoveRepoWritable } from '../../persistence/moving-wsl-projects/wsl-move-scope-guard'
|
||||
import { resolveManagedWorktreeCreateStartup } from './managed-worktree-create-startup'
|
||||
import {
|
||||
WorkspaceCreationRegistry,
|
||||
type WorkspaceCreationHandle
|
||||
} from '../sidebar-inputs/workspace-creation-registry'
|
||||
import { WorkspaceLifecycleQueue } from './workspace-lifecycle-queue'
|
||||
import type { WorkspaceCreationHandle } from './workspace-lifecycle-creations'
|
||||
import { workspaceLifecycleKey } from './workspace-lifecycle-commands'
|
||||
import { getRepoExecutionHostId } from '../../../shared/execution-host'
|
||||
import { ORCA_SIDEBAR_INPUTS_IS_MAIN } from '../../../shared/sidebar-inputs-switch'
|
||||
|
||||
export abstract class OrcaRuntimeWithCreateManagedWorktree extends OrcaRuntimeWithGetWorktreeTerminalProvisioningHost {
|
||||
@@ -35,8 +35,8 @@ export abstract class OrcaRuntimeWithCreateManagedWorktree extends OrcaRuntimeWi
|
||||
childRepo?: Repo
|
||||
): Promise<WorktreeLineageResolution>
|
||||
|
||||
// Switch point: only a build where main produces the sidebar's inputs lists creations.
|
||||
readonly workspaceCreations = ORCA_SIDEBAR_INPUTS_IS_MAIN ? new WorkspaceCreationRegistry() : null
|
||||
// Switch point: only a build where main produces the sidebar's inputs queues lifecycle commands.
|
||||
readonly workspaceLifecycle = ORCA_SIDEBAR_INPUTS_IS_MAIN ? new WorkspaceLifecycleQueue() : null
|
||||
|
||||
async createManagedWorktree(
|
||||
args: RuntimeManagedWorktreeCreateArgs
|
||||
@@ -68,7 +68,11 @@ export abstract class OrcaRuntimeWithCreateManagedWorktree extends OrcaRuntimeWi
|
||||
|
||||
const repo = await this.resolveRepoSelector(args.repoSelector)
|
||||
// Listed from here: a create refused before its project resolved has nothing to show.
|
||||
const creation = this.workspaceCreations?.open({ creationId: args.creationId, repoId: repo.id })
|
||||
const creation = this.workspaceLifecycle?.openCreate({
|
||||
creationId: args.creationId,
|
||||
repoId: repo.id,
|
||||
keyOf: (worktreeId) => workspaceLifecycleKey(getRepoExecutionHostId(repo), worktreeId)
|
||||
})
|
||||
try {
|
||||
return await this.createInResolvedProject(args, repo, creation, rearm, events)
|
||||
} finally {
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
import type { WorkspaceLifecycleKind } from '../../../shared/sidebar-list/workspace-lifecycle-hints'
|
||||
|
||||
export const WORKSPACE_DELETED_ERROR = 'workspace_deleted'
|
||||
|
||||
/** `unverifiable`: the execution host never confirmed the outcome, so the queue pauses. */
|
||||
export type LifecycleOutcome = 'settled' | 'unverifiable'
|
||||
|
||||
/** Delete options: an equal delete joins the pending one, a different one is refused. */
|
||||
export type LifecycleDeleteOptions = {
|
||||
key: string
|
||||
/** The explicit waiver (Force Delete, `--force`): runs past a pause, as it always has. */
|
||||
force: boolean
|
||||
/** Names the workspace in the "already in progress" refusal. */
|
||||
label: string
|
||||
}
|
||||
|
||||
export type LifecycleCommand<T> = {
|
||||
kind: Exclude<WorkspaceLifecycleKind, 'create'>
|
||||
run: () => Promise<T>
|
||||
/** Reads a reply; defaults to settled. */
|
||||
outcome?: (result: T) => LifecycleOutcome
|
||||
/** Whether a thrown error means contact was lost mid-run rather than a definite failure. */
|
||||
lostContact?: (error: unknown) => boolean
|
||||
/** After an unconfirmed outcome: re-reads the owner's state, throwing while it cannot answer. */
|
||||
observe?: () => Promise<void>
|
||||
delete?: LifecycleDeleteOptions
|
||||
}
|
||||
|
||||
/** Calls `done` once a listing that started after this call has applied. */
|
||||
export type LifecycleMembership = (done: () => void) => void
|
||||
|
||||
export type LifecycleAttempt = {
|
||||
outcome: LifecycleOutcome
|
||||
value?: unknown
|
||||
error?: unknown
|
||||
threw: boolean
|
||||
}
|
||||
|
||||
export async function attemptLifecycleCommand(
|
||||
command: LifecycleCommand<unknown>
|
||||
): Promise<LifecycleAttempt> {
|
||||
try {
|
||||
const value = await command.run()
|
||||
return { outcome: command.outcome?.(value) ?? 'settled', value, threw: false }
|
||||
} catch (error) {
|
||||
return {
|
||||
outcome: command.lostContact?.(error) ? 'unverifiable' : 'settled',
|
||||
error,
|
||||
threw: true
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { WorkspaceLifecycleQueue } from './workspace-lifecycle-queue'
|
||||
import {
|
||||
closeOutcome,
|
||||
deleteLostContact,
|
||||
observeTerminals,
|
||||
queueTerminalWorkspaceCommand,
|
||||
queueWorkspaceDelete,
|
||||
sleepOutcome,
|
||||
terminalLostContact,
|
||||
workspaceLifecycleKey
|
||||
} from './workspace-lifecycle-commands'
|
||||
|
||||
const flush = () => new Promise((resolve) => setTimeout(resolve, 0))
|
||||
|
||||
const repos = [
|
||||
{ id: 'repo-local', connectionId: null, executionHostId: null },
|
||||
{ id: 'repo-ssh', connectionId: 'box', executionHostId: null }
|
||||
]
|
||||
|
||||
const BY_NAME: Record<string, string> = { feature: 'repo-local::/a' }
|
||||
|
||||
function runtime(queue: WorkspaceLifecycleQueue | null) {
|
||||
return {
|
||||
workspaceLifecycle: queue,
|
||||
listRepos: () => repos,
|
||||
showManagedWorktree: async (selector: string) => {
|
||||
const id = BY_NAME[selector] ?? selector.replace(/^id:/, '')
|
||||
if (!id.includes('::')) {
|
||||
throw new Error('selector_not_found')
|
||||
}
|
||||
return { id, repoId: id.split('::')[0] }
|
||||
},
|
||||
listTerminals: async () => ({ totalCount: 0 })
|
||||
}
|
||||
}
|
||||
|
||||
const removal = {
|
||||
force: true,
|
||||
runHooks: false,
|
||||
allowUnverifiedPtyStop: false,
|
||||
allowFailedArchiveHook: false
|
||||
}
|
||||
|
||||
describe('workspace lifecycle commands', () => {
|
||||
it('runs directly in a build with no queue', async () => {
|
||||
let ran = false
|
||||
const reply = await queueTerminalWorkspaceCommand(runtime(null), 'id:repo-local::/a', {
|
||||
kind: 'sleep',
|
||||
run: async () => {
|
||||
ran = true
|
||||
return 'slept'
|
||||
}
|
||||
})
|
||||
expect([ran, reply]).toEqual([true, 'slept'])
|
||||
})
|
||||
|
||||
it('keys a command by the host its repo routes to, the same key a delete uses', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
let release!: () => void
|
||||
const gate = new Promise<void>((resolve) => (release = resolve))
|
||||
const sleeping = queueTerminalWorkspaceCommand(runtime(queue), 'id:repo-ssh::/a', {
|
||||
kind: 'sleep',
|
||||
run: () => gate
|
||||
})
|
||||
await flush()
|
||||
expect(queue.hints()).toEqual({
|
||||
[workspaceLifecycleKey('ssh:box', 'repo-ssh::/a')]: { kind: 'sleep', queuedBehind: 0 }
|
||||
})
|
||||
release()
|
||||
await sleeping
|
||||
})
|
||||
|
||||
it('makes a delete by name wait for the running sleep of the workspace it names', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const log: string[] = []
|
||||
let release!: () => void
|
||||
const gate = new Promise<void>((resolve) => (release = resolve))
|
||||
const sleeping = queueTerminalWorkspaceCommand(runtime(queue), 'id:repo-local::/a', {
|
||||
kind: 'sleep',
|
||||
run: async () => {
|
||||
log.push('sleep')
|
||||
await gate
|
||||
log.push('slept')
|
||||
}
|
||||
})
|
||||
const deleting = queueWorkspaceDelete(
|
||||
runtime(queue),
|
||||
'feature',
|
||||
undefined,
|
||||
removal,
|
||||
async () => {
|
||||
log.push('delete')
|
||||
}
|
||||
)
|
||||
await flush()
|
||||
expect(log).toEqual(['sleep'])
|
||||
expect(queue.hints()).toEqual({
|
||||
[workspaceLifecycleKey('local', 'repo-local::/a')]: { kind: 'delete', queuedBehind: 0 }
|
||||
})
|
||||
release()
|
||||
await Promise.all([sleeping, deleting])
|
||||
expect(log).toEqual(['sleep', 'slept', 'delete'])
|
||||
})
|
||||
|
||||
it('runs a delete of a workspace no lookup finds without queueing it', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
let ran = false
|
||||
await queueWorkspaceDelete(runtime(queue), 'gone', undefined, removal, async () => {
|
||||
ran = true
|
||||
})
|
||||
expect(ran).toBe(true)
|
||||
})
|
||||
|
||||
it('reads lost contact the same way for sleep, close and delete', async () => {
|
||||
expect(terminalLostContact(new Error('terminal_liveness_unavailable'))).toBe(true)
|
||||
expect(terminalLostContact(new Error('terminal_worktree_sleep_failed'))).toBe(false)
|
||||
expect(
|
||||
deleteLostContact(new Error('Timed out waiting for physical PTY teardown: repo::/a.'))
|
||||
).toBe(true)
|
||||
expect(deleteLostContact(new Error('Repo not found: repo'))).toBe(false)
|
||||
const silent = {
|
||||
listTerminals: async () => {
|
||||
throw new Error('terminal_liveness_unavailable')
|
||||
}
|
||||
}
|
||||
await expect(observeTerminals(silent, 'id:a')()).rejects.toThrow()
|
||||
const gone = {
|
||||
listTerminals: async () => {
|
||||
throw new Error('selector_not_found')
|
||||
}
|
||||
}
|
||||
// Any answer is the host answering.
|
||||
await expect(observeTerminals(gone, 'id:a')()).resolves.toBeUndefined()
|
||||
})
|
||||
|
||||
it('reads an unconfirmed sleep or close as unverifiable, and an observed result as settled', () => {
|
||||
const sleep = { stopped: 1, stoppedPtyIds: [], livePtyIds: [] }
|
||||
expect(
|
||||
sleepOutcome({
|
||||
...sleep,
|
||||
postStopVerified: false,
|
||||
postStopFailure: 'terminal_liveness_unavailable'
|
||||
})
|
||||
).toBe('unverifiable')
|
||||
// A terminal observed still running is a definite answer.
|
||||
expect(
|
||||
sleepOutcome({
|
||||
...sleep,
|
||||
postStopVerified: false,
|
||||
postStopFailure: 'terminal_worktree_sleep_still_live',
|
||||
remainingLivePtyIds: ['pty-1']
|
||||
})
|
||||
).toBe('settled')
|
||||
expect(
|
||||
sleepOutcome({ stopped: 1, stoppedPtyIds: [], livePtyIds: [], postStopVerified: true })
|
||||
).toBe('settled')
|
||||
const closed = { closed: 1, stopped: 1, retiredSurfaces: true as const }
|
||||
expect(closeOutcome({ ...closed, ptyStopVerdict: 'unverifiable' })).toBe('unverifiable')
|
||||
// A terminal observed still live is a definite answer, not an unconfirmed one.
|
||||
expect(closeOutcome({ ...closed, ptyStopVerdict: 'live' })).toBe('settled')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,230 @@
|
||||
import type {
|
||||
RuntimeWorktreeTerminalCloseResult,
|
||||
RuntimeWorktreeTerminalSleepResult
|
||||
} from '../../../shared/runtime-types'
|
||||
import {
|
||||
LOCAL_EXECUTION_HOST_ID,
|
||||
parseExecutionHostId,
|
||||
type ExecutionHostId
|
||||
} from '../../../shared/execution-host'
|
||||
import { splitWorktreeId } from '../../../shared/worktree/id'
|
||||
import {
|
||||
isProvenLivePtyRemovalError,
|
||||
isUnstoppedPtyRemovalError
|
||||
} from '../../../shared/worktree/removal'
|
||||
import {
|
||||
getExplicitWorktreeIdSelector,
|
||||
getRuntimeWorktreeRemovalOptionsKey
|
||||
} from '../runtime-worktree-selection'
|
||||
|
||||
import { hostSleepKey } from '../terminal/stop/worktree-terminal-sleep-key'
|
||||
import { resolveWorktreeHostRouting, type LaunchHostRepo } from '../worktree-launch-host-repo'
|
||||
import type {
|
||||
LifecycleCommand,
|
||||
LifecycleDeleteOptions,
|
||||
LifecycleOutcome
|
||||
} from './workspace-lifecycle-command'
|
||||
import type { WorkspaceLifecycleQueue } from './workspace-lifecycle-queue'
|
||||
|
||||
type RemovalFlag = 'force' | 'runHooks' | 'allowUnverifiedPtyStop' | 'allowFailedArchiveHook'
|
||||
|
||||
/** One key per host's copy of a workspace, the same one its terminals sleep under. */
|
||||
export function workspaceLifecycleKey(hostId: ExecutionHostId, worktreeId: string): string {
|
||||
return hostSleepKey(hostId, worktreeId)
|
||||
}
|
||||
|
||||
type LifecycleRuntime = {
|
||||
readonly workspaceLifecycle: WorkspaceLifecycleQueue | null
|
||||
listRepos: () => readonly LaunchHostRepo[]
|
||||
showManagedWorktree: (
|
||||
selector: string
|
||||
) => Promise<{ id: string; repoId: string; hostId?: string | null }>
|
||||
}
|
||||
|
||||
type ObservingRuntime = {
|
||||
listTerminals: (
|
||||
selector: string,
|
||||
limit: number,
|
||||
opts: { requireFreshPtyLiveness?: boolean }
|
||||
) => Promise<unknown>
|
||||
}
|
||||
|
||||
/** Runs `command` in its workspace's queue, or directly in a build with none. */
|
||||
export function runWorkspaceLifecycleCommand<T>(
|
||||
queue: WorkspaceLifecycleQueue | null,
|
||||
key: string,
|
||||
command: LifecycleCommand<T>
|
||||
): Promise<T> {
|
||||
return queue ? queue.submit(key, command) : command.run()
|
||||
}
|
||||
|
||||
const LIVENESS_UNAVAILABLE = 'terminal_liveness_unavailable'
|
||||
|
||||
/** A thrown terminal command that lost the host's liveness answer; any other error is definite. */
|
||||
export function terminalLostContact(error: unknown): boolean {
|
||||
return error instanceof Error && error.message === LIVENESS_UNAVAILABLE
|
||||
}
|
||||
|
||||
// The host could not say whether the stops took; a terminal observed still live is definite.
|
||||
export function sleepOutcome(result: RuntimeWorktreeTerminalSleepResult): LifecycleOutcome {
|
||||
return result.postStopVerified === false && result.postStopFailure === LIVENESS_UNAVAILABLE
|
||||
? 'unverifiable'
|
||||
: 'settled'
|
||||
}
|
||||
|
||||
export function closeOutcome(result: RuntimeWorktreeTerminalCloseResult): LifecycleOutcome {
|
||||
return result.ptyStopVerdict === 'unverifiable' ? 'unverifiable' : 'settled'
|
||||
}
|
||||
|
||||
/** A delete whose PTY stop went unproven (not seen live) lost contact; any other error is definite. */
|
||||
export function deleteLostContact(error: unknown): boolean {
|
||||
if (!(error instanceof Error)) {
|
||||
return false
|
||||
}
|
||||
return (
|
||||
error.message === LIVENESS_UNAVAILABLE ||
|
||||
(isUnstoppedPtyRemovalError(error.message) && !isProvenLivePtyRemovalError(error.message))
|
||||
)
|
||||
}
|
||||
|
||||
/** Re-reads the workspace's terminal liveness; throws only while the host still cannot answer. */
|
||||
export function observeTerminals(runtime: ObservingRuntime, selector: string) {
|
||||
return async (): Promise<void> => {
|
||||
try {
|
||||
await runtime.listTerminals(selector, 1, { requireFreshPtyLiveness: true })
|
||||
} catch (error) {
|
||||
// Any other answer (the workspace is gone, say) is the host answering.
|
||||
if (terminalLostContact(error)) {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** The sync key of an `id:` selector routed by the repo rows; null when only a lookup can say. */
|
||||
function explicitKey(
|
||||
repos: readonly LaunchHostRepo[],
|
||||
selector: string,
|
||||
hostId?: string | null
|
||||
): string | null {
|
||||
const worktreeId = getExplicitWorktreeIdSelector(selector)
|
||||
const repoId = worktreeId ? splitWorktreeId(worktreeId)?.repoId : undefined
|
||||
if (!worktreeId || !repoId) {
|
||||
return null
|
||||
}
|
||||
const named = parseExecutionHostId(hostId ?? undefined)?.id
|
||||
if (named) {
|
||||
return workspaceLifecycleKey(named, worktreeId)
|
||||
}
|
||||
const routing = resolveWorktreeHostRouting(repos, { repoId })
|
||||
if (routing.kind === 'ambiguous') {
|
||||
return null
|
||||
}
|
||||
return workspaceLifecycleKey(
|
||||
routing.kind === 'resolved' ? routing.hostId : LOCAL_EXECUTION_HOST_ID,
|
||||
worktreeId
|
||||
)
|
||||
}
|
||||
|
||||
/** The key a lookup finds: the workspace's own host, else as the repo rows route it. */
|
||||
async function lookedUpKey(
|
||||
runtime: LifecycleRuntime,
|
||||
selector: string,
|
||||
hostId?: string
|
||||
): Promise<string> {
|
||||
const worktree = await runtime.showManagedWorktree(selector)
|
||||
const named = parseExecutionHostId(hostId ?? worktree.hostId ?? undefined)?.id
|
||||
if (named) {
|
||||
return workspaceLifecycleKey(named, worktree.id)
|
||||
}
|
||||
const routing = resolveWorktreeHostRouting(runtime.listRepos(), worktree)
|
||||
if (routing.kind === 'ambiguous') {
|
||||
throw new Error('worktree_execution_host_unresolved')
|
||||
}
|
||||
return workspaceLifecycleKey(
|
||||
routing.kind === 'resolved' ? routing.hostId : LOCAL_EXECUTION_HOST_ID,
|
||||
worktree.id
|
||||
)
|
||||
}
|
||||
|
||||
/** `terminal.sleep`, `terminal.closeAll` and `worktree.sleep` as queued commands, in arrival order. */
|
||||
export function queueTerminalWorkspaceCommand<T>(
|
||||
runtime: LifecycleRuntime,
|
||||
selector: string,
|
||||
command: LifecycleCommand<T>
|
||||
): Promise<T> {
|
||||
const queue = runtime.workspaceLifecycle
|
||||
if (!queue) {
|
||||
return command.run()
|
||||
}
|
||||
const key = explicitKey(runtime.listRepos(), selector)
|
||||
return key
|
||||
? queue.submit(key, command)
|
||||
: queue.submitResolving(() => lookedUpKey(runtime, selector), command)
|
||||
}
|
||||
|
||||
/** `worktree.sleep`: a phone's or the CLI's sleep of the whole workspace. */
|
||||
export function queueWorktreeSleep(
|
||||
runtime: LifecycleRuntime &
|
||||
ObservingRuntime & {
|
||||
sleepManagedWorktree: (selector: string) => Promise<{ worktreeId: string }>
|
||||
},
|
||||
selector: string
|
||||
): Promise<{ worktreeId: string }> {
|
||||
return queueTerminalWorkspaceCommand(runtime, selector, {
|
||||
kind: 'sleep',
|
||||
run: () => runtime.sleepManagedWorktree(selector),
|
||||
lostContact: terminalLostContact,
|
||||
observe: observeTerminals(runtime, selector)
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* `worktree.rm`: keyed by the workspace's resolved id, whatever selector named it, so a delete by
|
||||
* name waits for that workspace's running sleep. A workspace no lookup finds runs unqueued.
|
||||
*/
|
||||
export function queueWorkspaceDelete<T>(
|
||||
runtime: LifecycleRuntime & ObservingRuntime,
|
||||
selector: string,
|
||||
hostId: string | undefined,
|
||||
requested: Partial<Record<RemovalFlag, boolean>>,
|
||||
/** Called with the normalized flags, plus the caller's host when it named one. */
|
||||
run: (removal: Record<RemovalFlag, boolean> & { hostId?: string }) => Promise<T>
|
||||
): Promise<T> {
|
||||
const removal: Record<RemovalFlag, boolean> = {
|
||||
force: requested.force === true,
|
||||
runHooks: requested.runHooks === true,
|
||||
allowUnverifiedPtyStop: requested.allowUnverifiedPtyStop === true,
|
||||
allowFailedArchiveHook: requested.allowFailedArchiveHook === true
|
||||
}
|
||||
const options: LifecycleDeleteOptions = {
|
||||
key: getRuntimeWorktreeRemovalOptionsKey(removal),
|
||||
force: removal.allowUnverifiedPtyStop,
|
||||
label: getExplicitWorktreeIdSelector(selector) ?? selector
|
||||
}
|
||||
const command: LifecycleCommand<T> = {
|
||||
kind: 'delete',
|
||||
run: () => run({ ...removal, ...(hostId ? { hostId } : {}) }),
|
||||
lostContact: deleteLostContact,
|
||||
observe: observeTerminals(runtime, selector),
|
||||
delete: options
|
||||
}
|
||||
const queue = runtime.workspaceLifecycle
|
||||
if (!queue) {
|
||||
return command.run()
|
||||
}
|
||||
const key = explicitKey(runtime.listRepos(), selector, hostId)
|
||||
return key
|
||||
? queue.submit(key, command)
|
||||
: queue.submitResolving(
|
||||
() =>
|
||||
lookedUpKey(runtime, selector, hostId).catch((error: unknown) => {
|
||||
// A delete legitimately reaches a workspace Git no longer lists; nothing orders it.
|
||||
if (error instanceof Error && error.message === 'selector_not_found') {
|
||||
return null
|
||||
}
|
||||
throw error
|
||||
}),
|
||||
command
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type {
|
||||
WorkspaceCreation,
|
||||
WorkspaceCreationsSnapshot
|
||||
} from '../../../shared/sidebar-list/workspace-creations'
|
||||
import type { ProducerRevision } from '../sidebar-inputs/producer-revision'
|
||||
import type { LifecycleMembership } from './workspace-lifecycle-command'
|
||||
|
||||
export type WorkspaceCreationHandle = {
|
||||
/** The workspace is persisted; called before the host announces it. */
|
||||
named: (worktreeId: string) => void
|
||||
close: () => void
|
||||
}
|
||||
|
||||
/** How a create holds its named workspace's queue while it runs. */
|
||||
export type CreationQueuePort<Token> = {
|
||||
revision: ProducerRevision
|
||||
membership: () => LifecycleMembership
|
||||
deletePending: (key: string) => boolean
|
||||
bind: (key: string) => Token
|
||||
finish: (key: string, token: Token) => void
|
||||
}
|
||||
|
||||
type Creation<Token> = WorkspaceCreation & {
|
||||
key?: string
|
||||
token?: Token
|
||||
settling?: boolean
|
||||
closed?: boolean
|
||||
}
|
||||
|
||||
/** The creates in flight: listed from open, bound to the workspace's queue once named. */
|
||||
export class WorkspaceLifecycleCreations<Token> {
|
||||
private readonly creations = new Map<string, Creation<Token>>()
|
||||
|
||||
constructor(private readonly port: CreationQueuePort<Token>) {}
|
||||
|
||||
/** Opened once the project is known; a caller without an id gets one, so a CLI create is listed. */
|
||||
open(start: {
|
||||
creationId?: string
|
||||
repoId: string
|
||||
/** The named workspace's queue key; commands sent to it then wait behind the create. */
|
||||
keyOf?: (worktreeId: string) => string
|
||||
}): WorkspaceCreationHandle {
|
||||
const { port } = this
|
||||
const creationId = start.creationId ?? randomUUID()
|
||||
const creation: Creation<Token> = { repoId: start.repoId }
|
||||
this.creations.set(creationId, creation)
|
||||
port.revision.bump()
|
||||
// Identity, not id: a concurrent retry under the same id owns the listing from then on.
|
||||
const owns = (): boolean => this.creations.get(creationId) === creation
|
||||
return {
|
||||
named: (worktreeId) => {
|
||||
if (creation.closed || !owns() || creation.worktreeId === worktreeId) {
|
||||
return
|
||||
}
|
||||
creation.worktreeId = worktreeId
|
||||
const key = start.keyOf?.(worktreeId)
|
||||
// Why not behind a delete: that delete is the previous occupant's, never this new one's.
|
||||
if (key && !creation.key && !port.deletePending(key)) {
|
||||
creation.key = key
|
||||
creation.token = port.bind(key)
|
||||
}
|
||||
port.revision.bump()
|
||||
},
|
||||
close: () => {
|
||||
if (creation.closed) {
|
||||
return
|
||||
}
|
||||
creation.closed = true
|
||||
// Its own queue entry always ends here, owned or not, so the workspace's queue moves on.
|
||||
if (creation.key && creation.token !== undefined) {
|
||||
port.finish(creation.key, creation.token)
|
||||
}
|
||||
if (!owns()) {
|
||||
return
|
||||
}
|
||||
const finish = (): void => {
|
||||
if (owns()) {
|
||||
this.creations.delete(creationId)
|
||||
}
|
||||
port.revision.bump()
|
||||
}
|
||||
if (creation.worktreeId === undefined) {
|
||||
finish()
|
||||
return
|
||||
}
|
||||
// Why held: the creation shows until a listing shows the workspace, in the same revision.
|
||||
creation.settling = true
|
||||
port.membership()(finish)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
snapshot(): WorkspaceCreationsSnapshot {
|
||||
const creations: WorkspaceCreationsSnapshot['creations'] = {}
|
||||
for (const [creationId, { repoId, worktreeId }] of this.creations) {
|
||||
creations[creationId] = worktreeId === undefined ? { repoId } : { repoId, worktreeId }
|
||||
}
|
||||
return { revision: this.port.revision.value, creations }
|
||||
}
|
||||
|
||||
/** Queue keys of finished creates still waiting for a listing to show them. */
|
||||
settlingKeys(): string[] {
|
||||
return [...this.creations.values()].flatMap(({ key, settling }) =>
|
||||
key && settling ? [key] : []
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,521 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { WorkspaceLifecycleQueue } from './workspace-lifecycle-queue'
|
||||
import { WORKSPACE_DELETED_ERROR, type LifecycleCommand } from './workspace-lifecycle-command'
|
||||
|
||||
const KEY = 'local|repo-1::/wt'
|
||||
|
||||
function deferred<T = void>() {
|
||||
let resolve!: (value: T) => void
|
||||
let reject!: (error: unknown) => void
|
||||
const promise = new Promise<T>((res, rej) => {
|
||||
resolve = res
|
||||
reject = rej
|
||||
})
|
||||
return { promise, resolve, reject }
|
||||
}
|
||||
|
||||
/** A command that records when it starts and finishes only when released. */
|
||||
function gated<T>(
|
||||
log: string[],
|
||||
name: string,
|
||||
kind: LifecycleCommand<T>['kind'],
|
||||
value: T,
|
||||
extra: Partial<LifecycleCommand<T>> = {}
|
||||
) {
|
||||
const gate = deferred()
|
||||
const command: LifecycleCommand<T> = {
|
||||
kind,
|
||||
run: async () => {
|
||||
log.push(`start ${name}`)
|
||||
await gate.promise
|
||||
log.push(`end ${name}`)
|
||||
return value
|
||||
},
|
||||
...extra
|
||||
}
|
||||
return { command, release: () => gate.resolve() }
|
||||
}
|
||||
|
||||
const flush = () => new Promise((resolve) => setTimeout(resolve, 0))
|
||||
|
||||
const deleteOptions = (key: string, force = false) => ({
|
||||
delete: { key, force, label: 'repo-1::/wt' }
|
||||
})
|
||||
|
||||
/** A queue whose retries after a silent owner wait for the test to release them. */
|
||||
function manualRetryQueue() {
|
||||
const retries: (() => void)[] = []
|
||||
const queue = new WorkspaceLifecycleQueue(
|
||||
() => new Promise<void>((resolve) => retries.push(resolve))
|
||||
)
|
||||
return { queue, retry: () => retries.splice(0).forEach((resolve) => resolve()) }
|
||||
}
|
||||
|
||||
describe('WorkspaceLifecycleQueue', () => {
|
||||
it('runs one command at a time, in the order accepted, without merging kinds', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const log: string[] = []
|
||||
const sleep = gated(log, 'sleep', 'sleep', 'slept')
|
||||
const close = gated(log, 'close', 'close', 'closed')
|
||||
const secondSleep = gated(log, 'sleep-2', 'sleep', 'slept-2')
|
||||
const replies = [
|
||||
queue.submit(KEY, sleep.command),
|
||||
queue.submit(KEY, close.command),
|
||||
queue.submit(KEY, secondSleep.command)
|
||||
]
|
||||
await flush()
|
||||
expect(log).toEqual(['start sleep'])
|
||||
expect(queue.hints()).toEqual({ [KEY]: { kind: 'sleep', queuedBehind: 2 } })
|
||||
|
||||
sleep.release()
|
||||
await flush()
|
||||
expect(log).toEqual(['start sleep', 'end sleep', 'start close'])
|
||||
expect(queue.hints()).toEqual({ [KEY]: { kind: 'close', queuedBehind: 1 } })
|
||||
close.release()
|
||||
await flush()
|
||||
secondSleep.release()
|
||||
expect(await Promise.all(replies)).toEqual(['slept', 'closed', 'slept-2'])
|
||||
expect(log).toEqual([
|
||||
'start sleep',
|
||||
'end sleep',
|
||||
'start close',
|
||||
'end close',
|
||||
'start sleep-2',
|
||||
'end sleep-2'
|
||||
])
|
||||
expect(queue.hints()).toEqual({})
|
||||
})
|
||||
|
||||
it('runs a repeated command again rather than merging it into the first', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
let runs = 0
|
||||
const sleep: LifecycleCommand<number> = { kind: 'sleep', run: async () => ++runs }
|
||||
expect(await Promise.all([queue.submit(KEY, sleep), queue.submit(KEY, sleep)])).toEqual([1, 2])
|
||||
})
|
||||
|
||||
it('keeps separate workspaces apart, including one id on two hosts', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const log: string[] = []
|
||||
const local = gated(log, 'local', 'sleep', 1)
|
||||
const ssh = gated(log, 'ssh', 'sleep', 2)
|
||||
void queue.submit(KEY, local.command)
|
||||
void queue.submit('ssh:box|repo-1::/wt', ssh.command)
|
||||
await flush()
|
||||
expect(log).toEqual(['start local', 'start ssh'])
|
||||
local.release()
|
||||
ssh.release()
|
||||
})
|
||||
|
||||
it('delete cancels what has not started and waits for what has', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const log: string[] = []
|
||||
const sleep = gated(log, 'sleep', 'sleep', 'slept')
|
||||
const wake = gated(log, 'wake', 'wake', 'woken')
|
||||
const remove = gated(log, 'delete', 'delete', 'deleted', deleteOptions('normal'))
|
||||
const sleeping = queue.submit(KEY, sleep.command)
|
||||
const waking = queue.submit(KEY, wake.command)
|
||||
await flush()
|
||||
const deleting = queue.submit(KEY, remove.command)
|
||||
// Acknowledged at once: its hint replaces the queued ones.
|
||||
expect(queue.hints()).toEqual({ [KEY]: { kind: 'delete', queuedBehind: 0 } })
|
||||
await expect(waking).rejects.toThrow(WORKSPACE_DELETED_ERROR)
|
||||
// A command after the delete is refused, not run on a workspace about to go.
|
||||
await expect(queue.submit(KEY, wake.command)).rejects.toThrow(WORKSPACE_DELETED_ERROR)
|
||||
expect(log).toEqual(['start sleep'])
|
||||
|
||||
sleep.release()
|
||||
await flush()
|
||||
expect(log).toEqual(['start sleep', 'end sleep', 'start delete'])
|
||||
remove.release()
|
||||
expect(await sleeping).toBe('slept')
|
||||
expect(await deleting).toBe('deleted')
|
||||
expect(log).not.toContain('start wake')
|
||||
})
|
||||
|
||||
it('joins a second delete with the same options, and refuses one with different options', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const log: string[] = []
|
||||
const remove = gated(log, 'delete', 'delete', 'deleted', deleteOptions('normal'))
|
||||
const first = queue.submit(KEY, remove.command)
|
||||
const second = queue.submit(
|
||||
KEY,
|
||||
gated(log, 'delete-2', 'delete', 'other', deleteOptions('normal')).command
|
||||
)
|
||||
await expect(
|
||||
queue.submit(KEY, gated(log, 'delete-3', 'delete', 'x', deleteOptions('hooks')).command)
|
||||
).rejects.toThrow('Worktree deletion already in progress: repo-1::/wt')
|
||||
remove.release()
|
||||
expect(await Promise.all([first, second])).toEqual(['deleted', 'deleted'])
|
||||
expect(log).toEqual(['start delete', 'end delete'])
|
||||
})
|
||||
|
||||
it('pauses on an unconfirmed outcome; re-observes when the owner answers, never replaying it', async () => {
|
||||
const { queue, retry } = manualRetryQueue()
|
||||
const log: string[] = []
|
||||
let reachable = false
|
||||
const sleep: LifecycleCommand<{ postStopVerified: boolean }> = {
|
||||
kind: 'sleep',
|
||||
run: async () => {
|
||||
log.push('sleep')
|
||||
return { postStopVerified: reachable }
|
||||
},
|
||||
outcome: (result) => (result.postStopVerified ? 'settled' : 'unverifiable'),
|
||||
observe: async () => {
|
||||
log.push('observe')
|
||||
if (!reachable) {
|
||||
throw new Error('terminal_liveness_unavailable')
|
||||
}
|
||||
}
|
||||
}
|
||||
// The requester gets the unconfirmed reply (today's toast); the workspace stays as observed.
|
||||
expect(await queue.submit(KEY, sleep)).toEqual({ postStopVerified: false })
|
||||
expect(queue.isPaused(KEY)).toBe(true)
|
||||
expect(queue.hints()).toEqual({ [KEY]: { kind: 'sleep', queuedBehind: 0, paused: true } })
|
||||
|
||||
// A command issued while paused waits behind the pause; it is neither refused nor run.
|
||||
const woke: string[] = []
|
||||
const waking = queue.submit(KEY, {
|
||||
kind: 'wake',
|
||||
run: async () => {
|
||||
woke.push('wake')
|
||||
return 'woken'
|
||||
}
|
||||
})
|
||||
await flush()
|
||||
expect(log).toEqual(['sleep', 'observe'])
|
||||
expect(woke).toEqual([])
|
||||
expect(queue.hints()).toEqual({ [KEY]: { kind: 'wake', queuedBehind: 0, paused: true } })
|
||||
|
||||
// Still silent on the next ask; then the owner answers and the queue carries on.
|
||||
retry()
|
||||
await flush()
|
||||
expect(log).toEqual(['sleep', 'observe', 'observe'])
|
||||
reachable = true
|
||||
retry()
|
||||
expect(await waking).toBe('woken')
|
||||
expect(log).toEqual(['sleep', 'observe', 'observe', 'observe'])
|
||||
expect(queue.isPaused(KEY)).toBe(false)
|
||||
expect(queue.hints()).toEqual({})
|
||||
})
|
||||
|
||||
it('re-runs a repeated command once contact is back', async () => {
|
||||
const { queue } = manualRetryQueue()
|
||||
let reachable = false
|
||||
let sleeps = 0
|
||||
const sleep: LifecycleCommand<boolean> = {
|
||||
kind: 'sleep',
|
||||
run: async () => {
|
||||
sleeps += 1
|
||||
return reachable
|
||||
},
|
||||
outcome: (ok) => (ok ? 'settled' : 'unverifiable'),
|
||||
observe: async () => {
|
||||
if (!reachable) {
|
||||
throw new Error('lost')
|
||||
}
|
||||
}
|
||||
}
|
||||
expect(await queue.submit(KEY, sleep)).toBe(false)
|
||||
reachable = true
|
||||
expect(await queue.submit(KEY, sleep)).toBe(true)
|
||||
expect(sleeps).toBe(2)
|
||||
expect(queue.isPaused(KEY)).toBe(false)
|
||||
})
|
||||
|
||||
it('pauses on lost contact mid-run, but not on a definite failure', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const lost = (message: string): LifecycleCommand<never> => ({
|
||||
kind: 'close',
|
||||
run: async () => {
|
||||
throw new Error(message)
|
||||
},
|
||||
lostContact: (error) => error instanceof Error && error.message === 'lost'
|
||||
})
|
||||
await expect(queue.submit(KEY, lost('refused'))).rejects.toThrow('refused')
|
||||
expect(queue.isPaused(KEY)).toBe(false)
|
||||
await expect(queue.submit(KEY, lost('lost'))).rejects.toThrow('lost')
|
||||
expect(queue.isPaused(KEY)).toBe(true)
|
||||
})
|
||||
|
||||
it('acknowledges a delete issued while paused and runs it once the owner answers', async () => {
|
||||
const { queue, retry } = manualRetryQueue()
|
||||
const log: string[] = []
|
||||
let reachable = false
|
||||
await queue.submit(KEY, {
|
||||
kind: 'sleep',
|
||||
run: async () => false,
|
||||
outcome: () => 'unverifiable',
|
||||
observe: async () => {
|
||||
if (!reachable) {
|
||||
throw new Error('lost')
|
||||
}
|
||||
}
|
||||
})
|
||||
const deleting = queue.submit(KEY, {
|
||||
kind: 'delete',
|
||||
run: async () => {
|
||||
log.push('delete')
|
||||
return 'deleted'
|
||||
},
|
||||
...deleteOptions('normal')
|
||||
})
|
||||
await flush()
|
||||
expect(queue.hints()).toEqual({ [KEY]: { kind: 'delete', queuedBehind: 0, paused: true } })
|
||||
expect(log).toEqual([])
|
||||
reachable = true
|
||||
retry()
|
||||
expect(await deleting).toBe('deleted')
|
||||
expect(queue.isPaused(KEY)).toBe(false)
|
||||
})
|
||||
|
||||
it('runs a forced delete past the pause at once, in place of an unforced one waiting', async () => {
|
||||
const { queue } = manualRetryQueue()
|
||||
const log: string[] = []
|
||||
await queue.submit(KEY, {
|
||||
kind: 'sleep',
|
||||
run: async () => false,
|
||||
outcome: () => 'unverifiable',
|
||||
observe: () => new Promise<void>(() => {})
|
||||
})
|
||||
const waiting = queue.submit(KEY, {
|
||||
kind: 'delete',
|
||||
run: async () => 'unforced',
|
||||
...deleteOptions('normal')
|
||||
})
|
||||
await flush()
|
||||
const forced = queue.submit(KEY, {
|
||||
kind: 'delete',
|
||||
run: async () => {
|
||||
log.push('forced')
|
||||
return 'deleted'
|
||||
},
|
||||
...deleteOptions('force', true)
|
||||
})
|
||||
await expect(waiting).rejects.toThrow(WORKSPACE_DELETED_ERROR)
|
||||
expect(await forced).toBe('deleted')
|
||||
expect(log).toEqual(['forced'])
|
||||
expect(queue.isPaused(KEY)).toBe(false)
|
||||
})
|
||||
|
||||
it('starts commands in arrival order even when an earlier one is still looking up its workspace', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const log: string[] = []
|
||||
const lookup = deferred<string | null>()
|
||||
const first = queue.submitResolving(() => lookup.promise, {
|
||||
kind: 'sleep',
|
||||
run: async () => {
|
||||
log.push('sleep (by name)')
|
||||
return 'slept'
|
||||
}
|
||||
})
|
||||
const second = queue.submit(KEY, {
|
||||
kind: 'wake',
|
||||
run: async () => {
|
||||
log.push('wake (by id)')
|
||||
return 'woken'
|
||||
}
|
||||
})
|
||||
await flush()
|
||||
expect(log).toEqual([])
|
||||
lookup.resolve(KEY)
|
||||
expect(await Promise.all([first, second])).toEqual(['slept', 'woken'])
|
||||
expect(log).toEqual(['sleep (by name)', 'wake (by id)'])
|
||||
})
|
||||
|
||||
it('lets a lookup stuck on a dead host hold another workspace only briefly', async () => {
|
||||
let clock = 0
|
||||
const queue = new WorkspaceLifecycleQueue(
|
||||
async (ms) => {
|
||||
clock += ms
|
||||
},
|
||||
() => clock
|
||||
)
|
||||
void queue.submitResolving(() => new Promise<string | null>(() => {}), {
|
||||
kind: 'sleep',
|
||||
run: async () => 'never'
|
||||
})
|
||||
const other = queue.submit('ssh:other|r::/w', { kind: 'wake', run: async () => 'woken' })
|
||||
expect(await other).toBe('woken')
|
||||
expect(clock).toBe(500)
|
||||
})
|
||||
|
||||
it('never cancels a command because its requester went away', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const log: string[] = []
|
||||
const sleep = gated(log, 'sleep', 'sleep', 'slept')
|
||||
// The requester drops its reply (a phone disconnect); the command still runs to its end.
|
||||
void queue.submit(KEY, sleep.command).catch(() => {})
|
||||
await flush()
|
||||
sleep.release()
|
||||
await flush()
|
||||
expect(log).toEqual(['start sleep', 'end sleep'])
|
||||
})
|
||||
|
||||
it('leaves nothing pending in a fresh queue, as after a restart', () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
expect(queue.hints()).toEqual({})
|
||||
expect(queue.creationsSnapshot().creations).toEqual({})
|
||||
})
|
||||
})
|
||||
|
||||
describe('WorkspaceLifecycleQueue creates', () => {
|
||||
it('lists a creation once opened, names its workspace, and drops it on close', () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1' })
|
||||
expect(queue.creationsSnapshot().creations).toEqual({ c1: { repoId: 'repo-1' } })
|
||||
const opened = queue.creationsSnapshot().revision
|
||||
creation.named('repo-1::/wt')
|
||||
expect(queue.creationsSnapshot().creations).toEqual({
|
||||
c1: { repoId: 'repo-1', worktreeId: 'repo-1::/wt' }
|
||||
})
|
||||
expect(queue.creationsSnapshot().revision).toBeGreaterThan(opened)
|
||||
creation.close()
|
||||
expect(queue.creationsSnapshot().creations).toEqual({})
|
||||
})
|
||||
|
||||
it('mints an id for a caller that sends none', () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const creation = queue.openCreate({ repoId: 'repo-1' })
|
||||
const ids = Object.keys(queue.creationsSnapshot().creations)
|
||||
expect(ids).toHaveLength(1)
|
||||
expect(ids[0]).not.toBe('')
|
||||
creation.close()
|
||||
expect(queue.creationsSnapshot().creations).toEqual({})
|
||||
})
|
||||
|
||||
it('leaves a concurrent retry under the same id listed when the first create closes', () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const first = queue.openCreate({ creationId: 'c1', repoId: 'repo-1' })
|
||||
const second = queue.openCreate({ creationId: 'c1', repoId: 'repo-1' })
|
||||
first.named('repo-1::/stale')
|
||||
first.close()
|
||||
expect(queue.creationsSnapshot().creations).toEqual({ c1: { repoId: 'repo-1' } })
|
||||
second.close()
|
||||
expect(queue.creationsSnapshot().creations).toEqual({})
|
||||
})
|
||||
|
||||
it('frees the workspace queue when a create closes after a retry took over its id', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const first = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY })
|
||||
first.named('repo-1::/wt')
|
||||
const retry = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY })
|
||||
const sleeping = queue.submit(KEY, { kind: 'sleep', run: async () => 'slept' })
|
||||
first.close()
|
||||
expect(await sleeping).toBe('slept')
|
||||
retry.close()
|
||||
})
|
||||
|
||||
it('closes once', () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1' })
|
||||
creation.close()
|
||||
const closed = queue.creationsSnapshot().revision
|
||||
creation.close()
|
||||
expect(queue.creationsSnapshot().revision).toBe(closed)
|
||||
})
|
||||
|
||||
it('binds to the workspace once named, so commands on the Creating row wait behind it', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const log: string[] = []
|
||||
const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY })
|
||||
creation.named('repo-1::/wt')
|
||||
expect(queue.hints()).toEqual({ [KEY]: { kind: 'create', queuedBehind: 0 } })
|
||||
const sleep = gated(log, 'sleep', 'sleep', 'slept')
|
||||
const sleeping = queue.submit(KEY, sleep.command)
|
||||
await flush()
|
||||
expect(log).toEqual([])
|
||||
expect(queue.hints()).toEqual({ [KEY]: { kind: 'create', queuedBehind: 1 } })
|
||||
creation.close()
|
||||
await flush()
|
||||
expect(log).toEqual(['start sleep'])
|
||||
sleep.release()
|
||||
expect(await sleeping).toBe('slept')
|
||||
})
|
||||
|
||||
it('lets a delete cancel what waits behind a create and run once the create ends', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const log: string[] = []
|
||||
const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY })
|
||||
creation.named('repo-1::/wt')
|
||||
const sleeping = queue.submit(KEY, gated(log, 'sleep', 'sleep', 'slept').command)
|
||||
const remove = gated(log, 'delete', 'delete', 'deleted')
|
||||
const deleting = queue.submit(KEY, remove.command)
|
||||
await expect(sleeping).rejects.toThrow(WORKSPACE_DELETED_ERROR)
|
||||
await flush()
|
||||
expect(log).toEqual([])
|
||||
creation.close()
|
||||
await flush()
|
||||
expect(log).toEqual(['start delete'])
|
||||
remove.release()
|
||||
expect(await deleting).toBe('deleted')
|
||||
})
|
||||
})
|
||||
|
||||
describe('WorkspaceLifecycleQueue create fencing', () => {
|
||||
it('never puts a new create behind the previous occupant’s delete', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const log: string[] = []
|
||||
const sleep = gated(log, 'sleep', 'sleep', 'slept')
|
||||
void queue.submit(KEY, sleep.command)
|
||||
const deleting = queue.submit(KEY, gated(log, 'delete', 'delete', 'deleted').command)
|
||||
const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY })
|
||||
creation.named('repo-1::/wt')
|
||||
// The delete still waits on the sleep alone; the create is not one of its commands.
|
||||
expect(queue.hints()).toEqual({ [KEY]: { kind: 'delete', queuedBehind: 0 } })
|
||||
creation.close()
|
||||
sleep.release()
|
||||
await flush()
|
||||
expect(log).toEqual(['start sleep', 'end sleep', 'start delete'])
|
||||
void deleting
|
||||
})
|
||||
})
|
||||
|
||||
describe('WorkspaceLifecycleQueue membership', () => {
|
||||
/** A listing whose fresh scan the test applies by hand. */
|
||||
function listing() {
|
||||
const waiters: (() => void)[] = []
|
||||
return {
|
||||
membership: (done: () => void) => waiters.push(done),
|
||||
applyFreshScan: () => waiters.splice(0).forEach((done) => done())
|
||||
}
|
||||
}
|
||||
|
||||
it('holds a finished create’s hint until a fresh listing, then drops it in that revision', () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const rows = listing()
|
||||
queue.setMembership(rows.membership)
|
||||
const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY })
|
||||
creation.named('repo-1::/wt')
|
||||
creation.close()
|
||||
// A scan that started before the create ended cannot drop the row: the creation still shows.
|
||||
expect(queue.creationsSnapshot().creations).toEqual({
|
||||
c1: { repoId: 'repo-1', worktreeId: 'repo-1::/wt' }
|
||||
})
|
||||
expect(queue.hints()).toEqual({ [KEY]: { kind: 'create', queuedBehind: 0 } })
|
||||
const bumps: number[] = []
|
||||
queue.revision.onChange(() => bumps.push(queue.revision.value))
|
||||
rows.applyFreshScan()
|
||||
expect(queue.creationsSnapshot().creations).toEqual({})
|
||||
expect(queue.hints()).toEqual({})
|
||||
// One synchronous bump, inside the listing's apply: the feed flushes both together.
|
||||
expect(bumps).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('drops a failed create at once, with nothing to list', () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const rows = listing()
|
||||
queue.setMembership(rows.membership)
|
||||
queue.openCreate({ creationId: 'c1', repoId: 'repo-1' }).close()
|
||||
expect(queue.creationsSnapshot().creations).toEqual({})
|
||||
})
|
||||
|
||||
it('shows a finished delete until a fresh listing drops the row', async () => {
|
||||
const queue = new WorkspaceLifecycleQueue()
|
||||
const rows = listing()
|
||||
queue.setMembership(rows.membership)
|
||||
expect(await queue.submit(KEY, { kind: 'delete', run: async () => 'deleted' })).toBe('deleted')
|
||||
expect(queue.hints()).toEqual({ [KEY]: { kind: 'delete', queuedBehind: 0 } })
|
||||
rows.applyFreshScan()
|
||||
expect(queue.hints()).toEqual({})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,362 @@
|
||||
import { ProducerRevision } from '../sidebar-inputs/producer-revision'
|
||||
import type {
|
||||
WorkspaceLifecycleHint,
|
||||
WorkspaceLifecycleKind
|
||||
} from '../../../shared/sidebar-list/workspace-lifecycle-hints'
|
||||
import type { WorkspaceCreationsSnapshot } from '../../../shared/sidebar-list/workspace-creations'
|
||||
import {
|
||||
WORKSPACE_DELETED_ERROR,
|
||||
attemptLifecycleCommand,
|
||||
type LifecycleAttempt,
|
||||
type LifecycleCommand,
|
||||
type LifecycleMembership
|
||||
} from './workspace-lifecycle-command'
|
||||
import {
|
||||
WorkspaceLifecycleCreations,
|
||||
type WorkspaceCreationHandle
|
||||
} from './workspace-lifecycle-creations'
|
||||
|
||||
type Entry = {
|
||||
kind: WorkspaceLifecycleKind
|
||||
/** Arrival order across every workspace, so a command waiting on its lookup keeps its place. */
|
||||
seq: number
|
||||
state: 'queued' | 'running'
|
||||
/** Null for a create, which runs outside the queue between open and close. */
|
||||
command: LifecycleCommand<unknown> | null
|
||||
reply: Promise<unknown> | null
|
||||
settle: (attempt: LifecycleAttempt) => void
|
||||
}
|
||||
|
||||
type Pause = { kind: WorkspaceLifecycleKind; observe: (() => Promise<void>) | null }
|
||||
|
||||
type WorkspaceQueue = {
|
||||
entries: Entry[]
|
||||
paused: Pause | null
|
||||
/** A finished create or delete, held until a fresh listing shows its membership. */
|
||||
settling: number
|
||||
draining: boolean
|
||||
/** Ends a wait on the pause early: a forced delete arrived. */
|
||||
wake: (() => void) | null
|
||||
}
|
||||
|
||||
type Signal = { promise: Promise<void>; notify: () => void }
|
||||
|
||||
const OBSERVE_RETRY_MS = 5_000
|
||||
// Long enough for a lookup on a reachable host, so its command keeps its place in the order.
|
||||
const LOOKUP_ORDER_WAIT_MS = 500
|
||||
|
||||
function signal(): Signal {
|
||||
let notify!: () => void
|
||||
const promise = new Promise<void>((resolve) => (notify = resolve))
|
||||
return { promise, notify }
|
||||
}
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms).unref?.())
|
||||
}
|
||||
|
||||
/**
|
||||
* One FIFO queue per host-qualified workspace for create, sleep, wake, close and delete. Commands
|
||||
* run one at a time in the order they arrived and are never merged. Delete cancels what has not
|
||||
* started and waits for what has. An unconfirmed outcome pauses the queue until the owner answers
|
||||
* again; main then re-observes and carries on, never replaying the paused command. Lives in
|
||||
* memory: after a restart nothing is pending and readers show observed state.
|
||||
*/
|
||||
export class WorkspaceLifecycleQueue {
|
||||
readonly revision = new ProducerRevision()
|
||||
private readonly queues = new Map<string, WorkspaceQueue>()
|
||||
private membership: LifecycleMembership = (done) => done()
|
||||
private nextSeq = 0
|
||||
/**
|
||||
* Arrivals still looking up their workspace, by when they arrived. A later arrival waits for
|
||||
* them only briefly: the lookup may name any workspace, and a dead host must not stall them all.
|
||||
*/
|
||||
private readonly resolving = new Map<number, number>()
|
||||
private resolved: Signal = signal()
|
||||
private readonly creations = new WorkspaceLifecycleCreations<Entry>({
|
||||
revision: this.revision,
|
||||
membership: () => this.membership,
|
||||
deletePending: (key) => this.deletePending(key),
|
||||
bind: (key) => this.bindCreate(key),
|
||||
finish: (key, entry) => this.finishEntry(key, entry)
|
||||
})
|
||||
|
||||
constructor(
|
||||
private readonly retryDelay: (ms: number) => Promise<void> = sleep,
|
||||
private readonly now: () => number = Date.now
|
||||
) {}
|
||||
|
||||
/** How long `seq` still waits for an earlier arrival's lookup; 0 once none holds it. */
|
||||
private lookupWaitMs(seq: number): number {
|
||||
const waits = [...this.resolving].flatMap(([earlier, arrivedAt]) =>
|
||||
earlier < seq ? [arrivedAt + LOOKUP_ORDER_WAIT_MS - this.now()] : []
|
||||
)
|
||||
return Math.max(0, ...waits)
|
||||
}
|
||||
|
||||
/** Where finished creates and deletes wait to be listed; until set they settle at once. */
|
||||
setMembership(membership: LifecycleMembership): void {
|
||||
this.membership = membership
|
||||
}
|
||||
|
||||
submit<T>(key: string, command: LifecycleCommand<T>): Promise<T> {
|
||||
return this.enqueue(key, command, this.nextSeq++)
|
||||
}
|
||||
|
||||
/**
|
||||
* For a command whose workspace is known only after a lookup: it takes its place on arrival.
|
||||
* A lookup answering null finds no workspace to order against, so the command runs directly.
|
||||
*/
|
||||
async submitResolving<T>(
|
||||
resolveKey: () => Promise<string | null>,
|
||||
command: LifecycleCommand<T>
|
||||
): Promise<T> {
|
||||
const seq = this.nextSeq++
|
||||
this.resolving.set(seq, this.now())
|
||||
let key: string | null
|
||||
try {
|
||||
key = await resolveKey()
|
||||
} finally {
|
||||
this.resolving.delete(seq)
|
||||
const resolved = this.resolved
|
||||
this.resolved = signal()
|
||||
resolved.notify()
|
||||
}
|
||||
return key === null ? command.run() : this.enqueue(key, command, seq)
|
||||
}
|
||||
|
||||
openCreate(
|
||||
start: Parameters<WorkspaceLifecycleCreations<Entry>['open']>[0]
|
||||
): WorkspaceCreationHandle {
|
||||
return this.creations.open(start)
|
||||
}
|
||||
|
||||
/** The per-workspace hints the feed publishes. */
|
||||
hints(): Record<string, WorkspaceLifecycleHint> {
|
||||
const hints: Record<string, WorkspaceLifecycleHint> = {}
|
||||
for (const [key, queue] of this.queues) {
|
||||
const deleting = queue.entries.some((entry) => entry.kind === 'delete')
|
||||
const head = queue.entries[0]
|
||||
const kind = deleting ? 'delete' : (head?.kind ?? queue.paused?.kind)
|
||||
if (kind === undefined) {
|
||||
// Only a settling delete is left: it shows until its row leaves the listing.
|
||||
if (queue.settling > 0) {
|
||||
hints[key] = { kind: 'delete', queuedBehind: 0 }
|
||||
}
|
||||
continue
|
||||
}
|
||||
hints[key] = {
|
||||
kind,
|
||||
queuedBehind: deleting ? 0 : Math.max(0, queue.entries.length - 1),
|
||||
...(queue.paused ? { paused: true as const } : {})
|
||||
}
|
||||
}
|
||||
for (const key of this.creations.settlingKeys()) {
|
||||
hints[key] ??= { kind: 'create', queuedBehind: 0 }
|
||||
}
|
||||
return hints
|
||||
}
|
||||
|
||||
/** The creations projection the `creations` readers already read. */
|
||||
creationsSnapshot(): WorkspaceCreationsSnapshot {
|
||||
return this.creations.snapshot()
|
||||
}
|
||||
|
||||
isPaused(key: string): boolean {
|
||||
return this.queues.get(key)?.paused != null
|
||||
}
|
||||
|
||||
private enqueue<T>(key: string, command: LifecycleCommand<T>, seq: number): Promise<T> {
|
||||
const queue = this.queueFor(key)
|
||||
const pendingDelete = queue.entries.find((entry) => entry.kind === 'delete')
|
||||
if (pendingDelete) {
|
||||
const pending = pendingDelete.command?.delete
|
||||
if (command.kind !== 'delete') {
|
||||
return Promise.reject(new Error(WORKSPACE_DELETED_ERROR))
|
||||
}
|
||||
if (pendingDelete.reply && pending?.key === command.delete?.key) {
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: an equal delete shares the pending delete's reply, as a removal in flight always has.
|
||||
return pendingDelete.reply as Promise<T>
|
||||
}
|
||||
// A forced delete replaces an unforced one still waiting (behind a pause); never a running one.
|
||||
if (!command.delete?.force || pendingDelete.state === 'running') {
|
||||
const label = command.delete?.label ?? key
|
||||
return Promise.reject(new Error(`Worktree deletion already in progress: ${label}`))
|
||||
}
|
||||
}
|
||||
if (command.kind === 'delete') {
|
||||
this.cancelQueued(queue)
|
||||
}
|
||||
let settle!: (attempt: LifecycleAttempt) => void
|
||||
const reply = new Promise<unknown>((resolve, reject) => {
|
||||
settle = (result) => (result.threw ? reject(result.error) : resolve(result.value))
|
||||
})
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: widened only to be stored; run's result is handed back to this command's own caller.
|
||||
const stored = command as LifecycleCommand<unknown>
|
||||
const entry: Entry = {
|
||||
kind: command.kind,
|
||||
seq,
|
||||
state: 'queued',
|
||||
command: stored,
|
||||
reply,
|
||||
settle
|
||||
}
|
||||
// In arrival order among what has not started; a late lookup goes ahead of later arrivals.
|
||||
const after = queue.entries.findIndex((queued) => queued.state === 'queued' && queued.seq > seq)
|
||||
queue.entries.splice(after === -1 ? queue.entries.length : after, 0, entry)
|
||||
this.revision.bump()
|
||||
if (command.delete?.force) {
|
||||
queue.wake?.()
|
||||
}
|
||||
void this.drain(key)
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: settled only with this command's run result.
|
||||
return reply as Promise<T>
|
||||
}
|
||||
|
||||
private deletePending(key: string): boolean {
|
||||
return this.queues.get(key)?.entries.some((entry) => entry.kind === 'delete') === true
|
||||
}
|
||||
|
||||
private queueFor(key: string): WorkspaceQueue {
|
||||
let queue = this.queues.get(key)
|
||||
if (!queue) {
|
||||
queue = { entries: [], paused: null, settling: 0, draining: false, wake: null }
|
||||
this.queues.set(key, queue)
|
||||
}
|
||||
return queue
|
||||
}
|
||||
|
||||
private bindCreate(key: string): Entry {
|
||||
const queue = this.queueFor(key)
|
||||
const entry: Entry = {
|
||||
kind: 'create',
|
||||
seq: this.nextSeq++,
|
||||
state: 'running',
|
||||
command: null,
|
||||
reply: null,
|
||||
settle: () => {}
|
||||
}
|
||||
// Ahead of everything not started: the create is already running.
|
||||
const firstQueued = queue.entries.findIndex((queued) => queued.state === 'queued')
|
||||
queue.entries.splice(firstQueued === -1 ? queue.entries.length : firstQueued, 0, entry)
|
||||
return entry
|
||||
}
|
||||
|
||||
private finishEntry(key: string, entry: Entry): void {
|
||||
const queue = this.queues.get(key)
|
||||
const index = queue?.entries.indexOf(entry) ?? -1
|
||||
if (queue && index !== -1) {
|
||||
queue.entries.splice(index, 1)
|
||||
this.revision.bump()
|
||||
void this.drain(key)
|
||||
}
|
||||
}
|
||||
|
||||
private cancelQueued(queue: WorkspaceQueue): void {
|
||||
const cancelled = queue.entries.filter((entry) => entry.state === 'queued')
|
||||
queue.entries = queue.entries.filter((entry) => entry.state !== 'queued')
|
||||
for (const entry of cancelled) {
|
||||
entry.settle({ outcome: 'settled', error: new Error(WORKSPACE_DELETED_ERROR), threw: true })
|
||||
}
|
||||
}
|
||||
|
||||
private async drain(key: string): Promise<void> {
|
||||
const queue = this.queues.get(key)
|
||||
if (!queue || queue.draining) {
|
||||
return
|
||||
}
|
||||
queue.draining = true
|
||||
try {
|
||||
for (let head = queue.entries[0]; head; head = queue.entries[0]) {
|
||||
if (head.state === 'running' || !head.command) {
|
||||
// A create running outside the queue; its close drains again.
|
||||
return
|
||||
}
|
||||
const waitMs = this.lookupWaitMs(head.seq)
|
||||
if (waitMs > 0) {
|
||||
await Promise.race([this.resolved.promise, this.retryDelay(waitMs)])
|
||||
continue
|
||||
}
|
||||
if (queue.paused && !head.command.delete?.force) {
|
||||
await this.awaitOwner(queue, queue.paused)
|
||||
continue
|
||||
}
|
||||
await this.runHead(key, queue, head, head.command)
|
||||
}
|
||||
} finally {
|
||||
queue.draining = false
|
||||
this.forgetIfIdle(key, queue)
|
||||
}
|
||||
}
|
||||
|
||||
private async runHead(
|
||||
key: string,
|
||||
queue: WorkspaceQueue,
|
||||
head: Entry,
|
||||
command: LifecycleCommand<unknown>
|
||||
): Promise<void> {
|
||||
head.state = 'running'
|
||||
this.revision.bump()
|
||||
const result = await attemptLifecycleCommand(command)
|
||||
queue.entries.splice(queue.entries.indexOf(head), 1)
|
||||
if (result.outcome === 'unverifiable') {
|
||||
queue.paused = { kind: head.kind, observe: command.observe ?? null }
|
||||
} else if (head.kind === 'delete' && !result.threw) {
|
||||
// Deleted: nothing is left to observe.
|
||||
queue.paused = null
|
||||
queue.settling += 1
|
||||
this.membership(() => {
|
||||
queue.settling -= 1
|
||||
this.revision.bump()
|
||||
this.forgetIfIdle(key, queue)
|
||||
})
|
||||
}
|
||||
head.settle(result)
|
||||
this.revision.bump()
|
||||
}
|
||||
|
||||
/**
|
||||
* Waits for the paused command's owner to answer again, then re-observes the listing and lifts
|
||||
* the pause. Returns early when a forced delete arrives; a silent owner is asked again later.
|
||||
*/
|
||||
private async awaitOwner(queue: WorkspaceQueue, paused: Pause): Promise<void> {
|
||||
if (!paused.observe) {
|
||||
queue.paused = null
|
||||
this.revision.bump()
|
||||
return
|
||||
}
|
||||
const woken = new Promise<'woken'>((resolve) => (queue.wake = () => resolve('woken')))
|
||||
try {
|
||||
const answered = await Promise.race([
|
||||
paused.observe().then(
|
||||
() => true,
|
||||
() => false
|
||||
),
|
||||
woken
|
||||
])
|
||||
if (answered === true) {
|
||||
await new Promise<void>((done) => this.membership(done))
|
||||
if (queue.paused === paused) {
|
||||
queue.paused = null
|
||||
this.revision.bump()
|
||||
}
|
||||
} else if (answered === false) {
|
||||
await Promise.race([this.retryDelay(OBSERVE_RETRY_MS), woken])
|
||||
}
|
||||
} finally {
|
||||
queue.wake = null
|
||||
}
|
||||
}
|
||||
|
||||
private forgetIfIdle(key: string, queue: WorkspaceQueue): void {
|
||||
if (
|
||||
queue.entries.length === 0 &&
|
||||
!queue.paused &&
|
||||
queue.settling === 0 &&
|
||||
!queue.draining &&
|
||||
this.queues.get(key) === queue
|
||||
) {
|
||||
this.queues.delete(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
import { z } from 'zod'
|
||||
|
||||
export const WORKSPACE_LIFECYCLE_KINDS = ['create', 'sleep', 'wake', 'close', 'delete'] as const
|
||||
|
||||
export type WorkspaceLifecycleKind = (typeof WORKSPACE_LIFECYCLE_KINDS)[number]
|
||||
|
||||
/** A lifecycle command main accepted for a workspace: a hint only, never saved, gone on restart. */
|
||||
const WorkspaceLifecycleHintSchema = z.object({
|
||||
kind: z.enum(WORKSPACE_LIFECYCLE_KINDS),
|
||||
/** Commands accepted after this one and still waiting. */
|
||||
queuedBehind: z.number(),
|
||||
/** The last command's outcome went unconfirmed by its host; the queue waits for proof. */
|
||||
paused: z.literal(true).optional()
|
||||
})
|
||||
|
||||
/** Keyed by the workspace's host-qualified key (`hostId|id`). */
|
||||
export const WorkspaceLifecycleHintsSchema = z.record(z.string(), WorkspaceLifecycleHintSchema)
|
||||
|
||||
export type WorkspaceLifecycleHint = z.infer<typeof WorkspaceLifecycleHintSchema>
|
||||
Reference in New Issue
Block a user