Chunks

Chunks#

Chunk workers process jobs together in groups, based on a size or a wait time, e.g. when 1,000 jobs are available or after 10 minutes have elapsed. Rather than handling one job at a time, a chunk worker receives a list of jobs and processes them in a single pass.

Multiple chunks can run in parallel within a single queue, and each chunk may contain many thousands of jobs. Aside from improved throughput for a single queue, chunks are ideal as the initial stage of data-ingestion and data-processing pipelines, or for batching calls to an external service such as ML inference.

Usage#

Declare a chunk worker with the chunk option, and define process_chunk rather than process. Here’s a worker that sends SMS messages in chunks rather than individually:

from oban import worker


@worker(queue="messages", chunk={"size": 100, "wait": 1})
class SendMessages:
    async def process_chunk(self, jobs):
        await messaging.send_batch([job.args for job in jobs])

The worker runs a chunk with up to 100 jobs, or whatever is available after one second, whichever comes first. Both options have defaults, a size of 100 and a wait of one second, so chunk=True is enough to get started.

Chunking requires a @worker class. Functions decorated with @job receive a single job’s arguments, so they can’t be chunked.

To process larger chunks less frequently, increase both values:

@worker(queue="messages", chunk={"size": 500, "wait": 5})

Now chunks run with up to 500 jobs or every five seconds.

Chunk jobs are enqueued like any other job, one at a time or in bulk:

await oban.enqueue_many([SendMessages.new({"to": to}) for to in recipients])

Chunk Partitioning#

By default, chunks are grouped by queue and worker, so each chunk contains jobs from the same worker, regardless of their args or meta. Use the by option to group chunks by specific fields instead. It takes the same form as queue partitioning:

# Chunk by worker, the default
chunk = {"by": "worker"}

# Chunk by an args key, without considering the worker
chunk = {"by": {"args": "account_id"}}

# Chunk by multiple args keys
chunk = {"by": {"args": ["account_id", "cohort"]}}

# Chunk by worker and an args key
chunk = {"by": {"worker": True, "args": "account_id"}}

# Chunk by a meta key
chunk = {"by": {"meta": "source_id"}}

Partitioning by args or meta without worker may produce chunks containing jobs from different workers. Regardless of the partitioning, a chunk always consists of jobs from the same queue.

Here’s an example that partitions by author_id to analyze recent messages per author:

@worker(chunk={"by": {"worker": True, "args": "author_id"}, "size": 50, "wait": 30})
class AnalyzeSentiment:
    async def process_chunk(self, jobs):
        author_id = jobs[0].args["author_id"]
        messages = await Message.fetch([job.args["message_id"] for job in jobs])
        sentiment = await gauge_sentiment(messages)

        await Author.record_sentiment(author_id, sentiment)

Results and Error Handling#

When process_chunk returns without raising, every job in the chunk is completed. To apply a different outcome to particular jobs, return a result from oban_pro.chunk with the jobs it applies to:

  • chunk.Cancel(reason, jobs) — cancel the listed jobs

  • chunk.Error(reason, jobs) — fail the listed jobs, so they’re retried, or discarded on their final attempt

  • chunk.Snooze(seconds, jobs) — reschedule the listed jobs to run again after a delay

Any jobs that aren’t listed are completed. For example, to cancel messages with a bad token and complete the rest:

from oban_pro import chunk


@worker(queue="messages", chunk={"size": 100, "wait": 1})
class SendMessages:
    async def process_chunk(self, jobs):
        responses = await messaging.send_batch([job.args for job in jobs])
        bad_token = [job for job, resp in zip(jobs, responses) if resp.bad_token]

        return chunk.Cancel("bad token", bad_token)

Return a list to apply several outcomes at once:

return [
    chunk.Cancel("bad token", bad_token),
    chunk.Snooze(60, throttled),
]

A job may only appear in one result. Listing a job twice is treated as an error for the whole chunk.

To apply an outcome to every job in the chunk, return a standard Cancel or Snooze, or raise an exception. A raised exception, including a worker timeout, fails every job in the chunk and each is retried according to its own attempts and the worker’s backoff.

Because process_chunk is a regular method, you can test it directly with jobs built from new():

async def test_cancelling_bad_tokens():
    jobs = [SendMessages.new({"to": to}) for to in recipients]
    result = await SendMessages().process_chunk(jobs)

    assert result.jobs == [jobs[1]]

Chunk Organization#

A chunk is run by a leader, which is the first job in the chunk to start executing (unrelated to cluster leadership). When the leader starts, it checks whether a full chunk is available. If not, it waits up to wait seconds for more jobs to arrive, then runs with as many jobs as it can find.

While waiting, the leader checks for a full chunk periodically, at most once a second and more often for short waits. The worker’s timeout, if any, applies to the entire chunk, including the time spent waiting.

Running Chunks Immediately#

A chunk won’t run until a full chunk is available or the wait has elapsed. When there are long gaps between incoming jobs, you may want a chunk to run as soon as the leader starts instead. That’s what the leading option is for. For example, to chunk all jobs that arrive within a five-minute window, but run immediately when more than five minutes have passed since the last chunk finished:

@worker(chunk={"size": 100, "wait": 300, "leading": True})

A leading chunk only runs immediately after a previous chunk has finished. The first chunk waits as usual.

Optimizing Chunks#

Queues with high concurrency and low throughput may start multiple chunk leaders simultaneously, rather than bundling jobs into a single chunk. The solution is to reduce the queue’s global concurrency so that new leaders don’t start while an existing chunk can grow:

[queues.chunked]
global_limit = 1

A low global limit reduces overall throughput. Partitioning the queue the same way as the chunk preserves throughput by allowing one concurrent leader per chunk. For example, when chunking by user_id in args, partition the queue with the same option:

[queues.chunked]
limit = 20
global_limit = 1
partition = { args = "user_id" }

Implementation Notes#

  • Only the leader counts toward a queue’s concurrency, global, and rate limits. The rest of the chunk runs within the leader.

  • Cancelling the leader cancels the whole chunk, provided process_chunk checks cancelled() on any of its jobs and returns Cancel. Cancelling another job while its chunk is running has no effect.

  • If a node shuts down while a chunk is running, the chunk’s jobs are retried, either after being failed or after being rescued like any other orphaned job.