diff --git a/docs/database-integration-task.md b/docs/database-integration-task.md new file mode 100644 index 0000000..fc85aa7 --- /dev/null +++ b/docs/database-integration-task.md @@ -0,0 +1,197 @@ +# Task: Go/PostgreSQL task queue service + +## Goal + +Implement a durable task-queue service in Go, backed by PostgreSQL. It must +store task state, atomically lease one pending task to a worker, persist result +metadata, and expose the data required by a future stitcher/report stage. + +This task implements the **queue and persistence layer** from the architecture +sketch. It does not implement the Worker Daemon or scientific/CV computation. + +## Technical baseline + +- Language: Go 1.22+. +- Database: PostgreSQL 15+. +- PostgreSQL driver and pool: `github.com/jackc/pgx/v5/pgxpool`. +- HTTP server: standard `net/http` is sufficient; do not introduce a framework + unless it solves a concrete requirement. +- Schema migrations: versioned SQL migrations using `golang-migrate`. +- Identifiers: UUID. +- Times: timezone-aware UTC timestamps. +- API boundary: the Go coordinator owns all database access. Python workers + communicate only over HTTP and never receive database credentials. + +If the coordinator application does not yet exist, create it with this minimal +layout. Do not add a distributed scheduler. + +```text +coordinator/ + cmd/coordinator/main.go + internal/config/ + internal/httpapi/ + internal/queue/ + internal/store/postgres/ + migrations/ +``` + +`main.go` must create one `pgxpool.Pool`, apply no migrations automatically in +production, wire dependencies, and perform graceful shutdown. Migrations are a +separate explicit command in CI/deployment. + +## Data model + +### `jobs` + +One user submission that may be split into several tasks. + +| Column | Type | Notes | +| --- | --- | --- | +| `id` | UUID PK | Generated by the service | +| `workload` | text | Allowed workload name | +| `input_uri` | text | Original dataset/video location | +| `parameters` | JSONB | Validated job-level parameters | +| `status` | enum | `pending`, `running`, `completed`, `failed`, `cancelled` | +| `created_at`, `completed_at` | timestamptz | UTC | + +### `tasks` + +One independently executable chunk. For an unsplit SciMesh workload, create one +task. For a video job, create one task per chunk. + +| Column | Type | Notes | +| --- | --- | --- | +| `id` | UUID PK | Task identifier returned to workers | +| `job_id` | UUID FK | References `jobs.id` | +| `chunk_index` | integer | Unique within a job | +| `workload` | text | Copied from job or an explicit task override | +| `input_uri` | text | Chunk or dataset artifact | +| `input_sha256` | text | Required checksum | +| `parameters` | JSONB | Task-specific validated parameters | +| `status` | enum | `pending`, `leased`, `completed`, `failed`, `cancelled` | +| `attempt` | integer | Starts at 0; incremented atomically on claim | +| `max_attempts` | integer | Default configurable, e.g. 3 | +| `lease_owner` | text nullable | Worker ID | +| `lease_expires_at` | timestamptz nullable | UTC | +| `result_uri` | text nullable | Result CSV/artifact location | +| `result_sha256` | text nullable | Checksum of result artifact | +| `metrics` | JSONB nullable | Safe execution metrics | +| `error_code`, `error_message` | text nullable | Sanitized failure information | +| `created_at`, `started_at`, `completed_at` | timestamptz | UTC | +| `version` | integer | Optimistic-concurrency/version marker | + +Required constraints and indexes: + +- unique `(job_id, chunk_index)`; +- index for claims: `(status, lease_expires_at, created_at)`; +- index on `job_id`; +- check `attempt >= 0` and `max_attempts > 0`; +- completed tasks must have `result_uri` and `result_sha256`; +- a leased task must have `lease_owner` and `lease_expires_at`. + +## Required Go service operations + +Keep HTTP handlers thin. Define a small queue service interface and a PostgreSQL +implementation; do not expose `pgx` rows or SQL details to handlers. + +```go +type QueueService interface { + CreateJobWithTasks(ctx context.Context, input CreateJobInput) (Job, error) + ClaimNextTask(ctx context.Context, input ClaimInput) (*ClaimedTask, error) + RenewLease(ctx context.Context, input RenewLeaseInput) (ClaimedTask, error) + CompleteTask(ctx context.Context, input CompleteTaskInput) (Task, error) + FailTask(ctx context.Context, input FailTaskInput) (Task, error) + ExpireLeases(ctx context.Context, now time.Time) (int64, error) + GetJobStatus(ctx context.Context, jobID uuid.UUID) (JobStatus, error) + ListCompletedResults(ctx context.Context, jobID uuid.UUID) ([]ResultManifest, error) +} +``` + +Every database call must receive a request-scoped `context.Context`. Configure +connection-pool size, database URL, and query/request timeouts via environment +variables. Use parameterized pgx queries only; do not build SQL with string +interpolation. + +`claim_next_task` must be atomic and safe with multiple coordinator processes. +Use one transaction with `SELECT ... FOR UPDATE SKIP LOCKED`, then update the +selected row to `leased`, increment `attempt`, and set the lease fields. Do not +implement claiming as `SELECT` followed by a separate unguarded `UPDATE`. + +Pseudo-SQL: + +```sql +WITH candidate AS ( + SELECT id + FROM tasks + WHERE status = 'pending' + AND attempt < max_attempts + AND workload = ANY($1) + ORDER BY created_at, chunk_index + FOR UPDATE SKIP LOCKED + LIMIT 1 +) +UPDATE tasks +SET status = 'leased', + attempt = attempt + 1, + lease_owner = $2, + lease_expires_at = $3, + started_at = COALESCE(started_at, $4), + version = version + 1 +FROM candidate +WHERE tasks.id = candidate.id +RETURNING tasks.*; +``` + +Expired leases must be returned to `pending` only when attempts remain; otherwise +mark the task `failed` with a lease-expired error. This should run before or as +part of each claim and as a periodic coordinator maintenance operation. + +## Go coordinator HTTP endpoints + +The Worker Daemon task depends on these endpoints. Claiming mutates state, so it +must be `POST` rather than `GET`. + +| Endpoint | Behavior | +| --- | --- | +| `POST /jobs` | Validate submission, create job and pending tasks transactionally | +| `POST /tasks/claim` | Atomically lease one compatible task; `204` when none exists | +| `POST /tasks/{task_id}/heartbeat` | Renew the current worker's lease | +| `POST /tasks/{task_id}/result` | Idempotently persist a completed result manifest | +| `POST /tasks/{task_id}/failure` | Record a safe failure or retryable state | +| `GET /jobs/{job_id}` | Return aggregate job/task progress | + +Every task-mutating handler must validate JSON, verify `worker_id` and +`attempt`, and map typed service errors to HTTP responses. Reject a stale or +foreign lease with `409 Conflict`. Completing the same task/attempt with the +same result manifest must be idempotent; a different manifest is a conflict. + +Use JSON request/response DTOs in `internal/httpapi`; map them to typed queue +inputs in the handler. Do not return raw database errors to clients. Return a +request ID in error responses and emit structured logs with the request ID, +task ID, worker ID, and operation. + +## Tests and acceptance criteria + +- `golang-migrate` upgrades an empty PostgreSQL database to the current schema + and can roll back the latest migration in a test database. +- Creating a job either creates all of its tasks or creates none. +- Concurrent claim tests prove that each pending task is leased to one worker + only; use a real PostgreSQL instance supplied through `TEST_DATABASE_URL`, + not SQLite or an in-memory mock. +- Lease expiry returns an unfinished task to the queue or fails it after the + final permitted attempt. +- A worker cannot renew, fail, or complete a task leased to another worker. +- Result submission is idempotent for the same attempt and manifest. +- `list_completed_results(job_id)` returns a deterministic order by + `chunk_index`, suitable for the future stitcher. +- API tests cover `204` for an empty queue, `409` for stale attempts, and + aggregate job status. +- `go test ./...` passes; run `go vet ./...` in CI. + +## Out of scope + +- Python Worker Daemon execution, input download, and result upload; +- video chunk generation and trajectory stitching; +- authentication provider, UI, object-storage implementation, and PDF report; +- network/distributed coordinator deployment beyond the local coordinator + process. diff --git a/docs/worker-daemon-task.md b/docs/worker-daemon-task.md new file mode 100644 index 0000000..e0900ce --- /dev/null +++ b/docs/worker-daemon-task.md @@ -0,0 +1,166 @@ +# Task: Worker Daemon + +## Goal + +Implement a standalone **Worker Daemon** that repeatedly obtains one task from +the central coordinator, runs it locally, and submits a result manifest. The +worker must not access the database directly. + +The target flow is: + +```text +Worker Daemon -> coordinator: claim task +coordinator -> Worker Daemon: task metadata + input location +Worker Daemon -> local SciMesh Core / CV runner: execute +Worker Daemon -> coordinator: submit result +``` + +The architecture sketch uses video chunks and CV, while the current SciMesh +repository contains local molecular workloads. Therefore the daemon must use a +small runner adapter: the first adapter may invoke a SciMesh CLI workload, and +future adapters may run a CV/video chunk processor. Do not put workload logic +inside the daemon. + +## Deliverables + +1. A Python module/package for the daemon and a console command, for example + `scimesh-worker`. +2. Configuration via environment variables and CLI overrides: + - `SCIMESH_COORDINATOR_URL` (required); + - `SCIMESH_WORKER_ID` (required, stable UUID or hostname-derived value); + - working directory for downloaded inputs and generated outputs; + - poll interval and request timeout; + - optional bearer token. +3. A `Runner` protocol and one `SciMeshRunner` implementation. The protocol + must make a future `VideoRunner` possible without changing daemon control + flow. +4. Structured logs containing `worker_id`, `task_id`, attempt number, state, + elapsed time, and error type. +5. Unit tests using a mocked HTTP coordinator and a fake runner. + +## Coordinator contract + +Use JSON over HTTPS. Claiming a task changes its state, so use `POST`, even if +the initial diagram labels the endpoint as `GET /get_task`. + +### Claim a task + +```http +POST /tasks/claim +Content-Type: application/json + +{ + "worker_id": "worker-01", + "capabilities": ["similarity-search", "similarity-graph"], + "max_concurrency": 1 +} +``` + +When no task is available, the coordinator returns `204 No Content`. + +When a task is available, it returns `200 OK`: + +```json +{ + "task_id": "0d2d5a53-4c7e-467e-93d2-45ed2dc18e46", + "attempt": 1, + "lease_expires_at": "2026-07-21T12:05:00Z", + "workload": "similarity-search", + "input": { + "uri": "https://coordinator.example/tasks/0d2d/input", + "sha256": "..." + }, + "parameters": { + "query_id": "CHEMBL939", + "top_k": 20 + } +} +``` + +`input.uri` may initially point to a coordinator download endpoint. Keep input +retrieval behind an `ArtifactClient` abstraction so it can later be replaced by +object storage without changing the daemon state machine. + +### Submit a result + +```http +POST /tasks/{task_id}/result +Content-Type: application/json + +{ + "worker_id": "worker-01", + "attempt": 1, + "status": "completed", + "result": { + "uri": "https://coordinator.example/tasks/0d2d/result.csv", + "sha256": "...", + "content_type": "text/csv" + }, + "metrics": { + "elapsed_seconds": 12.4, + "processed_rows": 10000 + } +} +``` + +For a failed execution, send `status: "failed"` with a short, sanitized +`error_code` and `error_message`. Never send a Python traceback, access token, +or local path outside the worker directory. + +## Required state machine + +```text +idle -> claiming -> downloading -> running -> uploading -> submitting -> idle + | | | | + +------------> failed <--------------------+ +``` + +- Poll only after a `204` response or a transient failure; use exponential + backoff with jitter and an upper bound. +- Verify the input checksum before running. +- Create one isolated task directory: `///`. +- Invoke the runner with an explicit argument list, never `shell=True`. +- Upload/submit exactly the produced result files listed by the runner. +- A timeout, network error, or rejected submission must leave the local task + directory available for diagnostics until a configurable cleanup period. +- Treat a duplicate successful submission as success when the coordinator + returns an idempotent response for the same `task_id` and `attempt`. + +## Runner interface + +The daemon owns task orchestration; the runner owns only local execution. + +```python +class Runner(Protocol): + def run(self, task: ClaimedTask, task_dir: Path) -> RunResult: + """Run one task and return output artifacts plus safe metrics.""" +``` + +`SciMeshRunner` should map `workload` and validated parameters to the existing +SciMesh CLI. For example, a `similarity-search` task invokes: + +```text +scimesh similarity-search --query-id ... --output /result.csv +``` + +Do not accept an arbitrary command from the coordinator. Maintain an allowlist +of registered workload names and validate every parameter before invocation. + +## Acceptance criteria + +- With a fake coordinator, the daemon claims one task, downloads a fixture, + invokes the fake runner once, and submits its CSV manifest. +- A `204` response does not create a task directory and waits before the next + poll. +- A bad input checksum prevents runner execution and reports a failed task. +- A transient claim/submit failure retries with bounded backoff. +- Two workers cannot both complete the same leased attempt; the daemon handles + a lease/submission conflict without corrupting local results. +- The daemon has no database driver or SQL queries. + +## Out of scope + +- FastAPI coordinator implementation; +- database schema and migrations; +- video segmentation, CV inference, and trajectory stitching; +- multiprocessing, distributed scheduling, and autoscaling.