fix(cutover): resume only on dst>=src; residual full-compare gate; test hygiene

Address Baz round (items 1-5):
- backfill.sh (#916, high): resume skips a window only when dst >= src. DIVERGENCE is
  no longer a resume criterion — a partially-copied window within tolerance would be
  skipped and its pre-anchor rows lost forever (delta only re-copies at/after
  backfill_start). The INSERT is idempotent, so re-copying a short window is safe;
  DIVERGENCE still governs the post-copy abort.
- README (#666, high): the pre-EXCHANGE fidelity gate must be a full compare
  (--sample-mod 1 --weeks-stride 1, no week narrowing); a single cross-project
  workspace-scoped residual row can be hashed out or week-skipped by sampling.
- delete_traffic.py (#890): comment referenced newest 500; now references REFILL_FETCH.
- TracesLocalV2CutoverTest (#901): wrapInDistributed() uses a text block + .formatted().
- _common.py (#922): delete the cutover-anchor trace in a finally block so discovery
  leaves no extra live trace skewing seeded counts.

Gate test 10/10.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: andrescrz

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

@@ -629,8 +629,11 @@ cheap (stage A); the bridge stays enabled so nothing is lost on a retry.
 - [ ] **Schema-parity guards green** — `cutoverCopiesEveryBaseColumn` and `successorMaterializedColumnsMatchSource` pass
       on the release, so the cutover copies every base column of `traces` and the two tables' base and materialized
       columns match.
-- [ ] **Fidelity verified** — `verify.sh` passes (full, or a documented representative sample) between source and
-      destination before the EXCHANGE. Re-run `delta_replay.sh` then `verify.sh` until it PASSES: while the buffer holds
+- [ ] **Fidelity verified** — `verify.sh` passes between source and destination before the EXCHANGE. This gate MUST be a
+      **full compare** (`--sample-mod 1 --weeks-stride 1`, no `--from-week`/`--to-week` narrowing): a cross-project
+      workspace-scoped residual row is a single key, so any sampling (`--sample-mod > 1`), week stride, or week narrowing
+      can hash it out or skip its week and still report `ok=1`. Reserve sampling/ranged runs for follow-up confidence
+      *after* the full gate passes. Re-run `delta_replay.sh` then `verify.sh` until it PASSES: while the buffer holds
       writes (or, on a rehearsal without it, once traffic is quiescent) the last delta must catch every in-flight write.
 - [ ] **`Distributed` wrap gated on app-readiness** — apply the wrap (step 4, part 2) only when the delete/read DAOs are
       sharding-aware; otherwise stop after the `EXCHANGE`, since a lightweight `DELETE` against a `Distributed` `traces`

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

@@ -210,9 +210,13 @@ insert_window() {
     src="$(count_src_uniq "$lo" "$hi")"
     dst="$(count_dst_uniq "$lo" "$hi")"
 
-    # Resume: destination already has this window (exact, ahead due to concurrent deletes, or within tolerance).
-    if [[ "$dst" != "0" ]] \
-        && [[ "$(awk -v s="$src" -v d="$dst" -v p="$DIVERGENCE" 'BEGIN { print (d >= s || (s - d) / s <= p) ? 1 : 0 }')" == "1" ]]; then
+    # Resume: skip only when the destination already holds at least as many logical rows as the source (exact, or ahead
+    # because concurrent deletes shrank the source). DIVERGENCE is NOT a resume criterion: a partially-copied window can
+    # sit a hair short of src yet within tolerance, and skipping it would leave those rows missing forever — the delta
+    # step only re-copies rows at/after backfill_start, so a pre-anchor gap is unrepairable. The backfill INSERT is
+    # idempotent (ReplacingMergeTree, mask-honoring), so re-copying a short window is safe and cheap. DIVERGENCE governs
+    # only the post-copy abort below.
+    if [[ "$dst" != "0" && "$dst" -ge "$src" ]]; then
         log "$label ($lo .. $hi): already present (src_uniq=$src dst_uniq=$dst), skipping"
         return
     fi

apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesLocalV2CutoverTest.java

@@ -987,10 +987,11 @@ private void exchangeTables() {
     // need not exist yet), then one atomic multi-target RENAME rotates the data to traces_local and the wrapper into
     // traces (the name freed by the first clause), so traces is never absent on a node.
     private void wrapInDistributed() {
-        execute(("CREATE TABLE traces_dist ON CLUSTER '{cluster}' AS traces "
-                + "ENGINE = Distributed('{cluster}', '" + DATABASE_NAME + "', 'traces_local', sipHash64(project_id))"),
-                _ -> {
-                });
+        execute("""
+                CREATE TABLE traces_dist ON CLUSTER '{cluster}' AS traces
+                ENGINE = Distributed('{cluster}', '%s', 'traces_local', sipHash64(project_id))
+                """.formatted(DATABASE_NAME), _ -> {
+        });
         execute("""
                 RENAME TABLE
                     traces TO traces_local,

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

@@ -62,15 +62,25 @@ def discover_workspace_and_project(
     opik_client.trace(id=anchor_id, name="cutover-anchor", project_name=project_name, input={"anchor": True}).end()
     opik_client.flush()
 
-    deadline = time.time() + timeout_s
-    while time.time() < deadline:
-        rows = ch.query(
-            "SELECT workspace_id, toString(project_id) FROM traces WHERE id = {id:String} LIMIT 1",
-            parameters={"id": anchor_id},
-        ).result_rows
-        if rows:
-            workspace_id, project_id = rows[0]
-            LOGGER.info("Resolved project '%s': workspace_id=%s project_id=%s", project_name, workspace_id, project_id)
-            return workspace_id, project_id
-        time.sleep(0.5)
-    raise TimeoutError(f"anchor trace for project '{project_name}' did not appear in ClickHouse within {timeout_s}s")
+    # Always remove the anchor: leaving it behind would add one live trace to the target project and skew seeded
+    # counts and delete/cutover verification. Cleanup failures are logged, not raised, so they don't mask a real error.
+    try:
+        deadline = time.time() + timeout_s
+        while time.time() < deadline:
+            rows = ch.query(
+                "SELECT workspace_id, toString(project_id) FROM traces WHERE id = {id:String} LIMIT 1",
+                parameters={"id": anchor_id},
+            ).result_rows
+            if rows:
+                workspace_id, project_id = rows[0]
+                LOGGER.info(
+                    "Resolved project '%s': workspace_id=%s project_id=%s", project_name, workspace_id, project_id
+                )
+                return workspace_id, project_id
+            time.sleep(0.5)
+        raise TimeoutError(f"anchor trace for project '{project_name}' did not appear in ClickHouse within {timeout_s}s")
+    finally:
+        try:
+            opik_client.rest_client.traces.delete_traces(ids=[anchor_id])
+        except Exception as exc:  # noqa: BLE001
+            LOGGER.warning("could not delete cutover-anchor trace %s: %s", anchor_id, exc)

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

@@ -77,9 +77,9 @@ def main(project, tps, duration, batch):
                 continue
             pool.extend(fetched)
             if not pool:
-                # An empty refill can be transient: the delete mask may not be visible to search yet, so the newest 500
-                # ids can all still be in `seen`/`pool`. Only stop after several consecutive empty refills, so a mask-lag
-                # blip doesn't end the run while thousands of lower-id traces remain undeleted.
+                # An empty refill can be transient: the delete mask may not be visible to search yet, so the newest
+                # REFILL_FETCH ids can all still be in `seen`/`pool`. Only stop after several consecutive empty refills,
+                # so a mask-lag blip doesn't end the run while thousands of lower-id traces remain undeleted.
                 empty_refills += 1
                 if empty_refills >= EMPTY_REFILL_LIMIT:
                     LOGGER.info("no more traces to delete after %d empty refills; stopping", empty_refills)