Example: a payments queue consumer

A Worker that consumes a payments queue and records each payment in D1. Nothing in it looks wrong, and nothing in an ordinary test would fail.

Run the check

From examples/payments-queue in the repository:

pnpm install
pnpm check

The check runs the scenario in payments.sim.ts inside workerd, injecting crashes, errors and lost replies at every effect the Worker performs and redelivering messages the way Cloudflare Queues does. It reports:

VIOLATED  payments  (seed payments)

Invariant payment-recorded-once is violated: D1:DB run happened twice for key "o-2" ...

Schedule (run seed ...), events kept: msg-2
  ambiguous at msg-2#1 effect #0 (D1:DB run)

The insert committed but its answer was lost, so the handler threw, the queue redelivered the message, and the order was recorded twice. (With --seed example it reports the other way to get there: a crash between the insert and ack().) The trace below the schedule shows every effect; a regression test, payments.regression.test.ts, is written next to the scenario and fails until the bug is fixed.

The fix

worker.fixed.ts makes the insert idempotent (ON CONFLICT (order_id) DO NOTHING over a unique order_id), so a redelivered message changes nothing. pnpm check:fixed runs the same scenario against it and reports BOUNDED: no violation within the stated bounds, which the report says plainly is not a proof.

Using it in your own project

Copy payments.sim.ts next to your Worker, replace the messages and the invariants with yours, and point worker at your entry and wrangler config. eutaxy, vitest and @cloudflare/vitest-plugin are the only dependencies.

The code

The complete project is examples/payments-queue in the repository.

worker.ts

/**
 * A queue consumer that records each payment in D1 and acknowledges the
 * message. It looks fine and passes ordinary tests. Run `eutaxy check
 * ./payments.sim.ts` to see what happens when the invocation dies between
 * the insert and the acknowledgement.
 */

export interface Env {
  readonly DB: D1Database;
}

export interface Payment {
  readonly orderId: string;
  readonly amountCents: number;
}

export default {
  async queue(batch: MessageBatch<Payment>, env: Env): Promise<void> {
    for (const message of batch.messages) {
      const { orderId, amountCents } = message.body;
      await env.DB.prepare("INSERT INTO payments (order_id, amount_cents) VALUES (?1, ?2)")
        .bind(orderId, amountCents)
        .run();
      message.ack();
    }
  },
} satisfies ExportedHandler<Env, Payment>;

payments.sim.ts

import {
  atLeastOncePerMessage,
  atMostOncePerKey,
  d1Param,
  d1Schema,
  d1Write,
  defineScenario,
  queue,
} from "eutaxy/cloudflare";

// The messages the queue will deliver. Eutaxy explores crashes, retries and
// lost replies around every effect the Worker performs while handling them.
const payments = queue("payments", [
  { body: { orderId: "o-1", amountCents: 1200 } },
  { body: { orderId: "o-2", amountCents: 500 } },
]);

export default defineScenario({
  name: "payments",
  worker: { entry: "./worker.ts", config: "./wrangler.jsonc" },
  setup: [d1Schema("DB", "./schema.sql")],
  drive: payments,
  invariants: [
    // No order may be recorded twice: the first bound parameter of each committed insert is the order id.
    atMostOncePerKey("payment-recorded-once", d1Write("DB"), d1Param(0)),
    // And no order may be lost.
    atLeastOncePerMessage("every-payment-recorded", payments, d1Write("DB"), d1Param(0), "orderId"),
  ],
  bounds: { maxFaultsPerRun: 1, maxRedeliveries: 2, maxConcurrentInvocations: 1, budgetMs: 2000 },
});

worker.fixed.ts

/**
 * The same consumer, made safe to redeliver: the insert is idempotent on
 * the order id, so a retried message changes nothing. `eutaxy check
 * ./payments.fixed.sim.ts` reports BOUNDED.
 */

import type { Env, Payment } from "./worker.ts";

export default {
  async queue(batch: MessageBatch<Payment>, env: Env): Promise<void> {
    for (const message of batch.messages) {
      const { orderId, amountCents } = message.body;
      await env.DB.prepare(
        "INSERT INTO payments (order_id, amount_cents) VALUES (?1, ?2) ON CONFLICT (order_id) DO NOTHING",
      )
        .bind(orderId, amountCents)
        .run();
      message.ack();
    }
  },
} satisfies ExportedHandler<Env, Payment>;