cap/picv rqueue operability v2

.env.example

@@ -12,6 +12,12 @@ ALLOWED_ORIGINS=
 # Password for the runtime web role. Generate with: openssl rand -hex 32
 APP_DB_PASSWORD=
 
+# Passwords for the least-privilege queue roles the API, worker, and retention
+# pass connect as. Generate each with: openssl rand -hex 32
+COMPUTE_PRODUCER_PASSWORD=
+COMPUTE_WORKER_PASSWORD=
+COMPUTE_PURGER_PASSWORD=
+
 # Browser-reachable MinIO endpoint for output download URLs.
 MINIO_PUBLIC_ENDPOINT=localhost:9000
 

.github/workflows/crash-recovery.yml

@@ -20,6 +20,9 @@ jobs:
         env:
             COMPUTE_API_TOKEN: compose-e2e-token
             APP_DB_PASSWORD: compose-e2e-app-db-password
+            COMPUTE_PRODUCER_PASSWORD: compose-e2e-producer-password
+            COMPUTE_WORKER_PASSWORD: compose-e2e-worker-password
+            COMPUTE_PURGER_PASSWORD: compose-e2e-purger-password
             BETTER_AUTH_SECRET: compose-e2e-better-auth-secret
             ORIGIN: http://localhost:3000
             MINIO_ACCESS_KEY: minioadmin

.github/workflows/release.yml

@@ -235,6 +235,9 @@ jobs:
             SIMULATION_ID: 4cfe522f-7e7d-46e0-96ca-7b98743fb9f5
             COMPUTE_API_TOKEN: compose-e2e-token
             APP_DB_PASSWORD: compose-e2e-app-db-password
+            COMPUTE_PRODUCER_PASSWORD: compose-e2e-producer-password
+            COMPUTE_WORKER_PASSWORD: compose-e2e-worker-password
+            COMPUTE_PURGER_PASSWORD: compose-e2e-purger-password
             BETTER_AUTH_SECRET: compose-e2e-better-auth-secret
             ORIGIN: http://localhost:3000
             MINIO_ACCESS_KEY: minioadmin

ARCHITECTURE.md

@@ -103,6 +103,31 @@ The web runtime role can read and write the web tables and read
 changes. The table definition in `apps/web/src/lib/server/db/compute.ts` is
 used only for reads and is excluded from web migrations.
 
+The compute service has three restricted runtime roles, provisioned after the
+compute and queue migrations by `tsdhn-queue-grants`. None can run schema
+changes, and row-level security scopes each to the deployment's queue:
+
+| Role | Process | Queue capability | `compute.jobs` |
+| --- | --- | --- | --- |
+| `COMPUTE_PRODUCER_ROLE` | API | enqueue and read (`PRODUCE`) | `SELECT`, `INSERT` |
+| `COMPUTE_WORKER_ROLE` | worker | claim and transition (`CONSUME`) | `SELECT`, `UPDATE` |
+| `COMPUTE_PURGER_ROLE` | worker retention | inspect plus delete queue jobs | none |
+
+The producer cannot claim a job, the worker cannot delete one, and the purger
+cannot reach `compute.jobs`. Deleting a queue job cascades to its attempt
+history, so retention uses its own credential rather than the consumer role.
+
+## Queue retention
+
+Terminal `task_queue.jobs` rows whose matching `compute.jobs` row is also
+terminal are deleted after seven days by an hourly pass in the worker process.
+Rows whose compute counterpart is still running, missing, or malformed remain
+available for reconciliation. The queue row has no application value once the
+handler records the outcome; the week preserves attempt history for operational
+inspection over a working week and weekend. Attempt and occurrence rows follow
+through `ON DELETE CASCADE`. `compute.jobs` is not purged because it holds the
+simulation result record.
+
 ## Identifiers
 
 | Name | Owner | Purpose |

docker-compose.yml

@@ -44,14 +44,21 @@ services:
         condition: service_healthy
     environment:
       COMPUTE_DATABASE_URL: postgresql://tsdhn:tsdhn@postgres:5432/tsdhn
+      COMPUTE_QUEUE: ${COMPUTE_QUEUE:-simulations}
       COMPUTE_QUEUE_SCHEMA: ${COMPUTE_QUEUE_SCHEMA:-task_queue}
       APP_DB_ROLE: ${APP_DB_ROLE:-tsdhn_app}
       APP_DB_PASSWORD: ${APP_DB_PASSWORD:?set APP_DB_PASSWORD in .env}
+      COMPUTE_PRODUCER_ROLE: ${COMPUTE_PRODUCER_ROLE:-tsdhn_producer}
+      COMPUTE_PRODUCER_PASSWORD: ${COMPUTE_PRODUCER_PASSWORD:?set COMPUTE_PRODUCER_PASSWORD in .env}
+      COMPUTE_WORKER_ROLE: ${COMPUTE_WORKER_ROLE:-tsdhn_worker}
+      COMPUTE_WORKER_PASSWORD: ${COMPUTE_WORKER_PASSWORD:?set COMPUTE_WORKER_PASSWORD in .env}
+      COMPUTE_PURGER_ROLE: ${COMPUTE_PURGER_ROLE:-tsdhn_purger}
+      COMPUTE_PURGER_PASSWORD: ${COMPUTE_PURGER_PASSWORD:?set COMPUTE_PURGER_PASSWORD in .env}
     command:
       [
         "sh",
         "-lc",
-        "uv run --no-dev tsdhn-compute-migrate && uv run --no-dev rqueue --database-url \"$$COMPUTE_DATABASE_URL\" --schema \"$$COMPUTE_QUEUE_SCHEMA\" migrate",
+        "uv run --no-dev tsdhn-compute-migrate && uv run --no-dev rqueue --database-url \"$$COMPUTE_DATABASE_URL\" --schema \"$$COMPUTE_QUEUE_SCHEMA\" migrate && uv run --no-dev tsdhn-queue-grants",
       ]
     restart: "no"
 
@@ -79,6 +86,8 @@ services:
       MINIO_BUCKET: ${MINIO_BUCKET:-tsdhn-results}
       COMPUTE_API_TOKEN: ${COMPUTE_API_TOKEN:?set COMPUTE_API_TOKEN in .env}
       ALLOWED_ORIGINS: ${ALLOWED_ORIGINS:-}
+      COMPUTE_PRODUCER_ROLE: ${COMPUTE_PRODUCER_ROLE:-tsdhn_producer}
+      COMPUTE_PRODUCER_PASSWORD: ${COMPUTE_PRODUCER_PASSWORD:?set COMPUTE_PRODUCER_PASSWORD in .env}
     ports:
       - "8000:8000"
 
@@ -105,6 +114,10 @@ services:
       MINIO_BUCKET: ${MINIO_BUCKET:-tsdhn-results}
       TSDHN_MODEL_DIR: /app/model
       TSDHN_JOBS_DIR: /var/tmp/jobs
+      COMPUTE_WORKER_ROLE: ${COMPUTE_WORKER_ROLE:-tsdhn_worker}
+      COMPUTE_WORKER_PASSWORD: ${COMPUTE_WORKER_PASSWORD:?set COMPUTE_WORKER_PASSWORD in .env}
+      COMPUTE_PURGER_ROLE: ${COMPUTE_PURGER_ROLE:-tsdhn_purger}
+      COMPUTE_PURGER_PASSWORD: ${COMPUTE_PURGER_PASSWORD:?set COMPUTE_PURGER_PASSWORD in .env}
     command: ["uv", "run", "--no-dev", "tsdhn-worker"]
     volumes:
       - jobs-data:/var/tmp/jobs

mise.toml

@@ -156,6 +156,7 @@ depends = ["db:start"]
 run = [
   "uv run tsdhn-compute-migrate",
   "uv run rqueue --database-url \"${COMPUTE_DATABASE_URL:-postgresql://tsdhn:tsdhn@127.0.0.1:5432/tsdhn}\" --schema \"${COMPUTE_QUEUE_SCHEMA:-task_queue}\" migrate",
+  "uv run tsdhn-queue-grants",
   "DATABASE_URL=\"${COMPUTE_DATABASE_URL:-postgresql://tsdhn:tsdhn@127.0.0.1:5432/tsdhn}\" bun --filter web db:migrate",
   "uv run tsdhn-web-grants",
 ]

packages/api/api/core/db.py

@@ -7,6 +7,7 @@
 `settings.worker_pool_size()` sizes the worker's pool the way it does.
 """
 
+import logging
 import uuid
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
@@ -15,9 +16,10 @@
 import asyncpg
 
 from api.core.errors import TransientInfraError
-from api.core.settings import COMPUTE_DATABASE_URL
+from api.core.settings import COMPUTE_DATABASE_URL, role_database_url
 
 __all__ = [
+    "COMPUTE_DATABASE_URL",
     "CONNECT_TIMEOUT",
     "JobRow",
     "acquire",
@@ -27,14 +29,21 @@
     "is_transient",
     "notify_channel",
     "open_pool",
+    "runtime_dsn",
     "transient_connection_errors",
 ]
 
+logger = logging.getLogger(__name__)
+
 JobRow = dict[str, Any]
 CONNECT_TIMEOUT = 2
 
 _pool: asyncpg.Pool | None = None
 
+# The DSN the open pool used, so `connect()` reaches the same role rather than
+# quietly falling back to the owner behind the pool's back.
+_dsn: str | None = None
+
 # Client-side failures: the server was never reached, or the socket died.
 _CLIENT_ERRORS = (ConnectionError, OSError, TimeoutError)
 
@@ -71,12 +80,27 @@ def is_transient(exc: BaseException) -> bool:
     return isinstance(sqlstate, str) and sqlstate[:2] in _TRANSIENT_SQLSTATE_CLASSES
 
 
-async def open_pool(*, min_size: int, max_size: int) -> asyncpg.Pool:
+def runtime_dsn(role: str, password: str) -> str | None:
+    """Resolve a runtime role's DSN, warning when it is not provisioned."""
+    dsn = role_database_url(role, password)
+    if dsn is None:
+        logger.warning(
+            "no password configured for role %s; connecting as the database "
+            "owner instead of the least-privilege role",
+            role or "<unset>",
+        )
+    return dsn
+
+
+async def open_pool(
+    *, min_size: int, max_size: int, dsn: str | None = None
+) -> asyncpg.Pool:
     """Create the process-wide pool. Idempotent within one process."""
-    global _pool
+    global _pool, _dsn
     if _pool is None:
+        _dsn = dsn or COMPUTE_DATABASE_URL
         _pool = await asyncpg.create_pool(
-            COMPUTE_DATABASE_URL,
+            _dsn,
             min_size=min_size,
             max_size=max_size,
             timeout=CONNECT_TIMEOUT,
@@ -85,10 +109,11 @@ async def open_pool(*, min_size: int, max_size: int) -> asyncpg.Pool:
 
 
 async def close_pool() -> None:
-    global _pool
+    global _pool, _dsn
     if _pool is not None:
         await _pool.close()
         _pool = None
+    _dsn = None
 
 
 def get_pool() -> asyncpg.Pool:
@@ -178,7 +203,9 @@ async def connect() -> asyncpg.Connection:
     watching browsers starve every other route.
     """
     try:
-        return await asyncpg.connect(COMPUTE_DATABASE_URL, timeout=CONNECT_TIMEOUT)
+        return await asyncpg.connect(
+            _dsn or COMPUTE_DATABASE_URL, timeout=CONNECT_TIMEOUT
+        )
     except Exception as e:
         if not is_transient(e):
             raise

packages/api/api/core/settings.py

@@ -1,12 +1,19 @@
 import os
 from pathlib import Path
+from urllib.parse import quote, urlsplit, urlunsplit
 
 __all__ = [
     "APP_DB_PASSWORD",
     "APP_DB_ROLE",
     "COMPUTE_DATABASE_URL",
+    "COMPUTE_PRODUCER_PASSWORD",
+    "COMPUTE_PRODUCER_ROLE",
+    "COMPUTE_PURGER_PASSWORD",
+    "COMPUTE_PURGER_ROLE",
     "COMPUTE_QUEUE",
     "COMPUTE_QUEUE_SCHEMA",
+    "COMPUTE_WORKER_PASSWORD",
+    "COMPUTE_WORKER_ROLE",
     "DB_POOL_MAX_SIZE",
     "DB_POOL_MIN_SIZE",
     "JOBS_DIR",
@@ -81,6 +88,43 @@ def worker_pool_size() -> tuple[int, int]:
 APP_DB_ROLE = os.environ.get("APP_DB_ROLE", "tsdhn_app")
 APP_DB_PASSWORD = os.environ.get("APP_DB_PASSWORD", "")
 
+# COMPUTE_DATABASE_URL names the schema owner: the role that runs migrations
+# and owns both `compute` and the queue schema. Runtime processes use these
+# roles instead, each provisioned with only the grants its process needs.
+COMPUTE_PRODUCER_ROLE = os.environ.get("COMPUTE_PRODUCER_ROLE", "tsdhn_producer")
+COMPUTE_PRODUCER_PASSWORD = os.environ.get("COMPUTE_PRODUCER_PASSWORD", "")
+
+COMPUTE_WORKER_ROLE = os.environ.get("COMPUTE_WORKER_ROLE", "tsdhn_worker")
+COMPUTE_WORKER_PASSWORD = os.environ.get("COMPUTE_WORKER_PASSWORD", "")
+
+# Retention deletes queue history, which no consuming role may do. Keeping it
+# on its own credential means a deployment can withhold it or move retention
+# to a maintenance container without changing the worker role.
+COMPUTE_PURGER_ROLE = os.environ.get("COMPUTE_PURGER_ROLE", "tsdhn_purger")
+COMPUTE_PURGER_PASSWORD = os.environ.get("COMPUTE_PURGER_PASSWORD", "")
+
+
+def role_database_url(role: str, password: str) -> str | None:
+    """Return the compute URL rewritten to connect as `role`.
+
+    A missing role or password means the role has not been provisioned. The
+    runtime caller can then retain the owner URL while warning the operator.
+    Credentials are percent-encoded because passwords commonly contain URL
+    punctuation.
+    """
+    if not role or not password:
+        return None
+    parts = urlsplit(COMPUTE_DATABASE_URL)
+    credentials = f"{quote(role, safe='')}:{quote(password, safe='')}"
+    # Preserve the authority verbatim. asyncpg accepts socket URLs with no
+    # hostname and multi-host authorities; accessing ``parts.hostname`` or
+    # ``parts.port`` would reject those valid DSN forms before asyncpg sees
+    # them. Existing userinfo is replaced by taking everything after the last
+    # @, while the raw host list (if any) remains untouched.
+    authority = parts.netloc.rsplit("@", 1)[-1]
+    return urlunsplit(parts._replace(netloc=f"{credentials}@{authority}"))
+
+
 MINIO_ENDPOINT = os.environ.get("MINIO_ENDPOINT", "localhost:9000")
 # Public endpoint differs from API endpoint for browser downloads.
 MINIO_PUBLIC_ENDPOINT = os.environ.get("MINIO_PUBLIC_ENDPOINT", MINIO_ENDPOINT)