fix: worker job buffer increased memory requirements causing excessive cpu for garbage collection (#6500)

🔒 Scanned for secrets using gitleaks 8.28.0

# Description
With the recent worker buffer modifications, each worker’s channel
capacity has increased from **1K** to **10K**. Each channel stores
struct values that occupy **56 bytes per slot**, even when the channel
is empty.
```go
type workerJob struct {
    job         *jobsdb.JobT              // 8 bytes (pointer)
    parameters  *routerutils.JobParameters// 8 bytes
    assignedAt  time.Time                 // 24 bytes (2 int64s + alignment)
    drainReason string                    // 16 bytes (2 words: pointer + length)
}
```
To optimize memory usage, struct pointers are now used instead of struct
values.
This reduces the memory footprint of empty channel slots from **56 bytes
per slot** to **8 bytes per slot**.

## Additional changes

- If `noOfJobsPerChannel` > `maxNoOfJobsPerChannel`, then
`maxNoOfJobsPerChannel` will automatically increase to match
`noOfJobsPerChannel`.
This eliminates the need to manually adjust `maxNoOfJobsPerChannel` when
increasing the channel size.
- Introduced `experimentalBufferSizeMinimum` (default: **500**) to
prevent the experimental calculator from going below this threshold,
provided the worker throughput exceeds 1 job/sec.
Extremely low buffer sizes have been observed to negatively impact
overall throughput.

## Linear Ticket

resolves PIPE-2554

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

Author: atzoum

router/handle_lifecycle.go

@@ -108,6 +108,14 @@ func (rt *Handle) Setup(
 	rt.noOfWorkers = getRouterConfigInt("noOfWorkers", destType, 64)
 	rt.maxNoOfJobsPerChannel = getRouterConfigInt("maxNoOfJobsPerChannel", destType, 10000)
 	rt.noOfJobsPerChannel = getRouterConfigInt("noOfJobsPerChannel", destType, 1000)
+	if rt.noOfJobsPerChannel > rt.maxNoOfJobsPerChannel { // if noOfJobsPerChannel is more than max, set it as the new max
+		rt.logger.Warnn("noOfJobsPerChannel is more than maxNoOfJobsPerChannel, setting maxNoOfJobsPerChannel to noOfJobsPerChannel value",
+			obskit.DestinationType(rt.destType),
+			logger.NewIntField("maxNoOfJobsPerChannel", int64(rt.maxNoOfJobsPerChannel)),
+			logger.NewIntField("noOfJobsPerChannel", int64(rt.noOfJobsPerChannel)),
+		)
+		rt.maxNoOfJobsPerChannel = rt.noOfJobsPerChannel
+	}
 	// Explicitly control destination types for which we want to support batching
 	// Avoiding stale configurations still having KAFKA batching enabled to cause issues with later versions of rudder-server
 	batchingSupportedDestinations := config.GetStringSliceVar([]string{"AM"}, "Router.batchingSupportedDestinations")
@@ -336,6 +344,7 @@ func (rt *Handle) setupReloadableVars() {
 	rt.reloadableConfig.oauthV2ExpirationTimeDiff = config.GetReloadableDurationVar(5, time.Minute, getRouterConfigKeys("oauth.expirationTimeDiff", rt.destType)...)
 	rt.reloadableConfig.enableExperimentalBufferSizeCalculator = config.GetReloadableBoolVar(false, getRouterConfigKeys("enableExperimentalBufferSizeCalculator", rt.destType)...)
 	rt.reloadableConfig.experimentalBufferSizeScalingFactor = config.GetReloadableFloat64Var(2.0, getRouterConfigKeys("experimentalBufferSizeScalingFactor", rt.destType)...)
+	rt.reloadableConfig.experimentalBufferSizeMinimum = config.GetReloadableIntVar(500, 1, getRouterConfigKeys("experimentalBufferSizeMinimum", rt.destType)...)
 	rt.diagnosisTickerTime = config.GetDurationVar(60, time.Second, "Diagnostics.routerTimePeriod", "Diagnostics.routerTimePeriodInS")
 	rt.netClientTimeout = config.GetDurationVar(10, time.Second,
 		"Router."+rt.destType+".httpTimeout",

router/partition_worker.go

@@ -55,6 +55,7 @@ func newPartitionWorker(ctx context.Context, rt *Handle, partition string) *part
 					workLoopThroughput,
 					rt.reloadableConfig.experimentalBufferSizeScalingFactor,
 					rt.noOfJobsPerChannel,
+					rt.reloadableConfig.experimentalBufferSizeMinimum,
 				),
 				&workerBufferStats{
 					onceEvery:       kitsync.NewOnceEvery(5 * time.Second),

router/types.go

@@ -73,4 +73,5 @@ type reloadableConfig struct {
 	oauthV2ExpirationTimeDiff              config.ValueLoader[time.Duration]
 	enableExperimentalBufferSizeCalculator config.ValueLoader[bool]    // whether to use the experimental worker buffer size calculator or not
 	experimentalBufferSizeScalingFactor    config.ValueLoader[float64] // scaling factor to scale up the buffer size in the experimental calculator
+	experimentalBufferSizeMinimum          config.ValueLoader[int]     // minimum buffer size in the experimental calculator
 }

router/worker_batch_loop.go

@@ -13,7 +13,7 @@ type workerBatchLoop struct {
 	ctx                      context.Context                                             // context for managing the lifecycle of the loop
 	jobsBatchTimeout         config.ValueLoader[time.Duration]                           // timeout for processing jobs in a batch
 	noOfJobsToBatchInAWorker config.ValueLoader[int]                                     // maximum number of jobs to batch in a worker before processing
-	inputCh                  <-chan workerJob                                            // channel to receive jobs for processing
+	inputCh                  <-chan *workerJob                                           // channel to receive jobs for processing
 	enableBatching           bool                                                        // whether to enable batching of jobs
 	batchTransform           func(routerJobs []types.RouterJobT) []types.DestinationJobT // function to transform router jobs into destination jobs in batch mode
 	transform                func(routerJobs []types.RouterJobT) []types.DestinationJobT // function to transform router jobs into destination jobs in non-batch mode
@@ -69,7 +69,7 @@ func (wl *workerBatchLoop) runLoop() {
 				doProcessRouterJobs()
 			}
 			start := time.Now()
-			if routerJob := wl.acceptWorkerJob(workerJob); routerJob != nil {
+			if routerJob := wl.acceptWorkerJob(*workerJob); routerJob != nil {
 				routerJobs = append(routerJobs, *routerJob)
 				if wl.noOfJobsToBatchInAWorker.Load() <= len(routerJobs) {
 					doProcessRouterJobs() // process the batch if it reaches the limit

router/worker_batch_loop_test.go

@@ -21,7 +21,7 @@ import (
 func TestWorkerBatchLoop(t *testing.T) {
 	t.Run("no batching and no transform at router, acceptWorkerJob returning always nil", func(t *testing.T) {
 		// Setup
-		inputCh := make(chan workerJob, 10)
+		inputCh := make(chan *workerJob, 10)
 		batchSize := 5
 
 		var processedJobs []types.DestinationJobT
@@ -75,7 +75,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 			params := &routerutils.JobParameters{
 				TransformAt: "processor", // Not "router"
 			}
-			inputCh <- workerJob{job: job, parameters: params}
+			inputCh <- &workerJob{job: job, parameters: params}
 		}
 
 		// Wait a bit to ensure jobs are processed and some timeouts are triggered too
@@ -104,7 +104,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 
 	t.Run("no batching and transform at router, then switch to no router transform", func(t *testing.T) {
 		// Setup
-		inputCh := make(chan workerJob, 10)
+		inputCh := make(chan *workerJob, 10)
 		batchSize := 5
 
 		var processedJobs []types.DestinationJobT
@@ -184,7 +184,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 			params := &routerutils.JobParameters{
 				TransformAt: "router",
 			}
-			inputCh <- workerJob{job: job, parameters: params}
+			inputCh <- &workerJob{job: job, parameters: params}
 		}
 
 		// Send a job with transform_at != "router" - this should trigger batch processing
@@ -195,7 +195,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 		params := &routerutils.JobParameters{
 			TransformAt: "processor", // Not "router"
 		}
-		inputCh <- workerJob{job: job, parameters: params}
+		inputCh <- &workerJob{job: job, parameters: params}
 
 		// Wait a bit to ensure processing
 		time.Sleep(200 * time.Millisecond)
@@ -226,7 +226,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 
 	t.Run("with batching enabled", func(t *testing.T) {
 		// Setup
-		inputCh := make(chan workerJob, 10)
+		inputCh := make(chan *workerJob, 10)
 		batchSize := 3
 
 		var processedJobs []types.DestinationJobT
@@ -292,7 +292,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 			params := &routerutils.JobParameters{
 				TransformAt: "router",
 			}
-			inputCh <- workerJob{job: job, parameters: params}
+			inputCh <- &workerJob{job: job, parameters: params}
 		}
 
 		// Wait for batch processing
@@ -306,7 +306,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 		params := &routerutils.JobParameters{
 			TransformAt: "router",
 		}
-		inputCh <- workerJob{job: job, parameters: params}
+		inputCh <- &workerJob{job: job, parameters: params}
 
 		// Wait for timeout to trigger processing of remaining jobs
 		time.Sleep(200*time.Millisecond + 50*time.Millisecond)
@@ -338,7 +338,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 
 	t.Run("jobsBatchTimeout resets even when no jobs during timeout", func(t *testing.T) {
 		// Setup
-		inputCh := make(chan workerJob, 10)
+		inputCh := make(chan *workerJob, 10)
 		batchSize := 5
 
 		var timeoutFireCount int
@@ -400,7 +400,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 			params := &routerutils.JobParameters{
 				TransformAt: "router",
 			}
-			inputCh <- workerJob{job: job, parameters: params}
+			inputCh <- &workerJob{job: job, parameters: params}
 
 			// Wait longer than timeout to ensure timeout fires
 			time.Sleep(60 * time.Millisecond)
@@ -433,7 +433,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 
 	t.Run("batch processing when batch size is reached", func(t *testing.T) {
 		// Setup
-		inputCh := make(chan workerJob, 10)
+		inputCh := make(chan *workerJob, 10)
 		batchSize := 2 // Small batch size for testing
 
 		var processedBatches [][]types.DestinationJobT
@@ -493,7 +493,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 			params := &routerutils.JobParameters{
 				TransformAt: "router",
 			}
-			inputCh <- workerJob{job: job, parameters: params}
+			inputCh <- &workerJob{job: job, parameters: params}
 		}
 
 		// Wait for processing
@@ -508,7 +508,7 @@ func TestWorkerBatchLoop(t *testing.T) {
 			params := &routerutils.JobParameters{
 				TransformAt: "router",
 			}
-			inputCh <- workerJob{job: job, parameters: params}
+			inputCh <- &workerJob{job: job, parameters: params}
 		}
 
 		// Wait for processing

router/worker_buffer.go

@@ -13,7 +13,7 @@ import (
 type workerBuffer struct {
 	maxCapacity    int
 	targetCapacity func() int
-	jobs           chan workerJob
+	jobs           chan *workerJob
 
 	stats        *workerBufferStats
 	mu           sync.RWMutex
@@ -36,7 +36,7 @@ func newWorkerBuffer(maxCapacity int, targetCapacity func() int, stats *workerBu
 	wb := &workerBuffer{
 		maxCapacity:    maxCapacity,
 		targetCapacity: targetCapacity,
-		jobs:           make(chan workerJob, maxCapacity),
+		jobs:           make(chan *workerJob, maxCapacity),
 		stats:          stats,
 	}
 	return wb
@@ -50,13 +50,13 @@ func newSimpleWorkerBuffer(capacity int) *workerBuffer {
 	wb := &workerBuffer{
 		maxCapacity:    capacity,
 		targetCapacity: func() int { return capacity },
-		jobs:           make(chan workerJob, capacity),
+		jobs:           make(chan *workerJob, capacity),
 		stats:          nil,
 	}
 	return wb
 }
 
-func (wb *workerBuffer) Jobs() <-chan workerJob {
+func (wb *workerBuffer) Jobs() <-chan *workerJob {
 	return wb.jobs
 }
 
@@ -127,7 +127,7 @@ func (rs *reservedSlot) Use(wj workerJob) {
 	rs.wb.mu.Lock()
 	defer rs.wb.mu.Unlock()
 	rs.wb.reservations--
-	rs.wb.jobs <- wj
+	rs.wb.jobs <- &wj
 }
 
 // Release releases the reserved slot from the worker's buffer

router/worker_buffer_calculator.go

@@ -44,12 +44,13 @@ func newExperimentalBufferSizeCalculator(
 	noOfJobsToBatchInAWorker config.ValueLoader[int], // number of jobs that a worker can batch together
 	workLoopThroughput metric.SimpleMovingAverage, // sliding average of work loop throughput
 	scalingFactor config.ValueLoader[float64], // scaling factor to scale up the buffer size
+	minBufferSize config.ValueLoader[int], // minimum buffer size
 ) bufferSizeCalculator {
 	return func() int {
-		const minBufferSize = 1
+		const one = 1
 		m1 := workLoopThroughput.Load() // at least the average throughput of the work loop
 		if m1 < 1 {                     // if there is no throughput yet, the throughput is less than 1 per second, set buffer to minBufferSize
-			return minBufferSize
+			return one
 		}
 		m2 := float64(jobQueryBatchSize.Load() / noOfWorkers) // at least the average number of jobs per worker during pickup
 		m3 := float64(noOfJobsToBatchInAWorker.Load())        // at least equal to the number of jobs to batch in a worker
@@ -59,7 +60,7 @@ func newExperimentalBufferSizeCalculator(
 				// calculate the maximum of the three metrics to determine the buffer size
 				math.Max(
 					math.Max(math.Max(m1, m2), m3)*scalingFactor.Load(), // scale up to provide some buffer
-					minBufferSize, // ensure buffer size is at least minBufferSize
+					math.Max(one, float64(minBufferSize.Load())),        // ensure buffer size is at least one or the configured minimum
 				)))
 	}
 }
@@ -74,13 +75,15 @@ func newBufferSizeCalculatorSwitcher(
 	workLoopThroughput metric.SimpleMovingAverage, // sliding average of work loop throughput
 	scalingFactor config.ValueLoader[float64], // scaling factor to scale up the buffer size
 	noOfJobsPerChannel int, // number of jobs per channel
+	minBufferSize config.ValueLoader[int], // minimum buffer size
 ) bufferSizeCalculator {
 	new := newExperimentalBufferSizeCalculator(
 		jobQueryBatchSize,
 		noOfWorkers,
 		noOfJobsToBatchInAWorker,
 		workLoopThroughput,
 		scalingFactor,
+		minBufferSize,
 	)
 	legacy := newStandardBufferSizeCalculator(
 		noOfJobsToBatchInAWorker,