Workers

Claim a task

POST /<queue>/claim with an optional worker id and a criteria object. The server hands you the oldest pending item that matches (exclusively, leased to you for the queue's leaseSeconds), or tells you the queue is empty.

curl -fsS https://task.example.com/resize/claim \
  -H 'authorization: Bearer USER_TOKEN' \
  -H 'content-type: application/json' \
  -d '{"worker":"box-1","criteria":{"format":"jpg","width":{"lte":2000}}}'

{
  "ok": true,
  "id": "9f2c1a0b7d4e6f83",
  "item": { "url": "https://ex.com/a.png", "format": "jpg", "width": 800 },
  "attempts": 1,
  "lease": { "worker": "box-1", "until": 1751500072000 }
}

When nothing matches you get { "ok": true, "empty": true }. Back off briefly and claim again. The worker id is a free-form label; it is recorded on the lease and is what you use to tell your workers apart.

Claiming by capability

The criteria object is how a worker pulls only the tasks it can process. It maps an item field to a matcher: a bare value (equality) or one operator object. An item is claimed only if every criterion matches; a missing field or an unknown operator fails closed, so you never get an item you didn't ask for.

{ "format": "png", "width": { "lte": 2000 }, "kind": { "in": ["thumb", "preview"] } }

Omit criteria (or pass {}) to take any pending item. This is how a fleet of unlike workers shares one queue: a GPU box claims { "format": "webp" }, a small box claims { "width": { "lte": 512 } }, and each drains only what it is suited to.

Report progress

POST /<queue>/items/<id>/progress with a record matching the queue's progress schema. Only the last progress value is kept, and it shows up in the producer's status and monitor feed. Progress on an already-terminal item is ignored.

curl -fsS https://task.example.com/resize/items/9f2c1a0b7d4e6f83/progress \
  -H 'authorization: Bearer USER_TOKEN' \
  -H 'content-type: application/json' \
  -d '{"percent":40}'
# -> {"ok":true,"accepted":true,"status":"processing"}

Post a result

POST /<queue>/items/<id>/result with a record matching the result schema. It finishes the task. The response's accepted tells you whether your result was the one that counted:

curl -fsS https://task.example.com/resize/items/9f2c1a0b7d4e6f83/result \
  -H 'authorization: Bearer USER_TOKEN' \
  -H 'content-type: application/json' \
  -d '{"output":"aGVsbG8="}'
# -> {"ok":true,"accepted":true,"status":"finished"}

A result is accepted whenever the item is not already terminal: even if your lease lapsed and the item was re-queued, your late result is still used, as long as no other worker finished it and the queue didn't time it out first. If someone (or the timeout) beat you, you get { "ok": true, "accepted": false, "status": "finished" } (or "failed") and your result is discarded. See the lifecycle for why this is safe.

Reporting task failures in-band

A task that can't be done (a bad URL, an unsupported format) is not a queue-level failure; it is a normal result that happens to describe an error. Put an error field in the result schema and finish the item with it:

curl -fsS https://task.example.com/resize/items/$ID/result \
  -H 'authorization: Bearer USER_TOKEN' -H 'content-type: application/json' \
  -d '{"error":"source image was not a PNG"}'

The item finishes normally (status: "finished") carrying your error in its result. That is distinct from the terminal "timeout" the server itself sets on the item's top-level error when a queue timeout fires: one is your application's verdict, the other is the queue giving up.

A worker loop over HTTP

const Q = "https://task.example.com/resize";
const H = { "authorization": "Bearer USER_TOKEN", "content-type": "application/json" };

async function post(path, body) {
  const res = await fetch(Q + path, { method: "POST", headers: H, body: JSON.stringify(body) });
  return res.json();
}

while (true) {
  const claim = await post("/claim", { worker: "box-1", criteria: { format: "jpg" } });
  if (claim.empty) { await new Promise(r => setTimeout(r, 2000)); continue; }

  try {
    await post(`/items/${claim.id}/progress`, { percent: 0 });
    const output = await doWork(claim.item);          // your work here
    await post(`/items/${claim.id}/result`, { output });
  } catch (err) {
    // application failure: report it in-band and move on
    await post(`/items/${claim.id}/result`, { error: String(err) });
  }
}

Run several of these in parallel (separate processes or boxes, each claiming for itself) to work the queue concurrently. There is no batch claim; parallelism comes from many single-item claims.

Working over a socket

The GET /<queue>/work WebSocket does the same three operations over one connection, and adds a safety net: if the socket drops, every item still leased to it is re-queued immediately, without waiting for the lease to lapse. Send claim / progress / result messages and read the acks.

const ws = new WebSocket("wss://task.example.com/resize/work?token=USER_TOKEN");

ws.onopen = () => ws.send(JSON.stringify({ claim: { worker: "box-1", criteria: { format: "jpg" } } }));

ws.onmessage = async (ev) => {
  const m = JSON.parse(ev.data);
  if (m.type === "empty") {
    setTimeout(() => ws.send(JSON.stringify({ claim: { worker: "box-1" } })), 2000);
  } else if (m.type === "item") {
    const output = await doWork(m.item);
    ws.send(JSON.stringify({ result: { id: m.id, value: { output } } }));
  } else if (m.type === "result-ack") {
    // m.accepted tells you if your result counted; claim the next one
    ws.send(JSON.stringify({ claim: { worker: "box-1", criteria: { format: "jpg" } } }));
  }
};