SDKs and CLI

Go SDK

Create jobs, run workers, define typed work and write workflows from Go.

The Go SDK uses only the standard library and needs Go 1.23 or later.

Shell

go get github.com/emit-run/sdk-go

This page covers version 0.2.0.

The client

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

c := emit.New(emit.Options{
	URL:   os.Getenv("EMIT_URL"),   // your cluster's endpoint, from its Overview page
	Token: os.Getenv("EMIT_TOKEN"), // an API token, from its Tokens tab
	Space: "orders",                // the space this client works in
})

billing := c.Space("billing") // the same client, another space

c.Jobs works with jobs in the client's space, and c.Admin manages spaces and tokens. Every call takes a context.Context first.

Jobs

created, err := c.Jobs.Create(ctx, emit.NewJob{
	Name:        "resize-image",
	Payload:     Resize{ImageID: "img_881", Width: 800},
	Key:         "resize-img_881-800",
	MaxAttempts: 5,
})

many, err := c.Jobs.CreateMany(ctx, []emit.NewJob{
	{Name: "resize-image", Payload: Resize{ImageID: "img_881", Width: 1600}},
	{Name: "resize-image", Payload: Resize{ImageID: "img_882", Width: 800}},
})

job, err := c.Jobs.Get(ctx, created.ID)            // Payload and Result are json.RawMessage
page, err := c.Jobs.List(ctx, emit.ListJobs{Name: "resize-image", Status: emit.Failed, Limit: 50})
next, err := c.Jobs.List(ctx, emit.ListJobs{Page: page.NextPage})
lines, err := c.Jobs.Lines(ctx, created.ID, emit.ReadLines{MinLevel: emit.LevelWarn})
outcomes, err := c.Jobs.Cancel(ctx, created.ID)    // needs the admin scope

NewJob has the fields in Creating a job: Key, DelayMs, RunAtMs, MaxAttempts, TimeoutMs, BackoffMs, Backoff and Callback. created.Outcome is "new", or "duplicate" when the key already made a job.

Waiting for a result

var out struct {
	URL string `json:"url"`
}
err := c.Jobs.Result(ctx, created.ID, &out)
var failed *emit.JobFailed
if errors.As(err, &failed) {
	log.Printf("job %s %s: %s", failed.ID, failed.Status, failed.Result)
}

Result follows the job until it ends and decodes its result into out. If the job failed or was cancelled, the error is a *emit.JobFailed.

Following a job

for event, err := range c.Jobs.Follow(ctx, created.ID, emit.FollowLines()) {
	if err != nil {
		return err
	}
	if event.Progress != nil {
		log.Printf("%.0f%% %s", event.Progress.Percent, event.Progress.Message)
	}
	if event.Finished != nil {
		log.Printf("finished: %s", event.Finished.Status)
	}
}

The first event is Now, the job as it is. Then one arrives per change (Claimed, Progress, Event, Lines, Retrying, Released), and Finished last. c.Jobs.FollowToken(ctx, id, time.Hour) makes a token that lets a browser follow the job.

Workers

worker := c.Worker(ctx, emit.WorkerOptions{Names: []string{"resize-image"}, Concurrency: 8}, func(ctx context.Context, job *emit.HeldJob) (any, error) {
	var in Resize
	if err := job.Decode(&in); err != nil {
		return nil, emit.Unrecoverable(err)
	}
	if err := job.Progress(ctx, 10, "downloading"); err != nil {
		return nil, err
	}
	// ...resize, passing ctx to anything that takes one...
	return map[string]string{"url": "https://images.example.com/" + in.ImageID + ".webp"}, nil
})

The handler returns	The job
`result, nil`	completes, with `result` as its result
`nil, err`	fails this attempt, and is retried while attempts are left
`nil, emit.Unrecoverable(err)`	fails for good
`nil, emit.RetryAfter(err, 30*time.Second)`	fails this attempt, and is retried after that wait (at most a day)
`nil, emit.Release`	goes back to the queue at once; the attempt isn't counted

A handler that panics fails its attempt, with the panic as the error, and the worker carries on.

job has ID, Name, Payload, Attempt, Decode(&v), and Progress, Log, Event and Note. The handler's ctx is cancelled when the worker loses the job, or is stopping: stop work, since nothing more from this attempt is accepted. A handler that returns ctx.Err() while the worker is stopping gives its job back.

A log line, a note or a progress message is at most 4,032 bytes, and an event 4 KiB as JSON: past that the call returns an error and sends nothing. An error message is cut to 4 KiB. A result is at most 4 MiB as JSON; a larger one fails the attempt.

Options

Names[]string

The job names to claim, at most 32. Leave it empty to claim any job in the space.

Concurrencyintdefault: 1

How many jobs the worker holds at once.

LeaseMsint64

The lease to ask for. By default, each job's own timeout. The worker renews it every half-lease.

WaitMsint64default: 30000

How long one claim waits for work before asking again.

Retryemit.RetryOptionsdefault: {Attempts: 3, BackoffMs: 250}

How a report that fails on the way, from a network error, a 5xx or a 429, is retried before the handler hears of it. The wait doubles each time. The last report of a job is retried until its lease ends.

OnErrorfunc(error)default: log.Printf

Told of a last report that never landed.

worker.Stop() stops claiming, cancels the handlers' ctx, waits for them, and returns what ended the loop, if anything did. <-worker.Done() is closed when the loop ends by itself, for example on a revoked token, and worker.Err() then says why.

Defined work

emit.Define names a kind of work with Go types for its input and output. Every process that defines it and calls c.Serve runs it, and whoever calls Run gets the result, wherever it ran.

resize := emit.Define(c, "resize-image", func(ctx context.Context, in Resize, job *emit.HeldJob) (Sized, error) {
	if err := job.Progress(ctx, 50, ""); err != nil {
		return Sized{}, err
	}
	return Sized{URL: "https://images.example.com/" + in.ImageID + ".webp"}, nil
})

worker := c.Serve(ctx, emit.WorkerOptions{Concurrency: 8}) // define everything first

out, err := resize.Run(ctx, Resize{ImageID: "img_881", Width: 800})                            // queue it, wait for the result
id, err := resize.Enqueue(ctx, Resize{ImageID: "img_882", Width: 800}, emit.WithKey("img_882")) // queue it, return its id
here, err := resize.Here(ctx, Resize{ImageID: "img_883", Width: 800})                          // run it in this process

Here runs the job in this process under a lease: if the process dies, a worker that serves the name takes the next attempt. emit.Ref[In, Out](c, name) is work that another service runs: Run and Enqueue work the same, and Serve doesn't claim it.

Options are functions: emit.WithKey, emit.WithDelayMs, emit.WithRunAtMs, emit.WithMaxAttempts, emit.WithBackoffMs, emit.WithTimeoutMs, emit.WithCallback, emit.WithSingleton, and, for a workflow's steps, emit.WithLabel.

Workflows

emit.Workflow(c, name, fn) defines a workflow, and c.Serve runs it. See Workflows for the whole model. The function gets the workflow's ctx, the input and wf *emit.Run:

Call	What it does
`step.Run(ctx, in, opts...)`	runs a step and returns its result
`h := step.Start(ctx, in)`; `h.Get(ctx)`, `h.Cancel(ctx)`	starts a step and returns a handle
`step.Enqueue(ctx, in)`	starts a step that outlives the run
`emit.All(ctx, step, inputs)` / `emit.AllSettled(...)`	one step per input, all at once
`emit.Do(ctx, name, fn)`	runs `fn` once, here, and keeps what it returns
`wf.Sleep(ctx, d)`, `wf.SleepUntil(ctx, t)`	pauses the run
`v, ok, err := emit.Wait[T](ctx, name, timeout)`	waits for a signal; `ok` is false if the timeout passes
`emit.Race(ctx, emit.Case(...), emit.WaitCase(...), emit.SleepCase(...))`	whichever finishes first
`wf.Note(ctx, text)`	adds a line to the run's timeline
`wf.Now()`, `wf.Rand()`	the time and a random number, the same every time the function runs
`return Out{}, wf.ContinueAsNew(input)`	starts the run over with a fresh history

emit.Workflow returns a *emit.Flow[In, Out] with Run, Enqueue, Result, Signal, SignalOrStart and Cancel. Signal a run with emit.ByID(id), emit.ByKey(key) or emit.BySingleton(value).

Admin

A token with the admin scope manages spaces and tokens:

admin := emit.New(emit.Options{URL: os.Getenv("EMIT_URL"), Token: os.Getenv("EMIT_ADMIN_TOKEN")}).Admin

spaces, err := admin.Spaces.List(ctx)
counts, err := admin.Spaces.Counts(ctx, "orders")
space, err := admin.Spaces.Get(ctx, "orders")
space.MaxUnfinished = 100_000
_, err = admin.Spaces.Configure(ctx, space)

token, err := admin.Tokens.Create(ctx, emit.Token{Name: "api-server", Spaces: []string{"orders"}, Scopes: []emit.Scope{emit.ScopeCreate, emit.ScopeRead}})
log.Print(token.Secret) // shown once
err = admin.Tokens.Revoke(ctx, token.ID)

Configure replaces the space's settings whole, so read them first, as above. See Spaces.

Webhook signatures

emit.VerifySignature(secret, header, body, tolerance) checks the Emit-Signature header of a callback or a wake-up against its raw body, and returns false for a bad signature or one older than tolerance. See Checking the signature.

The firehose

The firehose streams every job change in your cluster. It has its own address, your cluster's endpoint followed by /firehose, and its own token: the firehose token from the cluster's Tokens tab.

hose := emit.NewFirehose(os.Getenv("EMIT_FIREHOSE_URL"), os.Getenv("EMIT_FIREHOSE_TOKEN"))
for batch, err := range hose.Watch(ctx, emit.Filter{Spaces: []string{"orders"}, To: emit.Failed}) {
	if err != nil {
		return err
	}
	for _, change := range batch.Changes {
		log.Printf("%s %s: %s -> %s", change.Name, change.ID, change.From, change.To)
	}
	// keep batch.Cursor, and pass it as Filter.After to resume
}

Errors

Every refusal from emit is an *emit.Error with its Code (such as NotFound, PermissionDenied or ResourceExhausted), the HTTP Status, and RetryAfterMs when emit asks you to wait. Permanent() is true when trying again unchanged won't help: any 4xx but 429. A job that ended without completing is a *emit.JobFailed.

The SDK doesn't retry a Create for you. If you retry one, give the job a Key. See When emit is busy.