Skip to content

feat(processor): isolated destination processing pipeline - #7202

Merged
atzoum merged 1 commit into
masterfrom
feat.procPool
Jul 27, 2026
Merged

feat(processor): isolated destination processing pipeline#7202
atzoum merged 1 commit into
masterfrom
feat.procPool

Conversation

@atzoum

@atzoum atzoum commented Jul 21, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Description

Introduces the proc pool: a second processor worker pool that consumes a new intermediate proc jobsdb and runs the post-fan-out stages — rebuild → user transform → destination transform → store — isolated per destination, so a slow/failing destination no longer blocks the others. It reuses the existing processor stages as-is; only the rebuild entry stage is new.

This PR is the consumer side only. Nothing writes to the proc jobsdb yet — the gw-pool siphon that fans out and persists jobs into it is out of scope and will follow up in a separate pull request. With Processor.DestinationIsolation.enabled=false (default) the pool is a no-op.

Multi-consumer jobsdb. The proc jobsdb is multi-consumer: one job carries multiple consumer IDs (destination IDs), and each (job, consumer) pair has its own independent status lifecycle. The proc pool:

  • discovers partitions via GetDistinctConsumers (one partition = one destination),
  • picks up jobs scoped to a consumer (GetUnprocessed with Consumer), and
  • commits per-consumer statuses through the shared storeStage, redirected to write against procDB instead of gatewayDB.

Rebuild assumptions. Proc jobs persist only IDs + the event/metadata; at consume time rebuild re-hydrates Destination/Connection/Libraries/Credentials from the live backend config. Therefore:

  • Destination filtering and consent are not re-applied — they were already decided in the gw pool, and a job is only stored for the destinations that passed them.
  • A destination deleted/disabled between fan-out and consume is dropped gracefully to a terminal status. Drop is silent, i.e. it is not being reported, similar to what is happening already during the pretransform stage when we fan out events.

Linear Ticket

resolves PIPE-3048
resolves PIPE-3054

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:

@codecov

codecov Bot commented Jul 21, 2026 •
edited
Loading

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 77.97468% with 87 lines in your changes missing coverage. Please review.
✅ Project coverage is 79.68%. Comparing base (50df0f7) to head (b8b5d56).

Files with missing lines Patch % Lines
processor/proc_consumer.go 61.49% 63 Missing and 4 partials ⚠️
processor/proc_partition_worker.go 85.71% 7 Missing and 3 partials ⚠️
processor/proc_pipeline_worker.go 94.11% 3 Missing and 2 partials ⚠️
processor/processor.go 93.75% 2 Missing and 1 partial ⚠️
app/apphandlers/processorAppHandler.go 0.00% 2 Missing ⚠️
Additional details and impacted files 
@@            Coverage Diff             @@
##           master    #7202      +/-   ##
==========================================
- Coverage   79.82%   79.68%   -0.14%     
==========================================
  Files         596      599       +3     
  Lines       66446    66789     +343     
==========================================
+ Hits        53039    53221     +182     
- Misses      10280    10430     +150     
- Partials     3127     3138      +11

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

@atzoum
atzoum requested review from ktgowtham and mihir20 July 21, 2026 11:53
@atzoum atzoum mentioned this pull request Jul 22, 2026
Merged
1 task
@atzoum atzoum changed the title feat(processor): processor destination processing pipeline feat(processor): destination processing pipeline Jul 22, 2026
@atzoum atzoum changed the title feat(processor): destination processing pipeline feat(processor): isolated destination processing pipeline Jul 22, 2026
mihir20
mihir20 approved these changes Jul 27, 2026
View reviewed changes

@mihir20 mihir20 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM

@atzoum
atzoum merged commit a684641 into master Jul 27, 2026
113 of 115 checks passed
@atzoum
atzoum deleted the feat.procPool branch July 27, 2026 06:46
@atzoum atzoum mentioned this pull request Jul 30, 2026
Open
1 task
atzoum added a commit that referenced this pull request Jul 30, 2026
@atzoum
feat(processor): fork event processing for configured isolated destin…
6eb91b4 
…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>&nbsp;2&nbsp;</kbd> #7235 
- <kbd>&nbsp;1&nbsp;</kbd> #7207 👈 
<!-- GitButler Footer Boundary Bottom -->
This was referenced Aug 3, 2026
Merged
Merged
itsmihir pushed a commit that referenced this pull request Aug 3, 2026
@rudderstack-github-actions
chore: prerelease 1.82.0-rc.1 (#7242)
f1dc960 
🤖 I have created a release *beep* *boop*
---


##
[1.82.0-rc.1](v1.81.0...v1.82.0-rc.1)
(2026-08-03)


### Features

* **processor:** fork event processing for configured isolated
destinations
([#7207](#7207))
([6eb91b4](6eb91b4))
* **processor:** isolated destination processing pipeline
([#7202](#7202))
([a684641](a684641))
* **router:** detect and classify transformer-proxy contract breaches
(INT-6839)
([#7184](#7184))
([c7a15a8](c7a15a8))
* **router:** gate 296 delivered-with-warning behind per-workspace
rollout
([#7217](#7217))
([1d772a1](1d772a1))


### Bug Fixes

* apply configured client QPS and Burst settings in restConfig
([#7220](#7220))
([762bc06](762bc06))
* gcs manager allows non-service account credentials and warehouse ssl
keys can escape their containing path
([#7227](#7227))
([3e126b2](3e126b2))
* mid-run pod deaths under concurrent test runs
([#7212](#7212))
([5279b64](5279b64))
* normalize null BingAds offline conversion fields
([#7236](#7236))
([678f130](678f130))
* use correct JSON tag for adjustedConversionTime in bing-ads
([#7204](#7204))
([50df0f7](50df0f7))


### Miscellaneous

* add new readme logo
([#7200](#7200))
([ab86afd](ab86afd))
* **deps:** bump the go-deps group across 1 directory with 32 updates
([#7199](#7199))
([0eaa833](0eaa833))
* **jobsdb:** graceful shutdown of addNewDSLoop and refreshDSListLoop
([#7210](#7210))
([78e104c](78e104c))
* remove archiveInPreProcess processor toggle
([#7205](#7205))
([f471b1d](f471b1d))

---
This PR was generated with [Release
Please](https://github.com/googleapis/release-please). See
[documentation](https://github.com/googleapis/release-please#release-please).

Co-authored-by: rudderstack-github-actions[bot] <236995729+rudderstack-github-actions[bot]@users.noreply.github.com>
itsmihir pushed a commit that referenced this pull request Aug 4, 2026
@rudderstack-github-actions
chore: release 1.82.0 (#7241)
087532b 
🤖 I have created a release *beep* *boop*
---


##
[1.82.0](v1.81.0...v1.82.0)
(2026-08-03)


### Features

* **processor:** fork event processing for configured isolated
destinations
([#7207](#7207))
([6eb91b4](6eb91b4))
* **processor:** isolated destination processing pipeline
([#7202](#7202))
([a684641](a684641))
* **router:** detect and classify transformer-proxy contract breaches
(INT-6839)
([#7184](#7184))
([c7a15a8](c7a15a8))
* **router:** gate 296 delivered-with-warning behind per-workspace
rollout
([#7217](#7217))
([1d772a1](1d772a1))


### Bug Fixes

* apply configured client QPS and Burst settings in restConfig
([#7220](#7220))
([762bc06](762bc06))
* gcs manager allows non-service account credentials and warehouse ssl
keys can escape their containing path
([#7227](#7227))
([3e126b2](3e126b2))
* mid-run pod deaths under concurrent test runs
([#7212](#7212))
([5279b64](5279b64))
* normalize null BingAds offline conversion fields
([#7236](#7236))
([678f130](678f130))
* use correct JSON tag for adjustedConversionTime in bing-ads
([#7204](#7204))
([50df0f7](50df0f7))


### Miscellaneous

* add new readme logo
([#7200](#7200))
([ab86afd](ab86afd))
* **deps:** bump the go-deps group across 1 directory with 32 updates
([#7199](#7199))
([0eaa833](0eaa833))
* **jobsdb:** graceful shutdown of addNewDSLoop and refreshDSListLoop
([#7210](#7210))
([78e104c](78e104c))
* remove archiveInPreProcess processor toggle
([#7205](#7205))
([f471b1d](f471b1d))

---
This PR was generated with [Release
Please](https://github.com/googleapis/release-please). See
[documentation](https://github.com/googleapis/release-please#release-please).

Co-authored-by: rudderstack-github-actions[bot] <236995729+rudderstack-github-actions[bot]@users.noreply.github.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@ktgowtham ktgowtham ktgowtham approved these changes
@mihir20 mihir20 mihir20 approved these changes

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@atzoum @ktgowtham @mihir20