fix(ingest): break infinite-Nak loop on permanent delete errors

AGENTS.md

@@ -19,7 +19,7 @@ Twelve internal packages under `internal/`:
 - **`config/`** — YAML + env var config loading (cleanenv)
 - **`dedupe/`** — `Deduplicator` interface → `Embedded` (Pebble) — optional, controlled by `dedupe.enabled`
 - **`discovery/`** — `SchemaRegistry` that introspects ClickHouse `system.columns` + `Validate()` for ingest payloads
-- **`ingest/`** — Bento-based ingest pipeline (`bento.go`: JetStream input → per-table batch INSERT with DLQ output, plus delete handling) + `Sweeper` (Active Sweeper for NATS message lifecycle) + `EventMessage`/`BufferConsumerName` types (`types.go`)
+- **`ingest/`** — Bento-based ingest pipeline (`bento.go`: JetStream input → per-table batch INSERT with DLQ output, plus inline delete handling — failed deletes route to `dlq.<table>` and `DoubleAck` rather than `Nak`, to break the infinite-redelivery loop that a deterministic delete error would otherwise produce) + `Sweeper` (Active Sweeper for NATS message lifecycle) + `EventMessage`/`BufferConsumerName` types (`types.go`)
 - **`mq/`** — `Publisher`/`Subscriber` interfaces → `EmbeddedNATS` + `RemoteNATS`
 - **`observability/`** — OpenTelemetry pipeline: `InitProvider` wires trace/metric/log providers via OTLP gRPC (each signal independently gated). A top-level `Prometheus` config block drives an optional `/metrics` scrape endpoint that runs independently of OTLP push — standalone (Alloy/Mimir scrape, no collector), alongside OTLP, or off. `NewLogger` produces a slog handler that fans out to stdout AND OTLP (stdout always 100%, OTLP sample-rate-aware). `TraceHandler` injects trace_id/span_id from active spans. `tracer.go` provides W3C trace context propagation over NATS headers.
 - **`pipes/`** — Named query pipes: `NamedQuery` type + NATS KV store (`WAVEHOUSE_PIPES`) + `.sql` file bootstrap

CHANGELOG.md

@@ -6,6 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
+
+### Fixed
+- **Ingest worker no longer infinite-retries permanent delete errors** (`internal/ingest/bento.go`, `internal/ingest/bento_test.go`, `docs/src/content/docs/architecture.md`, `AGENTS.md`): `jsInput.Read`'s `action: "delete"` block called `m.Nak()` on every `chConn.Exec` failure, which JetStream interprets as "redeliver immediately." A delete whose error was *deterministic* (syntax error, unknown table, malformed identifier) would loop forever — clogging the buffer consumer, burning CPU, and spamming logs with the same message. Phase 1 of issue #91: every delete-Exec error is now treated as permanent. The original NATS envelope is published to `dlq.<table>` (reusing the existing `bentoDLQDropped` counter when the DLQ publish itself fails) and the message is `DoubleAck`'d so it leaves the main queue. Issue #91 stays open after this lands as the Phase 2 tracker for transient-vs-permanent error classification (timeouts and network errors should still `Nak()` for retry); Phase 1 alone is the stopgap, Phase 2 makes the trade-off acceptable in production.
+
 ### Removed
 - **`project-orchestrator.yml` workflow + its three composite-action artifacts** (`.github/workflows/project-orchestrator.yml`, `.github/actions/board-upsert-status/`, `.github/actions/set-linked-issues-status/`, `.github/scripts/board-fetch-item.sh`, `AGENTS.md`, `CHANGELOG.md`): −887 lines net. The orchestrator was the largest single source of cross-trigger complexity on this repo (3-4 workflow_run-chained runs per PR push, `statusCheckRollup` GraphQL perms quirks, integration-token `NONE` for private-org members) for behaviour that is mostly either provided natively by GitHub or a one-click manual operation on a 4-person team. Replaced by: reviewer-assign step in `housekeeping.yml` that fires once on `pull_request_target: opened` / `ready_for_review` (not per-synchronize, so it doesn't re-spam after `dismiss_stale_reviews_on_push`), plus GitHub's native Projects v2 workflows (`Auto-add to project`, `Item added`, `Pull request merged`) configured in the project UI. Trade-offs explicit in the PR body: drafts no longer auto-flip on bot-clean, `CHANGES_REQUESTED` doesn't auto-move the board card, linked-issue card mirroring is dropped. AGENTS.md §"Governance Files" + §"Task Board state machine" + §"Review tooling reference" all rewritten to match. `dependabot-automerge.yml` trimmed in parallel: no more board-upsert step (native handles placement), `PROJECT_BOARD_TOKEN` guard removed (no longer used in this workflow), reviewer list sourced from `board-config.env`'s `ADMINS` via `replace()`, major-bump comment uses the marker-comment upsert pattern from `housekeeping.yml`.
 - **`STATUS_*` and old `ADMINS` consumers in `board-config.env`** — STATUS option IDs had only orchestrator-side consumers and are now unreferenced. `ADMINS` was restored to `board-config.env` after the initial orchestrator-removal commit dropped it (Gemini and Claude both flagged the resulting drift across three inlined copies); both `housekeeping.yml` and `dependabot-automerge.yml` now load `ADMINS` from `board-config.env`. `admin-approval.yml` keeps its own inline copy with the documented latency-avoidance reasoning.

docs/src/content/docs/architecture.md

@@ -105,7 +105,7 @@ The API layer uses [Chi](https://github.com/go-chi/chi) for routing with standar
 
 ### `ingest/` — Bento Pipeline, DLQ & Sweeping
 
-- **bento.go** — `StartIngestWorker` launches a Bento-based ingest pipeline: a JetStream input (`jsInput`) reads from the `WAVEHOUSE` stream via a durable `buffer-consumer` pull consumer, batches events per table, and performs bulk INSERTs to ClickHouse. Delete actions (`action: "delete"`) are handled inline via `chConn.Exec`. Failed batches are routed to a DLQ output (`dlqOutput`) which publishes to `dlq.{table}` NATS subjects when DLQ is enabled. Unsafe table names are rejected via `safeIdentifierRe`.
+- **bento.go** — `StartIngestWorker` launches a Bento-based ingest pipeline: a JetStream input (`jsInput`) reads from the `WAVEHOUSE` stream via a durable `buffer-consumer` pull consumer, batches events per table, and performs bulk INSERTs to ClickHouse. Delete actions (`action: "delete"`) are handled inline via `chConn.Exec`; a failed delete is treated as permanent — the original envelope is routed to `dlq.{table}` and the message is `DoubleAck`'d to break the infinite-Nak retry loop a bad query would otherwise produce. Failed batches are routed to a DLQ output (`dlqOutput`) which publishes to `dlq.{table}` NATS subjects when DLQ is enabled. Unsafe table names are rejected via `safeIdentifierRe`. **DLQ shape note**: insert-failure messages on `dlq.{table}` carry the inner data payload only (`{"id":"abc","field":...}`) because `dlqOutput.WriteBatch` publishes the Bento message bytes (already-extracted `data`); delete-failure messages carry the full NATS envelope (`{"action":"delete","table_name":"...","id":"..."}`) so a consumer can re-issue the delete with full context. To stay BYOS-safe — a user row can have any column name, including `action` — consumers MUST discriminate via the `Wave-DLQ-Type` NATS header (set by the worker, non-user-controlled), not via any JSON field: `Wave-DLQ-Type: delete-envelope` → delete-failure shape; header absent → insert-payload shape. Delete-failure messages also carry W3C `traceparent`/`tracestate` headers via `observability.InjectNATS` so a consumer can stitch its trace to the worker's `clickhouse_delete` span. Phase 2 (issue #91) will tag insert-failure messages with `Wave-DLQ-Type: insert-payload` and may normalize the payload shapes.
 - **types.go** — `EventMessage` struct (TableName, ReceivedTimestamp, Data) and `BufferConsumerName` constant, shared across API handlers and the ingest pipeline.
 - **sweeper.go** — `Sweeper` implements the Active Sweeper pattern. It runs every minute and purges NATS JetStream messages that are **both** ACKed by the buffer consumer (written to ClickHouse) **and** older than the configurable gap window.
 
@@ -156,7 +156,9 @@ Bento ingest pipeline (StartIngestWorker):
   → Validate table name against safeIdentifierRe
   → Batch events per table, bulk INSERT to ClickHouse
   → Delete actions handled inline (chConn.Exec)
-  → On success: Ack messages
+    → On success: DoubleAck and continue
+    → On failure: route original envelope to dlq.{table} and DoubleAck (Phase 1: all delete errors treated as permanent to prevent infinite Nak loops)
+  → On success: DoubleAck messages
   → On failure: route to DLQ output (dlq.{table}), then Ack to prevent infinite retry
 
 Active Sweeper (async goroutine, every 60s):

internal/ingest/bento.go

@@ -56,6 +56,7 @@ type jsInput struct {
 	consumer jetstream.Consumer
 	iter     jetstream.MessagesContext
 	chConn   driver.Conn
+	js       jetstream.JetStream
 	inFlight atomic.Int32
 }
 
@@ -172,18 +173,58 @@ func (j *jsInput) Read(ctx context.Context) (*service.Message, service.AckFunc,
 
 			if err := j.chConn.Exec(spanCtx, delQuery, raw.ID); err != nil {
 				span.RecordError(err)
-				slog.ErrorContext(spanCtx, "clickhouse delete failed",
+				// Phase 1 (issue #91): treat every delete Exec error as permanent
+				// to break the infinite-Nak retry loop that a bad query (syntax
+				// error, missing table, etc.) would otherwise produce. Route the
+				// original NATS envelope to dlq.<table> and DoubleAck so the
+				// message is removed from the main queue.
+				//
+				// TODO(#91, Phase 2): classify transient (network/timeout) vs
+				// permanent errors and Nak the transient ones so they get retried.
+				// Until that lands, a ClickHouse blip during a delete drops the
+				// message to DLQ instead of retrying — accepted trade-off, see #91.
+				slog.ErrorContext(spanCtx, "clickhouse delete failed — routing to DLQ",
 					"table", raw.TableName,
 					"id", raw.ID,
 					"error", err,
 				)
 
-				if nakErr := m.Nak(); nakErr != nil {
-					slog.WarnContext(spanCtx, "nak failed after delete error", "error", nakErr)
+				// DLQ shape contract: `dlq.<table>` carries two structurally different
+				// payloads — insert-failure inner-data via dlqOutput.WriteBatch, and
+				// delete-failure full envelopes via this branch
+				// (`{"action":"delete","table_name":...,"id":...}`). Consumers must NOT
+				// branch on a JSON field to tell them apart: under BYOS a user row can
+				// carry any key, including `action`. We instead set a non-user-controlled
+				// NATS header here (`Wave-DLQ-Type: delete-envelope`) so consumers can
+				// detect the shape via header presence. The insert path has no header
+				// today; absence means insert-payload. Phase 2 (issue #91) will add
+				// `Wave-DLQ-Type: insert-payload` on the insert path and may normalize
+				// the shapes. observability.InjectNATS also propagates the
+				// `clickhouse_delete` span into the message so a Phase 2 consumer can
+				// stitch its own trace to the worker's.
+				dlqSubject := "dlq." + raw.TableName
+				dlqMsg := &nats.Msg{
+					Subject: dlqSubject,
+					Data:    m.Data(),
+					Header:  nats.Header{"Wave-DLQ-Type": []string{"delete-envelope"}},
+				}
+				observability.InjectNATS(spanCtx, dlqMsg)
+				if _, pubErr := j.js.PublishMsg(spanCtx, dlqMsg); pubErr != nil {
+					slog.ErrorContext(spanCtx, "DLQ publish failed for permanent delete error — message dropped",
+						"subject", dlqSubject,
+						"error", pubErr,
+					)
+					bentoDLQDropped.Add(spanCtx, 1, metric.WithAttributes(attribute.String("table", raw.TableName)))
+				} else {
+					slog.WarnContext(spanCtx, "sent failed delete to DLQ", "subject", dlqSubject)
+				}
+
+				if doubleAckErr := m.DoubleAck(spanCtx); doubleAckErr != nil {
+					slog.WarnContext(spanCtx, "double ack failed after permanent delete error", "error", doubleAckErr)
 				}
 
 				span.End()
-				return nil, nil, fmt.Errorf("execute delete: %w", err)
+				continue
 			}
 			// Log the success with all context
 			slog.InfoContext(spanCtx, "successfully deleted record",
@@ -485,6 +526,7 @@ func StartIngestWorker(ctx context.Context, nc *nats.Conn, chConn driver.Conn, c
 				return &jsInput{
 					consumer: cons,
 					chConn:   chConn,
+					js:       js,
 				}, nil
 			},
 		); err != nil {

internal/ingest/bento_test.go

@@ -24,7 +24,11 @@ import (
 // Different names from sweeper_test.go mocks to avoid redeclaration.
 // ---------------------------------------------------------------------------
 
-// bentoMockJS satisfies jetstream.JetStream; only Publish is overridden.
+// bentoMockJS satisfies jetstream.JetStream; Publish and PublishMsg are overridden.
+// Publish is used by dlqOutput.WriteBatch (insert-failure path).
+// PublishMsg is used by jsInput.Read's delete-failure branch, which needs to set
+// a non-user-controlled `Wave-DLQ-Type` header to discriminate DLQ payload shapes
+// under BYOS (see internal/ingest/bento.go).
 type bentoMockJS struct {
 	jetstream.JetStream
 	published []publishedMsg
@@ -33,13 +37,19 @@ type bentoMockJS struct {
 type publishedMsg struct {
 	Subject string
 	Data    []byte
+	Header  nats.Header
 }
 
 func (m *bentoMockJS) Publish(_ context.Context, subject string, payload []byte, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) {
 	m.published = append(m.published, publishedMsg{Subject: subject, Data: payload})
 	return &jetstream.PubAck{}, nil
 }
 
+func (m *bentoMockJS) PublishMsg(_ context.Context, msg *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) {
+	m.published = append(m.published, publishedMsg{Subject: msg.Subject, Data: msg.Data, Header: msg.Header})
+	return &jetstream.PubAck{}, nil
+}
+
 // bentoMockMsg satisfies jetstream.Msg for testing jsInput.Read.
 type bentoMockMsg struct {
 	jetstream.Msg
@@ -328,14 +338,76 @@ func TestJsInput_Read_DeleteExecError(t *testing.T) {
 	iter.msgs <- delMsg
 	iter.msgs <- insertMsg
 
-	input := &jsInput{iter: iter, chConn: conn}
+	js := &bentoMockJS{}
+	input := &jsInput{iter: iter, chConn: conn, js: js}
+	msg, _, err := input.Read(context.Background())
+	require.NoError(t, err, "delete error should be handled as permanent (no error returned)")
+	require.NotNil(t, msg, "Read should advance to the next message after a failed delete")
+
+	// Permanent-fail policy: original envelope routed to dlq.<table>, message DoubleAck'd, not Nak'd.
+	assert.False(t, delMsg.naked, "delete message must not be Nak'd (would cause infinite retry)")
+	assert.True(t, delMsg.doubleAcked, "delete message should be DoubleAck'd after routing to DLQ")
+
+	require.Len(t, js.published, 1, "failed delete should be published to DLQ exactly once")
+	assert.Equal(t, "dlq.clicks", js.published[0].Subject)
+	assert.JSONEq(t, string(delData), string(js.published[0].Data), "DLQ payload should preserve the original delete envelope")
+	// BYOS-safe shape discriminator: a `Wave-DLQ-Type: delete-envelope` NATS
+	// header tells consumers this is a full-envelope payload, not an insert
+	// row. Header is non-user-controlled — a user column named `action` in an
+	// insert row can't forge it.
+	assert.Equal(t, "delete-envelope", js.published[0].Header.Get("Wave-DLQ-Type"),
+		"delete failures must be tagged with Wave-DLQ-Type header so DLQ consumers can distinguish them from insert-failure payloads under BYOS")
+
+	// Next message in queue should be the follow-up insert.
+	table, _ := msg.MetaGet("table_name")
+	assert.Equal(t, "events", table)
+}
+
+func TestJsInput_Read_DeleteExecErrorDLQPublishFails(t *testing.T) {
+	t.Parallel()
+	// Even when the DLQ publish itself fails (NATS unavailable), the worker must
+	// still DoubleAck the message — Nak'ing here would re-create the infinite
+	// retry loop that the permanent-fail policy is designed to prevent.
+	del := map[string]any{"action": "delete", "table_name": "clicks", "id": "fail"}
+	delData, _ := json.Marshal(del)
+	delMsg := &bentoMockMsg{data: delData}
+
+	good := EventMessage{TableName: "events", Data: map[string]any{"id": "1"}}
+	goodData, _ := json.Marshal(good)
+	goodMsg := &bentoMockMsg{data: goodData}
+
+	conn := &bentoMockConn{
+		execFn: func(context.Context, string, ...any) error { return errors.New("syntax error") },
+	}
+
+	iter := &bentoMockIter{msgs: make(chan jetstream.Msg, 2)}
+	iter.msgs <- delMsg
+	iter.msgs <- goodMsg
+
+	js := &bentoMockJSWithError{publishErr: errors.New("NATS unavailable")}
+	input := &jsInput{iter: iter, chConn: conn, js: js}
+
 	msg, _, err := input.Read(context.Background())
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "execute delete: db error")
-	assert.Nil(t, msg)
+	require.NoError(t, err)
+	require.NotNil(t, msg)
+
+	assert.False(t, delMsg.naked, "delete must not be Nak'd even when DLQ publish fails")
+	assert.True(t, delMsg.doubleAcked, "delete must still be DoubleAck'd to avoid infinite redelivery")
+
+	// Prove the DLQ publish was actually attempted before we DoubleAck'd. A
+	// regression that skipped PublishMsg entirely (jumping straight to
+	// DoubleAck) would still satisfy the no-Nak/DoubleAck assertions above,
+	// but it would silently drop messages on every transient NATS hiccup
+	// instead of giving the operator a single DLQ-dropped error to grep on.
+	require.Len(t, js.published, 1, "DLQ publish must still be attempted once even when NATS is unavailable")
+	assert.Equal(t, "dlq.clicks", js.published[0].Subject)
+	assert.JSONEq(t, string(delData), string(js.published[0].Data),
+		"the attempted DLQ payload must still be the original delete envelope")
+	assert.Equal(t, "delete-envelope", js.published[0].Header.Get("Wave-DLQ-Type"),
+		"discriminator header must be set on the attempted publish, not only on successful ones")
 
-	// On exec error, message should be Nak'd for retry.
-	assert.True(t, delMsg.naked, "delete message should be nak'd on exec error")
+	table, _ := msg.MetaGet("table_name")
+	assert.Equal(t, "events", table)
 }
 
 func TestJsInput_Read_DeleteDrainSuccess(t *testing.T) {
@@ -580,12 +652,20 @@ func TestJsInput_Read_DeleteCancelledCtx(t *testing.T) {
 type bentoMockJSWithError struct {
 	jetstream.JetStream
 	publishErr error
+	published  []publishedMsg
 }
 
 func (m *bentoMockJSWithError) Publish(_ context.Context, _ string, _ []byte, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) {
 	return nil, m.publishErr
 }
 
+// Capture the published message before returning the error so tests can
+// distinguish "DLQ publish failed" from "DLQ publish never attempted".
+func (m *bentoMockJSWithError) PublishMsg(_ context.Context, msg *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) {
+	m.published = append(m.published, publishedMsg{Subject: msg.Subject, Data: msg.Data, Header: msg.Header})
+	return nil, m.publishErr
+}
+
 func TestDLQOutput_WriteBatch_PublishError(t *testing.T) {
 	t.Parallel()
 	js := &bentoMockJSWithError{publishErr: errors.New("NATS unavailable")}