From dd8cb74bf3c04ce55374d6e4f59ec2c11f86071f Mon Sep 17 00:00:00 2001 From: Simon Pinfold Date: Wed, 26 Aug 2026 12:02:27 -0700 Subject: [PATCH] fix(assets): narrow create_content integrity handling; report protected tags honestly (review-12, review2-18) --- app/assets/database/queries/records.py | 31 +++++++++++- app/assets/database/queries/tags.py | 6 ++- app/assets/services/tagging.py | 16 +++++- .../assets_test/queries/test_records.py | 27 ++++++++++ .../assets_test/services/test_tagging.py | 49 +++++++++++++++++++ 5 files changed, 126 insertions(+), 3 deletions(-) diff --git a/app/assets/database/queries/records.py b/app/assets/database/queries/records.py index d6c64a47c..694ce3908 100644 --- a/app/assets/database/queries/records.py +++ b/app/assets/database/queries/records.py @@ -35,6 +35,33 @@ class RecordPageSpec(NamedTuple): after: RecordCursorBoundary | None = None +_LIVE_PATH_UNIQUE_INDEX = "uq_asset_contents_path_live" + + +def _is_live_path_conflict(error: IntegrityError) -> bool: + """True only for a collision on the live-path uniqueness guard. + + ``create_content`` treats exactly one integrity failure as recoverable: two + live rows racing for the same ``path`` under the partial unique index + ``uq_asset_contents_path_live`` (``asset_contents(path) WHERE is_missing = 0``). + That case is resolved by handing back the row that won the race. Every other + integrity failure — notably the ``ck_asset_contents_size_nonneg`` and + ``ck_asset_contents_mtime_nonneg`` CHECK constraints — MUST propagate; treating + it as the race destroys the real diagnostic and surfaces a misleading + ``NoResultFound`` from the re-query (which finds no live row for the path). + + SQLite names the *column* in the message (``UNIQUE constraint failed: + asset_contents.path``) rather than the index, while Postgres exposes the index + name on ``orig.diag.constraint_name``; match either form. + """ + orig = error.orig + diag_name = getattr(getattr(orig, "diag", None), "constraint_name", None) + if diag_name == _LIVE_PATH_UNIQUE_INDEX: + return True + message = str(orig) + return "UNIQUE constraint failed" in message and "asset_contents.path" in message + + def create_content(session: Session, path: str, hash: str | None = None, size_bytes: int = 0, mtime_ns: int | None = None) -> AssetContent: content = AssetContent(path=path, hash=hash, size_bytes=size_bytes, mtime_ns=mtime_ns) try: @@ -42,7 +69,9 @@ def create_content(session: Session, path: str, hash: str | None = None, size_by session.add(content) session.flush() return content - except IntegrityError: + except IntegrityError as error: + if not _is_live_path_conflict(error): + raise winner = session.execute(sa.select(AssetContent).where(AssetContent.path == path, AssetContent.is_missing.is_(False))).scalar_one() return winner diff --git a/app/assets/database/queries/tags.py b/app/assets/database/queries/tags.py index 332400b64..a35254407 100644 --- a/app/assets/database/queries/tags.py +++ b/app/assets/database/queries/tags.py @@ -1,6 +1,6 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Iterable, Sequence @@ -32,6 +32,10 @@ class RemoveTagsResult: removed: list[str] not_present: list[str] total_tags: list[str] + # Tags that ARE present on the record but carry origin="automatic", so they + # cannot be removed via this API. Kept distinct from ``not_present`` so a + # caller can tell "the tag wasn't there" apart from "the tag is protected". + protected: list[str] = field(default_factory=list) def validate_tags_exist(session: Session, tags: list[str]) -> None: diff --git a/app/assets/services/tagging.py b/app/assets/services/tagging.py index 513cb4e23..a441c963c 100644 --- a/app/assets/services/tagging.py +++ b/app/assets/services/tagging.py @@ -81,6 +81,19 @@ def remove_tags( ) ) ) + # Requested tags that ARE present but carry origin="automatic": they + # cannot be removed via this API, so they belong in their own bucket + # rather than being lumped into not_present (which would falsely claim + # the tag was never on the record). + protected_tags = set( + session.scalars( + select(AssetTag.tag_name).where( + AssetTag.asset_id == reference_id, + AssetTag.origin == "automatic", + AssetTag.tag_name.in_(requested_tags), + ) + ) + ) if removable_tags: session.execute( delete(AssetTag).where( @@ -100,8 +113,9 @@ def remove_tags( return RemoveTagsResult( removed=sorted(removable_tags), - not_present=sorted(requested_tags - removable_tags), + not_present=sorted(requested_tags - removable_tags - protected_tags), total_tags=total_tags, + protected=sorted(protected_tags), ) diff --git a/tests-unit/assets_test/queries/test_records.py b/tests-unit/assets_test/queries/test_records.py index 5272b1b16..a096acfcd 100644 --- a/tests-unit/assets_test/queries/test_records.py +++ b/tests-unit/assets_test/queries/test_records.py @@ -1,6 +1,7 @@ """Tests for the B-schema record/content query layer.""" import pytest from sqlalchemy import create_engine, select, update +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.assets.database.models import Asset, AssetContent @@ -114,3 +115,29 @@ def test_concurrent_create_content_same_path(tmp_path): ).scalars() ) assert len(live_rows) == 1, f"Expected exactly one live row, got {len(live_rows)}" + + +def test_create_content_check_violation_surfaces_the_constraint_error(session): + """Given a create with a negative size_bytes (violates the + ck_asset_contents_size_nonneg CHECK), When create_content runs, Then the real + IntegrityError surfaces — it must NOT be misread as the live-path uniqueness + race and swallowed into a NoResultFound by the ``scalar_one()`` re-query.""" + with pytest.raises(IntegrityError): + create_content(session, path="/tmp/negative-size", size_bytes=-1) + + +def test_create_content_check_violation_on_mtime_surfaces(session): + """Same guarantee for the mtime_ns CHECK: a negative mtime raises the + constraint error, never NoResultFound.""" + with pytest.raises(IntegrityError): + create_content(session, path="/tmp/negative-mtime", size_bytes=0, mtime_ns=-1) + + +def test_create_content_uniqueness_race_returns_existing_live_row(session): + """Regression guard: a genuine live-path uniqueness collision still resolves + by returning the pre-existing live row, not by raising — the retire/dedup + callers in ingest depend on this behaviour.""" + first = create_content(session, path="/tmp/race") + second = create_content(session, path="/tmp/race") + + assert second.id == first.id diff --git a/tests-unit/assets_test/services/test_tagging.py b/tests-unit/assets_test/services/test_tagging.py index 9d1d6413e..76a6035cf 100644 --- a/tests-unit/assets_test/services/test_tagging.py +++ b/tests-unit/assets_test/services/test_tagging.py @@ -1,6 +1,8 @@ from sqlalchemy.orm import Session +from app.assets.database.models import AssetTag, Tag from app.assets.database.queries import create_content, create_record, fetch_record_tags +from app.assets.services.tagging import remove_tags def test_tags_are_birth_facts_of_a_record(session: Session) -> None: @@ -8,3 +10,50 @@ def test_tags_are_birth_facts_of_a_record(session: Session) -> None: record = create_record(session, content.id, "model", tags=["model", "checkpoint"]) assert fetch_record_tags(session, record.id) == ["checkpoint", "model"] + + +def _attach_automatic_tag(session: Session, record_id: str, tag_name: str) -> None: + if session.get(Tag, tag_name) is None: + session.add(Tag(name=tag_name)) + session.flush() + session.add( + AssetTag(asset_id=record_id, tag_name=tag_name, origin="automatic") + ) + + +def test_removing_present_automatic_tag_reports_it_protected_not_absent( + session: Session, mock_create_session +) -> None: + """Given a record carrying an origin='automatic' tag, When remove_tags is + asked to remove that tag, Then it is reported in the ``protected`` bucket — + present but not removable — and NOT as ``not_present``. A caller must be able + to tell 'the tag wasn't there' apart from 'the tag is there but protected'.""" + content = create_content(session, "/output/protected.png") + record = create_record(session, content.id, "protected-fixture") + _attach_automatic_tag(session, record.id, "auto") + session.commit() + + result = remove_tags(record.id, ["auto"]) + + assert result.protected == ["auto"] + assert result.not_present == [] + assert result.removed == [] + + +def test_remove_tags_separates_removed_protected_and_absent( + session: Session, mock_create_session +) -> None: + """A single remove_tags call across a manual (removable) tag, an automatic + (protected) tag, and a name that was never applied lands each in its own + bucket — removed / protected / not_present respectively.""" + content = create_content(session, "/output/mixed.png") + record = create_record(session, content.id, "mixed-fixture", tags=["keep-me"]) + _attach_automatic_tag(session, record.id, "auto") + session.commit() + + result = remove_tags(record.id, ["keep-me", "auto", "ghost"]) + + assert result.removed == ["keep-me"] + assert result.protected == ["auto"] + assert result.not_present == ["ghost"] + assert fetch_record_tags(session, record.id) == ["auto"]