This release adds a rules-based Pro Pruner, chunk workers for processing jobs in groups, and an Erlang notifier for direct pub/sub without Postgres, alongside fixes for global limits and acks across nodes. See the v0.7 Upgrade Guide for complete upgrade steps and migration caveats.
Pro Pruner
The Pro Pruner replaces the standard pruner with rules that retain specific queues, workers, or
states differently, by age or by length. Each job is pruned by the first rule that matches it, and
an implicit "default" rule catches everything else, keeping finished jobs for one day:
[pruner]
schedule = "*/5 * * * *"
[[pruner.rules]]
name = "events"
queue = "events"
max_len = 10_000
[[pruner.rules]]
name = "discarded"
state = "discarded"
max_age = "7 days"
archive = true
Rules run on a cron schedule, set their own limit and timeout, and can archive jobs to
oban_jobs_archive rather than deleting them. Jobs in an active workflow are preserved by
default, so pruning never breaks a running workflow.
Rules are stored in the database and can be inspected and changed at runtime through
oban_pro.pruner, without a restart:
from oban_pro import pruner
await pruner.insert(name="media", queue="media", max_age="1 week")
await pruner.update("media", paused=True)
The tables match Oban Pro for Elixir, so rules can be managed from Oban Web and shared with Elixir nodes.
Chunks
Chunk workers process jobs together in groups, running when a chunk reaches its size or after a
wait time, e.g. 100 jobs or every second. Declare one with the chunk option and define
process_chunk, which receives the list of jobs:
from oban import worker
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)
Every job in a chunk completes unless process_chunk says otherwise. Return chunk.Cancel,
chunk.Error, or chunk.Snooze to apply an outcome to particular jobs, or raise to fail the
whole chunk. Chunks are grouped by queue and worker by default, or by args and meta with the by
option, using the same format as queue partitioning.
Erlang Notifier
The Erlang notifier replaces Postgres LISTEN/NOTIFY with direct pub/sub between processes.
It avoids the 8000 byte payload limit, works through transaction poolers, and keeps notification
traffic off the database:
[notifier]
type = "erlang"
It runs entirely in Python, with no Erlang or Elixir required. Python processes discover each
other through queue producers, and because it speaks the Erlang distribution protocol, it also
joins clusters with Elixir nodes running Oban.Notifiers.PG.