fix: handle mid-run workflow cancellation, graceful server shutdown ordering, and profile attribute merging

src/repositories/index.ts

@@ -24,6 +24,13 @@ import {
 
 const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
 
+/**
+ * Applied when a profile is read, not when it is written: an upsert that
+ * leaves a field out must keep the stored value rather than reset it.
+ */
+const DEFAULT_LANGUAGE = "en";
+const DEFAULT_TIMEZONE = "UTC";
+
 // ─── Domain types ───────────────────────────────────────────────────────────
 
 export interface UserProfile {
@@ -90,8 +97,8 @@ export class UserRepository {
     const attrs = rows[0].attributes as any;
     return {
       userId: rows[0].externalId,
-      language: attrs.language,
-      timezone: attrs.timezone,
+      language: attrs.language ?? DEFAULT_LANGUAGE,
+      timezone: attrs.timezone ?? DEFAULT_TIMEZONE,
       email: attrs.email,
     };
   }
@@ -180,8 +187,8 @@ export class UserRepository {
 
       const record: UserRecord = {
         userId: userRow.externalId,
-        language: attrs.language,
-        timezone: attrs.timezone,
+        language: attrs.language ?? DEFAULT_LANGUAGE,
+        timezone: attrs.timezone ?? DEFAULT_TIMEZONE,
         email: attrs.email,
         segments: segmentsByUserId.get(internalId) || [],
         preferences: {
@@ -338,7 +345,9 @@ export class UserRepository {
             .onConflictDoUpdate({
               target: [users.projectId, users.externalId],
               set: {
-                attributes: drizzleSql`excluded.attributes`,
+                // Merged, not replaced: a field the caller left out (absent
+                // from the JSON) keeps its stored value.
+                attributes: drizzleSql`${users.attributes} || excluded.attributes`,
                 updatedAt: new Date(),
               },
             });
@@ -621,7 +630,8 @@ export class UserRepository {
     }
 
     if (filters?.search) {
-      const term = `%${filters.search.trim()}%`;
+      // `%`, `_` and `\` in the search are literal, not wildcards.
+      const term = `%${filters.search.trim().replace(/[\\%_]/g, "\\$&")}%`;
       conditions.push(
         drizzleSql`(${users.externalId} ILIKE ${term} OR (${users.attributes}->>'email') ILIKE ${term})`,
       );
@@ -650,8 +660,8 @@ export class UserRepository {
       const attrs = r.attributes as any;
       return {
         userId: r.externalId,
-        language: attrs.language,
-        timezone: attrs.timezone,
+        language: attrs.language ?? DEFAULT_LANGUAGE,
+        timezone: attrs.timezone ?? DEFAULT_TIMEZONE,
         email: attrs.email,
         createdAt: r.createdAt.getTime(),
       };

src/server.ts

@@ -273,44 +273,26 @@ export class NotifkitServer extends EventEmitter {
       ? ["api", "delivery", "engine", "enricher", "scheduler", "ai", "workflow", "events"]
       : this.options.services;
 
-    if (services.includes("api")) {
-      const { stopApiServer } = await import("./services/api/main.js");
-      await stopApiServer();
-    }
-
-    if (services.includes("delivery")) {
-      const { stopDeliveryWorker } = await import("./services/delivery/main.js");
-      await stopDeliveryWorker();
-    }
-
-    if (services.includes("engine")) {
-      const { stopEngineWorker } = await import("./services/engine/main.js");
-      await stopEngineWorker();
-    }
-
-    if (services.includes("enricher")) {
-      const { stopEnricherWorker } = await import("./services/enricher/main.js");
-      await stopEnricherWorker();
-    }
-
-    if (services.includes("scheduler")) {
-      const { stopSchedulerWorker } = await import("./services/scheduler/main.js");
-      await stopSchedulerWorker();
-    }
-
-    if (services.includes("ai")) {
-      const { stopAiWorker } = await import("./services/ai/main.js");
-      await stopAiWorker();
-    }
-
-    if (services.includes("workflow")) {
-      const { stopWorkflowWorker } = await import("./services/workflow/main.js");
-      await stopWorkflowWorker();
-    }
-
-    if (services.includes("events")) {
-      const { stopEventWorker } = await import("./services/events/main.js");
-      await stopEventWorker();
+    // Upstream first. With the pipeline fused, the enricher's task runs the
+    // engine and delivery inline, so delivery's connections must outlive it;
+    // stopping delivery first failed those sends mid-flight, after the
+    // provider may already have accepted them. Events goes last because it
+    // records what delivery did.
+    const shutdownOrder: [string, () => Promise<void>][] = [
+      ["api", async () => (await import("./services/api/main.js")).stopApiServer()],
+      ["workflow", async () => (await import("./services/workflow/main.js")).stopWorkflowWorker()],
+      ["enricher", async () => (await import("./services/enricher/main.js")).stopEnricherWorker()],
+      ["ai", async () => (await import("./services/ai/main.js")).stopAiWorker()],
+      ["engine", async () => (await import("./services/engine/main.js")).stopEngineWorker()],
+      [
+        "scheduler",
+        async () => (await import("./services/scheduler/main.js")).stopSchedulerWorker(),
+      ],
+      ["delivery", async () => (await import("./services/delivery/main.js")).stopDeliveryWorker()],
+      ["events", async () => (await import("./services/events/main.js")).stopEventWorker()],
+    ];
+    for (const [name, stop] of shutdownOrder) {
+      if (services.includes(name)) await stop();
     }
 
     if (this.pgContainer) {

src/services/api/handlers.ts

@@ -138,13 +138,15 @@ function contactsOf(user: InlineUserLike): ResolvedContact[] {
 /** Persist user records + their contacts in bulk. Shared by addUser and inline notify. */
 async function persistUsers(deps: Deps, users: InlineUserLike[], projectId: string): Promise<void> {
   const usersList = users.map((u) => {
-    const primaryEmail =
-      u.email?.[0] ?? u.contacts?.find((c) => c.channel === "email")?.target ?? null;
+    const primaryEmail = u.email?.[0] ?? u.contacts?.find((c) => c.channel === "email")?.target;
 
+    // Omitted fields stay omitted: the upsert merges what is given into the
+    // stored profile, so an inline `{ id, phone }` on notify must not reset the
+    // user's timezone, language or email. Defaults are applied when read.
     return {
       userId: u.id,
-      language: u.language ?? "en",
-      timezone: u.timezone ?? "UTC",
+      language: u.language,
+      timezone: u.timezone,
       email: primaryEmail,
       segments: u.segments ?? [],
       preferences: u.preferences ?? {},
@@ -255,7 +257,7 @@ function encodeLogCursor(ts: string, id: string): string {
 function decodeLogCursor(cursor: string | null | undefined): { ts: string; id: string } | null {
   if (!cursor) return null;
   // Cursors handed out before this format were a millisecond timestamp.
-  if (/^d+$/.test(cursor)) {
+  if (/^\d+$/.test(cursor)) {
     return {
       ts: new Date(Number(cursor)).toISOString(),
       id: "ffffffff-ffff-ffff-ffff-ffffffffffff",
@@ -277,6 +279,14 @@ function decodeLogCursor(cursor: string | null | undefined): { ts: string; id: s
   return null;
 }
 
+/**
+ * A LIKE pattern matching `term` anywhere, with the term's own `%`, `_` and
+ * `\` taken literally rather than as wildcards.
+ */
+function contains(term: string): string {
+  return `%${term.replace(/[\\%_]/g, "\\$&")}%`;
+}
+
 function getQueryParam(ctx: RouteContext, key: string): string | undefined {
   if (!ctx.query) return undefined;
   if (typeof ctx.query.get === "function") {
@@ -663,6 +673,15 @@ export function createHandlers(deps: Deps) {
       const eventTime = new Date(tsStr as string).getTime();
       const expMs = parseInt(expiryStr as string, 10) * 1000;
 
+      // An unreadable timestamp or expiry made every comparison false, which
+      // let the event through unchecked.
+      if (Number.isNaN(eventTime) || Number.isNaN(expMs)) {
+        sendJson(res, 400, {
+          error: "invalid_timestamp",
+          message: "x-timestamp and x-expiry must be a date and a number of seconds",
+        });
+        return;
+      }
       if (now > eventTime + expMs) {
         sendJson(res, 400, { error: "event_expired", message: "Webhook event is expired" });
         return;
@@ -754,10 +773,10 @@ export function createHandlers(deps: Deps) {
     if (search) {
       conditions.push(
         or(
-          like(messageLogs.taskId, `%${search}%`),
-          like(messageLogs.templateId, `%${search}%`),
-          like(messageLogs.providerMessageId, `%${search}%`),
-          like(messageLogs.campaignId, `%${search}%`),
+          like(messageLogs.taskId, contains(search)),
+          like(messageLogs.templateId, contains(search)),
+          like(messageLogs.providerMessageId, contains(search)),
+          like(messageLogs.campaignId, contains(search)),
         )!,
       );
     }
@@ -1195,7 +1214,7 @@ export function createHandlers(deps: Deps) {
     if (!deleted) return sendJson(res, 404, { error: "key_not_found" });
 
     await deps.redis.native.publish("apikey.invalidated", "*");
-    sendJson(res, 204, {});
+    sendNoContent(res);
   }
 
   // ── GET /v1/system/health — getSystemHealth ──────────────────────────────
@@ -1785,7 +1804,7 @@ code{background:#f4f4f5;padding:.1rem .35rem;border-radius:4px}</style>
     ];
 
     if (search) {
-      conditions.push(like(messageLogs.campaignId, `%${search}%`));
+      conditions.push(like(messageLogs.campaignId, contains(search)));
     }
     if (channel) {
       conditions.push(eq(messageLogs.channel, channel as any));
@@ -1958,7 +1977,7 @@ code{background:#f4f4f5;padding:.1rem .35rem;border-radius:4px}</style>
     const conditions = [eq(suppressions.projectId, ctx.projectId!)];
     if (channel) conditions.push(eq(suppressions.channel, channel as any));
     if (reason) conditions.push(eq(suppressions.reason, reason));
-    if (target) conditions.push(like(suppressions.target, `%${target.toLowerCase().trim()}%`));
+    if (target) conditions.push(like(suppressions.target, contains(target.toLowerCase().trim())));
 
     const rows = await deps.db
       .select()

src/services/api/main.ts

@@ -396,7 +396,7 @@ export async function startApiServer() {
     res.setHeader("Access-Control-Allow-Methods", "GET, POST, PUT, PATCH, DELETE, OPTIONS");
     res.setHeader(
       "Access-Control-Allow-Headers",
-      "Content-Type, Authorization, x-api-key, x-project-id",
+      "Content-Type, Authorization, x-api-key, x-project-id, x-idempotency-key",
     );
 
     if (req.method === "OPTIONS") {

src/services/delivery/main.ts

@@ -82,6 +82,14 @@ const DISPATCH_LOG_DELAY_MS = 1_000;
  */
 const RETRY_BACKOFF_MS = [30_000, 120_000, 600_000];
 
+/**
+ * How many times a send may be put back for want of provider capacity before
+ * it is failed. Separate from `maxAttempts`, which counts sends the provider
+ * refused: waiting in line is not a failed attempt, and a campaign several
+ * windows deeper than the provider's limit has to be able to wait that long.
+ */
+export const MAX_THROTTLE_DEFERRALS = 50;
+
 /**
  * Names the provider at `transports[i]` for its rate limit and circuit breaker.
  * The rate limit's budget lives in Redis and is shared by every delivery
@@ -390,6 +398,7 @@ export class DeliveryWorker extends BaseWorker {
     // channel is out of capacity only once every provider for it is.
     let startAt = -1;
     let retryAfterMs = Infinity;
+    let windowMs = Infinity;
     for (let i = 0; i < transports.length; i++) {
       const gate = await this.providerGate(task.channel, transports, i);
       if (gate.allowed) {
@@ -397,12 +406,12 @@ export class DeliveryWorker extends BaseWorker {
         break;
       }
       retryAfterMs = Math.min(retryAfterMs, gate.retryAfterMs);
+      windowMs = Math.min(windowMs, (transports[i].limits?.windowSeconds ?? 0) * 1000);
     }
     if (startAt === -1) {
       task.throttleAttemptCount = (task.throttleAttemptCount ?? 0) + 1;
-      const maxAttempts = task.deliveryOptions?.maxAttempts ?? 3;
 
-      if (task.throttleAttemptCount > maxAttempts) {
+      if (task.throttleAttemptCount > MAX_THROTTLE_DEFERRALS) {
         this.logger.warn(
           { messageId: message.id, taskId: task.taskId, attempts: task.throttleAttemptCount },
           "provider rate limit max attempts exceeded",
@@ -441,9 +450,15 @@ export class DeliveryWorker extends BaseWorker {
         return;
       }
 
+      // Spread over more windows the longer a task has waited. Every task turned
+      // away in one window used to come back at the same boundary, where the
+      // limit again let one window's worth through and bounced the rest.
+      const spreadMs = Number.isFinite(windowMs)
+        ? Math.random() * windowMs * Math.min(task.throttleAttemptCount, 10)
+        : 0;
       await this.reschedule(
         { ...original, throttleAttemptCount: task.throttleAttemptCount },
-        new Date(Date.now() + retryAfterMs),
+        new Date(Date.now() + retryAfterMs + spreadMs),
         "throttle",
       );
       return;

src/services/enricher/main.ts

@@ -149,6 +149,10 @@ export class EnricherWorker extends BaseWorker {
   }
 
   override async stop(): Promise<void> {
+    // The timers keep running until in-flight messages settle: each of them is
+    // waiting on one of these flushes, and clearing them first left the drain
+    // to the 30s stop timeout.
+    await super.stop();
     if (this.batchTimer) {
       clearTimeout(this.batchTimer);
       this.batchTimer = null;
@@ -164,7 +168,6 @@ export class EnricherWorker extends BaseWorker {
     await this.flushContactBatch();
     await this.flushUserBatch();
     await this.flushWorkerBuffers();
-    await super.stop();
   }
 
   private async flushWorkerBuffers(): Promise<void> {

src/services/workflow/main.ts

@@ -45,6 +45,39 @@ export const WORKFLOW_TIMERS_KEY = "notif:workflow:timers";
 const WORKFLOW_RESUME_RETRY_MS = 2_000;
 const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
 
+/** Seconds per unit. Everything is read as seconds first, then scheduled in ms. */
+const DURATION_UNIT_SECONDS: Record<string, number> = {
+  ms: 0.001,
+  s: 1,
+  m: 60,
+  h: 3_600,
+  d: 86_400,
+};
+
+/**
+ * `"500ms"`, `"0.5s"`, `"15m"`, `"1.5h"`, `"2.5d"` to seconds. Decimals are
+ * read exactly; anything that is not a number and a unit throws. The old
+ * parser matched on the last character, so "500ms" meant 500 seconds, and ran
+ * parseInt, so "1.5h" meant one hour.
+ */
+export function parseDurationSeconds(value: string, what: string): number {
+  const match = /^(\d+(?:\.\d+)?|\.\d+)(ms|s|m|h|d)$/.exec(value.trim());
+  if (!match) throw new Error(`Invalid ${what}: ${value}`);
+  return Number(match[1]) * DURATION_UNIT_SECONDS[match[2]!]!;
+}
+
+function durationToMs(value: string, what: string): number {
+  return Math.round(parseDurationSeconds(value, what) * 1000);
+}
+
+/** Thrown at a step boundary once the instance has been canceled. */
+class WorkflowCanceledError extends Error {
+  constructor() {
+    super("Workflow instance was canceled");
+    this.name = "WorkflowCanceledError";
+  }
+}
+
 loadEnv();
 const config = readBaseConfig();
 let logger: Logger;
@@ -94,12 +127,15 @@ export class WorkflowWorker extends BaseWorker {
   }
 
   override async stop(): Promise<void> {
+    // The flush interval keeps running until in-flight runs settle: a step's
+    // notify is waiting on it, and clearing it first left the drain to the 30s
+    // stop timeout with the instance lock still being renewed.
+    await super.stop();
     if (this.flushTimer) {
       clearInterval(this.flushTimer);
       this.flushTimer = null;
     }
     await this.flushWorkerBuffers();
-    await super.stop();
   }
 
   private async flushWorkerBuffers(): Promise<void> {
@@ -133,7 +169,10 @@ export class WorkflowWorker extends BaseWorker {
    * Re-delivers a resume through the timer ZSET a few seconds from now. The
    * poll loop publishes it once the instance is pending again, and drops it if
    * the instance has finished. The member carries no timestamp, so repeated
-   * deferrals of one instance collapse into a single entry.
+   * deferrals of one instance collapse into a single entry. It is marked
+   * `deferred` so it never equals a `wait` timer's member: sharing one let the
+   * ZADD here re-score a sleep's wake-up to seconds from now, and the early
+   * resume then re-suspended with no timer left to wake it.
    */
   private async deferResume(payload: any): Promise<void> {
     await this.redisCli.zadd(
@@ -144,6 +183,7 @@ export class WorkflowWorker extends BaseWorker {
         name: payload.name,
         projectId: payload.projectId,
         input: payload.input ?? {},
+        deferred: true,
       }),
     );
   }
@@ -271,10 +311,28 @@ export class WorkflowWorker extends BaseWorker {
         return;
       }
 
-      await this.dbConn
+      // Conditional, so a cancel that lands between the read above and here
+      // is not overwritten.
+      const claimed = await this.dbConn
         .update(workflowInstances)
         .set({ status: "running" })
-        .where(eq(workflowInstances.id, instanceId));
+        .where(and(eq(workflowInstances.id, instanceId), eq(workflowInstances.status, "pending")))
+        .returning({ id: workflowInstances.id });
+      if (claimed.length === 0) {
+        this.logger.info({ instanceId }, "Workflow is no longer pending, skipping");
+        return;
+      }
+
+      // A cancel can arrive while the handler runs. Checked before each new
+      // step, so a canceled instance stops at the next step boundary.
+      const stopIfCanceled = async () => {
+        const [row] = await this.dbConn
+          .select({ status: workflowInstances.status })
+          .from(workflowInstances)
+          .where(eq(workflowInstances.id, instanceId))
+          .limit(1);
+        if (row?.status === "canceled") throw new WorkflowCanceledError();
+      };
 
       // Load existing steps
       const existingSteps = await this.dbConn
@@ -292,6 +350,7 @@ export class WorkflowWorker extends BaseWorker {
         notify: async (args) => {
           const stepId = String(currentStepIndex++);
           if (stepOutputMap.has(stepId)) return stepOutputMap.get(stepId);
+          await stopIfCanceled();
 
           // The step payload is the same shape as notify(), but the wire event
           // is not — translate rather than spread.
@@ -348,14 +407,9 @@ export class WorkflowWorker extends BaseWorker {
             }
             return;
           }
+          await stopIfCanceled();
 
-          // Simple duration parse (e.g. '2h' -> ms)
-          let ms = 0;
-          if (duration.endsWith("d")) ms = parseInt(duration) * 24 * 60 * 60 * 1000;
-          else if (duration.endsWith("h")) ms = parseInt(duration) * 60 * 60 * 1000;
-          else if (duration.endsWith("m")) ms = parseInt(duration) * 60 * 1000;
-          else if (duration.endsWith("s")) ms = parseInt(duration) * 1000;
-          else throw new Error(`Invalid wait duration: ${duration}`);
+          const ms = durationToMs(duration, "wait duration");
 
           const resumeAt = Date.now() + ms;
 
@@ -397,17 +451,13 @@ export class WorkflowWorker extends BaseWorker {
             }
             return out;
           }
+          await stopIfCanceled();
 
           options = options || {};
           options.timeout = options.timeout || "24h";
           options.match = options.match || {};
 
-          let ms = 0;
-          if (options.timeout.endsWith("d")) ms = parseInt(options.timeout) * 24 * 60 * 60 * 1000;
-          else if (options.timeout.endsWith("h")) ms = parseInt(options.timeout) * 60 * 60 * 1000;
-          else if (options.timeout.endsWith("m")) ms = parseInt(options.timeout) * 60 * 1000;
-          else if (options.timeout.endsWith("s")) ms = parseInt(options.timeout) * 1000;
-          else throw new Error(`Invalid waitForEvent timeout: ${options.timeout}`);
+          const ms = durationToMs(options.timeout, "waitForEvent timeout");
 
           const resumeAt = Date.now() + ms;
 
@@ -448,6 +498,7 @@ export class WorkflowWorker extends BaseWorker {
         run: async (stepName, fn) => {
           const stepId = String(currentStepIndex++);
           if (stepOutputMap.has(stepId)) return stepOutputMap.get(stepId);
+          await stopIfCanceled();
 
           const result = await fn();
           await this.dbConn.insert(workflowSteps).values({
@@ -466,27 +517,29 @@ export class WorkflowWorker extends BaseWorker {
         event: (instance!.input as any) || { user: { id: "unknown" } },
       };
 
+      // Only from `running`: a cancel that landed mid-run must stand.
+      const finish = (status: "completed" | "pending" | "failed") =>
+        this.dbConn
+          .update(workflowInstances)
+          .set({ status })
+          .where(
+            and(eq(workflowInstances.id, instance!.id), eq(workflowInstances.status, "running")),
+          );
+
       try {
         await handler(ctx);
         // If we reach here, workflow completed
-        await this.dbConn
-          .update(workflowInstances)
-          .set({ status: "completed" })
-          .where(eq(workflowInstances.id, instance!.id));
+        await finish("completed");
         this.logger.info({ instanceId: instance!.id }, "Workflow completed successfully");
       } catch (err: any) {
-        if (err instanceof SuspendExecutionError || err.name === "SuspendExecutionError") {
-          await this.dbConn
-            .update(workflowInstances)
-            .set({ status: "pending" })
-            .where(eq(workflowInstances.id, instance!.id));
+        if (err instanceof WorkflowCanceledError) {
+          this.logger.info({ instanceId: instance!.id }, "Workflow canceled mid-run, stopping");
+        } else if (err instanceof SuspendExecutionError || err.name === "SuspendExecutionError") {
+          await finish("pending");
           this.logger.info({ instanceId: instance!.id, reason: err.reason }, "Workflow suspended");
         } else {
           this.logger.error({ err, instanceId: instance!.id }, "Workflow failed");
-          await this.dbConn
-            .update(workflowInstances)
-            .set({ status: "failed" })
-            .where(eq(workflowInstances.id, instance!.id));
+          await finish("failed");
         }
       }
     } finally {
@@ -578,7 +631,27 @@ export async function startWorkflowWorker() {
           if (inst.status === "running") continue;
 
           if (task.isEventTimeout) {
-            // It's a timeout for waitForEvent. Clean up only this step's waiter
+            // A timeout for one waitForEvent step. If that step already has an
+            // output the event came first and this timer is stale; acting on it
+            // would time out a later wait on the same event name.
+            if (task.stepId !== undefined) {
+              const [own] = await db
+                .select({ output: workflowSteps.output })
+                .from(workflowSteps)
+                .where(
+                  and(
+                    eq(workflowSteps.instanceId, task.instanceId),
+                    eq(workflowSteps.stepIndex, String(task.stepId)),
+                  ),
+                )
+                .limit(1);
+              if (!own || own.output !== null) {
+                await redis.native.zrem(WORKFLOW_TIMERS_KEY, taskStr);
+                continue;
+              }
+            }
+
+            // Clean up only this step's waiter
             const deleted = await db
               .delete(workflowWaiters)
               .where(
@@ -606,7 +679,10 @@ export async function startWorkflowWorker() {
                   eq(workflowSteps.action, "waitForEvent"),
                 ),
               );
-            const pendingStep = steps.find((s: any) => s.output === null);
+            const pendingStep =
+              task.stepId !== undefined
+                ? steps.find((s: any) => s.stepIndex === String(task.stepId) && s.output === null)
+                : steps.find((s: any) => s.output === null);
             if (pendingStep) {
               await db
                 .update(workflowSteps)

src/shared/events.ts

@@ -1,3 +1,6 @@
 import { EventEmitter } from "node:events";
 
 export const globalEmitter = new EventEmitter();
+// Every open SSE stream on the API listens here, so the count grows with
+// connected clients rather than signalling a leak.
+globalEmitter.setMaxListeners(0);

src/workers/index.ts

@@ -248,7 +248,7 @@ export abstract class BaseWorker {
         );
         globalEmitter.emit(
           "notification:failed",
-          message.id,
+          message.event.id,
           "Poison pill: max retries exceeded",
           message.event.type,
         );
@@ -294,7 +294,7 @@ export abstract class BaseWorker {
         await this.consumer.nack(message.id, message.event, stream, (err as Error).message);
         globalEmitter.emit(
           "notification:failed",
-          message.id,
+          message.event.id,
           (err as Error).message,
           message.event.type,
         );

tests/integration/api.test.ts

@@ -1082,3 +1082,105 @@ describe("admin dashboard", () => {
     }
   });
 });
+
+// ─── Regressions ────────────────────────────────────────────────────────────
+
+describe("regressions", () => {
+  it("keeps a user's stored profile when notify carries only part of it inline", async () => {
+    const p = await project();
+    await p.client.syncTemplates([{ id: "t", channel: "email", content: { subject: "s" } }]);
+    await p.client.addUser({ id: "u", email: "u@x.com", timezone: "Europe/Paris", language: "fr" });
+
+    const res = await p.api("POST", "/v1/notify", {
+      user: { id: "u", phone: ["+15550009"] },
+      template: "t",
+    });
+    expect(res.status).toBe(202);
+
+    const u = await p.client.getUser("u");
+    expect(u).toMatchObject({ timezone: "Europe/Paris", language: "fr", email: "u@x.com" });
+    expect(u.contacts!.map((c) => c.target).sort()).toEqual(["+15550009", "u@x.com"]);
+  });
+
+  it("still gives a brand-new inline user the default language and timezone", async () => {
+    const p = await project();
+    await p.client.syncTemplates([{ id: "t", channel: "email", content: { subject: "s" } }]);
+    await p.api("POST", "/v1/notify", { user: { id: "fresh", email: ["f@x.com"] }, template: "t" });
+    expect(await p.client.getUser("fresh")).toMatchObject({ language: "en", timezone: "UTC" });
+  });
+
+  it("honours a legacy millisecond cursor on the delivery log", async () => {
+    const p = await project();
+    await p.client.syncTemplates([{ id: "t", channel: "email", content: { subject: "s" } }]);
+    await p.client.addUser({ id: "u", email: "u@x.com" });
+    await p.client.notify({ user: "u", template: "t" });
+    await waitFor(
+      "logged",
+      async () => (await p.client.getNotificationLogs({ status: "delivered" })).logs.length === 1,
+    );
+    // A cursor from 1970 is older than every row, so the page is empty.
+    const res = await p.api("GET", "/v1/notifications/logs?cursor=1");
+    expect(res.body.logs).toEqual([]);
+  });
+
+  it("lets a browser send the idempotency key the API reads", async () => {
+    const res = await fetch(`${app.baseUrl}/v1/notify`, { method: "OPTIONS" });
+    expect(res.headers.get("access-control-allow-headers")?.toLowerCase()).toContain(
+      "x-idempotency-key",
+    );
+  });
+
+  it("answers a key deletion with an empty 204", async () => {
+    const p = await project();
+    const key = await app.admin("POST", `/v1/projects/${p.id}/keys`, {});
+    const res = await fetch(`${app.baseUrl}/v1/projects/${p.id}/keys/${key.body.id}`, {
+      method: "DELETE",
+      headers: { authorization: `Bearer ${ADMIN_KEY}` },
+    });
+    expect(res.status).toBe(204);
+    expect(res.headers.get("content-length") ?? "0").toBe("0");
+  });
+
+  it("refuses an event whose timestamp cannot be read", async () => {
+    const p = await project();
+    const res = await p.api(
+      "POST",
+      "/v1/events",
+      { name: "e", properties: {} },
+      { "x-timestamp": "not-a-date", "x-expiry": "60" },
+    );
+    expect(res.status).toBe(400);
+  });
+
+  it("treats % and _ in a search as literal characters", async () => {
+    const p = await project();
+    await p.client.createSuppression({ channel: "email", target: "a_b@x.com" });
+    await p.client.createSuppression({ channel: "email", target: "axb@x.com" });
+    const found = (await p.client.listSuppressions({ target: "a_b" })).suppressions;
+    expect(found.map((s) => s.target)).toEqual(["a_b@x.com"]);
+    expect((await p.client.listSuppressions({ target: "%" })).suppressions).toEqual([]);
+  });
+
+  it("serves many live event streams without leaking listener warnings", async () => {
+    const p = await project();
+    const warnings: string[] = [];
+    const onWarning = (w: Error) => warnings.push(w.name);
+    process.on("warning", onWarning);
+    const controllers = Array.from({ length: 12 }, () => new AbortController());
+    try {
+      await Promise.all(
+        controllers.map((c) =>
+          fetch(`${app.baseUrl}/v1/events/stream`, {
+            headers: { authorization: `Bearer ${p.apiKey}` },
+            signal: c.signal,
+          }),
+        ),
+      );
+      await settle(200);
+      expect(warnings).not.toContain("MaxListenersExceededWarning");
+    } finally {
+      for (const c of controllers) c.abort();
+      process.off("warning", onWarning);
+    }
+  });
+});

tests/integration/delivery-worker.test.ts

@@ -1,6 +1,6 @@
 import { describe, it, expect, beforeEach, afterEach } from "vitest";
 import { randomUUID } from "node:crypto";
-import { DeliveryWorker } from "@/services/delivery/main.js";
+import { DeliveryWorker, MAX_THROTTLE_DEFERRALS } from "@/services/delivery/main.js";
 import { StreamConsumer, PendingMessageScanner, StreamProducer } from "@/queue/index.js";
 import { IdempotencyGuard } from "@/idempotency/index.js";
 import { ContactRepository, TemplateRepository, UserRepository } from "@/repositories/index.js";
@@ -302,7 +302,7 @@ describe("DeliveryWorker", () => {
     expect(sends).toBe(1);
   });
 
-  it("defers sends over the provider's limit, and fails them once the retries run out", async () => {
+  it("defers sends over the provider's limit, and fails them once the deferrals run out", async () => {
     const limited = new FakeProvider("sms", () => ({ success: true, providerMessageId: "x" }), {
       limit: 1,
       windowSeconds: 60,
@@ -311,14 +311,26 @@ describe("DeliveryWorker", () => {
     await w.start();
     const first = dispatched({ channel: "sms", destination: "+1" });
     const second = dispatched({ channel: "sms", destination: "+2" });
-    const exhausted = dispatched({ channel: "sms", destination: "+3", throttleAttemptCount: 3 });
-    for (const ev of [first, second, exhausted]) await publish(ev);
+    // Waiting for capacity is not a failed attempt: a backlog several windows
+    // deep must keep waiting, not fail once it has waited maxAttempts times.
+    const waitedLong = dispatched({
+      channel: "sms",
+      destination: "+3",
+      throttleAttemptCount: 3,
+      deliveryOptions: { maxAttempts: 3, timeoutMs: 10_000 },
+    });
+    const exhausted = dispatched({
+      channel: "sms",
+      destination: "+4",
+      throttleAttemptCount: MAX_THROTTLE_DEFERRALS,
+    });
+    for (const ev of [first, second, waitedLong, exhausted]) await publish(ev);
 
     await waitFor("one sent", () => limited.calls === 1);
-    await waitFor("deferred", async () => (await infra.redis.xlen(STREAMS.SCHEDULED)) === 1);
-    const [row] = await infra.sql`SELECT payload FROM scheduled_payloads`;
-    expect(row!.payload.throttleAttemptCount).toBe(1);
-    expect(Date.parse(row!.payload.scheduledAt)).toBeGreaterThan(Date.now());
+    await waitFor("deferred", async () => (await infra.redis.xlen(STREAMS.SCHEDULED)) === 2);
+    const rows = await infra.sql`SELECT payload FROM scheduled_payloads`;
+    expect(rows.map((r) => r.payload.throttleAttemptCount).sort()).toEqual([1, 4]);
+    for (const r of rows) expect(Date.parse(r.payload.scheduledAt)).toBeGreaterThan(Date.now());
 
     await waitFor("exhausted one failed", async () =>
       (await events()).some(

tests/integration/worker-runtime.test.ts

@@ -464,22 +464,32 @@ describe("BaseWorker", () => {
       expect((await dlq()).map((d) => d.id)).toEqual([ev.id]);
       expect(await pending()).toBe(0);
       expect(w.seen.map((s) => s.attempt)).toEqual([1, 2]);
-      expect(failed).toHaveLength(1);
+      // The event's id, as every other emitter reports it — not the stream
+      // entry id, which means nothing to a listener.
+      expect(failed).toEqual([ev.id]);
     } finally {
       globalEmitter.off("notification:failed", onFailed);
     }
   });
 
-  it("dead-letters a NonRetryableError immediately", async () => {
-    const w = worker(async () => {
-      throw new NonRetryableError("bad input");
-    });
-    await w.start();
-    const ev = await publish(1);
-    await waitFor("dead-lettered", async () => (await dlq()).length === 1);
-    expect((await dlq())[0].id).toBe(ev.id);
-    expect(await pending()).toBe(0);
-    expect(w.seen).toHaveLength(1);
+  it("dead-letters a NonRetryableError immediately, reporting the event's id", async () => {
+    const failed: string[] = [];
+    const onFailed = (id: string) => failed.push(id);
+    globalEmitter.on("notification:failed", onFailed);
+    try {
+      const w = worker(async () => {
+        throw new NonRetryableError("bad input");
+      });
+      await w.start();
+      const ev = await publish(1);
+      await waitFor("dead-lettered", async () => (await dlq()).length === 1);
+      expect((await dlq())[0].id).toBe(ev.id);
+      expect(await pending()).toBe(0);
+      expect(w.seen).toHaveLength(1);
+      expect(failed).toEqual([ev.id]);
+    } finally {
+      globalEmitter.off("notification:failed", onFailed);
+    }
   });
 
   it("leaves a message pending, without counting an error, when the lock is held", async () => {
@@ -688,3 +698,131 @@ describe("startHealthReporter", () => {
     }
   });
 });
+
+// ─── Service workers: graceful stop ─────────────────────────────────────────
+
+describe("service workers drain their in-flight messages on stop", () => {
+  const PROJECT = "00000000-0000-4000-8000-000000000001";
+
+  /** Resolves "stopped", or "hung" if stop() outlasts `ms`. */
+  const stopWithin = (w: BaseWorker, ms: number) =>
+    Promise.race([w.stop().then(() => "stopped" as const), settle(ms).then(() => "hung" as const)]);
+
+  function streamWorkerParts(stream: any, group: any) {
+    const consumer = new StreamConsumer({
+      redis: infra.redis,
+      stream,
+      group,
+      consumer: "drain",
+      dlqStream: STREAMS.DEAD_LETTER,
+      blockMs: 100,
+      batchSize: 10,
+      logger,
+    });
+    const pendingScanner = new PendingMessageScanner({
+      redis: infra.redis,
+      stream,
+      group,
+      consumer: "drain",
+      logger,
+    });
+    return { consumer, pendingScanner, logger, recoveryIntervalMs: 60_000 };
+  }
+
+  async function publishTo(stream: string, type: string, payload: Record<string, unknown>) {
+    const ev = {
+      ...buildStreamEvent(type as any, payload, "test"),
+      id: randomUUID(),
+      timestamp: new Date().toISOString(),
+    };
+    await infra.redis.xadd(stream, "*", "data", JSON.stringify(ev));
+    return ev;
+  }
+
+  it("enricher: finishes a message that was mid-lookup when stop was called", async () => {
+    const { EnricherWorker } = await import("@/services/enricher/main.js");
+    let lookupStarted = false;
+    const published: unknown[] = [];
+    const producer = {
+      publishBatch: async (events: unknown[]) => {
+        published.push(...events);
+        return { messageIds: events.map(() => "1-0"), eventIds: events.map(() => randomUUID()) };
+      },
+    };
+    const w = new EnricherWorker({
+      ...streamWorkerParts(STREAMS.INBOUND_NORMAL, CONSUMER_GROUPS.ENRICHER),
+      concurrency: 10,
+      producers: { critical: producer, normal: producer, low: producer },
+      idempotency: new IdempotencyGuard({ redis: infra.redis, keyPrefix: "t:enr" }),
+      userRepo: {
+        findRecordsByIds: async () => {
+          lookupStarted = true;
+          await settle(300);
+          return [{ userId: "u1", preferences: {} }];
+        },
+      },
+      prefRepo: {},
+      contactRepo: {
+        findActiveByUserIds: async () =>
+          new Map([["u1", [{ id: "c1", channel: "email", target: "u1@x.com" }]]]),
+      },
+      templateCache: { getCachedTemplate: async () => null } as any,
+    });
+    await w.start();
+    await publishTo(STREAMS.INBOUND_NORMAL, "notification.requested", {
+      projectId: PROJECT,
+      target: { type: "user", userId: "u1" },
+      templateId: "t",
+      priority: "normal",
+      data: {},
+      fallback: false,
+    });
+    await waitFor("lookup in flight", () => lookupStarted);
+
+    expect(await stopWithin(w, 5_000)).toBe("stopped");
+    expect(published).toHaveLength(1);
+  }, 20_000);
+
+  it("workflow: finishes a run that was mid-step when stop was called", async () => {
+    const { WorkflowWorker } = await import("@/services/workflow/main.js");
+    const { workflow } = await import("@/workflows/index.js");
+    let stepStarted = false;
+    workflow("drain-on-stop", async ({ step }) => {
+      await step.run("slow", async () => {
+        stepStarted = true;
+        await settle(300);
+        return true;
+      });
+      await step.notify({ template: "t" });
+    });
+    const published: unknown[] = [];
+    const producer = {
+      publishBatch: async (events: unknown[]) => {
+        published.push(...events);
+        return { messageIds: events.map(() => "1-0"), eventIds: events.map(() => randomUUID()) };
+      },
+    };
+    const w = new WorkflowWorker({
+      ...streamWorkerParts(STREAMS.WORKFLOW_INBOUND, CONSUMER_GROUPS.WORKFLOW),
+      concurrency: 10,
+      redis: infra.redis,
+      db: infra.db,
+      workflowProducer: producer,
+      notificationProducer: producer,
+    });
+    await w.start();
+    const instanceId = randomUUID();
+    await publishTo(STREAMS.WORKFLOW_INBOUND, "workflow.triggered", {
+      projectId: PROJECT,
+      instanceId,
+      name: "drain-on-stop",
+      input: { user: { id: "u1" } },
+    });
+    await waitFor("step in flight", () => stepStarted);
+
+    expect(await stopWithin(w, 5_000)).toBe("stopped");
+    expect(published).toHaveLength(1);
+    const [row] = await infra.sql`SELECT status FROM workflow_instances WHERE id = ${instanceId}`;
+    expect(row!.status).toBe("completed");
+  }, 20_000);
+});

tests/integration/workflows.test.ts

@@ -51,6 +51,37 @@ workflow("await-go", async ({ step }) => {
   const r = await step.waitForEvent("go", { timeout: "1h" });
   await step.run("after", () => ({ got: r }));
 });
+workflow("quick-nap", async ({ step }) => {
+  await step.wait("500ms");
+  await step.run("woke", () => true);
+});
+workflow("fractional-wait", async ({ step }) => {
+  await step.wait("1.5h");
+});
+workflow("fractional-nap", async ({ step }) => {
+  await step.wait("0.5s");
+  await step.run("woke", () => true);
+});
+workflow("vague-wait", async ({ step }) => {
+  await step.wait("soon");
+});
+workflow("pays-twice", async ({ step }) => {
+  await step.waitForEvent("order.paid", { timeout: "3s" });
+  const second = await step.waitForEvent("order.paid", { timeout: "1h" });
+  await step.run("after", () => ({ second }));
+});
+
+// Holds the handler mid-run until the test lets it go.
+const gates = new Map<string, () => void>();
+const gateReached = new Set<string>();
+workflow("gated", async ({ step, event }) => {
+  await step.run("hold", async () => {
+    gateReached.add(event.tag);
+    await new Promise<void>((resolve) => gates.set(event.tag, resolve));
+    return true;
+  });
+  await step.notify({ template: "hello" });
+});
 
 beforeAll(async () => {
   app = await startNotifkit(infra, {
@@ -277,6 +308,106 @@ describe("code-defined workflows", () => {
     expect(email.for(p.id)).toEqual([]);
   }, 20_000);
 
+  it("stays canceled when canceled mid-run, and runs no further steps", async () => {
+    const p = await project();
+    const tag = randomUUID();
+    const id = await trigger(p.api, { name: "gated", user: "alice", input: { tag } });
+    await waitFor("handler mid-run", () => gateReached.has(tag));
+
+    expect((await p.api("DELETE", `/v1/workflows/instances/${id}`)).status).toBe(204);
+    gates.get(tag)!();
+
+    await settle(3_000);
+    expect((await instance(p.api, id)).status).toBe("canceled");
+    expect(email.for(p.id)).toEqual([]);
+  }, 20_000);
+
+  it("does not let an answered wait's timeout fire on a later wait for the same event", async () => {
+    const p = await project();
+    const id = await trigger(p.api, { name: "pays-twice", user: "alice" });
+    await waitForStatus(p.api, id, "pending");
+
+    // Answer the first wait; the workflow moves on to the second.
+    await sendEvent(p.api, "order.paid", { n: 1 });
+    await waitFor(
+      "suspended on the second wait",
+      async () => (await instance(p.api, id))?.steps.length === 2,
+    );
+    await waitForStatus(p.api, id, "pending");
+
+    // Past the first wait's 3s deadline plus a timer poll.
+    await settle(9_000);
+    const inst = await instance(p.api, id);
+    expect(inst.status).toBe("pending");
+    expect(inst.steps[1].output).toBeNull();
+    expect(inst.waiters).toHaveLength(1);
+  }, 40_000);
+
+  it("keeps a long sleep's wake-up when a resume is deferred for a busy instance", async () => {
+    const p = await project();
+    const id = await trigger(p.api, { name: "waits-an-hour", user: "alice" });
+    await waitForStatus(p.api, id, "pending");
+
+    // A duplicate resume finds the instance locked by another process.
+    await infra.redis.set(`lock:workflow:${id}`, "other-process", "EX", 60);
+    await infra.redis.xadd(
+      STREAMS.WORKFLOW_INBOUND,
+      "*",
+      "data",
+      JSON.stringify({
+        ...buildStreamEvent(
+          "workflow.resumed",
+          {
+            projectId: p.id,
+            instanceId: id,
+            name: "waits-an-hour",
+            input: { user: { id: "alice" } },
+          },
+          "test",
+        ),
+        id: randomUUID(),
+        timestamp: new Date().toISOString(),
+      }),
+    );
+    await settle(1_500);
+    await infra.redis.del(`lock:workflow:${id}`);
+
+    const timers = await infra.redis.zrange("notif:workflow:timers", "0", "-1", "WITHSCORES");
+    const scores: number[] = [];
+    for (let i = 0; i < timers.length; i += 2) {
+      if (timers[i]!.includes(id)) scores.push(Number(timers[i + 1]));
+    }
+    expect(Math.max(...scores)).toBeGreaterThan(Date.now() + 30 * 60_000);
+  }, 20_000);
+
+  it("reads a millisecond wait as milliseconds", async () => {
+    const p = await project();
+    const id = await trigger(p.api, { name: "quick-nap", user: "alice" });
+    // One timer poll (5s) is enough; "500ms" once meant 500 seconds.
+    await waitForStatus(p.api, id, "completed", 12_000);
+  }, 20_000);
+
+  it("reads a fractional duration exactly: 1.5h sleeps 90 minutes, not 1 hour", async () => {
+    const p = await project();
+    const id = await trigger(p.api, { name: "fractional-wait", user: "alice" });
+    await waitForStatus(p.api, id, "pending");
+    const wait = (await instance(p.api, id)).steps.find((s: any) => s.action === "wait");
+    const sleptMs = wait.output.scheduledAt - Date.parse(wait.createdAt);
+    expect(Math.abs(sleptMs - 90 * 60_000)).toBeLessThan(60_000);
+  });
+
+  it("wakes from a fractional wait in a small unit", async () => {
+    const p = await project();
+    const id = await trigger(p.api, { name: "fractional-nap", user: "alice" });
+    await waitForStatus(p.api, id, "completed", 12_000);
+  }, 20_000);
+
+  it("fails a wait whose duration is not a duration at all", async () => {
+    const p = await project();
+    const id = await trigger(p.api, { name: "vague-wait", user: "alice" });
+    await waitForStatus(p.api, id, "failed");
+  });
+
   it("does nothing for a workflow nobody defined", async () => {
     const p = await project();
     const id = await trigger(p.api, { name: "no-such-workflow", user: "alice" });

tests/redis.test.ts

@@ -160,6 +160,7 @@ describe("Redis Pre-compiled Commands (defineCommand)", () => {
         .mockResolvedValue([{ id: "inst-1", status: "pending", name: "flow", input: {} }]);
       qb.update = vi.fn().mockReturnValue(qb);
       qb.set = vi.fn().mockReturnValue(qb);
+      qb.returning = vi.fn().mockReturnValue(qb);
       qb.then = function (resolve: any) {
         resolve([{ id: "inst-1", status: "pending", name: "flow", input: {} }]);
       };

tests/repositories.mock.test.ts

@@ -31,10 +31,11 @@ describe("Repositories (Mocked Unit Tests)", () => {
       const repo = new UserRepository(mockDb as any);
       const user = await repo.findById("proj-1", "u1");
 
+      // Defaults are applied on read, since upserts leave omitted fields unset.
       expect(user).toEqual({
         userId: "u1",
-        language: undefined,
-        timezone: undefined,
+        language: "en",
+        timezone: "UTC",
         email: "a@b.c",
       });
     });

tests/server-shutdown.test.ts

@@ -0,0 +1,54 @@
+import { describe, it, expect, vi } from "vitest";
+
+// Records the order NotifkitServer.stop() stops its services in.
+const { stopped } = vi.hoisted(() => ({ stopped: [] as string[] }));
+
+function service(name: string, cap: string) {
+  return {
+    [`start${cap}`]: vi.fn().mockResolvedValue(undefined),
+    [`stop${cap}`]: vi.fn(async () => {
+      stopped.push(name);
+    }),
+    [`get${cap}`]: vi.fn(() => undefined),
+  };
+}
+
+vi.mock("../src/services/api/main.js", () => service("api", "ApiServer"));
+vi.mock("../src/services/enricher/main.js", () => service("enricher", "EnricherWorker"));
+vi.mock("../src/services/engine/main.js", () => service("engine", "EngineWorker"));
+vi.mock("../src/services/delivery/main.js", () => service("delivery", "DeliveryWorker"));
+vi.mock("../src/services/scheduler/main.js", () => service("scheduler", "SchedulerWorker"));
+vi.mock("../src/services/ai/main.js", () => service("ai", "AiWorker"));
+vi.mock("../src/services/workflow/main.js", () => service("workflow", "WorkflowWorker"));
+vi.mock("../src/services/events/main.js", () => service("events", "EventWorker"));
+
+import { NotifkitServer } from "@/server.js";
+
+describe("NotifkitServer.stop", () => {
+  it("stops services upstream first, so nothing in flight reaches a stage already shut", async () => {
+    const server = new NotifkitServer({
+      services: ["all"],
+      redisUrl: "redis://127.0.0.1:1",
+      databaseUrl: "postgres://x@127.0.0.1:1/x",
+      autoMigrate: false,
+      logLevel: "silent",
+    });
+    await server.start();
+    await server.stop();
+
+    const before = (a: string, b: string) => stopped.indexOf(a) < stopped.indexOf(b);
+    expect(stopped).toHaveLength(8);
+    expect(stopped[0]).toBe("api");
+    // A fused pipeline runs enricher → engine → delivery inside the enricher's
+    // task, so delivery's database must outlive the enricher.
+    expect(before("enricher", "engine")).toBe(true);
+    expect(before("engine", "delivery")).toBe(true);
+    // Workflow steps publish into the enricher's stream; AI and the scheduler
+    // hand work to delivery.
+    expect(before("workflow", "enricher")).toBe(true);
+    expect(before("ai", "delivery")).toBe(true);
+    expect(before("scheduler", "delivery")).toBe(true);
+    // Events records what delivery did, so it goes last.
+    expect(stopped.at(-1)).toBe("events");
+  });
+});

tests/workers-workflow.test.ts

@@ -367,7 +367,8 @@ describe("WorkflowWorker Edge Cases", () => {
           input: { user: { id: "u-1" } },
         },
       ])
-      .mockResolvedValueOnce([]);
+      // Then the per-step cancellation checks, which find nothing canceled.
+      .mockResolvedValue([]);
 
     await worker.process(msg);
 

tests/workers.test.ts

@@ -1,6 +1,6 @@
 import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
 import { AiWorker, PermanentAiError } from "@/services/ai/main.js";
-import { DeliveryWorker } from "@/services/delivery/main.js";
+import { DeliveryWorker, MAX_THROTTLE_DEFERRALS } from "@/services/delivery/main.js";
 import { PreferenceRepository } from "@/repositories/index.js";
 import {
   BaseWorker,
@@ -1531,13 +1531,28 @@ describe("DeliveryWorker", () => {
       expect(parkedPayload().throttleAttemptCount).toBe(3);
     });
 
-    it("gives up once the throttled retries exceed maxAttempts", async () => {
+    it("keeps waiting past maxAttempts: a deferral is not a failed attempt", async () => {
+      mockTransportRegistry.getAll.mockReturnValue([limitedTransport(vi.fn())]);
+      mockRedis.leaseApiRateLimit = vi.fn().mockResolvedValue(0);
+
+      await worker.process(
+        dispatched({ throttleAttemptCount: 3, deliveryOptions: { maxAttempts: 3 } }) as any,
+      );
+
+      expect(mockScheduledProducer.publish).toHaveBeenCalledTimes(1);
+      expect(parkedPayload().throttleAttemptCount).toBe(4);
+    });
+
+    it("gives up once the deferrals run out", async () => {
       const send = vi.fn();
       mockTransportRegistry.getAll.mockReturnValue([limitedTransport(send)]);
       mockRedis.leaseApiRateLimit = vi.fn().mockResolvedValue(0);
 
       await worker.process(
-        dispatched({ throttleAttemptCount: 3, deliveryOptions: { maxAttempts: 3 } }) as any,
+        dispatched({
+          throttleAttemptCount: MAX_THROTTLE_DEFERRALS,
+          deliveryOptions: { maxAttempts: 3 },
+        }) as any,
       );
 
       expect(send).not.toHaveBeenCalled();
@@ -1557,7 +1572,7 @@ describe("DeliveryWorker", () => {
 
       await worker.process(
         dispatched({
-          throttleAttemptCount: 9,
+          throttleAttemptCount: MAX_THROTTLE_DEFERRALS,
           deliveryOptions: { maxAttempts: 3 },
           fallbackChain: ["sms"],
           recipient,

tests/workflows.test.ts

@@ -149,7 +149,7 @@ describe("Workflow Engine Edge Cases", () => {
     mockDb.returning.mockResolvedValueOnce([
       { id: "inst-1", name: "test-wait", status: "pending", input: { user: { id: "user-1" } } },
     ]); // insert returning — the row stores the trigger input, so it carries the user
-    mockDb.then.mockImplementationOnce((res) => res()); // For the update to running
+    mockDb.then.mockImplementationOnce((res) => res([{ id: "claimed" }])); // update to running claims the instance
     mockDb.then.mockImplementationOnce((res) => res([])); // steps query
 
     await worker.process(msg);
@@ -190,7 +190,7 @@ describe("Workflow Engine Edge Cases", () => {
         { id: "inst-1", name: "test-wait", status: "pending", input: { user: { id: "user-1" } } },
       ]),
     );
-    mockDb.then.mockImplementationOnce((res) => res()); // For the update to running
+    mockDb.then.mockImplementationOnce((res) => res([{ id: "claimed" }])); // update to running claims the instance
 
     // Existing steps: notify and wait
     mockDb.then.mockImplementationOnce((res) =>
@@ -247,6 +247,7 @@ describe("Workflow Engine Edge Cases", () => {
         input: { user: { id: "user-1" } },
       },
     ]);
+    mockDb.then.mockImplementationOnce((res) => res([{ id: "claimed" }])); // update to running claims the instance
     mockDb.then.mockImplementationOnce((res) => res([])); // steps
 
     await worker.process(msg);
@@ -282,7 +283,7 @@ describe("Workflow Engine Edge Cases", () => {
         },
       ]),
     );
-    mockDb.then.mockImplementationOnce((res) => res()); // For the update to running
+    mockDb.then.mockImplementationOnce((res) => res([{ id: "claimed" }])); // update to running claims the instance
     // Steps now contain the event output injected by EventWorker
     mockDb.then.mockImplementationOnce((res) =>
       res([{ stepIndex: "0", action: "waitForEvent", output: { action: "clicked" } }]),
@@ -326,7 +327,7 @@ describe("Workflow Engine Edge Cases", () => {
         },
       ]),
     );
-    mockDb.then.mockImplementationOnce((res) => res()); // For the update to running
+    mockDb.then.mockImplementationOnce((res) => res([{ id: "claimed" }])); // update to running claims the instance
     // Steps contain NO output (null) because EventWorker didn't inject anything, the scheduler woke it up
     mockDb.then.mockImplementationOnce((res) =>
       res([{ stepIndex: "0", action: "waitForEvent", output: null }]),
@@ -366,7 +367,7 @@ describe("Workflow Engine Edge Cases", () => {
         },
       ]),
     );
-    mockDb.then.mockImplementationOnce((res) => res()); // For the update to running
+    mockDb.then.mockImplementationOnce((res) => res([{ id: "claimed" }])); // update to running claims the instance
     // Persisted step records timedOut: true
     mockDb.then.mockImplementationOnce((res) =>
       res([{ stepIndex: "0", action: "waitForEvent", output: { timedOut: true } }]),