Background workers

alembic/versions/6fb9b8fbac38_add_ingestion_status_updated_at.py

@@ -0,0 +1,40 @@
+"""add ingestion_status_updated_at
+
+Revision ID: 6fb9b8fbac38
+Revises: b2c52ee8ff12
+Create Date: 2026-07-29 14:34:12.166293
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "6fb9b8fbac38"
+down_revision: Union[str, Sequence[str], None] = "b2c52ee8ff12"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # A server_default lets the database fill existing rows when the NOT NULL
+    # column is added, so no separate backfill/alter step is needed.
+    with op.batch_alter_table("simulations", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "ingestion_status_updated_at",
+                sa.DateTime(),
+                nullable=False,
+                server_default=sa.text("CURRENT_TIMESTAMP"),
+            )
+        )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    with op.batch_alter_table("simulations", schema=None) as batch_op:
+        batch_op.drop_column("ingestion_status_updated_at")

alembic/versions/b2c52ee8ff12_add_ingestion_status.py

@@ -0,0 +1,71 @@
+"""Add ingestion status
+
+Revision ID: b2c52ee8ff12
+Revises: 28bee3aa2429
+Create Date: 2026-05-11 16:16:03.768893
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "b2c52ee8ff12"
+down_revision: Union[str, Sequence[str], None] = "28bee3aa2429"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    conn = op.get_bind()
+    dialect = conn.dialect.name
+    if dialect == "postgresql":
+        op.execute(
+            "CREATE TYPE ingestionstatus AS ENUM ('QUEUED', 'COPYING', 'COPIED', "
+            "'VALIDATING', 'VALIDATED', 'COMPLETED', 'COPY_FAILED', "
+            "'VALIDATION_FAILED')"
+        )
+    with op.batch_alter_table("simulations", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "ingestion_status",
+                sa.Enum(
+                    "QUEUED",
+                    "COPYING",
+                    "COPIED",
+                    "VALIDATING",
+                    "VALIDATED",
+                    "COMPLETED",
+                    "COPY_FAILED",
+                    "VALIDATION_FAILED",
+                    name="ingestionstatus",
+                ),
+                nullable=True,
+            )
+        )
+        batch_op.add_column(sa.Column("ingestion_version", sa.Integer(), nullable=True))
+    op.execute(
+        "UPDATE simulations SET ingestion_status = 'COMPLETED' WHERE ingestion_status "
+        "IS NULL"
+    )
+    op.execute(
+        "UPDATE simulations SET ingestion_version = 0 WHERE ingestion_version IS NULL"
+    )
+    with op.batch_alter_table("simulations", schema=None) as batch_op:
+        batch_op.alter_column("ingestion_status", nullable=False)
+        batch_op.alter_column("ingestion_version", nullable=False)
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    with op.batch_alter_table("simulations", schema=None) as batch_op:
+        batch_op.drop_column("ingestion_version")
+        batch_op.drop_column("ingestion_status")
+    conn = op.get_bind()
+    dialect = conn.dialect.name
+    if dialect == "postgresql":
+        op.execute("DROP TYPE ingestionstatus")

docs/celery.md

@@ -0,0 +1,115 @@
+# Celery async task processing
+
+SimDB uses [Celery](https://docs.celeryproject.org/) to run asynchronous background
+tasks such as copying simulation files and completing the ingestion pipeline.
+
+## Overview
+
+When simulations are uploaded via the REST API, the server offloads heavy operations
+to Celery workers instead of blocking the HTTP request. Tasks are defined in
+`src/simdb/workers/tasks.py`:
+
+- `copy_files_task` — copies input/output files from source locations to the server's
+  upload folder and updates the simulation's ingestion status.
+- `complete_ingestion_task` — marks a simulation as fully ingested.
+- `validate_imas_task` — runs validation checks on IMAS data (placeholder).
+- `send_email_task` — sends email notifications.
+- `fail_stale_ingestions_task` — periodic sweep (run by beat) that marks simulations
+  stuck in a non-terminal ingestion state as failed, so they cannot remain stuck
+  forever after a worker is hard-killed mid-ingestion. See "Recovering stuck
+  ingestions" below.
+
+Tasks can be chained in the API endpoint:
+
+```python
+copy_files = copy_files_task.si(simulation.uuid, ...)
+complete = complete_ingestion_task.si(simulation.uuid)
+_ = (copy_files | complete).apply_async()
+```
+
+## Configuration
+
+Celery is configured via `app.cfg`:
+
+| Section | Option                  | Required | Description                                      |
+|---------|-------------------------|----------|--------------------------------------------------|
+| celery  | broker_url              | no       | Redis URL for the message broker. Defaults to `redis://localhost:6379/0` |
+| celery  | result_backend          | no       | Redis URL for results storage. Defaults to `redis://localhost:6379/0`   |
+| celery  | task_soft_time_limit    | no       | Seconds before a running task is asked to stop (raises `SoftTimeLimitExceeded`, which ingestion tasks turn into a `COPY_FAILED` status). Defaults to `3600`. |
+| celery  | task_time_limit         | no       | Hard limit in seconds; the task is forcibly killed. Defaults to `3660`. |
+| celery  | stale_sweep_interval    | no       | How often (seconds) the beat scheduler runs `fail_stale_ingestions_task`. Defaults to `300`. |
+| celery  | stale_ingestion_timeout | no       | How long (seconds) a simulation may sit in a non-terminal ingestion state before the sweep fails it. Should be larger than `task_time_limit`. Defaults to `7200`. |
+
+Example:
+
+```ini
+[celery]
+broker_url = redis://localhost:6379/0
+result_backend = redis://localhost:6379/0
+task_soft_time_limit = 3600
+task_time_limit = 3660
+stale_sweep_interval = 300
+stale_ingestion_timeout = 7200
+```
+
+## Running workers
+
+### Standalone worker
+
+Start a Celery worker using the built-in CLI:
+
+```bash
+simdb_worker
+```
+
+### Worker with beat scheduler
+
+For periodic tasks (e.g. cleanup, reports), run both the worker and beat:
+
+```bash
+# Terminal 1: worker
+simdb_worker
+
+# Terminal 2: beat scheduler
+simdb_beat
+```
+
+### Flower monitoring
+
+[Flower](https://flower.readthedocs.io/) provides a web UI for monitoring Celery
+workers and tasks:
+
+```bash
+celery -A simdb.workers.celery flower --port=5555
+```
+
+## Recovering stuck ingestions
+
+A simulation moves through non-terminal ingestion states (`QUEUED`, `COPYING`, ...)
+before reaching a terminal one (`COMPLETED`, `COPY_FAILED`, `VALIDATION_FAILED`). A
+simulation is only deletable once it reaches a terminal state, so it is important
+that it never gets stuck. Three mechanisms guard against this:
+
+1. A task that raises is caught and the simulation is marked `COPY_FAILED`.
+2. A task that hangs is stopped by `task_soft_time_limit` and then marked
+   `COPY_FAILED` (the hard `task_time_limit` is an absolute backstop).
+3. A worker that is hard-killed (SIGKILL, OOM, node reboot) never runs its task's
+   error handling, so `fail_stale_ingestions_task` sweeps up any simulation left in a
+   non-terminal state longer than `stale_ingestion_timeout`. This requires the beat
+   scheduler to be running.
+
+The sweep relies on the `ingestion_status_updated_at` column, which is set on insert
+and refreshed on every update, so it reflects how long a simulation has been sitting
+in its current state.
+
+As a manual fallback, an admin can force-delete a stuck simulation regardless of its
+ingestion state:
+
+```
+DELETE /v1.3/simulation/<uuid>?force=true
+```
+
+## Testing with eager mode
+
+In tests, set `task_always_eager = True` to run tasks synchronously without a
+broker.

docs/developer_guide.md

@@ -45,6 +45,22 @@ simdb_server
 
 This will start a server on port 5000. You can test this server is running by opening http://localhost:5000 in a browser.
 
+## Running Celery workers
+
+For development, you typically want to run Celery tasks synchronously. This is
+enabled by setting `task_always_eager = True` in tests (see `tests/remote/api/v1.3/test_simulations3.py`).
+
+To run actual background workers during development:
+
+```bash
+# Worker
+simdb_worker
+
+# Beat scheduler (if needed)
+simdb_beat
+```
+
+See the [Celery documentation](celery.md) for full details.
 ## Swagger API documentation
 
 SimDB provides interactive Swagger API documentation for each API version. The documentation is automatically generated and accessible at different endpoints depending on the API version you want to explore.

docs/maintenance_guide.md

@@ -333,6 +333,12 @@ service nginx restart
 
 You should now be able to check the simdb server is running by going to the http address defined in your nginx site (localhost:80 in the example above).
 
+## Celery background workers
+
+SimDB uses Celery to run asynchronous background tasks such as copying simulation
+files. See the [Celery documentation](celery.md) for details on configuration and
+running workers.
+
 #### Nginx Request Entity Size
 
 You may need to increase the size of uploaded files that Nginx will accept. For SimDB this should be at least 100MB.

pyproject.toml

@@ -64,6 +64,8 @@ server = [
     "flask-restx>=1.0.0",
     "python-magic~=0.4",
     "simplejson~=3.0",
+    "celery>=5.3.0",
+    "redis>=5.0.0",
 ]
 auth-ad = [
     "easyad>=1.0",
@@ -93,12 +95,14 @@ postgres = [
     "psycopg2-binary>=2.8.0",
 ]
 all = [
-    "imas-simdb[server, imas-validator, postgres]"
+    "imas-simdb[server, imas-validator, postgres]",
 ]
 
 [project.scripts]
 simdb = "simdb.cli.simdb:main"
 simdb_server = "simdb.remote.wsgi:run"
+simdb_worker = "simdb.workers.cli:worker"
+simdb_beat = "simdb.workers.cli:beat"
 
 [project.urls]
 Homepage = "https://simdb.iter.org/dashboard/"

scripts/simdb-instance

@@ -19,6 +19,8 @@ Usage:
   scripts/simdb-instance deploy-develop
   scripts/simdb-instance deploy-pr PR_NUMBER [--port PORT]
   scripts/simdb-instance deploy-branch BRANCH [--name NAME] [--port PORT]
+  scripts/simdb-instance workers-start NAME
+  scripts/simdb-instance workers-stop NAME
   scripts/simdb-instance list
   scripts/simdb-instance status NAME
   scripts/simdb-instance logs NAME [SERVICE]
@@ -205,6 +207,7 @@ deploy_commit() {
         --project-directory "$worktree" \
         --file "$worktree/$COMPOSE_FILE" \
         --env-file "$pending_file" \
+        --profile with_workers \
         up -d --build --remove-orphans --wait; then
         mv "$pending_file" "$file"
     else
@@ -331,6 +334,16 @@ main() {
             require_command docker
             deploy_branch "$branch" "$name" "$port"
             ;;
+        workers-start)
+            (($# == 2)) || die "workers-start requires an instance name"
+            require_command docker
+            compose "$2" --profile with_workers up -d --wait worker beat
+            ;;
+        workers-stop)
+            (($# == 2)) || die "workers-stop requires an instance name"
+            require_command docker
+            compose "$2" --profile with_workers stop worker beat
+            ;;
         list)
             list_instances
             ;;