You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by zh...@apache.org on 2026/09/09 12:21:34 UTC
(maka) branch main updated: fix(runtime-host): retain connection teardown failures (#5106)
This is an automated email from the ASF dual-hosted git repository.
me2seeks pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/main by this push:
new b79a9946d7 fix(runtime-host): retain connection teardown failures (#5106)
b79a9946d7 is described below
commit b79a9946d76b725af7058e9180ea977409f7b114
Author: Zhang <96...@users.noreply.github.com>
AuthorDate: Wed Sep 9 20:21:29 2026 +0800
fix(runtime-host): retain connection teardown failures (#5106)
Generated-by: OpenAI Codex
---
.../src/__tests__/connection-session.test.ts | 61 +++++++++++++++++++++-
.../runtime-host/src/server/connection-session.ts | 38 +++++++++++---
.../src/server/serial-outbound-writer.ts | 6 +--
3 files changed, 93 insertions(+), 12 deletions(-)
diff --git a/packages/runtime-host/src/__tests__/connection-session.test.ts b/packages/runtime-host/src/__tests__/connection-session.test.ts
index 2785a7dc46..3321e037ac 100644
--- a/packages/runtime-host/src/__tests__/connection-session.test.ts
+++ b/packages/runtime-host/src/__tests__/connection-session.test.ts
@@ -364,8 +364,10 @@ test('outbound scheduling prioritizes controls, makes fair data progress, and fe
test('serial outbound writer fails once when its real transport is closed', async () => {
const pair = await openTransportPair();
let failureCalls = 0;
- const writer = new BoundedSerialOutboundWriter(pair.clientTransport, () => {
+ let reportedFailure: Error | undefined;
+ const writer = new BoundedSerialOutboundWriter(pair.clientTransport, (error) => {
failureCalls += 1;
+ reportedFailure = error;
});
try {
pair.clientTransport.abort();
@@ -373,6 +375,8 @@ test('serial outbound writer fails once when its real transport is closed', asyn
const receipt = writer.enqueue(statusResponse('closed-transport'));
await assert.rejects(receipt.flushed);
assert.equal(failureCalls, 1);
+ assert.ok(reportedFailure);
+ assert.match(reportedFailure.message, /closed|write/i);
assert.throws(() => writer.enqueue(statusResponse('after-failure')), /writer is closed/);
assert.equal(failureCalls, 1);
} finally {
@@ -623,6 +627,7 @@ test('clean read EOF drains an already dispatched response before closing', asyn
assert.equal(response.ok, true);
await withTimeout(fixture.run, 1_000, 'connection did not close after draining its response');
assert.equal(fixture.teardownCalls(), 1);
+ assert.deepEqual(fixture.diagnostics, []);
} finally {
await fixture.close();
}
@@ -647,6 +652,56 @@ test('a fatal transport close during clean EOF drain tears down exactly once', a
}
});
+test('records an unexpected accepted-connection failure before teardown', async () => {
+ const pair = await openTransportPair();
+ const teardownObserved = deferred();
+ const logs: string[] = [];
+ const session = new RuntimeHostConnectionSession({
+ transport: pair.serverTransport,
+ connection: acceptedConnection('failed-admission'),
+ resolveHandlers: () => ({
+ '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,
+ ...createUnavailableHostCoreOperationHandlers(),
+ ...createUnavailableDomainOperationHandlers(),
+ }),
+ resolveContinuity: () => undefined,
+ beginOperation: async () => {
+ throw new Error('api_key=sk-connection-secret123');
+ },
+ onDiagnostic: (diagnostic) => logs.push(diagnostic),
+ onTeardown: () => teardownObserved.resolve(),
+ });
+ const run = session.run();
+ try {
+ await writeProtocolFrame(pair.clientTransport, {
+ requestId: 'failed-admission-request',
+ operation: 'turn.query',
+ input: { sessionId: 'session', turnId: 'turn' },
+ });
+ await withTimeout(teardownObserved.promise, 1_000, 'connection did not tear down');
+ await withTimeout(run, 1_000, 'connection did not settle');
+ assert.equal(logs.length, 1);
+ assert.match(logs[0] ?? '', /connection session failed/);
+ assert.match(logs[0] ?? '', /\[redacted\]/i);
+ assert.doesNotMatch(logs[0] ?? '', /sk-connection-secret123/);
+ } finally {
+ pair.clientTransport.abort();
+ await Promise.allSettled([run, pair.close()]);
+ }
+});
+
test('a connection accepted before composition exists resolves ready handlers without reconnecting', async () => {
const base = await mkdtemp(join(tmpdir(), 'maka-runtime-host-pre-ready-'));
const root = join(base, 'root');
@@ -1419,6 +1474,7 @@ interface TransportPair {
interface HalfClosedDispatchedSession {
pair: TransportPair;
+ diagnostics: string[];
releaseHandler: Deferred;
teardownObserved: Deferred;
run: Promise<void>;
@@ -1459,6 +1515,7 @@ async function openHalfClosedDispatchedSession(
const handlerEntered = deferred();
const releaseHandler = deferred();
const teardownObserved = deferred();
+ const diagnostics: string[] = [];
let teardownCalls = 0;
const session = new RuntimeHostConnectionSession({
transport: pair.serverTransport,
@@ -1493,6 +1550,7 @@ async function openHalfClosedDispatchedSession(
seal() {},
finish() {},
}),
+ onDiagnostic: (diagnostic) => diagnostics.push(diagnostic),
onTeardown: () => {
teardownCalls += 1;
teardownObserved.resolve();
@@ -1511,6 +1569,7 @@ async function openHalfClosedDispatchedSession(
await withTimeout(readEnded, 1_000, 'Host did not observe Client read EOF');
return {
pair,
+ diagnostics,
releaseHandler,
teardownObserved,
run,
diff --git a/packages/runtime-host/src/server/connection-session.ts b/packages/runtime-host/src/server/connection-session.ts
index 59819c44bb..cf0e4e5217 100644
--- a/packages/runtime-host/src/server/connection-session.ts
+++ b/packages/runtime-host/src/server/connection-session.ts
@@ -25,6 +25,7 @@ import {
type HostOperationErrorCode,
type RequestFrame,
} from '../protocol/index.js';
+import { runtimeHostLogBuffer } from '../process-diagnostics.js';
import type { RuntimeHostMessageTransport } from '../transport/message-transport.js';
import {
dispatchOperation,
@@ -54,6 +55,7 @@ import {
authorizeRuntimeHostOperation,
hasRuntimeHostOperationGrant,
} from './connection-authority.js';
+import { boundedFailureDiagnostic } from './failure-diagnostic.js';
type AcceptedConnectionContext = Omit<
ConnectionContext,
@@ -78,12 +80,14 @@ export interface RuntimeHostConnectionSessionOptions {
resolveHostChanges?(): HostChangeFeed | undefined;
resolveSharedSessionId?(): string | undefined;
beginOperation(frame: RequestFrame): Promise<ConnectionOperationLease | HostOperationErrorCode>;
+ onDiagnostic?(diagnostic: string): void;
onTeardown(): void;
}
export class RuntimeHostConnectionSession {
readonly #options: RuntimeHostConnectionSessionOptions;
readonly #writer: BoundedSerialOutboundWriter;
+ readonly #onDiagnostic: (diagnostic: string) => void;
readonly #requests = new Map<string, Promise<void>>();
#transcriptPageTail: Promise<void> = Promise.resolve();
#inFlightStatusRequests = 0;
@@ -98,7 +102,9 @@ export class RuntimeHostConnectionSession {
constructor(options: RuntimeHostConnectionSessionOptions) {
this.#options = options;
- this.#writer = new BoundedSerialOutboundWriter(options.transport, () => this.#teardown());
+ this.#onDiagnostic =
+ options.onDiagnostic ?? ((diagnostic) => runtimeHostLogBuffer.append('error', diagnostic));
+ this.#writer = new BoundedSerialOutboundWriter(options.transport, (error) => this.#fail(error));
}
async run(): Promise<void> {
@@ -110,8 +116,8 @@ export class RuntimeHostConnectionSession {
if (!isReadEof(error)) throw error;
await this.#closeAfterDispatchedReplies();
}
- } catch {
- this.#teardown();
+ } catch (error) {
+ this.#fail(error);
} finally {
this.#teardown();
await Promise.allSettled(this.#requests.values());
@@ -154,7 +160,7 @@ export class RuntimeHostConnectionSession {
if (
!authorizeClientCapabilityFrame(this.#options.connection.authority, capabilityFrame)
) {
- this.#teardown();
+ this.#fail(new Error('Runtime Host Client Capability frame is not authorized'));
return;
}
this.#ensureClientCapabilities()?.accept(capabilityFrame);
@@ -165,11 +171,15 @@ export class RuntimeHostConnectionSession {
const usesLivenessReserve =
this.#requests.size === RUNTIME_HOST_MAX_IN_FLIGHT_DOMAIN_REQUESTS &&
(frame.operation === 'host.status' || this.#inFlightStatusRequests > 0);
+ if (this.#requests.has(frame.requestId)) {
+ this.#fail(new Error('Runtime Host Client reused an active request id'));
+ return;
+ }
if (
- this.#requests.has(frame.requestId) ||
- (this.#requests.size >= RUNTIME_HOST_MAX_IN_FLIGHT_DOMAIN_REQUESTS && !usesLivenessReserve)
+ this.#requests.size >= RUNTIME_HOST_MAX_IN_FLIGHT_DOMAIN_REQUESTS &&
+ !usesLivenessReserve
) {
- this.#teardown();
+ this.#fail(new Error('Runtime Host Client exceeded the in-flight request limit'));
return;
}
this.#dispatch(frame);
@@ -183,7 +193,7 @@ export class RuntimeHostConnectionSession {
? this.#transcriptPageTail.then(() => this.#handleRequest(frame))
: this.#handleRequest(frame);
const task = handling
- .catch(() => this.#teardown())
+ .catch((error: unknown) => this.#fail(error))
.finally(() => {
if (this.#requests.get(frame.requestId) === task) {
this.#requests.delete(frame.requestId);
@@ -398,6 +408,18 @@ export class RuntimeHostConnectionSession {
this.#hostChanges = undefined;
}
+ #fail(error: unknown): void {
+ if (this.#closed) return;
+ try {
+ this.#onDiagnostic(
+ `[runtime-host] connection session failed: ${boundedFailureDiagnostic(error)}`,
+ );
+ } catch {
+ // Diagnostics cannot keep an invalid connection alive.
+ }
+ this.#teardown();
+ }
+
#teardown(): void {
if (this.#closed) return;
this.#closed = true;
diff --git a/packages/runtime-host/src/server/serial-outbound-writer.ts b/packages/runtime-host/src/server/serial-outbound-writer.ts
index 1a3ac5a3f3..06d2da2a50 100644
--- a/packages/runtime-host/src/server/serial-outbound-writer.ts
+++ b/packages/runtime-host/src/server/serial-outbound-writer.ts
@@ -69,7 +69,7 @@ export class RuntimeHostOutboundQueueError extends Error {
export class BoundedSerialOutboundWriter {
readonly #transport: RuntimeHostMessageTransport;
- readonly #onFailure: () => void;
+ readonly #onFailure: (error: Error) => void;
readonly #queue: QueuedFrame[] = [];
#queuedBytes = 0;
#writing = false;
@@ -78,7 +78,7 @@ export class BoundedSerialOutboundWriter {
#controlBurst = 0;
#dataLane = 0;
- constructor(transport: RuntimeHostMessageTransport, onFailure: () => void) {
+ constructor(transport: RuntimeHostMessageTransport, onFailure: (error: Error) => void) {
this.#transport = transport;
this.#onFailure = onFailure;
}
@@ -157,7 +157,7 @@ export class BoundedSerialOutboundWriter {
#fail(error: Error): void {
if (this.#closed) return;
this.close(error);
- this.#onFailure();
+ this.#onFailure(error);
}
#nextFrame(): QueuedFrame | undefined {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org