From 56a3aebe022ec97779aa6a20687b8a7c7dcf2bfe Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 29 Aug 2026 20:44:13 +0000 Subject: [PATCH] fix(rsc-runtime): never let non-generation attempts supersede an activation, and bound activation phases The activation guard judged an in-flight activation superseded when any live higher-sequence attempt existed at commit-check time, or when a failed attempt had bumped the superseding sequence between the guard wait and the check. Neither produces a generation, so the newest successful compile was discarded with nothing to replace it and no retrigger - the permanent staleness wedge in #38. Supersession now tie-breaks only on the monotonic captured-cohort revision and the prepared-runtime authority digest. Generation materialization, the MCP activation reconcile, the final guard wait, and the prepared-runtime reconcile are now bounded by a time-scale-aware budget; a wedged phase fails the attempt loudly with the phase in its diagnostic (the page recovers via its failed-event bootstrap path) instead of silently hanging the provider tail and close(). A phase that settles after its budget releases its store or registry reservation so a stray late success cannot wedge later activations. --- .../src/dev/rsbuild-runtime-session.ts | 106 ++++++-- .../tests/dev-provider.integration.test.ts | 228 +++++++++++++++++- 2 files changed, 307 insertions(+), 27 deletions(-) diff --git a/examples/rsc-agent-runtime/src/dev/rsbuild-runtime-session.ts b/examples/rsc-agent-runtime/src/dev/rsbuild-runtime-session.ts index d447d9744..6d211870e 100644 --- a/examples/rsc-agent-runtime/src/dev/rsbuild-runtime-session.ts +++ b/examples/rsc-agent-runtime/src/dev/rsbuild-runtime-session.ts @@ -115,6 +115,23 @@ const claudePostToolUseFixture: DevRuntimeFixture = Object.freeze({ const claudeFixtures: readonly DevRuntimeFixture[] = Object.freeze([claudePostToolUseFixture]); const fixturesForHook = (host: 'claude' | 'codex'): readonly DevRuntimeFixture[] => host === 'claude' ? claudeFixtures : noFixtures; +/** + * Activation budgets mirror the repository's test time-scale rule: CI runners + * share two cores between Chrome, dev servers, and compiles, so fixed budgets + * tuned on many-core machines starve there. Scaling costs nothing on green + * runs - the activation resolves long before the deadline - while a wedged + * materialization or MCP reconcile becomes a loud `runtime.generation.failed` + * (with the phase in its diagnostic) instead of a silent permanent hang that + * also blocks `close()` behind the provider tail (#38). + */ +const localTimeScale = Number(process.env['AGENT_BUNDLE_TEST_TIME_SCALE'] ?? ''); +const runtimeTimeScale = process.env['CI'] !== undefined + ? 4 + : Number.isSafeInteger(localTimeScale) && localTimeScale >= 1 ? localTimeScale : 1; +const defaultActivationPhaseBudgetMs = 30_000 * runtimeTimeScale; + +type ActivationPhase = 'activation-guard' | 'generation-store' | 'mcp-registry' | 'prepared-runtime-reconcile'; + const withinDeadline = (promise: Promise, timeoutMs: number, message: string): Promise => new Promise((resolve, reject) => { const timeout = setTimeout(() => reject(new Error(message)), timeoutMs); @@ -585,6 +602,12 @@ export interface RsbuildRuntimeSessionStartTesting { readonly phase: 'store' | 'registry'; readonly session: RsbuildRuntimeSession; }>) => Promise | void; + /** + * Test-only barrier between the final activation-guard wait and the commit + * check, so supersession races (a newer attempt registering or failing + * while an activation is in flight) can be injected deterministically. + */ + readonly beforeActivationCommit?: () => Promise | void; readonly beforeAssetRead?: (input: Readonly<{ readonly request: DevRuntimeAssetRequest; readonly runtimeGenerationId: string; @@ -610,6 +633,12 @@ export interface RsbuildRuntimeSessionStartTesting { * `defaultMaximumRunHistory` runs. */ readonly maximumRunHistory?: number; + /** + * Test-only activation phase budget override so bounded-wedge suites do not + * need to wait out the scaled production budget; the public provider always + * uses `defaultActivationPhaseBudgetMs`. + */ + readonly activationPhaseBudgetMs?: number; } /** @@ -654,10 +683,10 @@ export class RsbuildRuntimeSession implements DevRuntimeSession { #failureTail: Promise = Promise.resolve(); #hmrReady = false; #latestAttemptSequence = 0; - #latestSupersedingAttemptSequence = 0; #latestPreparedRuntime: DevRuntimePreparedProject; #latestRscCohortRevision = 0; #invocationReservations = 0; + readonly #activationPhaseBudgetMs: number; #providerTail: Promise = Promise.resolve(); #server: StartDevServerResult['server'] | undefined; #status: DevRuntimeStatus; @@ -678,6 +707,7 @@ export class RsbuildRuntimeSession implements DevRuntimeSession { this.#latestPreparedRuntime = input.preparedRuntime; this.#testing = input.testing; this.#maximumRunHistory = input.testing.maximumRunHistory ?? defaultMaximumRunHistory; + this.#activationPhaseBudgetMs = input.testing.activationPhaseBudgetMs ?? defaultActivationPhaseBudgetMs; this.#ownedRunsRoot = input.ownedRunsRoot; this.#runRoot = input.ownedRunsRoot.root; this.#stateFile = join(resolve(input.context.storageRoot), 'state', `${stateStoreId}.jsonl`); @@ -2202,7 +2232,6 @@ export class RsbuildRuntimeSession implements DevRuntimeSession { barrier.settle(); return undefined; } - this.#latestSupersedingAttemptSequence = Math.max(this.#latestSupersedingAttemptSequence, barrier.sequence); const cohortRevision = ++this.#latestRscCohortRevision; const preparedRuntime = this.#latestPreparedRuntime; barrier.settle(); @@ -2259,7 +2288,6 @@ export class RsbuildRuntimeSession implements DevRuntimeSession { ): Promise { if (this.#failedAttempts.has(attemptId)) return; this.#failedAttempts.add(attemptId); - this.#latestSupersedingAttemptSequence = Math.max(this.#latestSupersedingAttemptSequence, this.#sequenceFor(attemptId)); const barrier = this.#attempts.get(attemptId); barrier?.settle(); const candidate = barrier?.candidate ?? this.#candidatesByAttempt.get(attemptId); @@ -2278,21 +2306,21 @@ export class RsbuildRuntimeSession implements DevRuntimeSession { #activationGuard(snapshot: RscRuntimeCapturedGenerationSnapshot): RuntimeGenerationActivationGuard { const preparedAuthorityDigest = preparedRuntimeAuthorityDigest(snapshot.preparedRuntime); - let waitedSequence = -1; return Object.freeze({ + // Supersession tie-breaks on monotonic ordinals only: a newer captured + // cohort (`#latestRscCohortRevision`) or a prepared-runtime authority + // change may discard this activation. Attempts that are merely live, or + // that failed or settled as no-ops, never produce a generation, so + // judging them as superseding would drop the newest successful compile + // with nothing to replace it - the permanent-staleness wedge in #38. check: () => !this.#closed && - waitedSequence === this.#latestSupersedingAttemptSequence && - ![...this.#attempts.values()].some((attempt) => attempt.sequence > this.#sequenceFor(snapshot.attemptId)) && snapshot.rscCohortRevision === this.#latestRscCohortRevision && preparedAuthorityDigest === preparedRuntimeAuthorityDigest(this.#latestPreparedRuntime), wait: async () => { while (!this.#closed) { const sequence = this.#sequenceFor(snapshot.attemptId); const pending = [...this.#attempts.values()].filter((attempt) => attempt.sequence > sequence); - if (pending.length === 0) { - waitedSequence = this.#latestSupersedingAttemptSequence; - return; - } + if (pending.length === 0) return; await Promise.all(pending.map((attempt) => attempt.settled)); } throw new Error('RSC runtime session is closed.'); @@ -2300,27 +2328,57 @@ export class RsbuildRuntimeSession implements DevRuntimeSession { }); } + /** + * Bounds one activation step with the scaled budget. A step that outlives + * its budget fails the attempt loudly (the page recovers through its + * `runtime.generation.failed` bootstrap path) instead of silently wedging + * the provider tail; if the abandoned step settles later, its resources are + * released so a stray success cannot leak store or registry reservations. + */ + async #boundedActivationPhase( + phase: ActivationPhase, + work: Promise, + abandon?: (value: T) => Promise, + ): Promise { + const budget = this.#activationPhaseBudgetMs; + try { + return await withinDeadline(work, budget, `RSC runtime ${phase} activation step exceeded ${String(budget)}ms.`); + } catch (error) { + if (abandon !== undefined) void work.then(abandon, () => undefined).catch(() => undefined); + throw error; + } + } + async #activate(snapshot: RscRuntimeCapturedGenerationSnapshot): Promise<'activated' | 'failed'> { const guard = this.#activationGuard(snapshot); let preparedGeneration: RuntimeGenerationPreparedActivation | undefined; let preparedRegistry: RuntimeMcpPreparedActivationReconcile | undefined; try { - preparedGeneration = await materializeRuntimeGeneration({ - guard, - snapshot, - stateStoreId, - store: this.#generationStore, - }); + preparedGeneration = await this.#boundedActivationPhase( + 'generation-store', + materializeRuntimeGeneration({ + guard, + snapshot, + stateStoreId, + store: this.#generationStore, + }), + (prepared) => this.#generationStore.abort(prepared), + ); await this.#testing.afterActivationPrepare?.(Object.freeze({ phase: 'store', session: this })); const metadata = preparedGeneration.generation.manifest.metadata; - preparedRegistry = await this.#mcpRegistry.prepareActivationReconcile({ - definitionDigest: metadata.definitionDigest, - runtimeGenerationId: preparedGeneration.generation.id, - servers: metadata.servers, - transportDigest: metadata.transportDigest, - }); + preparedRegistry = await this.#boundedActivationPhase( + 'mcp-registry', + this.#mcpRegistry.prepareActivationReconcile({ + definitionDigest: metadata.definitionDigest, + runtimeGenerationId: preparedGeneration.generation.id, + servers: metadata.servers, + transportDigest: metadata.transportDigest, + }), + (prepared) => this.#mcpRegistry.abortActivationReconcile(prepared), + ); await this.#testing.afterActivationPrepare?.(Object.freeze({ phase: 'registry', session: this })); - await guard.wait(preparedGeneration.generation.manifest); + await this.#boundedActivationPhase('activation-guard', guard.wait(preparedGeneration.generation.manifest)); + await this.#testing.beforeActivationCommit?.(); if (!guard.check(preparedGeneration.generation.manifest) || !this.#generationStore.canCommit(preparedGeneration)) { throw new Error('RSC runtime generation activation was superseded.'); } @@ -2380,7 +2438,7 @@ export class RsbuildRuntimeSession implements DevRuntimeSession { }); this.#setStatus('compiling'); try { - await this.#mcpRegistry.reconcile(input); + await this.#boundedActivationPhase('prepared-runtime-reconcile', this.#mcpRegistry.reconcile(input)); this.#updateSurfaces({ definition }, prepared); this.#updateSurfaceAssetApps(prepared); this.#setStatus('active'); diff --git a/examples/rsc-agent-runtime/tests/dev-provider.integration.test.ts b/examples/rsc-agent-runtime/tests/dev-provider.integration.test.ts index 1cb041f52..39c790ca5 100644 --- a/examples/rsc-agent-runtime/tests/dev-provider.integration.test.ts +++ b/examples/rsc-agent-runtime/tests/dev-provider.integration.test.ts @@ -3,7 +3,7 @@ import { tmpdir } from 'node:os'; import { basename, join } from 'node:path'; import { expect, test } from '@rstest/core'; -import type { createRsbuild, StartDevServerResult } from '@rsbuild/core'; +import { createRsbuild, type StartDevServerResult } from '@rsbuild/core'; import { ArtifactService, @@ -19,17 +19,20 @@ import { import { createDevRuntimeProvider } from '../src/dev/provider.js'; import { ResourceLedger, RsbuildRuntimeSession } from '../src/dev/rsbuild-runtime-session.js'; import { copyExample, type CopiedExample } from './support/copy-example.ts'; +import { timeScale } from './support/time-scale.ts'; const exampleRoot = process.cwd(); -const waitFor = async (predicate: () => boolean): Promise => { - const deadline = Date.now() + 15_000; +const waitForWithin = async (predicate: () => boolean, budgetMs: number): Promise => { + const deadline = Date.now() + budgetMs; while (!predicate()) { if (Date.now() >= deadline) throw new Error('Timed out waiting for the RSC runtime provider.'); await new Promise((resolve) => { setTimeout(resolve, 25); }); } }; +const waitFor = async (predicate: () => boolean): Promise => waitForWithin(predicate, 15_000); + const deferred = () => { let reject!: (reason?: unknown) => void; let resolve!: (value: T | PromiseLike) => void; @@ -70,6 +73,51 @@ const compileObserver = (onCompile: NonNullable { + let before: (() => void) | undefined; + let after: ((input: unknown) => Promise) | undefined; + const create = (async (input: Parameters[0]) => { + const plugins = (input?.config as Readonly<{ readonly plugins?: readonly unknown[] }> | undefined)?.plugins ?? []; + const plugin = plugins.find((candidate): candidate is Readonly<{ + readonly name: string; + setup(api: unknown): void; + }> => typeof candidate === 'object' && candidate !== null && + (candidate as { readonly name?: unknown }).name === 'agent-bundle:rsc-runtime-compile-observer'); + if (plugin === undefined) throw new Error('RSC compile observer plugin is unavailable.'); + plugin.setup({ + onAfterDevCompile: (callback: unknown) => { after = callback as (input: unknown) => Promise; }, + onBeforeDevCompile: (callback: unknown) => { before = callback as () => void; }, + }); + return createRsbuild(input); + }) as typeof createRsbuild; + return Object.freeze({ + beginAttempt: (): void => { + if (before === undefined) throw new Error('RSC compile observer hooks are unavailable.'); + before(); + }, + async completeAttempt(input: Readonly<{ + readonly children?: readonly Readonly<{ readonly hash: string; readonly name: string }>[]; + readonly hasErrors?: boolean; + }> = {}): Promise { + if (after === undefined) throw new Error('RSC compile observer hooks are unavailable.'); + await after({ + stats: { + hasErrors: () => input.hasErrors ?? false, + toJson: () => ({ children: input.children ?? [] }), + }, + }); + }, + create, + }); +}; + const snapshotFor = (attemptId: string, sourceRevision: string): RscRuntimeCompileSnapshot => Object.freeze({ attemptId, candidateId: attemptId, @@ -1297,6 +1345,180 @@ test('commits a compiled generation across an equivalent prepared-runtime revisi } }); +test('commits an activation while a later attempt is still live at the commit check', { timeout: 90_000 }, async () => { + const copied = await copyProviderExample(); + try { + const prepared = await new ProjectService({ includeDevRuntime: true, mode: 'development', root: copied.projectRoot }).prepare('dev'); + const observer = interceptCompileObserver(); + const commitReached = deferred(); + const allowCommit = deferred(); + let armCommitBarrier = false; + const events: Array<{ readonly runtimeGenerationId?: string; readonly type: string }> = []; + const session = await RsbuildRuntimeSession.start({ + ...startContext({ + projectRoot: copied.projectRoot, + preparedRuntime: prepared.devRuntime!, + providerSessionId: 'provider-live-attempt-commit', + signal: new AbortController().signal, + storageRoot: join(copied.projectRoot, '.agent-bundle', 'runtime-live-attempt-commit'), + }), + emit: (event) => { events.push(event); }, + }, { + beforeActivationCommit: async () => { + if (!armCommitBarrier) return; + armCommitBarrier = false; + commitReached.resolve(); + await allowCommit.promise; + }, + createRsbuild: observer.create, + }); + try { + await waitFor(() => session.status().state === 'active'); + const firstGeneration = session.status().activeVector!.runtimeGenerationId; + armCommitBarrier = true; + observer.beginAttempt(); + await observer.completeAttempt({ children: [{ hash: 'live-race-rsc', name: 'rsc' }, { hash: 'live-race-widget', name: 'widget' }] }); + await commitReached.promise; + // The #38 wedge: an undecided later attempt registers while the newest + // compile sits between its final guard wait and the commit check. It + // must not supersede the activation - it may still settle as a no-op or + // a failure, which would leave nothing to activate and no retrigger. + observer.beginAttempt(); + allowCommit.resolve(); + await waitFor(() => session.status().activeVector?.runtimeGenerationId !== firstGeneration); + const committed = session.status().activeVector?.runtimeGenerationId; + expect(committed).toEqual(expect.any(String)); + expect(session.status()).toMatchObject({ diagnostics: [], state: 'active' }); + expect(events.filter((event) => event.type === 'runtime.generation.activated' && event.runtimeGenerationId === committed)).toHaveLength(1); + await observer.completeAttempt({ hasErrors: true }); + expect(session.status().activeVector?.runtimeGenerationId).toBe(committed); + } finally { + await session.close(); + } + } finally { + await rm(copied.workspaceRoot, { force: true, recursive: true }); + } +}); + +test('commits an activation after a later broken attempt fails inside the commit window', { timeout: 90_000 }, async () => { + const copied = await copyProviderExample(); + try { + const prepared = await new ProjectService({ includeDevRuntime: true, mode: 'development', root: copied.projectRoot }).prepare('dev'); + const observer = interceptCompileObserver(); + const commitReached = deferred(); + const allowCommit = deferred(); + let armCommitBarrier = false; + const events: Array<{ readonly runtimeGenerationId?: string; readonly type: string }> = []; + const session = await RsbuildRuntimeSession.start({ + ...startContext({ + projectRoot: copied.projectRoot, + preparedRuntime: prepared.devRuntime!, + providerSessionId: 'provider-failed-attempt-commit', + signal: new AbortController().signal, + storageRoot: join(copied.projectRoot, '.agent-bundle', 'runtime-failed-attempt-commit'), + }), + emit: (event) => { events.push(event); }, + }, { + beforeActivationCommit: async () => { + if (!armCommitBarrier) return; + armCommitBarrier = false; + commitReached.resolve(); + await allowCommit.promise; + }, + createRsbuild: observer.create, + }); + try { + await waitFor(() => session.status().state === 'active'); + const firstGeneration = session.status().activeVector!.runtimeGenerationId; + armCommitBarrier = true; + observer.beginAttempt(); + await observer.completeAttempt({ children: [{ hash: 'failed-race-rsc', name: 'rsc' }, { hash: 'failed-race-widget', name: 'widget' }] }); + await commitReached.promise; + // A later broken compile fails and settles entirely inside the commit + // window. A failed attempt produces no generation, so it must not + // supersede the newest successful compile (#38) - discarding it here + // left the runtime permanently on the stale first generation. + observer.beginAttempt(); + await observer.completeAttempt({ hasErrors: true }); + allowCommit.resolve(); + await waitFor(() => session.status().activeVector?.runtimeGenerationId !== firstGeneration); + const committed = session.status().activeVector?.runtimeGenerationId; + expect(committed).toEqual(expect.any(String)); + expect(session.status()).toMatchObject({ state: 'active' }); + expect(events.filter((event) => event.type === 'runtime.generation.activated' && event.runtimeGenerationId === committed)).toHaveLength(1); + } finally { + await session.close(); + } + } finally { + await rm(copied.workspaceRoot, { force: true, recursive: true }); + } +}); + +test('fails a wedged activation reconcile within the budget and releases its late reservation', { timeout: 120_000 }, async () => { + const copied = await copyProviderExample(); + try { + const prepared = await new ProjectService({ includeDevRuntime: true, mode: 'development', root: copied.projectRoot }).prepare('dev'); + const wedgeReached = deferred(); + const releaseWedge = deferred(); + let armWedge = false; + const activationBudgetMs = 4_000 * timeScale; + const events: Array<{ readonly runtimeGenerationId?: string; readonly type: string }> = []; + const session = await RsbuildRuntimeSession.start({ + ...startContext({ + projectRoot: copied.projectRoot, + preparedRuntime: prepared.devRuntime!, + providerSessionId: 'provider-wedged-reconcile', + signal: new AbortController().signal, + storageRoot: join(copied.projectRoot, '.agent-bundle', 'runtime-wedged-reconcile'), + }), + emit: (event) => { events.push(event); }, + }, { + activationPhaseBudgetMs: activationBudgetMs, + beforeMcpRelist: async () => { + if (!armWedge) return; + wedgeReached.resolve(); + await releaseWedge.promise; + }, + }); + try { + await waitFor(() => session.status().state === 'active'); + const firstGeneration = session.status().activeVector!.runtimeGenerationId; + const mcp = await session.mcpRegistry.open({ serverName: 'timeline', target: 'portable' }); + try { + armWedge = true; + await changeDefinition(copied.projectRoot, 'Read state after a wedged activation reconcile.'); + await wedgeReached.promise; + // A wedged MCP reconcile must become a loud, phase-attributed failure + // within the scaled budget instead of the silent permanent hang from + // #38; the page recovers through its failed-event bootstrap path. + await waitForWithin( + () => session.status().diagnostics.some((diagnostic) => diagnostic.message.includes('mcp-registry activation step exceeded')), + activationBudgetMs + 15_000, + ); + expect(events.some((event) => event.type === 'runtime.generation.failed')).toBe(true); + expect(session.status().activeVector?.runtimeGenerationId).toBe(firstGeneration); + // Releasing the wedge lets the abandoned preparation settle late; its + // registry reservation must be released, or every later activation + // would wedge behind it. + armWedge = false; + releaseWedge.resolve(); + await changeWorkerImplementation(copied.projectRoot, 'post-wedge-activation'); + await waitForWithin( + () => session.status().activeVector?.runtimeGenerationId !== firstGeneration, + activationBudgetMs + 15_000, + ); + expect(session.status().activeVector?.runtimeGenerationId).not.toBe(firstGeneration); + } finally { + await mcp.close(); + } + } finally { + await session.close(); + } + } finally { + await rm(copied.workspaceRoot, { force: true, recursive: true }); + } +}); + test('retains a leased inactive generation through pruning and prunes it after the read releases', async () => { const copied = await copyProviderExample(); try {