Jobs¶

A job is work for later: put durably, claimed by a worker, and retried if it fails. The queue lives in the layout's database, with no broker.

Putting a job¶

Declare jobs = Jobs() on the service. With tx=, the job commits with your writes:

from myxo.jobs import Jobs, job


class CountWords(msgspec.Struct, frozen=True):
    note_id: int


class Notes(Service):
    name = "notes"
    jobs = Jobs()

    async def add(self, note: NewNote) -> NoteRef:
        insert = NOTES.insert().values(text=note.text).returning(NOTES.c.id)
        async with self.engine.begin() as tx:
            note_id = (await tx.execute(insert)).scalar_one()
            await self.jobs.put(CountWords(note_id), tx=tx, key=f"count:{note_id}")
        return NoteRef(note_id)

Argument	What it does
`tx`	puts the job in the caller's transaction; without it, the job commits on its own
`key`	a put does nothing while a job with that key is in the queue, waiting, running or failed
`delay`	holds the job back for that many seconds

A done job leaves the queue with its key, so a later put of the same key runs again. So a key guards against a repeated put, while the job itself must still be harmless to repeat. Nothing records that a done job ran: to see it, log from the handler.

Running jobs¶

A worker is a method:

from sqlalchemy import update


@job(CountWords, workers=2)
async def count(self, request: CountWords) -> None:
    note = await self.get(NoteRef(request.note_id))
    query = (
        update(NOTES)
        .where(NOTES.c.id == note.id)
        .values(words=len(note.text.split()))
    )
    async with self.engine.begin() as tx:
        await tx.execute(query)

Argument	Default	What it sets
`workers`	`1`	workers in each process that runs the service
`attempts`	`10`	runs before the job is kept, marked failed
`lease`	`30.0`	seconds a claim lasts, renewed while the job runs
`backoff`	`1.0`	seconds before the second attempt, doubling to 5 minutes, or to `backoff` if longer

The worker runs wherever the layout places the service. To keep the queue's work out of the request, split the service, or put the method in a service that runs in a worker process.

The guarantees¶

A job runs at least once. A job whose worker died is claimed again once its claim ends. A job may therefore run twice, after a crash or when it outlives its claim, so it must be harmless to repeat.

A job that raises is retried after its backoff, up to attempts times, then kept, marked failed, with its error. Find failed jobs in the table:

select id, topic, attempts, error from myxo_jobs where failed;

To run a failed job again, clear its mark:

update myxo_jobs set failed = false, attempts = 0 where id = 42;

How fast¶

On PostgreSQL, a put waits for the drive. A claim and a delete do not, since a lost one only runs the job again. On a server whose drive flushes each commit, the queue holds about 1,000 jobs per second (E12).

SQLite has one writer at a time, and each commit waits for the drive. On the same server, the queue kept up with 25 jobs per second beside 10 of the application's writes per second, or with 10 jobs beside 50 writes, and at most 35 to 75 jobs per second alone (E14). Past that, jobs wait for seconds, and some puts and writes fail with OperationalError after SQLite's 5 s timeout: the caller sees the failure. No job that a put had accepted was lost. That rate is the signal to move to PostgreSQL.