test: remove classification inventories and verify settled boundaries (#32216)

Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com>
This commit is contained in:
OrcaWin
2026-10-23 00:03:24 -07:00
committed by GitHub
co-authored by OrcaWin
parent a5b2543d85
commit 0841e09b64
8 changed files with 129 additions and 709 deletions
@@ -1,183 +0,0 @@
import { readdirSync, readFileSync } from 'node:fs'
import { join, sep } from 'node:path'
import { fileURLToPath } from 'node:url'
import ts from 'typescript-api'
import { afterEach, describe, expect, it } from 'vitest'
import { RelayAssignmentStore } from './assignment-store.js'
import type { RelayCellConfig } from './config.js'
import { openInMemoryRelayDatabase, type RelayDatabase } from './database.js'
// Step 6 (design §3.5): a phone on a `phonePath=memory` cell writes no `confirmation:` or
// `splice:` lease. Every reader of the lease table is classified by why that is safe; a new
// reader fails this test until someone decides. While the cell is recorded as reserve its
// control holds no lease either, so step 5's guards already cover it; after a flip back a
// memory phone looks like today's phone once its 90 s lease has expired.
const ACTIVITY_LEASE_READERS: Record<string, string> = {
// Reads control leases only, so phone leases never change the answer.
'assignment-store.ts:activateControlOnce': 'control-only',
'assignment-store.ts:reconcileIdleRegionalRehome': 'control-only',
'assignment-store.ts:renewTransactionalControlActivity': 'control-only',
'assignment-store.ts:startRegionalRehomeCandidate': 'control-only',
'control-renewal-statement.ts:CONTROL_RENEWAL_BATCH_SQL': 'control-only',
'idle-regional-rehome-selection.ts:selectIdleRegionalRehomes': 'control-only',
// Refuses or skips a cell recorded as reserve (or an unreadable set).
'assignment-store.ts:cellDeploymentStatus': 'refuses-or-skips-reserve-cells',
'assignment-store.ts:cellEvacuationCapacity': 'refuses-or-skips-reserve-cells',
'assignment-store.ts:releaseExpiredActivity': 'refuses-or-skips-reserve-cells',
'assignment-store.ts:releaseExpiredActivityLeases': 'refuses-or-skips-reserve-cells',
'assignment-store.ts:startEvacuation': 'refuses-or-skips-reserve-cells',
// Its only caller drops candidates whose source or target is a reserve cell.
'idle-regional-rehome-selection.ts:sourceRequestUnits': 'refuses-or-skips-reserve-cells',
// Only hosts in an open drain, migration or rehome, which no reserve cell can have:
// setCellAdmitMode refuses reserve while one is open, and every start refuses a reserve cell.
'assignment-store.ts:activeCellMigrations': 'open-flow-only',
'assignment-store.ts:additionalActivityCellIds': 'open-flow-only',
'assignment-store.ts:assertCellReservationAccounting': 'open-flow-only',
'assignment-store.ts:beginCellDrainSend': 'open-flow-only',
'assignment-store.ts:beginCellDrainSendOnce': 'open-flow-only',
'assignment-store.ts:completeReadyEvacuations': 'open-flow-only',
'assignment-store.ts:completeReadyRegionalRehomes': 'open-flow-only',
'assignment-store.ts:reconcileDrainMigrationAccounting': 'open-flow-only',
'assignment-store.ts:reconcileReservationAccounting': 'open-flow-only',
'assignment-store.ts:refreshDrainMigrationLeasesOnce': 'open-flow-only',
'registered-migration-abandonment.ts:ABANDONED_REGISTERED_MIGRATION': 'open-flow-only',
// One host's rows under lock; each caller is classified in the reserve-mode census.
'assignment-store.ts:lockAssignmentActivities': 'host-lock',
// Metrics and previews: they report, and act on nothing.
'assignment/assignment-inventory-snapshot.ts:readAssignmentInventorySnapshot': 'report-only',
'region-correction-preview.ts:previewRegionalRehomeEligibility': 'report-only',
'registered-migration-inventory.ts:readRegisteredMigrationInventory': 'report-only',
// Step 5's existing-only stranded release, fixed separately in #30609 before the first flip.
'assignment-store.ts:assignmentStrandedOnUnservedCell': 'step-5-stranded-release'
}
// A read is FROM, JOIN or USING on the table; `DELETE FROM` alone only writes. Not caught: a
// second read inside an already listed function, and counters derived from leases elsewhere.
const LEASE_TABLE = 'relay_assignment_activity_leases'
// Bare table-name values that never reach SQL.
const BARE_TABLE_NAME_USES: Record<string, string> = {
'database.ts:POSTGRES_TRANSACTION_PHASES': 'labels a failed statement by the table it names'
}
const LEASE_READ = /(?<!DELETE\s+)\b(?:FROM|JOIN|USING)\s+relay_assignment_activity_leases\b/
const CELLS: RelayCellConfig[] = [
{ id: 'cell-a', url: 'https://relay-a.example.com', capacityRequests: 10 },
{ id: 'cell-b', url: 'https://relay-b.example.com', capacityRequests: 10 }
]
describe('activity-lease reader census', () => {
let database: RelayDatabase | undefined
afterEach(async () => {
await database?.close()
database = undefined
})
it('classifies every reader of the activity-lease table', () => {
const { readers, bareTableNames } = activityLeaseReaders()
// A bare name could reach `FROM ${table}`, which the scan cannot see: write the table into the SQL.
expect(bareTableNames.filter((use) => !(use in BARE_TABLE_NAME_USES))).toEqual([])
expect(readers.filter((name) => !(name in ACTIVITY_LEASE_READERS))).toEqual([])
expect(Object.keys(ACTIVITY_LEASE_READERS).filter((name) => !readers.includes(name))).toEqual([])
})
async function setup(now: () => number) {
database = await openInMemoryRelayDatabase()
const store = new RelayAssignmentStore(database, now)
await store.reconcileCells(CELLS)
return store
}
// A host whose source cell kept no lease for it, as a memory cell's control and phones do.
async function registeredEvacuationWithoutSourceLeases(store: RelayAssignmentStore) {
const identity = { userId: 'user-a', relayHostId: 'host000000000001' }
const first = await store.assign(identity)
expect(first.cellId).toBe('cell-a')
const migration = await store.startEvacuation(identity, 'cell-b')
await store.activateControl(identity, {
cellId: 'cell-b',
assignmentEpoch: migration.assignmentEpoch,
generation: 1
})
await store.markMigrationTargetRegistered(identity, {
cellId: 'cell-b',
assignmentEpoch: migration.assignmentEpoch
})
await store.releaseActivity(identity, `control-pending:${first.assignmentEpoch}`)
return { identity, migration }
}
it('completes no evacuation from a reserve source, though it holds no source lease', async () => {
const now = () => 1_000
const store = await setup(now)
const { identity, migration } = await registeredEvacuationWithoutSourceLeases(store)
// setCellAdmitMode refuses this with the migration open; written directly to reach the second guard.
await database!.query(
`INSERT INTO relay_cell_admit_modes (cell_id, admit_mode, updated_at) VALUES ('cell-a', 'reserve', 1000)`
)
const fresh = new RelayAssignmentStore(database!, now)
await expect(fresh.completeEvacuation(identity, migration.assignmentEpoch)).rejects.toThrow(
'cell_in_reserve_mode'
)
await expect(fresh.completeReadyEvacuations()).rejects.toThrow('cell_in_reserve_mode')
expect(
await database!.query(`SELECT completed_at FROM relay_assignment_migrations`)
).toEqual([{ completed_at: null }])
// The same host on a db source completes: the guard, not the fixture, held it.
await store.setCellAdmitMode('cell-a', 'db')
expect(await store.completeReadyEvacuations()).toBe(1)
})
it('refuses evacuation status for a reserve cell, where activeCellMigrations would count no units', async () => {
const store = await setup(() => 1_000)
await store.setCellAdmitMode('cell-a', 'reserve')
await expect(store.cellEvacuationStatus('cell-a', 'cell-b', true)).rejects.toThrow(
'cell_in_reserve_mode'
)
await expect(store.cellEvacuationStatus('cell-b', 'cell-a', false)).rejects.toThrow(
'cell_in_reserve_mode'
)
})
})
function activityLeaseReaders(): { readers: string[]; bareTableNames: string[] } {
const directory = fileURLToPath(new URL('.', import.meta.url))
const readers = new Set<string>()
const bareTableNames: string[] = []
const files = readdirSync(directory, { recursive: true, encoding: 'utf8' }).filter(
(file) => file.endsWith('.ts') && !file.includes('.test.') && !file.split(/[\\/]/).includes('test-fixtures')
)
for (const relativeFile of files) {
const file = relativeFile.split(sep).join('/')
const text = readFileSync(join(directory, file), 'utf8')
if (!text.includes(LEASE_TABLE)) continue
const source = ts.createSourceFile(file, text, ts.ScriptTarget.ES2022, true)
const visit = (node: ts.Node): void => {
if (ts.isNoSubstitutionTemplateLiteral(node) || ts.isTemplateExpression(node) || ts.isStringLiteral(node)) {
if ((ts.isStringLiteral(node) || ts.isNoSubstitutionTemplateLiteral(node)) && node.text.trim() === LEASE_TABLE) {
bareTableNames.push(`${file}:${enclosingName(node)}`)
}
if (LEASE_READ.test(node.getText())) {
readers.add(`${file}:${enclosingName(node)}`)
return
}
}
ts.forEachChild(node, visit)
}
visit(source)
}
return { readers: [...readers].sort(), bareTableNames }
}
function enclosingName(node: ts.Node): string {
for (let current = node.parent; current; current = current.parent) {
if ((ts.isMethodDeclaration(current) || ts.isFunctionDeclaration(current)) && current.name) {
return current.name.getText()
}
// A module-level SQL constant, never a local `const rows = ...` inside a method.
if (ts.isVariableStatement(current) && ts.isSourceFile(current.parent)) {
return current.declarationList.declarations.map((declaration) => declaration.name.getText()).join(',')
}
}
return '<module>'
}
@@ -0,0 +1,76 @@
import { afterEach, describe, expect, it } from 'vitest'
import { RelayAssignmentStore } from './assignment-store.js'
import type { RelayCellConfig } from './config.js'
import { openInMemoryRelayDatabase, type RelayDatabase } from './database.js'
const CELLS: RelayCellConfig[] = [
{ id: 'cell-a', url: 'https://relay-a.example.com', capacityRequests: 10 },
{ id: 'cell-b', url: 'https://relay-b.example.com', capacityRequests: 10 }
]
describe('evacuation from reserve-mode cells', () => {
let database: RelayDatabase | undefined
afterEach(async () => {
await database?.close()
database = undefined
})
async function setup(now: () => number) {
database = await openInMemoryRelayDatabase()
const store = new RelayAssignmentStore(database, now)
await store.reconcileCells(CELLS)
return store
}
// A host whose source cell kept no lease for it, as a memory cell's control and phones do.
async function registeredEvacuationWithoutSourceLeases(store: RelayAssignmentStore) {
const identity = { userId: 'user-a', relayHostId: 'host000000000001' }
const first = await store.assign(identity)
expect(first.cellId).toBe('cell-a')
const migration = await store.startEvacuation(identity, 'cell-b')
await store.activateControl(identity, {
cellId: 'cell-b',
assignmentEpoch: migration.assignmentEpoch,
generation: 1
})
await store.markMigrationTargetRegistered(identity, {
cellId: 'cell-b',
assignmentEpoch: migration.assignmentEpoch
})
await store.releaseActivity(identity, `control-pending:${first.assignmentEpoch}`)
return { identity, migration }
}
it('completes no evacuation from a reserve source, though it holds no source lease', async () => {
const now = () => 1_000
const store = await setup(now)
const { identity, migration } = await registeredEvacuationWithoutSourceLeases(store)
// setCellAdmitMode refuses this with the migration open; written directly to reach the second guard.
await database!.query(
`INSERT INTO relay_cell_admit_modes (cell_id, admit_mode, updated_at) VALUES ('cell-a', 'reserve', 1000)`
)
const fresh = new RelayAssignmentStore(database!, now)
await expect(fresh.completeEvacuation(identity, migration.assignmentEpoch)).rejects.toThrow(
'cell_in_reserve_mode'
)
await expect(fresh.completeReadyEvacuations()).rejects.toThrow('cell_in_reserve_mode')
expect(
await database!.query(`SELECT completed_at FROM relay_assignment_migrations`)
).toEqual([{ completed_at: null }])
// The same host on a db source completes: the guard, not the fixture, held it.
await store.setCellAdmitMode('cell-a', 'db')
expect(await store.completeReadyEvacuations()).toBe(1)
})
it('refuses evacuation status for a reserve cell, where activeCellMigrations would count no units', async () => {
const store = await setup(() => 1_000)
await store.setCellAdmitMode('cell-a', 'reserve')
await expect(store.cellEvacuationStatus('cell-a', 'cell-b', true)).rejects.toThrow(
'cell_in_reserve_mode'
)
await expect(store.cellEvacuationStatus('cell-b', 'cell-a', false)).rejects.toThrow(
'cell_in_reserve_mode'
)
})
})
@@ -1,109 +1,15 @@
import { readFileSync } from 'node:fs'
import { fileURLToPath } from 'node:url'
import { ASSIGNMENT_LIMITS } from '@orca-cloud/relay-contract'
import ts from 'typescript-api'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { RelayAssignmentStore, RelayCellInReserveModeError } from './assignment-store.js'
import type { RelayCellConfig } from './config.js'
import { openInMemoryRelayDatabase, type RelayDatabase } from './database.js'
// Step 5 census (design §6.3): a reserve-mode cell keeps its hosts in memory, so Postgres's
// per-host state for it is stale. Every store method is classified below; a new one fails
// this test until someone decides how it treats a reserve-mode cell.
const RESERVE_MODE_CENSUS: Record<string, string> = {
// One host's connect, resolve or lease: a reserve-mode cell answers these from memory, and
// its ledger row is written after the fact.
acquireActivity: 'connect-path',
activateControl: 'connect-path',
// A flip back's re-registration: the cell is leaving reserve mode, so its rows are rebuilt.
activateControlDeferringCell: 'connect-path',
assign: 'connect-path',
changeActivity: 'connect-path',
exchangeRegionCorrection: 'connect-path',
hostWhereabouts: 'connect-path',
markMigrationTargetRegistered: 'connect-path',
releaseActivity: 'connect-path',
renewControlActivities: 'connect-path',
renewControlActivity: 'connect-path',
resolve: 'connect-path',
verifyCellAssignment: 'connect-path',
// Sweeps and per-cell host readers: skip reserve-mode cells, or refuse them.
cellDeploymentStatus: 'refuses-reserve-cells',
cellEvacuationCapacity: 'refuses-reserve-cells',
cellEvacuationStatus: 'refuses-reserve-cells',
evacuateDeadCells: 'skips-reserve-cells',
prepareCellDrainAttempt: 'refuses-reserve-cells',
rebalanceDormant: 'refuses-reserve-cells',
releaseExpiredActivity: 'skips-reserve-cells',
releaseExpiredActivityLeases: 'skips-reserve-cells',
selectIdleRegionalRehomeCandidates: 'skips-reserve-cells',
startActiveCellEvacuations: 'refuses-reserve-cells',
startEvacuation: 'refuses-reserve-cells',
// Steps of a drain, migration or rehome that only exist once a refused start above began one.
abortExpiredEvacuations: 'follows-a-refused-start',
abortExpiredRegionalRehomes: 'follows-a-refused-start',
abortUnarrivedRegionalRehomes: 'follows-a-refused-start',
beginCellDrainSend: 'follows-a-refused-start',
commitIdleRegionalRehome: 'follows-a-refused-start',
// A reserve source keeps no activity rows, so its "no source units" check proves nothing.
completeEvacuation: 'refuses-reserve-cells',
completeEvacuationFromDeadSource: 'refuses-reserve-cells',
completeReadyEvacuations: 'follows-a-refused-start',
completeReadyRegionalRehomes: 'follows-a-refused-start',
prepareCellDrainRecovery: 'follows-a-refused-start',
proveCellDrainNotDelivered: 'follows-a-refused-start',
reapRegionalRehomeAttempts: 'follows-a-refused-start',
reconcileIdleRegionalRehome: 'follows-a-refused-start',
recordCellDrainApplicationReceipt: 'follows-a-refused-start',
refreshRegionalRehomeLeases: 'follows-a-refused-start',
supersedeRegisteredCellEvacuations: 'follows-a-refused-start',
supersedeRegisteredEvacuation: 'follows-a-refused-start',
// Cell-level state, fences and controls: no per-host rows.
abortCellFenceAttempt: 'cell-level',
addMigrationCells: 'cell-level',
adoptLegacyCellFence: 'cell-level',
applyCellAdmissionSelector: 'cell-level',
applyRegionalRehomeControl: 'cell-level',
cellAdmitMode: 'cell-level',
cellHeartbeatFresh: 'cell-level',
recordCellAdmitEffective: 'cell-level',
commitCellReservationDelta: 'cell-level',
attestCellFence: 'cell-level',
attestCellFenceAttempt: 'cell-level',
bindCellFencePlanGeneration: 'cell-level',
cellFenceAttempt: 'cell-level',
commitLegacyCellFenceAdoption: 'cell-level',
configureCell: 'cell-level',
disableRegionalRehomeControl: 'cell-level',
inspectCellAdmissionSelector: 'cell-level',
inspectRegionalRehomeControl: 'cell-level',
prepareCellFenceAttempt: 'cell-level',
previewRegionCorrection: 'cell-level',
previewRegionalRehomeEligibility: 'cell-level',
pruneReleasedControlReservations: 'cell-level',
reconcileCells: 'cell-level',
reconcileCellsAtStartup: 'cell-level',
reserveModeCells: 'cell-level',
recordCellFenceOperation: 'cell-level',
recordCellHeartbeat: 'cell-level',
recordCellRegionalRehomeStatus: 'cell-level',
regionCatalog: 'cell-level',
regionCorrectionOutcomes: 'cell-level',
regionalRehomeFleetSafety: 'cell-level',
releaseExpiredRegionPreferences: 'cell-level',
seatFeedCells: 'cell-level',
setCellAdmissionState: 'cell-level',
setCellAdmitMode: 'cell-level',
setCellEnabled: 'cell-level',
startCellFenceApply: 'cell-level'
}
const CELLS: RelayCellConfig[] = [
{ id: 'cell-a', url: 'https://relay-a.example.com', capacityRequests: 10 },
{ id: 'cell-b', url: 'https://relay-b.example.com', capacityRequests: 10 }
]
describe('reserve-mode census', () => {
describe('assignment reserve-mode behavior', () => {
let database: RelayDatabase | undefined
afterEach(async () => {
@@ -111,12 +17,6 @@ describe('reserve-mode census', () => {
database = undefined
})
it('classifies every public store method for reserve-mode cells', () => {
const methods = publicStoreMethods()
expect(methods.filter((name) => !(name in RESERVE_MODE_CENSUS))).toEqual([])
expect(Object.keys(RESERVE_MODE_CENSUS).filter((name) => !methods.includes(name))).toEqual([])
})
async function setup(now: () => number, reserve: string[] = []) {
database = await openInMemoryRelayDatabase()
const store = new RelayAssignmentStore(database, now)
@@ -385,24 +285,3 @@ describe('reserve-mode census', () => {
expect(await store.cellAdmitMode('cell-a')).toMatchObject({ admitMode: 'reserve', updatedAt: 1_000 })
})
})
function publicStoreMethods(): string[] {
const path = fileURLToPath(new URL('./assignment-store.ts', import.meta.url))
const source = ts.createSourceFile(path, readFileSync(path, 'utf8'), ts.ScriptTarget.ES2022, true)
const names = new Set<string>()
source.forEachChild((node) => {
if (!ts.isClassDeclaration(node) || node.name?.text !== 'RelayAssignmentStore') return
for (const member of node.members) {
if (!ts.isMethodDeclaration(member) || !ts.isIdentifier(member.name)) continue
const hidden = ts
.getModifiers(member)
?.some(
(modifier) =>
modifier.kind === ts.SyntaxKind.PrivateKeyword ||
modifier.kind === ts.SyntaxKind.ProtectedKeyword
)
if (!hidden) names.add(member.name.text)
}
})
return [...names].sort()
}
@@ -423,14 +423,12 @@ What is still not covered: what the count-based raw-port inventory covers instea
reach `sendRequest`, and how often), native storage, transport skew, and the mutations listed under
_Known-open holes_ below.
Which subscriptions are covered is no longer stated here. It is held as data in
`mobile/src/transport/rpc-subscription-inventory.ts`, where every product `client.subscribe` is
classified as recorded, an unwritten scenario, or walled with the wall named, and
`rpc-subscription-boundary.test.ts` fails on a new site, a stale entry, a wrong method and a
`recorded` entry naming a family this manifest does not have. This paragraph is why: it said nine
sites when there were ten — the count was taken over `mobile/src`, and the host screen's
`accounts.subscribe` lives under `app/`. A count in prose cannot fail. Today four of the ten are
recorded, two are unwritten scenarios and four are walled, and the list is what says so.
Subscription coverage requires a scenario that drives the real consumer. The manifest includes
notifications, live worktree names, native chat, and host worktree refresh. Structured transcript,
structured status, and session tab streams still need scenarios. Account streams need native UI
or multi-host support; terminal and browser streams need device effects the recorder cannot model.
`rpc-client-ready-stream-release.test.ts` checks release before and after `ready` through both
direct and relay transports.
The frame plumbing is method-agnostic, so what stops a site is its consumer rather than the runner.
Blur is unrecorded across all ten subscribing sites: `useFocusEffect` is substituted as `useEffect`,
@@ -3,7 +3,7 @@ import { negotiateMobileRuntimeCapabilities } from './mobile-runtime-capability-
import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
import type { RpcResponse } from './types'
function negotiate(args: { reject: unknown; current?: boolean }): {
function negotiate(args: { reject: unknown }): {
onReady: ReturnType<typeof vi.fn>
onFailure: ReturnType<typeof vi.fn>
} {
@@ -11,7 +11,7 @@ function negotiate(args: { reject: unknown; current?: boolean }): {
const onFailure = vi.fn()
negotiateMobileRuntimeCapabilities({
sendRequest: () => Promise.reject(args.reject),
current: () => args.current ?? true,
current: () => true,
onReady,
onFailure
})
@@ -44,26 +44,42 @@ describe('mobile runtime capability negotiation', () => {
expect(onReady).not.toHaveBeenCalled()
})
it('leaves a replaced session alone on an unanswered request', async () => {
const timedOut = markRpcDeliveryUnknown(new Error('Request timed out'))
const { onReady, onFailure } = negotiate({ reject: timedOut, current: false })
it.each(['successful', 'unanswered', 'unsent'] as const)(
'leaves a replaced session alone after a %s request settles',
async (outcome) => {
const onReady = vi.fn()
const onFailure = vi.fn()
let resolveRequest!: (response: RpcResponse) => void
let rejectRequest!: (error: Error) => void
const request = new Promise<RpcResponse>((resolve, reject) => {
resolveRequest = resolve
rejectRequest = reject
})
let current = true
let checkedCurrent!: () => void
const settled = new Promise<void>((resolve) => {
checkedCurrent = resolve
})
negotiateMobileRuntimeCapabilities({
sendRequest: () => request,
current: () => {
checkedCurrent()
return current
},
onReady,
onFailure
})
await vi.waitFor(() => expect(onReady).not.toHaveBeenCalled())
expect(onFailure).not.toHaveBeenCalled()
})
it('leaves a replaced session alone on a successful response', async () => {
const onReady = vi.fn()
const onFailure = vi.fn()
negotiateMobileRuntimeCapabilities({
sendRequest: () =>
Promise.resolve({ id: 'capability-1', ok: true, result: {} } as RpcResponse),
current: () => false,
onReady,
onFailure
})
await vi.waitFor(() => expect(onReady).not.toHaveBeenCalled())
expect(onFailure).not.toHaveBeenCalled()
})
current = false
if (outcome === 'successful') {
resolveRequest({ id: 'capability-1', ok: true, result: {} })
} else {
const error = new Error('Connection interrupted')
rejectRequest(outcome === 'unanswered' ? markRpcDeliveryUnknown(error) : error)
}
await settled
expect(onReady).not.toHaveBeenCalled()
expect(onFailure).not.toHaveBeenCalled()
}
)
})
@@ -1,7 +1,6 @@
import { describe, expect, it } from 'vitest'
import { MobileRelayRpcStreams } from './mobile-relay-rpc-streams'
import { RpcClientStreamRegistry } from './rpc-client-stream-registry'
import { READY_STREAM_RELEASE_METHODS } from './rpc-client-server-subscription'
import type { RpcSuccess } from './types'
type SentFrame = { id: string; method: string; params?: unknown }
@@ -71,45 +70,13 @@ function relayTransport() {
}
}
function releases(sent: SentFrame[], method: string): unknown[] {
return sent.filter((frame) => frame.method === method).map((frame) => frame.params)
}
describe.each([
['direct', directTransport],
['relay', relayTransport]
])('%s transport releases the accounts stream', (_name, transport) => {
it('sends accounts.unsubscribe with the ready id on dispose', async () => {
const wire = transport()
const dispose = wire.subscribe('accounts.subscribe')
await Promise.resolve()
wire.reply(readyReply(wire.sent[0]!.id, 'accounts-conn-1'))
dispose()
expect(releases(wire.sent, 'accounts.unsubscribe')).toEqual([
{ subscriptionId: 'accounts-conn-1' }
])
})
it('holds a dispose that beat the ready and releases once the ready lands', async () => {
const wire = transport()
const dispose = wire.subscribe('accounts.subscribe')
await Promise.resolve()
dispose()
expect(releases(wire.sent, 'accounts.unsubscribe')).toEqual([])
wire.reply(readyReply(wire.sent[0]!.id, 'accounts-conn-1'))
expect(releases(wire.sent, 'accounts.unsubscribe')).toEqual([
{ subscriptionId: 'accounts-conn-1' }
])
})
})
// Iterates the mapping itself, so a method added to it is held to the same release on both routes.
describe.each(
[...READY_STREAM_RELEASE_METHODS].flatMap(([method, release]) => [
[
['accounts.subscribe', 'accounts.unsubscribe'],
['browser.screencast', 'browser.screencast.unsubscribe'],
['runtime.clientEvents.subscribe', 'runtime.clientEvents.unsubscribe'],
['notifications.subscribe', 'notifications.unsubscribe']
].flatMap(([method, release]) => [
['direct', method, release, directTransport] as const,
['relay', method, release, relayTransport] as const
])
@@ -125,6 +92,7 @@ describe.each(
const disposeBeforeReady = beforeReady.subscribe(method)
await Promise.resolve()
disposeBeforeReady()
expect(beforeReady.sent).toHaveLength(1)
beforeReady.reply(readyReply(beforeReady.sent[0]!.id, 'host-id-2'))
expect(afterReady.sent.slice(1)).toEqual([
@@ -1,202 +0,0 @@
import { readFileSync } from 'node:fs'
import { fileURLToPath } from 'node:url'
import { extname, join, relative, resolve } from 'node:path'
import ts from 'typescript'
import { describe, expect, it } from 'vitest'
import { censusSourceFiles } from '../test-support/census-source-files'
import { readScenarios } from '../test-support/rpc-recording/scenario-input'
import { READY_STREAM_RELEASE_METHODS } from './rpc-client-server-subscription'
import { RPC_SUBSCRIPTION_SITES, type RpcSubscriptionSite } from './rpc-subscription-inventory'
/**
* Makes the subscription inventory bind.
*
* Five failures, all of which mean "edit the list":
* - a file opens a stream and is not listed,
* - a listed file no longer opens one (stale entry — how allow-lists rot),
* - a listed file opens a different method than its entry claims,
* - the `ready-id` entries and the transport's release table name different methods,
* - a `recorded` entry names a family the scenario manifest does not have.
*
* The last one is what separates this from prose. A comment saying a stream is covered stays true
* forever; an entry that has to resolve against `pilot-scenarios.json` stops being true the moment
* the family is renamed or deleted.
*
* What this does NOT catch, all accepted:
* - A method that is not a string literal at the call site. `client.subscribe(method, …)` with a
* variable is invisible here, the same gap the raw-port ratchet accepts for a computed
* `sendRequest`. Every product site today spells its method.
* - Whether a `recorded` family's golden actually drives that file. `sites` in the manifest says
* so and no check ties the two together; that is one indirection further than this list is for.
* - Whether a wall is still real. A wall is prose by construction — it says why a recording
* cannot exist, and the only proof of the opposite is the recording.
* - `transport/` and `test-support/`, which implement and script the port rather than consuming
* it. A registry that forwards `subscribe` is not a call site.
*/
const mobileRoot = fileURLToPath(new URL('../..', import.meta.url))
const repoRoot = resolve(mobileRoot, '..')
const scannedRoots = ['app', 'src'].map((directory) => join(mobileRoot, directory))
const sourceExtensions = new Set(['.js', '.jsx', '.ts', '.tsx'])
/** The port's own implementation and the oracle that scripts it. Neither consumes a stream. */
const EXCLUDED_DIRECTORIES = ['src/transport/', 'src/test-support/']
function parse(path: string, source: string): ts.SourceFile {
const extension = extname(path)
return ts.createSourceFile(
path,
source,
ts.ScriptTarget.Latest,
true,
extension === '.tsx' || extension === '.jsx' ? ts.ScriptKind.TSX : ts.ScriptKind.TS
)
}
/** Every method this file opens a stream on, in source order. */
export function subscribedMethods(path: string, source: string): string[] {
const methods: string[] = []
const visit = (node: ts.Node): void => {
if (
ts.isCallExpression(node) &&
ts.isPropertyAccessExpression(node.expression) &&
node.expression.name.text === 'subscribe'
) {
const [method] = node.arguments
if (method && ts.isStringLiteral(method)) {
methods.push(method.text)
}
}
ts.forEachChild(node, visit)
}
visit(parse(path, source))
return methods
}
const scanned = scannedRoots
.flatMap(censusSourceFiles)
.filter((path) => sourceExtensions.has(extname(path)))
.filter((path) => !/\.test\.tsx?$/.test(path))
.map((path) => relative(mobileRoot, path).split(/[/\\]/).join('/'))
.filter((file) => !EXCLUDED_DIRECTORIES.some((directory) => file.startsWith(directory)))
const observed = new Map(
scanned
.map(
(file) =>
[
file,
subscribedMethods(join(mobileRoot, file), readFileSync(join(mobileRoot, file), 'utf8'))
] as const
)
.filter(([, methods]) => methods.length > 0)
)
const families = new Set(
readScenarios(join(repoRoot, 'mobile', 'rpc-foundation', 'pilot-scenarios.json')).scenarios.map(
(scenario) => scenario.family
)
)
function listedMethods(file: string): string[] {
return RPC_SUBSCRIPTION_SITES.filter((site) => site.file === file).map((site) => site.method)
}
describe('RPC subscription boundary', () => {
const probe = join(mobileRoot, 'src', 'session', 'probe.ts')
it('reads the method off each subscribe and ignores everything else', () => {
expect(
subscribedMethods(probe, "client.subscribe('terminal.subscribe', params, onData)")
).toEqual(['terminal.subscribe'])
expect(
subscribedMethods(probe, "entry.client.subscribe('accounts.subscribe', null, cb)")
).toEqual(['accounts.subscribe'])
expect(
subscribedMethods(probe, "a.subscribe('one', p, cb); b.subscribe('two', p, cb)")
).toEqual(['one', 'two'])
// A store listener, not a host stream: the first argument is not a method.
expect(subscribedMethods(probe, 'connectionLogStore.subscribe(selectedId, listener)')).toEqual(
[]
)
expect(subscribedMethods(probe, 'args.subscribe(handle)')).toEqual([])
expect(subscribedMethods(probe, "await client.sendRequest('worktree.ps', {})")).toEqual([])
expect(
subscribedMethods(probe, '// calls client.subscribe("x", p, cb) under the hood')
).toEqual([])
})
it('scans a plausible number of files', () => {
// A broken root or filter would make every check below vacuously pass.
expect(scanned.length).toBeGreaterThan(400)
expect(observed.size).toBeGreaterThan(5)
})
it('lists each file and method once', () => {
const seen = RPC_SUBSCRIPTION_SITES.map((site) => `${site.file}\0${site.method}`)
expect(seen.filter((key, index) => seen.indexOf(key) !== index)).toEqual([])
})
it('has no unlisted file opening a stream', () => {
const unlisted = [...observed.keys()].filter((file) => listedMethods(file).length === 0)
expect(
unlisted,
'A new subscription must be classified in rpc-subscription-inventory.ts: recorded, an unwritten scenario, or walled with the wall named.'
).toEqual([])
})
it('has no stale inventory entry', () => {
const stale = RPC_SUBSCRIPTION_SITES.filter((site) => !observed.has(site.file)).map(
(site) => site.file
)
expect(
stale,
'File no longer opens a stream — delete its line from rpc-subscription-inventory.ts.'
).toEqual([])
})
it('classifies every method each listed file opens', () => {
const mismatched = [...observed]
.filter(([file]) => listedMethods(file).length > 0)
.flatMap(([file, methods]) => {
const listed = [...listedMethods(file)].sort()
const found = [...new Set(methods)].sort()
return JSON.stringify(listed) === JSON.stringify(found)
? []
: [`${file}: listed ${listed.join(', ')}, found ${found.join(', ')}`]
})
expect(mismatched, 'The method an entry names is what the file opens.').toEqual([])
})
it('resolves every recorded family against the scenario manifest', () => {
const missing = RPC_SUBSCRIPTION_SITES.flatMap((site: RpcSubscriptionSite) =>
site.coverage.kind === 'recorded' && !families.has(site.coverage.family)
? [`${site.file}: no family ${site.coverage.family}`]
: []
)
expect(
missing,
'A recorded entry must name a family in pilot-scenarios.json, or the claim is prose.'
).toEqual([])
})
it('releases by ready id exactly the streams the transport release table names', () => {
const declared = [
...new Set(
RPC_SUBSCRIPTION_SITES.filter((site) => site.release === 'ready-id').map(
(site) => site.method
)
)
].sort()
expect(
declared,
'A stream whose host id arrives in `ready` is released only through READY_STREAM_RELEASE_METHODS.'
).toEqual([...READY_STREAM_RELEASE_METHODS.keys()].sort())
})
it('names the wall on every walled entry', () => {
const unnamed = RPC_SUBSCRIPTION_SITES.flatMap((site) =>
site.coverage.kind === 'walled' && site.coverage.wall.trim().length < 40 ? [site.file] : []
)
expect(unnamed, 'A wall has to say what it is; "not supported" is not a wall.').toEqual([])
})
})
@@ -1,132 +0,0 @@
/**
* Every product call site that opens a host stream, and what the recording oracle can see of it.
*
* The raw-request-port inventory next door counts down to zero; this one does not. A subscribe is
* not something a typed operation replaces — `RpcOperation` fixes a method, an acceptance policy
* and a reader for one reply, and a stream has many. What this list is for is the other half of
* the same question: which of these streams is a golden actually holding, and for the ones it is
* not, what exactly stops it. Left as prose in a README that answer went stale twice, because
* nothing failed when a new `client.subscribe` appeared.
*
* So each site is classified, and `rpc-subscription-boundary.test.ts` makes the classification
* bind: a new site with no entry fails, an entry whose file no longer subscribes fails, an entry
* naming the wrong method fails, a `ready-id` release the transport's table does not name fails,
* and a `recorded` entry whose family is not in the scenario manifest fails. A wall must name
* itself; "not recorded yet" and "cannot be recorded" are different claims and only one of them is
* a backlog item.
*/
export type RpcSubscriptionCoverage =
/** A golden holds this stream. `family` is a family in `pilot-scenarios.json`. */
| { readonly kind: 'recorded'; readonly family: string }
/** The recorder could mount this site; nobody has written the scenario. A backlog item. */
| { readonly kind: 'unwritten-scenario' }
/** Something structural stops a recording. Not a backlog item until the wall moves. */
| { readonly kind: 'walled'; readonly wall: string }
/**
* How the phone ends this stream on the host. `ready-id` must match `READY_STREAM_RELEASE_METHODS`:
* a stream whose host id arrives only in `ready` and is missing there is never released.
*/
export type RpcSubscriptionRelease =
/** The transport releases it by the id from the current `ready`. */
| 'ready-id'
/** The transport builds the unsubscribe from the subscribe params and request id. */
| 'params'
export type RpcSubscriptionSite = {
readonly file: string
readonly method: string
readonly release: RpcSubscriptionRelease
readonly coverage: RpcSubscriptionCoverage
}
export const RPC_SUBSCRIPTION_SITES: readonly RpcSubscriptionSite[] = [
// The account snapshot, opened twice. The home screen wires one per connected host; the host
// screen opens its own. Both decode the same snapshot, and neither is the wall — the loader
// reaches `decodeAccountsSnapshot` and it throws its own domain error on a bad one.
{
file: 'src/accounts/MobileAccountsScreen.tsx',
method: 'accounts.subscribe',
release: 'ready-id',
coverage: {
kind: 'walled',
wall: 'The screen renders `react-native.ScrollView`, not a substituted member, so the mount trap refuses on the first render: `Unsubstituted native member: react-native.ScrollView`.'
}
},
{
file: 'src/home/use-mobile-home-host-connections.ts',
method: 'accounts.subscribe',
release: 'ready-id',
coverage: {
kind: 'walled',
wall: 'Wired on a per-host client from `useAllHostClients`, and the runner hands an adapter one client rather than the multi-host context that hook reads.'
}
},
// The browser tab's screencast. Frames are pixels, not JSON.
{
file: 'src/browser/use-mobile-browser-stream.ts',
method: 'browser.screencast',
release: 'ready-id',
coverage: {
kind: 'walled',
wall: 'Writes to a webview terminal/browser ref this runner has no substitute for, and a substitute that shaped what the stream delivered would be inventing the device.'
}
},
{
file: 'src/notifications/mobile-notifications.ts',
method: 'notifications.subscribe',
release: 'ready-id',
coverage: { kind: 'recorded', family: 'notifications.desktop-stream' }
},
{
file: 'src/session/mobile-terminal-stream-subscribe.ts',
method: 'terminal.subscribe',
release: 'params',
coverage: {
kind: 'walled',
wall: 'Writes to a webview terminal ref this runner has no substitute for: the stream consumer calls `ref.init` and `dataRef.write`, so what a frame does is a device effect rather than an observation.'
}
},
{
file: 'src/session/use-live-worktree-name.ts',
method: 'runtime.clientEvents.subscribe',
release: 'ready-id',
coverage: { kind: 'recorded', family: 'live-worktree-name' }
},
{
file: 'src/session/use-mobile-native-chat-session.ts',
method: 'nativeChat.subscribe',
release: 'params',
coverage: { kind: 'recorded', family: 'session.native-chat-page' }
},
// The structured agent session's event stream. Mountable: its listener guards the payload, and
// the hold that precedes it is a plain request. What is missing is the scenario.
{
file: 'src/session/mobile-structured-transcript-subscription.ts',
method: 'agentSession.subscribe',
release: 'params',
coverage: { kind: 'unwritten-scenario' }
},
// The host's status stream, one per client, which the chat reads the host's "Stopping…" from.
// Mountable: the feed guards each frame and touches no device surface. The scenario is missing.
{
file: 'src/session/mobile-structured-session-status-feed.ts',
method: 'agentSession.subscribeStatus',
release: 'params',
coverage: { kind: 'unwritten-scenario' }
},
// The session tab snapshot. Mountable behind the reconciliation controller the hook already
// takes; no device surface is involved.
{
file: 'src/session/use-mobile-session-tabs-reconciliation.ts',
method: 'session.tabs.subscribe',
release: 'params',
coverage: { kind: 'unwritten-scenario' }
},
{
file: 'src/worktree/host-worktree-refresh.ts',
method: 'runtime.clientEvents.subscribe',
release: 'ready-id',
coverage: { kind: 'recorded', family: 'host-worktree-refresh' }
}
]