From 9ae528fa59b0e462514529c7d64342bd5ba39c28 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Tue, 12 May 2026 18:30:07 -0400 Subject: [PATCH 1/9] fix(ingest): break infinite-Nak loop on permanent delete errors (#91) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of issue #91 — every chConn.Exec failure on an `action: "delete"` message used to trigger m.Nak(), which JetStream interprets as "redeliver immediately." A deterministic error (syntax, unknown table, malformed identifier) would loop forever — clogging the buffer consumer, burning CPU, and spamming logs with the same message. Now the worker treats every delete-Exec error as permanent: it publishes the original NATS envelope to dlq. (reusing the bentoDLQDropped counter when the DLQ publish itself fails) and DoubleAcks so the message leaves the main queue. Phase 2 (transient vs permanent classification — timeouts and network errors should still Nak) remains as follow-up on the issue. Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 3 ++ docs/architecture.md | 4 ++- internal/ingest/bento.go | 28 +++++++++++++++--- internal/ingest/bento_test.go | 56 +++++++++++++++++++++++++++++++---- 4 files changed, 80 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e434e347..3fbe0ec1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,9 @@ 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`): `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.
` (reusing the existing `bentoDLQDropped` counter when the DLQ publish itself fails) and the message is `DoubleAck`'d so it leaves the main queue. Phase 2 (transient vs permanent classification — timeouts and network errors should still Nak) is tracked as follow-up on issue #91. Closes #91. + ### 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. diff --git a/docs/architecture.md b/docs/architecture.md index 9d62a1c1..8f0bdb28 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -120,7 +120,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`. - **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. @@ -162,6 +162,8 @@ 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: 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: Ack messages → On failure: route to DLQ output (dlq.{table}), then Ack to prevent infinite retry diff --git a/internal/ingest/bento.go b/internal/ingest/bento.go index 0d996067..19d05c1e 100644 --- a/internal/ingest/bento.go +++ b/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,36 @@ 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.
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) + 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", @@ -486,6 +505,7 @@ func StartIngestWorker(ctx context.Context, nc *nats.Conn, chConn driver.Conn, c return &jsInput{ consumer: cons, chConn: chConn, + js: js, }, nil }, ); err != nil { diff --git a/internal/ingest/bento_test.go b/internal/ingest/bento_test.go index c2e6a11a..385cb803 100644 --- a/internal/ingest/bento_test.go +++ b/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.
, 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) { From 26ba089c80491a02e6c39fd598908313c6a3079a Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Tue, 12 May 2026 18:37:09 -0400 Subject: [PATCH 2/9] docs(ingest): keep #91 open as Phase 2 tracker; AGENTS.md ingest blurb MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Claude review feedback on PR #122: - CHANGELOG.md no longer says "Closes #91" so the issue stays open after merge to track Phase 2 (transient-vs-permanent error classification). - AGENTS.md §Architecture's `ingest/` quick-reference now mentions the permanent-error policy on delete failures (DLQ + DoubleAck, not Nak). Co-Authored-By: Claude Opus 4.7 (1M context) --- AGENTS.md | 2 +- CHANGELOG.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 1ac1e502..5b5d7771 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -19,7 +19,7 @@ Ten 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.
` 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` - **`pipes/`** — Named query pipes: `NamedQuery` type + NATS KV store (`WAVEHOUSE_PIPES`) + `.sql` file bootstrap - **`policy/`** — Hasura-style access control: `Policy`/`TablePolicy`/`RolePermissions` types, `Evaluate()` engine with JWT claim templating, NATS KV store (`WAVEHOUSE_POLICY`) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3fbe0ec1..e7039fc2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased ### Fixed -- **Ingest worker no longer infinite-retries permanent delete errors** (`internal/ingest/bento.go`, `internal/ingest/bento_test.go`, `docs/architecture.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.
` (reusing the existing `bentoDLQDropped` counter when the DLQ publish itself fails) and the message is `DoubleAck`'d so it leaves the main queue. Phase 2 (transient vs permanent classification — timeouts and network errors should still Nak) is tracked as follow-up on issue #91. Closes #91. +- **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.
` (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`. From 34f742f3695a87d1d411085d652f34badd6ab391 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Wed, 13 May 2026 16:36:49 -0400 Subject: [PATCH 3/9] docs(ingest): document DLQ shape contract + MD022 + accurate DoubleAck wording MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses bot-review feedback on PR #122: - bento.go: comment above the delete-DLQ Publish noting the message-shape difference vs dlqOutput.WriteBatch — insert failures publish raw.Payload (inner data only), delete failures publish m.Data() (full envelope) so a consumer can re-issue the delete. Phase 2 (issue #91) may normalize. - docs/src/content/docs/architecture.md: explicit DLQ-shape contract on the ingest/ blurb so future consumers find it; and fix the data-flow line that read "Ack messages" — the code uses DoubleAck. - CHANGELOG.md: blank line above the new "### Fixed" heading (MD022). Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 1 + docs/src/content/docs/architecture.md | 4 ++-- internal/ingest/bento.go | 6 ++++++ 3 files changed, 9 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index da0fef34..58a8ca97 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ 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.
` (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. diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md index 699ab86d..9acfd888 100644 --- a/docs/src/content/docs/architecture.md +++ b/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`; 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`. +- **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. @@ -158,7 +158,7 @@ Bento ingest pipeline (StartIngestWorker): → Delete actions handled inline (chConn.Exec) → 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: Ack messages + → On success: DoubleAck messages → On failure: route to DLQ output (dlq.{table}), then Ack to prevent infinite retry Active Sweeper (async goroutine, every 60s): diff --git a/internal/ingest/bento.go b/internal/ingest/bento.go index 6a47acff..d9303ca6 100644 --- a/internal/ingest/bento.go +++ b/internal/ingest/bento.go @@ -186,6 +186,12 @@ func (j *jsInput) Read(ctx context.Context) (*service.Message, service.AckFunc, "error", err, ) + // 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.
` 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", From 89b1c0d3ecbbddb80b0bc37620b8fdea4b76816b Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Wed, 13 May 2026 16:44:47 -0400 Subject: [PATCH 4/9] docs(changelog): update docs path to docs/src/content/docs/architecture.md MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stale path artifact from the merge — main moved docs/ to Astro layout under docs/src/content/docs/ but my Phase-1 CHANGELOG entry still listed the pre-merge path. Caught by CodeRabbit on the round-2 review. Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 58a8ca97..49daba5d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## 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.
` (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. +- **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.
` (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`. From 5faba6ffd0566718ef915bba22d380c4a3c25624 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Wed, 13 May 2026 16:57:26 -0400 Subject: [PATCH 5/9] docs(ingest): TODO(#91, Phase 2) marker on the permanent-delete-error policy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Gemini's [SHOULD] on PR #122: makes the Phase 1 trade-off grep-able (TODO(#91, ...)) and explicit about the cost — a ClickHouse blip during a delete drops the message to DLQ instead of retrying, which Phase 2's transient-vs-permanent classifier will fix. Comment-only; no code logic change. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/ingest/bento.go | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/internal/ingest/bento.go b/internal/ingest/bento.go index d9303ca6..3e6c5ff4 100644 --- a/internal/ingest/bento.go +++ b/internal/ingest/bento.go @@ -177,9 +177,12 @@ func (j *jsInput) Read(ctx context.Context) (*service.Message, service.AckFunc, // 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.
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. + // 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, From 0c1a833c9c184de3e0600f9afea004d43792f9f9 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Thu, 14 May 2026 16:40:54 -0400 Subject: [PATCH 6/9] fix(ingest): BYOS-safe DLQ discriminator via NATS header + trace propagation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The delete-error branch was publishing the full NATS envelope to dlq.
while the insert-failure branch publishes the inner data payload only. Consumers can't tell them apart by branching on the JSON `action` field — under BYOS a user row can carry any column name, including `action`, so the discriminator would mis-classify insert payloads. Switch the delete-DLQ publish from j.js.Publish to j.js.PublishMsg with a non-user-controlled NATS header (Wave-DLQ-Type: delete-envelope). Insert path stays unchanged for Phase 1; absence of the header means insert-payload. Phase 2 (issue #91) will tag insert failures with Wave-DLQ-Type: insert-payload and may normalize both shapes. Also propagate the clickhouse_delete span into the DLQ message via observability.InjectNATS so a Phase 2 consumer can stitch its trace to the worker's. Trace context lives in the same nats.Header; the existing observability helper handles the W3C TraceContext injection. Test updates: bentoMockJS and bentoMockJSWithError now override PublishMsg (matching the new code path) and capture the message Header. The existing TestJsInput_Read_DeleteExecError gains an assertion that the published DLQ message carries Wave-DLQ-Type: delete-envelope. docs/src/content/docs/architecture.md: replace the "branch on action" guidance with the BYOS-safe header contract. Refs #91, addresses CodeRabbit Major (BYOS shape discriminator) + Gemini SHOULD (DLQ trace propagation) on PR #122. Co-Authored-By: Claude Opus 4.7 (1M context) --- docs/src/content/docs/architecture.md | 2 +- internal/ingest/bento.go | 27 ++++++++++++++++++++------- internal/ingest/bento_test.go | 22 +++++++++++++++++++++- 3 files changed, 42 insertions(+), 9 deletions(-) diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md index 9acfd888..b0704c93 100644 --- a/docs/src/content/docs/architecture.md +++ b/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`; 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. +- **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. diff --git a/internal/ingest/bento.go b/internal/ingest/bento.go index 3e6c5ff4..9951a406 100644 --- a/internal/ingest/bento.go +++ b/internal/ingest/bento.go @@ -189,14 +189,27 @@ func (j *jsInput) Read(ctx context.Context) (*service.Message, service.AckFunc, "error", err, ) - // 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.
` must branch on the `"action"` field to distinguish the - // two shapes; Phase 2 (issue #91) may normalize this. + // DLQ shape contract: `dlq.
` 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 - if _, pubErr := j.js.Publish(spanCtx, dlqSubject, m.Data()); pubErr != nil { + 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, diff --git a/internal/ingest/bento_test.go b/internal/ingest/bento_test.go index 385cb803..fc475e95 100644 --- a/internal/ingest/bento_test.go +++ b/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,6 +37,7 @@ 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) { @@ -40,6 +45,11 @@ func (m *bentoMockJS) Publish(_ context.Context, subject string, payload []byte, 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 @@ -341,6 +351,12 @@ func TestJsInput_Read_DeleteExecError(t *testing.T) { 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") @@ -630,6 +646,10 @@ func (m *bentoMockJSWithError) Publish(_ context.Context, _ string, _ []byte, _ return nil, m.publishErr } +func (m *bentoMockJSWithError) PublishMsg(_ context.Context, _ *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) { + return nil, m.publishErr +} + func TestDLQOutput_WriteBatch_PublishError(t *testing.T) { t.Parallel() js := &bentoMockJSWithError{publishErr: errors.New("NATS unavailable")} From c2e2cccf20c6b13a01229602738589ba14c8ce4c Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Thu, 14 May 2026 16:49:57 -0400 Subject: [PATCH 7/9] test(ingest): assert DLQ publish is attempted when NATS publish fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TestJsInput_Read_DeleteExecErrorDLQPublishFails covered the ack semantics (no Nak, DoubleAck) but never proved j.js.PublishMsg was actually called. A regression that skipped the DLQ publish entirely — jumping straight to DoubleAck on Exec error — would have silently passed the test while losing every "permanent delete failed" envelope without an operator-visible DLQ-dropped error log. Capture the message in bentoMockJSWithError.PublishMsg before returning the error, then assert the attempted publish: subject (dlq.
), payload (original envelope), and Wave-DLQ-Type: delete-envelope header. Per CodeRabbit on PR #122 / commit 0c1a833. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/ingest/bento_test.go | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/internal/ingest/bento_test.go b/internal/ingest/bento_test.go index fc475e95..e4358031 100644 --- a/internal/ingest/bento_test.go +++ b/internal/ingest/bento_test.go @@ -394,6 +394,18 @@ func TestJsInput_Read_DeleteExecErrorDLQPublishFails(t *testing.T) { 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") + table, _ := msg.MetaGet("table_name") assert.Equal(t, "events", table) } @@ -640,13 +652,17 @@ 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 } -func (m *bentoMockJSWithError) PublishMsg(_ context.Context, _ *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) { +// 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 } From 55da248211cca0ec1b86c6ed9fa6eedf04f68f3a Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Thu, 14 May 2026 17:14:27 -0400 Subject: [PATCH 8/9] =?UTF-8?q?test(ingest):=20integration=20coverage=20fo?= =?UTF-8?q?r=20delete=20action=20=E2=80=94=20happy=20path=20+=20DLQ=20roun?= =?UTF-8?q?d-trip?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The delete-action path through the worker had zero end-to-end coverage. Unit tests in internal/ingest/bento_test.go assert the right behavior on mocks (header set, payload preserved, DoubleAck called) but can't verify the wire contract: that NATS actually carries the Wave-DLQ-Type header across the publish/subscribe boundary, that the buffer consumer's ack-floor actually advances (proving no infinite-Nak loop in the live consumer), and that real ClickHouse actually applies the lightweight DELETE. This commit adds two integration tests: TestDelete_FailureRoutesToDLQWithHeader (tests/integration/dlq_test.go): publishes a delete envelope for a nonexistent table, subscribes to dlq.
via an ephemeral consumer on the WAVEHOUSE_DLQ stream, and asserts: - Wave-DLQ-Type: delete-envelope header survives the NATS round-trip (the BYOS-safe discriminator from PR #122 review). - Subject is dlq.
, payload is the original envelope bit-for-bit. - Buffer consumer state shows NumAckPending=0 and Delivered advanced, proving the worker DoubleAck'd (a Nak would leave the message pending until AckWait elapses — the infinite-Nak symptom this PR fixes). The Traceparent / W3C trace propagation assertion is intentionally omitted; see the inline comment for why (depends on OTEL globals that the integration harness leaves unset; round-trip is already covered by observability/tracer_test.go). TestDelete_HappyPath (tests/integration/ingest_test.go): POSTs an insert via the public API, waits for it to land in ClickHouse, publishes an action:"delete" envelope directly to JetStream (no public SDK/HTTP delete surface today — the delete is a NATS-only contract), and asserts the row is gone and no DLQ entry was teed. Also adds a small pollCount helper in ingest_test.go (DRYs the count-by-id polling pattern that this test and any future delete tests would otherwise duplicate). Refs #91, PR #122. Resilience tests (crash/resume, CH-down-mid-flow, NATS-unreachable) need framework changes and stay scoped for a separate follow-up. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/integration/dlq_test.go | 121 +++++++++++++++++++++++++++++++ tests/integration/ingest_test.go | 90 +++++++++++++++++++++++ 2 files changed, 211 insertions(+) diff --git a/tests/integration/dlq_test.go b/tests/integration/dlq_test.go index 65af392a..1b206a23 100644 --- a/tests/integration/dlq_test.go +++ b/tests/integration/dlq_test.go @@ -10,8 +10,12 @@ import ( "testing" "time" + "github.com/nats-io/nats.go/jetstream" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/Wave-RF/WaveHouse/internal/ingest" + "github.com/Wave-RF/WaveHouse/internal/mq" ) // TestDLQ_StatsEmptyOnFreshStart verifies the DLQ exposes an empty result @@ -84,3 +88,120 @@ func TestDLQ_PopulatedOnBentoFailure(t *testing.T) { return present }, 30*time.Second, 500*time.Millisecond, "DLQ should receive failed events within timeout") } + +// TestDelete_FailureRoutesToDLQWithHeader verifies the end-to-end Phase-1 +// permanent-delete-error policy (#91): +// +// 1. A bad delete (table doesn't exist in ClickHouse) reaches the worker. +// 2. The worker publishes the original NATS envelope to `dlq.
`. +// 3. The envelope carries a `Wave-DLQ-Type: delete-envelope` NATS header +// that survives the NATS round-trip — that's the BYOS-safe discriminator +// a downstream consumer must use (per PR #122 review). +// 4. The buffer consumer DoubleAcks the original message — no Nak loop. +// +// The unit tests in internal/ingest/bento_test.go assert the worker called +// PublishMsg with the right header on its mock. This test proves the same +// invariants over real JetStream wire format and consumer-state semantics. +func TestDelete_FailureRoutesToDLQWithHeader(t *testing.T) { + e := env(t) + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + js := e.embeddedMQ.JetStream() + + // Use a table name that intentionally doesn't exist in ClickHouse so the + // worker's DELETE Exec fails with the permanent-error path under test. + // Per-test suffix keeps state independent across runs / from other tests. + table := fmt.Sprintf("nonexistent_delete_dlq_%d", time.Now().UnixNano()) + deleteID := fmt.Sprintf("test-id-%d", time.Now().UnixNano()) + + // Ephemeral DLQ consumer scoped to our subject (no Durable name → no + // state leak across tests). DeliverAll covers the case where the worker + // publishes before we start listening on busy CI. + dlqCons, err := js.CreateOrUpdateConsumer(ctx, mq.DLQStreamName(), jetstream.ConsumerConfig{ + FilterSubject: "dlq." + table, + AckPolicy: jetstream.AckExplicitPolicy, + DeliverPolicy: jetstream.DeliverAllPolicy, + }) + require.NoError(t, err) + + // Snapshot the buffer consumer's ack state before publishing. After the + // DLQ receive we'll assert AckFloor caught up to Delivered — a DoubleAck + // advances both; a Nak advances Delivered but leaves AckFloor stuck + // until AckWait elapses (30s default), the symptom of the infinite-Nak + // loop this PR fixes. + bufferCons, err := js.Consumer(ctx, mq.StreamName(), ingest.BufferConsumerName) + require.NoError(t, err) + preInfo, err := bufferCons.Info(ctx) + require.NoError(t, err) + + // Publish the delete envelope directly to ingest.
. The worker + // picks it up, fails the CH Exec (no such table), and routes the + // envelope to dlq.
via PublishMsg + Wave-DLQ-Type header. + envelope := map[string]any{ + "action": "delete", + "table_name": table, + "id": deleteID, + } + payload, err := json.Marshal(envelope) + require.NoError(t, err) + _, err = js.Publish(ctx, "ingest."+table, payload) + require.NoError(t, err) + + // Wait up to 30s for the DLQ message — covers Bento's batch window plus + // loaded-CI slack. Fetch(1, …) is the simplest "pull next" primitive. + batch, err := dlqCons.Fetch(1, jetstream.FetchMaxWait(30*time.Second)) + require.NoError(t, err) + var got jetstream.Msg + for m := range batch.Messages() { + got = m + break + } + require.NoError(t, batch.Error()) + require.NotNil(t, got, "DLQ should receive the failed delete envelope within timeout") + defer func() { _ = got.Ack() }() + + // BYOS-safe discriminator: header is set by the worker (non-user- + // controlled). A user row with an `action` column can't forge this. + assert.Equal(t, "delete-envelope", got.Headers().Get("Wave-DLQ-Type"), + "Wave-DLQ-Type header must survive the NATS round-trip end-to-end") + + // Subject and payload are preserved bit-for-bit so a Phase-2 consumer + // can re-issue the delete with full context. + assert.Equal(t, "dlq."+table, got.Subject()) + assert.JSONEq(t, string(payload), string(got.Data()), + "DLQ envelope must equal the original ingest.
envelope verbatim") + + // Note: we intentionally do NOT assert on the W3C `traceparent` header + // here. observability.InjectNATS is called by the worker and writes + // traceparent in production (where cmd/wavehouse runs + // observability.InitProvider, installing a real TracerProvider + + // TraceContext propagator). The integration test harness in + // setup_test.go intentionally leaves both globals unset so otel_test.go + // can save/restore them around its own InitProvider invocations. Adding + // trace-propagation coverage to this test would require either wiring + // InitProvider into the shared harness (touches otel_test.go's + // save/restore dance) or installing a per-test recording tracer. + // The propagator round-trip is fully covered by + // observability/tracer_test.go:TestInjectExtractNATS_Roundtrip, so the + // signal-per-effort here is low. + + // Buffer consumer state: AckFloor must have caught up to Delivered, + // proving the worker DoubleAck'd our delete rather than Nak'ing it. + // Use Eventually with a short window — the DoubleAck and ack-floor + // update happen post-DLQ-publish but typically within a few ms. + require.Eventually(t, func() bool { + postInfo, err := bufferCons.Info(ctx) + if err != nil { + return false + } + // Our publish advanced Delivered by at least 1. + if postInfo.Delivered.Consumer < preInfo.Delivered.Consumer+1 { + return false + } + // Nothing should be stuck pending an ack — neither our message + // (which must have been DoubleAck'd) nor any earlier test's. + return postInfo.NumAckPending == 0 + }, 5*time.Second, 100*time.Millisecond, + "buffer consumer must DoubleAck the failed delete (NumAckPending=0); a Nak would leave it pending until AckWait elapses") +} diff --git a/tests/integration/ingest_test.go b/tests/integration/ingest_test.go index 91fa55e6..912de93a 100644 --- a/tests/integration/ingest_test.go +++ b/tests/integration/ingest_test.go @@ -15,6 +15,23 @@ import ( "github.com/stretchr/testify/require" ) +// pollCount returns the row count for `SELECT count() FROM
WHERE id = ''`, +// or -1 on query error. Used by the happy-path delete test to poll for both +// the insert landing and the row disappearing post-delete. +func pollCount(t *testing.T, table, id string) int64 { + t.Helper() + var count uint64 + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + err := env(t).chConn.QueryRow(ctx, + fmt.Sprintf("SELECT count() FROM %s WHERE id = '%s'", table, id), + ).Scan(&count) + if err != nil { + return -1 + } + return int64(count) +} + // TestIngest_FlowsToClickHouseWithoutDLQ exercises the happy path: POST // /v1/ingest/{table} is acknowledged synchronously, Bento batches the event // to ClickHouse, and the DLQ stays empty for that table. @@ -64,3 +81,76 @@ func TestIngest_FlowsToClickHouseWithoutDLQ(t *testing.T) { _, hasDLQ := tables["dlq."+table] assert.False(t, hasDLQ, "successful inserts should not produce DLQ entries") } + +// TestDelete_HappyPath exercises the worker's success branch for +// `action: "delete"`: an inserted row is removed end-to-end via a delete +// envelope published to ingest.
. There's no public SDK / HTTP delete +// surface today — the delete envelope is a NATS-only contract — so the test +// publishes directly to JetStream and observes ClickHouse state via the +// shared connection. +// +// Pairs with TestDelete_FailureRoutesToDLQWithHeader in dlq_test.go, which +// covers the failure branch. Together they give the worker's delete path +// real end-to-end coverage — previously zero. +func TestDelete_HappyPath(t *testing.T) { + e := env(t) + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + table := createTable(t, + "id String, page String", + "ORDER BY id", + ) + const rowID = "row-to-delete" + + // Insert via the public API so the test exercises the full ingest path + // (validation → JetStream → Bento → ClickHouse) before the delete. + body := fmt.Sprintf(`{"id":%q,"page":"/about"}`, rowID) + resp, err := http.Post( + e.server.URL+"/v1/ingest/"+table, + "application/json", + strings.NewReader(body), + ) + require.NoError(t, err) + resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + + // Wait for the insert to land. 30s upper bound on Bento's 5s batch + // window + loaded-runner slack, matching other tests in this package. + require.Eventually(t, func() bool { + return pollCount(t, table, rowID) == 1 + }, 30*time.Second, 500*time.Millisecond, "insert should land before delete") + + // Publish the delete envelope directly to JetStream. The worker reads + // it via the buffer-consumer, executes DELETE FROM `
` WHERE id=?, + // and DoubleAcks on Exec success (internal/ingest/bento.go). + envelope := map[string]any{ + "action": "delete", + "table_name": table, + "id": rowID, + } + payload, err := json.Marshal(envelope) + require.NoError(t, err) + _, err = e.embeddedMQ.JetStream().Publish(ctx, "ingest."+table, payload) + require.NoError(t, err) + + // ClickHouse lightweight DELETE (default in modern CH) marks rows + // invisible immediately after the mutation is applied. 30s upper bound + // is generous for a single-row mutation on the test runner. + require.Eventually(t, func() bool { + return pollCount(t, table, rowID) == 0 + }, 30*time.Second, 500*time.Millisecond, + "action:\"delete\" envelope should remove the row from ClickHouse") + + // Confirm the success branch didn't tee anything into the DLQ — the + // permanent-error policy only fires on Exec failures. + dlqResp, err := http.Get(e.server.URL + "/v1/dlq/stats") + require.NoError(t, err) + defer dlqResp.Body.Close() + var stats map[string]any + require.NoError(t, json.NewDecoder(dlqResp.Body).Decode(&stats)) + tables, ok := stats["tables"].(map[string]any) + require.True(t, ok) + _, hasDLQ := tables["dlq."+table] + assert.False(t, hasDLQ, "successful deletes must not produce DLQ entries") +} From 747a747d312c1babc1278e64a51e741f96f57435 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Thu, 14 May 2026 20:27:44 -0400 Subject: [PATCH 9/9] test(ingest): address review on the new delete-action integration tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two refinements from Claude's review of 55da248 on PR #122: pollCount in ingest_test.go used fmt.Sprintf with single-quote id interpolation and a raw (unquoted) table identifier. Current callers pass controlled constants so there's no live injection risk, but the function signature accepts arbitrary strings and would be copied verbatim by future test authors. Switch to the same pattern bento.go:172 uses in production: backtick-quoted table identifier + driver-bound ? parameter for the id. Comment notes why the table can't be a bound parameter (CH identifiers aren't bindable; backticks are the same defense the production path uses). TestDelete_FailureRoutesToDLQWithHeader's no-redelivery assertion used NumAckPending == 0, which is a global claim about the entire buffer consumer rather than a claim about our specific message. If some unrelated test left a Nak'd message whose AckWait hadn't elapsed, this would time out with a misleading "must DoubleAck the failed delete" error. Switch to capturing pubAck.Sequence from js.Publish and asserting AckFloor.Stream >= pubAck.Sequence — that targets THIS message's stream seq, immune to global consumer state. The pre/post Delivered snapshot is no longer needed; removed. Per Claude on PR #122 / commit 55da248. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/integration/dlq_test.go | 37 +++++++++++++++----------------- tests/integration/ingest_test.go | 10 +++++++-- 2 files changed, 25 insertions(+), 22 deletions(-) diff --git a/tests/integration/dlq_test.go b/tests/integration/dlq_test.go index 1b206a23..cfa4260f 100644 --- a/tests/integration/dlq_test.go +++ b/tests/integration/dlq_test.go @@ -125,19 +125,18 @@ func TestDelete_FailureRoutesToDLQWithHeader(t *testing.T) { }) require.NoError(t, err) - // Snapshot the buffer consumer's ack state before publishing. After the - // DLQ receive we'll assert AckFloor caught up to Delivered — a DoubleAck - // advances both; a Nak advances Delivered but leaves AckFloor stuck - // until AckWait elapses (30s default), the symptom of the infinite-Nak - // loop this PR fixes. bufferCons, err := js.Consumer(ctx, mq.StreamName(), ingest.BufferConsumerName) require.NoError(t, err) - preInfo, err := bufferCons.Info(ctx) - require.NoError(t, err) // Publish the delete envelope directly to ingest.
. The worker // picks it up, fails the CH Exec (no such table), and routes the // envelope to dlq.
via PublishMsg + Wave-DLQ-Type header. + // + // We capture the publish ack's stream sequence so the no-redelivery + // assertion below can target *this specific message*. A coarser + // "NumAckPending == 0" check would be a global claim about the buffer + // consumer's state — fragile if any other test (now or future) leaves a + // Nak'd message in pending state while AckWait elapses. envelope := map[string]any{ "action": "delete", "table_name": table, @@ -145,8 +144,9 @@ func TestDelete_FailureRoutesToDLQWithHeader(t *testing.T) { } payload, err := json.Marshal(envelope) require.NoError(t, err) - _, err = js.Publish(ctx, "ingest."+table, payload) + pubAck, err := js.Publish(ctx, "ingest."+table, payload) require.NoError(t, err) + require.NotNil(t, pubAck) // Wait up to 30s for the DLQ message — covers Bento's batch window plus // loaded-CI slack. Fetch(1, …) is the simplest "pull next" primitive. @@ -186,22 +186,19 @@ func TestDelete_FailureRoutesToDLQWithHeader(t *testing.T) { // observability/tracer_test.go:TestInjectExtractNATS_Roundtrip, so the // signal-per-effort here is low. - // Buffer consumer state: AckFloor must have caught up to Delivered, - // proving the worker DoubleAck'd our delete rather than Nak'ing it. - // Use Eventually with a short window — the DoubleAck and ack-floor - // update happen post-DLQ-publish but typically within a few ms. + // Buffer consumer state: AckFloor.Stream must have advanced to at least + // our publish's stream sequence, proving the worker DoubleAck'd *our* + // delete (not a global "everything is quiet" claim that could be tripped + // by an unrelated test leaving a Nak'd message pending). A Nak would + // leave AckFloor.Stream stuck below pubAck.Sequence until AckWait + // elapses (30s default) — the symptom of the infinite-Nak loop this PR + // fixes. require.Eventually(t, func() bool { postInfo, err := bufferCons.Info(ctx) if err != nil { return false } - // Our publish advanced Delivered by at least 1. - if postInfo.Delivered.Consumer < preInfo.Delivered.Consumer+1 { - return false - } - // Nothing should be stuck pending an ack — neither our message - // (which must have been DoubleAck'd) nor any earlier test's. - return postInfo.NumAckPending == 0 + return postInfo.AckFloor.Stream >= pubAck.Sequence }, 5*time.Second, 100*time.Millisecond, - "buffer consumer must DoubleAck the failed delete (NumAckPending=0); a Nak would leave it pending until AckWait elapses") + fmt.Sprintf("buffer consumer's AckFloor.Stream must reach %d (our publish seq), proving the worker DoubleAck'd this specific delete rather than Nak'ing it", pubAck.Sequence)) } diff --git a/tests/integration/ingest_test.go b/tests/integration/ingest_test.go index 912de93a..3cc9270b 100644 --- a/tests/integration/ingest_test.go +++ b/tests/integration/ingest_test.go @@ -15,16 +15,22 @@ import ( "github.com/stretchr/testify/require" ) -// pollCount returns the row count for `SELECT count() FROM
WHERE id = ''`, +// pollCount returns the row count for SELECT count() FROM `
` WHERE id = ?, // or -1 on query error. Used by the happy-path delete test to poll for both // the insert landing and the row disappearing post-delete. +// +// `id` is bound via the driver's parameterized form rather than interpolated +// into the SQL, mirroring the production pattern in internal/ingest/bento.go. +// `table` still uses fmt.Sprintf because ClickHouse identifiers can't be bound +// as parameters — backtick quoting is the same defense bento.go uses. func pollCount(t *testing.T, table, id string) int64 { t.Helper() var count uint64 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() err := env(t).chConn.QueryRow(ctx, - fmt.Sprintf("SELECT count() FROM %s WHERE id = '%s'", table, id), + fmt.Sprintf("SELECT count() FROM `%s` WHERE id = ?", table), + id, ).Scan(&count) if err != nil { return -1