Queues

A queue holds work until something takes it: an app hands jobs to a worker, one agent hands work to another, a burst of webhooks waits its turn instead of overwhelming what handles them. Messages are JSON and nothing is lost when a worker crashes: a message you take is leased to you, and if you don't acknowledge it before the lease runs out, it's delivered again.

That makes delivery at least once. A message can arrive twice (a worker finished but crashed before acknowledging it), so make the work safe to repeat: check a dict with put_if_absent, or use an id your database refuses twice.

Create a queue#

import noclick

jobs = noclick.Queue.from_name("invoices", create_if_missing=True)

from_name with create_if_missing=True creates the queue on first use and finds it after. A message is delivered up to max_attempts times (5 by default, 1 to 100) before it becomes a dead letter; pass max_attempts= to from_name or create to change it, and a value you pass to from_name is applied to a queue that already exists too. A name is 1 to 64 letters, digits and . _ -, unique for each user, and for your account; a queue created inside as_user belongs to that user. region= keeps its messages in a region, for good (your account's default, else us-east). Queues are kept in NoClick's own database in us-east for now, so that's the one region they take: another, or an account default region elsewhere, raises noclick.InvalidRequest naming it, and you pass region="us-east" to keep the queue there.

Put work on#

jobs = noclick.Queue.from_name("invoices", create_if_missing=True)
jobs.put({"invoice": "INV-1042", "customer": "acme"})
jobs.put({"check_payment": "INV-1042"}, delay_s=3600)
ids = jobs.put_many([{"invoice": f"INV-{n}"} for n in range(1043, 1053)])

A message is any JSON up to 256 KB. delay_s (up to 7 days) keeps it hidden until then, for a check later or a retry you schedule yourself. put returns the new message's id; put_many puts up to 100 in one call, all of them or none, and returns their ids in order.

Code that puts a message and then crashes may put it again when it restarts. A dedupe_key (up to 128 characters) makes that safe: for 24 hours, a second put with the same key adds nothing and returns the first message's id.

jobs = noclick.Queue.from_name("fulfilment", create_if_missing=True)
jobs.put({"order": "o_981"}, dedupe_key="order-o_981")

Take work off#

get takes the next message, waiting up to timeout_s seconds (at most 60) for one to arrive, and returns None if none does:

jobs = noclick.Queue.from_name("invoices")
msg = jobs.get(timeout_s=30, lease_s=120)
if msg is not None:
    print(msg.body["invoice"], msg.attempts)
    msg.ack()

The message is leased to you for lease_s seconds (60 by default, up to 12 hours), and nobody else gets it meanwhile. When the work is done, ack() deletes it for good. If it failed, nack(delay_s=..., error=...) hands it back for another try after that delay; error says why (up to 2 KB), and it's the last_error a dead letter keeps. If you do neither, the lease runs out and it's delivered again. Each delivery counts in attempts. An ack after the lease has run out raises noclick.Conflict, since the message may already be someone else's.

Choose a lease_s longer than the work takes. Messages come out roughly in the order they went in, but order isn't guaranteed.

When the failure isn't the message's, such as your account running out of money or a service every message needs being down, release(delay_s=...) hands it back without spending an attempt: attempts goes back to what it was, and last_error stays as it was. A backlog then waits out the problem instead of burning its tries and landing in the dead letters. A released message comes back until the cause is fixed, so stop taking more meanwhile:

recordings = noclick.Queue.from_name("recordings")
for msg in recordings.iterate(lease_s=600):
    try:
        words = noclick.AI.transcribe(msg.body["url"])
    except noclick.PaymentRequired:
        msg.release(delay_s=3600)
        break
    except Exception as exc:
        msg.nack(delay_s=60, error=str(exc))
    else:
        print(words.text)
        msg.ack()

Run a worker#

iterate takes messages as they arrive and waits while the queue is empty. It runs until you stop iterating:

jobs = noclick.Queue.from_name("invoices", create_if_missing=True)
for msg in jobs.iterate(lease_s=120):
    try:
        print("Sending", msg.body["invoice"])
    except Exception as exc:
        msg.nack(delay_s=30, error=str(exc))
    else:
        msg.ack()

In async code, iterate iterate.aio() and acknowledge with await msg.ack.aio(). Run as many workers as you like: each message goes to one of them at a time.

A batch job drains the queue and finishes: with stop_when_empty=True, iterate ends the first time nothing is waiting, instead of waiting for more. Messages leased to another worker, or delayed, don't keep it going:

jobs = noclick.Queue.from_name("invoices")
for msg in jobs.iterate(stop_when_empty=True):
    print("Sending", msg.body["invoice"])
    msg.ack()

Let an agent take the work#

Give an agent an OnQueue trigger and each message becomes a turn, with no worker to run:

invoices = noclick.Queue.from_name("invoices", create_if_missing=True)
support = noclick.Agent.from_name(
    "support-bot",
    create_if_missing=True,
    instructions="You chase unpaid invoices by email.",
    tools=["gmail.send_email", "stripe.list_invoices"],
    triggers=[noclick.OnQueue(invoices, message="Chase this unpaid invoice.")],
)

The agent reads message, then the message's body. A turn that finishes acknowledges its message; a turn that fails hands it back after a pause that grows with each attempt (30 seconds per attempt so far), and after max_attempts failures it's a dead letter, whose last_error is the turn's error. A turn that can't start because a budget or spend cap is used up is no failure: the trigger stops taking messages, which wait with their attempts unchanged, and shows status="error" saying why until there's money to spend again. concurrency= (1 by default, up to 50) lets that many turns run at once; messages a conversation= template keys to the same conversation still take their turns one at a time. With batch_size=10, one turn takes up to ten waiting messages at once, as a JSON list, for work that's cheaper in bulk. While the agent is paused, messages wait on the queue. The queue's used_by lists the agents that take its messages.

Look without taking#

jobs = noclick.Queue.from_name("invoices")
print(jobs.len(), "waiting")
for msg in jobs.peek(limit=5):
    print(msg.status, msg.attempts, msg.enqueued_at, msg.body)

len() counts the messages waiting, delayed ones included, read now. The queue's depth, in_flight and dead_letter_count are as of its last load. peek shows the oldest messages, waiting and in flight, without leasing them, so they can't be acknowledged. depth_history(hours=24) gives the depth every 15 minutes, which tells you whether your workers keep up.

Handle dead letters#

A message that fails max_attempts times moves to the dead letters, with why its last attempt failed:

jobs = noclick.Queue.from_name("invoices")
for letter in jobs.dead_letters():
    print(letter.body, letter.attempts, letter.last_error)
    if "unknown invoice" in (letter.last_error or ""):
        letter.discard()
    else:
        letter.retry()

retry() puts a dead letter back on the queue with its attempts reset, once you've fixed what made it fail; discard() deletes it. Dead letters are kept for 14 days.

Use it from an app#

An app reaches a queue through a binding: app.bind("JOBS", jobs) makes it available to the deployed code as JOBS, so the web app puts work on and a worker or an agent takes it off; the app's code can also find or make it by name with from_name. See Apps.

A preview's code gets an empty queue of its own the first time it uses one, so it never takes production's work, and production's consumers never see a preview's messages. See A preview's own data.

Clear, export and delete#

jobs = noclick.Queue.from_name("invoices")
jobs.clear()                    # waiting messages and dead letters gone; in-flight ones finish
jobs.delete()

delete removes the queue and its messages. A queue an agent's OnQueue trigger reads raises noclick.Conflict until you remove the trigger (a deleted agent's trigger goes with the queue).

jobs = noclick.Queue.from_name("invoices")
bundle = jobs.export().wait()   # JSON Lines: every message, dead letters too
print(bundle.url)

export is an operation that gives an export bundle: queues/<name>.jsonl, every message as the queue stood at one moment, oldest first, one per line with its status (waiting, delayed, in_flight or dead), attempts, last_error and body. Exporting takes nothing off the queue.

Limits and billing#

Queues cost nothing within these limits. Metered pricing, per thousand messages put, will come with larger limits. The turns an OnQueue trigger starts are billed like any other turn, and count toward the agent's budget and its user's cap.

Errors#

Every method is in the Queue reference: put, get, iterate, QueueMessage.ack, dead_letters, and the trigger OnQueue.