test(cutover): address second Baz review — wrap-only split-state guard + fixes

- exchange_and_wrap.sh --wrap-only now refuses when traces_local_v2 still exists
  (EXCHANGE done but post-swap RENAME not): wrapping then would orphan the old
  data under the wrong name. Prints the rename to finish first.
- seed_history: build BAD_ID_INSTANT from a fixed date so it can't raise
  ValueError at import on a Feb-29 run (2201 is not a leap year).
- delete_traffic: a transient search_traces failure now backs off and retries
  instead of being mistaken for "no more traces" and ending the run.
- README: document the mutually-exclusive wrap flags and EXCHANGE-only default;
  reword the awkward "EXCHANGEd" to "already-swapped".

Author: andrescrz

apps/opik-backend/data-migrations/traces-local-v2-cutover/README.md

@@ -155,8 +155,12 @@ new table before the EXCHANGE. The replay matches the **full key**, not `id` alo
 >
 > **Applying the deferred wrap later:** once the sharding-aware DAO has shipped, run
 > `exchange_and_wrap.sh --database opik --wrap-only` — it runs the settle gate and applies **only** the `RENAME` +
-> `Distributed` wrap on the already-EXCHANGEd `traces` (no second EXCHANGE, no new `cutover_start`). To roll the wrap
+> `Distributed` wrap on the already-swapped `traces` (no second EXCHANGE, no new `cutover_start`). To roll the wrap
 > back, use `rollback.sh --stage C`.
+>
+> **Wrap flags** (`exchange_and_wrap.sh`, mutually exclusive; default is EXCHANGE-only): omit them (or pass
+> `--skip-wrap`, an explicit alias) to run the EXCHANGE and stop; `--with-wrap` to also apply the wrap in the same run;
+> `--wrap-only` to apply just the deferred wrap later.
 
 **Dedup note.** After the delta, a row can have two physical versions on `traces_local_v2` (the backfilled one and the
 delta one). This is normal — `ReplacingMergeTree` collapses them on merge / under `FINAL` / `LIMIT 1 BY id`, highest

apps/opik-backend/data-migrations/traces-local-v2-cutover/scripts/exchange_and_wrap.sh

@@ -24,7 +24,7 @@
 #   --with-wrap       also apply the Distributed wrap in the same run (EXCHANGE + wrap). Use only once the delete/read
 #                     DAOs are sharding-aware. Mutually exclusive with --skip-wrap / --wrap-only.
 #   --skip-wrap       explicit alias for the default (EXCHANGE only); accepted for clarity and back-compat.
-#   --wrap-only       run ONLY the Distributed wrap on the already-EXCHANGEd `traces` (no EXCHANGE, no new cutover_start)
+#   --wrap-only       run ONLY the Distributed wrap on the already-swapped `traces` (no EXCHANGE, no new cutover_start)
 #                     — the deferred second half of a prior EXCHANGE-only run. Mutually exclusive with the above.
 #   --force           skip the replication-settle gate. By default the swap aborts while any replica still
 #                     has replication-queue backlog or an unfinished mutation on traces / traces_local_v2, since a
@@ -103,7 +103,10 @@ assert_pre_exchange_topology() {
     [[ -n "$(traces_engine traces_local_v2)" ]] || { echo "ERROR: successor 'traces_local_v2' not found; run the backfill + delta first." >&2; exit 1; }
 }
 
-# --wrap-only precondition: traces must be the post-EXCHANGE successor MergeTree (not the original, not already wrapped).
+# --wrap-only precondition: traces must be the post-EXCHANGE successor MergeTree (not the original, not already wrapped),
+# AND the post-swap RENAME must have completed. In the split state (EXCHANGE done, RENAME not) `traces` already holds
+# the successor schema but `traces_local_v2` still holds the old data — wrapping then would orphan the old data under
+# the wrong name (finalize.sh would misread it as the disposable successor). So refuse until the rename is finished.
 assert_pre_wrap_topology() {
     local engine end_time
     engine="$(traces_engine traces)"
@@ -117,6 +120,12 @@ assert_pre_wrap_topology() {
         echo "ERROR: --wrap-only expects the post-EXCHANGE state (traces = successor schema), but traces has Nullable end_time (the EXCHANGE has not run). Run without --wrap-only first." >&2
         exit 1
     }
+    if [[ -n "$(traces_engine traces_local_v2)" ]]; then
+        echo "ERROR: --wrap-only: 'traces_local_v2' still exists — the post-EXCHANGE RENAME did not complete, so wrapping" >&2
+        echo "       now would orphan the old data under the wrong name. Finish the rename first, then re-run --wrap-only:" >&2
+        echo "         clickhouse-client --database $DATABASE --query \"RENAME TABLE $DATABASE.traces_local_v2 TO $DATABASE.traces_pre_cutover_backup ON CLUSTER '{cluster}'\"" >&2
+        exit 1
+    fi
 }
 
 # Pre-EXCHANGE gate: the swap is metadata-only and near-instant, but each replica reads its own local parts afterwards.

tests_load/tests/traces-local-v2-cutover/delete_traffic.py

@@ -29,9 +29,9 @@ def _handle_sigint(_signum, _frame):
 def _fetch_ids(client, project, want, exclude):
     try:
         traces = client.search_traces(project_name=project, max_results=want, truncate=True)
-    except Exception as exc:  # search is best-effort; keep the loop alive
-        LOGGER.warning("search_traces failed: %s", exc)
-        return []
+    except Exception as exc:  # transient search failure: signal the caller to retry, not to treat the pool as drained
+        LOGGER.warning("search_traces failed (will retry): %s", exc)
+        return None
     return [t.id for t in traces if t.id not in exclude]
 
 
@@ -56,7 +56,12 @@ def main(project, tps, duration, batch):
         tick = time.time()
         if len(pool) < batch:
             # Exclude both already-deleted ids and those still queued in `pool`, so a refill can't requeue an in-flight id.
-            pool.extend(_fetch_ids(client, project, want=500, exclude=seen | set(pool)))
+            fetched = _fetch_ids(client, project, want=500, exclude=seen | set(pool))
+            if fetched is None:
+                # transient search failure — back off and retry rather than mistaking it for "no more traces".
+                time.sleep(interval if interval > 0 else 0.5)
+                continue
+            pool.extend(fetched)
             if not pool:
                 LOGGER.info("no more traces to delete; stopping")
                 break

tests_load/tests/traces-local-v2-cutover/seed_history.py

@@ -54,8 +54,9 @@
     "environment",
 ]
 
-# A far-future instant matching the litellm UUIDv7 bug (ids whose embedded timestamp lands around the year 2201).
-BAD_ID_INSTANT = utcnow().replace(year=2201)
+# A far-future instant matching the litellm UUIDv7 bug (ids whose embedded timestamp lands around the year 2201). Built
+# from a fixed date (not now().replace(year=2201)) so it never hits Feb 29 -> ValueError at import on a leap-day run.
+BAD_ID_INSTANT = datetime(2201, 6, 1, tzinfo=timezone.utc)
 
 _EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
 _TAG_POOL = ["prod", "llm", "rag", "eval", "v1", "v2", "canary", "batch", "stream", "agent"]