feat(jobsdb): non-blocking dataset compaction (#6967)

🔒 Scanned for secrets using gitleaks 8.30.1

# Description

Introduces a non-blocking compaction flow for the jobsdb migration loop,
gated behind the `nonBlockingCompaction` flag. When enabled it
dramatically shrinks the lock window taken during dataset migration:

- **`dsMigrationLock` is no longer taken.** Concurrent readers
(`getJobs`, `GetPileUpCounts`, `GetDistinctParameterValues`) are not
blocked by an in-flight compaction.
- **Caveat:** an `UpdateJobStatus` call that targets a source dataset
*while it is being compacted* will block at the PG level on the
`EXCLUSIVE` lock held against that source's status table, until the
compaction TX commits — bounded by the per-source `COPY` duration. After
commit, late writers landing on the old status table are fenced by the
readonly trigger and routed to the new destination via the existing
`ErrStaleDsList` retry path.

Compared to the legacy path, the writer side of `dsListLock` is reduced
from "entire migration TX, including bulk `COPY` of all non-terminal
jobs and `DROP TABLE` of every source" down to "`COMMIT` + a single
`getDSList` + a `MIN/MAX` scan of the new destination".

## New maintenance pool

Introducing a new maintenance pool which can be used by maintenance
operations, such as:

- adding new dataset
- refreshing dataset list
- compacting datasets
- compacting job status tables

This pool helps closing a connection-pool deadlock vector where jobsdb
readers and writers could fully exhaust the pool, blocking an active
maintenance goroutine while it still held a mutex (e.g. post-commit
compaction which requires acquiring a new connection to refresh the
dataset list). In addition, all maintenance goroutines were updated to
ensure they require no more than a single connection at any time,
eliminating pool-related deadlock risks caused by previously nested
connection usage.
If no maintenance pool is injected, the calls fall back to `dbHandle`
(backwards compatibility).

## Lock stats

`dsListLock` and `dsMigrationLock` now emit timing metrics, tagged by
lock type (`read`/`write`) and whether the acquisition was async:

| Metric | Description |
|---|---|
| `jobsdb_lock_wait_time` | Time spent waiting to acquire the lock |
| `jobsdb_lock_time` | Time the lock was held (excluding wait) |
| `jobsdb_lock_total_time` | End-to-end time (wait + hold) |

All metrics carry a `name` tag so lock contention can be tracked per
jobsdb instance.

## Flags

- `JobsDB.<prefix>.nonBlockingCompaction` (default `false`) — gates the
new flow. When off, `doMigrateDS` falls back to the legacy in-TX
migrate+drop path.
- `JobsDB.<prefix>.getJobsRetryOnCompaction` (default `true`) — gates
the `getJobs` snapshot revalidation. When on, if a dataset that
`getJobs` queried was compacted mid-read, the call returns
`ErrStaleDsList` and is retried against the freshly published list.
**Why:** while `getJobs` is reading the source's status table, an
`UpdateJobStatus` for one of the same jobs could commit against the new
dataset. Without the retry, `getJobs` would miss that status update,
risking out-of-order processing downstream. This option has no effect
unless `nonBlockingCompaction` is also on.

## Testing
#6979 runs all tests with non-blocking partition migration enabled

## Linear Ticket

resolves PIPE-2997

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

<!-- GitButler Footer Boundary Top -->
---
This is **part 3 of 4 in a stack** made with GitButler:
- <kbd>&nbsp;4&nbsp;</kbd> #6979 
- <kbd>&nbsp;3&nbsp;</kbd> #6967 👈 
- <kbd>&nbsp;2&nbsp;</kbd> #6962 
- <kbd>&nbsp;1&nbsp;</kbd> #6963 
<!-- GitButler Footer Boundary Bottom -->

Author: atzoum

app/apphandlers/embeddedAppHandler.go

@@ -143,8 +143,9 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 	})
 
 	var (
-		jobsdbPool   *sql.DB
-		priorityPool *sql.DB
+		jobsdbPool      *sql.DB
+		priorityPool    *sql.DB
+		maintenancePool *sql.DB
 	)
 	if config.GetBoolVar(true, "DB.embedded.Pool.enabled", "DB.Pool.enabled") {
 		jobsdbPool, err = misc.NewDatabaseConnectionPool(ctx, "embedded", misc.DatabaseConnectionPoolConfig{
@@ -172,6 +173,19 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 		}
 		defer priorityPool.Close()
 	}
+	if config.GetBoolVar(true, "DB.embedded.MaintenancePool.enabled", "DB.MaintenancePool.enabled") {
+		maintenancePool, err = misc.NewDatabaseConnectionPool(ctx, "emp", misc.DatabaseConnectionPoolConfig{
+			MaxOpenConns:    config.GetReloadableIntVar(20, 1, "DB.embedded.MaintenancePool.maxOpenConnections", "DB.MaintenancePool.maxOpenConnections"),
+			MaxIdleConns:    config.GetReloadableIntVar(2, 1, "DB.embedded.MaintenancePool.maxIdleConnections", "DB.MaintenancePool.maxIdleConnections"),
+			ConnMaxIdleTime: config.GetReloadableDurationVar(15, time.Minute, "DB.embedded.MaintenancePool.maxIdleTime", "DB.MaintenancePool.maxIdleTime"),
+			ConnMaxLifetime: config.GetReloadableDurationVar(0, time.Second, "DB.embedded.MaintenancePool.maxConnLifetime", "DB.MaintenancePool.maxConnLifetime"),
+			UpdateInterval:  config.GetDurationVar(60, time.Second, "DB.embedded.MaintenancePool.updateInterval", "DB.MaintenancePool.updateInterval"),
+		}, config, statsFactory)
+		if err != nil {
+			return err
+		}
+		defer maintenancePool.Close()
+	}
 	partitionCount := config.GetIntVar(0, 1, "JobsDB.partitionCount")
 
 	pendingEventsRegistry := rmetrics.NewPendingEventsRegistry()
@@ -186,6 +200,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 		jobsdb.WithDBHandle(jobsdbPool),
 		jobsdb.WithNumPartitions(partitionCount),
 		jobsdb.WithPriorityPoolDB(priorityPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 	)
 	defer gwWOHandle.Close()
 	if err = gwWOHandle.Start(); err != nil {
@@ -201,6 +216,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),
 		jobsdb.WithPriorityPoolDB(priorityPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 		jobsdb.WithNumPartitions(partitionCount),
 	)
 	defer gwROHandle.Close()
@@ -214,6 +230,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),
 		jobsdb.WithPriorityPoolDB(priorityPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 		jobsdb.WithNumPartitions(partitionCount),
 	)
 	defer rtRWHandle.Close()
@@ -227,6 +244,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),
 		jobsdb.WithPriorityPoolDB(priorityPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 		jobsdb.WithNumPartitions(partitionCount),
 	)
 	defer brtRWHandle.Close()
@@ -239,6 +257,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 	)
 	defer eschRWDB.Close()
 
@@ -250,6 +269,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithJobMaxAge(config.GetReloadableDurationVar(24, time.Hour, "archival.jobRetention")),
 		jobsdb.WithDBHandle(jobsdbPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 	)
 	defer arcRWDB.Close()
 
@@ -271,7 +291,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 	}
 
 	// setup partition migrator
-	ppmSetup, err := setupProcessorPartitionMigrator(ctx, shutdownFn, jobsdbPool, priorityPool,
+	ppmSetup, err := setupProcessorPartitionMigrator(ctx, shutdownFn, jobsdbPool, priorityPool, maintenancePool,
 		config, statsFactory,
 		gwRODB, gwWODB,
 		rtRWDB, brtRWDB,

app/apphandlers/gatewayAppHandler.go

@@ -69,7 +69,10 @@ func (a *gatewayApp) StartRudderCore(ctx context.Context, _ func(), options *app
 	}
 	defer sourceHandle.Stop()
 
-	var jobsdbPool *sql.DB
+	var (
+		jobsdbPool      *sql.DB
+		maintenancePool *sql.DB
+	)
 	if config.GetBoolVar(true, "DB.gateway.Pool.enabled", "DB.Pool.enabled") {
 		jobsdbPool, err = misc.NewDatabaseConnectionPool(ctx, "gateway", misc.DatabaseConnectionPoolConfig{
 			MaxOpenConns:    config.GetReloadableIntVar(20, 1, "DB.gateway.Pool.maxOpenConnections", "DB.Pool.maxOpenConnections"),
@@ -83,6 +86,19 @@ func (a *gatewayApp) StartRudderCore(ctx context.Context, _ func(), options *app
 		}
 		defer jobsdbPool.Close()
 	}
+	if config.GetBoolVar(true, "DB.gateway.MaintenancePool.enabled", "DB.MaintenancePool.enabled") {
+		maintenancePool, err = misc.NewDatabaseConnectionPool(ctx, "gmp", misc.DatabaseConnectionPoolConfig{
+			MaxOpenConns:    config.GetReloadableIntVar(6, 1, "DB.gateway.MaintenancePool.maxOpenConnections", "DB.MaintenancePool.maxOpenConnections"),
+			MaxIdleConns:    config.GetReloadableIntVar(1, 1, "DB.gateway.MaintenancePool.maxIdleConnections", "DB.MaintenancePool.maxIdleConnections"),
+			ConnMaxIdleTime: config.GetReloadableDurationVar(15, time.Minute, "DB.gateway.MaintenancePool.maxIdleTime", "DB.MaintenancePool.maxIdleTime"),
+			ConnMaxLifetime: config.GetReloadableDurationVar(0, time.Second, "DB.gateway.MaintenancePool.maxConnLifetime", "DB.MaintenancePool.maxConnLifetime"),
+			UpdateInterval:  config.GetDurationVar(60, time.Second, "DB.gateway.MaintenancePool.updateInterval", "DB.MaintenancePool.updateInterval"),
+		}, config, statsFactory)
+		if err != nil {
+			return err
+		}
+		defer maintenancePool.Close()
+	}
 	partitionCount := config.GetIntVar(0, 1, "JobsDB.partitionCount")
 
 	gwWOHandle := jobsdb.NewForWrite(
@@ -91,6 +107,7 @@ func (a *gatewayApp) StartRudderCore(ctx context.Context, _ func(), options *app
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "Gateway.jobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 		jobsdb.WithNumPartitions(partitionCount),
 	)
 	defer gwWOHandle.Close()
@@ -112,7 +129,7 @@ func (a *gatewayApp) StartRudderCore(ctx context.Context, _ func(), options *app
 	if err != nil {
 		return fmt.Errorf("resolving mode provider: %w", err)
 	}
-	partitionMigrator, gwDB, err := setupGatewayPartitionMigrator(ctx, jobsdbPool, config, statsFactory, gwWODB, modeProvider.EtcdClient)
+	partitionMigrator, gwDB, err := setupGatewayPartitionMigrator(ctx, jobsdbPool, maintenancePool, config, statsFactory, gwWODB, modeProvider.EtcdClient)
 	if err != nil {
 		return fmt.Errorf("setting up partition migrator: %w", err)
 	}

app/apphandlers/processorAppHandler.go

@@ -150,8 +150,9 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 	})
 
 	var (
-		jobsdbPool   *sql.DB
-		priorityPool *sql.DB
+		jobsdbPool      *sql.DB
+		priorityPool    *sql.DB
+		maintenancePool *sql.DB
 	)
 	if config.GetBoolVar(true, "DB.processor.Pool.enabled", "DB.Pool.enabled") {
 		jobsdbPool, err = misc.NewDatabaseConnectionPool(ctx, "processor", misc.DatabaseConnectionPoolConfig{
@@ -179,6 +180,19 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 		}
 		defer priorityPool.Close()
 	}
+	if config.GetBoolVar(true, "DB.processor.MaintenancePool.enabled", "DB.MaintenancePool.enabled") {
+		maintenancePool, err = misc.NewDatabaseConnectionPool(ctx, "pmp", misc.DatabaseConnectionPoolConfig{
+			MaxOpenConns:    config.GetReloadableIntVar(20, 1, "DB.processor.MaintenancePool.maxOpenConnections", "DB.MaintenancePool.maxOpenConnections"),
+			MaxIdleConns:    config.GetReloadableIntVar(2, 1, "DB.processor.MaintenancePool.maxIdleConnections", "DB.MaintenancePool.maxIdleConnections"),
+			ConnMaxIdleTime: config.GetReloadableDurationVar(15, time.Minute, "DB.processor.MaintenancePool.maxIdleTime", "DB.MaintenancePool.maxIdleTime"),
+			ConnMaxLifetime: config.GetReloadableDurationVar(0, time.Second, "DB.processor.MaintenancePool.maxConnLifetime", "DB.MaintenancePool.maxConnLifetime"),
+			UpdateInterval:  config.GetDurationVar(60, time.Second, "DB.processor.MaintenancePool.updateInterval", "DB.MaintenancePool.updateInterval"),
+		}, config, statsFactory)
+		if err != nil {
+			return err
+		}
+		defer maintenancePool.Close()
+	}
 	partitionCount := config.GetIntVar(0, 1, "JobsDB.partitionCount")
 	pendingEventsRegistry := rmetrics.NewPendingEventsRegistry()
 
@@ -189,6 +203,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),
 		jobsdb.WithPriorityPoolDB(priorityPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 		jobsdb.WithNumPartitions(partitionCount),
 	)
 	defer gwROHandle.Close()
@@ -201,6 +216,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),
 		jobsdb.WithPriorityPoolDB(priorityPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 		jobsdb.WithNumPartitions(partitionCount),
 	)
 	defer rtRWHandle.Close()
@@ -214,6 +230,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),
 		jobsdb.WithPriorityPoolDB(priorityPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 		jobsdb.WithNumPartitions(partitionCount),
 	)
 	defer brtRWHandle.Close()
@@ -224,6 +241,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 		jobsdb.WithDSLimit(a.config.eschDSLimit),
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(jobsdbPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 	)
 	defer eschRWDB.Close()
 
@@ -235,6 +253,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithJobMaxAge(config.GetReloadableDurationVar(24, time.Hour, "archival.jobRetention")),
 		jobsdb.WithDBHandle(jobsdbPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 	)
 	defer arcRWDB.Close()
 
@@ -256,7 +275,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 	}
 
 	// setup partition migrator
-	ppmSetup, err := setupProcessorPartitionMigrator(ctx, shutdownFn, jobsdbPool, priorityPool,
+	ppmSetup, err := setupProcessorPartitionMigrator(ctx, shutdownFn, jobsdbPool, priorityPool, maintenancePool,
 		config, statsFactory,
 		gwRODB, nil,
 		rtRWDB, brtRWDB,

app/apphandlers/setup_partitionmigration.go

@@ -41,6 +41,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 	shutdownFn func(), // called to initiate shutdown of the app
 	dbPool *sql.DB, // database handle
 	priorityPool *sql.DB, // priority database handle
+	maintenancePool *sql.DB, // maintenance database handle (may be nil; jobsdb falls back to dbPool)
 	config *config.Config, stats stats.Stats,
 	gwRODB, gwWODB, // gateway reader and writer jobsDB handles. if gwWODB is nil, gwRODB is used for reading and a new writer gw DB is created internally
 	rtRWDB, brtRWDB jobsdb.JobsDB,
@@ -95,6 +96,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 			jobsdb.WithStats(stats),
 			jobsdb.WithDBHandle(dbPool),
 			jobsdb.WithPriorityPoolDB(priorityPool),
+			jobsdb.WithMaintenancePoolDB(maintenancePool),
 		)
 		if err := gwBuffRWHandle.Start(); err != nil {
 			return ppmSetup, fmt.Errorf("starting gw buffer jobsdb handle: %w", err)
@@ -112,6 +114,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 			jobsdb.WithStats(stats),
 			jobsdb.WithDBHandle(dbPool),
 			jobsdb.WithPriorityPoolDB(priorityPool),
+			jobsdb.WithMaintenancePoolDB(maintenancePool),
 			jobsdb.WithNumPartitions(partitionCount),
 		)
 		gwBuffROHandle := jobsdb.NewForRead(
@@ -121,6 +124,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 			jobsdb.WithStats(stats),
 			jobsdb.WithDBHandle(dbPool),
 			jobsdb.WithPriorityPoolDB(priorityPool),
+			jobsdb.WithMaintenancePoolDB(maintenancePool),
 		)
 		gwSetupOpt = partitionbuffer.WithReaderOnlyAndFlushJobsDBs(gwRODB, gwBuffROHandle, gwWODB)
 	}
@@ -150,6 +154,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		jobsdb.WithStats(stats),
 		jobsdb.WithDBHandle(dbPool),
 		jobsdb.WithPriorityPoolDB(priorityPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 	)
 	rtPartitionBuffer, err := partitionbuffer.NewJobsDBPartitionBuffer(ctx,
 		partitionbuffer.WithReadWriteJobsDBs(rtRWDB, rtBuffRWHandle),
@@ -176,6 +181,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		jobsdb.WithStats(stats),
 		jobsdb.WithDBHandle(dbPool),
 		jobsdb.WithPriorityPoolDB(priorityPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 	)
 	brtPartitionBuffer, err := partitionbuffer.NewJobsDBPartitionBuffer(ctx,
 		partitionbuffer.WithReadWriteJobsDBs(brtRWDB, brtBuffRWHandle),
@@ -272,6 +278,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 
 func setupGatewayPartitionMigrator(ctx context.Context,
 	dbPool *sql.DB, // database handle pool
+	maintenancePool *sql.DB, // maintenance database handle (may be nil; jobsdb falls back to dbPool)
 	config *config.Config, stats stats.Stats,
 	gwWODB jobsdb.JobsDB,
 	etcdClientProvider func() (etcdclient.Client, error),
@@ -291,6 +298,7 @@ func setupGatewayPartitionMigrator(ctx context.Context,
 		jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.gw_buf.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
 		jobsdb.WithStats(stats),
 		jobsdb.WithDBHandle(dbPool),
+		jobsdb.WithMaintenancePoolDB(maintenancePool),
 	)
 	if err := gwBuffWOHandle.Start(); err != nil {
 		return nil, nil, fmt.Errorf("starting gw buffer jobsdb handle: %w", err)

jobsdb/admin.go

@@ -152,7 +152,7 @@ func (jd *Handle) doCleanup(ctx context.Context, l lock.LockToken) error {
 	{
 		deleteStmt := "DELETE FROM %s_journal WHERE start_time < NOW() - INTERVAL '%d DAY'"
 		var journalEntryCount int64
-		res, err := jd.getDB(ctx).ExecContext(
+		res, err := jd.maintenanceDB().ExecContext(
 			ctx,
 			fmt.Sprintf(
 				deleteStmt,
@@ -180,7 +180,7 @@ func (jd *Handle) abortOldJobs(ctx context.Context, dsList []dataSetT) error {
 	maxAgeStatusResponse := `{"reason": "job max age exceeded"}`
 	maxAge := jd.conf.jobMaxAge.Load()
 	for _, ds := range dsList {
-		res, err := jd.getDB(ctx).ExecContext(
+		res, err := jd.maintenanceDB().ExecContext(
 			ctx,
 			fmt.Sprintf(
 				`INSERT INTO %[1]q (job_id, job_state, error_response)