From 23103e511a379a54957e194072ffc0df4df786c8 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Wed, 14 Oct 2026 17:13:29 -0400 Subject: [PATCH] perf(layout): re-project only the written partition in the layout stream (#28845) The layout stream (layout.subscribe, still unused) re-loaded every session partition and ran the load's save-and-diff report after each write. A save now names the partition it wrote, the stream re-loads only that partition without the report, and the renderer layout client no longer re-subscribes on a timer. Adds a latency spec variant with a subscriber attached. Co-authored-by: Claude Opus 5.5 --- ...sistence-host-partitioned-sessions.test.ts | 14 ++ .../loading-store/session-host-partitions.ts | 19 ++- .../session-snapshot-operations.ts | 3 +- .../loading-store/store-runtime-state.ts | 3 +- .../workspace-session-snapshot-publication.ts | 5 +- .../workspace-session-write-listeners.ts | 14 +- .../loading-store/write-scheduling.ts | 14 +- .../runtime/workspace-layout-stream.test.ts | 132 +++++++++++++++++- src/main/runtime/workspace-layout-stream.ts | 86 ++++++++---- .../src/runtime/runtime-layout-client.test.ts | 67 +++------ .../src/runtime/runtime-layout-client.ts | 37 ++--- .../workspace-layout/workspace-layout-load.ts | 14 +- .../workspace-layout-oracle-latency.spec.ts | 130 +++++++++++++---- 13 files changed, 383 insertions(+), 155 deletions(-) diff --git a/src/main/persistence-host-partitioned-sessions.test.ts b/src/main/persistence-host-partitioned-sessions.test.ts index 9c3aad9620f..1314d8e2ff5 100644 --- a/src/main/persistence-host-partitioned-sessions.test.ts +++ b/src/main/persistence-host-partitioned-sessions.test.ts @@ -314,6 +314,20 @@ describe('Store host-partitioned workspace sessions', () => { expect(store.getWorkspaceSession('local').activeRepoId).toBe('repo-local') }) + it('names the one partition each session write changed', async () => { + const store = await createStore() + store.setWorkspaceSession(makeHostSession('repo-a'), 'runtime:env-a') + const written: (string | null)[] = [] + store.onWorkspaceSessionWritten((hostId) => written.push(hostId)) + + store.setWorkspaceSession(makeHostSession('repo-local'), 'local') + store.setWorkspaceSession(makeHostSession('repo-a2'), 'runtime:env-a') + store.patchWorkspaceSession({ activeTabId: 'tab-a' }, 'runtime:env-a') + store.removeWorkspaceSessionHost('runtime:env-a') + + expect(written).toEqual(['local', 'runtime:env-a', 'runtime:env-a', 'runtime:env-a']) + }) + it('patches a single host partition without touching the others', async () => { const store = await createStore() store.setWorkspaceSession(makeHostSession('repo-local'), 'local') diff --git a/src/main/persistence/loading-store/session-host-partitions.ts b/src/main/persistence/loading-store/session-host-partitions.ts index aace1ce4b68..12c5b7bc241 100644 --- a/src/main/persistence/loading-store/session-host-partitions.ts +++ b/src/main/persistence/loading-store/session-host-partitions.ts @@ -93,9 +93,11 @@ export class SessionHostPartitionOperations { const { [hostId]: _removed, ...remaining } = partitions runtime.state.workspaceSessionsByHostId = remaining invalidateLocalWorktreeMetadataPruneInputs() - scheduleSave(this[sessionHostPartitionOperationsContext].scheduling, [ - 'workspaceSessionsByHostId' - ]) + scheduleSave( + this[sessionHostPartitionOperationsContext].scheduling, + ['workspaceSessionsByHostId'], + hostId + ) } readTerminalScrollbackSnapshot(ref: string): string | null { @@ -170,7 +172,8 @@ export function removeWorkspaceSessionOwnerInPartition( } scheduleSave( owner[sessionHostPartitionOperationsContext].scheduling, - resolved === LOCAL_EXECUTION_HOST_ID ? ['workspaceSession'] : ['workspaceSessionsByHostId'] + resolved === LOCAL_EXECUTION_HOST_ID ? ['workspaceSession'] : ['workspaceSessionsByHostId'], + resolved ) } @@ -234,9 +237,11 @@ export function setHostWorkspaceSession( ...owner[sessionHostPartitionOperationsContext].runtime.state.workspaceSessionsByHostId, [hostId]: withRequiredWorkspaceSessionMaps(orderTerminalRowsByTabGroups(pruned)) } - scheduleSave(owner[sessionHostPartitionOperationsContext].scheduling, [ - 'workspaceSessionsByHostId' - ]) + scheduleSave( + owner[sessionHostPartitionOperationsContext].scheduling, + ['workspaceSessionsByHostId'], + hostId + ) } /** diff --git a/src/main/persistence/loading-store/session-snapshot-operations.ts b/src/main/persistence/loading-store/session-snapshot-operations.ts index 88c209485b3..c7320cc346e 100644 --- a/src/main/persistence/loading-store/session-snapshot-operations.ts +++ b/src/main/persistence/loading-store/session-snapshot-operations.ts @@ -128,7 +128,8 @@ export class SessionSnapshotOperations { } scheduleSave( scheduling, - hostId === LOCAL_EXECUTION_HOST_ID ? ['workspaceSession'] : ['workspaceSessionsByHostId'] + hostId === LOCAL_EXECUTION_HOST_ID ? ['workspaceSession'] : ['workspaceSessionsByHostId'], + hostId ) } diff --git a/src/main/persistence/loading-store/store-runtime-state.ts b/src/main/persistence/loading-store/store-runtime-state.ts index f72bba42ffe..ce557228c29 100644 --- a/src/main/persistence/loading-store/store-runtime-state.ts +++ b/src/main/persistence/loading-store/store-runtime-state.ts @@ -1,6 +1,7 @@ import { removeStaleDurableWriteTempFiles } from '../../durable-file-write' import type { GlobalSettings } from '../../../shared/global-settings-types' import type { PersistedState } from '../../../shared/persisted-state-types' +import type { WorkspaceSessionWriteListener } from './workspace-session-write-listeners' import type { ActiveViewPreference } from '../../active-view-preference' import { getProfileTerminalScrollbackSnapshotRoot, @@ -84,7 +85,7 @@ export class StoreRuntimeState { >() uiChangeListeners = new Set<(ui: PersistedState['ui']) => void>() /** Observers of persisted workspace-session writes; see notifyWorkspaceSessionWritten. */ - readonly workspaceSessionWriteListeners = new Set<() => void>() + readonly workspaceSessionWriteListeners = new Set() projectHostOperations: ProjectHostPersistenceOperations | null = null projectGroupOperations: ProjectGroupPersistenceOperations | null = null folderWorkspaceOperations: FolderWorkspacePersistenceOperations | null = null diff --git a/src/main/persistence/loading-store/workspace-session-snapshot-publication.ts b/src/main/persistence/loading-store/workspace-session-snapshot-publication.ts index d1446c701c4..f73fa61444c 100644 --- a/src/main/persistence/loading-store/workspace-session-snapshot-publication.ts +++ b/src/main/persistence/loading-store/workspace-session-snapshot-publication.ts @@ -124,11 +124,12 @@ export function setLocalWorkspaceSession( remappedManualUnread.changed ) { // UI remaps include protected fields, so keep the complete serializer boundary. - scheduleSave(context.scheduling) + scheduleSave(context.scheduling, undefined, LOCAL_EXECUTION_HOST_ID) } else { scheduleSave( context.scheduling, - remappedLeases.changed ? ['workspaceSession', 'sshRemotePtyLeases'] : ['workspaceSession'] + remappedLeases.changed ? ['workspaceSession', 'sshRemotePtyLeases'] : ['workspaceSession'], + LOCAL_EXECUTION_HOST_ID ) } } diff --git a/src/main/persistence/loading-store/workspace-session-write-listeners.ts b/src/main/persistence/loading-store/workspace-session-write-listeners.ts index 88407105e7b..9f8d4880012 100644 --- a/src/main/persistence/loading-store/workspace-session-write-listeners.ts +++ b/src/main/persistence/loading-store/workspace-session-write-listeners.ts @@ -1,19 +1,27 @@ +import type { ExecutionHostId } from '../../../shared/execution-host' import type { StoreRuntimeState } from './store-runtime-state' +/** Gets the one partition a write changed, or null when any partition may have changed. */ +export type WorkspaceSessionWriteListener = (hostId: ExecutionHostId | null) => void + const WORKSPACE_SESSION_DOMAINS: readonly string[] = [ 'workspaceSession', 'workspaceSessionsByHostId' ] -/** `dirtyDomains` undefined means a full save, which may carry an in-place session edit. */ +/** + * `dirtyDomains` undefined means a full save, which may carry an in-place session edit. + * `hostId` names the only partition the write changed; without it every partition is suspect. + */ export function notifyWorkspaceSessionWritten( runtime: Pick, - dirtyDomains?: readonly string[] + dirtyDomains?: readonly string[], + hostId?: ExecutionHostId ): void { if (dirtyDomains && !dirtyDomains.some((domain) => WORKSPACE_SESSION_DOMAINS.includes(domain))) { return } for (const listener of runtime.workspaceSessionWriteListeners) { - listener() + listener(hostId ?? null) } } diff --git a/src/main/persistence/loading-store/write-scheduling.ts b/src/main/persistence/loading-store/write-scheduling.ts index edd2754ddd9..9c622b64ee0 100644 --- a/src/main/persistence/loading-store/write-scheduling.ts +++ b/src/main/persistence/loading-store/write-scheduling.ts @@ -1,7 +1,11 @@ import type { StoreRuntimeState } from './store-runtime-state' import type { PrimaryStateWriteOperations } from './primary-state-writes' import { enqueueWrite } from './primary-state-writes' -import { notifyWorkspaceSessionWritten } from './workspace-session-write-listeners' +import type { ExecutionHostId } from '../../../shared/execution-host' +import { + notifyWorkspaceSessionWritten, + type WorkspaceSessionWriteListener +} from './workspace-session-write-listeners' const SAVE_DEBOUNCE_MS = 1_000 const SAVE_MAX_WAIT_MS = 5_000 @@ -40,7 +44,7 @@ export class WriteSchedulingOperations { } /** Called after each in-memory workspace-session write; listeners must not throw or block. */ - onWorkspaceSessionWritten(listener: () => void): () => void { + onWorkspaceSessionWritten(listener: WorkspaceSessionWriteListener): () => void { const { runtime } = this[writeSchedulingOperationsContext] runtime.workspaceSessionWriteListeners.add(listener) return () => { @@ -49,12 +53,14 @@ export class WriteSchedulingOperations { } } +/** `writtenHostId` names the one session partition this save changed, if it changed only one. */ export function scheduleSave( owner: WriteSchedulingOperations, - dirtyDomains?: readonly string[] + dirtyDomains?: readonly string[], + writtenHostId?: ExecutionHostId ): void { const { runtime, writes } = owner[writeSchedulingOperationsContext] - notifyWorkspaceSessionWritten(runtime, dirtyDomains) + notifyWorkspaceSessionWritten(runtime, dirtyDomains, writtenHostId) runtime.automationListProjectionCache = null const trackedDomains = runtime.dirtyProfileStateDomains if (dirtyDomains === undefined) { diff --git a/src/main/runtime/workspace-layout-stream.test.ts b/src/main/runtime/workspace-layout-stream.test.ts index fc9bd1ac572..b85a29efebb 100644 --- a/src/main/runtime/workspace-layout-stream.test.ts +++ b/src/main/runtime/workspace-layout-stream.test.ts @@ -3,39 +3,77 @@ import { LOCAL_EXECUTION_HOST_ID, type ExecutionHostId } from '../../shared/exec import type { WorkspaceSessionState } from '../../shared/workspace-session-state-types' import { localDesktopSession, - relaySshSession + relaySshSession, + serverRuntimeSession } from '../../shared/workspace-layout/workspace-layout-profile.test-fixture' import { emptySession, FOLDER_KEY, GIT_KEY, + SERVER_KEY, SSH_KEY } from '../../shared/workspace-layout/workspace-layout-session.test-fixture' import type { WorkspaceLayoutEvent } from '../../shared/workspace-layout/workspace-layout-stream-frames' +import { loadWorkspaceLayout } from '../../shared/workspace-layout/workspace-layout-load' +import { nameBasedLoadContext } from '../../shared/workspace-layout/workspace-layout-minted-ids' +import { + publishWorkspaceLayout, + type PublishedWorkspaceLayout +} from '../../shared/workspace-layout/workspace-layout-published' import { WorkspaceLayoutStream } from './workspace-layout-stream' const SSH_HOST: ExecutionHostId = 'ssh:target-1' +const SERVER_HOST: ExecutionHostId = 'runtime:env-1' function fakeStore(partitions: Map) { - const listeners = new Set<() => void>() - const onWorkspaceSessionWritten = vi.fn((listener: () => void) => { + const listeners = new Set<(hostId: ExecutionHostId | null) => void>() + const onWorkspaceSessionWritten = vi.fn((listener: (hostId: ExecutionHostId | null) => void) => { listeners.add(listener) return () => listeners.delete(listener) }) + const getWorkspaceSession = vi.fn((hostId: ExecutionHostId) => partitions.get(hostId)!) return { store: { - getWorkspaceSession: (hostId: ExecutionHostId) => partitions.get(hostId)!, + getWorkspaceSession, getWorkspaceSessionHostIds: () => [...partitions.keys()], onWorkspaceSessionWritten }, listeners, + /** A save that names the partition it wrote. */ write(hostId: ExecutionHostId, session: WorkspaceSessionState) { partitions.set(hostId, session) - listeners.forEach((listener) => listener()) + listeners.forEach((listener) => listener(hostId)) + }, + /** A save that names no partition, so every partition may have changed. */ + writeUnnamed(hostId: ExecutionHostId, session: WorkspaceSessionState) { + partitions.set(hostId, session) + listeners.forEach((listener) => listener(null)) + }, + /** A removed host's partition, named by the save that drops it. */ + removePartition(hostId: ExecutionHostId) { + partitions.delete(hostId) + listeners.forEach((listener) => listener(hostId)) } } } +/** Today's projection: every partition loaded with its save-and-diff report, then filtered by home. */ +function fullProjection( + partitions: Map, + homeHostId: (key: string) => ExecutionHostId | null +): [string, PublishedWorkspaceLayout][] { + const projected = new Map() + for (const [hostId, session] of partitions) { + const { layout } = loadWorkspaceLayout(hostId, session, nameBasedLoadContext()) + for (const [key, workspace] of Object.entries(layout.workspaces)) { + if (homeHostId(key) === hostId) { + projected.set(key, publishWorkspaceLayout(workspace, hostId)) + } + } + } + return [...projected] +} + function setup() { const partitions = new Map([ [LOCAL_EXECUTION_HOST_ID, localDesktopSession()], @@ -51,7 +89,7 @@ function setup() { store: () => fake.store, homeHostId: (key) => homes[key] ?? null }) - return { ...fake, stream, homes } + return { ...fake, stream, homes, partitions } } const flush = () => new Promise((resolve) => queueMicrotask(resolve)) @@ -199,4 +237,86 @@ describe('WorkspaceLayoutStream', () => { expect(JSON.stringify(later.tabs)).toBe(JSON.stringify(earlier.tabs.slice(1))) } }) + + it('re-projects only the partition a write names', async () => { + const { stream, store, write, writeUnnamed } = setup() + stream.subscribe(() => {}) + store.getWorkspaceSession.mockClear() + write(SSH_HOST, relaySshSession('target-1')) + await flush() + expect(store.getWorkspaceSession.mock.calls).toEqual([[SSH_HOST]]) + store.getWorkspaceSession.mockClear() + writeUnnamed(SSH_HOST, relaySshSession('target-1')) + await flush() + expect(store.getWorkspaceSession.mock.calls).toEqual([[LOCAL_EXECUTION_HOST_ID], [SSH_HOST]]) + }) + + it('publishes what a full re-projection with the load report would, write after write', async () => { + const { stream, write, writeUnnamed, removePartition, homes, partitions } = setup() + const homeHostId = (key: string) => homes[key] ?? null + let cache = new Map() + const { snapshot } = stream.subscribe((event) => { + if (event.type === 'removed') { + cache.delete(event.key) + } else { + cache.set(event.key, event.layout) + } + }) + cache = new Map(snapshot.map((entry) => [entry.key, entry.layout])) + const expectFullProjection = async (): Promise => { + await flush() + const expected = fullProjection(partitions, homeHostId) + expect(Object.fromEntries(cache)).toEqual(Object.fromEntries(expected)) + // A late subscriber's snapshot keeps today's order too. + const late = stream.subscribe(() => {}) + expect(late.snapshot.map((entry) => [entry.key, entry.layout])).toEqual(expected) + late.unsubscribe() + } + await expectFullProjection() + + const local = localDesktopSession() + local.tabGroups![GIT_KEY]![0]!.tabOrder.reverse() + write(LOCAL_EXECUTION_HOST_ID, local) + await expectFullProjection() + + // A named write and an unnamed one in the same task. + const ssh = relaySshSession('target-1') + delete ssh.tabsByWorktree[SSH_KEY] + delete ssh.unifiedTabs![SSH_KEY] + write(SSH_HOST, ssh) + writeUnnamed(LOCAL_EXECUTION_HOST_ID, localDesktopSession()) + await expectFullProjection() + + // A new partition, a home that moves between partitions, then the partition removed. + homes[SERVER_KEY] = SERVER_HOST + write(SERVER_HOST, serverRuntimeSession('env-1')) + await expectFullProjection() + homes[GIT_KEY] = SERVER_HOST + write(SSH_HOST, relaySshSession('target-1')) + await expectFullProjection() + homes[GIT_KEY] = LOCAL_EXECUTION_HOST_ID + removePartition(SERVER_HOST) + await expectFullProjection() + }) + + it('re-projects every partition after a failed projection', async () => { + const { stream, store, write } = setup() + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const events: WorkspaceLayoutEvent[] = [] + stream.subscribe((event) => events.push(event)) + store.getWorkspaceSession.mockImplementationOnce(() => { + throw new Error('read failed') + }) + const local = localDesktopSession() + local.tabGroups![GIT_KEY]![0]!.tabOrder.reverse() + write(LOCAL_EXECUTION_HOST_ID, local) + await flush() + expect(events).toEqual([]) + // The local edit the failed projection missed is published by a write elsewhere. + write(SSH_HOST, relaySshSession('target-1')) + await flush() + expect(events.map((event) => [event.type, event.key])).toEqual([['workspace', GIT_KEY]]) + expect(warn).toHaveBeenCalledTimes(1) + warn.mockRestore() + }) }) diff --git a/src/main/runtime/workspace-layout-stream.ts b/src/main/runtime/workspace-layout-stream.ts index 8c20e38cd18..fb3b4a0eb72 100644 --- a/src/main/runtime/workspace-layout-stream.ts +++ b/src/main/runtime/workspace-layout-stream.ts @@ -1,9 +1,10 @@ // The read-only layout stream behind `layout.subscribe`: each workspace's published layout, -// re-projected from the Store's sessions after every write while anyone subscribes. Nothing reads -// it until a client opts in; with no subscriber it does no work. +// re-projected from the Store's sessions after every write while anyone subscribes. A write that +// names its partition re-projects only that partition. Nothing reads it until a client opts in; +// with no subscriber it does no work. import type { ExecutionHostId } from '../../shared/execution-host' -import { loadWorkspaceLayout } from '../../shared/workspace-layout/workspace-layout-load' +import { loadWorkspaceLayoutWithoutReport } from '../../shared/workspace-layout/workspace-layout-load' import { stableJson } from '../../shared/workspace-layout/workspace-layout-load-report' import { publishWorkspaceLayout, @@ -26,8 +27,11 @@ export class WorkspaceLayoutStream { private readonly listeners = new Set() /** Workspace key → its published layout, with its JSON for the by-value comparison. */ private published = new Map() + /** Partition → every workspace it holds, its own copy of a workspace it does not own included. */ + private projectedPartitions = new Map>() private stopWatching: (() => void) | null = null - private dirty = false + /** Partitions to re-project at the next flush; 'all' when a write did not name its partition. */ + private dirty: Set | 'all' | null = null private failureLogged = false constructor( @@ -45,9 +49,10 @@ export class WorkspaceLayoutStream { } { if (this.listeners.size === 0) { this.stopWatching = - this.deps.store()?.onWorkspaceSessionWritten?.(() => this.markDirty()) ?? null + this.deps.store()?.onWorkspaceSessionWritten?.((hostId) => this.markDirty(hostId)) ?? null this.published = new Map() - this.dirty = true + this.projectedPartitions = new Map() + this.dirty = 'all' } // A write not yet published reaches existing listeners first, so the snapshot is current. this.flush() @@ -62,28 +67,38 @@ export class WorkspaceLayoutStream { this.stopWatching?.() this.stopWatching = null this.published.clear() + this.projectedPartitions.clear() } } } - private markDirty(): void { - if (this.dirty) { - return + private markDirty(hostId: ExecutionHostId | null): void { + const scheduled = this.dirty !== null + if (hostId === null) { + this.dirty = 'all' + } else if (this.dirty === null) { + this.dirty = new Set([hostId]) + } else if (this.dirty !== 'all') { + this.dirty.add(hostId) + } + if (!scheduled) { + // Coalesces a burst of writes in one task into one projection. + queueMicrotask(() => this.flush()) } - this.dirty = true - // Coalesces a burst of writes in one task into one projection. - queueMicrotask(() => this.flush()) } private flush(): void { - if (!this.dirty) { + const dirty = this.dirty + if (dirty === null) { return } - this.dirty = false + this.dirty = null // A failed projection is logged once; it never reaches the write that notified it. try { - this.reconcile() + this.reconcile(dirty) } catch (error) { + // The cache may be part-updated; the next write re-projects every partition. + this.projectedPartitions.clear() this.logFailure('projection', error) } } @@ -95,28 +110,43 @@ export class WorkspaceLayoutStream { } } - private project(): Map { - const store = this.deps.store() + private projectPartition(hostId: ExecutionHostId): Map { const projected = new Map() - for (const hostId of store?.getWorkspaceSessionHostIds?.() ?? []) { - const session = store?.getWorkspaceSession?.(hostId) - if (!session) { - continue - } - const { layout } = loadWorkspaceLayout(hostId, session, nameBasedLoadContext()) - for (const [key, workspace] of Object.entries(layout.workspaces)) { + const session = this.deps.store()?.getWorkspaceSession?.(hostId) + if (!session) { + return projected + } + const { layout } = loadWorkspaceLayoutWithoutReport(hostId, session, nameBasedLoadContext()) + for (const [key, workspace] of Object.entries(layout.workspaces)) { + const published = publishWorkspaceLayout(workspace, hostId) + projected.set(key, { layout: published, json: stableJson(published) }) + } + return projected + } + + private project(dirty: Set | 'all'): Map { + const hostIds = this.deps.store()?.getWorkspaceSessionHostIds?.() ?? [] + const partitions = new Map>() + for (const hostId of hostIds) { + const cached = + dirty === 'all' || dirty.has(hostId) ? undefined : this.projectedPartitions.get(hostId) + partitions.set(hostId, cached ?? this.projectPartition(hostId)) + } + this.projectedPartitions = partitions + const projected = new Map() + for (const [hostId, workspaces] of partitions) { + for (const [key, workspace] of workspaces) { // Only the owning partition's copy is published; a stray copy elsewhere is not shown. if (this.deps.homeHostId(key) === hostId) { - const layout = publishWorkspaceLayout(workspace, hostId) - projected.set(key, { layout, json: stableJson(layout) }) + projected.set(key, workspace) } } } return projected } - private reconcile(): void { - const next = this.project() + private reconcile(dirty: Set | 'all'): void { + const next = this.project(dirty) const events: WorkspaceLayoutEvent[] = [] for (const key of this.published.keys()) { if (!next.has(key)) { diff --git a/src/renderer/src/runtime/runtime-layout-client.test.ts b/src/renderer/src/runtime/runtime-layout-client.test.ts index b28173419b5..e76b69b1aec 100644 --- a/src/renderer/src/runtime/runtime-layout-client.test.ts +++ b/src/renderer/src/runtime/runtime-layout-client.test.ts @@ -142,15 +142,6 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta return host.streams[count - 1]! } - /** Whether a retry opens exactly `delayMs` after the previous loss. */ - async function expectRetryAfter(delayMs: number): Promise { - const count = host.streams.length - await vi.advanceTimersByTimeAsync(delayMs - 1) - expect(host.streams).toHaveLength(count) - await vi.advanceTimersByTimeAsync(1) - return opened(count + 1) - } - function snapshot(workspaces: { key: string; layout: PublishedWorkspaceLayout }[] = []) { return { type: 'snapshot', subscriptionId: 'layout-1', workspaces } } @@ -186,7 +177,7 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta expect(stream.unsubscribed).toBe(false) }) - it('after a lost stream, resubscribes and the new snapshot replaces the cache', async () => { + it('after a lost stream, keeps the last state and never reopens it', async () => { subscribe() const first = await opened(1) first.send( @@ -197,36 +188,27 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta ) first.drop() expect(first.unsubscribed).toBe(true) - // The last state stays until the runtime sends a new one; nothing is cleared or guessed. - expect(cache).toEqual({ a: layout('a', ['t1', 't2']), b: layout('b') }) - const second = await expectRetryAfter(250) - // A late frame from the dropped stream changes nothing. + expect(status).toEqual({ state: 'lost' }) + await vi.advanceTimersByTimeAsync(60_000) + expect(host.streams).toHaveLength(1) + // The last state stays; a late frame from the dropped stream changes nothing. first.send({ type: 'removed', key: 'a' }) - expect(cache).toHaveProperty('a') - second.send(snapshot([{ key: 'a', layout: layout('a', ['t2']) }])) - expect(cache).toEqual({ a: layout('a', ['t2']) }) + expect(cache).toEqual({ a: layout('a', ['t1', 't2']), b: layout('b') }) }) - it('backs off on failed subscribes and error frames, and a snapshot alone does not reset it', async () => { - host.rejectNextSubscribe() - subscribe() - await vi.advanceTimersByTimeAsync(0) - expect(host.streams).toHaveLength(0) - const first = await expectRetryAfter(250) - first.fail('stream failed') - expect(first.unsubscribed).toBe(true) - // A host that snapshots then drops at once keeps backing off instead of looping fast. - let stream = await expectRetryAfter(500) - for (const delay of [1000, 2000, 4000, 5000, 5000]) { - stream.send(snapshot()) - stream.drop() - stream = await expectRetryAfter(delay) - } - // A change after the snapshot proves the stream stayed up; the next loss starts over. - stream.send(snapshot()) - stream.send({ type: 'workspace', key: 'a', layout: layout('a') }) - stream.drop() - await expectRetryAfter(250) + it.each([ + ['a failed subscribe', () => host.rejectNextSubscribe(), async () => {}], + ['an error frame', () => {}, async () => (await opened(1)).fail('stream failed')] + ])('stops after %s without retrying', async (_case, before, fail) => { + before() + const subscription = subscribe() + await fail() + await vi.advanceTimersByTimeAsync(60_000) + expect(status).toEqual({ state: 'lost' }) + expect(host.streams.length).toBeLessThanOrEqual(1) + expect(host.streams.every((stream) => stream.unsubscribed)).toBe(true) + subscription.close() + expect(status).toEqual({ state: 'lost' }) }) it.each([ @@ -260,7 +242,7 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta expect(status).toMatchObject({ state: 'refused', code: 'method_not_found' }) }) - it('close unsubscribes, stops retrying and ignores later frames', async () => { + it('close unsubscribes and ignores later frames', async () => { const subscription = subscribe() const stream = await opened(1) stream.send(snapshot()) @@ -274,15 +256,6 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta expect(cache).toEqual({}) }) - it('close while waiting to retry cancels the retry', async () => { - const subscription = subscribe() - ;(await opened(1)).drop() - expect(status).toEqual({ state: 'retrying' }) - subscription.close() - await vi.advanceTimersByTimeAsync(10_000) - expect(host.streams).toHaveLength(1) - }) - it('close before the subscribe settles releases the stream it then gets', async () => { const subscription = subscribe() subscription.close() diff --git a/src/renderer/src/runtime/runtime-layout-client.ts b/src/renderer/src/runtime/runtime-layout-client.ts index 01e6eab01df..fc1609438ba 100644 --- a/src/renderer/src/runtime/runtime-layout-client.ts +++ b/src/renderer/src/runtime/runtime-layout-client.ts @@ -11,20 +11,19 @@ import { import type { RuntimeClientTarget } from '../../../shared/runtime-client-target' import { subscribeRuntimeRpc } from './runtime-rpc-subscribe' -/** `retrying`: waiting to reopen a lost stream. `refused`: the host answered that it will never - * serve this stream, so the client stopped. */ +/** `lost`: the stream ended without a refusal and the client stopped. `refused`: the host answered + * that it will never serve this stream, so the client stopped. */ export type RuntimeLayoutStreamStatus = - | { state: 'connecting' | 'live' | 'retrying' | 'closed' } + | { state: 'connecting' | 'live' | 'lost' | 'closed' } | { state: 'refused'; code: string } -const RETRY_BASE_MS = 250 -const RETRY_MAX_MS = 5000 // An older host without the method, or a caller scope denied it: a retry gets the same answer. const REFUSAL_CODES = ['method_not_found', 'forbidden'] as const /** - * Opens the stream and keeps it open: after a lost stream it resubscribes, and the new snapshot - * replaces the cache. `onFrame` gets a snapshot first, then that stream's changes. + * Opens the stream once; `onFrame` gets a snapshot first, then that stream's changes. A local + * stream ends only on close, document retirement or a refusal, so nothing reopens it. A remote + * stream that drops reports `lost`; reopening it belongs to the connection's reconnect. */ export function subscribeRuntimeLayout( target: RuntimeClientTarget, @@ -35,8 +34,6 @@ export function subscribeRuntimeLayout( // Bumped whenever a stream is given up, so its late callbacks change nothing. let generation = 0 let handle: { unsubscribe: () => void } | null = null - let retryTimer: ReturnType | null = null - let failures = 0 let state: RuntimeLayoutStreamStatus['state'] = 'connecting' const setStatus = (status: RuntimeLayoutStreamStatus): void => { @@ -52,10 +49,6 @@ export function subscribeRuntimeLayout( const stop = (status: RuntimeLayoutStreamStatus): void => { release() - if (retryTimer !== null) { - clearTimeout(retryTimer) - retryTimer = null - } setStatus(status) } @@ -67,18 +60,7 @@ export function subscribeRuntimeLayout( console.warn('[workspace-layout] layout stream lost:', error) } const refusal = REFUSAL_CODES.find((code) => hasRuntimeRpcErrorCode(error, code)) - if (refusal) { - stop({ state: 'refused', code: refusal }) - return - } - release() - setStatus({ state: 'retrying' }) - const delay = Math.min(RETRY_BASE_MS * 2 ** failures, RETRY_MAX_MS) - failures += 1 - retryTimer = setTimeout(() => { - retryTimer = null - void open() - }, delay) + stop(refusal ? { state: 'refused', code: refusal } : { state: 'lost' }) } const receive = (streamGeneration: number, frame: WorkspaceLayoutStreamFrame | null): void => { @@ -95,9 +77,6 @@ export function subscribeRuntimeLayout( } else if (state !== 'live') { // The host sends the snapshot first; a change before it has nothing to replace. return - } else { - // A change after the snapshot proves the stream stays up, unlike a snapshot then a drop. - failures = 0 } onFrame(frame) } @@ -124,7 +103,7 @@ export function subscribeRuntimeLayout( void open() return { close: () => { - if (state !== 'closed' && state !== 'refused') { + if (state === 'connecting' || state === 'live') { stop({ state: 'closed' }) } } diff --git a/src/shared/workspace-layout/workspace-layout-load.ts b/src/shared/workspace-layout/workspace-layout-load.ts index 70844117c1e..2c0931e3937 100644 --- a/src/shared/workspace-layout/workspace-layout-load.ts +++ b/src/shared/workspace-layout/workspace-layout-load.ts @@ -22,6 +22,7 @@ import { } from './workspace-layout-load-bindings' import { diffStoredSession } from './workspace-layout-load-report' import type { + LoadedWorkspaceLayout, WorkspaceLayoutLoadContext, WorkspaceLayoutLoadResult } from './workspace-layout-load-types' @@ -120,6 +121,16 @@ export function loadWorkspaceLayout( stored: WorkspaceSessionState, context: WorkspaceLayoutLoadContext ): WorkspaceLayoutLoadResult { + const loaded = loadWorkspaceLayoutWithoutReport(hostId, stored, context) + return { ...loaded, changes: diffStoredSession(stored, saveWorkspaceLayout(loaded)) } +} + +/** The same load without the save and diff, for a reader that never writes the result back. */ +export function loadWorkspaceLayoutWithoutReport( + hostId: ExecutionHostId, + stored: WorkspaceSessionState, + context: WorkspaceLayoutLoadContext +): LoadedWorkspaceLayout { // The Loader edits its own copy; the Store's object stays untouched. const session = structuredClone(stored) const desktopView: DesktopLayoutView = { @@ -174,6 +185,5 @@ export function loadWorkspaceLayout( loadLeaves(layout, { session, layouts }, { view: desktopView, facts, carried }, context) unbindDuplicateTerminals(layout) layout = applyLegacySurfaceTombstones(layout, session, carried) - const loaded = { layout, desktopView, facts, carried } - return { ...loaded, changes: diffStoredSession(stored, saveWorkspaceLayout(loaded)) } + return { layout, desktopView, facts, carried } } diff --git a/tests/e2e/workspace-layout-oracle-latency.spec.ts b/tests/e2e/workspace-layout-oracle-latency.spec.ts index 18fd201b0e7..3a690a62761 100644 --- a/tests/e2e/workspace-layout-oracle-latency.spec.ts +++ b/tests/e2e/workspace-layout-oracle-latency.spec.ts @@ -2,8 +2,9 @@ * Measures command → drawn update for layout commands, idle and with three panes flooding output * (`yes`), one of them in a hidden worktree, and time to first drawn after relaunching a large * profile. Measuring only: it records each number in the oracle report and never fails on one, - * so a later budget can gate on it. Runs with no layout subscriber, so main is measured as users - * run it. The committed main baseline is workspace-layout-oracle-latency-baseline.json. + * so a later budget can gate on it. The plain run has no layout subscriber, so main is measured as + * users run it; the subscribed run holds one `layout.subscribe` stream open in the window, as the + * layout client will. The committed main baseline is workspace-layout-oracle-latency-baseline.json. * * Each command is timed twice. The in-app clock (helpers/layout-latency-probe.ts) stops in the * window at React's commit and at the next frame. The outer clock stops when the test sees the @@ -12,7 +13,7 @@ import { mkdirSync, writeFileSync } from 'node:fs' import path from 'node:path' -import type { Page, TestInfo } from '@stablyai/playwright-test' +import type { ElectronApplication, Page, TestInfo } from '@stablyai/playwright-test' import { expect, test } from './helpers/orca-app' import { createRestartSession } from './helpers/orca-restart' import { bootstrapFirstLaunch, seededRepoPathOrSkip } from './helpers/terminal-restart-persistence' @@ -57,6 +58,53 @@ const LARGE_PROFILE = { hiddenWorktrees: 12, tabsPerWorktree: 3, activePanes: 4 type Condition = 'idle' | 'flood' type Sample = { outerMs: number; inApp: InAppSample } type Samples = Record +/** Main's memory at a phase's start and end, in MB; no GC is forced, so read it as a trend. */ +type MainMemory = { heapUsedMb: [number, number]; rssMb: [number, number] } +/** What the window's layout stream received during a phase. */ +type StreamTraffic = { frames: number; kb: number } + +declare global { + // oxlint-disable-next-line typescript-eslint/consistent-type-definitions -- declaration merging requires interface + interface Window { + __layoutStreamTraffic?: { frames: number; bytes: number; errors: number } + } +} + +async function readMainMemory( + app: ElectronApplication +): Promise<{ heapUsed: number; rss: number }> { + return app.evaluate(() => { + const { heapUsed, rss } = process.memoryUsage() + return { heapUsed, rss } + }) +} + +/** Holds one `layout.subscribe` stream open in the window for the rest of the run. */ +async function attachLayoutSubscriber(page: Page): Promise { + await page.evaluate(async () => { + const traffic = { frames: 0, bytes: 0, errors: 0 } + window.__layoutStreamTraffic = traffic + await window.api.runtime.subscribe({ method: 'layout.subscribe', params: {} }, (response) => { + if (!response.ok) { + traffic.errors += 1 + return + } + traffic.frames += 1 + traffic.bytes += JSON.stringify(response.result).length + }) + }) + await expect + .poll(() => page.evaluate(() => window.__layoutStreamTraffic?.frames ?? 0)) + .toBeGreaterThan(0) +} + +async function readStreamTraffic(page: Page): Promise<{ frames: number; bytes: number }> { + const traffic = await page.evaluate(() => window.__layoutStreamTraffic ?? null) + if (traffic && traffic.errors > 0) { + throw new Error(`layout stream reported ${traffic.errors} error frame(s)`) + } + return { frames: traffic?.frames ?? 0, bytes: traffic?.bytes ?? 0 } +} /** Reads can briefly fail while the window re-publishes its terminals; only reads are retried. */ async function read(client: RuntimeClient, method: string, params: unknown): Promise { @@ -243,9 +291,11 @@ type Stats = ReturnType type StartupSample = { firstDrawnMs: number; eventLoopDelayMax: number } type LatencyReport = | { - scenario: 'latency' + scenario: 'latency' | 'latency-subscribed' summary: Record eventLoopDelay: Record + mainMemory: Record + streamTraffic?: Record closeConfirms: Record samples: Samples } @@ -271,9 +321,7 @@ function writeLatencyReport(testInfo: TestInfo, { samples, ...summary }: Latency ) } -// oxlint-disable-next-line no-empty-pattern -- The measurement owns its launch. -test('layout oracle: command to drawn latency', async ({}, testInfo) => { - test.setTimeout(90 * 60_000) +async function measureLatency(testInfo: TestInfo, subscribed: boolean): Promise { const repoPath = seededRepoPathOrSkip() const session = createRestartSession(testInfo) const { app, page } = await session.launch() @@ -284,12 +332,22 @@ test('layout oracle: command to drawn latency', async ({}, testInfo) => { const worktree = `id:${worktreeId}` const client = new RuntimeClient(session.userDataDir, 30_000) await waitForBoundPanes(page, 1) - const phases: Record = { - idle: { from: Date.now(), to: 0, confirms: 0 }, - flood: { from: 0, to: 0, confirms: 0 } + if (subscribed) { + await attachLayoutSubscriber(page) } - phases.idle.confirms = await measureCondition(page, client, worktreeId, 'idle', samples) - phases.idle.to = Date.now() + type Mark = { at: number; heapUsed: number; rss: number; frames: number; bytes: number } + const mark = async (): Promise => ({ + at: Date.now(), + ...(await readMainMemory(app)), + ...(await readStreamTraffic(page)) + }) + type Phase = { from: Mark; to: Mark; confirms: number } + const measurePhase = async (condition: Condition): Promise => { + const from = await mark() + const confirms = await measureCondition(page, client, worktreeId, condition, samples) + return { from, to: await mark(), confirms } + } + const idle = await measurePhase('idle') // Two flooding tabs beside the measured one and one in a worktree that is never shown. const list = await read(client, 'terminal.list', { worktree }) @@ -305,9 +363,7 @@ test('layout oracle: command to drawn latency', async ({}, testInfo) => { } await client.call('terminal.focus', { terminal: measured.handle, navigation: 'host' }) await waitForActiveTerminalManager(page) - phases.flood.from = Date.now() - phases.flood.confirms = await measureCondition(page, client, worktreeId, 'flood', samples) - phases.flood.to = Date.now() + const flood = await measurePhase('flood') const summary = Object.fromEntries( Object.entries(samples).map(([action, values]) => [ @@ -320,18 +376,30 @@ test('layout oracle: command to drawn latency', async ({}, testInfo) => { } ]) ) - const eventLoopDelay = Object.fromEntries( - Object.entries(phases).map(([condition, phase]) => [ - condition, - summarizeEventLoopDelay(session.userDataDir, 'main', phase.from, phase.to) - ]) - ) - const closeConfirms = { idle: phases.idle.confirms, flood: phases.flood.confirms } + const perCondition = (read: (phase: Phase) => T): Record => ({ + idle: read(idle), + flood: read(flood) + }) + const mb = (bytes: number): number => Math.round(bytes / 1024 / 102.4) / 10 writeLatencyReport(testInfo, { - scenario: 'latency', + scenario: subscribed ? 'latency-subscribed' : 'latency', summary, - eventLoopDelay, - closeConfirms, + eventLoopDelay: perCondition(({ from, to }) => + summarizeEventLoopDelay(session.userDataDir, 'main', from.at, to.at) + ), + mainMemory: perCondition(({ from, to }) => ({ + heapUsedMb: [mb(from.heapUsed), mb(to.heapUsed)], + rssMb: [mb(from.rss), mb(to.rss)] + })), + ...(subscribed + ? { + streamTraffic: perCondition(({ from, to }) => ({ + frames: to.frames - from.frames, + kb: Math.round((to.bytes - from.bytes) / 1024) + })) + } + : {}), + closeConfirms: perCondition(({ confirms }) => confirms), samples }) } finally { @@ -341,6 +409,18 @@ test('layout oracle: command to drawn latency', async ({}, testInfo) => { await session.close(app) await session.dispose() } +} + +// oxlint-disable-next-line no-empty-pattern -- The measurement owns its launch. +test('layout oracle: command to drawn latency', async ({}, testInfo) => { + test.setTimeout(90 * 60_000) + await measureLatency(testInfo, false) +}) + +// oxlint-disable-next-line no-empty-pattern -- The measurement owns its launch. +test('layout oracle: command to drawn latency with a layout subscriber', async ({}, testInfo) => { + test.setTimeout(90 * 60_000) + await measureLatency(testInfo, true) }) // oxlint-disable-next-line no-empty-pattern -- The measurement owns its launches.