TypeScript API

The TypeScript SDK (npm package @synaptiai/cronoq, in sdks/typescript) is a client for a cronoq server, built on Connect-ES. It runs in two places:

The package is published to GitHub Packages, whose npm registry needs a GitHub token to install any package, public ones included. Create a personal access token (classic) with the read:packages scope, put it in GITHUB_TOKEN, and point the @synaptiai scope at the registry:

echo '@synaptiai:registry=https://npm.pkg.github.com' >> .npmrc
echo '//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}' >> .npmrc
npm install @synaptiai/cronoq

Each GitHub release also carries the package tarball, which installs without a token:

npm install https://github.com/synaptiai/cronoq/releases/download/v0.2.1/synaptiai-cronoq-0.2.1.tgz

Submitting and waiting

import { connect, text } from "@synaptiai/cronoq/node";

const c = connect("http://127.0.0.1:7070", { token: process.env.CRONOQ_TOKEN });
const sub = await c.submit({ payload: { to: "ada" }, key: "welcome-ada", priority: "high" });
const res = await c.wait(sub.id, 30);
if (res.status === "succeeded") console.log(text(res.output));

Task ids are bigints: they use all 64 bits, which a number cannot hold. A task reference is an id (bigint) or a key (string). Times are in seconds.

connect(endpoint, options) (Node)

endpoint is http://host:port or https://host:port. The connection opens on the first call.

connectWeb(baseUrl, { token }) (browser)

baseUrl is the server's gRPC listener, for example https://cronoq.example.com:7070. The server must list the page's origin in [server.grpc_web] (gRPC-Web). A token in a page is visible to whoever uses the page: give browsers their own token, limited to the pools and role they need.

Client

Every method except watch also takes { signal, timeout } call options; watch takes only { signal }. client.tasks and client.admin are the generated Connect clients, for calls the class does not wrap (such as addRecurring or listWorkers); pb exports the generated messages.

Task fields: payload, contentType, kind, pool, key, tenant, priority ("critical", "high", "normal", "low"), timeout (per attempt), deadline (relative), startIn, after (references, or { ref, cond } with cond "on_success", "on_complete" or "on_failure"), 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).

Payloads and outputs: a Uint8Array is sent as is, a string as UTF-8 text, null or undefined as empty, and anything else as JSON. text(bytes) and json(bytes) decode them.

TaskInfo, Result and Event carry a status ("ready", "running", "succeeded", ...; isDone(status) is true for terminal ones). TaskPage.unavailableShards names shards that could not be reached (Sharding).

Errors

Failed calls reject with a CronoqError: code is the gRPC status name in lower case ("not_found", "unavailable", "cancelled", ...), and metadata holds the response metadata (Errors). retryable is true for unavailable (not the leader, shutting down) and resource_exhausted (queue full); timeout for a wait that ended while the task was still running; notFound for unknown tasks, pools and dependencies. In browsers, only the metadata the server exposes to CORS requests is visible.

Workers (Node)

import { runWorker } from "@synaptiai/cronoq/node";

const stop = new AbortController();
process.on("SIGTERM", () => stop.abort());
await runWorker(
  { target: "http://127.0.0.1:7070", pool: "default", slots: 8, token: process.env.CRONOQ_WORKER_TOKEN, signal: stop.signal },
  async (lease) => handle(lease.json()),
);

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 signal aborts, it stops taking leases and lets running handlers finish for up to drainTimeout, then aborts the rest and resolves. An aborted handler that returns within 5 more seconds still has its outcome reported; after that its outcome is dropped (the server retries the task). It rejects 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, deadline, signal, and text() and json(). signal aborts 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 by the server the outcome is not reported; after the drain timeout it is reported if the handler returns within 5 seconds. The handler returns the output, or throws:

(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 MAX_REDIRECTS 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.