mirror of
https://github.com/stablyai/orca.git
synced 2026-10-24 00:04:01 +00:00
fix(session): defer session writes suppressed by a direct-SSH apply
The debounced session writer cleared its pending changed-field set whenever the persist gate was shut, after it had already advanced its identity-based detection baseline. Because detection is `prev[key] !== next[key]`, a field dropped there could never be re-detected, so a mutation made during an SSH apply (up to 30s plus a 1s tail, on every connect and reconnect) was lost for good — including a closed-tab tombstone, which is exactly the state that stops the host from resurrecting the tab. Keep the pending set instead, and add a one-shot wake-up from the gate owner so a deferred write is not stranded when the suppression tail expires with no store update behind it.
This commit is contained in:
@@ -1,7 +1,10 @@
|
||||
import { useEffect } from 'react'
|
||||
import { translate } from '@/i18n/i18n'
|
||||
import { useAppStore } from '../store'
|
||||
import { isDirectSshRemoteWorkspaceApplyInProgress } from '../hooks/remote-workspace-snapshot-apply'
|
||||
import {
|
||||
isDirectSshRemoteWorkspaceApplyInProgress,
|
||||
onDirectSshRemoteWorkspaceApplyWindowClosed
|
||||
} from '../hooks/remote-workspace-snapshot-apply'
|
||||
import { createSessionWriteSubscriber } from '../lib/session-write-subscriber'
|
||||
import { buildActiveViewUnloadPatch } from '../lib/active-view-persist'
|
||||
import {
|
||||
@@ -84,6 +87,7 @@ export function useAppSessionPersistence(): void {
|
||||
return createSessionWriteSubscriber({
|
||||
store: useAppStore,
|
||||
shouldSchedulePersist: () => !isDirectSshRemoteWorkspaceApplyInProgress(),
|
||||
subscribeToPersistGateOpen: onDirectSshRemoteWorkspaceApplyWindowClosed,
|
||||
persist: ({ patch }) => {
|
||||
const state = useAppStore.getState()
|
||||
// Why: route each host's worktree-scoped slice to its own partition; return the local write so the remote-workspace upload chain below keeps its ordering.
|
||||
|
||||
+251
@@ -0,0 +1,251 @@
|
||||
/**
|
||||
* Drives the real apply path with the real session writer wired exactly as the app wires it.
|
||||
*
|
||||
* The write gate closes for the whole apply plus a 1s suppression tail that expires on a wall
|
||||
* clock — no store update follows it. So a tab the user closes on an unrelated target during
|
||||
* target A's reconnect has to be held and written when the tail ends, or the close and its
|
||||
* tombstone are lost until the next unrelated mutation (a crash or kill in between loses both).
|
||||
*/
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type * as AgentStatusModule from '@/lib/agent-status'
|
||||
import type { RemoteWorkspaceSnapshot } from '../../../shared/remote-workspace-types'
|
||||
import type { DirectSshAuthority, SshProviderEpoch } from '../../../shared/ssh-types'
|
||||
import { createTestStore, makeWorktree } from '../store/slices/store-test-helpers'
|
||||
import {
|
||||
createSessionWriteSubscriber,
|
||||
type WorkspaceSessionWrite
|
||||
} from '../lib/session-write-subscriber'
|
||||
import {
|
||||
applyDirectSshRemoteWorkspaceSnapshot,
|
||||
isDirectSshRemoteWorkspaceApplyInProgress,
|
||||
onDirectSshRemoteWorkspaceApplyWindowClosed
|
||||
} from './remote-workspace-snapshot-apply'
|
||||
import type { DirectSshSnapshotApplyToken } from './direct-ssh-reconnect-coordinator-types'
|
||||
|
||||
vi.mock('sonner', () => ({
|
||||
toast: { info: vi.fn(), success: vi.fn(), error: vi.fn() }
|
||||
}))
|
||||
vi.mock('@/lib/agent-status', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof AgentStatusModule>()
|
||||
return {
|
||||
...actual,
|
||||
detectAgentStatusFromTitle: vi.fn().mockReturnValue(null)
|
||||
}
|
||||
})
|
||||
|
||||
const TARGET_A = 'ssh-target-a'
|
||||
const PATH_A = '/srv/proj/target-a'
|
||||
const WORKTREE_A = `repoA::${PATH_A}`
|
||||
const WORKTREE_B = 'repoB::/srv/other/target-b'
|
||||
|
||||
const authority: DirectSshAuthority = {
|
||||
targetId: TARGET_A,
|
||||
providerEpoch: 'provider-epoch-1' as SshProviderEpoch,
|
||||
connectionGeneration: 1
|
||||
}
|
||||
|
||||
type TestStore = ReturnType<typeof createTestStore>
|
||||
|
||||
function token(snapshotRevision: number): DirectSshSnapshotApplyToken {
|
||||
return {
|
||||
authority,
|
||||
catalogRevision: 0,
|
||||
repoFingerprint: 'fp',
|
||||
authorityRequirement: 'required',
|
||||
snapshotRevision,
|
||||
outcome: 'complete'
|
||||
}
|
||||
}
|
||||
|
||||
function snapshot(revision: number): RemoteWorkspaceSnapshot {
|
||||
return {
|
||||
namespace: 'workspace',
|
||||
revision,
|
||||
updatedAt: revision,
|
||||
schemaVersion: 1,
|
||||
session: {
|
||||
activeWorktreePath: PATH_A,
|
||||
activeTabId: 'tab-a',
|
||||
tabsByWorktreePath: {
|
||||
[PATH_A]: [
|
||||
{
|
||||
id: 'tab-a',
|
||||
worktreePath: PATH_A,
|
||||
ptyId: 'pty-tab-a',
|
||||
title: 'Terminal 1',
|
||||
customTitle: null,
|
||||
color: null,
|
||||
sortOrder: 0,
|
||||
createdAt: 1
|
||||
}
|
||||
]
|
||||
},
|
||||
terminalLayoutsByTabId: {},
|
||||
activeWorktreePathsOnShutdown: [],
|
||||
activeTabIdByWorktreePath: { [PATH_A]: 'tab-a' },
|
||||
remoteSessionIdsByTabId: { 'tab-a': 'pty-tab-a' },
|
||||
lastVisitedAtByWorktreePath: { [PATH_A]: revision },
|
||||
defaultTerminalTabsAppliedByWorktreePath: { [PATH_A]: true }
|
||||
}
|
||||
} satisfies RemoteWorkspaceSnapshot
|
||||
}
|
||||
|
||||
function seedCatalog(store: TestStore): void {
|
||||
store.setState({
|
||||
worktreesByRepo: {
|
||||
repoA: [
|
||||
makeWorktree({
|
||||
id: WORKTREE_A,
|
||||
repoId: 'repoA',
|
||||
path: PATH_A,
|
||||
hostId: `ssh:${TARGET_A}`
|
||||
} as never)
|
||||
]
|
||||
},
|
||||
repos: [
|
||||
{
|
||||
id: 'repoA',
|
||||
path: '/srv/proj',
|
||||
displayName: 'Proj',
|
||||
badgeColor: '#000',
|
||||
addedAt: 0,
|
||||
connectionId: TARGET_A
|
||||
} as never
|
||||
],
|
||||
reconnectPersistedTerminals: (async () => {}) as never,
|
||||
markRemoteWorkspaceHydrated: (() => {}) as never,
|
||||
setRemoteWorkspaceSyncStatus: (() => {}) as never
|
||||
})
|
||||
}
|
||||
|
||||
/** A second target's tab the user closes while target A is mid-apply. */
|
||||
function closeUnrelatedTargetTab(store: TestStore): void {
|
||||
store.setState({
|
||||
tabsByWorktree: { ...store.getState().tabsByWorktree, [WORKTREE_B]: [] },
|
||||
closedTerminalTabTombstonesByTabId: {
|
||||
'tab-b': { closedAt: Date.now(), worktreeId: WORKTREE_B }
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/** Everything here is timer-driven, so fake timers make the tail deterministic and instant. */
|
||||
describe('session writes deferred by a direct-SSH apply', () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
async function applyTargetA(store: TestStore, onApplied: () => void): Promise<void> {
|
||||
await applyDirectSshRemoteWorkspaceSnapshot({
|
||||
store,
|
||||
snapshot: snapshot(1),
|
||||
token: token(1),
|
||||
arrival: 1,
|
||||
isArrivalCurrent: () => true,
|
||||
isPreparationTokenCurrent: () => true,
|
||||
waitForWorkspaceSessionReady: async () => true,
|
||||
// Runs inside the apply, which is when the write gate is shut.
|
||||
finalizeHydratedTerminals: () => {
|
||||
onApplied()
|
||||
return 0
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
function seedBothTargets(store: TestStore): void {
|
||||
seedCatalog(store)
|
||||
store.setState({
|
||||
tabsByWorktree: {
|
||||
[WORKTREE_B]: [
|
||||
{
|
||||
id: 'tab-b',
|
||||
ptyId: 'pty-tab-b',
|
||||
worktreeId: WORKTREE_B,
|
||||
title: 'Other host',
|
||||
customTitle: null,
|
||||
color: null,
|
||||
sortOrder: 0,
|
||||
createdAt: 1
|
||||
} as never
|
||||
]
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
it('lands an unrelated target\u2019s tab close once the suppression tail expires', async () => {
|
||||
const store = createTestStore()
|
||||
seedBothTargets(store)
|
||||
|
||||
const persist = vi.fn<(payload: WorkspaceSessionWrite) => void>()
|
||||
const cleanup = createSessionWriteSubscriber({
|
||||
store,
|
||||
persist,
|
||||
shouldSchedulePersist: () => !isDirectSshRemoteWorkspaceApplyInProgress(),
|
||||
subscribeToPersistGateOpen: onDirectSshRemoteWorkspaceApplyWindowClosed
|
||||
})
|
||||
try {
|
||||
store.setState({ workspaceSessionReady: true, hydrationSucceeded: true })
|
||||
vi.advanceTimersByTime(200)
|
||||
expect(persist).toHaveBeenCalled()
|
||||
persist.mockClear()
|
||||
|
||||
await applyTargetA(store, () => closeUnrelatedTargetTab(store))
|
||||
|
||||
expect(
|
||||
isDirectSshRemoteWorkspaceApplyInProgress(),
|
||||
'the suppression tail should still be holding writes here'
|
||||
).toBe(true)
|
||||
expect(persist).not.toHaveBeenCalled()
|
||||
|
||||
// Tail (1000ms) + the notice's 1ms margin + the writer's 150ms debounce.
|
||||
vi.advanceTimersByTime(1_200)
|
||||
|
||||
expect(persist, 'the close made during the apply never reached disk').toHaveBeenCalledTimes(1)
|
||||
const patch = persist.mock.calls[0][0].patch
|
||||
expect(patch.closedTerminalTabTombstonesByTabId?.['tab-b']?.worktreeId).toBe(WORKTREE_B)
|
||||
expect(patch.tabsByWorktree?.[WORKTREE_B]).toEqual([])
|
||||
expect(patch.tabsByWorktree?.[WORKTREE_A]).toHaveLength(1)
|
||||
} finally {
|
||||
cleanup()
|
||||
}
|
||||
})
|
||||
|
||||
it('still wakes the deferred write when the wall clock steps back during the tail', async () => {
|
||||
const store = createTestStore()
|
||||
seedBothTargets(store)
|
||||
const startedAt = Date.now()
|
||||
|
||||
const persist = vi.fn<(payload: WorkspaceSessionWrite) => void>()
|
||||
const cleanup = createSessionWriteSubscriber({
|
||||
store,
|
||||
persist,
|
||||
shouldSchedulePersist: () => !isDirectSshRemoteWorkspaceApplyInProgress(),
|
||||
subscribeToPersistGateOpen: onDirectSshRemoteWorkspaceApplyWindowClosed
|
||||
})
|
||||
try {
|
||||
store.setState({ workspaceSessionReady: true, hydrationSucceeded: true })
|
||||
vi.advanceTimersByTime(200)
|
||||
persist.mockClear()
|
||||
|
||||
await applyTargetA(store, () => closeUnrelatedTargetTab(store))
|
||||
|
||||
// The notice's delay is wall-clock arithmetic on a monotonic timer, so an NTP step back
|
||||
// leaves the deadline in the future when it fires. It must wait, not give up.
|
||||
vi.setSystemTime(startedAt - 5_000)
|
||||
vi.advanceTimersByTime(1_200)
|
||||
expect(persist, 'the notice fired before its own deadline').not.toHaveBeenCalled()
|
||||
|
||||
vi.advanceTimersByTime(5_200)
|
||||
|
||||
expect(persist, 'a clock step back stranded the deferred write').toHaveBeenCalledTimes(1)
|
||||
expect(
|
||||
persist.mock.calls[0][0].patch.closedTerminalTabTombstonesByTabId?.['tab-b']
|
||||
).toBeDefined()
|
||||
} finally {
|
||||
cleanup()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -27,6 +27,45 @@ export function isDirectSshRemoteWorkspaceApplyInProgress(): boolean {
|
||||
return snapshotApplyDepth > 0 || Date.now() < snapshotWriteSuppressUntil
|
||||
}
|
||||
|
||||
const applyWindowCloseListeners = new Set<() => void>()
|
||||
|
||||
/**
|
||||
* Fires once after the write-suppression tail expires. Why: the tail ends on a wall clock with no
|
||||
* store update behind it, so a session write deferred during an apply has nothing else to wake it.
|
||||
*/
|
||||
export function onDirectSshRemoteWorkspaceApplyWindowClosed(listener: () => void): () => void {
|
||||
applyWindowCloseListeners.add(listener)
|
||||
return () => {
|
||||
applyWindowCloseListeners.delete(listener)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* One timer in flight per apply. It never polls: the tail deadline only moves forward, and each
|
||||
* re-arm waits exactly the time still left on it, so the wait converges instead of spinning.
|
||||
*/
|
||||
function scheduleApplyWindowClosedNotice(): void {
|
||||
// +1ms because the gate compares `Date.now() < suppressUntil`; fire strictly past the deadline.
|
||||
const delayMs = Math.max(0, snapshotWriteSuppressUntil - Date.now()) + 1
|
||||
setTimeout(() => {
|
||||
// An apply still in flight schedules the next notice from its own `finally`.
|
||||
if (snapshotApplyDepth > 0) {
|
||||
return
|
||||
}
|
||||
// Why re-arm rather than return: `delayMs` is wall-clock arithmetic handed to a monotonic
|
||||
// timer, so a clock step back (NTP) can leave the deadline in the future when this fires.
|
||||
// Dropping the notice there would strand every write deferred in this window.
|
||||
if (Date.now() < snapshotWriteSuppressUntil) {
|
||||
scheduleApplyWindowClosedNotice()
|
||||
return
|
||||
}
|
||||
// Safe to iterate live: Set iteration tolerates a listener unsubscribing itself.
|
||||
for (const listener of applyWindowCloseListeners) {
|
||||
listener()
|
||||
}
|
||||
}, delayMs)
|
||||
}
|
||||
|
||||
type RemoteWorkspaceSnapshotApplyInput = {
|
||||
store: Pick<StoreApi<AppState>, 'getState'>
|
||||
snapshot: RemoteWorkspaceSnapshot
|
||||
@@ -195,5 +234,6 @@ export async function applyDirectSshRemoteWorkspaceSnapshot({
|
||||
} finally {
|
||||
snapshotWriteSuppressUntil = Date.now() + REMOTE_WORKSPACE_SNAPSHOT_WRITE_SUPPRESS_MS
|
||||
snapshotApplyDepth -= 1
|
||||
scheduleApplyWindowClosedNotice()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,299 @@
|
||||
/**
|
||||
* A session mutation made while the direct-SSH write gate is closed must be deferred, never
|
||||
* dropped.
|
||||
*
|
||||
* Why this is data loss and not a delay: the subscriber advances its change-detection baseline
|
||||
* (`prev = next`) before it consults the gate, and detection is identity-based
|
||||
* (`prev[key] !== next[key]`). A field cleared out of the pending set at that point can never be
|
||||
* re-detected, so the mutation is gone until something else happens to touch the same field. The
|
||||
* gate is closed for up to SNAPSHOT_TERMINAL_RECONNECT_TIMEOUT_MS plus a 1s tail on every SSH
|
||||
* connect and reconnect — and a tab closed in that window loses both the removal and the tombstone
|
||||
* that was supposed to stop the host from resurrecting it.
|
||||
*/
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { useAppStore, type AppState } from '@/store'
|
||||
import {
|
||||
createSessionWriteSubscriber,
|
||||
type WorkspaceSessionWrite
|
||||
} from './session-write-subscriber'
|
||||
|
||||
let initialState: AppState
|
||||
|
||||
function terminalTab(id: string, worktreeId: string): AppState['tabsByWorktree'][string][number] {
|
||||
return {
|
||||
id,
|
||||
ptyId: `pty-${id}`,
|
||||
worktreeId,
|
||||
title: id,
|
||||
defaultTitle: id,
|
||||
customTitle: null,
|
||||
color: null,
|
||||
sortOrder: 0,
|
||||
createdAt: 1
|
||||
}
|
||||
}
|
||||
|
||||
/** Stands in for the direct-SSH apply gate: `isOpen` is the flag, and the wake-up fires on demand
|
||||
* the way the apply's suppression tail fires it. */
|
||||
function createTestPersistGate(isOpen: () => boolean) {
|
||||
const listeners: (() => void)[] = []
|
||||
return {
|
||||
deps: {
|
||||
shouldSchedulePersist: isOpen,
|
||||
subscribeToPersistGateOpen: (onGateOpen: () => void) => {
|
||||
listeners.push(onGateOpen)
|
||||
return () => {
|
||||
listeners.length = 0
|
||||
}
|
||||
}
|
||||
},
|
||||
listenerCount: (): number => listeners.length,
|
||||
notifyGateOpen: (): void => {
|
||||
for (const listener of listeners) {
|
||||
listener()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
describe('session write subscriber defers writes across a closed persistence gate', () => {
|
||||
beforeEach(() => {
|
||||
initialState = useAppStore.getState()
|
||||
vi.useFakeTimers()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
useAppStore.setState(initialState, true)
|
||||
})
|
||||
|
||||
it('writes a mutation made while the gate was closed once a later mutation reopens it', () => {
|
||||
const persist = vi.fn<(payload: WorkspaceSessionWrite) => void>()
|
||||
let gateOpen = true
|
||||
const cleanup = createSessionWriteSubscriber({
|
||||
store: useAppStore,
|
||||
persist,
|
||||
...createTestPersistGate(() => gateOpen).deps
|
||||
})
|
||||
|
||||
useAppStore.setState({
|
||||
workspaceSessionReady: true,
|
||||
hydrationSucceeded: true
|
||||
})
|
||||
vi.advanceTimersByTime(200)
|
||||
persist.mockClear()
|
||||
|
||||
// The user closes an SSH tab while a snapshot apply is running.
|
||||
gateOpen = false
|
||||
useAppStore.setState({
|
||||
activeTabId: 'still-open-tab',
|
||||
closedTerminalTabTombstonesByTabId: {
|
||||
'closed-during-apply': {
|
||||
closedAt: Date.now(),
|
||||
worktreeId: 'wt-remote'
|
||||
}
|
||||
}
|
||||
})
|
||||
vi.advanceTimersByTime(200)
|
||||
expect(persist).not.toHaveBeenCalled()
|
||||
|
||||
// The apply finishes and an unrelated field changes. The deferred close must ride along.
|
||||
gateOpen = true
|
||||
useAppStore.setState({ activeRepoId: 'repo-1' })
|
||||
vi.advanceTimersByTime(200)
|
||||
|
||||
expect(persist).toHaveBeenCalledTimes(1)
|
||||
const patch = persist.mock.calls[0][0].patch
|
||||
expect(
|
||||
patch.closedTerminalTabTombstonesByTabId,
|
||||
'the tombstone written during the apply window was discarded, not deferred'
|
||||
).toEqual({
|
||||
'closed-during-apply': {
|
||||
closedAt: expect.any(Number),
|
||||
worktreeId: 'wt-remote'
|
||||
}
|
||||
})
|
||||
expect(patch.activeTabId).toBe('still-open-tab')
|
||||
expect(patch.activeRepoId).toBe('repo-1')
|
||||
cleanup()
|
||||
})
|
||||
|
||||
it('writes a mutation whose debounce expired inside the apply window', () => {
|
||||
const persist = vi.fn<(payload: WorkspaceSessionWrite) => void>()
|
||||
let gateOpen = true
|
||||
const cleanup = createSessionWriteSubscriber({
|
||||
store: useAppStore,
|
||||
persist,
|
||||
...createTestPersistGate(() => gateOpen).deps
|
||||
})
|
||||
|
||||
useAppStore.setState({
|
||||
workspaceSessionReady: true,
|
||||
hydrationSucceeded: true
|
||||
})
|
||||
vi.advanceTimersByTime(200)
|
||||
persist.mockClear()
|
||||
|
||||
// Mutation lands with the gate open, but the apply starts before the debounce expires.
|
||||
useAppStore.setState({ activeTabId: 'closed-mid-debounce' })
|
||||
vi.advanceTimersByTime(50)
|
||||
gateOpen = false
|
||||
vi.advanceTimersByTime(200)
|
||||
expect(persist).not.toHaveBeenCalled()
|
||||
|
||||
gateOpen = true
|
||||
useAppStore.setState({ activeRepoId: 'repo-2' })
|
||||
vi.advanceTimersByTime(200)
|
||||
|
||||
expect(persist).toHaveBeenCalledTimes(1)
|
||||
expect(
|
||||
persist.mock.calls[0][0].patch.activeTabId,
|
||||
'the debounce fired inside the apply window and threw the mutation away'
|
||||
).toBe('closed-mid-debounce')
|
||||
cleanup()
|
||||
})
|
||||
|
||||
it('keeps an unrelated target’s tab close across another target’s apply', () => {
|
||||
const persist = vi.fn<(payload: WorkspaceSessionWrite) => void>()
|
||||
let gateOpen = true
|
||||
const cleanup = createSessionWriteSubscriber({
|
||||
store: useAppStore,
|
||||
persist,
|
||||
...createTestPersistGate(() => gateOpen).deps
|
||||
})
|
||||
|
||||
useAppStore.setState({
|
||||
workspaceSessionReady: true,
|
||||
hydrationSucceeded: true,
|
||||
tabsByWorktree: {
|
||||
'wt-target-a': [terminalTab('tab-a', 'wt-target-a')],
|
||||
'wt-target-b': [terminalTab('tab-b', 'wt-target-b')]
|
||||
}
|
||||
})
|
||||
vi.advanceTimersByTime(200)
|
||||
persist.mockClear()
|
||||
|
||||
// The gate is global: target A's apply suppresses writes for every other target too.
|
||||
gateOpen = false
|
||||
useAppStore.setState({
|
||||
tabsByWorktree: {
|
||||
'wt-target-a': [terminalTab('tab-a', 'wt-target-a')],
|
||||
'wt-target-b': []
|
||||
},
|
||||
closedTerminalTabTombstonesByTabId: {
|
||||
'tab-b': { closedAt: Date.now(), worktreeId: 'wt-target-b' }
|
||||
}
|
||||
})
|
||||
vi.advanceTimersByTime(200)
|
||||
expect(persist).not.toHaveBeenCalled()
|
||||
|
||||
gateOpen = true
|
||||
useAppStore.setState({ activeRepoId: 'repo-3' })
|
||||
vi.advanceTimersByTime(200)
|
||||
|
||||
expect(persist).toHaveBeenCalledTimes(1)
|
||||
const patch = persist.mock.calls[0][0].patch
|
||||
expect(
|
||||
patch.tabsByWorktree?.['wt-target-b'],
|
||||
'target B lost a tab close because target A was mid-apply'
|
||||
).toEqual([])
|
||||
expect(patch.tabsByWorktree?.['wt-target-a']).toHaveLength(1)
|
||||
expect(patch.closedTerminalTabTombstonesByTabId?.['tab-b']?.worktreeId).toBe('wt-target-b')
|
||||
cleanup()
|
||||
})
|
||||
|
||||
it('writes on the gate-open wake-up when no further store update follows', () => {
|
||||
const persist = vi.fn<(payload: WorkspaceSessionWrite) => void>()
|
||||
let gateOpen = true
|
||||
const gate = createTestPersistGate(() => gateOpen)
|
||||
const cleanup = createSessionWriteSubscriber({
|
||||
store: useAppStore,
|
||||
persist,
|
||||
...gate.deps
|
||||
})
|
||||
|
||||
useAppStore.setState({
|
||||
workspaceSessionReady: true,
|
||||
hydrationSucceeded: true
|
||||
})
|
||||
vi.advanceTimersByTime(200)
|
||||
persist.mockClear()
|
||||
|
||||
gateOpen = false
|
||||
useAppStore.setState({ activeTabId: 'closed-during-apply' })
|
||||
vi.advanceTimersByTime(200)
|
||||
expect(persist).not.toHaveBeenCalled()
|
||||
|
||||
// The suppression tail expires on a wall clock, so nothing touches the store afterwards.
|
||||
gateOpen = true
|
||||
expect(gate.listenerCount(), 'the subscriber never registered for the gate-open wake-up').toBe(
|
||||
1
|
||||
)
|
||||
gate.notifyGateOpen()
|
||||
vi.advanceTimersByTime(200)
|
||||
|
||||
expect(persist).toHaveBeenCalledTimes(1)
|
||||
expect(persist.mock.calls[0][0].patch.activeTabId).toBe('closed-during-apply')
|
||||
cleanup()
|
||||
})
|
||||
|
||||
it('arms no timer while the gate stays closed', () => {
|
||||
const persist = vi.fn<(payload: WorkspaceSessionWrite) => void>()
|
||||
let gateOpen = true
|
||||
const cleanup = createSessionWriteSubscriber({
|
||||
store: useAppStore,
|
||||
persist,
|
||||
...createTestPersistGate(() => gateOpen).deps
|
||||
})
|
||||
|
||||
useAppStore.setState({
|
||||
workspaceSessionReady: true,
|
||||
hydrationSucceeded: true
|
||||
})
|
||||
vi.advanceTimersByTime(200)
|
||||
persist.mockClear()
|
||||
|
||||
gateOpen = false
|
||||
useAppStore.setState({ activeTabId: 'closed-during-apply' })
|
||||
vi.advanceTimersByTime(10_000)
|
||||
|
||||
expect(persist).not.toHaveBeenCalled()
|
||||
// A gate that never reopens must not turn deferral into a polling loop.
|
||||
expect(vi.getTimerCount(), 'the deferred write is polling on a re-armed timer').toBe(0)
|
||||
cleanup()
|
||||
})
|
||||
|
||||
it('does not let an unrelated update reset a debounce armed by the gate-open wake-up', () => {
|
||||
const persist = vi.fn<(payload: WorkspaceSessionWrite) => void>()
|
||||
let gateOpen = true
|
||||
const gate = createTestPersistGate(() => gateOpen)
|
||||
const cleanup = createSessionWriteSubscriber({
|
||||
store: useAppStore,
|
||||
persist,
|
||||
...gate.deps
|
||||
})
|
||||
|
||||
useAppStore.setState({ workspaceSessionReady: true, hydrationSucceeded: true })
|
||||
vi.advanceTimersByTime(200)
|
||||
persist.mockClear()
|
||||
|
||||
gateOpen = false
|
||||
useAppStore.setState({ activeTabId: 'closed-during-apply' })
|
||||
vi.advanceTimersByTime(200)
|
||||
expect(persist).not.toHaveBeenCalled()
|
||||
|
||||
// The wake-up arms the debounce; agent-status and usage ticks keep arriving while it runs.
|
||||
gateOpen = true
|
||||
gate.notifyGateOpen()
|
||||
vi.advanceTimersByTime(100)
|
||||
useAppStore.getState().setCacheTimerStartedAt('tab-1:pane-1', Date.now())
|
||||
vi.advanceTimersByTime(60)
|
||||
|
||||
expect(
|
||||
persist,
|
||||
'an unrelated update pushed the deferred write past its debounce'
|
||||
).toHaveBeenCalledTimes(1)
|
||||
expect(persist.mock.calls[0][0].patch.activeTabId).toBe('closed-during-apply')
|
||||
cleanup()
|
||||
})
|
||||
})
|
||||
@@ -62,6 +62,9 @@ function makeTerminalSessionState(title: string, label = title): Partial<AppStat
|
||||
}
|
||||
}
|
||||
|
||||
/** These cases reopen the gate through a store update, so they need no wake-up of their own. */
|
||||
const noGateOpenWakeUp = (): (() => void) => () => {}
|
||||
|
||||
describe('createSessionWriteSubscriber', () => {
|
||||
beforeEach(() => {
|
||||
initialState = useAppStore.getState()
|
||||
@@ -571,13 +574,14 @@ describe('createSessionWriteSubscriber', () => {
|
||||
cleanup()
|
||||
})
|
||||
|
||||
it('updates its baseline without scheduling when shouldSchedulePersist returns false', () => {
|
||||
it('defers rather than drops a change made while shouldSchedulePersist returns false', () => {
|
||||
const persist = vi.fn<(payload: WorkspaceSessionWrite) => void>()
|
||||
let shouldSchedule = false
|
||||
const cleanup = createSessionWriteSubscriber({
|
||||
store: useAppStore,
|
||||
persist,
|
||||
shouldSchedulePersist: () => shouldSchedule
|
||||
shouldSchedulePersist: () => shouldSchedule,
|
||||
subscribeToPersistGateOpen: noGateOpenWakeUp
|
||||
})
|
||||
|
||||
useAppStore.setState({ workspaceSessionReady: true, hydrationSucceeded: true })
|
||||
@@ -588,25 +592,34 @@ describe('createSessionWriteSubscriber', () => {
|
||||
useAppStore.setState({ activeTabId: 'tab-1' })
|
||||
vi.advanceTimersByTime(200)
|
||||
expect(persist).toHaveBeenCalledTimes(1)
|
||||
// The suppressed first pass owed every field; only activeTabId changed after the gate opened.
|
||||
expect(persist.mock.calls[0][0].patch.tabsByWorktree).toBeDefined()
|
||||
cleanup()
|
||||
})
|
||||
|
||||
it('cancels a pending debounce when shouldSchedulePersist returns false', () => {
|
||||
it('holds a write whose change arrived while shouldSchedulePersist returned false', () => {
|
||||
const persist = vi.fn<(payload: WorkspaceSessionWrite) => void>()
|
||||
let shouldSchedule = true
|
||||
const cleanup = createSessionWriteSubscriber({
|
||||
store: useAppStore,
|
||||
persist,
|
||||
shouldSchedulePersist: () => shouldSchedule
|
||||
shouldSchedulePersist: () => shouldSchedule,
|
||||
subscribeToPersistGateOpen: noGateOpenWakeUp
|
||||
})
|
||||
|
||||
useAppStore.setState({ workspaceSessionReady: true })
|
||||
useAppStore.setState({ workspaceSessionReady: true, hydrationSucceeded: true })
|
||||
vi.advanceTimersByTime(50)
|
||||
shouldSchedule = false
|
||||
useAppStore.setState({ activeTabId: 'remote-tab' })
|
||||
vi.advanceTimersByTime(200)
|
||||
|
||||
expect(persist).not.toHaveBeenCalled()
|
||||
|
||||
shouldSchedule = true
|
||||
useAppStore.setState({ activeRepoId: 'repo-after-remote-pull' })
|
||||
vi.advanceTimersByTime(200)
|
||||
expect(persist).toHaveBeenCalledTimes(1)
|
||||
expect(persist.mock.calls[0][0].patch.activeTabId).toBe('remote-tab')
|
||||
cleanup()
|
||||
})
|
||||
|
||||
@@ -616,7 +629,8 @@ describe('createSessionWriteSubscriber', () => {
|
||||
const cleanup = createSessionWriteSubscriber({
|
||||
store: useAppStore,
|
||||
persist,
|
||||
shouldSchedulePersist: () => shouldSchedule
|
||||
shouldSchedulePersist: () => shouldSchedule,
|
||||
subscribeToPersistGateOpen: noGateOpenWakeUp
|
||||
})
|
||||
|
||||
useAppStore.setState({ workspaceSessionReady: true, hydrationSucceeded: true })
|
||||
@@ -629,6 +643,13 @@ describe('createSessionWriteSubscriber', () => {
|
||||
vi.advanceTimersByTime(200)
|
||||
|
||||
expect(persist).not.toHaveBeenCalled()
|
||||
|
||||
// The suppressed write is owed, not forgotten: it lands when the gate reopens.
|
||||
shouldSchedule = true
|
||||
useAppStore.setState({ activeTabId: 'tab-after-remote-pull' })
|
||||
vi.advanceTimersByTime(200)
|
||||
expect(persist).toHaveBeenCalledTimes(1)
|
||||
expect(persist.mock.calls[0][0].patch.activeWorktreeId).toBe('wt-before-remote-pull')
|
||||
cleanup()
|
||||
})
|
||||
|
||||
|
||||
@@ -77,15 +77,30 @@ export type WorkspaceSessionWrite = {
|
||||
patch: WorkspaceSessionPatch
|
||||
}
|
||||
|
||||
/**
|
||||
* Why the two are one unit and not two optional callbacks: a gate can reopen on a wall clock (the
|
||||
* direct-SSH apply's suppression tail) with no store update behind it. A caller that supplies the
|
||||
* gate but no wake-up would strand every write deferred in that window until some unrelated
|
||||
* mutation happened along — the failure this deferral exists to prevent. Pairing them in the type
|
||||
* makes that combination unwriteable rather than merely discouraged.
|
||||
*/
|
||||
type SessionWritePersistGate =
|
||||
| { shouldSchedulePersist?: undefined; subscribeToPersistGateOpen?: undefined }
|
||||
| {
|
||||
/** False defers the write; it never drops it. */
|
||||
shouldSchedulePersist: () => boolean
|
||||
/** Called when the gate may have reopened. Returns an unsubscribe. */
|
||||
subscribeToPersistGateOpen: (onGateOpen: () => void) => () => void
|
||||
}
|
||||
|
||||
export type SessionWriteSubscriberDeps = {
|
||||
store: {
|
||||
subscribe: (listener: (state: AppState) => void) => () => void
|
||||
getState: () => AppState
|
||||
}
|
||||
persist: (payload: WorkspaceSessionWrite) => void
|
||||
shouldSchedulePersist?: () => boolean
|
||||
debounceMs?: number
|
||||
}
|
||||
} & SessionWritePersistGate
|
||||
|
||||
/**
|
||||
* Why: factored out so a vitest can drive the real Zustand store and assert
|
||||
@@ -97,6 +112,7 @@ export function createSessionWriteSubscriber({
|
||||
store,
|
||||
persist,
|
||||
shouldSchedulePersist,
|
||||
subscribeToPersistGateOpen,
|
||||
debounceMs = 150
|
||||
}: SessionWriteSubscriberDeps): () => void {
|
||||
let timer: ReturnType<typeof setTimeout> | null = null
|
||||
@@ -108,10 +124,50 @@ export function createSessionWriteSubscriber({
|
||||
// reuse the prior identity while a real session change keeps fresh tabs for
|
||||
// the eventual getState() patch build. `null` makes the first fire proceed.
|
||||
let prev: Record<string, unknown> | null = null
|
||||
// Why: this set is the only record that a mutation still owes a write — `prev` has already
|
||||
// advanced past it, and change detection is identity-based, so a field dropped from here can
|
||||
// never be re-detected. It is retired only by a flush that reached `persist` (or found nothing
|
||||
// left to write), and by unsubscribe. A closed gate never retires it.
|
||||
const pendingChangedFields = new Set<SessionRelevantField>()
|
||||
const terminalTabsProjection = createTerminalSessionTabsProjection()
|
||||
const unifiedTabsProjection = createUnifiedSessionTabsProjection()
|
||||
|
||||
const flushPendingWrite = (): void => {
|
||||
timer = null
|
||||
// Why: rebuild from the freshest store state rather than the snapshot
|
||||
// captured when this timer was scheduled. Today this is equivalent
|
||||
// because buildWorkspaceSessionPayload reads only SESSION_RELEVANT_FIELDS
|
||||
// (the same fields gating the timer reset), so the captured `state` is
|
||||
// already current for those fields. Calling getState() guards against a
|
||||
// future refactor that adds a non-relevant field read to the payload
|
||||
// builder — without this, such a change would silently start emitting
|
||||
// stale values for that field.
|
||||
const fresh = store.getState()
|
||||
// Why: a closed gate defers, it never discards. Returning with the pending set intact leaves
|
||||
// the write owed; the next store update or gate-open wake-up re-arms it. Nothing re-arms from
|
||||
// here, so a gate that never reopens costs no timer.
|
||||
if (!shouldPersistWorkspaceSession(fresh)) {
|
||||
return
|
||||
}
|
||||
if (shouldSchedulePersist && !shouldSchedulePersist()) {
|
||||
return
|
||||
}
|
||||
const changed = new Set(pendingChangedFields)
|
||||
pendingChangedFields.clear()
|
||||
const patch = buildWorkspaceSessionPatch(fresh, changed)
|
||||
if (Object.keys(patch).length === 0) {
|
||||
return
|
||||
}
|
||||
persist({ patch })
|
||||
}
|
||||
|
||||
const armFlushTimer = (): void => {
|
||||
if (timer !== null) {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
timer = setTimeout(flushPendingWrite, debounceMs)
|
||||
}
|
||||
|
||||
const unsub = store.subscribe((state) => {
|
||||
if (!shouldPersistWorkspaceSession(state)) {
|
||||
return
|
||||
@@ -130,7 +186,7 @@ export function createSessionWriteSubscriber({
|
||||
prev === null
|
||||
? [...SESSION_RELEVANT_FIELDS]
|
||||
: SESSION_RELEVANT_FIELDS.filter((key) => prev?.[key] !== next[key])
|
||||
if (changedFields.length === 0) {
|
||||
if (changedFields.length === 0 && pendingChangedFields.size === 0) {
|
||||
return
|
||||
}
|
||||
prev = next
|
||||
@@ -138,47 +194,26 @@ export function createSessionWriteSubscriber({
|
||||
pendingChangedFields.add(field)
|
||||
}
|
||||
if (shouldSchedulePersist && !shouldSchedulePersist()) {
|
||||
if (timer !== null) {
|
||||
clearTimeout(timer)
|
||||
timer = null
|
||||
}
|
||||
pendingChangedFields.clear()
|
||||
return
|
||||
}
|
||||
if (timer !== null) {
|
||||
clearTimeout(timer)
|
||||
// Why: an unrelated update may wake a deferred write but must never reset an armed debounce —
|
||||
// that reset storm is exactly what the changed-field gate above exists to prevent.
|
||||
if (timer !== null && changedFields.length === 0) {
|
||||
return
|
||||
}
|
||||
timer = setTimeout(() => {
|
||||
timer = null
|
||||
// Why: rebuild from the freshest store state rather than the snapshot
|
||||
// captured when this timer was scheduled. Today this is equivalent
|
||||
// because buildWorkspaceSessionPayload reads only SESSION_RELEVANT_FIELDS
|
||||
// (the same fields gating the timer reset), so the captured `state` is
|
||||
// already current for those fields. Calling getState() guards against a
|
||||
// future refactor that adds a non-relevant field read to the payload
|
||||
// builder — without this, such a change would silently start emitting
|
||||
// stale values for that field.
|
||||
const fresh = store.getState()
|
||||
if (!shouldPersistWorkspaceSession(fresh)) {
|
||||
pendingChangedFields.clear()
|
||||
return
|
||||
}
|
||||
if (shouldSchedulePersist && !shouldSchedulePersist()) {
|
||||
pendingChangedFields.clear()
|
||||
return
|
||||
}
|
||||
const changed = new Set(pendingChangedFields)
|
||||
pendingChangedFields.clear()
|
||||
const patch = buildWorkspaceSessionPatch(fresh, changed)
|
||||
if (Object.keys(patch).length === 0) {
|
||||
return
|
||||
}
|
||||
persist({ patch })
|
||||
}, debounceMs)
|
||||
armFlushTimer()
|
||||
})
|
||||
|
||||
const unsubGateOpen = subscribeToPersistGateOpen?.(() => {
|
||||
if (pendingChangedFields.size === 0 || timer !== null) {
|
||||
return
|
||||
}
|
||||
armFlushTimer()
|
||||
})
|
||||
|
||||
return () => {
|
||||
unsub()
|
||||
unsubGateOpen?.()
|
||||
if (timer !== null) {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user