ElementsSignal

Once

Competing workers, visibility leases, retries, dead letters, and per-key ordering for signal.once.

signal.once processes background work with competing consumers — exactly one worker claims each message; failed attempts retry, then dead-letter.

For developers shipping jobs on okengine — declare the Signal, bind on(signal, flow), emit with fx.emit.

The one rule

Physics live on the Signal (retries, deadLetter, optional). The Flow is the subscriber. Visibility lease defaults to 30s — an inflight worker that dies is reclaimed on the next claim.

once — claim, lease, reclaim

leaseMs · 30s default
pending
no sweeper — next claim query reclaims
  • worker-a

    waiting to claim

    Claim sets lockedBy + leaseExpiresAt. Finish side effects before the lease ends — or another worker may reclaim while you still run.

  • worker-b

    idle until lease expires

    Next claim query takes pending or expired inflight. Make the handler idempotent.

Smallest Example

One handle, three independent uses

emailTask is a shared const. Declare it, bind a worker, and emit — different files, any order. Binding is not "step 2" after declare; emit is not "step 3".

Declare

src/signals/email.ts
import { signal } from "okengine";
import { z } from "zod";

export const emailTask = signal.once("tasks.email", {
  schema: z.object({ to: z.string().email(), body: z.string() }),
  retries: 3,
  deadLetter: true,
});

Bind a worker

src/flows/workers/email.ts
import { on, flow } from "okengine";
import { emailTask } from "@/signals/email";
import { rawEmail } from "@/channels/email";

export const processEmail = on(
  emailTask,
  flow("workers.email", {
    do: async ({ to, body }, fx) => {
      await fx.send(rawEmail, { to, body });
    },
  }),
);

Emit

await fx.emit(emailTask, { to: "alice@example.com", body: "Welcome" });

The emit resolves when the outbox commits. The worker runs asynchronously — the producer does not wait for fx.send to finish.

Competing consumers (once vs broadcast)

This is the most common mix-up. signal.once is a competing-consumer work queue: exactly one worker claims each message. It is not fan-out.

Three differently-named Flows bound to the same once signal:

src/flows/orders/side-effects.ts
import { on, flow } from "okengine";
import { orderPlaced } from "@/signals/orders";

export const charge = on(
  orderPlaced,
  flow("orders.charge", {
    do: async ({ orderId }, fx) => {
      await fx.call(chargeOrder, { orderId });
    },
  }),
);

export const ship = on(
  orderPlaced,
  flow("orders.ship", {
    do: async ({ orderId }, fx) => {
      await fx.call(shipOrder, { orderId });
    },
  }),
);

export const notify = on(
  orderPlaced,
  flow("orders.notify", {
    do: async ({ orderId }, fx) => {
      await fx.call(notifyOrder, { orderId });
    },
  }),
);
await fx.emit(orderPlaced, { orderId: "ord_99", amount: 150, userId: "usr_1" });
ExpectationReal result
All three Flows runNo. Exactly one of the three runs
Always orders.charge (first on)No. The winner is whichever claim lands first — not a fixed Flow
A sticky assignment to one FlowNo. There is no owner — only an exclusive claim per message

Consequence: if every bound Flow should independently receive its own copy, use signal.broadcast. That is the correct fix for this exact mistake.

Two or more different Flow definitions on the same once signal fail OKE1071.

Cause: Once signal "{signal}" is bound to more than one Flow ({flows}).

Fix: Use signal.broadcast if each flow should independently receive this event, or bind only one flow if these should compete for the same work.

Load-balancing competing consumers is the same Flow on many process replicas — still one on() in source. That case is not a second Flow definition.

Progressive Patterns

From a minimal job to ordered partitions and DLQ inspection:

Defaults are retries: 3, deadLetter: true, optional: false:

src/signals/orders.ts
import { signal } from "okengine";
import { z } from "zod";

export const orderPlaced = signal.once("orders.placed", {
  schema: z.object({
    orderId: z.string(),
    amount: z.number(),
    userId: z.string(),
  }),
});
src/flows/orders/fulfill.ts
import { on, flow } from "okengine";
import { orderPlaced } from "@/signals/orders";

export const fulfill = on(
  orderPlaced,
  flow("orders.fulfill", {
    do: async ({ orderId }, fx) => {
      await fx.call(chargeAndShip, { orderId });
    },
  }),
);

Delivery Reference

SurfaceSignaturePurpose
Declaresignal.once(name, options?)Competing queue — one claim per message
Bindon(signalHandle, flow)Worker Flow; payload is do's input
Emitfx.emit(signal, payload?, { key? })Enrol in the outbox; records emits
Inspect DLQfx.deadLetters(signal)Exhausted messages; records reads signal:<name>
NeedUse instead
Every active subscriber gets a copybroadcast
Browser / SSE resume after disconnectlive
Durable multi-step work with a journalDurable Workflows

Options for once

Optional second argument to signal.once. Delivery is the helper name — not an option.

OptionTypeDefaultMeaning
schemaStandard SchemaomittedEmit contract — validated at fx.emit (OKE1250). Workers inherit payload typing.
retriesnumber3Extra attempts after the first (retries + 1 total)
deadLetterbooleantrueKeep exhausted messages; not a queue name
optionalbooleanfalseAllow emit with zero subscribers
descriptionstringthe nameConsole / docs blurb

retention is a type error on signal.once — that option is live-only:

signal.once("…"): retention is only valid with signal.live

Consequence: deadLetter is a boolean flag, not a DLQ signal name string.

Binding Workers

Each worker binds with on(signalHandle, flow(...)). The Signal carries delivery physics; the Flow is only the handler.

Name the Signal and attach retry / DLQ / schema policy:

src/signals/email.ts
import { signal } from "okengine";
import { z } from "zod";

export const emailTask = signal.once("tasks.email", {
  schema: z.object({ to: z.string().email(), body: z.string() }),
  retries: 3,
  deadLetter: true,
});

Emit

Before any worker runs, fx.emit validates schema (when set) and enrols the message in the outbox. The call resolves on commit — not when the handler finishes.

src/flows/orders/create.ts
import { on, flow, http } from "okengine";
import { z } from "zod";
import { orderPlaced } from "@/signals/orders";

export const create = on(
  http.post({
    in: z.object({ userId: z.string(), amount: z.number() }),
  }),
  flow({
    do: async ({ userId, amount }, fx) => {
      const orderId = fx.id();
      await fx.emit(orderPlaced, { orderId, amount, userId }, { key: userId });
      return { orderId };
    },
  }),
);
CallRecordsMeaning
fx.emit(signal, payload?)emitsCompeting pool — no ordering
fx.emit(signal, payload, { key })emitsPer-key FIFO for that (signal, key)
fx.deadLetters(signal)reads signal:<name>Inspect dead messages

parentRunId is stamped automatically from the producer run for Console trace chains — you do not set it by hand in app code.

Lease and reclaim

Detailed section

If you only need the default 30s lease, jump to Ordering. Claims set lockedBy + leaseExpiresAt. There is no background sweeper — reclaim happens lazily on the next claim after expiry.

ConceptDefaultMeaning
Visibility lease30_000 msHow long an inflight claim holds the message
Reclaimnext claim after expiryAnother worker may take the same message
Status pathpending → inflight → delivered | deadOperator inspect / Console

Consequence: treat every once handler as at-least-once. Prefer idempotent do bodies (or durable steps for side effects that must not double-fire).

Ordering

Detailed section

Partition with {key} only when you need per-tenant / per-user FIFO. Unkeyed messages stay a competing pool and may run concurrently.

src/flows/orders/ship.ts
import { on, flow } from "okengine";
import { orderPlaced } from "@/signals/orders";

export const shipOrder = on(
  orderPlaced,
  flow("orders.ship", {
    do: async ({ orderId }, fx) => {
      await fx.call(fulfillOrder, { orderId });
    },
  }),
);
await fx.emit(orderPlaced, { orderId: "ord_1", userId: "usr_1" }, { key: "usr_1" });
EmitConcurrencyOrder
No keyCompeting — may overlapNone
Same keySerialized by leaseFIFO for that key
Different keysMay overlapIndependent

Retries and dead letters

Detailed section

If you only need defaults (retries: 3, deadLetter: true), jump to Idempotency. Retries requeue immediately — there is no delay backoff between attempts.

Idempotency

Lease reclaim and retries mean a handler can run more than once for the same message.

ApproachWhen
Idempotent doSide effects are safe to repeat (upsert, set-once flags)
durable: true + fx.stepMulti-step work that must not double-fire
Short handlersFinish before the 30s lease so reclaim does not overlap
src/flows/payments/sync.ts
import { on, flow } from "okengine";
import { syncPayment } from "@/signals/payments";

export const runSync = on(
  syncPayment,
  flow("payments.sync", {
    durable: true,
    do: async ({ chargeId }, fx) => {
      await fx.step("charge", async () => {
        await fx.call(applyCharge, { chargeId });
      });
      await fx.step("receipt", async () => {
        await fx.call(sendReceipt, { chargeId });
      });
    },
  }),
);

See Durable Workflows.

Troubleshooting

Learn more

  • Signal Overview — delivery matrix and drivers
  • Broadcast — fan-out without leases
  • Live — retained SSE tapes
  • Consumers — on(signal) next to Clock / CDC
  • Workflows — durable + fx.step for idempotent side effects
  • fx — fx.emit, fx.deadLetters
  • Errors — OKE1071 · OKE1072 · OKE1240 · OKE1250 · OKE1001

Next

Broadcast

Ephemeral fan-out across subscribed Flows.

Live

Retained SSE tapes for browsers.

Consumers

Signal workers, Clock jobs, and SQL CDC.

Overview

Data in motion — once, broadcast, and live SSE tapes under one typed declaration with HTTP-shaped delivery helpers.

Broadcast

Ephemeral fan-out — every active subscriber gets a copy; offline listeners miss the event.

On this page

Smallest ExampleDeclareBind a workerEmitCompeting consumers (once vs broadcast)Progressive PatternsDelivery ReferenceOptions for onceBinding WorkersEmitLease and reclaimOrderingRetries and dead lettersIdempotencyTroubleshootingLearn moreNext