Improve perf hot paths
Deduplicate watcher cleanup, reduce redundant terminal capture/diagnostics, bound remote terminal snapshot replay memory, and suppress redundant git/upstream status work.
This commit is contained in:
parent
7cf3b168fd
commit
2e5c03c31e
|
|
@ -194,4 +194,37 @@ describe('registerFilesystemWatcherHandlers', () => {
|
|||
)
|
||||
expect(unwatchMock).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('registers one destroyed listener for many SSH worktree watches', async () => {
|
||||
const destroyedCallbacks: (() => void)[] = []
|
||||
const sender = {
|
||||
isDestroyed: () => false,
|
||||
send: vi.fn(),
|
||||
once: vi.fn((event: string, callback: () => void) => {
|
||||
if (event === 'destroyed') {
|
||||
destroyedCallbacks.push(callback)
|
||||
}
|
||||
}),
|
||||
id: 99
|
||||
}
|
||||
const unwatchMock = vi.fn()
|
||||
const watchMock = vi.fn().mockResolvedValue(unwatchMock)
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock })
|
||||
|
||||
for (let i = 0; i < 12; i += 1) {
|
||||
await handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: `/home/me/repo-${i}`, connectionId: 'conn-1' }
|
||||
)
|
||||
}
|
||||
|
||||
// Why: WebContents warns after 10 listeners. The cleanup work still covers
|
||||
// every remote watch by scanning the shared remote watcher registry.
|
||||
expect(sender.once).toHaveBeenCalledTimes(1)
|
||||
expect(destroyedCallbacks).toHaveLength(1)
|
||||
|
||||
destroyedCallbacks[0]()
|
||||
|
||||
expect(unwatchMock).toHaveBeenCalledTimes(12)
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -50,10 +50,9 @@ const watchedRoots = new Map<string, WatchedRoot>()
|
|||
// repeated "Failed to read changes" / "watchman not found" errors.
|
||||
const unwatchableRoots = new Set<string>()
|
||||
|
||||
// Why: the `destroyed` listener was previously registered per-root on the
|
||||
// same WebContents. With 11+ worktrees, this exceeded Node's default
|
||||
// MaxListeners of 10. Track which senders already have a single cleanup
|
||||
// listener so we register exactly once per sender.
|
||||
// Why: watcher cleanup is keyed to the renderer WebContents, not to a specific
|
||||
// watched root. One listener per sender avoids MaxListeners warnings when a
|
||||
// workspace has many local and SSH-backed worktrees open.
|
||||
const senderCleanupRegistered = new Set<number>()
|
||||
|
||||
// Why: on Windows, tearing down and recreating @parcel/watcher subscriptions
|
||||
|
|
@ -311,6 +310,41 @@ async function createWatcher(rootKey: string, rootPath: string): Promise<Watched
|
|||
|
||||
// ── Subscribe / Unsubscribe ──────────────────────────────────────────
|
||||
|
||||
function cleanupLocalWatchersForSender(senderId: number): void {
|
||||
for (const [key, watchedRoot] of watchedRoots) {
|
||||
if (watchedRoot.listeners.has(senderId)) {
|
||||
watchedRoot.listeners.delete(senderId)
|
||||
if (watchedRoot.listeners.size === 0) {
|
||||
// Cancel any pending grace-period teardown for this root.
|
||||
const pending = pendingTeardowns.get(key)
|
||||
if (pending) {
|
||||
clearTimeout(pending)
|
||||
pendingTeardowns.delete(key)
|
||||
}
|
||||
if (watchedRoot.batch.timer) {
|
||||
clearTimeout(watchedRoot.batch.timer)
|
||||
}
|
||||
void watchedRoot.subscription.unsubscribe().catch((err: unknown) => {
|
||||
console.error(`[filesystem-watcher] unsubscribe error for ${key}:`, err)
|
||||
})
|
||||
watchedRoots.delete(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function registerSenderCleanup(sender: WebContents): void {
|
||||
if (senderCleanupRegistered.has(sender.id)) {
|
||||
return
|
||||
}
|
||||
senderCleanupRegistered.add(sender.id)
|
||||
sender.once('destroyed', () => {
|
||||
senderCleanupRegistered.delete(sender.id)
|
||||
cleanupLocalWatchersForSender(sender.id)
|
||||
cleanupRemoteWatchersForSender(sender.id)
|
||||
})
|
||||
}
|
||||
|
||||
async function subscribe(worktreePath: string, sender: WebContents): Promise<void> {
|
||||
const rootKey = normalizeRootPath(worktreePath)
|
||||
|
||||
|
|
@ -364,38 +398,7 @@ async function subscribe(worktreePath: string, sender: WebContents): Promise<voi
|
|||
}
|
||||
|
||||
root.listeners.set(sender.id, sender)
|
||||
|
||||
// Why: register a single `destroyed` listener per sender (not per-root).
|
||||
// The old code registered one listener per root, so 11+ worktrees would
|
||||
// exceed Node's default MaxListeners of 10 on the same WebContents. A
|
||||
// single listener that iterates all roots avoids the warning and is
|
||||
// equivalent — `destroyed` fires once when the renderer process exits.
|
||||
if (!senderCleanupRegistered.has(sender.id)) {
|
||||
senderCleanupRegistered.add(sender.id)
|
||||
sender.once('destroyed', () => {
|
||||
senderCleanupRegistered.delete(sender.id)
|
||||
for (const [key, watchedRoot] of watchedRoots) {
|
||||
if (watchedRoot.listeners.has(sender.id)) {
|
||||
watchedRoot.listeners.delete(sender.id)
|
||||
if (watchedRoot.listeners.size === 0) {
|
||||
// Cancel any pending grace-period teardown for this root.
|
||||
const pending = pendingTeardowns.get(key)
|
||||
if (pending) {
|
||||
clearTimeout(pending)
|
||||
pendingTeardowns.delete(key)
|
||||
}
|
||||
if (watchedRoot.batch.timer) {
|
||||
clearTimeout(watchedRoot.batch.timer)
|
||||
}
|
||||
void watchedRoot.subscription.unsubscribe().catch((err: unknown) => {
|
||||
console.error(`[filesystem-watcher] unsubscribe error for ${key}:`, err)
|
||||
})
|
||||
watchedRoots.delete(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
registerSenderCleanup(sender)
|
||||
}
|
||||
|
||||
function unsubscribe(worktreePath: string, senderId: number): void {
|
||||
|
|
@ -455,9 +458,7 @@ function addRemoteWatchListener(key: string, sender: WebContents): void {
|
|||
return
|
||||
}
|
||||
state.listeners.set(sender.id, sender)
|
||||
sender.once('destroyed', () => {
|
||||
releaseRemoteWatchListener(key, sender.id)
|
||||
})
|
||||
registerSenderCleanup(sender)
|
||||
}
|
||||
|
||||
function releaseRemoteWatchListener(key: string, senderId: number): void {
|
||||
|
|
@ -473,6 +474,12 @@ function releaseRemoteWatchListener(key: string, senderId: number): void {
|
|||
remoteWatchers.delete(key)
|
||||
}
|
||||
|
||||
function cleanupRemoteWatchersForSender(senderId: number): void {
|
||||
for (const key of Array.from(remoteWatchers.keys())) {
|
||||
releaseRemoteWatchListener(key, senderId)
|
||||
}
|
||||
}
|
||||
|
||||
type RemoteWatcherInstallResult = 'installed' | 'unavailable' | 'cancelled'
|
||||
|
||||
async function installRemoteWatcher(
|
||||
|
|
@ -637,6 +644,8 @@ export function registerFilesystemWatcherHandlers(): void {
|
|||
|
||||
/** Tear down all watchers on app shutdown. */
|
||||
export async function closeAllWatchers(): Promise<void> {
|
||||
senderCleanupRegistered.clear()
|
||||
|
||||
// Cancel any pending grace-period teardowns — we're tearing down everything.
|
||||
for (const timer of pendingTeardowns.values()) {
|
||||
clearTimeout(timer)
|
||||
|
|
|
|||
|
|
@ -58,7 +58,8 @@ async function usePollingOnce(
|
|||
useEffect: (effect: () => void | (() => void)) => {
|
||||
effect()
|
||||
},
|
||||
useMemo: (factory: () => unknown) => factory()
|
||||
useMemo: (factory: () => unknown) => factory(),
|
||||
useRef: <T>(initial: T) => ({ current: initial })
|
||||
}
|
||||
})
|
||||
|
||||
|
|
@ -159,4 +160,87 @@ describe('useGitStatusPolling', () => {
|
|||
expect(gitStatus).not.toHaveBeenCalled()
|
||||
expect(state.setGitStatus).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does not overlap slow git status polls and runs one trailing refresh', async () => {
|
||||
vi.resetModules()
|
||||
let intervalCallback: (() => void) | null = null
|
||||
let resolveFirst!: (value: GitStatusResult) => void
|
||||
const firstStatus = new Promise<GitStatusResult>((resolve) => {
|
||||
resolveFirst = resolve
|
||||
})
|
||||
const state: PollState = {
|
||||
activeWorktreeId: worktree.id,
|
||||
updateWorktreeGitIdentity: vi.fn(),
|
||||
setGitStatus: vi.fn(),
|
||||
fetchUpstreamStatus: vi.fn().mockResolvedValue(undefined),
|
||||
setUpstreamStatus: vi.fn(),
|
||||
setConflictOperation: vi.fn(),
|
||||
gitConflictOperationByWorktree: {},
|
||||
sshConnectionStates: new Map()
|
||||
}
|
||||
const status: GitStatusResult = {
|
||||
entries: [],
|
||||
conflictOperation: 'unknown',
|
||||
head: 'abc123',
|
||||
branch: 'refs/heads/main'
|
||||
}
|
||||
const gitStatus = vi.fn().mockReturnValueOnce(firstStatus).mockResolvedValue(status)
|
||||
|
||||
vi.doMock('react', async () => {
|
||||
const actual = await vi.importActual<typeof React>('react')
|
||||
return {
|
||||
...actual,
|
||||
useCallback: (callback: unknown) => callback,
|
||||
useEffect: (effect: () => void | (() => void)) => {
|
||||
effect()
|
||||
},
|
||||
useMemo: (factory: () => unknown) => factory(),
|
||||
useRef: <T>(initial: T) => ({ current: initial })
|
||||
}
|
||||
})
|
||||
|
||||
vi.doMock('@/store', () => ({
|
||||
useAppStore: Object.assign((selector: (s: PollState) => unknown) => selector(state), {
|
||||
getState: () => ({ settings: null })
|
||||
})
|
||||
}))
|
||||
vi.doMock('@/store/selectors', () => ({
|
||||
useActiveWorktree: () => worktree,
|
||||
useAllWorktrees: () => [worktree],
|
||||
useRepoById: () => repo,
|
||||
useRepoMap: () => new Map([[repo.id, repo]])
|
||||
}))
|
||||
vi.doMock('@/lib/connection-context', () => ({
|
||||
getConnectionId: () => undefined
|
||||
}))
|
||||
|
||||
vi.stubGlobal('window', {
|
||||
api: { git: { status: gitStatus } },
|
||||
addEventListener: vi.fn(),
|
||||
removeEventListener: vi.fn()
|
||||
})
|
||||
vi.stubGlobal('document', { hasFocus: () => true })
|
||||
vi.stubGlobal(
|
||||
'setInterval',
|
||||
vi.fn((callback: () => void) => {
|
||||
intervalCallback = callback
|
||||
return 1
|
||||
})
|
||||
)
|
||||
vi.stubGlobal('clearInterval', vi.fn())
|
||||
|
||||
const { useGitStatusPolling: runPolling } = await import('./useGitStatusPolling')
|
||||
GitStatusPollingHarness({ runPolling })
|
||||
await vi.waitFor(() => expect(gitStatus).toHaveBeenCalledTimes(1))
|
||||
|
||||
expect(intervalCallback).toBeTypeOf('function')
|
||||
const tick = intervalCallback as unknown as () => void
|
||||
tick()
|
||||
tick()
|
||||
expect(gitStatus).toHaveBeenCalledTimes(1)
|
||||
|
||||
resolveFirst(status)
|
||||
await vi.waitFor(() => expect(gitStatus).toHaveBeenCalledTimes(2))
|
||||
await vi.waitFor(() => expect(state.setGitStatus).toHaveBeenCalledTimes(2))
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import { useCallback, useEffect, useMemo } from 'react'
|
||||
import { useCallback, useEffect, useMemo, useRef } from 'react'
|
||||
import { useAppStore } from '@/store'
|
||||
import { useActiveWorktree, useAllWorktrees, useRepoById, useRepoMap } from '@/store/selectors'
|
||||
import type { GitConflictOperation } from '../../../../shared/types'
|
||||
|
|
@ -21,6 +21,9 @@ export function useGitStatusPolling(): void {
|
|||
const conflictOperationByWorktree = useAppStore((s) => s.gitConflictOperationByWorktree)
|
||||
const sshConnectionStates = useAppStore((s) => s.sshConnectionStates)
|
||||
const repoMap = useRepoMap()
|
||||
const statusPollInFlightRef = useRef(false)
|
||||
const statusPollRerunRef = useRef(false)
|
||||
const fetchStatusRef = useRef<() => void>(() => {})
|
||||
|
||||
const worktreePath = activeWorktree?.path ?? null
|
||||
const activeRepoId = activeWorktree?.repoId ?? null
|
||||
|
|
@ -55,7 +58,7 @@ export function useGitStatusPolling(): void {
|
|||
return result
|
||||
}, [allWorktrees, conflictOperationByWorktree, activeWorktreeId, repoMap])
|
||||
|
||||
const fetchStatus = useCallback(async () => {
|
||||
const runFetchStatus = useCallback(async () => {
|
||||
if (!activeWorktreeId || !worktreePath || !activeRepoSupportsGit) {
|
||||
return
|
||||
}
|
||||
|
|
@ -91,6 +94,25 @@ export function useGitStatusPolling(): void {
|
|||
updateWorktreeGitIdentity
|
||||
])
|
||||
|
||||
const fetchStatus = useCallback(() => {
|
||||
if (statusPollInFlightRef.current) {
|
||||
statusPollRerunRef.current = true
|
||||
return
|
||||
}
|
||||
statusPollInFlightRef.current = true
|
||||
// Why: git status can exceed the 3s poll interval on large repos. Keep at
|
||||
// most one subprocess chain in flight, then run one trailing refresh if a
|
||||
// tick was skipped so the UI catches up without process pileups.
|
||||
void runFetchStatus().finally(() => {
|
||||
statusPollInFlightRef.current = false
|
||||
if (statusPollRerunRef.current) {
|
||||
statusPollRerunRef.current = false
|
||||
fetchStatusRef.current()
|
||||
}
|
||||
})
|
||||
}, [runFetchStatus])
|
||||
fetchStatusRef.current = fetchStatus
|
||||
|
||||
useEffect(() => {
|
||||
void fetchStatus()
|
||||
// Why: skip IPC-heavy git status calls when the window is not focused.
|
||||
|
|
|
|||
|
|
@ -333,6 +333,26 @@ describe('connectPanePty', () => {
|
|||
.cancelAnimationFrame
|
||||
}
|
||||
delete (globalThis as unknown as { window?: unknown }).window
|
||||
delete (globalThis as Record<string, unknown>).__ptyConnectDiag
|
||||
})
|
||||
|
||||
it('does not retain PTY connect diagnostics unless e2e debug state is enabled', async () => {
|
||||
const { connectPanePty } = await import('./pty-connection')
|
||||
const logSpy = vi.spyOn(console, 'log').mockImplementation(() => {})
|
||||
const transport = createMockTransport()
|
||||
transportFactoryQueue.push(transport)
|
||||
mockStoreState = {
|
||||
...mockStoreState,
|
||||
tabsByWorktree: { 'wt-1': [{ id: 'tab-1', ptyId: null }] },
|
||||
ptyIdsByTabId: { 'tab-1': [] }
|
||||
}
|
||||
|
||||
connectPanePty(createPane(1) as never, createManager(1) as never, createDeps() as never)
|
||||
await flushAsyncTicks()
|
||||
|
||||
expect((globalThis as Record<string, unknown>).__ptyConnectDiag).toBeUndefined()
|
||||
expect(logSpy).not.toHaveBeenCalledWith(expect.stringContaining('[pty-connect]'))
|
||||
logSpy.mockRestore()
|
||||
})
|
||||
|
||||
it('does not send startup command via sendInput for local connections', async () => {
|
||||
|
|
|
|||
|
|
@ -26,10 +26,25 @@ import {
|
|||
} from '@/lib/pane-manager/pane-terminal-output-scheduler'
|
||||
import { makePaneKey } from '../../../../shared/stable-pane-id'
|
||||
import { createTerminalCommandLifecycle } from './terminal-command-lifecycle'
|
||||
import { e2eConfig } from '@/lib/e2e-config'
|
||||
|
||||
const pendingSpawnByPaneKey = new Map<string, Promise<string | null>>()
|
||||
const SSH_SESSION_EXPIRED_ERROR = 'SSH_SESSION_EXPIRED'
|
||||
const REMOTE_PTY_ID_PREFIX = 'remote:'
|
||||
const PTY_CONNECT_DIAG_LIMIT = 200
|
||||
|
||||
function recordPtyConnectDiagnostic(message: string): void {
|
||||
if (!e2eConfig.exposeStore) {
|
||||
return
|
||||
}
|
||||
console.log(`[pty-connect] ${message}`)
|
||||
const target = globalThis as Record<string, unknown>
|
||||
const diag = (target.__ptyConnectDiag ??= [] as string[]) as string[]
|
||||
diag.push(message)
|
||||
if (diag.length > PTY_CONNECT_DIAG_LIMIT) {
|
||||
diag.splice(0, diag.length - PTY_CONNECT_DIAG_LIMIT)
|
||||
}
|
||||
}
|
||||
|
||||
// Why: when multiple panes/tabs need the same deferred SSH connection,
|
||||
// the first one calls ssh.connect() and subsequent ones must wait for it
|
||||
|
|
@ -1086,17 +1101,13 @@ export function connectPanePty(
|
|||
isSessionOwnedByWorktree(candidateReattachSessionId, deps.worktreeId)
|
||||
? candidateReattachSessionId
|
||||
: null
|
||||
const _diagMsg = `pane=${pane.id} tab=${deps.tabId} restored=${restoredPtyId} existing=${existingPtyId} detached=${detachedLivePtyId} reattach=${deferredReattachSessionId} hasTransport=${hasExistingPaneTransport} pendingKey=${pendingSpawnKey}`
|
||||
console.log(`[pty-connect] ${_diagMsg}`)
|
||||
;((globalThis as Record<string, unknown>).__ptyConnectDiag ??= [] as string[]) as string[]
|
||||
;((globalThis as Record<string, unknown>).__ptyConnectDiag as string[]).push(_diagMsg)
|
||||
recordPtyConnectDiagnostic(
|
||||
`pane=${pane.id} tab=${deps.tabId} restored=${restoredPtyId} existing=${existingPtyId} detached=${detachedLivePtyId} reattach=${deferredReattachSessionId} hasTransport=${hasExistingPaneTransport} pendingKey=${pendingSpawnKey}`
|
||||
)
|
||||
|
||||
if (deferredReattachSessionId) {
|
||||
allowInitialIdleCacheSeed = true
|
||||
console.log(`[pty-connect] pane=${pane.id} → REATTACH ${deferredReattachSessionId}`)
|
||||
;((globalThis as Record<string, unknown>).__ptyConnectDiag as string[])?.push(
|
||||
`pane=${pane.id} → REATTACH`
|
||||
)
|
||||
recordPtyConnectDiagnostic(`pane=${pane.id} -> REATTACH ${deferredReattachSessionId}`)
|
||||
|
||||
// Why: reattach also pre-signals so the cooperation gate suppresses
|
||||
// the daemon seed for this paneKey. Reattach paths register their
|
||||
|
|
@ -1174,10 +1185,7 @@ export function connectPanePty(
|
|||
startFreshSpawn()
|
||||
})
|
||||
} else if (detachedLivePtyId) {
|
||||
console.log(`[pty-connect] pane=${pane.id} → ATTACH detached=${detachedLivePtyId}`)
|
||||
;((globalThis as Record<string, unknown>).__ptyConnectDiag as string[])?.push(
|
||||
`pane=${pane.id} → ATTACH ${detachedLivePtyId}`
|
||||
)
|
||||
recordPtyConnectDiagnostic(`pane=${pane.id} -> ATTACH detached=${detachedLivePtyId}`)
|
||||
allowInitialIdleCacheSeed = false
|
||||
// Why: surface synchronous attach failures (e.g., the PTY died between
|
||||
// mount and remount, so window.api.pty.resize rejects) through
|
||||
|
|
@ -1210,10 +1218,7 @@ export function connectPanePty(
|
|||
allowInitialIdleCacheSeed = false
|
||||
const pendingSpawn = pendingSpawnByPaneKey.get(pendingSpawnKey)
|
||||
if (pendingSpawn) {
|
||||
console.log(`[pty-connect] pane=${pane.id} → PENDING SPAWN (waiting on same leaf)`)
|
||||
;((globalThis as Record<string, unknown>).__ptyConnectDiag as string[])?.push(
|
||||
`pane=${pane.id} → PENDING SPAWN`
|
||||
)
|
||||
recordPtyConnectDiagnostic(`pane=${pane.id} -> PENDING SPAWN`)
|
||||
void pendingSpawn
|
||||
.then((spawnedPtyId) => {
|
||||
if (disposed) {
|
||||
|
|
@ -1254,10 +1259,7 @@ export function connectPanePty(
|
|||
reportError(err instanceof Error ? err.message : String(err))
|
||||
})
|
||||
} else {
|
||||
console.log(`[pty-connect] pane=${pane.id} → FRESH SPAWN`)
|
||||
;((globalThis as Record<string, unknown>).__ptyConnectDiag as string[])?.push(
|
||||
`pane=${pane.id} → FRESH SPAWN`
|
||||
)
|
||||
recordPtyConnectDiagnostic(`pane=${pane.id} -> FRESH SPAWN`)
|
||||
startFreshSpawn()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -91,6 +91,24 @@ describe('createRemoteRuntimePtyTransport', () => {
|
|||
)
|
||||
}
|
||||
|
||||
function emitSnapshotFrame(
|
||||
streamId: number,
|
||||
opcode:
|
||||
| TerminalStreamOpcode.SnapshotStart
|
||||
| TerminalStreamOpcode.SnapshotChunk
|
||||
| TerminalStreamOpcode.SnapshotEnd,
|
||||
payload: Uint8Array<ArrayBufferLike>
|
||||
): void {
|
||||
subscriptionCallbacks?.onBinary?.(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode,
|
||||
streamId,
|
||||
seq: 1,
|
||||
payload
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetModules()
|
||||
vi.clearAllMocks()
|
||||
|
|
@ -509,4 +527,37 @@ describe('createRemoteRuntimePtyTransport', () => {
|
|||
expect(onBell).not.toHaveBeenCalled()
|
||||
expect(onConnect).toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('bounds oversized binary snapshots without closing the live stream', async () => {
|
||||
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
|
||||
const onReplayData = vi.fn()
|
||||
const onData = vi.fn()
|
||||
const onError = vi.fn()
|
||||
const onConnect = vi.fn()
|
||||
const transport = createRemoteRuntimePtyTransport('env-1', {
|
||||
worktreeId: 'wt-1'
|
||||
})
|
||||
|
||||
await transport.connect({ url: '', callbacks: { onReplayData, onData, onError, onConnect } })
|
||||
await vi.waitFor(() => expect(subscriptionSendBinary).toHaveBeenCalled())
|
||||
const { streamId } = latestSubscribePayload()
|
||||
|
||||
emitSnapshotFrame(
|
||||
streamId,
|
||||
TerminalStreamOpcode.SnapshotStart,
|
||||
encodeTerminalStreamJson({ kind: 'scrollback' })
|
||||
)
|
||||
emitSnapshotFrame(streamId, TerminalStreamOpcode.SnapshotChunk, new Uint8Array(1024 * 1024))
|
||||
emitSnapshotFrame(streamId, TerminalStreamOpcode.SnapshotChunk, new Uint8Array(1024 * 1024))
|
||||
emitSnapshotFrame(streamId, TerminalStreamOpcode.SnapshotChunk, new Uint8Array(1))
|
||||
emitSnapshotFrame(streamId, TerminalStreamOpcode.SnapshotEnd, new Uint8Array())
|
||||
emitOutput(streamId, 'live-after-overflow')
|
||||
|
||||
expect(onReplayData).not.toHaveBeenCalled()
|
||||
expect(onError).toHaveBeenCalledWith(
|
||||
'Remote terminal snapshot exceeded the 2 MiB replay limit; live output will continue.'
|
||||
)
|
||||
expect(onConnect).toHaveBeenCalled()
|
||||
expect(onData).toHaveBeenCalledWith('live-after-overflow')
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -63,9 +63,14 @@ type RemoteRuntimeMultiplexedTerminalState = {
|
|||
terminal: string
|
||||
callbacks: RemoteRuntimeMultiplexedTerminalCallbacks
|
||||
snapshotChunks: Uint8Array<ArrayBufferLike>[]
|
||||
snapshotBytes: number
|
||||
snapshotOverflowed: boolean
|
||||
}
|
||||
|
||||
const CONTROL_STREAM_ID = 0
|
||||
const MAX_REMOTE_TERMINAL_SNAPSHOT_BYTES = 2 * 1024 * 1024
|
||||
const REMOTE_TERMINAL_SNAPSHOT_TOO_LARGE =
|
||||
'Remote terminal snapshot exceeded the 2 MiB replay limit; live output will continue.'
|
||||
|
||||
class RemoteRuntimeTerminalMultiplexer {
|
||||
private readonly streams = new Map<number, RemoteRuntimeMultiplexedTerminalState>()
|
||||
|
|
@ -89,7 +94,9 @@ class RemoteRuntimeTerminalMultiplexer {
|
|||
streamId,
|
||||
terminal: args.terminal,
|
||||
callbacks: args.callbacks,
|
||||
snapshotChunks: []
|
||||
snapshotChunks: [],
|
||||
snapshotBytes: 0,
|
||||
snapshotOverflowed: false
|
||||
}
|
||||
this.streams.set(streamId, state)
|
||||
|
||||
|
|
@ -216,10 +223,12 @@ class RemoteRuntimeTerminalMultiplexer {
|
|||
return
|
||||
}
|
||||
if (event.type === 'end') {
|
||||
clearSnapshot(stream)
|
||||
this.streams.delete(event.streamId)
|
||||
stream.callbacks.onEnd?.()
|
||||
this.closeIfIdle()
|
||||
} else if (event.type === 'error') {
|
||||
clearSnapshot(stream)
|
||||
stream.callbacks.onError?.(
|
||||
typeof event.message === 'string' ? event.message : 'Remote terminal stream failed.'
|
||||
)
|
||||
|
|
@ -258,20 +267,33 @@ class RemoteRuntimeTerminalMultiplexer {
|
|||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.SnapshotStart) {
|
||||
stream.snapshotChunks = []
|
||||
clearSnapshot(stream)
|
||||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.SnapshotChunk) {
|
||||
if (stream.snapshotOverflowed) {
|
||||
return
|
||||
}
|
||||
stream.snapshotBytes += frame.payload.byteLength
|
||||
if (stream.snapshotBytes > MAX_REMOTE_TERMINAL_SNAPSHOT_BYTES) {
|
||||
clearSnapshot(stream)
|
||||
stream.snapshotOverflowed = true
|
||||
stream.callbacks.onError?.(REMOTE_TERMINAL_SNAPSHOT_TOO_LARGE)
|
||||
return
|
||||
}
|
||||
stream.snapshotChunks.push(frame.payload)
|
||||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.SnapshotEnd) {
|
||||
stream.callbacks.onSnapshot(decodeTerminalStreamText(concatBytes(stream.snapshotChunks)))
|
||||
stream.snapshotChunks = []
|
||||
if (!stream.snapshotOverflowed) {
|
||||
stream.callbacks.onSnapshot(decodeTerminalStreamText(concatBytes(stream.snapshotChunks)))
|
||||
}
|
||||
clearSnapshot(stream)
|
||||
stream.callbacks.onSubscribed?.()
|
||||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.Error) {
|
||||
clearSnapshot(stream)
|
||||
stream.callbacks.onError?.(decodeTerminalStreamText(frame.payload))
|
||||
}
|
||||
}
|
||||
|
|
@ -318,6 +340,7 @@ class RemoteRuntimeTerminalMultiplexer {
|
|||
this.subscription = null
|
||||
this.streams.clear()
|
||||
for (const stream of streams) {
|
||||
clearSnapshot(stream)
|
||||
stream.callbacks.onTransportClose?.()
|
||||
if (message) {
|
||||
stream.callbacks.onError?.(message)
|
||||
|
|
@ -364,6 +387,12 @@ function concatBytes(chunks: Uint8Array<ArrayBufferLike>[]): Uint8Array<ArrayBuf
|
|||
return out
|
||||
}
|
||||
|
||||
function clearSnapshot(stream: RemoteRuntimeMultiplexedTerminalState): void {
|
||||
stream.snapshotChunks = []
|
||||
stream.snapshotBytes = 0
|
||||
stream.snapshotOverflowed = false
|
||||
}
|
||||
|
||||
function isTerminalDriverState(
|
||||
value: unknown
|
||||
): value is { kind: 'idle' } | { kind: 'desktop' } | { kind: 'mobile'; clientId: string } {
|
||||
|
|
|
|||
|
|
@ -866,6 +866,24 @@ describe('createEditorSlice remote branch actions', () => {
|
|||
})
|
||||
})
|
||||
|
||||
it('does not notify subscribers when upstream status is unchanged', () => {
|
||||
const store = createEditorStore()
|
||||
const status = {
|
||||
hasUpstream: true,
|
||||
upstreamName: 'origin/main',
|
||||
ahead: 2,
|
||||
behind: 1
|
||||
}
|
||||
|
||||
store.getState().setUpstreamStatus('wt-1', status)
|
||||
const listener = vi.fn()
|
||||
const unsubscribe = store.subscribe(listener)
|
||||
store.getState().setUpstreamStatus('wt-1', { ...status })
|
||||
unsubscribe()
|
||||
|
||||
expect(listener).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('runs pull and refreshes status + upstream on success', async () => {
|
||||
const store = createEditorStore()
|
||||
store.getState().setGitStatus('wt-1', {
|
||||
|
|
|
|||
|
|
@ -1960,12 +1960,17 @@ export const createEditorSlice: StateCreator<AppState, [], [], EditorSlice> = (s
|
|||
}),
|
||||
remoteStatusesByWorktree: {},
|
||||
setUpstreamStatus: (worktreeId, status) =>
|
||||
set((s) => ({
|
||||
remoteStatusesByWorktree: {
|
||||
...s.remoteStatusesByWorktree,
|
||||
[worktreeId]: status
|
||||
set((s) => {
|
||||
if (areUpstreamStatusesEqual(s.remoteStatusesByWorktree[worktreeId], status)) {
|
||||
return s
|
||||
}
|
||||
})),
|
||||
return {
|
||||
remoteStatusesByWorktree: {
|
||||
...s.remoteStatusesByWorktree,
|
||||
[worktreeId]: status
|
||||
}
|
||||
}
|
||||
}),
|
||||
isRemoteOperationActive: false,
|
||||
remoteOperationDepth: 0,
|
||||
inFlightRemoteOpKind: null,
|
||||
|
|
@ -2525,6 +2530,19 @@ function areTrackedConflictMapsEqual(
|
|||
return prevKeys.length === nextKeys.length && prevKeys.every((key) => prev[key] === next[key])
|
||||
}
|
||||
|
||||
function areUpstreamStatusesEqual(
|
||||
prev: GitUpstreamStatus | undefined,
|
||||
next: GitUpstreamStatus
|
||||
): boolean {
|
||||
return (
|
||||
prev !== undefined &&
|
||||
prev.hasUpstream === next.hasUpstream &&
|
||||
prev.upstreamName === next.upstreamName &&
|
||||
prev.ahead === next.ahead &&
|
||||
prev.behind === next.behind
|
||||
)
|
||||
}
|
||||
|
||||
function reconcileOpenFilesForStatus(
|
||||
openFiles: OpenFile[],
|
||||
worktreeId: string,
|
||||
|
|
|
|||
|
|
@ -60,6 +60,7 @@ import {
|
|||
makeWorktree,
|
||||
seedStore
|
||||
} from './store-test-helpers'
|
||||
import { shutdownBufferCaptures } from '@/components/terminal-pane/shutdown-buffer-captures'
|
||||
|
||||
// ─── Tests ────────────────────────────────────────────────────────────
|
||||
|
||||
|
|
@ -1366,6 +1367,28 @@ describe('shutdownWorktreeTerminals (sleep) — agent status hygiene', () => {
|
|||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
mockApi.pty.kill.mockResolvedValue(undefined)
|
||||
shutdownBufferCaptures.clear()
|
||||
})
|
||||
|
||||
it('asks sleep-time buffer capture to skip local scrollback serialization', async () => {
|
||||
const store = createTestStore()
|
||||
const wt = 'repo1::/path/wt1'
|
||||
const capture = vi.fn()
|
||||
|
||||
seedStore(store, {
|
||||
worktreesByRepo: {
|
||||
repo1: [makeWorktree({ id: wt, repoId: 'repo1', path: '/path/wt1' })]
|
||||
},
|
||||
tabsByWorktree: {
|
||||
[wt]: [makeTab({ id: 'tab-1', worktreeId: wt, ptyId: 'pty-1' })]
|
||||
},
|
||||
ptyIdsByTabId: { 'tab-1': ['pty-1'] }
|
||||
})
|
||||
shutdownBufferCaptures.set('tab-1', capture)
|
||||
|
||||
await store.getState().shutdownWorktreeTerminals(wt, { keepIdentifiers: true })
|
||||
|
||||
expect(capture).toHaveBeenCalledWith({ includeLocalBuffers: false })
|
||||
})
|
||||
|
||||
it('drops live agentStatusByPaneKey entries on sleep so the working row disappears', async () => {
|
||||
|
|
|
|||
|
|
@ -1111,15 +1111,15 @@ export const createTerminalSlice: StateCreator<AppState, [], [], TerminalSlice>
|
|||
// subsequent set must use a functional updater spreading
|
||||
// s.terminalLayoutsByTabId, not a captured snapshot). For SSH this is
|
||||
// load-bearing — the relay drops the remote PTY on kill so there's no
|
||||
// on-disk history dir to cold-restore from. For local daemon it's
|
||||
// defense-in-depth alongside the on-disk history dir preserved by
|
||||
// keepHistory below.
|
||||
// on-disk history dir to cold-restore from. Local daemon scrollback is
|
||||
// intentionally skipped because the session payload prunes it and daemon
|
||||
// history/checkpoints are authoritative.
|
||||
if (keepIdentifiers) {
|
||||
for (const tab of tabs) {
|
||||
const capture = shutdownBufferCaptures.get(tab.id)
|
||||
if (capture) {
|
||||
try {
|
||||
capture()
|
||||
capture({ includeLocalBuffers: false })
|
||||
} catch {
|
||||
// Don't let one tab's capture failure block the rest.
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue