diff --git a/backend/fluksio/alembic/versions/e5b8c2f4a913_run_artifact_filename.py b/backend/fluksio/alembic/versions/e5b8c2f4a913_run_artifact_filename.py new file mode 100644 index 0000000..67a987e --- /dev/null +++ b/backend/fluksio/alembic/versions/e5b8c2f4a913_run_artifact_filename.py @@ -0,0 +1,38 @@ +"""run_artifact.filename + +The row is keyed by the message the bytes left on, and that was also the only +name it could answer with. A reference rebuilt from it was then the same bytes +under a different name from the one the producing node gave them. + +Revision ID: e5b8c2f4a913 +Revises: a4d9e2b71c68 +Create Date: 2026-08-26 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "e5b8c2f4a913" +down_revision = "a4d9e2b71c68" +branch_labels = None +depends_on = None + + +def upgrade(): + # Nullable: the rows already here never recorded one, and the message name + # is what they keep answering with. + op.add_column( + "run_artifact", + sa.Column( + "filename", + sqlmodel.sql.sqltypes.AutoString(length=255), + nullable=True, + ), + ) + + +def downgrade(): + op.drop_column("run_artifact", "filename") diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index a984bb5..18e534e 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -184,13 +184,17 @@ def resolve_references( with Session(db_engine) as session: for key, text in pending.items(): if text.startswith(RUN_REF_PREFIX): - resolved[key] = _from_run(session, key, text[len(RUN_REF_PREFIX) :]) + resolved[key] = _from_run( + session, key, text[len(RUN_REF_PREFIX) :], artifacts + ) else: resolved[key] = _from_digest(session, key, text, artifacts) return resolved -def _from_run(session: Session, key: str, spelling: str) -> Any: +def _from_run( + session: Session, key: str, spelling: str, artifacts: ArtifactStore | None = None +) -> Any: """``.`` as the value that run produced.""" run_id, _, output = spelling.partition(".") if not run_id or not output: @@ -210,18 +214,25 @@ def _from_run(session: Session, key: str, spelling: str) -> Any: return copy.deepcopy(result[output]) # The artifact rows are the fallback: bytes a node made that the flow never - # declared as an output. They carry the message name instead, which loads - # the same bytes either way. + # declared as an output. A row records what the node called the file, so + # the reference rebuilt here is the one its producer made — apart from the + # rows written before there was a column to keep it in, which answer with + # the message name as they always did. rows = session.exec( select(RunArtifact).where(col(RunArtifact.run_id) == run_id) ).all() for row in rows: if output in (row.name, row.name.rsplit(".", 1)[-1]): + if artifacts is not None and artifacts.path(row.digest) is None: + raise RunRejected( + f"Parameter '{key}': run '{run_id}' made '{output}', but its " + "bytes are gone from this installation's store" + ) return { "digest": row.digest, "size": row.size, "media_type": row.media_type or "application/octet-stream", - "name": row.name, + "name": row.filename or row.name, } known = ", ".join(sorted({*result, *(row.name for row in rows)})) or "none" raise RunRejected( @@ -252,7 +263,7 @@ def _from_digest( "digest": row.digest, "size": row.size, "media_type": row.media_type or "application/octet-stream", - "name": row.name, + "name": row.filename or row.name, } @@ -982,6 +993,7 @@ class RunService: RunArtifact( run_id=run_id, name=message[:255], + filename=str(ref.get("name") or "")[:255] or None, node=outcome.node[:255], digest=str(ref.get("digest") or "")[:71], size=int(ref.get("size") or 0), diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 3f8e8d7..91a5d45 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -445,7 +445,11 @@ class RunArtifact(SQLModel, table=True): __tablename__ = "run_artifact" run_id: str = Field(primary_key=True, max_length=64) + #: 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 + #: rebuilt from this row is otherwise the same bytes under another name. + filename: str | None = Field(default=None, max_length=255) node: str = Field(default="", max_length=255) digest: str = Field(default="", index=True, max_length=71) size: int = 0 diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 01d0481..95c382c 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -140,6 +140,7 @@ def made_artifact(): RunArtifact( run_id=run_id, name="prepare.dataset", + filename="cities.csv", node="load", digest=digest, size=12, @@ -159,11 +160,29 @@ def test_a_run_reference_resolves_to_what_that_run_produced(made_artifact): artifact_flow(), {"dataset": f"@run:{run_id}.dataset"} ) - # The producer's own reference, file name and all — not one rebuilt from - # the row, which carries the message name instead. + # The producer's own reference, whatever type it made it — the row beside + # it is the fallback, not the first answer. assert resolved["dataset"] == reference +def test_the_artifact_row_answers_with_the_name_the_node_gave_the_file(made_artifact): + """Bytes the flow never declared as an output are reachable through the row.""" + run_id, reference = made_artifact + with Session(db_engine) as session: + run = session.get(Run, run_id) + run.result = {} + session.add(run) + session.commit() + + resolved = resolve_references( + artifact_flow(), {"dataset": f"@run:{run_id}.dataset"} + ) + + assert resolved["dataset"]["digest"] == reference["digest"] + # The file name, not the message it happened to leave on. + assert resolved["dataset"]["name"] == "cities.csv" + + def test_a_bare_digest_resolves_to_the_bytes_under_it(made_artifact): _run_id, reference = made_artifact resolved = resolve_references(artifact_flow(), {"dataset": reference["digest"]})