feat(processor): per-destination visibility at the destination-filter boundary

[mihir20]

Description

Part of Pipeline Inspector: gives per-destination visibility at the destination-filter boundary so fully- and partially-dropped events stop vanishing from the reporting funnel without a trace.

What changed

Single classification site. The per-destType destination discovery in the fan-out loop is replaced by one classifyDestinations call that splits a source's candidate destinations into available and excluded (with the exclusion reason and status code). A RETL-stamped destination_id narrows the candidate set before classification. isDestinationAvailable (the preprocess guard) and classifyDestinations agree by construction, which is pinned by tests.

Per-source destination snapshot. classifyDestinations runs per event, so it now consumes a sourceDestinations snapshot (destTypes DisplayName → definition for FilterClientIntegrations, byTypeName definition-name → enabled destinations in backend-config order) built once per source under a single config RLock, instead of taking two RLocks and re-scanning the source's destination list for every event. Only the event-dependent work (client-integration filtering, consent filtering) remains per event. Filtering semantics, ordering and RETL narrowing are unchanged.

One flag: Processor.earlyDestinationFilter (reloadable, default true).

• true (default): behavior is byte-identical to today. Events with no surviving destination drop at the preprocess guard (pre tracking-plan) with the existing source-level destination_filter filtered/298 row. No new rows are emitted anywhere.
• false: the preprocess guard is skipped, so every event reaches source hydration and tracking-plan validation, and the destination-filter decision moves to fan-out — which becomes the single drop point. Per-destination visibility comes with it unconditionally:
  • destination_enter succeeded/200 for every candidate destination (survivors and excluded alike), so destination_enter − filtered_* = user_transformer input holds per destination;
  • per-destination destination_filter rows for excluded candidates: filtered_integration/298 (integrations opt-out) and filtered_consent/297 (consent denied);
  • filtered_no_destination/298 (source-level, empty destination_id) for zero-candidate events — a source with no destinations, or a RETL event whose stamped destination is unavailable. Without this carve-out those events would disappear from the funnel entirely.

The flag is snapshotted once per batch in preprocessStage and threaded through the stage messages, so an in-flight batch never observes mixed values when the reloadable config flips.

Hard cutover, no dual-emit. With the flag false, the old source-level filtered row stops entirely — there is no intermediate state where the drop moves to fan-out but keeps the old row shape. Consumers of that row (reporting-service filteredEvents, dashboards) must be ready before the flip.

in_pu is left empty on all new rows. The field is slated for deprecation, so no chain value is computed for the rows introduced here, and the inPU values of pre-existing rows (gateway, tracking-plan, source-hydration, user-transformer) are untouched regardless of the flag.

Proc rebuild stage: disabled destinations keep flowing (review follow-up). DESTINATION_ENTER is recorded at fan-out, but a forked event whose destination got disabled before the proc rebuild stage consumed it used to be dropped there with no reporting row — leaving the reporting chain dangling. The rebuild stage now re-hydrates a disabled destination as-is and lets the event continue through the pipeline, so it reaches the router/batchrouter queue and is aborted there with the usual drain reporting ("destination is disabled"). A destination deleted from the config is still dropped at rebuild to a terminal Filtered status: without its config there is no destination type, hence no transformation to run and no router queue to store to.

New reloadable flag Processor.DestinationIsolation.dropEventsForDisabledDestAtProcRebuild (default false) opts back into the early drop for disabled destinations — useful for draining a proc-table backlog quickly after the user disables a misbehaving destination.

Behavior summary for a fully-dropped event

Processor.earlyDestinationFilter	Drop point	Rows emitted
true (default)	preprocess guard, pre-TP	old source-level filtered/298 only — identical to today
false	fan-out, post-TP	destination_enter ×N + filtered_* ×N; zero-candidate events emit filtered_no_destination/298

Side effects of flipping to false, for rollout notes: zero-destination events now incur source-hydration/TP transformer cost for sources that have those configured, and TP incoming/violation counts increase accordingly (arguably a fix — TP health no longer silently excludes events whose user opted out of every destination).

Tests

• TestClassifyDestinations: unit coverage of the classifier — integrations/consent exclusion, RETL narrowing, ordering, disabled destinations excluded from the candidate set, and an invariant subtest that len(available) > 0 agrees with isDestinationAvailable across a case table.
• TestDestinationVisibilityReporting: stage-level coverage of the fan-out reporting — enter/filter row shapes, the zero-candidate and RETL carve-outs, forked-destination enter rows, both flag states, and mid-flight flag flips on the same handle.
• TestProcRebuildStage: config-drift coverage of the rebuild stage — a disabled destination flows through re-hydrated as-is by default, is dropped as Filtered when dropEventsForDisabledDestAtProcRebuild is set, and a deleted destination is always dropped.
• integration_test/reporting_dropped_events: an end-to-end scenario driving one server through both flag states against a real Postgres reports table (preprocess row byte-compat, enter rows, partial exclusion, zero-candidate).

Linear Ticket

pipe-3286

Security

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

[mihir20]

Stack (merge bottom-up)

• feat(processor): per-destination visibility at the destination-filter boundary #7283 pipeline-inspector/02-per-destination-visibility

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 79.79%. Comparing base (f2655fb) to head (00ddf24).

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #7283      +/-   ##
==========================================
- Coverage   80.12%   79.79%   -0.34%     
==========================================
  Files         601      601              
  Lines       67217    67301      +84     
==========================================
- Hits        53857    53701     -156     
- Misses      10184    10434     +250     
+ Partials     3176     3166      -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.

[atzoum]

why does this require a new status code?

[mihir20]

It's as per the original design proposal 😅. We are distinguishing filtering at a stage with different status codes for different statuses.

[atzoum]

I am still trying to remember why we need this, I mean we are using different statuses already

[atzoum]

I guess I was influenced by the reportingv2 solution where counter labels are stable and we classify through special category codes. But not sure there is any benefit in the current design by inventing yet another status code.

[atzoum]

In practice destinations are soft deleted, never hard deleted & soft deleted destinations arrive to rudder-server as disabled

[atzoum]

what are sibling destinations?

[atzoum]

I guess I was influenced by the reportingv2 solution where counter labels are stable and we classify through special category codes. But not sure there is any benefit in the current design by inventing yet another status code.

[atzoum]

@mihir20 I think that ideally, destination-specific reports should be persisted on the same tx that writes jobs to rt & batchrt jobsdb. Not a hard requirement though