patrick/Polaris
Watch
1
0
Fork
You've already forked Polaris
 0
Code Issues 35 Pull requests 3 Projects Releases Packages Wiki Activity Actions 3

PH0.3 — Worker skeleton: claim loop, task handler registry, jobs discovery, heartbeat #53

Merged
bart merged 6 commits from feature/worker-skeleton into main 2026-09-11 11:13:31 +00:00
Conversation 0 Commits 6 Files changed 27 +1089
bart commented 2026-09-11 10:40:44 +00:00
Collaborator
Copy link

Summary

  • worker/claim.py: claim_next_task() — SELECT ... FOR UPDATE SKIP LOCKED so several worker processes can poll the task table concurrently without double-claiming a row.
  • worker/retry.py + worker/loop.py: exponential, capped backoff (MAX_ATTEMPTS = 5); process_one() claims + dispatches + records the outcome (done / retried / terminally failed) on the task row; run_claim_loop() wraps it in an injectable poll loop.
  • worker/task_handlers.py + worker/handlers/: @task_handler("<kind>") decorator + pkgutil.walk_packages discovery, same conflict-free seam polaris.api uses for routers. handlers/noop.py is the concrete proof, mirroring api/routers/platform/health.py.
  • worker/job_discovery.py + worker/jobs/: modules exporting a schedule (APScheduler trigger) + run(), discovered the same way. jobs/heartbeat.py is the required real deliverable (AC1).
  • worker/scheduler.py + worker/main.py + worker/__main__.py: build_scheduler() wires discovered jobs into a BackgroundScheduler; main() starts it alongside run_claim_loop(); runnable via python -m polaris.worker.
  • domain/tasks.py (new domain package, this file only): enqueue() — the single helper other domain services use to push work onto the task table, via a TaskRepository protocol it defines itself (polaris.domain must not import sqlalchemy.orm.Session directly, import rules table in docs/12-implementation-plan.md).
  • db/repositories/task.py: SqlAlchemyTaskRepository, the concrete implementation — an explicitly-scoped extension to polaris.db called for by the ticket brief (PH0.2 didn't need a repository layer yet).

Acceptance criteria (issue #6)

  • Worker starts against the compose stack and logs a heartbeat on schedule — not fully provable yet: infra/compose/docker-compose.yml's worker service runs command: ["polaris", "worker"], but there is no cli/worker.py subcommand (confirmed: docker run <image> worker → invalid choice: 'worker'). backend/src/polaris/cli/** is sealed to this branch per my ticket brief. Filed as seam request #52 (cli/worker.py, ~15 lines mirroring cli/api.py, wraps polaris.worker.main:main()). Everything else needed for this criterion is done and tested: handler/job discovery, the scheduler, and the heartbeat job itself all work (verified with a standalone smoke test and inside the built Docker image).
  • A task enqueued via the domain helper is claimed exactly once by one of two concurrent workers and its handler runs
  • A failing handler is retried with backoff and ends in a failed state after the retry budget; state changes are visible on the task row
  • A handler module dropped into the handlers package is registered without editing a shared file; same for a job module
  • Unit tests cover claim, retry and registry; ruff/mypy/pytest green

Test plan

  • uv run ruff format --check . / uv run ruff check . — clean
  • uv run mypy --strict src — clean
  • uv run pytest — 60/60 passing (18 pure unit + 12 integration for worker/domain, rest pre-existing), including a genuine two-connection SKIP LOCKED concurrency test (test_two_concurrent_claimants_get_the_task_at_most_once) against a real Postgres testcontainer
  • docker build backend/ — succeeds; python -m polaris.worker imports cleanly inside the built image

Notes / seam requests

  • #52 — cli/worker.py needed for docker compose up worker to actually start the process (see above).
  • Did not use the Forgejo stopwatch on this ticket, per the corrected guidance in CLAUDE.md (one clock per account, shared across parallel sessions on this token) relayed mid-task.
  • Assignee on #6 is patrick, matching the convention used on the already-merged #4.
  • Branched from the old develop tip (pre-merge into main); fast-forwarded onto origin/main once develop was retired mid-task, so this PR targets main as instructed.

Closes #6

🤖 Generated with Claude Code

https://claude.ai/code/session_013YioTVKBPoE6thZqbnTtnM

## Summary - `worker/claim.py`: `claim_next_task()` — `SELECT ... FOR UPDATE SKIP LOCKED` so several worker processes can poll the `task` table concurrently without double-claiming a row. - `worker/retry.py` + `worker/loop.py`: exponential, capped backoff (`MAX_ATTEMPTS = 5`); `process_one()` claims + dispatches + records the outcome (done / retried / terminally failed) on the task row; `run_claim_loop()` wraps it in an injectable poll loop. - `worker/task_handlers.py` + `worker/handlers/`: `@task_handler("<kind>")` decorator + `pkgutil.walk_packages` discovery, same conflict-free seam `polaris.api` uses for routers. `handlers/noop.py` is the concrete proof, mirroring `api/routers/platform/health.py`. - `worker/job_discovery.py` + `worker/jobs/`: modules exporting a `schedule` (APScheduler trigger) + `run()`, discovered the same way. `jobs/heartbeat.py` is the required real deliverable (AC1). - `worker/scheduler.py` + `worker/main.py` + `worker/__main__.py`: `build_scheduler()` wires discovered jobs into a `BackgroundScheduler`; `main()` starts it alongside `run_claim_loop()`; runnable via `python -m polaris.worker`. - `domain/tasks.py` (new `domain` package, this file only): `enqueue()` — the single helper other domain services use to push work onto the `task` table, via a `TaskRepository` protocol it defines itself (`polaris.domain` must not import `sqlalchemy.orm.Session` directly, import rules table in `docs/12-implementation-plan.md`). - `db/repositories/task.py`: `SqlAlchemyTaskRepository`, the concrete implementation — an explicitly-scoped extension to `polaris.db` called for by the ticket brief (PH0.2 didn't need a repository layer yet). ## Acceptance criteria (issue #6) - [ ] Worker starts against the compose stack and logs a heartbeat on schedule — **not fully provable yet**: `infra/compose/docker-compose.yml`'s `worker` service runs `command: ["polaris", "worker"]`, but there is no `cli/worker.py` subcommand (confirmed: `docker run <image> worker` → `invalid choice: 'worker'`). `backend/src/polaris/cli/**` is sealed to this branch per my ticket brief. Filed as seam request **#52** (`cli/worker.py`, ~15 lines mirroring `cli/api.py`, wraps `polaris.worker.main:main()`). Everything else needed for this criterion is done and tested: handler/job discovery, the scheduler, and the heartbeat job itself all work (verified with a standalone smoke test and inside the built Docker image). - [x] A task enqueued via the domain helper is claimed exactly once by one of two concurrent workers and its handler runs - [x] A failing handler is retried with backoff and ends in a failed state after the retry budget; state changes are visible on the task row - [x] A handler module dropped into the handlers package is registered without editing a shared file; same for a job module - [x] Unit tests cover claim, retry and registry; ruff/mypy/pytest green ## Test plan - `uv run ruff format --check .` / `uv run ruff check .` — clean - `uv run mypy --strict src` — clean - `uv run pytest` — 60/60 passing (18 pure unit + 12 integration for worker/domain, rest pre-existing), including a genuine two-connection `SKIP LOCKED` concurrency test (`test_two_concurrent_claimants_get_the_task_at_most_once`) against a real Postgres testcontainer - `docker build backend/` — succeeds; `python -m polaris.worker` imports cleanly inside the built image ## Notes / seam requests - **#52** — `cli/worker.py` needed for `docker compose up worker` to actually start the process (see above). - Did **not** use the Forgejo stopwatch on this ticket, per the corrected guidance in `CLAUDE.md` (one clock per account, shared across parallel sessions on this token) relayed mid-task. - Assignee on #6 is `patrick`, matching the convention used on the already-merged #4. - Branched from the old `develop` tip (pre-merge into `main`); fast-forwarded onto `origin/main` once `develop` was retired mid-task, so this PR targets `main` as instructed. Closes #6 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_013YioTVKBPoE6thZqbnTtnM
polaris.domain must not import sqlalchemy.orm.Session directly (import
rules, docs/12-implementation-plan.md), so domain/tasks.py (next commit)
defines a TaskRepository protocol instead. PH0.2 did not need a
repository layer yet; this is the minimal, explicitly-scoped extension
to polaris.db the PH0.3 ticket brief calls for.

Refs: #6

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013YioTVKBPoE6thZqbnTtnM
domain/tasks.py is the single entry point future domain services use
to push work onto the task table (CLAUDE.md: "Long work goes through
the task table to the worker — never run the solver inside a
request"). Creates the domain package with only this file, per the
PH0.3 ticket brief.

Refs: #6

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013YioTVKBPoE6thZqbnTtnM
@task_handler(kind) in worker/handlers/<kind>.py and a schedule + run()
in worker/jobs/<name>.py are found by pkgutil.walk_packages, the same
conflict-free seam polaris.api uses for routers - no shared file to
edit when a new handler or job module is added. noop.py and
heartbeat.py are this ticket's concrete proof, matching how
api/routers/platform/health.py proves router discovery.

Refs: #6

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013YioTVKBPoE6thZqbnTtnM
claim_next_task() uses SELECT ... FOR UPDATE SKIP LOCKED so several
worker processes can poll the task table concurrently without
double-claiming a row (docs/07-architecture.md). process_one() claims
one task, dispatches it to its registered handler, and records the
outcome on the row: done, retried with exponential capped backoff, or
failed once MAX_ATTEMPTS is exhausted. run_claim_loop() wraps it in a
poll loop, injectable (handlers, iterations, sleep) for deterministic
tests.

Refs: #6

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013YioTVKBPoE6thZqbnTtnM
build_scheduler() runs discovered jobs/ modules on an APScheduler
BackgroundScheduler alongside the claim loop. worker.main:main() wires
handler discovery, the scheduler and run_claim_loop together into the
worker process; python -m polaris.worker runs it directly.

No cli/worker.py subcommand is added here: backend/src/polaris/cli/**
is sealed to this branch per the PH0.3 ticket brief, even though
infra/compose/docker-compose.yml's worker service already expects
`polaris worker` to exist. Filed as seam request #52 (comment on #6);
main() above is what a future cli/worker.py trivially wraps.

Refs: #6

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013YioTVKBPoE6thZqbnTtnM
Unit tests for backoff policy, task/job discovery (synthetic packages
plus the real noop/heartbeat modules as concrete proof), and
build_scheduler. Integration tests (real Postgres via testcontainers,
per docs/10-dev-workflow.md - not mocks for anything DB-shaped) cover
claim_next_task including a genuine two-connection SKIP LOCKED
concurrency test, process_one's done/retry/terminal-failure state
transitions on the task row, and domain.tasks.enqueue() against both a
fake repository and the real SqlAlchemyTaskRepository. 60/60 backend
tests, ruff and mypy --strict green.

Closes: #6

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013YioTVKBPoE6thZqbnTtnM
bart merged commit 5c95587374 into main 2026-09-11 11:13:31 +00:00
Sign in to join this conversation.
Reviewers
No reviewers
Labels
Clear labels   
ready-for-agent
Self-contained ticket an agent can pick up

No labels
ready-for-agent
Milestone
Clear milestone
No items
No milestone
Projects
Clear projects
No items
No project
Assignees
Clear assignees
No assignees
1 participant
Notifications
Due date
The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set

Reference
patrick/Polaris!53
No description provided.