From 4c423f9f688227745dea3d58018de46a45dba751 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 20 Jun 2026 05:45:09 -0700 Subject: [PATCH 1/2] Structure relay device persistence errors Co-authored-by: codex --- infra/relay/src/agentActivity/Devices.test.ts | 78 +++++ infra/relay/src/agentActivity/Devices.ts | 302 +++++++++++------- 2 files changed, 261 insertions(+), 119 deletions(-) diff --git a/infra/relay/src/agentActivity/Devices.test.ts b/infra/relay/src/agentActivity/Devices.test.ts index bcc627d8f90..553899da178 100644 --- a/infra/relay/src/agentActivity/Devices.test.ts +++ b/infra/relay/src/agentActivity/Devices.test.ts @@ -223,4 +223,82 @@ describe("Devices", () => { Effect.provide(Devices.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)))), ); }); + + it.effect("identifies the failed device registration stage", () => { + const cause = new Error("push-token claim failed"); + const fakeDb = { + update: () => ({ + set: (values: Record) => ({ + where: () => ("pushToken" in values ? Effect.fail(cause) : Effect.void), + }), + }), + } as unknown as RelayDb.RelayDb["Service"]; + + return Effect.gen(function* () { + const devices = yield* Devices.Devices; + const error = yield* devices.register({ userId: "user-2", registration }).pipe(Effect.flip); + + expect(error).toMatchObject({ + userId: "user-2", + deviceId: "device-1", + stage: "claim-push-token", + }); + expect(error.cause).toBe(cause); + expect(error.message).toBe( + "Failed to persist mobile device registration for user-2/device-1 during claim-push-token.", + ); + }).pipe( + Effect.provide(Devices.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)))), + ); + }); + + it.effect("identifies the failed device unregistration stage", () => { + const cause = new Error("live activity delete failed"); + const fakeDb = { + delete: (table: unknown) => ({ + where: () => (table === relayLiveActivities ? Effect.fail(cause) : Effect.void), + }), + } as unknown as RelayDb.RelayDb["Service"]; + + return Effect.gen(function* () { + const devices = yield* Devices.Devices; + const error = yield* devices + .unregister({ userId: "user-2", deviceId: "device-1" }) + .pipe(Effect.flip); + + expect(error).toMatchObject({ + userId: "user-2", + deviceId: "device-1", + stage: "delete-live-activity", + }); + expect(error.cause).toBe(cause); + expect(error.message).toBe( + "Failed to unregister mobile device user-2/device-1 during delete-live-activity.", + ); + }).pipe( + Effect.provide(Devices.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)))), + ); + }); + + it.effect("attaches the user to device list failures", () => { + const cause = new Error("device list failed"); + const fakeDb = { + select: () => ({ + from: () => ({ + where: () => Effect.fail(cause), + }), + }), + } as unknown as RelayDb.RelayDb["Service"]; + + return Effect.gen(function* () { + const devices = yield* Devices.Devices; + const error = yield* devices.listForUser({ userId: "user-2" }).pipe(Effect.flip); + + expect(error).toMatchObject({ userId: "user-2" }); + expect(error.cause).toBe(cause); + expect(error.message).toBe("Failed to list mobile devices for user-2."); + }).pipe( + Effect.provide(Devices.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)))), + ); + }); }); diff --git a/infra/relay/src/agentActivity/Devices.ts b/infra/relay/src/agentActivity/Devices.ts index 973c430832c..e6cd8914d28 100644 --- a/infra/relay/src/agentActivity/Devices.ts +++ b/infra/relay/src/agentActivity/Devices.ts @@ -15,28 +15,41 @@ import { relayLiveActivities, relayMobileDevices } from "../persistence/schema.t export class DeviceRegistrationPersistenceError extends Schema.TaggedErrorClass()( "DeviceRegistrationPersistenceError", - { cause: Schema.Defect() }, + { + userId: Schema.String, + deviceId: Schema.String, + stage: Schema.Literals(["claim-push-token", "claim-push-to-start-token", "upsert-device"]), + cause: Schema.Defect(), + }, ) { override get message(): string { - return "Failed to persist mobile device registration"; + return `Failed to persist mobile device registration for ${this.userId}/${this.deviceId} during ${this.stage}.`; } } export class DeviceUnregistrationPersistenceError extends Schema.TaggedErrorClass()( "DeviceUnregistrationPersistenceError", - { cause: Schema.Defect() }, + { + userId: Schema.String, + deviceId: Schema.String, + stage: Schema.Literals(["delete-live-activity", "delete-device"]), + cause: Schema.Defect(), + }, ) { override get message(): string { - return "Failed to unregister mobile device"; + return `Failed to unregister mobile device ${this.userId}/${this.deviceId} during ${this.stage}.`; } } export class DeviceListPersistenceError extends Schema.TaggedErrorClass()( "DeviceListPersistenceError", - { cause: Schema.Defect() }, + { + userId: Schema.String, + cause: Schema.Defect(), + }, ) { override get message(): string { - return "Failed to list mobile devices"; + return `Failed to list mobile devices for ${this.userId}.`; } } @@ -61,130 +74,181 @@ export const make = Effect.gen(function* () { const db = yield* RelayDb.RelayDb; return Devices.of({ - register: Effect.fn("relay.devices.register")( - function* (input) { - yield* Effect.annotateCurrentSpan({ - "relay.mobile.device_id": input.registration.deviceId, - }); - const updatedAt = DateTime.formatIso(yield* DateTime.now); - const registration = input.registration; + register: Effect.fn("relay.devices.register")(function* (input) { + yield* Effect.annotateCurrentSpan({ + "relay.mobile.device_id": input.registration.deviceId, + }); + const updatedAt = DateTime.formatIso(yield* DateTime.now); + const registration = input.registration; - yield* Effect.all( - [ - registration.pushToken - ? db - .update(relayMobileDevices) - .set({ pushToken: null, updatedAt }) - .where(eq(relayMobileDevices.pushToken, registration.pushToken)) - : Effect.void, - registration.pushToStartToken - ? db - .update(relayMobileDevices) - .set({ pushToStartToken: null, updatedAt }) - .where(eq(relayMobileDevices.pushToStartToken, registration.pushToStartToken)) - : Effect.void, - ], - { concurrency: 2, discard: true }, - ); + yield* Effect.all( + [ + registration.pushToken + ? db + .update(relayMobileDevices) + .set({ pushToken: null, updatedAt }) + .where(eq(relayMobileDevices.pushToken, registration.pushToken)) + .pipe( + Effect.mapError( + (cause) => + new DeviceRegistrationPersistenceError({ + userId: input.userId, + deviceId: registration.deviceId, + stage: "claim-push-token", + cause, + }), + ), + ) + : Effect.void, + registration.pushToStartToken + ? db + .update(relayMobileDevices) + .set({ pushToStartToken: null, updatedAt }) + .where(eq(relayMobileDevices.pushToStartToken, registration.pushToStartToken)) + .pipe( + Effect.mapError( + (cause) => + new DeviceRegistrationPersistenceError({ + userId: input.userId, + deviceId: registration.deviceId, + stage: "claim-push-to-start-token", + cause, + }), + ), + ) + : Effect.void, + ], + { concurrency: 2, discard: true }, + ); - yield* db - .insert(relayMobileDevices) - .values({ - userId: input.userId, - deviceId: registration.deviceId, - label: registration.label, + yield* db + .insert(relayMobileDevices) + .values({ + userId: input.userId, + deviceId: registration.deviceId, + label: registration.label, + platform: registration.platform, + iosMajorVersion: registration.iosMajorVersion, + appVersion: registration.appVersion ?? null, + pushToken: registration.pushToken ?? null, + pushToStartToken: registration.pushToStartToken ?? null, + preferencesJson: registration.preferences, + createdAt: updatedAt, + updatedAt, + }) + .onConflictDoUpdate({ + target: [relayMobileDevices.userId, relayMobileDevices.deviceId], + set: { platform: registration.platform, + label: registration.label, iosMajorVersion: registration.iosMajorVersion, appVersion: registration.appVersion ?? null, - pushToken: registration.pushToken ?? null, - pushToStartToken: registration.pushToStartToken ?? null, - preferencesJson: registration.preferences, - createdAt: updatedAt, - updatedAt, - }) - .onConflictDoUpdate({ - target: [relayMobileDevices.userId, relayMobileDevices.deviceId], - set: { - platform: registration.platform, - label: registration.label, - iosMajorVersion: registration.iosMajorVersion, - appVersion: registration.appVersion ?? null, - pushToken: sql`coalesce(excluded.push_token, ${relayMobileDevices.pushToken})`, - pushToStartToken: sql`coalesce( + pushToken: sql`coalesce(excluded.push_token, ${relayMobileDevices.pushToken})`, + pushToStartToken: sql`coalesce( excluded.push_to_start_token, ${relayMobileDevices.pushToStartToken} )`, - preferencesJson: registration.preferences, - updatedAt, - }, - }); - }, - Effect.mapError((cause) => new DeviceRegistrationPersistenceError({ cause })), - ), - unregister: Effect.fn("relay.devices.unregister")( - function* (input) { - yield* Effect.annotateCurrentSpan({ - "relay.mobile.device_id": input.deviceId, - }); - yield* Effect.all( - [ - db - .delete(relayLiveActivities) - .where( - and( - eq(relayLiveActivities.userId, input.userId), - eq(relayLiveActivities.deviceId, input.deviceId), - ), + preferencesJson: registration.preferences, + updatedAt, + }, + }) + .pipe( + Effect.mapError( + (cause) => + new DeviceRegistrationPersistenceError({ + userId: input.userId, + deviceId: registration.deviceId, + stage: "upsert-device", + cause, + }), + ), + ); + }), + unregister: Effect.fn("relay.devices.unregister")(function* (input) { + yield* Effect.annotateCurrentSpan({ + "relay.mobile.device_id": input.deviceId, + }); + yield* Effect.all( + [ + db + .delete(relayLiveActivities) + .where( + and( + eq(relayLiveActivities.userId, input.userId), + eq(relayLiveActivities.deviceId, input.deviceId), ), - db - .delete(relayMobileDevices) - .where( - and( - eq(relayMobileDevices.userId, input.userId), - eq(relayMobileDevices.deviceId, input.deviceId), - ), + ) + .pipe( + Effect.mapError( + (cause) => + new DeviceUnregistrationPersistenceError({ + userId: input.userId, + deviceId: input.deviceId, + stage: "delete-live-activity", + cause, + }), ), - ], - { concurrency: 2, discard: true }, + ), + db + .delete(relayMobileDevices) + .where( + and( + eq(relayMobileDevices.userId, input.userId), + eq(relayMobileDevices.deviceId, input.deviceId), + ), + ) + .pipe( + Effect.mapError( + (cause) => + new DeviceUnregistrationPersistenceError({ + userId: input.userId, + deviceId: input.deviceId, + stage: "delete-device", + cause, + }), + ), + ), + ], + { concurrency: 2, discard: true }, + ); + }), + listForUser: Effect.fn("relay.devices.listForUser")(function* (input) { + const rows = yield* db + .select({ + deviceId: relayMobileDevices.deviceId, + label: relayMobileDevices.label, + platform: relayMobileDevices.platform, + iosMajorVersion: relayMobileDevices.iosMajorVersion, + appVersion: relayMobileDevices.appVersion, + preferences: relayMobileDevices.preferencesJson, + updatedAt: relayMobileDevices.updatedAt, + }) + .from(relayMobileDevices) + .where(eq(relayMobileDevices.userId, input.userId)) + .pipe( + Effect.mapError( + (cause) => new DeviceListPersistenceError({ userId: input.userId, cause }), + ), ); - }, - Effect.mapError((cause) => new DeviceUnregistrationPersistenceError({ cause })), - ), - listForUser: Effect.fn("relay.devices.listForUser")( - function* (input) { - const rows = yield* db - .select({ - deviceId: relayMobileDevices.deviceId, - label: relayMobileDevices.label, - platform: relayMobileDevices.platform, - iosMajorVersion: relayMobileDevices.iosMajorVersion, - appVersion: relayMobileDevices.appVersion, - preferences: relayMobileDevices.preferencesJson, - updatedAt: relayMobileDevices.updatedAt, - }) - .from(relayMobileDevices) - .where(eq(relayMobileDevices.userId, input.userId)); - return rows.map((row) => ({ - deviceId: row.deviceId, - label: row.label, - platform: row.platform, - iosMajorVersion: row.iosMajorVersion, - appVersion: row.appVersion, - notifications: { - enabled: row.preferences.notificationsEnabled, - notifyOnApproval: row.preferences.notifyOnApproval, - notifyOnInput: row.preferences.notifyOnInput, - notifyOnCompletion: row.preferences.notifyOnCompletion, - notifyOnFailure: row.preferences.notifyOnFailure, - }, - liveActivities: { - enabled: row.preferences.liveActivitiesEnabled, - }, - updatedAt: row.updatedAt, - })); - }, - Effect.mapError((cause) => new DeviceListPersistenceError({ cause })), - ), + return rows.map((row) => ({ + deviceId: row.deviceId, + label: row.label, + platform: row.platform, + iosMajorVersion: row.iosMajorVersion, + appVersion: row.appVersion, + notifications: { + enabled: row.preferences.notificationsEnabled, + notifyOnApproval: row.preferences.notifyOnApproval, + notifyOnInput: row.preferences.notifyOnInput, + notifyOnCompletion: row.preferences.notifyOnCompletion, + notifyOnFailure: row.preferences.notifyOnFailure, + }, + liveActivities: { + enabled: row.preferences.liveActivitiesEnabled, + }, + updatedAt: row.updatedAt, + })); + }), }); }); From 142db89571352da45abda83b5428fc2aa5504679 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 20 Jun 2026 06:30:12 -0700 Subject: [PATCH 2/2] Make relay device failure stages deterministic Co-authored-by: codex --- infra/relay/src/agentActivity/Devices.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/infra/relay/src/agentActivity/Devices.ts b/infra/relay/src/agentActivity/Devices.ts index e6cd8914d28..86e3564d5be 100644 --- a/infra/relay/src/agentActivity/Devices.ts +++ b/infra/relay/src/agentActivity/Devices.ts @@ -118,7 +118,7 @@ export const make = Effect.gen(function* () { ) : Effect.void, ], - { concurrency: 2, discard: true }, + { discard: true }, ); yield* db @@ -209,7 +209,7 @@ export const make = Effect.gen(function* () { ), ), ], - { concurrency: 2, discard: true }, + { discard: true }, ); }), listForUser: Effect.fn("relay.devices.listForUser")(function* (input) {