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:34 UTC
(maka) 01/03: fix(runtime-host): flush subscription opens before activation
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