SciMesh/docs/database-integration-task.md

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.
• Result storage: a coordinator-managed local artifact directory in the first version; keep its interface replaceable by object storage later.
• 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.

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.

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:

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; return the new lease_expires_at
PUT /tasks/{task_id}/artifacts/{filename}	Stream one result artifact into coordinator storage; verify the current worker 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.

The artifact endpoint receives binary content with Content-Type, X-Worker-ID, and X-Task-Attempt headers. It must stream the request body to the coordinator artifact directory instead of buffering it in memory, calculate or verify its SHA-256, and return 201 with the durable artifact uri. Only the worker holding the current lease may upload. POST /result accepts only URIs returned by this endpoint for the same task and attempt.

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 and input download;
• video chunk generation and trajectory stitching;
• authentication provider, UI, object-storage implementation, and PDF report;
• network/distributed coordinator deployment beyond the local coordinator process.