diff --git a/config/tsconfig.tc.web.json b/config/tsconfig.tc.web.json index a534df96414..442cca7e353 100644 --- a/config/tsconfig.tc.web.json +++ b/config/tsconfig.tc.web.json @@ -28,6 +28,7 @@ "../src/main/gitlab/mappers.ts", "../src/main/runtime/sidebar-inputs/sidebar-clock.ts", "../src/main/runtime/sidebar-inputs/sidebar-temporal-owner.ts", + "../src/main/runtime/sidebar-inputs/producer-revision.ts", "../src/main/jira/adf-markdown.ts", "../src/main/jira/adf-text-marks.ts", "../src/main/jira/adf-media-destination.ts", diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 8a04e231a8a..70a259b50ad 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -41,6 +41,9 @@ import { RepoIconDataUrls } from './sidebar-inputs/repo-icon-data-urls' import { getMainHttpClient } from '../network/http-client' import { getRuntimeDesktopSurface } from './runtime-desktop-surface' import { CollapseWriteOrder } from './workspace-view-collapse-writes' +import { SidebarInputsFeed } from './sidebar-inputs/sidebar-inputs-feed' +import { sidebarInputsFeedDomains } from './sidebar-inputs/sidebar-inputs-feed-domains' +import { requireSwitchedOn } from './sidebar-inputs/require-switched-on' import { HEADLESS_RUNTIME_WINDOW_ID } from '../../shared/runtime-session-contracts' import { getHostDisplayLabelOverrides } from '../../shared/host-setting-overrides' @@ -55,13 +58,14 @@ class OrcaRuntimeService extends OrcaRuntimeSidebarBase { get sidebarFacts() { return sidebarFactsRuntime(this) } - // Switch point. Idle until relay hosts attach; the fallback refresh runs while a window shows the sidebar. + // Switch point. Idle until relay hosts attach; the fallback refresh runs while a window shows the sidebar or a phone reads the feed. readonly pairedServerRows = ORCA_SIDEBAR_INPUTS_IS_MAIN ? new PairedServerRowsProducer({ hasDemand: () => - this.authoritativeWindowId !== null && - this.authoritativeWindowId !== HEADLESS_RUNTIME_WINDOW_ID && - this.store?.getUI?.().sidebarOpen !== false, + this.sidebarInputsFeed?.subscribed === true || + (this.authoritativeWindowId !== null && + this.authoritativeWindowId !== HEADLESS_RUNTIME_WINDOW_ID && + this.store?.getUI?.().sidebarOpen !== false), hostLabelOverrides: () => getHostDisplayLabelOverrides(this.getClientSettings()) }) : null @@ -81,6 +85,20 @@ class OrcaRuntimeService extends OrcaRuntimeSidebarBase { : null // Switch point: the per-writer ordering behind ui.setCollapsed. readonly collapseWriteOrder = ORCA_SIDEBAR_INPUTS_IS_MAIN ? new CollapseWriteOrder() : null + // Switch point: the phone's feed of the producers above; idle until this host's own phone reads. + readonly sidebarInputsFeed = ORCA_SIDEBAR_INPUTS_IS_MAIN + ? new SidebarInputsFeed( + () => + sidebarInputsFeedDomains({ + facts: requireSwitchedOn(this.sidebarFacts), + pairedServerRows: requireSwitchedOn(this.pairedServerRows), + workspaceCreations: requireSwitchedOn(this.workspaceCreations), + store: () => this.store, + onClientEvent: (listener, options) => this.onClientEvent(listener, options) + }), + (event) => this.emitClientEvent(event) + ) + : null private async selectReferenceCatalogWorkspace(selector: string, cwd?: string) { const catalog = listReferenceWorkspaces( diff --git a/src/main/runtime/rpc/methods/sidebar-inputs-read.test.ts b/src/main/runtime/rpc/methods/sidebar-inputs-read.test.ts new file mode 100644 index 00000000000..ab934698908 --- /dev/null +++ b/src/main/runtime/rpc/methods/sidebar-inputs-read.test.ts @@ -0,0 +1,95 @@ +import '../unused-default-rpc-methods.test-fixture' +import { describe, expect, it } from 'vitest' +import { MOBILE_BRIDGE_MAX_FRAME_BYTES } from '../../../../shared/mobile-bridge-frame-bytes' +import { SIDEBAR_INPUTS_PAGE_MAX_BYTES } from '../../../../shared/sidebar-list/sidebar-inputs-feed-contract' +import type { OrcaRuntimeService } from '../../orca-runtime' +import { SidebarInputsFeed } from '../../sidebar-inputs/sidebar-inputs-feed' +import { testDomain } from '../../sidebar-inputs/sidebar-inputs-feed.test-fixture' +import type { RpcRequest } from '../core' +import { RpcDispatcher } from '../dispatcher' +import type { RpcDispatchStreamingOptions } from '../dispatcher-stream-options' +import { SIDEBAR_INPUT_METHODS } from './sidebar-inputs' + +function makeHost() { + // Rows far larger than one page, so the first reply is a full page. + const rows = testDomain( + 'rows', + Object.fromEntries( + Array.from({ length: 3_000 }, (_, index) => [`row-${index}`, 'r'.padEnd(300, '中')]) + ) + ) + const feed = new SidebarInputsFeed( + () => [rows.domain], + () => {} + ) + const cleanups = new Map void>() + const fake = { + getRuntimeId: () => 'test-runtime', + sidebarInputsFeed: feed, + registerSubscriptionCleanup: (id: string, cleanup: () => void) => cleanups.set(id, cleanup) + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: sidebarInputs.read reads only the members above. + const runtime = fake as unknown as OrcaRuntimeService + const dispatcher = new RpcDispatcher({ runtime, methods: SIDEBAR_INPUT_METHODS }) + let requestId = 0 + const read = (context: RpcDispatchStreamingOptions) => { + const request: RpcRequest = { + id: `req-${++requestId}`, + authToken: 'tok', + method: 'sidebarInputs.read', + params: { maxBytes: SIDEBAR_INPUTS_PAGE_MAX_BYTES } + } + return dispatcher.dispatch(request, context) + } + return { rows, read, cleanups } +} + +describe('sidebarInputs.read', () => { + it('serves the host’s own phone a page that fits its bridge frame with the RPC envelope', async () => { + const host = makeHost() + const response = await host.read({ clientKind: 'mobile', connectionId: 'conn-1' }) + expect(response).toMatchObject({ + ok: true, + result: { kind: 'snapshot', next: expect.any(String) } + }) + const frame = Buffer.byteLength(JSON.stringify(response), 'utf8') + expect(frame).toBeGreaterThan(SIDEBAR_INPUTS_PAGE_MAX_BYTES * 0.9) + expect(frame).toBeLessThanOrEqual(MOBILE_BRIDGE_MAX_FRAME_BYTES) + }) + + it('refuses a delegated phone, a paired desktop and a caller without a connection', async () => { + const host = makeHost() + for (const context of [ + { clientKind: 'mobile', connectionId: 'conn-1', pairedDeviceParentId: 'desktop-device' }, + { clientKind: 'runtime', connectionId: 'conn-1' }, + { clientKind: 'mobile' } + ] satisfies RpcDispatchStreamingOptions[]) { + expect(await host.read(context)).toMatchObject({ ok: false, error: { code: 'forbidden' } }) + } + expect(host.rows.reads()).toBe(0) + }) + + it('subscribes a connection on its first read and releases it when the connection closes', async () => { + const host = makeHost() + await host.read({ clientKind: 'mobile', connectionId: 'conn-1' }) + await host.read({ clientKind: 'mobile', connectionId: 'conn-1' }) + expect([...host.cleanups.keys()]).toEqual(['sidebar-inputs-feed-conn-1']) + expect(host.rows.watchers()).toBe(1) + host.cleanups.get('sidebar-inputs-feed-conn-1')?.() + expect(host.rows.watchers()).toBe(0) + }) + + it('does not subscribe a connection that closed before its read ran', async () => { + const host = makeHost() + const closed = new AbortController() + closed.abort() + const response = await host.read({ + clientKind: 'mobile', + connectionId: 'conn-1', + signal: closed.signal + }) + expect(response).toMatchObject({ ok: false }) + expect(host.cleanups.size).toBe(0) + expect(host.rows.watchers()).toBe(0) + }) +}) diff --git a/src/main/runtime/rpc/methods/sidebar-inputs.ts b/src/main/runtime/rpc/methods/sidebar-inputs.ts index e59ced9849a..35e161ad128 100644 --- a/src/main/runtime/rpc/methods/sidebar-inputs.ts +++ b/src/main/runtime/rpc/methods/sidebar-inputs.ts @@ -1,19 +1,14 @@ import { defineMethod } from '../core' -import { getRegisteredSshState, listRegisteredSshTargets } from '../../../ssh/ssh-target-registry' -import { getWslHostService } from '../../../wsl/wsl-host-service-instance' -import { readDesktopSidebarFacts } from '../../sidebar-inputs/desktop-sidebar-facts' +import { requireSwitchedOn } from '../../sidebar-inputs/require-switched-on' +import { readRuntimeDesktopSidebarFacts } from '../../sidebar-inputs/sidebar-inputs-feed-domains' +import { isHostOwnPhone } from '../host-own-phone' +import { SidebarInputsReadParams } from '../../../../shared/sidebar-list/sidebar-inputs-feed-contract' import { SidebarSessionBeginParams } from '../../../../shared/rpc-contract/sidebar-session-params' import { RepoIconDataUrlsParams } from '../../../../shared/sidebar-list/repo-icon-data-urls-contract' import { SetCollapsedParams } from '../../../../shared/rpc-contract/workspace-view-collapse-params' import { setCollapsedSection } from '../../workspace-view-collapse-writes' // Registered only with ORCA_SIDEBAR_INPUTS=main, so the producers below always exist. -function requireSwitchedOn(producer: Producer | null): Producer { - if (producer === null) { - throw new Error('sidebar inputs are off in this build') - } - return producer -} export const SIDEBAR_INPUT_METHODS = [ defineMethod({ @@ -33,13 +28,7 @@ export const SIDEBAR_INPUT_METHODS = [ name: 'sidebarInputs.desktopFacts', permission: 'workspace', params: null, - handler: async (_params, { runtime }) => - readDesktopSidebarFacts({ - store: runtime.requireStore(), - sshTargets: listRegisteredSshTargets(), - sshStateOf: getRegisteredSshState, - wslStatuses: getWslHostService()?.getStatuses() ?? [] - }) + handler: async (_params, { runtime }) => readRuntimeDesktopSidebarFacts(runtime.requireStore()) }), defineMethod({ name: 'sidebarInputs.pairedServerRows', @@ -61,6 +50,35 @@ export const SIDEBAR_INPUT_METHODS = [ handler: async (_params, { runtime }) => requireSwitchedOn(runtime.sidebarFacts).getSidebarTemporalSnapshot() }), + defineMethod({ + name: 'sidebarInputs.read', + permission: 'workspace', + params: SidebarInputsReadParams, + // The feed is this host's own view. A delegated phone lists through its own desktop's feed, + // which carries this host's rows; other desktops and web clients build their own view. + handler: async (params, context) => { + const feed = requireSwitchedOn(context.runtime.sidebarInputsFeed) + const { connectionId } = context + if (!isHostOwnPhone(context) || connectionId === undefined) { + throw Object.assign(new Error('Only this host’s own phone reads its sidebar feed.'), { + code: 'forbidden' + }) + } + // A connection already closed has run its cleanups; subscribing it now would never release. + if (context.signal?.aborted) { + throw new Error('client_disconnected') + } + // Subscribed from a connection's first read until it closes. + if (feed.acquire(connectionId)) { + context.runtime.registerSubscriptionCleanup( + `sidebar-inputs-feed-${connectionId}`, + () => feed.release(connectionId), + connectionId + ) + } + return feed.read(params) + } + }), defineMethod({ name: 'sidebarSession.begin', permission: 'workspace', diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-paired-desktop-methods.ts b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-paired-desktop-methods.ts index 176f138093a..1b2062024c2 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-paired-desktop-methods.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-paired-desktop-methods.ts @@ -108,6 +108,6 @@ export const PAIRED_DESKTOP_METHODS = [ 'settings.updateTerminalQuickCommands', 'terminal.getAutoRestoreFit', 'terminal.setAutoRestoreFit', - // Switch point: the per-section collapse write exists only with ORCA_SIDEBAR_INPUTS=main. - ...(ORCA_SIDEBAR_INPUTS_IS_MAIN ? (['ui.setCollapsed'] as const) : []) + // Switch point: the per-section collapse write and the sidebar feed exist only with ORCA_SIDEBAR_INPUTS=main. + ...(ORCA_SIDEBAR_INPUTS_IS_MAIN ? (['ui.setCollapsed', 'sidebarInputs.read'] as const) : []) ] as const diff --git a/src/main/runtime/sidebar-inputs/orca-runtime-sidebar-inputs-feed.test.ts b/src/main/runtime/sidebar-inputs/orca-runtime-sidebar-inputs-feed.test.ts new file mode 100644 index 00000000000..236d6445d55 --- /dev/null +++ b/src/main/runtime/sidebar-inputs/orca-runtime-sidebar-inputs-feed.test.ts @@ -0,0 +1,54 @@ +// Runs on plain Node, as orcad does: the feed reads the runtime's own producers. +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { RuntimeClientEvent } from '../../../shared/runtime-client-events' +import { OrcaRuntimeService } from '../orca-runtime' +import { requireSwitchedOn } from './require-switched-on' +import { TestPhone } from './sidebar-inputs-feed.test-fixture' + +vi.mock('../../../shared/sidebar-inputs-switch', () => ({ ORCA_SIDEBAR_INPUTS_IS_MAIN: true })) + +beforeEach(() => { + vi.useFakeTimers() + vi.spyOn(console, 'warn').mockImplementation(() => {}) +}) + +afterEach(() => { + vi.useRealTimers() + vi.restoreAllMocks() +}) + +describe('the runtime sidebar inputs feed', () => { + it('publishes the producers’ snapshots and announces a change on the client-event bus', async () => { + const runtime = new OrcaRuntimeService(null) + const feed = requireSwitchedOn(runtime.sidebarInputsFeed) + const events: RuntimeClientEvent[] = [] + runtime.onClientEvent((event) => events.push(event)) + const phone = new TestPhone(feed, 'conn-1') + phone.refresh() + await phone.settle() + expect(phone.view().temporal).toEqual({ sessionId: null, sortSnapshot: null, retention: null }) + + const creation = requireSwitchedOn(runtime.workspaceCreations).open({ + creationId: 'c1', + repoId: 'repo-1' + }) + await vi.advanceTimersByTimeAsync(60) + const notice = events.findLast((event) => event.type === 'sidebarInputsChanged') + expect(notice).toEqual({ + type: 'sidebarInputsChanged', + generation: feed.generation, + revision: expect.any(Number) + }) + if (notice?.type !== 'sidebarInputsChanged') { + throw new Error('no change notice') + } + phone.notice(notice) + await phone.settle() + expect(phone.replies.at(-1)).toMatchObject({ + kind: 'delta', + domains: { creations: { upserts: { c1: { repoId: 'repo-1' } } } } + }) + creation.close() + phone.disconnect() + }) +}) diff --git a/src/main/runtime/sidebar-inputs/orca-runtime-sidebar-temporal.ts b/src/main/runtime/sidebar-inputs/orca-runtime-sidebar-temporal.ts index c2963df9877..d2ef7946c4b 100644 --- a/src/main/runtime/sidebar-inputs/orca-runtime-sidebar-temporal.ts +++ b/src/main/runtime/sidebar-inputs/orca-runtime-sidebar-temporal.ts @@ -34,7 +34,7 @@ export abstract class OrcaRuntimeWithSidebarTemporal extends OrcaRuntimeWithWork constructor(...args: ConstructorParameters) { super(...args) - this.workspaceActivityFacts.onChange(() => this.sidebarTemporal.noteActivityChange()) + this.workspaceActivityFacts.revision.onChange(() => this.sidebarTemporal.noteActivityChange()) this.store?.onUIChanged?.(() => this.sidebarTemporal.noteViewChange()) // Every catalog change, from IPC or the runtime, reaches clients through this one bus. this.clientEvents.on( @@ -68,6 +68,10 @@ export abstract class OrcaRuntimeWithSidebarTemporal extends OrcaRuntimeWithWork return this.sidebarTemporal.snapshot() } + onSidebarTemporalChange(listener: () => void): () => void { + return this.sidebarTemporal.revision.onChange(listener) + } + beginSidebarListSession(sessionId: string): void { this.sidebarTemporal.beginSession(sessionId) } diff --git a/src/main/runtime/sidebar-inputs/paired-server-rows-producer.ts b/src/main/runtime/sidebar-inputs/paired-server-rows-producer.ts index d788d282d95..567e080c26b 100644 --- a/src/main/runtime/sidebar-inputs/paired-server-rows-producer.ts +++ b/src/main/runtime/sidebar-inputs/paired-server-rows-producer.ts @@ -1,3 +1,4 @@ +import { ProducerRevision } from './producer-revision' import { toRuntimeExecutionHostId, type ExecutionHostId } from '../../../shared/execution-host' import { describePairedServerHosts } from '../../../shared/paired-server-hosts' import type { MobileRelayServerWorktreeRow } from '../../../shared/mobile-relay-hosts-contract' @@ -49,7 +50,7 @@ type ServerEntry = { export class PairedServerRowsProducer implements ServerWorktreeRows { private readonly servers = new Map() private hosts: MobileDesktopRelayHosts | null = null - private revision = 0 + readonly revision = new ProducerRevision() constructor( private readonly options: { @@ -82,7 +83,7 @@ export class PairedServerRowsProducer implements ServerWorktreeRows { } this.servers.clear() this.hosts = null - this.revision += 1 + this.revision.bump() } } @@ -99,7 +100,7 @@ export class PairedServerRowsProducer implements ServerWorktreeRows { if (fences.get(environmentId) !== entry.fence) { entry.watch?.close() this.servers.delete(environmentId) - this.revision += 1 + this.revision.bump() } } // Why the shown hosts: a managed server reached over its SSH host's relay is not in contact here. @@ -117,7 +118,7 @@ export class PairedServerRowsProducer implements ServerWorktreeRows { refreshing: null, refreshAgain: false }) - this.revision += 1 + this.revision.bump() } } for (const entry of this.servers.values()) { @@ -125,7 +126,7 @@ export class PairedServerRowsProducer implements ServerWorktreeRows { if (inContact !== entry.inContact) { entry.inContact = inContact entry.confirmed = false - this.revision += 1 + this.revision.bump() } // Why current(): a replaced transport can end the stream without telling it. if (inContact && !entry.watch?.current()) { @@ -173,7 +174,7 @@ export class PairedServerRowsProducer implements ServerWorktreeRows { ]) ) return { - revision: this.revision, + revision: this.revision.value, servers: [...this.servers.values()].map((entry) => { const hostId = toRuntimeExecutionHostId(entry.environmentId) return { @@ -209,7 +210,7 @@ export class PairedServerRowsProducer implements ServerWorktreeRows { private unconfirm(entry: ServerEntry): void { if (entry.confirmed) { entry.confirmed = false - this.revision += 1 + this.revision.bump() } } @@ -245,6 +246,6 @@ export class PairedServerRowsProducer implements ServerWorktreeRows { entry.rows = stampServerWorktreeRows(entry.environmentId, reply.data.worktrees) entry.confirmed = entry.inContact // Every answer moves the revision: rows carry activity that changes without a catalog event. - this.revision += 1 + this.revision.bump() } } diff --git a/src/main/runtime/sidebar-inputs/producer-revision.ts b/src/main/runtime/sidebar-inputs/producer-revision.ts new file mode 100644 index 00000000000..4fd0deebe4d --- /dev/null +++ b/src/main/runtime/sidebar-inputs/producer-revision.ts @@ -0,0 +1,23 @@ +/** A producer's revision. Listeners hear each bump synchronously, possibly mid-change, so they read later. */ +export class ProducerRevision { + private count = 0 + private readonly listeners = new Set<() => void>() + + get value(): number { + return this.count + } + + bump(): void { + this.count += 1 + for (const listener of this.listeners) { + listener() + } + } + + onChange(listener: () => void): () => void { + this.listeners.add(listener) + return () => { + this.listeners.delete(listener) + } + } +} diff --git a/src/main/runtime/sidebar-inputs/require-switched-on.ts b/src/main/runtime/sidebar-inputs/require-switched-on.ts new file mode 100644 index 00000000000..25fabe29819 --- /dev/null +++ b/src/main/runtime/sidebar-inputs/require-switched-on.ts @@ -0,0 +1,7 @@ +/** A producer that exists only with ORCA_SIDEBAR_INPUTS=main, read where that build is certain. */ +export function requireSwitchedOn(producer: Producer | null): Producer { + if (producer === null) { + throw new Error('sidebar inputs are off in this build') + } + return producer +} diff --git a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-domains.ts b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-domains.ts new file mode 100644 index 00000000000..0fa23c6f561 --- /dev/null +++ b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-domains.ts @@ -0,0 +1,124 @@ +import { getRegisteredSshState, listRegisteredSshTargets } from '../../ssh/ssh-target-registry' +import { getWslHostService } from '../../wsl/wsl-host-service-instance' +import type { RuntimeClientEvent } from '../../../shared/runtime-client-events' +import type { DesktopSidebarFacts } from '../../../shared/sidebar-list/desktop-sidebar-facts' +import type { RuntimeStore } from '../runtime-store-contract' +import type { OrcaRuntimeWithSidebarTemporal } from './orca-runtime-sidebar-temporal' +import type { PairedServerRowsProducer } from './paired-server-rows-producer' +import type { WorkspaceCreationRegistry } from './workspace-creation-registry' +import type { SidebarInputsFeedDomain } from './sidebar-inputs-feed' +import { readDesktopSidebarFacts } from './desktop-sidebar-facts' + +/** This host's store and connection registries, read the way sidebarInputs.desktopFacts reads them. */ +export function readRuntimeDesktopSidebarFacts(store: RuntimeStore): DesktopSidebarFacts { + return readDesktopSidebarFacts({ + store, + sshTargets: listRegisteredSshTargets(), + sshStateOf: getRegisteredSshState, + wslStatuses: getWslHostService()?.getStatuses() ?? [] + }) +} + +// The bus every catalog, view and SSH change reaches clients through. +const CATALOG_EVENTS = new Set([ + 'worktreesChanged', + 'reposChanged', + 'workspaceViewChanged', + 'sshStateChanged' +]) + +export type SidebarInputsFeedSources = { + facts: OrcaRuntimeWithSidebarTemporal + pairedServerRows: PairedServerRowsProducer + workspaceCreations: WorkspaceCreationRegistry + store: () => RuntimeStore | null + onClientEvent: ( + listener: (event: RuntimeClientEvent) => void, + options: { consumesTerminalSideEffects: false } + ) => () => void +} + +/** Every merged producer's snapshot, keyed per entity: a change re-sends only what it touched. */ +export function sidebarInputsFeedDomains( + sources: SidebarInputsFeedSources +): SidebarInputsFeedDomain[] { + const { facts, pairedServerRows, workspaceCreations } = sources + const watchCatalog = (changed: () => void): (() => void) => + sources.onClientEvent( + (event) => { + if (CATALOG_EVENTS.has(event.type)) { + changed() + } + }, + { consumesTerminalSideEffects: false } + ) + const requireStore = (): RuntimeStore => { + const store = sources.store() + if (!store) { + throw new Error('no store') + } + return store + } + return [ + { + name: 'activity', + read: () => facts.getWorkspaceActivityFacts().facts, + watch: (changed) => facts.workspaceActivityFacts.revision.onChange(changed) + }, + { + name: 'temporal', + read: () => { + const { sessionId, sortSnapshot, retention } = facts.getSidebarTemporalSnapshot() + return { sessionId, sortSnapshot, retention } + }, + watch: (changed) => facts.onSidebarTemporalChange(changed) + }, + { + name: 'creations', + read: () => workspaceCreations.snapshot().creations, + watch: (changed) => workspaceCreations.revision.onChange(changed) + }, + { + name: 'pairedServerHosts', + read: () => + Object.fromEntries( + pairedServerRows + .snapshot() + .servers.map(({ hostId, freshness, host }) => [hostId, { freshness, host }]) + ), + watch: (changed) => pairedServerRows.revision.onChange(changed) + }, + { + name: 'pairedServerRows', + read: () => + Object.fromEntries( + pairedServerRows + .snapshot() + .servers.flatMap(({ rows }) => rows.map((row) => [row.key, row])) + ), + watch: (changed) => pairedServerRows.revision.onChange(changed) + }, + { + name: 'desktop', + // Each field its own entity; the WSL registry announces no change. + read: () => readRuntimeDesktopSidebarFacts(requireStore()), + watch: (changed) => { + const store = sources.store() + const detaches = [ + watchCatalog(changed), + store?.onUIChanged?.(changed), + store?.onWorkspaceSessionWritten?.(changed) + ] + return () => detaches.forEach((detach) => detach?.()) + }, + rereadOnRead: true + }, + { + name: 'review', + read: async () => (await facts.getWorkspaceReviewFacts()).facts, + // The window's PR cache saves announce no change. + watch: watchCatalog, + rereadOnRead: true + } + ] +} diff --git a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-log.ts b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-log.ts new file mode 100644 index 00000000000..07141c3a89e --- /dev/null +++ b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-log.ts @@ -0,0 +1,62 @@ +import type { FeedChange, FeedEntity } from './sidebar-inputs-feed-pages' + +export const FEED_LOG_MAX_REVISIONS = 256 +export const FEED_LOG_MAX_BYTES = 2 * 1024 * 1024 + +/** Per domain, each changed key's new entity, or null when it was removed. */ +export type FeedRevisionChanges = Map> + +type LogEntry = { revision: number; changes: FeedRevisionChanges; bytes: number } + +/** The last revisions' changes, bounded by count and bytes; older cursors read a snapshot. */ +export class FeedChangeLog { + private readonly entries: LogEntry[] = [] + private bytes = 0 + + /** `floor` is the revision the log starts from: the oldest cursor it can bring current. */ + constructor(private floor: number) {} + + covers(cursor: number, revision: number): boolean { + return cursor >= this.floor && cursor <= revision + } + + append(revision: number, changes: FeedRevisionChanges): void { + let bytes = 0 + for (const changed of changes.values()) { + for (const [key, entity] of changed) { + bytes += key.length + (entity?.bytes ?? 0) + } + } + this.entries.push({ revision, changes, bytes }) + this.bytes += bytes + while (this.entries.length > FEED_LOG_MAX_REVISIONS || this.bytes > FEED_LOG_MAX_BYTES) { + const dropped = this.entries.shift() + if (!dropped) { + break + } + this.bytes -= dropped.bytes + this.floor = dropped.revision + } + } + + /** Every key changed after `cursor`, in its latest form. */ + *since(cursor: number): Iterable { + const merged: FeedRevisionChanges = new Map() + for (const entry of this.entries) { + if (entry.revision > cursor) { + for (const [name, changed] of entry.changes) { + const into = merged.get(name) ?? new Map() + merged.set(name, into) + for (const [key, entity] of changed) { + into.set(key, entity) + } + } + } + } + for (const [name, changed] of merged) { + for (const [key, entity] of changed) { + yield [name, key, entity] + } + } + } +} diff --git a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-pages.test.ts b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-pages.test.ts new file mode 100644 index 00000000000..b34ce5b05c7 --- /dev/null +++ b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-pages.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from 'vitest' +import { + SIDEBAR_INPUTS_PAGE_MAX_BYTES, + SIDEBAR_INPUTS_PAGE_MIN_BYTES, + SIDEBAR_INPUTS_SCHEMA_VERSION, + SIDEBAR_INPUTS_SEMANTICS_VERSION +} from '../../../shared/sidebar-list/sidebar-inputs-feed-contract' +import { + feedEntityFromJson, + paginateFeedChanges, + type FeedChange +} from './sidebar-inputs-feed-pages' + +const feedEntity = (value: unknown) => feedEntityFromJson(JSON.stringify(value)) + +// Worst cases for escaping: quotes, backslashes, 3-byte and 4-byte characters. +const NASTY = '"\\é中\u{1f600}
' + +function encodedPages(changes: FeedChange[], budget: number) { + return paginateFeedChanges(changes, budget).map((body, index, bodies) => + JSON.stringify({ + generation: '00000000-0000-4000-8000-000000000000', + revision: Number.MAX_SAFE_INTEGER, + schemaVersion: SIDEBAR_INPUTS_SCHEMA_VERSION, + semanticsVersion: SIDEBAR_INPUTS_SEMANTICS_VERSION, + kind: 'snapshot', + from: Number.MAX_SAFE_INTEGER, + domains: body.domains, + next: index + 1 < bodies.length ? `${'f'.repeat(36)}.${index + 1}` : null + }) + ) +} + +describe('paginateFeedChanges', () => { + for (const budget of [SIDEBAR_INPUTS_PAGE_MIN_BYTES, SIDEBAR_INPUTS_PAGE_MAX_BYTES]) { + it(`keeps every encoded page within ${budget} bytes and rejoins oversized entities`, () => { + const changes: FeedChange[] = [ + ['rows', 'small', feedEntity({ name: NASTY })], + ['rows', `huge${NASTY}`, feedEntity({ text: NASTY.repeat(budget / 4) })], + ...Array.from({ length: 500 }, (_, index): FeedChange => [ + index % 2 ? 'rows' : `domain${NASTY}`, + `key-${index}${NASTY}`, + index % 7 ? feedEntity({ index, text: NASTY.repeat(index) }) : null + ]) + ] + const pages = encodedPages(changes, budget) + expect(pages.length).toBeGreaterThan(2) + for (const page of pages) { + expect(Buffer.byteLength(page, 'utf8')).toBeLessThanOrEqual(budget) + } + const joined = pages + .map((page) => JSON.parse(page).domains.rows?.parts?.[`huge${NASTY}`] ?? '') + .join('') + expect(JSON.parse(joined)).toEqual({ text: NASTY.repeat(budget / 4) }) + }) + } + + it('sends one empty page for no changes', () => { + expect(paginateFeedChanges([], SIDEBAR_INPUTS_PAGE_MAX_BYTES)).toEqual([ + { domains: {}, bytes: 0 } + ]) + }) +}) diff --git a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-pages.ts b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-pages.ts new file mode 100644 index 00000000000..3bfb40771d5 --- /dev/null +++ b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-pages.ts @@ -0,0 +1,100 @@ +import type { SidebarInputsDomainChange } from '../../../shared/sidebar-list/sidebar-inputs-feed-contract' + +/** One published entity: its value and that value's JSON, which the feed diffs and measures. */ +export type FeedEntity = { value: unknown; json: string; bytes: number } + +/** An entity's change: its new form, or null when it was removed. */ +export type FeedChange = readonly [domain: string, key: string, entity: FeedEntity | null] + +export type FeedPageBody = { domains: Record; bytes: number } + +/** Room left in every page for the reply's other fields; a page token is under 100 bytes. */ +export const FEED_PAGE_ENVELOPE_BYTES = 512 +// `"name":{"upserts":{},"removals":[],"parts":{}},` less the name. +const DOMAIN_SHELL_BYTES = 44 +// Each entry's quotes, colon and comma. +const ENTRY_SHELL_BYTES = 4 +// JSON text holds no raw control characters, so a code unit escapes to at most 3 UTF-8 bytes. +const MAX_ESCAPED_BYTES_PER_CODE_UNIT = 3 + +export function feedEntityFromJson(json: string): FeedEntity { + return { value: JSON.parse(json), json, bytes: Buffer.byteLength(json, 'utf8') } +} + +function jsonBytes(text: string): number { + return Buffer.byteLength(JSON.stringify(text), 'utf8') +} + +/** + * Splits changes into pages of at most `budgetBytes` encoded, envelope included. An entity too + * large for any page travels as JSON text parts, each filling its own page. + */ +export function paginateFeedChanges(changes: Iterable, budgetBytes: number) { + const room = budgetBytes - FEED_PAGE_ENVELOPE_BYTES + const pages: FeedPageBody[] = [] + let page: FeedPageBody = { domains: {}, bytes: 0 } + const domainOf = (name: string): { change: SidebarInputsDomainChange; cost: number } => { + const existing = page.domains[name] + return existing + ? { change: existing, cost: 0 } + : { + change: { upserts: {}, removals: [], parts: {} }, + cost: jsonBytes(name) + DOMAIN_SHELL_BYTES + } + } + const place = ( + name: string, + cost: number, + write: (change: SidebarInputsDomainChange) => void + ) => { + let domain = domainOf(name) + if (page.bytes + domain.cost + cost > room && page.bytes > 0) { + pages.push(page) + page = { domains: {}, bytes: 0 } + domain = domainOf(name) + } + page.domains[name] = domain.change + page.bytes += domain.cost + cost + write(domain.change) + } + for (const [name, key, entity] of changes) { + const keyBytes = jsonBytes(key) + ENTRY_SHELL_BYTES + if (entity === null) { + place(name, keyBytes, (change) => change.removals.push(key)) + } else if (keyBytes + entity.bytes + jsonBytes(name) + DOMAIN_SHELL_BYTES <= room) { + place(name, keyBytes + entity.bytes, (change) => { + change.upserts[key] = entity.value + }) + } else { + const partRoom = room - keyBytes - jsonBytes(name) - DOMAIN_SHELL_BYTES - 2 + for (const part of splitJsonText(entity.json, partRoom)) { + place(name, keyBytes + jsonBytes(part), (change) => { + change.parts[key] = part + }) + // Why: a page holds one part of an entity, so the reader joins parts in page order. + pages.push(page) + page = { domains: {}, bytes: 0 } + } + } + } + if (page.bytes > 0 || pages.length === 0) { + pages.push(page) + } + return pages +} + +/** Cuts text into pieces whose JSON string form fits `maxBytes`, never inside a surrogate pair. */ +function splitJsonText(text: string, maxBytes: number): string[] { + const units = Math.max(2, Math.floor(maxBytes / MAX_ESCAPED_BYTES_PER_CODE_UNIT)) + const parts: string[] = [] + for (let start = 0; start < text.length;) { + let end = Math.min(text.length, start + units) + const last = text.charCodeAt(end - 1) + if (end < text.length && last >= 0xd800 && last <= 0xdbff) { + end -= 1 + } + parts.push(text.slice(start, end)) + start = end + } + return parts +} diff --git a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-pins.ts b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-pins.ts new file mode 100644 index 00000000000..7fcc46e0ec5 --- /dev/null +++ b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-pins.ts @@ -0,0 +1,63 @@ +import type { SidebarInputsReadResult } from '../../../shared/sidebar-list/sidebar-inputs-feed-contract' + +/** A multi-page batch stays readable this long after its last page read. */ +export const FEED_PIN_IDLE_MS = 30_000 +export const FEED_PIN_MAX_BYTES = 16 * 1024 * 1024 +export const FEED_PIN_MAX_BATCHES = 8 + +type Pin = { pages: readonly SidebarInputsReadResult[]; bytes: number; expiresAt: number } + +/** + * Batches whose later pages a reader fetches by token, unchanged and shared by every reader. + * They stay within their bounds, except one batch larger than them all, pinned alone. + */ +export class FeedPins { + /** By batch key, oldest first. */ + private readonly pins = new Map() + + /** The batch pinned under `key`, if still readable. */ + find(key: string): readonly SidebarInputsReadResult[] | null { + const pin = this.live(key) + return pin ? pin.pages : null + } + + /** The page a token names; null once its batch expired or was evicted. */ + page(token: string): SidebarInputsReadResult | null { + const split = token.lastIndexOf('#') + return this.live(token.slice(0, split))?.pages[Number(token.slice(split + 1))] ?? null + } + + add(key: string, pages: readonly SidebarInputsReadResult[], bytes: number): void { + const now = Date.now() + let total = bytes + for (const [existing, pin] of this.pins) { + if (pin.expiresAt <= now) { + this.pins.delete(existing) + } else { + total += pin.bytes + } + } + for (const [existing, pin] of this.pins) { + if (total <= FEED_PIN_MAX_BYTES && this.pins.size < FEED_PIN_MAX_BATCHES) { + break + } + this.pins.delete(existing) + total -= pin.bytes + } + this.pins.set(key, { pages, bytes, expiresAt: now + FEED_PIN_IDLE_MS }) + } + + private live(key: string): Pin | null { + const pin = this.pins.get(key) + if (!pin || pin.expiresAt <= Date.now()) { + return null + } + pin.expiresAt = Date.now() + FEED_PIN_IDLE_MS + return pin + } +} + +/** The token of a pinned batch's page. */ +export function feedPageToken(key: string, index: number): string { + return `${key}#${index}` +} diff --git a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed.test-fixture.ts b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed.test-fixture.ts new file mode 100644 index 00000000000..41e9ada923d --- /dev/null +++ b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed.test-fixture.ts @@ -0,0 +1,161 @@ +import type { + SidebarInputsChangedEvent, + SidebarInputsReadResult +} from '../../../shared/sidebar-list/sidebar-inputs-feed-contract' +import { + INITIAL_SIDEBAR_INPUTS_FEED_STATE, + refreshSidebarInputs, + sidebarInputsChanged, + sidebarInputsReplied, + type SidebarInputsFeedRequest, + type SidebarInputsFeedState, + type SidebarInputsFeedStep +} from '../../../shared/sidebar-list/sidebar-inputs-feed-installer' +import type { SidebarInputsFeed, SidebarInputsFeedDomain } from './sidebar-inputs-feed' + +/** A domain whose entities a test edits; counts its reads and live watchers. */ +export function testDomain( + name: string, + initial: Record = {}, + options: { rereadOnRead?: boolean } = {} +) { + let entities: Record = { ...initial } + const watchers = new Set<() => void>() + let reads = 0 + let failing = false + let gate: Promise | null = null + const domain: SidebarInputsFeedDomain = { + name, + rereadOnRead: options.rereadOnRead, + read: async () => { + reads += 1 + await gate + if (failing) { + throw new Error('read failed') + } + return { ...entities } + }, + watch: (changed) => { + watchers.add(changed) + return () => watchers.delete(changed) + } + } + const announce = (): void => watchers.forEach((changed) => changed()) + return { + domain, + entities: () => entities, + set(key: string, value: unknown) { + entities = { ...entities, [key]: value } + announce() + }, + remove(key: string) { + entities = Object.fromEntries( + Object.entries(entities).filter(([existing]) => existing !== key) + ) + announce() + }, + fail(next: boolean) { + failing = next + }, + /** Edits without announcing, as a source with no change event does. */ + setQuietly(key: string, value: unknown) { + entities = { ...entities, [key]: value } + }, + /** Holds every read until the returned release runs. */ + hold(): () => void { + let release = (): void => {} + gate = new Promise((resolve) => { + release = () => { + gate = null + resolve() + } + }) + return release + }, + reads: () => reads, + watchers: () => watchers.size + } +} + +/** A phone driving the shared installer against one feed, one reply at a time. */ +export class TestPhone { + state: SidebarInputsFeedState = INITIAL_SIDEBAR_INPUTS_FEED_STATE + pending: SidebarInputsFeedRequest | null = null + readonly replies: SidebarInputsReadResult[] = [] + + constructor( + public feed: SidebarInputsFeed, + readonly connectionId: string + ) { + feed.acquire(connectionId) + } + + private apply(step: SidebarInputsFeedStep): void { + this.state = step.state + this.pending = step.request ?? this.pending + } + + refresh(): void { + this.apply(refreshSidebarInputs(this.state)) + } + + notice(event: Pick): void { + this.apply(sidebarInputsChanged(this.state, event)) + } + + /** The host answers the pending read now; the phone sees the answer only when `deliver` runs. */ + async serveOne(): Promise<{ reply: SidebarInputsReadResult; deliver: () => void }> { + const request = this.pending + if (!request) { + throw new Error('no read pending') + } + this.pending = null + const reply = await this.feed.read(request.params) + return { + reply, + deliver: () => { + this.replies.push(reply) + this.apply(sidebarInputsReplied(this.state, request.id, reply)) + } + } + } + + async answerOne(): Promise { + const { reply, deliver } = await this.serveOne() + deliver() + return reply + } + + async settle(): Promise { + while (this.pending) { + await this.answerOne() + } + } + + /** The connection closed with whatever was in flight; the next ready supersedes it. */ + disconnect(): void { + this.feed.release(this.connectionId) + this.pending = null + } + + view(): Record> { + const view: Record> = {} + for (const [name, entities] of this.state.installed?.domains ?? []) { + if (entities.size > 0) { + view[name] = Object.fromEntries(entities) + } + } + return view + } +} + +/** What a phone should hold: every non-empty domain's current entities. */ +export function truthOf(...domains: ReturnType[]) { + const truth: Record> = {} + for (const domain of domains) { + if (Object.keys(domain.entities()).length > 0) { + truth[domain.domain.name] = domain.entities() + } + } + return truth +} diff --git a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed.test.ts b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed.test.ts new file mode 100644 index 00000000000..6ab3a01da2f --- /dev/null +++ b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed.test.ts @@ -0,0 +1,390 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { + SidebarInputsChangedEvent, + SidebarInputsReadResult +} from '../../../shared/sidebar-list/sidebar-inputs-feed-contract' +import { sidebarInputsReplied } from '../../../shared/sidebar-list/sidebar-inputs-feed-installer' +import { SidebarInputsFeed } from './sidebar-inputs-feed' +import { FEED_LOG_MAX_REVISIONS } from './sidebar-inputs-feed-log' +import { FEED_PIN_IDLE_MS, FEED_PIN_MAX_BATCHES } from './sidebar-inputs-feed-pins' +import { TestPhone, testDomain, truthOf } from './sidebar-inputs-feed.test-fixture' + +// Three ~100 KiB entities: a snapshot of them spans two 256 KiB pages. +const BIG = 'x'.repeat(100 * 1024) + +function makeHost(initial: Record = { a: 1 }) { + const rows = testDomain('rows', initial) + const view = testDomain('view', { sort: 'name' }) + const events: SidebarInputsChangedEvent[] = [] + const feed = new SidebarInputsFeed( + () => [rows.domain, view.domain], + (event) => events.push(event) + ) + const lastEvent = (): SidebarInputsChangedEvent => { + const event = events.at(-1) + if (!event) { + throw new Error('no change notice yet') + } + return event + } + return { rows, view, feed, events, lastEvent, truth: () => truthOf(rows, view) } +} + +/** Lets the feed coalesce and publish what the domains announced. */ +const flush = () => vi.advanceTimersByTimeAsync(60) + +beforeEach(() => { + vi.useFakeTimers() +}) + +afterEach(() => { + vi.useRealTimers() +}) + +describe('sidebar inputs feed', () => { + it('installs a snapshot, then a delta after a change', async () => { + const host = makeHost() + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.settle() + expect(phone.replies.at(-1)?.kind).toBe('snapshot') + expect(phone.view()).toEqual(host.truth()) + + host.rows.set('b', 2) + host.rows.remove('a') + await flush() + phone.notice(host.lastEvent()) + await phone.settle() + expect(phone.replies.at(-1)).toMatchObject({ + kind: 'delta', + domains: { rows: { upserts: { b: 2 }, removals: ['a'] } } + }) + expect(phone.view()).toEqual(host.truth()) + }) + + it('installs a paged snapshot whole, then catches up a change made while paging', async () => { + const host = makeHost({ a: BIG, b: BIG, c: BIG }) + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + const first = await phone.answerOne() + expect(first.next).not.toBeNull() + expect(phone.state.installed).toBeNull() + + host.rows.set('d', 'late') + await flush() + phone.notice(host.lastEvent()) + await phone.settle() + expect(phone.replies.map((reply) => reply.kind)).toEqual(['snapshot', 'snapshot', 'delta']) + expect(phone.view()).toEqual(host.truth()) + }) + + it('ignores a duplicate or late notice, and reads on a newer one', async () => { + const host = makeHost() + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.settle() + host.rows.set('a', 2) + await flush() + const notice = host.lastEvent() + phone.notice(notice) + await phone.settle() + + phone.notice(notice) + phone.notice({ ...notice, revision: notice.revision - 1 }) + expect(phone.pending).toBeNull() + phone.notice({ ...notice, revision: notice.revision + 1 }) + expect(phone.pending?.params).toMatchObject({ cursor: notice.revision }) + }) + + it('coalesces notices into one read in flight and one more after it', async () => { + const host = makeHost() + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + const served = await phone.serveOne() + for (let value = 2; value < 5; value += 1) { + host.rows.set('a', value) + await flush() + phone.notice(host.lastEvent()) + } + expect(phone.pending).toBeNull() + served.deliver() + await phone.settle() + expect(phone.replies.map((reply) => reply.kind)).toEqual(['snapshot', 'delta']) + expect(phone.view()).toEqual(host.truth()) + }) + + it('needs no second read when the reply already covers a notice', async () => { + const host = makeHost() + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.settle() + host.rows.set('a', 2) + await flush() + phone.refresh() + phone.notice(host.lastEvent()) + await phone.settle() + expect(phone.replies).toHaveLength(2) + expect(phone.view()).toEqual(host.truth()) + }) + + it('drops a reply to a request it no longer waits on', async () => { + const host = makeHost() + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + const stale = phone.pending + phone.disconnect() + phone.feed.acquire(phone.connectionId) + phone.refresh() + if (!stale) { + throw new Error('expected a read') + } + const reply = await host.feed.read(stale.params) + expect(sidebarInputsReplied(phone.state, stale.id, reply).state).toBe(phone.state) + }) + + it('reads a snapshot when a delta does not start at the installed revision', async () => { + const host = makeHost() + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.settle() + host.rows.set('a', 2) + await flush() + phone.notice(host.lastEvent()) + const request = phone.pending + if (!request) { + throw new Error('expected a read') + } + const reply = await host.feed.read(request.params) + const step = sidebarInputsReplied(phone.state, request.id, { + ...reply, + from: (reply.from ?? 0) - 1 + }) + expect(step.state.installed).toBe(phone.state.installed) + expect(step.request?.params).toEqual({ maxBytes: expect.any(Number) }) + }) + + it('reads a snapshot once the cursor fell out of the log', async () => { + const host = makeHost() + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.settle() + for (let value = 0; value <= FEED_LOG_MAX_REVISIONS; value += 1) { + host.rows.set('a', value) + await flush() + } + phone.notice(host.lastEvent()) + await phone.settle() + expect(phone.replies.at(-1)?.kind).toBe('snapshot') + expect(phone.view()).toEqual(host.truth()) + }) + + it('starts over when a pinned snapshot expired mid-paging', async () => { + const host = makeHost({ a: BIG, b: BIG, c: BIG }) + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.answerOne() + await vi.advanceTimersByTimeAsync(FEED_PIN_IDLE_MS + 1) + expect((await phone.answerOne()).kind).toBe('reset') + await phone.settle() + expect(phone.view()).toEqual(host.truth()) + }) + + it('keeps what it installed across a disconnect and reads a snapshot from a new session', async () => { + const host = makeHost({ a: BIG, b: BIG, c: BIG }) + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.settle() + const installed = phone.state.installed + + host.rows.set('d', 1) + await flush() + phone.notice(host.lastEvent()) + expect(phone.pending).not.toBeNull() + phone.disconnect() + expect(phone.state.installed).toBe(installed) + + // The last subscriber left: the session's state and watches are gone. + expect(host.rows.watchers()).toBe(0) + host.rows.set('e', 1) + expect(host.rows.reads()).toBe(2) + + phone.feed.acquire('conn-2') + phone.refresh() + await phone.settle() + expect(phone.replies.at(-1)?.kind).toBe('snapshot') + expect(phone.view()).toEqual(host.truth()) + }) + + it('reads a delta over a re-paired connection while the session lives on', async () => { + const host = makeHost() + const other = new TestPhone(host.feed, 'conn-other') + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.settle() + phone.disconnect() + host.rows.set('a', 2) + await flush() + + phone.feed.acquire('conn-repaired') + phone.refresh() + await phone.settle() + expect(phone.replies.at(-1)?.kind).toBe('delta') + expect(phone.view()).toEqual(host.truth()) + other.disconnect() + }) + + it('serves two phones from one session, which ends with the last of them', async () => { + const host = makeHost() + const first = new TestPhone(host.feed, 'conn-1') + const second = new TestPhone(host.feed, 'conn-2') + first.refresh() + second.refresh() + await Promise.all([first.settle(), second.settle()]) + expect(host.rows.reads()).toBe(1) + + first.disconnect() + host.rows.set('a', 2) + await flush() + second.notice(host.lastEvent()) + await second.settle() + expect(second.view()).toEqual(host.truth()) + + second.disconnect() + expect(host.rows.watchers()).toBe(0) + }) + + it('replaces everything after a runtime restart, whose generation is new', async () => { + const host = makeHost() + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.settle() + + const restarted = makeHost({ z: 1 }) + phone.feed = restarted.feed + restarted.feed.acquire('conn-1') + phone.notice({ generation: restarted.feed.generation, revision: 1 }) + expect(phone.pending?.params).toMatchObject({ generation: host.feed.generation }) + await phone.settle() + expect(phone.replies.at(-1)?.kind).toBe('snapshot') + expect(phone.state.installed?.generation).toBe(restarted.feed.generation) + expect(phone.view()).toEqual(restarted.truth()) + }) + + it('sends an entity larger than a page as parts, and a large revision as a paged delta', async () => { + const rankedIds = Array.from( + { length: 10_000 }, + (_, index) => `runtime:env-${index}::/w/${index}` + ) + const host = makeHost({ order: { rankedIds } }) + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.settle() + expect(phone.replies.length).toBeGreaterThan(2) + expect(phone.view()).toEqual(host.truth()) + + for (let index = 0; index < 2_000; index += 1) { + host.rows.set(`row-${index}`, { name: `workspace ${index}`.padEnd(300, '.') }) + } + await flush() + phone.notice(host.lastEvent()) + const before = phone.replies.length + await phone.settle() + expect(phone.replies.slice(before).every((reply) => reply.kind === 'delta')).toBe(true) + expect(phone.replies.length - before).toBeGreaterThan(1) + expect(phone.view()).toEqual(host.truth()) + }) + + it('keeps a domain as last published when its read fails', async () => { + const host = makeHost() + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.settle() + host.rows.fail(true) + host.rows.set('a', 2) + await flush() + phone.refresh() + await phone.settle() + expect(phone.view().rows).toEqual({ a: 1 }) + }) + + it('bounds pinned batches, evicting the oldest', async () => { + const host = makeHost({ a: BIG, b: BIG, c: BIG }) + host.feed.acquire('conn-1') + const firsts: SidebarInputsReadResult[] = [] + for (let index = 0; index <= FEED_PIN_MAX_BATCHES; index += 1) { + // A different page size is a different batch. + firsts.push(await host.feed.read({ maxBytes: 200 * 1024 + index })) + } + const oldest = firsts[0].next + const newest = firsts.at(-1)?.next + if (!oldest || !newest) { + throw new Error('expected paged batches') + } + expect((await host.feed.read({ pageToken: oldest, maxBytes: 1 })).kind).toBe('reset') + expect((await host.feed.read({ pageToken: newest, maxBytes: 1 })).kind).toBe('snapshot') + }) + + it('reads nothing and announces nothing until a phone subscribes', async () => { + const host = makeHost() + host.rows.set('a', 2) + await flush() + expect(host.rows.reads()).toBe(0) + expect(host.rows.watchers()).toBe(0) + expect(host.events).toEqual([]) + }) + + it('re-reads a quiet source on every read, and joins concurrent reads into one flush', async () => { + const quiet = testDomain('quiet', { a: 1 }, { rereadOnRead: true }) + const feed = new SidebarInputsFeed( + () => [quiet.domain], + () => {} + ) + feed.acquire('conn-1') + await feed.read({ maxBytes: 1 }) + quiet.setQuietly('a', 2) + const release = quiet.hold() + const reads = [feed.read({ maxBytes: 1 }), feed.read({ maxBytes: 1 })] + release() + const [first, second] = await Promise.all(reads) + expect(quiet.reads()).toBe(3) + expect(first).toMatchObject({ kind: 'snapshot', domains: { quiet: { upserts: { a: 2 } } } }) + expect(second.revision).toBe(first.revision) + }) + + it('answers reset to a read whose session ended while it waited', async () => { + const host = makeHost() + host.feed.acquire('conn-1') + const release = host.rows.hold() + const read = host.feed.read({ maxBytes: 1 }) + host.feed.release('conn-1') + release() + expect((await read).kind).toBe('reset') + }) + + it('refuses a page token from a session that ended', async () => { + const host = makeHost({ a: BIG, b: BIG, c: BIG }) + host.feed.acquire('conn-1') + const first = await host.feed.read({ maxBytes: 1 }) + host.feed.release('conn-1') + host.feed.acquire('conn-2') + if (!first.next) { + throw new Error('expected a paged snapshot') + } + expect((await host.feed.read({ pageToken: first.next, maxBytes: 1 })).kind).toBe('reset') + }) + + it('reads a snapshot past a revision larger than the whole log', async () => { + const host = makeHost() + const phone = new TestPhone(host.feed, 'conn-1') + phone.refresh() + await phone.settle() + for (let index = 0; index < 30; index += 1) { + host.rows.set(`big-${index}`, BIG) + } + await flush() + phone.notice(host.lastEvent()) + await phone.settle() + expect(phone.replies.at(-1)?.kind).toBe('snapshot') + expect(phone.view()).toEqual(host.truth()) + }) +}) diff --git a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed.ts b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed.ts new file mode 100644 index 00000000000..7128ddfb182 --- /dev/null +++ b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed.ts @@ -0,0 +1,322 @@ +import { randomUUID } from 'node:crypto' +import { + SIDEBAR_INPUTS_PAGE_MAX_BYTES, + SIDEBAR_INPUTS_PAGE_MIN_BYTES, + SIDEBAR_INPUTS_SCHEMA_VERSION, + SIDEBAR_INPUTS_SEMANTICS_VERSION, + type SidebarInputsChangedEvent, + type SidebarInputsReadParams, + type SidebarInputsReadResult +} from '../../../shared/sidebar-list/sidebar-inputs-feed-contract' +import { + feedEntityFromJson, + paginateFeedChanges, + type FeedChange, + type FeedEntity +} from './sidebar-inputs-feed-pages' +import { FeedChangeLog, type FeedRevisionChanges } from './sidebar-inputs-feed-log' +import { FeedPins, feedPageToken } from './sidebar-inputs-feed-pins' + +/** One producer's snapshot, as entities keyed so that none is needlessly large. */ +export type SidebarInputsFeedDomain = { + name: string + read: () => Record | Promise> + /** Calls `changed` after anything `read` depends on may have changed; returns the detach. */ + watch: (changed: () => void) => () => void + /** Some of its sources announce no change, so every phone read re-reads it first. */ + rereadOnRead?: boolean +} + +// One revision per burst of producer changes. +const FLUSH_COALESCE_MS = 50 + +/** + * The phone's copy of the sidebar inputs, served from the producers' own snapshots. Everything + * here (published entities, log, pinned pages) exists only while a phone is subscribed; the + * producers run regardless, since the desktop reads them too. + */ +export class SidebarInputsFeed { + readonly generation = randomUUID() + private lastRevision = 0 + private readonly subscribers = new Set() + private session: FeedSession | null = null + + constructor( + private readonly domains: () => readonly SidebarInputsFeedDomain[], + private readonly emit: (event: SidebarInputsChangedEvent) => void + ) {} + + get subscribed(): boolean { + return this.session !== null + } + + /** True when `subscriberId` was not yet subscribed, so its caller owns the matching release. */ + acquire(subscriberId: string): boolean { + if (this.subscribers.has(subscriberId)) { + return false + } + this.subscribers.add(subscriberId) + // The empty set is a revision of its own, so a cursor from before this session reads afresh. + this.session ??= new FeedSession( + this.domains(), + this.generation, + this.lastRevision + 1, + this.emit + ) + return true + } + + release(subscriberId: string): void { + if (this.subscribers.delete(subscriberId) && this.subscribers.size === 0 && this.session) { + this.lastRevision = this.session.dispose() + this.session = null + } + } + + async read(params: SidebarInputsReadParams): Promise { + const session = this.session + if (!session) { + throw new Error('sidebar inputs feed has no subscriber') + } + return session.read(params) + } +} + +function feedReply( + generation: string, + kind: SidebarInputsReadResult['kind'], + revision: number, + from: number | null, + domains: SidebarInputsReadResult['domains'], + next: string | null +): SidebarInputsReadResult { + return { + generation, + revision, + schemaVersion: SIDEBAR_INPUTS_SCHEMA_VERSION, + semanticsVersion: SIDEBAR_INPUTS_SEMANTICS_VERSION, + kind, + from, + domains, + next + } +} + +class FeedSession { + private readonly published = new Map>() + private readonly dirty = new Set() + // Counts dirty marks; `flushedMarks` is the count the last finished flush had taken. + private marks = 0 + private flushedMarks = 0 + private readonly log: FeedChangeLog + private readonly pins = new FeedPins() + private readonly detaches: (() => void)[] + private timer: ReturnType | null = null + private flushing: Promise | null = null + private disposed = false + + constructor( + private readonly domains: readonly SidebarInputsFeedDomain[], + private readonly generation: string, + private revision: number, + private readonly emit: (event: SidebarInputsChangedEvent) => void + ) { + this.log = new FeedChangeLog(revision) + for (const domain of domains) { + this.markDirty(domain.name) + } + this.detaches = domains.map((domain) => domain.watch(() => this.invalidate(domain.name))) + } + + /** Returns the last revision, which the next session continues from. */ + dispose(): number { + this.disposed = true + if (this.timer) { + clearTimeout(this.timer) + } + for (const detach of this.detaches) { + detach() + } + return this.revision + } + + async read(params: SidebarInputsReadParams): Promise { + const { generation } = this + const reset = (): SidebarInputsReadResult => + feedReply(generation, 'reset', this.revision, null, {}, null) + if (params.pageToken !== undefined) { + return this.pins.page(params.pageToken) ?? reset() + } + for (const domain of this.domains) { + if (domain.rereadOnRead) { + this.markDirty(domain.name) + } + } + await this.flushDirty() + // A release during the flush discarded what this read would serve. + if (this.disposed) { + return reset() + } + const budget = Math.min( + SIDEBAR_INPUTS_PAGE_MAX_BYTES, + Math.max(SIDEBAR_INPUTS_PAGE_MIN_BYTES, params.maxBytes) + ) + const { cursor } = params + // Generations have no order: any other one, or a cursor the log no longer reaches, reads afresh. + if ( + params.generation === generation && + cursor !== undefined && + this.log.covers(cursor, this.revision) + ) { + return this.batch(`delta:${cursor}`, cursor, budget, () => this.log.since(cursor)) + } + return this.batch('snapshot', null, budget, () => this.everything()) + } + + private invalidate(name: string): void { + if (this.disposed) { + return + } + this.markDirty(name) + this.timer ??= setTimeout(() => { + this.timer = null + void this.flushDirty() + }, FLUSH_COALESCE_MS) + } + + private markDirty(name: string): void { + this.dirty.add(name) + this.marks += 1 + } + + /** Resolves once everything dirty at the call is published, waiting on at most two flushes. */ + private async flushDirty(): Promise { + if (this.timer) { + clearTimeout(this.timer) + this.timer = null + } + const marks = this.marks + // A flush already running may have taken its domains before this call's marks. + while (this.flushedMarks < marks && !this.disposed) { + this.flushing ??= this.flush().finally(() => { + this.flushing = null + }) + await this.flushing + } + } + + private async flush(): Promise { + const marks = this.marks + const due = this.domains.filter((domain) => this.dirty.has(domain.name)) + this.dirty.clear() + const reads = await Promise.all(due.map((domain) => readDomain(domain))) + this.flushedMarks = marks + if (this.disposed) { + return + } + const changes: FeedRevisionChanges = new Map() + due.forEach((domain, index) => { + const changed = this.publish(domain.name, reads[index]) + if (changed.size > 0) { + changes.set(domain.name, changed) + } + }) + if (changes.size > 0) { + this.revision += 1 + this.log.append(this.revision, changes) + this.emit({ + type: 'sidebarInputsChanged', + generation: this.generation, + revision: this.revision + }) + } + } + + /** Replaces a domain's published entities; returns what changed. A failed read changes nothing. */ + private publish(name: string, read: Map | null): Map { + const changed = new Map() + if (!read) { + return changed + } + const before = this.published.get(name) ?? new Map() + const after = new Map() + for (const [key, json] of read) { + const kept = before.get(key) + if (kept?.json === json) { + after.set(key, kept) + } else { + // Parsed from its own JSON, so a producer's later in-place edit cannot reach a page. + const entity = feedEntityFromJson(json) + after.set(key, entity) + changed.set(key, entity) + } + } + for (const key of before.keys()) { + if (!after.has(key)) { + changed.set(key, null) + } + } + this.published.set(name, after) + return changed + } + + private *everything(): Iterable { + for (const domain of this.domains) { + for (const [key, entity] of this.published.get(domain.name) ?? []) { + yield [domain.name, key, entity] + } + } + } + + /** The batch at this revision, pinned for its later pages; a reader at the same place shares it. */ + private batch( + startsFrom: string, + from: number | null, + budget: number, + changes: () => Iterable + ): SidebarInputsReadResult { + const key = `${startsFrom}:${this.revision}:${budget}` + const pinned = this.pins.find(key) + if (pinned) { + return pinned[0] + } + const bodies = paginateFeedChanges(changes(), budget) + const pages = bodies.map((body, index) => + feedReply( + this.generation, + from === null ? 'snapshot' : 'delta', + this.revision, + from, + body.domains, + index + 1 < bodies.length ? feedPageToken(key, index + 1) : null + ) + ) + if (pages.length > 1) { + this.pins.add( + key, + pages, + bodies.reduce((sum, body) => sum + body.bytes, 0) + ) + } + return pages[0] + } +} + +/** Each entity's JSON; null when the read failed. */ +async function readDomain(domain: SidebarInputsFeedDomain): Promise | null> { + try { + const entities = new Map() + for (const [key, value] of Object.entries(await domain.read())) { + // What JSON cannot hold (undefined, a function) is absent, as it would be on the wire. + const json: string | undefined = JSON.stringify(value) + if (json !== undefined) { + entities.set(key, json) + } + } + return entities + } catch (error) { + // The domain keeps what it last published; its next change or read tries again. + console.warn(`[sidebar-inputs-feed] could not read ${domain.name}`, error) + return null + } +} diff --git a/src/main/runtime/sidebar-inputs/sidebar-temporal-owner.ts b/src/main/runtime/sidebar-inputs/sidebar-temporal-owner.ts index adad22ec416..c04c20e47db 100644 --- a/src/main/runtime/sidebar-inputs/sidebar-temporal-owner.ts +++ b/src/main/runtime/sidebar-inputs/sidebar-temporal-owner.ts @@ -1,3 +1,4 @@ +import { ProducerRevision } from './producer-revision' import type { Repo } from '../../../shared/repo-types' import type { Worktree } from '../../../shared/worktree/types' import { getWorktreeHostIdentity } from '../../../shared/worktree/host-qualified-identity' @@ -60,7 +61,7 @@ const RETENTION_DEADLINE = 'sleeping-retention' * compared against it and never persists ranks. It stays idle until a window begins a session. */ export class SidebarTemporalOwner { - private revision = 0 + readonly revision = new ProducerRevision() private sessionId: string | null = null private disposed = false private settleState = INITIAL_SORT_SETTLE_STATE @@ -83,7 +84,7 @@ export class SidebarTemporalOwner { snapshot(): SidebarTemporalSnapshot { return { - revision: this.revision, + revision: this.revision.value, sessionId: this.sessionId, sortSnapshot: this.sortSnapshot, retention: this.retention @@ -102,7 +103,7 @@ export class SidebarTemporalOwner { // Until the new session's first sample lands, a reader must not take the old warm ranks. this.sortSnapshot = null this.retentionState = createSleepingSweepRetentionState() - this.revision += 1 + this.revision.bump() this.retentionPass() this.settleNow() } @@ -231,7 +232,7 @@ export class SidebarTemporalOwner { this.identityById = new Map( sample.worktrees.map((worktree) => [worktree.id, getWorktreeHostIdentity(worktree)]) ) - this.revision += 1 + this.revision.bump() this.scheduleCreateGraceBoundary(ranking.sorted, now) } @@ -265,7 +266,7 @@ export class SidebarTemporalOwner { this.clock.cancel(RETENTION_DEADLINE) if (this.retention !== null) { this.retention = null - this.revision += 1 + this.revision.bump() } return } @@ -289,7 +290,7 @@ export class SidebarTemporalOwner { const previous = this.retention?.retainedIds if (!previous || previous.length !== ids.length || previous.some((id, i) => id !== ids[i])) { this.retention = { retainedIds: ids, retentionDecidedAt: now } - this.revision += 1 + this.revision.bump() } } } diff --git a/src/main/runtime/sidebar-inputs/workspace-activity-facts-producer.ts b/src/main/runtime/sidebar-inputs/workspace-activity-facts-producer.ts index 6fc48d8d16b..2dc7e5c3a23 100644 --- a/src/main/runtime/sidebar-inputs/workspace-activity-facts-producer.ts +++ b/src/main/runtime/sidebar-inputs/workspace-activity-facts-producer.ts @@ -1,3 +1,4 @@ +import { ProducerRevision } from './producer-revision' import { AGENT_STATUS_STALE_AFTER_MS, isFreshNonDoneAgentStatus, @@ -43,8 +44,7 @@ export class WorkspaceActivityFactsProducer { private readonly agentPanes = new Map() private readonly agentPanesByWorktree = new Map>() private agentStore: AgentStatusStoreView | null = null - private readonly changeListeners = new Set<() => void>() - private revision = 0 + readonly revision = new ProducerRevision() private expiryAt = Infinity constructor( @@ -55,14 +55,8 @@ export class WorkspaceActivityFactsProducer { } ) {} - /** Told after every new revision. */ - onChange(listener: () => void): () => void { - this.changeListeners.add(listener) - return () => this.changeListeners.delete(listener) - } - snapshot(): WorkspaceActivityFactsSnapshot { - return { revision: this.revision, facts: Object.fromEntries(this.facts) } + return { revision: this.revision.value, facts: Object.fromEntries(this.facts) } } /** Re-reads the workspace's surfaces; the runtime calls this per publication. */ @@ -195,10 +189,7 @@ export class WorkspaceActivityFactsProducer { } this.facts.set(worktreeId, next) } - this.revision += 1 - for (const listener of this.changeListeners) { - listener() - } + this.revision.bump() } // One deadline for the earliest lease; a renewed lease only moves later, so an early fire re-arms. diff --git a/src/main/runtime/sidebar-inputs/workspace-creation-registry.ts b/src/main/runtime/sidebar-inputs/workspace-creation-registry.ts index e2dfc82fad9..2240c62ccac 100644 --- a/src/main/runtime/sidebar-inputs/workspace-creation-registry.ts +++ b/src/main/runtime/sidebar-inputs/workspace-creation-registry.ts @@ -1,3 +1,4 @@ +import { ProducerRevision } from './producer-revision' import { randomUUID } from 'node:crypto' import type { WorkspaceCreation, @@ -14,27 +15,27 @@ export type WorkspaceCreationHandle = { /** Every creation this host is running, whoever asked: its own window, a phone, the CLI. */ export class WorkspaceCreationRegistry { private readonly creations = new Map() - private revision = 0 + readonly revision = new ProducerRevision() /** Opened once the project is known. A caller without an id gets one, so a CLI create is listed. */ open(start: { creationId?: string; repoId: string }): WorkspaceCreationHandle { const creationId = start.creationId ?? randomUUID() const entry: WorkspaceCreation = { repoId: start.repoId } this.creations.set(creationId, entry) - this.revision += 1 + this.revision.bump() // Identity, not id: a concurrent retry under the same id owns the entry from then on. const owns = (): boolean => this.creations.get(creationId) === entry return { named: (worktreeId) => { if (owns() && entry.worktreeId !== worktreeId) { entry.worktreeId = worktreeId - this.revision += 1 + this.revision.bump() } }, close: () => { if (owns()) { this.creations.delete(creationId) - this.revision += 1 + this.revision.bump() } } } @@ -45,6 +46,6 @@ export class WorkspaceCreationRegistry { for (const [creationId, entry] of this.creations) { creations[creationId] = { ...entry } } - return { revision: this.revision, creations } + return { revision: this.revision.value, creations } } } diff --git a/src/renderer/src/hooks/ipc-events/runtime-client-ipc-bridge.ts b/src/renderer/src/hooks/ipc-events/runtime-client-ipc-bridge.ts index 4fd0f8fb5a4..3ce59b73613 100644 --- a/src/renderer/src/hooks/ipc-events/runtime-client-ipc-bridge.ts +++ b/src/renderer/src/hooks/ipc-events/runtime-client-ipc-bridge.ts @@ -163,6 +163,10 @@ export function registerRuntimeClientIpcBridge( }) return } + // This window builds its sidebar from its own store, never from a host's phone feed. + if (event.type === 'sidebarInputsChanged') { + return + } // Older hosts broadcast local/CLI activation without an address; that is not this viewer's intent. if (!event.navigation || !navigationTargetsClients(event.navigation)) { return diff --git a/src/shared/rpc-contract/rpc-params-catalog.generated.ts b/src/shared/rpc-contract/rpc-params-catalog.generated.ts index f844d9a97fc..f6c9d9a82ab 100644 --- a/src/shared/rpc-contract/rpc-params-catalog.generated.ts +++ b/src/shared/rpc-contract/rpc-params-catalog.generated.ts @@ -625,6 +625,7 @@ import { } from './worktree-params' import { WorktreeAdoptProvisionedRoot } from './worktree-provisioned-root-params' import { RepoIconDataUrlsParams } from '../sidebar-list/repo-icon-data-urls-contract' +import { SidebarInputsReadParams } from '../sidebar-list/sidebar-inputs-feed-contract' import { SkillBundleInstallRequestSchema } from '../skill-bundle-install-contract' import { SkillDeleteRequestSchema } from '../skill-delete-contract' import { @@ -1254,6 +1255,7 @@ export const RPC_PARAMS_BY_METHOD = { 'sidebarInputs.activityFacts': null, 'sidebarInputs.desktopFacts': null, 'sidebarInputs.pairedServerRows': null, + 'sidebarInputs.read': SidebarInputsReadParams, 'sidebarInputs.repoIcons': RepoIconDataUrlsParams, 'sidebarInputs.reviewFacts': null, 'sidebarInputs.temporal': null, diff --git a/src/shared/runtime-client-events.ts b/src/shared/runtime-client-events.ts index af26fb90a1a..4c2a58fe1fb 100644 --- a/src/shared/runtime-client-events.ts +++ b/src/shared/runtime-client-events.ts @@ -9,6 +9,7 @@ import type { TerminalSideEffectBatch } from './terminal-side-effect-facts' import type { RuntimeNativeChatLaunchDraftResolution } from './runtime-types' import type { RuntimeNavigationTarget } from './runtime-navigation' import type { WorktreeSortOrderUpdate } from './worktree/sort-order-update' +import type { SidebarInputsChangedEvent } from './sidebar-list/sidebar-inputs-feed-contract' export type RuntimeClientEvent = | { type: 'reposChanged' } @@ -23,6 +24,8 @@ export type RuntimeClientEvent = } | ({ type: 'nativeChatLaunchDraftResolved' } & RuntimeNativeChatLaunchDraftResolution) | { type: 'terminalSideEffects'; batch: TerminalSideEffectBatch } + /** Only while this host's own phone reads the sidebar feed; every other reader drops it. */ + | SidebarInputsChangedEvent // Why: SSH connections live on the runtime host; paired clients have no IPC // channel for ssh:state-changed, so without this event their reconnect // overlays never learn the host connected (STA-1468). diff --git a/src/shared/sidebar-list/sidebar-inputs-feed-contract.ts b/src/shared/sidebar-list/sidebar-inputs-feed-contract.ts new file mode 100644 index 00000000000..8ca0b7bf471 --- /dev/null +++ b/src/shared/sidebar-list/sidebar-inputs-feed-contract.ts @@ -0,0 +1,58 @@ +import { z } from 'zod' +import { openEnum } from '../zod-salvage' + +/** The reply's shape; an added optional field keeps it. */ +export const SIDEBAR_INPUTS_SCHEMA_VERSION = 1 +/** What the inputs mean: bumps whenever the same inputs would build a different list. */ +export const SIDEBAR_INPUTS_SEMANTICS_VERSION = 1 + +/** A page's encoded ceiling. A phone's bridge frame (640 KiB) holds it with the RPC envelope. */ +export const SIDEBAR_INPUTS_PAGE_MAX_BYTES = 256 * 1024 +export const SIDEBAR_INPUTS_PAGE_MIN_BYTES = 16 * 1024 + +/** + * With no pageToken: a delta from `cursor` when it is still in this generation's log, else a + * snapshot. A pageToken continues the batch whose page named it. + */ +export const SidebarInputsReadParams = z.object({ + generation: z.string().optional(), + cursor: z.number().int().nonnegative().optional(), + pageToken: z.string().optional(), + maxBytes: z.number().int().positive() +}) + +export type SidebarInputsReadParams = z.infer + +/** One domain's entities as the batch changes them. */ +const SidebarInputsDomainChangeSchema = z.object({ + upserts: z.record(z.string(), z.unknown()), + removals: z.array(z.string()), + /** Pieces of entities too large for one page, as JSON text joined across the batch's pages. */ + parts: z.record(z.string(), z.string()) +}) + +export const SidebarInputsReadResultSchema = z.object({ + /** A new id each time the runtime starts; ids have no order. */ + generation: z.string(), + /** The revision this batch brings the reader to; a reset names the current one. */ + revision: z.number().int(), + schemaVersion: z.number().int(), + semanticsVersion: z.number().int(), + // A kind this build has not heard of reads as a reset: re-read from what is installed. + kind: openEnum(['snapshot', 'delta', 'reset'], 'reset'), + /** A delta applies only on exactly this revision; null otherwise. */ + from: z.number().int().nullable(), + domains: z.record(z.string(), SidebarInputsDomainChangeSchema), + /** The batch's next page; null on its last. A batch installs only once it is whole. */ + next: z.string().nullable() +}) + +export type SidebarInputsDomainChange = z.infer +export type SidebarInputsReadResult = z.infer + +/** Carries no data: a reader re-reads through sidebarInputs.read. */ +export type SidebarInputsChangedEvent = { + type: 'sidebarInputsChanged' + generation: string + revision: number +} diff --git a/src/shared/sidebar-list/sidebar-inputs-feed-installer.test.ts b/src/shared/sidebar-list/sidebar-inputs-feed-installer.test.ts new file mode 100644 index 00000000000..cd26db9341e --- /dev/null +++ b/src/shared/sidebar-list/sidebar-inputs-feed-installer.test.ts @@ -0,0 +1,190 @@ +import { describe, expect, it } from 'vitest' +import { + SIDEBAR_INPUTS_SCHEMA_VERSION, + SIDEBAR_INPUTS_SEMANTICS_VERSION, + type SidebarInputsReadResult +} from './sidebar-inputs-feed-contract' +import { + INITIAL_SIDEBAR_INPUTS_FEED_STATE, + refreshSidebarInputs, + sidebarInputsChanged, + sidebarInputsReadFailed, + sidebarInputsReplied, + type SidebarInputsFeedStep +} from './sidebar-inputs-feed-installer' + +function page( + fields: Partial & Pick +): SidebarInputsReadResult { + return { + generation: 'g1', + schemaVersion: SIDEBAR_INPUTS_SCHEMA_VERSION, + semanticsVersion: SIDEBAR_INPUTS_SEMANTICS_VERSION, + from: null, + domains: {}, + next: null, + ...fields + } +} + +const rows = (upserts: Record, removals: string[] = []) => ({ + rows: { upserts, removals, parts: {} } +}) + +function reply( + step: SidebarInputsFeedStep, + result: SidebarInputsReadResult +): SidebarInputsFeedStep { + if (!step.request) { + throw new Error('expected a read') + } + return sidebarInputsReplied(step.state, step.request.id, result) +} + +/** Installed at g1 revision 3 with rows {a: 1}. */ +function installed(): SidebarInputsFeedStep { + return reply( + refreshSidebarInputs(INITIAL_SIDEBAR_INPUTS_FEED_STATE), + page({ kind: 'snapshot', revision: 3, domains: rows({ a: 1 }) }) + ) +} + +const rowsOf = (step: SidebarInputsFeedStep) => + Object.fromEntries(step.state.installed?.domains.get('rows') ?? []) + +describe('sidebar inputs installer', () => { + it('installs a snapshot, then applies a delta from exactly its revision', () => { + const base = installed() + expect(base.request).toBeNull() + const read = sidebarInputsChanged(base.state, { generation: 'g1', revision: 4 }) + expect(read.request?.params).toMatchObject({ generation: 'g1', cursor: 3 }) + const next = reply( + read, + page({ kind: 'delta', from: 3, revision: 4, domains: rows({ b: 2 }, ['a']) }) + ) + expect(rowsOf(next)).toEqual({ b: 2 }) + expect(next.state.installed?.revision).toBe(4) + }) + + it('supersedes a read in flight on a stream ready and drops the stale reply', () => { + const first = refreshSidebarInputs(INITIAL_SIDEBAR_INPUTS_FEED_STATE) + const second = refreshSidebarInputs(first.state) + const stale = first.request + if (!stale) { + throw new Error('expected a read') + } + const dropped = sidebarInputsReplied( + second.state, + stale.id, + page({ kind: 'snapshot', revision: 1 }) + ) + expect(dropped.state).toBe(second.state) + expect( + rowsOf(reply(second, page({ kind: 'snapshot', revision: 2, domains: rows({ a: 1 }) }))) + ).toEqual({ a: 1 }) + }) + + it('reads once more after a notice its reply does not cover, and not after one it does', () => { + const base = installed() + const read = sidebarInputsChanged(base.state, { generation: 'g1', revision: 4 }) + const covered = sidebarInputsChanged(read.state, { generation: 'g1', revision: 5 }) + expect(covered.request).toBeNull() + expect( + reply({ ...read, state: covered.state }, page({ kind: 'delta', from: 3, revision: 5 })) + .request + ).toBeNull() + const uncovered = sidebarInputsChanged(read.state, { generation: 'g1', revision: 6 }) + expect( + reply({ ...read, state: uncovered.state }, page({ kind: 'delta', from: 3, revision: 5 })) + .request?.params + ).toMatchObject({ + cursor: 5 + }) + }) + + it('forgets a failed read, so the next notice or ready reads again', () => { + const base = installed() + const read = sidebarInputsChanged(base.state, { generation: 'g1', revision: 4 }) + const noticed = sidebarInputsChanged(read.state, { generation: 'g1', revision: 5 }) + if (!read.request) { + throw new Error('expected a read') + } + const failed = sidebarInputsReadFailed(noticed.state, read.request.id) + expect(failed).toMatchObject({ inFlight: null, staged: null, noticed: null }) + expect( + sidebarInputsChanged(failed, { generation: 'g1', revision: 5 }).request?.params + ).toMatchObject({ + cursor: 3 + }) + expect(refreshSidebarInputs(failed).request?.params).toMatchObject({ cursor: 3 }) + }) + + it('stages pages and installs them only once the last lands', () => { + const first = reply( + refreshSidebarInputs(INITIAL_SIDEBAR_INPUTS_FEED_STATE), + page({ kind: 'snapshot', revision: 2, domains: rows({ a: 1 }), next: 'p#1' }) + ) + expect(first.state.installed).toBeNull() + expect(first.request?.params).toMatchObject({ pageToken: 'p#1' }) + const done = reply(first, page({ kind: 'snapshot', revision: 2, domains: rows({ b: 2 }) })) + expect(rowsOf(done)).toEqual({ a: 1, b: 2 }) + }) + + it('joins an entity’s parts across pages', () => { + const json = JSON.stringify({ rankedIds: ['x', 'y'] }) + const part = (text: string) => ({ rows: { upserts: {}, removals: [], parts: { order: text } } }) + const first = reply( + refreshSidebarInputs(INITIAL_SIDEBAR_INPUTS_FEED_STATE), + page({ kind: 'snapshot', revision: 2, domains: part(json.slice(0, 5)), next: 'p#1' }) + ) + const done = reply(first, page({ kind: 'snapshot', revision: 2, domains: part(json.slice(5)) })) + expect(rowsOf(done)).toEqual({ order: { rankedIds: ['x', 'y'] } }) + }) + + it('reads a snapshot on a delta from elsewhere, a page from another batch, or unjoinable parts', () => { + const base = installed() + const read = sidebarInputsChanged(base.state, { generation: 'g1', revision: 5 }) + const mismatch = reply(read, page({ kind: 'delta', from: 2, revision: 5 })) + expect(mismatch.state.installed).toBe(base.state.installed) + expect(mismatch.request?.params).toEqual({ maxBytes: expect.any(Number) }) + + const paging = reply(read, page({ kind: 'delta', from: 3, revision: 5, next: 'p#1' })) + expect(reply(paging, page({ kind: 'delta', from: 3, revision: 6 })).request?.params).toEqual({ + maxBytes: expect.any(Number) + }) + + const broken = reply( + read, + page({ + kind: 'delta', + from: 3, + revision: 5, + domains: { rows: { upserts: {}, removals: [], parts: { a: '{"unterminated' } } } + }) + ) + expect(broken.state.installed).toBe(base.state.installed) + expect(broken.request?.params).toEqual({ maxBytes: expect.any(Number) }) + }) + + it('re-reads from what is installed on a reset, and on a notice from a new generation', () => { + const base = installed() + const read = refreshSidebarInputs(base.state) + expect(reply(read, page({ kind: 'reset', revision: 9 })).request?.params).toMatchObject({ + cursor: 3 + }) + const restarted = sidebarInputsChanged(base.state, { generation: 'g2', revision: 1 }) + expect(restarted.request?.params).toMatchObject({ generation: 'g1', cursor: 3 }) + const next = reply( + restarted, + page({ generation: 'g2', kind: 'snapshot', revision: 1, domains: rows({ z: 1 }) }) + ) + expect(next.state.installed?.generation).toBe('g2') + expect(rowsOf(next)).toEqual({ z: 1 }) + }) + + it('never installs a snapshot older than what it has', () => { + const base = installed() + const late = reply(refreshSidebarInputs(base.state), page({ kind: 'snapshot', revision: 2 })) + expect(late.state.installed).toBe(base.state.installed) + }) +}) diff --git a/src/shared/sidebar-list/sidebar-inputs-feed-installer.ts b/src/shared/sidebar-list/sidebar-inputs-feed-installer.ts new file mode 100644 index 00000000000..552f3baf5c6 --- /dev/null +++ b/src/shared/sidebar-list/sidebar-inputs-feed-installer.ts @@ -0,0 +1,214 @@ +import { + SIDEBAR_INPUTS_PAGE_MAX_BYTES, + type SidebarInputsChangedEvent, + type SidebarInputsDomainChange, + type SidebarInputsReadParams, + type SidebarInputsReadResult +} from './sidebar-inputs-feed-contract' + +/** The inputs a reader has installed whole: every domain's entities at one revision. Its consumer + * compares the versions with its own and shows "Reopen to update" outside its range. */ +export type InstalledSidebarInputs = { + generation: string + revision: number + schemaVersion: number + semanticsVersion: number + domains: ReadonlyMap> +} + +type StagedBatch = Pick & { + pages: readonly SidebarInputsReadResult['domains'][] +} + +type FeedPosition = Pick + +/** + * A phone's feed reader for one host connection. One read is awaited at a time and a refresh + * supersedes it; a notice the read may not cover, during paging included, costs one more read. + */ +export type SidebarInputsFeedState = { + installed: InstalledSidebarInputs | null + /** The pages of the batch being read, installed only once its last page lands. */ + staged: StagedBatch | null + /** The read in flight; a reply to any other request is stale and dropped. */ + inFlight: number | null + /** The newest notice that arrived after the read in flight was sent. */ + noticed: FeedPosition | null + lastRequestId: number +} + +export type SidebarInputsFeedRequest = { id: number; params: SidebarInputsReadParams } + +export type SidebarInputsFeedStep = { + state: SidebarInputsFeedState + /** The read to send now, if any. */ + request: SidebarInputsFeedRequest | null +} + +export const INITIAL_SIDEBAR_INPUTS_FEED_STATE: SidebarInputsFeedState = { + installed: null, + staged: null, + inFlight: null, + noticed: null, + lastRequestId: 0 +} + +function covers(position: FeedPosition | null, notice: FeedPosition): boolean { + return position?.generation === notice.generation && position.revision >= notice.revision +} + +function send( + state: SidebarInputsFeedState, + params: Omit +): SidebarInputsFeedStep { + const id = state.lastRequestId + 1 + return { + state: { ...state, inFlight: id, lastRequestId: id }, + request: { id, params: { ...params, maxBytes: SIDEBAR_INPUTS_PAGE_MAX_BYTES } } + } +} + +/** A fresh read: the host serves it at its current revision, past every notice so far. */ +function readFromInstalled(state: SidebarInputsFeedState): SidebarInputsFeedStep { + const { installed } = state + return send( + { ...state, staged: null, noticed: null }, + installed ? { generation: installed.generation, cursor: installed.revision } : {} + ) +} + +/** On every stream ready and return to the foreground; supersedes a read in flight. */ +export function refreshSidebarInputs(state: SidebarInputsFeedState): SidebarInputsFeedStep { + return readFromInstalled(state) +} + +/** A change notice; one this reader already has or is fetching is a duplicate or arrived late. */ +export function sidebarInputsChanged( + state: SidebarInputsFeedState, + notice: FeedPosition +): SidebarInputsFeedStep { + if (covers(state.staged ?? state.installed, notice) || covers(state.noticed, notice)) { + return { state, request: null } + } + return state.inFlight === null + ? readFromInstalled(state) + : { state: { ...state, noticed: notice }, request: null } +} + +/** The read failed; the next ready, foreground or notice reads again. */ +export function sidebarInputsReadFailed( + state: SidebarInputsFeedState, + requestId: number +): SidebarInputsFeedState { + return requestId === state.inFlight + ? { ...state, inFlight: null, staged: null, noticed: null } + : state +} + +export function sidebarInputsReplied( + state: SidebarInputsFeedState, + requestId: number, + reply: SidebarInputsReadResult +): SidebarInputsFeedStep { + if (requestId !== state.inFlight) { + return { state, request: null } + } + const settled = { ...state, inFlight: null } + if (reply.kind === 'reset') { + return readFromInstalled(settled) + } + const { staged, installed } = state + const continues = staged + ? staged.generation === reply.generation && + staged.revision === reply.revision && + staged.kind === reply.kind && + staged.from === reply.from + : reply.kind === 'snapshot' || + (installed?.generation === reply.generation && reply.from === installed.revision) + if (!continues) { + // Not the batch this reader asked for: only a whole snapshot is safe now. + return send({ ...settled, staged: null, noticed: null }, {}) + } + const batch: StagedBatch = { + generation: reply.generation, + revision: reply.revision, + kind: reply.kind, + from: reply.from, + pages: [...(staged?.pages ?? []), reply.domains] + } + if (reply.next !== null) { + return send({ ...settled, staged: batch }, { pageToken: reply.next }) + } + // A batch older than what is installed (a late snapshot) installs nothing. + const stale = installed?.generation === batch.generation && batch.revision < installed.revision + const next = stale ? installed : installBatch(installed, batch, reply) + if (!next) { + return send({ ...settled, staged: null, noticed: null }, {}) + } + const done = { ...settled, staged: null, installed: next } + return done.noticed && !covers(next, done.noticed) + ? readFromInstalled(done) + : { state: { ...done, noticed: null }, request: null } +} + +/** Applies a whole batch as one step; null when an entity's joined parts are not JSON. */ +function installBatch( + installed: InstalledSidebarInputs | null, + batch: StagedBatch, + last: Pick +): InstalledSidebarInputs | null { + const domains = new Map(batch.kind === 'snapshot' ? [] : (installed?.domains ?? [])) + const touched = new Map>() + const parts = new Map>() + const entitiesOf = (name: string): Map => { + let entities = touched.get(name) + if (!entities) { + entities = new Map(domains.get(name) ?? []) + touched.set(name, entities) + } + return entities + } + for (const page of batch.pages) { + for (const [name, change] of Object.entries(page)) { + applyDomainChange(entitiesOf(name), change, parts, name) + } + } + for (const [name, texts] of parts) { + for (const [key, text] of texts) { + try { + entitiesOf(name).set(key, JSON.parse(text)) + } catch { + return null + } + } + } + for (const [name, entities] of touched) { + domains.set(name, entities) + } + return { + generation: batch.generation, + revision: batch.revision, + schemaVersion: last.schemaVersion, + semanticsVersion: last.semanticsVersion, + domains + } +} + +function applyDomainChange( + entities: Map, + change: SidebarInputsDomainChange, + parts: Map>, + name: string +): void { + for (const [key, value] of Object.entries(change.upserts)) { + entities.set(key, value) + } + for (const key of change.removals) { + entities.delete(key) + } + for (const [key, text] of Object.entries(change.parts)) { + const texts = parts.get(name) ?? new Map() + parts.set(name, texts) + texts.set(key, (texts.get(key) ?? '') + text) + } +} diff --git a/tests/e2e/sidebar-inputs-feed.spec.ts b/tests/e2e/sidebar-inputs-feed.spec.ts new file mode 100644 index 00000000000..a5566301a60 --- /dev/null +++ b/tests/e2e/sidebar-inputs-feed.spec.ts @@ -0,0 +1,111 @@ +import { test, expect } from './helpers/orca-app' +import { SIDEBAR_INPUTS_BUILD } from './helpers/sidebar-shadow' +import { waitForActiveWorktree, waitForSessionReady } from './helpers/store' +import { createDesktopMobilePairingOffer } from './helpers/paired-electron-client' +import { pairMobileClient, type PairedMobileSocket } from './helpers/paired-mobile-client' +import type { SidebarInputsReadResult } from '../../src/shared/sidebar-list/sidebar-inputs-feed-contract' +import { + INITIAL_SIDEBAR_INPUTS_FEED_STATE, + refreshSidebarInputs, + sidebarInputsChanged, + sidebarInputsReplied, + type SidebarInputsFeedState, + type SidebarInputsFeedStep +} from '../../src/shared/sidebar-list/sidebar-inputs-feed-installer' + +/** Sends each read the installer asks for over the phone's one socket until none is left. */ +async function settle(socket: PairedMobileSocket, step: SidebarInputsFeedStep) { + const kinds: string[] = [] + let { state, request } = step + while (request) { + const frameId = `feed-${request.id}` + socket.send(frameId, 'sidebarInputs.read', request.params) + await expect.poll(() => socket.frames.some((frame) => frame.id === frameId)).toBe(true) + const frame = socket.frames.find((candidate) => candidate.id === frameId) + expect(frame, JSON.stringify(frame?.error)).toMatchObject({ ok: true }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: an ok sidebarInputs.read frame carries its result. + const reply = frame?.result as SidebarInputsReadResult + kinds.push(reply.kind) + ;({ state, request } = sidebarInputsReplied(state, request.id, reply)) + } + return { state, kinds } +} + +function pinned(state: SidebarInputsFeedState): unknown { + return state.installed?.domains.get('desktop')?.get('pinnedWorkspaceIds') +} + +test.describe('switch on', () => { + test.skip( + !SIDEBAR_INPUTS_BUILD, + 'sidebarInputs.read exists only in an ORCA_SIDEBAR_INPUTS=main build' + ) + + test('the host’s own phone reads a whole snapshot, then a delta after a change', async ({ + orcaPage + }) => { + await waitForSessionReady(orcaPage) + const worktreeId = await waitForActiveWorktree(orcaPage) + const phone = pairMobileClient(await createDesktopMobilePairingOffer(orcaPage)) + try { + const socket = await phone.openSocket() + socket.send('events', 'runtime.clientEvents.subscribe', null) + + const first = await settle(socket, refreshSidebarInputs(INITIAL_SIDEBAR_INPUTS_FEED_STATE)) + expect(first.kinds.every((kind) => kind === 'snapshot')).toBe(true) + const installed = first.state.installed + expect(installed?.domains.get('temporal')?.has('sortSnapshot')).toBe(true) + expect(pinned(first.state)).not.toContain(worktreeId) + + await orcaPage.evaluate( + (id) => + window.api.runtime.call({ + method: 'worktree.set', + params: { worktree: `id:${id}`, isPinned: true } + }), + worktreeId + ) + const notice = (): { generation: string; revision: number } | undefined => + socket.frames + .map((frame) => frame.result) + .findLast( + ( + result + ): result is { type: 'sidebarInputsChanged'; generation: string; revision: number } => + typeof result === 'object' && + result !== null && + 'type' in result && + result.type === 'sidebarInputsChanged' && + 'revision' in result && + typeof result.revision === 'number' && + result.revision > (installed?.revision ?? 0) + ) + await expect.poll(() => notice() !== undefined).toBe(true) + const event = notice() + if (!event) { + throw new Error('no change notice') + } + const second = await settle(socket, sidebarInputsChanged(first.state, event)) + expect(second.kinds).toEqual(['delta']) + expect(second.state.installed?.generation).toBe(installed?.generation) + expect(pinned(second.state)).toContain(worktreeId) + } finally { + phone.dispose() + } + }) +}) + +test.describe('switch off', () => { + test.skip(SIDEBAR_INPUTS_BUILD, 'checks the shipped method set') + + test('a switch-off build refuses sidebarInputs.read to the phone', async ({ orcaPage }) => { + await waitForSessionReady(orcaPage) + const phone = pairMobileClient(await createDesktopMobilePairingOffer(orcaPage)) + try { + const response = await phone.request('sidebarInputs.read', { maxBytes: 262_144 }) + expect(response).toMatchObject({ ok: false }) + } finally { + phone.dispose() + } + }) +})