feat(processor): fork event processing for rsources pipelines

processor/proc_fork.go

@@ -54,6 +54,14 @@ func (proc *Handle) shouldForkDestination(destinationID string) bool {
 	return enabled
 }
 
+// forkableEvent reports whether the given event is eligible to be forked to the
+// intermediate (proc) jobsdb at all, independent of the per-destination
+// shouldForkDestination check. rsources (retl) jobs — those carrying a SourceJobRunID —
+// stay inline by default, unless configured otherwise.
+func (proc *Handle) forkableEvent(event *types.TransformerEvent) bool {
+	return proc.config.forkRsourcesTrackedJobs || event.Metadata.SourceJobRunID == ""
+}
+
 // newForkedJob builds a single intermediate (proc) job for one source event fanned out to
 // forkedDestIDs. The payload carries the source-level message + metadata (destination is
 // re-hydrated per consumer at drain time, see procRebuildStage), the forked destination

processor/proc_fork_test.go

@@ -59,6 +59,31 @@ func TestShouldForkDestination(t *testing.T) {
 	})
 }
 
+func TestForkableEvent(t *testing.T) {
+	retlEvent := &types.TransformerEvent{
+		Metadata: types.Metadata{SourceJobRunID: "jr1"},
+	}
+	nonRetlEvent := &types.TransformerEvent{
+		Metadata: types.Metadata{SourceJobRunID: ""},
+	}
+
+	t.Run("rsources-tracked events stay inline by default", func(t *testing.T) {
+		proc := newTestProcHandle()
+		require.False(t, proc.config.forkRsourcesTrackedJobs)
+
+		require.False(t, proc.forkableEvent(retlEvent))
+		require.True(t, proc.forkableEvent(nonRetlEvent))
+	})
+
+	t.Run("rsources-tracked events fork once opted in", func(t *testing.T) {
+		proc := newTestProcHandle()
+		proc.config.forkRsourcesTrackedJobs = true
+
+		require.True(t, proc.forkableEvent(retlEvent))
+		require.True(t, proc.forkableEvent(nonRetlEvent))
+	})
+}
+
 func TestNewForkedJob(t *testing.T) {
 	proc := newTestProcHandle()
 	event := &types.TransformerEvent{

processor/processor.go

@@ -204,6 +204,7 @@ type Handle struct {
 		pythonTransformConfig                     transformerutils.PythonTransformConfig
 		userTransformationMirroringBlockedIDs     config.ValueLoader[[]string]
 		storeSamplerEnabled                       config.ValueLoader[bool]
+		forkRsourcesTrackedJobs                   bool
 	}
 
 	drainConfig struct {
@@ -827,6 +828,7 @@ func (proc *Handle) loadConfig() {
 	// GWCustomVal is used as a key in the jobsDB customval column
 	proc.config.GWCustomVal = proc.conf.GetStringVar("GW", "Gateway.CustomVal")
 	proc.config.pythonTransformConfig = transformerutils.LoadPythonTransformConfig(proc.conf)
+	proc.config.forkRsourcesTrackedJobs = proc.conf.GetBoolVar(false, "Processor.DestinationIsolation.forkRsourcesTrackedJobs")
 	proc.loadReloadableConfig(defaultPayloadLimit, defaultMaxEventsToProcess)
 }
 
@@ -2380,10 +2382,9 @@ func (proc *Handle) pretransformStage(partition string, preTrans *preTransformat
 			workspaceLibraries := proc.getWorkspaceLibraries(workspaceID)
 
 			// Destinations whose events are siphoned to the intermediate (proc) jobsdb
-			// rather than transformed inline. rsources (retl) jobs stay inline until the
-			// intermediate stage is modelled in rsources accounting.
+			// rather than transformed inline.
 			var forkedDestIDs []string
-			forkable := event.Metadata.SourceJobRunID == ""
+			forkable := proc.forkableEvent(event)
 			eventFanout := 0
 
 			for _, destType := range enabledDestTypes {
@@ -3003,6 +3004,9 @@ func (proc *Handle) storeStage(partition string, pipelineIndex int, in *storeMes
 		}
 	}
 	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)
 	statusDB := proc.statusUpdateDB(in)
 	commitStatuses := func(ctx context.Context) error {
 		return statusDB.WithUpdateSafeTx(ctx, func(tx jobsdb.UpdateSafeTx) error {