Workflows

A workflow is a function that runs steps, sleeps, and waits for people and events. You write it top to bottom as ordinary code, and it keeps going through crashes, deploys and restarts. While a run waits, no worker holds it, and the waiting isn't billed.

A run of a workflow is a job that waits. Each step it calls is a job of its own, which any worker that serves the step's name can do, in any language. While the run waits for a step, a sleep or a signal, no worker holds it. When the wait is over, a worker picks the run up and it carries on from where it was.

Workflows are in both SDKs. Workers that run workflows need Node.js 20.16 or later, or Bun, in TypeScript, and Go 1.23 or later in Go. Starting and signalling runs works from anywhere, curl included.

A first workflow

A trial that sends a welcome email, waits up to 14 days for the customer to upgrade, and sends a reminder if they don't:

import { Emit } from "@emit.run/sdk";
import { z } from "zod";

const emit = new Emit({ url: process.env.EMIT_URL!, token: process.env.EMIT_TOKEN!, space: "accounts" });

const Email = z.object({ to: z.string(), template: z.string() });

// A step: ordinary work with a name, served by any worker that defines it with a handler.
const sendEmail = emit.define("send-email", { input: Email }, async ({ to, template }, job) => {
  await job.log("info", `sending ${template} to ${to}`);
  // ...call your email provider...
  return { sent: true };
});

export const trial = emit.workflow(
  "trial",
  { input: z.object({ userId: z.string(), email: z.string() }) },
  async (user, wf) => {
    await sendEmail.run({ to: user.email, template: "welcome" });

    const upgrade = await wf.wait("upgraded", { timeout: "14d" }); // undefined if 14 days pass first
    if (upgrade !== undefined) return { converted: true };

    await sendEmail.run({ to: user.email, template: "trial-ended" });
    return { converted: false };
  },
);

emit.serve(); // this process now runs send-email steps and trial runs

package main

import (
	"context"
	"log"
	"os"
	"os/signal"
	"time"

	emit "github.com/emit-run/sdk-go"
)

type Email struct {
	To       string `json:"to"`
	Template string `json:"template"`
}

type User struct {
	UserID string `json:"userId"`
	Email  string `json:"email"`
}

type Outcome struct {
	Converted bool `json:"converted"`
}

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
	defer stop()
	c := emit.New(emit.Options{URL: os.Getenv("EMIT_URL"), Token: os.Getenv("EMIT_TOKEN"), Space: "accounts"})

	// A step: ordinary work with a name, served by any worker that defines it with a handler.
	sendEmail := emit.Define(c, "send-email", func(ctx context.Context, e Email, job *emit.HeldJob) (bool, error) {
		// ...call your email provider...
		return true, job.Log(ctx, emit.LevelInfo, "sending "+e.Template+" to "+e.To)
	})

	emit.Workflow(c, "trial", func(ctx context.Context, user User, wf *emit.Run) (Outcome, error) {
		if _, err := sendEmail.Run(ctx, Email{To: user.Email, Template: "welcome"}); err != nil {
			return Outcome{}, err
		}
		_, upgraded, err := emit.Wait[map[string]string](ctx, "upgraded", 14*24*time.Hour)
		if err != nil {
			return Outcome{}, err
		}
		if upgraded {
			return Outcome{Converted: true}, nil
		}
		if _, err := sendEmail.Run(ctx, Email{To: user.Email, Template: "trial-ended"}); err != nil {
			return Outcome{}, err
		}
		return Outcome{Converted: false}, nil
	})

	worker := c.Serve(ctx, emit.WorkerOptions{Concurrency: 8}) // runs send-email steps and trial runs
	<-ctx.Done()
	if err := worker.Stop(); err != nil {
		log.Fatal(err)
	}
}

Start a run when someone signs up, and signal it when they upgrade:

// in your sign-up handler
await trial.enqueue({ userId: "usr_42", email: "ada@example.com" }, { key: "trial-usr_42" });

// in your billing webhook
await trial.signal({ key: "trial-usr_42" }, "upgraded", { plan: "pro" });

// trial is what emit.Workflow returned.
id, err := trial.Enqueue(ctx, User{UserID: "usr_42", Email: "ada@example.com"}, emit.WithKey("trial-usr_42"))

// in your billing webhook
took, err := trial.Signal(ctx, emit.ByKey("trial-usr_42"), "upgraded", map[string]string{"plan": "pro"})

# A run is a job with the workflow's name and "workflow": true.
curl -s "$EMIT_URL/v1/spaces/accounts/jobs" \
  -H "Authorization: Bearer $EMIT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name": "trial", "payload": {"userId": "usr_42", "email": "ada@example.com"}, "key": "trial-usr_42", "workflow": true}'

curl -s "$EMIT_URL/v1/spaces/accounts/jobs:signal" \
  -H "Authorization: Bearer $EMIT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"key": "trial-usr_42", "name": "upgraded", "data": {"plan": "pro"}}'
# {"delivered":[true],"ids":["01M3QSR386017JCY0DC11GTG7D"]}

Steps

A step is ordinary work, made with define (emit.Define in Go), with a handler: the function that does the work. Nothing about it is special to workflows: the same send-email can be run from a workflow, from an API route, or enqueued from a script. Inside a workflow, run makes the step a job of the run and returns its result.

A handler means "this process does this work." serve() runs every name defined with a handler in the process. If another service does the work, define the step without one. The name and the input and output are the contract, and whichever worker serves that name runs it, in any language:

// Served by a Go service: this process only calls it.
const transcode = emit.define("transcode", {
  input: z.object({ url: z.string(), height: z.number() }),
  output: z.object({ url: z.string() }),
});

// Served by a TypeScript service: this process only calls it.
transcode := emit.Ref[Transcode, Video](c, "transcode")

Don't give a step that runs elsewhere a placeholder handler that throws. This process would then take those jobs and fail them.

Step options are the usual job options: timeoutMs, maxAttempts, backoffMs and delayMs, plus label, a few words (at most 64 bytes) shown on the step's row when you look at the run, such as the photo's file name. A step belongs to its run, so it takes no key.

Writing the function

A run can be picked up many times: after each wait, and after a crash or a deploy. Each time, your function is called again with the same input, and every step that already finished returns its saved result at once, without running again. So the steps run once, but the code between them can run more than once. Four rules follow.

1. Put anything that touches the outside world in a step, or in wf.do. Charging a card, sending an email, writing to your database. Code outside a step may run again.

   TypeScriptGo

   Copy

   await mailer.send(user.email, "welcome");                                // wrong: may be sent twice
   await wf.do("welcome-email", () => mailer.send(user.email, "welcome"));  // right: sent once

   mailer.Send(user.Email, "welcome") // wrong: may be sent twice
   
   // right: sent once
   _, err := emit.Do(ctx, "welcome-email", func(ctx context.Context) (bool, error) {
       return true, mailer.Send(user.Email, "welcome")
   })

   wf.do (emit.Do in Go) runs a function in this process and keeps what it returns, so later runs of the function get the saved value instead of calling it again. If it throws, it's tried again, three times in all unless you say otherwise. If the worker dies while the function is running, the next attempt calls it again, so pass an idempotency key to the service where it takes one. Use wf.do for small things that don't deserve a step of their own. Its name must be unique within the run: in a loop, add the index, like `reminder-${n}`.

2. Decide only on the input and on what steps returned. In TypeScript, Date and Math.random() inside a workflow are made safe for you: they give the same answers every time the function runs. In Go, use wf.Now() and wf.Rand() instead of time.Now() and math/rand. Anything else that can change, such as crypto.randomUUID(), an environment variable or a database read, belongs in a step or wf.do.

3. Call steps in the same order for the same results. Steps are matched to their saved results by order: the first send-email call, the second, and so on. Loops and parallel calls are fine, as long as the same input and the same step results lead to the same calls.

4. Let errors you don't recognise go up. Catch a step's failure (JobFailed) to handle it, and rethrow everything else.

In TypeScript, also: don't start timers or listeners in the body, and await only emit's calls (steps, wf.do, wf.sleep, wf.wait, wf.race, wf.note) and Promise.all or Promise.allSettled of them. Put a fetch or any other promise in a wf.do.

In Go, also: pass the workflow's ctx to every call, don't start goroutines in the function (use Start and emit.All for parallel steps), and keep side effects out of defer, which runs each time the run pauses.

Plain steps have none of these rules. A step's handler runs once per attempt, like any job.

Sleeping

wf.sleep pauses the run for a duration: a number of milliseconds, or a string such as "30s", "2h" or "7d". wf.sleepUntil pauses it until a date. In Go, wf.Sleep(ctx, d) and wf.SleepUntil(ctx, t). A sleep is at most 366 days.

A sleeping run holds no worker and costs nothing. The sleep ends on time whether or not a worker is up, and the run then waits for a worker like any job.

Signals

A signal is how your own backend tells a run that something happened: a payment arrived, a document was signed. The run waits for it by name, with an optional timeout:

const Payment = z.object({ invoiceId: z.string(), amountCents: z.number() });

// in the workflow: undefined if 7 days pass first
const payment = await wf.wait("invoice-paid", { timeout: "7d", schema: Payment });
if (!payment) return await sendDunningEmail.run(invoice);

// in the workflow: paid is false if 7 days pass first
payment, paid, err := emit.Wait[Payment](ctx, "invoice-paid", 7*24*time.Hour)
if err != nil {
	return Result{}, err
}
if !paid {
	return sendDunningEmail.Run(ctx, invoice)
}

Send a signal to a run by its id, or by the key or singleton it was started with:

const took = await subscription.signal({ key: "sub-cus_318" }, "invoice-paid", { invoiceId: "inv_77", amountCents: 4900 });

took, err := subscription.Signal(ctx, emit.ByKey("sub-cus_318"), "invoice-paid", Payment{InvoiceID: "inv_77", AmountCents: 4900})

curl -s "$EMIT_URL/v1/spaces/billing/jobs:signal" \
  -H "Authorization: Bearer $EMIT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"key": "sub-cus_318", "name": "invoice-paid", "data": {"invoiceId": "inv_77", "amountCents": 4900}}'

• A signal sent before the run reaches its wait is kept, not lost. Several signals of one name are taken in the order they arrived, one per wait.
• The call returns true if a run took the signal, and false if there's no unfinished run to take it.
• A signal is delivered at least once: if you send it again because the answer was lost, the run's next wait of that name takes it again. Send data a second delivery can't hurt, or include your own event id.
• Signal, or start. signal({ singleton }, name, data, { start: input }) (Go: SignalOrStart) delivers the signal to the open run, or, if there's none, starts a run with input instead. One call, with no race between "is it running?" and "start it".

Use signals for your own backend. For a person clicking a button, or a service calling a webhook, use a link.

Running steps in parallel

Start steps together and wait for all of them:

const [hd, sd] = await Promise.all([
  transcode.run({ url, height: 1080 }),
  transcode.run({ url, height: 480 }),
]);

const thumbnails = await Promise.all(photos.map((photo) => thumbnail.run(photo, { label: photo.name })));

thumbnails, err := emit.All(ctx, thumbnail, photos) // one step per photo, all at once

Steps started together are created together, run in parallel on whatever workers serve them, and the run wakes once, when the last one finishes. Promise.all and emit.All fail with the first step that fails. Use Promise.allSettled (Go: emit.AllSettled) to carry on when some fail.

A run with more than about 10,000 steps gets slower to pick up. For more, split the work into child workflows, each handling a batch, and run those in parallel.

Pass references, not data. Inputs and results are at most 4 MiB each. When steps hand each other files or large JSON, store them in object storage or your database and pass the key.

Steps in the background, and races

step.start() starts a step without waiting for it, and returns a handle you can await later or cancel. wf.race waits for whichever of several started steps, waits and sleeps finishes first:

const first = await wf.race({
  answer: customerReply.start(quote),
  revised: wf.wait("quote-revised"),
  expired: wf.sleep("14d"),
});
if ("expired" in first) return { status: "expired" };

Use wf.race, never Promise.race, with steps: emit decides which finished first, so the answer is the same every time the function runs. In Go, emit.Race with emit.Case, emit.WaitCase and emit.SleepCase.

When a run finishes, steps it started and didn't wait for are cancelled. Steps it started with enqueue keep going on their own.

People and outside services

Some steps are finished by someone outside your system: a manager approving a refund, an e-signature service calling back. Write them as a step that hands out a link and lets go:

const approveRefund = emit.define(
  "approve-refund",
  { input: z.object({ amountCents: z.number(), customer: z.string(), teamLead: z.string(), manager: z.string() }), output: z.boolean() },
  async (refund, job) => {
    const who = job.attempt === 1 ? refund.teamLead : refund.manager; // no answer in a day? ask the manager
    await chat.post(who, `Refund ${refund.amountCents / 100} to ${refund.customer}?`, {
      approve: await job.link.complete(true),
      deny: await job.link.complete(false),
    });
    return job.detach(); // the step stays open for a day; a link finishes it
  },
);

// in a workflow: a day for the team lead, then a day for the manager
const approved = await approveRefund.run(refund, { maxAttempts: 2 });

approveRefund := emit.Define(c, "approve-refund", func(ctx context.Context, r Refund, job *emit.HeldJob) (bool, error) {
	who := r.TeamLead
	if job.Attempt > 1 { // no answer in a day? ask the manager
		who = r.Manager
	}
	yes, err := job.Link.Complete(ctx, true)
	if err != nil {
		return false, err
	}
	no, err := job.Link.Complete(ctx, false)
	if err != nil {
		return false, err
	}
	chat.Post(who, fmt.Sprintf("Refund %d to %s?", r.AmountCents/100, r.Customer), yes, no)
	return false, emit.Detach() // the step stays open for a day; a link finishes it
})

// in a workflow: a day for the team lead, then a day for the manager
approved, err := approveRefund.Run(ctx, refund, emit.WithMaxAttempts(2))

• job.link.complete(value) is a URL. Opening it shows a confirmation page, and confirming completes the step with value. job.link.fail(reason) fails it instead. job.link.webhook() completes the step with whatever JSON is POSTed to it: give it to a payment provider or an e-signature service.
• Links need no API token and can't be forged. Each works once, for one attempt of one step. A link of an earlier attempt, or of a step that already finished, does nothing.
• A detached step holds no worker. Each attempt lasts a day, or what detach({ for: "3d" }) says, up to 366 days. When an attempt runs out, the handler runs again with job.attempt one higher: send a reminder, or ask someone else. After the last attempt the step fails, and run throws JobFailed: catch it to handle "nobody answered".
• A link expires with its attempt, or sooner if you say: job.link.complete(true, { expires: "2h" }).

When a link finishes the step, the step's result is the link's value, so don't also wait for a signal about the same thing.

Starting, keys and cancelling

enqueue starts a run and returns its id. run starts one and waits for its result. result(id) waits for a run started earlier.

Two options stop a second run from starting:

trial.cancel(id) (Go: trial.Cancel) cancels a run and everything it started, all the way down: its steps, its child workflows and their steps. Cancelling from outside a run needs a token with the admin scope.

Watching a run

In the dashboard, a space's Runs tab shows each run as a timeline: its steps, sleeps and signals, what it's waiting for, and its notes. From a terminal, emit runs tree <id> --follow draws the same, live.

A note is a line you write for people to read: wf.note("Payment confirmed") in a workflow, or job.note(...) in a step. Notes show on the run's timeline, including to a browser following the run without an API token. See Show progress in a browser.

Failures

• A step is retried by its own policy. If it fails for good, run throws JobFailed (Go: returns *emit.JobFailed). Catch it to undo what came before:

  TypeScriptCopy

  const receipt = await charge.run(order);
  try {
    await ship.run(order);
  } catch (error) {
    if (!(error instanceof JobFailed)) throw error;
    await refundCharge.run(receipt);
    return { status: "refunded" };
  }

• A step nobody serves, because of a typo or a service that's gone, fails after 7 days unclaimed.

• A JobFailed you don't catch fails the run for good.

• Any other error fails this attempt of the run. The run is tried again by its own retry policy, and the steps that finished stay finished. Throw Unrecoverable to fail it for good.

Changing a workflow

emit doesn't store your workflow's code: a run that continues after a deploy runs the new code.

Serve each workflow from one language. Steps can be served from any.

Limits

A workflow that loops forever, like a nightly report, should start itself over now and then with return wf.continueAsNew(input) (Go: return Out{}, wf.ContinueAsNew(input)): same run id, fresh history.

Billing

On plans that charge per job, every job counts: a run, each step, each wf.do and each signal. Sleeping, waiting and picking a run back up are free.