From 2082adeb0d8a9edb0023ff85361075c115883ee1 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 24 Oct 2026 08:24:29 -0500 Subject: [PATCH 1/8] fix(relay): retry APNs pushes that fail with 429, 5xx or no answer A signed APNs delivery job that got a 429, a 5xx or no response was marked completed and its handler succeeded, so the queue acked it and a redelivered copy was skipped as a duplicate. The alert was lost. Those jobs now release their claim and fail, so Cloudflare Queues redelivers them under the existing retry policy. Permanent failures and JWT errors still complete as before. Co-Authored-By: Claude Opus 5.5 --- .../src/agentActivity/ApnsDeliveries.test.ts | 147 +++++++++++++++++- .../relay/src/agentActivity/ApnsDeliveries.ts | 68 +++++++- .../src/agentActivity/DeliveryAttempts.ts | 41 ++++- .../agentActivity/MobileRegistrations.test.ts | 1 + 4 files changed, 250 insertions(+), 7 deletions(-) diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index ddabfe557fa51..1f88dc093726e 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -10,7 +10,13 @@ import * as NodeCrypto from "node:crypto"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Redacted from "effect/Redacted"; -import { FetchHttpClient, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http"; +import { + FetchHttpClient, + HttpClient, + HttpClientError, + HttpClientRequest, + HttpClientResponse, +} from "effect/http"; import { makeApnsDeliveryJobPayload, @@ -159,7 +165,7 @@ function layerFor(input: { readonly currentActivityStates?: ReadonlyArray; readonly execute?: ( request: HttpClientRequest.HttpClientRequest, - ) => Effect.Effect; + ) => Effect.Effect; }) { return ApnsDeliveries.layer.pipe( Layer.provide(ApnsClient.layer), @@ -203,6 +209,18 @@ function layerFor(input: { input.attempts.push(attempt); return "claimed"; }), + releaseSourceJob: ({ sourceJobId }) => + Effect.sync(() => { + const index = input.attempts.findIndex( + (row) => + row.sourceJobId === sourceJobId && + row.apnsStatus === undefined && + row.transportError === undefined, + ); + if (index !== -1) { + input.attempts.splice(index, 1); + } + }), completeSourceJob: (completion) => Effect.sync(() => { const attempt = input.attempts.find( @@ -979,6 +997,131 @@ describe("ApnsDeliveries", () => { ); }); + it.effect("fails a signed push job for redelivery when APNs answers 503", () => { + const attempts: Array = []; + const statuses = [503, 200]; + const payload = makeApnsDeliveryJobPayload({ + kind: "push_notification", + userId: target.user_id, + deviceId: target.device_id, + token: "apns-device-token", + aggregate: null, + notification: { + title: "Thread", + body: "Completed: Project", + environmentId: "env", + threadId: "thread", + deepLink: "/", + }, + createdAt: "1970-01-01T00:00:00.000Z", + expiresAt: "1970-01-01T00:10:00.000Z", + jobId: "job-push-unavailable", + }); + const signed = signApnsDeliveryJob({ + secret: config.apnsDeliveryJobSigningSecret, + payload, + }); + const execute = (request: HttpClientRequest.HttpClientRequest) => + Effect.sync(() => { + const status = statuses.shift() ?? 200; + return HttpClientResponse.fromWeb( + request, + status === 200 + ? new Response("", { status }) + : Response.json({ reason: "ServiceUnavailable" }, { status }), + ); + }); + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const failure = yield* deliveries.processSignedJob(signed).pipe(Effect.flip); + + expect(failure).toMatchObject({ + _tag: "ApnsDeliveryJobRetryableFailure", + sourceJobId: "job-push-unavailable", + apnsStatus: 503, + apnsReason: "ServiceUnavailable", + }); + expect(attempts).toEqual([]); + + const redelivered = yield* deliveries.processSignedJob(signed); + + expect(redelivered).toMatchObject({ ok: true, apnsStatus: 200 }); + expect(statuses).toEqual([]); + expect(attempts).toMatchObject([{ sourceJobId: "job-push-unavailable", apnsStatus: 200 }]); + }).pipe( + Effect.provide( + layerFor({ + attempts, + currentTargets: [ + { + ...target, + push_token: "apns-device-token", + }, + ], + config: signingConfig, + execute, + }), + ), + ); + }); + + it.effect("fails a signed Live Activity job for redelivery when APNs is unreachable", () => { + const attempts: Array = []; + const markedDeliveries: Array< + Parameters[0] + > = []; + let executeCount = 0; + const payload = makeApnsDeliveryJobPayload({ + kind: "live_activity_update", + userId: target.user_id, + deviceId: target.device_id, + token: target.activity_push_token ?? "activity-token", + aggregate, + createdAt: "1970-01-01T00:00:00.000Z", + expiresAt: "1970-01-01T00:10:00.000Z", + jobId: "job-update-unreachable", + }); + const signed = signApnsDeliveryJob({ + secret: config.apnsDeliveryJobSigningSecret, + payload, + }); + const execute = (request: HttpClientRequest.HttpClientRequest) => { + executeCount += 1; + return executeCount === 1 + ? Effect.fail( + new HttpClientError.HttpClientError({ + reason: new HttpClientError.TransportError({ + request, + description: "connection reset", + }), + }), + ) + : Effect.succeed(HttpClientResponse.fromWeb(request, new Response("", { status: 200 }))); + }; + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const failure = yield* deliveries.processSignedJob(signed).pipe(Effect.flip); + + expect(failure).toMatchObject({ + _tag: "ApnsDeliveryJobRetryableFailure", + sourceJobId: "job-update-unreachable", + apnsStatus: null, + }); + expect(attempts).toEqual([]); + expect(markedDeliveries).toEqual([]); + + const redelivered = yield* deliveries.processSignedJob(signed); + + expect(redelivered).toMatchObject({ ok: true, apnsStatus: 200 }); + expect(executeCount).toBe(2); + expect(markedDeliveries).toMatchObject([{ kind: "live_activity_update" }]); + }).pipe( + Effect.provide(layerFor({ attempts, markedDeliveries, config: signingConfig, execute })), + ); + }); + it.effect("invalidates dead device push tokens after permanent APNs alert failures", () => { const attempts: Array = []; const invalidatedTokens: Array< diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 6e48ac3663f71..8e0b69343df11 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -133,6 +133,22 @@ export class ApnsDeliveryTransportError extends Schema.TaggedError()( + "ApnsDeliveryJobRetryableFailure", + { + sourceJobId: Schema.String, + kind: RelayDeliveryKindSchema, + apnsStatus: Schema.NullOr(Schema.Number), + apnsReason: Schema.NullOr(Schema.String), + }, +) { + override get message(): string { + return `APNs ${this.kind} delivery job '${this.sourceJobId}' failed and will be retried`; + } +} + +export type ApnsSignedJobError = ApnsDeliveryError | ApnsDeliveryJobRetryableFailure; + const decodeRelayAgentActivityAggregateStateJson = Schema.decodeUnknownOption( Schema.fromJsonString(RelayAgentActivityAggregateStateSchema), ); @@ -337,6 +353,17 @@ function isPermanentApnsTokenFailure(result: Apns.ApnsDeliveryResult): boolean { ); } +// APNs asks senders to retry these later. +function isRetryableApnsStatus(status: number): boolean { + return status === 429 || status >= 500; +} + +type ApnsSendResult = Apns.ApnsDeliveryResult & { readonly retryable: boolean }; + +function classifyApnsResponse(result: Apns.ApnsDeliveryResult): ApnsSendResult { + return { ...result, retryable: !result.ok && isRetryableApnsStatus(result.status) }; +} + function duplicateJobResult(input: { readonly deviceId: string; readonly kind: RelayDeliveryKind; @@ -381,7 +408,7 @@ const recoverApnsDeliveryTransportError = ( readonly sourceJobId: string | null; }, cause: Apns.ApnsError, -): Effect.Effect => { +): Effect.Effect => { const error = new ApnsDeliveryTransportError({ deviceId: input.deviceId, kind: input.kind, @@ -406,6 +433,11 @@ const recoverApnsDeliveryTransportError = ( status: 0, reason: cause.message, apnsId: null, + // A request that got no answer may succeed later; JWT errors are + // configuration problems that a retry cannot fix. + retryable: + cause._tag === "ApnsHttpRequestError" && + (cause.status === null || isRetryableApnsStatus(cause.status)), }), ); }; @@ -519,16 +551,16 @@ export class ApnsDeliveries extends Context.Service< }) => Effect.Effect; readonly sendLiveActivity: ( input: SendLiveActivityDeliveryInput, - ) => Effect.Effect; + ) => Effect.Effect; readonly processSignedJob: ( body: unknown, - ) => Effect.Effect; + ) => Effect.Effect; readonly sendPushNotification: (input: { readonly target: LiveActivityDeliveryTarget; readonly token: string; readonly sourceJobId?: string | null; readonly notification: ApnsNotificationPayload; - }) => Effect.Effect; + }) => Effect.Effect; } >()("t3code-relay/agentActivity/ApnsDeliveries") {} @@ -676,6 +708,22 @@ export const make = Effect.gen(function* () { ); }); + // The queue redelivers a job only when its handler fails. Releasing the claim + // lets the redelivered copy send instead of waiting out the in-flight lease. + const failForRetry = Effect.fnUntraced(function* (input: { + readonly sourceJobId: string; + readonly kind: RelayDeliveryKind; + readonly result: ApnsSendResult; + }) { + yield* attempts.releaseSourceJob({ sourceJobId: input.sourceJobId }); + return yield* new ApnsDeliveryJobRetryableFailure({ + sourceJobId: input.sourceJobId, + kind: input.kind, + apnsStatus: input.result.status === 0 ? null : input.result.status, + apnsReason: input.result.reason ?? null, + }); + }); + const sendLiveActivity: ApnsDeliveries["Service"]["sendLiveActivity"] = Effect.fn( "relay.apns_deliveries.send_live_activity", )(function* (input) { @@ -813,12 +861,16 @@ export const make = Effect.gen(function* () { issuedAtUnixSeconds: epochSeconds, }) .pipe( + Effect.map(classifyApnsResponse), Effect.catchTags({ ApnsJwtEncodingError: recoverTransportError, ApnsJwtSigningError: recoverTransportError, ApnsHttpRequestError: recoverTransportError, }), ); + if (input.sourceJobId && result.retryable) { + return yield* failForRetry({ sourceJobId: input.sourceJobId, kind: input.kind, result }); + } if (result.ok) { yield* liveActivities.markDelivery({ userId: input.target.user_id, @@ -981,12 +1033,20 @@ export const make = Effect.gen(function* () { issuedAtUnixSeconds: epochSeconds, }) .pipe( + Effect.map(classifyApnsResponse), Effect.catchTags({ ApnsJwtEncodingError: recoverTransportError, ApnsJwtSigningError: recoverTransportError, ApnsHttpRequestError: recoverTransportError, }), ); + if (input.sourceJobId && result.retryable) { + return yield* failForRetry({ + sourceJobId: input.sourceJobId, + kind: "push_notification", + result, + }); + } if (isPermanentApnsTokenFailure(result)) { yield* liveActivities.invalidateDeliveryToken({ userId: input.target.user_id, diff --git a/infra/relay/src/agentActivity/DeliveryAttempts.ts b/infra/relay/src/agentActivity/DeliveryAttempts.ts index 3e938617a5bed..ed2374ad81eaa 100644 --- a/infra/relay/src/agentActivity/DeliveryAttempts.ts +++ b/infra/relay/src/agentActivity/DeliveryAttempts.ts @@ -13,7 +13,12 @@ import { relayDeliveryAttempts } from "../persistence/schema.ts"; export class DeliveryAttemptRecordPersistenceError extends Schema.TaggedError()( "DeliveryAttemptRecordPersistenceError", { - operation: Schema.Literals(["record", "claim-source-job", "complete-source-job"]), + operation: Schema.Literals([ + "record", + "claim-source-job", + "release-source-job", + "complete-source-job", + ]), sourceJobId: Schema.NullOr(Schema.String), userId: Schema.NullOr(Schema.String), environmentId: Schema.NullOr(Schema.String), @@ -61,6 +66,9 @@ export class DeliveryAttempts extends Context.Service< readonly claimSourceJob: ( input: DeliveryAttemptInput & { readonly sourceJobId: string }, ) => Effect.Effect; + readonly releaseSourceJob: (input: { + readonly sourceJobId: string; + }) => Effect.Effect; readonly completeSourceJob: ( input: DeliveryAttemptCompletionInput, ) => Effect.Effect; @@ -217,6 +225,37 @@ export const make = Effect.gen(function* () { ), ); }), + // Drops an unfinished claim so a redelivered copy of the job can claim it + // again before the lease runs out. A completed row stays as the dedupe record. + releaseSourceJob: Effect.fn("relay.delivery_attempts.release_source_job")(function* (input) { + yield* Effect.annotateCurrentSpan({ "relay.delivery.job_id": input.sourceJobId }); + yield* db + .delete(relayDeliveryAttempts) + .where( + and( + eq(relayDeliveryAttempts.sourceJobId, input.sourceJobId), + isNull(relayDeliveryAttempts.apnsStatus), + isNull(relayDeliveryAttempts.apnsReason), + isNull(relayDeliveryAttempts.apnsId), + isNull(relayDeliveryAttempts.transportError), + ), + ) + .pipe( + Effect.mapError( + (cause) => + new DeliveryAttemptRecordPersistenceError({ + operation: "release-source-job", + sourceJobId: input.sourceJobId, + userId: null, + environmentId: null, + threadId: null, + deviceId: null, + kind: null, + cause, + }), + ), + ); + }), completeSourceJob: Effect.fn("relay.delivery_attempts.complete_source_job")(function* (input) { yield* Effect.annotateCurrentSpan({ "relay.delivery.job_id": input.sourceJobId }); const completedAt = DateTime.formatIso(yield* DateTime.now); diff --git a/infra/relay/src/agentActivity/MobileRegistrations.test.ts b/infra/relay/src/agentActivity/MobileRegistrations.test.ts index 38840cf02b5b6..ecc7526502531 100644 --- a/infra/relay/src/agentActivity/MobileRegistrations.test.ts +++ b/infra/relay/src/agentActivity/MobileRegistrations.test.ts @@ -130,6 +130,7 @@ function makeDeliveryAttempts( return { record: () => Effect.void, claimSourceJob: () => Effect.succeed("claimed"), + releaseSourceJob: () => Effect.void, completeSourceJob: () => Effect.void, ...overrides, }; From f86f1a93f58e674a5bd14e9c16e9fdfc6088195b Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 24 Oct 2026 08:53:50 -0500 Subject: [PATCH 2/8] fix(relay): retry a failed APNs job without retrying its batch The APNs queue consumer let a job's failure end the batch stream, so Alchemy retried every message in the batch. A job that APNs throttled sent the jobs behind it back to the queue unsent, and each of those spent one of its own retries. The consumer now acks or retries each message on its own, as the FCM consumer does. Co-Authored-By: Claude Opus 5.5 --- .../src/agentActivity/ApnsDeliveries.test.ts | 88 +++++++++++++++++++ .../ApnsDeliveryQueueConsumer.ts | 25 ++++++ infra/relay/src/worker.ts | 8 +- 3 files changed, 115 insertions(+), 6 deletions(-) create mode 100644 infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index 1f88dc093726e..81d6fe614dca7 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -4,12 +4,15 @@ import type { RelayAgentActivityState, } from "@t3tools/contracts/relay"; import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto"; +import type * as Cloudflare from "alchemy/Cloudflare"; import { describe, expect, it } from "@effect/vitest"; // @effect-diagnostics-next-line nodeBuiltinImport:off -- Effect's Crypto has no generateKeyPairSync. import * as NodeCrypto from "node:crypto"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Redacted from "effect/Redacted"; +import * as Stream from "effect/Stream"; import { FetchHttpClient, HttpClient, @@ -29,6 +32,7 @@ import * as RelayConfiguration from "../Config.ts"; import * as ApnsDeliveryQueue from "./ApnsDeliveryQueue.ts"; import * as AgentActivityRows from "./AgentActivityRows.ts"; import * as ApnsDeliveries from "./ApnsDeliveries.ts"; +import * as ApnsDeliveryQueueConsumer from "./ApnsDeliveryQueueConsumer.ts"; import * as ApnsClient from "./ApnsClient.ts"; import * as ApnsProviderTokens from "./ApnsProviderTokens.ts"; @@ -1280,6 +1284,90 @@ describe("ApnsDeliveries", () => { }); }); +describe("APNs queue message isolation", () => { + it.effect("retries only the message APNs throttled and delivers the rest of its batch", () => { + const attempts: Array = []; + const outcomes = new Map(); + const requestedTokens: Array = []; + const signedPush = (deviceId: string, token: string) => + signApnsDeliveryJob({ + secret: config.apnsDeliveryJobSigningSecret, + payload: makeApnsDeliveryJobPayload({ + kind: "push_notification", + userId: target.user_id, + deviceId, + token, + aggregate: null, + notification: { + title: "Thread", + body: "Completed: Project", + environmentId: "env", + threadId: "thread", + deepLink: "/", + }, + createdAt: "1970-01-01T00:00:00.000Z", + expiresAt: "1970-01-01T00:10:00.000Z", + jobId: `job-${deviceId}`, + }), + }); + const message = (id: string, body: unknown): Cloudflare.Queues.Message => ({ + id, + body, + timestamp: DateTime.toDateUtc(DateTime.makeUnsafe(0)), + attempts: 1, + ack: () => { + if (!outcomes.has(id)) outcomes.set(id, "ack"); + }, + retry: () => { + if (!outcomes.has(id)) outcomes.set(id, "retry"); + }, + }); + const batch = [ + message("throttled", signedPush("device-1", "throttled-token")), + message("healthy", signedPush("device-2", "healthy-token")), + ]; + const execute = (request: HttpClientRequest.HttpClientRequest) => + Effect.sync(() => { + const token = request.url.split("/").at(-1) ?? ""; + requestedTokens.push(token); + return HttpClientResponse.fromWeb( + request, + token === "throttled-token" + ? Response.json({ reason: "TooManyRequests" }, { status: 429 }) + : new Response("", { status: 200 }), + ); + }); + + return Effect.gen(function* () { + yield* Stream.fromIterable(batch).pipe( + Stream.runForEach(ApnsDeliveryQueueConsumer.processMessage), + ); + // Alchemy acknowledges the batch after a successful stream. Cloudflare + // ignores those acknowledgements for messages explicitly retried earlier. + for (const item of batch) item.ack(); + + expect([...outcomes]).toEqual([ + ["throttled", "retry"], + ["healthy", "ack"], + ]); + expect(requestedTokens).toEqual(["throttled-token", "healthy-token"]); + expect(attempts).toMatchObject([{ sourceJobId: "job-device-2", apnsStatus: 200 }]); + }).pipe( + Effect.provide( + layerFor({ + attempts, + currentTargets: [ + { ...target, device_id: "device-1", push_token: "throttled-token" }, + { ...target, device_id: "device-2", push_token: "healthy-token" }, + ], + config: signingConfig, + execute, + }), + ), + ); + }); +}); + describe("live activity alert decisions", () => { const preferences = { liveActivitiesEnabled: true, diff --git a/infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts b/infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts new file mode 100644 index 0000000000000..a11109cae956e --- /dev/null +++ b/infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts @@ -0,0 +1,25 @@ +import type * as Cloudflare from "alchemy/Cloudflare"; +import * as Effect from "effect/Effect"; + +import * as ApnsDeliveries from "./ApnsDeliveries.ts"; + +export const processMessage = Effect.fn("relay.apn_delivery_queue.process_message")(function* ( + message: Cloudflare.Queues.Message, +) { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + yield* deliveries.processSignedJob(message.body).pipe( + Effect.matchEffect({ + onFailure: (error) => + Effect.gen(function* () { + // Decide this message's outcome before Alchemy acknowledges the batch. + // Cloudflare keeps the first ack/retry decision for each message. + message.retry(); + yield* Effect.logWarning("APNs queue delivery failed; retrying message", { + messageId: message.id, + errorTag: error._tag, + }); + }), + onSuccess: () => Effect.sync(() => message.ack()), + }), + ); +}); diff --git a/infra/relay/src/worker.ts b/infra/relay/src/worker.ts index a910a30649a22..a83edc3a774a2 100644 --- a/infra/relay/src/worker.ts +++ b/infra/relay/src/worker.ts @@ -57,6 +57,7 @@ import * as ApnsClient from "./agentActivity/ApnsClient.ts"; import * as ApnsProviderTokens from "./agentActivity/ApnsProviderTokens.ts"; import * as ApnsDeliveryQueue from "./agentActivity/ApnsDeliveryQueue.ts"; import * as ApnsDeliveries from "./agentActivity/ApnsDeliveries.ts"; +import * as ApnsDeliveryQueueConsumer from "./agentActivity/ApnsDeliveryQueueConsumer.ts"; import * as CloudflareApiPause from "./environments/CloudflareApiPause.ts"; import * as EnvironmentConnector from "./environments/EnvironmentConnector.ts"; import * as EnvironmentLinker from "./environments/EnvironmentLinker.ts"; @@ -360,12 +361,7 @@ export const layer = Api.make( (stream) => stream.pipe( Stream.withSpan("relay.apn_delivery_queue.process_batch"), - Stream.runForEach((message) => - ApnsDeliveries.ApnsDeliveries.pipe( - Effect.flatMap((deliveries) => deliveries.processSignedJob(message.body)), - Effect.withSpan("relay.apn_delivery_queue.process_message"), - ), - ), + Stream.runForEach(ApnsDeliveryQueueConsumer.processMessage), Effect.provide(layerRuntime), ), ); From 66dfe2372725041b926cf7035d3976554e5378b2 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 15:36:02 +0000 Subject: [PATCH 3/8] fix(relay): bound retry errors and verify APNs redelivery --- .../src/agentActivity/ApnsDeliveries.test.ts | 129 +++++++++--------- .../relay/src/agentActivity/ApnsDeliveries.ts | 9 +- .../ApnsDeliveryQueueConsumer.ts | 19 ++- .../agentActivity/DeliveryAttempts.test.ts | 25 ++++ 4 files changed, 102 insertions(+), 80 deletions(-) diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index 81d6fe614dca7..3b2bdbf8098e6 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -1001,74 +1001,77 @@ describe("ApnsDeliveries", () => { ); }); - it.effect("fails a signed push job for redelivery when APNs answers 503", () => { - const attempts: Array = []; - const statuses = [503, 200]; - const payload = makeApnsDeliveryJobPayload({ - kind: "push_notification", - userId: target.user_id, - deviceId: target.device_id, - token: "apns-device-token", - aggregate: null, - notification: { - title: "Thread", - body: "Completed: Project", - environmentId: "env", - threadId: "thread", - deepLink: "/", - }, - createdAt: "1970-01-01T00:00:00.000Z", - expiresAt: "1970-01-01T00:10:00.000Z", - jobId: "job-push-unavailable", - }); - const signed = signApnsDeliveryJob({ - secret: config.apnsDeliveryJobSigningSecret, - payload, - }); - const execute = (request: HttpClientRequest.HttpClientRequest) => - Effect.sync(() => { - const status = statuses.shift() ?? 200; - return HttpClientResponse.fromWeb( - request, - status === 200 - ? new Response("", { status }) - : Response.json({ reason: "ServiceUnavailable" }, { status }), - ); + it.effect.each([429, 500, 503, 599])( + "fails a signed push job for redelivery when APNs answers %i", + (status) => { + const attempts: Array = []; + const statuses = [status, 200]; + const payload = makeApnsDeliveryJobPayload({ + kind: "push_notification", + userId: target.user_id, + deviceId: target.device_id, + token: "apns-device-token", + aggregate: null, + notification: { + title: "Thread", + body: "Completed: Project", + environmentId: "env", + threadId: "thread", + deepLink: "/", + }, + createdAt: "1970-01-01T00:00:00.000Z", + expiresAt: "1970-01-01T00:10:00.000Z", + jobId: "job-push-unavailable", }); + const signed = signApnsDeliveryJob({ + secret: config.apnsDeliveryJobSigningSecret, + payload, + }); + const execute = (request: HttpClientRequest.HttpClientRequest) => + Effect.sync(() => { + const status = statuses.shift() ?? 200; + return HttpClientResponse.fromWeb( + request, + status === 200 + ? new Response("", { status }) + : Response.json({ reason: "ServiceUnavailable" }, { status }), + ); + }); - return Effect.gen(function* () { - const deliveries = yield* ApnsDeliveries.ApnsDeliveries; - const failure = yield* deliveries.processSignedJob(signed).pipe(Effect.flip); + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const failure = yield* deliveries.processSignedJob(signed).pipe(Effect.flip); - expect(failure).toMatchObject({ - _tag: "ApnsDeliveryJobRetryableFailure", - sourceJobId: "job-push-unavailable", - apnsStatus: 503, - apnsReason: "ServiceUnavailable", - }); - expect(attempts).toEqual([]); + expect(failure).toMatchObject({ + _tag: "ApnsDeliveryJobRetryableFailure", + sourceJobId: "job-push-unavailable", + apnsStatus: status, + }); + expect(failure).not.toHaveProperty("apnsReason"); + expect(attempts).toEqual([]); - const redelivered = yield* deliveries.processSignedJob(signed); + const redelivered = yield* deliveries.processSignedJob(signed); - expect(redelivered).toMatchObject({ ok: true, apnsStatus: 200 }); - expect(statuses).toEqual([]); - expect(attempts).toMatchObject([{ sourceJobId: "job-push-unavailable", apnsStatus: 200 }]); - }).pipe( - Effect.provide( - layerFor({ - attempts, - currentTargets: [ - { - ...target, - push_token: "apns-device-token", - }, - ], - config: signingConfig, - execute, - }), - ), - ); - }); + expect(redelivered).toMatchObject({ ok: true, apnsStatus: 200 }); + expect(statuses).toEqual([]); + expect(attempts).toMatchObject([{ sourceJobId: "job-push-unavailable", apnsStatus: 200 }]); + }).pipe( + Effect.provide( + layerFor({ + attempts, + currentTargets: [ + { + ...target, + push_token: "apns-device-token", + }, + ], + config: signingConfig, + execute, + }), + ), + ); + }, + ); it.effect("fails a signed Live Activity job for redelivery when APNs is unreachable", () => { const attempts: Array = []; diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 93aaad84c387a..3da77a91e95ff 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -139,7 +139,6 @@ export class ApnsDeliveryJobRetryableFailure extends Schema.TaggedError= 500; + return status === 429 || (status >= 500 && status < 600); } type ApnsSendResult = Apns.ApnsDeliveryResult & { readonly retryable: boolean }; @@ -691,9 +689,7 @@ export const make = Effect.gen(function* () { ); }); - // The queue redelivers a job only when its handler fails. Releasing the claim - // lets the redelivered copy send instead of waiting out the in-flight lease. - const failForRetry = Effect.fnUntraced(function* (input: { + const failForRetry = Effect.fn("relay.apns_deliveries.fail_for_retry")(function* (input: { readonly sourceJobId: string; readonly kind: RelayDeliveryKind; readonly result: ApnsSendResult; @@ -703,7 +699,6 @@ export const make = Effect.gen(function* () { sourceJobId: input.sourceJobId, kind: input.kind, apnsStatus: input.result.status === 0 ? null : input.result.status, - apnsReason: input.result.reason ?? null, }); }); diff --git a/infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts b/infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts index a11109cae956e..7a6f27b4b6cc9 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts @@ -9,16 +9,15 @@ export const processMessage = Effect.fn("relay.apn_delivery_queue.process_messag const deliveries = yield* ApnsDeliveries.ApnsDeliveries; yield* deliveries.processSignedJob(message.body).pipe( Effect.matchEffect({ - onFailure: (error) => - Effect.gen(function* () { - // Decide this message's outcome before Alchemy acknowledges the batch. - // Cloudflare keeps the first ack/retry decision for each message. - message.retry(); - yield* Effect.logWarning("APNs queue delivery failed; retrying message", { - messageId: message.id, - errorTag: error._tag, - }); - }), + // The enclosing process_message span covers this handler. + onFailure: Effect.fnUntraced(function* (error) { + // Cloudflare keeps the first ack/retry decision, before Alchemy acknowledges the batch. + message.retry(); + yield* Effect.logWarning("APNs queue delivery failed; retrying message", { + messageId: message.id, + errorTag: error._tag, + }); + }), onSuccess: () => Effect.sync(() => message.ack()), }), ); diff --git a/infra/relay/src/agentActivity/DeliveryAttempts.test.ts b/infra/relay/src/agentActivity/DeliveryAttempts.test.ts index 056b906bdb436..936bb0bc953e0 100644 --- a/infra/relay/src/agentActivity/DeliveryAttempts.test.ts +++ b/infra/relay/src/agentActivity/DeliveryAttempts.test.ts @@ -7,6 +7,31 @@ import * as RelayDb from "../db.ts"; import * as DeliveryAttempts from "./DeliveryAttempts.ts"; describe("DeliveryAttempts", () => { + it.effect("reports claim-release persistence failures without completing the job", () => { + const cause = new Error("database unavailable"); + const fakeDb = { + delete: () => ({ where: () => Effect.fail(cause) }), + } as unknown as RelayDb.RelayDb["Service"]; + + return Effect.gen(function* () { + const attempts = yield* DeliveryAttempts.DeliveryAttempts; + const error = yield* attempts.releaseSourceJob({ sourceJobId: "job-1" }).pipe(Effect.flip); + expect(error).toMatchObject({ + _tag: "DeliveryAttemptRecordPersistenceError", + operation: "release-source-job", + sourceJobId: "job-1", + cause, + }); + }).pipe( + Effect.provide( + DeliveryAttempts.layer.pipe( + Layer.provide(NodeCryptoLayer.layer), + Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)), + ), + ), + ); + }); + it.effect("reports completed source jobs when the durable claim already exists", () => { const fakeDb = { insert: () => ({ From fbbb864cc1f4a5da7c4f2655c4f382db8204a3b0 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 08:10:36 +0000 Subject: [PATCH 4/8] fix(relay): align APNs consumer with cloudflare package migration --- infra/relay/src/agentActivity/ApnsDeliveries.test.ts | 2 +- infra/relay/src/agentActivity/ApnsDeliveries.ts | 2 -- infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts | 2 +- 3 files changed, 2 insertions(+), 4 deletions(-) diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index 2e0e702d6da41..87f1b0c564caa 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -5,7 +5,7 @@ import type { RelayAgentActivityState, } from "@t3tools/contracts/relay"; import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto"; -import type * as Cloudflare from "alchemy/Cloudflare"; +import type * as Cloudflare from "@alchemy.run/cloudflare"; import { describe, expect, it } from "@effect/vitest"; // @effect-diagnostics-next-line nodeBuiltinImport:off -- Effect's Crypto has no generateKeyPairSync. import * as NodeCrypto from "node:crypto"; diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 8392b3126b5e5..24d0882cadcd9 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -123,8 +123,6 @@ export class ApnsDeliveryTransportError extends Schema.TaggedError()( "ApnsDeliveryJobRetryableFailure", { diff --git a/infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts b/infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts index 7a6f27b4b6cc9..42726982afca0 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveryQueueConsumer.ts @@ -1,4 +1,4 @@ -import type * as Cloudflare from "alchemy/Cloudflare"; +import type * as Cloudflare from "@alchemy.run/cloudflare"; import * as Effect from "effect/Effect"; import * as ApnsDeliveries from "./ApnsDeliveries.ts"; From d933c176fefd567771340971d38540386d929521 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 8 Nov 2026 09:41:58 +0000 Subject: [PATCH 5/8] fix(relay): retry a failed APNs claim release before redelivery If releaseSourceJob fails, the claim stays in_flight for its 10-minute lease. The signed job expires within that same window, so every queue redelivery is rejected and the alert ends in the dead-letter queue. failForRetry now retries the release a few times with a short backoff. Co-Authored-By: Claude Opus 5.5 --- .../src/agentActivity/ApnsDeliveries.test.ts | 87 ++++++++++++++++++- .../relay/src/agentActivity/ApnsDeliveries.ts | 7 +- 2 files changed, 92 insertions(+), 2 deletions(-) diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index 87f1b0c564caa..cf21b0e972ab9 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -11,9 +11,11 @@ import { describe, expect, it } from "@effect/vitest"; import * as NodeCrypto from "node:crypto"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Redacted from "effect/Redacted"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { FetchHttpClient, HttpClient, @@ -151,6 +153,7 @@ const target: LiveActivities.TargetRow = { function layerFor(input: { readonly attempts: Array; readonly sourceJobClaims?: ReadonlyMap; + readonly releaseFailures?: { remaining: number }; readonly queuedJobs?: Array; readonly markedDeliveries?: Array< Parameters[0] @@ -206,7 +209,22 @@ function layerFor(input: { return "claimed"; }), releaseSourceJob: ({ sourceJobId }) => - Effect.sync(() => { + Effect.suspend(() => { + if (input.releaseFailures && input.releaseFailures.remaining > 0) { + input.releaseFailures.remaining -= 1; + return Effect.fail( + new DeliveryAttempts.DeliveryAttemptRecordPersistenceError({ + operation: "release-source-job", + sourceJobId, + userId: null, + environmentId: null, + threadId: null, + deviceId: null, + kind: null, + cause: new Error("connection reset"), + }), + ); + } const index = input.attempts.findIndex( (row) => row.sourceJobId === sourceJobId && @@ -216,6 +234,7 @@ function layerFor(input: { if (index !== -1) { input.attempts.splice(index, 1); } + return Effect.void; }), completeSourceJob: (completion) => Effect.sync(() => { @@ -993,6 +1012,72 @@ describe("ApnsDeliveries", () => { }, ); + it.effect("releases the claim of a retryable job after a failed release attempt", () => { + const attempts: Array = []; + const releaseFailures = { remaining: 1 }; + const statuses = [503, 200]; + const payload = makeApnsDeliveryJobPayload({ + kind: "push_notification", + userId: target.user_id, + deviceId: target.device_id, + token: "apns-device-token", + aggregate: null, + notification: { + title: "Thread", + body: "Completed: Project", + environmentId: "env", + threadId: "thread", + deepLink: "/", + }, + createdAt: "1970-01-01T00:00:00.000Z", + expiresAt: "1970-01-01T00:10:00.000Z", + jobId: "job-push-release-retry", + }); + const signed = signApnsDeliveryJob({ + secret: config.apnsDeliveryJobSigningSecret, + payload, + }); + const execute = (request: HttpClientRequest.HttpClientRequest) => + Effect.sync(() => { + const status = statuses.shift() ?? 200; + return HttpClientResponse.fromWeb( + request, + status === 200 + ? new Response("", { status }) + : Response.json({ reason: "ServiceUnavailable" }, { status }), + ); + }); + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const fiber = yield* deliveries.processSignedJob(signed).pipe(Effect.flip, Effect.forkChild); + yield* TestClock.adjust("1 second"); + const failure = yield* Fiber.join(fiber); + + expect(failure).toMatchObject({ + _tag: "ApnsDeliveryJobRetryableFailure", + sourceJobId: "job-push-release-retry", + }); + expect(releaseFailures.remaining).toBe(0); + expect(attempts).toEqual([]); + + const redelivered = yield* deliveries.processSignedJob(signed); + + expect(redelivered).toMatchObject({ ok: true, apnsStatus: 200 }); + expect(attempts).toMatchObject([{ sourceJobId: "job-push-release-retry", apnsStatus: 200 }]); + }).pipe( + Effect.provide( + layerFor({ + attempts, + releaseFailures, + currentTargets: [{ ...target, push_token: "apns-device-token" }], + config: signingConfig, + execute, + }), + ), + ); + }); + it.effect("fails a signed Live Activity job for redelivery when APNs is unreachable", () => { const attempts: Array = []; const markedDeliveries: Array< diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 24d0882cadcd9..05ebbdde96b62 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -15,6 +15,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Redacted from "effect/Redacted"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import { @@ -661,7 +662,11 @@ const make = Effect.gen(function* () { readonly kind: RelayDeliveryKind; readonly result: ApnsSendResult; }) { - yield* attempts.releaseSourceJob({ sourceJobId: input.sourceJobId }); + // A claim left in place blocks redelivery until its lease ends, and the signed + // job expires by then, so a failed release would drop the alert. + yield* attempts + .releaseSourceJob({ sourceJobId: input.sourceJobId }) + .pipe(Effect.retry({ times: 3, schedule: Schedule.exponential("100 millis") })); return yield* new ApnsDeliveryJobRetryableFailure({ sourceJobId: input.sourceJobId, kind: input.kind, From e6dedca26574675abd87144c8a5fc41fa6410336 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 8 Nov 2026 09:44:57 +0000 Subject: [PATCH 6/8] test(relay): cover the release retry as a row of the APNs status table Co-Authored-By: Claude Opus 5.5 --- .../src/agentActivity/ApnsDeliveries.test.ts | 92 ++++--------------- 1 file changed, 19 insertions(+), 73 deletions(-) diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index cf21b0e972ab9..845088fde7341 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -153,7 +153,7 @@ const target: LiveActivities.TargetRow = { function layerFor(input: { readonly attempts: Array; readonly sourceJobClaims?: ReadonlyMap; - readonly releaseFailures?: { remaining: number }; + readonly releaseFailures?: number; readonly queuedJobs?: Array; readonly markedDeliveries?: Array< Parameters[0] @@ -169,6 +169,7 @@ function layerFor(input: { request: HttpClientRequest.HttpClientRequest, ) => Effect.Effect; }) { + let releaseFailuresLeft = input.releaseFailures ?? 0; return ApnsDeliveries.layer.pipe( Layer.provide(ApnsClient.layer), Layer.provide(ApnsProviderTokens.layer), @@ -210,8 +211,8 @@ function layerFor(input: { }), releaseSourceJob: ({ sourceJobId }) => Effect.suspend(() => { - if (input.releaseFailures && input.releaseFailures.remaining > 0) { - input.releaseFailures.remaining -= 1; + if (releaseFailuresLeft > 0) { + releaseFailuresLeft -= 1; return Effect.fail( new DeliveryAttempts.DeliveryAttemptRecordPersistenceError({ operation: "release-source-job", @@ -940,9 +941,15 @@ describe("ApnsDeliveries", () => { ); }); - it.effect.each([429, 500, 503, 599])( - "fails a signed push job for redelivery when APNs answers %i", - (status) => { + it.effect.each([ + { status: 429, releaseFailures: 0 }, + { status: 500, releaseFailures: 0 }, + { status: 503, releaseFailures: 0 }, + { status: 599, releaseFailures: 0 }, + { status: 503, releaseFailures: 1 }, + ])( + "fails a signed push job for redelivery when APNs answers $status (release failures: $releaseFailures)", + ({ status, releaseFailures }) => { const attempts: Array = []; const statuses = [status, 200]; const payload = makeApnsDeliveryJobPayload({ @@ -979,7 +986,11 @@ describe("ApnsDeliveries", () => { return Effect.gen(function* () { const deliveries = yield* ApnsDeliveries.ApnsDeliveries; - const failure = yield* deliveries.processSignedJob(signed).pipe(Effect.flip); + const fiber = yield* deliveries + .processSignedJob(signed) + .pipe(Effect.flip, Effect.forkChild); + yield* TestClock.adjust("1 second"); + const failure = yield* Fiber.join(fiber); expect(failure).toMatchObject({ _tag: "ApnsDeliveryJobRetryableFailure", @@ -998,6 +1009,7 @@ describe("ApnsDeliveries", () => { Effect.provide( layerFor({ attempts, + releaseFailures, currentTargets: [ { ...target, @@ -1012,72 +1024,6 @@ describe("ApnsDeliveries", () => { }, ); - it.effect("releases the claim of a retryable job after a failed release attempt", () => { - const attempts: Array = []; - const releaseFailures = { remaining: 1 }; - const statuses = [503, 200]; - const payload = makeApnsDeliveryJobPayload({ - kind: "push_notification", - userId: target.user_id, - deviceId: target.device_id, - token: "apns-device-token", - aggregate: null, - notification: { - title: "Thread", - body: "Completed: Project", - environmentId: "env", - threadId: "thread", - deepLink: "/", - }, - createdAt: "1970-01-01T00:00:00.000Z", - expiresAt: "1970-01-01T00:10:00.000Z", - jobId: "job-push-release-retry", - }); - const signed = signApnsDeliveryJob({ - secret: config.apnsDeliveryJobSigningSecret, - payload, - }); - const execute = (request: HttpClientRequest.HttpClientRequest) => - Effect.sync(() => { - const status = statuses.shift() ?? 200; - return HttpClientResponse.fromWeb( - request, - status === 200 - ? new Response("", { status }) - : Response.json({ reason: "ServiceUnavailable" }, { status }), - ); - }); - - return Effect.gen(function* () { - const deliveries = yield* ApnsDeliveries.ApnsDeliveries; - const fiber = yield* deliveries.processSignedJob(signed).pipe(Effect.flip, Effect.forkChild); - yield* TestClock.adjust("1 second"); - const failure = yield* Fiber.join(fiber); - - expect(failure).toMatchObject({ - _tag: "ApnsDeliveryJobRetryableFailure", - sourceJobId: "job-push-release-retry", - }); - expect(releaseFailures.remaining).toBe(0); - expect(attempts).toEqual([]); - - const redelivered = yield* deliveries.processSignedJob(signed); - - expect(redelivered).toMatchObject({ ok: true, apnsStatus: 200 }); - expect(attempts).toMatchObject([{ sourceJobId: "job-push-release-retry", apnsStatus: 200 }]); - }).pipe( - Effect.provide( - layerFor({ - attempts, - releaseFailures, - currentTargets: [{ ...target, push_token: "apns-device-token" }], - config: signingConfig, - execute, - }), - ), - ); - }); - it.effect("fails a signed Live Activity job for redelivery when APNs is unreachable", () => { const attempts: Array = []; const markedDeliveries: Array< From d006cd1c9fa004b437d8249becdb87bd87d19575 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 8 Nov 2026 09:57:57 +0000 Subject: [PATCH 7/8] test(relay): pin the persistence error when every APNs claim release fails Co-Authored-By: Claude Opus 5.5 --- .../src/agentActivity/ApnsDeliveries.test.ts | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index 845088fde7341..e9cc51613d4b4 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -1024,6 +1024,66 @@ describe("ApnsDeliveries", () => { }, ); + it.effect("fails with the persistence error when every claim release attempt fails", () => { + const attempts: Array = []; + let executeCount = 0; + const payload = makeApnsDeliveryJobPayload({ + kind: "push_notification", + userId: target.user_id, + deviceId: target.device_id, + token: "apns-device-token", + aggregate: null, + notification: { + title: "Thread", + body: "Completed: Project", + environmentId: "env", + threadId: "thread", + deepLink: "/", + }, + createdAt: "1970-01-01T00:00:00.000Z", + expiresAt: "1970-01-01T00:10:00.000Z", + jobId: "job-push-release-exhausted", + }); + const signed = signApnsDeliveryJob({ + secret: config.apnsDeliveryJobSigningSecret, + payload, + }); + const execute = (request: HttpClientRequest.HttpClientRequest) => + Effect.sync(() => { + executeCount += 1; + return HttpClientResponse.fromWeb( + request, + Response.json({ reason: "ServiceUnavailable" }, { status: 503 }), + ); + }); + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const fiber = yield* deliveries.processSignedJob(signed).pipe(Effect.flip, Effect.forkChild); + yield* TestClock.adjust("1 second"); + const failure = yield* Fiber.join(fiber); + + expect(failure).toMatchObject({ + _tag: "DeliveryAttemptRecordPersistenceError", + operation: "release-source-job", + sourceJobId: "job-push-release-exhausted", + }); + expect(executeCount).toBe(1); + expect(attempts).toMatchObject([{ sourceJobId: "job-push-release-exhausted" }]); + expect(attempts[0]).not.toHaveProperty("apnsStatus"); + }).pipe( + Effect.provide( + layerFor({ + attempts, + releaseFailures: 4, + currentTargets: [{ ...target, push_token: "apns-device-token" }], + config: signingConfig, + execute, + }), + ), + ); + }); + it.effect("fails a signed Live Activity job for redelivery when APNs is unreachable", () => { const attempts: Array = []; const markedDeliveries: Array< From e21c3c581e874d22812bffb241b1f51ff037fd1b Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 8 Nov 2026 12:42:04 +0000 Subject: [PATCH 8/8] test(relay): yield the signed APNs jobs after main's signer became an Effect Main #30501 made signApnsDeliveryJob an Effect.fn. The branch's tests passed the Effect object as the job body, so decode failed at runtime. Co-Authored-By: Claude Opus 5.5 --- .../src/agentActivity/ApnsDeliveries.test.ts | 24 ++++++++++--------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index b2812cb8443de..fe04a9ac09cf9 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -987,9 +987,8 @@ describe("ApnsDeliveries", () => { return Effect.gen(function* () { const deliveries = yield* ApnsDeliveries.ApnsDeliveries; - const fiber = yield* deliveries - .processSignedJob(signed) - .pipe(Effect.flip, Effect.forkChild); + const job = yield* signed; + const fiber = yield* deliveries.processSignedJob(job).pipe(Effect.flip, Effect.forkChild); yield* TestClock.adjust("1 second"); const failure = yield* Fiber.join(fiber); @@ -1001,7 +1000,7 @@ describe("ApnsDeliveries", () => { expect(failure).not.toHaveProperty("apnsReason"); expect(attempts).toEqual([]); - const redelivered = yield* deliveries.processSignedJob(signed); + const redelivered = yield* deliveries.processSignedJob(job); expect(redelivered).toMatchObject({ ok: true, apnsStatus: 200 }); expect(statuses).toEqual([]); @@ -1060,7 +1059,9 @@ describe("ApnsDeliveries", () => { return Effect.gen(function* () { const deliveries = yield* ApnsDeliveries.ApnsDeliveries; - const fiber = yield* deliveries.processSignedJob(signed).pipe(Effect.flip, Effect.forkChild); + const fiber = yield* deliveries + .processSignedJob(yield* signed) + .pipe(Effect.flip, Effect.forkChild); yield* TestClock.adjust("1 second"); const failure = yield* Fiber.join(fiber); @@ -1121,7 +1122,8 @@ describe("ApnsDeliveries", () => { return Effect.gen(function* () { const deliveries = yield* ApnsDeliveries.ApnsDeliveries; - const failure = yield* deliveries.processSignedJob(signed).pipe(Effect.flip); + const job = yield* signed; + const failure = yield* deliveries.processSignedJob(job).pipe(Effect.flip); expect(failure).toMatchObject({ _tag: "ApnsDeliveryJobRetryableFailure", @@ -1131,7 +1133,7 @@ describe("ApnsDeliveries", () => { expect(attempts).toEqual([]); expect(markedDeliveries).toEqual([]); - const redelivered = yield* deliveries.processSignedJob(signed); + const redelivered = yield* deliveries.processSignedJob(job); expect(redelivered).toMatchObject({ ok: true, apnsStatus: 200 }); expect(executeCount).toBe(2); @@ -1294,10 +1296,6 @@ describe("APNs queue message isolation", () => { if (!outcomes.has(id)) outcomes.set(id, "retry"); }, }); - const batch = [ - message("throttled", signedPush("device-1", "throttled-token")), - message("healthy", signedPush("device-2", "healthy-token")), - ]; const execute = (request: HttpClientRequest.HttpClientRequest) => Effect.sync(() => { const token = request.url.split("/").at(-1) ?? ""; @@ -1311,6 +1309,10 @@ describe("APNs queue message isolation", () => { }); return Effect.gen(function* () { + const batch = [ + message("throttled", yield* signedPush("device-1", "throttled-token")), + message("healthy", yield* signedPush("device-2", "healthy-token")), + ]; yield* Stream.fromIterable(batch).pipe( Stream.runForEach(ApnsDeliveryQueueConsumer.processMessage), );