# 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:

```python
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:

```python
@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:

```python
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](smart_concurrency.md):

```python
# 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:

```python
@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:

```python
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:

```python
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()`:

```python
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:

```python
@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:

```toml
[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:

```toml
[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.