fix(relay): bound stalled push requests

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<void>();
+          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" },
+          });
+        }),
+      );
+    }
+  }
 });

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({

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<DeliveryAttempts.DeliveryAttemptInput> = [];
+      const markedDeliveries: Array<
+        Parameters<LiveActivities.LiveActivities["Service"]["markDelivery"]>[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<void>();
+      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<DeliveryAttempts.DeliveryAttemptInput> = [];
     const payload = makeApnsDeliveryJobPayload({