authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-04 10:58:14-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 10:58:14-07:00
log088cc3f95599b7b933289d719fa1e9b0608cb4cd
treebb56f1e47b4214ad1096f0ce0998e7e4ebb09049
parent19c1710a88a2c87c159d76269dacdf0b17ddd9f4
signature Signed by PGP key B5690EEEBB952194

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.

3 files changed, 162 insertions(+), 0 deletions(-)

infra/relay/src/agentActivity/ApnsClient.test.ts+79
...@@ -4,10 +4,13 @@ import { EnvironmentId, ThreadId } from "@t3tools/contracts";...@@ -4,10 +4,13 @@ import { EnvironmentId, ThreadId } from "@t3tools/contracts";
4import type { RelayAgentActivityAggregateState } from "@t3tools/contracts/relay";4import type { RelayAgentActivityAggregateState } from "@t3tools/contracts/relay";
5import { describe, expect, it } from "@effect/vitest";5import { describe, expect, it } from "@effect/vitest";
6import * as DateTime from "effect/DateTime";6import * as DateTime from "effect/DateTime";
7import * as Deferred from "effect/Deferred";
7import * as Effect from "effect/Effect";8import * as Effect from "effect/Effect";
9import * as Fiber from "effect/Fiber";
8import * as Layer from "effect/Layer";10import * as Layer from "effect/Layer";
9import * as Redacted from "effect/Redacted";11import * as Redacted from "effect/Redacted";
10import * as Schema from "effect/Schema";12import * as Schema from "effect/Schema";
13import * as TestClock from "effect/testing/TestClock";
11import * as HttpClient from "effect/unstable/http/HttpClient";14import * as HttpClient from "effect/unstable/http/HttpClient";
12import * as HttpClientError from "effect/unstable/http/HttpClientError";15import * as HttpClientError from "effect/unstable/http/HttpClientError";
13import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse";16import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse";
...@@ -361,4 +364,80 @@ describe("ApnsClient", () => {...@@ -361,4 +364,80 @@ describe("ApnsClient", () => {
361 ApnsProviderTokens.__resetApnsProviderTokenCacheForTest();364 ApnsProviderTokens.__resetApnsProviderTokenCacheForTest();
362 }).pipe(Effect.provide(layer));365 }).pipe(Effect.provide(layer));
363 });366 });
367
368 for (const requestKind of ["live-activity", "push-notification"] as const) {
369 for (const stage of ["send", "read-response"] as const) {
370 it.effect(`aborts a stalled ${requestKind} ${stage} after ten seconds`, () =>
371 Effect.gen(function* () {
372 const started = yield* Deferred.make<void>();
373 const signals: AbortSignal[] = [];
374 const stalledHttpClient = HttpClient.make((request, _url, signal) => {
375 signals.push(signal);
376 const stall = Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never));
377 if (stage === "send") return stall;
378 const response = HttpClientResponse.fromWeb(request, new Response("", { status: 200 }));
379 Object.defineProperty(response, "text", { value: stall });
380 return Effect.succeed(response);
381 });
382 const layer = ApnsClient.layer.pipe(
383 Layer.provide(Layer.succeed(HttpClient.HttpClient, stalledHttpClient)),
384 Layer.provide(
385 Layer.succeed(ApnsProviderTokens.ApnsProviderTokens, {
386 getJwt: () => Effect.succeed("test-jwt"),
387 }),
388 ),
389 );
390 const apns = yield* ApnsClient.ApnsClient.pipe(Effect.provide(layer));
391 const credentials = {
392 teamId: "team-timeout",
393 keyId: "key-timeout",
394 privateKey: Redacted.make("unused-test-key"),
395 bundleId: "com.t3tools.test",
396 environment: "sandbox",
397 } satisfies ApnsCredentials;
398 const send =
399 requestKind === "live-activity"
400 ? apns.sendLiveActivityRequest({
401 credentials,
402 issuedAtUnixSeconds: 123,
403 request: apns.makeLiveActivityRequest({
404 event: "update",
405 token: "long-push-token",
406 state,
407 nowEpochSeconds: 123,
408 nowIso: DateTime.formatIso(now),
409 }),
410 })
411 : apns.sendPushNotificationRequest({
412 credentials,
413 issuedAtUnixSeconds: 123,
414 request: apns.makePushNotificationRequest({
415 token: "long-push-token",
416 notification: {
417 title: "Thread",
418 body: "Done",
419 environmentId: "env",
420 threadId: "thread",
421 deepLink: "/",
422 },
423 }),
424 });
425 const fiber = yield* send.pipe(Effect.flip, Effect.forkChild);
426 yield* Deferred.await(started);
427 yield* TestClock.adjust("10 seconds");
428 expect(signals[0]?.aborted).toBe(true);
429 const error = yield* Fiber.join(fiber);
430 expect(error).toMatchObject({
431 _tag: "ApnsHttpRequestError",
432 requestKind,
433 event: requestKind === "live-activity" ? "update" : null,
434 stage,
435 status: stage === "read-response" ? 200 : null,
436 tokenSuffix: "sh-token",
437 cause: { _tag: "TimeoutError" },
438 });
439 }),
440 );
441 }
442 }
364});443});
infra/relay/src/agentActivity/ApnsClient.ts+6
...@@ -15,6 +15,8 @@ import * as ApnsProviderTokens from "./ApnsProviderTokens.ts";...@@ -15,6 +15,8 @@ import * as ApnsProviderTokens from "./ApnsProviderTokens.ts";
15export { ApnsJwtEncodingError, ApnsJwtSigningError } from "./apnsJwt.ts";15export { ApnsJwtEncodingError, ApnsJwtSigningError } from "./apnsJwt.ts";
1616
17const LIVE_ACTIVITY_NAME = "AgentActivity";17const LIVE_ACTIVITY_NAME = "AgentActivity";
18// Bound sending and reading separately so neither stage can hold a batch open.
19const APNS_HTTP_STAGE_TIMEOUT = "10 seconds";
18// Updates only flow on domain events, so a healthy agent can be silent for20// Updates only flow on domain events, so a healthy agent can be silent for
19// minutes (long tool calls, pending approvals). Two minutes made iOS dim21// minutes (long tool calls, pending approvals). Two minutes made iOS dim
20// perfectly healthy activities; ten minutes still bounds how long a dead22// perfectly healthy activities; ten minutes still bounds how long a dead
...@@ -246,6 +248,7 @@ export const make = Effect.gen(function* () {...@@ -246,6 +248,7 @@ export const make = Effect.gen(function* () {
246 }),248 }),
247 HttpClientRequest.bodyJson(input.request.payload),249 HttpClientRequest.bodyJson(input.request.payload),
248 Effect.flatMap(httpClient.execute),250 Effect.flatMap(httpClient.execute),
251 Effect.timeout(APNS_HTTP_STAGE_TIMEOUT),
249 Effect.mapError(252 Effect.mapError(
250 (cause) =>253 (cause) =>
251 new ApnsHttpRequestError({254 new ApnsHttpRequestError({
...@@ -261,6 +264,7 @@ export const make = Effect.gen(function* () {...@@ -261,6 +264,7 @@ export const make = Effect.gen(function* () {
261 ),264 ),
262 );265 );
263 const responseText = yield* response.text.pipe(266 const responseText = yield* response.text.pipe(
267 Effect.timeout(APNS_HTTP_STAGE_TIMEOUT),
264 Effect.mapError(268 Effect.mapError(
265 (cause) =>269 (cause) =>
266 new ApnsHttpRequestError({270 new ApnsHttpRequestError({
...@@ -306,6 +310,7 @@ export const make = Effect.gen(function* () {...@@ -306,6 +310,7 @@ export const make = Effect.gen(function* () {
306 }),310 }),
307 HttpClientRequest.bodyJson(input.request.payload),311 HttpClientRequest.bodyJson(input.request.payload),
308 Effect.flatMap(httpClient.execute),312 Effect.flatMap(httpClient.execute),
313 Effect.timeout(APNS_HTTP_STAGE_TIMEOUT),
309 Effect.mapError(314 Effect.mapError(
310 (cause) =>315 (cause) =>
311 new ApnsHttpRequestError({316 new ApnsHttpRequestError({
...@@ -321,6 +326,7 @@ export const make = Effect.gen(function* () {...@@ -321,6 +326,7 @@ export const make = Effect.gen(function* () {
321 ),326 ),
322 );327 );
323 const responseText = yield* response.text.pipe(328 const responseText = yield* response.text.pipe(
329 Effect.timeout(APNS_HTTP_STAGE_TIMEOUT),
324 Effect.mapError(330 Effect.mapError(
325 (cause) =>331 (cause) =>
326 new ApnsHttpRequestError({332 new ApnsHttpRequestError({
infra/relay/src/agentActivity/ApnsDeliveries.test.ts+77
...@@ -5,11 +5,15 @@ import type {...@@ -5,11 +5,15 @@ import type {
5import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto";5import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto";
6import { describe, expect, it } from "@effect/vitest";6import { describe, expect, it } from "@effect/vitest";
7import * as NodeCrypto from "node:crypto";7import * as NodeCrypto from "node:crypto";
8import * as Deferred from "effect/Deferred";
8import * as Effect from "effect/Effect";9import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
9import * as Layer from "effect/Layer";11import * as Layer from "effect/Layer";
10import * as Logger from "effect/Logger";12import * as Logger from "effect/Logger";
11import * as Redacted from "effect/Redacted";13import * as Redacted from "effect/Redacted";
12import * as References from "effect/References";14import * as References from "effect/References";
15import * as Stream from "effect/Stream";
16import * as TestClock from "effect/testing/TestClock";
13import {17import {
14 FetchHttpClient,18 FetchHttpClient,
15 HttpClient,19 HttpClient,
...@@ -975,6 +979,79 @@ describe("ApnsDeliveries", () => {...@@ -975,6 +979,79 @@ describe("ApnsDeliveries", () => {
975 );979 );
976 });980 });
977981
982 it.effect("continues a signed delivery batch after an APNs timeout", () =>
983 Effect.gen(function* () {
984 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
985 const markedDeliveries: Array<
986 Parameters<LiveActivities.LiveActivities["Service"]["markDelivery"]>[0]
987 > = [];
988 const secondTarget = {
989 ...target,
990 device_id: "device-2",
991 activity_push_token: "second-token",
992 };
993 const jobs = [target, secondTarget].map((device) =>
994 signApnsDeliveryJob({
995 secret: signingConfig.apnsDeliveryJobSigningSecret,
996 payload: makeApnsDeliveryJobPayload({
997 kind: "live_activity_update",
998 userId: device.user_id,
999 deviceId: device.device_id,
1000 token: device.activity_push_token!,
1001 aggregate,
1002 createdAt: "1970-01-01T00:00:00.000Z",
1003 expiresAt: "1970-01-01T00:10:00.000Z",
1004 jobId: `job-timeout-${device.device_id}`,
1005 }),
1006 }),
1007 );
1008 const started = yield* Deferred.make<void>();
1009 const results: Array<{ deviceId: string; ok: boolean }> = [];
1010 const layer = makeLayer({
1011 attempts,
1012 markedDeliveries,
1013 currentTargets: [target, secondTarget],
1014 config: signingConfig,
1015 execute: (request) =>
1016 request.url.endsWith("/activity-token")
1017 ? Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never))
1018 : Effect.succeed(
1019 HttpClientResponse.fromWeb(request, new Response("", { status: 200 })),
1020 ),
1021 });
1022 const batch = yield* Effect.gen(function* () {
1023 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
1024 yield* Stream.fromIterable(jobs).pipe(
1025 Stream.runForEach((job) =>
1026 deliveries.processSignedJob(job).pipe(
1027 Effect.tap((result) =>
1028 Effect.sync(() => {
1029 results.push(result);
1030 }),
1031 ),
1032 ),
1033 ),
1034 );
1035 }).pipe(Effect.provide(layer), Effect.forkChild);
1036
1037 yield* Deferred.await(started);
1038 yield* TestClock.adjust("10 seconds");
1039 yield* Fiber.join(batch);
1040 expect(results).toMatchObject([
1041 { deviceId: "device-1", ok: false },
1042 { deviceId: "device-2", ok: true },
1043 ]);
1044 expect(attempts).toMatchObject([
1045 {
1046 sourceJobId: "job-timeout-device-1",
1047 apnsReason: expect.stringContaining("request failed"),
1048 },
1049 { sourceJobId: "job-timeout-device-2", apnsStatus: 200 },
1050 ]);
1051 expect(markedDeliveries).toMatchObject([{ deviceId: "device-2" }]);
1052 }),
1053 );
1054
978 it.effect("processes signed push notification jobs through APNs and records attempts", () => {1055 it.effect("processes signed push notification jobs through APNs and records attempts", () => {
979 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];1056 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
980 const payload = makeApnsDeliveryJobPayload({1057 const payload = makeApnsDeliveryJobPayload({