fix(ae-trigger): escape a PollLimit-saturated watermark boundary (from #67)

ae-prod's boundary handling accumulates the seen-fingerprint set on a no-progress
tick, but if >PollLimit rows share one normalized event_time the SQL
(>= watermark ORDER BY time LIMIT N, no secondary key) returns the same N boundary
rows every poll → rows beyond N at that timestamp are never emitted (infinite
boundary). Detect all-skipped-at-capacity and advance the watermark by 1ns to make
forward progress (fingerprint dedup already tolerates the 1ns overlap).

Cherry-picked the trigger fix + its test from the stale CodeRabbit-chat PR #67
(687851d); dropped that PR's unrelated gen-pod.tmpl.yaml churn. #67 itself is
NOT mergeable (77 commits behind ae-prod, re-adds deleted terraform).

Author: pixie-agent

src/vizier/services/adaptive_export/internal/trigger/clickhouse.go

@@ -273,13 +273,15 @@ func (t *ClickHouseHTTP) run(ctx context.Context, out chan<- kubescape.Event) {
 		// with the time-based throttle inside flushWatermark, this
 		// produces at most one persistent INSERT per WatermarkSaveInterval.
 		const saveEveryN = 256
+		skippedAtBoundary := 0
 		for i, row := range rows {
 			fp := rowFingerprint(row)
 			// Cursor comparisons are in NORMALIZED nanos (F8): the raw
 			// event_time unit is not enforced, so compare on the same scale
 			// as the SQL filter (chNormEventTimeNanos) and maxSeen.
 			evn := normalizeEventTimeNanos(row.EventTime)
 			if evn == watermark && seenAtBoundary[fp] {
+				skippedAtBoundary++
 				continue // already pushed in a prior poll at this exact boundary
 			}
 			ev, err := kubescape.Extract(row)
@@ -314,6 +316,23 @@ func (t *ClickHouseHTTP) run(ctx context.Context, out chan<- kubescape.Event) {
 			for fp := range nextSeen {
 				seenAtBoundary[fp] = true
 			}
+			// Paging escape: if every row returned was a boundary-skip AND
+			// the response was at PollLimit capacity, there may be additional
+			// rows at the same normalized event_time that we will never reach
+			// (the SQL ORDER BY has no secondary key, so LIMIT always returns
+			// the same PollLimit rows from the boundary). Advance the watermark
+			// by 1 nanosecond to escape the boundary. In practice this means
+			// at most one nanosecond's worth of events are not re-delivered on
+			// the next poll, which is acceptable: the fingerprint dedup already
+			// tolerates boundary overlap, and we prefer forward progress over
+			// an infinite loop.
+			if skippedAtBoundary > 0 && len(nextSeen) == 0 && len(rows) >= t.cfg.PollLimit {
+				watermark++
+				seenAtBoundary = map[string]bool{}
+				dirty = true
+				log.WithField("watermark", watermark).
+					Warn("trigger: boundary paging escape — advanced watermark by 1ns to unblock poll")
+			}
 		}
 		// Final flush at end of pollOnce — also throttled.
 		flushWatermark()

src/vizier/services/adaptive_export/internal/trigger/clickhouse_internal_test.go

@@ -22,6 +22,7 @@ import (
 	"net/http/httptest"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 )
@@ -61,9 +62,13 @@ func TestNormalizeEventTimeNanos(t *testing.T) {
 // stops a larger-unit row from poisoning the watermark (F8). It captures the
 // query the trigger sends to ClickHouse.
 func TestFetchSinceFiltersOnNormalizedEventTime(t *testing.T) {
+	var mu sync.Mutex
 	var gotQuery string
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		gotQuery = r.URL.Query().Get("query")
+		q := r.URL.Query().Get("query")
+		mu.Lock()
+		gotQuery = q
+		mu.Unlock()
 		w.WriteHeader(200) // empty body = 0 rows, valid JSONEachRow
 	}))
 	defer srv.Close()
@@ -80,16 +85,20 @@ func TestFetchSinceFiltersOnNormalizedEventTime(t *testing.T) {
 		t.Fatalf("fetchSince: %v", err)
 	}
 
-	if !strings.Contains(gotQuery, chNormEventTimeNanos) {
-		t.Errorf("query does not normalize event_time; want %q in:\n%s", chNormEventTimeNanos, gotQuery)
+	mu.Lock()
+	q := gotQuery
+	mu.Unlock()
+
+	if !strings.Contains(q, chNormEventTimeNanos) {
+		t.Errorf("query does not normalize event_time; want %q in:\n%s", chNormEventTimeNanos, q)
 	}
 	// The >= bound must compare the normalized expression against the nanos
 	// watermark, not the raw column.
 	wantPred := chNormEventTimeNanos + " >= " + strconv.FormatUint(wmNanos, 10)
-	if !strings.Contains(gotQuery, wantPred) {
-		t.Errorf("query filter is not normalized-vs-nanos-watermark; want %q in:\n%s", wantPred, gotQuery)
+	if !strings.Contains(q, wantPred) {
+		t.Errorf("query filter is not normalized-vs-nanos-watermark; want %q in:\n%s", wantPred, q)
 	}
-	if strings.Contains(gotQuery, "event_time >= ") {
-		t.Errorf("query still uses RAW event_time filter (poison-prone):\n%s", gotQuery)
+	if strings.Contains(q, "event_time >= ") {
+		t.Errorf("query still uses RAW event_time filter (poison-prone):\n%s", q)
 	}
 }