feat(core): capture the final recorded error for failed rETL records

[kuldeep0020]

What

First slice of "show all errors in rETL" (spec: rudder-specs #109, show-all-errors-rudder-server-lld.md): capture the final recorded error per failed rETL record — discarded today at the rsources collector — cap and sanitize it, gate it, and carry it additively to the internal v2 failed-records API as the error leaf.

Changes

• Additive error_response text NOT NULL DEFAULT '' column on rsources_failed_keys_v2_records (ADD COLUMN IF NOT EXISTS, mirrors the code precedent; no backfill).
• Envelope unwrap (reason/response/Error/error, exact-key match) → rune-safe byte cap (Rsources.failedKeys.maxErrorLength, default 2048, reloadable) → UTF-8 sanitize, in that order.
• Capture predicate: Rsources.failedKeys.captureErrorDetail (default false) AND per-connection opt-in on the job's ParametersT AND reloadable blockedConnections/blockedWorkspaces operator blacklists (<sourceID>:<destID> keys). Suppression drops only the message; {record, code} and retries are untouched.
• Opt-in resolution (the spec's Option-2 revision — no wire signal): the processor resolves config.source.syncSettings.errorDetailsConfig.enabled from the connection object in its own backend-config subscription (connectionConfigMap), at the ParametersT marshal site. The first resolution per job_run_id is pinned (24h TTL, bounded store — run ids are client-mintable via the payload, so the store must not grow unboundedly), so every record of one sync carries the same decision and a toggle takes effect from the next sync. Fail-closed on unknown connection / absent path / non-bool. Nothing client-supplied feeds the decision.
• Aggregate capture telemetry per publish (rsources_failed_records_error_{captured,clipped,suppressed} + summary log). Message bodies are never logged.

Carrier history (why the diff shows gateway deletions)

The opt-in had two earlier carriers on this branch, both retired in review: the context.sources.capture_error payload stamp (leaks through old servers to destinations), then the X-Rudder-Capture-Error-Detail request header. Under Option 2 the gateway intake and the gateway→processor carrier (EventParams/Metadata) are deleted; the flag is born server-side at the processor. A forged header or payload field is inert because no code reads one — SEC-L1 is closed by construction, pinned by tests on both the gateway and processor sides.

Notes for review

• The JSON leaf tag is error (the cross-service wire contract in the spec README); error_response is the column name. The LLD §3.2 snippet says error_response — flagged on the spec PR as a spec correction.
• reason outranks response in the unwrap: a drained job carries both a stale destination body and the drain reason, and the reason is the final recorded error (pinned by test).
• The collector still does no backend-config cross-check of its own: it parses only ParametersT.capture_error, which the processor now stamps from config. The durable landing gate lives in rudder-sources.
• The pin store is capped (50k) rather than operator-configured: at the cap, resolution still happens per record and stays fail-closed — only whole-run uniformity lapses for runs arriving while full.

Rollout

Ships first with captureErrorDetail off everywhere. Do not enable until the rudder-sources landing release is live (ordering invariant in the spec README).

Testing

Test plan A1–A4 from the spec: envelope shapes, cap/UTF-8/NUL handling, predicate + runtime-reloadable blacklists, v2 transport both directions (old↔new interop) — including round-trips against real postgres. A3 (rewritten for Option 2): resolution true/false/fail-closed shapes, per-run pin across a live config flip (driven through a real connectionConfigMap rebuild), pin TTL/sweep/cap lifecycle, forged header + payload signals inert on both gateway and processor. New tests race-clean; go build ./..., full gateway/processor/rsources suites, and the Makefile-pinned golangci-lint clean.

Linear: PRO-5904

[codecov]

Codecov Report

❌ Patch coverage is 97.62846% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 79.81%. Comparing base (f2655fb) to head (5e819c4).
⚠️ Report is 11 commits behind head on master.

Files with missing lines	Patch %	Lines
services/rsources/stats_collector.go	92.15%	2 Missing and 2 partials ⚠️
services/rsources/handler.go	87.50%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #7295      +/-   ##
==========================================
- Coverage   80.12%   79.81%   -0.32%     
==========================================
  Files         601      606       +5     
  Lines       67217    67893     +676     
==========================================
+ Hits        53857    54186     +329     
- Misses      10184    10521     +337     
- Partials     3176     3186      +10     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[kuldeep0020]

Adopted — you were right, and the precedent held up all the way down. As of d92738e the opt-in travels as a X-Rudder-Capture-Error-Detail: true request header (named after the captureErrorDetail config key), parsed in augmentAuthRequestContext beside the run-id headers with the same verbatim-value idiom (strict == "true"). The payload mechanism is gone entirely: no stamp, no first-event read, and the per-event strip is deleted — nothing touches context.sources anymore.

Two knock-on simplifications came free:

• rudder-sources' producer side (#3536) now sets the header per request beside X-Rudder-Job-Run-Id, and its deployment kill switch is deleted: the switch existed to keep the stamp from leaking through old servers into destination payloads, and headers an old server ignores can't leak — both version-skew directions are now additive.
• The "no opted-out connection's error text enters storage" invariant, which payload forgery undermined (SEC-L1), now holds structurally: row data can't set a header. The residual surface is a client holding the connection's valid write key — the same trust class as X-Rudder-Job-Run-Id itself.

Spec updated on rudder-specs#109 (fb4fe93) to the header contract.

One follow-up your review of this area might have an opinion on: the honor gate still keys on sourcesJobRunID after the first event's payload may have overridden it (pre-existing behavior, pinned by test). Gating on authContext.SourceJobRunID instead would make "authenticated rETL only" purely header-derived — happy to do that here or separately.

Note: this push deliberately addresses only the carrier rework. The architects-panel review on the spec PR (B1 query-string redaction, M1 first-wins upsert) is being handled separately — B1/M1 fixes follow after the design discussion rather than riding this change.

[kuldeep0020]

Update since my last comment here: we went a step further than the header. There's no wire signal at all anymore. The processor resolves errorDetailsConfig.enabled from the connection in its own backend-config subscription at the point it builds ParametersT, and pins the first answer per job run id so a mid-sync toggle can't split a run (24h TTL, bounded map, since run ids are client-mintable). The gateway intake and the EventParams/Metadata plumbing are gone as of e54b728, so a forged header or payload flag is inert, nothing reads them.

Spec is updated on rudder-specs#109 (server LLD §3.6). Your original comment is what pushed it here: once the opt-in moved out of the payload, the natural end state was not trusting the client for it at all.

[atzoum]

rETL error-capture decision store

Problem

The current opt-in mechanism is wrong:

• The decision lives in memory. A rudder-server restart during a sync loses it.
• Multiple rudder-server nodes take part in one sync. Each node makes its own decision.
• The opt-in bookkeeping leaked into statsCollector (captureErrorJobIds).

Goal

Build one component. This component is the authority for one question: must a sync run (jobRunID) on a given connection (sourceID and destinationID) capture error responses?

Design

The component has two inputs:

• A pinned decision per jobRunID, stored in a central database table. Every node and every restart sees the same value.
• Operator settings from reloadable config, applied on each call.

See internal/drain-config for a component with a similar shape. Do not copy its code. Use the same idea: a shared or local database, with a maximum of 2 connections.

Interface

The component exposes one method:

GetErrorResponse(ctx context.Context, key statKey, status *jobsdb.JobStatusT) (string, error)

The method returns the error text to store, or "" when capture is off for this combination. A database error propagates to the caller.

The method applies its checks in this order:

1. If the raw error response is empty, '', or {}, return "". Do not touch the cache or the database.
2. If the enabled/disabled flag is off, return "". Do not touch the cache or the database. Do not insert a row.
3. Read the pinned decision for the jobRunID. If the decision is false, return "".
4. If the connection or the workspace is on a blocklist, return "".
5. Extract the error text. Clip it to the max length. Sanitize it. Return it.

Pinned decision

The decision value comes from backend config: source.syncSettings.errorDetailsConfig.enabled. The component subscribes to backend config when it starts. If backend config does not contain the connection, the decision is false.

The decision is stored in one table with 3 columns: job_run_id, store_error_responses, created_at. The key is job_run_id alone. This is sufficient because in rETL one jobRunID maps to one source.

Keep database queries and inserts to a minimum:

• On startup, load all entries from the database into a memory cache.
• On a cache miss for a jobRunID:
  1. Begin a transaction.
  2. Take a transaction-scoped advisory lock: pg_advisory_xact_lock(classid, objid). Use a dedicated class id for this component and a hash of the jobRunID as the object id. The two-argument form has its own keyspace. It cannot collide with the single-argument locks that jobsdb (hash-derived ids) and the rsources handler (id 100020001) already use.
  3. Look up the jobRunID in the table.
  4. If the row is not found, compute the decision from backend config and insert it.
  5. Commit the transaction and cache the result.

Do not limit the number of jobRunIDs in the memory cache. The memory overhead is not important at this stage.

Operator settings

Move all configuration logic that now lives in statsCollector (errorCaptureSettings) into the new component:

• The enabled/disabled flag (Rsources.failedKeys.captureErrorDetail), used in step 2.
• The blocklists, used in step 4.
• The max-length cap, used in step 5.

Do not overengineer the solution. Do not use sync.Map or settings snapshotting at this stage. Plain reloadable config reads are sufficient.

Cleanup

The component periodically deletes entries older than a configured age. The default is 24 hours. drainConfigManager does something similar, but do not copy its patterns without review.

A run that is longer than the retention age loses its row. A later cache miss computes the decision again, and the value can change mid-run. This is acceptable.

Lifecycle

Create one instance per process, at application startup. Pass it to every NewStatsCollector caller. The component owns its cleanup goroutine. The application starts and stops the component.

Integration

Inject the component into statsCollector. Replace captureErrorResponse with a call to the new method. Remove captureErrorJobIds and errorCaptureSettings from statsCollector.

Change CollectFailedRecords to accept a context.Context and return an error. Errors from GetErrorResponse propagate through it.

Remove the plumbing that the new component makes obsolete:

• processor/error_capture_optin.go (captureOptInPins), its wiring in processor, and its tests.
• gateway/handle_forged_capture_signals_test.go. The gateway header code is already gone. Only this test remains.

Names

Proposed names: table rsources_sync_settings, component SyncSettingDelegate. Both are open to better alternatives.

[atzoum]

we don't need this file do we?

[atzoum]

there should be no diff in the gateway package