Compare commits

...
Author SHA1 Message Date
Emil a201dd5ef9 Reject stale sessions before worker key creation
users / test (push) Canceled after 0s
2026-07-27 22:28:02 +03:00
Efremenko Arhip e2a57175a0 fix(users): use errors.Is in worker key name test (errorlint) 2026-07-27 16:57:49 +03:00
Efremenko Arhip 2991ed202b feat(demo): add demo-reset to wipe the demo's data volumes
demo-down / stop intentionally keep the Postgres and artifact volumes, so a
fresh start still carried old workers/jobs/tasks. Add a reset action (compose
down -v) and a make demo-reset target for a truly pristine restart.
2026-07-27 16:48:06 +03:00
Efremenko Arhip 9b235282fc fix(coordinator/ui): use the entered machine name in the generated worker command
The Add-your-machine page sent the typed name only to the worker key, while
the copyable command hard-coded --worker-name my-machine, so every self-service
worker registered as my-machine regardless of what the user entered. Thread the
name into the command (shell-quoted; falls back to my-machine when blank).
2026-07-27 16:48:06 +03:00
Efremenko Arhip 3a1461315f feat: self-service worker enrollment bound to a user account
Let a signed-in user turn their own machine into a worker without the
shared token. The coordinator already binds a JWT-authenticated
registration to owner_id as untrusted; this adds the missing pieces.

userservice: long-lived worker keys (scimesh_wk_live_*, hash-at-rest)
with create/list/revoke and a public /worker-tokens/exchange that trades
a key for a short-lived JWT carrying the owner current role/verified.

python worker: SCIMESH_WORKER_KEY + SCIMESH_USERSERVICE_URL; a token
provider exchanges the key and refreshes the JWT proactively and on 401,
so a long-running worker survives token expiry. Static bearer token path
is unchanged.

coordinator UI: an "add your machine" page that mints a key and shows a
ready-to-run command, proxying key management to the userservice; the
dashboard gains an owner-scoped "my machines" section.

docs: how to run a worker from your account, plus the untrusted/quorum/
verified trust model.
2026-07-27 16:11:07 +03:00
Efremenko Arhip 172ff76fb8 fix(test): update NewCompleteTask call to new signature
Pass workers/results repos and quorum in the postgres integration
test, matching the constructor change that added quorum voting.
2026-07-27 11:29:52 +03:00
Efremenko Arhip 6f14eeb32e fix(coordinator): silence nilerr on unresolvable-worker trust fallback 2026-07-27 11:05:36 +03:00
Efremenko Arhip 18d58cce84 feat(coordinator): quorum verification for untrusted (volunteer) results
Replaces the C1 quarantine with real verification. Trusted results (lab token,
verified, or admin worker) are accepted directly as before. An untrusted
worker's result is recorded as one vote per (task, owner) in a new task_results
table; the task only completes once QUORUM_SIZE distinct owners submit the same
result hash, otherwise it returns to the queue for another independent compute.

- migration 0013 task_results (one vote per owner, quorum by result_sha256)
- CompleteTask branches on worker trust; unknown worker defaults trusted (safe:
  completing needs the lease, whose owner is always a known registered worker)
- claim drops the quarantine and excludes chunks the owner already voted on
- domain Task.ReleaseAfterVote; QUORUM_SIZE config (default 2)
- unit tests: trusted direct-complete, untrusted needs-quorum, can-claim

Reducer and job done/total logic untouched — still one completed task per chunk.
2026-07-26 22:55:51 +03:00
Efremenko Arhip dcabfcd0c3 feat(coordinator): business metrics — jobs/tasks/workers by status
A scrape-time collector reports scimesh_tasks/jobs/workers gauges keyed by
status, sourced from cheap GROUP BY queries (StatsRepo), zero-filled across all
known statuses so the dashboard shows flat zeros instead of gaps. A failed
query yields no samples for that scrape rather than crashing it.

Metrics is now built in main so the DB-backed collector can be registered
(NewServer takes *metrics.Metrics; nil self-provisions for tests). Grafana
dashboard gains a Domain state row: tasks/jobs/workers by status and a queue-
depth stat.
2026-07-26 22:16:35 +03:00
Efremenko Arhip e9cf6f0842 build(coordinator): go 1.25 (prometheus deps require it), bump Docker base
The metrics deps pull in modules whose go directive is 1.25, so go mod tidy
raised the module to go 1.25.0. The build image is bumped golang:1.24 ->
golang:1.25-alpine to match (the image runs GOTOOLCHAIN=local and can't
auto-fetch a newer toolchain). CI reads go-version-file, so it follows along.
client_golang pinned to v1.19.1 (stable, same API).
2026-07-26 21:59:49 +03:00
Efremenko Arhip 779ff8c10e feat(demo): Prometheus + Grafana with a provisioned dashboard
make demo-ui now also starts Prometheus (scrapes coordinator:8080/metrics) and
Grafana with a provisioned datasource and a SciMesh Coordinator dashboard
(request rate & p95 by route, status mix, goroutines, RSS). Grafana allows
anonymous viewing so the dashboard opens without a login; admin/admin to edit.

- monitoring/prometheus.yml + grafana provisioning + dashboard JSON
- docker-compose.monitoring.yml overlay (third -f in demo-ui.sh)
- demo prints the Grafana and Prometheus URLs
2026-07-26 21:49:52 +03:00
Efremenko Arhip e584cfc481 feat(coordinator): Prometheus /metrics endpoint
Adds internal/metrics: a private registry with the Go runtime + process
collectors and HTTP RED instrumentation (scimesh_http_requests_total and
request_duration_seconds), labelled by method/status and a normalized route so
per-id paths collapse to {id} and never blow up label cardinality.

/metrics is unauthenticated (like /health) for a Prometheus scraper; the
middleware wraps the whole router so every request is measured once.
2026-07-26 21:47:27 +03:00
Efremenko Arhip 4ac19999a9 feat(coordinator): profile page at /ui/profile
Shows the signed-in user's id, email, role, verified status, and created-at by
proxying the session token to the userservice /me endpoint (email/created_at
are not in the JWT). Profile link added to the dashboard, job, and admin headers.
Tests: /me proxy forwards the bearer and renders the account; redirect without a session.
2026-07-26 21:39:08 +03:00
Efremenko Arhip df4bdc9de9 feat(coordinator): admin panel at /ui/admin
Admin-only page to promote/demote/verify/unverify a user by id. Actions proxy
to the userservice forwarding the admin's session JWT, which the userservice
re-checks (defense in depth). Non-admins are redirected off the panel; the
Admin link shows only for admins. Admins already see all jobs on the dashboard.

Tests: requireAdmin gate, bearer forwarding, action/id validation, admin link.
2026-07-26 21:27:39 +03:00
Efremenko Arhip 9a458ec4ef feat(users): admin promote/demote endpoints
POST /users/{id}/promote and /demote set a user's role (admin/user), admin-only
(403 otherwise). Mirrors the verify endpoints: SetRole use case + repo method,
validated role. Unit, admin-flow, and integration tests included.
2026-07-26 21:14:32 +03:00
Efremenko Arhip 49eb662798 feat(coordinator): logout button in the operator UI
The dashboard and job pages show a 'Log out' control (POST /ui/logout) and a
'Signed in · <role>' label when a userservice session is active. Under basic
auth (no session) neither appears, so the fallback UI is unchanged.

Threads a template-only Session view (json:"-") from authctx into the
dashboard and job views. Tests assert the control renders only in session mode.
2026-07-26 21:03:50 +03:00
Efremenko Arhip 5a9a10c681 feat(demo): bring up coordinator + userservice, seed root admin
make demo-ui now starts the userservice (own Postgres + migrations) beside the
coordinator on a shared JWT secret, and seeds a root admin. The coordinator runs
in UI session mode, so /ui opens a login page instead of a basic-auth prompt.

- docker-compose.users.yml overlay: userservice stack + coordinator JWT wiring
- demo-ui.sh: waits for the userservice, logs in as the seeded admin to poll the
  now session-gated dashboard, and prints the admin credentials
- validated with docker compose config (6 services, merged env)
2026-07-26 20:55:54 +03:00
Efremenko Arhip c8c6455caf feat(coordinator): owner-scope UI views by logged-in user
With a UI session, the dashboard and job pages are scoped to the caller:
- Overview lists only the user's own jobs (admin/basic-auth operator: all)
- JobDetail, artifact download, and preview 404 on another user's job
- scoping keys off authctx: no requester (basic auth) still sees everything,
  so the fallback operator UI is unchanged

ListJobs gains an owner filter (SQL WHERE) so paging stays correct per user.
Tests cover Overview scoping and cross-user JobDetail rejection.
2026-07-26 20:12:49 +03:00
Efremenko Arhip 33f629f387 feat(coordinator): UI login/register via userservice (cookie session)
When JWT_SECRET + USERSERVICE_URL are set, the operator UI authenticates
through userservice login/registration instead of the static UI_AUTH_TOKEN:
- /ui/login, /ui/register, /ui/logout pages proxy to the userservice
- successful login stores the JWT in an httpOnly, /ui-scoped cookie
- withUISession verifies the cookie locally and stamps the requester
- unset -> falls back to basic auth, so the team's existing flow is unchanged

Tests cover the session gate, cookie set/clear, and the login/register proxy.
2026-07-26 19:54:39 +03:00
Efremenko Arhip a7e949a0a7 feat(users): bootstrap first admin on startup
BOOTSTRAP_ADMIN_EMAIL/PASSWORD seed a role=admin account at boot if absent —
the only way to get the first admin, since /register makes plain users and
promotion needs an existing admin. Idempotent and race-safe. Tests included.
2026-07-26 19:44:06 +03:00
Efremenko Arhip 163cbe14bf fix(coordinator): bind JWT caller to worker at claim (close quarantine bypass)
The trust tier was read off the caller-supplied worker_id, so a JWT user who
knew any trusted worker's id could claim as it — draining and poisoning the
trusted queue and bypassing the untrusted-worker quarantine entirely.

Claim now requires a JWT caller to own the worker it acts as; a shared-token
caller (lab operator) may still act as any worker. Claim is the sole grantor of
a lease, so this also protects the downstream heartbeat/result/failure paths.

Tests: reject claim as another user's worker; allow claim as own worker.
2026-07-26 19:26:30 +03:00
Efremenko Arhip 80ff72a0fe feat(coordinator): worker trust tiers (C1) — enroll volunteers, quarantine untrusted
- migration 0012: workers.owner_id + trust_level (trusted/untrusted)
- verifier/authctx read the JWT verified claim; IsTrusted() = admin||verified
- /workers/register resolves trust from auth: service token or verified/admin
  JWT -> trusted; plain user JWT -> untrusted, tagged with owner_id
- claim quarantines untrusted workers (no tasks) until quorum (C2) lands
- unit tests for trust resolution, quarantine, and the verified claim

Additive and backward compatible: shared-token workers stay trusted, so the
existing worker flow and team tests are unchanged. Quorum verification (C2)
is deferred.
2026-07-26 19:18:50 +03:00
Efremenko Arhip c6a66747eb feat(users): add admin-granted verified badge for trusted contributors
- migration 0002: users.verified boolean, default false
- verified rides in the JWT (role + verified claims)
- POST /users/{id}/verify + /unverify, admin-only (403 otherwise)
- Issue now takes the whole user so trust claims travel in the token
- unit + integration + admin-flow tests
2026-07-26 19:10:01 +03:00
Efremenko Arhip 0c1f5f06d4 chore(users): remove coordinator leftovers, tidy for userservice
- delete api/requests.http and ARCHITECTURE.md (coordinator content)
- rewrite README.md and .env.example for the userservice
- drop dead RunPeriodic (reaper machinery userservice has no use for)
- fix .gitignore/.dockerignore/.golangci.yml module + artifact names
- degeneralize stale copied comments that said "coordinator"
2026-07-26 16:36:17 +03:00
Efremenko Arhip 67407220c3 ci(users): add pipeline (vet, gofmt, race, lint, migrate, integration)
Mirrors the coordinator workflow against a scimesh_users Postgres service.
Switches the test request helper to http.NewRequestWithContext so the noctx
linter passes on the go1.22 module.
2026-07-26 16:27:24 +03:00
Efremenko Arhip 1b1b971378 fix(coordinator): scan owner_id in UIReadRepo.ListJobs
jobColumns gained owner_id but ListJobs' Scan still read 12 targets,
so the query returned 13 columns and pgx failed at runtime. Only the
integration tests (real DB) caught it; memstore-backed unit tests did not.
2026-07-26 16:24:34 +03:00
Efremenko Arhip 73196579e8 test(users): cover config, usecase error paths, and /me 500 path 2026-07-26 16:15:36 +03:00
Efremenko Arhip a3db1a1e67 add users logic 2026-07-26 14:38:24 +03:00
Emil 16db1e41f7 Link team profiles in README 2026-07-25 15:02:28 +03:00
Emil ad9cc8f95c Add team section to README 2026-07-25 14:54:50 +03:00
Emil 746958884e Document local pipeline demo 2026-07-24 17:37:46 +03:00
Emil 7ad28b939d Show live processing speed chart 2026-07-24 17:28:16 +03:00
Emil 1012f5d95a Allow demo worker count override 2026-07-24 17:21:45 +03:00
Emil f5baec507c Add final result preview button 2026-07-24 17:19:31 +03:00
Emil 5a1414bee9 Add local pipeline demo launcher 2026-07-24 17:13:26 +03:00
Emil 8b738efd5d Add safe artifact previews 2026-07-24 17:04:07 +03:00
Emil d0aeb7fc95 Upgrade pipeline observability UI
coordinator / test (push) Canceled after 0s
2026-07-24 15:13:59 +03:00
155 changed files with 9295 additions and 224 deletions
+66
View File
@@ -0,0 +1,66 @@
name: users
on:
push:
paths:
- "users/**"
- ".github/workflows/users.yml"
pull_request:
paths:
- "users/**"
- ".github/workflows/users.yml"
defaults:
run:
working-directory: users
jobs:
test:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:16-alpine
env:
POSTGRES_USER: scimesh
POSTGRES_PASSWORD: scimesh
POSTGRES_DB: scimesh_users
ports:
- 5432:5432
options: >-
--health-cmd "pg_isready -U scimesh"
--health-interval 5s
--health-timeout 3s
--health-retries 10
env:
TEST_DATABASE_URL: postgres://scimesh:scimesh@localhost:5432/scimesh_users?sslmode=disable
steps:
- uses: actions/checkout@v4
- uses: actions/setup-go@v5
with:
go-version-file: users/go.mod
cache-dependency-path: users/go.sum
- name: go vet
run: go vet ./...
- name: gofmt
run: test -z "$(gofmt -l .)" || (gofmt -l . && exit 1)
- name: unit tests (race)
run: go test -race ./...
- name: lint
run: go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.12.2 run --build-tags=integration ./...
- name: install migrate CLI
run: go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@v4.17.1
- name: apply migrations
run: migrate -path migrations -database "$TEST_DATABASE_URL" up
- name: integration tests
run: go test -tags=integration ./internal/storage/postgres/ -v
+1
View File
@@ -15,3 +15,4 @@ test_structures/
# Local coordinator-worker execution state
worker-data*/
scimesh-worker-data/
coordinator/.demo/
+24
View File
@@ -0,0 +1,24 @@
.DEFAULT_GOAL := help
.PHONY: help demo-ui demo-down demo-logs
help:
@printf '%s\n' \
'SciMesh developer commands:' \
' make demo-ui Start the local UI pipeline demo with 2 workers.' \
' make demo-ui WORKERS=3 Start the demo with 3 local workers.' \
' make demo-logs Follow coordinator logs for the demo.' \
' make demo-down Stop demo containers and workers.' \
'' \
'After make demo-ui: open http://localhost:18080/ui (operator / demo-ui-secret).'
# Convenient entry points from the repository root. Extra settings are passed
# through, for example: make demo-ui WORKERS=3
demo-ui:
$(MAKE) -C coordinator demo-ui
demo-down:
$(MAKE) -C coordinator demo-down
demo-logs:
$(MAKE) -C coordinator demo-logs
+28
View File
@@ -44,6 +44,27 @@ scimesh similarity-search --help
scimesh similarity-graph --help
```
## Manual pipeline demo
To inspect the coordinator, Web UI, and distributed `similarity-search`
pipeline by hand, install development dependencies once and start the isolated
demo from the repository root:
```bash
python3 -m venv .venv
.venv/bin/pip install -e '.[dev]'
make demo-ui
```
Open `http://localhost:18080/ui` and sign in with username `operator` and
password `demo-ui-secret`. The command starts PostgreSQL, the coordinator, and
two local reference workers. Upload a small ChEMBL TSV, then use the job page
to follow shard progress, inspect bounded **Preview CSV** results, and see a
live processing-speed chart in shards per minute. To change the worker count,
run `make demo-ui WORKERS=3`; stop everything with `make demo-down`.
Run `make help` to display these commands in the terminal.
## Similarity search
`similarity-search` finds the top-k molecules most similar to a query. The query is supplied either by ChEMBL ID or by SMILES. It uses Morgan fingerprints with `radius=2` and `fpSize=2048`, Tanimoto similarity, streaming TSV reads, and a bounded heap. Invalid SMILES and the query molecule are skipped.
@@ -111,3 +132,10 @@ pytest
```
The package separates common dataset parsing and fingerprints from independent workloads. Add future workloads through the workload registry without changing the main CLI.
## Team
- [Emil](https://github.com/emil28092005) — Project Lead
- [Kristina](https://github.com/kristtma) — Tech Lead
- [Arkhip](https://github.com/hIpa-ussr) — Programmer
- [Makar](https://github.com/RERAN4K) — Programmer
+1 -1
View File
@@ -38,7 +38,7 @@ real PostgreSQL smoke test) passed on 2026-07-24.
| CTX-08 Distributed similarity-search | Implemented | Python planner resolves `query_id` once, creates deterministic shard plans, worker adapter emits exact partial top-k CSVs/metrics, and reducer matches the local reference. |
| CTX-09 Reducer and final-result API | Implemented | Atomic `reducing` claim, deterministic coordinator-side top-k reducer, sanitized reducer failure, final artifact persistence, `result_uri`, and final CSV download. |
| CTX-10 Distributed similarity-graph | Not started | Local reference exists. |
| CTX-11 Dashboard/operator view | Implemented | Protected local view: job/task/worker status, validated similarity-search upload, partial-artifact diagnostics, final-result download, and bounded polling. |
| CTX-11 Dashboard/operator view | Implemented | Protected live control room: recent-run/worker overview, real pipeline-stage visualization, shard attempts and safe failures, validated similarity-search upload, coordinator artifacts, final-result download, and bounded polling. |
| CTX-12 Reliability, security, CI | In progress | Unit, race, PostgreSQL integration, and smoke checks exist; CI hardening remains. |
## Next recommended assignment
+1 -1
View File
@@ -5,7 +5,7 @@
# build fails with "the --mount option requires BuildKit".
# --- build stage ----------------------------------------------------------
FROM golang:1.24-alpine AS build
FROM golang:1.25-alpine AS build
WORKDIR /src
+66 -1
View File
@@ -1,4 +1,6 @@
.PHONY: build run test test-integration vet lint tidy check migrate-up migrate-down up down down-clean logs ps rebuild psql smoke
.DEFAULT_GOAL := help
.PHONY: help build run test test-integration vet lint tidy check migrate-up migrate-down up down down-clean logs ps rebuild psql smoke demo-ui demo-down demo-reset demo-logs
# `check` deliberately uses its own Compose project and host ports. This keeps
# it from connecting to or replacing a developer's local PostgreSQL instance.
@@ -10,6 +12,69 @@ CHECK_TOKEN ?= dev-token
CHECK_DATABASE_URL ?= postgres://scimesh:scimesh@localhost:$(CHECK_POSTGRES_PORT)/scimesh?sslmode=disable
CHECK_COMPOSE = POSTGRES_PORT=$(CHECK_POSTGRES_PORT) COORDINATOR_PORT=$(CHECK_COORDINATOR_PORT) docker compose -p $(CHECK_PROJECT)
# --- local manual demo ---------------------------------------------------
# A separate project and ports mean this demo cannot collide with the normal
# `make up` stack or a developer's local PostgreSQL on 5432.
DEMO_PROJECT ?= scimesh-demo
DEMO_POSTGRES_PORT ?= 55432
DEMO_COORDINATOR_PORT ?= 18080
DEMO_UI_TOKEN ?= demo-ui-secret
DEMO_WORKER_TOKEN ?= demo-worker-token
DEMO_WORKERS ?= 2
# Short public knob for `make demo-ui WORKERS=3`; DEMO_WORKERS remains useful
# for scripts and backwards-compatible documentation.
WORKERS ?= $(DEMO_WORKERS)
DEMO_DIR ?= .demo
help:
@printf '%s\n' \
'SciMesh coordinator commands:' \
' make up / make down Start or stop the normal coordinator stack.' \
' make demo-ui [WORKERS=3] Start isolated UI demo services and local workers.' \
' make demo-logs Follow coordinator logs for the UI demo.' \
' make demo-down Stop the demo services and workers.' \
' make demo-reset Stop the demo and wipe its data volumes.' \
' make test / make vet Run Go verification.' \
'' \
'Demo UI: http://localhost:18080/ui (login page; admin root@scimesh.local / rootpassword).'
demo-ui:
@DEMO_PROJECT="$(DEMO_PROJECT)" \
DEMO_POSTGRES_PORT="$(DEMO_POSTGRES_PORT)" \
DEMO_COORDINATOR_PORT="$(DEMO_COORDINATOR_PORT)" \
DEMO_UI_TOKEN="$(DEMO_UI_TOKEN)" \
DEMO_WORKER_TOKEN="$(DEMO_WORKER_TOKEN)" \
DEMO_WORKERS="$(WORKERS)" \
DEMO_DIR="$(DEMO_DIR)" \
./scripts/demo-ui.sh start
demo-down:
@DEMO_PROJECT="$(DEMO_PROJECT)" \
DEMO_POSTGRES_PORT="$(DEMO_POSTGRES_PORT)" \
DEMO_COORDINATOR_PORT="$(DEMO_COORDINATOR_PORT)" \
DEMO_UI_TOKEN="$(DEMO_UI_TOKEN)" \
DEMO_WORKER_TOKEN="$(DEMO_WORKER_TOKEN)" \
DEMO_DIR="$(DEMO_DIR)" \
./scripts/demo-ui.sh stop
demo-reset:
@DEMO_PROJECT="$(DEMO_PROJECT)" \
DEMO_POSTGRES_PORT="$(DEMO_POSTGRES_PORT)" \
DEMO_COORDINATOR_PORT="$(DEMO_COORDINATOR_PORT)" \
DEMO_UI_TOKEN="$(DEMO_UI_TOKEN)" \
DEMO_WORKER_TOKEN="$(DEMO_WORKER_TOKEN)" \
DEMO_DIR="$(DEMO_DIR)" \
./scripts/demo-ui.sh reset
demo-logs:
@DEMO_PROJECT="$(DEMO_PROJECT)" \
DEMO_POSTGRES_PORT="$(DEMO_POSTGRES_PORT)" \
DEMO_COORDINATOR_PORT="$(DEMO_COORDINATOR_PORT)" \
DEMO_UI_TOKEN="$(DEMO_UI_TOKEN)" \
DEMO_WORKER_TOKEN="$(DEMO_WORKER_TOKEN)" \
DEMO_DIR="$(DEMO_DIR)" \
./scripts/demo-ui.sh logs
# --- build / run ---------------------------------------------------------
build:
go build ./...
+39 -4
View File
@@ -76,10 +76,45 @@ UI_AUTH_TOKEN='local-ui-secret' make up
```
The UI is disabled by default and never accepts the worker bearer token.
It shows recent jobs, task/worker state, diagnostic shard artifacts, and the
final CSV for completed similarity-search jobs. The coordinator enters
`reducing` after the last shard completes, then exposes the final deterministic
global top-k result when merging succeeds.
The **control room** shows live workers, recent runs, shard state/attempts,
safe failures, coordinator artifacts, and the final CSV for completed
similarity-search jobs. The job page follows the real stages: TSV accepted →
shards execute → workers return CSVs → `reducing` → final deterministic global
top-k result. It polls only its own coordinator read-model and never controls
or exposes worker processes.
For a hands-on run, open `/ui`, choose **New similarity search**, select a
small ChEMBL-style TSV, then leave one or more `scimesh-worker` processes
running in separate terminals. The detail page updates every two seconds and
stops polling after a completed, failed, or cancelled job. Use **Preview CSV**
to inspect a bounded first page of a partial or completed final result before
downloading it. The UI never exposes source datasets or shard inputs; partial
CSVs remain available only as diagnostics.
### One-command manual demo
From the repository root, create the Python environment once, then start a
self-contained UI demo with two local reference workers:
```sh
python3 -m venv .venv
.venv/bin/pip install -e '.[dev]'
make demo-ui
```
This uses a separate Docker project and ports `18080` (coordinator) and
`55432` (PostgreSQL), so it does not conflict with the normal stack. Open
`http://localhost:18080/ui`, use username `operator` and password
`demo-ui-secret`, upload a small ChEMBL TSV, and observe the workers process
it. Change the worker count with `make demo-ui WORKERS=3`; stop all demo
services and workers with `make demo-down`.
The job page shows a live **Processing speed** graph in completed shards per
minute. It uses the coordinator snapshots observed by the open browser tab, so
it is a transparent local-session measurement rather than a persisted metric.
Use **Preview CSV** before downloading a partial diagnostic or completed final
result. Run `make help` from either the repository root or this directory for
the full list of demo commands.
`up` starts three services in order: Postgres waits until `pg_isready` passes, a
one-shot `migrate` container applies the schema and exits, and only then does the
+21 -9
View File
@@ -9,6 +9,7 @@ import (
"syscall"
"github.com/emil28092005/SciMesh/coordinator/internal/infra"
"github.com/emil28092005/SciMesh/coordinator/internal/metrics"
"github.com/emil28092005/SciMesh/coordinator/internal/storage/blob"
"github.com/emil28092005/SciMesh/coordinator/internal/storage/postgres"
httptransport "github.com/emil28092005/SciMesh/coordinator/internal/transport/http"
@@ -59,13 +60,14 @@ func run() error {
}
var (
clk = infra.NewClock()
tx = postgres.NewTxManager(pool)
taskRepo = postgres.NewTaskRepo(pool)
jobRepo = postgres.NewJobRepo(pool)
workerRepo = postgres.NewWorkerRepo(pool)
artifactRepo = postgres.NewArtifactRepo(pool)
uiReadRepo = postgres.NewUIReadRepo(pool)
clk = infra.NewClock()
tx = postgres.NewTxManager(pool)
taskRepo = postgres.NewTaskRepo(pool)
jobRepo = postgres.NewJobRepo(pool)
workerRepo = postgres.NewWorkerRepo(pool)
artifactRepo = postgres.NewArtifactRepo(pool)
uiReadRepo = postgres.NewUIReadRepo(pool)
taskResultRepo = postgres.NewTaskResultRepo(pool)
)
useCases := httptransport.UseCases{
@@ -74,7 +76,7 @@ func run() error {
SubmitDataset: usecase.NewSubmitDataset(blobStore, artifactRepo, jobRepo, taskRepo, tx, clk, cfg.DefaultMaxAttempts),
ClaimTask: usecase.NewClaimTask(taskRepo, jobRepo, workerRepo, tx, clk, cfg.LeaseDuration),
RenewLease: usecase.NewRenewLease(taskRepo, workerRepo, tx, clk, cfg.LeaseDuration),
CompleteTask: usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, tx, clk),
CompleteTask: usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, workerRepo, taskResultRepo, tx, clk, cfg.QuorumSize),
ReduceJob: usecase.NewReduceJob(jobRepo, taskRepo, artifactRepo, blobStore, tx, clk),
FailTask: usecase.NewFailTask(taskRepo, jobRepo, tx, clk),
GetJobStatus: usecase.NewGetJobStatus(jobRepo, taskRepo),
@@ -84,6 +86,7 @@ func run() error {
GetJobResult: usecase.NewGetJobResult(jobRepo, usecase.NewDownloadArtifact(artifactRepo, blobStore)),
GetTaskInput: usecase.NewGetTaskInput(taskRepo, artifactRepo, blobStore),
Dashboard: usecase.NewDashboard(uiReadRepo),
PreviewArtifact: usecase.NewPreviewArtifact(uiReadRepo, blobStore),
}
// Background reapers are tracked so shutdown can wait for them. Without this
@@ -107,9 +110,18 @@ func run() error {
}(r.name, r.fn)
}
// Business metrics: gauges of tasks/jobs/workers by status, sampled from the
// database on every Prometheus scrape.
statsRepo := postgres.NewStatsRepo(pool)
m := metrics.New()
m.RegisterBusiness(func(ctx context.Context) (metrics.Stats, error) {
tasks, jobs, workers, err := statsRepo.Counts(ctx)
return metrics.Stats{Tasks: tasks, Jobs: jobs, Workers: workers}, err
})
// pool.Ping backs /health: readiness means the database answers, not just
// that the process is alive.
api := httptransport.NewServer(useCases, log, cfg.RequestTimeout, cfg.HeartbeatInterval, cfg.MaxUploadBytes, pool.Ping)
api := httptransport.NewServer(useCases, log, cfg.RequestTimeout, cfg.HeartbeatInterval, cfg.MaxUploadBytes, cfg.JWTSecret, cfg.UserserviceURL, m, pool.Ping, cfg.PublicCoordinatorURL, cfg.PublicUserserviceURL)
err = infra.RunServer(ctx, log, cfg.Addr, api.Handler(cfg.Token, cfg.UIToken))
// Shutdown order matters, and defers alone cannot express it (they run
+30
View File
@@ -0,0 +1,30 @@
# Demo overlay: Prometheus scrapes the coordinator's /metrics, Grafana shows the
# provisioned SciMesh dashboard. Merged by scripts/demo-ui.sh with a third -f.
# Both share the coordinator's compose network, so Prometheus reaches it by name.
services:
prometheus:
image: prom/prometheus:v2.54.1
volumes:
- ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml:ro
ports:
- "${PROMETHEUS_PORT:-19090}:9090"
restart: unless-stopped
grafana:
image: grafana/grafana:11.2.0
depends_on:
- prometheus
environment:
GF_SECURITY_ADMIN_USER: admin
GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_PASSWORD:-admin}
# Anonymous viewing so the demo dashboard opens without a login.
GF_AUTH_ANONYMOUS_ENABLED: "true"
GF_AUTH_ANONYMOUS_ORG_ROLE: Viewer
GF_USERS_DEFAULT_THEME: dark
volumes:
- ./monitoring/grafana/provisioning:/etc/grafana/provisioning:ro
- ./monitoring/grafana/dashboards:/var/lib/grafana/dashboards:ro
ports:
- "${GRAFANA_PORT:-13000}:3000"
restart: unless-stopped
+71
View File
@@ -0,0 +1,71 @@
# Demo overlay: adds the userservice (its own Postgres + migrations) alongside
# the coordinator and wires the two together with a shared JWT secret, so the
# operator UI authenticates through userservice login/registration.
#
# Used only by scripts/demo-ui.sh, merged onto docker-compose.yml with a second
# -f. Not part of the plain `make up` stack.
services:
postgres-users:
image: postgres:16-alpine
environment:
POSTGRES_USER: ${POSTGRES_USER:-scimesh}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-scimesh}
POSTGRES_DB: scimesh_users
healthcheck:
test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-scimesh} -d scimesh_users"]
interval: 5s
timeout: 3s
retries: 10
start_period: 5s
migrate-users:
image: migrate/migrate:v4.17.1
depends_on:
postgres-users:
condition: service_healthy
volumes:
- ../users/migrations:/migrations:ro
command:
- -path=/migrations
- -database=postgres://${POSTGRES_USER:-scimesh}:${POSTGRES_PASSWORD:-scimesh}@postgres-users:5432/scimesh_users?sslmode=disable
- up
restart: on-failure
userservice:
build:
context: ../users
depends_on:
postgres-users:
condition: service_healthy
migrate-users:
condition: service_completed_successfully
environment:
USERSERVICE_ADDR: ":8081"
DATABASE_URL: postgres://${POSTGRES_USER:-scimesh}:${POSTGRES_PASSWORD:-scimesh}@postgres-users:5432/scimesh_users?sslmode=disable
JWT_SECRET: ${JWT_SECRET}
# Seeds the first admin the very first time it boots (idempotent after).
BOOTSTRAP_ADMIN_EMAIL: ${BOOTSTRAP_ADMIN_EMAIL:-root@scimesh.local}
BOOTSTRAP_ADMIN_PASSWORD: ${BOOTSTRAP_ADMIN_PASSWORD}
LOG_LEVEL: ${LOG_LEVEL:-info}
ports:
- "${USERSERVICE_PORT:-18081}:8081"
healthcheck:
test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8081/health"]
interval: 10s
timeout: 3s
retries: 3
start_period: 5s
restart: unless-stopped
# Turn the coordinator UI into session mode: the same shared secret verifies
# userservice tokens locally, and USERSERVICE_URL is where login/register proxy.
coordinator:
environment:
JWT_SECRET: ${JWT_SECRET}
USERSERVICE_URL: http://userservice:8081
# Browser/host-facing URLs for the "add your machine" command. A user's
# worker runs on the host, so it reaches the published ports on localhost,
# not the in-cluster service names.
PUBLIC_COORDINATOR_URL: http://localhost:${COORDINATOR_PORT:-8080}
PUBLIC_USERSERVICE_URL: http://localhost:${USERSERVICE_PORT:-8081}
+13 -3
View File
@@ -1,23 +1,33 @@
module github.com/emil28092005/SciMesh/coordinator
go 1.22
go 1.25.0
require (
github.com/Masterminds/squirrel v1.5.4
github.com/cenkalti/backoff/v4 v4.3.0
github.com/golang-jwt/jwt/v5 v5.3.1
github.com/google/uuid v1.6.0
github.com/jackc/pgx/v5 v5.6.0
github.com/joho/godotenv v1.5.1
github.com/prometheus/client_golang v1.19.1
gopkg.in/natefinch/lumberjack.v2 v2.2.1
)
require (
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
github.com/jackc/puddle/v2 v2.2.1 // indirect
github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect
github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/prometheus/client_model v0.6.2 // indirect
github.com/prometheus/common v0.55.0 // indirect
github.com/prometheus/procfs v0.21.1 // indirect
golang.org/x/crypto v0.17.0 // indirect
golang.org/x/sync v0.1.0 // indirect
golang.org/x/text v0.14.0 // indirect
golang.org/x/sync v0.22.0 // indirect
golang.org/x/sys v0.47.0 // indirect
golang.org/x/text v0.40.0 // indirect
google.golang.org/protobuf v1.36.11 // indirect
)
+28 -6
View File
@@ -1,10 +1,18 @@
github.com/Masterminds/squirrel v1.5.4 h1:uUcX/aBc8O7Fg9kaISIUsHXdKuqehiXAMQTYX8afzqM=
github.com/Masterminds/squirrel v1.5.4/go.mod h1:NNaOrjSoIDfDA40n7sr2tPNZRfjzjA400rg+riTZj10=
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY=
github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
@@ -21,20 +29,34 @@ github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 h1:SOEGU9fKiNWd/HOJuq
github.com/lann/builder v0.0.0-20180802200727-47ae307949d0/go.mod h1:dXGbAdH5GtBTC4WfIxhKZfyBF/HBFgRZSWwZ9g/He9o=
github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 h1:P6pPBnrTSX3DEVR4fDembhRWSsG5rVo6hYhAB/ADZrk=
github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0/go.mod h1:vmVJ0l/dxyfGW6FmdpVm2joNMFikkuWg0EoCKLGUMNw=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prometheus/client_golang v1.19.1 h1:wZWJDwK+NameRJuPGDhlnFgx8e8HN3XHQeLaYJFJBOE=
github.com/prometheus/client_golang v1.19.1/go.mod h1:mP78NwGzrVks5S2H6ab8+ZZGJLZUq1hoULYBAYBw1Ho=
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G1dc=
github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8=
github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc=
gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc=
+43
View File
@@ -0,0 +1,43 @@
// Package authctx carries the authenticated requester across the transport and
// use-case layers without either one importing the other. The HTTP middleware
// stamps a Requester after verifying a user's JWT; the job use cases read it to
// record ownership and to enforce that a non-admin only touches their own jobs.
package authctx
import (
"context"
"github.com/google/uuid"
)
// Requester is the identity behind a request, derived from a verified JWT.
// A request authenticated only by the shared worker/service token carries no
// Requester at all (From returns ok=false), which is how worker traffic and
// legacy unauthenticated-user traffic stay owner-less.
type Requester struct {
UserID uuid.UUID
Role string
Verified bool
}
// IsAdmin reports whether the requester may act on any user's jobs.
func (r Requester) IsAdmin() bool { return r.Role == "admin" }
// IsTrusted reports whether workers this requester registers produce results
// the coordinator accepts without quorum. Admins and verified contributors are
// trusted; a plain unverified user is not.
func (r Requester) IsTrusted() bool { return r.IsAdmin() || r.Verified }
type ctxKey struct{}
// With returns a copy of ctx carrying r.
func With(ctx context.Context, r Requester) context.Context {
return context.WithValue(ctx, ctxKey{}, r)
}
// From returns the requester stamped by the middleware, or ok=false when the
// request was not authenticated as a user.
func From(ctx context.Context) (Requester, bool) {
r, ok := ctx.Value(ctxKey{}).(Requester)
return r, ok
}
+9 -1
View File
@@ -19,7 +19,11 @@ const (
// Job is one user submission that fans out into one or more tasks.
type Job struct {
ID uuid.UUID
ID uuid.UUID
// OwnerID is the userservice user who submitted the job (JWT `sub`). nil
// when the job was created without user authentication. Not a foreign key:
// users live in a separate service/database.
OwnerID *uuid.UUID
Workload string
InputURI string // external input URI; empty for uploaded datasets
InputArtifactID *uuid.UUID // uploaded input artifact; nil for URI submissions
@@ -119,6 +123,10 @@ func (p JobProgress) DeriveStatus() JobStatus {
switch {
case p.Job.Status == JobCancelled:
return JobCancelled
case p.Job.Status == JobFailed:
// A reducer may fail after every shard has completed. That terminal
// failure must not be overwritten by an otherwise-complete task count.
return JobFailed
case p.Job.Status == JobReducing:
return JobReducing
case p.Total == 0:
+1
View File
@@ -82,6 +82,7 @@ func TestDeriveStatus(t *testing.T) {
{"done and failed", JobProgress{Total: 3, Done: 2, Failed: 1}, JobFailed},
{"failed but work remains", JobProgress{Total: 3, Pending: 1, Failed: 2}, JobRunning},
{"cancelled job wins over task histogram", JobProgress{Job: Job{Status: JobCancelled}, Total: 3, Done: 1, Cancelled: 2}, JobCancelled},
{"persisted reducer failure wins over completed tasks", JobProgress{Job: Job{Status: JobFailed}, Total: 3, Done: 3}, JobFailed},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
+31
View File
@@ -24,6 +24,10 @@ const (
// ErrCodeLeaseExpired marks tasks failed by the reaper rather than by a worker.
const ErrCodeLeaseExpired = "lease_expired"
// ErrCodeQuorumFailed marks a task whose untrusted results never reached a
// verifying quorum before its attempts ran out.
const ErrCodeQuorumFailed = "quorum_failed"
// Task is one independently executable chunk of a job.
//
// Nullable columns are pointers so "no lease" stays distinguishable from
@@ -216,6 +220,33 @@ func (t *Task) CompleteWith(resultArtifactID uuid.UUID, metrics map[string]any,
return nil
}
// ReleaseAfterVote returns an untrusted worker's task to the queue after its
// result was recorded as a quorum vote but quorum was not yet reached, so a
// different owner can compute it independently. When no attempts remain the task
// fails: its untrusted results could not be verified.
func (t *Task) ReleaseAfterVote(worker string, attempt int, now time.Time) error {
if t.Status == TaskCompleted {
return nil // settled by a concurrent quorum
}
if err := t.verifyLease(worker, attempt, now); err != nil {
return err
}
t.LeaseOwner = nil
t.LeaseExpiresAt = nil
t.Version++
if t.CanRetry() {
t.Status = TaskPending
return nil
}
code, msg := ErrCodeQuorumFailed, "untrusted results did not reach quorum"
t.ErrorCode = &code
t.ErrorMessage = &msg
t.Status = TaskFailed
t.CompletedAt = &now
return nil
}
// Fail records a worker-reported failure. A retryable failure with attempts
// left returns the task to the queue; otherwise it terminates as failed.
func (t *Task) Fail(worker string, attempt int, code, message string, retryable bool, now time.Time) error {
+24 -4
View File
@@ -14,14 +14,30 @@ const (
WorkerOffline WorkerStatus = "offline"
)
// WorkerTrust says whether a worker's results are accepted directly or must
// clear quorum cross-checking.
type WorkerTrust string
const (
// WorkerTrusted — lab machine (shared token) or a verified/admin contributor.
WorkerTrusted WorkerTrust = "trusted"
// WorkerUntrusted — a plain enthusiast; results are quarantined until quorum.
WorkerUntrusted WorkerTrust = "untrusted"
)
// Worker is a registered process/machine allowed to claim tasks. Its
// capabilities are the allowlisted workload names it can run; the coordinator
// never hands it a task outside that set.
type Worker struct {
ID uuid.UUID
Name string
Capabilities []string
Status WorkerStatus
ID uuid.UUID
Name string
Capabilities []string
Status WorkerStatus
// OwnerID is the userservice user who registered this worker; nil for a
// worker registered with the shared service token.
OwnerID *uuid.UUID
// TrustLevel decides whether this worker's results need quorum.
TrustLevel WorkerTrust
LastHeartbeatAt time.Time
CreatedAt time.Time
UpdatedAt time.Time
@@ -29,6 +45,9 @@ type Worker struct {
// NewWorker registers a worker. A worker with no capabilities could never be
// handed a task, so an empty set is rejected rather than silently stored.
//
// Trust defaults to WorkerTrusted (the shared-token lab worker); the caller
// overrides it for a volunteer registered through the userservice.
func NewWorker(name string, capabilities []string, now time.Time) (*Worker, error) {
if len(capabilities) == 0 {
return nil, ErrInvalidInput
@@ -38,6 +57,7 @@ func NewWorker(name string, capabilities []string, now time.Time) (*Worker, erro
Name: name,
Capabilities: capabilities,
Status: WorkerOnline,
TrustLevel: WorkerTrusted,
LastHeartbeatAt: now,
CreatedAt: now,
UpdatedAt: now,
+51 -14
View File
@@ -26,6 +26,24 @@ type Config struct {
Token string
// Local operator UI credential. Empty disables the embedded UI entirely.
UIToken string
// Shared HS256 secret used to verify userservice-issued JWTs. When set, a
// submitter may authenticate with a JWT (in addition to workers using the
// shared token) and their jobs are stamped with owner_id. Empty disables
// user-JWT auth entirely — the pre-userservice behaviour. Must match the
// userservice's JWT_SECRET.
JWTSecret string
// Base URL of the userservice, e.g. http://userservice:8081. When set
// together with JWTSecret, the operator UI authenticates via userservice
// login/registration (cookie session) instead of the static UI_AUTH_TOKEN
// basic auth. Empty keeps the basic-auth UI.
UserserviceURL string
// Browser-facing base URLs used to render the "add your machine" command on
// the UI. They must be reachable from a user's own machine, which is not
// necessarily the in-cluster address the coordinator uses for UserserviceURL.
// PublicCoordinatorURL empty lets the page fall back to its own origin;
// PublicUserserviceURL empty falls back to UserserviceURL.
PublicCoordinatorURL string
PublicUserserviceURL string
// Minimum log level: debug, info, warn, error.
LogLevel string
@@ -50,6 +68,9 @@ type Config struct {
LeaseDuration time.Duration
// Default attempt ceiling for newly created tasks.
DefaultMaxAttempts int
// How many distinct owners must agree on an untrusted result before it is
// accepted (trusted workers are accepted directly).
QuorumSize int
// How often the background lease-reaper runs.
ReaperInterval time.Duration
// A worker silent for longer than this is marked offline by the reaper.
@@ -78,20 +99,25 @@ func LoadConfig() (Config, error) {
DatabaseURL: os.Getenv("DATABASE_URL"),
// COORDINATOR_TOKEN is the contract name; WORKER_AUTH_TOKEN is the
// former name, still honoured so existing .env files keep working.
Token: getEnv("COORDINATOR_TOKEN", os.Getenv("WORKER_AUTH_TOKEN")),
UIToken: os.Getenv("UI_AUTH_TOKEN"),
LogLevel: getEnv("LOG_LEVEL", "info"),
LogFile: os.Getenv("LOG_FILE"),
StorageDir: getEnv("COORDINATOR_STORAGE_DIR", "./data"),
MaxUploadBytes: 1 << 30, // 1 GiB
DBMaxConns: 10,
DBConnectTimeout: 30 * time.Second,
RequestTimeout: 15 * time.Second,
HeartbeatInterval: 15 * time.Second,
LeaseDuration: 2 * time.Minute,
DefaultMaxAttempts: 3,
ReaperInterval: 30 * time.Second,
WorkerOfflineAfter: 1 * time.Minute,
Token: getEnv("COORDINATOR_TOKEN", os.Getenv("WORKER_AUTH_TOKEN")),
UIToken: os.Getenv("UI_AUTH_TOKEN"),
JWTSecret: os.Getenv("JWT_SECRET"),
UserserviceURL: os.Getenv("USERSERVICE_URL"),
PublicCoordinatorURL: os.Getenv("PUBLIC_COORDINATOR_URL"),
PublicUserserviceURL: getEnv("PUBLIC_USERSERVICE_URL", os.Getenv("USERSERVICE_URL")),
LogLevel: getEnv("LOG_LEVEL", "info"),
LogFile: os.Getenv("LOG_FILE"),
StorageDir: getEnv("COORDINATOR_STORAGE_DIR", "./data"),
MaxUploadBytes: 1 << 30, // 1 GiB
DBMaxConns: 10,
DBConnectTimeout: 30 * time.Second,
RequestTimeout: 15 * time.Second,
HeartbeatInterval: 15 * time.Second,
LeaseDuration: 2 * time.Minute,
DefaultMaxAttempts: 3,
QuorumSize: 2,
ReaperInterval: 30 * time.Second,
WorkerOfflineAfter: 1 * time.Minute,
}
if cfg.DatabaseURL == "" {
@@ -100,6 +126,11 @@ func LoadConfig() (Config, error) {
if cfg.UIToken != "" && cfg.Token != "" && cfg.UIToken == cfg.Token {
return Config{}, fmt.Errorf("UI_AUTH_TOKEN must differ from the worker auth token")
}
// A short secret makes the HMAC brute-forceable; refuse a weak one rather
// than verify tokens against it.
if cfg.JWTSecret != "" && len(cfg.JWTSecret) < 32 {
return Config{}, fmt.Errorf("JWT_SECRET must be at least 32 bytes")
}
var err error
if cfg.DBMaxConns, err = getEnvInt32("DB_MAX_CONNS", cfg.DBMaxConns); err != nil {
@@ -129,6 +160,12 @@ func LoadConfig() (Config, error) {
if cfg.DefaultMaxAttempts, err = getEnvInt("DEFAULT_MAX_ATTEMPTS", cfg.DefaultMaxAttempts); err != nil {
return Config{}, err
}
if cfg.QuorumSize, err = getEnvInt("QUORUM_SIZE", cfg.QuorumSize); err != nil {
return Config{}, err
}
if cfg.QuorumSize < 1 {
return Config{}, fmt.Errorf("QUORUM_SIZE must be positive")
}
if cfg.DefaultMaxAttempts < 1 {
return Config{}, fmt.Errorf("DEFAULT_MAX_ATTEMPTS must be positive")
}
+33
View File
@@ -417,3 +417,36 @@ func contains(ss []string, s string) bool {
}
return false
}
// TaskResultRepo is an in-memory usecase.TaskResultRepository: one vote per
// (task, owner).
type TaskResultRepo struct {
mu sync.Mutex
votes map[uuid.UUID]map[uuid.UUID]string // taskID -> ownerID -> sha256
}
func NewTaskResultRepo() *TaskResultRepo {
return &TaskResultRepo{votes: make(map[uuid.UUID]map[uuid.UUID]string)}
}
func (r *TaskResultRepo) RecordVote(_ context.Context, taskID, ownerID uuid.UUID, sha256 string, _ uuid.UUID) error {
r.mu.Lock()
defer r.mu.Unlock()
if r.votes[taskID] == nil {
r.votes[taskID] = make(map[uuid.UUID]string)
}
r.votes[taskID][ownerID] = sha256
return nil
}
func (r *TaskResultRepo) CountAgreeing(_ context.Context, taskID uuid.UUID, sha256 string) (int, error) {
r.mu.Lock()
defer r.mu.Unlock()
n := 0
for _, s := range r.votes[taskID] {
if s == sha256 {
n++
}
}
return n, nil
}
+36 -5
View File
@@ -27,7 +27,7 @@ var _ usecase.UIReadRepository = (*UIReadRepo)(nil)
func (r *UIReadRepo) GetJob(ctx context.Context, id uuid.UUID) (*domain.Job, error) {
return r.jobs.Get(ctx, id)
}
func (r *UIReadRepo) ListJobs(_ context.Context, limit int) ([]domain.Job, error) {
func (r *UIReadRepo) ListJobs(_ context.Context, owner *uuid.UUID, limit int) ([]domain.Job, error) {
if limit < 1 || limit > 100 {
return nil, domain.ErrInvalidInput
}
@@ -35,6 +35,9 @@ func (r *UIReadRepo) ListJobs(_ context.Context, limit int) ([]domain.Job, error
defer r.jobs.mu.Unlock()
out := make([]domain.Job, 0, len(r.jobs.jobs))
for _, job := range r.jobs.jobs {
if owner != nil && (job.OwnerID == nil || *job.OwnerID != *owner) {
continue
}
out = append(out, *job)
}
sort.Slice(out, func(i, j int) bool {
@@ -84,16 +87,44 @@ func (r *UIReadRepo) ListWorkers(_ context.Context, limit int) ([]domain.Worker,
copy.Capabilities = append([]string(nil), worker.Capabilities...)
out = append(out, copy)
}
sortWorkers(out)
if len(out) > limit {
out = out[:limit]
}
return out, nil
}
func (r *UIReadRepo) ListWorkersByOwner(_ context.Context, owner uuid.UUID, limit int) ([]domain.Worker, error) {
if limit < 1 || limit > 100 {
return nil, domain.ErrInvalidInput
}
r.workers.mu.Lock()
defer r.workers.mu.Unlock()
out := []domain.Worker{}
for _, worker := range r.workers.workers {
if worker.OwnerID == nil || *worker.OwnerID != owner {
continue
}
copy := *worker
copy.Capabilities = append([]string(nil), worker.Capabilities...)
out = append(out, copy)
}
sortWorkers(out)
if len(out) > limit {
out = out[:limit]
}
return out, nil
}
// sortWorkers orders workers most-recently-seen first, breaking ties on id so
// the order is deterministic across calls.
func sortWorkers(out []domain.Worker) {
sort.Slice(out, func(i, j int) bool {
if out[i].LastHeartbeatAt.Equal(out[j].LastHeartbeatAt) {
return out[i].ID.String() > out[j].ID.String()
}
return out[i].LastHeartbeatAt.After(out[j].LastHeartbeatAt)
})
if len(out) > limit {
out = out[:limit]
}
return out, nil
}
func (r *UIReadRepo) ListArtifactsByJob(_ context.Context, jobID uuid.UUID) ([]domain.Artifact, error) {
r.artifacts.mu.Lock()
+66
View File
@@ -0,0 +1,66 @@
package metrics
import (
"context"
"time"
"github.com/prometheus/client_golang/prometheus"
)
// Stats is a point-in-time snapshot of the coordinator's domain state: counts of
// tasks, jobs, and workers keyed by their status. Maps are expected to be
// zero-filled by the provider so every known status is always present, giving
// the dashboard flat zero lines instead of gaps.
type Stats struct {
Tasks map[string]int
Jobs map[string]int
Workers map[string]int
}
// StatsFunc returns the current snapshot. It is called on every scrape, so it
// must be a cheap aggregate query.
type StatsFunc func(context.Context) (Stats, error)
// RegisterBusiness registers a collector that reports domain-state gauges
// (scimesh_tasks/jobs/workers by status) sourced from collect on each scrape.
// Deriving the gauges at scrape time keeps them fresh without a background
// goroutine, and a failed query simply yields no samples for that scrape.
func (m *Metrics) RegisterBusiness(collect StatsFunc) {
m.reg.MustRegister(&businessCollector{
collect: collect,
tasks: prometheus.NewDesc("scimesh_tasks", "Tasks by status.", []string{"status"}, nil),
jobs: prometheus.NewDesc("scimesh_jobs", "Jobs by status.", []string{"status"}, nil),
workers: prometheus.NewDesc("scimesh_workers", "Workers by status.", []string{"status"}, nil),
})
}
type businessCollector struct {
collect StatsFunc
tasks, jobs, workers *prometheus.Desc
}
func (c *businessCollector) Describe(ch chan<- *prometheus.Desc) {
ch <- c.tasks
ch <- c.jobs
ch <- c.workers
}
func (c *businessCollector) Collect(ch chan<- prometheus.Metric) {
// A bounded query so one slow scrape cannot stall Prometheus.
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
s, err := c.collect(ctx)
if err != nil {
return // no samples this scrape; Prometheus keeps the last value
}
emit(ch, c.tasks, s.Tasks)
emit(ch, c.jobs, s.Jobs)
emit(ch, c.workers, s.Workers)
}
func emit(ch chan<- prometheus.Metric, desc *prometheus.Desc, counts map[string]int) {
for status, n := range counts {
ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, float64(n), status)
}
}
@@ -0,0 +1,51 @@
package metrics
import (
"context"
"errors"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
func scrape(t *testing.T, m *Metrics) string {
t.Helper()
rec := httptest.NewRecorder()
req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, "/metrics", nil)
m.Handler().ServeHTTP(rec, req)
return rec.Body.String()
}
func TestBusinessCollectorEmitsGauges(t *testing.T) {
m := New()
m.RegisterBusiness(func(context.Context) (Stats, error) {
return Stats{
Tasks: map[string]int{"pending": 3, "running": 1, "completed": 0},
Jobs: map[string]int{"running": 2},
Workers: map[string]int{"online": 4},
}, nil
})
body := scrape(t, m)
for _, want := range []string{
`scimesh_tasks{status="pending"} 3`,
`scimesh_tasks{status="completed"} 0`,
`scimesh_jobs{status="running"} 2`,
`scimesh_workers{status="online"} 4`,
} {
if !strings.Contains(body, want) {
t.Errorf("metrics missing %q\n%s", want, body)
}
}
}
func TestBusinessCollectorSkipsOnError(t *testing.T) {
m := New()
m.RegisterBusiness(func(context.Context) (Stats, error) {
return Stats{}, errors.New("db down")
})
if strings.Contains(scrape(t, m), "scimesh_tasks") {
t.Error("a failed snapshot must emit no business samples")
}
}
+112
View File
@@ -0,0 +1,112 @@
// Package metrics exposes Prometheus instrumentation for the coordinator: an
// HTTP RED middleware (rate, errors, duration) plus the standard Go runtime and
// process collectors, all on a private registry so nothing leaks in from global
// state.
package metrics
import (
"net/http"
"regexp"
"strconv"
"strings"
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/collectors"
"github.com/prometheus/client_golang/prometheus/promhttp"
)
type Metrics struct {
reg *prometheus.Registry
requests *prometheus.CounterVec
duration *prometheus.HistogramVec
}
// New builds the registry and registers the runtime, process, and HTTP metrics.
func New() *Metrics {
reg := prometheus.NewRegistry()
reg.MustRegister(
collectors.NewGoCollector(),
collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}),
)
requests := prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: "scimesh",
Subsystem: "http",
Name: "requests_total",
Help: "HTTP requests, labelled by method, normalized route, and status.",
}, []string{"method", "route", "status"})
duration := prometheus.NewHistogramVec(prometheus.HistogramOpts{
Namespace: "scimesh",
Subsystem: "http",
Name: "request_duration_seconds",
Help: "HTTP request duration in seconds.",
Buckets: prometheus.DefBuckets,
}, []string{"method", "route"})
reg.MustRegister(requests, duration)
return &Metrics{reg: reg, requests: requests, duration: duration}
}
// Handler serves the metrics in Prometheus text format.
func (m *Metrics) Handler() http.Handler {
return promhttp.HandlerFor(m.reg, promhttp.HandlerOpts{})
}
// Registry exposes the registry so callers can register extra collectors.
func (m *Metrics) Registry() *prometheus.Registry { return m.reg }
// Middleware records one request into the RED metrics. It normalizes the path
// so per-id routes collapse to a single low-cardinality label.
func (m *Metrics) Middleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
start := time.Now()
rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
next.ServeHTTP(rec, r)
route := normalizeRoute(r.URL.Path)
m.requests.WithLabelValues(r.Method, route, strconv.Itoa(rec.status)).Inc()
m.duration.WithLabelValues(r.Method, route).Observe(time.Since(start).Seconds())
})
}
type statusRecorder struct {
http.ResponseWriter
status int
}
func (s *statusRecorder) WriteHeader(code int) {
s.status = code
s.ResponseWriter.WriteHeader(code)
}
var uuidRe = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$`)
// normalizeRoute collapses uuid and numeric path segments to {id}, keeping the
// route label cardinality bounded (otherwise every job/task id would be its own
// time series).
func normalizeRoute(path string) string {
if path == "" {
return "/"
}
segs := strings.Split(path, "/")
for i, s := range segs {
if s == "" {
continue
}
if uuidRe.MatchString(s) || isAllDigits(s) {
segs[i] = "{id}"
}
}
return strings.Join(segs, "/")
}
func isAllDigits(s string) bool {
for _, r := range s {
if r < '0' || r > '9' {
return false
}
}
return s != ""
}
@@ -0,0 +1,47 @@
package metrics
import (
"context"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
func TestNormalizeRoute(t *testing.T) {
cases := map[string]string{
"/health": "/health",
"/jobs/3f2504e0-4f89-41d3-9a0c-0305e82c3301": "/jobs/{id}",
"/tasks/3f2504e0-4f89-41d3-9a0c-0305e82c3301/result": "/tasks/{id}/result",
"/ui/jobs/12345": "/ui/jobs/{id}",
"/": "/",
}
for in, want := range cases {
if got := normalizeRoute(in); got != want {
t.Errorf("normalizeRoute(%q) = %q, want %q", in, got, want)
}
}
}
func TestMiddlewareAndHandler(t *testing.T) {
m := New()
h := m.Middleware(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusCreated)
}))
req, _ := http.NewRequestWithContext(context.Background(), http.MethodPost, "/jobs/3f2504e0-4f89-41d3-9a0c-0305e82c3301", nil)
h.ServeHTTP(httptest.NewRecorder(), req)
// Scrape and confirm the request was recorded under the normalized route.
rec := httptest.NewRecorder()
greq, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, "/metrics", nil)
m.Handler().ServeHTTP(rec, greq)
body := rec.Body.String()
if !strings.Contains(body, `scimesh_http_requests_total{method="POST",route="/jobs/{id}",status="201"}`) {
t.Errorf("requests_total not recorded as expected; body:\n%s", body)
}
if !strings.Contains(body, "go_goroutines") {
t.Error("Go runtime collector not registered")
}
}
@@ -133,6 +133,33 @@ func TestClaimReductionIsAtomic(t *testing.T) {
}
}
func TestUIReadRepoListsReducerFields(t *testing.T) {
pool := testPool(t)
job, _ := seedJob(t, pool, 1)
jobs := NewJobRepo(pool)
ctx := context.Background()
if err := jobs.UpdateStatus(ctx, job.ID, domain.JobReducing, nil); err != nil {
t.Fatal(err)
}
if claimed, err := jobs.ClaimReduction(ctx, job.ID, time.Now().UTC()); err != nil || !claimed {
t.Fatalf("claim reduction = (%v, %v)", claimed, err)
}
listed, err := NewUIReadRepo(pool).ListJobs(ctx, nil, 20)
if err != nil {
t.Fatalf("list UI jobs: %v", err)
}
for _, item := range listed {
if item.ID != job.ID {
continue
}
if item.Status != domain.JobReducing || item.ReducerStartedAt == nil {
t.Fatalf("UI reducer projection = %+v", item)
}
return
}
t.Fatalf("seeded job %s is missing from UI list", job.ID)
}
// A job must land whole or not at all: a half-created job leaves chunks no
// worker could ever complete.
func TestCreateJobRollsBackOnFailure(t *testing.T) {
@@ -379,8 +406,9 @@ func TestCompleteTaskReplayIsIdempotent(t *testing.T) {
job, _ := seedJob(t, pool, 1)
tasks, jobs, artifacts, tx := NewTaskRepo(pool), NewJobRepo(pool), NewArtifactRepo(pool), NewTxManager(pool)
workers, results := NewWorkerRepo(pool), NewTaskResultRepo(pool)
clk := fixedClock{now: time.Now().UTC()}
uc := usecase.NewCompleteTask(tasks, jobs, artifacts, tx, clk)
uc := usecase.NewCompleteTask(tasks, jobs, artifacts, workers, results, tx, clk, 2)
claimed, err := tasks.ClaimNext(ctx, usecase.ClaimFilter{
Owner: "worker-1", Now: clk.now, LeaseUntil: clk.now.Add(time.Minute),
@@ -28,14 +28,15 @@ var _ usecase.JobRepository = (*JobRepo)(nil)
var jobColumns = []string{
"id", "workload", "input_uri", "parameters", "status", "created_at", "completed_at",
"input_artifact_id", "result_artifact_id", "error_code", "error_message", "reducer_started_at",
"owner_id",
}
// Insert runs inside the caller's transaction, alongside the job's tasks — that
// is what makes "all tasks or none" hold.
func (r *JobRepo) Insert(ctx context.Context, j *domain.Job) error {
sql, args, err := psql.Insert("jobs").
Columns("id", "workload", "input_uri", "parameters", "status", "created_at").
Values(j.ID, j.Workload, j.InputURI, jsonbOrEmpty(j.Parameters), string(j.Status), j.CreatedAt).
Columns("id", "workload", "input_uri", "parameters", "status", "created_at", "owner_id").
Values(j.ID, j.Workload, j.InputURI, jsonbOrEmpty(j.Parameters), string(j.Status), j.CreatedAt, j.OwnerID).
ToSql()
if err != nil {
return err
@@ -59,7 +60,8 @@ func (r *JobRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Job, error) {
)
err = conn(ctx, r.pool).QueryRow(ctx, sql, args...).Scan(
&j.ID, &j.Workload, &j.InputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt,
&j.InputArtifactID, &j.ResultArtifactID, &j.ErrorCode, &j.ErrorMessage, &j.ReducerStartedAt)
&j.InputArtifactID, &j.ResultArtifactID, &j.ErrorCode, &j.ErrorMessage, &j.ReducerStartedAt,
&j.OwnerID)
if errors.Is(err, pgx.ErrNoRows) {
return nil, domain.ErrJobNotFound
}
@@ -0,0 +1,65 @@
package postgres
import (
"context"
"fmt"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
)
// Known statuses per entity, so counts are zero-filled and every status is
// always present in the metrics (a flat 0 line beats a gap on the dashboard).
var (
taskStatuses = []string{string(domain.TaskPending), string(domain.TaskLeased), string(domain.TaskRunning), string(domain.TaskCompleted), string(domain.TaskFailed), string(domain.TaskCancelled)}
jobStatuses = []string{string(domain.JobPending), string(domain.JobRunning), string(domain.JobReducing), string(domain.JobCompleted), string(domain.JobFailed), string(domain.JobCancelled)}
workerStatuses = []string{string(domain.WorkerOnline), string(domain.WorkerBusy), string(domain.WorkerOffline)}
)
// StatsRepo answers the aggregate status counts the business metrics report. It
// runs one cheap GROUP BY per entity; the collector calls this on every scrape.
type StatsRepo struct {
pool *pgxpool.Pool
}
func NewStatsRepo(pool *pgxpool.Pool) *StatsRepo {
return &StatsRepo{pool: pool}
}
// Counts returns status->count maps for tasks, jobs, and workers, each
// zero-filled across its known statuses.
func (r *StatsRepo) Counts(ctx context.Context) (tasks, jobs, workers map[string]int, err error) {
if tasks, err = r.countByStatus(ctx, "tasks", taskStatuses); err != nil {
return nil, nil, nil, err
}
if jobs, err = r.countByStatus(ctx, "jobs", jobStatuses); err != nil {
return nil, nil, nil, err
}
if workers, err = r.countByStatus(ctx, "workers", workerStatuses); err != nil {
return nil, nil, nil, err
}
return tasks, jobs, workers, nil
}
func (r *StatsRepo) countByStatus(ctx context.Context, table string, known []string) (map[string]int, error) {
out := make(map[string]int, len(known))
for _, s := range known {
out[s] = 0 // zero-fill
}
// table is a fixed internal constant, never user input — safe to format.
rows, err := r.pool.Query(ctx, fmt.Sprintf("SELECT status, count(*) FROM %s GROUP BY status", table))
if err != nil {
return nil, fmt.Errorf("count %s by status: %w", table, err)
}
defer rows.Close()
for rows.Next() {
var status string
var n int
if err := rows.Scan(&status, &n); err != nil {
return nil, err
}
out[status] = n // an unknown status still shows up, which is a useful signal
}
return out, rows.Err()
}
@@ -84,6 +84,9 @@ WITH candidate AS (
WHERE status = 'pending'
AND attempt < max_attempts
AND (cardinality($1::text[]) = 0 OR workload = ANY($1))
AND ($5::uuid IS NULL OR NOT EXISTS (
SELECT 1 FROM task_results tr
WHERE tr.task_id = tasks.id AND tr.owner_id = $5))
ORDER BY created_at, chunk_index
FOR UPDATE SKIP LOCKED
LIMIT 1
@@ -108,7 +111,7 @@ func (r *TaskRepo) ClaimNext(ctx context.Context, f usecase.ClaimFilter) (*domai
var task *domain.Task
err := withRetry(ctx, func(ctx context.Context) error {
row := conn(ctx, r.pool).QueryRow(ctx, claimNextSQL, workloads, f.Owner, f.LeaseUntil, f.Now)
row := conn(ctx, r.pool).QueryRow(ctx, claimNextSQL, workloads, f.Owner, f.LeaseUntil, f.Now, f.VoterOwner)
t, err := scanTask(row)
if errors.Is(err, pgx.ErrNoRows) {
task = nil
@@ -0,0 +1,45 @@
package postgres
import (
"context"
"fmt"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgxpool"
)
// TaskResultRepo records and tallies quorum votes for untrusted task results.
type TaskResultRepo struct {
pool *pgxpool.Pool
}
func NewTaskResultRepo(pool *pgxpool.Pool) *TaskResultRepo {
return &TaskResultRepo{pool: pool}
}
// RecordVote stores (or replaces) one owner's vote for a task's result.
func (r *TaskResultRepo) RecordVote(ctx context.Context, taskID, ownerID uuid.UUID, sha256 string, artifactID uuid.UUID) error {
const sql = `
INSERT INTO task_results (task_id, owner_id, result_sha256, result_artifact_id)
VALUES ($1, $2, $3, $4)
ON CONFLICT (task_id, owner_id) DO UPDATE
SET result_sha256 = EXCLUDED.result_sha256,
result_artifact_id = EXCLUDED.result_artifact_id,
created_at = now()`
if _, err := conn(ctx, r.pool).Exec(ctx, sql, taskID, ownerID, sha256, artifactID); err != nil {
return fmt.Errorf("record vote: %w", err)
}
return nil
}
// CountAgreeing returns how many distinct owners have voted for the given result
// hash on this task — the size of the agreeing set the quorum is measured
// against.
func (r *TaskResultRepo) CountAgreeing(ctx context.Context, taskID uuid.UUID, sha256 string) (int, error) {
const sql = `SELECT count(DISTINCT owner_id) FROM task_results WHERE task_id = $1 AND result_sha256 = $2`
var n int
if err := conn(ctx, r.pool).QueryRow(ctx, sql, taskID, sha256).Scan(&n); err != nil {
return 0, fmt.Errorf("count agreeing: %w", err)
}
return n, nil
}
@@ -24,11 +24,15 @@ func (r *UIReadRepo) GetJob(ctx context.Context, id uuid.UUID) (*domain.Job, err
return job, err
}
func (r *UIReadRepo) ListJobs(ctx context.Context, limit int) ([]domain.Job, error) {
func (r *UIReadRepo) ListJobs(ctx context.Context, owner *uuid.UUID, limit int) ([]domain.Job, error) {
if limit < 1 || limit > 100 {
return nil, domain.ErrInvalidInput
}
sql, args, err := psql.Select(jobColumns...).From("jobs").OrderBy("created_at DESC", "id DESC").Limit(uint64(limit)).ToSql()
q := psql.Select(jobColumns...).From("jobs")
if owner != nil {
q = q.Where(sq.Eq{"owner_id": *owner})
}
sql, args, err := q.OrderBy("created_at DESC", "id DESC").Limit(uint64(limit)).ToSql()
if err != nil {
return nil, err
}
@@ -41,13 +45,13 @@ func (r *UIReadRepo) ListJobs(ctx context.Context, limit int) ([]domain.Job, err
for rows.Next() {
var j domain.Job
var status string
var inputURI *string
if err := rows.Scan(&j.ID, &j.Workload, &inputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt); err != nil {
if err := rows.Scan(
&j.ID, &j.Workload, &j.InputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt,
&j.InputArtifactID, &j.ResultArtifactID, &j.ErrorCode, &j.ErrorMessage, &j.ReducerStartedAt,
&j.OwnerID,
); err != nil {
return nil, err
}
if inputURI != nil {
j.InputURI = *inputURI
}
j.Status = domain.JobStatus(status)
jobs = append(jobs, j)
}
@@ -124,6 +128,32 @@ func (r *UIReadRepo) ListWorkers(ctx context.Context, limit int) ([]domain.Worke
return workers, rows.Err()
}
func (r *UIReadRepo) ListWorkersByOwner(ctx context.Context, owner uuid.UUID, limit int) ([]domain.Worker, error) {
if limit < 1 || limit > 100 {
return nil, domain.ErrInvalidInput
}
sql, args, err := psql.Select(workerColumns...).From("workers").
Where(sq.Eq{"owner_id": owner}).
OrderBy("last_heartbeat_at DESC", "id DESC").Limit(uint64(limit)).ToSql()
if err != nil {
return nil, err
}
rows, err := conn(ctx, r.pool).Query(ctx, sql, args...)
if err != nil {
return nil, fmt.Errorf("list workers by owner: %w", err)
}
defer rows.Close()
workers := make([]domain.Worker, 0)
for rows.Next() {
worker, err := scanWorker(rows)
if err != nil {
return nil, err
}
workers = append(workers, *worker)
}
return workers, rows.Err()
}
func (r *UIReadRepo) ListArtifactsByJob(ctx context.Context, jobID uuid.UUID) ([]domain.Artifact, error) {
sql, args, err := psql.Select(artifactColumns...).From("artifacts").Where(sq.Eq{"job_id": jobID}).OrderBy("created_at ASC", "id ASC").ToSql()
if err != nil {
@@ -23,13 +23,13 @@ func NewWorkerRepo(pool *pgxpool.Pool) *WorkerRepo {
return &WorkerRepo{pool: pool}
}
var workerColumns = []string{"id", "name", "capabilities", "status", "last_heartbeat_at", "created_at", "updated_at"}
var workerColumns = []string{"id", "name", "capabilities", "status", "owner_id", "trust_level", "last_heartbeat_at", "created_at", "updated_at"}
func (r *WorkerRepo) Insert(ctx context.Context, w *domain.Worker) error {
sql, args, err := psql.Insert("workers").
Columns(workerColumns...).
// capabilities is a jsonb column; pgx marshals the []string to a JSON array.
Values(w.ID, w.Name, w.Capabilities, string(w.Status),
Values(w.ID, w.Name, w.Capabilities, string(w.Status), w.OwnerID, string(w.TrustLevel),
w.LastHeartbeatAt, w.CreatedAt, w.UpdatedAt).
ToSql()
if err != nil {
@@ -95,11 +95,13 @@ func scanWorker(row pgx.Row) (*domain.Worker, error) {
var (
w domain.Worker
status string
trust string
)
if err := row.Scan(&w.ID, &w.Name, &w.Capabilities, &status,
if err := row.Scan(&w.ID, &w.Name, &w.Capabilities, &status, &w.OwnerID, &trust,
&w.LastHeartbeatAt, &w.CreatedAt, &w.UpdatedAt); err != nil {
return nil, err
}
w.Status = domain.WorkerStatus(status)
w.TrustLevel = domain.WorkerTrust(trust)
return &w, nil
}
+60
View File
@@ -0,0 +1,60 @@
// Package token verifies the HS256 JWTs minted by the userservice. The
// coordinator only ever *verifies* — it never issues — so this is a deliberately
// small counterpart to the userservice's issuer. Verification is local: the
// shared secret is enough, with no runtime call back to the userservice.
package token
import (
"fmt"
"github.com/golang-jwt/jwt/v5"
"github.com/google/uuid"
)
// Claims is the subset of a userservice token the coordinator cares about.
type Claims struct {
UserID uuid.UUID
Role string
Verified bool
}
// Verifier checks tokens against the shared HS256 secret.
type Verifier struct {
secret []byte
}
// NewVerifier returns a Verifier, or nil when secret is empty — a nil Verifier
// means user-JWT auth is disabled and only the shared service token is accepted.
func NewVerifier(secret string) *Verifier {
if secret == "" {
return nil
}
return &Verifier{secret: []byte(secret)}
}
type claims struct {
Role string `json:"role"`
Verified bool `json:"verified"`
jwt.RegisteredClaims
}
// Verify checks the signature and expiry and returns the identity. It pins the
// algorithm to HMAC, rejecting a token that asks for "none" or an RS256 public
// key — the classic algorithm-substitution attack.
func (v *Verifier) Verify(raw string) (Claims, error) {
var c claims
_, err := jwt.ParseWithClaims(raw, &c, func(t *jwt.Token) (any, error) {
if _, ok := t.Method.(*jwt.SigningMethodHMAC); !ok {
return nil, fmt.Errorf("unexpected signing method: %v", t.Header["alg"])
}
return v.secret, nil
})
if err != nil {
return Claims{}, err
}
id, err := uuid.Parse(c.Subject)
if err != nil {
return Claims{}, fmt.Errorf("token subject is not a uuid: %w", err)
}
return Claims{UserID: id, Role: c.Role, Verified: c.Verified}, nil
}
@@ -0,0 +1,98 @@
package token
import (
"testing"
"time"
"github.com/golang-jwt/jwt/v5"
"github.com/google/uuid"
)
const secret = "coordinator-verify-secret-32-bytes!!"
func sign(t *testing.T, method jwt.SigningMethod, key any, sub, role string, exp time.Time) string {
t.Helper()
return signVerified(t, method, key, sub, role, false, exp)
}
func signVerified(t *testing.T, method jwt.SigningMethod, key any, sub, role string, verified bool, exp time.Time) string {
t.Helper()
tok := jwt.NewWithClaims(method, claims{
Role: role,
Verified: verified,
RegisteredClaims: jwt.RegisteredClaims{
Subject: sub,
ExpiresAt: jwt.NewNumericDate(exp),
},
})
raw, err := tok.SignedString(key)
if err != nil {
t.Fatalf("sign: %v", err)
}
return raw
}
func TestVerifyCarriesVerifiedClaim(t *testing.T) {
v := NewVerifier(secret)
raw := signVerified(t, jwt.SigningMethodHS256, []byte(secret), uuid.New().String(), "user", true, time.Now().Add(time.Hour))
claims, err := v.Verify(raw)
if err != nil {
t.Fatalf("verify: %v", err)
}
if !claims.Verified {
t.Error("verified claim not read from token")
}
}
func TestNewVerifierNilWhenNoSecret(t *testing.T) {
if NewVerifier("") != nil {
t.Error("empty secret must yield a nil verifier (auth disabled)")
}
}
func TestVerifyRoundTrip(t *testing.T) {
v := NewVerifier(secret)
id := uuid.New()
raw := sign(t, jwt.SigningMethodHS256, []byte(secret), id.String(), "admin", time.Now().Add(time.Hour))
claims, err := v.Verify(raw)
if err != nil {
t.Fatalf("verify: %v", err)
}
if claims.UserID != id {
t.Errorf("UserID = %v, want %v", claims.UserID, id)
}
if claims.Role != "admin" {
t.Errorf("Role = %q, want admin", claims.Role)
}
}
func TestVerifyRejectsExpired(t *testing.T) {
v := NewVerifier(secret)
raw := sign(t, jwt.SigningMethodHS256, []byte(secret), uuid.New().String(), "user", time.Now().Add(-time.Minute))
if _, err := v.Verify(raw); err == nil {
t.Error("expired token accepted")
}
}
func TestVerifyRejectsWrongSecret(t *testing.T) {
raw := sign(t, jwt.SigningMethodHS256, []byte(secret), uuid.New().String(), "user", time.Now().Add(time.Hour))
if _, err := NewVerifier("another-secret-also-at-least-32-byte").Verify(raw); err == nil {
t.Error("token verified under the wrong secret")
}
}
func TestVerifyRejectsNoneAlg(t *testing.T) {
raw := sign(t, jwt.SigningMethodNone, jwt.UnsafeAllowNoneSignatureType, uuid.New().String(), "admin", time.Now().Add(time.Hour))
if _, err := NewVerifier(secret).Verify(raw); err == nil {
t.Error("none-signed token accepted")
}
}
func TestVerifyRejectsNonUUIDSubject(t *testing.T) {
raw := sign(t, jwt.SigningMethodHS256, []byte(secret), "not-a-uuid", "user", time.Now().Add(time.Hour))
if _, err := NewVerifier(secret).Verify(raw); err == nil {
t.Error("non-uuid subject accepted")
}
}
@@ -12,6 +12,7 @@ import (
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
)
@@ -56,10 +57,24 @@ func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) {
return
}
worker, err := s.uc.RegisterWorker.Execute(ctx, usecase.RegisterWorkerInput{
// Resolve the worker's trust tier from how the caller authenticated:
// - shared service token (no requester) -> trusted lab worker
// - verified/admin user JWT -> trusted volunteer
// - plain user JWT -> untrusted (quarantined)
in := usecase.RegisterWorkerInput{
Name: req.Name,
Capabilities: req.Capabilities,
})
TrustLevel: domain.WorkerTrusted,
}
if requester, ok := authctx.From(ctx); ok {
id := requester.UserID
in.OwnerID = &id
if !requester.IsTrusted() {
in.TrustLevel = domain.WorkerUntrusted
}
}
worker, err := s.uc.RegisterWorker.Execute(ctx, in)
if err != nil {
s.writeError(w, r, err)
return
@@ -9,6 +9,9 @@ import (
"net/http"
"strings"
"time"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
tokenpkg "github.com/emil28092005/SciMesh/coordinator/internal/token"
)
type ctxKey string
@@ -41,25 +44,46 @@ func newRequestID() string {
// withAuth enforces the shared bearer token every worker presents.
// An empty token disables the check (local development only).
func withAuth(token string) func(http.Handler) http.Handler {
// withAuth authenticates a request one of two ways. Workers (and legacy
// submitters) present the shared service token. When user-JWT auth is enabled
// (verifier != nil), a submitter may instead present a userservice JWT; on
// success the requester is stamped into the context so the job use cases can
// record owner_id and enforce ownership. An empty token with no verifier
// disables auth entirely (dev only).
func withAuth(token string, verifier *tokenpkg.Verifier) func(http.Handler) http.Handler {
return func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if token == "" {
if token == "" && verifier == nil {
next.ServeHTTP(w, r)
return
}
presented := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
// Constant-time compare: a byte-by-byte early exit would let an
// attacker recover the token by timing responses.
if subtle.ConstantTimeCompare([]byte(presented), []byte(token)) != 1 {
w.Header().Set("WWW-Authenticate", "Bearer")
writeJSON(w, http.StatusUnauthorized, errorResponse{
Error: "unauthorized",
RequestID: requestIDFrom(r.Context()),
})
// Shared service token: constant-time compare so a byte-by-byte
// early exit cannot leak the token through response timing.
if token != "" && subtle.ConstantTimeCompare([]byte(presented), []byte(token)) == 1 {
next.ServeHTTP(w, r)
return
}
next.ServeHTTP(w, r)
// Otherwise try a user JWT, if that path is configured.
if verifier != nil && presented != "" {
if claims, err := verifier.Verify(presented); err == nil {
ctx := authctx.With(r.Context(), authctx.Requester{
UserID: claims.UserID,
Role: claims.Role,
Verified: claims.Verified,
})
next.ServeHTTP(w, r.WithContext(ctx))
return
}
}
w.Header().Set("WWW-Authenticate", "Bearer")
writeJSON(w, http.StatusUnauthorized, errorResponse{
Error: "unauthorized",
RequestID: requestIDFrom(r.Context()),
})
})
}
}
+114 -20
View File
@@ -7,8 +7,11 @@ import (
"context"
"log/slog"
"net/http"
"strings"
"time"
"github.com/emil28092005/SciMesh/coordinator/internal/metrics"
tokenpkg "github.com/emil28092005/SciMesh/coordinator/internal/token"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
)
@@ -31,6 +34,7 @@ type UseCases struct {
DownloadArtifact *usecase.DownloadArtifact
GetTaskInput *usecase.GetTaskInput
Dashboard *usecase.Dashboard
PreviewArtifact *usecase.PreviewArtifact
}
type Server struct {
@@ -39,21 +43,62 @@ type Server struct {
requestTimeout time.Duration
heartbeatInterval time.Duration
maxUploadBytes int64
// verifier validates userservice JWTs. nil disables user-JWT auth, leaving
// only the shared service token — the pre-userservice behaviour.
verifier *tokenpkg.Verifier
// userserviceURL is the base URL the UI proxies login/registration to. Empty
// keeps the static basic-auth UI.
userserviceURL string
// publicCoordinatorURL / publicUserserviceURL are the browser-facing URLs
// rendered into the worker-enrollment command. Either may be empty; the
// template falls back (own origin / userserviceURL respectively).
publicCoordinatorURL string
publicUserserviceURL string
// httpClient makes the login/register calls to the userservice.
httpClient *http.Client
// metrics holds the Prometheus registry and HTTP instrumentation.
metrics *metrics.Metrics
// ready probes downstream dependencies (the database) for /health. Kept as
// a func so the transport layer never imports pgx.
ready func(context.Context) error
}
func NewServer(uc UseCases, log *slog.Logger, requestTimeout, heartbeatInterval time.Duration,
maxUploadBytes int64, ready func(context.Context) error) *Server {
return &Server{
uc: uc,
log: log,
requestTimeout: requestTimeout,
heartbeatInterval: heartbeatInterval,
maxUploadBytes: maxUploadBytes,
ready: ready,
maxUploadBytes int64, jwtSecret, userserviceURL string, m *metrics.Metrics, ready func(context.Context) error,
publicURLs ...string) *Server {
if m == nil {
m = metrics.New()
}
// publicURLs is variadic so existing callers/tests need no change: [0] is the
// public coordinator URL, [1] the public userservice URL; both optional.
var publicCoordinatorURL, publicUserserviceURL string
if len(publicURLs) > 0 {
publicCoordinatorURL = strings.TrimRight(publicURLs[0], "/")
}
if len(publicURLs) > 1 {
publicUserserviceURL = strings.TrimRight(publicURLs[1], "/")
}
return &Server{
uc: uc,
log: log,
requestTimeout: requestTimeout,
heartbeatInterval: heartbeatInterval,
maxUploadBytes: maxUploadBytes,
verifier: tokenpkg.NewVerifier(jwtSecret),
userserviceURL: strings.TrimRight(userserviceURL, "/"),
publicCoordinatorURL: publicCoordinatorURL,
publicUserserviceURL: publicUserserviceURL,
httpClient: &http.Client{Timeout: 10 * time.Second},
metrics: m,
ready: ready,
}
}
// uiSessionMode reports whether the operator UI authenticates via userservice
// login (cookie session) rather than the static basic-auth token. It needs both
// a verifier (to check the JWT locally) and a userservice URL (to issue it).
func (s *Server) uiSessionMode() bool {
return s.verifier != nil && s.userserviceURL != ""
}
// Handler builds the router. Go 1.22's ServeMux matches on method and path
@@ -76,17 +121,65 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("GET /health", s.handleHealth)
if len(uiToken) > 0 && uiToken[0] != "" && s.uc.Dashboard != nil {
// Unauthenticated like /health, so a Prometheus scraper needs no credential.
mux.Handle("GET /metrics", s.metrics.Handler())
hasBasicAuth := len(uiToken) > 0 && uiToken[0] != ""
if s.uc.Dashboard != nil && (s.uiSessionMode() || hasBasicAuth) {
ui := http.NewServeMux()
ui.HandleFunc("GET /ui", s.handleUIHome)
ui.HandleFunc("GET /ui/jobs/new", s.handleUINewJob)
ui.HandleFunc("GET /ui/jobs/{job_id}", s.handleUIJob)
ui.HandleFunc("GET /ui/api/jobs/{job_id}", s.handleUIJobJSON)
ui.HandleFunc("POST /ui/api/jobs/{job_id}/cancel", s.handleCancelJob)
ui.HandleFunc("POST /ui/api/jobs/upload", s.handleUploadDataset)
ui.HandleFunc("GET /ui/jobs/{job_id}/artifacts/{artifact_id}", s.handleUIArtifactDownload)
mux.Handle("/ui", chain(ui, withRequestID, withAccessLog(s.log), withBasicAuth(uiToken[0]), withSameOrigin))
mux.Handle("/ui/", chain(ui, withRequestID, withAccessLog(s.log), withBasicAuth(uiToken[0]), withSameOrigin))
// The operator application routes, all requiring an authenticated caller.
app := []struct {
pattern string
handler http.HandlerFunc
}{
{"GET /ui", s.handleUIHome},
{"GET /ui/jobs/new", s.handleUINewJob},
{"GET /ui/jobs/{job_id}", s.handleUIJob},
{"GET /ui/api/overview", s.handleUIOverviewJSON},
{"GET /ui/api/jobs/{job_id}", s.handleUIJobJSON},
{"POST /ui/api/jobs/{job_id}/cancel", s.handleCancelJob},
{"POST /ui/api/jobs/upload", s.handleUploadDataset},
{"GET /ui/jobs/{job_id}/artifacts/{artifact_id}", s.handleUIArtifactDownload},
{"GET /ui/jobs/{job_id}/artifacts/{artifact_id}/preview", s.handleUIArtifactPreview},
}
if s.uiSessionMode() {
// Public auth pages — reachable without a session so a user can log in.
ui.HandleFunc("GET /ui/login", s.handleUILoginForm)
ui.HandleFunc("POST /ui/login", s.handleUILogin)
ui.HandleFunc("GET /ui/register", s.handleUIRegisterForm)
ui.HandleFunc("POST /ui/register", s.handleUIRegister)
ui.HandleFunc("POST /ui/logout", s.handleUILogout)
gate := withUISession(s.verifier)
for _, rt := range app {
ui.Handle(rt.pattern, gate(rt.handler))
}
ui.Handle("GET /ui/profile", gate(http.HandlerFunc(s.handleUIProfile)))
// Worker enrollment: a user creates/lists/revokes their own worker keys
// and copies a ready-to-run command. Session-only — it proxies to the
// userservice with the caller's token, so it has no meaning under basic
// auth (which has no userservice).
ui.Handle("GET /ui/workers/new", gate(http.HandlerFunc(s.handleUIAddWorker)))
ui.Handle("GET /ui/api/worker-keys", gate(http.HandlerFunc(s.handleUIWorkerKeysList)))
ui.Handle("POST /ui/api/worker-keys", gate(http.HandlerFunc(s.handleUIWorkerKeyCreate)))
ui.Handle("POST /ui/api/worker-keys/{id}/revoke", gate(http.HandlerFunc(s.handleUIWorkerKeyRevoke)))
// Admin panel: session + admin role.
ui.Handle("GET /ui/admin", chain(http.HandlerFunc(s.handleUIAdmin), gate, requireAdmin))
ui.Handle("POST /ui/admin/user-action", chain(http.HandlerFunc(s.handleUIAdminUserAction), gate, requireAdmin))
} else {
for _, rt := range app {
ui.HandleFunc(rt.pattern, rt.handler)
}
}
common := []func(http.Handler) http.Handler{withRequestID, withAccessLog(s.log)}
if !s.uiSessionMode() {
common = append(common, withBasicAuth(uiToken[0]))
}
common = append(common, withSameOrigin)
mux.Handle("/ui", chain(ui, common...))
mux.Handle("/ui/", chain(ui, common...))
} else {
// More specific than the protected catch-all: UI absence is not an auth
// failure and does not disclose that a UI feature is configured elsewhere.
@@ -96,9 +189,10 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
mux.Handle("/", chain(protected,
withRequestID, // outermost: every response gets an ID,
withAccessLog(s.log), // including the 401s below
withAuth(token),
withAuth(token, s.verifier),
))
return mux
// Measure every request once, outermost, with a normalized route label.
return s.metrics.Middleware(mux)
}
// handleHealth reports readiness. It probes the database so an orchestrator
@@ -50,7 +50,7 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur
SubmitDataset: usecase.NewSubmitDataset(blobs, arts, jobs, tasks, tx, clk, 3),
ClaimTask: usecase.NewClaimTask(tasks, jobs, work, tx, clk, lease),
RenewLease: usecase.NewRenewLease(tasks, work, tx, clk, lease),
CompleteTask: usecase.NewCompleteTask(tasks, jobs, arts, tx, clk),
CompleteTask: usecase.NewCompleteTask(tasks, jobs, arts, work, memstore.NewTaskResultRepo(), tx, clk, 2),
ReduceJob: usecase.NewReduceJob(jobs, tasks, arts, blobs, tx, clk),
FailTask: usecase.NewFailTask(tasks, jobs, tx, clk),
GetJobStatus: usecase.NewGetJobStatus(jobs, tasks),
@@ -60,6 +60,7 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur
DownloadArtifact: downloadArtifact,
GetTaskInput: usecase.NewGetTaskInput(tasks, arts, blobs),
Dashboard: usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, work, arts)),
PreviewArtifact: usecase.NewPreviewArtifact(memstore.NewUIReadRepo(jobs, tasks, work, arts), blobs),
}
worker, err := uc.RegisterWorker.Execute(context.Background(), usecase.RegisterWorkerInput{
Name: "test-worker", Capabilities: []string{"w", "similarity-search"},
@@ -67,7 +68,7 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur
if err != nil {
t.Fatalf("register test worker: %v", err)
}
srv := coordhttp.NewServer(uc, slog.New(slog.NewTextHandler(io.Discard, nil)), 5*time.Second, 15*time.Second, 1<<30, ready)
srv := coordhttp.NewServer(uc, slog.New(slog.NewTextHandler(io.Discard, nil)), 5*time.Second, 15*time.Second, 1<<30, "", "", nil, ready)
ts := httptest.NewServer(srv.Handler(token, configuredUIToken))
t.Cleanup(ts.Close)
return &env{ts: ts, blobs: blobs, workerID: worker.ID.String()}
@@ -152,11 +153,36 @@ func TestUIRequiresDistinctCredentialAndRendersDashboard(t *testing.T) {
t.Fatalf("UI status: %d", resp.StatusCode)
}
body, _ := io.ReadAll(resp.Body)
if !strings.Contains(string(body), "SciMesh operator dashboard") {
if !strings.Contains(string(body), "SciMesh control room") {
t.Errorf("dashboard body missing title")
}
}
func TestUIOverviewReturnsLiveSafeProjection(t *testing.T) {
e := newEnv(t, healthy)
code, _ := e.do(t, "POST", "/jobs", `{"workload":"w","input_uri":"s3://in","chunks":[{"chunk_index":0,"input_uri":"s3://c","input_sha256":"sha"}]}`)
if code != http.StatusCreated {
t.Fatalf("create job: %d", code)
}
req, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/api/overview", nil)
req.SetBasicAuth("operator", uiToken)
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
var overview map[string]any
if err := json.NewDecoder(resp.Body).Decode(&overview); err != nil {
t.Fatal(err)
}
if resp.StatusCode != http.StatusOK || overview["active_jobs"].(float64) != 1 || overview["online_workers"].(float64) != 1 {
t.Fatalf("overview = (%d, %v)", resp.StatusCode, overview)
}
if _, leaked := overview["worker_auth_token"]; leaked {
t.Fatal("overview must not expose authentication configuration")
}
}
func TestUIDisabledReturnsNotFound(t *testing.T) {
e := newEnvWithUIToken(t, healthy, "")
resp := e.get(t, "/ui")
@@ -437,6 +463,77 @@ func TestSimilaritySearchLifecyclePublishesFinalResult(t *testing.T) {
if resp.StatusCode != http.StatusOK || string(body) != "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.900000\n" {
t.Fatalf("final result = (%d, %q)", resp.StatusCode, body)
}
uiRequest, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/jobs/"+jobID, nil)
uiRequest.SetBasicAuth("operator", uiToken)
uiResponse, err := http.DefaultClient.Do(uiRequest)
if err != nil {
t.Fatal(err)
}
defer uiResponse.Body.Close()
uiBody, _ := io.ReadAll(uiResponse.Body)
if uiResponse.StatusCode != http.StatusOK || !strings.Contains(string(uiBody), "Final result ready") || !strings.Contains(string(uiBody), "Preview CSV") || !strings.Contains(string(uiBody), "Processing speed") {
t.Fatalf("final UI = (%d, %q)", uiResponse.StatusCode, uiBody)
}
jsonRequest, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/api/jobs/"+jobID, nil)
jsonRequest.SetBasicAuth("operator", uiToken)
jsonResponse, err := http.DefaultClient.Do(jsonRequest)
if err != nil {
t.Fatal(err)
}
defer jsonResponse.Body.Close()
var detail map[string]any
if err := json.NewDecoder(jsonResponse.Body).Decode(&detail); err != nil {
t.Fatal(err)
}
if jsonResponse.StatusCode != http.StatusOK || detail["final_result_available"] != true {
t.Fatalf("final UI JSON = (%d, %v)", jsonResponse.StatusCode, detail)
}
artifacts := detail["artifacts"].([]any)
var finalID string
for _, raw := range artifacts {
artifact := raw.(map[string]any)
if artifact["kind"] == "final_result" && artifact["downloadable"] == true {
finalID = artifact["id"].(string)
break
}
}
if finalID == "" {
t.Fatalf("artifacts = %v, want downloadable final result", artifacts)
}
var inputID string
for _, raw := range artifacts {
artifact := raw.(map[string]any)
if artifact["kind"] == "input" {
inputID = artifact["id"].(string)
break
}
}
if inputID == "" {
t.Fatalf("artifacts = %v, want input artifact", artifacts)
}
inputRequest, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/jobs/"+jobID+"/artifacts/"+inputID, nil)
inputRequest.SetBasicAuth("operator", uiToken)
inputResponse, err := http.DefaultClient.Do(inputRequest)
if err != nil {
t.Fatal(err)
}
defer inputResponse.Body.Close()
if inputResponse.StatusCode != http.StatusNotFound {
t.Fatalf("UI input download = %d, want 404", inputResponse.StatusCode)
}
previewRequest, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/jobs/"+jobID+"/artifacts/"+finalID+"/preview", nil)
previewRequest.SetBasicAuth("operator", uiToken)
previewResponse, err := http.DefaultClient.Do(previewRequest)
if err != nil {
t.Fatal(err)
}
defer previewResponse.Body.Close()
previewBody, _ := io.ReadAll(previewResponse.Body)
if previewResponse.StatusCode != http.StatusOK || !strings.Contains(string(previewBody), "Final result preview") || !strings.Contains(string(previewBody), "0.900000") {
t.Fatalf("final preview = (%d, %q)", previewResponse.StatusCode, previewBody)
}
}
func TestForeignArtifactResultConflict(t *testing.T) {
@@ -0,0 +1,56 @@
{{define "add-worker.html"}}
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Add your machine · SciMesh</title>
<style>
:root{color:#e5efff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 10% -8%,#183f77 0,transparent 32rem),#08111f}.page{max-width:980px;margin:auto;padding:28px 22px 64px}a{color:#94bdff}.back{text-decoration:none}.eyebrow{margin:28px 0 4px;color:#7baaff;font-size:.78rem;font-weight:800;letter-spacing:.14em;text-transform:uppercase}h1{margin:0;color:#f4f8ff;font-size:clamp(2rem,5vw,3.25rem);letter-spacing:-.055em}.lead{max-width:720px;margin:10px 0 0;color:#aabed9;font-size:1.06rem}.layout{display:grid;grid-template-columns:1.45fr .8fr;gap:15px;margin-top:28px}.card,.aside,.notice{border:1px solid #294662;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #00000021}.card{padding:22px}.aside,.notice{padding:18px}.aside h2,.notice h2{margin:0;color:#f1f6ff;font-size:1rem}.aside p,.notice p{color:#9fb3cf}.aside ol{margin:13px 0 0;padding-left:20px;color:#aebfda}.aside li{margin:10px 0}label{display:block;margin:18px 0 5px;color:#eaf2ff;font-weight:750}input{width:100%;border:1px solid #42617f;border-radius:9px;padding:10px 11px;background:#0a1626;color:#e5efff;font:inherit}input:focus{outline:2px solid #5d97f5;outline-offset:1px}.hint{margin:5px 0;color:#92a9c6;font-size:.87rem}.button{display:inline-flex;margin-top:16px;border:0;border-radius:10px;padding:11px 15px;background:#67e3b8;color:#062018;font:inherit;font-weight:850;cursor:pointer}.button.secondary{background:#23344d;color:#dce8ff}.button:disabled{opacity:.6;cursor:wait}.command{margin-top:18px;border:1px solid #2c8c70;border-radius:11px;padding:14px;background:#0c2b2a;color:#a8f1d0}.command strong{color:#e6fff4}.command pre{margin:10px 0 0;padding:12px;overflow-x:auto;border-radius:8px;background:#061a19;color:#c8ffe8;font:.82rem/1.5 ui-monospace,SFMono-Regular,monospace;white-space:pre;word-break:normal}.keys{margin-top:14px;display:grid;gap:9px}.key{display:flex;align-items:center;justify-content:space-between;gap:12px;border:1px solid #294662;border-radius:11px;padding:12px 14px;background:#0a1626}.key .kn{color:#f3f7ff;font-weight:700}.key .kp{margin-top:3px;color:#8196b3;font-family:ui-monospace,SFMono-Regular,monospace;font-size:.76rem}.key .kd{color:#8fa6c3;font-size:.8rem}.revoke{border:1px solid #6a2a3a;border-radius:8px;padding:7px 11px;background:#2a1420;color:#ff9bad;font:inherit;font-weight:700;cursor:pointer}.empty{padding:20px;border:1px dashed #35516f;border-radius:12px;color:#9ab0cb;text-align:center}.error{margin:12px 0 0;color:#ffacba}.warn{color:#ffd08a}.hidden{display:none}.cap{display:inline-block;margin:4px 5px 0 0;border:1px solid #375978;border-radius:5px;padding:2px 6px;color:#b5d3f5;font:.76rem ui-monospace,SFMono-Regular,monospace}@media(max-width:720px){.layout{grid-template-columns:1fr}.page{padding:22px 14px}}
</style>
</head>
<body data-coordinator="{{.CoordinatorURL}}" data-userservice="{{.UserserviceURL}}">
<main class="page">
<a class="back" href="/ui">← Back to control room</a><p class="eyebrow">Contribute compute</p><h1>Turn this computer into a worker</h1><p class="lead">Create a key, install the worker, and run one command. The worker binds to your account and pulls tasks whenever it is online.</p>
<div class="layout">
<section class="card">
<h2 style="margin:0 0 4px;color:#f1f6ff">Your worker keys</h2>
<p class="hint">A key is long-lived and does not expire like a login. The worker trades it for short-lived tokens automatically. Revoke a key to stop its machines.</p>
<form id="create" novalidate>
<label for="key-name">Name this machine <small>(optional)</small></label>
<input id="key-name" name="name" maxlength="100" placeholder="e.g. home-desktop" autocomplete="off">
<button class="button" id="create-btn" type="submit">Create key →</button>
<p id="error" class="error" role="alert"></p>
</form>
<div id="command" class="command hidden"></div>
<div id="keys" class="keys"></div>
</section>
<aside class="aside">
<h2>Set it up</h2>
<ol>
<li><strong>Create a key</strong><br>Use the form; copy the command it generates.</li>
<li><strong>Paste it in a terminal</strong><br>The command clones the project, sets up a Python environment, installs the worker, and starts it. The machine then appears under <a href="/ui">My machines</a>.</li>
</ol>
<h2 style="margin-top:24px">Will my results count?</h2>
<p>Your worker is <strong>untrusted</strong> by default: its results are cross-checked and accepted once a second independent worker computes the same answer (quorum), or once an admin marks your account <strong>verified</strong> — then your workers are trusted and results count immediately.</p>
<p><span class="cap">similarity-search</span> is the only workload a volunteer worker runs today.</p>
</aside>
</div>
</main>
<script>
const coord=(document.body.dataset.coordinator||location.origin).replace(/\/+$/,'');
const users=(document.body.dataset.userservice||'').replace(/\/+$/,'');
const keysBox=document.querySelector('#keys'),cmdBox=document.querySelector('#command'),form=document.querySelector('#create'),nameInput=document.querySelector('#key-name'),createBtn=document.querySelector('#create-btn'),error=document.querySelector('#error');
const node=(tag,text,cls)=>{const n=document.createElement(tag);if(text!==undefined)n.textContent=text;if(cls)n.className=cls;return n};
const shq=s=>"'"+String(s).replace(/'/g,"'\\''")+"'";
const buildCommand=(key,name)=>['git clone https://github.com/emil28092005/SciMesh.git','cd SciMesh','python -m venv .venv','source .venv/bin/activate','pip install -e .','','SCIMESH_COORDINATOR_URL='+coord+' \\','SCIMESH_USERSERVICE_URL='+(users||'<your userservice URL>')+' \\','SCIMESH_WORKER_KEY='+key+' \\','scimesh-worker --worker-name '+shq(name||'my-machine')].join('\n');
const showCommand=(key,name)=>{cmdBox.replaceChildren();cmdBox.append(node('strong','Key created — copy it now, it is shown only once.'));const pre=node('pre',buildCommand(key,name));cmdBox.append(pre);const copy=node('button','Copy command','button secondary');copy.type='button';copy.addEventListener('click',()=>{navigator.clipboard&&navigator.clipboard.writeText(buildCommand(key,name)).then(()=>{copy.textContent='Copied ✓'},()=>{copy.textContent='Press Ctrl+C to copy'})});cmdBox.append(document.createElement('br'),copy);if(!users){cmdBox.append(node('p','Set SCIMESH_USERSERVICE_URL to a userservice URL your machine can reach.','warn'))}cmdBox.classList.remove('hidden')};
const revoke=async id=>{const r=await fetch('/ui/api/worker-keys/'+encodeURIComponent(id)+'/revoke',{method:'POST'});if(r.status===204||r.ok){loadKeys()}else{error.textContent='Could not revoke the key.'}};
const renderKeys=keys=>{keysBox.replaceChildren();if(!keys.length){keysBox.append(node('div','No keys yet. Create one above to connect a machine.','empty'));return}for(const k of keys){const row=node('div',undefined,'key'),left=node('div');left.append(node('div',k.name||'unnamed','kn'),node('div',k.prefix+'…','kp'),node('div','Created '+new Date(k.created_at).toLocaleString()+(k.last_used_at?' · last used '+new Date(k.last_used_at).toLocaleString():' · never used'),'kd'));const btn=node('button','Revoke','revoke');btn.type='button';btn.addEventListener('click',()=>revoke(k.id));row.append(left,btn);keysBox.append(row)}};
const loadKeys=async()=>{try{const r=await fetch('/ui/api/worker-keys',{headers:{Accept:'application/json'}});if(!r.ok)throw Error();const data=await r.json();renderKeys(data.worker_keys||[])}catch(_){keysBox.replaceChildren(node('div','Could not load your keys.','empty'))}};
form.addEventListener('submit',async event=>{event.preventDefault();error.textContent='';createBtn.disabled=true;const machineName=nameInput.value.trim();try{const r=await fetch('/ui/api/worker-keys',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({name:machineName})}),data=await r.json().catch(()=>({}));if(!r.ok)throw Error(data.error||'Could not create the key.');showCommand(data.key,machineName);nameInput.value='';loadKeys()}catch(err){error.textContent=err.message}finally{createBtn.disabled=false}});
loadKeys();
</script>
</body>
</html>
{{end}}
@@ -0,0 +1,46 @@
{{define "admin.html"}}
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Admin · SciMesh</title>
<style>
:root{color:#e5efff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 10% -8%,#183f77 0,transparent 32rem),#08111f}.page{max-width:820px;margin:auto;padding:28px 22px 64px}a{color:#94bdff}.top{display:flex;justify-content:space-between;align-items:center;gap:12px}.eyebrow{margin:0;color:#7baaff;font-size:.78rem;font-weight:800;letter-spacing:.14em;text-transform:uppercase}h1{margin:6px 0 0;color:#f4f8ff;font-size:clamp(1.8rem,4vw,2.6rem);letter-spacing:-.04em}.lead{max-width:640px;margin:10px 0 0;color:#aabed9}.card{margin-top:24px;border:1px solid #294662;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #00000021;padding:22px}.card h2{margin:0 0 4px;color:#f1f6ff;font-size:1.1rem}.card p{margin:0;color:#9fb3cf;font-size:.92rem}label{display:block;margin:16px 0 5px;color:#eaf2ff;font-weight:750}input{width:100%;border:1px solid #42617f;border-radius:9px;padding:10px 11px;background:#0a1626;color:#e5efff;font:inherit}input:focus{outline:2px solid #5d97f5;outline-offset:1px}.actions{display:flex;flex-wrap:wrap;gap:9px;margin-top:16px}.btn{border:0;border-radius:9px;padding:10px 14px;font:inherit;font-weight:800;cursor:pointer}.btn-primary{background:#67e3b8;color:#062018}.btn-muted{background:#23344d;color:#dce8ff}.notice{margin-top:16px;border-radius:10px;padding:11px 13px;font-weight:700}.ok{background:#123f34;color:#76efb5}.err{background:#552334;color:#ff9bad}.muted{color:#8ba2c2}.hint{margin-top:4px;color:#92a9c6;font-size:.85rem}</style>
</head>
<body>
<main class="page">
<header class="top">
<div><p class="eyebrow">Admin panel</p><h1>User &amp; run control</h1></div>
<div style="display:flex;gap:10px;align-items:center"><a href="/ui">← Dashboard</a><a href="/ui/profile">Profile</a><form method="post" action="/ui/logout" style="margin:0"><button class="btn btn-muted" type="submit">Log out</button></form></div>
</header>
<p class="lead">Signed in as <strong>{{.Role}}</strong>. Promote or verify a user by their id, and control every job from the dashboard.</p>
{{if .Msg}}<div class="notice ok">{{.Msg}}</div>{{end}}
{{if .Error}}<div class="notice err">{{.Error}}</div>{{end}}
<section class="card">
<h2>Manage a user</h2>
<p>Paste the user id (the JWT <code>sub</code> / the value shown at registration). Actions are applied immediately.</p>
<form method="post" action="/ui/admin/user-action">
<label for="user_id">User id</label>
<input id="user_id" name="user_id" placeholder="00000000-0000-0000-0000-000000000000" autocomplete="off" required>
<p class="hint">Promote makes them an admin; Verify marks them a trusted contributor (their workers skip quorum).</p>
<div class="actions">
<button class="btn btn-primary" name="action" value="promote" type="submit">Make admin</button>
<button class="btn btn-muted" name="action" value="demote" type="submit">Remove admin</button>
<button class="btn btn-primary" name="action" value="verify" type="submit">Verify</button>
<button class="btn btn-muted" name="action" value="unverify" type="submit">Unverify</button>
</div>
</form>
</section>
<section class="card">
<h2>Jobs &amp; tasks</h2>
<p>As an admin you already see <strong>every user's jobs</strong> on the dashboard, with per-task status and job cancellation. A regular user sees only their own.</p>
<div class="actions"><a class="btn btn-muted" href="/ui" style="text-decoration:none">Open the dashboard →</a></div>
</section>
</main>
</body>
</html>
{{end}}
@@ -0,0 +1,35 @@
{{define "artifact-preview.html"}}
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>SciMesh · artifact preview</title>
<style>
:root{color:#e4eeff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}body{margin:0;background:radial-gradient(circle at 10% -5%,#173f76 0,transparent 34rem),#08111f}.page{max-width:1180px;margin:auto;padding:32px 20px 56px}a{color:#8ab5ff}.back{text-decoration:none}h1{margin:18px 0 4px;color:#f4f8ff;font-size:1.6rem;word-break:break-word}.muted{color:#9cb0cb}.notice{margin:16px 0;padding:15px 17px;border:1px solid #aa8844;border-radius:10px;background:#302610;color:#f2dd9a}.table-wrap{overflow-x:auto;border:1px solid #294662;border-radius:10px;background:#0d1a2cdc;margin-top:16px}table{width:100%;border-collapse:collapse}td,th{padding:10px 12px;border-bottom:1px solid #203a55;text-align:left;vertical-align:top;white-space:pre-wrap;word-break:break-word}th{color:#9cb9dc;font-size:.78rem;text-transform:uppercase;letter-spacing:.06em;background:#10253d}tr:last-child td{border:0}.empty{padding:24px;text-align:center;color:#9cb0cb}
</style>
</head>
<body>
<main class="page">
<a class="back" href="/ui/jobs/{{.JobID}}">← Back to job</a>
<h1>Preview: {{.Filename}}</h1>
{{if .Diagnostic}}
<p class="muted">Diagnostic preview — a shard-level partial result, not the final molecular-search answer. At most {{.RowLimit}} rows and {{.ByteLimit}} bytes are read from storage.</p>
{{else}}
<p class="muted">Final result preview. At most {{.RowLimit}} rows and {{.ByteLimit}} bytes are read from storage.</p>
{{end}}
{{if not .Previewable}}
<div class="notice">{{.Reason}}</div>
{{else}}
{{if .Truncated}}<div class="notice">Truncated: showing at most the first {{.RowLimit}} rows or {{.ByteLimit}} bytes. Download the artifact for its full contents.</div>{{end}}
<div class="table-wrap">
<table>
<tr>{{range .Headers}}<th>{{.}}</th>{{end}}</tr>
{{range .Rows}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{else}}<tr><td class="empty" colspan="99">No data rows.</td></tr>{{end}}
</table>
</div>
{{end}}
</main>
</body>
</html>
{{end}}
@@ -4,20 +4,39 @@
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>SciMesh operator dashboard</title>
<title>SciMesh control room</title>
<style>
:root{color:#172033;background:#f6f8fc;font:16px/1.5 system-ui,sans-serif}body{margin:0}.page{max-width:1180px;margin:auto;padding:32px 20px 56px}.top{display:flex;justify-content:space-between;gap:24px;align-items:start}.eyebrow{margin:0;color:#50617d;font-size:.86rem;font-weight:700;text-transform:uppercase;letter-spacing:.08em}h1{margin:.2rem 0;font-size:2rem}h2{margin:32px 0 12px;font-size:1.28rem}.lead{margin:0;color:#56657c}.button{display:inline-block;border:0;border-radius:8px;padding:11px 15px;background:#1f5eff;color:#fff;font-weight:700;text-decoration:none;white-space:nowrap}.notice{margin-top:24px;padding:16px 18px;border:1px solid #f2cb72;border-radius:10px;background:#fff8e6}.notice strong{display:block}.steps{display:grid;grid-template-columns:repeat(3,1fr);gap:12px;margin-top:14px}.step,.card{padding:16px;border:1px solid #dfe5f0;border-radius:10px;background:#fff}.step b{display:block;color:#1f5eff}.table-wrap{overflow-x:auto;background:#fff;border:1px solid #dfe5f0;border-radius:10px}table{width:100%;border-collapse:collapse}td,th{padding:13px 14px;border-bottom:1px solid #e8ecf4;text-align:left;vertical-align:top}th{color:#50617d;font-size:.78rem;text-transform:uppercase;letter-spacing:.06em}tr:last-child td{border:0}a{color:#174ecf}small,.muted{color:#68758b}.status{display:inline-block;border-radius:999px;padding:3px 9px;font-size:.84rem;font-weight:700}.status-success{background:#dff6e9;color:#126b3d}.status-danger{background:#ffe4e6;color:#a31135}.status-active{background:#e4edff;color:#174ecf}.status-waiting{background:#edf0f5;color:#50617d}.bar{height:7px;min-width:120px;margin-top:7px;overflow:hidden;border-radius:999px;background:#e6eaf1}.bar>span{display:block;height:100%;background:#1f5eff}.kicker{font-variant-numeric:tabular-nums}.empty{padding:28px;text-align:center;color:#68758b}.worker{display:grid;grid-template-columns:1.3fr .8fr 2fr 1fr;gap:12px;align-items:center}.worker+.worker{border-top:1px solid #e8ecf4;padding-top:12px;margin-top:12px}@media(max-width:760px){.top,.steps{display:block}.button{margin-top:12px}.step{margin-top:10px}.worker{grid-template-columns:1fr}.hide-mobile{display:none}}
:root{color:#dce8ff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 15% -10%,#163d77 0,transparent 32rem),radial-gradient(circle at 95% 5%,#123e39 0,transparent 29rem),#08111f}.page{max-width:1240px;margin:auto;padding:28px 22px 64px}.top{display:flex;align-items:flex-start;justify-content:space-between;gap:22px}.eyebrow{margin:0;color:#78a9ff;font-size:.77rem;font-weight:750;letter-spacing:.14em;text-transform:uppercase}.title{max-width:700px;margin:8px 0;font-size:clamp(2rem,5vw,3.6rem);line-height:1.04;letter-spacing:-.055em}.lead{max-width:690px;margin:0;color:#aabbd5;font-size:1.05rem}.button{display:inline-flex;align-items:center;gap:8px;border:0;border-radius:10px;padding:12px 16px;background:#4f8cff;color:#071224;font:inherit;font-weight:800;text-decoration:none;box-shadow:0 12px 30px #163d7766}.live{display:inline-flex;align-items:center;gap:7px;margin-top:18px;color:#8ba2c2;font-size:.87rem}.pulse{width:8px;height:8px;border-radius:50%;background:#5ee6a6;box-shadow:0 0 0 5px #5ee6a622}.summary{display:grid;grid-template-columns:1.4fr repeat(3,1fr);gap:13px;margin:32px 0}.panel,.metric,.flow-step,.job,.worker{border:1px solid #26415f;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #0000001f}.panel{padding:20px}.pipeline{display:grid;grid-template-columns:repeat(4,1fr);gap:10px;margin-top:16px}.flow-step{position:relative;padding:14px;min-height:116px}.flow-step:not(:last-child):after{content:"";position:absolute;top:35px;right:-10px;width:10px;height:2px;background:#3c5d83}.flow-step b{display:block;color:#f2f7ff}.flow-step small{display:block;margin-top:6px;color:#91a8c6}.flow-step .dot{display:inline-block;width:9px;height:9px;margin-right:7px;border-radius:50%;background:#5ee6a6}.metric{padding:16px}.metric b{display:block;margin-top:7px;color:#f5f8ff;font-size:2rem;line-height:1}.metric span{color:#9bb0cc;font-size:.84rem}.section-head{display:flex;align-items:baseline;justify-content:space-between;gap:12px;margin:36px 0 12px}.section-head h2{margin:0;color:#f3f7ff;font-size:1.18rem}.section-head p{margin:0;color:#8fa6c3;font-size:.9rem}.jobs{display:grid;gap:10px}.job{display:grid;grid-template-columns:minmax(210px,1.35fr) minmax(150px,.9fr) minmax(180px,1fr) auto;gap:18px;align-items:center;padding:17px 18px;text-decoration:none;color:inherit;transition:border-color .15s,transform .15s}.job:hover{border-color:#5d96ee;transform:translateY(-1px)}.job-name{color:#f3f7ff;font-weight:750}.job-id{margin-top:3px;color:#8196b3;font-family:ui-monospace,SFMono-Regular,monospace;font-size:.76rem}.badge{display:inline-flex;align-items:center;border-radius:999px;padding:4px 9px;font-size:.78rem;font-weight:800}.badge-waiting{background:#23344d;color:#b9cce9}.badge-active{background:#173d77;color:#9fc7ff}.badge-success{background:#123f34;color:#76efb5}.badge-danger{background:#552334;color:#ff9bad}.bar{height:7px;margin-top:8px;overflow:hidden;border-radius:999px;background:#20344e}.bar span{display:block;height:100%;border-radius:inherit;background:linear-gradient(90deg,#5a92ff,#61e1bd)}.numbers{color:#afc0d9;font-size:.88rem}.arrow{color:#79aaff;font-size:1.35rem}.workers{display:grid;grid-template-columns:repeat(3,1fr);gap:11px}.worker{padding:15px}.worker-head{display:flex;justify-content:space-between;gap:8px}.worker strong{color:#f3f7ff}.worker small,.worker p{color:#95a9c4}.worker p{margin:12px 0 0}.cap{display:inline-block;margin:4px 5px 0 0;border:1px solid #365576;border-radius:5px;padding:2px 6px;color:#a9c9f4;font:.75rem ui-monospace,SFMono-Regular,monospace}.empty{padding:30px;border:1px dashed #35516f;border-radius:14px;color:#9ab0cb;text-align:center}.offline{color:#faafbd}.sr{position:absolute;width:1px;height:1px;clip:rect(0,0,0,0);overflow:hidden;white-space:nowrap}@media(max-width:820px){.top,.section-head{display:block}.button{margin-top:18px}.summary{grid-template-columns:1fr 1fr}.panel{grid-column:span 2}.pipeline{grid-template-columns:1fr 1fr}.flow-step:not(:last-child):after{display:none}.job{grid-template-columns:1fr 1fr}.arrow{display:none}.workers{grid-template-columns:1fr 1fr}}@media(max-width:540px){.page{padding:22px 14px}.summary,.workers{grid-template-columns:1fr}.panel{grid-column:auto}.pipeline,.job{grid-template-columns:1fr}.title{font-size:2.35rem}}
</style>
</head>
<body>
<main class="page">
<header class="top"><div><p class="eyebrow">Local coordinator</p><h1>SciMesh operator dashboard</h1><p class="lead">See where a computation is and what should happen next.</p></div><a class="button" href="/ui/jobs/new">Start a check</a></header>
<section class="notice" aria-label="Pipeline flow"><strong>Similarity-search jobs produce a final CSV.</strong><span>Workers return shard-level candidates; after every shard succeeds, the coordinator deterministically merges them into one global top-k result.</span><div class="steps"><div class="step"><b>1. Upload TSV</b>The coordinator splits the file into shard tasks.</div><div class="step"><b>2. Wait for a worker</b>A worker claims a shard, calculates similarity, and returns a CSV.</div><div class="step"><b>3. Download result</b>When merging finishes, download the final CSV from the job page.</div></div></section>
<h2>Recent jobs</h2>
<div class="table-wrap"><table><tr><th>Computation</th><th>State</th><th>Progress</th><th class="hide-mobile">Created</th></tr>{{range .Jobs}}<tr><td><a href="/ui/jobs/{{.ID}}"><strong>{{workloadLabel .Workload}}</strong></a><br><small>Open job details</small></td><td><span class="status status-{{statusClass .Status}}">{{statusLabel .Status}}</span><br><small>{{statusHint .Status}}</small></td><td class="kicker"><strong>{{.Completed}} / {{.Total}}</strong> complete{{if gt .Failed 0}} · <span style="color:#a31135">failed: {{.Failed}}</span>{{end}}{{if gt .Cancelled 0}} · <span>stopped: {{.Cancelled}}</span>{{end}}<div class="bar"><span style="width:{{progressPercent .Completed .Failed .Cancelled .Total}}%"></span></div></td><td class="hide-mobile"><small>{{time .CreatedAt}}</small></td></tr>{{else}}<tr><td colspan="4" class="empty"><strong>No jobs yet.</strong><br>Click “Start a check”, upload a small TSV, and leave a worker running.</td></tr>{{end}}</table></div>
<h2>Workers</h2>
<section class="card">{{range .Workers}}<div class="worker"><div><strong>{{.Name}}</strong><br><small>{{.ID}}</small></div><div><span class="status status-{{if eq .Status "online"}}success{{else}}waiting{{end}}">{{workerStatusLabel .Status}}</span></div><div>{{range .Capabilities}}<code>{{.}}</code> {{end}}</div><div class="muted">Last signal<br>{{time .LastHeartbeatAt}}</div></div>{{else}}<div class="empty"><strong>No worker is registered yet.</strong><br>Run <code>scimesh-worker</code> with the coordinator URL and worker token.</div>{{end}}</section>
<header class="top">
<div><p class="eyebrow">Local scientific compute</p><h1 class="title">SciMesh control room</h1><p class="lead">Follow the real path from a molecular TSV to a globally reduced similarity result—without reading coordinator logs.</p><div class="live"><i class="pulse"></i><span id="refresh-state">Live overview · refreshes every 2 seconds</span></div></div>
<div style="display:flex;gap:10px;align-items:center;flex-wrap:wrap">{{if .Session}}<span class="live" style="margin-top:0">Signed in · {{.Session.Role}}</span>{{end}}{{if .Session}}<a class="button" href="/ui/profile" style="background:#23344d;color:#dce8ff;box-shadow:none">Profile</a>{{end}}{{if and .Session (eq .Session.Role "admin")}}<a class="button" href="/ui/admin" style="background:#23344d;color:#dce8ff;box-shadow:none">Admin</a>{{end}}{{if .Session}}<a class="button" href="/ui/workers/new" style="background:#23344d;color:#dce8ff;box-shadow:none">🖥 Add your machine</a>{{end}}<a class="button" href="/ui/jobs/new"> New similarity search</a>{{if .Session}}<form method="post" action="/ui/logout" style="margin:0"><button class="button" type="submit" style="background:#23344d;color:#dce8ff;box-shadow:none">Log out</button></form>{{end}}</div>
</header>
<section class="summary" aria-label="Pipeline summary">
<div class="panel"><strong>How a search becomes a result</strong><div class="pipeline"><div class="flow-step"><span><i class="dot"></i>01</span><b>Upload TSV</b><small>The coordinator validates and slices the dataset.</small></div><div class="flow-step"><span><i class="dot"></i>02</span><b>Run shards</b><small>Workers fingerprint molecules and return shard top-k CSVs.</small></div><div class="flow-step"><span><i class="dot"></i>03</span><b>Merge exactly</b><small>The coordinator ranks retained candidates deterministically.</small></div><div class="flow-step"><span><i class="dot"></i>04</span><b>Download CSV</b><small>A checksum-protected global result is ready.</small></div></div></div>
<div class="metric"><span>Active runs</span><b id="active-jobs">{{.ActiveJobs}}</b><small>waiting, running, or merging</small></div>
<div class="metric"><span>Available workers</span><b id="online-workers">{{.OnlineWorkers}}</b><small>recently registered</small></div>
<div class="metric"><span>Finished runs</span><b id="finished-jobs">{{.FinishedJobs}}</b><small>in the latest 20</small></div>
</section>
<section><div class="section-head"><h2>Recent computations</h2><p id="job-count">{{len .Jobs}} shown · newest first</p></div><div id="jobs" class="jobs">{{range .Jobs}}<a class="job" href="/ui/jobs/{{.ID}}"><div><div class="job-name">{{workloadLabel .Workload}}</div><div class="job-id">{{.ID}}</div></div><div><span class="badge badge-{{statusClass .Status}}">{{statusLabel .Status}}</span><div class="job-id">{{statusHint .Status}}</div></div><div><div class="numbers"><b>{{.Completed}}</b> / {{.Total}} shards complete{{if gt .Failed 0}} · <span class="offline">{{.Failed}} failed</span>{{end}}</div><div class="bar"><span style="width:{{progressPercent .Completed .Failed .Cancelled .Total}}%"></span></div></div><span class="arrow" aria-hidden="true"></span></a>{{else}}<div class="empty"><strong>No computations yet.</strong><br>Start a small similarity search, then keep one or more workers running to watch this dashboard come alive.</div>{{end}}</div></section>
{{if and .Session (ne .Session.Role "admin")}}<section><div class="section-head"><h2>My machines</h2><p>Workers you registered. <a href="/ui/workers/new" style="color:#79aaff">Add your machine →</a></p></div><div id="my-workers" class="workers">{{range .MyWorkers}}<article class="worker"><div class="worker-head"><strong>{{.Name}}</strong><span class="badge badge-{{workerStatusClass .Status}}">{{workerStatusLabel .Status}}</span></div><small>{{.ID}}</small><p>{{range .Capabilities}}<span class="cap">{{.}}</span>{{end}}</p><p>Last signal · {{time .LastHeartbeatAt}}</p></article>{{else}}<div class="empty"><strong>No machine of yours is connected.</strong><br><a href="/ui/workers/new" style="color:#79aaff">Turn this computer into a worker →</a></div>{{end}}</div></section>{{end}}
<section><div class="section-head"><h2>Worker fleet</h2><p>Workers register themselves; this page never controls their processes.</p></div><div id="workers" class="workers">{{range .Workers}}<article class="worker"><div class="worker-head"><strong>{{.Name}}</strong><span class="badge badge-{{workerStatusClass .Status}}">{{workerStatusLabel .Status}}</span></div><small>{{.ID}}</small><p>{{range .Capabilities}}<span class="cap">{{.}}</span>{{end}}</p><p>Last signal · {{time .LastHeartbeatAt}}</p></article>{{else}}<div class="empty"><strong>No worker is registered.</strong><br>Start <code>scimesh-worker</code> in another terminal, then return here.</div>{{end}}</div></section>
</main>
<script>
const statusInfo={pending:['Waiting for a worker','waiting'],leased:['Assigned to a worker','active'],running:['Running','active'],reducing:['Merging results','active'],completed:['Completed','success'],failed:['Needs attention','danger'],cancelled:['Stopped','waiting']};
const pct=j=>j.total?Math.min(100,Math.floor((j.completed+j.failed+j.cancelled)*100/j.total)):0;
const node=(tag,text,cls)=>{const n=document.createElement(tag);if(text!==undefined)n.textContent=text;if(cls)n.className=cls;return n};
const renderJobs=jobs=>{const box=document.querySelector('#jobs');box.replaceChildren();if(!jobs.length){const empty=node('div',undefined,'empty');empty.append(node('strong','No computations yet.'),document.createElement('br'),document.createTextNode('Start a small similarity search, then keep one or more workers running to watch this dashboard come alive.'));box.append(empty);return}for(const job of jobs){const info=statusInfo[job.status]||[job.status,'waiting'],link=node('a',undefined,'job');link.href='/ui/jobs/'+encodeURIComponent(job.id);const intro=node('div');intro.append(node('div',job.workload==='similarity-search'?'Molecule similarity search':job.workload,'job-name'),node('div',job.id,'job-id'));const state=node('div');state.append(node('span',info[0],'badge badge-'+info[1]),node('div',job.status==='reducing'?'Every shard is complete; coordinator is ranking the global top-k.':'Live coordinator state','job-id'));const progress=node('div'),numbers=node('div',undefined,'numbers');numbers.append(node('b',String(job.completed)),document.createTextNode(' / '+job.total+' shards complete'));if(job.failed){numbers.append(document.createTextNode(' · '),node('span',job.failed+' failed','offline'))}const bar=node('div',undefined,'bar'),fill=node('span');fill.style.width=pct(job)+'%';bar.append(fill);progress.append(numbers,bar);link.append(intro,state,progress,node('span','→','arrow'));box.append(link)}};
const workerCard=worker=>{const card=node('article',undefined,'worker'),head=node('div',undefined,'worker-head'),left=node('div'),workerInfo=worker.status==='online'?['Available','success']:worker.status==='busy'?['Busy','active']:['Offline','waiting'];left.append(node('strong',worker.name),node('small',worker.id));head.append(left,node('span',workerInfo[0],'badge badge-'+workerInfo[1]));const caps=node('p');for(const capability of worker.capabilities||[])caps.append(node('span',capability,'cap'));card.append(head,caps,node('p','Last signal · '+new Date(worker.last_heartbeat_at).toLocaleString()));return card};
const renderWorkers=workers=>{const box=document.querySelector('#workers');box.replaceChildren();if(!workers.length){const empty=node('div',undefined,'empty');empty.append(node('strong','No worker is registered.'),document.createElement('br'),document.createTextNode('Start scimesh-worker in another terminal, then return here.'));box.append(empty);return}for(const worker of workers)box.append(workerCard(worker))};
const renderMyWorkers=workers=>{const box=document.querySelector('#my-workers');if(!box)return;box.replaceChildren();if(!workers.length){const empty=node('div',undefined,'empty'),link=node('a','Turn this computer into a worker →');link.href='/ui/workers/new';link.style.color='#79aaff';empty.append(node('strong','No machine of yours is connected.'),document.createElement('br'),link);box.append(empty);return}for(const worker of workers)box.append(workerCard(worker))};
let timer;const refresh=async()=>{try{const response=await fetch('/ui/api/overview',{headers:{Accept:'application/json'}});if(!response.ok)throw Error();const view=await response.json();document.querySelector('#active-jobs').textContent=view.active_jobs;document.querySelector('#online-workers').textContent=view.online_workers;document.querySelector('#finished-jobs').textContent=view.finished_jobs;document.querySelector('#job-count').textContent=view.jobs.length+' shown · newest first';renderJobs(view.jobs);renderWorkers(view.workers);renderMyWorkers(view.my_workers||[]);document.querySelector('#refresh-state').textContent='Live overview · updated just now'}catch(_){document.querySelector('#refresh-state').textContent='Connection interrupted · trying again automatically'}};
const start=()=>{if(!timer&&!document.hidden)timer=setInterval(refresh,2000)};document.addEventListener('visibilitychange',()=>{if(document.hidden&&timer){clearInterval(timer);timer=undefined}else start()});start();
</script>
</body>
</html>
{{end}}
File diff suppressed because one or more lines are too long
@@ -0,0 +1,27 @@
{{define "login.html"}}
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Sign in · SciMesh</title>
<style>
:root{color:#e5efff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;min-height:100vh;display:grid;place-items:center;background:radial-gradient(circle at 10% -8%,#183f77 0,transparent 32rem),#08111f}a{color:#94bdff}.card{width:min(92vw,380px);border:1px solid #294662;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #00000021;padding:28px}.eyebrow{margin:0 0 4px;color:#7baaff;font-size:.78rem;font-weight:800;letter-spacing:.14em;text-transform:uppercase}h1{margin:0 0 18px;color:#f4f8ff;font-size:1.7rem;letter-spacing:-.03em}label{display:block;margin:14px 0 5px;color:#eaf2ff;font-weight:750}input{width:100%;border:1px solid #42617f;border-radius:9px;padding:10px 11px;background:#0a1626;color:#e5efff;font:inherit}input:focus{outline:2px solid #5d97f5;outline-offset:1px}.button{display:block;width:100%;margin-top:22px;border:0;border-radius:10px;padding:12px 16px;background:#67e3b8;color:#062018;font:inherit;font-weight:850;cursor:pointer}.error{margin:14px 0 0;color:#ffacba}.alt{margin:18px 0 0;color:#9fb3cf;font-size:.92rem}</style>
</head>
<body>
<main class="card">
<p class="eyebrow">SciMesh</p>
<h1>Sign in</h1>
<form method="post" action="/ui/login">
<label for="email">Email</label>
<input id="email" name="email" type="email" autocomplete="username" required autofocus>
<label for="password">Password</label>
<input id="password" name="password" type="password" autocomplete="current-password" required>
<button class="button" type="submit">Sign in</button>
</form>
{{if .Error}}<p class="error">{{.Error}}</p>{{end}}
<p class="alt">No account? <a href="/ui/register">Register</a></p>
</main>
</body>
</html>
{{end}}
@@ -4,27 +4,20 @@
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Create a check — SciMesh</title>
<title>New similarity search · SciMesh</title>
<style>
:root{color:#172033;background:#f6f8fc;font:16px/1.5 system-ui,sans-serif}body{margin:0}.page{max-width:760px;margin:auto;padding:32px 20px 56px}a{color:#174ecf}.back{text-decoration:none}.eyebrow{margin:28px 0 4px;color:#50617d;font-size:.86rem;font-weight:700;text-transform:uppercase;letter-spacing:.08em}h1{margin:0;font-size:2rem}.lead{color:#56657c}.notice{margin:22px 0;padding:16px 18px;border:1px solid #f2cb72;border-radius:10px;background:#fff8e6}.notice strong{display:block}.card{padding:22px;border:1px solid #dfe5f0;border-radius:12px;background:#fff}label{display:block;margin:18px 0 4px;font-weight:700}input{box-sizing:border-box;width:100%;padding:10px;border:1px solid #bac5d8;border-radius:7px;font:inherit}input[type=file]{padding:8px;background:#f8faff}.hint{margin:4px 0;color:#68758b;font-size:.9rem}.button{margin-top:22px;border:0;border-radius:8px;padding:11px 16px;background:#1f5eff;color:#fff;font:inherit;font-weight:700;cursor:pointer}.button:disabled{opacity:.6;cursor:wait}.error{margin-top:16px;color:#a31135}.working{margin-top:16px;color:#174ecf}.checklist{margin:8px 0;padding-left:20px;color:#56657c}.checklist li{margin:5px 0}
:root{color:#e5efff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 10% -8%,#183f77 0,transparent 32rem),#08111f}.page{max-width:980px;margin:auto;padding:28px 22px 64px}a{color:#94bdff}.back{text-decoration:none}.eyebrow{margin:28px 0 4px;color:#7baaff;font-size:.78rem;font-weight:800;letter-spacing:.14em;text-transform:uppercase}h1{margin:0;color:#f4f8ff;font-size:clamp(2rem,5vw,3.25rem);letter-spacing:-.055em}.lead{max-width:720px;margin:10px 0 0;color:#aabed9;font-size:1.06rem}.layout{display:grid;grid-template-columns:1.45fr .8fr;gap:15px;margin-top:28px}.card,.aside,.notice{border:1px solid #294662;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #00000021}.card{padding:22px}.aside,.notice{padding:18px}.aside h2,.notice h2{margin:0;color:#f1f6ff;font-size:1rem}.aside p,.notice p{color:#9fb3cf}.aside ol{margin:13px 0 0;padding-left:20px;color:#aebfda}.aside li{margin:10px 0}label{display:block;margin:18px 0 5px;color:#eaf2ff;font-weight:750}input,select{width:100%;border:1px solid #42617f;border-radius:9px;padding:10px 11px;background:#0a1626;color:#e5efff;font:inherit}input:focus,select:focus{outline:2px solid #5d97f5;outline-offset:1px}input[type=file]{padding:8px}.hint{margin:5px 0;color:#92a9c6;font-size:.87rem}.split{display:grid;grid-template-columns:1fr 1fr;gap:12px}.run-preview{margin-top:18px;border:1px solid #2c8c70;border-radius:11px;padding:13px;background:#0c2b2a;color:#a8f1d0}.run-preview strong{color:#e6fff4}.button{display:inline-flex;margin-top:22px;border:0;border-radius:10px;padding:12px 16px;background:#67e3b8;color:#062018;font:inherit;font-weight:850;cursor:pointer}.button:disabled{opacity:.6;cursor:wait}.working{margin:14px 0 0;color:#9fc5ff}.error{margin:12px 0 0;color:#ffacba}.hidden{display:none}.cap{display:inline-block;margin:4px 5px 0 0;border:1px solid #375978;border-radius:5px;padding:2px 6px;color:#b5d3f5;font:.76rem ui-monospace,SFMono-Regular,monospace}@media(max-width:720px){.layout,.split{grid-template-columns:1fr}.page{padding:22px 14px}}
</style>
</head>
<body>
<main class="page">
<a class="back" href="/ui">← Back to jobs</a><p class="eyebrow">Guided run</p><h1>Search for similar molecules</h1><p class="lead">Creates a diagnostic <code>similarity-search</code> job: a worker finds the top-k molecules most similar to a target SMILES.</p>
<section class="notice"><strong>Before starting</strong><ul class="checklist"><li>Keep at least one <code>scimesh-worker</code> running.</li><li>Use a small TSV for a hands-on check.</li><li><b>“Rows per shard” does not limit the file size.</b> It splits the entire upload into tasks: a full ChEMBL TSV at 1,000 rows per shard creates thousands of tasks.</li></ul></section>
<form id="run" class="card">
<label for="file">ChEMBL TSV</label><input id="file" type="file" name="file" required accept=".tsv,.txt,text/tab-separated-values"><p class="hint">Expected columns: <code>chembl_id</code> and <code>canonical_smiles</code>.</p>
<label for="query-smiles">Target molecule (SMILES)</label><input id="query-smiles" name="query_smiles" required maxlength="200" value="CCO" autocomplete="off"><p class="hint"><code>CCO</code> is ethanol. For gefitinib, use its SMILES here or the local CLI with <code>--query-id</code>.</p>
<label for="top-k">Matches to return</label><input id="top-k" name="top_k" type="number" min="1" max="100000" value="20" required><p class="hint">This is the top-k within each shard, not a global top-k for the whole dataset yet.</p>
<label for="chunk-rows">Rows per shard</label><input id="chunk-rows" name="chunk_rows" type="number" min="1" max="100000" value="1000" required><p class="hint">Fewer rows mean more tasks and more visible progress; more rows mean fewer, longer tasks.</p>
<label for="max-rows">Maximum dataset rows to process <small>(optional)</small></label><input id="max-rows" name="max_rows" type="number" min="1" max="10000000" placeholder="For example: 500"><p class="hint">Useful for a quick check of a large TSV. The coordinator creates shards from only the first N data rows; it still stores the original upload.</p>
<button class="button" id="submit" type="submit">Upload file and create job</button><p id="working" class="working" hidden aria-live="polite">Uploading the file and creating shard tasks… Keep this page open.</p><p id="error" class="error" role="alert"></p>
</form>
<a class="back" href="/ui">← Back to control room</a><p class="eyebrow">New computation</p><h1>Similarity search, end to end</h1><p class="lead">Upload a ChEMBL-style TSV. Workers calculate shard candidates; after every shard succeeds, SciMesh merges the exact global top-k into one final CSV.</p>
<div class="layout"><form id="run" class="card" novalidate><label for="file">ChEMBL TSV</label><input id="file" type="file" name="file" required accept=".tsv,.txt,text/tab-separated-values"><p class="hint">Required columns: <code>chembl_id</code> and <code>canonical_smiles</code>.</p><label for="query-smiles">Target molecule (SMILES)</label><input id="query-smiles" name="query_smiles" required maxlength="200" value="CCO" autocomplete="off"><p class="hint">Use a valid SMILES. The coordinator shares this exact query with every shard.</p><div class="split"><div><label for="top-k">Global top-k</label><input id="top-k" name="top_k" type="number" min="1" max="100000" value="20" required><p class="hint">How many final molecules to retain.</p></div><div><label for="chunk-rows">Rows per shard</label><input id="chunk-rows" name="chunk_rows" type="number" min="1" max="100000" value="1000" required><p class="hint">Smaller shards make more visible tasks.</p></div></div><div class="split"><div><label for="threshold">Similarity threshold <small>(optional)</small></label><input id="threshold" name="threshold" type="number" min="0" max="1" step="0.01" placeholder="For example: 0.70"><p class="hint">Leave blank to rank every valid candidate.</p></div><div><label for="direction">Keep molecules</label><select id="direction" name="threshold_direction"><option value="greater">more similar (≥ threshold)</option><option value="less">less similar (≤ threshold)</option></select><p class="hint">“Less” helps explore dissimilar molecules.</p></div></div><label for="max-rows">Maximum dataset rows <small>(optional quick run)</small></label><input id="max-rows" name="max_rows" type="number" min="1" max="10000000" placeholder="For example: 500"><p class="hint">Only the first N data rows become shards; the original upload remains stored by the coordinator.</p><div id="preview" class="run-preview"><strong>Ready to plan a run.</strong><br>Select a TSV to see the file that will be sent to the coordinator.</div><button class="button" id="submit" type="submit">Create pipeline run →</button><p id="working" class="working hidden" aria-live="polite">Uploading TSV and creating coordinator-owned shards…</p><p id="error" class="error" role="alert"></p></form><aside class="aside"><h2>What you will observe</h2><ol><li><strong>Input accepted</strong><br>TSV is validated and split into durable shard artifacts.</li><li><strong>Workers claim tasks</strong><br>Each worker downloads one shard, fingerprints it, and uploads a partial CSV.</li><li><strong>Global reduction</strong><br>The coordinator compares exact scores from all partial results.</li><li><strong>Final download</strong><br>The result page exposes a checksum-protected global CSV.</li></ol><h2 style="margin-top:26px">Before you submit</h2><p>Keep at least one worker running in another terminal. The browser cannot start or control worker processes.</p><p><span class="cap">similarity-search</span> is currently the only distributed workload available here.</p></aside></div>
</main>
<script>
const form=document.querySelector('#run'),button=document.querySelector('#submit'),working=document.querySelector('#working'),error=document.querySelector('#error');
form.addEventListener('submit',async event=>{event.preventDefault();error.textContent='';const fields=new FormData(form),file=fields.get('file'),maxRows=String(fields.get('max_rows')||'').trim();if(!(file instanceof File)||file.size===0){error.textContent='Choose a non-empty TSV file.';return}const parameters={query_smiles:fields.get('query_smiles'),top_k:Number(fields.get('top_k')),progress_every:0},upload=new FormData();upload.append('workload','similarity-search');upload.append('parameters',JSON.stringify(parameters));upload.append('chunk_rows',fields.get('chunk_rows'));if(maxRows)upload.append('max_rows',maxRows);upload.append('file',file,file.name);button.disabled=true;working.hidden=false;try{const response=await fetch('/ui/api/jobs/upload',{method:'POST',body:upload}),data=await response.json();if(!response.ok)throw Error(data.error||'Unable to create the job.');location.href='/ui/jobs/'+data.job_id}catch(err){error.textContent=err.message==='invalid input'?'Check the TSV and fields: the coordinator could not accept this request.':err.message;button.disabled=false;working.hidden=true}});
const form=document.querySelector('#run'),button=document.querySelector('#submit'),working=document.querySelector('#working'),error=document.querySelector('#error'),preview=document.querySelector('#preview'),fileInput=document.querySelector('#file');
const bytes=size=>size<1024?size+' B':size<1024*1024?(size/1024).toFixed(1)+' KiB':(size/(1024*1024)).toFixed(1)+' MiB';fileInput.addEventListener('change',()=>{const file=fileInput.files&&fileInput.files[0];preview.replaceChildren();if(!file){preview.append(document.createTextNode('Select a TSV to see the file that will be sent to the coordinator.'));return}const strong=document.createElement('strong');strong.textContent='Source ready: '+file.name;preview.append(strong,document.createElement('br'),document.createTextNode(bytes(file.size)+' · the coordinator will validate its header before creating tasks.'))});
form.addEventListener('submit',async event=>{event.preventDefault();error.textContent='';const fields=new FormData(form),file=fields.get('file'),query=String(fields.get('query_smiles')||'').trim(),topK=Number(fields.get('top_k')),chunkRows=Number(fields.get('chunk_rows')),threshold=String(fields.get('threshold')||'').trim(),maxRows=String(fields.get('max_rows')||'').trim();if(!(file instanceof File)||file.size===0){error.textContent='Choose a non-empty TSV file.';return}if(!query||query.length>200||!Number.isInteger(topK)||topK<1||!Number.isInteger(chunkRows)||chunkRows<1){error.textContent='Enter a target SMILES, a positive global top-k, and a positive rows-per-shard value.';return}if(threshold&&(Number.isNaN(Number(threshold))||Number(threshold)<0||Number(threshold)>1)){error.textContent='Similarity threshold must be between 0 and 1.';return}const parameters={query_smiles:query,top_k:topK,threshold_direction:fields.get('threshold_direction'),progress_every:0};if(threshold)parameters.threshold=Number(threshold);const upload=new FormData();upload.append('workload','similarity-search');upload.append('parameters',JSON.stringify(parameters));upload.append('chunk_rows',String(chunkRows));if(maxRows)upload.append('max_rows',maxRows);upload.append('file',file,file.name);button.disabled=true;working.classList.remove('hidden');try{const response=await fetch('/ui/api/jobs/upload',{method:'POST',body:upload}),data=await response.json();if(!response.ok)throw Error(data.error||'Unable to create the job.');location.href='/ui/jobs/'+encodeURIComponent(data.job_id)}catch(err){error.textContent=err.message==='invalid input'?'The coordinator could not accept this run. Check the TSV columns and form values.':err.message;button.disabled=false;working.classList.add('hidden')}});
</script>
</body>
</html>
@@ -0,0 +1,32 @@
{{define "profile.html"}}
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Profile · SciMesh</title>
<style>
:root{color:#e5efff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 10% -8%,#183f77 0,transparent 32rem),#08111f}.page{max-width:720px;margin:auto;padding:28px 22px 64px}a{color:#94bdff}.top{display:flex;justify-content:space-between;align-items:center;gap:12px}.eyebrow{margin:0;color:#7baaff;font-size:.78rem;font-weight:800;letter-spacing:.14em;text-transform:uppercase}h1{margin:6px 0 0;color:#f4f8ff;font-size:clamp(1.8rem,4vw,2.6rem);letter-spacing:-.04em}.btn{border:0;border-radius:9px;padding:10px 14px;font:inherit;font-weight:800;cursor:pointer;text-decoration:none}.btn-muted{background:#23344d;color:#dce8ff}.card{margin-top:24px;border:1px solid #294662;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #00000021;padding:6px 22px}.err{margin-top:24px;border-radius:10px;padding:12px 14px;background:#552334;color:#ff9bad;font-weight:700}.row{display:flex;justify-content:space-between;gap:16px;padding:15px 0;border-bottom:1px solid #1d3350}.row:last-child{border-bottom:0}.k{color:#9fb3cf}.v{color:#f2f7ff;font-weight:700;text-align:right;word-break:break-all}.mono{font-family:ui-monospace,SFMono-Regular,monospace;font-size:.9rem}.pill{display:inline-block;border-radius:999px;padding:3px 10px;font-size:.82rem;font-weight:800}.pill-yes{background:#123f34;color:#76efb5}.pill-no{background:#23344d;color:#b9cce9}.hint{margin-top:14px;color:#8ba2c2;font-size:.86rem}</style>
</head>
<body>
<main class="page">
<header class="top">
<div><p class="eyebrow">Account</p><h1>Your profile</h1></div>
<div style="display:flex;gap:10px;align-items:center"><a href="/ui">← Dashboard</a><form method="post" action="/ui/logout" style="margin:0"><button class="btn btn-muted" type="submit">Log out</button></form></div>
</header>
{{if .Error}}<div class="err">{{.Error}}</div>{{end}}
{{with .Profile}}
<section class="card">
<div class="row"><span class="k">User id</span><span class="v mono">{{.ID}}</span></div>
<div class="row"><span class="k">Email</span><span class="v">{{.Email}}</span></div>
<div class="row"><span class="k">Role</span><span class="v">{{.Role}}</span></div>
<div class="row"><span class="k">Verified contributor</span><span class="v">{{if .Verified}}<span class="pill pill-yes">yes</span>{{else}}<span class="pill pill-no">no</span>{{end}}</span></div>
<div class="row"><span class="k">Member since</span><span class="v mono">{{.CreatedAt}}</span></div>
</section>
<p class="hint">Your user id is what the coordinator stores as the owner of every job you submit. Give it to an admin to be promoted or verified.</p>
{{end}}
</main>
</body>
</html>
{{end}}
@@ -0,0 +1,28 @@
{{define "register.html"}}
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Register · SciMesh</title>
<style>
:root{color:#e5efff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;min-height:100vh;display:grid;place-items:center;background:radial-gradient(circle at 10% -8%,#183f77 0,transparent 32rem),#08111f}a{color:#94bdff}.card{width:min(92vw,380px);border:1px solid #294662;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #00000021;padding:28px}.eyebrow{margin:0 0 4px;color:#7baaff;font-size:.78rem;font-weight:800;letter-spacing:.14em;text-transform:uppercase}h1{margin:0 0 18px;color:#f4f8ff;font-size:1.7rem;letter-spacing:-.03em}label{display:block;margin:14px 0 5px;color:#eaf2ff;font-weight:750}input{width:100%;border:1px solid #42617f;border-radius:9px;padding:10px 11px;background:#0a1626;color:#e5efff;font:inherit}input:focus{outline:2px solid #5d97f5;outline-offset:1px}.hint{margin:5px 0 0;color:#92a9c6;font-size:.85rem}.button{display:block;width:100%;margin-top:22px;border:0;border-radius:10px;padding:12px 16px;background:#67e3b8;color:#062018;font:inherit;font-weight:850;cursor:pointer}.error{margin:14px 0 0;color:#ffacba}.alt{margin:18px 0 0;color:#9fb3cf;font-size:.92rem}</style>
</head>
<body>
<main class="card">
<p class="eyebrow">SciMesh</p>
<h1>Create account</h1>
<form method="post" action="/ui/register">
<label for="email">Email</label>
<input id="email" name="email" type="email" autocomplete="username" required autofocus>
<label for="password">Password</label>
<input id="password" name="password" type="password" autocomplete="new-password" minlength="8" maxlength="72" required>
<p class="hint">At least 8 characters.</p>
<button class="button" type="submit">Register</button>
</form>
{{if .Error}}<p class="error">{{.Error}}</p>{{end}}
<p class="alt">Already have an account? <a href="/ui/login">Sign in</a></p>
</main>
</body>
</html>
{{end}}
+56 -1
View File
@@ -26,6 +26,7 @@ var uiTemplates = template.Must(template.New("ui").Funcs(template.FuncMap{
"taskErrorLabel": uiTaskErrorLabel,
"taskErrorHint": uiTaskErrorHint,
"workerStatusLabel": uiWorkerStatusLabel,
"workerStatusClass": uiWorkerStatusClass,
"workloadLabel": uiWorkloadLabel,
"progressPercent": uiProgressPercent,
"cancellable": uiCancellable,
@@ -101,6 +102,8 @@ func uiWorkerStatusLabel(status string) string {
switch status {
case "online":
return "Available"
case "busy":
return "Busy"
case "offline":
return "Offline"
default:
@@ -108,6 +111,17 @@ func uiWorkerStatusLabel(status string) string {
}
}
func uiWorkerStatusClass(status string) string {
switch status {
case "online":
return "success"
case "busy":
return "active"
default:
return "waiting"
}
}
// uiTaskErrorLabel deliberately maps worker implementation errors to an
// operator-facing diagnosis. Raw subprocess commands and local paths belong in
// the worker terminal, not in the web UI.
@@ -207,6 +221,20 @@ func (s *Server) handleUIHome(w http.ResponseWriter, r *http.Request) {
s.renderUI(w, "dashboard.html", view)
}
// handleUIOverviewJSON is the bounded polling projection used by the operator
// dashboard. It intentionally returns only the safe UI read model, never
// worker tokens, storage keys, or database entities.
func (s *Server) handleUIOverviewJSON(w http.ResponseWriter, r *http.Request) {
ctx, cancel := s.reqCtx(r)
defer cancel()
view, err := s.uc.Dashboard.Overview(ctx, 20)
if err != nil {
s.writeError(w, r, err)
return
}
writeJSON(w, http.StatusOK, view)
}
func (s *Server) handleUINewJob(w http.ResponseWriter, r *http.Request) {
s.renderUI(w, "new-job.html", nil)
}
@@ -257,7 +285,7 @@ func (s *Server) handleUIArtifactDownload(w http.ResponseWriter, r *http.Request
}
ctx, cancel := s.reqCtx(r)
defer cancel()
belongs, err := s.uc.Dashboard.ArtifactBelongsToJob(ctx, jobID, artifactID)
belongs, err := s.uc.Dashboard.DownloadableArtifactBelongsToJob(ctx, jobID, artifactID)
if err != nil {
s.writeError(w, r, err)
return
@@ -283,3 +311,30 @@ func (s *Server) handleUIArtifactDownload(w http.ResponseWriter, r *http.Request
w.Header().Set("X-Checksum-SHA256", art.SHA256)
_, _ = io.Copy(w, body)
}
// handleUIArtifactPreview renders a bounded CSV preview. Its use case owns
// the job-scoped access rule, including the requirement that a final artifact
// is the persisted result of a completed job.
func (s *Server) handleUIArtifactPreview(w http.ResponseWriter, r *http.Request) {
jobID, ok := s.uiJobID(w, r)
if !ok {
return
}
artifactID, err := uuid.Parse(r.PathValue("artifact_id"))
if err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
if s.uc.PreviewArtifact == nil {
http.NotFound(w, r)
return
}
ctx, cancel := s.reqCtx(r)
defer cancel()
view, err := s.uc.PreviewArtifact.Execute(ctx, jobID, artifactID)
if err != nil {
s.writeError(w, r, err)
return
}
s.renderUI(w, "artifact-preview.html", view)
}
@@ -0,0 +1,114 @@
package http
import (
"context"
"io"
"net/http"
"net/url"
"strings"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
)
// adminUserActions are the userservice endpoints the admin panel may invoke, by
// their path suffix. A whitelist so a crafted form can never proxy an arbitrary
// path.
var adminUserActions = map[string]bool{
"promote": true,
"demote": true,
"verify": true,
"unverify": true,
}
// requireAdmin gates a route on the session caller being an admin. It runs
// inside withUISession, which has already stamped the requester. A non-admin is
// sent back to the dashboard rather than shown the panel.
func requireAdmin(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if req, ok := authctx.From(r.Context()); !ok || !req.IsAdmin() {
http.Redirect(w, r, "/ui", http.StatusSeeOther)
return
}
next.ServeHTTP(w, r)
})
}
func (s *Server) handleUIAdmin(w http.ResponseWriter, r *http.Request) {
role := ""
if req, ok := authctx.From(r.Context()); ok {
role = req.Role
}
s.renderUI(w, "admin.html", map[string]any{
"Role": role,
"Msg": r.URL.Query().Get("msg"),
"Error": r.URL.Query().Get("error"),
})
}
// handleUIAdminUserAction proxies a user-management action to the userservice,
// forwarding the admin's session token so the userservice re-checks the role.
// The user id and action come from the form, so a single static form action can
// drive every operation.
func (s *Server) handleUIAdminUserAction(w http.ResponseWriter, r *http.Request) {
userID := strings.TrimSpace(r.FormValue("user_id"))
action := r.FormValue("action")
if !adminUserActions[action] {
http.Redirect(w, r, "/ui/admin?error=unknown+action", http.StatusSeeOther)
return
}
if _, err := uuid.Parse(userID); err != nil {
http.Redirect(w, r, "/ui/admin?error=invalid+user+id", http.StatusSeeOther)
return
}
c, err := r.Cookie(sessionCookie)
if err != nil {
redirectToLogin(w, r)
return
}
status, _, err := s.callUserserviceAuthed(r.Context(), http.MethodPost, "/users/"+userID+"/"+action, c.Value)
if err != nil {
s.log.Error("admin action proxy", "err", err, "action", action)
http.Redirect(w, r, "/ui/admin?error=service+unavailable", http.StatusSeeOther)
return
}
switch status {
case http.StatusNoContent:
http.Redirect(w, r, "/ui/admin?msg="+url.QueryEscape(action+" applied"), http.StatusSeeOther)
case http.StatusNotFound:
http.Redirect(w, r, "/ui/admin?error=user+not+found", http.StatusSeeOther)
case http.StatusForbidden, http.StatusUnauthorized:
http.Redirect(w, r, "/ui/admin?error=not+authorized", http.StatusSeeOther)
default:
http.Redirect(w, r, "/ui/admin?error=action+failed", http.StatusSeeOther)
}
}
// callUserserviceAuthed makes an authenticated call to the userservice, passing
// the caller's JWT through as a bearer token. Used for admin actions; login and
// registration use the unauthenticated callUserservice.
func (s *Server) callUserserviceAuthed(ctx context.Context, method, path, bearer string) (int, []byte, error) {
// path is not attacker-controlled: the caller composes it only from a
// uuid-validated id and an action from a fixed whitelist, and the host is
// the operator-configured userservice — so the SSRF taint gosec sees here
// cannot reach an arbitrary destination.
req, err := http.NewRequestWithContext(ctx, method, s.userserviceURL+path, nil) //nolint:gosec // G704: path is validated, host is config
if err != nil {
return 0, nil, err
}
req.Header.Set("Authorization", "Bearer "+bearer)
resp, err := s.httpClient.Do(req) //nolint:gosec // G704: see above
if err != nil {
return 0, nil, err
}
defer func() { _ = resp.Body.Close() }()
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
if err != nil {
return 0, nil, err
}
return resp.StatusCode, body, nil
}
@@ -0,0 +1,113 @@
package http
import (
"context"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"testing"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
)
func adminReq(t *testing.T, role string) *http.Request {
t.Helper()
req := newReq(http.MethodGet, "/ui/admin", nil)
return req.WithContext(authctx.With(context.Background(), authctx.Requester{UserID: uuid.New(), Role: role}))
}
func TestRequireAdminAllowsAdminOnly(t *testing.T) {
reached := false
h := requireAdmin(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { reached = true }))
// Admin passes through.
h.ServeHTTP(httptest.NewRecorder(), adminReq(t, "admin"))
if !reached {
t.Error("admin must reach the handler")
}
// Plain user is redirected to the dashboard.
reached = false
rec := httptest.NewRecorder()
h.ServeHTTP(rec, adminReq(t, "user"))
if reached {
t.Error("non-admin must not reach the handler")
}
if rec.Code != http.StatusSeeOther || rec.Header().Get("Location") != "/ui" {
t.Errorf("non-admin got %d -> %q, want 303 -> /ui", rec.Code, rec.Header().Get("Location"))
}
}
func TestAdminUserActionForwardsBearer(t *testing.T) {
targetID := uuid.NewString()
var gotAuth, gotPath string
stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotAuth = r.Header.Get("Authorization")
gotPath = r.URL.Path
w.WriteHeader(http.StatusNoContent)
}))
defer stub.Close()
s := newLoginServer(stub)
req := newReq(http.MethodPost, "/ui/admin/user-action",
strings.NewReader(url.Values{"user_id": {targetID}, "action": {"promote"}}.Encode()))
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "admin.jwt.token"})
rec := httptest.NewRecorder()
s.handleUIAdminUserAction(rec, req)
if gotAuth != "Bearer admin.jwt.token" {
t.Errorf("forwarded auth = %q, want the admin bearer", gotAuth)
}
if gotPath != "/users/"+targetID+"/promote" {
t.Errorf("forwarded path = %q", gotPath)
}
if rec.Code != http.StatusSeeOther || !strings.Contains(rec.Header().Get("Location"), "msg=") {
t.Errorf("got %d -> %q, want 303 with a success msg", rec.Code, rec.Header().Get("Location"))
}
}
func TestAdminUserActionRejectsUnknownAction(t *testing.T) {
s := newLoginServer(httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {
t.Fatal("userservice must not be called for an invalid action")
})))
req := newReq(http.MethodPost, "/ui/admin/user-action",
strings.NewReader(url.Values{"user_id": {uuid.NewString()}, "action": {"delete"}}.Encode()))
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "x"})
rec := httptest.NewRecorder()
s.handleUIAdminUserAction(rec, req)
if !strings.Contains(rec.Header().Get("Location"), "error=") {
t.Errorf("unknown action redirect = %q, want an error", rec.Header().Get("Location"))
}
}
func TestAdminUserActionRejectsBadID(t *testing.T) {
s := newLoginServer(httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {
t.Fatal("userservice must not be called for an invalid id")
})))
req := newReq(http.MethodPost, "/ui/admin/user-action",
strings.NewReader(url.Values{"user_id": {"not-a-uuid"}, "action": {"promote"}}.Encode()))
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "x"})
rec := httptest.NewRecorder()
s.handleUIAdminUserAction(rec, req)
if !strings.Contains(rec.Header().Get("Location"), "error=") {
t.Errorf("bad id redirect = %q, want an error", rec.Header().Get("Location"))
}
}
func TestDashboardAdminLinkOnlyForAdmin(t *testing.T) {
admin := render(t, "dashboard.html", usecase.DashboardView{Session: &usecase.SessionView{Role: "admin"}})
if !strings.Contains(admin, "/ui/admin") {
t.Error("admin must see the Admin link")
}
user := render(t, "dashboard.html", usecase.DashboardView{Session: &usecase.SessionView{Role: "user"}})
if strings.Contains(user, "/ui/admin") {
t.Error("a plain user must not see the Admin link")
}
}
@@ -0,0 +1,171 @@
package http
import (
"bytes"
"context"
"encoding/json"
"io"
"net/http"
"time"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
tokenpkg "github.com/emil28092005/SciMesh/coordinator/internal/token"
)
// sessionCookie holds the userservice JWT for the operator UI. It is httpOnly so
// page scripts cannot read the token, and scoped to /ui so it never rides along
// with worker API calls.
const sessionCookie = "scimesh_session"
// withUISession gates the operator UI on a valid userservice session cookie.
// A missing or invalid token redirects to the login page rather than returning
// 401, because the caller here is a browser, not an API client. On success it
// stamps the requester so downstream handlers can scope views by owner.
func withUISession(v tokenVerifier) func(http.Handler) http.Handler {
return func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
c, err := r.Cookie(sessionCookie)
if err != nil || c.Value == "" {
redirectToLogin(w, r)
return
}
claims, err := v.Verify(c.Value)
if err != nil {
// Expired or tampered: drop the stale cookie and re-authenticate.
clearSessionCookie(w, r)
redirectToLogin(w, r)
return
}
ctx := authctx.With(r.Context(), authctx.Requester{
UserID: claims.UserID,
Role: claims.Role,
Verified: claims.Verified,
})
next.ServeHTTP(w, r.WithContext(ctx))
})
}
}
// tokenVerifier is satisfied by *token.Verifier; taking an interface keeps the
// UI auth testable with a stub.
type tokenVerifier interface {
Verify(raw string) (tokenpkg.Claims, error)
}
func (s *Server) handleUILoginForm(w http.ResponseWriter, r *http.Request) {
s.renderUI(w, "login.html", map[string]any{"Error": r.URL.Query().Get("error")})
}
func (s *Server) handleUIRegisterForm(w http.ResponseWriter, r *http.Request) {
s.renderUI(w, "register.html", map[string]any{"Error": r.URL.Query().Get("error")})
}
// handleUILogin exchanges the submitted credentials for a userservice token and
// stores it in the session cookie. The coordinator never sees or stores the
// password beyond forwarding it once.
func (s *Server) handleUILogin(w http.ResponseWriter, r *http.Request) {
email, password := r.FormValue("email"), r.FormValue("password")
status, body, err := s.callUserservice(r.Context(), "/login", email, password)
if err != nil {
s.log.Error("userservice login call", "err", err)
http.Redirect(w, r, "/ui/login?error=service+unavailable", http.StatusSeeOther)
return
}
if status != http.StatusOK {
http.Redirect(w, r, "/ui/login?error=invalid+email+or+password", http.StatusSeeOther)
return
}
var resp struct {
Token string `json:"token"`
}
if err := json.Unmarshal(body, &resp); err != nil || resp.Token == "" {
http.Redirect(w, r, "/ui/login?error=service+unavailable", http.StatusSeeOther)
return
}
setSessionCookie(w, r, resp.Token)
http.Redirect(w, r, "/ui", http.StatusSeeOther)
}
// handleUIRegister creates an account through the userservice, then sends the
// user to the login page. The new account is a plain user until an admin
// promotes or verifies it.
func (s *Server) handleUIRegister(w http.ResponseWriter, r *http.Request) {
email, password := r.FormValue("email"), r.FormValue("password")
status, _, err := s.callUserservice(r.Context(), "/register", email, password)
if err != nil {
s.log.Error("userservice register call", "err", err)
http.Redirect(w, r, "/ui/register?error=service+unavailable", http.StatusSeeOther)
return
}
switch status {
case http.StatusCreated:
http.Redirect(w, r, "/ui/login?error=registered,+please+log+in", http.StatusSeeOther)
case http.StatusConflict:
http.Redirect(w, r, "/ui/register?error=email+already+registered", http.StatusSeeOther)
default:
http.Redirect(w, r, "/ui/register?error=invalid+email+or+password", http.StatusSeeOther)
}
}
func (s *Server) handleUILogout(w http.ResponseWriter, r *http.Request) {
clearSessionCookie(w, r)
http.Redirect(w, r, "/ui/login", http.StatusSeeOther)
}
// callUserservice POSTs credentials to the userservice and returns its status
// and body. It is the only runtime dependency on the userservice — login and
// registration; token verification stays local.
func (s *Server) callUserservice(ctx context.Context, path, email, password string) (int, []byte, error) {
payload, _ := json.Marshal(map[string]string{"email": email, "password": password})
req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.userserviceURL+path, bytes.NewReader(payload))
if err != nil {
return 0, nil, err
}
req.Header.Set("Content-Type", "application/json")
resp, err := s.httpClient.Do(req)
if err != nil {
return 0, nil, err
}
defer func() { _ = resp.Body.Close() }()
// Cap the response; login/register bodies are tiny.
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
if err != nil {
return 0, nil, err
}
return resp.StatusCode, body, nil
}
func setSessionCookie(w http.ResponseWriter, r *http.Request, token string) {
// Secure is set under TLS; a local demo runs plain HTTP, where forcing
// Secure would stop the browser from ever sending the cookie back.
http.SetCookie(w, &http.Cookie{ //nolint:gosec // G124: Secure follows r.TLS by design
Name: sessionCookie,
Value: token,
Path: "/ui",
HttpOnly: true,
Secure: r.TLS != nil,
SameSite: http.SameSiteLaxMode,
Expires: time.Now().Add(24 * time.Hour),
})
}
func clearSessionCookie(w http.ResponseWriter, r *http.Request) {
http.SetCookie(w, &http.Cookie{ //nolint:gosec // G124: Secure follows r.TLS by design
Name: sessionCookie,
Value: "",
Path: "/ui",
HttpOnly: true,
Secure: r.TLS != nil,
SameSite: http.SameSiteLaxMode,
MaxAge: -1,
})
}
func redirectToLogin(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, "/ui/login", http.StatusSeeOther)
}
@@ -0,0 +1,175 @@
package http
import (
"context"
"errors"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"testing"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
tokenpkg "github.com/emil28092005/SciMesh/coordinator/internal/token"
)
// newReq builds a request carrying a context, which http.NewRequestWithContext
// provides on go1.22 (httptest.NewRequestWithContext needs go1.23).
func newReq(method, target string, body io.Reader) *http.Request {
req, err := http.NewRequestWithContext(context.Background(), method, target, body)
if err != nil {
panic(err)
}
return req
}
type stubVerifier struct {
claims tokenpkg.Claims
err error
}
func (s stubVerifier) Verify(string) (tokenpkg.Claims, error) { return s.claims, s.err }
func TestWithUISessionRedirectsWithoutCookie(t *testing.T) {
h := withUISession(stubVerifier{})(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {
t.Fatal("handler must not run without a session")
}))
rec := httptest.NewRecorder()
h.ServeHTTP(rec, newReq(http.MethodGet, "/ui", nil))
if rec.Code != http.StatusSeeOther {
t.Fatalf("got %d, want 303", rec.Code)
}
if loc := rec.Header().Get("Location"); loc != "/ui/login" {
t.Errorf("redirect = %q, want /ui/login", loc)
}
}
func TestWithUISessionAcceptsValidCookieAndStampsRequester(t *testing.T) {
id := uuid.New()
verifier := stubVerifier{claims: tokenpkg.Claims{UserID: id, Role: "admin", Verified: true}}
var gotReq authctx.Requester
var ok bool
h := withUISession(verifier)(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) {
gotReq, ok = authctx.From(r.Context())
}))
req := newReq(http.MethodGet, "/ui", nil)
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "valid.jwt"})
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if !ok || gotReq.UserID != id || gotReq.Role != "admin" || !gotReq.Verified {
t.Errorf("requester = %+v (ok=%v), want id=%v admin verified", gotReq, ok, id)
}
}
func TestWithUISessionClearsInvalidCookie(t *testing.T) {
h := withUISession(stubVerifier{err: errors.New("expired")})(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {
t.Fatal("handler must not run with an invalid token")
}))
req := newReq(http.MethodGet, "/ui", nil)
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "stale.jwt"})
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code != http.StatusSeeOther {
t.Fatalf("got %d, want 303", rec.Code)
}
if c := rec.Result().Cookies(); len(c) == 0 || c[0].MaxAge >= 0 {
t.Error("stale cookie must be cleared (MaxAge < 0)")
}
}
// newLoginServer builds a Server whose userservice calls hit stub.
func newLoginServer(stub *httptest.Server) *Server {
return &Server{
log: slog.New(slog.NewTextHandler(io.Discard, nil)),
userserviceURL: strings.TrimRight(stub.URL, "/"),
httpClient: stub.Client(),
}
}
func postForm(path string, form url.Values) *http.Request {
req := newReq(http.MethodPost, path, strings.NewReader(form.Encode()))
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
return req
}
func TestHandleUILoginSetsCookieOnSuccess(t *testing.T) {
stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/login" {
t.Errorf("unexpected path %q", r.URL.Path)
}
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"token":"issued.jwt.here"}`))
}))
defer stub.Close()
s := newLoginServer(stub)
rec := httptest.NewRecorder()
s.handleUILogin(rec, postForm("/ui/login", url.Values{"email": {"a@b.com"}, "password": {"password123"}}))
if rec.Code != http.StatusSeeOther || rec.Header().Get("Location") != "/ui" {
t.Fatalf("got %d -> %q, want 303 -> /ui", rec.Code, rec.Header().Get("Location"))
}
cookies := rec.Result().Cookies()
if len(cookies) == 0 || cookies[0].Name != sessionCookie || cookies[0].Value != "issued.jwt.here" {
t.Errorf("session cookie not set: %+v", cookies)
}
if !cookies[0].HttpOnly {
t.Error("session cookie must be httpOnly")
}
}
func TestHandleUILoginRejectsBadCredentials(t *testing.T) {
stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusUnauthorized)
}))
defer stub.Close()
s := newLoginServer(stub)
rec := httptest.NewRecorder()
s.handleUILogin(rec, postForm("/ui/login", url.Values{"email": {"a@b.com"}, "password": {"wrong"}}))
if rec.Code != http.StatusSeeOther || !strings.HasPrefix(rec.Header().Get("Location"), "/ui/login?error=") {
t.Fatalf("got %d -> %q, want 303 -> /ui/login?error=", rec.Code, rec.Header().Get("Location"))
}
if len(rec.Result().Cookies()) != 0 {
t.Error("no cookie must be set on failed login")
}
}
func TestHandleUIRegisterConflict(t *testing.T) {
stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusConflict)
}))
defer stub.Close()
s := newLoginServer(stub)
rec := httptest.NewRecorder()
s.handleUIRegister(rec, postForm("/ui/register", url.Values{"email": {"dup@b.com"}, "password": {"password123"}}))
if got := rec.Header().Get("Location"); !strings.Contains(got, "already+registered") {
t.Errorf("register conflict redirect = %q", got)
}
}
func TestHandleUILogoutClearsCookie(t *testing.T) {
s := &Server{log: slog.New(slog.NewTextHandler(io.Discard, nil))}
rec := httptest.NewRecorder()
s.handleUILogout(rec, newReq(http.MethodPost, "/ui/logout", nil))
if rec.Header().Get("Location") != "/ui/login" {
t.Errorf("logout redirect = %q", rec.Header().Get("Location"))
}
c := rec.Result().Cookies()
if len(c) == 0 || c[0].MaxAge >= 0 {
t.Error("logout must clear the session cookie")
}
}
@@ -0,0 +1,42 @@
package http
import (
"bytes"
"strings"
"testing"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
)
func render(t *testing.T, name string, data any) string {
t.Helper()
var buf bytes.Buffer
if err := uiTemplates.ExecuteTemplate(&buf, name, data); err != nil {
t.Fatalf("render %s: %v", name, err)
}
return buf.String()
}
func TestDashboardLogoutOnlyInSession(t *testing.T) {
withSession := render(t, "dashboard.html", usecase.DashboardView{Session: &usecase.SessionView{Role: "admin"}})
if !strings.Contains(withSession, "/ui/logout") || !strings.Contains(withSession, "Log out") {
t.Error("dashboard must show a logout control in session mode")
}
noSession := render(t, "dashboard.html", usecase.DashboardView{})
if strings.Contains(noSession, "/ui/logout") {
t.Error("dashboard must not show logout under basic auth (no session)")
}
}
func TestJobLogoutOnlyInSession(t *testing.T) {
withSession := render(t, "job.html", usecase.JobDetailView{Session: &usecase.SessionView{Role: "user"}})
if !strings.Contains(withSession, "/ui/logout") {
t.Error("job page must show a logout control in session mode")
}
noSession := render(t, "job.html", usecase.JobDetailView{})
if strings.Contains(noSession, "/ui/logout") {
t.Error("job page must not show logout under basic auth (no session)")
}
}
@@ -0,0 +1,49 @@
package http
import (
"encoding/json"
"net/http"
)
// profileView is the account data shown on the profile page, mirroring the
// userservice /me response.
type profileView struct {
ID string `json:"id"`
Email string `json:"email"`
Role string `json:"role"`
Verified bool `json:"verified"`
CreatedAt string `json:"created_at"`
}
// handleUIProfile shows the signed-in user's own account. It proxies the
// session token to the userservice /me endpoint, which is the authority on the
// account (email and created_at are not in the JWT).
func (s *Server) handleUIProfile(w http.ResponseWriter, r *http.Request) {
c, err := r.Cookie(sessionCookie)
if err != nil {
redirectToLogin(w, r)
return
}
status, body, err := s.callUserserviceAuthed(r.Context(), http.MethodGet, "/me", c.Value)
if err != nil {
s.log.Error("profile /me proxy", "err", err)
s.renderUI(w, "profile.html", map[string]any{"Error": "userservice unavailable"})
return
}
if status == http.StatusUnauthorized {
clearSessionCookie(w, r)
redirectToLogin(w, r)
return
}
if status != http.StatusOK {
s.renderUI(w, "profile.html", map[string]any{"Error": "could not load your account"})
return
}
var p profileView
if err := json.Unmarshal(body, &p); err != nil {
s.renderUI(w, "profile.html", map[string]any{"Error": "could not read your account"})
return
}
s.renderUI(w, "profile.html", map[string]any{"Profile": p})
}
@@ -0,0 +1,43 @@
package http
import (
"net/http"
"net/http/httptest"
"strings"
"testing"
)
func TestProfileProxiesMe(t *testing.T) {
var gotAuth, gotPath string
stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotAuth, gotPath = r.Header.Get("Authorization"), r.URL.Path
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"id":"11111111-1111-1111-1111-111111111111","email":"me@example.com","role":"user","verified":false,"created_at":"2026-07-26T00:00:00Z"}`))
}))
defer stub.Close()
s := newLoginServer(stub)
req := newReq(http.MethodGet, "/ui/profile", nil)
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "my.jwt"})
rec := httptest.NewRecorder()
s.handleUIProfile(rec, req)
if gotAuth != "Bearer my.jwt" || gotPath != "/me" {
t.Fatalf("proxy: auth=%q path=%q", gotAuth, gotPath)
}
body := rec.Body.String()
if !strings.Contains(body, "me@example.com") || !strings.Contains(body, "11111111-1111-1111-1111-111111111111") {
t.Error("profile page must show the email and id")
}
}
func TestProfileRedirectsWithoutCookie(t *testing.T) {
s := newLoginServer(httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {
t.Fatal("must not call userservice without a session")
})))
rec := httptest.NewRecorder()
s.handleUIProfile(rec, newReq(http.MethodGet, "/ui/profile", nil))
if rec.Code != http.StatusSeeOther {
t.Errorf("no cookie: got %d, want 303 redirect", rec.Code)
}
}
@@ -43,3 +43,12 @@ func TestUITaskErrorPresentationDoesNotExposeCommand(t *testing.T) {
t.Error("error hint must explain the failure")
}
}
func TestUIWorkerStatusPresentation(t *testing.T) {
if got := uiWorkerStatusLabel("busy"); got != "Busy" {
t.Errorf("busy worker label = %q", got)
}
if got := uiWorkerStatusClass("busy"); got != "active" {
t.Errorf("busy worker class = %q", got)
}
}
@@ -0,0 +1,118 @@
package http
import (
"bytes"
"context"
"io"
"net/http"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
)
// handleUIAddWorker renders the "add your machine" page: instructions, the
// user's existing worker keys, and a ready-to-run command carrying a freshly
// minted key. All key operations happen client-side against the JSON endpoints
// below; this handler only supplies the browser-facing URLs.
func (s *Server) handleUIAddWorker(w http.ResponseWriter, r *http.Request) {
data := map[string]any{
"CoordinatorURL": s.publicCoordinatorURL,
"UserserviceURL": s.publicUserserviceURL,
}
if req, ok := authctx.From(r.Context()); ok {
data["Session"] = &usecase.SessionView{Role: req.Role, Verified: req.Verified}
}
s.renderUI(w, "add-worker.html", data)
}
// handleUIWorkerKeysList proxies the caller's live worker keys from the
// userservice, forwarding their session token.
func (s *Server) handleUIWorkerKeysList(w http.ResponseWriter, r *http.Request) {
c, err := r.Cookie(sessionCookie)
if err != nil {
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "not signed in"})
return
}
status, body, err := s.callUserserviceAuthed(r.Context(), http.MethodGet, "/worker-keys", c.Value)
if err != nil {
s.log.Error("worker-keys list proxy", "err", err)
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "userservice unavailable"})
return
}
proxyJSON(w, status, body)
}
// handleUIWorkerKeyCreate mints a new worker key via the userservice and returns
// its response — including the one-time plaintext key — straight to the browser.
func (s *Server) handleUIWorkerKeyCreate(w http.ResponseWriter, r *http.Request) {
c, err := r.Cookie(sessionCookie)
if err != nil {
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "not signed in"})
return
}
body, _ := io.ReadAll(io.LimitReader(r.Body, 1<<12))
if len(body) == 0 {
body = []byte("{}")
}
status, respBody, err := s.callUserserviceAuthedBody(r.Context(), http.MethodPost, "/worker-keys", c.Value, body)
if err != nil {
s.log.Error("worker-key create proxy", "err", err)
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "userservice unavailable"})
return
}
proxyJSON(w, status, respBody)
}
// handleUIWorkerKeyRevoke retires one of the caller's keys via the userservice.
// The id is validated as a UUID so the proxied path can never be attacker-shaped.
func (s *Server) handleUIWorkerKeyRevoke(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
if _, err := uuid.Parse(id); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid worker key id"})
return
}
c, err := r.Cookie(sessionCookie)
if err != nil {
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "not signed in"})
return
}
status, _, err := s.callUserserviceAuthed(r.Context(), http.MethodDelete, "/worker-keys/"+id, c.Value)
if err != nil {
s.log.Error("worker-key revoke proxy", "err", err)
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "userservice unavailable"})
return
}
w.WriteHeader(status)
}
// proxyJSON forwards a userservice JSON response verbatim, preserving its status.
func proxyJSON(w http.ResponseWriter, status int, body []byte) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_, _ = w.Write(body)
}
// callUserserviceAuthedBody is callUserserviceAuthed with a JSON request body,
// used for the create call. Kept separate so the bodyless admin/profile callers
// stay unchanged.
func (s *Server) callUserserviceAuthedBody(ctx context.Context, method, path, bearer string, body []byte) (int, []byte, error) {
req, err := http.NewRequestWithContext(ctx, method, s.userserviceURL+path, bytes.NewReader(body)) //nolint:gosec // G704: path is a fixed literal, host is config
if err != nil {
return 0, nil, err
}
req.Header.Set("Authorization", "Bearer "+bearer)
req.Header.Set("Content-Type", "application/json")
resp, err := s.httpClient.Do(req) //nolint:gosec // G704: see above
if err != nil {
return 0, nil, err
}
defer func() { _ = resp.Body.Close() }()
respBody, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
if err != nil {
return 0, nil, err
}
return resp.StatusCode, respBody, nil
}
@@ -0,0 +1,106 @@
package http
import (
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
func TestWorkerKeyCreateProxiesWithBody(t *testing.T) {
var gotAuth, gotPath, gotMethod, gotBody string
stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotAuth, gotPath, gotMethod = r.Header.Get("Authorization"), r.URL.Path, r.Method
b, _ := io.ReadAll(r.Body)
gotBody = string(b)
w.WriteHeader(http.StatusCreated)
_, _ = w.Write([]byte(`{"id":"11111111-1111-1111-1111-111111111111","name":"box","prefix":"scimesh_wk_live_ab","created_at":"2026-07-26T00:00:00Z","key":"scimesh_wk_live_secret"}`))
}))
defer stub.Close()
s := newLoginServer(stub)
req := newReq(http.MethodPost, "/ui/api/worker-keys", strings.NewReader(`{"name":"box"}`))
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "my.jwt"})
rec := httptest.NewRecorder()
s.handleUIWorkerKeyCreate(rec, req)
if gotAuth != "Bearer my.jwt" || gotPath != "/worker-keys" || gotMethod != http.MethodPost {
t.Fatalf("proxy: auth=%q path=%q method=%q", gotAuth, gotPath, gotMethod)
}
if !strings.Contains(gotBody, `"name":"box"`) {
t.Errorf("request body not forwarded: %q", gotBody)
}
if rec.Code != http.StatusCreated || !strings.Contains(rec.Body.String(), "scimesh_wk_live_secret") {
t.Errorf("response not passed through: %d %s", rec.Code, rec.Body.String())
}
}
func TestWorkerKeysListProxies(t *testing.T) {
stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet || r.URL.Path != "/worker-keys" {
t.Errorf("unexpected upstream call %s %s", r.Method, r.URL.Path)
}
_, _ = w.Write([]byte(`{"worker_keys":[{"id":"1","name":"box","prefix":"scimesh_wk_live_ab","created_at":"2026-07-26T00:00:00Z"}]}`))
}))
defer stub.Close()
s := newLoginServer(stub)
req := newReq(http.MethodGet, "/ui/api/worker-keys", nil)
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "my.jwt"})
rec := httptest.NewRecorder()
s.handleUIWorkerKeysList(rec, req)
if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "worker_keys") {
t.Errorf("list not passed through: %d %s", rec.Code, rec.Body.String())
}
}
func TestWorkerKeyRevokeProxiesDelete(t *testing.T) {
const id = "22222222-2222-2222-2222-222222222222"
var gotPath, gotMethod string
stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotPath, gotMethod = r.URL.Path, r.Method
w.WriteHeader(http.StatusNoContent)
}))
defer stub.Close()
s := newLoginServer(stub)
req := newReq(http.MethodPost, "/ui/api/worker-keys/"+id+"/revoke", nil)
req.SetPathValue("id", id)
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "my.jwt"})
rec := httptest.NewRecorder()
s.handleUIWorkerKeyRevoke(rec, req)
if gotMethod != http.MethodDelete || gotPath != "/worker-keys/"+id {
t.Fatalf("proxy: method=%q path=%q", gotMethod, gotPath)
}
if rec.Code != http.StatusNoContent {
t.Errorf("revoke status = %d, want 204", rec.Code)
}
}
func TestWorkerKeyRevokeRejectsBadID(t *testing.T) {
s := newLoginServer(httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {
t.Fatal("must not call userservice for an invalid id")
})))
req := newReq(http.MethodPost, "/ui/api/worker-keys/not-a-uuid/revoke", nil)
req.SetPathValue("id", "not-a-uuid")
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "my.jwt"})
rec := httptest.NewRecorder()
s.handleUIWorkerKeyRevoke(rec, req)
if rec.Code != http.StatusBadRequest {
t.Errorf("bad id: got %d, want 400", rec.Code)
}
}
func TestWorkerKeysRequireSession(t *testing.T) {
s := newLoginServer(httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {
t.Fatal("must not call userservice without a session cookie")
})))
rec := httptest.NewRecorder()
s.handleUIWorkerKeysList(rec, newReq(http.MethodGet, "/ui/api/worker-keys", nil))
if rec.Code != http.StatusUnauthorized {
t.Errorf("no cookie: got %d, want 401", rec.Code)
}
}
+7
View File
@@ -4,6 +4,8 @@ import (
"io"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
)
// Use-case boundary types. Adapters map their wire formats onto these, so the
@@ -28,6 +30,11 @@ type ChunkInput struct {
type RegisterWorkerInput struct {
Name string
Capabilities []string
// OwnerID is the userservice user registering this worker; nil for a
// shared-token registration. TrustLevel is resolved by the transport layer
// from how the caller authenticated.
OwnerID *uuid.UUID
TrustLevel domain.WorkerTrust
}
type ClaimTaskInput struct {
+7
View File
@@ -53,6 +53,7 @@ func (uc *CreateJob) Execute(ctx context.Context, in CreateJobInput) (*domain.Jo
if err != nil {
return nil, err
}
job.OwnerID = ownerFromContext(ctx)
err = uc.tx.WithinTx(ctx, func(ctx context.Context) error {
if err := uc.jobs.Insert(ctx, job); err != nil {
@@ -97,6 +98,9 @@ func (uc *CancelJob) Execute(ctx context.Context, jobID uuid.UUID) (int64, error
if err != nil {
return err
}
if err := authorizeJobAccess(ctx, job); err != nil {
return err
}
if job.Status == domain.JobCancelled {
return nil
}
@@ -132,6 +136,9 @@ func (uc *GetJobStatus) Execute(ctx context.Context, jobID uuid.UUID) (domain.Jo
if err != nil {
return domain.JobProgress{}, err
}
if err := authorizeJobAccess(ctx, job); err != nil {
return domain.JobProgress{}, err
}
counts, err := uc.tasks.CountByStatus(ctx, jobID)
if err != nil {
return domain.JobProgress{}, err
+51
View File
@@ -0,0 +1,51 @@
package usecase
import (
"context"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
)
// ownerFromContext returns the authenticated user id to stamp on a new job, or
// nil when the request was not authenticated as a user — worker or legacy
// traffic, or user-JWT auth disabled. A nil owner is stored as NULL.
func ownerFromContext(ctx context.Context) *uuid.UUID {
if r, ok := authctx.From(ctx); ok {
id := r.UserID
return &id
}
return nil
}
// uiOwnerFilter returns the owner a UI listing must be restricted to: nil for an
// operator/admin or an unauthenticated (basic-auth) session, which see all jobs,
// or the caller's id for a plain user, who sees only their own.
func uiOwnerFilter(ctx context.Context) *uuid.UUID {
r, ok := authctx.From(ctx)
if !ok || r.IsAdmin() {
return nil
}
id := r.UserID
return &id
}
// authorizeJobAccess enforces that a non-admin user may only act on their own
// job. It returns ErrJobNotFound — not a 403 — on a mismatch, so the response
// never reveals that another user's job exists.
//
// Requests with no authenticated user (worker/legacy traffic, or JWT auth
// disabled) are not restricted here: the shared service token already gated
// them, and worker endpoints legitimately operate across all jobs.
func authorizeJobAccess(ctx context.Context, job *domain.Job) error {
r, ok := authctx.From(ctx)
if !ok || r.IsAdmin() {
return nil
}
if job.OwnerID == nil || *job.OwnerID != r.UserID {
return domain.ErrJobNotFound
}
return nil
}
@@ -0,0 +1,66 @@
package usecase
import (
"context"
"errors"
"testing"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
)
func TestOwnerFromContext(t *testing.T) {
if ownerFromContext(context.Background()) != nil {
t.Error("no requester must yield a nil owner")
}
id := uuid.New()
ctx := authctx.With(context.Background(), authctx.Requester{UserID: id, Role: "user"})
got := ownerFromContext(ctx)
if got == nil || *got != id {
t.Errorf("owner = %v, want %v", got, id)
}
}
func TestAuthorizeJobAccess(t *testing.T) {
owner := uuid.New()
other := uuid.New()
job := &domain.Job{ID: uuid.New(), OwnerID: &owner}
ctxOf := func(id uuid.UUID, role string) context.Context {
return authctx.With(context.Background(), authctx.Requester{UserID: id, Role: role})
}
cases := []struct {
name string
ctx context.Context
wantErr bool
}{
{"no requester (worker/legacy) allowed", context.Background(), false},
{"owner allowed", ctxOf(owner, "user"), false},
{"admin allowed", ctxOf(other, "admin"), false},
{"non-owner denied", ctxOf(other, "user"), true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
err := authorizeJobAccess(tc.ctx, job)
if tc.wantErr {
if !errors.Is(err, domain.ErrJobNotFound) {
t.Errorf("got %v, want ErrJobNotFound", err)
}
} else if err != nil {
t.Errorf("unexpected error: %v", err)
}
})
}
}
func TestAuthorizeJobAccessNilOwner(t *testing.T) {
// A legacy job with no owner must not be readable by an arbitrary user.
job := &domain.Job{ID: uuid.New(), OwnerID: nil}
ctx := authctx.With(context.Background(), authctx.Requester{UserID: uuid.New(), Role: "user"})
if err := authorizeJobAccess(ctx, job); !errors.Is(err, domain.ErrJobNotFound) {
t.Errorf("got %v, want ErrJobNotFound", err)
}
}
+9
View File
@@ -23,6 +23,15 @@ type ClaimFilter struct {
Owner string // worker ID taking the lease
Now time.Time
LeaseUntil time.Time
// VoterOwner, when set, excludes tasks this owner has already voted on, so
// an untrusted worker never verifies its own chunk twice.
VoterOwner *uuid.UUID
}
// TaskResultRepository records and tallies quorum votes for untrusted results.
type TaskResultRepository interface {
RecordVote(ctx context.Context, taskID, ownerID uuid.UUID, sha256 string, artifactID uuid.UUID) error
CountAgreeing(ctx context.Context, taskID uuid.UUID, sha256 string) (int, error)
}
// TaskRepository persists tasks.
+151
View File
@@ -0,0 +1,151 @@
package usecase
import (
"context"
"encoding/csv"
"errors"
"io"
"mime"
"strings"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
)
// The preview is deliberately a diagnostic aid, never a full artifact
// viewer. These limits bound both memory use and storage reads.
const (
previewMaxRows = 30
previewMaxBytes = 64 * 1024
)
// ArtifactPreviewView is the safe, bounded data rendered by the operator UI.
// It deliberately contains neither storage keys nor worker-local details.
type ArtifactPreviewView struct {
JobID string
ArtifactID string
Filename string
Diagnostic bool
Previewable bool
Reason string
Headers []string
Rows [][]string
Truncated bool
RowLimit int
ByteLimit int64
}
// PreviewArtifact reads the beginning of a job-scoped CSV result. Partial
// results are diagnostic; a final result is available only after the reducer
// has persisted it as this job's completed result.
type PreviewArtifact struct {
read UIReadRepository
blobs BlobStore
}
func NewPreviewArtifact(read UIReadRepository, blobs BlobStore) *PreviewArtifact {
return &PreviewArtifact{read: read, blobs: blobs}
}
func (p *PreviewArtifact) Execute(ctx context.Context, jobID, artifactID uuid.UUID) (ArtifactPreviewView, error) {
job, err := p.read.GetJob(ctx, jobID)
if err != nil {
return ArtifactPreviewView{}, err
}
// Another user's job (and not admin): report not-found, matching the
// artifact-absent response so nothing about it leaks.
if err := authorizeJobAccess(ctx, job); err != nil {
return ArtifactPreviewView{}, domain.ErrArtifactNotFound
}
artifacts, err := p.read.ListArtifactsByJob(ctx, jobID)
if err != nil {
return ArtifactPreviewView{}, err
}
var artifact *domain.Artifact
for i := range artifacts {
if artifacts[i].ID == artifactID {
artifact = &artifacts[i]
break
}
}
if artifact == nil || !previewableArtifact(*job, *artifact) {
// Use one response for an unknown artifact, another job's artifact, and
// an artifact that is not yet public. This avoids leaking its state.
return ArtifactPreviewView{}, domain.ErrArtifactNotFound
}
view := ArtifactPreviewView{
JobID: jobID.String(),
ArtifactID: artifact.ID.String(),
Filename: artifact.Filename,
Diagnostic: artifact.Kind == domain.ArtifactPartialResult,
RowLimit: previewMaxRows,
ByteLimit: previewMaxBytes,
}
if !isCSVArtifact(artifact) {
view.Reason = "This artifact is not a CSV file, so it cannot be shown as text here. Download it instead."
return view, nil
}
if artifact.SizeBytes == 0 {
view.Reason = "This artifact is empty."
return view, nil
}
body, err := p.blobs.Open(ctx, artifact.StorageKey)
if err != nil {
return ArtifactPreviewView{}, err
}
defer func() { _ = body.Close() }()
limited := &io.LimitedReader{R: body, N: previewMaxBytes}
reader := csv.NewReader(limited)
reader.FieldsPerRecord = -1 // a byte limit may end inside a record
headers, err := reader.Read()
if err != nil {
view.Reason = "This artifact could not be read as CSV."
return view, nil
}
view.Headers = append([]string(nil), headers...)
view.Rows = make([][]string, 0, previewMaxRows)
for len(view.Rows) < previewMaxRows {
record, readErr := reader.Read()
if readErr != nil {
if !errors.Is(readErr, io.EOF) {
view.Truncated = true
}
break
}
view.Rows = append(view.Rows, append([]string(nil), record...))
}
if artifact.SizeBytes > previewMaxBytes {
view.Truncated = true
} else if len(view.Rows) == previewMaxRows {
if _, readErr := reader.Read(); readErr == nil {
view.Truncated = true
}
}
view.Previewable = true
return view, nil
}
func previewableArtifact(job domain.Job, artifact domain.Artifact) bool {
if artifact.Kind == domain.ArtifactPartialResult {
return true
}
return artifact.Kind == domain.ArtifactFinalResult &&
job.Status == domain.JobCompleted &&
job.ResultArtifactID != nil &&
*job.ResultArtifactID == artifact.ID
}
func isCSVArtifact(artifact *domain.Artifact) bool {
mediaType, _, err := mime.ParseMediaType(artifact.ContentType)
if err == nil && strings.EqualFold(mediaType, "text/csv") {
return true
}
return strings.HasSuffix(strings.ToLower(artifact.Filename), ".csv")
}
@@ -0,0 +1,121 @@
package usecase_test
import (
"context"
"errors"
"strconv"
"strings"
"testing"
"time"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/memstore"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
)
func newPreviewHarness() (*usecase.PreviewArtifact, *memstore.JobRepo, *memstore.ArtifactRepo, *memstore.BlobStore) {
jobs := memstore.NewJobRepo()
tasks := memstore.NewTaskRepo()
workers := memstore.NewWorkerRepo()
artifacts := memstore.NewArtifactRepo()
blobs := memstore.NewBlobStore()
return usecase.NewPreviewArtifact(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts), blobs), jobs, artifacts, blobs
}
func previewJob(t *testing.T, jobs *memstore.JobRepo, status domain.JobStatus) uuid.UUID {
t.Helper()
job := &domain.Job{ID: uuid.New(), Workload: "similarity-search", Status: status, CreatedAt: time.Now().UTC()}
if err := jobs.Insert(context.Background(), job); err != nil {
t.Fatalf("insert preview job: %v", err)
}
return job.ID
}
func previewArtifact(t *testing.T, artifacts *memstore.ArtifactRepo, blobs *memstore.BlobStore, jobID uuid.UUID, kind domain.ArtifactKind, filename, contentType, contents string) uuid.UUID {
t.Helper()
id := uuid.New()
sha, size, err := blobs.Put(context.Background(), id.String(), strings.NewReader(contents))
if err != nil {
t.Fatalf("store preview artifact: %v", err)
}
artifact := &domain.Artifact{ID: id, JobID: jobID, Kind: kind, Filename: filename, StorageKey: id.String(), ContentType: contentType, SizeBytes: size, SHA256: sha, CreatedAt: time.Now().UTC()}
if err := artifacts.Insert(context.Background(), artifact); err != nil {
t.Fatalf("insert preview artifact: %v", err)
}
return id
}
func TestPreviewArtifactRendersBoundedCSV(t *testing.T) {
preview, jobs, artifacts, blobs := newPreviewHarness()
jobID := previewJob(t, jobs, domain.JobRunning)
var csv strings.Builder
csv.WriteString("chembl_id,score\n")
for i := 0; i < 40; i++ {
csv.WriteString("CHEMBL" + strconv.Itoa(i) + ",0.9\n")
}
artifactID := previewArtifact(t, artifacts, blobs, jobID, domain.ArtifactPartialResult, "partial.csv", "text/csv; charset=utf-8", csv.String())
view, err := preview.Execute(context.Background(), jobID, artifactID)
if err != nil {
t.Fatalf("preview: %v", err)
}
if !view.Previewable || !view.Diagnostic || !view.Truncated || len(view.Rows) != 30 || view.Headers[0] != "chembl_id" || view.Rows[0][0] != "CHEMBL0" {
t.Fatalf("unexpected preview: %+v", view)
}
}
func TestPreviewArtifactCapsStorageReadAndHandlesInvalidCSV(t *testing.T) {
preview, jobs, artifacts, blobs := newPreviewHarness()
jobID := previewJob(t, jobs, domain.JobRunning)
// Fewer than 30 oversized records force the byte cap, rather than the row
// cap, to stop parsing.
large := "id,value\n" + strings.Repeat("row,"+strings.Repeat("x", 5*1024)+"\n", 20)
largeID := previewArtifact(t, artifacts, blobs, jobID, domain.ArtifactPartialResult, "large.csv", "text/csv", large)
view, err := preview.Execute(context.Background(), jobID, largeID)
if err != nil || !view.Previewable || !view.Truncated || len(view.Rows) > 30 {
t.Fatalf("large preview = (%+v, %v)", view, err)
}
invalidID := previewArtifact(t, artifacts, blobs, jobID, domain.ArtifactPartialResult, "broken.csv", "text/csv", "\"unterminated")
invalid, err := preview.Execute(context.Background(), jobID, invalidID)
if err != nil || invalid.Previewable || invalid.Reason == "" {
t.Fatalf("invalid preview = (%+v, %v)", invalid, err)
}
}
func TestPreviewArtifactRejectsOtherJobsAndNonResults(t *testing.T) {
preview, jobs, artifacts, blobs := newPreviewHarness()
jobA := previewJob(t, jobs, domain.JobRunning)
jobB := previewJob(t, jobs, domain.JobRunning)
partialID := previewArtifact(t, artifacts, blobs, jobA, domain.ArtifactPartialResult, "partial.csv", "text/csv", "a,b\n1,2\n")
if _, err := preview.Execute(context.Background(), jobB, partialID); !errors.Is(err, domain.ErrArtifactNotFound) {
t.Fatalf("cross-job preview error = %v", err)
}
inputID := previewArtifact(t, artifacts, blobs, jobA, domain.ArtifactInput, "input.csv", "text/csv", "a,b\n1,2\n")
if _, err := preview.Execute(context.Background(), jobA, inputID); !errors.Is(err, domain.ErrArtifactNotFound) {
t.Fatalf("input preview error = %v", err)
}
}
func TestPreviewArtifactExposesOnlyPersistedCompletedFinalResult(t *testing.T) {
preview, jobs, artifacts, blobs := newPreviewHarness()
jobID := previewJob(t, jobs, domain.JobReducing)
finalID := previewArtifact(t, artifacts, blobs, jobID, domain.ArtifactFinalResult, "final.csv", "text/csv", "rank,chembl_id\n1,CHEMBL1\n")
if _, err := preview.Execute(context.Background(), jobID, finalID); !errors.Is(err, domain.ErrArtifactNotFound) {
t.Fatalf("uncompleted final preview error = %v", err)
}
if err := jobs.CompleteWithResult(context.Background(), jobID, finalID, time.Now().UTC()); err != nil {
t.Fatal(err)
}
view, err := preview.Execute(context.Background(), jobID, finalID)
if err != nil || !view.Previewable || view.Diagnostic {
t.Fatalf("completed final preview = (%+v, %v)", view, err)
}
if err := jobs.FailReduction(context.Background(), jobID, "reducer_failed", "final result reduction failed", time.Now().UTC()); err != nil {
t.Fatal(err)
}
if _, err := preview.Execute(context.Background(), jobID, finalID); !errors.Is(err, domain.ErrArtifactNotFound) {
t.Fatalf("failed reducer preview error = %v", err)
}
}
+3
View File
@@ -126,6 +126,9 @@ func (uc *GetJobResult) Execute(ctx context.Context, jobID uuid.UUID) (*domain.A
if err != nil {
return nil, nil, err
}
if err := authorizeJobAccess(ctx, job); err != nil {
return nil, nil, err
}
if job.Status != domain.JobCompleted || job.ResultArtifactID == nil {
return nil, nil, domain.ErrArtifactNotFound
}
+120 -15
View File
@@ -2,10 +2,12 @@ package usecase
import (
"context"
"errors"
"time"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
)
@@ -46,11 +48,28 @@ func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.Cl
return nil, domain.ErrInvalidInput
}
workloads := in.Workloads
var voterOwner *uuid.UUID
if workerID, err := uuid.Parse(in.WorkerID); err == nil {
worker, err := uc.workers.Get(ctx, workerID)
if err != nil {
return nil, err
}
// Bind the caller to the worker it claims as. A JWT-authenticated
// volunteer may operate only its own workers; without this the trust
// tier would be read off a caller-supplied worker_id, letting anyone who
// knows a trusted worker's id claim as it. A shared-token caller (no
// requester) is a lab operator and may act as any worker.
if r, ok := authctx.From(ctx); ok {
if worker.OwnerID == nil || *worker.OwnerID != r.UserID {
// Don't disclose that another user's worker exists.
return nil, domain.ErrWorkerNotFound
}
}
// An untrusted volunteer may claim, but never a chunk its owner has
// already voted on — so quorum needs genuinely independent computations.
if worker.TrustLevel == domain.WorkerUntrusted {
voterOwner = worker.OwnerID
}
// Never trust caller-supplied capabilities: registration is the durable
// worker identity and its allowlist.
workloads = worker.Capabilities
@@ -71,6 +90,7 @@ func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.Cl
Owner: in.WorkerID,
Now: now,
LeaseUntil: now.Add(uc.leaseDuration),
VoterOwner: voterOwner,
})
if err != nil {
return err
@@ -141,13 +161,22 @@ type CompleteTask struct {
tasks TaskRepository
jobs JobRepository
artifacts ArtifactRepository
workers WorkerRepository
results TaskResultRepository
tx TxManager
clock Clock
// quorum is how many distinct owners must agree on an untrusted result
// before it is accepted; a trusted worker's result is accepted directly.
quorum int
}
func NewCompleteTask(tasks TaskRepository, jobs JobRepository, artifacts ArtifactRepository,
tx TxManager, clock Clock) *CompleteTask {
return &CompleteTask{tasks: tasks, jobs: jobs, artifacts: artifacts, tx: tx, clock: clock}
workers WorkerRepository, results TaskResultRepository, tx TxManager, clock Clock, quorum int) *CompleteTask {
if quorum < 1 {
quorum = 2
}
return &CompleteTask{tasks: tasks, jobs: jobs, artifacts: artifacts, workers: workers,
results: results, tx: tx, clock: clock, quorum: quorum}
}
// Execute applies the result and, when that was the job's last outstanding
@@ -166,24 +195,35 @@ func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*dom
}
// Rule 10: never trust a worker-supplied artifact reference. The result
// must be an artifact the coordinator itself stored for *this* task.
if err := uc.verifyResultArtifact(ctx, in.TaskID, in.Attempt, in.ResultArtifactID); err != nil {
art, err := uc.verifyResultArtifact(ctx, in.TaskID, in.Attempt, in.ResultArtifactID)
if err != nil {
return err
}
trusted, ownerID, err := uc.workerTrust(ctx, in.WorkerID)
if err != nil {
return err
}
now := uc.clock.Now()
// Untrusted (volunteer) worker: record a vote and only complete once a
// quorum of distinct owners agree; otherwise return the task to the queue.
if !trusted {
return uc.recordVote(ctx, task, in, art, ownerID, now, &out)
}
// Trusted worker (lab token, verified, or admin): accept directly.
before := task.Version
if err := task.CompleteWith(in.ResultArtifactID, in.Metrics,
in.WorkerID, in.Attempt, now); err != nil {
if err := task.CompleteWith(in.ResultArtifactID, in.Metrics, in.WorkerID, in.Attempt, now); err != nil {
return err
}
out = task
// A replay of an already-recorded result leaves the entity untouched.
// Writing anyway would fail the optimistic-concurrency guard (the stored
// version already equals ours) and turn an idempotent call into a 409.
if task.Version == before {
return nil
}
if err := uc.tasks.Update(ctx, task); err != nil {
return err
}
@@ -195,18 +235,83 @@ func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*dom
return out, nil
}
// verifyResultArtifact enforces that the referenced artifact was stored by the
// coordinator for this exact task. It stops a worker from completing task B with
// an artifact it uploaded for task A, and from naming an id that isn't a result.
func (uc *CompleteTask) verifyResultArtifact(ctx context.Context, taskID uuid.UUID, attempt int, artifactID uuid.UUID) error {
art, err := uc.artifacts.Get(ctx, artifactID)
// recordVote handles an untrusted result: it stores the vote, then completes the
// task when the submitter's result hash has reached quorum, or returns the task
// to the queue so another owner can compute it independently.
func (uc *CompleteTask) recordVote(ctx context.Context, task *domain.Task, in CompleteTaskInput,
art *domain.Artifact, ownerID uuid.UUID, now time.Time, out **domain.Task) error {
*out = task
if task.Status == domain.TaskCompleted {
return nil // already settled by an earlier quorum; nothing to record
}
if err := uc.results.RecordVote(ctx, task.ID, ownerID, art.SHA256, in.ResultArtifactID); err != nil {
return err
}
agree, err := uc.results.CountAgreeing(ctx, task.ID, art.SHA256)
if err != nil {
return err
}
if art.TaskID == nil || *art.TaskID != taskID || art.Attempt == nil || *art.Attempt != attempt || art.Kind != domain.ArtifactPartialResult {
return domain.ErrResultConflict
if agree >= uc.quorum {
// The submitter's own (already verified) artifact carries the winning
// hash, so complete with it.
if err := task.CompleteWith(in.ResultArtifactID, in.Metrics, in.WorkerID, in.Attempt, now); err != nil {
return err
}
} else if err := task.ReleaseAfterVote(in.WorkerID, in.Attempt, now); err != nil {
return err
}
return nil
if err := uc.tasks.Update(ctx, task); err != nil {
return err
}
return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now)
}
// workerTrust reports whether the worker's results are accepted directly, and
// the owner to attribute a vote to when they are not.
func (uc *CompleteTask) workerTrust(ctx context.Context, workerID string) (trusted bool, ownerID uuid.UUID, err error) {
// When the worker can't be resolved, default to trusted — the pre-quorum
// behaviour. This is safe because completing a task requires holding its
// lease, and the lease owner is always a real registered worker whose trust
// is therefore known; only an untrusted worker ever takes the quorum path.
id, err := uuid.Parse(workerID)
if err != nil {
// An unparseable worker id means the worker can't be resolved; fall back
// to the trusted default rather than surfacing the parse error.
return true, uuid.Nil, nil //nolint:nilerr // unresolvable worker → trusted (pre-quorum default)
}
w, err := uc.workers.Get(ctx, id)
if err != nil {
if errors.Is(err, domain.ErrWorkerNotFound) {
return true, uuid.Nil, nil
}
return false, uuid.Nil, err
}
if w.TrustLevel != domain.WorkerUntrusted {
return true, uuid.Nil, nil
}
if w.OwnerID == nil {
// An untrusted worker always has an owner (it registered via a user JWT);
// a missing one is a data error, not a silent trust upgrade.
return false, uuid.Nil, domain.ErrInvalidInput
}
return false, *w.OwnerID, nil
}
// verifyResultArtifact enforces that the referenced artifact was stored by the
// coordinator for this exact task. It stops a worker from completing task B with
// an artifact it uploaded for task A, and from naming an id that isn't a result.
func (uc *CompleteTask) verifyResultArtifact(ctx context.Context, taskID uuid.UUID, attempt int, artifactID uuid.UUID) (*domain.Artifact, error) {
art, err := uc.artifacts.Get(ctx, artifactID)
if err != nil {
return nil, err
}
if art.TaskID == nil || *art.TaskID != taskID || art.Attempt == nil || *art.Attempt != attempt || art.Kind != domain.ArtifactPartialResult {
return nil, domain.ErrResultConflict
}
return art, nil
}
// --- FailTask ------------------------------------------------------------
+193 -27
View File
@@ -2,10 +2,12 @@ package usecase
import (
"context"
"fmt"
"time"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
)
@@ -13,25 +15,34 @@ import (
// It intentionally exposes no storage paths or credentials.
type UIReadRepository interface {
GetJob(ctx context.Context, jobID uuid.UUID) (*domain.Job, error)
ListJobs(ctx context.Context, limit int) ([]domain.Job, error)
// ListJobs returns the most recent jobs. A non-nil owner restricts the list
// to that user's jobs; nil returns all (operator/admin view).
ListJobs(ctx context.Context, owner *uuid.UUID, limit int) ([]domain.Job, error)
ListTasksByJob(ctx context.Context, jobID uuid.UUID) ([]domain.Task, error)
ListTasksByJobs(ctx context.Context, jobIDs []uuid.UUID) (map[uuid.UUID][]domain.Task, error)
ListWorkers(ctx context.Context, limit int) ([]domain.Worker, error)
// ListWorkersByOwner returns the most recent workers registered by one user,
// for the "my machines" section of the dashboard.
ListWorkersByOwner(ctx context.Context, owner uuid.UUID, limit int) ([]domain.Worker, error)
ListArtifactsByJob(ctx context.Context, jobID uuid.UUID) ([]domain.Artifact, error)
}
type JobCard struct {
ID string `json:"id"`
Workload string `json:"workload"`
Status string `json:"status"`
CreatedAt time.Time `json:"created_at"`
Total int `json:"total"`
Pending int `json:"pending"`
Leased int `json:"leased"`
Running int `json:"running"`
Completed int `json:"completed"`
Failed int `json:"failed"`
Cancelled int `json:"cancelled"`
ID string `json:"id"`
Workload string `json:"workload"`
Status string `json:"status"`
CreatedAt time.Time `json:"created_at"`
CompletedAt *time.Time `json:"completed_at,omitempty"`
ReducerStartedAt *time.Time `json:"reducer_started_at,omitempty"`
ErrorCode string `json:"error_code,omitempty"`
ErrorMessage string `json:"error_message,omitempty"`
Total int `json:"total"`
Pending int `json:"pending"`
Leased int `json:"leased"`
Running int `json:"running"`
Completed int `json:"completed"`
Failed int `json:"failed"`
Cancelled int `json:"cancelled"`
}
type TaskCard struct {
@@ -42,10 +53,20 @@ type TaskCard struct {
MaxAttempts int `json:"max_attempts"`
LeaseOwner string `json:"lease_owner,omitempty"`
LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"`
StartedAt *time.Time `json:"started_at,omitempty"`
CompletedAt *time.Time `json:"completed_at,omitempty"`
ErrorCode string `json:"error_code,omitempty"`
ErrorMessage string `json:"error_message,omitempty"`
}
// ParameterCard is an intentionally small allowlist of run configuration that
// helps an operator verify what is being computed without exposing arbitrary
// job payloads to the browser.
type ParameterCard struct {
Label string `json:"label"`
Value string `json:"value"`
}
type ArtifactCard struct {
ID string `json:"id"`
Kind string `json:"kind"`
@@ -65,14 +86,43 @@ type WorkerCard struct {
}
type DashboardView struct {
Jobs []JobCard
Workers []WorkerCard
Jobs []JobCard `json:"jobs"`
Workers []WorkerCard `json:"workers"`
// MyWorkers is the signed-in user's own registered workers. Empty for an
// admin or a basic-auth operator, who instead see the whole fleet in Workers.
MyWorkers []WorkerCard `json:"my_workers"`
ActiveJobs int `json:"active_jobs"`
FinishedJobs int `json:"finished_jobs"`
OnlineWorkers int `json:"online_workers"`
// Session is the signed-in user, when the UI runs in session mode. nil under
// basic auth. Template-only, never serialised to the polling JSON.
Session *SessionView `json:"-"`
}
// SessionView is the minimal identity the UI header needs to show who is signed
// in and to offer a logout control.
type SessionView struct {
Role string
Verified bool
}
// sessionViewFrom builds the header session info from the request context, or
// nil when the caller is not an authenticated user (basic-auth operator).
func sessionViewFrom(ctx context.Context) *SessionView {
r, ok := authctx.From(ctx)
if !ok {
return nil
}
return &SessionView{Role: r.Role, Verified: r.Verified}
}
type JobDetailView struct {
JobCard
Tasks []TaskCard `json:"tasks"`
Artifacts []ArtifactCard `json:"artifacts"`
FinalResultAvailable bool `json:"final_result_available"`
Tasks []TaskCard `json:"tasks"`
Artifacts []ArtifactCard `json:"artifacts"`
Parameters []ParameterCard `json:"parameters"`
FinalResultAvailable bool `json:"final_result_available"`
Session *SessionView `json:"-"`
}
type Dashboard struct{ read UIReadRepository }
@@ -80,7 +130,7 @@ type Dashboard struct{ read UIReadRepository }
func NewDashboard(read UIReadRepository) *Dashboard { return &Dashboard{read: read} }
func (d *Dashboard) Overview(ctx context.Context, limit int) (DashboardView, error) {
jobs, err := d.read.ListJobs(ctx, limit)
jobs, err := d.read.ListJobs(ctx, uiOwnerFilter(ctx), limit)
if err != nil {
return DashboardView{}, err
}
@@ -98,19 +148,57 @@ func (d *Dashboard) Overview(ctx context.Context, limit int) (DashboardView, err
return DashboardView{}, err
}
for _, job := range jobs {
out.Jobs = append(out.Jobs, jobCard(job, tasksByJob[job.ID]))
card := jobCard(job, tasksByJob[job.ID])
out.Jobs = append(out.Jobs, card)
switch card.Status {
case string(domain.JobCompleted), string(domain.JobFailed), string(domain.JobCancelled):
out.FinishedJobs++
default:
out.ActiveJobs++
}
}
for _, worker := range workers {
out.Workers = append(out.Workers, WorkerCard{ID: worker.ID.String(), Name: worker.Name, Status: string(worker.Status), Capabilities: worker.Capabilities, LastHeartbeatAt: worker.LastHeartbeatAt})
out.Workers = append(out.Workers, workerCard(worker))
if worker.Status == domain.WorkerOnline || worker.Status == domain.WorkerBusy {
out.OnlineWorkers++
}
}
// A plain user also gets a dedicated "my machines" list scoped to their own
// registrations; an admin/operator sees only the fleet above.
if owner := uiOwnerFilter(ctx); owner != nil {
mine, err := d.read.ListWorkersByOwner(ctx, *owner, limit)
if err != nil {
return DashboardView{}, err
}
out.MyWorkers = make([]WorkerCard, 0, len(mine))
for _, worker := range mine {
out.MyWorkers = append(out.MyWorkers, workerCard(worker))
}
}
out.Session = sessionViewFrom(ctx)
return out, nil
}
func workerCard(w domain.Worker) WorkerCard {
return WorkerCard{
ID: w.ID.String(),
Name: w.Name,
Status: string(w.Status),
Capabilities: w.Capabilities,
LastHeartbeatAt: w.LastHeartbeatAt,
}
}
func (d *Dashboard) JobDetail(ctx context.Context, jobID uuid.UUID) (JobDetailView, error) {
job, err := d.read.GetJob(ctx, jobID)
if err != nil {
return JobDetailView{}, err
}
// A plain user may only open their own job; a mismatch reads as not-found so
// the page never reveals another user's job exists.
if err := authorizeJobAccess(ctx, job); err != nil {
return JobDetailView{}, err
}
tasks, err := d.read.ListTasksByJob(ctx, jobID)
if err != nil {
return JobDetailView{}, err
@@ -119,11 +207,28 @@ func (d *Dashboard) JobDetail(ctx context.Context, jobID uuid.UUID) (JobDetailVi
if err != nil {
return JobDetailView{}, err
}
out := JobDetailView{JobCard: jobCard(*job, tasks), Tasks: make([]TaskCard, 0, len(tasks)), Artifacts: make([]ArtifactCard, 0, len(artifacts))}
workers, err := d.read.ListWorkers(ctx, 100)
if err != nil {
return JobDetailView{}, err
}
workerNames := make(map[string]string, len(workers))
for _, worker := range workers {
workerNames[worker.ID.String()] = worker.Name
}
out := JobDetailView{
JobCard: jobCard(*job, tasks),
Tasks: make([]TaskCard, 0, len(tasks)),
Artifacts: make([]ArtifactCard, 0, len(artifacts)),
Parameters: uiParameters(job.Parameters),
Session: sessionViewFrom(ctx),
}
for _, task := range tasks {
card := TaskCard{ID: task.ID.String(), ChunkIndex: task.ChunkIndex, Status: string(task.Status), Attempt: task.Attempt, MaxAttempts: task.MaxAttempts, LeaseExpiresAt: task.LeaseExpiresAt}
card := TaskCard{ID: task.ID.String(), ChunkIndex: task.ChunkIndex, Status: string(task.Status), Attempt: task.Attempt, MaxAttempts: task.MaxAttempts, LeaseExpiresAt: task.LeaseExpiresAt, StartedAt: task.StartedAt, CompletedAt: task.CompletedAt}
if task.LeaseOwner != nil {
card.LeaseOwner = *task.LeaseOwner
card.LeaseOwner = workerNames[*task.LeaseOwner]
if card.LeaseOwner == "" {
card.LeaseOwner = "Worker " + shortID(*task.LeaseOwner)
}
}
if task.ErrorCode != nil {
card.ErrorCode = *task.ErrorCode
@@ -135,7 +240,7 @@ func (d *Dashboard) JobDetail(ctx context.Context, jobID uuid.UUID) (JobDetailVi
}
for _, artifact := range artifacts {
diagnostic := artifact.Kind == domain.ArtifactPartialResult
downloadable := diagnostic || (artifact.Kind == domain.ArtifactFinalResult && out.Status == string(domain.JobCompleted))
downloadable := previewableArtifact(*job, artifact)
out.Artifacts = append(out.Artifacts, ArtifactCard{ID: artifact.ID.String(), Kind: string(artifact.Kind), Filename: artifact.Filename, SizeBytes: artifact.SizeBytes, SHA256: artifact.SHA256, Downloadable: downloadable, Diagnostic: diagnostic})
if artifact.Kind == domain.ArtifactFinalResult && downloadable {
out.FinalResultAvailable = true
@@ -144,13 +249,25 @@ func (d *Dashboard) JobDetail(ctx context.Context, jobID uuid.UUID) (JobDetailVi
return out, nil
}
func (d *Dashboard) ArtifactBelongsToJob(ctx context.Context, jobID, artifactID uuid.UUID) (bool, error) {
// DownloadableArtifactBelongsToJob applies the same policy used by the UI
// projection: partial diagnostics and the persisted final result are public to
// the operator; source inputs and shards are not exposed through a guessed UI
// URL.
func (d *Dashboard) DownloadableArtifactBelongsToJob(ctx context.Context, jobID, artifactID uuid.UUID) (bool, error) {
job, err := d.read.GetJob(ctx, jobID)
if err != nil {
return false, err
}
// Not the caller's job (and not admin): treat as if the artifact is absent.
if err := authorizeJobAccess(ctx, job); err != nil {
return false, nil //nolint:nilerr // masking the authz error as "not found" is intentional
}
artifacts, err := d.read.ListArtifactsByJob(ctx, jobID)
if err != nil {
return false, err
}
for _, a := range artifacts {
if a.ID == artifactID {
if a.ID == artifactID && previewableArtifact(*job, a) {
return true, nil
}
}
@@ -158,7 +275,13 @@ func (d *Dashboard) ArtifactBelongsToJob(ctx context.Context, jobID, artifactID
}
func jobCard(job domain.Job, tasks []domain.Task) JobCard {
c := JobCard{ID: job.ID.String(), Workload: job.Workload, CreatedAt: job.CreatedAt}
c := JobCard{ID: job.ID.String(), Workload: job.Workload, CreatedAt: job.CreatedAt, CompletedAt: job.CompletedAt, ReducerStartedAt: job.ReducerStartedAt}
if job.ErrorCode != nil {
c.ErrorCode = *job.ErrorCode
}
if job.ErrorMessage != nil {
c.ErrorMessage = *job.ErrorMessage
}
for _, task := range tasks {
c.Total++
switch task.Status {
@@ -180,3 +303,46 @@ func jobCard(job domain.Job, tasks []domain.Task) JobCard {
c.Status = string(p.DeriveStatus())
return c
}
func uiParameters(parameters map[string]any) []ParameterCard {
keys := []struct {
key string
label string
}{
{"query_smiles", "Target SMILES"},
{"query_id", "Target ChEMBL ID"},
{"top_k", "Global top-k"},
{"threshold", "Similarity threshold"},
{"threshold_direction", "Threshold direction"},
}
out := make([]ParameterCard, 0, len(keys))
for _, entry := range keys {
value, ok := parameters[entry.key]
if !ok {
continue
}
formatted, ok := formatUIParameter(value)
if ok {
out = append(out, ParameterCard{Label: entry.label, Value: formatted})
}
}
return out
}
func formatUIParameter(value any) (string, bool) {
switch typed := value.(type) {
case string:
return typed, true
case int, int64, float64, bool:
return fmt.Sprint(typed), true
default:
return "", false
}
}
func shortID(value string) string {
if len(value) <= 8 {
return value
}
return value[:8]
}
@@ -0,0 +1,19 @@
package usecase
import "testing"
func TestUIParametersAreAllowlisted(t *testing.T) {
parameters := uiParameters(map[string]any{
"query_smiles": "CCO",
"top_k": float64(20),
"internal_storage_key": "must-not-reach-browser",
"nested": map[string]any{"secret": "no"},
})
if len(parameters) != 2 {
t.Fatalf("parameters = %#v, want only two allowlisted values", parameters)
}
if parameters[0] != (ParameterCard{Label: "Target SMILES", Value: "CCO"}) ||
parameters[1] != (ParameterCard{Label: "Global top-k", Value: "20"}) {
t.Fatalf("parameters = %#v", parameters)
}
}
@@ -0,0 +1,82 @@
package usecase_test
import (
"context"
"errors"
"testing"
"time"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/memstore"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
)
func newDashboard() (*usecase.Dashboard, *memstore.JobRepo) {
jobs := memstore.NewJobRepo()
tasks := memstore.NewTaskRepo()
workers := memstore.NewWorkerRepo()
artifacts := memstore.NewArtifactRepo()
return usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts)), jobs
}
func ownedJob(t *testing.T, jobs *memstore.JobRepo, owner uuid.UUID) uuid.UUID {
t.Helper()
o := owner
job := &domain.Job{ID: uuid.New(), Workload: "similarity-search", Status: domain.JobRunning, OwnerID: &o, CreatedAt: time.Now().UTC()}
if err := jobs.Insert(context.Background(), job); err != nil {
t.Fatalf("insert owned job: %v", err)
}
return job.ID
}
func userCtx(id uuid.UUID, role string) context.Context {
return authctx.With(context.Background(), authctx.Requester{UserID: id, Role: role})
}
func TestOverviewScopesJobsByOwner(t *testing.T) {
dash, jobs := newDashboard()
alice, bob := uuid.New(), uuid.New()
ownedJob(t, jobs, alice)
ownedJob(t, jobs, bob)
// A plain user sees only their own job.
v, err := dash.Overview(userCtx(alice, "user"), 20)
if err != nil {
t.Fatal(err)
}
if len(v.Jobs) != 1 {
t.Errorf("alice sees %d jobs, want 1", len(v.Jobs))
}
// An admin sees every job.
if v, _ := dash.Overview(userCtx(uuid.New(), "admin"), 20); len(v.Jobs) != 2 {
t.Errorf("admin sees %d jobs, want 2", len(v.Jobs))
}
// No requester (basic-auth operator) sees every job — unchanged behaviour.
if v, _ := dash.Overview(context.Background(), 20); len(v.Jobs) != 2 {
t.Errorf("operator sees %d jobs, want 2", len(v.Jobs))
}
}
func TestJobDetailRejectsAnotherUsersJob(t *testing.T) {
dash, jobs := newDashboard()
alice, bob := uuid.New(), uuid.New()
jobID := ownedJob(t, jobs, alice)
// Bob cannot open Alice's job.
if _, err := dash.JobDetail(userCtx(bob, "user"), jobID); !errors.Is(err, domain.ErrJobNotFound) {
t.Errorf("bob: got %v, want ErrJobNotFound", err)
}
// Alice can.
if _, err := dash.JobDetail(userCtx(alice, "user"), jobID); err != nil {
t.Errorf("alice: unexpected error %v", err)
}
// Admin can.
if _, err := dash.JobDetail(userCtx(uuid.New(), "admin"), jobID); err != nil {
t.Errorf("admin: unexpected error %v", err)
}
}
@@ -0,0 +1,66 @@
package usecase_test
import (
"context"
"testing"
"time"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/memstore"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
)
func newDashboardWithWorkers() (*usecase.Dashboard, *memstore.WorkerRepo) {
jobs := memstore.NewJobRepo()
tasks := memstore.NewTaskRepo()
workers := memstore.NewWorkerRepo()
artifacts := memstore.NewArtifactRepo()
return usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts)), workers
}
func seedWorker(t *testing.T, workers *memstore.WorkerRepo, owner *uuid.UUID, name string) {
t.Helper()
w := &domain.Worker{
ID: uuid.New(),
Name: name,
Capabilities: []string{"similarity-search"},
Status: domain.WorkerOnline,
OwnerID: owner,
LastHeartbeatAt: time.Now().UTC(),
}
if err := workers.Insert(context.Background(), w); err != nil {
t.Fatalf("insert worker: %v", err)
}
}
func TestOverviewSplitsMyWorkers(t *testing.T) {
dash, workers := newDashboardWithWorkers()
alice, bob := uuid.New(), uuid.New()
seedWorker(t, workers, &alice, "alice-box")
seedWorker(t, workers, &bob, "bob-box")
seedWorker(t, workers, nil, "lab-shared") // owner-less shared-token worker
// A plain user sees the whole fleet, but MyWorkers holds only their own.
v, err := dash.Overview(userCtx(alice, "user"), 20)
if err != nil {
t.Fatal(err)
}
if len(v.Workers) != 3 {
t.Errorf("fleet shows %d workers, want 3", len(v.Workers))
}
if len(v.MyWorkers) != 1 || v.MyWorkers[0].Name != "alice-box" {
t.Errorf("MyWorkers = %+v, want only alice-box", v.MyWorkers)
}
// An admin is not owner-scoped: they get the fleet and no personal list.
if av, _ := dash.Overview(userCtx(uuid.New(), "admin"), 20); len(av.MyWorkers) != 0 || len(av.Workers) != 3 {
t.Errorf("admin MyWorkers=%d Workers=%d, want 0 and 3", len(av.MyWorkers), len(av.Workers))
}
// A basic-auth operator (no requester) also gets no personal list.
if ov, _ := dash.Overview(context.Background(), 20); len(ov.MyWorkers) != 0 {
t.Errorf("operator MyWorkers=%d, want 0", len(ov.MyWorkers))
}
}
+1
View File
@@ -43,6 +43,7 @@ func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (Su
if err != nil {
return SubmitDatasetResult{}, err
}
job.OwnerID = ownerFromContext(ctx)
// Everything written to blob storage, so a failed transaction can undo it.
var putKeys []string
+164 -13
View File
@@ -11,6 +11,7 @@ import (
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/memstore"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
@@ -34,12 +35,13 @@ func (s expiringBlobStore) Put(ctx context.Context, key string, body io.Reader)
// harness wires every use case to in-memory stores so orchestration can be
// tested without a database.
type harness struct {
tasks *memstore.TaskRepo
jobs *memstore.JobRepo
work *memstore.WorkerRepo
arts *memstore.ArtifactRepo
blobs *memstore.BlobStore
clk *memstore.Clock
tasks *memstore.TaskRepo
jobs *memstore.JobRepo
work *memstore.WorkerRepo
arts *memstore.ArtifactRepo
blobs *memstore.BlobStore
clk *memstore.Clock
taskResults *memstore.TaskResultRepo
createJob *usecase.CreateJob
submit *usecase.SubmitDataset
@@ -61,19 +63,20 @@ type harness struct {
func newHarness() *harness {
h := &harness{
tasks: memstore.NewTaskRepo(),
jobs: memstore.NewJobRepo(),
work: memstore.NewWorkerRepo(),
arts: memstore.NewArtifactRepo(),
blobs: memstore.NewBlobStore(),
clk: memstore.NewClock(time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC)),
tasks: memstore.NewTaskRepo(),
jobs: memstore.NewJobRepo(),
work: memstore.NewWorkerRepo(),
arts: memstore.NewArtifactRepo(),
blobs: memstore.NewBlobStore(),
clk: memstore.NewClock(time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC)),
taskResults: memstore.NewTaskResultRepo(),
}
tx := memstore.Tx{}
h.createJob = usecase.NewCreateJob(h.jobs, h.tasks, tx, h.clk)
h.submit = usecase.NewSubmitDataset(h.blobs, h.arts, h.jobs, h.tasks, tx, h.clk, 3)
h.claim = usecase.NewClaimTask(h.tasks, h.jobs, h.work, tx, h.clk, lease)
h.renew = usecase.NewRenewLease(h.tasks, h.work, tx, h.clk, lease)
h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, tx, h.clk)
h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, h.work, h.taskResults, tx, h.clk, 2)
h.fail = usecase.NewFailTask(h.tasks, h.jobs, tx, h.clk)
h.status = usecase.NewGetJobStatus(h.jobs, h.tasks)
h.results = usecase.NewListResults(h.tasks)
@@ -265,6 +268,154 @@ func TestClaimEmptyQueueReturnsNil(t *testing.T) {
}
}
func TestRegisterWorkerDefaultsToTrusted(t *testing.T) {
h := newHarness()
// A shared-token registration carries no owner and no explicit trust.
w, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{
Name: "lab", Capabilities: []string{"w"},
})
if err != nil {
t.Fatal(err)
}
if w.TrustLevel != domain.WorkerTrusted {
t.Errorf("trust = %q, want trusted", w.TrustLevel)
}
if w.OwnerID != nil {
t.Errorf("owner = %v, want nil for a shared-token worker", w.OwnerID)
}
}
func TestRegisterWorkerRecordsOwnerAndUntrusted(t *testing.T) {
h := newHarness()
owner := uuid.New()
w, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{
Name: "volunteer", Capabilities: []string{"w"},
OwnerID: &owner, TrustLevel: domain.WorkerUntrusted,
})
if err != nil {
t.Fatal(err)
}
if w.TrustLevel != domain.WorkerUntrusted {
t.Errorf("trust = %q, want untrusted", w.TrustLevel)
}
if w.OwnerID == nil || *w.OwnerID != owner {
t.Errorf("owner = %v, want %v", w.OwnerID, owner)
}
}
func TestJWTCallerCannotClaimAsAnotherUsersWorker(t *testing.T) {
h := newHarness()
h.seedJob(t, "w", 1)
// A trusted lab worker owned by nobody (shared-token registration).
victim, _ := h.register.Execute(ctx, usecase.RegisterWorkerInput{Name: "lab", Capabilities: []string{"w"}})
// An attacker authenticated as a JWT user tries to claim as the lab worker.
attacker := authctx.With(ctx, authctx.Requester{UserID: uuid.New(), Role: "user"})
claimed, err := h.claim.Execute(attacker, usecase.ClaimTaskInput{WorkerID: victim.ID.String()})
if !errors.Is(err, domain.ErrWorkerNotFound) {
t.Fatalf("claim as another's worker = (%v, %v), want ErrWorkerNotFound", claimed, err)
}
}
func TestJWTCallerClaimsAsOwnTrustedWorker(t *testing.T) {
h := newHarness()
h.seedJob(t, "w", 1)
owner := uuid.New()
// The user's own worker, trusted (e.g. a verified contributor).
mine, _ := h.register.Execute(ctx, usecase.RegisterWorkerInput{
Name: "mine", Capabilities: []string{"w"}, OwnerID: &owner, TrustLevel: domain.WorkerTrusted,
})
callerCtx := authctx.With(ctx, authctx.Requester{UserID: owner, Role: "user", Verified: true})
got, err := h.claim.Execute(callerCtx, usecase.ClaimTaskInput{WorkerID: mine.ID.String()})
if err != nil || got == nil {
t.Fatalf("own trusted worker claim = (%v, %v), want a task", got, err)
}
}
func TestUntrustedWorkerCanClaim(t *testing.T) {
h := newHarness()
h.seedJob(t, "w", 1)
owner := uuid.New()
worker, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{
Name: "volunteer", Capabilities: []string{"w"},
OwnerID: &owner, TrustLevel: domain.WorkerUntrusted,
})
if err != nil {
t.Fatal(err)
}
// Volunteers are no longer quarantined — they may claim; their results are
// gated by quorum at completion, not by withholding work.
claimed, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{WorkerID: worker.ID.String()})
if err != nil || claimed == nil {
t.Fatalf("untrusted claim = (%v, %v), want a task", claimed, err)
}
}
// registerUntrusted registers a volunteer worker under a fresh owner.
func (h *harness) registerUntrusted(t *testing.T, name, workload string) (*domain.Worker, uuid.UUID) {
t.Helper()
owner := uuid.New()
w, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{
Name: name, Capabilities: []string{workload},
OwnerID: &owner, TrustLevel: domain.WorkerUntrusted,
})
if err != nil {
t.Fatal(err)
}
return w, owner
}
func TestUntrustedResultNeedsQuorum(t *testing.T) {
h := newHarness()
jobID := h.seedJob(t, "w", 1)
if err := h.jobs.UpdateStatus(ctx, jobID, domain.JobRunning, nil); err != nil {
t.Fatal(err)
}
w1, _ := h.registerUntrusted(t, "v1", "w")
w2, _ := h.registerUntrusted(t, "v2", "w")
// First volunteer computes and submits — one vote, not yet quorum (2).
taskID, attempt := h.leaseOne(t, w1.ID.String(), "w")
art1 := h.uploadResult(t, taskID, w1.ID.String(), attempt)
if _, err := h.complete.Execute(ctx, usecase.CompleteTaskInput{TaskID: taskID, WorkerID: w1.ID.String(), Attempt: attempt, ResultArtifactID: art1}); err != nil {
t.Fatalf("first vote: %v", err)
}
if tk, _ := h.tasks.Get(ctx, taskID); tk.Status != domain.TaskPending {
t.Fatalf("after one vote status = %s, want pending", tk.Status)
}
// Second volunteer (distinct owner) computes the same bytes -> quorum -> done.
taskID2, attempt2 := h.leaseOne(t, w2.ID.String(), "w")
art2 := h.uploadResult(t, taskID2, w2.ID.String(), attempt2)
if _, err := h.complete.Execute(ctx, usecase.CompleteTaskInput{TaskID: taskID2, WorkerID: w2.ID.String(), Attempt: attempt2, ResultArtifactID: art2}); err != nil {
t.Fatalf("second vote: %v", err)
}
if tk, _ := h.tasks.Get(ctx, taskID); tk.Status != domain.TaskCompleted {
t.Fatalf("after quorum status = %s, want completed", tk.Status)
}
}
func TestTrustedResultCompletesDirectly(t *testing.T) {
h := newHarness()
jobID := h.seedJob(t, "w", 1)
if err := h.jobs.UpdateStatus(ctx, jobID, domain.JobRunning, nil); err != nil {
t.Fatal(err)
}
// A trusted (default) worker's single result completes the task immediately.
worker, _ := h.register.Execute(ctx, usecase.RegisterWorkerInput{Name: "lab", Capabilities: []string{"w"}})
taskID, attempt := h.leaseOne(t, worker.ID.String(), "w")
art := h.uploadResult(t, taskID, worker.ID.String(), attempt)
if _, err := h.complete.Execute(ctx, usecase.CompleteTaskInput{TaskID: taskID, WorkerID: worker.ID.String(), Attempt: attempt, ResultArtifactID: art}); err != nil {
t.Fatal(err)
}
if tk, _ := h.tasks.Get(ctx, taskID); tk.Status != domain.TaskCompleted {
t.Fatalf("trusted result status = %s, want completed", tk.Status)
}
}
func TestClaimRequiresWorkerID(t *testing.T) {
h := newHarness()
if _, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{}); !errors.Is(err, domain.ErrInvalidInput) {
+7
View File
@@ -22,6 +22,13 @@ func (uc *RegisterWorker) Execute(ctx context.Context, in RegisterWorkerInput) (
if err != nil {
return nil, err
}
w.OwnerID = in.OwnerID
// The transport layer resolves trust from the caller's credentials; fall
// back to the domain default (trusted) only when it was left unset, so a
// zero-value input never silently downgrades a shared-token worker.
if in.TrustLevel != "" {
w.TrustLevel = in.TrustLevel
}
if err := uc.workers.Insert(ctx, w); err != nil {
return nil, err
}
@@ -0,0 +1,6 @@
BEGIN;
DROP INDEX IF EXISTS ix_jobs_owner;
ALTER TABLE jobs DROP COLUMN IF EXISTS owner_id;
COMMIT;
@@ -0,0 +1,14 @@
BEGIN;
-- Who submitted this job. Equals users.id from the userservice, taken from the
-- JWT `sub` claim. NOT a foreign key: users live in a separate service/database,
-- so integrity is guaranteed by the signed token, not by the DB.
--
-- Nullable because rows created before auth existed have no owner; new inserts
-- must supply it (enforced in the app, not the schema, during the MVP).
ALTER TABLE jobs ADD COLUMN owner_id uuid;
-- "List my jobs" / "admin filters by owner" scans by owner.
CREATE INDEX ix_jobs_owner ON jobs (owner_id);
COMMIT;
@@ -0,0 +1,8 @@
BEGIN;
DROP INDEX IF EXISTS ix_workers_owner;
ALTER TABLE workers DROP COLUMN IF EXISTS trust_level;
ALTER TABLE workers DROP COLUMN IF EXISTS owner_id;
DROP TYPE IF EXISTS worker_trust;
COMMIT;
@@ -0,0 +1,18 @@
BEGIN;
-- Whether a worker's results are accepted directly or must clear quorum.
-- 'trusted' — lab machine (shared token) or a verified/admin contributor.
-- 'untrusted' — a plain enthusiast; results are quarantined until quorum (C2).
CREATE TYPE worker_trust AS ENUM ('trusted', 'untrusted');
-- Who registered this worker (userservice user id, from the JWT sub). NULL for
-- workers registered with the shared service token. Not a foreign key: users
-- live in a separate service/database.
ALTER TABLE workers ADD COLUMN owner_id uuid;
-- Existing rows were all shared-token lab workers, hence 'trusted'.
ALTER TABLE workers ADD COLUMN trust_level worker_trust NOT NULL DEFAULT 'trusted';
CREATE INDEX ix_workers_owner ON workers (owner_id);
COMMIT;
@@ -0,0 +1,5 @@
BEGIN;
DROP TABLE IF EXISTS task_results;
COMMIT;
@@ -0,0 +1,23 @@
BEGIN;
-- Quorum votes for a task computed by untrusted (volunteer) workers. A trusted
-- worker's result completes the task directly and never lands here; an untrusted
-- result is recorded as one vote, and the task is only completed once enough
-- distinct owners submit the same result_sha256.
--
-- One vote per (task, owner): a single volunteer cannot stuff the ballot by
-- running many workers under one account. A resubmission updates their vote.
CREATE TABLE task_results (
task_id uuid NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
owner_id uuid NOT NULL,
result_sha256 text NOT NULL,
result_artifact_id uuid NOT NULL REFERENCES artifacts(id) ON DELETE CASCADE,
created_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (task_id, owner_id)
);
-- Quorum check groups a task's votes by result_sha256.
CREATE INDEX ix_task_results_quorum ON task_results (task_id, result_sha256);
COMMIT;
@@ -0,0 +1,154 @@
{
"annotations": { "list": [] },
"editable": true,
"graphTooltip": 1,
"schemaVersion": 39,
"tags": ["scimesh"],
"time": { "from": "now-15m", "to": "now" },
"refresh": "5s",
"title": "SciMesh Coordinator",
"uid": "scimesh-coordinator",
"panels": [
{
"type": "timeseries",
"title": "HTTP request rate by route",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 },
"fieldConfig": { "defaults": { "unit": "reqps" }, "overrides": [] },
"targets": [
{
"refId": "A",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"expr": "sum by (route) (rate(scimesh_http_requests_total[1m]))",
"legendFormat": "{{route}}"
}
]
},
{
"type": "timeseries",
"title": "p95 latency by route",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 },
"fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] },
"targets": [
{
"refId": "A",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"expr": "histogram_quantile(0.95, sum by (le, route) (rate(scimesh_http_request_duration_seconds_bucket[5m])))",
"legendFormat": "{{route}}"
}
]
},
{
"type": "timeseries",
"title": "Requests by status",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"gridPos": { "h": 8, "w": 12, "x": 0, "y": 8 },
"fieldConfig": { "defaults": { "unit": "reqps" }, "overrides": [] },
"targets": [
{
"refId": "A",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"expr": "sum by (status) (rate(scimesh_http_requests_total[1m]))",
"legendFormat": "{{status}}"
}
]
},
{
"type": "timeseries",
"title": "Goroutines",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"gridPos": { "h": 8, "w": 6, "x": 12, "y": 8 },
"fieldConfig": { "defaults": { "unit": "short" }, "overrides": [] },
"targets": [
{
"refId": "A",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"expr": "go_goroutines{job=\"coordinator\"}",
"legendFormat": "goroutines"
}
]
},
{
"type": "timeseries",
"title": "Resident memory",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"gridPos": { "h": 8, "w": 6, "x": 18, "y": 8 },
"fieldConfig": { "defaults": { "unit": "bytes" }, "overrides": [] },
"targets": [
{
"refId": "A",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"expr": "process_resident_memory_bytes{job=\"coordinator\"}",
"legendFormat": "rss"
}
]
},
{
"type": "row",
"title": "Domain state",
"collapsed": false,
"gridPos": { "h": 1, "w": 24, "x": 0, "y": 16 }
},
{
"type": "timeseries",
"title": "Tasks by status",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"gridPos": { "h": 8, "w": 12, "x": 0, "y": 17 },
"fieldConfig": { "defaults": { "unit": "short", "custom": { "stacking": { "mode": "normal" }, "fillOpacity": 30 } }, "overrides": [] },
"targets": [
{
"refId": "A",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"expr": "scimesh_tasks",
"legendFormat": "{{status}}"
}
]
},
{
"type": "timeseries",
"title": "Jobs by status",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"gridPos": { "h": 8, "w": 12, "x": 12, "y": 17 },
"fieldConfig": { "defaults": { "unit": "short", "custom": { "stacking": { "mode": "normal" }, "fillOpacity": 30 } }, "overrides": [] },
"targets": [
{
"refId": "A",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"expr": "scimesh_jobs",
"legendFormat": "{{status}}"
}
]
},
{
"type": "stat",
"title": "Queue depth (pending tasks)",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"gridPos": { "h": 8, "w": 6, "x": 0, "y": 25 },
"fieldConfig": { "defaults": { "unit": "short", "color": { "mode": "thresholds" }, "thresholds": { "steps": [ { "color": "green", "value": null }, { "color": "yellow", "value": 50 }, { "color": "red", "value": 500 } ] } }, "overrides": [] },
"targets": [
{
"refId": "A",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"expr": "sum(scimesh_tasks{status=\"pending\"})",
"legendFormat": "pending"
}
]
},
{
"type": "timeseries",
"title": "Workers by status",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"gridPos": { "h": 8, "w": 18, "x": 6, "y": 25 },
"fieldConfig": { "defaults": { "unit": "short", "custom": { "stacking": { "mode": "normal" }, "fillOpacity": 30 } }, "overrides": [] },
"targets": [
{
"refId": "A",
"datasource": { "type": "prometheus", "uid": "prometheus" },
"expr": "scimesh_workers",
"legendFormat": "{{status}}"
}
]
}
]
}
@@ -0,0 +1,10 @@
apiVersion: 1
providers:
- name: SciMesh
type: file
disableDeletion: false
allowUiUpdates: true
options:
path: /var/lib/grafana/dashboards
foldersFromFilesStructure: false
@@ -0,0 +1,10 @@
apiVersion: 1
datasources:
- name: Prometheus
uid: prometheus
type: prometheus
access: proxy
url: http://prometheus:9090
isDefault: true
editable: false
+10
View File
@@ -0,0 +1,10 @@
# Prometheus scrape config for the SciMesh demo. Prometheus runs in the same
# compose network as the coordinator, so it reaches it by service name.
global:
scrape_interval: 5s
evaluation_interval: 5s
scrape_configs:
- job_name: coordinator
static_configs:
- targets: ["coordinator:8080"]
+195
View File
@@ -0,0 +1,195 @@
#!/usr/bin/env bash
# Start a self-contained local UI demo with coordinator, PostgreSQL, and local
# reference workers. It is intentionally for a developer's machine only.
set -euo pipefail
action=${1:-start}
script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
coordinator_dir=$(CDPATH= cd -- "$script_dir/.." && pwd)
repo_dir=$(CDPATH= cd -- "$coordinator_dir/.." && pwd)
project=${DEMO_PROJECT:-scimesh-demo}
postgres_port=${DEMO_POSTGRES_PORT:-55432}
coordinator_port=${DEMO_COORDINATOR_PORT:-18080}
userservice_port=${DEMO_USERSERVICE_PORT:-18081}
prometheus_port=${DEMO_PROMETHEUS_PORT:-19090}
grafana_port=${DEMO_GRAFANA_PORT:-13000}
ui_token=${DEMO_UI_TOKEN:-demo-ui-secret}
worker_token=${DEMO_WORKER_TOKEN:-demo-worker-token}
# Shared HS256 secret; the coordinator verifies userservice tokens with it. Must
# be at least 32 bytes (both services refuse a shorter one).
jwt_secret=${DEMO_JWT_SECRET:-demo-jwt-secret-please-change-me-0123456789}
# The first admin, seeded into the userservice on first boot.
admin_email=${DEMO_ADMIN_EMAIL:-root@scimesh.local}
admin_password=${DEMO_ADMIN_PASSWORD:-rootpassword}
workers=${DEMO_WORKERS:-2}
demo_dir=${DEMO_DIR:-.demo}
case "$demo_dir" in
/*) ;;
*) demo_dir="$coordinator_dir/$demo_dir" ;;
esac
worker_bin=${SCIMESH_WORKER_BIN:-"$repo_dir/.venv/bin/scimesh-worker"}
pid_file="$demo_dir/workers.pids"
logs_dir="$demo_dir/logs"
compose() {
POSTGRES_PORT="$postgres_port" \
COORDINATOR_PORT="$coordinator_port" \
USERSERVICE_PORT="$userservice_port" \
PROMETHEUS_PORT="$prometheus_port" \
GRAFANA_PORT="$grafana_port" \
UI_AUTH_TOKEN="$ui_token" \
WORKER_AUTH_TOKEN="$worker_token" \
JWT_SECRET="$jwt_secret" \
BOOTSTRAP_ADMIN_EMAIL="$admin_email" \
BOOTSTRAP_ADMIN_PASSWORD="$admin_password" \
docker compose -p "$project" \
-f "$coordinator_dir/docker-compose.yml" \
-f "$coordinator_dir/docker-compose.users.yml" \
-f "$coordinator_dir/docker-compose.monitoring.yml" "$@"
}
stop_workers() {
[[ -f "$pid_file" ]] || return 0
while IFS= read -r pid; do
[[ "$pid" =~ ^[0-9]+$ ]] || continue
command_line=$(ps -p "$pid" -o args= 2>/dev/null || true)
# Never kill a recycled PID or a worker launched outside this demo.
if [[ "$command_line" == *"$demo_dir/worker-"* ]]; then
kill "$pid" 2>/dev/null || true
fi
done < "$pid_file"
rm -f "$pid_file"
}
wait_for_coordinator() {
local attempt=0
until curl --fail --silent --show-error "http://localhost:$coordinator_port/health" >/dev/null; do
attempt=$((attempt + 1))
if (( attempt >= 45 )); then
echo "Coordinator did not become ready. Recent logs:" >&2
compose logs --tail=80 coordinator >&2 || true
exit 1
fi
sleep 1
done
}
wait_for_userservice() {
local attempt=0
until curl --fail --silent --show-error "http://localhost:$userservice_port/health" >/dev/null; do
attempt=$((attempt + 1))
if (( attempt >= 45 )); then
echo "Userservice did not become ready. Recent logs:" >&2
compose logs --tail=80 userservice >&2 || true
exit 1
fi
sleep 1
done
}
wait_for_workers() {
local attempt=0 registered overview cookie="$demo_dir/session.cookies"
# The dashboard API is behind a userservice session now, not basic auth. Log in
# as the seeded admin (who sees every worker) to obtain a session cookie.
curl --fail --silent -c "$cookie" \
--data-urlencode "email=$admin_email" \
--data-urlencode "password=$admin_password" \
"http://localhost:$coordinator_port/ui/login" >/dev/null 2>&1 || true
until false; do
overview=$(curl --fail --silent --show-error -b "$cookie" \
"http://localhost:$coordinator_port/ui/api/overview" 2>/dev/null || true)
# The overview contains no jobs at demo startup, so every `id` belongs to
# a registered worker. Avoid adding jq just for this local helper.
registered=$(printf '%s' "$overview" | grep -o '"id"' | wc -l | tr -d ' ' || true)
if [[ "$registered" =~ ^[0-9]+$ ]] && (( registered >= workers )); then
return 0
fi
attempt=$((attempt + 1))
if (( attempt >= 20 )); then
echo "Only $registered of $workers demo workers registered. Recent worker logs:" >&2
tail -n 40 "$logs_dir"/worker-*.log 2>/dev/null >&2 || true
exit 1
fi
sleep 1
done
}
start() {
if ! [[ "$workers" =~ ^[1-9][0-9]*$ ]]; then
echo "DEMO_WORKERS must be a positive integer (got $workers)." >&2
exit 2
fi
if [[ ! -x "$worker_bin" ]]; then
echo "Reference worker not found: $worker_bin" >&2
echo "Create it first from the repository root: python3 -m venv .venv && .venv/bin/pip install -e '.[dev]'" >&2
exit 2
fi
command -v docker >/dev/null || { echo "Docker is required." >&2; exit 2; }
command -v curl >/dev/null || { echo "curl is required." >&2; exit 2; }
stop_workers
mkdir -p "$logs_dir"
compose up -d --build
echo "Waiting for the coordinator on http://localhost:$coordinator_port ..."
wait_for_coordinator
echo "Waiting for the userservice on http://localhost:$userservice_port ..."
wait_for_userservice
: > "$pid_file"
for index in $(seq 1 "$workers"); do
work_dir="$demo_dir/worker-$index"
mkdir -p "$work_dir"
SCIMESH_COORDINATOR_URL="http://localhost:$coordinator_port" \
SCIMESH_BEARER_TOKEN="$worker_token" \
"$worker_bin" \
--worker-name "demo-worker-$index" \
--work-dir "$work_dir" \
>"$logs_dir/worker-$index.log" 2>&1 &
echo "$!" >> "$pid_file"
done
wait_for_workers
cat <<EOF
SciMesh manual demo is ready.
UI: http://localhost:$coordinator_port/ui (shows a login page)
Admin login: $admin_email / $admin_password
Userservice: http://localhost:$userservice_port
Grafana: http://localhost:$grafana_port (anonymous view; admin/${GRAFANA_PASSWORD:-admin} to edit)
Prometheus: http://localhost:$prometheus_port
Workers: $workers local reference workers
Sign in with the admin above, or register a new account from the login page.
The admin sees every job; a plain user sees only their own. Upload a small
ChEMBL TSV through “New similarity search”, then watch the job page update.
Worker logs are in $logs_dir. Stop everything with:
make demo-down
EOF
}
case "$action" in
start) start ;;
stop)
stop_workers
compose down
echo "SciMesh manual demo stopped."
;;
reset)
# Like stop, but also drops the data volumes so the next start is pristine
# (empty Postgres, no leftover workers/jobs/tasks, no cached artifacts).
stop_workers
compose down -v
echo "SciMesh manual demo stopped and data volumes removed."
;;
logs)
echo "Worker logs: $logs_dir"
compose logs -f coordinator
;;
*)
echo "Usage: $0 {start|stop|reset|logs}" >&2
exit 2
;;
esac
+52 -3
View File
@@ -33,14 +33,61 @@ Everything below fills in the details.
## 0. Auth
Every request except `GET /health` carries a shared bearer token:
Every request except `GET /health` carries a bearer token:
```
Authorization: Bearer <token>
```
There are two ways to obtain that token.
### Shared coordinator token (lab / operator workers)
```
Authorization: Bearer <COORDINATOR_TOKEN>
```
The token is handed to you out of band (env var / secret) — the same string the
coordinator was started with. Never log it, never send it in an error body.
coordinator was started with. A worker using it registers **owner-less and
trusted**: its results are accepted without quorum. Never log it, never send it
in an error body.
### Worker key (run a worker bound to your own account)
Any signed-in user can turn their machine into a worker without the shared
secret:
1. In the web UI, open **“Add your machine”** (`/ui/workers/new`) and create a
**worker key** (`scimesh_wk_live_…`). It is shown once — copy it.
2. Install and run the reference worker with the copied command:
```
git clone https://github.com/emil28092005/SciMesh.git
cd SciMesh
python -m venv .venv
source .venv/bin/activate
pip install -e .
SCIMESH_COORDINATOR_URL=<coordinator> \
SCIMESH_USERSERVICE_URL=<userservice> \
SCIMESH_WORKER_KEY=scimesh_wk_live_xxx \
scimesh-worker --worker-name my-machine
```
The worker ships in this repository, not on PyPI, so it is installed from a
clone (`pip install -e .`) rather than `pip install scimesh`.
Under the hood the worker trades the key at `POST /worker-tokens/exchange` for a
short-lived JWT and refreshes it automatically before it expires — so unlike a
raw login token, a worker key keeps a long-running worker authenticated. Revoke
the key in the UI to cut a machine off.
**Trust and quorum.** A worker registered with a plain user's key is
**untrusted**: its result is quarantined and only accepted once a second,
independent worker (a different owner) computes the same answer — the quorum
(default 2). If an admin marks your account **verified**, your workers become
trusted and their results count immediately; re-register the worker after being
verified so it picks up the upgraded trust.
## 1. Register (once, at startup)
@@ -191,7 +238,9 @@ Per the worker contract, at minimum:
- `SCIMESH_COORDINATOR_URL` (e.g. `http://coordinator:8080`)
- worker name (the coordinator returns its `worker_id` at registration;
`SCIMESH_WORKER_ID` is only a legacy/test override)
- the bearer token
- the credential — either `SCIMESH_BEARER_TOKEN` (shared token or a raw JWT) or
`SCIMESH_WORKER_KEY` together with `SCIMESH_USERSERVICE_URL` (a worker key the
worker exchanges and refreshes; see §0)
- poll interval and request timeout
- a working directory for downloaded inputs and generated outputs
+29 -17
View File
@@ -12,16 +12,28 @@ for a trusted local team. The coordinator remains the only process with direct
database and artifact-storage access; the browser never calls PostgreSQL and
never receives a worker bearer token.
The first release must be useful before CTX-07--CTX-10 are complete. Therefore
it has two visibly different modes:
## Current delivered scope
The initial operator UI and CTX-09 final reduction are now implemented. The
control room polls a bounded, coordinator-owned read model every two seconds
while a tab is visible. It shows the worker fleet, recent jobs, safe shard
diagnostics, the actual `reducing` phase, and final-result availability. A job
detail page renders the concrete pipeline stages—input accepted, shards,
worker CSVs, reduction, final CSV—from coordinator state and replaces task and
artifact views as work changes. All browser mutations remain limited to
validated dataset upload and operator cancellation.
The interface must distinguish an in-progress distributed search from a run
whose reducer has produced a durable final result:
| Mode | What it proves | What it must not claim |
| --- | --- | --- |
| **Pipeline check** | Upload, task creation, claim, heartbeat, artifact upload, task completion, retries, and downloads work end-to-end. | That multiple shard results have been scientifically reduced into one answer. |
| **Final run** | A reducer has produced a durable final CSV for the full job. | Available only after CTX-09, and for graph only after CTX-10. |
| **In-progress run** | Upload, task creation, claim, heartbeat, artifact upload, task completion, retries, and shard diagnostics work end-to-end. | That the partial CSVs are a global scientific answer. |
| **Final run** | A reducer has produced a durable final CSV for the full job. | Available for `similarity-search` after CTX-09; graph remains unavailable until CTX-10. |
Never label a partial artifact as a final molecular result. The UI must show a
clear `Pipeline check — partial results` badge while a reducer is unavailable.
clear waiting or `reducing` stage until a final artifact exists and the job is
`completed`.
## 2. Constraints and decisions
@@ -74,7 +86,7 @@ clear `Pipeline check — partial results` badge while a reducer is unavailable.
| Worker registration/lease flow | Implemented | Add a read-only worker list; no browser worker controls. |
| Task diagnostics | No public list/detail response | Add sanitized job task list with attempt, status, lease owner, expiry and error. |
| Artifact download | Worker endpoint exists | Add UI-authorized, job-scoped download proxy. |
| Final result | Reducer is not implemented | Gate behind CTX-09; show partial diagnostic artifacts meanwhile. |
| Final result | CTX-09 final artifact and download route exist | Show the `reducing` stage, then make the final CSV prominent only for `completed`. |
| Distributed graph correctness | Planner/reducer unavailable | Do not advertise a multi-shard graph as final until CTX-10. |
## 5. Proposed structure
@@ -186,11 +198,10 @@ Rules:
Inputs: exactly one `query_smiles` or `query_id`, `top_k`, optional threshold,
threshold direction, `max_rows`, and `progress_every`.
For a runnable manual pipeline check before CTX-08, offer `query_smiles` and
default `chunk_rows` large enough to create one shard. A `query_id` across
multiple shards is disabled with an explanation until CTX-07 resolves it once
before fan-out. The detail page calls an artifact a **partial top-k CSV**, not
a global top-k, until CTX-09 reduction exists.
The current upload form accepts `query_smiles`, because resolving a
cross-shard `query_id` has not yet been connected to coordinator uploads. The
detail page calls an artifact a **partial top-k CSV** until all shards are
complete and CTX-09 reduction stores the final global result.
### 8.3 Similarity graph
@@ -283,7 +294,7 @@ checksum/size metadata display, and prominent partial/final labels.
file; `Content-Disposition` is safe; preview never loads an unbounded CSV; no
final-result button exists before CTX-09.
### WUI-06 — Final-result UX after CTX-09
### WUI-06 — Final-result UX after CTX-09 — implemented
**Depends on:** CTX-09 and WUI-05.
@@ -365,8 +376,9 @@ that the interface exists today.
## 13. Definition of done for the first hand-testable release
WUI-00 through WUI-05 are complete when a clean local checkout can run a
trusted, authenticated local UI; display coordinator readiness, workers, jobs,
tasks and safe errors; submit a valid small search pipeline check; poll it to a
terminal task state; and download/preview the coordinator-owned partial CSV.
The page must make the absence of final reduction impossible to miss.
The hand-testable release is complete when a clean local checkout can run a
trusted, authenticated local UI; display workers, jobs, pipeline stages, tasks
and safe errors; submit a valid small search; poll it through `reducing`; and
download the coordinator-owned final CSV only after completion. The page must
make the distinction between partial diagnostics and the final result
impossible to miss.
+43 -7
View File
@@ -7,9 +7,11 @@ import http.client
import json
from pathlib import Path
from typing import Protocol
from urllib.error import HTTPError
from urllib.parse import quote, urljoin, urlsplit
from urllib.request import Request, build_opener
from .auth import StaticTokenProvider, TokenProvider
from .coordinator import CoordinatorConflictError
from .models import ClaimedTask, ProducedArtifact, UploadedArtifact
from .transport import SameOriginAuthRedirectHandler, origin
@@ -29,23 +31,51 @@ class ArtifactClient(Protocol):
class HttpArtifactClient:
"""Transfers artifacts through the coordinator without leaking credentials."""
def __init__(self, coordinator_url: str, timeout: float, bearer_token: str | None = None) -> None:
def __init__(
self,
coordinator_url: str,
timeout: float,
bearer_token: str | None = None,
*,
token_provider: TokenProvider | None = None,
) -> None:
self.coordinator_url = coordinator_url.rstrip("/")
self.timeout = timeout
self.bearer_token = bearer_token
self._tokens: TokenProvider = token_provider or StaticTokenProvider(bearer_token)
self.coordinator_origin = origin(coordinator_url)
self._opener = build_opener(SameOriginAuthRedirectHandler(self.coordinator_origin))
@property
def bearer_token(self) -> str | None:
return self._tokens.token()
def download(self, uri: str, destination: Path) -> None:
destination.parent.mkdir(parents=True, exist_ok=True)
resolved_uri = urljoin(f"{self.coordinator_url}/", uri)
self._download_once(resolved_uri, destination, allow_refresh=True)
def _download_once(self, resolved_uri: str, destination: Path, *, allow_refresh: bool) -> None:
request = Request(resolved_uri, headers=self._auth_headers_for(resolved_uri))
with self._opener.open(request, timeout=self.timeout) as response, destination.open("wb") as target:
while chunk := response.read(1024 * 1024):
target.write(chunk)
try:
with self._opener.open(request, timeout=self.timeout) as response, destination.open("wb") as target:
while chunk := response.read(1024 * 1024):
target.write(chunk)
except HTTPError as error:
# Refresh an expired token and retry once, mirroring the coordinator
# client, so a token that lapses mid-task does not fail the download.
if error.code == 401 and allow_refresh:
self._tokens.refresh()
self._download_once(resolved_uri, destination, allow_refresh=False)
return
raise
def upload(
self, task: ClaimedTask, worker_id: str, artifact: ProducedArtifact
) -> UploadedArtifact:
return self._upload_once(task, worker_id, artifact, allow_refresh=True)
def _upload_once(
self, task: ClaimedTask, worker_id: str, artifact: ProducedArtifact, *, allow_refresh: bool
) -> UploadedArtifact:
"""Stream an artifact and require durable coordinator-owned metadata."""
url = (
@@ -76,6 +106,11 @@ class HttpArtifactClient:
connection.send(chunk)
response = connection.getresponse()
body = response.read()
if response.status == 401 and allow_refresh:
# Token lapsed mid-task: refresh and retry the upload once.
self._tokens.refresh()
connection.close()
return self._upload_once(task, worker_id, artifact, allow_refresh=False)
if response.status == 409:
raise CoordinatorConflictError("artifact upload rejected because the task lease was lost")
if response.status != 200:
@@ -93,8 +128,9 @@ class HttpArtifactClient:
def _auth_headers_for(self, uri: str) -> dict[str, str]:
"""Only coordinator-owned URLs receive the coordinator bearer token."""
if self.bearer_token and origin(uri) == self.coordinator_origin:
return {"Authorization": f"Bearer {self.bearer_token}"}
token = self._tokens.token()
if token and origin(uri) == self.coordinator_origin:
return {"Authorization": f"Bearer {token}"}
return {}
def sha256_file(path: Path) -> str:
+128
View File
@@ -0,0 +1,128 @@
"""Bearer-token strategies for the worker's coordinator calls.
A worker authenticates in one of two ways:
* a *static* token the shared service token or a directly supplied JWT, fixed
for the life of the process; or
* a *worker key* a long-lived per-user credential the worker trades for a
short-lived JWT at the userservice, refreshing before that JWT expires.
Both are exposed through the small ``TokenProvider`` protocol so the HTTP
clients neither know nor care which one is in play.
"""
from __future__ import annotations
import json
import time
from typing import Callable, Protocol
from urllib.error import HTTPError, URLError
from urllib.request import Request, build_opener
from .transport import NoRedirectHandler
class TokenExchangeError(RuntimeError):
"""The userservice refused or failed to exchange a worker key."""
class TokenProvider(Protocol):
def token(self) -> str | None:
"""Return the current bearer token, refreshing it if necessary."""
def refresh(self) -> None:
"""Force the next token to be re-fetched (e.g. after a 401)."""
class StaticTokenProvider:
"""Serves a fixed token forever. ``None`` means "send no Authorization"."""
def __init__(self, token: str | None) -> None:
self._token = token
def token(self) -> str | None:
return self._token
def refresh(self) -> None: # noqa: D401 - nothing to refresh
return None
class WorkerKeyTokenProvider:
"""Exchanges a long-lived worker key for short-lived JWTs and refreshes them.
The token is cached until roughly ``1 - refresh_leeway`` of its lifetime has
elapsed, so the worker renews ahead of expiry instead of waiting for a 401.
A monotonic clock is injectable to keep tests deterministic.
"""
def __init__(
self,
userservice_url: str,
worker_key: str,
timeout: float,
*,
refresh_leeway: float = 0.2,
now: Callable[[], float] = time.monotonic,
) -> None:
self._url = userservice_url.rstrip("/")
self._key = worker_key
self._timeout = timeout
self._leeway = refresh_leeway
self._now = now
self._token: str | None = None
self._refresh_at: float = 0.0
self._opener = build_opener(NoRedirectHandler())
def token(self) -> str:
if self._token is None or self._now() >= self._refresh_at:
self._exchange()
assert self._token is not None # _exchange sets it or raises
return self._token
def refresh(self) -> None:
self._exchange()
def _exchange(self) -> None:
request = Request(
f"{self._url}/worker-tokens/exchange",
data=json.dumps({"key": self._key}).encode(),
method="POST",
headers={"Content-Type": "application/json"},
)
try:
with self._opener.open(request, timeout=self._timeout) as response:
raw = response.read()
data = json.loads(raw) if raw else {}
except HTTPError as error:
# A revoked or unknown key is a permanent 401; there is nothing the
# worker can do but stop, so surface it rather than retry forever.
raise TokenExchangeError(
f"worker key exchange rejected with status {error.code}"
) from error
except (URLError, TimeoutError, json.JSONDecodeError) as error:
raise TokenExchangeError("worker key exchange request failed") from error
token = data.get("token")
if not isinstance(token, str) or not token:
raise TokenExchangeError("worker key exchange response is missing a token")
expires_in = data.get("expires_in")
ttl = float(expires_in) if isinstance(expires_in, (int, float)) and expires_in > 0 else 0.0
self._token = token
# Renew once ~(1 - leeway) of the lifetime is gone. An unknown TTL falls
# back to re-exchanging on the next call — correct, just chattier.
self._refresh_at = self._now() + ttl * (1.0 - self._leeway)
def provider_from_config(
*,
worker_key: str | None,
userservice_url: str | None,
bearer_token: str | None,
request_timeout: float,
) -> TokenProvider:
"""Pick the token strategy: a worker key (exchange mode) wins over a static
bearer token, which in turn wins over no credential at all."""
if worker_key and userservice_url:
return WorkerKeyTokenProvider(userservice_url, worker_key, request_timeout)
return StaticTokenProvider(bearer_token)
+26 -4
View File
@@ -7,6 +7,7 @@ import logging
from pathlib import Path
from .artifacts import HttpArtifactClient
from .auth import provider_from_config
from .config import WorkerConfig
from .coordinator import HttpCoordinatorClient
from .daemon import WorkerDaemon
@@ -22,14 +23,23 @@ def build_parser() -> argparse.ArgumentParser:
"SCIMESH_WORKER_NAME, SCIMESH_CPU_COUNT, SCIMESH_MEMORY_MB, "
"SCIMESH_POLL_INTERVAL, SCIMESH_REQUEST_TIMEOUT, "
"SCIMESH_HEARTBEAT_INTERVAL, SCIMESH_CLEANUP_AFTER_SECONDS, "
"SCIMESH_MAX_TASKS, and SCIMESH_BEARER_TOKEN. "
"SCIMESH_WORKER_ID is a legacy/test override."
"SCIMESH_MAX_TASKS, SCIMESH_BEARER_TOKEN, SCIMESH_WORKER_KEY, and "
"SCIMESH_USERSERVICE_URL. SCIMESH_WORKER_ID is a legacy/test override."
),
)
parser.add_argument("--coordinator-url")
parser.add_argument("--worker-id")
parser.add_argument("--work-dir")
parser.add_argument("--worker-name")
parser.add_argument(
"--worker-key",
help="Long-lived worker key from the web UI; the worker exchanges it for "
"short-lived tokens, binding it to your account. Requires --userservice-url.",
)
parser.add_argument(
"--userservice-url",
help="Base URL of the userservice that issues tokens for --worker-key",
)
parser.add_argument("--cpu-count", type=int)
parser.add_argument("--memory-mb", type=int)
parser.add_argument("--poll-interval", type=float)
@@ -68,11 +78,23 @@ def main(argv: list[str] | None = None) -> int:
except (TypeError, ValueError) as error:
parser.error(str(error))
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
client = HttpCoordinatorClient(config.coordinator_url, config.request_timeout, config.bearer_token)
# One shared token strategy backs both clients: a worker key (exchanged and
# refreshed) or a static bearer token, decided by what the config carries.
tokens = provider_from_config(
worker_key=config.worker_key,
userservice_url=config.userservice_url,
bearer_token=config.bearer_token,
request_timeout=config.request_timeout,
)
client = HttpCoordinatorClient(
config.coordinator_url, config.request_timeout, token_provider=tokens
)
completed_without_interruption = WorkerDaemon(
config,
client,
HttpArtifactClient(config.coordinator_url, config.request_timeout, config.bearer_token),
HttpArtifactClient(
config.coordinator_url, config.request_timeout, token_provider=tokens
),
SciMeshRunner(),
).run_forever()
return 0 if completed_without_interruption else 130
+21
View File
@@ -11,6 +11,14 @@ from typing import Mapping
from urllib.parse import urlsplit
def _clean_url(value: object | None) -> str | None:
"""Normalise an optional URL: drop a blank one, strip a trailing slash."""
if value is None:
return None
text = str(value).strip()
return text.rstrip("/") or None
def _positive_number(value: object, name: str, *, allow_zero: bool = False) -> None:
if (
isinstance(value, bool)
@@ -35,6 +43,11 @@ class WorkerConfig:
request_timeout: float = 30.0
heartbeat_interval: float = 15.0
bearer_token: str | None = None
# A long-lived per-user credential. When set (with userservice_url), the
# worker exchanges it for short-lived JWTs instead of using bearer_token,
# binding the worker to that user's account.
worker_key: str | None = None
userservice_url: str | None = None
cleanup_after_seconds: float | None = None
max_tasks: int | None = None
exit_when_idle: bool = False
@@ -54,6 +67,12 @@ class WorkerConfig:
raise ValueError("coordinator_url must be an absolute HTTP(S) URL")
if not isinstance(self.worker_name, str) or not self.worker_name.strip():
raise ValueError("worker_name must be non-empty")
if self.userservice_url is not None:
us = urlsplit(self.userservice_url)
if us.scheme not in {"http", "https"} or not us.hostname:
raise ValueError("userservice_url must be an absolute HTTP(S) URL")
if self.worker_key is not None and not self.userservice_url:
raise ValueError("worker_key requires userservice_url (SCIMESH_USERSERVICE_URL)")
if isinstance(self.cpu_count, bool) or not isinstance(self.cpu_count, int) or self.cpu_count < 1:
raise ValueError("cpu_count must be positive")
if self.worker_id is not None and not isinstance(self.worker_id, str):
@@ -114,6 +133,8 @@ class WorkerConfig:
request_timeout=float(value("request_timeout", "SCIMESH_REQUEST_TIMEOUT", "30")),
heartbeat_interval=float(value("heartbeat_interval", "SCIMESH_HEARTBEAT_INTERVAL", "15")),
bearer_token=value("bearer_token", "SCIMESH_BEARER_TOKEN"),
worker_key=value("worker_key", "SCIMESH_WORKER_KEY"),
userservice_url=_clean_url(value("userservice_url", "SCIMESH_USERSERVICE_URL")),
cleanup_after_seconds=float(cleanup) if cleanup else None,
max_tasks=int(max_tasks) if max_tasks is not None else None,
exit_when_idle=bool(values.get("exit_when_idle", False)),
+28 -3
View File
@@ -7,6 +7,7 @@ from typing import Any, Protocol
from urllib.error import HTTPError, URLError
from urllib.request import Request, build_opener
from .auth import StaticTokenProvider, TokenProvider
from .models import ClaimedTask, RegisteredWorker
from .transport import NoRedirectHandler
@@ -38,12 +39,25 @@ class CoordinatorClient(Protocol):
class HttpCoordinatorClient:
def __init__(self, base_url: str, timeout: float, bearer_token: str | None = None) -> None:
def __init__(
self,
base_url: str,
timeout: float,
bearer_token: str | None = None,
*,
token_provider: TokenProvider | None = None,
) -> None:
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.bearer_token = bearer_token
# A bearer_token argument keeps older call sites working; internally
# everything goes through a provider so refresh is uniform.
self._tokens: TokenProvider = token_provider or StaticTokenProvider(bearer_token)
self._opener = build_opener(NoRedirectHandler())
@property
def bearer_token(self) -> str | None:
return self._tokens.token()
def register(
self, name: str, capabilities: tuple[str, ...], cpu_count: int, memory_mb: int | None
) -> RegisteredWorker:
@@ -102,6 +116,11 @@ class HttpCoordinatorClient:
return lease_expires_at
def _request(self, method: str, path: str, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]:
return self._request_once(method, path, payload, allow_refresh=True)
def _request_once(
self, method: str, path: str, payload: dict[str, Any], *, allow_refresh: bool
) -> tuple[int, dict[str, Any]]:
request = Request(
f"{self.base_url}{path}", data=json.dumps(payload).encode(), method=method,
headers={"Content-Type": "application/json", **self._auth_header()},
@@ -114,6 +133,11 @@ class HttpCoordinatorClient:
except json.JSONDecodeError as error:
raise CoordinatorError("coordinator returned invalid JSON") from error
except HTTPError as error:
# A 401 usually means the short-lived JWT expired; mint a fresh one
# and retry exactly once so an in-flight worker rides over the gap.
if error.code == 401 and allow_refresh:
self._tokens.refresh()
return self._request_once(method, path, payload, allow_refresh=False)
if error.code >= 500:
raise CoordinatorTransientError(f"coordinator returned {error.code}") from error
return error.code, {}
@@ -121,4 +145,5 @@ class HttpCoordinatorClient:
raise CoordinatorTransientError("coordinator request failed") from error
def _auth_header(self) -> dict[str, str]:
return {"Authorization": f"Bearer {self.bearer_token}"} if self.bearer_token else {}
token = self._tokens.token()
return {"Authorization": f"Bearer {token}"} if token else {}
+16
View File
@@ -52,6 +52,22 @@ HELP_TEXT = dedent(
--block-size 250 \\
--output results/similarity_graph.csv
6. Start a local distributed pipeline demo with a coordinator, PostgreSQL,
and two reference workers. Run this from the repository root after
installing the project with its development dependencies:
make demo-ui
# Use more workers when you want to compare throughput:
make demo-ui WORKERS=3
# Open http://localhost:18080/ui
# Username: operator Password: demo-ui-secret
The job page includes a live completed-shards-per-minute chart. It is
measured from coordinator snapshots observed by the open browser tab.
Stop the local demo with: make demo-down
Use --max-rows for quick local tests; omit it to process the full dataset.
For all options, run:
+1
View File
@@ -9,3 +9,4 @@ def test_help_command_prints_runnable_examples(capsys: object) -> None:
assert "scimesh similarity-search" in output
assert "scimesh similarity-graph" in output
assert "mkdir -p results" in output
assert "make demo-ui" in output
+205
View File
@@ -0,0 +1,205 @@
"""Tests for worker token strategies and the client's 401 refresh."""
from __future__ import annotations
import json
from pathlib import Path
from urllib.error import HTTPError
import pytest
from scimesh.worker.auth import (
StaticTokenProvider,
TokenExchangeError,
WorkerKeyTokenProvider,
provider_from_config,
)
from scimesh.worker.config import WorkerConfig
from scimesh.worker.coordinator import HttpCoordinatorClient
class FakeResponse:
def __init__(self, status: int, body: bytes) -> None:
self.status = status
self._body = body
def read(self) -> bytes:
return self._body
def __enter__(self) -> "FakeResponse":
return self
def __exit__(self, *exc) -> bool:
return False
class SeqOpener:
"""Returns/raises a scripted sequence of responses, recording each request."""
def __init__(self, actions: list) -> None:
self.actions = list(actions)
self.requests: list = []
def open(self, request, timeout=None):
self.requests.append(request)
action = self.actions.pop(0)
if isinstance(action, Exception):
raise action
return action
def _exchange_response(token: str, expires_in: int) -> FakeResponse:
return FakeResponse(200, json.dumps({"token": token, "expires_in": expires_in}).encode())
def test_static_provider_returns_fixed_token_and_never_refreshes():
provider = StaticTokenProvider("tok")
assert provider.token() == "tok"
provider.refresh()
assert provider.token() == "tok"
def test_static_provider_none_means_no_auth():
assert StaticTokenProvider(None).token() is None
def test_worker_key_provider_exchanges_once_then_caches():
clock = {"t": 1000.0}
provider = WorkerKeyTokenProvider(
"http://users", "scimesh_wk_live_x", timeout=5, now=lambda: clock["t"]
)
provider._opener = SeqOpener([_exchange_response("jwt-1", 100)])
# First call exchanges; a second call well within the TTL reuses the cache.
assert provider.token() == "jwt-1"
clock["t"] = 1050.0 # 50s later, TTL 100s with 0.2 leeway → refresh at +80s
assert provider.token() == "jwt-1"
assert len(provider._opener.requests) == 1
def test_worker_key_provider_refreshes_after_leeway():
clock = {"t": 0.0}
provider = WorkerKeyTokenProvider(
"http://users", "k", timeout=5, now=lambda: clock["t"]
)
provider._opener = SeqOpener([
_exchange_response("jwt-1", 100),
_exchange_response("jwt-2", 100),
])
assert provider.token() == "jwt-1"
clock["t"] = 85.0 # past the 80s refresh point
assert provider.token() == "jwt-2"
assert len(provider._opener.requests) == 2
def test_worker_key_provider_force_refresh():
provider = WorkerKeyTokenProvider("http://users", "k", timeout=5, now=lambda: 0.0)
provider._opener = SeqOpener([
_exchange_response("jwt-1", 100),
_exchange_response("jwt-2", 100),
])
assert provider.token() == "jwt-1"
provider.refresh()
assert provider.token() == "jwt-2"
def test_worker_key_provider_raises_on_rejected_key():
provider = WorkerKeyTokenProvider("http://users", "bad", timeout=5, now=lambda: 0.0)
provider._opener = SeqOpener([HTTPError("http://users", 401, "unauthorized", {}, None)])
with pytest.raises(TokenExchangeError):
provider.token()
def test_worker_key_provider_raises_when_token_missing():
provider = WorkerKeyTokenProvider("http://users", "k", timeout=5, now=lambda: 0.0)
provider._opener = SeqOpener([FakeResponse(200, json.dumps({"expires_in": 100}).encode())])
with pytest.raises(TokenExchangeError):
provider.token()
def test_provider_from_config_selects_worker_key_mode():
provider = provider_from_config(
worker_key="scimesh_wk_live_x",
userservice_url="http://users",
bearer_token="ignored",
request_timeout=5,
)
assert isinstance(provider, WorkerKeyTokenProvider)
def test_provider_from_config_falls_back_to_static():
provider = provider_from_config(
worker_key=None, userservice_url=None, bearer_token="tok", request_timeout=5
)
assert isinstance(provider, StaticTokenProvider)
assert provider.token() == "tok"
class RefreshCountingProvider:
def __init__(self) -> None:
self.tokens = ["stale", "fresh"]
self.index = 0
self.refreshes = 0
def token(self) -> str:
return self.tokens[min(self.index, len(self.tokens) - 1)]
def refresh(self) -> None:
self.refreshes += 1
self.index += 1
def test_coordinator_client_refreshes_and_retries_once_on_401():
provider = RefreshCountingProvider()
client = HttpCoordinatorClient("http://coord", timeout=5, token_provider=provider)
client._opener = SeqOpener([
HTTPError("http://coord/tasks/claim", 401, "unauthorized", {}, None),
FakeResponse(204, b""),
])
status, _ = client._request("POST", "/tasks/claim", {"worker_id": "w"})
assert status == 204
assert provider.refreshes == 1
# The retry carried the refreshed token.
assert provider.index == 1
def test_coordinator_client_does_not_loop_on_persistent_401():
provider = RefreshCountingProvider()
client = HttpCoordinatorClient("http://coord", timeout=5, token_provider=provider)
client._opener = SeqOpener([
HTTPError("http://coord/x", 401, "unauthorized", {}, None),
HTTPError("http://coord/x", 401, "unauthorized", {}, None),
])
status, _ = client._request("POST", "/x", {})
# One refresh, one retry, then the second 401 is surfaced rather than retried.
assert status == 401
assert provider.refreshes == 1
def _base_config(**extra) -> dict:
return {
"coordinator_url": "http://coord",
"worker_id": None,
"work_dir": Path("."),
**extra,
}
def test_worker_key_requires_userservice_url():
with pytest.raises(ValueError, match="userservice_url"):
WorkerConfig(**_base_config(worker_key="scimesh_wk_live_x"))
def test_worker_key_with_userservice_url_is_valid():
cfg = WorkerConfig(**_base_config(worker_key="scimesh_wk_live_x", userservice_url="http://users"))
assert cfg.worker_key == "scimesh_wk_live_x"
assert cfg.userservice_url == "http://users"
def test_userservice_url_must_be_absolute():
with pytest.raises(ValueError, match="userservice_url"):
WorkerConfig(**_base_config(userservice_url="not-a-url"))
+16
View File
@@ -0,0 +1,16 @@
# Keep the build context small and never bake secrets or local state into an image.
.env
.git
.gitignore
*.md
Makefile
docker-compose.yml
Dockerfile
.dockerignore
# Local build artifacts
/userservice
/bin/
*.out
/data/
/logs/
+29
View File
@@ -0,0 +1,29 @@
# Copy to .env and adjust. All settings are read from the environment.
USERSERVICE_ADDR=:8081
DATABASE_URL=postgres://scimesh:scimesh@localhost:5433/scimesh_users?sslmode=disable
# Shared HS256 secret used to sign JWTs. The coordinator verifies tokens with
# this SAME secret, so the two values must match exactly. Minimum 32 bytes.
JWT_SECRET=change-me-to-a-long-random-secret-min-32-bytes
# How long an issued token stays valid.
JWT_TTL=24h
# bcrypt work factor. Empty/0 uses the library default (10).
# BCRYPT_COST=10
# First-admin bootstrap. When both are set and no such account exists, the
# service creates it with role=admin on startup (idempotent). This is the only
# way to get the first admin. Leave empty in production once seeded.
# BOOTSTRAP_ADMIN_EMAIL=root@scimesh.local
# BOOTSTRAP_ADMIN_PASSWORD=change-me-strong
# Logging. LOG_LEVEL: debug|info|warn|error. LOG_FILE empty = stdout only;
# set a path to also write a size-rotated file (kept across restarts).
LOG_LEVEL=info
# LOG_FILE=./logs/userservice.log
# Optional tuning (defaults shown).
DB_MAX_CONNS=10
# How long to keep retrying the initial DB connection while Postgres boots.
DB_CONNECT_TIMEOUT=30s
REQUEST_TIMEOUT=15s
+6
View File
@@ -0,0 +1,6 @@
/userservice
/bin/
.env
*.out
/logs/
/data/
+54
View File
@@ -0,0 +1,54 @@
version: "2"
run:
timeout: 3m
linters:
# "standard" = errcheck, govet, ineffassign, staticcheck, unused.
default: standard
enable:
# Catches `err == ErrFoo` where errors.Is is required. Directly relevant
# here: domain exposes sentinel errors that use cases may wrap with %w.
- errorlint
# Returning nil after checking a non-nil error — a silent bug factory.
- nilerr
# http.Get/Do without a context: every outbound call must be cancellable.
- noctx
# Unclosed response bodies leak connections.
- bodyclose
# Common security mistakes (weak crypto, unhandled file perms).
- gosec
# Style and naming consistency.
- revive
- misspell
- unconvert
settings:
errcheck:
# Deferred Close/Rollback are intentionally ignored in a few places
# (rollback after commit is a documented no-op).
check-type-assertions: true
revive:
rules:
- name: exported
disabled: true # internal packages need no exported-symbol comments
gosec:
excludes:
- G404 # math/rand is fine for jitter; nothing here is security-sensitive
exclusions:
rules:
# Tests may skip error checks and use long literals freely.
- path: _test\.go
linters:
- errcheck
- gosec
formatters:
enable:
- gofmt
- goimports
settings:
goimports:
local-prefixes:
- github.com/emil28092005/SciMesh/users
+52
View File
@@ -0,0 +1,52 @@
# syntax=docker/dockerfile:1
#
# Requires BuildKit (the RUN --mount cache lines below). Docker 23+ enables it
# by default when the buildx plugin is present; install `docker-buildx` if a
# build fails with "the --mount option requires BuildKit".
# --- build stage ----------------------------------------------------------
FROM golang:1.24-alpine AS build
WORKDIR /src
# Copy manifests first: this layer stays cached until dependencies actually
# change, so editing Go sources does not re-download the module graph.
COPY go.mod go.sum ./
RUN --mount=type=cache,target=/go/pkg/mod go mod download
COPY . .
# The cache mounts persist the module cache and the compiler's build cache
# *across* builds, so a rebuild after a code edit recompiles only what changed
# instead of the whole dependency tree.
#
# CGO_ENABLED=0 produces a fully static binary, so the runtime image needs no
# libc. -trimpath strips local paths; -s -w drop the symbol table and DWARF.
RUN --mount=type=cache,target=/go/pkg/mod \
--mount=type=cache,target=/root/.cache/go-build \
CGO_ENABLED=0 GOOS=linux go build \
-trimpath -ldflags="-s -w" \
-o /out/userservice ./cmd/userservice
# --- runtime stage --------------------------------------------------------
FROM alpine:3.20
# ca-certificates for outbound TLS; wget backs the container healthcheck.
RUN apk add --no-cache ca-certificates wget \
&& adduser -D -H -u 10001 userservice \
# Pre-create the log dir owned by the non-root user. A named volume mounted
# here inherits this ownership from the image, so the process can write to it —
# a host bind mount, owned by root, could not.
&& mkdir -p /var/log/scimesh \
&& chown -R userservice:userservice /var/log/scimesh
COPY --from=build /out/userservice /usr/local/bin/userservice
# Never run as root: a compromised process should not own the container.
USER userservice
EXPOSE 8081
# Exec form, not shell: the binary becomes PID 1 and receives SIGTERM directly,
# which is what its graceful shutdown depends on.
ENTRYPOINT ["/usr/local/bin/userservice"]

Some files were not shown because too many files have changed in this diff Show More