diff --git a/backend/fluksio/alembic/versions/ccb21eacf5d5_run_child_foreign_keys.py b/backend/fluksio/alembic/versions/ccb21eacf5d5_run_child_foreign_keys.py new file mode 100644 index 0000000..c7e15c6 --- /dev/null +++ b/backend/fluksio/alembic/versions/ccb21eacf5d5_run_child_foreign_keys.py @@ -0,0 +1,38 @@ +"""run_node, run_metric, run_artifact → run.id + +A foreign key from each run child table to its run, cascading on delete, so +deleting a run takes its rows with it and nothing can write one for a run that +is not there. + +Revision ID: ccb21eacf5d5 +Revises: 2e7ca7d10044 +Create Date: 2026-10-15 + +""" + +from alembic import op + +# revision identifiers, used by Alembic. +revision = "ccb21eacf5d5" +down_revision = "2e7ca7d10044" +branch_labels = None +depends_on = None + +TABLES = ("run_node", "run_metric", "run_artifact") + + +def upgrade(): + for table in TABLES: + # Rows a late write left behind after their run was deleted. Nothing + # reads them, and the constraint would refuse them. + op.execute(f"DELETE FROM {table} WHERE run_id NOT IN (SELECT id FROM run)") + with op.batch_alter_table(table) as batch_op: + batch_op.create_foreign_key( + f"fk_{table}_run_id_run", "run", ["run_id"], ["id"], ondelete="CASCADE" + ) + + +def downgrade(): + for table in TABLES: + with op.batch_alter_table(table) as batch_op: + batch_op.drop_constraint(f"fk_{table}_run_id_run", type_="foreignkey") diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index 64479c9..b843e0f 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -851,10 +851,10 @@ async def retry_run(run_id: str, request: Request, user: CurrentUser) -> Any: def _drop_runs(session: Session, chosen: Any) -> None: """A selection of runs and everything hanging off them. - The run tables carry a plain string ``run_id`` and no foreign key, so - nothing cascades on its own and these four statements are the whole of it. - ``flow_run``, ``metric_minute`` and ``engine_event`` stay — they are the - observability record and are pruned on their own window. + The run tables' foreign keys cascade, so deleting the runs takes their + nodes, metrics and artifact rows with them. ``flow_run``, ``metric_minute`` + and ``engine_event`` stay — they are the observability record and are + pruned on their own window. Two things this costs, both deliberate. ``RunNode.outputs`` *is* the stage cache, so a later run loses hits these would have served. And a node @@ -864,9 +864,6 @@ def _drop_runs(session: Session, chosen: Any) -> None: keeps whatever a ``run_artifact`` row or a live message still names, so dropping the rows is enough and the hourly sweep reclaims the blobs. """ - session.execute(delete(RunNode).where(col(RunNode.run_id).in_(chosen))) - session.execute(delete(RunMetric).where(col(RunMetric.run_id).in_(chosen))) - session.execute(delete(RunArtifact).where(col(RunArtifact.run_id).in_(chosen))) session.execute(delete(Run).where(col(Run.id).in_(chosen))) diff --git a/backend/fluksio/core/db.py b/backend/fluksio/core/db.py index 0b2d91a..833f8ba 100644 --- a/backend/fluksio/core/db.py +++ b/backend/fluksio/core/db.py @@ -66,8 +66,9 @@ def _sqlite_pragmas(dbapi_connection: Any, _record: Any) -> None: # Commit without waiting for the platter. A power cut can lose the last # transactions; it cannot corrupt the file. cursor.execute("PRAGMA synchronous=NORMAL") - # Off by default in SQLite, and `ondelete="CASCADE"` on the OAuth tables is - # the whole of revoking a client. + # Off by default in SQLite, and `ondelete="CASCADE"` is the whole of + # revoking an OAuth client and of deleting a run's nodes, metrics and + # artifacts with it. cursor.execute("PRAGMA foreign_keys=ON") # A second writer waits rather than raising "database is locked". cursor.execute("PRAGMA busy_timeout=30000") @@ -109,17 +110,30 @@ def _alembic_config(connection: Any) -> Config: def migrate(engine: Engine) -> None: """Bring the schema up to date, creating it if there is nothing there.""" - with engine.begin() as connection: - config = _alembic_config(connection) - tables = inspect(connection).get_table_names() - if not tables: - # Nothing to upgrade from. Building the schema from the models is - # both faster and exactly what the revisions would have produced. - SQLModel.metadata.create_all(connection) - command.stamp(config, "head") - logger.info("created the database schema") - return - command.upgrade(config, "head") + with engine.connect() as connection: + # Foreign keys off while the schema changes. A batch migration rebuilds + # a table by dropping it, and dropping `run` with them on cascades into + # every run's nodes, metrics and artifacts. SQLite ignores the pragma + # inside a transaction, so it is set before the migration's opens and + # reset after it closes. + connection.exec_driver_sql("PRAGMA foreign_keys=OFF") + connection.commit() + try: + with connection.begin(): + config = _alembic_config(connection) + tables = inspect(connection).get_table_names() + if not tables: + # Nothing to upgrade from. Building the schema from the + # models is both faster and exactly what the revisions + # would have produced. + SQLModel.metadata.create_all(connection) + command.stamp(config, "head") + logger.info("created the database schema") + return + command.upgrade(config, "head") + finally: + connection.exec_driver_sql("PRAGMA foreign_keys=ON") + connection.commit() def init_db(session: Session) -> None: diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 9ac0ff2..5ee58c3 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -585,6 +585,9 @@ class MetricSink: ) ) session.commit() + except IntegrityError: + # Deleted while it was still reporting, as `_record_node` explains. + logger.info("Run %s is gone; not writing its metrics", self.run_id) except Exception: logger.exception( "Could not write %d metric(s) of %s", len(rows), self.run_id @@ -1568,6 +1571,10 @@ class RunService: ) ) session.commit() + except IntegrityError: + # The run was deleted while this node was still going, say once it + # had been swept as abandoned. Its rows have no run to belong to. + logger.info("Run %s is gone; not recording node '%s'", run_id, outcome.node) except Exception: logger.exception( "Could not record node '%s' of run %s", outcome.node, run_id diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 6665392..ff6902e 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -432,7 +432,9 @@ class RunNode(SQLModel, table=True): __tablename__ = "run_node" - run_id: str = Field(primary_key=True, max_length=64) + run_id: str = Field( + primary_key=True, max_length=64, foreign_key="run.id", ondelete="CASCADE" + ) node: str = Field(primary_key=True, max_length=255) #: ok, error, skipped, cached or cancelled. status: str = Field(default="ok", max_length=16) @@ -470,7 +472,9 @@ class RunMetric(SQLModel, table=True): __tablename__ = "run_metric" - run_id: str = Field(primary_key=True, max_length=64) + run_id: str = Field( + primary_key=True, max_length=64, foreign_key="run.id", ondelete="CASCADE" + ) name: str = Field(primary_key=True, max_length=128) #: -1 for a value with no step of its own — a final score. step: int = Field(primary_key=True) @@ -484,7 +488,9 @@ class RunArtifact(SQLModel, table=True): __tablename__ = "run_artifact" - run_id: str = Field(primary_key=True, max_length=64) + run_id: str = Field( + primary_key=True, max_length=64, foreign_key="run.id", ondelete="CASCADE" + ) #: The message the bytes left the node on, which is what addresses them. name: str = Field(primary_key=True, max_length=255) #: What the node called the file, when it said. Kept because a reference diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 902d87c..dfc8764 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -7,6 +7,7 @@ The pipeline half — what a hit restores and what a key is made of — is in import base64 import csv import json +import logging import threading from datetime import UTC, datetime, timedelta @@ -41,6 +42,10 @@ def test_a_run_cache_finds_what_an_earlier_run_recorded(tmp_path): reference = store.put([b"payload"], name="data.bin") plain, with_artifact, collected = "k-plain", "k-artifact", "k-collected" with Session(db_engine) as session: + for run_id in ("cache-1", "cache-2", "cache-3"): + session.add( + Run(id=run_id, flow="study", status="ok", created_at=datetime.now(UTC)) + ) session.add( RunNode( run_id="cache-1", @@ -891,7 +896,7 @@ def test_overview_is_not_read_as_a_run_id(client, superuser_token_headers): # ----------------------------------------------------------------------------- # Deleting a run # -# The route owns the four statements; what these guard is that it takes the +# The foreign keys cascade; what these guard is that deleting a run takes the # children with it and refuses a run the driver is still writing to. # ----------------------------------------------------------------------------- @@ -928,7 +933,7 @@ def deletable_run(): def test_deleting_a_run_takes_its_children_with_it( client, superuser_token_headers, deletable_run ): - """No foreign key cascades here, so the route has to do it itself.""" + """The foreign keys cascade, which needs them on for every connection.""" answer = client.delete( f"{settings.API_V1_STR}/runs/{deletable_run}", headers=superuser_token_headers ) @@ -943,6 +948,18 @@ def test_deleting_a_run_takes_its_children_with_it( assert left == [], f"{table.__name__} rows outlived the run" +def test_a_node_finishing_after_its_run_was_deleted_is_dropped_quietly(caplog): + """The constraint refuses its row, and that is a no-op rather than an error.""" + service = RunService(controller=_OneFlow(FlowDef(name="gone")), queue=_Collect()) + with caplog.at_level(logging.INFO, logger="fluksio.flow.runs"): + service._record_node("del-gone", NodeOutcome(node="gone.a", ok=True)) + + with Session(db_engine) as session: + assert session.get(RunNode, {"run_id": "del-gone", "node": "gone.a"}) is None + assert "is gone" in caplog.text + assert not [record for record in caplog.records if record.levelno >= logging.ERROR] + + def test_deleting_a_run_that_is_not_there_is_a_404(client, superuser_token_headers): answer = client.delete( f"{settings.API_V1_STR}/runs/nope-1", headers=superuser_token_headers