Run child tables cascade from run.id
Migration ccb21eacf5d5 deletes rows whose run is gone, then adds foreign keys with ON DELETE CASCADE from run_node, run_metric and run_artifact to run.id. Deleting runs is now the one statement. A node or metric written after its run was deleted is dropped with an info line. Migrations run with foreign keys off, since a batch rebuild of run would otherwise cascade into every child row.
This commit is contained in:
@@ -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")
|
||||
@@ -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)))
|
||||
|
||||
|
||||
|
||||
+27
-13
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user