catalog: repair dependent recorded_at only after winning the episodes race

[chintondutta]

Fixes #46.

What was broken

Two callers appending the identical outcome concurrently to Catalog (e.g. a retried or duplicate-dispatched Airflow batch task) both pass the exists("episodes/...") early-return check -- it and the dependent-table write loop are not one atomic operation. The old fallback treated any dependent that lost its create-if-absent race as certain crash debris and unconditionally deleted + overwrote it with the current caller's own recorded_at:

if not self.location.store_file_if_absent(staged_file, dependent_key):
    # A crashed earlier append left this dependent behind ...
    self.location.delete(dependent_key)
    self.location.store_file_if_absent(staged_file, dependent_key)

That reasoning can't actually distinguish "stale debris from an abandoned attempt" from "a live sibling racer computing the identical outcome right now" -- both look identical from here (dependent exists, episodes doesn't). Each of the 4 dependent tables raced and "repaired" independently, so the episodes file and its dependents could end up with different recorded_at values for one outcome, and the dependents could even disagree with each other.

This isn't just cosmetic: curation.py's episodes_latest/measurements_latest views each independently rank rows by (recorded_at DESC, run_fingerprint DESC) per table. On an episode with more than one recorded run, a mismatch introduced by this race could make the two views pick different run_fingerprints as "latest" and the episodes view's join would then stitch together rows from two different runs.

Fix

Don't guess at dependent-write time. store_file_if_absent on episodes is atomic, so at most one caller ever observes True for a given file_stem -- defer any repair until that unique winner is known:

episodes_created = self.location.store_file_if_absent(...)
if episodes_created:
    # We are now the unique, durable owner of this file_stem.
    for table_name in (...):
        self.location.publish(staged_file, dependent_key)  # unconditional atomic replace

The loser of the episodes race does nothing further -- its own dependent writes, if any transiently won, get overwritten by the winner's repair pass. StorageRoot.publish is already an unconditional atomic replace used elsewhere for exactly this "overwrite is correct" case, so no new storage primitive was needed.

This correctly handles both scenarios that need different treatment:

• True crash-repair (sequential retry after a crash, no live racer): the retry is the only caller running, so it always wins episodes and repairs the stale dependents left by the crashed attempt -- unchanged behavior, still covered by the existing test_crash_repaired_append_keeps_one_recorded_at_across_tables.
• Concurrent race (two callers truly in parallel): exactly one wins episodes and repairs all dependents to its own recorded_at, including ones the other caller initially won -- the new bug this PR fixes.

Testing

New regression test test_concurrent_append_of_the_identical_outcome_keeps_one_recorded_at (tests/test_catalog_curation.py) uses a small storage-boundary test double (a barrier gating store_file_if_absent) to force two threads to interleave reliably rather than intermittently, then asserts all 5 tables carry one recorded_at. Confirmed it fails against the pre-fix code (mixed recorded_at across tables: {...}, 2 distinct values) and passes with the fix.

uv run pytest tests/test_catalog_curation.py -q   # 17 passed
uv run pytest -q                                  # unrelated: tests/test_ffmpeg.py fails/errors in this sandbox (no ffmpeg/ffprobe on PATH); everything else passes: 277 passed, 2 skipped
uv run ruff check --fix
uv run ruff format
uv run ty check

No stored-format changes -- this only affects which recorded_at a dependent's row content carries, never which rows exist or what they otherwise contain (guaranteed identical by file_stem = episode_id-run_fingerprint, which already encodes the full observable outcome).

[kstonekuan]

Thank you @chintondutta, another excellent find and fix. What I validated locally:

• Full quality gate clean and all 304 tests pass (including the ffmpeg suite your sandbox could not run).
• Confirmed the new regression test fails against pre-fix catalog.py (mixed recorded_at, 2 distinct values) and passes with the fix.
• Traced the interleavings: the loser's create-if-absent writes can never overwrite the winner's repair pass (create-only loses to an existing file), and the winner's publish is ordered after the unique episodes win, so the final state is always the winner's. The crash-retry path is unchanged and still covered by the existing test.

The barrier-gated storage double is a great pattern: it forces the exact interleaving instead of hoping a stress loop hits it.

One observation, not a blocker: there is a residual double-failure window. If the winner crashes mid-repair (after episodes landed) while stale debris from an earlier crash existed, a later retry early-returns on exists(episodes) and never repairs the mixed dependents. That window is strictly narrower than the race this fixes and needs two failures to hit, so it is fine to leave; if it ever matters we can revisit with repair-on-read in the curation views. Feel free to open an issue for it if you want to track it.

Merging now.