fix(migrations): never redefine a compatibility view whose old name a project model now uses

A recorded compatibility view whose old name is a project model's
destination (for example after an interrupted early drop) is no longer
refreshed, rebound, or released by builds of the destination. Before, when
the project's model built at that name before the destination, the
destination's refresh overwrote the project's view with the compatibility
SQL; concurrent builds made that order likely.

Refs CHI-391

Author: kvlonge

src/sqlbuild/compiler/planner/_helpers/migrations/old_names.py

@@ -119,7 +119,7 @@ def plan_old_name_views(
         if history.move.target_name in (None, target_name)
     )
     views: tuple[OldNameView, ...] = tuple(
-        _recorded_view(runtime=runtime, history=history)
+        _recorded_view(runtime=runtime, history=history, models=runtime.project.models)
         for history in histories
         if history.created is not None and history.dropped is None
     )
@@ -562,7 +562,9 @@ def claim_message(*, model_name: str, view: OldNameView, now: datetime) -> str:
     )
 
 
-def _recorded_view(*, runtime: PlannerRuntime, history: OldNameViewHistory) -> OldNameView:
+def _recorded_view(
+    *, runtime: PlannerRuntime, history: OldNameViewHistory, models: Iterable[CompiledModel]
+) -> OldNameView:
     return OldNameView(
         destination_model=history.move.destination_model,
         old=_location(runtime=runtime, relation=history.old),
@@ -571,6 +573,14 @@ def _recorded_view(*, runtime: PlannerRuntime, history: OldNameViewHistory) -> O
         column_aliases=() if history.created is None else history.created.column_aliases,
         migration_event_id=history.move.event_id,
         target_name=history.move.target_name,
+        name_reused_by=next(
+            (
+                model.name
+                for model in models
+                if migration_relation_for_location(model.destination).matches(history.old)
+            ),
+            None,
+        ),
     )
 
 

src/sqlbuild/compiler/planner/models.py

@@ -987,6 +987,7 @@ class OldNameView:
     column_aliases: tuple[tuple[str, str], ...]
     migration_event_id: str
     target_name: str | None
+    name_reused_by: str | None = None
 
 
 @dataclass(frozen=True)

src/sqlbuild/executor/migrations/_helpers/old_name_views.py

@@ -421,7 +421,7 @@ def _migration_event_id(*, entry: OldNameViewPlanEntry, run_id: str) -> str:
 def views_reading(
     *, location: CompiledRelationLocation, recorded_views: tuple[OldNameView, ...]
 ) -> tuple[OldNameViewSource, ...]:
-    """Return recorded views that read one relation, then views chained onto those, in order."""
+    """Return recorded views reading one relation, then chained views; reused names are skipped."""
 
     ordered: list[OldNameViewSource] = []
     seen: set[tuple[str | None, str, str]] = set()
@@ -639,7 +639,8 @@ def _direct_readers(
     return tuple(
         view
         for view in recorded_views
-        if migration_relation_for_location(view.new).matches(relation)
+        if view.name_reused_by is None
+        and migration_relation_for_location(view.new).matches(relation)
     )
 
 

tests/integration/src/sqlbuild/cli/commands/main/old_name_views/_test_types.py

@@ -45,6 +45,7 @@ class OldNameJanitorClaimTestCase:
     description: str
     janitor_args: tuple[str, ...]
     install_build_fault: Callable[[pytest.MonkeyPatch], None]
+    claiming_builds: tuple[tuple[str, ...], ...]
     expected_janitor_fragment: str
     expected_old_name_type: str
     expected_final_facts: tuple[str, ...]

tests/integration/src/sqlbuild/cli/commands/main/old_name_views/test_old_name_janitor_resume.py

@@ -20,6 +20,7 @@
     OldNameJanitorResumeTestCase,
 )
 from tests.integration.src.sqlbuild.cli.commands.main.old_name_views.helpers import (
+    DESTINATION_MODEL,
     JANITOR_PROJECT_TOML,
     ORIGIN_MODEL,
     claiming_view_models,
@@ -31,6 +32,7 @@
     relation_type_in,
 )
 
+_ONE_BUILD: tuple[tuple[str, ...], ...] = (("build",),)
 _CLAIMED: str = "└── record  dropped  (name now used by another relation, model:revenue)"
 
 
@@ -94,6 +96,7 @@ def test_given_janitor_crash_after_drop_when_rerunning_then_drop_is_recorded_as_
             description="plain janitor run records the claimed name without touching it",
             janitor_args=("janitor", "--auto-approve"),
             install_build_fault=no_build_fault,
+            claiming_builds=_ONE_BUILD,
             expected_janitor_fragment=_CLAIMED,
             expected_old_name_type="VIEW",
             expected_final_facts=("required", "origin_archived", "view_created", "view_dropped"),
@@ -102,6 +105,7 @@ def test_given_janitor_crash_after_drop_when_rerunning_then_drop_is_recorded_as_
             description="repeated early drop leaves the project's view in place",
             janitor_args=("janitor", "--auto-approve", "--drop-old-name-view", "dev.revenue"),
             install_build_fault=no_build_fault,
+            claiming_builds=_ONE_BUILD,
             expected_janitor_fragment=_CLAIMED,
             expected_old_name_type="VIEW",
             expected_final_facts=("required", "origin_archived", "view_created", "view_dropped"),
@@ -110,6 +114,19 @@ def test_given_janitor_crash_after_drop_when_rerunning_then_drop_is_recorded_as_
             description="repeated early drop spares a project view built without a fingerprint",
             janitor_args=("janitor", "--auto-approve", "--drop-old-name-view", "dev.revenue"),
             install_build_fault=fail_view_fingerprint_write,
+            claiming_builds=_ONE_BUILD,
+            expected_janitor_fragment=_CLAIMED,
+            expected_old_name_type="VIEW",
+            expected_final_facts=("required", "origin_archived", "view_created", "view_dropped"),
+        ),
+        OldNameJanitorClaimTestCase(
+            description="destination built after the project's view leaves that view alone",
+            janitor_args=("janitor", "--auto-approve", "--drop-old-name-view", "dev.revenue"),
+            install_build_fault=no_build_fault,
+            claiming_builds=(
+                ("build", "--select", ORIGIN_MODEL),
+                ("build", "--select", DESTINATION_MODEL),
+            ),
             expected_janitor_fragment=_CLAIMED,
             expected_old_name_type="VIEW",
             expected_final_facts=("required", "origin_archived", "view_created", "view_dropped"),
@@ -140,7 +157,9 @@ def test_given_interrupted_early_drop_when_project_builds_the_name_then_janitor_
     )
     with monkeypatch.context() as patch:
         test_case.install_build_fault(patch)
-        _ = build_ok(project_dir=tmp_path, capsys=capsys)
+        for args in test_case.claiming_builds:
+            claiming: CliRun = run_sqb(project_dir=tmp_path, args=args, capsys=capsys)
+            assert claiming.exit_code == 0, claiming.output
 
     janitor: CliRun = run_sqb(project_dir=tmp_path, args=test_case.janitor_args, capsys=capsys)
     old_name_type: str | None = relation_type_in(