You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/08/26 09:33:33 UTC
(maka) branch fix/runtime-host-flapping-root created (now 70f366df7)
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a change to branch fix/runtime-host-flapping-root
in repository https://gitbox.apache.org/repos/asf/maka.git
at 70f366df7 fix(desktop): bound one IPC replacement lifetime
This branch includes the following new commits:
new 22dbea497 fix(runtime-host): flush subscription opens before activation
new 5a85e8a0b fix(runtime-host): quiesce after reconnect gaps
new 70f366df7 fix(desktop): bound one IPC replacement lifetime
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
(maka) 03/03: fix(desktop): bound one IPC replacement lifetime
Posted by wy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch fix/runtime-host-flapping-root
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 70f366df78e7cee8160856833a3bcf7191fa9fd2
Author: M4n5ter <m4...@gmail.com>
AuthorDate: Wed Aug 26 17:31:52 2026 +0800
fix(desktop): bound one IPC replacement lifetime
Co-authored-by: me2seeks <me...@gmail.com>
Co-authored-by: hqhq1025 <15...@qq.com>
Generated-by: Codex
---
.../runtime-host-reconnecting-ipc-main.test.ts | 59 ++++++++++++++++++++-
.../src/main/runtime-host-reconnecting-ipc-main.ts | 60 ++++++++++++----------
2 files changed, 92 insertions(+), 27 deletions(-)
diff --git a/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts
index 75181fca2..6cd4575bd 100644
--- a/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts
+++ b/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts
@@ -30,6 +30,7 @@ import {
} from "../ipc-reconnect-policy.js";
import * as ipcReconnectPolicy from "../ipc-reconnect-policy.js";
import {
+ RuntimeHostHandlerUnavailableError,
RuntimeHostReconnectingIpcMain,
RuntimeHostTargetChangedError,
} from "../runtime-host-reconnecting-ipc-main.js";
@@ -127,7 +128,7 @@ test("reconciles a dispatched control on a replacement without replaying it", as
test("bounds reconciliation when no replacement candidate becomes available", async () => {
const ipc = ipcHarness();
const router = new RuntimeHostReconnectingIpcMain(ipc, {
- reconciliationWaitTimeoutMs: 5,
+ replacementWaitTimeoutMs: 5,
});
const firstTarget = router.createTarget("target-a") as ReconciledControlTarget;
let dispatches = 0;
@@ -338,6 +339,62 @@ test("holds an invocation across a Runtime Host candidate replacement", async ()
assert.equal(ipc.size, 0);
});
+test("bounds an invocation while an active Runtime Host has no handler", async () => {
+ const ipc = ipcHarness();
+ const router = new RuntimeHostReconnectingIpcMain(ipc, {
+ replacementWaitTimeoutMs: 5,
+ });
+ const target = router.createTarget("target-a");
+ target.handle("sessions:send", async () => "sent");
+ router.activate("target-a");
+ target.removeHandler("sessions:send");
+
+ await assert.rejects(
+ () => ipc.invoke("sessions:send", scope("target-a")),
+ RuntimeHostHandlerUnavailableError,
+ );
+ target.handle("sessions:send", async () => "retried");
+ assert.equal(await ipc.invoke("sessions:send", scope("target-a")), "retried");
+ router.close();
+});
+
+test("does not reset one reconnectable read deadline across failed replacements", async (t) => {
+ const ipc = ipcHarness();
+ const router = new RuntimeHostReconnectingIpcMain(ipc, {
+ replacementWaitTimeoutMs: 15,
+ });
+ let monotonicNow = 0;
+ t.mock.method(performance, "now", () => monotonicNow);
+ router.activate("target-a");
+ let attempts = 0;
+ const maximumAttempts = 20;
+ const installFailingTarget = (): void => {
+ const target = router.createTarget("target-a");
+ target.handleReconnectableRead?.("projects:getSnapshot", async () => {
+ attempts += 1;
+ monotonicNow += 5;
+ target.removeHandler("projects:getSnapshot");
+ if (attempts < maximumAttempts) installFailingTarget();
+ throw new RuntimeHostOperationError(
+ "project.catalog.query",
+ "host_draining",
+ "Runtime Host is draining",
+ );
+ });
+ };
+ installFailingTarget();
+
+ try {
+ await assert.rejects(
+ () => ipc.invoke("projects:getSnapshot", scope("target-a")),
+ RuntimeHostHandlerUnavailableError,
+ );
+ assert.equal(attempts, 4);
+ } finally {
+ router.close();
+ }
+});
+
test("does not return a late read from a replaced Runtime Host candidate", async () => {
const ipc = ipcHarness();
const router = new RuntimeHostReconnectingIpcMain(ipc);
diff --git a/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts b/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts
index 3875fb5f4..22a5390b3 100644
--- a/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts
+++ b/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts
@@ -33,16 +33,16 @@ type ReconcileIpcHandler = (
type ReconciliationUnavailableIpcHandler = ReconcileIpcHandler;
-const DEFAULT_RECONCILIATION_WAIT_TIMEOUT_MS = 15_000;
+const DEFAULT_REPLACEMENT_WAIT_TIMEOUT_MS = 15_000;
export interface RuntimeHostReconnectingIpcMainOptions {
- readonly reconciliationWaitTimeoutMs?: number;
+ readonly replacementWaitTimeoutMs?: number;
}
-class ReconciliationWaitExpiredError extends Error {
+class HandlerWaitExpiredError extends Error {
constructor() {
- super("Runtime Host reconciliation replacement wait expired");
- this.name = "ReconciliationWaitExpiredError";
+ super("Runtime Host replacement wait expired");
+ this.name = "HandlerWaitExpiredError";
}
}
@@ -81,6 +81,13 @@ export class RuntimeHostTargetChangedError extends Error {
}
}
+export class RuntimeHostHandlerUnavailableError extends Error {
+ constructor() {
+ super("Runtime Host handler remained unavailable after the reconnection window");
+ this.name = "RuntimeHostHandlerUnavailableError";
+ }
+}
+
/**
* Keeps Electron IPC registration stable across reconnects while fencing each
* target generation. Reconnectable reads may move to a replacement candidate,
@@ -90,7 +97,7 @@ export class RuntimeHostReconnectingIpcMain {
readonly #ipcMain: Pick<IpcMain, "handle" | "removeHandler">;
readonly #slots = new Map<string, HandlerSlot>();
readonly #activeEpochs = new Set<string>();
- readonly #reconciliationWaitTimeoutMs: number;
+ readonly #replacementWaitTimeoutMs: number;
#closed = false;
constructor(
@@ -98,12 +105,12 @@ export class RuntimeHostReconnectingIpcMain {
options: RuntimeHostReconnectingIpcMainOptions = {},
) {
this.#ipcMain = ipcMain;
- const reconciliationWaitTimeoutMs =
- options.reconciliationWaitTimeoutMs ?? DEFAULT_RECONCILIATION_WAIT_TIMEOUT_MS;
- if (!Number.isSafeInteger(reconciliationWaitTimeoutMs) || reconciliationWaitTimeoutMs <= 0) {
- throw new TypeError("Runtime Host reconciliation wait timeout must be positive");
+ const replacementWaitTimeoutMs =
+ options.replacementWaitTimeoutMs ?? DEFAULT_REPLACEMENT_WAIT_TIMEOUT_MS;
+ if (!Number.isSafeInteger(replacementWaitTimeoutMs) || replacementWaitTimeoutMs <= 0) {
+ throw new TypeError("Runtime Host replacement wait timeout must be positive");
}
- this.#reconciliationWaitTimeoutMs = reconciliationWaitTimeoutMs;
+ this.#replacementWaitTimeoutMs = replacementWaitTimeoutMs;
}
createTarget(epoch: string): RuntimeHostTargetIpcMain {
@@ -232,26 +239,28 @@ export class RuntimeHostReconnectingIpcMain {
args: readonly unknown[],
): Promise<unknown> {
const epoch = this.#requireTargetEpoch(args[0]);
- let handler: BoundHandler =
- slot.handlers.get(epoch) ?? await this.#waitForHandler(slot, epoch);
- let reconciliationContext: unknown;
let reconciling = false;
- let reconciliationDeadline: number | undefined;
+ let replacementDeadline: number | undefined;
const waitForReplacement = async (
- previous: BoundHandler,
+ previous?: BoundHandler,
): Promise<BoundHandler | undefined> => {
- if (!reconciling) return this.#waitForHandler(slot, epoch, previous);
- const remainingMs = Math.max(
- 0,
- (reconciliationDeadline ?? Date.now()) - Date.now(),
- );
+ // One invocation gets one monotonic replacement window across every
+ // candidate it visits; flapping must not restart its lifetime.
+ replacementDeadline ??= performance.now() + this.#replacementWaitTimeoutMs;
+ const remainingMs = Math.max(0, replacementDeadline - performance.now());
try {
+ if (remainingMs <= 0) throw new HandlerWaitExpiredError();
return await this.#waitForHandler(slot, epoch, previous, remainingMs);
} catch (error) {
- if (error instanceof ReconciliationWaitExpiredError) return undefined;
- throw error;
+ if (!(error instanceof HandlerWaitExpiredError)) throw error;
+ if (reconciling) return undefined;
+ throw new RuntimeHostHandlerUnavailableError();
}
};
+ const initialHandler = slot.handlers.get(epoch) ?? await waitForReplacement();
+ if (!initialHandler) throw new RuntimeHostHandlerUnavailableError();
+ let handler: BoundHandler = initialHandler;
+ let reconciliationContext: unknown;
const unavailable = (): Promise<unknown> =>
requireReconciliationUnavailableHandler(handler)(
reconciliationContext,
@@ -278,7 +287,6 @@ export class RuntimeHostReconnectingIpcMain {
if (step.kind === "completed") return step.value;
reconciliationContext = step.context;
reconciling = true;
- reconciliationDeadline = Date.now() + this.#reconciliationWaitTimeoutMs;
const replacement = await waitForReplacement(handler);
if (!replacement) return unavailable();
handler = replacement;
@@ -325,7 +333,7 @@ export class RuntimeHostReconnectingIpcMain {
return Promise.resolve(current);
}
if (timeoutMs !== undefined && timeoutMs <= 0) {
- return Promise.reject(new ReconciliationWaitExpiredError());
+ return Promise.reject(new HandlerWaitExpiredError());
}
return new Promise((resolve, reject) => {
let timeout: ReturnType<typeof setTimeout> | undefined;
@@ -344,7 +352,7 @@ export class RuntimeHostReconnectingIpcMain {
if (timeoutMs !== undefined) {
timeout = setTimeout(() => {
if (!slot.waiters.delete(waiter)) return;
- waiter.reject(new ReconciliationWaitExpiredError());
+ waiter.reject(new HandlerWaitExpiredError());
}, timeoutMs);
}
});
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
(maka) 02/03: fix(runtime-host): quiesce after reconnect gaps
Posted by wy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch fix/runtime-host-flapping-root
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 5a85e8a0b7102d751cb629ecf7a45f91126438f3
Author: M4n5ter <m4...@gmail.com>
AuthorDate: Wed Aug 26 17:31:40 2026 +0800
fix(runtime-host): quiesce after reconnect gaps
Generated-by: Codex
---
.../__tests__/runtime-host-desktop-manager.test.ts | 37 ++++++++++++++++++++++
.../src/main/runtime-host-desktop-manager.ts | 2 +-
.../src/__tests__/reconnecting-connection.test.ts | 31 +++++++++++++++++-
.../runtime-host/src/client/reconnect-lifecycle.ts | 14 ++++++--
4 files changed, 79 insertions(+), 5 deletions(-)
diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts
index 4f9cdef92..07b0dfc95 100644
--- a/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts
+++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts
@@ -141,6 +141,43 @@ test('quiesces reconnect and waits for the Host process before update install',
await owner.close();
});
+test('waits through a reconnect gap before quiescing Host retirement', async () => {
+ const first = candidateHarness();
+ const replacement = candidateHarness({ disconnectOnPrepare: true });
+ let starts = 0;
+ let reportReplacementStart!: () => void;
+ let releaseReplacement!: () => void;
+ const replacementStarted = new Promise<void>((resolve) => {
+ reportReplacementStart = resolve;
+ });
+ const replacementReleased = new Promise<void>((resolve) => {
+ releaseReplacement = resolve;
+ });
+ const owner = await startRuntimeHostDesktopManager({} as DesktopRuntimeHostCandidateStartInput, {
+ startCandidate: async () => {
+ starts += 1;
+ if (starts === 1) return ready(first.candidate);
+ reportReplacementStart();
+ await replacementReleased;
+ return ready(replacement.candidate);
+ },
+ waitForHostExit: async () => {},
+ });
+
+ first.disconnect();
+ await replacementStarted;
+ const retirement = owner.retireOwnedLocalHost('interrupt_active_work');
+ await new Promise<void>((resolve) => setImmediate(resolve));
+ assert.equal(replacement.prepareRetirementCalls, 0);
+
+ releaseReplacement();
+ assert.equal((await retirement).kind, 'retired');
+ assert.equal(replacement.prepareRetirementCalls, 1);
+ assert.deepEqual(replacement.retirementModes, ['interrupt_active_work']);
+ assert.equal(starts, 2);
+ await owner.close();
+});
+
test('retires the owned ephemeral Host before Desktop quit', async () => {
const events: string[] = [];
const current = candidateHarness({
diff --git a/apps/desktop/src/main/runtime-host-desktop-manager.ts b/apps/desktop/src/main/runtime-host-desktop-manager.ts
index 61ab1868a..cc347473c 100644
--- a/apps/desktop/src/main/runtime-host-desktop-manager.ts
+++ b/apps/desktop/src/main/runtime-host-desktop-manager.ts
@@ -557,7 +557,7 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
const lifecycle = this.#requireLifecycle(
this.#requireTarget(LOCAL_RUNTIME_HOST_PROFILE.id),
);
- const quiescence = lifecycle.quiesce();
+ const quiescence = await lifecycle.quiesce();
let hostPid = quiescence.current.hostPid;
let launchBarrierPaused = false;
const resume = () => {
diff --git a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts
index a13b13a39..59ac6eca8 100644
--- a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts
+++ b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts
@@ -399,7 +399,7 @@ test('reconnect lifecycle quiescence suppresses replacement until it is resumed'
},
});
- const quiescence = lifecycle.quiesce();
+ const quiescence = await lifecycle.quiesce();
assert.equal(quiescence.current, first.connection);
first.disconnect();
await new Promise<void>((resolve) => setImmediate(resolve));
@@ -411,6 +411,35 @@ test('reconnect lifecycle quiescence suppresses replacement until it is resumed'
await lifecycle.close();
});
+test('reconnect lifecycle quiescence waits through a connection gap before freezing', async () => {
+ const first = connectionHarness('first', () => undefined);
+ const replacement = connectionHarness('replacement', () => undefined);
+ const reconnecting = deferredValue<RuntimeHostConnection>();
+ const connectStarted = deferred();
+ let connectCalls = 0;
+ const lifecycle = await startRuntimeHostReconnectLifecycle({
+ initial: first.connection,
+ connect: async () => {
+ connectCalls += 1;
+ connectStarted.resolve();
+ return reconnecting.promise;
+ },
+ });
+
+ first.disconnect();
+ await connectStarted.promise;
+ const quiescenceTask = lifecycle.quiesce();
+ reconnecting.resolve(replacement.connection);
+ const quiescence = await quiescenceTask;
+ assert.equal(quiescence.current, replacement.connection);
+
+ replacement.disconnect();
+ await yieldToEventLoop();
+ assert.equal(connectCalls, 1);
+ quiescence.resume();
+ await lifecycle.close();
+});
+
test('reconnect delay escalates past maxMs while the Host never stabilizes', async () => {
const first = connectionHarness('first', () => undefined);
const delays: number[] = [];
diff --git a/packages/runtime-host/src/client/reconnect-lifecycle.ts b/packages/runtime-host/src/client/reconnect-lifecycle.ts
index 5a1cec6b8..ef252fac8 100644
--- a/packages/runtime-host/src/client/reconnect-lifecycle.ts
+++ b/packages/runtime-host/src/client/reconnect-lifecycle.ts
@@ -49,7 +49,7 @@ export interface RuntimeHostReconnectLifecycle<T extends RuntimeHostReconnectRes
readonly current: T | undefined;
waitForCurrent(previous?: T, signal?: AbortSignal): Promise<T>;
subscribe(listener: (current: T | undefined) => void): () => void;
- quiesce(): RuntimeHostReconnectQuiescence<T>;
+ quiesce(): Promise<RuntimeHostReconnectQuiescence<T>>;
close(): Promise<void>;
}
@@ -190,13 +190,21 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
return () => this.#listeners.delete(listener);
}
- quiesce(): RuntimeHostReconnectQuiescence<T> {
+ async quiesce(): Promise<RuntimeHostReconnectQuiescence<T>> {
+ while (!this.#current) {
+ if (this.#closed || this.#terminalError) {
+ throw new Error('Runtime Host reconnect lifecycle is closed');
+ }
+ if (this.#quiesced) {
+ throw new Error('Runtime Host reconnect lifecycle is already quiesced');
+ }
+ await this.waitForCurrent();
+ }
if (this.#closed || this.#terminalError) {
throw new Error('Runtime Host reconnect lifecycle is closed');
}
if (this.#quiesced) throw new Error('Runtime Host reconnect lifecycle is already quiesced');
const current = this.#current;
- if (!current) throw new Error('Runtime Host has no current connection to quiesce');
this.#quiesced = true;
let active = true;
return {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
(maka) 01/03: fix(runtime-host): flush subscription opens before activation
Posted by wy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch fix/runtime-host-flapping-root
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 22dbea4976a1905f8cf0a045742744be1499db6b
Author: M4n5ter <m4...@gmail.com>
AuthorDate: Wed Aug 26 17:31:19 2026 +0800
fix(runtime-host): flush subscription opens before activation
Generated-by: Codex
---
.../src/__tests__/connection-session.test.ts | 226 +++++++++++++++++++++
.../runtime-host/src/server/connection-session.ts | 5 +-
2 files changed, 230 insertions(+), 1 deletion(-)
diff --git a/packages/runtime-host/src/__tests__/connection-session.test.ts b/packages/runtime-host/src/__tests__/connection-session.test.ts
index 5760ad57e..4f2c4c399 100644
--- a/packages/runtime-host/src/__tests__/connection-session.test.ts
+++ b/packages/runtime-host/src/__tests__/connection-session.test.ts
@@ -41,7 +41,9 @@ import {
RUNTIME_HOST_COMPATIBILITY_EPOCH,
RUNTIME_HOST_MAX_IN_FLIGHT_DOMAIN_REQUESTS,
RUNTIME_HOST_PROTOCOL_VERSION,
+ SESSION_CONTINUITY_SCHEMA_VERSION,
type ClientFrame,
+ type EncodedProtocolMessage,
type HostFrame,
type ResponseFrame,
type TurnSnapshot,
@@ -53,6 +55,7 @@ import type {
ClientCapabilityService,
} from '../server/client-capability-service.js';
import { RuntimeHostConnectionSession } from '../server/connection-session.js';
+import type { SessionContinuityService } from '../server/session-continuity-service.js';
import {
createUnavailableAccessAuthorityOperationHandlers,
createUnavailableDomainOperationHandlers,
@@ -68,6 +71,7 @@ import {
RuntimeHostOutboundQueueError,
} from '../server/serial-outbound-writer.js';
import { FramedTransport } from '../transport/framed-transport.js';
+import type { RuntimeHostMessageTransport } from '../transport/message-transport.js';
const CURRENT_PROTOCOL = {
min: RUNTIME_HOST_PROTOCOL_VERSION,
@@ -361,6 +365,191 @@ test('serial outbound writer reports its 2 MiB byte bound before its frame bound
}
});
+test('flushes concurrent subscription opens before activating their live frame streams', async () => {
+ const releaseWrites = deferred();
+ const requestsEntered = deferred();
+ const allWrites = deferred();
+ const inbound = Array.from({ length: 16 }, (_, index) => ({
+ requestId: `open-${index}`,
+ operation: 'subscription.open',
+ input: { sessionId: `session-${index}`, transcript: { kind: 'none' } },
+ }));
+ const written: EncodedProtocolMessage[] = [];
+ let aborted = false;
+ let resolveClosed!: () => void;
+ let rejectRead: ((error: Error) => void) | undefined;
+ const closed = new Promise<void>((resolve) => {
+ resolveClosed = resolve;
+ });
+ const transport: RuntimeHostMessageTransport = {
+ closed,
+ read: async () => {
+ const frame = inbound.shift();
+ if (frame) return frame;
+ return new Promise<never>((_resolve, reject) => {
+ rejectRead = reject;
+ });
+ },
+ write: async (message) => {
+ await releaseWrites.promise;
+ written.push(message);
+ if (written.length === 32) allWrites.resolve();
+ },
+ closeAfterFlush: () => {
+ resolveClosed();
+ },
+ abort: (error) => {
+ if (aborted) return;
+ aborted = true;
+ rejectRead?.(error ?? new Error('in-memory transport aborted'));
+ resolveClosed();
+ },
+ };
+ let openCalls = 0;
+ let sink: Parameters<SessionContinuityService['attachConnection']>[1] | undefined;
+ const largeSnapshot = (sessionId: string) => {
+ const snapshot = canonicalProjection(sessionId);
+ return {
+ ...snapshot,
+ schemaVersion: SESSION_CONTINUITY_SCHEMA_VERSION,
+ projectionRevision: 1,
+ queue: {
+ ...snapshot.queue,
+ followup: Array.from({ length: 2 }, (_, index) => ({
+ entryId: `entry-${sessionId}-${index}`,
+ messageId: `message-${sessionId}-${index}`,
+ content: { text: 'q'.repeat(25 * 1024), quotes: [] },
+ placement: 'next_turn' as const,
+ state: 'queued' as const,
+ })),
+ },
+ };
+ };
+ const continuity: SessionContinuityService = {
+ handlers: {
+ 'subscription.open': async (input) => {
+ openCalls += 1;
+ if (openCalls === 16) requestsEntered.resolve();
+ return {
+ ok: true,
+ result: {
+ hostEpoch: 'host-epoch',
+ subscriptionId: `subscription-${input.sessionId}`,
+ nextSequence: 1,
+ snapshot: largeSnapshot(input.sessionId),
+ activeAssistantStreams: Array.from({ length: 180 }, (_, index) => ({
+ kind: 'text' as const,
+ turnId: `turn-${input.sessionId}`,
+ messageId: `stream-${input.sessionId}-${index}`,
+ })),
+ transcript: transcriptBootstrapFor(input.sessionId),
+ },
+ };
+ },
+ 'subscription.close': async (input) => ({
+ ok: true,
+ result: { subscriptionId: input.subscriptionId },
+ }),
+ 'session.transcript.overlay.release': async () => ({
+ ok: false,
+ error: { code: 'operation_unavailable', message: 'not used' },
+ }),
+ 'session.transcript.page': async () => ({
+ ok: false,
+ error: { code: 'operation_unavailable', message: 'not used' },
+ }),
+ },
+ attachConnection: (_connectionId, attachedSink) => {
+ sink = attachedSink;
+ return {
+ activate: (subscriptionId) => {
+ const sessionId = subscriptionId.slice('subscription-'.length);
+ void sink
+ ?.send({
+ kind: 'subscription.session_projection',
+ hostEpoch: 'host-epoch',
+ subscriptionId,
+ sequence: 1,
+ snapshot: largeSnapshot(sessionId),
+ })
+ .catch(() => undefined);
+ },
+ abort() {},
+ close() {},
+ };
+ },
+ };
+ const handlers: OperationHandlerMap = {
+ 'host.status': async () => ({
+ ok: true,
+ result: {
+ hostEpoch: 'host-epoch',
+ compositionId: 'maka.interactive',
+ compositionRevision: '1',
+ state: 'ready',
+ connections: 1,
+ activeOperations: 0,
+ activeResidencies: 0,
+ },
+ }),
+ ...UNUSED_HOST_DIAGNOSTICS_HANDLER,
+ ...createUnavailableAccessAuthorityOperationHandlers(),
+ ...createHandlers(async (input) => ({
+ ok: true,
+ result: runningSnapshot(input.sessionId, input.turnId),
+ })),
+ ...continuity.handlers,
+ };
+ const session = new RuntimeHostConnectionSession({
+ transport,
+ connection: acceptedConnection('concurrent-subscription-opens'),
+ resolveHandlers: () => handlers,
+ resolveContinuity: () => continuity,
+ beginOperation: async () => ({
+ acquireResidency: () => ({ release() {} }),
+ seal() {},
+ finish() {},
+ }),
+ onTeardown() {},
+ });
+ const run = session.run();
+ try {
+ await withTimeout(requestsEntered.promise, 1_000, 'subscription opens were not dispatched');
+ await new Promise<void>((resolve) => setImmediate(resolve));
+ assert.equal(aborted, false);
+
+ releaseWrites.resolve();
+ await withTimeout(allWrites.promise, 2_000, 'subscription frames were not flushed');
+ const frames = written.map((message) =>
+ decodeHostFrame(JSON.parse(message.toString('utf8')) as unknown),
+ );
+ const openResponseBytes = written.reduce(
+ (total, message, index) =>
+ !('kind' in frames[index]!) && frames[index]!.operation === 'subscription.open'
+ ? total + message.byteLength
+ : total,
+ 0,
+ );
+ assert.ok(openResponseBytes < 2 * 1024 * 1024);
+ assert.ok(written.reduce((total, message) => total + message.byteLength, 0) > 2 * 1024 * 1024);
+ assert.equal(
+ frames.filter((frame) => !('kind' in frame) && frame.operation === 'subscription.open')
+ .length,
+ 16,
+ );
+ assert.equal(
+ frames.filter((frame) => 'kind' in frame && frame.kind === 'subscription.session_projection')
+ .length,
+ 16,
+ );
+ assert.equal(aborted, false);
+ } finally {
+ releaseWrites.resolve();
+ transport.abort();
+ await run;
+ }
+});
+
test('clean read EOF drains an already dispatched response before closing', async () => {
const fixture = await openHalfClosedDispatchedSession('half-close');
try {
@@ -1439,6 +1628,43 @@ function canonicalProjection(sessionId: string): CanonicalSessionProjection {
};
}
+function transcriptBootstrapFor(sessionId: string) {
+ const contents = Buffer.from('t'.repeat(16 * 1024));
+ return {
+ throughSequence: 0,
+ overlayMessageCount: 0,
+ durable: {
+ kind: 'page' as const,
+ sessionId,
+ source: 'durable' as const,
+ direction: 'older' as const,
+ throughSequence: 0,
+ rawBytes: contents.byteLength,
+ fragments: [
+ {
+ kind: 'durable' as const,
+ sequence: 0,
+ byteOffset: 0,
+ totalBytes: contents.byteLength,
+ payloadDigest: null,
+ data: contents.toString('base64'),
+ },
+ ],
+ nextCursor: null,
+ },
+ overlay: {
+ kind: 'page' as const,
+ sessionId,
+ source: 'overlay' as const,
+ direction: 'older' as const,
+ throughSequence: 0,
+ rawBytes: 0,
+ fragments: [],
+ nextCursor: null,
+ },
+ };
+}
+
function connectionTextEvent(sessionId: string, index: number, messageId?: string) {
return {
type: 'text_delta' as const,
diff --git a/packages/runtime-host/src/server/connection-session.ts b/packages/runtime-host/src/server/connection-session.ts
index 717699b3e..4f54f122d 100644
--- a/packages/runtime-host/src/server/connection-session.ts
+++ b/packages/runtime-host/src/server/connection-session.ts
@@ -233,9 +233,12 @@ export class RuntimeHostConnectionSession {
response.ok && response.operation === 'subscription.open'
? response.result.subscriptionId
: undefined;
- if (openedSubscriptionId) continuity?.activate(openedSubscriptionId);
try {
await receipt.flushed;
+ // Subscriber-local queues retain pre-activation events. Expose them
+ // only after the open result leaves the connection-wide writer, or a
+ // restore fan-out can make legal responses and first frames overflow it.
+ if (openedSubscriptionId) continuity?.activate(openedSubscriptionId);
} catch (error) {
if (openedSubscriptionId) continuity?.abort(openedSubscriptionId);
throw error;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org