diff --git a/CLAUDE.md b/CLAUDE.md index d3bae23..8c71e82 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -5,8 +5,9 @@ Rust (axum) API with PostgreSQL. Shipped so far: the production-grade skeleton, GitHub-OAuth authentication subsystem, the **Repository + Workflow Management modules** (GitHub App integration, webhook-driven sync, workflow YAML parsing/validation, Monaco workspace with dependency graph), and the **Pipeline Execution subsystem** (event-driven -scheduler, HMAC-signed runner WebSocket protocol, live log streaming, Cloudflare R2 -artifacts, a reference Docker runner in `runner/`, and the full Pipelines UI), and the +scheduler, HMAC-signed runner WebSocket protocol, live log streaming, S3-compatible object +storage for artifacts — MinIO as the default/primary store with Cloudflare R2 as the +fallback — a reference Docker runner in `runner/`, and the full Pipelines UI), and the **Secrets Management module** (envelope-encrypted write-only workspace/repository/ environment secrets, dispatch-time injection with unconditional log masking, dedicated `secrets.read`/`secrets.manage` RBAC, full catalog + detail UI), and the **Environments @@ -103,12 +104,23 @@ on (job, seq)) → broadcast to browser subscribers on `/ws/workspaces/{ws}/pipe (strict Origin check + session cookie + `content.read` BEFORE upgrade; snapshot then live events with `createdAt`; server pings every 30 s and answers client `{"type":"ping"}` with a pong; `log_gap` on lag → client backfills over REST, paging until exhausted). Artifacts -AND completed-job log archives live in **Cloudflare R2** (S3 API; artifact presigned -PUT/GET minted server-side, HeadObject verification before rows flip to `uploaded`; logs -gzip'd server-side to `logs/{ws}/{pipeline}/{job}-{attempt}.log.gz` on job completion — -feature-gated on R2 env, clean denial/Postgres-only without it; runner convention: files in +AND completed-job log archives live in **S3-compatible object storage** +(`services/object_store.rs`: one generic `S3Store` client with two flavors — **MinIO is +the default/primary backend whenever configured** (`services/minio.rs`; explicit endpoint, +path-style addressing, startup bucket auto-create) and **Cloudflare R2 the fallback** +(`services/r2.rs`; derived account endpoint, region `auto`) — R2 alone keeps its historical +primary role. The `Storage` router in AppState handles it: server-side writes (log +archives, logos) reactively fall back to the secondary store on failure; presigned upload +grants route via a 60 s-cached HeadBucket health probe of the primary; and every stored +object carries a `storage_backend` marker ('minio'|'r2', migration `20260716100001`) so +presigns/HeadObject/deletes always target the store that holds it — presigned URLs are +host-specific, and NULL/legacy markers read as 'r2'. Artifact presigned PUT/GET minted +server-side with the declared Content-Length bound into the signature, HeadObject +verification before rows flip to `uploaded`; logs gzip'd server-side to +`logs/{ws}/{pipeline}/{job}-{attempt}.log.gz` on job completion — feature-gated on +MinIO/R2 env, clean denial/Postgres-only without either; runner convention: files in `.overup/artifacts/`). `services/janitor.rs` (hourly) expires uploaded artifacts after -`ARTIFACT_RETENTION_DAYS` (deleting the R2 object), removes stale pending artifact rows, +`ARTIFACT_RETENTION_DAYS` (deleting the stored object), removes stale pending artifact rows, prunes archived log chunks past `LOG_HOT_RETENTION_DAYS` (the raw-log download then 307-redirects to a presigned R2 GET), and purges sessions/oauth states. The ledger list API filters on repository, workflow, status, conclusion, trigger, branch, actor, runner, date @@ -350,14 +362,16 @@ overup/ │ # secrets (ciphertext-only + RBAC backfill), │ # environments (+secrets.environment_id scope + │ # RBAC backfill), secret value_set_at rotation clock, - │ # notifications (+preferences/projector cursor) + │ # notifications (+preferences/projector cursor), + │ # storage_backend markers (artifacts/pipeline_jobs/ + │ # workspaces — MinIO/R2 object routing) └── src/ ├── main.rs # bootstrap: env, tracing, pool, migrate, orphan recovery, │ # scheduler spawn, janitor, serve ├── config.rs # all env-driven configuration (GitHub App, signing key, │ # execution budgets, optional R2 group) ├── state.rs # AppState: pool, config, oauth, http, github_app, - │ # log_hub, runner_hub, scheduler, r2 + │ # log_hub, runner_hub, scheduler, storage ├── error.rs # AppError → sanitized JSON responses ├── db/ # parameterized sqlx queries only, one module per table ├── models/ # FromRow rows + camelCase response DTOs, per resource @@ -372,7 +386,9 @@ overup/ # auth_flow, workspace, authz (RBAC), repo_sync, # workflow_parse, pipeline_plan, pipeline_run (state # machine), scheduler, log_hub (mask+cap+broadcast), - # runner_hub, r2 (presign + HeadObject), + # runner_hub, object_store (generic S3Store + Storage + # router: MinIO primary / R2 fallback, presign + + # HeadObject) + minio/r2 (per-backend constructors), # secrets_crypto (AES-256-GCM envelope encryption), # notification (mapping) + notification_hub (per-user # fan-out) + notification_projector (audit tail) @@ -459,8 +475,11 @@ Secrets subsystem; DEKs and decrypted values ride in `zeroize::Zeroizing` buffer `serde_yaml_ng` (maintained serde_yaml fork; workflow parsing under strict budgets), `axum` with the **`ws` feature** (runner + browser WebSocket upgrades), `dashmap` (RunnerHub connection registry + LogHub broadcast/mask maps), -`futures-util` (WS stream splitting), `aws-sdk-s3` (Cloudflare R2 via its S3 API — custom -endpoint, region `auto`, presigned URLs; isolated in `services/r2.rs`), and the local +`futures-util` (WS stream splitting), `aws-sdk-s3` (one generic client for both object +stores — MinIO via explicit endpoint + path-style addressing, Cloudflare R2 via its +account endpoint + region `auto`; presigned URLs, HeadObject/HeadBucket; isolated in +`services/object_store.rs` with per-backend constructors in `services/minio.rs` / +`services/r2.rs`), and the local `protocol` crate (shared WS message types + HMAC helpers). The `runner/` crate adds `bollard` 0.21 (Docker Engine API: image pull, container lifecycle, exec streams), `tokio-tungstenite` (rustls), `tar` + `flate2` + `bytes` (repackaging the source tarball @@ -475,7 +494,10 @@ into a traversal-safe tar streamed into the container via the Docker archive API - Exact redirect-URI allow-list (one registered callback URL) - Code exchange server-to-server over TLS (rustls), HTTP redirects disabled - Session tokens: 32 bytes OS RNG; **only SHA-256 hashes** in the database -- Session rotation on every login; absolute expiry; hourly janitor purges expired rows +- Session rotation on every login; absolute expiry PLUS an idle timeout + (`SESSION_IDLE_TIMEOUT_HOURS`, default 72, 0 disables — rides the `last_seen_at` + column touched on every request, so a leaked token dies after inactivity); hourly + janitor purges expired AND idle-expired rows with the same predicate - Cookie: `HttpOnly`, `Secure` (prod), `SameSite=Lax`, `Path=/`; with `COOKIE_SECURE=true` the name is auto-prefixed `__Host-` (binds the cookie to the exact host — no subdomain planting/fixation) @@ -509,7 +531,20 @@ into a traversal-safe tar streamed into the container via the Docker archive API responses, never in Postgres; the private key PEM loads once at startup - Webhooks: constant-time HMAC-SHA256 over the raw body (`X-Hub-Signature-256`) before any parsing; `X-GitHub-Delivery` primary key makes redeliveries no-ops; payloads are parsed - into minimal typed envelopes and never logged + into minimal typed envelopes and never logged; `GITHUB_WEBHOOK_SECRET` must be ≥ 16 + bytes at boot (signing-key parity — a guessable secret would let anyone forge deliveries) +- Request tracing records method + PATH only (custom `MakeSpan` in routes) — the query + string carries secrets on some routes (`?code=`/`?state=` on the OAuth callback, + `?ticket=` on WS upgrades) and must never land in spans, even at debug level +- **Object storage routing is marker-driven**: every stored object (artifact, log archive, + logo) records which backend holds it (`storage_backend` columns, 'minio'|'r2', + NULL/legacy → 'r2'); presigns/HeadObject/deletes always target that store — presigned + URLs are host-specific, so a marker mix-up would 404, never leak. Server-side writes + fall back MinIO→R2 reactively; presigned upload grants route on a cached health probe. + Presigned PUTs bind the runner-declared Content-Length (and content type) into the + signed headers, so the store rejects any different body size at the edge; HeadObject + re-verifies afterwards. MINIO_ENDPOINT is validated at boot and plain http on a + non-loopback host draws a startup warning (cleartext credentials) - Setup redirect: `installation_id` is length-capped, numeric-validated, then verified against the GitHub API with an app JWT + account/installer match before linking - Workflow YAML parsing is deterministic and side-effect free: 512 KB cap, 20k node budget, @@ -585,8 +620,9 @@ into a traversal-safe tar streamed into the container via the Docker archive API - Storage hygiene: artifacts carry an immutable `expires_at` computed at upload from per-kind workspace retention policies (`artifact_retention_policies`, 1–400 days, kind row → `default` row → `ARTIFACT_RETENTION_DAYS` env); the hourly janitor deletes - expired/abandoned R2 objects and rows, and prunes archived log chunks only when the R2 - archive exists (`LOG_HOT_RETENTION_DAYS`). Artifact `kind` is classified SERVER-side + expired/abandoned stored objects (routed per-object to MinIO or R2 by marker) and rows, + and prunes archived log chunks only when the object-storage archive exists + (`LOG_HOT_RETENTION_DAYS`). Artifact `kind` is classified SERVER-side from the validated name (`services/artifact_kind.rs`); runner-reported archive manifests (entries/uncompressed size/file count) are capped (1000 entries, 96 KB JSON, 1 TiB/1M ceilings) and dropped whole on any violation — the upload itself still succeeds @@ -650,9 +686,14 @@ into a traversal-safe tar streamed into the container via the Docker archive API # (openssl rand -hex 32) enables the Secrets module — without it secret # creation is denied and pipelines for repos WITH stored secrets fail # closed (secrets_unavailable); losing it makes stored values permanently -# undecryptable (re-enter values to recover). R2_* vars are optional — -# without them pipelines run but artifact uploads are denied and logs -# stay in Postgres (no archival/pruning). Retention knobs: +# undecryptable (re-enter values to recover). Object storage is optional +# and S3-compatible with two backends: MINIO_* vars (endpoint/key/secret/ +# bucket — the local docker-compose MinIO is http://localhost:9000 with +# overup / overup-minio; bucket auto-created at startup) make MinIO the +# DEFAULT/primary store, and R2_* vars configure Cloudflare R2 as the +# fallback (or the primary when MinIO is absent). Without either, +# pipelines run but artifact uploads are denied and logs stay in +# Postgres (no archival/pruning). Retention knobs: # ARTIFACT_RETENTION_DAYS / ARTIFACT_PENDING_TTL_HOURS / # LOG_HOT_RETENTION_DAYS — per-kind artifact retention (1–400 days) is # also configurable per workspace in the Artifacts UI and takes diff --git a/backend/.env.example b/backend/.env.example index e515418..0e785b3 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -151,9 +151,33 @@ MAX_LOG_BYTES_PER_JOB=10485760 MAX_ARTIFACT_BYTES=104857600 MAX_ARTIFACTS_PER_JOB=10 -# Cloudflare R2 storage (S3-compatible API) for artifacts AND gzip'd log -# archives. All four together or none: without R2, pipelines run fine but -# artifact uploads are denied and logs simply stay in Postgres. +# Object storage for artifacts AND gzip'd log archives (and workspace +# logos). Two S3-compatible backends are supported; configure either, both, +# or neither: +# - MinIO (below) is the DEFAULT / primary store whenever it is configured. +# - Cloudflare R2 (further below) is the fallback when MinIO is also +# configured, and the primary when it is the only one. +# With both set, server-side writes fall back to R2 automatically when the +# MinIO write fails, and every object remembers which store holds it. With +# neither, pipelines run fine but artifact uploads are denied and logs +# simply stay in Postgres. + +# MinIO (or any S3-compatible endpoint). All four together or none. The +# bucket is created at startup when missing. The local docker-compose MinIO +# uses: endpoint http://localhost:9000, credentials overup / overup-minio. +# Plain http is fine on localhost; a non-loopback http endpoint draws a +# startup warning (credentials would cross the network in cleartext). +#MINIO_ENDPOINT=http://localhost:9000 +#MINIO_ACCESS_KEY_ID= +#MINIO_SECRET_ACCESS_KEY= +#MINIO_BUCKET=overup +# Optional: region (default us-east-1) and path-style addressing (default +# true — the standard MinIO deployment shape). +#MINIO_REGION=us-east-1 +#MINIO_FORCE_PATH_STYLE=true + +# Cloudflare R2. All four together or none: acts as the fallback store when +# MinIO is configured, or the primary when it is the only backend. # Create an API token at Cloudflare -> R2 -> Manage R2 API Tokens. #R2_ACCOUNT_ID= #R2_ACCESS_KEY_ID= @@ -169,9 +193,13 @@ ARTIFACT_RETENTION_DAYS=30 # Hours a never-confirmed (pending) artifact row may linger before cleanup. ARTIFACT_PENDING_TTL_HOURS=24 # Days a finished job's log chunks stay hot in Postgres after archival to -# R2 (only pruned when R2 is configured). +# object storage (only pruned when MinIO or R2 is configured). LOG_HOT_RETENTION_DAYS=7 +# Session hardening. Sessions expire after SESSION_TTL_HOURS absolutely, and +# ALSO after this many hours of inactivity (0 disables the idle timeout). +SESSION_IDLE_TIMEOUT_HOURS=72 + # Notification Center retention (independent from audit retention — the # audit ledger is permanent; notifications are operational awareness). # Days an ARCHIVED notification is kept before hard deletion. diff --git a/backend/migrations/20260716100001_storage_backend_markers.sql b/backend/migrations/20260716100001_storage_backend_markers.sql new file mode 100644 index 0000000..e59de54 --- /dev/null +++ b/backend/migrations/20260716100001_storage_backend_markers.sql @@ -0,0 +1,22 @@ +-- Per-object storage-backend markers. Presigned URLs, HeadObject and +-- DeleteObject are host-specific, so once MinIO (primary) and R2 (fallback) +-- coexist every stored object must remember which store holds it. Existing +-- rows were all written when R2 was the only store, hence the defaults; the +-- nullable columns read as 'r2' when NULL (legacy rows). + +ALTER TABLE artifacts + ADD COLUMN storage_backend TEXT NOT NULL DEFAULT 'r2' + CONSTRAINT artifacts_storage_backend_check + CHECK (storage_backend IN ('minio', 'r2')); + +-- Set alongside logs_archived_at going forward; NULL = legacy archive in R2. +ALTER TABLE pipeline_jobs + ADD COLUMN logs_archive_backend TEXT + CONSTRAINT pipeline_jobs_logs_archive_backend_check + CHECK (logs_archive_backend IN ('minio', 'r2')); + +-- Set alongside logo_key going forward; NULL = legacy logo in R2. +ALTER TABLE workspaces + ADD COLUMN logo_storage_backend TEXT + CONSTRAINT workspaces_logo_storage_backend_check + CHECK (logo_storage_backend IN ('minio', 'r2')); diff --git a/backend/src/config.rs b/backend/src/config.rs index c3d58f6..fa90df5 100644 --- a/backend/src/config.rs +++ b/backend/src/config.rs @@ -63,9 +63,17 @@ pub struct Config { pub notification_retention_days: i32, /// Days after a notification is read before the janitor auto-archives it. pub notification_auto_archive_days: i32, + /// Idle timeout for sessions: a session unused for this many hours is + /// invalid even before its absolute expiry (0 disables). Uses the + /// `last_seen_at` column that is already touched on every request. + pub session_idle_timeout_hours: i64, + /// MinIO (or any S3-compatible endpoint) storage — all-or-none optional + /// group. When configured, MinIO is the DEFAULT / primary object store; + /// R2 becomes the fallback. + pub minio: Option, /// Cloudflare R2 storage (artifacts + gzip'd log archives) — all-or-none - /// optional group; without it, artifact grants are cleanly denied and - /// logs simply stay in Postgres. + /// optional group; without it (and without MinIO), artifact grants are + /// cleanly denied and logs simply stay in Postgres. pub r2: Option, /// Hosted-runner provisioning (Docker containers spawned by the control /// plane) — optional group enabled with RUNNER_PROVISIONER=docker; @@ -88,6 +96,21 @@ pub struct R2Config { pub bucket: String, } +#[derive(Clone)] +pub struct MinioConfig { + /// Full http(s) endpoint, e.g. http://localhost:9000. Plain http on a + /// non-loopback host draws a startup warning (credentials in cleartext). + pub endpoint: String, + pub access_key_id: String, + pub secret_access_key: String, + pub bucket: String, + /// MinIO's own default region (MINIO_REGION, default us-east-1). + pub region: String, + /// Path-style addressing (MINIO_FORCE_PATH_STYLE, default true) — the + /// standard MinIO deployment shape. + pub force_path_style: bool, +} + #[derive(Clone)] pub struct RunnerProvisionerConfig { /// Runner image to run, e.g. ghcr.io/botcoder254/overup-runner:latest @@ -170,6 +193,56 @@ impl Config { anyhow::bail!("RUNNER_JOB_SIGNING_KEY must be at least 32 bytes"); } + // Parity with the signing-key rule: a guessable webhook secret would + // let anyone forge GitHub deliveries, so a weak one fails at boot. + let github_webhook_secret = required("GITHUB_WEBHOOK_SECRET")?; + if github_webhook_secret.len() < 16 { + anyhow::bail!( + "GITHUB_WEBHOOK_SECRET must be at least 16 bytes (generate one with `openssl rand -hex 32` and set it on the GitHub App too)" + ); + } + + let session_idle_timeout_hours: i64 = optional("SESSION_IDLE_TIMEOUT_HOURS", "72") + .parse() + .context("SESSION_IDLE_TIMEOUT_HOURS must be an integer (0 disables)")?; + if session_idle_timeout_hours < 0 { + anyhow::bail!("SESSION_IDLE_TIMEOUT_HOURS must be 0 (disabled) or positive"); + } + + // MinIO settings are all-or-none, mirroring the R2 group. MinIO is + // the default/primary object store whenever it is configured. + let minio_keys = [ + "MINIO_ENDPOINT", + "MINIO_ACCESS_KEY_ID", + "MINIO_SECRET_ACCESS_KEY", + "MINIO_BUCKET", + ]; + let minio_present = minio_keys + .iter() + .filter(|key| std::env::var(key).is_ok()) + .count(); + let minio = match minio_present { + 0 => None, + 4 => { + let endpoint = required("MINIO_ENDPOINT")?.trim_end_matches('/').to_string(); + validate_minio_endpoint(&endpoint)?; + let force_path_style: bool = optional("MINIO_FORCE_PATH_STYLE", "true") + .parse() + .context("MINIO_FORCE_PATH_STYLE must be true or false")?; + Some(MinioConfig { + endpoint, + access_key_id: required("MINIO_ACCESS_KEY_ID")?, + secret_access_key: required("MINIO_SECRET_ACCESS_KEY")?, + bucket: required("MINIO_BUCKET")?, + region: optional("MINIO_REGION", "us-east-1"), + force_path_style, + }) + } + _ => anyhow::bail!( + "MinIO configuration is incomplete: set all of MINIO_ENDPOINT, MINIO_ACCESS_KEY_ID, MINIO_SECRET_ACCESS_KEY, MINIO_BUCKET or none" + ), + }; + // R2 settings are all-or-none: a partial configuration is a // deployment mistake, not a feature toggle. let r2_keys = [ @@ -349,7 +422,7 @@ impl Config { .context("TRUST_PROXY must be true or false")?, github_app_client_id: required("GITHUB_APP_CLIENT_ID")?, github_app_private_key_pem, - github_webhook_secret: required("GITHUB_WEBHOOK_SECRET")?, + github_webhook_secret, github_app_slug: required("GITHUB_APP_SLUG")?, runner_job_signing_key, default_job_image, @@ -373,6 +446,8 @@ impl Config { log_hot_retention_days, notification_retention_days, notification_auto_archive_days, + session_idle_timeout_hours, + minio, r2, runner_provisioner, secrets_master_key, @@ -384,6 +459,40 @@ fn required(key: &str) -> anyhow::Result { std::env::var(key).with_context(|| format!("missing required environment variable {key}")) } +/// MINIO_ENDPOINT must be a syntactically sane http(s) URL. Plain http is +/// allowed (the standard local `docker compose` shape) but draws a loud +/// warning on non-loopback hosts: S3 credentials would cross the network in +/// cleartext. +fn validate_minio_endpoint(endpoint: &str) -> anyhow::Result<()> { + let rest = if let Some(rest) = endpoint.strip_prefix("https://") { + rest + } else if let Some(rest) = endpoint.strip_prefix("http://") { + let host = rest + .split(['/', ':']) + .next() + .unwrap_or_default() + .trim_start_matches('[') + .trim_end_matches(']'); + let loopback = host == "localhost" + || host + .parse::() + .is_ok_and(|ip| ip.is_loopback()); + if !loopback { + tracing::warn!( + "MINIO_ENDPOINT uses plain http on a non-loopback host — S3 credentials and \ + presigned uploads cross the network in cleartext; put MinIO behind TLS" + ); + } + rest + } else { + anyhow::bail!("MINIO_ENDPOINT must start with http:// or https://"); + }; + if rest.is_empty() || rest.starts_with('/') { + anyhow::bail!("MINIO_ENDPOINT is missing a host"); + } + Ok(()) +} + fn optional(key: &str, default: &str) -> String { std::env::var(key).unwrap_or_else(|_| default.to_string()) } @@ -420,7 +529,24 @@ fn parse_prepull_list(raw: &str) -> anyhow::Result> { #[cfg(test)] mod tests { - use super::parse_prepull_list; + use super::{parse_prepull_list, validate_minio_endpoint}; + + #[test] + fn minio_endpoint_accepts_http_and_https_urls() { + assert!(validate_minio_endpoint("http://localhost:9000").is_ok()); + assert!(validate_minio_endpoint("http://127.0.0.1:9000").is_ok()); + assert!(validate_minio_endpoint("https://minio.example.com").is_ok()); + // Non-loopback http is allowed (warn-only at startup). + assert!(validate_minio_endpoint("http://10.0.0.5:9000").is_ok()); + } + + #[test] + fn minio_endpoint_rejects_malformed_urls() { + assert!(validate_minio_endpoint("localhost:9000").is_err()); + assert!(validate_minio_endpoint("ftp://minio.example.com").is_err()); + assert!(validate_minio_endpoint("http://").is_err()); + assert!(validate_minio_endpoint("https:///bucket").is_err()); + } #[test] fn prepull_list_trims_dedupes_and_drops_empties() { diff --git a/backend/src/db/artifacts.rs b/backend/src/db/artifacts.rs index 8fa6aa9..3f21771 100644 --- a/backend/src/db/artifacts.rs +++ b/backend/src/db/artifacts.rs @@ -17,6 +17,7 @@ pub async fn insert_pending( job_id: Uuid, name: &str, r2_key: &str, + storage_backend: &str, size_bytes: i64, content_type: &str, kind: &str, @@ -25,12 +26,13 @@ pub async fn insert_pending( sqlx::query_as::<_, Artifact>( r#" INSERT INTO artifacts - (workspace_id, pipeline_id, job_id, name, r2_key, size_bytes, content_type, kind, expires_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + (workspace_id, pipeline_id, job_id, name, r2_key, storage_backend, size_bytes, content_type, kind, expires_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) ON CONFLICT ON CONSTRAINT artifacts_job_name_key DO UPDATE SET size_bytes = EXCLUDED.size_bytes, content_type = EXCLUDED.content_type, kind = EXCLUDED.kind, + storage_backend = EXCLUDED.storage_backend, status = 'pending', checksum_sha256 = NULL, uncompressed_bytes = NULL, @@ -45,6 +47,7 @@ pub async fn insert_pending( .bind(job_id) .bind(name) .bind(r2_key) + .bind(storage_backend) .bind(size_bytes) .bind(content_type) .bind(kind) @@ -53,6 +56,22 @@ pub async fn insert_pending( .await } +/// The pending row for one (job, name) pair — the verification path needs +/// its key and storage-backend marker before HeadObject. +pub async fn find_pending( + pool: &PgPool, + job_id: Uuid, + name: &str, +) -> sqlx::Result> { + sqlx::query_as::<_, Artifact>( + "SELECT * FROM artifacts WHERE job_id = $1 AND name = $2 AND status = 'pending'", + ) + .bind(job_id) + .bind(name) + .fetch_optional(pool) + .await +} + pub async fn count_for_job(pool: &PgPool, job_id: Uuid) -> sqlx::Result { let (count,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM artifacts WHERE job_id = $1") @@ -209,6 +228,7 @@ const CATALOG_PROVENANCE: &str = r#" /// stay cheap and the manifest remains a detail-only payload. const CATALOG_LIST_COLUMNS: &str = r#" SELECT a.id, a.workspace_id, a.pipeline_id, a.job_id, a.name, a.r2_key, + a.storage_backend, a.size_bytes, a.content_type, a.checksum_sha256, a.status, a.kind, a.uncompressed_bytes, a.file_count, NULL::jsonb AS entries, a.created_at, a.expires_at, diff --git a/backend/src/db/pipeline_jobs.rs b/backend/src/db/pipeline_jobs.rs index 388a49d..e1e6e86 100644 --- a/backend/src/db/pipeline_jobs.rs +++ b/backend/src/db/pipeline_jobs.rs @@ -608,12 +608,19 @@ pub async fn add_log_bytes(pool: &PgPool, job_id: Uuid, delta: i64) -> sqlx::Res Ok(total) } -/// Guarded archive marker: set once, never overwritten. -pub async fn mark_logs_archived(pool: &PgPool, job_id: Uuid) -> sqlx::Result { +/// Guarded archive marker: set once, never overwritten. Records which +/// object store took the archive so downloads route to the right host. +pub async fn mark_logs_archived( + pool: &PgPool, + job_id: Uuid, + storage_backend: &str, +) -> sqlx::Result { let result = sqlx::query( - "UPDATE pipeline_jobs SET logs_archived_at = now() WHERE id = $1 AND logs_archived_at IS NULL", + "UPDATE pipeline_jobs SET logs_archived_at = now(), logs_archive_backend = $2 \ + WHERE id = $1 AND logs_archived_at IS NULL", ) .bind(job_id) + .bind(storage_backend) .execute(pool) .await?; Ok(result.rows_affected() == 1) diff --git a/backend/src/db/sessions.rs b/backend/src/db/sessions.rs index 7c9b362..4ebbce2 100644 --- a/backend/src/db/sessions.rs +++ b/backend/src/db/sessions.rs @@ -41,11 +41,14 @@ pub struct SessionRow { pub is_current: bool, } -/// All live (unexpired) sessions for a user, most recently seen first. +/// All live (unexpired, non-idle) sessions for a user, most recently seen +/// first. Uses the same idle predicate as `find_valid_user`, so a session +/// this list shows can actually still authenticate. pub async fn list_for_user( pool: &PgPool, user_id: Uuid, current_hash: &str, + idle_timeout_hours: i64, ) -> sqlx::Result> { sqlx::query_as::<_, SessionRow>( r#" @@ -53,11 +56,13 @@ pub async fn list_for_user( (token_hash = $2) AS is_current FROM sessions WHERE user_id = $1 AND expires_at > now() + AND ($3 = 0 OR COALESCE(last_seen_at, created_at) > now() - ($3 * interval '1 hour')) ORDER BY (token_hash = $2) DESC, last_seen_at DESC NULLS LAST, created_at DESC "#, ) .bind(user_id) .bind(current_hash) + .bind(idle_timeout_hours) .fetch_all(pool) .await } @@ -91,9 +96,16 @@ pub async fn delete_all_for_user_except( Ok(result.rows_affected()) } -/// Resolve a session token hash to its user, only while the session is alive. -/// Touches `last_seen_at` in the same round-trip. -pub async fn find_valid_user(pool: &PgPool, token_hash: &str) -> sqlx::Result> { +/// Resolve a session token hash to its user, only while the session is alive +/// AND has been used within the idle window (`idle_timeout_hours`, 0 +/// disables — a stolen token then stops working after inactivity even +/// before the absolute expiry). Touches `last_seen_at` in the same +/// round-trip; a never-used session ages from its creation instant. +pub async fn find_valid_user( + pool: &PgPool, + token_hash: &str, + idle_timeout_hours: i64, +) -> sqlx::Result> { sqlx::query_as::<_, User>( r#" UPDATE sessions s @@ -101,11 +113,13 @@ pub async fn find_valid_user(pool: &PgPool, token_hash: &str) -> sqlx::Result now() + AND ($2 = 0 OR COALESCE(s.last_seen_at, s.created_at) > now() - ($2 * interval '1 hour')) AND u.id = s.user_id RETURNING u.* "#, ) .bind(token_hash) + .bind(idle_timeout_hours) .fetch_optional(pool) .await } @@ -118,9 +132,18 @@ pub async fn delete_by_token_hash(pool: &PgPool, token_hash: &str) -> sqlx::Resu Ok(()) } -pub async fn delete_expired(pool: &PgPool) -> sqlx::Result { - let result = sqlx::query("DELETE FROM sessions WHERE expires_at <= now()") - .execute(pool) - .await?; +/// Purge sessions past their absolute expiry OR idle-expired ones (the same +/// predicate `find_valid_user` rejects on, so unusable rows don't linger). +pub async fn delete_expired(pool: &PgPool, idle_timeout_hours: i64) -> sqlx::Result { + let result = sqlx::query( + r#" + DELETE FROM sessions + WHERE expires_at <= now() + OR ($1 > 0 AND COALESCE(last_seen_at, created_at) <= now() - ($1 * interval '1 hour')) + "#, + ) + .bind(idle_timeout_hours) + .execute(pool) + .await?; Ok(result.rows_affected()) } diff --git a/backend/src/db/users.rs b/backend/src/db/users.rs index 3cce1e1..8b4b357 100644 --- a/backend/src/db/users.rs +++ b/backend/src/db/users.rs @@ -68,9 +68,9 @@ pub async fn update_profile( /// key, and deleting the workspace cascades to every workspace-scoped /// resource (members, repositories, pipelines, secrets, environments, /// runners). Sessions cascade with the user row itself. Returns the logo -/// keys of deleted workspaces so the caller can best-effort remove the R2 -/// objects. -pub async fn delete_account(pool: &PgPool, user_id: Uuid) -> sqlx::Result> { +/// `(key, storage_backend)` pairs of deleted workspaces so the caller can +/// best-effort remove the objects from the stores that hold them. +pub async fn delete_account(pool: &PgPool, user_id: Uuid) -> sqlx::Result> { let mut tx = pool.begin().await?; sqlx::query( r#" @@ -81,8 +81,9 @@ pub async fn delete_account(pool: &PgPool, user_id: Uuid) -> sqlx::Result = sqlx::query_scalar( - "DELETE FROM workspaces WHERE created_by = $1 RETURNING COALESCE(logo_key, '')", + let logo_keys: Vec<(String, String)> = sqlx::query_as( + "DELETE FROM workspaces WHERE created_by = $1 \ + RETURNING COALESCE(logo_key, ''), COALESCE(logo_storage_backend, 'r2')", ) .bind(user_id) .fetch_all(&mut *tx) @@ -92,5 +93,5 @@ pub async fn delete_account(pool: &PgPool, user_id: Uuid) -> sqlx::Result, -) -> sqlx::Result>> { - let previous: Option<(Option,)> = sqlx::query_as( + storage_backend: Option<&str>, +) -> sqlx::Result>> { + let previous: Option<(Option, String)> = sqlx::query_as( r#" UPDATE workspaces w - SET logo_key = $2, updated_at = now() - FROM (SELECT id, logo_key FROM workspaces WHERE id = $1 FOR UPDATE) old + SET logo_key = $2, logo_storage_backend = $3, updated_at = now() + FROM (SELECT id, logo_key, logo_storage_backend FROM workspaces WHERE id = $1 FOR UPDATE) old WHERE w.id = old.id - RETURNING old.logo_key + RETURNING old.logo_key, COALESCE(old.logo_storage_backend, 'r2') "#, ) .bind(workspace_id) .bind(logo_key) + .bind(storage_backend) .fetch_optional(pool) .await?; - Ok(previous.map(|(key,)| key)) + Ok(previous.map(|(key, backend)| key.map(|key| (key, backend)))) } /// Members with their role, owner first, then by join date. diff --git a/backend/src/handlers/artifacts.rs b/backend/src/handlers/artifacts.rs index 1a15b39..cc8fc4a 100644 --- a/backend/src/handlers/artifacts.rs +++ b/backend/src/handlers/artifacts.rs @@ -326,13 +326,16 @@ pub async fn remove( .ok_or(AppError::NotFound)?; if matches!(artifact.status.as_str(), "uploaded" | "pending") - && let Some(r2) = &state.r2 - && let Err(error) = r2.delete_object(&artifact.r2_key).await + && let Some(storage) = &state.storage + && let Err(error) = storage + .store_for(&artifact.storage_backend) + .delete_object(&artifact.r2_key) + .await { tracing::warn!( %artifact_id, error = ?error, - "failed to delete artifact object from R2 during operator delete" + "failed to delete artifact object from storage during operator delete" ); } diff --git a/backend/src/handlers/dashboard_ws.rs b/backend/src/handlers/dashboard_ws.rs index 8f76a5b..5a9de0c 100644 --- a/backend/src/handlers/dashboard_ws.rs +++ b/backend/src/handlers/dashboard_ws.rs @@ -49,11 +49,15 @@ pub async fn authenticate_browser( } let jar = axum_extra::extract::cookie::CookieJar::from_headers(headers); let token = jar.get(&state.config.cookie_name)?.value().to_string(); - db::sessions::find_valid_user(&state.pool, &session::hash_token(&token)) - .await - .ok() - .flatten() - .map(|user| user.id) + db::sessions::find_valid_user( + &state.pool, + &session::hash_token(&token), + state.config.session_idle_timeout_hours, + ) + .await + .ok() + .flatten() + .map(|user| user.id) } /// Browsers only send tiny control frames. diff --git a/backend/src/handlers/github_installations.rs b/backend/src/handlers/github_installations.rs index 166e19d..c58458a 100644 --- a/backend/src/handlers/github_installations.rs +++ b/backend/src/handlers/github_installations.rs @@ -46,7 +46,12 @@ pub async fn setup( else { return fail("no session cookie"); }; - let user = match db::sessions::find_valid_user(&state.pool, &session::hash_token(&token)).await + let user = match db::sessions::find_valid_user( + &state.pool, + &session::hash_token(&token), + state.config.session_idle_timeout_hours, + ) + .await { Ok(Some(user)) => user, Ok(None) => return fail("session invalid"), diff --git a/backend/src/handlers/jobs.rs b/backend/src/handlers/jobs.rs index 042f878..9fdcbc2 100644 --- a/backend/src/handlers/jobs.rs +++ b/backend/src/handlers/jobs.rs @@ -208,6 +208,7 @@ mod tests { position: 0, metrics: None, logs_archived_at: None, + logs_archive_backend: None, queued_at: Utc::now(), assigned_at: None, started_at: None, diff --git a/backend/src/handlers/me.rs b/backend/src/handlers/me.rs index 424816d..fb8be6a 100644 --- a/backend/src/handlers/me.rs +++ b/backend/src/handlers/me.rs @@ -158,10 +158,11 @@ pub async fn delete_me( let logo_keys = db::users::delete_account(&state.pool, user.id).await?; - // Best-effort R2 cleanup of workspace logos; row state is already final. - if let Some(r2) = state.r2.as_ref() { - for key in logo_keys { - if let Err(error) = r2.delete_object(&key).await { + // Best-effort storage cleanup of workspace logos; row state is already + // final. Each delete routes to the store that holds the object. + if let Some(storage) = state.storage.as_ref() { + for (key, backend) in logo_keys { + if let Err(error) = storage.store_for(&backend).delete_object(&key).await { tracing::warn!(%key, error = ?error, "failed to delete logo object for deleted account"); } } diff --git a/backend/src/handlers/pipelines.rs b/backend/src/handlers/pipelines.rs index e69a88e..33a2a19 100644 --- a/backend/src/handlers/pipelines.rs +++ b/backend/src/handlers/pipelines.rs @@ -640,11 +640,14 @@ pub async fn job_logs_raw( // presigned GET (the gzip'd file; browsers download it as-is). if chunks.is_empty() && job.logs_archived_at.is_some() - && let Some(r2) = &state.r2 + && let Some(storage) = &state.storage { let key = crate::services::log_archive::log_key_for(workspace_id, &job); let filename = format!("{}-{}.log.gz", job.job_key, job.attempt); - let url = r2 + // NULL legacy markers read as 'r2' — every pre-marker archive + // was written when R2 was the only store. + let url = storage + .store_for(job.logs_archive_backend.as_deref().unwrap_or("r2")) .presign_get(&key, &filename) .await .map_err(AppError::Internal)?; @@ -715,11 +718,12 @@ pub async fn artifact_download( if artifact.status != "uploaded" { return Err(AppError::Conflict("artifact is not available for download")); } - let Some(r2) = &state.r2 else { + let Some(storage) = &state.storage else { return Err(AppError::Conflict("artifact storage is not configured")); }; - let url = r2 + let url = storage + .store_for(&artifact.storage_backend) .presign_get(&artifact.r2_key, &artifact.name) .await .map_err(AppError::Internal)?; diff --git a/backend/src/handlers/runner_ws.rs b/backend/src/handlers/runner_ws.rs index 5fdfcc9..b829ff1 100644 --- a/backend/src/handlers/runner_ws.rs +++ b/backend/src/handlers/runner_ws.rs @@ -640,7 +640,7 @@ async fn handle_artifact_request( deny("job is not assigned to this runner"); return Ok(()); }; - let Some(r2) = &state.r2 else { + let Some(storage) = &state.storage else { deny("artifact storage is not configured"); return Ok(()); }; @@ -666,13 +666,19 @@ async fn handle_artifact_request( let pipeline = db::pipelines::find_by_id(&state.pool, job.pipeline_id) .await? .ok_or_else(|| anyhow::anyhow!("pipeline vanished"))?; - let key = crate::services::r2::R2::artifact_key( + let key = crate::services::object_store::artifact_key( pipeline.workspace_id, pipeline.id, job.id, &name, ); + // Presigning is an offline signature, so the grant routes up front: + // the primary store while its cached health probe passes, else the + // fallback. The chosen backend is recorded on the row — verification, + // downloads and deletes all follow the marker. + let store = storage.store_for_upload().await; + // Server-side classification drives per-kind retention and catalog // filters; the runner never influences it beyond the validated name. let kind = crate::services::artifact_kind::classify(&name); @@ -689,16 +695,22 @@ async fn handle_artifact_request( job.id, &name, &key, + store.backend().as_str(), size_bytes as i64, &content_type, kind, // Retention clock starts at upload request; the janitor deletes the - // R2 object and flips the row to expired once it lapses. + // stored object and flips the row to expired once it lapses. Some(chrono::Utc::now() + chrono::Duration::days(retention_days)), ) .await?; - let put_url = match r2.presign_put(&artifact.r2_key, &content_type).await { + // The declared size rides the signed headers: the store rejects a body + // of any other length at the edge (HeadObject re-verifies afterwards). + let put_url = match store + .presign_put(&artifact.r2_key, &content_type, size_bytes as i64) + .await + { Ok(url) => url, Err(error) => { tracing::warn!(error = ?error, "artifact presign failed"); @@ -715,7 +727,7 @@ async fn handle_artifact_request( put_url, key: artifact.r2_key, expires_at: chrono::Utc::now() - + chrono::Duration::from_std(crate::services::r2::UPLOAD_URL_TTL) + + chrono::Duration::from_std(crate::services::object_store::UPLOAD_URL_TTL) .unwrap_or(chrono::Duration::minutes(15)), }, ); @@ -783,7 +795,7 @@ async fn handle_artifact_done( let Some(job) = db::pipeline_jobs::find_assigned(&state.pool, job_id, runner_id).await? else { return Ok(()); }; - let Some(r2) = &state.r2 else { + let Some(storage) = &state.storage else { return Ok(()); }; if !crate::services::github_app::is_safe_name_segment(&name) { @@ -800,15 +812,16 @@ async fn handle_artifact_done( let pipeline = db::pipelines::find_by_id(&state.pool, job.pipeline_id) .await? .ok_or_else(|| anyhow::anyhow!("pipeline vanished"))?; - let key = crate::services::r2::R2::artifact_key( - pipeline.workspace_id, - pipeline.id, - job.id, - &name, - ); + + // The pending row carries the key AND the storage-backend marker the + // grant recorded — verification must ask the store that was granted. + let Some(pending) = db::artifacts::find_pending(&state.pool, job.id, &name).await? else { + return Ok(()); + }; + let store = storage.store_for(&pending.storage_backend); // Trust the bucket, not the runner: the object must exist and fit. - let verified_size = match r2.head_size(&key).await { + let verified_size = match store.head_size(&pending.r2_key).await { Ok(Some(size)) if size > 0 && size <= state.config.max_artifact_bytes => size, Ok(_) => { db::artifacts::mark_failed(&state.pool, job.id, &name).await?; diff --git a/backend/src/handlers/sessions.rs b/backend/src/handlers/sessions.rs index edb6fd6..38fb87c 100644 --- a/backend/src/handlers/sessions.rs +++ b/backend/src/handlers/sessions.rs @@ -66,7 +66,13 @@ pub async fn list( jar: CookieJar, ) -> AppResult> { let current_hash = current_token_hash(&state, &jar)?; - let rows = db::sessions::list_for_user(&state.pool, user.id, ¤t_hash).await?; + let rows = db::sessions::list_for_user( + &state.pool, + user.id, + ¤t_hash, + state.config.session_idle_timeout_hours, + ) + .await?; let sessions: Vec = rows .into_iter() .map(|row| SessionResponse { diff --git a/backend/src/handlers/workspaces.rs b/backend/src/handlers/workspaces.rs index c4784eb..630122c 100644 --- a/backend/src/handlers/workspaces.rs +++ b/backend/src/handlers/workspaces.rs @@ -12,7 +12,7 @@ use crate::db::workspaces::ProvisionOutcome; use crate::error::{AppError, AppResult}; use crate::middleware::auth::CurrentUser; use crate::models::workspace::{WorkspaceMemberResponse, WorkspaceResponse}; -use crate::services::r2::R2; +use crate::services::object_store; use crate::services::workspace as workspace_service; use crate::services::{authz, image_sniff}; use crate::state::AppState; @@ -209,8 +209,8 @@ pub async fn upload_logo( body: Bytes, ) -> AppResult> { authz::require_permission(&state.pool, user.id, workspace_id, authz::WORKSPACE_MANAGE).await?; - let r2 = state - .r2 + let storage = state + .storage .as_ref() .ok_or(AppError::Conflict(STORAGE_NOT_CONFIGURED))?; @@ -227,17 +227,23 @@ pub async fn upload_logo( }; // Server-generated key — the object lands before the row points at it, - // so a crash in between leaves only an unreferenced object. - let key = R2::logo_key(workspace_id, ext); - r2.put_object(&key, body.to_vec(), content_type).await?; - - let previous = db::workspaces::set_logo_key(&state.pool, workspace_id, Some(&key)) - .await? - .ok_or(AppError::NotFound)?; + // so a crash in between leaves only an unreferenced object. The write + // falls back to the secondary store; the accepting backend is recorded. + let key = object_store::logo_key(workspace_id, ext); + let backend = storage.put_object(&key, body.to_vec(), content_type).await?; + + let previous = db::workspaces::set_logo_key( + &state.pool, + workspace_id, + Some(&key), + Some(backend.as_str()), + ) + .await? + .ok_or(AppError::NotFound)?; // Best-effort removal of the replaced object; the row already moved on. - if let Some(old_key) = previous.filter(|old| old != &key) - && let Err(error) = r2.delete_object(&old_key).await + if let Some((old_key, old_backend)) = previous.filter(|(old, _)| old != &key) + && let Err(error) = storage.store_for(&old_backend).delete_object(&old_key).await { tracing::warn!(key = %old_key, error = ?error, "failed to delete replaced workspace logo"); } @@ -257,7 +263,10 @@ pub async fn upload_logo( .execute(&state.pool) .await?; - let url = r2.presign_get_inline(&key).await?; + let url = storage + .store_for(backend.as_str()) + .presign_get_inline(&key) + .await?; Ok(Json(json!({ "logoUrl": url }))) } @@ -270,13 +279,13 @@ pub async fn remove_logo( ) -> AppResult { authz::require_permission(&state.pool, user.id, workspace_id, authz::WORKSPACE_MANAGE).await?; - let previous = db::workspaces::set_logo_key(&state.pool, workspace_id, None) + let previous = db::workspaces::set_logo_key(&state.pool, workspace_id, None, None) .await? .ok_or(AppError::NotFound)?; - if let Some(old_key) = previous { - if let Some(r2) = state.r2.as_ref() - && let Err(error) = r2.delete_object(&old_key).await + if let Some((old_key, old_backend)) = previous { + if let Some(storage) = state.storage.as_ref() + && let Err(error) = storage.store_for(&old_backend).delete_object(&old_key).await { tracing::warn!(key = %old_key, error = ?error, "failed to delete removed workspace logo"); } @@ -312,8 +321,13 @@ pub async fn logo_url( .await? .ok_or(AppError::NotFound)?; - let url = match (workspace.logo_key.as_deref(), state.r2.as_ref()) { - (Some(key), Some(r2)) => Some(r2.presign_get_inline(key).await?), + let url = match (workspace.logo_key.as_deref(), state.storage.as_ref()) { + (Some(key), Some(storage)) => Some( + storage + .store_for(workspace.logo_storage_backend.as_deref().unwrap_or("r2")) + .presign_get_inline(key) + .await?, + ), _ => None, }; Ok(Json(json!({ "url": url }))) diff --git a/backend/src/main.rs b/backend/src/main.rs index 4140073..8a10abf 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -42,6 +42,13 @@ async fn main() -> anyhow::Result<()> { .context("failed to run database migrations")?; let mut state = AppState::new(pool.clone(), config.clone())?; + // Object storage bootstrap: create missing MinIO buckets (fresh local + // deployments start empty). Warn-only — storage stays enabled and any + // real outage surfaces on the first write. + if let Some(storage) = &state.storage { + tracing::info!(primary = %storage.primary_backend(), "object storage configured"); + storage.ensure_buckets().await; + } // Optional hosted-runner provisioner. Constructed whenever configured; // its reconnect loop owns the Docker connection, so a daemon outage (at // boot or later) degrades to a clean 409 on the hosted-runner endpoint diff --git a/backend/src/middleware/auth.rs b/backend/src/middleware/auth.rs index ec0dbc2..614180b 100644 --- a/backend/src/middleware/auth.rs +++ b/backend/src/middleware/auth.rs @@ -23,9 +23,13 @@ impl FromRequestParts for CurrentUser { .map(|cookie| cookie.value().to_string()) .ok_or(AppError::Unauthorized)?; - let user = db::sessions::find_valid_user(&state.pool, &session::hash_token(&token)) - .await? - .ok_or(AppError::Unauthorized)?; + let user = db::sessions::find_valid_user( + &state.pool, + &session::hash_token(&token), + state.config.session_idle_timeout_hours, + ) + .await? + .ok_or(AppError::Unauthorized)?; Ok(CurrentUser(user)) } diff --git a/backend/src/models/artifact.rs b/backend/src/models/artifact.rs index 8f27bc7..ce00cf4 100644 --- a/backend/src/models/artifact.rs +++ b/backend/src/models/artifact.rs @@ -13,6 +13,9 @@ pub struct Artifact { pub job_id: Uuid, pub name: String, pub r2_key: String, + /// Which object store holds the blob ('minio' | 'r2') — internal + /// routing marker, never exposed to clients (like r2_key). + pub storage_backend: String, pub size_bytes: Option, pub content_type: Option, pub checksum_sha256: Option, diff --git a/backend/src/models/pipeline.rs b/backend/src/models/pipeline.rs index 7cfca2d..8c73ffc 100644 --- a/backend/src/models/pipeline.rs +++ b/backend/src/models/pipeline.rs @@ -120,9 +120,12 @@ pub struct PipelineJob { pub position: i32, /// Runner-reported resource metrics, clamped server-side before storage. pub metrics: Option, - /// Set once the full log has been archived to R2; NULL chunks are the - /// only copy and must never be pruned. + /// Set once the full log has been archived to object storage; NULL + /// chunks are the only copy and must never be pruned. pub logs_archived_at: Option>, + /// Which store holds the archive ('minio' | 'r2'); NULL legacy rows + /// read as 'r2'. Internal routing marker, never serialized outward. + pub logs_archive_backend: Option, pub queued_at: DateTime, pub assigned_at: Option>, pub started_at: Option>, diff --git a/backend/src/models/workspace.rs b/backend/src/models/workspace.rs index 2f61d89..9e5057c 100644 --- a/backend/src/models/workspace.rs +++ b/backend/src/models/workspace.rs @@ -13,6 +13,9 @@ pub struct Workspace { pub description: Option, pub created_by: Uuid, pub logo_key: Option, + /// Which store holds the logo ('minio' | 'r2'); NULL legacy rows read + /// as 'r2'. Internal routing marker, never serialized outward. + pub logo_storage_backend: Option, pub created_at: DateTime, pub updated_at: DateTime, } diff --git a/backend/src/routes/mod.rs b/backend/src/routes/mod.rs index aa68cd3..1499cc7 100644 --- a/backend/src/routes/mod.rs +++ b/backend/src/routes/mod.rs @@ -564,7 +564,19 @@ pub fn build_router(state: AppState) -> anyhow::Result { security_headers::security_headers, )) .layer(cors) - .layer(TraceLayer::new_for_http()) + // The default span records the full URI including the query string, + // which carries secrets on some routes (?code=/&state= on the OAuth + // callback, ?ticket= on WS upgrades). Record method + PATH only so + // raising the log filter to debug can never capture them. + .layer(TraceLayer::new_for_http().make_span_with( + |request: &Request| { + tracing::info_span!( + "request", + method = %request.method(), + path = %request.uri().path(), + ) + }, + )) .layer(PropagateRequestIdLayer::new(X_REQUEST_ID)) .layer(SetRequestIdLayer::new(X_REQUEST_ID, MakeRequestUuid)) .with_state(state); diff --git a/backend/src/services/janitor.rs b/backend/src/services/janitor.rs index b016b50..12f4124 100644 --- a/backend/src/services/janitor.rs +++ b/backend/src/services/janitor.rs @@ -22,27 +22,33 @@ pub async fn run(state: AppState) { async fn pass(state: &AppState) { // 1. Expired sessions and abandoned login transactions. - if let Err(error) = db::sessions::delete_expired(&state.pool).await { + if let Err(error) = + db::sessions::delete_expired(&state.pool, state.config.session_idle_timeout_hours).await + { tracing::warn!(error = ?error, "failed to purge expired sessions"); } if let Err(error) = db::oauth_states::delete_expired(&state.pool).await { tracing::warn!(error = ?error, "failed to purge expired oauth states"); } - // 2. Uploaded artifacts whose retention lapsed: delete the R2 object - // best-effort, then flip the row so downloads stop immediately even - // if the delete failed (the next pass retries nothing — the object - // becomes unreachable garbage at worst, never a live leak). + // 2. Uploaded artifacts whose retention lapsed: delete the stored + // object best-effort (routed to the store that holds it), then flip + // the row so downloads stop immediately even if the delete failed + // (the next pass retries nothing — the object becomes unreachable + // garbage at worst, never a live leak). match db::artifacts::find_expired(&state.pool, BATCH).await { Ok(expired) => { for artifact in expired { - if let Some(r2) = &state.r2 - && let Err(error) = r2.delete_object(&artifact.r2_key).await + if let Some(storage) = &state.storage + && let Err(error) = storage + .store_for(&artifact.storage_backend) + .delete_object(&artifact.r2_key) + .await { tracing::warn!( artifact_id = %artifact.id, error = ?error, - "failed to delete expired artifact object from R2" + "failed to delete expired artifact object from storage" ); } if let Err(error) = db::artifacts::mark_expired(&state.pool, artifact.id).await { @@ -65,13 +71,16 @@ async fn pass(state: &AppState) { { Ok(stale) => { for artifact in stale { - if let Some(r2) = &state.r2 - && let Err(error) = r2.delete_object(&artifact.r2_key).await + if let Some(storage) = &state.storage + && let Err(error) = storage + .store_for(&artifact.storage_backend) + .delete_object(&artifact.r2_key) + .await { tracing::warn!( artifact_id = %artifact.id, error = ?error, - "failed to delete abandoned artifact object from R2" + "failed to delete abandoned artifact object from storage" ); } if let Err(error) = @@ -84,10 +93,10 @@ async fn pass(state: &AppState) { Err(error) => tracing::warn!(error = ?error, "failed to scan stale pending artifacts"), } - // 4. Hot log chunks whose R2 archive exists and whose hot window has - // lapsed. Only runs with R2 configured — without it, Postgres is the - // only copy and is never pruned. - if state.r2.is_some() { + // 4. Hot log chunks whose object-storage archive exists and whose hot + // window has lapsed. Only runs with storage configured — without it, + // Postgres is the only copy and is never pruned. + if state.storage.is_some() { match db::pipeline_jobs::find_prunable_archived( &state.pool, state.config.log_hot_retention_days, diff --git a/backend/src/services/log_archive.rs b/backend/src/services/log_archive.rs index 9446812..4bade0d 100644 --- a/backend/src/services/log_archive.rs +++ b/backend/src/services/log_archive.rs @@ -1,11 +1,14 @@ -//! Post-completion log archival to R2. +//! Post-completion log archival to object storage. //! -//! When a job reaches a terminal state and R2 is configured, its persisted -//! (already masked) log chunks are concatenated, gzip-compressed, and stored -//! at `logs/{workspace}/{pipeline}/{job}-{attempt}.log.gz`. Postgres remains +//! When a job reaches a terminal state and object storage is configured +//! (MinIO primary and/or R2 fallback), its persisted (already masked) log +//! chunks are concatenated, gzip-compressed, and stored at +//! `logs/{workspace}/{pipeline}/{job}-{attempt}.log.gz`. Postgres remains //! the source of truth until the janitor prunes chunks whose hot-retention //! window has lapsed — a failed upload just leaves the job unarchived, and -//! unarchived jobs are never pruned. +//! unarchived jobs are never pruned. The write falls back to the secondary +//! store automatically; whichever store accepted the archive is recorded on +//! the job row so downloads presign against the right host. use std::io::Write; @@ -13,13 +16,13 @@ use uuid::Uuid; use crate::db; use crate::models::pipeline::PipelineJob; -use crate::services::r2::R2; +use crate::services::object_store; use crate::state::AppState; /// Fire-and-forget archival for one finished job. Skipped jobs and jobs /// that never produced output are left alone. pub fn spawn_archive(state: &AppState, job: &PipelineJob) { - if state.r2.is_none() { + if state.storage.is_none() { return; } if job.conclusion.as_deref() == Some("skipped") || job.log_bytes == 0 { @@ -39,7 +42,7 @@ pub fn spawn_archive(state: &AppState, job: &PipelineJob) { } async fn archive_job(state: &AppState, job: &PipelineJob) -> anyhow::Result<()> { - let Some(r2) = &state.r2 else { + let Some(storage) = &state.storage else { return Ok(()); }; @@ -73,13 +76,15 @@ async fn archive_job(state: &AppState, job: &PipelineJob) -> anyhow::Result<()> .await??; let key = log_key_for(pipeline.workspace_id, job); - r2.put_object(&key, compressed, "application/gzip").await?; - db::pipeline_jobs::mark_logs_archived(&state.pool, job.id).await?; + let backend = storage + .put_object(&key, compressed, "application/gzip") + .await?; + db::pipeline_jobs::mark_logs_archived(&state.pool, job.id, backend.as_str()).await?; - tracing::debug!(job_id = %job.id, key = %key, "job log archived to R2"); + tracing::debug!(job_id = %job.id, key = %key, backend = %backend, "job log archived"); Ok(()) } pub fn log_key_for(workspace_id: Uuid, job: &PipelineJob) -> String { - R2::log_key(workspace_id, job.pipeline_id, job.id, job.attempt) + object_store::log_key(workspace_id, job.pipeline_id, job.id, job.attempt) } diff --git a/backend/src/services/minio.rs b/backend/src/services/minio.rs new file mode 100644 index 0000000..149e909 --- /dev/null +++ b/backend/src/services/minio.rs @@ -0,0 +1,22 @@ +//! MinIO flavor of the shared S3 object store. +//! +//! MinIO is a drop-in S3 replacement: only construction differs — an +//! explicit endpoint URL (nothing is derived from an account id) and +//! path-style addressing (`http://host:9000/bucket/key`), which MinIO's +//! standard deployment requires. When configured, MinIO is the DEFAULT / +//! primary object store; R2 (when also configured) becomes the fallback. + +use crate::config::MinioConfig; +use crate::services::object_store::{S3Store, StorageBackend}; + +pub fn store(config: &MinioConfig) -> S3Store { + S3Store::new( + StorageBackend::Minio, + &config.endpoint, + &config.region, + &config.access_key_id, + &config.secret_access_key, + &config.bucket, + config.force_path_style, + ) +} diff --git a/backend/src/services/mod.rs b/backend/src/services/mod.rs index cdf4445..9e63fe5 100644 --- a/backend/src/services/mod.rs +++ b/backend/src/services/mod.rs @@ -7,10 +7,12 @@ pub mod image_sniff; pub mod janitor; pub mod log_archive; pub mod log_hub; +pub mod minio; pub mod notification; pub mod notification_hub; pub mod notification_projector; pub mod pipeline_plan; +pub mod object_store; pub mod pipeline_run; pub mod r2; pub mod repo_sync; diff --git a/backend/src/services/object_store.rs b/backend/src/services/object_store.rs new file mode 100644 index 0000000..37dbd87 --- /dev/null +++ b/backend/src/services/object_store.rs @@ -0,0 +1,477 @@ +//! S3-compatible object storage (artifacts, log archives, workspace logos). +//! +//! Two backends share one generic [`S3Store`] client: **MinIO** (any +//! S3-compatible endpoint — the default/primary store when configured) and +//! **Cloudflare R2** (fully supported; acts as the fallback when both are +//! configured, or as the primary when it is the only one). The control plane +//! never proxies artifact bytes: runners upload through short-lived presigned +//! PUT URLs and browsers download through presigned GET URLs, both verified +//! server-side (HeadObject) before rows flip to `uploaded`. +//! +//! Presigned URLs are host-specific, so every stored object carries a +//! `storage_backend` marker ('minio' | 'r2') in its row; reads, deletes and +//! verification always route to the store that actually holds the object +//! ([`Storage::store_for`]). Server-side writes (log archives, logos) fall +//! back to the secondary store when the primary write fails; presigned +//! upload grants pick the primary while a cached HeadBucket probe says it is +//! healthy and the fallback otherwise. + +use std::time::{Duration, Instant}; + +use anyhow::Context; +use aws_sdk_s3::config::{BehaviorVersion, Credentials, Region}; +use aws_sdk_s3::presigning::PresigningConfig; +use uuid::Uuid; + +/// Presigned PUT grants expire quickly; the runner uploads immediately. +pub const UPLOAD_URL_TTL: Duration = Duration::from_secs(15 * 60); +/// Download links are minted per request. +pub const DOWNLOAD_URL_TTL: Duration = Duration::from_secs(10 * 60); +/// How long one primary-health probe result is trusted before re-probing. +const HEALTH_PROBE_TTL: Duration = Duration::from_secs(60); + +/// Which physical store an object lives in. The string forms are the DB +/// values of every `storage_backend` marker column. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StorageBackend { + Minio, + R2, +} + +impl StorageBackend { + pub fn as_str(self) -> &'static str { + match self { + StorageBackend::Minio => "minio", + StorageBackend::R2 => "r2", + } + } + + /// Parse a stored marker. Unknown / legacy values read as R2 — every + /// pre-marker row was written when R2 was the only store. + pub fn parse(value: &str) -> StorageBackend { + match value { + "minio" => StorageBackend::Minio, + _ => StorageBackend::R2, + } + } +} + +impl std::fmt::Display for StorageBackend { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Deterministic object key; every segment is a UUID and the name is +/// allow-list validated by the caller before it gets here. +pub fn artifact_key(workspace_id: Uuid, pipeline_id: Uuid, job_id: Uuid, name: &str) -> String { + format!("artifacts/{workspace_id}/{pipeline_id}/{job_id}/{name}") +} + +/// Deterministic key for a job's archived (gzip) log; every segment is a +/// UUID or an integer. +pub fn log_key(workspace_id: Uuid, pipeline_id: Uuid, job_id: Uuid, attempt: i32) -> String { + format!("logs/{workspace_id}/{pipeline_id}/{job_id}-{attempt}.log.gz") +} + +/// Server-generated key for a workspace logo. Every segment is a UUID; +/// the extension comes from server-side magic-byte detection, never from +/// the client's filename or declared content type. +pub fn logo_key(workspace_id: Uuid, ext: &str) -> String { + format!("logos/{workspace_id}/{}.{ext}", Uuid::new_v4()) +} + +/// One S3-compatible store. Generic over the endpoint so MinIO and R2 share +/// every code path; only construction differs (see the [`Self::new`] callers +/// in `services/minio.rs` and `services/r2.rs`). +pub struct S3Store { + client: aws_sdk_s3::Client, + bucket: String, + backend: StorageBackend, +} + +impl S3Store { + /// `force_path_style` is required by MinIO's standard addressing + /// (`http://host:9000/bucket/key`); R2 uses virtual-host addressing and + /// leaves it off. Credentials never appear in Debug output or logs. + pub fn new( + backend: StorageBackend, + endpoint_url: &str, + region: &str, + access_key_id: &str, + secret_access_key: &str, + bucket: &str, + force_path_style: bool, + ) -> Self { + let credentials = Credentials::new( + access_key_id, + secret_access_key, + None, + None, + match backend { + StorageBackend::Minio => "minio-static", + StorageBackend::R2 => "r2-static", + }, + ); + let config = aws_sdk_s3::Config::builder() + .behavior_version(BehaviorVersion::latest()) + .region(Region::new(region.to_string())) + .endpoint_url(endpoint_url) + .credentials_provider(credentials) + .force_path_style(force_path_style) + .build(); + Self { + client: aws_sdk_s3::Client::from_conf(config), + bucket: bucket.to_string(), + backend, + } + } + + pub fn backend(&self) -> StorageBackend { + self.backend + } + + /// Server-side upload for control-plane-generated objects (log + /// archives, logos). Bodies are small — bounded by the per-job log cap. + pub async fn put_object( + &self, + key: &str, + body: Vec, + content_type: &str, + ) -> anyhow::Result<()> { + self.client + .put_object() + .bucket(&self.bucket) + .key(key) + .content_type(content_type) + .body(body.into()) + .send() + .await + .with_context(|| format!("failed to upload object to {} storage", self.backend))?; + Ok(()) + } + + /// Best-effort delete; a missing object counts as success (DeleteObject + /// is idempotent on S3-compatible stores). + pub async fn delete_object(&self, key: &str) -> anyhow::Result<()> { + self.client + .delete_object() + .bucket(&self.bucket) + .key(key) + .send() + .await + .with_context(|| format!("failed to delete object from {} storage", self.backend))?; + Ok(()) + } + + /// Presigned PUT bound to the exact content type AND content length the + /// runner declared — both ride the signed headers, so the store rejects + /// a body of any other size or type at the edge (the after-the-fact + /// HeadObject check remains as defense-in-depth). + pub async fn presign_put( + &self, + key: &str, + content_type: &str, + content_length: i64, + ) -> anyhow::Result { + let presigned = self + .client + .put_object() + .bucket(&self.bucket) + .key(key) + .content_type(content_type) + .content_length(content_length) + .presigned(PresigningConfig::expires_in(UPLOAD_URL_TTL)?) + .await + .context("failed to presign artifact upload")?; + Ok(presigned.uri().to_string()) + } + + pub async fn presign_get(&self, key: &str, filename: &str) -> anyhow::Result { + let presigned = self + .client + .get_object() + .bucket(&self.bucket) + .key(key) + .response_content_disposition(format!("attachment; filename=\"{filename}\"")) + .presigned(PresigningConfig::expires_in(DOWNLOAD_URL_TTL)?) + .await + .context("failed to presign artifact download")?; + Ok(presigned.uri().to_string()) + } + + /// Presigned GET for inline display (no attachment disposition) — used + /// for workspace logos rendered in `` tags. Same short TTL; the SPA + /// re-mints through the authenticated logo-url endpoint. + pub async fn presign_get_inline(&self, key: &str) -> anyhow::Result { + let presigned = self + .client + .get_object() + .bucket(&self.bucket) + .key(key) + .response_content_disposition("inline") + .presigned(PresigningConfig::expires_in(DOWNLOAD_URL_TTL)?) + .await + .context("failed to presign inline download")?; + Ok(presigned.uri().to_string()) + } + + /// Size of the uploaded object, or None when it does not exist. + pub async fn head_size(&self, key: &str) -> anyhow::Result> { + match self + .client + .head_object() + .bucket(&self.bucket) + .key(key) + .send() + .await + { + Ok(output) => Ok(output.content_length()), + Err(err) => { + if let aws_sdk_s3::error::SdkError::ServiceError(service_err) = &err + && service_err.err().is_not_found() + { + return Ok(None); + } + Err(anyhow::Error::new(err).context("artifact HeadObject failed")) + } + } + } + + /// Liveness probe: can this store answer for its bucket right now? + async fn bucket_reachable(&self) -> bool { + self.client + .head_bucket() + .bucket(&self.bucket) + .send() + .await + .is_ok() + } + + /// Create the bucket when it does not exist yet (MinIO deployments + /// commonly start empty; R2 buckets are dashboard-managed, so callers + /// only invoke this for MinIO). Warn-only at the call site. + pub async fn ensure_bucket(&self) -> anyhow::Result { + if self.bucket_reachable().await { + return Ok(false); + } + match self + .client + .create_bucket() + .bucket(&self.bucket) + .send() + .await + { + Ok(_) => Ok(true), + // A concurrent boot may have won the race — owned is success. + Err(aws_sdk_s3::error::SdkError::ServiceError(service_err)) + if matches!( + service_err.err(), + aws_sdk_s3::operation::create_bucket::CreateBucketError::BucketAlreadyOwnedByYou(_) + | aws_sdk_s3::operation::create_bucket::CreateBucketError::BucketAlreadyExists(_) + ) => + { + Ok(false) + } + Err(err) => Err(anyhow::Error::new(err) + .context(format!("failed to create bucket on {} storage", self.backend))), + } + } +} + +struct HealthCache { + checked_at: Option, + healthy: bool, +} + +/// The storage router held in `AppState`. `None` in state keeps every +/// clean-denial path exactly as before; when present, `primary` is MinIO +/// whenever MinIO is configured (R2 otherwise) and `fallback` is R2 when +/// both are configured. +pub struct Storage { + primary: S3Store, + fallback: Option, + /// Cached primary liveness so grant routing never probes per-request. + /// A tokio Mutex on purpose: concurrent expirees coalesce into one probe. + primary_health: tokio::sync::Mutex, +} + +impl Storage { + pub fn new(primary: S3Store, fallback: Option) -> Self { + Self { + primary, + fallback, + primary_health: tokio::sync::Mutex::new(HealthCache { + checked_at: None, + healthy: true, + }), + } + } + + pub fn primary_backend(&self) -> StorageBackend { + self.primary.backend() + } + + /// The store that owns an object, by its row's `storage_backend` marker. + /// Unknown/legacy values read as R2; a marker whose store is no longer + /// configured falls back to the primary (the operation then fails or + /// misses cleanly instead of panicking on a missing client). + pub fn store_for(&self, backend: &str) -> &S3Store { + let wanted = StorageBackend::parse(backend); + if self.primary.backend() == wanted { + return &self.primary; + } + if let Some(fallback) = &self.fallback + && fallback.backend() == wanted + { + return fallback; + } + tracing::debug!( + wanted = %wanted, + using = %self.primary.backend(), + "storage backend marker points at an unconfigured store; using primary" + ); + &self.primary + } + + /// Server-side write with reactive fallback: try the primary; if it + /// fails and a fallback exists, try that. Returns the backend that + /// accepted the object so the caller can record the marker. + pub async fn put_object( + &self, + key: &str, + body: Vec, + content_type: &str, + ) -> anyhow::Result { + let Some(fallback) = &self.fallback else { + self.primary.put_object(key, body, content_type).await?; + return Ok(self.primary.backend()); + }; + match self.primary.put_object(key, body.clone(), content_type).await { + Ok(()) => Ok(self.primary.backend()), + Err(primary_error) => { + tracing::warn!( + key = %key, + error = ?primary_error, + primary = %self.primary.backend(), + fallback = %fallback.backend(), + "primary storage write failed; trying fallback" + ); + fallback + .put_object(key, body, content_type) + .await + .with_context(|| { + format!( + "both storage backends rejected the write \ + (primary {}: {primary_error:#})", + self.primary.backend() + ) + })?; + Ok(fallback.backend()) + } + } + } + + /// The store presigned upload grants should target. Presigning is an + /// offline signature (it cannot fail over reactively), so routing keys + /// off a cached HeadBucket probe of the primary: healthy → primary, + /// unreachable → fallback. Without a fallback the primary is always + /// used — exactly the single-store behavior. + pub async fn store_for_upload(&self) -> &S3Store { + let Some(fallback) = &self.fallback else { + return &self.primary; + }; + let mut cache = self.primary_health.lock().await; + let stale = cache + .checked_at + .is_none_or(|at| at.elapsed() >= HEALTH_PROBE_TTL); + if stale { + let healthy = self.primary.bucket_reachable().await; + if healthy != cache.healthy { + tracing::warn!( + primary = %self.primary.backend(), + healthy, + "primary storage health changed" + ); + } + cache.checked_at = Some(Instant::now()); + cache.healthy = healthy; + } + if cache.healthy { &self.primary } else { fallback } + } + + /// Startup bucket bootstrap: MinIO buckets are created when missing + /// (a fresh `docker compose` MinIO starts empty); R2 buckets are + /// dashboard-managed and left alone. Warn-only — storage stays enabled + /// either way and writes surface their own errors. + pub async fn ensure_buckets(&self) { + for store in std::iter::once(&self.primary).chain(self.fallback.as_ref()) { + if store.backend() != StorageBackend::Minio { + continue; + } + match store.ensure_bucket().await { + Ok(true) => tracing::info!("created MinIO storage bucket"), + Ok(false) => {} + Err(error) => tracing::warn!( + error = ?error, + "could not verify/create the MinIO bucket; uploads may fail until it exists" + ), + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn store(backend: StorageBackend) -> S3Store { + S3Store::new( + backend, + "http://localhost:9000", + "us-east-1", + "test", + "test-secret", + "overup", + true, + ) + } + + #[test] + fn backend_parse_defaults_legacy_values_to_r2() { + assert_eq!(StorageBackend::parse("minio"), StorageBackend::Minio); + assert_eq!(StorageBackend::parse("r2"), StorageBackend::R2); + assert_eq!(StorageBackend::parse(""), StorageBackend::R2); + assert_eq!(StorageBackend::parse("something-else"), StorageBackend::R2); + } + + #[test] + fn store_for_routes_by_marker_and_falls_back_to_primary() { + let storage = Storage::new( + store(StorageBackend::Minio), + Some(store(StorageBackend::R2)), + ); + assert_eq!(storage.store_for("minio").backend(), StorageBackend::Minio); + assert_eq!(storage.store_for("r2").backend(), StorageBackend::R2); + // Legacy/unknown markers read as R2. + assert_eq!(storage.store_for("legacy").backend(), StorageBackend::R2); + + // R2-only deployment: a minio marker has no store — primary wins. + let r2_only = Storage::new(store(StorageBackend::R2), None); + assert_eq!(r2_only.store_for("minio").backend(), StorageBackend::R2); + assert_eq!(r2_only.store_for("r2").backend(), StorageBackend::R2); + } + + #[test] + fn object_keys_are_stable() { + let ws = Uuid::nil(); + let p = Uuid::nil(); + let j = Uuid::nil(); + assert_eq!( + artifact_key(ws, p, j, "dist.tar.gz"), + format!("artifacts/{ws}/{p}/{j}/dist.tar.gz") + ); + assert_eq!(log_key(ws, p, j, 2), format!("logs/{ws}/{p}/{j}-2.log.gz")); + assert!(logo_key(ws, "png").starts_with(&format!("logos/{ws}/"))); + } +} diff --git a/backend/src/services/r2.rs b/backend/src/services/r2.rs index 7652f7f..c7db73c 100644 --- a/backend/src/services/r2.rs +++ b/backend/src/services/r2.rs @@ -1,159 +1,24 @@ -//! Cloudflare R2 storage (artifacts + log archives) via the S3 API. +//! Cloudflare R2 flavor of the shared S3 object store. //! -//! The control plane never proxies artifact bytes: runners upload through -//! short-lived presigned PUT URLs and browsers download through presigned -//! GET URLs. Presigning happens client-side against the R2 S3 endpoint -//! (`https://.r2.cloudflarestorage.com`) — the only domain R2 -//! presigning supports. Uploads are verified server-side with HeadObject -//! before the artifact row flips to `uploaded`. Log archives are small -//! (bounded by the per-job log cap) and are PUT server-side after a job -//! completes; expired objects are deleted by the janitor. - -use std::time::Duration; - -use anyhow::Context; -use aws_sdk_s3::config::{BehaviorVersion, Credentials, Region}; -use aws_sdk_s3::presigning::PresigningConfig; -use uuid::Uuid; - -/// Presigned PUT grants expire quickly; the runner uploads immediately. -pub const UPLOAD_URL_TTL: Duration = Duration::from_secs(15 * 60); -/// Download links are minted per request. -pub const DOWNLOAD_URL_TTL: Duration = Duration::from_secs(10 * 60); - -pub struct R2 { - client: aws_sdk_s3::Client, - bucket: String, -} - -impl R2 { - pub fn new(account_id: &str, access_key_id: &str, secret_access_key: &str, bucket: &str) -> Self { - let credentials = Credentials::new(access_key_id, secret_access_key, None, None, "r2-static"); - let config = aws_sdk_s3::Config::builder() - .behavior_version(BehaviorVersion::latest()) - .region(Region::new("auto")) - .endpoint_url(format!("https://{account_id}.r2.cloudflarestorage.com")) - .credentials_provider(credentials) - .build(); - Self { - client: aws_sdk_s3::Client::from_conf(config), - bucket: bucket.to_string(), - } - } - - /// Deterministic object key; every segment is a UUID and the name is - /// allow-list validated by the caller before it gets here. - pub fn artifact_key(workspace_id: Uuid, pipeline_id: Uuid, job_id: Uuid, name: &str) -> String { - format!("artifacts/{workspace_id}/{pipeline_id}/{job_id}/{name}") - } - - /// Deterministic key for a job's archived (gzip) log; every segment is a - /// UUID or an integer. - pub fn log_key(workspace_id: Uuid, pipeline_id: Uuid, job_id: Uuid, attempt: i32) -> String { - format!("logs/{workspace_id}/{pipeline_id}/{job_id}-{attempt}.log.gz") - } - - /// Server-generated key for a workspace logo. Every segment is a UUID; - /// the extension comes from server-side magic-byte detection, never from - /// the client's filename or declared content type. - pub fn logo_key(workspace_id: Uuid, ext: &str) -> String { - format!("logos/{workspace_id}/{}.{ext}", Uuid::new_v4()) - } - - /// Server-side upload for control-plane-generated objects (log - /// archives). Bodies are small — bounded by the per-job log cap. - pub async fn put_object( - &self, - key: &str, - body: Vec, - content_type: &str, - ) -> anyhow::Result<()> { - self.client - .put_object() - .bucket(&self.bucket) - .key(key) - .content_type(content_type) - .body(body.into()) - .send() - .await - .context("failed to upload object to R2")?; - Ok(()) - } - - /// Best-effort delete; a missing object counts as success (DeleteObject - /// is idempotent on S3-compatible stores). - pub async fn delete_object(&self, key: &str) -> anyhow::Result<()> { - self.client - .delete_object() - .bucket(&self.bucket) - .key(key) - .send() - .await - .context("failed to delete object from R2")?; - Ok(()) - } - - pub async fn presign_put(&self, key: &str, content_type: &str) -> anyhow::Result { - let presigned = self - .client - .put_object() - .bucket(&self.bucket) - .key(key) - .content_type(content_type) - .presigned(PresigningConfig::expires_in(UPLOAD_URL_TTL)?) - .await - .context("failed to presign artifact upload")?; - Ok(presigned.uri().to_string()) - } - - pub async fn presign_get(&self, key: &str, filename: &str) -> anyhow::Result { - let presigned = self - .client - .get_object() - .bucket(&self.bucket) - .key(key) - .response_content_disposition(format!("attachment; filename=\"{filename}\"")) - .presigned(PresigningConfig::expires_in(DOWNLOAD_URL_TTL)?) - .await - .context("failed to presign artifact download")?; - Ok(presigned.uri().to_string()) - } - - /// Presigned GET for inline display (no attachment disposition) — used - /// for workspace logos rendered in `` tags. Same short TTL; the SPA - /// re-mints through the authenticated logo-url endpoint. - pub async fn presign_get_inline(&self, key: &str) -> anyhow::Result { - let presigned = self - .client - .get_object() - .bucket(&self.bucket) - .key(key) - .response_content_disposition("inline") - .presigned(PresigningConfig::expires_in(DOWNLOAD_URL_TTL)?) - .await - .context("failed to presign inline download")?; - Ok(presigned.uri().to_string()) - } - - /// Size of the uploaded object, or None when it does not exist. - pub async fn head_size(&self, key: &str) -> anyhow::Result> { - match self - .client - .head_object() - .bucket(&self.bucket) - .key(key) - .send() - .await - { - Ok(output) => Ok(output.content_length()), - Err(err) => { - if let aws_sdk_s3::error::SdkError::ServiceError(service_err) = &err - && service_err.err().is_not_found() - { - return Ok(None); - } - Err(anyhow::Error::new(err).context("artifact HeadObject failed")) - } - } - } +//! R2 presigning only works against the account-scoped S3 endpoint +//! (`https://.r2.cloudflarestorage.com`), with the literal region +//! `auto` and virtual-host addressing. Everything else — presigned PUT/GET, +//! HeadObject verification, deletes — is the generic +//! [`crate::services::object_store::S3Store`]. R2 stays fully supported: +//! it is the primary store when it is the only one configured, and the +//! fallback when MinIO is also present. + +use crate::config::R2Config; +use crate::services::object_store::{S3Store, StorageBackend}; + +pub fn store(config: &R2Config) -> S3Store { + S3Store::new( + StorageBackend::R2, + &format!("https://{}.r2.cloudflarestorage.com", config.account_id), + "auto", + &config.access_key_id, + &config.secret_access_key, + &config.bucket, + false, + ) } diff --git a/backend/src/state.rs b/backend/src/state.rs index a043445..08697a0 100644 --- a/backend/src/state.rs +++ b/backend/src/state.rs @@ -11,7 +11,7 @@ use crate::services::github_app::GitHubApp; use crate::services::log_hub::LogHub; use crate::services::notification_hub::NotificationHub; use crate::services::notification_projector::NotificationProjector; -use crate::services::r2::R2; +use crate::services::object_store::Storage; use crate::services::runner_hub::RunnerHub; use crate::services::runner_provisioner::RunnerProvisioner; use crate::services::scheduler::Scheduler; @@ -48,8 +48,9 @@ pub struct AppState { pub notification_hub: Arc, /// Wake handle for the audit-tail notification projector loop. pub notification_projector: Arc, - /// Artifact storage; None disables artifact grants cleanly. - pub r2: Option>, + /// Object storage router (MinIO primary / R2 fallback when both are + /// configured); None disables artifact grants cleanly. + pub storage: Option>, /// One-time tickets for cross-origin browser WebSocket auth /// (deployments whose proxy cannot forward upgrades, e.g. Netlify). pub ws_tickets: Arc, @@ -96,14 +97,19 @@ impl AppState { .map(|key| SecretsCrypto::new(key).map(Arc::new)) .transpose()?; - let r2 = config.r2.as_ref().map(|r2| { - Arc::new(R2::new( - &r2.account_id, - &r2.access_key_id, - &r2.secret_access_key, - &r2.bucket, - )) - }); + // MinIO (when configured) is the default/primary object store and + // R2 the fallback; R2 alone keeps its historical primary role. + let storage = match (config.minio.as_ref(), config.r2.as_ref()) { + (Some(minio), r2) => Some(Arc::new(Storage::new( + crate::services::minio::store(minio), + r2.map(crate::services::r2::store), + ))), + (None, Some(r2)) => Some(Arc::new(Storage::new( + crate::services::r2::store(r2), + None, + ))), + (None, None) => None, + }; // Every WorkspaceHub::publish marks its workspace dirty on this // indexer and pokes the notification projector, so the hub is @@ -127,7 +133,7 @@ impl AppState { search_indexer, notification_hub: Arc::new(NotificationHub::default()), notification_projector, - r2, + storage, ws_tickets: Arc::new(WsTicketStore::default()), // Requires async Docker probing; main fills it in right after. runner_provisioner: None, diff --git a/docker-compose.yml b/docker-compose.yml index 7b178d7..4000a10 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -17,5 +17,34 @@ services: timeout: 5s retries: 10 + # MinIO — the default/primary object store (artifacts + log archives + + # logos). Point the backend at it with: + # MINIO_ENDPOINT=http://localhost:9000 + # MINIO_ACCESS_KEY_ID=overup + # MINIO_SECRET_ACCESS_KEY=overup-minio + # MINIO_BUCKET=overup + # The backend creates the bucket at startup if it is missing. Console UI + # at http://localhost:9001 (same credentials). Dev credentials only — + # never reuse them in a real deployment. + minio: + image: minio/minio:latest + container_name: overup-minio + restart: unless-stopped + command: server /data --console-address ":9001" + environment: + MINIO_ROOT_USER: overup + MINIO_ROOT_PASSWORD: overup-minio + ports: + - "9000:9000" + - "9001:9001" + volumes: + - overup_minio:/data + healthcheck: + test: ["CMD", "mc", "ready", "local"] + interval: 5s + timeout: 5s + retries: 10 + volumes: overup_pgdata: + overup_minio: