diff --git a/cloud/apps/relay/src/assignment-store.ts b/cloud/apps/relay/src/assignment-store.ts index 716a0a92146..c002db6cfdb 100644 --- a/cloud/apps/relay/src/assignment-store.ts +++ b/cloud/apps/relay/src/assignment-store.ts @@ -1084,8 +1084,10 @@ export class RelayAssignmentStore { // Only a known member is ever written: the close reason arrives as free text from the host's // socket, so anything unrecognised is dropped here rather than stored and replayed at a phone. - // The timestamp fences the write, which is what lets the caller fire it and forget it - a record - // that loses the race to a later one cannot overwrite it. + // One rule governs this and the clear below, which is what lets both be fired and forgotten: + // the newer event wins, and an equal timestamp goes to the close. A close cannot precede the + // proof of the socket it closes, so same-millisecond pairs are proof-then-close and the reason + // is real; a reason older than the proof is stale and loses, whichever write lands first. async recordHostCloseReason( identity: AssignmentIdentity, reason: unknown, @@ -1097,22 +1099,27 @@ export class RelayAssignmentStore { `UPDATE relay_assignments SET last_host_close_reason = ?, last_host_close_reason_at = ? WHERE user_id = ? AND relay_host_id = ? - AND (last_host_close_reason_at IS NULL OR last_host_close_reason_at < ?)`, + AND (last_host_close_reason_at IS NULL OR last_host_close_reason_at <= ?)`, [parsed, at, identity.userId, identity.relayHostId, at] ) return parsed } - // A host that proved itself is not signed out, whatever it said last. Fenced on `provedAt` so a - // clear still in flight cannot erase a reason the same host recorded after this proof, and - // predicated on a reason being there at all so the common case writes no row version. + // A host that proved itself is not signed out, whatever it said last. The proof advances the + // fence rather than erasing it, and does so whether or not there is a reason to clear: leaving + // the column NULL would let a record still in flight from an older close match `IS NULL` and + // land on a host that has since proved itself. That is why this writes a row version on every + // control connect. It is not folded into the UPDATE `activateControl` already makes on this row + // (`touchAssignment`, `adjustActivityCount`), which would make it free, because that statement + // is awaited and load bearing: on a boot that deferred the column addition it would raise 42703 + // and fail every host control connect, where a statement of its own merely fails and is logged. async clearHostCloseReason(identity: AssignmentIdentity, provedAt: number): Promise { await this.database.query( `UPDATE relay_assignments - SET last_host_close_reason = NULL, last_host_close_reason_at = NULL + SET last_host_close_reason = NULL, last_host_close_reason_at = ? WHERE user_id = ? AND relay_host_id = ? - AND last_host_close_reason IS NOT NULL AND last_host_close_reason_at <= ?`, - [identity.userId, identity.relayHostId, provedAt] + AND (last_host_close_reason_at IS NULL OR last_host_close_reason_at < ?)`, + [provedAt, identity.userId, identity.relayHostId, provedAt] ) } diff --git a/cloud/apps/relay/src/host-session-registry.ts b/cloud/apps/relay/src/host-session-registry.ts index ed93b4b52f9..2d0ac384f5f 100644 --- a/cloud/apps/relay/src/host-session-registry.ts +++ b/cloud/apps/relay/src/host-session-registry.ts @@ -1177,6 +1177,12 @@ export class HostSessionRegistry { return } if (existing) this.observer.recordReconnect() + // The last point both admissions share: a rebind below returns without ever reaching the new + // session built at the end. A host that proved itself is not signed out, whatever it said + // last, and a close within the orphan grace leaves a session a later rebind reuses. Not + // awaited: the row is fenced on this timestamp, so no ordering against a close still being + // written can leave a reason on a host that proved itself at or after it. + this.clearHostCloseReason({ userId: identity.sub, relayHostId: identity.relayHostId }) if (rebind && existing) { const previousSocket = existing.socket if (existing.orphanTimer) clearTimeout(existing.orphanTimer) @@ -1257,10 +1263,6 @@ export class HostSessionRegistry { regionalDrainExpiresAt: null } const sessionKey = this.key(identity.sub, identity.relayHostId) - // A host that proved itself again is not signed out, whatever it said last. Not awaited: - // the row is fenced on this timestamp, so a slow clear cannot erase a later close, and the - // stale reason it leaves behind in the meantime is only readable while the host is absent. - this.clearHostCloseReason({ userId: identity.sub, relayHostId: identity.relayHostId }) this.sessions.set(sessionKey, session) this.wireActiveControl(session) this.sendHelloAck(session) diff --git a/cloud/apps/relay/src/host-signed-out-rejection.test.ts b/cloud/apps/relay/src/host-signed-out-rejection.test.ts index 99c18bd3992..fe2032c3996 100644 --- a/cloud/apps/relay/src/host-signed-out-rejection.test.ts +++ b/cloud/apps/relay/src/host-signed-out-rejection.test.ts @@ -135,7 +135,32 @@ function createCell( ) => Promise } ).activate(socket, identity, null, generation, false, 1, '1.4.173') - return { registry, activate, assignments, readHostCloseReason } + // A reconnect the registry treats as a resume: the orphaned session is reused rather than + // replaced, which is the branch that returns before a new session is ever built. + const activateRebind = (socket: WebSocket): Promise => { + const sessions = (registry as unknown as { sessions: Map }) + .sessions + const existing = sessions.get(`${identity.sub}\0${identity.relayHostId}`) + if (!existing) throw new Error('no session to rebind onto') + return ( + registry as unknown as { + activate: ( + socket: WebSocket, + identity: RelayTokenClaims, + existing: unknown, + generation: number, + rebind: boolean, + assignmentEpoch: number, + appVersion: string + ) => Promise + } + ).activate(socket, identity, existing, existing.generation, true, 1, '1.4.173') + } + const sessionFor = (): unknown => + (registry as unknown as { sessions: Map }).sessions.get( + `${identity.sub}\0${identity.relayHostId}` + ) + return { registry, activate, activateRebind, sessionFor, assignments, readHostCloseReason } } async function dialPhone(registry: HostSessionRegistry): Promise { @@ -249,7 +274,9 @@ describe('host sign-out reason on phone rejection', () => { ) )[0] expect(row?.['last_host_close_reason']).toBeNull() - expect(row?.['last_host_close_reason_at']).toBeNull() + // The fence still carries the proof this host gave when it connected; it is a timestamp, and + // the invented text reached neither column. + expect(typeof row?.['last_host_close_reason_at']).toBe('number') }) it('forgets the sign-out once the host proves itself again', async () => { @@ -324,6 +351,101 @@ describe('host sign-out reason on phone rejection', () => { ) }) + it('clears the sign-out when the host rebinds onto its orphaned session', async () => { + const { registry, activate, activateRebind, sessionFor, assignments } = createCell(database) + const control = new FakeSocket() + await activate(control as unknown as WebSocket, 1) + const orphaned = sessionFor() + control.close(1000, RELAY_HOST_CLOSE_REASON.SIGNED_OUT) + // Inside the orphan grace, so the session is orphaned rather than gone and the reconnect + // resumes it. The branch that does so returns before the new-session path the clear used + // to sit on. + await settle(1) + expect(await assignments.readHostCloseReason(assignmentIdentity)).toBe( + RELAY_HOST_CLOSE_REASON.SIGNED_OUT + ) + + const rebound = new FakeSocket() + await activateRebind(rebound as unknown as WebSocket) + await settle(0) + // Not vacuous: a replaced session would make this a different object and prove nothing. + expect(sessionFor()).toBe(orphaned) + expect(await assignments.readHostCloseReason(assignmentIdentity)).toBeNull() + + // Drop it the way a network death would, so only a stale reason could still name a cause. + rebound.terminate() + await settle() + const phone = await dialPhone(registry) + expect(phone.close).toHaveBeenCalledWith( + RELAY_CLOSE_CODE.HOST_OFFLINE, + 'relay connection rejected' + ) + }) + + // Both writes are unawaited, so either order can reach the row. These pin the rule that decides + // the outcome on timestamps alone, which is why they need no concurrency to be meaningful. + describe('when a close and a proof race to the row', () => { + const PROOF_AT = 2_000 + const CLOSE_AT = 1_000 + + it('drops a close still being written when the host has since proved itself', async () => { + const store = new RelayAssignmentStore(database) + await store.clearHostCloseReason(assignmentIdentity, PROOF_AT) + await store.recordHostCloseReason( + assignmentIdentity, + RELAY_HOST_CLOSE_REASON.SIGNED_OUT, + CLOSE_AT + ) + + expect(await store.readHostCloseReason(assignmentIdentity)).toBeNull() + }) + + it('clears a reason the host recorded before it proved itself', async () => { + const store = new RelayAssignmentStore(database) + await store.recordHostCloseReason( + assignmentIdentity, + RELAY_HOST_CLOSE_REASON.SIGNED_OUT, + CLOSE_AT + ) + await store.clearHostCloseReason(assignmentIdentity, PROOF_AT) + + expect(await store.readHostCloseReason(assignmentIdentity)).toBeNull() + }) + + // The rule has to keep real sign-outs, or it would trade one wrong verdict for another. + it('keeps a close that lands at the same instant as the proof, in either order', async () => { + const store = new RelayAssignmentStore(database) + await store.clearHostCloseReason(assignmentIdentity, PROOF_AT) + await store.recordHostCloseReason( + assignmentIdentity, + RELAY_HOST_CLOSE_REASON.SIGNED_OUT, + PROOF_AT + ) + expect(await store.readHostCloseReason(assignmentIdentity)).toBe( + RELAY_HOST_CLOSE_REASON.SIGNED_OUT + ) + + await store.clearHostCloseReason(assignmentIdentity, PROOF_AT) + expect(await store.readHostCloseReason(assignmentIdentity)).toBe( + RELAY_HOST_CLOSE_REASON.SIGNED_OUT + ) + }) + + it('keeps a close that happened after the proof', async () => { + const store = new RelayAssignmentStore(database) + await store.clearHostCloseReason(assignmentIdentity, CLOSE_AT) + await store.recordHostCloseReason( + assignmentIdentity, + RELAY_HOST_CLOSE_REASON.SIGNED_OUT, + PROOF_AT + ) + + expect(await store.readHostCloseReason(assignmentIdentity)).toBe( + RELAY_HOST_CLOSE_REASON.SIGNED_OUT + ) + }) + }) + // What makes the column addition deferrable: a boot that could not take the lock still serves, // with the verdict it gave before these columns existed. it('serves phones normally on a boot that has not applied the columns', async () => {