Lesson 9 — background processing for slow work¶
What you will build¶
The same triage, run behind a job row. The lesson's provider is deliberately slow — four hundred milliseconds of pretending to think — because that is the whole reason this lesson exists: a model call can take seconds, and a support desk that holds a connection open for it is a support desk that falls over when a provider slows down.
The sequence in the transcript is the ask: 202 and a queued job come back before the model
answers, the first poll sees running, and the job ends as done with the ticket at
awaiting_approval — with nothing sent, because nothing in this path sends.
The runtime's contribution is create_task_in_context: the task carries the caller's context, so the
job's log lines and its correlation id belong to the request that started it. Everything else is
application code, including the part people forget — shutdown waits for work in flight.
What you need¶
Nothing new.
examples/support_desk/jobs.py
"""Background triage: the request returns a job, and the work outlives the request.
A model call can take seconds. Holding an HTTP connection open for it is how a support desk
becomes unavailable when a provider gets slow, so the route submits the work and answers `202`,
and a job row is what a client polls.
``create_task_in_context`` is the runtime's contribution: the task carries the caller's context,
so the job's log lines and its correlation id belong to the request that started it. The
application owns everything else — the row, the states, and the fact that shutdown waits for work
in flight rather than abandoning it half-written.
"""
from __future__ import annotations
import asyncio
from typing import Annotated
from sqlalchemy import insert, select, update
from support_desk.errors import TicketNotFoundError
from support_desk.security import STAFF, Principal, require_role
from support_desk.service import TicketService
from support_desk.tables import jobs
from support_desk.triage import Triage
from xyberos import Inject, create_task_in_context
from xyberos.database import DatabaseProvider
from xyberos.errors import XyberosError
QUEUED = "queued"
RUNNING = "running"
DONE = "done"
FAILED = "failed"
class TriageJobs:
"""Submits triage as a background task, and reports where each one got to."""
def __init__(
self,
database: Annotated[DatabaseProvider, Inject()],
tickets: Annotated[TicketService, Inject()],
triage: Annotated[Triage, Inject()],
) -> None:
self._database = database
self._tickets = tickets
self._triage = triage
self._tasks: set[asyncio.Task[None]] = set()
async def submit(self, principal: Principal, ticket_id: int) -> dict[str, object]:
"""Accept the work and return the job, before any model has been called.
The ticket is loaded first, so a ticket in another tenant is refused with the same 404 every
other route gives rather than becoming a failing background job.
"""
require_role(principal, *STAFF)
await self._tickets.get_ticket(principal, ticket_id)
async with self._database.transaction() as connection:
result = await connection.execute(
insert(jobs).values(
tenant=principal.tenant,
ticket_id=ticket_id,
submitted_by=principal.subject,
state=QUEUED,
)
)
inserted = result.inserted_primary_key
if inserted is None:
raise AssertionError("the insert reported no primary key")
job_id = int(inserted[0])
task = create_task_in_context(
self._run(job_id, principal, ticket_id), name=f"triage-{job_id}"
)
self._tasks.add(task)
task.add_done_callback(self._tasks.discard)
return {"job": job_id, "state": QUEUED, "ticket": ticket_id}
async def job(self, principal: Principal, job_id: int) -> dict[str, object]:
"""Return one job in the caller's tenant, or report it as not found."""
require_role(principal, *STAFF)
async with self._database.transaction() as connection:
row = (
(
await connection.execute(
select(jobs).where(jobs.c.id == job_id, jobs.c.tenant == principal.tenant)
)
)
.mappings()
.first()
)
if row is None:
raise TicketNotFoundError(job_id)
return dict(row)
async def wait(self) -> None:
"""Wait for the work in flight, which is what shutdown has to do to be honest."""
while self._tasks:
tasks = list(self._tasks)
self._tasks.clear()
await asyncio.gather(*tasks, return_exceptions=True)
async def aclose(self) -> None:
"""Close the component the way the runtime closes a provided resource."""
await self.wait()
async def _run(self, job_id: int, principal: Principal, ticket_id: int) -> None:
"""Run the workflow and record the outcome on the job row, whatever it was."""
await self._set(job_id, RUNNING, "")
try:
result = await self._triage.run(principal, ticket_id)
except XyberosError as error:
# A model failure is the job's failure, not the request's: the row names the error
# class, and the message stays on the server rather than travelling to a client.
await self._set(job_id, FAILED, type(error).__name__)
else:
await self._set(job_id, DONE, str(result["state"]))
async def _set(self, job_id: int, state: str, detail: str) -> None:
async with self._database.transaction() as connection:
await connection.execute(
update(jobs).where(jobs.c.id == job_id).values(state=state, detail=detail)
)
examples/learn_hybrid/hybrid09_background.py
"""Lesson 9 — background processing: the request returns a job, and the work outlives it.
Run it with:
python -m learn_hybrid.hybrid09_background
"""
from __future__ import annotations
import tempfile
import time
from pathlib import Path
from support_desk.app import create_app
from support_desk.jobs import DONE, FAILED
from support_desk.migrate import migrate
from support_desk.security import PrincipalVerifier
from xyberos.ai import FakeModelProvider, ModelReply
from xyberos.database import SqliteSettings
from xyberos.http import HttpAdapter
from xyberos.http.testing import create_test_client
SECRET = "a-secret-the-deployment-keeps"
VERIFIER = PrincipalVerifier(SECRET)
ADA = VERIFIER.headers("ada@example.com", "acme", ["requester"])
SAM = VERIFIER.headers("sam@desk", "acme", ["support_agent"])
LEE = VERIFIER.headers("lee@desk", "acme", ["support_lead"])
#: A slow provider on purpose: the reason this lesson exists is that a model call takes seconds.
SLOW = 0.4
#: How long a client waits for the job, and how often it asks. The budget is a deadline rather than
#: a number of attempts: the work takes real time, so counting polls counts the wrong thing.
POLL_BUDGET = 15.0
POLL_INTERVAL = 0.01
def provider() -> FakeModelProvider:
"""A classification that takes ``SLOW`` seconds, then an instant draft."""
return FakeModelProvider(
[
ModelReply.json_payload({"category": "refund", "confidence": "high"}, delay=SLOW),
ModelReply.ok("A refund takes three business days to appear."),
]
)
def main() -> None:
"""Submit triage, poll while the model is still thinking, and read where it finished."""
with tempfile.TemporaryDirectory() as directory:
settings = SqliteSettings(path=Path(directory) / "support.db")
migrate(settings)
app = create_app(settings=settings, secret=SECRET, provider=provider())
with create_test_client(HttpAdapter(app)) as client:
client.post("/knowledge", headers=LEE)
client.post(
"/tickets",
json={"subject": "Where is my refund?", "body": "Order 42 came back on Monday."},
headers=ADA,
)
submitted = client.post("/tickets/1/triage", headers=SAM)
accepted = submitted.json()
print("POST /tickets/1/triage ->", submitted.status_code, accepted)
job_id = accepted["job"]
seen = [str(accepted["state"])]
job = accepted
deadline = time.monotonic() + POLL_BUDGET
while time.monotonic() < deadline:
job = client.get(f"/jobs/{job_id}", headers=SAM).json()
if job["state"] not in seen:
seen.append(str(job["state"]))
if job["state"] in {DONE, FAILED}:
break
time.sleep(POLL_INTERVAL)
print("the job went:", " -> ".join(seen))
print("it finished as:", job["state"], "with", job["detail"])
print("the ticket is:", client.get("/tickets/1", headers=SAM).json()["state"])
print(
"the draft is waiting for a person:",
client.get("/tickets/1/draft", headers=SAM).json()["body"],
)
print("nothing was sent:", client.get("/tickets/1/sent", headers=SAM).json()["count"])
if __name__ == "__main__":
main()
How to run it¶
What it prints¶
POST /tickets/1/triage -> 202 {'job': 1, 'state': 'queued', 'ticket': 1}
the job went: queued -> running -> done
it finished as: done with awaiting_approval
the ticket is: awaiting_approval
the draft is waiting for a person: A refund takes three business days to appear.
nothing was sent: 0
Three states, and each one earns its place: queued is what the request was answered with, running
is what a client polls while it waits, and done carries the ticket's state as its detail. A job row
that only had "finished" would leave a client unable to tell a slow job from a lost one.
The detail on the failed path is the error's class name, not its message: the row is server-side
data, and provider messages travel further than people expect.
Common mistakes¶
- Holding the request open. The cheapest way to keep
p95honest is to not wait for something slow in the request at all. - A task with no row. Work that exists only inside an event loop cannot be polled, retried, or reviewed after a restart.
- Abandoning the work at shutdown.
aclosewaits. A half-triaged ticket is worse than a slow shutdown, and the write is what makes it half. - Letting the job's failure become the ticket's failure. The job records what happened; the ticket is untouched until a step succeeds. A failed job leaves a ticket somebody can try again.
- Polling forever with no terminal states. Say which states are terminal — here
doneandfailed— and let the client stop.
Next¶
Lesson 10 counts what all of this cost, and what the spend bought.