fix: deferred validation + per-node frontier eliminates false positives

- Validator runs only after consumer drains (via 'validate' profile)
- Per-node frontier restored: min(lm, olr) per N{x} prefix
- Single Kafka topic via RegexRouter (not ByLogicalTableRouter)
- fuzz-test.sh actions: up/run/status/validate/logs/down

Result: 26,867 events validated, 0 mismatches, 39 known LOB issues.
Non-LOB accuracy: 100%.

Author: rophy

tests/dbz-twin/rac/docker-compose-fuzz.yaml

@@ -59,6 +59,8 @@ services:
     environment:
       KAFKA_BOOTSTRAP: localhost:9092
       SQLITE_DB: /app/data/fuzz.db
+      LM_TOPIC: lm-events
+      OLR_TOPIC: olr-events
     volumes:
       - ./kafka-consumer.py:/app/kafka-consumer.py:ro
       - fuzz-data:/app/data
@@ -67,6 +69,7 @@ services:
     image: python:3.12-slim
     container_name: fuzz-validator
     network_mode: host
+    profiles: ["validate"]
     depends_on:
       consumer:
         condition: service_started

tests/dbz-twin/rac/fuzz-test.sh

@@ -286,43 +286,47 @@ action_status() {
 }
 
 action_validate() {
-    echo "=== Waiting for validator to complete ==="
-    echo "  Validator will idle-timeout after processing all events..."
-    echo ""
-
-    # Wait for validator to complete
-    docker wait fuzz-validator > /dev/null 2>&1 || true
-    local exit_code
-    exit_code=$(docker inspect fuzz-validator --format '{{.State.ExitCode}}' 2>/dev/null || echo "1")
-
-    # Save log
-    local vlog="/tmp/fuzz-validator-$(date +%Y%m%d-%H%M%S).log"
-    docker logs fuzz-validator > "$vlog" 2>&1
-    echo "  Validator log: $vlog"
-    echo ""
-
-    # Show summary
-    tail -15 "$vlog"
+    echo "=== Running validation ==="
+
+    # Wait for consumer to catch up (no new events for 30s)
+    echo "  Waiting for consumer to drain..."
+    local prev_count=0 idle_count=0
+    while true; do
+        local cur_count
+        cur_count=$(docker logs --tail 1 fuzz-consumer 2>/dev/null | grep -oP 'LM=\K[0-9]+' || echo "0")
+        if [[ "$cur_count" == "$prev_count" ]]; then
+            idle_count=$(( idle_count + 1 ))
+            [[ $idle_count -ge 6 ]] && break  # 30s idle
+        else
+            idle_count=0
+            prev_count=$cur_count
+        fi
+        sleep 5
+    done
+    echo "  Consumer drained (LM events: $cur_count)"
 
+    # Start validator (uses 'validate' profile)
+    docker compose -f "$COMPOSE_FILE" run --rm validator
+    local exit_code=$?
     echo ""
     echo "  OLR memory: $(_olr_memory_mb) MB"
 
     # OLR errors
     local olr_errors
     olr_errors=$(ssh $_SSH_OPTS "${VM_USER}@${VM_HOST}" \
-        "podman logs $OLR_CONTAINER 2>&1 | grep -c 'ERROR\|ASAN\|AddressSanitizer'" 2>/dev/null || echo "0")
+        "podman logs $OLR_CONTAINER 2>&1 | grep -c 'ERROR\|ASAN\|AddressSanitizer'" 2>/dev/null | tr -d '[:space:]' || echo "0")
+    [[ -z "$olr_errors" ]] && olr_errors=0
     if [[ "$olr_errors" -gt 0 ]]; then
         echo "  WARNING: $olr_errors OLR errors detected"
         ssh $_SSH_OPTS "${VM_USER}@${VM_HOST}" \
             "podman logs $OLR_CONTAINER 2>&1 | grep 'ERROR\|ASAN' | tail -5" 2>/dev/null
     fi
 
     echo ""
-    if [[ "$exit_code" == "0" ]]; then
+    if [[ "$exit_code" -eq 0 ]]; then
         echo "=== PASS: Fuzz test completed ==="
     else
         echo "=== FAIL: Fuzz test found mismatches ==="
-        echo "  Investigate: ./fuzz-test.sh logs validator"
     fi
 
     return "$exit_code"

tests/dbz-twin/rac/validator.py

@@ -138,42 +138,48 @@ def main():
                 prev_lm_count = lm_count
                 prev_olr_count = olr_count
 
-            # Find safe frontier: min of max event_id on each side.
-            # With single-topic Kafka delivery, events arrive in commit order
-            # within each adapter, so max(event_id) is a reliable watermark.
-            lm_max = conn.execute(
-                "SELECT MAX(event_id) FROM lm_events").fetchone()[0]
-            olr_max = conn.execute(
-                "SELECT MAX(event_id) FROM olr_events").fetchone()[0]
-
-            if lm_max is None or olr_max is None:
+            # Find safe frontier per node: min(lm, olr) for each N{x} prefix.
+            # Event_ids from two RAC nodes interleave non-monotonically in
+            # commit order, so a global frontier would validate events before
+            # the other side has delivered them.
+            node_frontiers = {}
+            for node_prefix in ('N1', 'N2'):
+                lm_node_max = conn.execute(
+                    "SELECT MAX(event_id) FROM lm_events WHERE event_id LIKE ?",
+                    (f'{node_prefix}_%',)).fetchone()[0]
+                olr_node_max = conn.execute(
+                    "SELECT MAX(event_id) FROM olr_events WHERE event_id LIKE ?",
+                    (f'{node_prefix}_%',)).fetchone()[0]
+                if lm_node_max and olr_node_max:
+                    node_frontiers[node_prefix] = min(lm_node_max, olr_node_max)
+
+            if not node_frontiers:
                 continue
 
-            frontier = min(lm_max, olr_max)
+            frontier = max(node_frontiers.values())
             if frontier <= cursor_event_id:
-                # Check idle timeout
                 if time.time() - last_new_events > IDLE_TIMEOUT:
                     print(f"[validator] Idle timeout ({IDLE_TIMEOUT}s). "
                           f"Final validation pass...", flush=True)
-                    frontier = min(lm_max, olr_max)
                     if frontier <= cursor_event_id:
                         break
                 else:
                     continue
 
-            # Fetch distinct event_ids in range from both sides
+            # Fetch event_ids within each node's safe frontier
             lm_ids = set()
             olr_ids = set()
-            for r in conn.execute(
-                "SELECT DISTINCT event_id FROM lm_events "
-                "WHERE event_id > ? AND event_id <= ? ORDER BY event_id",
-                (cursor_event_id, frontier)).fetchall():
-                lm_ids.add(r['event_id'])
-            for r in conn.execute(
-                "SELECT DISTINCT event_id FROM olr_events "
-                "WHERE event_id > ? AND event_id <= ? ORDER BY event_id",
-                (cursor_event_id, frontier)).fetchall():
-                olr_ids.add(r['event_id'])
+            for node_prefix, nf in node_frontiers.items():
+                for r in conn.execute(
+                    "SELECT DISTINCT event_id FROM lm_events "
+                    "WHERE event_id > ? AND event_id <= ? AND event_id LIKE ?",
+                    (cursor_event_id, nf, f'{node_prefix}_%')).fetchall():
+                    lm_ids.add(r['event_id'])
+                for r in conn.execute(
+                    "SELECT DISTINCT event_id FROM olr_events "
+                    "WHERE event_id > ? AND event_id <= ? AND event_id LIKE ?",
+                    (cursor_event_id, nf, f'{node_prefix}_%')).fetchall():
+                    olr_ids.add(r['event_id'])
 
             all_ids = sorted(lm_ids | olr_ids)