jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Disconnect daemons locally when Hub is unreachable (#3321)

Browse Source 
* fix(hub): make daemon disconnect unilateral

Local relationship removal must not depend on reaching the Hub. Fence execution authority, make one bounded courtesy revocation attempt, then clear local state so status and reconnect eligibility agree immediately.

* fix(hub): preserve disconnect intent across crashes

Persist the terminal disconnect marker only while the bounded Hub notification is in flight. Startup removes that marker, so a crash cannot restore the old relationship or create a revocation retry loop.
This commit is contained in:
Mohamed Boudra
2026-08-13 14:00:37 +02:00
committed by GitHub
parent 9ca6f24315
commit d0d60a4052
7 changed files with 195 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/hub/relationship-controller.test.ts
+90 -25
View File
@@ -344,16 +344,18 @@ describe("Hub relationship", () => {
const credential = relationship.relationshipFile()?.credential?.secret;
relationship.loseEnrollmentResponse();
await connecting.result;
relationship.failRevocations(2);
relationship.failRevocations(1);
const disconnected = await relationship.disconnect();
expect(disconnected.state).toBe("disconnecting");
expect(relationship.relationshipFile()?.state).toBe("disconnecting");
const reconnected = await relationship.beginConnect("fresh-token").result;
await relationship.restartDaemon();
await relationship.retry();
expect(relationship.revocationAttempts()).toBe(3);
expect(disconnected).toMatchObject({
state: "not_connected",
hub: null,
warning: expect.stringContaining("server-side revocation may remain pending"),
});
expect(reconnected.state).toBe("connecting");
expect(relationship.revocationAttempts()).toBe(1);
expect(relationship.latestRevocation()).toEqual(
expect.objectContaining({
daemonId: enrollment.daemonId,
@@ -361,7 +363,10 @@ describe("Hub relationship", () => {
credential,
}),
);
expect(relationship.relationshipFile()).toBeNull();
expect(relationship.pendingRelationshipRetries()).toBe(0);
expect(relationship.loggableValues(disconnected)).toContain(
"Failed to notify Hub before removing local relationship",
);
});
test("disconnect revokes only after an in-flight enrollment settles", async () => {
@@ -371,7 +376,8 @@ describe("Hub relationship", () => {
const enrollment = await relationship.enrollmentBegins();
const disconnecting = relationship.beginDisconnect();
await relationship.relationshipStateBecomes("disconnecting");
await relationship.connectionStateBecomes("disconnecting");
expect(relationship.relationshipFile()?.state).toBe("disconnecting");
expect(relationship.revocationAttempts()).toBe(0);
relationship.completeEnrollment();
@@ -385,6 +391,40 @@ describe("Hub relationship", () => {
expect(relationship.relationshipFile()).toBeNull();
});
test("disconnect persists terminal intent only while remote notification is in flight", async () => {
relationship = await HubRelationshipHarness.start();
await relationship.beginConnect().result;
relationship.holdRevocation();
const disconnecting = relationship.beginDisconnect();
await relationship.relationshipStateBecomes("disconnecting");
expect(await relationship.status()).toMatchObject({ state: "disconnecting" });
expect(relationship.pendingRelationshipRetries()).toBe(0);
relationship.completeRevocation();
await expect(disconnecting.result).resolves.toMatchObject({ state: "not_connected" });
expect(relationship.relationshipFile()).toBeNull();
});
test("force disconnect does not wait for or notify an in-flight enrollment", async () => {
relationship = await HubRelationshipHarness.start();
relationship.holdEnrollment();
const connecting = relationship.beginConnect("one-time-token");
await relationship.enrollmentBegins();
const disconnected = await relationship.disconnect(true);
expect(disconnected).toMatchObject({ state: "not_connected", hub: null });
expect(relationship.revocationAttempts()).toBe(0);
expect(relationship.relationshipFile()).toBeNull();
relationship.completeEnrollment();
await connecting.result;
expect(relationship.socketAttempts()).toBe(0);
expect(await relationship.status()).toMatchObject({ state: "not_connected", hub: null });
});
test("daemon restart reconnects the same durable relationship", async () => {
relationship = await HubRelationshipHarness.start();
await relationship.beginConnect().result;
@@ -729,48 +769,73 @@ describe("Hub relationship", () => {
expect(loggable).not.toContain(enrollment.idempotencyKey);
});
test("offline disconnect retries across runtime and restart without opening a socket", async () => {
test("offline disconnect removes local authority after one notify attempt", async () => {
relationship = await HubRelationshipHarness.start();
await relationship.beginConnect().result;
relationship.connectLatestSocket();
relationship.failRevocations(3);
const disconnecting = await relationship.disconnect();
await relationship.retry();
const disconnected = await relationship.disconnect();
await relationship.restartDaemon();
await relationship.retry();
expect(disconnecting.state).toBe("disconnecting");
expect(relationship.revocationAttempts()).toBe(4);
expect(disconnected).toMatchObject({
state: "not_connected",
hub: null,
warning: expect.stringContaining("server-side revocation may remain pending"),
});
expect(await relationship.status()).toMatchObject({ state: "not_connected", hub: null });
expect(relationship.revocationAttempts()).toBe(1);
expect(relationship.socketAttempts()).toBe(1);
expect(relationship.pendingRelationshipRetries()).toBe(0);
expect(relationship.relationshipFile()).toBeNull();
});
test("successful disconnect clears a transient revocation error", async () => {
test("successful disconnect notifies the Hub and returns clean local status", async () => {
relationship = await HubRelationshipHarness.start();
await relationship.beginConnect().result;
relationship.failRevocations(1);
const disconnecting = await relationship.disconnect();
await relationship.retry();
const disconnected = await relationship.disconnect();
expect(disconnecting).toMatchObject({ state: "disconnecting", error: expect.any(String) });
expect(disconnected).toMatchObject({ state: "not_connected", hub: null, error: null });
expect(disconnected).not.toHaveProperty("warning");
expect(relationship.revocationAttempts()).toBe(1);
expect(await relationship.status()).toMatchObject({ state: "not_connected", error: null });
});
test("force disconnect removes local authority and reports the remote warning", async () => {
test("force disconnect removes local authority without notifying the Hub", async () => {
relationship = await HubRelationshipHarness.start();
await relationship.beginConnect().result;
relationship.failRevocations(1);
await relationship.disconnect();
const forced = await relationship.disconnect(true);
expect(forced.state).toBe("not_connected");
expect(forced.warning).toContain("remote revocation");
expect(forced).toMatchObject({ state: "not_connected", hub: null });
expect(forced).not.toHaveProperty("warning");
expect(relationship.revocationAttempts()).toBe(0);
expect(relationship.loggableValues(forced)).not.toContain(
"Failed to notify Hub before removing local relationship",
);
expect(relationship.relationshipFile()).toBeNull();
});
test("startup removes a legacy persisted disconnecting relationship", async () => {
relationship = await HubRelationshipHarness.start();
await relationship.beginConnect().result;
const active = relationship.relationshipFile();
await relationship.corruptRelationshipFile(
JSON.stringify({ ...active, state: "disconnecting" }),
);
await relationship.startStoppedDaemon();
expect(await relationship.status()).toMatchObject({ state: "not_connected", hub: null });
expect(relationship.relationshipFile()).toBeNull();
expect(relationship.revocationAttempts()).toBe(0);
expect(relationship.pendingRelationshipRetries()).toBe(0);
expect(relationship.loggableValues(await relationship.status())).toContain(
"Removed legacy disconnecting Hub relationship during startup",
);
});
test("disconnect leaves no intent artifacts and shutdown closes the owned agent", async () => {
relationship = await HubRelationshipHarness.start();
await relationship.beginConnect().result;
packages/server/src/server/hub/relationship-controller.ts
+30 -32
View File
@@ -197,11 +197,19 @@ export class HubRelationshipController implements HubRelationshipManagement {
this.clock = options.clock ?? systemClock;
this.retryPolicy = options.retryPolicy ?? new BoundedExponentialHubRetryPolicy();
this.record = this.load();
if (this.record?.state === "disconnecting") {
// COMPAT(hubUnilateralDisconnect): added in v0.4.0, remove after 2027-02-13 once legacy records have aged out.
this.options.logger.warn(
{ daemonId: this.record.relationship.daemonId },
"Removed legacy disconnecting Hub relationship during startup",
);
rmSync(this.filePath, { force: true });
this.record = null;
}
if (this.record?.state === "revoked") {
this.state = "revoked";
this.lastError = this.record.reason ?? null;
} else if (this.record?.state === "disconnecting") this.state = "disconnecting";
else if (this.record) this.state = "connecting";
} else if (this.record) this.state = "connecting";
}
async start(): Promise<void> {
@@ -218,7 +226,6 @@ export class HubRelationshipController implements HubRelationshipManagement {
);
}
}
if (this.record?.state === "disconnecting") await this.tryRevocation(this.record);
}
async stop(): Promise<void> {
@@ -291,14 +298,6 @@ export class HubRelationshipController implements HubRelationshipManagement {
await pendingCreateCleanup;
return { status: this.status() };
}
if (input.force) {
this.remove();
await pendingCreateCleanup;
return {
status: this.status(),
warning: "Local Hub credential removed; remote revocation may remain pending.",
};
}
const disconnecting: DisconnectingRecord = {
version: 1,
state: "disconnecting",
@@ -309,12 +308,29 @@ export class HubRelationshipController implements HubRelationshipManagement {
this.persist(disconnecting);
this.record = disconnecting;
this.state = "disconnecting";
if (waitForEnrollment) {
if (waitForEnrollment && !input.force) {
await Promise.all(this.inFlightEnrollments);
}
await this.tryRevocation(disconnecting);
let warning: string | undefined;
if (!input.force) {
try {
await this.options.remote.revoke({
daemonId: disconnecting.relationship.daemonId,
hubOrigin: disconnecting.relationship.hubOrigin,
credential: disconnecting.credential.secret,
});
} catch (error) {
this.options.logger.warn(
{ err: error, daemonId: disconnecting.relationship.daemonId },
"Failed to notify Hub before removing local relationship",
);
warning =
"Hub could not be reached; local relationship removed, but server-side revocation may remain pending.";
}
}
this.remove();
await pendingCreateCleanup;
return { status: this.status() };
return { status: this.status(), ...(warning ? { warning } : {}) };
}
private async tryEnrollment(pending: PendingRecord, enrollmentGeneration: number): Promise<void> {
@@ -457,24 +473,6 @@ export class HubRelationshipController implements HubRelationshipManagement {
});
}
private async tryRevocation(record: DisconnectingRecord): Promise<void> {
const generation = this.generation;
try {
await this.options.remote.revoke({
daemonId: record.relationship.daemonId,
hubOrigin: record.relationship.hubOrigin,
credential: record.credential.secret,
});
if (generation !== this.generation) return;
this.remove();
} catch (error) {
if (generation !== this.generation) return;
this.lastError = error instanceof Error ? error.message : String(error);
this.state = "disconnecting";
this.schedule(() => void this.tryRevocation(record));
}
}
private schedule(task: () => void): void {
this.retry?.cancel();
const generation = this.generation;
packages/server/src/server/hub/test-utils/relationship-harness.ts
+28
View File
@@ -284,6 +284,7 @@ class InMemoryHubRelationships implements HubRelationshipRemote {
private enrollmentRejection: 401 | 403 | null = null;
private enrollmentScopes = ["hub.execution.*"];
private revokeFailures = 0;
private revocationGate: Deferred<void> | null = null;
private readonly relationships = new Set<string>();
readonly enrollmentSnapshots: RelationshipInvocationSnapshot[] = [];
readonly socketSnapshots: RelationshipInvocationSnapshot[] = [];
@@ -353,8 +354,19 @@ class InMemoryHubRelationships implements HubRelationshipRemote {
this.revokeFailures = count;
}
holdRevocation(): void {
this.revocationGate = deferred<void>();
}
completeRevocation(): void {
if (!this.revocationGate) throw new Error("No revocation is waiting");
this.revocationGate.resolve();
this.revocationGate = null;
}
async revoke(input: HubRevocation): Promise<void> {
this.revocations.push({ ...input });
await this.revocationGate?.promise;
if (this.revokeFailures > 0) {
this.revokeFailures--;
throw new Error("Hub is offline");
@@ -510,6 +522,14 @@ export class HubRelationshipHarness {
}
}
async connectionStateBecomes(expected: string): Promise<void> {
for (let attempt = 0; attempt < 100; attempt++) {
if ((await this.status()).state === expected) return;
await new Promise((resolve) => setTimeout(resolve, 10));
}
throw new Error(`Hub connection state did not become ${expected}`);
}
async manageRelationshipFromExternalSocket(): Promise<SessionOutboundMessage[]> {
const address = Object.values(networkInterfaces())
.flat()
@@ -596,6 +616,14 @@ export class HubRelationshipHarness {
this.remote.failRevocations(count);
}
holdRevocation(): void {
this.remote.holdRevocation();
}
completeRevocation(): void {
this.remote.completeRevocation();
}
failProviderPromptStart(prompt = "Create through the Hub"): void {
this.promptsToFail.add(prompt);
}