feat(processor): fork event processing for configured isolated destinations

[atzoum]

Description

Adds the write side of the destination-isolated pipeline: the gw pool now forks events for configured destinations into the intermediate proc jobsdb (consumed by the proc pool from #7202) instead of transforming them inline, so a slow/failing destination's events are siphoned off and no longer hold up the shared pipeline.

Forking is opt-in per destination via a hierarchical, config which is by default disabled:

Processor.DestinationIsolation.enabledDestinations.<destinationID>
Processor.DestinationIsolation.enabledDestinations.all

The fork. At the pretransform fan-out stage, each event's forked destinations are split off from the inline groupedEvents. One proc job is written per forked event, carrying the forked destination IDs as its consumers (multi-consumer job) and a source-level payload. The forked-job list is carried over through the transform stages and finally committed during the store stage.

Atomicity. The proc write commits in the same transaction as the gateway status update so there is no room for duplicate jobs to be written in proc jobsdb in case of a crash.

rsources/retl excluded (for now). Events carrying a sourceJobRunID stay inline until the intermediate stage is modelled in rsources accounting, otherwise a retl source could be reported complete while its events still sit in the queue.

Supporting jobsdb changes

• Post-commit listeners register only after the store/update succeeds (internalStoreJobsInTx, pendingEventsJobsDB). With WithStoreSafeTxFromTx the caller's transaction is reused across the internal stale-dataset-list retry, so registering before the op double-fired the pending-events / cache / stat listeners on every retry.
• Single-consumer handles now allow persisting an explicit consumers array. The partition-migration buffer (proc_buf) is a single-consumer handle that relays multi-consumer proc jobs between nodes; it previously collapsed their consumers to the legacy '', so migrated jobs resurfaced under an unknown destination and were dropped. proc is the first multi-consumer jobsdb to be migrated, so this was never hit before.
• checkIfCompactDS rewrite: the terminal-job-count check (single- and multi-consumer) now does an index-only grouped scan instead of a per-job correlated subquery against the v_last_* views — pure perf, same result. Needed once proc compaction is exercised at the volumes below.
• WithDefaultSkipStatusCompaction: pass-through jobsdbs (gw, esch, arc, proc, and their _buf migration-buffer counterparts, err_idx) skip status compaction by default — they have no router-style retry backlog, so there's nothing to reclaim.

Deciding when to enable: fan-out visibility

An event on a source wired to N destinations produces N forked jobs, each needing its own proc consumer status update. So isolating all destinations is cheap on narrowly-wired sources but gets progressively more expensive as its source's fan-out grows — even if most of those destinations are filtered out downstream by a user transformation before they'd ever reach the router. For narrow-fanout scenarios we can safely isolate all destinations, while for wide-fanout ones we may need to isolate only specific slow/failing destinations rather than paying the fan-out cost for all of them. Two new stats provide visibility to make that call: processor_source_to_destination_fanout (histogram) and processor_max_source_to_destination_fanout (gauge), recorded per event at fan-out, before filtering.

BenchmarkProcessorDestinationIsolation quantifies this trade-off directly: 100k events fanned out to 10/20/50 destinations (all but 2 filtered downstream), with and without forking, across a few JobsDB.proc.maxDSSize settings.

Benchmark results

Single-run wall-clock drain times (local Docker Postgres + transformer — directional, not a production SLA):

Destinations	No fork	Forked (best maxDSSize)	Overhead
10	32.9s	36.7s (20k)	+12%
20	58.8s	1m08.5s (10k)	+16%
50	2m12.9s	3m12.9s (10k)	+45%

Full results

Scenario	Time
10 dest, no fork	32.90s
10 dest, forked, maxDSSize=100k	43.52s
10 dest, forked, maxDSSize=20k	36.71s
10 dest, forked, maxDSSize=10k	37.23s
20 dest, no fork	58.81s
20 dest, forked, maxDSSize=100k	1m16.45s
20 dest, forked, maxDSSize=20k	1m10.40s
20 dest, forked, maxDSSize=10k	1m08.48s
50 dest, no fork	2m12.95s
50 dest, forked, maxDSSize=20k	3m48.67s
50 dest, forked, maxDSSize=10k	3m12.88s
50 dest, forked, maxDSSize=5k	3m18.04s

Tests

• Processor integration test (TestProcessorDestinationIsolation), generalized to N destinations with partial forking and user-transformation filtering: all-forked / mixed / none / wide-fanout-mostly-filtered — asserts every event reaches the router regardless of path, the expected proc job count and per-consumer fan-out, and full drain.
• BenchmarkProcessorDestinationIsolation — proc consumer status-update throughput at scale, exercising maxProcDSSize/compaction settings.
• jobsdb unit tests for both supporting fixes.
• Partition-migration tests (embedded + gw/proc) extended with a forked destination (plus a second, router-aborted destination so proc jobs carry multiple consumers) to verify migration moves all pending consumers.

Linear Ticket

resolves PIPE-3049

Security

• The code changed/added as part of this pull request won't create any security issues with how the software is being used.

---

This is part 1 of 2 in a stack made with GitButler:

•  2  feat(processor): fork event processing for rsources pipelines #7235
•  1  feat(processor): fork event processing for configured isolated destinations #7207 👈

[codecov]

Codecov Report

❌ Patch coverage is 91.27907% with 15 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.34%. Comparing base (3e126b2) to head (3c0cd35).

Files with missing lines	Patch %	Lines
processor/processor.go	88.23%	3 Missing and 3 partials ⚠️
processor/proc_fork.go	92.00%	2 Missing and 2 partials ⚠️
...r/partitionbuffer/jobsdb_partition_buffer_store.go	85.71%	1 Missing and 1 partial ⚠️
jobsdb/jobsdb_pending_events.go	66.66%	1 Missing and 1 partial ⚠️
app/apphandlers/setup_partitionmigration.go	85.71%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #7207      +/-   ##
==========================================
+ Coverage   79.78%   80.34%   +0.56%     
==========================================
  Files         599      600       +1     
  Lines       66862    67012     +150     
==========================================
+ Hits        53343    53839     +496     
+ Misses      10380    10014     -366     
- Partials     3139     3159      +20     

☔ 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.

[ktgowtham]

LGTM 👍

[mihir20]

LGTM