You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/10/09 12:45:06 UTC
(maka) branch main updated: fix(runtime-host): await initial connection cleanup on close (#5976)
This is an automated email from the ASF dual-hosted git repository.
Astro-Han 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 7763f8d186 fix(runtime-host): await initial connection cleanup on close (#5976)
7763f8d186 is described below
commit 7763f8d186983e49622a8dfe07a3af0782bee432
Author: yum <ya...@chihum.dev>
AuthorDate: Fri Oct 9 20:45:01 2026 +0800
fix(runtime-host): await initial connection cleanup on close (#5976)
Track initial acquisition alongside reconnect work and complete closed only
once all owned cleanup has settled. Register tasks before callbacks can
reenter, and keep explicit cancellation out of the fatal error path.
Close Desktop candidates that fail admission before handoff, and prevent
removed target generations from reactivating or publishing stale failures.
Generated-by: Codex
test(desktop): cover a replaced Local generation's late failure
The ownership guards were only exercised with the manager closed. The
initial Local start is not a serialized target mutation, so once it
marks its generation invalid, retryLocalStart can replace that
generation while the old #markUnavailable still waits on observation
teardown.
Hold that teardown open from the startCandidate seam, let the retry
reach ready, then release the old generation. Its epoch stays inactive
and its late unavailable state is not published over the replacement.
Without the #publishState ownership guard, the old generation publishes
unavailable after the replacement is ready.
---
.../__tests__/runtime-host-desktop-manager.test.ts | 191 +++++++++++++++++++++
.../src/main/runtime-host-desktop-manager.ts | 59 ++++---
.../src/__tests__/reconnecting-connection.test.ts | 184 ++++++++++++++++++++
.../runtime-host/src/client/reconnect-lifecycle.ts | 65 +++++--
4 files changed, 464 insertions(+), 35 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 c39a9ffac0..0d646df20b 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
@@ -1756,6 +1756,131 @@ test('cancelling a live handoff does not authorize any replacement', async () =>
await owner.close();
});
+test('close waits for an installed candidate to finish cleanup', { timeout: 5_000 }, async () => {
+ const host = candidateHarness();
+ const cleanup = deferred<void>();
+ const closeCandidate = host.candidate.close.bind(host.candidate);
+ let cleanupStarted = false;
+ host.candidate.close = async () => {
+ cleanupStarted = true;
+ await cleanup.promise;
+ await closeCandidate();
+ };
+ const manager = createRuntimeHostDesktopManager(LOCAL_INPUT, {
+ startCandidate: async () => ready(host.candidate),
+ });
+ let closing: Promise<void> | undefined;
+ try {
+ await manager.start();
+ let closeSettled = false;
+ closing = manager.close().then(() => { closeSettled = true; });
+ await new Promise<void>((resolve) => setImmediate(resolve));
+ assert.equal(cleanupStarted, true);
+ assert.equal(closeSettled, false);
+ assert.equal(host.closeCalls, 0);
+
+ cleanup.resolve();
+ await closing;
+ assert.equal(host.closeCalls, 1);
+ assert.deepEqual(manager.entries(), []);
+ } finally {
+ cleanup.resolve();
+ await Promise.allSettled([closing ?? manager.close()]);
+ }
+});
+
+test('close waits for a late initial candidate and its cleanup without publishing a failure', { timeout: 5_000 }, async () => {
+ const host = candidateHarness();
+ const connected = deferred<DesktopRuntimeHostCandidateStartResult>();
+ const cleanup = deferred<void>();
+ const closeCandidate = host.candidate.close.bind(host.candidate);
+ let cleanupStarted = false;
+ host.candidate.close = async () => {
+ cleanupStarted = true;
+ await cleanup.promise;
+ await closeCandidate();
+ };
+ const readiness: string[] = [];
+ const failures: Error[] = [];
+ const manager = createRuntimeHostDesktopManager(LOCAL_INPUT, {
+ startCandidate: () => connected.promise,
+ onTargetStateChanged: (state) => readiness.push(state.readiness),
+ onFatalError: (error) => failures.push(error),
+ });
+ const starting = manager.start();
+ let closeSettled = false;
+ const closing = manager.close().then(() => { closeSettled = true; });
+ try {
+ await new Promise<void>((resolve) => setImmediate(resolve));
+ assert.equal(closeSettled, false);
+
+ connected.resolve(ready(host.candidate));
+ await new Promise<void>((resolve) => setImmediate(resolve));
+ assert.equal(cleanupStarted, true);
+ assert.equal(closeSettled, false);
+ assert.equal(host.closeCalls, 0);
+
+ cleanup.resolve();
+ await Promise.all([starting, closing]);
+ await new Promise<void>((resolve) => setImmediate(resolve));
+ assert.equal(host.closeCalls, 1);
+ assert.deepEqual(manager.entries(), []);
+ assert.deepEqual(failures, []);
+ assert.deepEqual(readiness, ['connecting']);
+ } finally {
+ connected.resolve(ready(host.candidate));
+ cleanup.resolve();
+ await Promise.allSettled([starting, closing]);
+ // A regression may drop the late candidate before any owner can close it.
+ if (host.closeCalls === 0) await closeCandidate();
+ }
+});
+
+test('close waits for an aborted initial factory to finish cleanup without publishing a failure', { timeout: 5_000 }, async () => {
+ const host = candidateHarness();
+ const cleanup = deferred<void>();
+ let cleanupStarted = false;
+ const readiness: string[] = [];
+ const failures: Error[] = [];
+ const manager = createRuntimeHostDesktopManager(LOCAL_INPUT, {
+ startCandidate: async (input) => {
+ const signal = input.signal!;
+ if (!signal.aborted) {
+ await new Promise<void>((resolve) => {
+ signal.addEventListener('abort', () => resolve(), { once: true });
+ });
+ }
+ cleanupStarted = true;
+ await cleanup.promise;
+ await host.candidate.close();
+ throw signal.reason;
+ },
+ onTargetStateChanged: (state) => readiness.push(state.readiness),
+ onFatalError: (error) => failures.push(error),
+ });
+ const starting = manager.start();
+ let closeSettled = false;
+ const closing = manager.close().then(() => { closeSettled = true; });
+ try {
+ await new Promise<void>((resolve) => setImmediate(resolve));
+ assert.equal(cleanupStarted, true);
+ assert.equal(closeSettled, false);
+ assert.equal(host.closeCalls, 0);
+
+ cleanup.resolve();
+ await Promise.all([starting, closing]);
+ await new Promise<void>((resolve) => setImmediate(resolve));
+ assert.equal(host.closeCalls, 1);
+ assert.deepEqual(manager.entries(), []);
+ assert.deepEqual(failures, []);
+ assert.deepEqual(readiness, ['connecting']);
+ } finally {
+ cleanup.resolve();
+ await Promise.allSettled([starting, closing]);
+ if (host.closeCalls === 0) await host.candidate.close();
+ }
+});
+
test('a scope published before the first connection waits for that connection', async () => {
const host = candidateHarness();
const connected = deferred<DesktopRuntimeHostCandidateStartResult>();
@@ -1834,6 +1959,72 @@ test('recovers a degraded Local start through a fresh target generation', async
await owner.close();
});
+test('a replaced Local generation does not publish its late failure over the retry', { timeout: 5_000 }, async () => {
+ const recovered = candidateHarness();
+ const unobserveStarted = deferred<void>();
+ const unobserved = deferred<void>();
+ let starts = 0;
+ const epochActive: Array<() => boolean> = [];
+ const published: Array<{ readiness: string; epoch: string }> = [];
+ const failures: Error[] = [];
+ const manager = createRuntimeHostDesktopManager(LOCAL_INPUT, {
+ startCandidate: async (input, observations) => {
+ starts += 1;
+ epochActive.push(input.isTargetActive!);
+ if (starts > 1) return ready(recovered.candidate);
+ // Hold the failed generation's observation teardown open, so its
+ // failure publication is still pending when the retry replaces it.
+ await observations.attach({
+ observe: async () => undefined,
+ unobserve: () => {
+ unobserveStarted.resolve();
+ return unobserved.promise;
+ },
+ });
+ await observations.observe('session-1', 'observer-1', {
+ id: 1, send() {}, once() {}, off() {},
+ });
+ throw new Error('connect failed');
+ },
+ onFatalError: (error) => failures.push(error),
+ onTargetStateChanged: (state) => {
+ published.push({ readiness: state.readiness, epoch: state.epoch });
+ },
+ });
+ const failedEpoch = manager.entries()[0]!.epoch;
+ const starting = manager.start();
+ try {
+ // The initial start is not a serialized target mutation; once it marks
+ // its generation invalid, a retry may replace that generation.
+ await unobserveStarted.promise;
+ await manager.retryLocalStart();
+ const replacement = manager.current();
+ assert.equal(replacement?.readiness, 'ready');
+ assert.notEqual(replacement?.epoch, failedEpoch);
+
+ unobserved.resolve();
+ await starting;
+ await new Promise<void>((resolve) => setImmediate(resolve));
+ assert.deepEqual(published, [
+ { readiness: 'connecting', epoch: failedEpoch },
+ { readiness: 'connecting', epoch: replacement!.epoch },
+ { readiness: 'ready', epoch: replacement!.epoch },
+ ]);
+ assert.deepEqual(
+ manager.entries().map((entry) => [entry.readiness, entry.epoch]),
+ [['ready', replacement!.epoch]],
+ );
+ assert.deepEqual(epochActive.map((isActive) => isActive()), [false, true]);
+ assert.equal(manager.ownsScope({ hostId: 'test-host', targetEpoch: failedEpoch }), false);
+ assert.equal(manager.ownsScope({ hostId: 'test-host', targetEpoch: replacement!.epoch }), true);
+ assert.deepEqual(failures.map((error) => error.message), ['connect failed']);
+ } finally {
+ unobserved.resolve();
+ await Promise.allSettled([starting]);
+ await manager.close();
+ }
+});
+
test('keeps a known repair actionable when its first authority inspection fails', async () => {
const repaired = candidateHarness({ ownership: 'supervised' });
let inspected = false;
diff --git a/apps/desktop/src/main/runtime-host-desktop-manager.ts b/apps/desktop/src/main/runtime-host-desktop-manager.ts
index 45b8aac153..b9491fd503 100644
--- a/apps/desktop/src/main/runtime-host-desktop-manager.ts
+++ b/apps/desktop/src/main/runtime-host-desktop-manager.ts
@@ -754,6 +754,7 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
target: DesktopRuntimeHostTargetGeneration,
error: unknown,
): Promise<void> {
+ if (!this.#ownsTarget(target)) return;
const alreadyUnavailable = target.state.readiness === 'unavailable';
target.valid = false;
this.#ipcMain.deactivate(target.epoch);
@@ -1186,27 +1187,36 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
throw error;
}
if (result.kind === 'ready') {
- ipcMain.completeRegistration();
- const previous = target.lastCandidate;
- const retainedOwnedProcess =
- previous?.hostId === result.candidate.client.hostId &&
- previous.hostEpoch === result.candidate.client.hostEpoch &&
- previous.ownership === 'owned_ephemeral' &&
- result.candidate.hostOwnership === 'owned_ephemeral' &&
- previous.ownedProcess?.pid === result.candidate.hostPid
- ? previous.ownedProcess
- : undefined;
- target.lastCandidate = {
- hostId: result.candidate.client.hostId,
- hostEpoch: result.candidate.client.hostEpoch,
- ownership: result.candidate.hostOwnership,
- ...(result.candidate.ownedProcess
- ? { ownedProcess: trackOwnedProcess(result.candidate.ownedProcess) }
- : retainedOwnedProcess
- ? { ownedProcess: retainedOwnedProcess }
- : {}),
- };
- return { kind: 'ready', value: result.candidate };
+ try {
+ signal.throwIfAborted();
+ if (!this.#ownsTarget(target) || !target.valid) {
+ throw new Error('Runtime Host target is no longer active');
+ }
+ ipcMain.completeRegistration();
+ const previous = target.lastCandidate;
+ const retainedOwnedProcess =
+ previous?.hostId === result.candidate.client.hostId &&
+ previous.hostEpoch === result.candidate.client.hostEpoch &&
+ previous.ownership === 'owned_ephemeral' &&
+ result.candidate.hostOwnership === 'owned_ephemeral' &&
+ previous.ownedProcess?.pid === result.candidate.hostPid
+ ? previous.ownedProcess
+ : undefined;
+ target.lastCandidate = {
+ hostId: result.candidate.client.hostId,
+ hostEpoch: result.candidate.client.hostEpoch,
+ ownership: result.candidate.hostOwnership,
+ ...(result.candidate.ownedProcess
+ ? { ownedProcess: trackOwnedProcess(result.candidate.ownedProcess) }
+ : retainedOwnedProcess
+ ? { ownedProcess: retainedOwnedProcess }
+ : {}),
+ };
+ return { kind: 'ready', value: result.candidate };
+ } catch (error) {
+ await result.candidate.close().catch(() => undefined);
+ throw error;
+ }
}
if (result.kind === 'incompatible' || result.kind === 'upgrade_required') {
const conflict = result;
@@ -1434,6 +1444,7 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
this.#publishState(generation, generation.state);
},
onFatalError: (error) => {
+ if (!this.#ownsTarget(generation)) return;
if (starting) {
fatalDuringStart = error;
} else if (generation.valid) {
@@ -1518,6 +1529,7 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
}
#activate(target: DesktopRuntimeHostTargetGeneration): void {
+ if (!this.#ownsTarget(target) || !target.valid) return;
this.#ipcMain.activate(target.epoch);
const profile = target.target.profile;
if (profile.kind === 'remote' && profile.transport.kind === 'libp2p-direct') {
@@ -1602,10 +1614,15 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
return target;
}
+ #ownsTarget(target: DesktopRuntimeHostTargetGeneration): boolean {
+ return !this.#closed && this.#targets.get(target.target.profile.id) === target;
+ }
+
#publishState(
target: DesktopRuntimeHostTargetGeneration,
state: RuntimeHostDesktopTargetState,
): void {
+ if (!this.#ownsTarget(target)) return;
// A retry starting is not evidence of recovery. Keep its last failure until
// a connection succeeds (or a newer failure replaces it).
if (state.readiness !== 'ready' && target.state.readiness !== 'ready') {
diff --git a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts
index 337cd01a71..800f5e30b3 100644
--- a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts
+++ b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts
@@ -21,6 +21,7 @@ import { deferred } from '@maka/core/test-only/async-primitives';
import assert from 'node:assert/strict';
import test from 'node:test';
import {
+ createRuntimeHostReconnectLifecycle,
createRuntimeHostReconnectingConnection,
remoteRuntimeHostUnavailableError,
RuntimeHostOperationError,
@@ -428,6 +429,189 @@ test('reconnect lifecycle close waits for a resource returned after cancellation
assert.equal(closeSettled, true);
});
+for (const closeDuringConnect of [false, true]) {
+ test(`reconnect lifecycle close waits for its initial connection and late resource cleanup (${closeDuringConnect ? 'reentrant' : 'caller'} close)`, {
+ timeout: 5_000,
+ }, async () => {
+ const acquired = deferredValue<RuntimeHostConnection>();
+ const closeEntered = deferred();
+ const releaseClose = deferred();
+ const late = connectionHarness('late-initial', () => undefined);
+ const lifecycle = createRuntimeHostReconnectLifecycle({
+ connect: async () => {
+ if (closeDuringConnect) void lifecycle.close();
+ return acquired.promise;
+ },
+ });
+ const starting = lifecycle.start();
+ let closeSettled = false;
+ let closedSettled = false;
+ const closing = lifecycle.close().then(() => {
+ closeSettled = true;
+ });
+ void lifecycle.closed.then(() => {
+ closedSettled = true;
+ });
+ const resource = {
+ ...late.connection,
+ close: async () => {
+ closeEntered.resolve();
+ await releaseClose.promise;
+ await late.connection.close();
+ },
+ };
+ try {
+ acquired.resolve(resource);
+ await closeEntered.promise;
+ await new Promise<void>((resolve) => setImmediate(resolve));
+ assert.equal(closeSettled, false);
+ assert.equal(closedSettled, false);
+ assert.equal(lifecycle.current, undefined);
+ releaseClose.resolve();
+ await Promise.all([starting, closing, lifecycle.closed, late.connection.closed]);
+ assert.equal(closeSettled, true);
+ assert.equal(closedSettled, true);
+ } finally {
+ acquired.resolve(resource);
+ releaseClose.resolve();
+ await Promise.allSettled([starting, closing]);
+ await late.connection.close();
+ }
+ });
+}
+
+test('reconnect lifecycle close waits for initial abort cleanup without reporting a fatal failure', {
+ timeout: 5_000,
+}, async () => {
+ const cleanupEntered = deferred();
+ const releaseCleanup = deferred();
+ const fatalErrors: Error[] = [];
+ let cleaned = false;
+ let reentrantCloseSettled = false;
+ const lifecycle = createRuntimeHostReconnectLifecycle({
+ connect: async (signal): Promise<RuntimeHostConnection> => {
+ await new Promise<void>((resolve) =>
+ signal.addEventListener(
+ 'abort',
+ () => {
+ void lifecycle.close().then(() => {
+ reentrantCloseSettled = true;
+ });
+ resolve();
+ },
+ { once: true },
+ ),
+ );
+ cleanupEntered.resolve();
+ await releaseCleanup.promise;
+ cleaned = true;
+ throw signal.reason;
+ },
+ onFatalError: (error) => fatalErrors.push(error),
+ });
+ const starting = lifecycle.start().catch((error: unknown) => error);
+ let closedSettled = false;
+ let closeSettled = false;
+ void lifecycle.closed.then(() => {
+ closedSettled = true;
+ });
+ const closing = lifecycle.close().then(() => {
+ closeSettled = true;
+ });
+ try {
+ await cleanupEntered.promise;
+ await new Promise<void>((resolve) => setImmediate(resolve));
+ assert.equal(closeSettled, false);
+ assert.equal(closedSettled, false);
+ assert.equal(reentrantCloseSettled, false);
+ assert.equal(cleaned, false);
+ releaseCleanup.resolve();
+ await Promise.all([starting, closing, lifecycle.closed]);
+ assert.equal(cleaned, true);
+ assert.equal(closeSettled, true);
+ assert.equal(closedSettled, true);
+ assert.equal(reentrantCloseSettled, true);
+ assert.deepEqual(fatalErrors, []);
+ } finally {
+ releaseCleanup.resolve();
+ await Promise.allSettled([starting, closing]);
+ }
+});
+
+test('reconnect lifecycle closes an unstarted initial resource and cannot start after close', {
+ timeout: 5_000,
+}, async () => {
+ const initial = connectionHarness('unstarted', () => undefined);
+ let connectCalls = 0;
+ const lifecycle = createRuntimeHostReconnectLifecycle({
+ initial: initial.connection,
+ connect: async () => {
+ connectCalls += 1;
+ return initial.connection;
+ },
+ });
+ let resourceClosed = false;
+ void initial.connection.closed.then(() => {
+ resourceClosed = true;
+ });
+ try {
+ await lifecycle.close();
+ await lifecycle.closed;
+ assert.equal(resourceClosed, true);
+ await assert.rejects(lifecycle.start(), /lifecycle is closed/u);
+ assert.equal(connectCalls, 0);
+ assert.equal(lifecycle.current, undefined);
+ } finally {
+ await initial.connection.close();
+ }
+});
+
+test('reconnect lifecycle repeated start acquires only one initial resource', {
+ timeout: 5_000,
+}, async () => {
+ const resource = connectionHarness('initial', () => undefined);
+ const acquired = deferredValue<RuntimeHostConnection>();
+ let connectCalls = 0;
+ const lifecycle = createRuntimeHostReconnectLifecycle({
+ connect: async () => {
+ connectCalls += 1;
+ return acquired.promise;
+ },
+ });
+ const first = lifecycle.start();
+ const second = lifecycle.start();
+ try {
+ acquired.resolve(resource.connection);
+ await Promise.all([first, second]);
+ assert.equal(connectCalls, 1);
+ assert.equal(lifecycle.current, resource.connection);
+ } finally {
+ acquired.resolve(resource.connection);
+ await Promise.allSettled([first, second]);
+ await lifecycle.close();
+ }
+});
+
+test('reconnect lifecycle initial permanent failure rejects waiters and completes close', {
+ timeout: 5_000,
+}, async () => {
+ const failure = new RuntimeHostPermanentReconnectError('initial admission rejected');
+ const fatalErrors: Error[] = [];
+ const lifecycle = createRuntimeHostReconnectLifecycle<RuntimeHostConnection>({
+ connect: async () => {
+ throw failure;
+ },
+ onFatalError: (error) => {
+ fatalErrors.push(error);
+ },
+ });
+ const waiting = assert.rejects(lifecycle.waitForCurrent(), (error) => error === failure);
+ await assert.rejects(lifecycle.start(), (error) => error === failure);
+ await Promise.all([waiting, lifecycle.closed, lifecycle.close()]);
+ assert.deepEqual(fatalErrors, [failure]);
+ assert.equal(lifecycle.current, undefined);
+});
+
test('reconnect lifecycle quiescence suppresses replacement until it is resumed', async () => {
const first = connectionHarness('first', () => undefined);
const replacement = connectionHarness('replacement', () => undefined);
diff --git a/packages/runtime-host/src/client/reconnect-lifecycle.ts b/packages/runtime-host/src/client/reconnect-lifecycle.ts
index 11a1c8a3de..d7210961c3 100644
--- a/packages/runtime-host/src/client/reconnect-lifecycle.ts
+++ b/packages/runtime-host/src/client/reconnect-lifecycle.ts
@@ -112,7 +112,7 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
implements RuntimeHostReconnectLifecycle<T>
{
readonly closed: Promise<void>;
- readonly #initial: T | undefined;
+ #initial: T | undefined;
readonly #connect: (signal: AbortSignal) => Promise<T>;
readonly #retryInitialFailure: boolean | ((error: Error) => boolean);
#connectedOnce = false;
@@ -135,6 +135,7 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
#closed = false;
#quiesced = false;
#terminalError: Error | undefined;
+ #startTask: Promise<void> | undefined;
#reconnectTask: Promise<void> | undefined;
#reconnectAbort: AbortController | undefined;
#discardTask: Promise<void> = Promise.resolve();
@@ -177,10 +178,22 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
return this.#current;
}
- async start(): Promise<void> {
+ start(): Promise<void> {
+ if (this.#closed)
+ return Promise.reject(new Error('Runtime Host reconnect lifecycle is closed'));
+ if (this.#startTask) return this.#startTask;
+ const task = deferredTask();
+ this.#startTask = task.promise;
+ void this.#start().then(task.resolve, task.reject);
+ return task.promise;
+ }
+
+ async #start(): Promise<void> {
try {
- if (this.#initial) {
- this.#install(this.#initial);
+ const initial = this.#initial;
+ this.#initial = undefined;
+ if (initial) {
+ this.#install(initial);
return;
}
this.#initialSignal?.throwIfAborted();
@@ -189,6 +202,7 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
: this.#abort.signal;
this.#install(await this.#connect(signal));
} catch (error) {
+ if (this.#closed) throw error;
const failure = asError(error);
if (
(typeof this.#retryInitialFailure === 'function'
@@ -281,8 +295,11 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
}
close(): Promise<void> {
- this.#closeTask ??= this.#close();
- return this.#closeTask;
+ if (this.#closeTask) return this.#closeTask;
+ const task = deferredTask();
+ this.#closeTask = task.promise;
+ void this.#close().then(task.resolve, task.reject);
+ return task.promise;
}
async #close(): Promise<void> {
@@ -293,9 +310,11 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
this.#abort.abort();
const error = new Error('Runtime Host reconnect lifecycle is closed');
this.#rejectWaiters(error);
- const current = this.#current;
+ const current = this.#current ?? this.#initial;
+ this.#initial = undefined;
this.#setCurrent(undefined);
await current?.close().catch(() => undefined);
+ await this.#startTask?.catch(() => undefined);
await this.#reconnectTask?.catch(() => undefined);
await this.#discardTask;
this.#resolveClosed();
@@ -337,17 +356,21 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
)
return;
const reconnectAbort = new AbortController();
- const task = this.#reconnect(AbortSignal.any([this.#abort.signal, reconnectAbort.signal]));
+ const task = deferredTask();
this.#reconnectAbort = reconnectAbort;
- this.#reconnectTask = task;
+ this.#reconnectTask = task.promise;
+ void this.#reconnect(AbortSignal.any([this.#abort.signal, reconnectAbort.signal])).then(
+ task.resolve,
+ task.reject,
+ );
const finalize = () => {
- if (this.#reconnectTask === task) {
+ if (this.#reconnectTask === task.promise) {
this.#reconnectTask = undefined;
this.#reconnectAbort = undefined;
}
this.#scheduleReconnect();
};
- void task.then(finalize, finalize);
+ void task.promise.then(finalize, finalize);
}
async #reconnect(signal: AbortSignal): Promise<void> {
@@ -443,12 +466,12 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
}
#failPermanently(error: Error): void {
- if (this.#terminalError) return;
+ if (this.#closed || this.#terminalError) return;
this.#terminalError = error;
- this.#abort.abort();
this.#rejectWaiters(error);
+ // Acquisition tasks must settle before close can finish joining them.
+ void this.close();
notifyError(this.#onFatalError, error);
- this.#resolveClosed();
}
#rejectWaiters(error: Error): void {
@@ -460,6 +483,20 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
}
}
+function deferredTask(): {
+ promise: Promise<void>;
+ resolve: () => void;
+ reject: (error: unknown) => void;
+} {
+ let resolve!: () => void;
+ let reject!: (error: unknown) => void;
+ const promise = new Promise<void>((resolveTask, rejectTask) => {
+ resolve = resolveTask;
+ reject = rejectTask;
+ });
+ return { promise, resolve, reject };
+}
+
function notifyError(listener: ((error: Error) => void) | undefined, error: Error): void {
try {
listener?.(error);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org