Scale media collection concurrency with available resources

.env.example

@@ -10,6 +10,8 @@ WEB_PORT=5173
 API_PROXY_URL=http://localhost:3000
 MEDIA_ROOT=./data/media
 CREDENTIALS_ROOT=./data/credentials
+# Optional override; when unset, collection concurrency is sized from CPU and memory.
+# COLLECTION_CONCURRENCY=4
 MAX_ASSET_BYTES=104857600
 MAX_COLLECTION_BYTES=524288000
 MAX_MEDIA_STORAGE_BYTES=4294967296

.pi/skills/railway-cli/SKILL.md

@@ -0,0 +1,142 @@
+---
+name: railway-cli
+description: Operate Railway from the command line to inspect projects, services, deployments, logs, metrics, variables, domains, and resources; deploy local code; provision infrastructure; and troubleshoot build or runtime failures. Use whenever a task requires Railway CLI state or mutations, especially deployment and production-resource work.
+compatibility: Requires the Railway CLI and either interactive Railway login or RAILWAY_TOKEN/RAILWAY_API_TOKEN authentication.
+---
+
+# Railway CLI
+
+Use the Railway CLI for operations that depend on the current repository, its linked project, local deployments, exact command output, SSH, or `railway run`.
+
+Official references:
+
+- CLI: <https://docs.railway.com/cli>
+- Deploying: <https://docs.railway.com/cli/deploying>
+- Variables: <https://docs.railway.com/cli/variable>
+- Logs: <https://docs.railway.com/cli/logs>
+
+## Resource model
+
+Railway resources are scoped as workspace → project → environment → service → deployment. Most commands use the project, environment, and service linked to the current directory. Explicit `--project`, `--environment`, and `--service` scopes override linked context.
+
+## Preflight
+
+For inspection, configuration, and debugging work, establish the executable, authentication, version, and linked scope before mutation:
+
+```bash
+command -v railway
+railway --version
+railway whoami --json
+railway status --json
+```
+
+If the CLI is unavailable, install it using one of Railway's official methods:
+
+```bash
+npm install --global @railway/cli
+# or on macOS
+brew install railway
+```
+
+For an interactive local session, use `railway login`. Use `railway login --browserless` only on a genuinely headless machine. In CI or unattended execution, use exactly one token type:
+
+- `RAILWAY_TOKEN`: project-scoped operations
+- `RAILWAY_API_TOKEN`: account/workspace-scoped operations
+
+Never print, persist, or commit token or secret values. Never read `.env` merely to discover secrets when Railway references or existing configuration are sufficient.
+
+If the user asks to deploy the current directory, `railway up` may be run directly because it handles authentication and project/service setup. Do not add a redundant failing `whoami` preflight in that flow.
+
+## Operating rules
+
+1. Prefer `--json` for bounded reads and machine-readable results.
+2. Resolve the exact project, environment, and service before every mutation.
+3. When several services could match, inspect with `railway service list --json`; do not guess.
+4. State the intended scope before changing production resources or configuration.
+5. Require explicit user confirmation before destructive actions such as deleting services, deployments, databases, buckets, volumes, domains, or variables.
+6. Do not use `--yes` to bypass destructive confirmations unless the user explicitly authorized that exact action.
+7. After every mutation, perform a scoped read-back to verify it.
+8. Never claim deployment success until its terminal status is `SUCCESS`. A detached upload only proves the build was queued.
+9. Prefer committed, reviewable application configuration over ad hoc production changes when the requested behavior belongs in source control.
+10. Before changing concurrency or resource-sensitive behavior, inspect both Railway metrics/limits and the application's actual bottlenecks. Treat CPU, memory, database connections, Redis load, external subprocesses, and persistent-volume semantics as separate constraints.
+
+## Common inspections
+
+```bash
+railway status --json
+railway whoami --json
+railway project list --json
+railway service list --json
+railway deployment list --json
+railway metrics --service <service> --since 1h --json
+railway logs --service <service> --lines 200 --json
+railway logs --service <service> --build --latest --lines 200 --json
+railway logs --service <service> --http --since 1h --json
+railway variable list --service <service> --json
+railway domain list --service <service> --json
+```
+
+Use longer metrics windows when making capacity decisions; compare idle, normal, and peak periods where possible. Use `railway <command> --help` when the installed CLI's syntax differs from documentation—the installed version is authoritative.
+
+## Deployments
+
+Deploy source from the current repository with `railway up`, not `railway deploy` (`deploy` installs templates):
+
+```bash
+railway up --service <service> --environment <environment>
+railway up --detach --service <service> --environment <environment>
+```
+
+For a detached deployment, poll the newest scoped deployment:
+
+```bash
+railway deployment list --service <service> --environment <environment> --json
+```
+
+Report `SUCCESS` as deployed. For `FAILED` or `CRASHED`, inspect build and runtime logs. Report any other state exactly rather than treating it as success.
+
+## Variables and local commands
+
+```bash
+railway variable list --service <service> --environment <environment> --json
+railway variable set KEY=value --service <service> --environment <environment>
+printf %s "$SECRET_VALUE" | railway variable set SECRET_KEY --stdin --service <service> --environment <environment>
+railway variable delete KEY --service <service> --environment <environment>
+railway run --service <service> --environment <environment> -- <command>
+```
+
+Variable mutations can stage or trigger deployments. Verify the resulting variable/deployment state. Avoid placing sensitive values directly in command text when stdin is supported because shell history and tool logs may retain arguments.
+
+## Services and infrastructure
+
+```bash
+railway add --service <name> --json
+railway add --database postgres --json
+railway add --database redis --json
+railway domain list --service <service> --json
+railway logs --service <service> --network --lines 200 --json
+```
+
+Always inspect existing services before provisioning to avoid duplicates.
+
+## Resource-aware application changes
+
+When asked to tune throughput based on Railway resources:
+
+1. Inspect linked context and identify the relevant service.
+2. Query service metrics over a representative period and inspect recent logs/deployments.
+3. Inspect source code to find current limits and why they exist.
+4. Preserve data-integrity constraints; increase parallelism only around work that is safe to overlap.
+5. Prefer a bounded, configurable limit with a conservative default. If runtime resource detection is appropriate, cap it and reserve headroom for the API, database clients, queues, and subprocess peaks.
+6. Use platform-provided limits when available. If only runtime CPU/memory detection is available, document that it reflects container limits and add explicit environment overrides.
+7. Run the project's lint, typecheck, build, and requested checks before proposing deployment.
+8. If the user requested a PR rather than deployment, do not mutate production behavior directly; commit the source-controlled change and open the PR after verification.
+
+## Completion response
+
+Summarize:
+
+1. Action and exact Railway scope.
+2. Observed result or status, omitting secrets.
+3. Source changes and verification performed.
+4. Next action, such as a PR link, deployment approval, or a concrete unresolved blocker.

README.md

@@ -42,6 +42,8 @@ COOKIE_SECURE=true
 WEB_ORIGIN=http://localhost:5173
 MEDIA_ROOT=/data/media
 CREDENTIALS_ROOT=/data/credentials
+# Optional: override resource-aware collection concurrency (1-8).
+# COLLECTION_CONCURRENCY=4
 MAX_ASSET_BYTES=104857600
 MAX_COLLECTION_BYTES=524288000
 MAX_MEDIA_STORAGE_BYTES=4294967296
@@ -88,7 +90,9 @@ Profile discovery uses `gallery-dl` in metadata-only mode and returns at most 24
 
 Extraction defaults to a 100 MiB asset limit and a 500 MiB collection limit. Videos are downloaded at up to 720p when that source is available, then normalized to H.264/AAC with a maximum 1280px dimension and 30 FPS. Still images are converted to WebP with a maximum 1920px dimension; animated GIF, WebP, APNG, and other multi-frame images are preserved unchanged. An optimized file replaces its source only when it is smaller or the source exceeds the configured dimensions. `OPTIMIZATION_TIMEOUT_MS` bounds each FFmpeg operation.
 
-`MAX_MEDIA_STORAGE_BYTES` defaults to 4 GiB. After a collection completes, oldest media is removed when database-tracked usage exceeds `MEDIA_RETENTION_TRIGGER_PERCENT` (80% by default) until it reaches `MEDIA_RETENTION_TARGET_PERCENT` (70%). The worker processes one collection at a time so quota decisions cannot race each other. Metadata probing uses `METADATA_CONCURRENCY` to avoid launching an unbounded number of processes.
+`MAX_MEDIA_STORAGE_BYTES` defaults to 4 GiB. After a collection completes, oldest media is removed when database-tracked usage exceeds `MEDIA_RETENTION_TRIGGER_PERCENT` (80% by default) until it reaches `MEDIA_RETENTION_TARGET_PERCENT` (70%). Retention is serialized independently from collection processing, so collections can run concurrently without overlapping quota decisions.
+
+Collection concurrency is automatically sized when the worker starts. It budgets two CPU threads and 1.5 GiB per collection, reserves 1.5 GiB for the co-located API and worker overhead, and caps automatic concurrency at four. The current Railway backend allocation of 8 vCPU and 8 GiB therefore processes up to four collections concurrently. Set `COLLECTION_CONCURRENCY` to a value from 1 through 8 only when an explicit override is needed. Metadata probing within each collection remains bounded by `METADATA_CONCURRENCY`.
 
 ## S3-compatible media storage
 

apps/worker/src/index.ts

@@ -1,3 +1,4 @@
+import { availableParallelism, totalmem } from 'node:os';
 import { Worker } from 'bullmq';
 import { Redis } from 'ioredis';
 import { loadWorkerConfig, mediaStorageOptions } from '@media-scraper/config';
@@ -10,9 +11,38 @@ import { MediaStorage } from '@media-scraper/storage';
 import { processCollection } from './process-collection.js';
 import { processMediaMaintenance } from './storage-retention.js';
 
+const AUTO_COLLECTION_CONCURRENCY_LIMIT = 4;
+const BYTES_PER_GIBIBYTE = 1024 ** 3;
+const CPU_THREADS_PER_COLLECTION = 2;
 const MAINTENANCE_INTERVAL_MS = 30_000;
+const MEMORY_PER_COLLECTION_BYTES = 1.5 * BYTES_PER_GIBIBYTE;
+const RESERVED_MEMORY_BYTES = 1.5 * BYTES_PER_GIBIBYTE;
 
 const config = loadWorkerConfig();
+const cpuCount = availableParallelism();
+const constrainedMemoryBytes = process.constrainedMemory();
+const memoryLimitBytes =
+  constrainedMemoryBytes > 0 && constrainedMemoryBytes <= totalmem()
+    ? constrainedMemoryBytes
+    : totalmem();
+const automaticCollectionConcurrency = Math.max(
+  1,
+  Math.min(
+    AUTO_COLLECTION_CONCURRENCY_LIMIT,
+    Math.floor(cpuCount / CPU_THREADS_PER_COLLECTION),
+    Math.floor(
+      (memoryLimitBytes - RESERVED_MEMORY_BYTES) / MEMORY_PER_COLLECTION_BYTES,
+    ),
+  ),
+);
+const collectionConcurrency =
+  config.COLLECTION_CONCURRENCY ?? automaticCollectionConcurrency;
+const concurrencySource = config.COLLECTION_CONCURRENCY
+  ? 'configured override'
+  : `${String(cpuCount)} CPUs and ${(memoryLimitBytes / BYTES_PER_GIBIBYTE).toFixed(1)} GiB memory`;
+console.info(
+  `Collection worker concurrency set to ${String(collectionConcurrency)} (${concurrencySource})`,
+);
 const database = createDatabase(config.DATABASE_URL);
 const redis = new Redis(config.REDIS_URL, { maxRetriesPerRequest: null });
 const shutdownController = new AbortController();
@@ -34,7 +64,7 @@ const worker = new Worker<CollectionJobPayload>(
       storage,
       videoMaxDimension: config.VIDEO_MAX_DIMENSION,
     }),
-  { connection: redis, concurrency: 1 },
+  { connection: redis, concurrency: collectionConcurrency },
 );
 
 let maintenancePromise: Promise<void> | undefined;

docker-compose.yml

@@ -75,6 +75,7 @@ services:
       REDIS_URL: redis://redis:6379
       MEDIA_ROOT: /data/media
       CREDENTIALS_ROOT: /data/credentials
+      COLLECTION_CONCURRENCY: ${COLLECTION_CONCURRENCY:-}
       MAX_ASSET_BYTES: ${MAX_ASSET_BYTES:-104857600}
       MAX_COLLECTION_BYTES: ${MAX_COLLECTION_BYTES:-524288000}
       MAX_MEDIA_STORAGE_BYTES: ${MAX_MEDIA_STORAGE_BYTES:-4294967296}

packages/config/src/index.ts

@@ -7,6 +7,7 @@ const WORKSPACE_MARKER = 'pnpm-workspace.yaml';
 const DEFAULT_MAX_ASSET_BYTES = 100 * 1024 * 1024;
 const DEFAULT_MAX_COLLECTION_BYTES = 500 * 1024 * 1024;
 const DEFAULT_MAX_MEDIA_STORAGE_BYTES = 4 * 1024 * 1024 * 1024;
+const MAX_COLLECTION_CONCURRENCY = 8;
 const DEFAULT_OPTIMIZATION_TIMEOUT_MS = 10 * 60 * 1_000;
 const DEFAULT_PRESIGNED_URL_TTL_SECONDS = 15 * 60;
 const DEFAULT_PROFILE_DISCOVERY_CACHE_TTL_SECONDS = 10 * 60;
@@ -33,6 +34,10 @@ if (existsSync(environmentPath)) process.loadEnvFile(environmentPath);
 
 const positiveInteger = z.coerce.number().int().positive();
 const percentage = z.coerce.number().int().min(1).max(100);
+const optionalPositiveInteger = z.preprocess(
+  (value) => (value === '' ? undefined : value),
+  positiveInteger.optional(),
+);
 const optionalEnvironmentString = z.preprocess(
   (value) => (value === '' ? undefined : value),
   z.string().min(1).optional(),
@@ -103,6 +108,10 @@ const apiEnvironmentSchema = commonEnvironmentSchema
 
 const workerEnvironmentSchema = commonEnvironmentSchema
   .extend({
+    COLLECTION_CONCURRENCY: optionalPositiveInteger.refine(
+      (value) => value === undefined || value <= MAX_COLLECTION_CONCURRENCY,
+      `Collection concurrency cannot exceed ${String(MAX_COLLECTION_CONCURRENCY)}`,
+    ),
     EXTRACTION_TIMEOUT_MS: positiveInteger.default(1_800_000),
     IMAGE_MAX_DIMENSION: positiveInteger.default(DEFAULT_IMAGE_MAX_DIMENSION),
     MAX_ASSET_BYTES: positiveInteger.default(DEFAULT_MAX_ASSET_BYTES),