Go API

The Go SDK (github.com/synaptiai/cronoq/sdks/go, package cronoq) is a client for a cronoq server: it submits, inspects, waits for and watches tasks, and runs workers. It needs Go 1.26 or newer.

go get github.com/synaptiai/cronoq/sdks/go

The generated gRPC stubs are in the cronoqpb package, for calls the SDK does not wrap (such as AddRecurring or ListWorkers); Client.Conn() gives the connection to use with them.

Submitting and waiting

c, err := cronoq.Dial("http://127.0.0.1:7070", cronoq.Options{Token: os.Getenv("CRONOQ_TOKEN")})
if err != nil {
	return err
}
defer c.Close()

sub, err := c.Submit(ctx, cronoq.Task{Payload: []byte(`{"to":"ada"}`), Key: "welcome-ada", Priority: cronoq.High})
res, err := c.Wait(ctx, cronoq.ID(sub.ID), 30*time.Second)
if err == nil && res.OK() {
	fmt.Println(string(res.Output))
}

Dial(target, Options)

target is http://host:port, https://host:port or host:port (plain). The connection is opened on the first call.

Client

Ref names a task: cronoq.ID(id) or cronoq.Key(key).

TaskPage.NextAfter is 0 on the last page.

Task fields: Payload, ContentType, Kind, Pool, Key, Tenant, Priority (cronoq.Critical, High, Normal, Low), Timeout (per attempt), Deadline (relative), StartIn, After ([]Dep{{Ref: cronoq.Key("extract"), Cond: cronoq.OnSuccess}}), Resources, Tags, Labels (worker labels the task requires) and MaxAttempts (replaces the pool's retry policy with this many attempts and the server's default backoff).

TaskInfo, Result and Event carry a Status (cronoq.Ready, Running, Succeeded, ...; Status.Done() is true for terminal ones). TaskPage.UnavailableShards names shards that could not be reached (Sharding).

Errors

Failed calls return a *cronoq.Error with the gRPC Code and Message (Errors):

var e *cronoq.Error
if errors.As(err, &e) && e.Timeout() { /* the task is still running */ }

Retryable() is true for UNAVAILABLE (not the leader, shutting down) and RESOURCE_EXHAUSTED (queue full); NotFound() for unknown tasks, pools and dependencies.

Workers

err := cronoq.RunWorker(ctx, cronoq.WorkerOptions{
	Target: "http://127.0.0.1:7070",
	Pool:   "default",
	Slots:  8,
	Token:  os.Getenv("CRONOQ_WORKER_TOKEN"),
}, func(ctx context.Context, l *cronoq.Lease) ([]byte, error) {
	return process(ctx, l.Payload)
})

RunWorker registers for the pool and runs the handler for each lease, at most Slots at a time (or fewer if the server grants fewer). It heartbeats every running lease, and reconnects with backoff when the connection drops (right away when the server announces a leadership change). When the pool belongs to another shard, it connects to that shard's endpoint; see below. When ctx ends, it stops taking leases and lets running handlers finish for up to DrainTimeout, then cancels the rest and returns nil. A cancelled handler that returns within 5 more seconds still has its outcome reported; after that its outcome is dropped (the server retries the task). It returns an error only for failures that retrying cannot fix, such as a rejected token or an unknown pool.

Handlers

The handler gets a *Lease: ID, TaskID, Attempt, Payload, ContentType, Kind, Key, Pool, Priority, Tags, Timeout and Deadline. Its context ends when the server aborts the attempt (timeout, cancellation, lease expiry), when the connection to the server ends, or when the drain timeout passes. After an abort the outcome is not reported; after the drain timeout it is reported if the handler returns within 5 seconds. It returns the output, or an error:

(Lease.TaskID, Lease.Attempt) is a fencing token: attempt numbers never repeat for a task, also across failovers. Record the pair with the handler's side effect and let only the newest attempt write (Exactly-once side effects).

Shards

A worker that connects to another shard's server is refused with the endpoint of its pool's shard, and RunWorker connects there (up to cronoq.MaxRedirects times in a row). It never follows a redirect from https:// to http://, sends its token over plain http:// only to loopback unless InsecureRedirectHTTP is set, and refuses endpoints with userinfo or a path. When that shard cannot be reached, it asks Target again. See Sharding.