Add worker and database implementation briefs

2026-07-21 17:08:07 +03:00
parent 134c1010f1
commit ccf11403bf
2 changed files with 363 additions and 0 deletions
@@ -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.
@@ -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: `<work-dir>/<task-id>/<attempt>/`.
+- 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 <local-input> --query-id ... --output <task-dir>/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.