Emil_Shanaty/SciMesh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into Workers

Browse Source
This commit is contained in:
Emil
2026-07-22 18:45:30 +03:00
parent 99dc321c69 ccf11403bf
commit 167b9e2aaa
2 changed files with 363 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/database-integration-task.md
+197
View File
@@ -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.
docs/worker-daemon-task.md
+166
View File
@@ -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.