feat(sidebar): dark phone feed of the sidebar inputs (sidebarInputs.read) (#30161)

* feat(sidebar): dark phone feed of the sidebar inputs (sidebarInputs.read)

* fix(sidebar): feed review: fresh read on ready, notices kept per revision, one flush owner, closed connections never subscribe

* fix(sidebar): feed skips values JSON cannot hold; a failed read forgets its notice

* fix(sidebar): name the feed batch key by where the batch starts from
This commit is contained in:
Jinwoo Hong
2026-10-17 20:46:10 -07:00
committed by GitHub
parent 1b3113de8e
commit 3aad2b17b0
28 changed files with 2136 additions and 55 deletions
+1
View File
@@ -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",
+22 -4
View File
@@ -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(
@@ -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<string, () => 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)
})
})
+34 -16
View File
@@ -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: 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',
@@ -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
@@ -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()
})
})
@@ -34,7 +34,7 @@ export abstract class OrcaRuntimeWithSidebarTemporal extends OrcaRuntimeWithWork
constructor(...args: ConstructorParameters<typeof OrcaRuntimeWithWorkspaceReviewFacts>) {
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)
}
@@ -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<string, ServerEntry>()
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()
}
}
@@ -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)
}
}
}
@@ -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: Producer | null): Producer {
if (producer === null) {
throw new Error('sidebar inputs are off in this build')
}
return producer
}
@@ -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<RuntimeClientEvent['type']>([
'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
}
]
}
@@ -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<string, Map<string, FeedEntity | null>>
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<FeedChange> {
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<string, FeedEntity | null>()
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]
}
}
}
}
@@ -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 }
])
})
})
@@ -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<string, SidebarInputsDomainChange>; 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<FeedChange>, 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
}
@@ -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<string, Pin>()
/** 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}`
}
@@ -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<string, unknown> = {},
options: { rereadOnRead?: boolean } = {}
) {
let entities: Record<string, unknown> = { ...initial }
const watchers = new Set<() => void>()
let reads = 0
let failing = false
let gate: Promise<void> | 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<SidebarInputsChangedEvent, 'generation' | 'revision'>): 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<SidebarInputsReadResult> {
const { reply, deliver } = await this.serveOne()
deliver()
return reply
}
async settle(): Promise<void> {
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<string, Record<string, unknown>> {
const view: Record<string, Record<string, unknown>> = {}
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<typeof testDomain>[]) {
const truth: Record<string, Record<string, unknown>> = {}
for (const domain of domains) {
if (Object.keys(domain.entities()).length > 0) {
truth[domain.domain.name] = domain.entities()
}
}
return truth
}
@@ -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<string, unknown> = { 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())
})
})
@@ -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<string, unknown> | Promise<Record<string, unknown>>
/** 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<string>()
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<SidebarInputsReadResult> {
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<string, Map<string, FeedEntity>>()
private readonly dirty = new Set<string>()
// 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<typeof setTimeout> | null = null
private flushing: Promise<void> | 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<SidebarInputsReadResult> {
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<void> {
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<void> {
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<string, string> | null): Map<string, FeedEntity | null> {
const changed = new Map<string, FeedEntity | null>()
if (!read) {
return changed
}
const before = this.published.get(name) ?? new Map<string, FeedEntity>()
const after = new Map<string, FeedEntity>()
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<FeedChange> {
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<FeedChange>
): 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<Map<string, string> | null> {
try {
const entities = new Map<string, string>()
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
}
}
@@ -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()
}
}
}
@@ -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<string, AgentPane>()
private readonly agentPanesByWorktree = new Map<string, Map<string, AgentPane>>()
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.
@@ -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<string, WorkspaceCreation>()
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 }
}
}
@@ -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
+2
View File
@@ -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,
+3
View File
@@ -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).
@@ -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<typeof SidebarInputsReadParams>
/** 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<typeof SidebarInputsDomainChangeSchema>
export type SidebarInputsReadResult = z.infer<typeof SidebarInputsReadResultSchema>
/** Carries no data: a reader re-reads through sidebarInputs.read. */
export type SidebarInputsChangedEvent = {
type: 'sidebarInputsChanged'
generation: string
revision: number
}
@@ -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<SidebarInputsReadResult> & Pick<SidebarInputsReadResult, 'kind' | 'revision'>
): SidebarInputsReadResult {
return {
generation: 'g1',
schemaVersion: SIDEBAR_INPUTS_SCHEMA_VERSION,
semanticsVersion: SIDEBAR_INPUTS_SEMANTICS_VERSION,
from: null,
domains: {},
next: null,
...fields
}
}
const rows = (upserts: Record<string, unknown>, 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)
})
})
@@ -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<string, ReadonlyMap<string, unknown>>
}
type StagedBatch = Pick<SidebarInputsReadResult, 'generation' | 'revision' | 'kind' | 'from'> & {
pages: readonly SidebarInputsReadResult['domains'][]
}
type FeedPosition = Pick<SidebarInputsChangedEvent, 'generation' | 'revision'>
/**
* 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<SidebarInputsReadParams, 'maxBytes'>
): 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<SidebarInputsReadResult, 'schemaVersion' | 'semanticsVersion'>
): InstalledSidebarInputs | null {
const domains = new Map(batch.kind === 'snapshot' ? [] : (installed?.domains ?? []))
const touched = new Map<string, Map<string, unknown>>()
const parts = new Map<string, Map<string, string>>()
const entitiesOf = (name: string): Map<string, unknown> => {
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<string, unknown>,
change: SidebarInputsDomainChange,
parts: Map<string, Map<string, string>>,
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<string, string>()
parts.set(name, texts)
texts.set(key, (texts.get(key) ?? '') + text)
}
}
+111
View File
@@ -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()
}
})
})