Repair idempotent dispatch failures (#51)

## Summary

- re-dispatch stored `queued` and `scheduled` emails when an identical
idempotent request is replayed
- preserve the original email ID and stored schedule during recovery
- rely on the existing atomic send lease and final-state checks to
collapse duplicate SQS deliveries
- document the ambiguous-acceptance recovery procedure and at-least-once
boundary

## Root cause

Email persistence and SQS/EventBridge dispatch are separate operations.
If persistence succeeded and dispatch then failed, the API returned an
error but retained the email and idempotency claim. An identical retry
returned that stored record without restoring an immediate or
short-delay job, so the email could remain queued forever.

## Impact

Clients can safely retry the identical payload with the same idempotency
key after an ambiguous API failure. HayaSend returns the same ID and
restores dispatch for eligible non-final records. Replays never
re-dispatch `sending`, canceled, suppressed, or terminal records.

This remains an immediate risk-reduction bridge. The accepted #81/#84
direction replaces client-triggered repair with durable outbox
reconciliation in the provider-neutral semantics series.

## Validation

- `npm ci`
- `npm run check`
- `npm test` — 22 files, 170 tests
- `npm run build`
- `npm run lint:openapi`
- `npm audit --omit=dev --audit-level=high` — 0 vulnerabilities
- `npm outdated --json` — no direct dependency updates available
- `git diff --check`

Closes #50

Signed-off-by: Yusuke Hayashi <yusuke8h@gmail.com>

Author: yhay81

CHANGELOG.md

@@ -8,6 +8,8 @@ minor releases before v1.0.
 
 - Stop retrying permanent SES request rejections while preserving retries for
   throttling, provider availability, network, timeout, and unknown failures.
+- Repair missing SQS or EventBridge dispatch when an identical idempotent
+  replay finds a stored queued or scheduled email.
 
 ## 0.1.0 - 2026-07-26
 

docs/architecture.md

@@ -41,6 +41,12 @@ Scheduler names are derived from email IDs. Rescheduling replaces the same
 one-time schedule, cancellation deletes it, and stale SQS deliveries reload
 the current DynamoDB record before doing any work.
 
+If persistence succeeds but SQS or Scheduler dispatch fails, an identical
+request with the same idempotency key re-dispatches the stored non-final email
+using its stored schedule. This can create a duplicate SQS job when the first
+dispatch actually succeeded but its response was lost. The worker lease and
+final-state check are designed for that at-least-once condition.
+
 There is still an unavoidable narrow failure window if SES accepts a message
 and the worker stops before recording the provider ID. A later retry can
 produce a duplicate. HayaSend documents this at-least-once boundary instead

docs/operations.md

@@ -142,6 +142,19 @@ If a received email is visible in S3 but absent from the API:
 Never paste the raw MIME into logs or issue trackers. Use object identifiers
 and request IDs during diagnosis.
 
+## Ambiguous send acceptance
+
+If `POST /emails` or `POST /emails/batch` fails after the request reached
+HayaSend, retry the identical payload with the same idempotency key. HayaSend
+returns the original email ID and re-dispatches a stored `queued` or
+`scheduled` record, repairing a failure between persistence and SQS or
+Scheduler acceptance. A replay can create a duplicate SQS job, which the
+worker lease and final-state check normally collapse.
+
+Do not change the payload or generate a new key during this recovery. A
+different payload conflicts with the existing claim, while a new key creates
+a different email and can result in two deliveries.
+
 ## Dead-letter queue
 
 1. Pause the producer if the failure is systemic.

src/services/email-service.ts

@@ -242,9 +242,9 @@ export class EmailService {
         await this.webhooks.publish("email.scheduled", record);
       }
     } else if (
-      created.record.status === "scheduled" &&
-      created.record.scheduled_at &&
-      secondsUntil(created.record.scheduled_at, now) > 900
+      created.record.status === "queued" ||
+      (created.record.status === "scheduled" &&
+        created.record.scheduled_at !== undefined)
     ) {
       await this.scheduler.schedule(
         created.record.id,

tests/app.test.ts

@@ -504,7 +504,7 @@ describe("HTTP API", () => {
     });
   });
 
-  it("replays identical idempotent requests without a second job", async () => {
+  it("replays identical idempotent requests with the same id and a repair job", async () => {
     const { queue, request } = fixture();
     const init = {
       method: "POST",
@@ -518,7 +518,7 @@ describe("HTTP API", () => {
       id: string;
     };
     expect(second.id).toBe(first.id);
-    expect(queue.jobs).toHaveLength(1);
+    expect(queue.jobs).toHaveLength(2);
 
     const conflict = await request("/emails", {
       ...init,

tests/email-service.test.ts

@@ -3,7 +3,7 @@ import { MemoryStore } from "../src/adapters/memory-store.js";
 import { MemoryAttachmentStorage } from "../src/adapters/attachment-storage.js";
 import { QueueEmailScheduler } from "../src/adapters/email-scheduler.js";
 import { CapturingJobQueue } from "../src/adapters/sqs-job-queue.js";
-import type { EmailRecord } from "../src/core/types.js";
+import type { EmailRecord, EmailStatus } from "../src/core/types.js";
 import type {
   MailTransport,
   MailTransportResult,
@@ -35,6 +35,7 @@ class RecordingEmailScheduler implements EmailScheduler {
     emailId: string;
     scheduledAt: string;
   }> = [];
+  scheduleFailures = 0;
   onFirstReschedule?: () => Promise<void>;
 
   constructor(private readonly queueScheduler: QueueEmailScheduler) {}
@@ -44,6 +45,10 @@ class RecordingEmailScheduler implements EmailScheduler {
     scheduledAt?: string,
     now?: Date,
   ): Promise<void> {
+    if (this.scheduleFailures > 0) {
+      this.scheduleFailures -= 1;
+      throw new Error("temporary scheduler failure");
+    }
     await this.queueScheduler.schedule(emailId, scheduledAt, now);
   }
 
@@ -125,7 +130,7 @@ describe("EmailService", () => {
   });
 
   it("keeps a template send idempotent across later publications", async () => {
-    const { queue, service, templates } = fixture();
+    const { queue, service, templates, transport } = fixture();
     const template = await templates.create({
       name: "Idempotent template",
       from: "sender@example.com",
@@ -154,6 +159,116 @@ describe("EmailService", () => {
         html: "<p>One</p>",
       },
     });
+    expect(queue.jobs).toHaveLength(2);
+
+    await Promise.all([
+      service.processSend(first.record.id),
+      service.processSend(first.record.id),
+    ]);
+    expect(transport.sent).toHaveLength(1);
+  });
+
+  it("repairs an immediate dispatch failure through idempotent replay", async () => {
+    const { queue, scheduler, service, store } = fixture();
+    scheduler.scheduleFailures = 1;
+
+    await expect(
+      service.create(input, "repair-immediate"),
+    ).rejects.toThrow("temporary scheduler failure");
+    const storedAfterFailure = await store.listEmails(100);
+    expect(storedAfterFailure.data).toHaveLength(1);
+    expect(storedAfterFailure.data[0]).toMatchObject({
+      status: "queued",
+      attempts: 0,
+    });
+    expect(queue.jobs).toHaveLength(0);
+
+    const replay = await service.create(input, "repair-immediate");
+
+    expect(replay).toMatchObject({
+      replayed: true,
+      record: {
+        id: storedAfterFailure.data[0]?.id,
+        status: "queued",
+      },
+    });
+    expect(queue.jobs).toEqual([
+      {
+        job: {
+          type: "send_email",
+          email_id: storedAfterFailure.data[0]?.id,
+        },
+        delaySeconds: 0,
+      },
+    ]);
+    await expect(store.listEmails(100)).resolves.toMatchObject({
+      data: [{ id: storedAfterFailure.data[0]?.id }],
+    });
+  });
+
+  it("repairs a short scheduled dispatch with its stored delay", async () => {
+    const { queue, scheduler, service, store } = fixture();
+    const now = new Date("2026-07-26T00:00:00.000Z");
+    const scheduledAt = "2026-07-26T00:10:00.000Z";
+    const scheduledInput = { ...input, scheduled_at: scheduledAt };
+    scheduler.scheduleFailures = 1;
+
+    await expect(
+      service.create(scheduledInput, "repair-short-schedule", now),
+    ).rejects.toThrow("temporary scheduler failure");
+    const storedAfterFailure = await store.listEmails(100);
+    expect(storedAfterFailure.data).toHaveLength(1);
+    expect(queue.jobs).toHaveLength(0);
+
+    const replay = await service.create(
+      scheduledInput,
+      "repair-short-schedule",
+      now,
+    );
+
+    expect(replay).toMatchObject({
+      replayed: true,
+      record: {
+        id: storedAfterFailure.data[0]?.id,
+        status: "scheduled",
+        scheduled_at: scheduledAt,
+      },
+    });
+    expect(queue.jobs).toEqual([
+      {
+        job: {
+          type: "send_email",
+          email_id: storedAfterFailure.data[0]?.id,
+        },
+        delaySeconds: 600,
+      },
+    ]);
+  });
+
+  it.each<EmailStatus>([
+    "sending",
+    "sent",
+    "delivered",
+    "delivery_delayed",
+    "opened",
+    "clicked",
+    "bounced",
+    "complained",
+    "failed",
+    "canceled",
+    "suppressed",
+  ])("does not redispatch a replayed %s record", async (status) => {
+    const { queue, service, store } = fixture();
+    const created = await service.create(input, `final-${status}`);
+    expect(queue.jobs).toHaveLength(1);
+    await store.updateEmail(created.record.id, {
+      status,
+      last_event: status,
+    });
+
+    const replay = await service.create(input, `final-${status}`);
+
+    expect(replay.replayed).toBe(true);
     expect(queue.jobs).toHaveLength(1);
   });