refactor(processor): enhance handling of disabled destinations during rehydration and update related logic

🔒 Scanned for secrets using gitleaks 8.28.0

Author: mihir20

processor/proc_consumer.go

@@ -167,9 +167,18 @@ func (proc *Handle) procStoreStage(partition string, pipelineIndex int, in *stor
 // re-hydrates Destination/Connection/Libraries/Credentials from live backendConfig.
 // Dest-filter and consent are NOT re-applied here — they were already decided in gw
 // pool (preprocess/fan-out) and the proc 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. Surviving events are grouped per (source,destination)
-// so the reused transform stages operate on them unchanged.
+// passed them.
+//
+// Config drift between fan-out and consume: a destination deleted from the config
+// cannot be re-hydrated (its type and config are gone, so there isn't even a router
+// queue to store to) and is dropped to a terminal status. A destination merely
+// disabled keeps flowing through the pipeline so it reaches the router/batchrouter
+// queue and is aborted there with the usual drain reporting — DESTINATION_ENTER was
+// already recorded at fan-out, so dropping here would leave the reporting chain
+// dangling. Processor.DestinationIsolation.dropEventsForDisabledDestAtProcRebuild opts back
+// into dropping disabled destinations here, e.g. to drain a proc-table backlog quickly
+// after the user disables a misbehaving destination. Surviving events are grouped per
+// (source,destination) so the reused transform stages operate on them unchanged.
 func (proc *Handle) procRebuildStage(destinationID string, in subJob) (*transformationMessage, error) { //nolint: unparam
 	s := time.Now()
 	defer func() {
@@ -183,6 +192,7 @@ func (proc *Handle) procRebuildStage(destinationID string, in subJob) (*transfor
 	statusList := make([]*jobsdb.JobStatusT, 0, len(in.subJobs))
 	var reportMetrics []*reportingtypes.PUReportedMetric
 	var totalEvents int
+	dropDisabled := proc.config.dropEventsForDisabledDestAtProcRebuild.Load()
 
 	for _, job := range in.subJobs {
 		var payload procJobPayload
@@ -194,11 +204,17 @@ func (proc *Handle) procRebuildStage(destinationID string, in subJob) (*transfor
 		totalEvents++
 		sourceID := payload.Metadata.SourceID
 
-		// Hydrate the destination from live config. Config drift: a destination
-		// deleted/disabled since fan-out is dropped gracefully to a terminal status.
-		dest, ok := proc.getEnabledDestinationByID(sourceID, destinationID)
+		// Re-hydrate the destination from live config. Config drift: a destination
+		// deleted since fan-out cannot be re-hydrated and is dropped to a terminal
+		// status; a disabled one keeps flowing (aborted with reporting at the
+		// router/batchrouter) unless dropEventsForDisabledDestAtProcRebuild is set.
+		dest, ok := proc.getDestinationByID(sourceID, destinationID)
 		if !ok {
-			statusList = append(statusList, procJobStatus(job, destinationID, jobsdb.Filtered.State, `{"reason":"destination not found or disabled"}`))
+			statusList = append(statusList, procJobStatus(job, destinationID, jobsdb.Filtered.State, `{"reason":"destination not found"}`))
+			continue
+		}
+		if !dest.Enabled && dropDisabled {
+			statusList = append(statusList, procJobStatus(job, destinationID, jobsdb.Filtered.State, `{"reason":"destination disabled"}`))
 			continue
 		}
 
@@ -259,14 +275,14 @@ func (proc *Handle) procRebuildStage(destinationID string, in subJob) (*transfor
 	}, nil
 }
 
-// getEnabledDestinationByID returns the live, enabled destination for the given
-// (source, destination) connection, or false when it no longer exists / is disabled.
-func (proc *Handle) getEnabledDestinationByID(sourceID, destinationID string) (backendconfig.DestinationT, bool) {
+// getDestinationByID returns the live destination for the given (source, destination)
+// connection — enabled or not — or false when it no longer exists in the config.
+func (proc *Handle) getDestinationByID(sourceID, destinationID string) (backendconfig.DestinationT, bool) {
 	proc.config.configSubscriberLock.RLock()
 	defer proc.config.configSubscriberLock.RUnlock()
 	for i := range proc.config.sourceIdDestinationMap[sourceID] {
 		dest := &proc.config.sourceIdDestinationMap[sourceID][i]
-		if dest.ID == destinationID && dest.Enabled {
+		if dest.ID == destinationID {
 			return *dest, true
 		}
 	}

processor/proc_consumer_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/rudderlabs/rudder-go-kit/config"
 	"github.com/rudderlabs/rudder-go-kit/jsonrs"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
@@ -26,6 +27,7 @@ func newTestProcHandle() *Handle {
 	proc.config.connectionConfigMap = map[connection]backendconfig.Connection{}
 	proc.config.workspaceLibrariesMap = map[string]backendconfig.LibrariesT{}
 	proc.config.credentialsMap = map[string][]types.Credential{}
+	proc.config.dropEventsForDisabledDestAtProcRebuild = config.SingleValueLoader(false)
 	return proc
 }
 
@@ -135,14 +137,31 @@ func TestProcRebuildStage(t *testing.T) {
 		require.Equal(t, "defID-"+dstID, ev.Metadata.DestinationDefinitionID)
 	})
 
-	t.Run("config drift: disabled destination is dropped (filtered), not paniced", func(t *testing.T) {
+	t.Run("config drift: disabled destination keeps flowing by default, to be aborted with reporting at the router", func(t *testing.T) {
 		proc := newTestProcHandle()
 		proc.config.sourceIdDestinationMap[srcID] = []backendconfig.DestinationT{testDestination(dstID, "my-webhook", false)}
 
 		job := newProcJob(1, procJobPayload{Metadata: types.Metadata{SourceID: srcID, MessageID: "msg-1"}})
 		out, err := proc.procRebuildStage(dstID, procRebuildInput(job))
 		require.NoError(t, err)
 
+		key := getKeyFromSourceAndDest(srcID, dstID)
+		require.Len(t, out.groupedEvents[key], 1)
+		require.False(t, out.groupedEvents[key][0].Destination.Enabled, "the disabled destination is re-hydrated as-is")
+		require.Len(t, out.statusList, 1)
+		require.Equal(t, jobsdb.Succeeded.State, out.statusList[0].JobState)
+		require.Equal(t, dstID, out.statusList[0].Consumer)
+	})
+
+	t.Run("config drift: disabled destination is dropped (filtered) when dropEventsForDisabledDestAtProcRebuild is set", func(t *testing.T) {
+		proc := newTestProcHandle()
+		proc.config.dropEventsForDisabledDestAtProcRebuild = config.SingleValueLoader(true)
+		proc.config.sourceIdDestinationMap[srcID] = []backendconfig.DestinationT{testDestination(dstID, "my-webhook", false)}
+
+		job := newProcJob(1, procJobPayload{Metadata: types.Metadata{SourceID: srcID, MessageID: "msg-1"}})
+		out, err := proc.procRebuildStage(dstID, procRebuildInput(job))
+		require.NoError(t, err)
+
 		require.Empty(t, out.groupedEvents)
 		require.Len(t, out.statusList, 1)
 		require.Equal(t, jobsdb.Filtered.State, out.statusList[0].JobState)

processor/processor.go

@@ -207,6 +207,8 @@ type Handle struct {
 		forkRsourcesTrackedJobs                   bool
 		reportingDedupMetricsEnabled              config.ValueLoader[bool]
 		earlyDestinationFilter                    config.ValueLoader[bool]
+
+		dropEventsForDisabledDestAtProcRebuild config.ValueLoader[bool]
 	}
 
 	drainConfig struct {
@@ -855,6 +857,9 @@ func (proc *Handle) loadReloadableConfig(defaultPayloadLimit int64, defaultMaxEv
 	proc.config.storeSamplerEnabled = proc.conf.GetReloadableBoolVar(false, "Processor.storeSamplerEnabled")
 	proc.config.reportingDedupMetricsEnabled = proc.conf.GetReloadableBoolVar(false, "Reporting.dedupMetrics.enabled")
 	proc.config.earlyDestinationFilter = proc.conf.GetReloadableBoolVar(true, "Processor.earlyDestinationFilter")
+	// Opt-in early drop at the proc rebuild stage for destinations disabled since fan-out;
+	// by default such events keep flowing so the router/batchrouter aborts them with reporting.
+	proc.config.dropEventsForDisabledDestAtProcRebuild = proc.conf.GetReloadableBoolVar(false, "Processor.DestinationIsolation.dropEventsForDisabledDestAtProcRebuild")
 }
 
 type connection struct {