sink/clickhouse: detect silent-write-drop from CH summary

Live evidence from the deployed operator (pod
adaptive-export-6cfb95b7-pbmxg, log line 2026-05-23T20:58:39Z):

  time="2026-05-23T20:58:39Z" level=info
    msg="sink: pixie write completed"
    body_bytes=2098817
    rows_sent=1658
    table=redis_events
    ch_summary="{
      \"read_rows\":\"0\",\"read_bytes\":\"0\",
      \"written_rows\":\"0\",\"written_bytes\":\"0\",
      \"total_rows_to_read\":\"0\",\"result_rows\":\"0\",
      \"result_bytes\":\"0\",\"elapsed_ns\":\"23034181\"
    }"

The operator sent 1658 rows (~2 MiB), ClickHouse returned 2xx, but
its X-ClickHouse-Summary response header reports written_rows=0 —
i.e. CH silently dropped every row. The current WritePixieRows
treats the 2xx as durable success and logs "pixie write completed";
the analyst sees the gap days later.

Fix: parse X-ClickHouse-Summary and return an error when
written_rows < rows_sent. Header absence is tolerated (older CH
versions / proxies strip it); only an EXPLICIT zero-of-non-zero
(or partial-of-non-zero) is treated as data loss. The error message
includes both rows_sent and the full summary so the next iteration
can correlate the dropped batch with a request id.

TDD:
- TestWritePixieRows_DetectsSilentZeroWriteDrop  — replays the live
  summary header verbatim; failed before the fix, green after.
- TestWritePixieRows_DetectsPartialWriteDrop     — same class, 100
  of 200 written.
- TestWritePixieRows_HappyPath                   — pins the contract:
  written_rows>=rows_sent → nil error.
- TestWritePixieRows_NoSummaryHeaderIsTolerated  — older CH won't
  send the header; no false positive.
- TestWritePixieRows_EmptyBatchShortCircuits     — len(rows)==0 path
  never hits HTTP.

Doesn't address the root cause (why CH dropped those rows — likely
a per-batch type-mismatch silently swallowed by JSONEachRow under
input_format_skip_unknown_fields or a parse-error 2xx); that needs
a separate investigation with the request id from the next caught
event. The detection here at least stops us from reporting durable
writes that never happened.

Author: entlein

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

@@ -211,9 +211,52 @@ func (s *ClickHouseHTTP) WritePixieRows(ctx context.Context, table string, rows
 		"ch_summary":     summary,
 		"first_row_keys": strings.Join(firstRowKeys, ","),
 	}).Info("sink: pixie write completed")
+	// Detect the silent-drop class: CH returns 2xx but
+	// X-ClickHouse-Summary.written_rows < len(rows). Observed live on
+	// 2026-05-23T20:58Z (redis_events: rows_sent=1658, written_rows=0)
+	// — the operator reported success and the analyst saw the gap days
+	// later. Header absence is tolerated (older CH versions / proxies
+	// strip it); only an EXPLICIT zero-of-non-zero counts.
+	if writeMismatch := summaryWroteFewerThan(summary, len(rows)); writeMismatch != nil {
+		return fmt.Errorf("sink: pixie write to %s reported %d rows_sent but CH summary written_rows=%d (silent drop): %s",
+			table, len(rows), writeMismatch.writtenRows, summary)
+	}
 	return nil
 }
 
+// summaryDelta carries the parsed write counters from CH's
+// X-ClickHouse-Summary response header.
+type summaryDelta struct {
+	writtenRows int64
+}
+
+// summaryWroteFewerThan returns non-nil when the X-ClickHouse-Summary
+// header is present, parseable, and reports written_rows < rowsSent.
+// Returns nil when the header is missing, unparseable, or the count
+// matches/exceeds rowsSent — those are not data-loss signals.
+func summaryWroteFewerThan(summary string, rowsSent int) *summaryDelta {
+	if summary == "" {
+		return nil
+	}
+	var parsed struct {
+		WrittenRows json.Number `json:"written_rows"`
+	}
+	if err := json.Unmarshal([]byte(summary), &parsed); err != nil {
+		return nil
+	}
+	if parsed.WrittenRows == "" {
+		return nil
+	}
+	wrote, err := parsed.WrittenRows.Int64()
+	if err != nil {
+		return nil
+	}
+	if wrote >= int64(rowsSent) {
+		return nil
+	}
+	return &summaryDelta{writtenRows: wrote}
+}
+
 // normalisePixieValue coerces pxapi-emitted Go values into JSON-friendly
 // shapes ClickHouse parses cleanly. time.Time → "YYYY-MM-DD HH:MM:SS.NNN…"
 // (CH's DateTime64 input format); []byte → string; everything else → as-is.

src/vizier/services/adaptive_export/internal/sink/clickhouse_test.go

@@ -451,3 +451,138 @@ func TestParseActiveRows_RoundTripFromBytes(t *testing.T) {
 		t.Fatalf("round-trip mismatch: %+v", rows)
 	}
 }
+
+// pixieRow returns a minimal-but-valid map shaped like a pxapi row.
+func pixieRow() map[string]any {
+	return map[string]any{
+		"time_":       time.Unix(0, 1700000000000000000).UTC(),
+		"upid":        "1234:5678:9",
+		"namespace":   "redis",
+		"pod":         "redis/redis-1",
+		"req_cmd":     "GET",
+		"resp":        "OK",
+		"latency":     int64(123456),
+		"remote_addr": "10.0.0.1",
+		"remote_port": int64(6379),
+		"local_addr":  "10.0.0.2",
+		"local_port":  int64(34567),
+		"trace_role":  int64(2),
+		"encrypted":   false,
+		"px_info_":    "",
+		"req_args":    "",
+	}
+}
+
+// TestWritePixieRows_HappyPath — happy path: CH returns 200 with a
+// non-zero `written_rows` in X-ClickHouse-Summary; WritePixieRows
+// returns nil. Pins the contract the regression test below inverts.
+func TestWritePixieRows_HappyPath(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("X-ClickHouse-Summary",
+			`{"read_rows":"1","read_bytes":"100","written_rows":"1","written_bytes":"100",`+
+				`"total_rows_to_read":"0","result_rows":"1","result_bytes":"100","elapsed_ns":"1000000"}`)
+		w.WriteHeader(200)
+	}))
+	defer srv.Close()
+	s, err := New(Config{Endpoint: srv.URL})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	if err := s.WritePixieRows(context.Background(), "redis_events", []map[string]any{pixieRow()}); err != nil {
+		t.Fatalf("WritePixieRows: %v", err)
+	}
+}
+
+// TestWritePixieRows_DetectsSilentZeroWriteDrop — regression for the
+// silent-data-loss bug observed on the live operator:
+//
+//	sink: pixie write completed
+//	    rows_sent=1658
+//	    body_bytes=2098817
+//	    ch_summary="{...,"written_rows":"0",...}"
+//	    table=redis_events
+//
+// CH returned 2xx but `X-ClickHouse-Summary.written_rows` was zero
+// for a 1658-row payload — i.e. CH silently dropped every row. The
+// operator must NOT report success in that case; otherwise the
+// caller treats the batch as durably persisted and we lose data.
+func TestWritePixieRows_DetectsSilentZeroWriteDrop(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Real CH summary header from the operator-pod log on
+		// 2026-05-23T20:58:39Z, table=redis_events.
+		w.Header().Set("X-ClickHouse-Summary",
+			`{"read_rows":"0","read_bytes":"0","written_rows":"0","written_bytes":"0",`+
+				`"total_rows_to_read":"0","result_rows":"0","result_bytes":"0","elapsed_ns":"23034181"}`)
+		w.WriteHeader(200)
+	}))
+	defer srv.Close()
+	s, err := New(Config{Endpoint: srv.URL})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	// Send a real (non-zero) batch — a zero-input batch short-circuits
+	// before the HTTP call so the assertion would never fire.
+	batch := make([]map[string]any, 1658)
+	for i := range batch {
+		batch[i] = pixieRow()
+	}
+	err = s.WritePixieRows(context.Background(), "redis_events", batch)
+	if err == nil {
+		t.Fatalf("expected error from silent-drop (rows_sent=%d, written_rows=0), got nil", len(batch))
+	}
+	if !strings.Contains(err.Error(), "0") || !strings.Contains(err.Error(), "1658") {
+		t.Fatalf("error should mention both written_rows=0 and rows_sent=1658 for diagnosis; got: %v", err)
+	}
+}
+
+// TestWritePixieRows_DetectsPartialWriteDrop — CH wrote SOME rows
+// but not all. Same data-loss class as the zero-write case; reject.
+func TestWritePixieRows_DetectsPartialWriteDrop(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("X-ClickHouse-Summary",
+			`{"read_rows":"100","read_bytes":"10000","written_rows":"100","written_bytes":"10000",`+
+				`"total_rows_to_read":"0","result_rows":"100","result_bytes":"10000","elapsed_ns":"1000000"}`)
+		w.WriteHeader(200)
+	}))
+	defer srv.Close()
+	s, _ := New(Config{Endpoint: srv.URL})
+	batch := make([]map[string]any, 200) // sent 200, CH says wrote 100
+	for i := range batch {
+		batch[i] = pixieRow()
+	}
+	err := s.WritePixieRows(context.Background(), "redis_events", batch)
+	if err == nil {
+		t.Fatalf("expected error on partial write (sent=200, written=100); got nil")
+	}
+}
+
+// TestWritePixieRows_NoSummaryHeaderIsTolerated — older CH versions
+// (or proxies) may strip the X-ClickHouse-Summary header. Absence is
+// NOT a failure signal — only an explicit zero-of-non-zero is.
+func TestWritePixieRows_NoSummaryHeaderIsTolerated(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(200) // no summary header at all
+	}))
+	defer srv.Close()
+	s, _ := New(Config{Endpoint: srv.URL})
+	if err := s.WritePixieRows(context.Background(), "redis_events", []map[string]any{pixieRow()}); err != nil {
+		t.Fatalf("missing summary header must not error; got: %v", err)
+	}
+}
+
+// TestWritePixieRows_EmptyBatchShortCircuits — zero-row input never
+// hits HTTP and never produces a "silent drop" false positive.
+func TestWritePixieRows_EmptyBatchShortCircuits(t *testing.T) {
+	called := false
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		called = true
+	}))
+	defer srv.Close()
+	s, _ := New(Config{Endpoint: srv.URL})
+	if err := s.WritePixieRows(context.Background(), "redis_events", nil); err != nil {
+		t.Fatalf("empty WritePixieRows: %v", err)
+	}
+	if called {
+		t.Fatalf("empty batch made an HTTP call")
+	}
+}