Concepts

Jobs

What a job is, the statuses it moves through, and how retries, delays, idempotency keys and callbacks work.

A job is one piece of work: a name that says what kind of work it is, like charge-invoice, and a JSON payload with what the worker needs, like {"invoiceId": "inv_1042"}. emit gives each job an id, a 26-character string such as 01M3QSGN0103P7EGVBX9D5NTSN, and keeps it in a space.

A worker claims jobs by name, so one space can hold many kinds of work, each served by the workers that know how to do it.

Statuses

A job is always in one of five statuses.

Status	What it means
pending	Waiting for a worker, or for a time: a delay, or the wait before a retry.
running	A worker holds it and is working on it.
completed	The worker finished it. Its result is what the worker returned.
failed	It ran out of attempts, or a worker failed it for good. Its result is the last error.
cancelled	Someone cancelled it before it finished.

The last three are final: a job that has finished never changes again.

text

pending ──claimed──▶ running ──returned──▶ completed
   ▲                    │
   └── failed, with ────┤──── out of attempts, or failed for good ──▶ failed
       attempts left    │
anything unfinished ──cancelled──▶ cancelled

Creating a job

Create a job with its name and payload. Everything else is optional.

TypeScriptGocurl

const { id, outcome } = await emit.jobs.create(
  "charge-invoice",
  { invoiceId: "inv_1042", amountCents: 4900 },
  {
    key: "charge-inv_1042", // a repeat returns this job instead of making another
    delayMs: 60_000,        // not before a minute from now
    maxAttempts: 5,         // the first try and up to four retries
    timeoutMs: 120_000,     // how long one attempt may take
  },
);

created, err := c.Jobs.Create(ctx, emit.NewJob{
	Name:        "charge-invoice",
	Payload:     Invoice{ID: "inv_1042", AmountCents: 4900},
	Key:         "charge-inv_1042", // a repeat returns this job instead of making another
	DelayMs:     60_000,            // not before a minute from now
	MaxAttempts: 5,                 // the first try and up to four retries
	TimeoutMs:   120_000,           // how long one attempt may take
})
if err != nil {
	return err
}
log.Printf("job %s (%s)", created.ID, created.Outcome)

curl -s "$EMIT_URL/v1/spaces/billing/jobs" \
  -H "Authorization: Bearer $EMIT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "charge-invoice",
    "payload": {"invoiceId": "inv_1042", "amountCents": 4900},
    "key": "charge-inv_1042",
    "delayMs": 60000,
    "maxAttempts": 5,
    "timeoutMs": 120000
  }'

To create many jobs, send them in one call: emit.jobs.createMany([...]) in TypeScript, c.Jobs.CreateMany in Go, or a JSON array to the same endpoint. The answers come back in the order you sent the jobs. One call with a hundred jobs is much faster than a hundred calls.

A payload is at most 4 MiB of JSON. For anything larger, store the data yourself and put its URL or key in the payload.

Retries

A job gets attempts: four unless you or its space say otherwise, counting the first. When an attempt fails and attempts are left, the job goes back to pending and waits before it's tried again. The first wait is one second and each one after doubles, up to an hour. Change the policy for a space in its settings, or for one job with backoffMs (the first wait) and backoff (how the waits grow: exponential, linear or constant, with a ceiling and optional jitter).

What your handler does decides what happens to the job:

The handler	The job
returns a value	completes, with that value as its result
throws, or returns an error	fails this attempt, and is retried while attempts are left
throws `Unrecoverable`	fails for good, whatever attempts are left
throws `RetryAfter`	fails this attempt, and is retried after the wait you give (at most a day) instead of the policy's
throws `Release`	goes back to the queue at once, and the attempt isn't counted

TypeScriptGocurl

import { RetryAfter, Unrecoverable } from "@emit.run/sdk";

type Invoice = { invoiceId: string; amountCents: number };

emit.worker<Invoice>({ names: ["charge-invoice"] }, async (job) => {
  const response = await fetch("https://payments.example.com/v1/charges", {
    method: "POST",
    headers: { "content-type": "application/json", "idempotency-key": job.payload.invoiceId },
    body: JSON.stringify(job.payload),
    signal: job.signal, // stop if emit gives this job to another worker
  });
  if (response.status === 429) {
    const seconds = Number(response.headers.get("retry-after") ?? "30");
    throw new RetryAfter("rate limited by the payment provider", seconds * 1000);
  }
  if (response.status === 402) throw new Unrecoverable("card declined");
  if (!response.ok) throw new Error(`payment provider answered ${response.status}`);
  return await response.json(); // the job's result
});

worker := c.Worker(ctx, emit.WorkerOptions{Names: []string{"charge-invoice"}}, func(ctx context.Context, job *emit.HeldJob) (any, error) {
	request, err := http.NewRequestWithContext(ctx, "POST", "https://payments.example.com/v1/charges", bytes.NewReader(job.Payload))
	if err != nil {
		return nil, emit.Unrecoverable(err)
	}
	request.Header.Set("Content-Type", "application/json")
	response, err := http.DefaultClient.Do(request)
	if err != nil {
		return nil, err // a network error: try again later
	}
	defer response.Body.Close()
	switch {
	case response.StatusCode == http.StatusTooManyRequests:
		seconds, err := strconv.Atoi(response.Header.Get("Retry-After"))
		if err != nil || seconds <= 0 {
			seconds = 30
		}
		return nil, emit.RetryAfter(errors.New("rate limited by the payment provider"), time.Duration(seconds)*time.Second)
	case response.StatusCode == http.StatusPaymentRequired:
		return nil, emit.Unrecoverable(errors.New("card declined"))
	case response.StatusCode >= 300:
		return nil, fmt.Errorf("payment provider answered %d", response.StatusCode)
	}
	var charge map[string]any
	if err := json.NewDecoder(response.Body).Decode(&charge); err != nil {
		return nil, err
	}
	return charge, nil // the job's result
})

# Fail this attempt and ask for the next one in 30 seconds:
curl -s "$EMIT_URL/v1/spaces/billing/jobs:report" \
  -H "Authorization: Bearer $EMIT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"id": "01M3QSMMXQ00ZSGYA6VWM3NGQA", "attempt": 1, "fail": {"error": "rate limited", "retryAfterMs": 30000}}'

# Fail it for good:
curl -s "$EMIT_URL/v1/spaces/billing/jobs:report" \
  -H "Authorization: Bearer $EMIT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"id": "01M3QSMMXQ00ZSGYA6VWM3NGQA", "attempt": 1, "fail": {"error": "card declined", "unrecoverable": true}}'

An error message is kept up to 4 KiB. A result is at most 4 MiB of JSON: a larger one fails the attempt.

Leases and timeouts

A worker holds a job under a lease, which starts as the job's timeout: five minutes unless you or its space say otherwise. The SDKs renew the lease while your handler runs, so a long job is fine as long as the worker is alive.

If the worker dies or loses its connection, the lease runs out, the attempt counts as failed, and another worker gets the next attempt. So a crashed job is picked up again within its timeout. The first worker may also be alive but slow, and still running when the second starts. The SDKs tell it to stop (job.signal aborts in TypeScript; the handler's ctx is cancelled in Go), and emit ignores anything more it reports.

Make handlers safe to run twice

A job can run more than once: after a crash, a lost lease or a retry. Write each handler so that a second run does no harm. Pass an idempotency key to the services you call, as the payment example above does, or check whether the work is already done.

Delays and schedules

A job can wait before it's first handed out. Give either delayMs, a wait from now, or runAtMs, a time in Unix milliseconds. Either can be up to a year away. A delayed job is pending until its time, then queues like any other.

Idempotency keys

A key makes a create safe to repeat. Create a job with a key that an earlier job in the space already has, and emit doesn't make a second job: it answers with the earlier job's id, and outcome says duplicate instead of new (DUPLICATE and NEW over HTTP).

Use a key whenever you might send the same create twice, for example when you retry after a timeout: charge-inv_1042 for the charge of invoice 1042.

A key holds while its job is unfinished, and for a window after the job finishes. On emit cloud that window is 10 minutes. A Dedicated cluster can raise it to a day in its node settings. After the window, the same key makes a new job.

Jobs nobody claims

A job that no worker claims for 7 days after it could first be claimed fails for good, with the error not claimed within 7d. It catches a typo in a job name, or a service you retired, instead of leaving jobs waiting forever. Change the window, or turn it off, with the space's unclaimed setting.

Progress, logs and events

While it runs, a worker can report on the job:

Progress: a percentage and a short message. Only the latest is kept.
Log lines: a level (trace to fatal; upper case over HTTP) and a message of up to 4 KiB each, kept in order and read back a page at a time.
Events: small JSON objects, up to 4 KiB each and 1,000 a job unless the space says otherwise, all kept in order.

TypeScriptGocurl

emit.worker<{ importId: string; rows: number }>({ names: ["import-contacts"] }, async (job) => {
  await job.log("info", "import started", { metadata: { rows: job.payload.rows } });
  for (let done = 0; done < job.payload.rows; done += 500) {
    // ...import the next 500 rows...
    await job.progress(Math.round((done / job.payload.rows) * 100), `${done} rows`);
  }
  await job.event({ stage: "imported" });
  return { imported: job.payload.rows };
});

worker := c.Worker(ctx, emit.WorkerOptions{Names: []string{"import-contacts"}}, func(ctx context.Context, job *emit.HeldJob) (any, error) {
	var in struct {
		ImportID string `json:"importId"`
		Rows     int    `json:"rows"`
	}
	if err := job.Decode(&in); err != nil {
		return nil, emit.Unrecoverable(err)
	}
	if err := job.Log(ctx, emit.LevelInfo, "import started"); err != nil {
		return nil, err
	}
	for done := 0; done < in.Rows; done += 500 {
		// ...import the next 500 rows...
		if err := job.Progress(ctx, float64(done)*100/float64(in.Rows), fmt.Sprintf("%d rows", done)); err != nil {
			return nil, err
		}
	}
	if err := job.Event(ctx, map[string]string{"stage": "imported"}); err != nil {
		return nil, err
	}
	return map[string]int{"imported": in.Rows}, nil
})

curl -s "$EMIT_URL/v1/spaces/contacts/jobs:report" \
  -H "Authorization: Bearer $EMIT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "id": "01M3QSMMXQ00ZSGYA6VWM3NGQA",
    "attempt": 1,
    "touch": {},
    "progress": {"percent": 40, "message": "2000 rows"},
    "lines": [{"level": "INFO", "message": "import started"}],
    "event": {"stage": "importing"}
  }'

Read them back with emit.jobs.get(id) and emit.jobs.lines(id), or follow the job as it changes with emit.jobs.follow(id). A browser can follow a job too, with no API token: see Show progress in a browser.

Cancelling

Cancelling a job that hasn't finished makes it cancelled at once. A worker that's running it finds out when it next renews the lease, and the SDKs then tell your handler to stop, as they do for a lost lease. Anything it reports afterwards is ignored. Cancelling needs a token with the admin scope.

TypeScriptGocurl

const [outcome] = await emit.jobs.cancel("01M3QSMMXQ00ZSGYA6VWM3NGQA");
// { status: "cancelled" }, or { rejection: { reason: "already_finished" } }

outcomes, err := c.Jobs.Cancel(ctx, "01M3QSMMXQ00ZSGYA6VWM3NGQA")

curl -s "$EMIT_URL/v1/spaces/billing/jobs:cancel" \
  -H "Authorization: Bearer $EMIT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '["01M3QSMMXQ00ZSGYA6VWM3NGQA"]'

Callbacks

A callback is a webhook emit sends when a job finishes, however it finishes. Give the job a URL, and optionally headers to send with it:

TypeScriptGocurl

await emit.jobs.create("render-report", { reportId: "rep_881" }, {
  callback: {
    url: "https://api.example.com/webhooks/emit",
    headers: { "x-tenant": "acme" },
  },
});

_, err := c.Jobs.Create(ctx, emit.NewJob{
	Name:     "render-report",
	Payload:  map[string]string{"reportId": "rep_881"},
	Callback: &emit.Callback{URL: "https://api.example.com/webhooks/emit", Headers: map[string]string{"x-tenant": "acme"}},
})

curl -s "$EMIT_URL/v1/spaces/reports/jobs" \
  -H "Authorization: Bearer $EMIT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "render-report",
    "payload": {"reportId": "rep_881"},
    "callback": {"url": "https://api.example.com/webhooks/emit", "headers": {"x-tenant": "acme"}}
  }'

When the job ends, emit POSTs a JSON body to the URL:

JSON

{
  "jobId": "01M3QSMMXQ00ZSGYA6VWM3NGQA",
  "space": "reports",
  "name": "render-report",
  "status": "completed",
  "attempt": 1,
  "timestamp": 1790726272524,
  "result": { "url": "https://files.example.com/rep_881.pdf" }
}

status is completed, failed or cancelled, and result is the job's result, or its last error. Answer with any 2xx status. Anything else, or no answer, is tried again: after 5 seconds, then doubling up to an hour between tries, for about a day.

The URL must be http or https and reachable from the internet. Header values can carry credentials, so emit never shows them again: reading the job back gives the header names with empty values.

Checking the signature

Every callback carries an Emit-Signature header, signed with the space's webhook secret. A space you set up in the dashboard gets a secret, shown once. A space created by its first job has none until you set one: type a new secret in the space's settings, or send webhookSecret through the API. Check the signature before you trust the body, against the body exactly as it arrived:

TypeScriptGoShell

import { verifySignature } from "@emit.run/sdk";

const webhookSecret = process.env.EMIT_WEBHOOK_SECRET!;

export async function handleEmitCallback(request: Request): Promise<Response> {
  const body = await request.text(); // the raw body: don't parse it first
  if (!(await verifySignature(webhookSecret, request.headers.get("emit-signature"), body))) {
    return new Response("invalid signature", { status: 401 });
  }
  const callback = JSON.parse(body) as { jobId: string; status: string; result: unknown };
  console.log(`job ${callback.jobId} ${callback.status}`);
  return new Response(null, { status: 204 });
}

webhookSecret := os.Getenv("EMIT_WEBHOOK_SECRET")

http.HandleFunc("POST /webhooks/emit", func(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(r.Body) // the raw body: don't decode it first
	if err != nil {
		http.Error(w, "unreadable body", http.StatusBadRequest)
		return
	}
	if !emit.VerifySignature(webhookSecret, r.Header.Get("Emit-Signature"), string(body), 5*time.Minute) {
		http.Error(w, "invalid signature", http.StatusUnauthorized)
		return
	}
	var callback struct {
		JobID  string `json:"jobId"`
		Status string `json:"status"`
	}
	if err := json.Unmarshal(body, &callback); err != nil {
		http.Error(w, "not JSON", http.StatusBadRequest)
		return
	}
	log.Printf("job %s %s", callback.JobID, callback.Status)
	w.WriteHeader(http.StatusNoContent)
})

# The header is t=<Unix ms>,v1=<hex HMAC-SHA256 of "<t>.<raw body>", keyed with the webhook secret>.
# Given the header's t, and the body saved to body.json:
printf '%s.%s' "$t" "$(cat body.json)" | openssl dgst -sha256 -hmac "$EMIT_WEBHOOK_SECRET"
# Compare the hex digest with the header's v1, and reject a t more than a few minutes old.

verifySignature and VerifySignature refuse a signature more than five minutes old, which stops an old request from being replayed. The TypeScript function takes a different tolerance, in milliseconds, as its fourth argument.

When emit is busy

A space or a plan can cap the jobs in flight: jobs created but not yet finished. At the cap, emit holds new jobs back until workers bring the count down: it answers HTTP 429 with a Retry-After header, and the body's retryAfterMs says how long to wait. Nothing from that call is kept, so the same call is sent again after the wait. (A single batch larger than the cap is refused.)

The SDKs do this for you. A create tries again as emit says, for up to a minute, and only then returns the 429 as an error. Set how long it waits with roomWaitMs in TypeScript or RoomWait in Go. Without an SDK, send the same request again after the wait. With a key on each job, a repeat within the key's window won't make a job twice.

TypeScriptGocurl

const emit = new Emit({
  url: process.env.EMIT_URL!,
  token: process.env.EMIT_TOKEN!,
  space: "imports",
  roomWaitMs: 5 * 60_000, // wait up to five minutes for room
});

c := emit.New(emit.Options{
	URL:      os.Getenv("EMIT_URL"),
	Token:    os.Getenv("EMIT_TOKEN"),
	Space:    "imports",
	RoomWait: 5 * time.Minute, // wait up to five minutes for room
})

curl -s -i "$EMIT_URL/v1/spaces/imports/jobs" \
  -H "Authorization: Bearer $EMIT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name": "import-row", "payload": {"importId": "imp_7", "row": 18}, "key": "imp_7-row-18"}'
# HTTP/1.1 429 Too Many Requests
# retry-after: 1
# {"code":"ResourceExhausted","message":"...","retryAfterMs":1000}

Claims and reports aren't refused this way: they're what brings the count down.