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/09/02 02:48:52 UTC
(maka) 01/01: fix(runtime-host): avoid false Guest reconnects
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch fix/session-guest-reconnect-liveness
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 2a4295cfacaf938ff6253b106d29c2e5bf36efb6
Author: Wang <m4...@gmail.com>
AuthorDate: Wed Sep 2 10:48:37 2026 +0800
fix(runtime-host): avoid false Guest reconnects
Treat authenticated Session traffic as liveness while periodic route status is delayed, and keep transcript replay out of the Host readiness boundary during recovery.
Generated-by: Codex
---
.../runtime-host-session-observer.test.ts | 68 +++++++++++++++++++
.../runtime-host-session-observation-registry.ts | 66 ++++++++++++-------
.../__tests__/session-subscription-client.test.ts | 76 +++++++++++++++++++---
packages/runtime-host/src/client/connection.ts | 27 +++++++-
4 files changed, 199 insertions(+), 38 deletions(-)
diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts
index b4f7c03a9c..1adc4475dc 100644
--- a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts
+++ b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts
@@ -37,6 +37,7 @@ import type { DesktopRuntimeHostSession } from "../runtime-host-client.js";
import {
DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES,
type DesktopTranscriptBatch,
+ type DesktopTranscriptOpenResult,
} from '../../preload/transcript-contract.js';
import { RuntimeHostSessionObservationRegistry } from "../runtime-host-session-observation-registry.js";
import {
@@ -485,6 +486,73 @@ test('restores transcript consumers across Host replacement', async () => {
await observations.close();
});
+test('does not hold Host observation recovery on transcript replay', async () => {
+ const observations = new RuntimeHostSessionObservationRegistry();
+ const target = eventTarget(18);
+ const transcriptTarget: RuntimeHostTranscriptTarget = {
+ id: 19,
+ send() {},
+ once() {},
+ off() {},
+ };
+ const transcriptResult = (generation: string) => ({
+ sessionId: 'session-1',
+ generation,
+ hostEpoch: `host-${generation}`,
+ readThroughMessageId: null,
+ });
+ const source = (generation: string) => ({
+ async observe() {},
+ async unobserve() {},
+ async openTranscript() {
+ return transcriptResult(generation);
+ },
+ async loadTranscriptBefore() {},
+ async loadTranscriptAround() {},
+ async closeTranscript() {},
+ });
+ const first = source('first');
+ await observations.attach(first);
+ await observations.observe('session-1', 'observer-1', target);
+ await observations.openTranscript('session-1', 'consumer-1', transcriptTarget);
+ observations.detach(first);
+
+ const observationSeed = deferred<void>();
+ const transcriptReplay = deferred<DesktopTranscriptOpenResult>();
+ let transcriptReplayStarted = false;
+ let transcriptReplayCompleted = false;
+ const replacement = {
+ async observe() {
+ await observationSeed.promise;
+ },
+ async unobserve() {},
+ async openTranscript() {
+ transcriptReplayStarted = true;
+ const result = await transcriptReplay.promise;
+ transcriptReplayCompleted = true;
+ return result;
+ },
+ async loadTranscriptBefore() {},
+ async loadTranscriptAround() {},
+ async closeTranscript() {},
+ };
+ const attaching = observations.attach(replacement);
+ let attached = false;
+ void attaching.then(() => {
+ attached = true;
+ });
+ await waitFor(() => transcriptReplayStarted);
+ assert.equal(attached, false);
+
+ observationSeed.resolve();
+ assert.deepEqual(await attaching, ['session-1']);
+ assert.equal(attached, true);
+
+ transcriptReplay.resolve(transcriptResult('second'));
+ await waitFor(() => transcriptReplayCompleted);
+ await observations.close();
+});
+
test('fences transcript range failures to the current registration and Host source', async () => {
const observations = new RuntimeHostSessionObservationRegistry();
const target: RuntimeHostTranscriptTarget = {
diff --git a/apps/desktop/src/main/runtime-host-session-observation-registry.ts b/apps/desktop/src/main/runtime-host-session-observation-registry.ts
index 289c154b63..724a4f0213 100644
--- a/apps/desktop/src/main/runtime-host-session-observation-registry.ts
+++ b/apps/desktop/src/main/runtime-host-session-observation-registry.ts
@@ -145,6 +145,13 @@ export class RuntimeHostSessionObservationRegistry {
}
this.#source = source;
this.#bindTarget = bindTarget;
+ // Transcript replay is a recoverable renderer read replica, not part of
+ // the Runtime Host availability boundary. Start it alongside Session
+ // observation seeding, but do not keep the whole Host in `reconnecting`
+ // while a large transcript is replayed and acknowledged. The existing
+ // consumer remains registered and receives its replacement generation as
+ // soon as replay completes.
+ this.#restoreTranscripts(source);
const restored = await Promise.all(
[...this.#registrations].map(async ([observerId, registration]) => {
try {
@@ -177,31 +184,6 @@ export class RuntimeHostSessionObservationRegistry {
}
}),
);
- await Promise.all(
- [...this.#transcripts].map(async ([consumerId, registration]) => {
- try {
- const transcriptSource = requireTranscriptSource(source);
- const result = await transcriptSource.openTranscript(
- registration.sessionId,
- consumerId,
- this.#bindTranscriptTarget(registration.target),
- );
- if (this.#source === source && this.#transcripts.get(consumerId) === registration) {
- registration.lifecycle = 'active';
- registration.ready.resolve(result);
- } else {
- await transcriptSource.closeTranscript(consumerId);
- }
- } catch (error) {
- if (this.#source === source && this.#transcripts.get(consumerId) === registration) {
- if (registration.lifecycle === 'pending') {
- this.#deleteTranscript(consumerId, registration);
- }
- this.#onError(error);
- }
- }
- }),
- );
return [...new Set(restored.filter((sessionId): sessionId is string => !!sessionId))];
}
@@ -486,6 +468,40 @@ export class RuntimeHostSessionObservationRegistry {
}
}
+ #restoreTranscripts(source: SessionObservationSource): void {
+ for (const [consumerId, registration] of this.#transcripts) {
+ void this.#restoreTranscript(source, consumerId, registration);
+ }
+ }
+
+ async #restoreTranscript(
+ source: SessionObservationSource,
+ consumerId: string,
+ registration: TranscriptRegistration,
+ ): Promise<void> {
+ try {
+ const transcriptSource = requireTranscriptSource(source);
+ const result = await transcriptSource.openTranscript(
+ registration.sessionId,
+ consumerId,
+ this.#bindTranscriptTarget(registration.target),
+ );
+ if (this.#source === source && this.#transcripts.get(consumerId) === registration) {
+ registration.lifecycle = 'active';
+ registration.ready.resolve(result);
+ } else {
+ await transcriptSource.closeTranscript(consumerId);
+ }
+ } catch (error) {
+ if (this.#source === source && this.#transcripts.get(consumerId) === registration) {
+ if (registration.lifecycle === 'pending') {
+ this.#deleteTranscript(consumerId, registration);
+ }
+ this.#onError(error);
+ }
+ }
+ }
+
#deleteTranscript(consumerId: string, registration: TranscriptRegistration): void {
if (this.#transcripts.get(consumerId) !== registration) return;
this.#transcripts.delete(consumerId);
diff --git a/packages/runtime-host/src/__tests__/session-subscription-client.test.ts b/packages/runtime-host/src/__tests__/session-subscription-client.test.ts
index 7f5ba135a3..3f88772151 100644
--- a/packages/runtime-host/src/__tests__/session-subscription-client.test.ts
+++ b/packages/runtime-host/src/__tests__/session-subscription-client.test.ts
@@ -25,7 +25,7 @@ import { createServer, type Server } from 'node:net';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { test } from 'node:test';
-import { setImmediate as delayImmediate } from 'node:timers/promises';
+import { setImmediate as delayImmediate, setTimeout as delay } from 'node:timers/promises';
import {
prepareStorageRootControlDirectory,
resolveStorageRoot,
@@ -1310,6 +1310,58 @@ test('probes an otherwise idle accepted Runtime Host connection', { timeout: 2_0
);
});
+test('accepts Session traffic as liveness while a route status observation is delayed', {
+ timeout: 5_000,
+}, async () => {
+ const observed = deferred<HostStatusResult>();
+ await withProtocolPeer(
+ async (transport, hostEpoch, rootId) => {
+ const hello = decodeClientFrame(await transport.read(1_000));
+ assert.ok('kind' in hello && hello.kind === 'hello');
+ await writeProtocolFrame(transport, {
+ kind: 'accepted',
+ rootId,
+ hostEpoch,
+ connectionId: 'connection-active-liveness',
+ selectedProtocol: RUNTIME_HOST_PROTOCOL_VERSION,
+ compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH,
+ compositionId: 'maka.interactive',
+ compositionRevision: '1',
+ state: 'ready',
+ });
+ const probe = decodeClientFrame(await transport.read(1_000));
+ assert.ok(!('kind' in probe));
+ assert.equal(probe.operation, 'host.status');
+
+ // Keep the status request pending past its original two-second deadline.
+ // A valid Session frame in between proves the same authenticated
+ // transport is alive and must extend that deadline.
+ await delay(1_100);
+ await writeProtocolFrame(transport, {
+ kind: 'session.catalog.changed',
+ revision: 1,
+ sessionId: 'shared-session',
+ });
+ await delay(1_100);
+ await writeProtocolFrame(transport, {
+ requestId: probe.requestId,
+ operation: 'host.status',
+ ok: true,
+ result: hostStatus(hostEpoch),
+ });
+ await answerStatus(transport, hostEpoch);
+ },
+ async (connection) => {
+ await observed.promise;
+ assert.equal((await connection.status()).hostEpoch, connection.hostEpoch);
+ },
+ {
+ livenessIntervalMs: 20,
+ onHostStatus: observed.resolve,
+ },
+ );
+});
+
async function withProtocolPeer(
serve: (transport: FramedTransport, hostEpoch: string, rootId: string) => Promise<void>,
run: (connection: RuntimeHostConnection) => Promise<void>,
@@ -1426,18 +1478,22 @@ async function answerStatus(transport: FramedTransport, hostEpoch: string): Prom
requestId: request.requestId,
operation: 'host.status',
ok: true,
- result: {
- hostEpoch,
- compositionId: 'maka.interactive',
- compositionRevision: '1',
- state: 'ready',
- connections: 1,
- activeOperations: 1,
- activeResidencies: 0,
- },
+ result: hostStatus(hostEpoch),
});
}
+function hostStatus(hostEpoch: string): HostStatusResult {
+ return {
+ hostEpoch,
+ compositionId: 'maka.interactive',
+ compositionRevision: '1',
+ state: 'ready',
+ connections: 1,
+ activeOperations: 1,
+ activeResidencies: 0,
+ };
+}
+
function openResult(
hostEpoch: string,
subscriptionId: string,
diff --git a/packages/runtime-host/src/client/connection.ts b/packages/runtime-host/src/client/connection.ts
index 9f13a46141..abbd00d4f5 100644
--- a/packages/runtime-host/src/client/connection.ts
+++ b/packages/runtime-host/src/client/connection.ts
@@ -100,7 +100,10 @@ export interface ConnectRuntimeHostInput {
handshakeTimeoutMs?: number;
/**
* Maximum quiet interval before an end-to-end Host liveness probe.
- * Any valid inbound frame restarts the interval. Injectable so tests can
+ * Any valid inbound frame restarts the quiet interval. A Host status
+ * observer additionally uses this as its periodic observation cadence;
+ * inbound traffic then extends an outstanding probe's progress deadline
+ * without suppressing future status observations. Injectable so tests can
* exercise the cadence without waiting the real default (2s).
*/
livenessIntervalMs?: number;
@@ -355,6 +358,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection {
readonly #sessionCatalogChangeListeners = new Set<(frame: SessionCatalogChangedFrame) => void>();
readonly #scheduledTaskChangeListeners = new Set<(frame: ScheduledTaskChangedFrame) => void>();
#livenessTimer: NodeJS.Timeout | undefined;
+ #livenessProbeDeadline: NodeJS.Timeout | undefined;
#livenessProbePending = false;
#inFlightDomainRequests = 0;
#terminalError: Error | undefined;
@@ -824,6 +828,9 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection {
}
#resetLivenessCheck(): void {
+ // Any authenticated Host frame proves the transport is still making
+ // progress, even when a periodic status observation is slow.
+ this.#resetLivenessProbeDeadline();
// A status observer needs periodic route observations even while other
// Session traffic keeps the connection active.
if (this.#onHostStatus) return;
@@ -845,10 +852,11 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection {
#startLivenessProbe(): void {
if (this.#terminalError || this.#livenessProbePending) return;
this.#livenessProbePending = true;
+ this.#resetLivenessProbeDeadline();
void this.#requestOperation(
'host.status',
{},
- DEFAULT_LIVENESS_TIMEOUT_MS,
+ undefined,
(status) => {
this.#validateHostStatusIdentity(status);
try {
@@ -858,15 +866,26 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection {
// connection it is watching.
}
},
- 'connection',
+ 'request',
)
.catch((error: unknown) => this.#fail(asError(error)))
.finally(() => {
+ if (this.#livenessProbeDeadline) clearTimeout(this.#livenessProbeDeadline);
+ this.#livenessProbeDeadline = undefined;
this.#livenessProbePending = false;
this.#scheduleLivenessCheck();
});
}
+ #resetLivenessProbeDeadline(): void {
+ if (!this.#livenessProbePending || this.#terminalError) return;
+ if (this.#livenessProbeDeadline) clearTimeout(this.#livenessProbeDeadline);
+ this.#livenessProbeDeadline = setTimeout(() => {
+ this.#livenessProbeDeadline = undefined;
+ this.#fail(requestTimeoutError('host.status'));
+ }, DEFAULT_LIVENESS_TIMEOUT_MS);
+ }
+
#acceptSubscriptionFrame(frame: SubscriptionFrame): void {
const subscription = this.#subscriptions.get(frame.subscriptionId);
if (!subscription) {
@@ -931,6 +950,8 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection {
this.#terminalError = error;
if (this.#livenessTimer) clearTimeout(this.#livenessTimer);
this.#livenessTimer = undefined;
+ if (this.#livenessProbeDeadline) clearTimeout(this.#livenessProbeDeadline);
+ this.#livenessProbeDeadline = undefined;
this.#queuedDomainFrames.length = 0;
this.#inFlightDomainRequests = 0;
for (const pending of this.#pendingRequests.values()) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org