Reduce email retry counts; handle known unretriable emails

.changeset/email-retry-policy.md

@@ -0,0 +1,5 @@
+---
+'hive': minor
+---
+
+Emails are handed to the mail provider at most 5 times per job, about 90 seconds of retries, instead of graphile-worker's default of 25 attempts spread over days; the links they carry expire well before that. A refusal of the message itself ends the job at once: Postmark rejecting the recipient or the message (an invalid or inactive recipient, a malformed message) or an SMTP server's permanent reply to the recipient or the message (5xx, except 503). Such an attempt is recorded as the job's last one, so the invitations table and the Email Deliveries page show "attempt 1 of 1" and that Hive has stopped retrying, rather than promising retries that will not come. Rate limits, provider outages and transport failures are still retried, and so is a refusal of Hive's own account, credentials or message stream: that misconfiguration fails the job after its retries and shows up in the job error metrics.

integration-tests/tests/api/organization/email-deliveries.spec.ts

@@ -139,7 +139,7 @@ test.concurrent(
       status_code: 200,
       status_text: 'OK',
       attempt: 1,
-      max_attempts: 25,
+      max_attempts: 5,
       metadata: {
         subject: `You have been invited to join ${organization.slug}`,
         job_id: expect.any(String),
@@ -152,24 +152,27 @@ test.concurrent(
       acceptedInvitation.ok!.createdOrganizationInvitation.code,
     );
 
-    // The refused address is retried, so there may be more than one row by now.
-    expect(rows.filter(row => row.destination === refused)).toContainEqual({
-      delivery_type: 'EMAIL',
-      destination: refused,
-      event: 'organization_invitation',
-      idempotency_key: `organization-invitation:${refusedInvitation.ok!.createdOrganizationInvitation.createdAt}:${refused}`,
-      ok: 0,
-      status_code: 422,
-      status_text:
-        'You tried to send to a recipient that has been marked as inactive. (error code 406)',
-      attempt: 1,
-      max_attempts: 25,
-      metadata: {
-        subject: `You have been invited to join ${organization.slug}`,
-        job_id: expect.any(String),
-        hive_request_id: expect.any(String),
+    // An inactive recipient cannot change, so the job gives up after this attempt: the row
+    // records it as the last one, and no further attempt follows.
+    expect(rows.filter(row => row.destination === refused)).toEqual([
+      {
+        delivery_type: 'EMAIL',
+        destination: refused,
+        event: 'organization_invitation',
+        idempotency_key: `organization-invitation:${refusedInvitation.ok!.createdOrganizationInvitation.createdAt}:${refused}`,
+        ok: 0,
+        status_code: 422,
+        status_text:
+          'You tried to send to a recipient that has been marked as inactive. (error code 406)',
+        attempt: 1,
+        max_attempts: 1,
+        metadata: {
+          subject: `You have been invited to join ${organization.slug}`,
+          job_id: expect.any(String),
+          hive_request_id: expect.any(String),
+        },
       },
-    });
+    ]);
 
     const result = await execute({
       document: InvitationDeliveriesQuery,
@@ -180,7 +183,7 @@ test.concurrent(
 
     expect(invitations).toContainEqual({
       email: accepted,
-      delivery: { ok: true, statusCode: 200, statusText: 'OK', attempt: 1, maxAttempts: 25 },
+      delivery: { ok: true, statusCode: 200, statusText: 'OK', attempt: 1, maxAttempts: 5 },
     });
     expect(invitations).toContainEqual({
       email: refused,
@@ -189,8 +192,8 @@ test.concurrent(
         statusCode: 422,
         statusText:
           'You tried to send to a recipient that has been marked as inactive. (error code 406)',
-        attempt: expect.any(Number),
-        maxAttempts: 25,
+        attempt: 1,
+        maxAttempts: 1,
       },
     });
 
@@ -213,10 +216,8 @@ test.concurrent(
       ],
       isLimited: false,
     });
-    // The refused address keeps being retried, so only the accepted count is exact.
     const stats = deliveries.emailDeliveryStats!;
-    expect(stats.totalFailures).toBeGreaterThanOrEqual(1);
-    expect(stats.totalDeliveries - stats.totalFailures).toBe(1);
+    expect(stats).toEqual({ totalDeliveries: 2, totalFailures: 1 });
     expect(deliveries.emailDeliveryHealth).toEqual({
       lastSuccessAt: expect.any(String),
       lastFailureAt: expect.any(String),
@@ -227,7 +228,7 @@ test.concurrent(
       ok: true,
       statusCode: 200,
       attempt: 1,
-      maxAttempts: 25,
+      maxAttempts: 5,
     });
     expect(deliveries.emailDeliveryLog).toContainEqual(
       expect.objectContaining({ destination: refused, ok: false, statusCode: 422 }),
@@ -330,16 +331,17 @@ test.concurrent(
     expect(secondKey).not.toBe(firstKey);
     await pollFor(async () => (await countAttempts(secondKey)) > 0);
 
-    // The first invitation's job keeps retrying for hours after the re-invite, on an exponential
-    // backoff. This is the row it writes later: newer than anything the new job has recorded.
+    // A job that was refused for a transient reason keeps retrying for a while after the
+    // re-invite. This is the row the first invitation's job would write later: newer than anything
+    // the new job has recorded.
     await clickHouseInsert(`
       INSERT INTO "outbound_deliveries" (
         "timestamp", "organization_id", "delivery_type", "destination", "event", "idempotency_key",
         "attempt", "max_attempts", "ok", "status_code", "status_text", "duration_ms", "metadata"
       )
       VALUES (
         now() + toIntervalMinute(1), '${organization.id}', 'EMAIL', '${refused}',
-        'organization_invitation', '${firstKey}', 9, 25, 0, 422, 'Retry of the earlier invitation',
+        'organization_invitation', '${firstKey}', 3, 5, 0, 503, 'Service Unavailable',
         5, map('subject', 'You have been invited to join ${organization.slug}')
       )
     `);
@@ -359,8 +361,8 @@ test.concurrent(
           statusCode: 422,
           statusText:
             'You tried to send to a recipient that has been marked as inactive. (error code 406)',
-          attempt: expect.any(Number),
-          maxAttempts: 25,
+          attempt: 1,
+          maxAttempts: 1,
         },
       },
     ]);

packages/services/api/src/modules/deliveries/module.graphql.ts

@@ -72,7 +72,9 @@ export default gql`
     """
     attempt: Int!
     """
-    The retry budget of the job that made this attempt.
+    The retry budget of the job that made this attempt. When the sender gave the message up after
+    this attempt, because the refusal could not change, this is the attempt's own number: attempt
+    equals maxAttempts and nothing follows.
     """
     maxAttempts: Int!
     """

packages/services/workflows/src/kit.spec.ts

@@ -1,14 +1,23 @@
 import type { JobHelpers } from 'graphile-worker';
 import { z } from 'zod';
 import type { Context } from './context.js';
-import { defineTask, implementTask, type TaskDefinition } from './kit.js';
+import { defineTask, implementTask, TaskScheduler, type TaskDefinition } from './kit.js';
+import { EMAIL_MAX_ATTEMPTS } from './lib/emails/retry-policy.js';
 import { EvaluateMetricAlertRulesTask } from './tasks/evaluate-metric-alert-rules.js';
 import { FlushTargetTokenLastUsedTask } from './tasks/flush-target-token-last-used.js';
+import { OrganizationInvitationTask } from './tasks/organization-invitation.js';
 import { PurgeExpiredAlertStateLogTask } from './tasks/purge-expired-alert-state-log.js';
 import { PurgeExpiredDedupeKeysTask } from './tasks/purge-expired-dedupe-keys.js';
 import { PurgeExpiredSchemaChecks } from './tasks/purge-expired-schema-checks.js';
 import { PurgeExpiredSchemaRevisions } from './tasks/purge-expired-schema-revisions.js';
 
+const addJob = vi.fn(async () => ({ id: 'job-1' }));
+// Only the worker utils are stubbed; the logger bridge needs the package's own Logger class.
+vi.mock('graphile-worker', async importOriginal => ({
+  ...(await importOriginal<typeof import('graphile-worker')>()),
+  makeWorkerUtils: async () => ({ addJob }),
+}));
+
 const helpers = {
   job: { id: '1', job_queue_id: null, attempts: 1, max_attempts: 25, priority: 0, task_id: 1 },
 } as unknown as JobHelpers;
@@ -49,3 +58,44 @@ describe('implementTask', () => {
     expect(() => task(cronPayload, helpers)).toThrow();
   });
 });
+
+describe('TaskScheduler', () => {
+  function createScheduler() {
+    const pgPool = { getPgPoolCompat: () => ({}) } as any;
+    const logger = { info: vi.fn(), child: () => logger } as any;
+    return new TaskScheduler(pgPool, logger);
+  }
+
+  test('gives a job the attempts its task definition asks for', async () => {
+    addJob.mockClear();
+
+    await createScheduler().scheduleTask(
+      OrganizationInvitationTask,
+      {
+        organizationId: 'o1',
+        organizationName: 'Acme',
+        code: 'code-1',
+        email: 'invitee@example.com',
+        link: 'https://app/join/code-1',
+      },
+      { requestId: 'req-1' },
+    );
+
+    expect(addJob).toHaveBeenCalledWith(
+      'organizationInvitation',
+      expect.objectContaining({ requestId: 'req-1' }),
+      { maxAttempts: EMAIL_MAX_ATTEMPTS },
+    );
+  });
+
+  test("leaves the attempts to graphile's default for a task without a budget", async () => {
+    addJob.mockClear();
+
+    await createScheduler().scheduleTask(
+      defineTask({ name: 'plain', schema: z.object({ id: z.string() }) }),
+      { id: 'x' },
+    );
+
+    expect(addJob).toHaveBeenCalledWith('plain', expect.anything(), { maxAttempts: undefined });
+  });
+});

packages/services/workflows/src/kit.ts

@@ -10,6 +10,8 @@ import type { Context } from './context';
 export type TaskDefinition<TName extends string, TSchema extends z.ZodType> = {
   name: TName;
   schema: TSchema;
+  /** Attempts graphile-worker gives each job of this task; its default is 25. */
+  maxAttempts?: number;
 };
 
 type TaskImplementationArgs<TPayload> = {
@@ -163,10 +165,11 @@ export class TaskScheduler {
       }
     }
 
-    const job = await tools.addJob(taskDefinition.name, {
-      requestId: opts?.requestId,
-      input,
-    });
+    const job = await tools.addJob(
+      taskDefinition.name,
+      { requestId: opts?.requestId, input },
+      { maxAttempts: taskDefinition.maxAttempts },
+    );
 
     this.logger.info(
       {

packages/services/workflows/src/lib/delivery-recorder.spec.ts

@@ -65,6 +65,32 @@ describe('trackDelivery', () => {
     );
   });
 
+  test('records a failure the sender will not retry as the last attempt of the job', async () => {
+    const record = vi.fn(async () => {});
+    const error = new Error('Response code 422 (Unprocessable Entity)');
+
+    await expect(
+      trackDelivery(
+        { record },
+        meta,
+        () => Promise.reject(error),
+        () => ({
+          ...deliveryOutcome(422, 'Unprocessable Entity'),
+          final: true,
+        }),
+      ),
+    ).rejects.toBe(error);
+
+    expect(record).toHaveBeenCalledWith(
+      expect.objectContaining({
+        ok: false,
+        statusCode: 422,
+        attempt: meta.attempt,
+        maxAttempts: meta.attempt,
+      }),
+    );
+  });
+
   test('masks an error the sender did not recognise', async () => {
     const record = vi.fn(async () => {});
     const error = new TypeError("Cannot read properties of undefined (reading 'id')");

packages/services/workflows/src/lib/delivery-recorder.ts

@@ -22,10 +22,12 @@ export interface DeliveryOutcome {
 /**
  * How a sender reads the errors of the client it uses: what the destination answered, or
  * `Timed out` when nothing did. Null for anything it does not recognise, which `trackDelivery`
- * masks as `Unexpected error`. Each sender exports its own, next to the client whose error shapes
- * it knows.
+ * masks as `Unexpected error`. `final` says the sender will not try again after this answer.
+ * Each sender exports its own, next to the client whose error shapes it knows.
  */
-export type DescribeDeliveryError = (error: unknown) => DeliveryOutcome | null;
+export type DescribeDeliveryError = (
+  error: unknown,
+) => (DeliveryOutcome & { final?: boolean }) | null;
 
 /** An outcome whose text is bounded to what the column holds. */
 export function deliveryOutcome(statusCode: number | null, statusText: string): DeliveryOutcome {
@@ -58,7 +60,10 @@ export interface DeliveryMeta {
   idempotencyKey: string | null;
   /** graphile-worker attempt number of the job making the call. */
   attempt: number;
-  /** The retry budget of that job. */
+  /**
+   * The retry budget of that job. A failure the sender will not retry is recorded with the budget
+   * ending at its own attempt, so a reader sees it was the last.
+   */
   maxAttempts: number;
   /**
    * Exactly what was sent to identify the call: the request headers, the Slack metadata
@@ -163,8 +168,15 @@ export async function trackDelivery<T>(
     });
     return value;
   } catch (error) {
-    const failure = describeError(error) ?? deliveryOutcome(null, UNEXPECTED_ERROR);
-    await recorder.record({ ...meta, ...failure, ok: false, durationMs: Date.now() - startedAt });
+    const { final = false, ...failure }: DeliveryOutcome & { final?: boolean } =
+      describeError(error) ?? deliveryOutcome(null, UNEXPECTED_ERROR);
+    await recorder.record({
+      ...meta,
+      ...failure,
+      maxAttempts: final ? meta.attempt : meta.maxAttempts,
+      ok: false,
+      durationMs: Date.now() - startedAt,
+    });
     throw error;
   }
 }

packages/services/workflows/src/lib/emails/retry-policy.spec.ts

@@ -0,0 +1,48 @@
+import { EmailSendError } from './email-send-error.js';
+import { isPermanentRefusal } from './retry-policy.js';
+
+function refusal(
+  provider: 'postmark' | 'mock' | 'smtp' | 'sendmail',
+  statusCode: number | null,
+  errorCode: string | null = null,
+): EmailSendError {
+  return new EmailSendError('refused', { provider, statusCode, statusText: 'refused', errorCode });
+}
+
+describe('isPermanentRefusal', () => {
+  test.each<[string, EmailSendError, boolean]>([
+    [
+      'Postmark rejects the request, e.g. an invalid recipient',
+      refusal('postmark', 422, '300'),
+      true,
+    ],
+    ['Postmark rejects an inactive recipient', refusal('postmark', 422, '406'), true],
+    ['the mock provider mirrors that', refusal('mock', 422, '406'), true],
+    ['the Postmark account is pending approval', refusal('postmark', 422, '412'), false],
+    ['the Postmark account may not send', refusal('postmark', 422, '413'), false],
+    ['the Postmark message stream does not exist', refusal('postmark', 422, '1235'), false],
+    ['Postmark rejects the request without saying why', refusal('postmark', 422), false],
+    ['Postmark rate-limits the call', refusal('postmark', 429), false],
+    ['Postmark fails on its side', refusal('postmark', 500), false],
+    ['Postmark is down for maintenance', refusal('postmark', 503), false],
+    ['an SMTP server rejects the recipient for good', refusal('smtp', 550, 'EENVELOPE'), true],
+    ['an SMTP server rejects the message for good', refusal('smtp', 554, 'EMESSAGE'), true],
+    ['an SMTP server rejects the credentials', refusal('smtp', 535, 'EAUTH'), false],
+    ['an SMTP server requires authentication', refusal('smtp', 530, 'EAUTH'), false],
+    [
+      'an SMTP server refuses the session in its greeting',
+      refusal('smtp', 554, 'EPROTOCOL'),
+      false,
+    ],
+    ['an SMTP server asks to try later', refusal('smtp', 451, 'EENVELOPE'), false],
+    [
+      'an SMTP server only complains about the command order',
+      refusal('smtp', 503, 'EENVELOPE'),
+      false,
+    ],
+    ['sendmail relays a permanent SMTP reply', refusal('sendmail', 550), true],
+    ['a refusal without a code', refusal('sendmail', null), false],
+  ])('%s', (_name, error, expected) => {
+    expect(isPermanentRefusal(error)).toBe(expected);
+  });
+});