# Timers & Concurrency

## Timers

Use queue messages as the trigger source, then sleep durably inside the workflow.

index.ts:

```ts
import { actor, queue, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

type Reminder = {
  text: string;
  at: number;
};

export const reminderActor = actor({
  state: {
    fired: [] as string[],
  },
  queues: {
    reminders: queue<Reminder>(),
  },
  run: workflow(async (ctx) => {
    await ctx.loop("reminder-loop", async (loopCtx) => {
        const message = await loopCtx.queue.next("wait-reminder");

        const runAt = Math.max(Date.now(), message.body.at);
        await loopCtx.sleepUntil("wait-until-reminder", runAt);

        await loopCtx.step("record-reminder", async (step) => {
          step.state.fired.push(message.body.text);
        });

      });
  }),
  actions: {
    getState: (c) => c.state,
  },
});

export const registry = setup({ use: { reminderActor } });
```

client.ts:

```ts
import { createClient } from "rivetkit/client";
import type { registry } from "./index";

const client = createClient<typeof registry>("http://localhost:6420");
const handle = client.reminderActor.getOrCreate(["main"]);

await handle.send("reminders", {
  text: "send weekly report",
  at: Date.now() + 1_000,
});

await new Promise((resolve) => setTimeout(resolve, 1_300));
console.log(await handle.getState());
```

## Timeouts

Use step timeouts and retries for slow or flaky dependencies.

Step timeouts are critical by default and fail immediately. Set `retryOnTimeout: true` if a timeout should retry like any other error using `maxRetries`.

Workflows use roll-forward semantics everywhere. When a step throws, any `state` or `vars` mutations it made before failing are never rolled back, whether the step retries or the failure is caught by `tryStep` or `try`. The next attempt observes whatever the failed attempt already wrote, so write steps idempotently: check before you increment, or move the mutation after the fallible work.

examples/docs/actors-workflows/timeouts.ts:

```ts
import { actor, queue, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

async function chargeCard(orderId: string): Promise<string> {
  return `charge-${orderId}`;
}

export const timeoutActor = actor({
  state: {
    lastChargeId: null as string | null,
  },
  queues: {
    charge: queue<{ orderId: string }>(),
  },
  run: workflow(async (ctx) => {
    await ctx.loop("charge-loop", async (loopCtx) => {
        const message = await loopCtx.queue.next("wait-charge");

        const chargeId = await loopCtx.step<string>({
          name: "charge-card",
          timeout: 5_000,
          retryOnTimeout: true,
          maxRetries: 5,
          retryBackoffBase: 200,
          retryBackoffMax: 2_000,
          run: async (loopCtx) => await chargeCard(message.body.orderId),
        });

        await loopCtx.step("save-charge", async (step) => {
          step.state.lastChargeId = chargeId;
        });

      });
  }),
});

export const registry = setup({ use: { timeoutActor } });
```

## Join

Use `join` when several independent tasks can run in parallel.

index.ts:

```ts
import { actor, queue, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

export const dashboardActor = actor({
  state: {
    summary: null as null | {
      users: number;
      orders: number;
      revenue: number;
    },
  },
  queues: {
    refresh: queue<Record<string, never>>(),
  },
  run: workflow(async (ctx) => {
    await ctx.loop("dashboard-loop", async (loopCtx) => {
        await loopCtx.queue.next("wait-refresh");

        const summary = await loopCtx.join("fetch-summary", {
          users: {
            run: async (branchCtx) => {
              return await branchCtx.step("fetch-users", (branchCtx) => fetchCount("/users"));
            },
          },
          orders: {
            run: async (branchCtx) => {
              return await branchCtx.step("fetch-orders", (branchCtx) => fetchCount("/orders"));
            },
          },
          revenue: {
            run: async (branchCtx) => {
              return await branchCtx.step("fetch-revenue", (branchCtx) => fetchCount("/revenue"));
            },
          },
        });

        await loopCtx.step("save-summary", async (step) => {
          step.state.summary = summary;
        });

      });
  }),
  actions: {
    getState: (c) => c.state,
  },
});

async function fetchCount(path: string): Promise<number> {
  const res = await fetch(`https://api.example.com${path}`);
  if (!res.ok) throw new Error(`fetch ${path} failed: ${res.status}`);
  return ((await res.json()) as { count: number }).count;
}

export const registry = setup({ use: { dashboardActor } });
```

client.ts:

```ts
import { createClient } from "rivetkit/client";
import type { registry } from "./index";

const client = createClient<typeof registry>("http://localhost:6420");
const handle = client.dashboardActor.getOrCreate(["main"]);

await handle.send("refresh", {});
console.log(await handle.getState());
```

## Race

Use `race` when you need first-winner behavior.

index.ts:

```ts
import { actor, queue, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

export const auctionActor = actor({
  state: { result: null as "sold" | "expired" | null },
  queues: {
    bids: queue<{ amount: number }>(),
  },
  run: workflow(async (ctx) => {
    await ctx.step("list-item", (ctx) => listItem("item-123"));

    const { winner } = await ctx.race("bid-or-expire", [
      {
        name: "bid",
        run: async (branchCtx) => {
          const bid = await branchCtx.queue.next("wait-bid");
          return bid.body.amount;
        },
      },
      {
        name: "expire",
        run: async (branchCtx) => {
          await branchCtx.sleep("auction-timeout", 24 * 60 * 60 * 1000);
          return 0;
        },
      },
    ]);

    await ctx.step("finalize", async (step) => {
      await finalizeAuction("item-123", winner);
      step.state.result = winner === "bid" ? "sold" : "expired";
    });
  }),
  actions: {
    getState: (c) => c.state,
  },
});

async function listItem(itemId: string): Promise<void> {
  await fetch(`https://api.example.com/auctions/${itemId}`, {
    method: "POST",
  });
}

async function finalizeAuction(
  itemId: string,
  outcome: string,
): Promise<void> {
  await fetch(`https://api.example.com/auctions/${itemId}/finalize`, {
    method: "POST",
    body: JSON.stringify({ outcome }),
  });
}

export const registry = setup({ use: { auctionActor } });
```

client.ts:

```ts
import { createClient } from "rivetkit/client";
import type { registry } from "./index";

const client = createClient<typeof registry>("http://localhost:6420");
const handle = client.auctionActor.getOrCreate(["item-123"]);

await handle.send("bids", { amount: 100 });
console.log(await handle.getState());
```

## Fan-out / fan-in (join)

Use this when independent work items can run in parallel and you need a single merged result.

examples/docs/actors-workflows/fan-in-out.ts:

```ts
import { actor, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

export const fanInOutActor = actor({
  state: {
    total: 0,
  },
  run: workflow(async (ctx) => {
    await ctx.loop("join-loop", async (loopCtx) => {
        const [message] = await loopCtx.queue.nextBatch("wait-refresh", {
          timeout: 30_000,
        });

        if (!message) return;

        const joined = await loopCtx.join("parallel-work", {
          users: {
            run: async (branchCtx) =>
              await branchCtx.step("fetch-users", (branchCtx) => fetchCount("/users")),
          },
          orders: {
            run: async (branchCtx) =>
              await branchCtx.step("fetch-orders", (branchCtx) => fetchCount("/orders")),
          },
          invoices: {
            run: async (branchCtx) =>
              await branchCtx.step("fetch-invoices", (branchCtx) => fetchCount("/invoices")),
          },
        });

        await loopCtx.step("merge-results", async (step) => {
          step.state.total =
            joined.users + joined.orders + joined.invoices;
        });

      });
  }),
  actions: {
    getState: (c) => c.state,
  },
});

async function fetchCount(path: string): Promise<number> {
  const res = await fetch(`https://api.example.com${path}`);
  if (!res.ok) throw new Error(`fetch ${path} failed: ${res.status}`);
  return ((await res.json()) as { count: number }).count;
}

export const registry = setup({ use: { fanInOutActor } });
```

## Scatter-gather across actors

Use this when multiple actors can process independent parts of a request in parallel, then return a merged response.

examples/docs/actors-workflows/scatter-gather.ts:

```ts
import { actor, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

type ScatterMessage = { input: number };

export const shardActor = actor({
  actions: {
    compute: async (_c, input: number) => input * 10,
  },
});

export const scatterGatherActor = actor({
  state: {
    lastSum: 0,
  },
  run: workflow(async (ctx) => {
    await ctx.loop("scatter-gather-loop", async (loopCtx) => {
        const [message] = await loopCtx.queue.nextBatch("wait-scatter", {
          timeout: 30_000,
        });

        if (!message) return;
        const scatter = message.body as ScatterMessage;

        const gathered = await loopCtx.join("gather", {
          shardA: {
            run: async (joinCtx) =>
              await joinCtx.step("call-shard-a", async (step) =>
                callShard(step, "a", scatter.input),
              ),
          },
          shardB: {
            run: async (joinCtx) =>
              await joinCtx.step("call-shard-b", async (step) =>
                callShard(step, "b", scatter.input),
              ),
          },
          shardC: {
            run: async (joinCtx) =>
              await joinCtx.step("call-shard-c", async (step) =>
                callShard(step, "c", scatter.input),
              ),
          },
        });

        await loopCtx.step("aggregate", async (step) => {
          step.state.lastSum = gathered.shardA + gathered.shardB + gathered.shardC;
        });

      });
  }),
  actions: {
    getState: (c) => c.state,
  },
});

async function callShard(
  ctx: WorkflowStepContextOf<typeof scatterGatherActor>,
  shardId: "a" | "b" | "c",
  input: number,
): Promise<number> {
  const client = ctx.client();
  const handle = client.shardActor.getOrCreate([shardId]);
  return await handle.compute(input);
}

export const registry = setup({ use: { scatterGatherActor, shardActor } });
```

## Poll + backoff loop

Use this when an external dependency has variable availability and retries should slow down after failures.

examples/docs/actors-workflows/poll-backoff.ts:

```ts
import { actor, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

async function pollExternal(attempt: number): Promise<boolean> {
  return attempt % 3 === 0;
}

export const pollBackoffActor = actor({
  state: {
    attempts: 0,
    backoffMs: 100,
    status: "unknown" as "unknown" | "healthy" | "retrying",
  },
  run: workflow(async (ctx) => {
    await ctx.loop("poll-loop", async (loopCtx) => {
        const success = await loopCtx.step("poll-target", async (step) => {
          step.state.attempts += 1;
          return pollExternal(step.state.attempts);
        });

        if (success) {
          await loopCtx.step("reset-backoff", async (step) => {
            step.state.status = "healthy";
            step.state.backoffMs = 100;
          });
          await loopCtx.sleep("healthy-interval", 1_000);
          return;
        }

        const retryDelay = await loopCtx.step("grow-backoff", async (ctx) => {
          ctx.state.status = "retrying";
          ctx.state.backoffMs = Math.min(ctx.state.backoffMs * 2, 5_000);
          return ctx.state.backoffMs;
        });

        await loopCtx.sleep("retry-delay", retryDelay);
      });
  }),
  actions: {
    getState: (c) => c.state,
  },
});

export const registry = setup({ use: { pollBackoffActor } });
```

## Coordinator -> worker RPC

Use this when one actor orchestrates work by calling actions on other actors.

examples/docs/actors-workflows/coordinator-worker.ts:

```ts
import { actor, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

type TaskMessage = {
  taskId: string;
  workerId: string;
  value: number;
};

export const workerActor = actor({
  actions: {
    runTask: async (_c, value: number) => value * 2,
  },
});

export const coordinatorActor = actor({
  state: {
    lastTaskId: null as string | null,
    lastResult: 0,
  },
  run: workflow(async (ctx) => {
    await ctx.loop("orchestrator-loop", async (loopCtx) => {
        const [message] = await loopCtx.queue.nextBatch("wait-task", {
          timeout: 30_000,
        });

        if (!message) return;
        const task = message.body as TaskMessage;

        const result = await loopCtx.step("dispatch-rpc", async (step) =>
          dispatchTask(step, task),
        );

        await loopCtx.step("record-result", async (step) => {
          step.state.lastTaskId = task.taskId;
          step.state.lastResult = result as number;
        });

      });
  }),
  actions: {
    getState: (c) => c.state,
  },
});

async function dispatchTask(
  ctx: WorkflowStepContextOf<typeof coordinatorActor>,
  task: TaskMessage,
): Promise<number> {
  const client = ctx.client();
  const worker = client.workerActor.getOrCreate([task.workerId]);
  return await worker.runTask(task.value);
}

export const registry = setup({ use: { coordinatorActor, workerActor } });
```

## Child worker orchestration

Use this when one workflow coordinates many child workers (actors or worker workflows) and manages their lifecycle.

examples/docs/actors-workflows/child-worker-orchestration.ts:

```ts
import { actor, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

type BatchMessage = { payload: number };

export const childWorkerActor = actor({
  actions: {
    process: async (_c, payload: number) => payload * 3,
  },
});

export const orchestratorActor = actor({
  state: {
    lastTotal: 0,
  },
  run: workflow(async (ctx) => {
    await ctx.step("start-children", async (step) => startChildren(step));

    await ctx.loop("orchestrate-loop", async (loopCtx) => {
        const [message] = await loopCtx.queue.nextBatch("wait-batch", {
          timeout: 30_000,
        });

        if (!message) return;
        const batch = message.body as BatchMessage;

        const results = await loopCtx.join("collect-updates", {
          a: {
            run: async (joinCtx) =>
              await joinCtx.step("run-child-a", async (step) =>
                runChildWorker(step, "child-a", batch.payload),
              ),
          },
          b: {
            run: async (joinCtx) =>
              await joinCtx.step("run-child-b", async (step) =>
                runChildWorker(step, "child-b", batch.payload),
              ),
          },
          c: {
            run: async (joinCtx) =>
              await joinCtx.step("run-child-c", async (step) =>
                runChildWorker(step, "child-c", batch.payload),
              ),
          },
        });

        await loopCtx.step("reconcile", async (step) => {
          step.state.lastTotal = results.a + results.b + results.c;
        });

      });
  }),
  actions: {
    getState: (c) => c.state,
  },
});

async function startChildren(
  ctx: WorkflowStepContextOf<typeof orchestratorActor>,
): Promise<void> {
  const client = ctx.client();
  await client.childWorkerActor.getOrCreate(["child-a"]).process(0);
  await client.childWorkerActor.getOrCreate(["child-b"]).process(0);
  await client.childWorkerActor.getOrCreate(["child-c"]).process(0);
}

async function runChildWorker(
  ctx: WorkflowStepContextOf<typeof orchestratorActor>,
  workerId: "child-a" | "child-b" | "child-c",
  payload: number,
): Promise<number> {
  const client = ctx.client();
  return await client.childWorkerActor.getOrCreate([workerId]).process(payload);
}

export const registry = setup({ use: { orchestratorActor, childWorkerActor } });
```

## Timeout + fallback actor

Use this when a primary actor call might be slow or unavailable and you need a deterministic fallback path.

examples/docs/actors-workflows/timeout-fallback.ts:

```ts
import { actor, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

export const primaryServiceActor = actor({
  actions: {
    fetchValue: async () => {
      await new Promise((resolve) => setTimeout(resolve, 500));
      return "primary";
    },
  },
});

export const fallbackServiceActor = actor({
  actions: {
    fetchValue: async () => "fallback",
  },
});

export const timeoutFallbackActor = actor({
  state: {
    lastSource: "none" as "none" | "primary" | "fallback",
    lastValue: "",
  },
  run: workflow(async (ctx) => {
    await ctx.loop("timeout-loop", async (loopCtx) => {
        await loopCtx.queue.nextBatch("wait-request", {
          timeout: 30_000,
        });

        const winner = await loopCtx.race("primary-vs-timeout", [
          {
            name: "primary",
            run: async (raceCtx) =>
              await raceCtx.step("call-primary", async (step) =>
                callPrimaryValue(step),
              ),
          },
          {
            name: "timeout",
            run: async (raceCtx) => {
              await raceCtx.sleep("primary-timeout", 200);
              return "timeout";
            },
          },
        ]);

        let value = winner.value as string;
        let source: "primary" | "fallback" = "primary";

        if (winner.winner === "timeout") {
          value = (await loopCtx.step("fallback-call", async (step) =>
            callFallbackValue(step),
          )) as string;
          source = "fallback";
        }

        await loopCtx.step("record-choice", async (step) => {
          step.state.lastSource = source;
          step.state.lastValue = value;
        });

      });
  }),
  actions: {
    getState: (c) => c.state,
  },
});

async function callPrimaryValue(
  ctx: WorkflowStepContextOf<typeof timeoutFallbackActor>,
): Promise<string> {
  const client = ctx.client();
  const primary = client.primaryServiceActor.getOrCreate(["main"]);
  return await primary.fetchValue();
}

async function callFallbackValue(
  ctx: WorkflowStepContextOf<typeof timeoutFallbackActor>,
): Promise<string> {
  const client = ctx.client();
  const fallback = client.fallbackServiceActor.getOrCreate(["main"]);
  return await fallback.fetchValue();
}

export const registry = setup({
  use: { timeoutFallbackActor, primaryServiceActor, fallbackServiceActor },
});
```