Concepts

Failure & Recovery

Error hooks, rollback, and compensating actions across actors.

Error hooks

Use onError when you want a best-effort notification for workflow failures.

Step failures include the attempt number, retry counts, whether the step will retry, and the next retry delay.
Workflow failures also include terminal errors outside steps, such as rollback failures or code/history mismatches.
The hook is observational. It is not part of workflow replay, so use it for logging, metrics, or updating non-critical actor state.
This is also a good place to forward workflow failures to Sentry or another error reporting pipeline.

import { actor, event, setup } from "rivetkit";
import { workflow, type WorkflowErrorEvent } from "rivetkit/workflow";

export const errorHookActor = actor({
  state: {
    lastError: null as WorkflowErrorEvent | null,
  },
  events: {
    workflowError: event<[WorkflowErrorEvent]>(),
  },
  run: workflow(
    async (ctx) => {
      await ctx.step({
        name: "sync-ledger",
        maxRetries: 3,
        retryBackoffBase: 250,
        retryBackoffMax: 1_000,
        run: async (ctx) => {
          throw new Error("ledger unavailable");
        },
      });
    },
    {
      onError: (c, event) => {
        c.state.lastError = event;
        c.broadcast("workflowError", event);
      },
    },
  ),
  actions: {
    getState: (c) => c.state,
  },
});

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

Rollback

Use rollback checkpoints before steps that have compensating actions.

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

export const checkoutActor = actor({
  state: { status: "pending" as string },
  queues: {
    orders: queue<{ orderId: string }>(),
  },
  run: workflow(async (ctx) => {
    await ctx.loop("checkout-loop", async (loopCtx) => {
        const message = await loopCtx.queue.next("wait-order");

        await loopCtx.rollbackCheckpoint("checkout-checkpoint");

        await loopCtx.step<string>({
          name: "reserve-inventory",
          run: (loopCtx) => reserveInventory(message.body.orderId),
          rollback: async (_rollbackCtx, id) => {
            await releaseInventory(id as string);
          },
        });

        await loopCtx.step<string>({
          name: "charge-card",
          run: (loopCtx) => chargeCard(message.body.orderId),
          rollback: async (_rollbackCtx, chargeId) => {
            await refundCharge(chargeId as string);
          },
        });

        await loopCtx.step("confirm", async (step) => {
          step.state.status = "confirmed";
        });

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

async function reserveInventory(orderId: string): Promise<string> {
  const res = await fetch("https://api.example.com/inventory/reserve", {
    method: "POST",
    body: JSON.stringify({ orderId }),
  });
  return ((await res.json()) as { reservationId: string }).reservationId;
}

async function releaseInventory(reservationId: string): Promise<void> {
  await fetch(`https://api.example.com/inventory/${reservationId}/release`, {
    method: "POST",
  });
}

async function chargeCard(orderId: string): Promise<string> {
  const res = await fetch("https://api.stripe.com/v1/charges", {
    method: "POST",
    headers: { Authorization: `Bearer ${process.env.STRIPE_KEY}` },
    body: JSON.stringify({ orderId }),
  });
  return ((await res.json()) as { id: string }).id;
}

async function refundCharge(chargeId: string): Promise<void> {
  await fetch("https://api.stripe.com/v1/refunds", {
    method: "POST",
    headers: { Authorization: `Bearer ${process.env.STRIPE_KEY}` },
    body: JSON.stringify({ charge: chargeId }),
  });
}

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

Cross-actor saga (compensating actions)

Use this when a workflow spans multiple actors and each side effect may need compensation.

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

type CheckoutMessage = {
  orderId: string;
  amount: number;
};

export const inventoryActor = actor({
  actions: {
    reserve: async (_c, orderId: string) => `reserve-${orderId}`,
    release: async (_c, reservationId: string) => reservationId,
  },
});

export const billingActor = actor({
  actions: {
    charge: async (_c, amount: number) => `charge-${amount}`,
    refund: async (_c, chargeId: string) => chargeId,
  },
});

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

        if (!message) return;
        const checkout = message.body as CheckoutMessage;

        await loopCtx.rollbackCheckpoint("checkout-saga");

        await loopCtx.step({
          name: "reserve-inventory",
          run: async (ctx) => reserveInventoryForCheckout(ctx, checkout.orderId),
          // Rollback callbacks only receive a rollback context, not actor
          // APIs like client(). Compensate with direct external calls.
          rollback: async (_rollbackCtx, output) => {
            await releaseInventoryForCheckout(output as string);
          },
        });

        await loopCtx.step({
          name: "charge-card",
          run: async (ctx) => chargeCheckout(ctx, checkout.amount),
          rollback: async (_rollbackCtx, output) => {
            await refundCheckout(output as string);
          },
        });

        await loopCtx.step("mark-complete", async (step) =>
          markOrderComplete(step),
        );

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

async function reserveInventoryForCheckout(
  ctx: WorkflowStepContextOf<typeof checkoutSagaActor>,
  orderId: string,
): Promise<string> {
  const client = ctx.client();
  const inventory = client.inventoryActor.getOrCreate(["main"]);
  return await inventory.reserve(orderId);
}

async function releaseInventoryForCheckout(
  reservationId: string,
): Promise<void> {
  await fetch("https://api.example.com/inventory/release", {
    method: "POST",
    body: JSON.stringify({ reservationId }),
  });
}

async function chargeCheckout(
  ctx: WorkflowStepContextOf<typeof checkoutSagaActor>,
  amount: number,
): Promise<string> {
  const client = ctx.client();
  const billing = client.billingActor.getOrCreate(["main"]);
  return await billing.charge(amount);
}

async function refundCheckout(
  chargeId: string,
): Promise<void> {
  await fetch("https://api.example.com/billing/refund", {
    method: "POST",
    body: JSON.stringify({ chargeId }),
  });
}

function markOrderComplete(
  ctx: WorkflowStepContextOf<typeof checkoutSagaActor>,
): void {
  ctx.state.completedOrders += 1;
}

export const registry = setup({
  use: { checkoutSagaActor, inventoryActor, billingActor },
});