feat: renumber lib reference into 4-pillar anatomy
Renumber the 15 reference docs into pillar groups (recipe -> anatomy): - 01 Build Recipe (entry), 02 Composition Root - domains: 10 Layering Rules - contract: 20 Spec-First Routing, 21 CRUD Template, 22 Auth Integration - interfaces/infra: 30 Infra Adapters - UI: 40 UI: Auto-Rendered Admin & Custom Pages (new; @aetoskia packages, client template, x- catalog via react-openapi MCP module) - pipelines: 50 Pipeline Scaffold, 51 Worker & Progress, 52 Human-in-the-Loop - delivery/meta: 90 Drone Pipelines, 91 Versioning & Changelog, 92 Genericity Matrix, 93 Library Registry, 94 Repository Inventory MCP module names are unchanged (NN_ prefix stripped); all cross-references (wiki, templates, hooks, registry) updated; lib index rewritten pillar-first; build recipe bootstraps the client template. Bundle + HTML sites regenerated.
This commit is contained in:
95
standards/src/lib/51_worker_and_progress.md
Normal file
95
standards/src/lib/51_worker_and_progress.md
Normal file
@@ -0,0 +1,95 @@
|
||||
# Worker & Progress
|
||||
|
||||
## Summary
|
||||
|
||||
Long-running work runs in a background worker. Jobs are durable documents that
|
||||
move through an explicit status machine. Progress is streamed to clients.
|
||||
|
||||
---
|
||||
|
||||
## The status machine
|
||||
|
||||
```
|
||||
pending -> processing -> completed
|
||||
\-> failed (retried up to N times)
|
||||
\-> paused (waiting for human input)
|
||||
```
|
||||
|
||||
- `pending` — queued, eligible for pickup.
|
||||
- `processing` — claimed by a worker.
|
||||
- `paused` — blocked on a human-in-the-loop question.
|
||||
- `failed` — error; `retry_count` increments until a max, then skipped.
|
||||
- `completed` — terminal success.
|
||||
|
||||
## The poll loop
|
||||
|
||||
```python
|
||||
async def poll(resources):
|
||||
backoff = MIN_POLL
|
||||
while True:
|
||||
jobs = await job_svc.search(filter={"status": "pending"}, limit=100)
|
||||
backoff = MIN_POLL if jobs else min(backoff * 2, MAX_POLL)
|
||||
for job in jobs:
|
||||
if (job.retry_count or 0) >= MAX_RETRIES:
|
||||
continue
|
||||
await job_svc.update_status(job.id, "processing")
|
||||
try:
|
||||
await process(job, resources)
|
||||
except PauseForResolution:
|
||||
await job_svc.update_status(job.id, "paused")
|
||||
except Exception as e:
|
||||
await job_svc.update_status(
|
||||
job.id, "failed", error=str(e),
|
||||
retry_count=(job.retry_count or 0) + 1,
|
||||
)
|
||||
await asyncio.sleep(backoff)
|
||||
```
|
||||
|
||||
Tunables live in environment variables: `WORKER_POLL_MIN`, `WORKER_POLL_MAX`,
|
||||
`WORKER_MAX_RETRIES`.
|
||||
|
||||
## Progress events
|
||||
|
||||
A worker publishes structured progress events to a **bus** (a port). The HTTP
|
||||
edge exposes a streaming endpoint that replays events for a given job id.
|
||||
|
||||
```python
|
||||
class ProgressEvent(BaseModel):
|
||||
job_id: str
|
||||
step: str
|
||||
status: str # started | progress | completed | skipped | paused
|
||||
message: dict | None = None
|
||||
```
|
||||
|
||||
The bus is an in-memory adapter by default; because it is a port, it can be
|
||||
swapped for a durable broker. Emit from the orchestrator at each stage:
|
||||
|
||||
```python
|
||||
await self.bus.publish(ProgressEvent(job_id=jid, step="normalize",
|
||||
status="completed", message={"count": n}))
|
||||
```
|
||||
|
||||
## Streaming endpoint
|
||||
|
||||
```python
|
||||
async def stream_job_events(request, id):
|
||||
bus = request.app.state.db_resources.get("progress_bus")
|
||||
if bus is None:
|
||||
return {"events": []}
|
||||
|
||||
async def gen():
|
||||
sent = 0
|
||||
while True:
|
||||
events = bus.get_events(id)
|
||||
for e in events[sent:]:
|
||||
yield f"data: {e.model_dump_json(exclude_none=True)}\n\n"
|
||||
sent += 1
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
return StreamingResponse(gen(), media_type="text/event-stream")
|
||||
```
|
||||
|
||||
> **Known limitation.** The default in-memory bus is per-process. If the worker
|
||||
> and the API run as separate processes, events published by the worker are not
|
||||
> visible to the API. For production, back the bus with a shared broker. The
|
||||
> port boundary makes this a one-adapter change.
|
||||
Reference in New Issue
Block a user