diff --git a/packages/server/src/server/hub/relationship-controller.test.ts b/packages/server/src/server/hub/relationship-controller.test.ts index 23f24fbed0..febc6456c8 100644 --- a/packages/server/src/server/hub/relationship-controller.test.ts +++ b/packages/server/src/server/hub/relationship-controller.test.ts @@ -172,6 +172,30 @@ describe("Hub relationship", () => { await secondResult; }); + test("a stale enrollment rejection cannot discard a fresh pending ceremony", async () => { + relationship = await HubRelationshipHarness.start(); + relationship.holdEnrollment(); + const expiredConnect = relationship.beginConnect("expired-token"); + await relationship.enrollmentBegins(); + + relationship.holdEnrollment(); + const freshConnect = relationship.beginConnect("fresh-token"); + const freshEnrollment = await relationship.enrollmentBegins(); + relationship.rejectEnrollment(0, 403); + await expiredConnect.result; + + expect(relationship.relationshipFile()?.enrollment?.token).toBe("fresh-token"); + expect(relationship.pendingRelationshipRetries()).toBe(0); + + relationship.completeEnrollment(); + await freshConnect.result; + + expect(freshEnrollment.token).toBe("fresh-token"); + expect(relationship.relationshipFile()?.state).toBe("active"); + expect(relationship.enrollmentAttempts()).toHaveLength(2); + expect(relationship.pendingRelationshipRetries()).toBe(0); + }); + test("a rejected pending enrollment is discarded without blocking daemon restart", async () => { relationship = await HubRelationshipHarness.start(); relationship.holdEnrollment(); diff --git a/packages/server/src/server/hub/relationship-controller.ts b/packages/server/src/server/hub/relationship-controller.ts index a475c97357..0dc7a1a5b9 100644 --- a/packages/server/src/server/hub/relationship-controller.ts +++ b/packages/server/src/server/hub/relationship-controller.ts @@ -158,6 +158,7 @@ export class HubRelationshipController implements HubRelationshipManagement { private socket: HubSocketConnection | null = null; private retry: ScheduledRelationshipTask | null = null; private generation = 0; + private enrollmentGeneration = 0; private retryAttempt = 0; private executions: { relationshipId: string; value: HubExecutions } | null = null; @@ -176,8 +177,9 @@ export class HubRelationshipController implements HubRelationshipManagement { async start(): Promise { if (this.record?.state === "active") this.openSocket(this.record, false); if (this.record?.state === "pending") { + const enrollmentGeneration = this.beginEnrollmentAttempt(); try { - await this.tryEnrollment(this.record); + await this.tryEnrollment(this.record, enrollmentGeneration); } catch (error) { if (!(error instanceof HubEnrollmentRejectedError)) throw error; this.options.logger.warn( @@ -212,8 +214,11 @@ export class HubRelationshipController implements HubRelationshipManagement { throw new Error("A pending Hub enrollment already exists for a different Hub"); } this.record = { ...this.record, enrollment: { token: input.token } }; + const enrollmentGeneration = this.beginEnrollmentAttempt(); this.persist(this.record); - await this.tryEnrollment(this.record); + this.state = "connecting"; + this.lastError = null; + await this.tryEnrollment(this.record, enrollmentGeneration); return this.status(); } if (this.record && this.record.state !== "revoked") { @@ -236,7 +241,8 @@ export class HubRelationshipController implements HubRelationshipManagement { this.persist(pending); this.record = pending; this.state = "connecting"; - await this.tryEnrollment(pending); + this.lastError = null; + await this.tryEnrollment(pending, this.beginEnrollmentAttempt()); return this.status(); } @@ -271,8 +277,8 @@ export class HubRelationshipController implements HubRelationshipManagement { return { status: this.status() }; } - private async tryEnrollment(pending: PendingRecord): Promise { - const generation = this.generation; + private async tryEnrollment(pending: PendingRecord, enrollmentGeneration: number): Promise { + if (enrollmentGeneration !== this.enrollmentGeneration) return; const verifier = createHash("sha256").update(pending.credential.secret).digest("base64url"); try { const enrollment = await this.options.remote.enroll({ @@ -285,7 +291,7 @@ export class HubRelationshipController implements HubRelationshipManagement { credentialVerifier: verifier, scopes: pending.relationship.scopes, }); - if (generation !== this.generation) return; + if (enrollmentGeneration !== this.enrollmentGeneration) return; if ( enrollment.relationshipId !== pending.relationship.id || !enrollment.scopes.includes("hub.*") @@ -301,16 +307,17 @@ export class HubRelationshipController implements HubRelationshipManagement { }; this.persist(active); this.record = active; + this.retry = null; this.retryAttempt = 0; this.openSocket(active, false); } catch (error) { - if (generation !== this.generation) return; + if (enrollmentGeneration !== this.enrollmentGeneration) return; if (error instanceof HubEnrollmentRejectedError) { this.remove(); throw error; } this.lastError = error instanceof Error ? error.message : String(error); - this.scheduleEnrollment(pending); + this.scheduleEnrollment(pending, enrollmentGeneration); } } @@ -380,9 +387,15 @@ export class HubRelationshipController implements HubRelationshipManagement { this.schedule(() => this.openSocket(record, true)); } - private scheduleEnrollment(record: PendingRecord): void { + private scheduleEnrollment(record: PendingRecord, enrollmentGeneration: number): void { + if (enrollmentGeneration !== this.enrollmentGeneration) return; this.state = "reconnecting"; - this.schedule(() => void this.tryEnrollment(record)); + this.retry?.cancel(); + const delay = this.retryPolicy.delay(this.retryAttempt++); + this.retry = this.clock.schedule(delay, () => { + if (enrollmentGeneration !== this.enrollmentGeneration) return; + void this.tryEnrollment(record, enrollmentGeneration); + }); } private async tryRevocation(record: DisconnectingRecord): Promise { @@ -435,10 +448,18 @@ export class HubRelationshipController implements HubRelationshipManagement { private cancelLifecycle(): void { ++this.generation; + ++this.enrollmentGeneration; this.retry?.cancel(); this.retry = null; } + private beginEnrollmentAttempt(): number { + this.retry?.cancel(); + this.retry = null; + this.retryAttempt = 0; + return ++this.enrollmentGeneration; + } + private persist(record: HubRelationshipRecord): void { writePrivateFileAtomicSync(this.filePath, `${JSON.stringify(record, null, 2)}\n`); } diff --git a/packages/server/src/server/hub/test-utils/relationship-harness.ts b/packages/server/src/server/hub/test-utils/relationship-harness.ts index c08a1dc2da..78ffd2dfe6 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -174,6 +174,10 @@ class TestRelationshipClock implements HubRelationshipClock, HubRelationshipRetr } } } + + pendingTasks(): number { + return this.tasks.filter((task) => !task.cancelled).length; + } } class MemoryHubSocket extends EventEmitter implements WebSocketLike, HubSocketConnection { @@ -307,7 +311,11 @@ class InMemoryHubRelationships implements HubRelationshipRemote { enrollments: HubEnrollment[] = []; revocations: HubRevocation[] = []; sockets: SocketAttempt[] = []; - private enrollmentGate: Deferred | null = null; + private readonly enrollmentGates: Array> = []; + private readonly heldEnrollments: Array<{ + input: HubEnrollment; + gate: Deferred; + }> = []; private enrollmentObserved = deferred(); private socketObserved = deferred(); private enrollmentRejection: 401 | 403 | null = null; @@ -319,7 +327,7 @@ class InMemoryHubRelationships implements HubRelationshipRemote { constructor(private readonly captureRelationship: () => RelationshipInvocationSnapshot) {} holdEnrollment(): void { - this.enrollmentGate = deferred(); + this.enrollmentGates.push(deferred()); } rejectNextEnrollment(statusCode: 401 | 403): void { @@ -336,21 +344,33 @@ class InMemoryHubRelationships implements HubRelationshipRemote { this.enrollmentRejection = null; throw new HubEnrollmentRejectedError(statusCode); } - if (this.enrollmentGate) return this.enrollmentGate.promise; + const gate = this.enrollmentGates.shift(); + if (gate) { + this.heldEnrollments.push({ input, gate }); + return gate.promise; + } return this.enrollmentResult(input); } completeEnrollment(): void { - const input = this.enrollments.at(-1); - if (!input || !this.enrollmentGate) throw new Error("No enrollment is waiting"); - this.enrollmentGate.resolve(this.enrollmentResult(input)); - this.enrollmentGate = null; + const held = this.heldEnrollments.at(-1); + if (!held) throw new Error("No enrollment is waiting"); + held.gate.resolve(this.enrollmentResult(held.input)); + this.heldEnrollments.splice(this.heldEnrollments.indexOf(held), 1); } loseEnrollmentResponse(): void { - if (!this.enrollmentGate) throw new Error("No enrollment is waiting"); - this.enrollmentGate.reject(new Error("Enrollment response was lost")); - this.enrollmentGate = null; + const held = this.heldEnrollments.at(-1); + if (!held) throw new Error("No enrollment is waiting"); + held.gate.reject(new Error("Enrollment response was lost")); + this.heldEnrollments.splice(this.heldEnrollments.indexOf(held), 1); + } + + rejectEnrollment(index: number, statusCode: 401 | 403): void { + const held = this.heldEnrollments[index]; + if (!held) throw new Error(`Enrollment ${index} is not waiting`); + held.gate.reject(new HubEnrollmentRejectedError(statusCode)); + this.heldEnrollments.splice(index, 1); } async enrollmentAt(index: number): Promise { @@ -535,6 +555,14 @@ export class HubRelationshipHarness { this.remote.loseEnrollmentResponse(); } + rejectEnrollment(index: number, statusCode: 401 | 403): void { + this.remote.rejectEnrollment(index, statusCode); + } + + pendingRelationshipRetries(): number { + return this.clock.pendingTasks(); + } + rejectNextEnrollment(statusCode: 401 | 403): void { this.remote.rejectNextEnrollment(statusCode); }