authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-09-08 16:39:48-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-08 16:39:48-07:00
log3dfc134e6c1e2359400c56ea415e74fecf9d5c12
tree714428671226282cdea453ae955518066d9f2119
parentfdf34c4018352df41aa22fbaa774c25c6c1ff7da
signature Signed by PGP key B5690EEEBB952194

fix(relay): recheck queued iOS alerts and retain fast completions (#10849)

3 files changed, 192 insertions(+), 34 deletions(-)

infra/relay/src/agentActivity/AgentActivityPublisher.ts+1
......@@ -130,6 +130,7 @@ export const make = Effect.gen(function* () {
130130 target,
131131 aggregate,
132132 nowMs: now.epochMilliseconds,
133 replay: true,
133134 });
134135 }),
135136 publish: Effect.fn("relay.agent_activity_publisher.publish")(function* (input) {
infra/relay/src/agentActivity/ApnsDeliveries.test.ts+110
......@@ -1862,3 +1862,113 @@ describe("live activity alert decisions", () => {
18621862 ).toBeNull();
18631863 });
18641864});
1865
1866describe("queued iOS alert policy", () => {
1867 for (const scenario of ["enabled", "muted", "late"] as const) {
1868 it.effect(`checks the current policy for a ${scenario} completion`, () => {
1869 let sent = 0;
1870 const completed = { ...state, phase: "completed" as const };
1871 const prefs = JSON.parse(enabledPreferences);
1872 if (scenario === "muted") prefs.notifyOnCompletion = false;
1873 const payload = makeApnsDeliveryJobPayload({
1874 kind: "push_notification",
1875 userId: target.user_id,
1876 deviceId: target.device_id,
1877 token: "push",
1878 aggregate: null,
1879 notification: {
1880 title: "Thread",
1881 body: "Done: Project",
1882 environmentId: "env",
1883 threadId: "thread",
1884 deepLink: "/",
1885 phase: "completed",
1886 updatedAt: completed.updatedAt,
1887 },
1888 createdAt: completed.updatedAt,
1889 expiresAt: "1970-01-01T00:10:00.000Z",
1890 jobId: `delivery-policy-${scenario}`,
1891 });
1892 const signed = signApnsDeliveryJob({ secret: config.apnsDeliveryJobSigningSecret, payload });
1893 return Effect.gen(function* () {
1894 if (scenario === "late") yield* TestClock.adjust("3 minutes");
1895 const d = yield* ApnsDeliveries.ApnsDeliveries;
1896 yield* d.processSignedJob(signed);
1897 expect(sent).toBe(scenario === "enabled" ? 1 : 0);
1898 }).pipe(
1899 Effect.provide(
1900 makeLayer({
1901 attempts: [],
1902 config: signingConfig,
1903 currentTargets: [
1904 { ...target, push_token: "push", preferences_json: JSON.stringify(prefs) },
1905 ],
1906 currentActivityStates: [completed],
1907 execute: (request) =>
1908 Effect.sync(() => {
1909 sent++;
1910 return HttpClientResponse.fromWeb(request, new Response("", { status: 200 }));
1911 }),
1912 }),
1913 ),
1914 );
1915 });
1916 }
1917});
1918
1919describe("fast completion delivery", () => {
1920 it.effect("keeps a completion alert when work finishes before running delivery", () => {
1921 const queuedJobs: SignedApnsDeliveryJob[] = [];
1922 const old = {
1923 ...aggregate,
1924 activeCount: 0,
1925 activities: [
1926 {
1927 ...aggregate.activities[0]!,
1928 threadId: "old" as RelayAgentActivityState["threadId"],
1929 phase: "completed" as const,
1930 },
1931 ],
1932 };
1933 const device = { ...target, last_aggregate_json: JSON.stringify(old) };
1934 const done = {
1935 ...aggregate,
1936 activeCount: 0,
1937 activities: [{ ...aggregate.activities[0]!, phase: "completed" as const }],
1938 };
1939 return Effect.gen(function* () {
1940 const d = yield* ApnsDeliveries.ApnsDeliveries;
1941 yield* d.sendForTarget({ target: device, aggregate, nowMs: 0 });
1942 yield* d.sendForTarget({ target: device, aggregate: done, nowMs: 0 });
1943 expect(
1944 queuedJobs.some((x) => x.payload.alert !== null && x.payload.alert !== undefined),
1945 ).toBe(true);
1946 }).pipe(Effect.provide(makeLayer({ attempts: [], queuedJobs, currentTargets: [device] })));
1947 });
1948 it.effect("replays a newly visible completion without alerting", () => {
1949 const queuedJobs: SignedApnsDeliveryJob[] = [];
1950 const done = {
1951 ...aggregate,
1952 activeCount: 0,
1953 activities: [{ ...aggregate.activities[0]!, phase: "completed" as const }],
1954 };
1955 const previous = {
1956 ...aggregate,
1957 activities: [
1958 { ...aggregate.activities[0]!, threadId: "other" as RelayAgentActivityState["threadId"] },
1959 ],
1960 };
1961 const device = { ...target, last_aggregate_json: JSON.stringify(previous) };
1962 return Effect.gen(function* () {
1963 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
1964 yield* deliveries.sendForTarget({
1965 target: device,
1966 aggregate: done,
1967 nowMs: 0,
1968 replay: true,
1969 });
1970 expect(queuedJobs).toHaveLength(1);
1971 expect(queuedJobs[0]?.payload.alert).toBeUndefined();
1972 }).pipe(Effect.provide(makeLayer({ attempts: [], queuedJobs })));
1973 });
1974});
infra/relay/src/agentActivity/ApnsDeliveries.ts+81-34
......@@ -181,7 +181,7 @@ function shouldUpdateLiveActivity(input: {
181181 // A thread finishing must never be throttled away: when a completion and a
182182 // new start land in the same window, activeCount is unchanged and the Done
183183 // transition (and its alert) would otherwise be suppressed.
184 if (newlyTerminalRows(input.previousAggregate, input.nextAggregate).length > 0) {
184 if (newlyTerminalRows(input.previousAggregate, input.nextAggregate, true).length > 0) {
185185 return true;
186186 }
187187 const lastDeliveryAtMs =
......@@ -228,6 +228,7 @@ function chooseLiveActivityDelivery(input: {
228228 readonly target: LiveActivities.TargetRow;
229229 readonly aggregate: RelayAgentActivityAggregateState | null;
230230 readonly nowMs: number;
231 readonly replay?: boolean;
231232}): ChosenLiveActivityDelivery | "suppressed" | null {
232233 const preferences = parsePreferences(input.target.preferences_json);
233234 if (preferences?.liveActivitiesEnabled === false) {
......@@ -286,18 +287,20 @@ function chooseLiveActivityDelivery(input: {
286287 kind: "live_activity_update",
287288 token: input.target.activity_push_token,
288289 aggregate: nextAggregate,
289 alert:
290 alertForAttentionTransition({
291 previousAggregate,
292 nextAggregate,
293 preferences,
294 }) ??
295 alertForNewlyTerminal({
296 previousAggregate,
297 nextAggregate,
298 preferences,
299 nowMs: input.nowMs,
300 }),
290 alert: input.replay
291 ? null
292 : (alertForAttentionTransition({
293 previousAggregate,
294 nextAggregate,
295 preferences,
296 }) ??
297 alertForNewlyTerminal({
298 previousAggregate,
299 nextAggregate,
300 preferences,
301 nowMs: input.nowMs,
302 includeUnobserved: true,
303 })),
301304 }
302305 : "suppressed";
303306}
......@@ -306,6 +309,7 @@ function chooseDelivery(input: {
306309 readonly target: LiveActivities.TargetRow;
307310 readonly aggregate: RelayAgentActivityAggregateState | null;
308311 readonly nowMs: number;
312 readonly replay?: boolean;
309313}): ChosenDelivery | null {
310314 const liveActivityDelivery = chooseLiveActivityDelivery(input);
311315 if (liveActivityDelivery === "suppressed") {
......@@ -314,7 +318,7 @@ function chooseDelivery(input: {
314318 if (liveActivityDelivery) {
315319 return liveActivityDelivery;
316320 }
317 const notification = notificationForAggregate(input);
321 const notification = input.replay ? null : notificationForAggregate(input);
318322 return notification && input.target.push_token
319323 ? {
320324 kind: "push_notification",
......@@ -519,6 +523,7 @@ export class ApnsDeliveries extends Context.Service<
519523 readonly target: LiveActivities.TargetRow;
520524 readonly aggregate: RelayAgentActivityAggregateState | null;
521525 readonly nowMs: number;
526 readonly replay?: boolean;
522527 }) => Effect.Effect<RelayDeliveryResult | null, ApnsDeliveryError>;
523528 readonly sendPushNotificationForTarget: (input: {
524529 readonly target: LiveActivities.TargetRow;
......@@ -652,7 +657,7 @@ export const make = Effect.gen(function* () {
652657 });
653658 });
654659
655 const isCurrentSignedJobToken = Effect.fnUntraced(function* (input: {
660 const currentSignedJobTarget = Effect.fnUntraced(function* (input: {
656661 readonly target: LiveActivityDeliveryTarget;
657662 readonly kind: RelayDeliveryKind;
658663 readonly token: string;
......@@ -660,10 +665,10 @@ export const make = Effect.gen(function* () {
660665 return yield* liveActivities.listTargets({ userId: input.target.user_id }).pipe(
661666 Effect.map((targets) => {
662667 const currentTarget = targets.find((row) => row.device_id === input.target.device_id);
663 return (
664 currentTarget !== undefined &&
668 return currentTarget &&
665669 expectedCurrentToken({ target: currentTarget, kind: input.kind }) === input.token
666 );
670 ? currentTarget
671 : null;
667672 }),
668673 );
669674 });
......@@ -679,11 +684,7 @@ export const make = Effect.gen(function* () {
679684 const now = yield* DateTime.now;
680685 const aggregate =
681686 input.aggregate === null ? null : sanitizeAgentActivityAggregateState(input.aggregate);
682 const { epochSeconds, iso, request } = makeLiveActivityDeliveryRequest(
683 apns,
684 { ...input, aggregate } as SendLiveActivityDeliveryInput,
685 now,
686 );
687 let alert = input.alert ?? null;
687688 const recoverTransportError = (cause: Apns.ApnsError) =>
688689 recoverApnsDeliveryTransportError(
689690 {
......@@ -709,18 +710,37 @@ export const make = Effect.gen(function* () {
709710 if (claim === "in_flight") {
710711 return yield* new ApnsDeliveryJobClaimInFlight({ sourceJobId: input.sourceJobId });
711712 }
712 const tokenIsCurrent = yield* isCurrentSignedJobToken({
713 const currentTarget = yield* currentSignedJobTarget({
713714 target: input.target,
714715 kind: input.kind,
715716 token: input.token,
716717 });
717 if (!tokenIsCurrent) {
718 if (!currentTarget) {
718719 yield* attempts.completeSourceJob({
719720 sourceJobId: input.sourceJobId,
720721 apnsReason: "Stale APNs delivery job skipped.",
721722 });
722723 return staleJobResult({ deviceId: input.target.device_id, kind: input.kind });
723724 }
725 if (alert) {
726 const preferences = parsePreferences(currentTarget.preferences_json);
727 const previousAggregate = parseAggregate(currentTarget.last_aggregate_json);
728 alert =
729 !preferences?.notificationsEnabled || !aggregate
730 ? null
731 : (alertForAttentionTransition({
732 previousAggregate,
733 nextAggregate: aggregate,
734 preferences,
735 }) ??
736 alertForNewlyTerminal({
737 previousAggregate,
738 nextAggregate: aggregate,
739 preferences,
740 nowMs: now.epochMilliseconds,
741 includeUnobserved: true,
742 }));
743 }
724744 if (
725745 input.kind !== "live_activity_start" &&
726746 aggregate !== null &&
......@@ -752,6 +772,11 @@ export const make = Effect.gen(function* () {
752772 }
753773 return staleJobResult({ deviceId: input.target.device_id, kind: input.kind });
754774 }
775 const { epochSeconds, iso, request } = makeLiveActivityDeliveryRequest(
776 apns,
777 { ...input, aggregate, alert } as SendLiveActivityDeliveryInput,
778 now,
779 );
755780 const result = yield* apns
756781 .sendLiveActivityRequest({
757782 credentials: credentialsForTarget(config.apns, input.target),
......@@ -859,12 +884,12 @@ export const make = Effect.gen(function* () {
859884 if (claim === "in_flight") {
860885 return yield* new ApnsDeliveryJobClaimInFlight({ sourceJobId: input.sourceJobId });
861886 }
862 const tokenIsCurrent = yield* isCurrentSignedJobToken({
887 const currentTarget = yield* currentSignedJobTarget({
863888 target: input.target,
864889 kind: "push_notification",
865890 token: input.token,
866891 });
867 if (!tokenIsCurrent) {
892 if (!currentTarget) {
868893 yield* attempts.completeSourceJob({
869894 sourceJobId: input.sourceJobId,
870895 apnsReason: "Stale APNs delivery job skipped.",
......@@ -889,6 +914,24 @@ export const make = Effect.gen(function* () {
889914 kind: "push_notification",
890915 });
891916 }
917 const preferences = parsePreferences(currentTarget.preferences_json);
918 const alertAllowed =
919 notification.phase !== undefined && notification.updatedAt !== undefined
920 ? shouldAlertForActivity({
921 ...notification,
922 phase: notification.phase,
923 updatedAt: notification.updatedAt,
924 preferences,
925 nowMs: now.epochMilliseconds,
926 })
927 : preferences?.notificationsEnabled === true;
928 if (!alertAllowed) {
929 yield* attempts.completeSourceJob({
930 sourceJobId: input.sourceJobId,
931 apnsReason: "Notification is disabled or no longer fresh.",
932 });
933 return staleJobResult({ deviceId: input.target.device_id, kind: "push_notification" });
934 }
892935 }
893936 const result = yield* apns
894937 .sendPushNotificationRequest({
......@@ -1057,6 +1100,7 @@ export const make = Effect.gen(function* () {
10571100 target: input.target,
10581101 aggregate: input.aggregate,
10591102 nowMs: input.nowMs,
1103 replay: input.replay ?? false,
10601104 });
10611105 if (!delivery) {
10621106 return null;
......@@ -1072,17 +1116,20 @@ export const make = Effect.gen(function* () {
10721116 });
10731117 return result;
10741118 }
1075 const notification = notificationForAggregate({
1076 target: input.target,
1077 aggregate: input.aggregate,
1078 nowMs: input.nowMs,
1079 });
1119 const notification = input.replay
1120 ? null
1121 : notificationForAggregate({
1122 target: input.target,
1123 aggregate: input.aggregate,
1124 nowMs: input.nowMs,
1125 });
10801126 // The end event doubles as the "task finished" moment. When a companion
10811127 // push notification is about to ring the device (below), the activity end
10821128 // stays silent; otherwise the end itself carries the alert so LA-only
10831129 // users still get the buzz.
1084 const alert =
1085 delivery.kind === "live_activity_end"
1130 const alert = input.replay
1131 ? null
1132 : delivery.kind === "live_activity_end"
10861133 ? notification && input.target.push_token
10871134 ? null
10881135 : alertForTerminalAggregate({