From 088cc3f95599b7b933289d719fa1e9b0608cb4cd Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Fri, 4 Sep 2026 10:58:14 -0700 Subject: [PATCH] fix(relay): bound stalled push requests (#9734) Give HTTP sends and response-body reads ten seconds each. Keep the existing typed transport errors, delivery records, and queue policy. Abort stalled requests and let the next signed job run. JWT retrieval is outside these deadlines. Do not add retries for an uncertain response result. Created with GPT-6 Astra (preview) in Codex. --- .../src/agentActivity/ApnsClient.test.ts | 79 +++++++++++++++++++ infra/relay/src/agentActivity/ApnsClient.ts | 6 ++ .../src/agentActivity/ApnsDeliveries.test.ts | 77 ++++++++++++++++++ 3 files changed, 162 insertions(+) diff --git a/infra/relay/src/agentActivity/ApnsClient.test.ts b/infra/relay/src/agentActivity/ApnsClient.test.ts index 9f9877533515dad45f8037139619ed73d1afa545..03010ce48ce53a05b1c0e78f65b03678f57ca922 100644 --- a/infra/relay/src/agentActivity/ApnsClient.test.ts +++ b/infra/relay/src/agentActivity/ApnsClient.test.ts @@ -4,10 +4,13 @@ import { EnvironmentId, ThreadId } from "@t3tools/contracts"; import type { RelayAgentActivityAggregateState } from "@t3tools/contracts/relay"; import { describe, expect, it } from "@effect/vitest"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; 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 Schema from "effect/Schema"; +import * as TestClock from "effect/testing/TestClock"; import * as HttpClient from "effect/unstable/http/HttpClient"; import * as HttpClientError from "effect/unstable/http/HttpClientError"; import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse"; @@ -361,4 +364,80 @@ describe("ApnsClient", () => { ApnsProviderTokens.__resetApnsProviderTokenCacheForTest(); }).pipe(Effect.provide(layer)); }); + + for (const requestKind of ["live-activity", "push-notification"] as const) { + for (const stage of ["send", "read-response"] as const) { + it.effect(`aborts a stalled ${requestKind} ${stage} after ten seconds`, () => + Effect.gen(function* () { + const started = yield* Deferred.make(); + const signals: AbortSignal[] = []; + const stalledHttpClient = HttpClient.make((request, _url, signal) => { + signals.push(signal); + const stall = Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)); + if (stage === "send") return stall; + const response = HttpClientResponse.fromWeb(request, new Response("", { status: 200 })); + Object.defineProperty(response, "text", { value: stall }); + return Effect.succeed(response); + }); + const layer = ApnsClient.layer.pipe( + Layer.provide(Layer.succeed(HttpClient.HttpClient, stalledHttpClient)), + Layer.provide( + Layer.succeed(ApnsProviderTokens.ApnsProviderTokens, { + getJwt: () => Effect.succeed("test-jwt"), + }), + ), + ); + const apns = yield* ApnsClient.ApnsClient.pipe(Effect.provide(layer)); + const credentials = { + teamId: "team-timeout", + keyId: "key-timeout", + privateKey: Redacted.make("unused-test-key"), + bundleId: "com.t3tools.test", + environment: "sandbox", + } satisfies ApnsCredentials; + const send = + requestKind === "live-activity" + ? apns.sendLiveActivityRequest({ + credentials, + issuedAtUnixSeconds: 123, + request: apns.makeLiveActivityRequest({ + event: "update", + token: "long-push-token", + state, + nowEpochSeconds: 123, + nowIso: DateTime.formatIso(now), + }), + }) + : apns.sendPushNotificationRequest({ + credentials, + issuedAtUnixSeconds: 123, + request: apns.makePushNotificationRequest({ + token: "long-push-token", + notification: { + title: "Thread", + body: "Done", + environmentId: "env", + threadId: "thread", + deepLink: "/", + }, + }), + }); + const fiber = yield* send.pipe(Effect.flip, Effect.forkChild); + yield* Deferred.await(started); + yield* TestClock.adjust("10 seconds"); + expect(signals[0]?.aborted).toBe(true); + const error = yield* Fiber.join(fiber); + expect(error).toMatchObject({ + _tag: "ApnsHttpRequestError", + requestKind, + event: requestKind === "live-activity" ? "update" : null, + stage, + status: stage === "read-response" ? 200 : null, + tokenSuffix: "sh-token", + cause: { _tag: "TimeoutError" }, + }); + }), + ); + } + } }); diff --git a/infra/relay/src/agentActivity/ApnsClient.ts b/infra/relay/src/agentActivity/ApnsClient.ts index c401eb6c2e4e7b25484973ba2bf4d38d2304bd45..4276b7062555a5d037eb4df54b0f0e3f31d9eaa9 100644 --- a/infra/relay/src/agentActivity/ApnsClient.ts +++ b/infra/relay/src/agentActivity/ApnsClient.ts @@ -15,6 +15,8 @@ import * as ApnsProviderTokens from "./ApnsProviderTokens.ts"; export { ApnsJwtEncodingError, ApnsJwtSigningError } from "./apnsJwt.ts"; const LIVE_ACTIVITY_NAME = "AgentActivity"; +// Bound sending and reading separately so neither stage can hold a batch open. +const APNS_HTTP_STAGE_TIMEOUT = "10 seconds"; // Updates only flow on domain events, so a healthy agent can be silent for // minutes (long tool calls, pending approvals). Two minutes made iOS dim // perfectly healthy activities; ten minutes still bounds how long a dead @@ -246,6 +248,7 @@ export const make = Effect.gen(function* () { }), HttpClientRequest.bodyJson(input.request.payload), Effect.flatMap(httpClient.execute), + Effect.timeout(APNS_HTTP_STAGE_TIMEOUT), Effect.mapError( (cause) => new ApnsHttpRequestError({ @@ -261,6 +264,7 @@ export const make = Effect.gen(function* () { ), ); const responseText = yield* response.text.pipe( + Effect.timeout(APNS_HTTP_STAGE_TIMEOUT), Effect.mapError( (cause) => new ApnsHttpRequestError({ @@ -306,6 +310,7 @@ export const make = Effect.gen(function* () { }), HttpClientRequest.bodyJson(input.request.payload), Effect.flatMap(httpClient.execute), + Effect.timeout(APNS_HTTP_STAGE_TIMEOUT), Effect.mapError( (cause) => new ApnsHttpRequestError({ @@ -321,6 +326,7 @@ export const make = Effect.gen(function* () { ), ); const responseText = yield* response.text.pipe( + Effect.timeout(APNS_HTTP_STAGE_TIMEOUT), Effect.mapError( (cause) => new ApnsHttpRequestError({ diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index 1c89b56da5c968147d7723219e59d0c80fdca047..e34dcc55e37ef54f23af04f84506a97187bc6e4c 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -5,11 +5,15 @@ import type { import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto"; import { describe, expect, it } from "@effect/vitest"; import * as NodeCrypto from "node:crypto"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; import * as Redacted from "effect/Redacted"; import * as References from "effect/References"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { FetchHttpClient, HttpClient, @@ -975,6 +979,79 @@ describe("ApnsDeliveries", () => { ); }); + it.effect("continues a signed delivery batch after an APNs timeout", () => + Effect.gen(function* () { + const attempts: Array = []; + const markedDeliveries: Array< + Parameters[0] + > = []; + const secondTarget = { + ...target, + device_id: "device-2", + activity_push_token: "second-token", + }; + const jobs = [target, secondTarget].map((device) => + signApnsDeliveryJob({ + secret: signingConfig.apnsDeliveryJobSigningSecret, + payload: makeApnsDeliveryJobPayload({ + kind: "live_activity_update", + userId: device.user_id, + deviceId: device.device_id, + token: device.activity_push_token!, + aggregate, + createdAt: "1970-01-01T00:00:00.000Z", + expiresAt: "1970-01-01T00:10:00.000Z", + jobId: `job-timeout-${device.device_id}`, + }), + }), + ); + const started = yield* Deferred.make(); + const results: Array<{ deviceId: string; ok: boolean }> = []; + const layer = makeLayer({ + attempts, + markedDeliveries, + currentTargets: [target, secondTarget], + config: signingConfig, + execute: (request) => + request.url.endsWith("/activity-token") + ? Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.succeed( + HttpClientResponse.fromWeb(request, new Response("", { status: 200 })), + ), + }); + const batch = yield* Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + yield* Stream.fromIterable(jobs).pipe( + Stream.runForEach((job) => + deliveries.processSignedJob(job).pipe( + Effect.tap((result) => + Effect.sync(() => { + results.push(result); + }), + ), + ), + ), + ); + }).pipe(Effect.provide(layer), Effect.forkChild); + + yield* Deferred.await(started); + yield* TestClock.adjust("10 seconds"); + yield* Fiber.join(batch); + expect(results).toMatchObject([ + { deviceId: "device-1", ok: false }, + { deviceId: "device-2", ok: true }, + ]); + expect(attempts).toMatchObject([ + { + sourceJobId: "job-timeout-device-1", + apnsReason: expect.stringContaining("request failed"), + }, + { sourceJobId: "job-timeout-device-2", apnsStatus: 200 }, + ]); + expect(markedDeliveries).toMatchObject([{ deviceId: "device-2" }]); + }), + ); + it.effect("processes signed push notification jobs through APNs and records attempts", () => { const attempts: Array = []; const payload = makeApnsDeliveryJobPayload({ -- 2.54.0