From f913e4fdc9feb6adbba899cc5deb5b5b5489b9a8 Mon Sep 17 00:00:00 2001 From: Mason Sharp Date: Thu, 9 Apr 2026 09:59:27 -0700 Subject: [PATCH 1/2] Prevent overlapping scheduled job runs with singleton mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scheduled jobs (repset-diff, table-diff, schema-diff) had no overlap protection. If a job took longer than its run_frequency interval, gocron would start a new instance concurrently, stacking connection pools — e.g. 4 overlapping runs × 2 nodes × 10 max_connections = 80 database connections despite a max_connections cap of 10. Add WithSingletonMode(LimitModeReschedule) so gocron skips a tick when the previous run is still in progress. Co-Authored-By: Claude Opus 4.6 (1M context) --- internal/jobs/scheduler.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/internal/jobs/scheduler.go b/internal/jobs/scheduler.go index b6ad2b5..1bf5c3a 100644 --- a/internal/jobs/scheduler.go +++ b/internal/jobs/scheduler.go @@ -69,9 +69,17 @@ func (m *Manager) Run(ctx context.Context) error { switch { case job.Cron != "": - gJob, err = m.scheduler.NewJob(gocron.CronJob(job.Cron, false), gocron.NewTask(runFn)) + gJob, err = m.scheduler.NewJob( + gocron.CronJob(job.Cron, false), + gocron.NewTask(runFn), + gocron.WithSingletonMode(gocron.LimitModeReschedule), + ) case job.Frequency > 0: - gJob, err = m.scheduler.NewJob(gocron.DurationJob(job.Frequency), gocron.NewTask(runFn)) + gJob, err = m.scheduler.NewJob( + gocron.DurationJob(job.Frequency), + gocron.NewTask(runFn), + gocron.WithSingletonMode(gocron.LimitModeReschedule), + ) default: return fmt.Errorf("scheduler: job %q requires either frequency or cron", job.Name) } From d993bd718b685bef2e14c69e7edfcc9cc89b6971 Mon Sep 17 00:00:00 2001 From: Mason Sharp Date: Thu, 9 Apr 2026 10:04:56 -0700 Subject: [PATCH 2/2] Apply max_connections cap to repset-diff and schema-diff discovery pools The discovery and metadata pools in repset-diff and schema-diff were created with empty ConnectionOptions, bypassing the max_connections setting. Also add YAML config fallback in Validate() for both commands so the CLI path is consistent with the scheduled-job and HTTP paths. In practice these pools only open 1-2 short-lived connections, so this aligns them with the table-diff pool code rather than fixing a user-facing issue. Co-Authored-By: Claude Opus 4.6 (1M context) --- internal/consistency/diff/repset_diff.go | 7 ++++++- internal/consistency/diff/schema_diff.go | 9 +++++++-- 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/internal/consistency/diff/repset_diff.go b/internal/consistency/diff/repset_diff.go index 644abb0..8eeb94c 100644 --- a/internal/consistency/diff/repset_diff.go +++ b/internal/consistency/diff/repset_diff.go @@ -27,6 +27,7 @@ import ( "github.com/pgedge/ace/db/queries" "github.com/pgedge/ace/internal/infra/db" utils "github.com/pgedge/ace/pkg/common" + "github.com/pgedge/ace/pkg/config" "github.com/pgedge/ace/pkg/logger" "github.com/pgedge/ace/pkg/taskstore" "github.com/pgedge/ace/pkg/types" @@ -88,7 +89,7 @@ func NewRepsetDiffTask() *RepsetDiffCmd { } func (c *RepsetDiffCmd) connOpts() auth.ConnectionOptions { - return auth.ConnectionOptions{} + return auth.ConnectionOptions{PoolSize: c.MaxConnections} } func (c *RepsetDiffCmd) parseSkipList() error { @@ -231,6 +232,10 @@ func (c *RepsetDiffCmd) Validate() error { if c.RepsetName == "" { return fmt.Errorf("repset name is required") } + cfg := config.Get() + if c.MaxConnections == 0 && cfg != nil { + c.MaxConnections = cfg.TableDiff.MaxConnections + } if c.MaxConnections < 0 { return fmt.Errorf("max_connections must be >= 1 (or 0 to derive from concurrency factor)") } diff --git a/internal/consistency/diff/schema_diff.go b/internal/consistency/diff/schema_diff.go index 8402dfd..ff52175 100644 --- a/internal/consistency/diff/schema_diff.go +++ b/internal/consistency/diff/schema_diff.go @@ -28,6 +28,7 @@ import ( "github.com/pgedge/ace/db/queries" "github.com/pgedge/ace/internal/infra/db" utils "github.com/pgedge/ace/pkg/common" + "github.com/pgedge/ace/pkg/config" "github.com/pgedge/ace/pkg/logger" "github.com/pgedge/ace/pkg/taskstore" "github.com/pgedge/ace/pkg/types" @@ -178,6 +179,10 @@ func (c *SchemaDiffCmd) Validate() error { return fmt.Errorf("schema-diff needs at least two nodes to compare") } + cfg := config.Get() + if c.MaxConnections == 0 && cfg != nil { + c.MaxConnections = cfg.TableDiff.MaxConnections + } if c.MaxConnections < 0 { return fmt.Errorf("max_connections must be >= 1 (or 0 to derive from concurrency factor)") } @@ -220,7 +225,7 @@ func (c *SchemaDiffCmd) RunChecks(skipValidation bool) error { } } - pool, err := auth.GetClusterNodeConnection(c.Ctx, nodeWithDBInfo, auth.ConnectionOptions{}) + pool, err := auth.GetClusterNodeConnection(c.Ctx, nodeWithDBInfo, auth.ConnectionOptions{PoolSize: c.MaxConnections}) if err != nil { return fmt.Errorf("could not connect to node %s: %w", nodeName, err) } @@ -300,7 +305,7 @@ func (task *SchemaDiffCmd) schemaObjectDiff() error { } } - pool, err := auth.GetClusterNodeConnection(task.Ctx, nodeWithDBInfo, auth.ConnectionOptions{}) + pool, err := auth.GetClusterNodeConnection(task.Ctx, nodeWithDBInfo, auth.ConnectionOptions{PoolSize: task.MaxConnections}) if err != nil { logger.Warn("could not connect to node %s: %v. Skipping.", nodeName, err) continue