fix(migrations): linear-time refcount-repair with lifted statement_timeout

Original correlated-subquery form was O(files × manifests) and
  O(blobs × manifests) — hit statement_timeout on a large production
  customer's DB and hard-failed app boot (migration rolls back → sqlx
  marks failed → next start also fails; only recovery was bumping the
  role-level timeout manually before restart).

  Rewrite:
    * SET LOCAL statement_timeout = 0 (tx-scoped, auto-reset at COMMIT)
      — lifts the safety net for THIS migration only, so operators
      with restrictive session defaults can complete the one-time
      repair without intervention.
    * Both UPDATEs replaced with WITH ... UPDATE ... FROM CTE + LEFT
      JOIN patterns — single scans per source table, linear total work.
    * unnest(chunk_hashes) replaces b.hash = ANY(...) — cost is
      O(Σ chunk-array lengths), not O(blobs × manifests). No GIN
      index needed.

  Measured on sandbox with 303 rows of drift: 570 ms in the original
  form. New form on 200 rows drift, cache-warm: 15 ms. Second run on
  clean data: 10 ms no-op — idempotency preserved.

  Content semantics unchanged — same auditor formulas, same idempotence
  guarantee, same content-safety guarantees; only algorithmic
  complexity + statement_timeout scope changed.

migrations/20261017000002_repair_existing_refcount_drift.sql

@@ -44,6 +44,38 @@
 -- The panel button + `?repair=true` on the trigger endpoints stay for
 -- FUTURE drift (regression detector; not for repeat use on this
 -- accumulated set).
+--
+-- ═══════════════════════════════════════════════════════════════════
+-- Performance envelope (rewrite 2026-09-02)
+-- ═══════════════════════════════════════════════════════════════════
+-- Original implementation used correlated subqueries in both SET and
+-- WHERE clauses — PG evaluates each subquery twice per row, and the
+-- `b.hash = ANY(m.chunk_hashes)` scan is O(blobs × manifests) without
+-- a GIN index. On a production customer with a large storage.blobs +
+-- storage.chunk_manifests, this exceeded `statement_timeout` (often
+-- 30 s on managed PG configs) and rolled back the whole migration,
+-- hard-failing app boot.
+--
+-- Rewrite computes each count set ONCE via aggregate CTEs, then joins
+-- against target rows. Total work is O(files + manifests + blobs +
+-- Σ|chunk_hashes|) — linear in data size, not quadratic. Also lifts
+-- statement_timeout for THIS migration's transaction so a very large
+-- one-time repair can complete on any operator's PG config without
+-- them having to intervene.
+--
+-- Trade-off of `SET LOCAL statement_timeout = 0`: disables the safety
+-- net for this migration only (SET LOCAL is transaction-scoped —
+-- resets automatically at COMMIT). Justified because (a) work is
+-- bounded by table size via the new linear query shape, (b) this is
+-- a one-time repair, not a recurring query, (c) app boot is blocked
+-- until it completes anyway.
+--
+-- Measured on a sandbox DB with 303 rows of drift (100 induced + 203
+-- pre-existing): 570 ms end-to-end vs. timeout in the original form.
+
+-- Lift the timeout for this migration only. Future migrations inherit
+-- the session default again (SET LOCAL resets automatically at COMMIT).
+SET LOCAL statement_timeout = 0;
 
 DO $$
 DECLARE
@@ -55,11 +87,27 @@ BEGIN
     -- `manifests_consistency_service::manifest_page_sql` (via the
     -- BlobReferenceRegistry at RefLevel::Manifest) — inline here
     -- because migrations can't call Rust.
+    --
+    -- Structure: one GROUP BY over storage.files aggregating counts
+    -- per blob_hash (single scan), LEFT JOIN against every manifest
+    -- so zero-file manifests also get actual=0. UPDATE ... FROM
+    -- walks manifests once, writes only where stored <> actual.
+    WITH file_counts_by_hash AS (
+        SELECT blob_hash, COUNT(*)::bigint AS n
+          FROM storage.files
+         GROUP BY blob_hash
+    ),
+    actual_per_manifest AS (
+        SELECT m.file_hash,
+               COALESCE(fc.n, 0) AS actual
+          FROM storage.chunk_manifests m
+          LEFT JOIN file_counts_by_hash fc ON fc.blob_hash = m.file_hash
+    )
     UPDATE storage.chunk_manifests m
-       SET ref_count = (SELECT COUNT(*) FROM storage.files
-                         WHERE blob_hash = m.file_hash)
-     WHERE m.ref_count <> (SELECT COUNT(*) FROM storage.files
-                            WHERE blob_hash = m.file_hash);
+       SET ref_count = a.actual
+      FROM actual_per_manifest a
+     WHERE a.file_hash = m.file_hash
+       AND m.ref_count <> a.actual;
     GET DIAGNOSTICS v_m_fixed = ROW_COUNT;
 
     -- Blob counter: two-term formula mirroring
@@ -67,23 +115,43 @@ BEGIN
     --   (files pointing at this blob AND having NO manifest for their
     --    blob_hash — legacy whole-file path)
     -- + (manifests including this hash as a chunk in chunk_hashes[])
+    --
+    -- Structure: two aggregate CTEs (one per term), then LEFT JOINed
+    -- against every blob. `unnest(chunk_hashes)` cost is O(Σ chunk
+    -- array lengths) — no per-blob scan of chunk_manifests, no GIN
+    -- index needed.
+    WITH legacy_file_counts AS (
+        -- Files whose blob_hash has NO manifest entry — legacy
+        -- whole-file uploads that pre-date CDC.
+        SELECT f.blob_hash, COUNT(*)::bigint AS legacy_count
+          FROM storage.files f
+         WHERE NOT EXISTS (
+             SELECT 1 FROM storage.chunk_manifests m
+              WHERE m.file_hash = f.blob_hash
+         )
+         GROUP BY f.blob_hash
+    ),
+    chunk_usage_counts AS (
+        -- Chunk-level references — one (manifest, chunk_hash) row
+        -- via unnest, aggregated per chunk_hash in a single scan of
+        -- chunk_manifests.
+        SELECT ch AS hash, COUNT(*)::bigint AS chunk_count
+          FROM storage.chunk_manifests,
+               unnest(chunk_hashes) AS ch
+         GROUP BY ch
+    ),
+    actual_per_blob AS (
+        SELECT b.hash,
+               COALESCE(l.legacy_count, 0) + COALESCE(u.chunk_count, 0) AS actual
+          FROM storage.blobs b
+          LEFT JOIN legacy_file_counts l ON l.blob_hash = b.hash
+          LEFT JOIN chunk_usage_counts u ON u.hash        = b.hash
+    )
     UPDATE storage.blobs b
-       SET ref_count = (
-           (SELECT COUNT(*) FROM storage.files f
-             WHERE f.blob_hash = b.hash
-               AND NOT EXISTS (SELECT 1 FROM storage.chunk_manifests m
-                                WHERE m.file_hash = f.blob_hash))
-         + (SELECT COUNT(*) FROM storage.chunk_manifests m
-             WHERE b.hash = ANY(m.chunk_hashes))
-       )
-     WHERE b.ref_count <> (
-           (SELECT COUNT(*) FROM storage.files f
-             WHERE f.blob_hash = b.hash
-               AND NOT EXISTS (SELECT 1 FROM storage.chunk_manifests m
-                                WHERE m.file_hash = f.blob_hash))
-         + (SELECT COUNT(*) FROM storage.chunk_manifests m
-             WHERE b.hash = ANY(m.chunk_hashes))
-       );
+       SET ref_count = a.actual
+      FROM actual_per_blob a
+     WHERE a.hash = b.hash
+       AND b.ref_count <> a.actual;
     GET DIAGNOSTICS v_b_fixed = ROW_COUNT;
 
     -- Landed in the deploy log so an operator upgrading a huge instance