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/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. A DLQ consumer must branch on the presence of `"action"` to distinguish the two shapes — Phase 2 (issue #91) may normalize this.
 - **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,42 @@ 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. Phase 2 will classify
+				// transient (network/timeout) vs permanent errors and Nak the
+				// transient ones.
+				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 note: insert failures route through dlqOutput.WriteBatch
+				// and publish `raw.Payload` (the inner data field only). This path
+				// publishes the full NATS envelope (`{"action":"delete","table_name":...,"id":...}`)
+				// so a DLQ consumer can re-issue the delete with full context. A consumer
+				// of `dlq.<table>` must branch on the `"action"` field to distinguish the
+				// two shapes; Phase 2 (issue #91) may normalize this.
+				dlqSubject := "dlq." + raw.TableName
+				if _, pubErr := j.js.Publish(spanCtx, dlqSubject, m.Data()); 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 +510,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

@@ -328,14 +328,58 @@ 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.Error(t, err)
-	assert.Contains(t, err.Error(), "execute delete: db error")
-	assert.Nil(t, msg)
+	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")
+
+	// 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}
 
-	// On exec error, message should be Nak'd for retry.
-	assert.True(t, delMsg.naked, "delete message should be nak'd on exec error")
+	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.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")
+
+	table, _ := msg.MetaGet("table_name")
+	assert.Equal(t, "events", table)
 }
 
 func TestJsInput_Read_DeleteDrainSuccess(t *testing.T) {