rudderlabs · atzoum · Jul 30, 2026 · Jul 22, 2026
diff --git a/app/apphandlers/apphandlers_test.go b/app/apphandlers/apphandlers_test.go
@@ -226,7 +226,7 @@ func setDefaultEnv(t *testing.T) {
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "SourceDebugger.disableEventUploads"), "true")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "TransformationDebugger.disableTransformationStatusUploads"), "true")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.backup.enabled"), "false")
-	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.migrateDSLoopSleepDuration"), "60m")
+	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.compactionLoopSleepDuration"), "60m")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "archival.Enabled"), "false")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "Reporting.syncer.enabled"), "false")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "BatchRouter.mainLoopFreq"), "1s")

diff --git a/app/apphandlers/embeddedAppHandler.go b/app/apphandlers/embeddedAppHandler.go
@@ -208,6 +208,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 	gwWOHandle := jobsdb.NewForWrite(
 		"gw",
 		jobsdb.WithClearDB(options.ClearDB),
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw jobsdb
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),
 		jobsdb.WithNumPartitions(partitionCount),
@@ -223,6 +224,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 
 	gwROHandle := jobsdb.NewForRead(
 		"gw",
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw jobsdb
 		jobsdb.WithDSLimit(a.config.gwDSLimit),
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "Gateway.jobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(statsFactory),
@@ -265,6 +267,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 	eschRWDB := jobsdb.NewForReadWrite(
 		"esch",
 		jobsdb.WithClearDB(options.ClearDB),
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in esch jobsdb
 		jobsdb.WithDSLimit(a.config.eschDSLimit),
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(statsFactory),
@@ -276,6 +279,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 	arcRWDB := jobsdb.NewForReadWrite(
 		"arc",
 		jobsdb.WithClearDB(options.ClearDB),
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in arc jobsdb
 		jobsdb.WithDSLimit(a.config.arcDSLimit),
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(statsFactory),
@@ -291,6 +295,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 			"proc",
 			jobsdb.WithMultiConsumer(),
 			jobsdb.WithClearDB(options.ClearDB),
+			jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in proc jobsdb
 			jobsdb.WithDSLimit(a.config.procDSLimit),
 			jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
 			jobsdb.WithStats(statsFactory),

diff --git a/app/apphandlers/gatewayAppHandler.go b/app/apphandlers/gatewayAppHandler.go
@@ -104,6 +104,7 @@ func (a *gatewayApp) StartRudderCore(ctx context.Context, _ func(), options *app
 	gwWOHandle := jobsdb.NewForWrite(
 		"gw",
 		jobsdb.WithClearDB(options.ClearDB),
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw jobsdb
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "Gateway.jobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),

diff --git a/app/apphandlers/processorAppHandler.go b/app/apphandlers/processorAppHandler.go
@@ -210,6 +210,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 
 	gwROHandle := jobsdb.NewForRead(
 		"gw",
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw jobsdb
 		jobsdb.WithDSLimit(a.config.gwDSLimit),
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "Gateway.jobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(statsFactory),
@@ -250,6 +251,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 	eschRWDB := jobsdb.NewForReadWrite(
 		"esch",
 		jobsdb.WithClearDB(options.ClearDB),
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in esch jobsdb
 		jobsdb.WithDSLimit(a.config.eschDSLimit),
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),
@@ -260,6 +262,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 	arcRWDB := jobsdb.NewForReadWrite(
 		"arc",
 		jobsdb.WithClearDB(options.ClearDB),
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in arc jobsdb
 		jobsdb.WithDSLimit(a.config.arcDSLimit),
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(statsFactory),
@@ -275,6 +278,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 			"proc",
 			jobsdb.WithMultiConsumer(),
 			jobsdb.WithClearDB(options.ClearDB),
+			jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in proc jobsdb
 			jobsdb.WithDSLimit(a.config.procDSLimit),
 			jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
 			jobsdb.WithStats(statsFactory),

diff --git a/app/apphandlers/setup_partitionmigration.go b/app/apphandlers/setup_partitionmigration.go
@@ -95,6 +95,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		gwBuffRWHandle := jobsdb.NewForReadWrite(
 			"gw_buf",
 			jobsdb.WithClearDB(false),
+			jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw_buf jobsdb
 			jobsdb.WithDSLimit(config.GetReloadableIntVar(0, 1, "JobsDB.gw_buf.dsLimit", "JobsDB.dsLimit")),
 			jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.gw_buf.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
 			jobsdb.WithStats(stats),
@@ -115,6 +116,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		gwWODB = jobsdb.NewForWrite(
 			"gw",
 			jobsdb.WithClearDB(false),
+			jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw jobsdb
 			jobsdb.WithStats(stats),
 			jobsdb.WithDBHandle(dbPool),
 			jobsdb.WithPriorityPoolDB(priorityPool),
@@ -123,6 +125,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		)
 		gwBuffROHandle := jobsdb.NewForRead(
 			"gw_buf",
+			jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw_buf jobsdb
 			jobsdb.WithDSLimit(config.GetReloadableIntVar(0, 1, "JobsDB.gw_buf.dsLimit", "JobsDB.dsLimit")),
 			jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.gw_buf.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
 			jobsdb.WithStats(stats),
@@ -153,6 +156,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 	rtBuffRWHandle := jobsdb.NewForReadWrite(
 		"rt_buf",
 		jobsdb.WithClearDB(false),
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in rt_buf jobsdb
 		jobsdb.WithDSLimit(config.GetReloadableIntVar(0, 1, "JobsDB.rt_buff.dsLimit", "JobsDB.dsLimit")),
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.rt_buff.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(stats),
@@ -180,6 +184,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 	brtBuffRWHandle := jobsdb.NewForReadWrite(
 		"batch_rt_buf",
 		jobsdb.WithClearDB(false),
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in batch_rt_buf jobsdb
 		jobsdb.WithDSLimit(config.GetReloadableIntVar(0, 1, "JobsDB.batch_rt_buff.dsLimit", "JobsDB.dsLimit")),
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.batch_rt_buff.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(stats),
@@ -208,7 +213,10 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 	if procRWDB != nil {
 		procBuffRWHandle := jobsdb.NewForReadWrite(
 			"proc_buf",
+			// proc_buf doesn't need to be multi-consumer because it is used only as a temporary storage.
+			// During flush, a single consumer will move the jobs to the proc jobsdb which is multi-consumer.
 			jobsdb.WithClearDB(false),
+			jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in proc_buf jobsdb
 			jobsdb.WithDSLimit(config.GetReloadableIntVar(0, 1, "JobsDB.proc_buf.dsLimit", "JobsDB.dsLimit")),
 			jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.proc_buf.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
 			jobsdb.WithStats(stats),
@@ -347,6 +355,7 @@ func setupGatewayPartitionMigrator(ctx context.Context,
 	gwBuffWOHandle := jobsdb.NewForWrite(
 		"gw_buf",
 		jobsdb.WithClearDB(false),
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw_buf jobsdb
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.gw_buf.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(stats),
 		jobsdb.WithDBHandle(dbPool),

diff --git a/archiver/archiver_isolation_test.go b/archiver/archiver_isolation_test.go
@@ -160,7 +160,7 @@ func ArchivalScenario(
 	config.Set("SourceDebugger.disableEventUploads", true)
 	config.Set("TransformationDebugger.disableTransformationStatusUploads", true)
 	config.Set("JobsDB.backup.enabled", false)
-	config.Set("JobsDB.migrateDSLoopSleepDuration", "60m")
+	config.Set("JobsDB.compactionLoopSleepDuration", "60m")
 	config.Set("JobsDB.enableWriterQueue", false)
 	config.Set("RUDDER_TMPDIR", os.TempDir())
 	config.Set("archival.ArchiveSleepDuration", "1s")
@@ -322,7 +322,9 @@ func insertJobs(
 	configMap map[string]backendconfig.ConfigT,
 	numJobsPerSource int,
 ) (map[string][]*jobsdb.JobT, int) {
-	gwJobsDB := jobsdb.NewForWrite("gw", jobsdb.WithStats(stats.NOP))
+	gwJobsDB := jobsdb.NewForWrite("gw", jobsdb.WithStats(stats.NOP),
+		jobsdb.WithDefaultSkipStatusCompaction(true), // must match the default used by every other "gw" jobsdb construction
+	)
 	require.NoError(t, gwJobsDB.Start(), "it should be able to start the jobsdb")
 	defer gwJobsDB.Stop()
 

diff --git a/cluster/migrator/processor/targetnode/targetnode_migrator_builder_test.go b/cluster/migrator/processor/targetnode/targetnode_migrator_builder_test.go
@@ -230,3 +230,7 @@ func (m *mockBufferedJobsDB) RefreshBufferedPartitions(ctx context.Context) erro
 func (m *mockBufferedJobsDB) FlushBufferedPartitions(ctx context.Context, partitionIds []string) error {
 	return nil
 }
+
+func (m *mockBufferedJobsDB) WithStoreConsistency(ctx context.Context, fn func() error) error {
+	return fn()
+}
diff --git a/cluster/partitionbuffer/jobsdb_partition_buffer.go b/cluster/partitionbuffer/jobsdb_partition_buffer.go
@@ -31,6 +31,10 @@ type JobsDBPartitionBuffer interface {
 	RefreshBufferedPartitions(ctx context.Context) error
 	// FlushBufferedPartitions flushes the buffered data for the provided partition ids to the database and unmarks them as buffered.
 	FlushBufferedPartitions(ctx context.Context, partitionIds []string) error
+	// WithStoreConsistency runs fn while keeping Store routing consistent with concurrent flush switchovers.
+	// Callers storing into an externally-owned transaction (via WithStoreSafeTxFromTx + StoreInTx) must wrap
+	// that whole transaction with it.
+	WithStoreConsistency(ctx context.Context, fn func() error) error
 }
 
 type jobsDBPartitionBuffer struct {

diff --git a/cluster/partitionbuffer/jobsdb_partition_buffer_store.go b/cluster/partitionbuffer/jobsdb_partition_buffer_store.go
@@ -27,23 +27,21 @@ func (b *jobsDBPartitionBuffer) WithStoreSafeTx(ctx context.Context, fn func(tx
 		return ErrStoreNotSupported
 	}
 	for {
-		if !b.bufferedPartitionsMu.RTryLockWithContext(ctx) {
-			return fmt.Errorf("acquiring a buffered partitions read lock: %w", ctx.Err())
-		}
-		err := b.primaryWriteJobsDB.WithStoreSafeTx(ctx, func(tx jobsdb.StoreSafeTx) (err error) {
-			if !b.differentBufferDBs { // no need to check for stale version
+		err := b.WithStoreConsistency(ctx, func() error {
+			return b.primaryWriteJobsDB.WithStoreSafeTx(ctx, func(tx jobsdb.StoreSafeTx) (err error) {
+				if !b.differentBufferDBs { // no need to check for stale version
+					return fn(tx)
+				}
+				diff, err := b.versionDiff(ctx, tx.Tx()) // get the version difference
+				if err != nil {
+					return err
+				}
+				if diff != 0 { // stale version
+					return errStaleBufferedPartitions
+				}
 				return fn(tx)
-			}
-			diff, err := b.versionDiff(ctx, tx.Tx()) // get the version difference
-			if err != nil {
-				return err
-			}
-			if diff != 0 { // stale version
-				return errStaleBufferedPartitions
-			}
-			return fn(tx)
+			})
 		})
-		b.bufferedPartitionsMu.RUnlock()
 		if !errors.Is(err, errStaleBufferedPartitions) {
 			return err
 		}
@@ -54,6 +52,24 @@ func (b *jobsDBPartitionBuffer) WithStoreSafeTx(ctx context.Context, fn func(tx
 	}
 }
 
+// WithStoreConsistency runs fn while holding the buffered-partitions read lock, so that Store
+// routing decisions (see splitJobs) stay consistent with a concurrent flush switchover, which
+// drains and unbuffers partitions under the corresponding write lock.
+//
+// The lock must be acquired before the transaction fn stores into is opened and held until that
+// transaction commits: a buffered write only becomes visible to a switchover's drain (a separate
+// connection) once it commits, so releasing the lock any earlier lets a switchover unbuffer the
+// partition in between and orphan the job. Store/WithStoreSafeTx use this internally; callers that
+// store into an externally-owned transaction (via WithStoreSafeTxFromTx + StoreInTx) must wrap that
+// whole transaction with it themselves.
+func (b *jobsDBPartitionBuffer) WithStoreConsistency(ctx context.Context, fn func() error) error {
+	if !b.bufferedPartitionsMu.RTryLockWithContext(ctx) {
+		return fmt.Errorf("acquiring a buffered partitions read lock: %w", ctx.Err())
+	}
+	defer b.bufferedPartitionsMu.RUnlock()
+	return fn()
+}
+
 // StoreInTx stores the provided jobs into the appropriate JobsDBs based on their partition buffering status within the provided StoreSafeTx
 func (b *jobsDBPartitionBuffer) StoreInTx(ctx context.Context, tx jobsdb.StoreSafeTx, jobList []*jobsdb.JobT) error {
 	if !b.canStore {

diff --git a/enterprise/reporting/error_index/error_index_reporting.go b/enterprise/reporting/error_index/error_index_reporting.go
@@ -201,6 +201,7 @@ func (eir *ErrorIndexReporter) DatabaseSyncer(c types.SyncerConfig) types.Report
 	errIndexDB := jobsdb.NewForReadWrite(
 		"err_idx",
 		jobsdb.WithDBHandle(dbHandle),
+		jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in err_idx jobsdb
 		jobsdb.WithDSLimit(eir.conf.GetReloadableIntVar(0, 1, "Reporting.errorIndexReporting.dsLimit")),
 		jobsdb.WithConfig(eir.conf),
 		jobsdb.WithSkipMaintenanceErr(eir.conf.GetBoolVar(false, "Reporting.errorIndexReporting.skipMaintenanceError")),

diff --git a/gateway/gateway_integration_test.go b/gateway/gateway_integration_test.go
@@ -167,7 +167,7 @@ func runGateway(
 
 	config.Set("Gateway.webPort", strconv.Itoa(port))
 	config.Set("JobsDB.backup.enabled", false)
-	config.Set("JobsDB.migrateDSLoopSleepDuration", "60m")
+	config.Set("JobsDB.compactionLoopSleepDuration", "60m")
 	config.Set("RUDDER_TMPDIR", os.TempDir())
 	config.Set("recovery.storagePath", path.Join(tmpDir, "/recovery_data.json"))
 	config.Set("recovery.enabled", false)

diff --git a/integration_test/activationrecordsreporting/activation_records_reporting_test.go b/integration_test/activationrecordsreporting/activation_records_reporting_test.go
@@ -370,7 +370,7 @@ func runRudderServer(t testing.TB, ctx context.Context, cancel context.CancelFun
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "SourceDebugger.disableEventUploads"), "true")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "TransformationDebugger.disableTransformationStatusUploads"), "true")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.backup.enabled"), "false")
-	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.migrateDSLoopSleepDuration"), "60m")
+	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.compactionLoopSleepDuration"), "60m")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "archival.Enabled"), "false")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "Reporting.syncer.enabled"), "false")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "Gateway.webPort"), strconv.Itoa(tc.gwPort))

diff --git a/integration_test/backendconfigunavailability/backend_config_unavailability_test.go b/integration_test/backendconfigunavailability/backend_config_unavailability_test.go
@@ -179,7 +179,7 @@ func runRudderServer(
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "SourceDebugger.disableEventUploads"), "true")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "TransformationDebugger.disableTransformationStatusUploads"), "true")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.backup.enabled"), "false")
-	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.migrateDSLoopSleepDuration"), "60m")
+	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.compactionLoopSleepDuration"), "60m")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "archival.Enabled"), "false")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "Reporting.syncer.enabled"), "false")
 	t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "BatchRouter.pingFrequency"), "1s")

diff --git a/integration_test/bqstreamallevents/bqstreamallevents_test.go b/integration_test/bqstreamallevents/bqstreamallevents_test.go
@@ -1033,7 +1033,7 @@ func runRudderServer(
 	config.Set("SourceDebugger.disableEventUploads", true)
 	config.Set("TransformationDebugger.disableTransformationStatusUploads", true)
 	config.Set("JobsDB.backup.enabled", false)
-	config.Set("JobsDB.migrateDSLoopSleepDuration", "60m")
+	config.Set("JobsDB.compactionLoopSleepDuration", "60m")
 	config.Set("archival.Enabled", false)
 	config.Set("Reporting.syncer.enabled", false)
 	config.Set("Gateway.webPort", strconv.Itoa(port))