feat(processor): fork event processing for rsources pipelines - #7235
Open
atzoum wants to merge 1 commit into
Open
feat(processor): fork event processing for rsources pipelines#7235atzoum wants to merge 1 commit into
atzoum wants to merge 1 commit into
Conversation
1 task
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #7235 +/- ##
==========================================
- Coverage 79.88% 79.85% -0.04%
==========================================
Files 600 600
Lines 67013 67034 +21
==========================================
- Hits 53534 53530 -4
- Misses 10335 10352 +17
- Partials 3144 3152 +8 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
atzoum
marked this pull request as draft
July 30, 2026 10:03
atzoum
added a commit
that referenced
this pull request
Jul 30, 2026
…ations (#7207) # 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% | <details> <summary>Full results</summary> | 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 | </details> ### 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 - [x] The code changed/added as part of this pull request won't create any security issues with how the software is being used. <!-- GitButler Footer Boundary Top --> --- This is **part 1 of 2 in a stack** made with GitButler: - <kbd> 2 </kbd> #7235 - <kbd> 1 </kbd> #7207 👈 <!-- GitButler Footer Boundary Bottom -->
atzoum
force-pushed
the
feat.procIsolationRsources
branch
from
July 30, 2026 19:41
8262038 to
d9d02a0
Compare
atzoum
marked this pull request as ready for review
July 30, 2026 19:41
atzoum
force-pushed
the
feat.procIsolationRsources
branch
from
August 3, 2026 07:57
d9d02a0 to
847d9b5
Compare
atzoum
force-pushed
the
feat.procIsolationRsources
branch
from
August 4, 2026 07:06
847d9b5 to
3ce5be1
Compare
mihir20
reviewed
Aug 4, 2026
| in.rsourcesStats.CollectStats(statusList) | ||
| // forkedJobs are only ever populated in the gw pool (the proc pool never re-forks), | ||
| // so this is a no-op when storeStage is invoked via procStoreStage. | ||
| in.rsourcesStats.JobsForked(in.forkedJobs) |
Contributor
There was a problem hiding this comment.
Will this mess up the stats for Rsources?
For this scenario, an event is ingested from a source connected to 2 destinations. Earlier we would have reported in = 1 and out = 1 for source-level stats. but now we will report in = 3 and out = 3
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Closes the rsources accounting gap #7207 left open: forking siphons an event to the intermediate
procjobsdb before it's transformed, but until now nothing recorded that in rsources until the proc pool eventually drained it. An arbitrarily later, unbounded window in which a retljobRunIdcould be reportedcompletedwhile its forked events were still sitting unaccounted-for inproc. That's why forking was hard-excluded for any event carrying aSourceJobRunID.The mechanism
No new stage is introduced, it piggybacks on the existing source-level
In/Outthe same way dropped-job accounting already does (JobsDropped).JobsForkedincrementsInby the number of consumers a job is forked to (e.g. one job forked to 2 destinations bumpsInby 2), committed in the same transaction as theprocwrite itself so there's no window where the fork is durable but unaccounted-for.Outis unaffected: the proc pool's ownrsources.StatsCollector(proc_partition_worker.go) already callsCollectStatsonce per (job, consumer) as each forked job drains to a terminal status. A source now only reportscompletedonce every forked consumer has drained, in addition to the existing per-destination router/batch-router accounting, which is untouched.Still opt-in
Forking retl-tracked events is gated by its own config, disabled by default:
Verification
A new integration test
TestProcessorRsourcesStatscovers both the forked and non-forked paths end to end against the rsources job-status API:inline- baseline, isolation off.retl_excluded- retl events still stay inline and the numbers matchinlineexactly.all_forked- every destination is forked, proving the forkInbalances against the proc consumer's drainedOut.mixed- a destination that's both forked and filtered by a user transformation (a proc-consumer status update with nothing ever reaching the router), and another one is both forked and aborted at the router.Each scenario asserts exact
In/Out/Failedat source and destination level,Completedvia/internal/v2/job-status, failed-records resolution back to the sentrecordIdfor aborted destinations, jobsdb drain counts, and webhook delivery counts.Linear Ticket
resolves PIPE-3055
Security