Cap database connection pool size for table-diff and repset-diff

ace.yaml

@@ -28,6 +28,7 @@ table_diff:
   diff_batch_size: 1
   max_diff_batch_size: 1000
   compare_unit_size: 10000
+  max_connections: 0            # max DB connections per node (0 = derive from concurrency factor)
 
 mtree:
   cdc:

docs/api.md

@@ -33,6 +33,7 @@ Compares a table between nodes and generates a diff report.
 | `--output`             | `-o`  | Output format (`json` or `html`)                                    | json    |
 | `--nodes`              | `-n`  | Nodes to include in the diff (comma-separated, or "all")            | all     |
 | `--table-filter`       | `-F`  | `WHERE` clause expression to use while diffing tables               |         |
+| `--max-connections`    | `-m`  | Maximum database connections per node                               | derived |
 | `--override-block-size`| `-B`  | Allow block sizes outside `ace.yaml` guardrails                     | false   |
 | `--quiet`              | `-q`  | Suppress progress output                                            | false   |
 | `--debug`              | `-v`  | Enable debug logging                                                | false   |
@@ -147,6 +148,7 @@ Compares schemas across nodes in a pgEdge cluster. By default, `schema-diff` per
 | `--output`             | `-o`  | Output format (`json` or `html`)                                        | json    |
 | `--nodes`              | `-n`  | Nodes to include in the diff (comma-separated, or "all")                | all     |
 | `--table-filter`       | `-F`  | `WHERE` clause expression to use while diffing tables                   |         |
+| `--max-connections`    | `-m`  | Maximum database connections per node                                   | derived |
 | `--override-block-size`| `-B`  | Allow block sizes outside `ace.yaml` guardrails                         | false   |
 | `--quiet`              | `-q`  | Suppress progress output                                                | false   |
 | `--debug`              | `-v`  | Enable debug logging                                                    | false   |
@@ -189,6 +191,7 @@ Performs a `table-diff` on every table in a replication set and reports differen
 | `--concurrency-factor`  | `-c`  | CPU ratio for concurrency (0.0–4.0)                                          | 0.5     |
 | `--compare-unit-size`   | `-u`  | Recursive split size for mismatched blocks                                  | 10000   |
 | `--output`              | `-o`  | Per-table diff output format (`json` or `html`)                             | json    |
+| `--max-connections`     | `-m`  | Maximum database connections per node                                       | derived |
 | `--override-block-size` | `-B`  | Allow block sizes outside `ace.yaml` guardrails                             | false   |
 | `--quiet`               | `-q`  | Suppress output                                                             | false   |
 | `--debug`               | `-v`  | Enable debug logging                                                        | false   |

docs/commands/diff/repset-diff.md

@@ -25,6 +25,7 @@ Runs `table-diff` on every table in a replication set and reports differences.
 | `--concurrency-factor <float>` | `-c` | CPU ratio for concurrency (0.0–4.0). Default `0.5`. |
 | `--compare-unit-size <int>` | `-u` | Recursive split size for mismatched blocks. Default `10000`. |
 | `--output <json\|html>` | `-o` | Per-table diff report format. Default `json`. |
+| `--max-connections <int>` | `-m` | Maximum database connections per node. Caps the pool regardless of concurrency factor. | derived |
 | `--override-block-size` | `-B` | Allow block sizes outside `ace.yaml` guardrails. |
 | `--quiet` | `-q` | Suppress output | `false` |
 | `--debug` | `-v` | Debug logging | `false` |

docs/commands/diff/table-diff.md

@@ -24,6 +24,7 @@ This command compares the data in the specified table across nodes in a cluster
 | `--table-filter <WHERE>` | `-F` | Optional SQL `WHERE` clause applied on every node before hashing. |
 | `--against-origin <node>` |  | Limit the diff to rows whose `node_origin` matches this Spock node id or name (useful for failed-node recovery). |
 | `--until <timestamp>` |  | Optional commit timestamp fence (RFC3339) applied with `--against-origin` and `--table-filter`; excludes newer rows. |
+| `--max-connections <int>` | `-m` | Maximum database connections per node. When set, caps the connection pool regardless of concurrency factor. Default: derived from `--concurrency-factor`. |
 | `--override-block-size` | `-B` | Skip block-size safety checks defined in `ace.yaml`. |
 | `--quiet` | `-q` | Suppress progress output. Results still write to the diff file. |
 | `--debug` | `-v` | Enable verbose logging. |

docs/configuration.md

@@ -32,6 +32,7 @@ The [`ace.yaml` file](https://github.com/pgEdge/ace/blob/main/ace.yaml) defines
 | table_diff --> min_diff_block_size | Minimum diff block (row chunk) size. **Default: 1** |
 | table_diff --> max_diff_block_size | Maximum diff block size. **Default: 1000000** |
 | table_diff --> compare_unit_size | Unit size for smallest comparison chunk. **Default: 10000** |
+| table_diff --> max_connections | Maximum database connections per node during diff operations. When set, caps the connection pool regardless of concurrency factor. **Default: 0** (derive from concurrency factor) |
 | mtree → cdc --> slot_name | Logical decoding slot name for mtree CDC. **Default: "ace_mtree_slot"** |
 | mtree → cdc --> publication_name | Publication used for mtree CDC. **Default: "ace_mtree_pub"** |
 | mtree → cdc --> cdc_processing_timeout | CDC processing timeout (s). **Default: 30** |

docs/design/table_diff.md

@@ -162,7 +162,8 @@ flowchart LR
 ### Resource Utilisation and Tuning
 
 - **block_size**: Larger blocks reduce hash tasks and recursion but increase memory/IO per hash and slow mismatch localisation; smaller blocks do the opposite (more queries, finer locality).
-- **concurrency_factor**: CPU ratio (0.0–4.0) that scales workers relative to `NumCPU` (e.g. 0.5 on a 16-CPU host spawns 8 workers). Higher = faster hashing but more load on DB backends, network, and local CPU; can contend with other workloads and connection limits.
+- **concurrency_factor**: CPU ratio (0.0–4.0) that scales workers relative to `NumCPU` (e.g. 0.5 on a 16-CPU host spawns 8 workers). Higher = faster hashing but more load on DB backends, network, and local CPU; can contend with other workloads and connection limits. The connection pool per node is sized to match the worker count (minimum 4).
+- **max_connections**: Hard cap on the connection pool size per node. When set, overrides the concurrency-derived pool size. Useful for environments with limited `max_connections` on the database server. Workers that exceed the pool size will queue for a connection rather than fail.
 - **compare_unit_size**: Lower values push recursion deeper (more queries, smaller fetches); higher values stop earlier (fewer queries, larger fetches on mismatched ranges).
 - **max_diff_rows**: Early-exit guardrail. Lower caps keep runs short and reports small on divergent tables; raising/removing can grow memory and report size when drift is large.
 - **table_filter**: Narrows scope and cost; enables accurate `COUNT(*)` on the filtered view. Must be identical across nodes to avoid false positives.

docs/http-api.md

@@ -93,6 +93,7 @@ Request body:
 | `concurrency_factor` | float | no | CPU ratio (0.0–4.0). Defaults to `table_diff.concurrency_factor` or `0.5`. |
 | `compare_unit_size` | int | no | Defaults to `table_diff.compare_unit_size` or `10000`. |
 | `max_diff_rows` | int64 | no | Defaults to `table_diff.max_diff_rows` or `0` (no limit). |
+| `max_connections` | int | no | Max DB connections per node. Defaults to `table_diff.max_connections` or derived from concurrency factor. |
 | `table_filter` | string | no | SQL `WHERE` predicate (without `WHERE`). |
 | `override_block_size` | bool | no | Bypass block-size guardrails. |
 | `quiet` | bool | no | Suppress progress output. |
@@ -192,6 +193,7 @@ Request body:
 | `block_size` | int | no | Defaults to `table_diff.diff_block_size` or `100000`. |
 | `concurrency_factor` | float | no | CPU ratio (0.0–4.0). Defaults to `table_diff.concurrency_factor` or `0.5`. |
 | `compare_unit_size` | int | no | Defaults to `table_diff.compare_unit_size` or `10000`. |
+| `max_connections` | int | no | Max DB connections per node. Defaults to `table_diff.max_connections` or derived from concurrency factor. |
 | `output` | string | no | `json` (default) or `html`. |
 | `override_block_size` | bool | no | Bypass block-size guardrails. |
 | `quiet` | bool | no | Suppress output. |
@@ -213,6 +215,7 @@ Request body:
 | `block_size` | int | no | Defaults to `table_diff.diff_block_size` or `100000`. |
 | `concurrency_factor` | float | no | CPU ratio (0.0–4.0). Defaults to `table_diff.concurrency_factor` or `0.5`. |
 | `compare_unit_size` | int | no | Defaults to `table_diff.compare_unit_size` or `10000`. |
+| `max_connections` | int | no | Max DB connections per node. Defaults to `table_diff.max_connections` or derived from concurrency factor. |
 | `output` | string | no | `json` (default) or `html`. |
 | `override_block_size` | bool | no | Bypass block-size guardrails. |
 | `quiet` | bool | no | Suppress output. |

docs/openapi.yaml

@@ -465,6 +465,9 @@ components:
         max_diff_rows:
           type: integer
           description: Defaults to table_diff.max_diff_rows or 0 (no limit).
+        max_connections:
+          type: integer
+          description: Maximum database connections per node. Defaults to table_diff.max_connections or derived from concurrency factor.
         table_filter:
           type: string
           description: SQL WHERE predicate (without WHERE).
@@ -567,6 +570,9 @@ components:
         compare_unit_size:
           type: integer
           description: Defaults to table_diff.compare_unit_size or 10000.
+        max_connections:
+          type: integer
+          description: Maximum database connections per node. Defaults to table_diff.max_connections or derived from concurrency factor.
         output:
           type: string
           enum: [json, html]
@@ -601,6 +607,9 @@ components:
         compare_unit_size:
           type: integer
           description: Defaults to table_diff.compare_unit_size or 10000.
+        max_connections:
+          type: integer
+          description: Maximum database connections per node. Defaults to table_diff.max_connections or derived from concurrency factor.
         output:
           type: string
           enum: [json, html]

docs/performance.md

@@ -34,6 +34,9 @@ When invoking [ACE commands](commands/index.md), review the available command op
 - **`--compare-unit-size`**
     Sets the minimum block size used when ACE recursively drills into mismatched blocks. Smaller values provide more granular comparisons at the cost of additional round-trips. Default is `10000`.
 
+- **`--max-connections`**
+    Caps the number of database connections ACE opens per node. By default, the pool size is derived from `--concurrency-factor` and the number of CPUs. On machines with many cores, this can result in a large number of connections. Use `--max-connections` to set a hard upper limit, or set `table_diff.max_connections` in `ace.yaml` to apply it globally. This is especially useful when running ACE against databases with limited `max_connections` or when sharing the database with other applications.
+
 - **`--override-block-size` (`-B`)**
     Override the safety limits defined in `ace.yaml`. Use cautiously: extremely large blocks can lead to `array_agg` memory pressure.
 

internal/cli/cli.go

@@ -102,6 +102,12 @@ func SetupCLI() *cli.Command {
 			Usage:   "Override block size",
 			Value:   false,
 		},
+		&cli.IntFlag{
+			Name:    "max-connections",
+			Aliases: []string{"m"},
+			Usage:   "Maximum number of database connections per node (default: derived from concurrency factor)",
+			Value:   0,
+		},
 	}
 
 	rerunOnlyFlags := []cli.Flag{
@@ -879,6 +885,7 @@ func TableDiffCLI(cmd *cli.Command) error {
 	task.DBName = cmd.String("dbname")
 	task.BlockSize = int(blockSizeInt)
 	task.ConcurrencyFactor = cmd.Float64("concurrency-factor")
+	task.MaxConnections = cmd.Int("max-connections")
 	task.CompareUnitSize = cmd.Int("compare-unit-size")
 	task.Output = strings.ToLower(cmd.String("output"))
 	task.Nodes = cmd.String("nodes")
@@ -1350,6 +1357,7 @@ func RepsetDiffCLI(cmd *cli.Command) error {
 
 	task.BlockSize = int(blockSizeInt)
 	task.ConcurrencyFactor = cmd.Float64("concurrency-factor")
+	task.MaxConnections = cmd.Int("max-connections")
 	task.CompareUnitSize = cmd.Int("compare-unit-size")
 	task.Output = cmd.String("output")
 	task.OverrideBlockSize = cmd.Bool("override-block-size")

internal/consistency/diff/repset_diff.go

@@ -50,6 +50,7 @@ type RepsetDiffCmd struct {
 	database          types.Database
 	ConnectionPool    *pgxpool.Pool
 	ConcurrencyFactor float64
+	MaxConnections    int
 	BlockSize         int
 	CompareUnitSize   int
 	Output            string
@@ -367,6 +368,7 @@ func RepsetDiff(task *RepsetDiffCmd) (err error) {
 		tdTask.Nodes = task.Nodes
 		tdTask.QualifiedTableName = tableName
 		tdTask.ConcurrencyFactor = task.ConcurrencyFactor
+		tdTask.MaxConnections = task.MaxConnections
 		tdTask.BlockSize = task.BlockSize
 		tdTask.CompareUnitSize = task.CompareUnitSize
 		tdTask.Output = task.Output
@@ -422,6 +424,7 @@ func (task *RepsetDiffCmd) CloneForSchedule(ctx context.Context) *RepsetDiffCmd
 	clone.Quiet = task.Quiet
 	clone.BlockSize = task.BlockSize
 	clone.ConcurrencyFactor = task.ConcurrencyFactor
+	clone.MaxConnections = task.MaxConnections
 	clone.CompareUnitSize = task.CompareUnitSize
 	clone.Output = task.Output
 	clone.TableFilter = task.TableFilter

internal/consistency/diff/table_diff.go

@@ -70,6 +70,7 @@ type TableDiffTask struct {
 
 	BlockSize         int
 	ConcurrencyFactor float64
+	MaxConnections    int
 	Output            string
 	TableFilter       string
 	QuietMode         bool
@@ -1027,6 +1028,7 @@ func (t *TableDiffTask) CloneForSchedule(ctx context.Context) *TableDiffTask {
 	cloned.Nodes = t.Nodes
 	cloned.BlockSize = t.BlockSize
 	cloned.ConcurrencyFactor = t.ConcurrencyFactor
+	cloned.MaxConnections = t.MaxConnections
 	cloned.Output = t.Output
 	cloned.TableFilter = t.TableFilter
 	cloned.QuietMode = t.QuietMode
@@ -1046,8 +1048,19 @@ func (t *TableDiffTask) CloneForSchedule(ctx context.Context) *TableDiffTask {
 	return cloned
 }
 
+func (t *TableDiffTask) maxPoolSize() int {
+	poolSize := int(math.Round(float64(runtime.NumCPU()) * t.ConcurrencyFactor))
+	if poolSize < 4 {
+		poolSize = 4
+	}
+	if t.MaxConnections > 0 && t.MaxConnections < poolSize {
+		poolSize = t.MaxConnections
+	}
+	return poolSize
+}
+
 func (t *TableDiffTask) connOpts() auth.ConnectionOptions {
-	return auth.ConnectionOptions{}
+	return auth.ConnectionOptions{PoolSize: t.maxPoolSize()}
 }
 
 func (t *TableDiffTask) CheckColumnSize() error {

internal/jobs/config.go

@@ -109,6 +109,9 @@ func buildTableDiffJob(cfg *config.Config, def config.JobDef, spec scheduleSpec)
 	if v := intArg(def.Args, "max_diff_rows", 0); v > 0 {
 		base.MaxDiffRows = int64(v)
 	}
+	if v := intArg(def.Args, "max_connections", 0); v > 0 {
+		base.MaxConnections = v
+	}
 	if out := stringArg(def.Args, "output"); out != "" {
 		base.Output = out
 	}
@@ -133,6 +136,9 @@ func buildTableDiffJob(cfg *config.Config, def config.JobDef, spec scheduleSpec)
 	if base.CompareUnitSize == 0 && cfg.TableDiff.CompareUnitSize > 0 {
 		base.CompareUnitSize = cfg.TableDiff.CompareUnitSize
 	}
+	if base.MaxConnections == 0 && cfg.TableDiff.MaxConnections > 0 {
+		base.MaxConnections = cfg.TableDiff.MaxConnections
+	}
 
 	return Job{
 		Name:       jobName(def, "table-diff", base.QualifiedTableName),
@@ -252,6 +258,9 @@ func buildRepsetDiffJob(cfg *config.Config, def config.JobDef, spec scheduleSpec
 	if out := stringArg(def.Args, "output"); out != "" {
 		base.Output = out
 	}
+	if v := intArg(def.Args, "max_connections", 0); v > 0 {
+		base.MaxConnections = v
+	}
 	base.TableFilter = stringArg(def.Args, "table_filter")
 	base.OverrideBlockSize = boolArg(def.Args, "override_block_size", base.OverrideBlockSize)
 	base.SkipDBUpdate = boolArg(def.Args, "skip_db_update", base.SkipDBUpdate)
@@ -272,6 +281,9 @@ func buildRepsetDiffJob(cfg *config.Config, def config.JobDef, spec scheduleSpec
 	if base.CompareUnitSize == 0 && cfg.TableDiff.CompareUnitSize > 0 {
 		base.CompareUnitSize = cfg.TableDiff.CompareUnitSize
 	}
+	if base.MaxConnections == 0 && cfg.TableDiff.MaxConnections > 0 {
+		base.MaxConnections = cfg.TableDiff.MaxConnections
+	}
 
 	return Job{
 		Name:       jobName(def, "repset-diff", base.RepsetName),

pkg/config/config.go

@@ -54,6 +54,7 @@ type DiffConfig struct {
 	BatchSize         int     `yaml:"diff_batch_size"`
 	MaxBatchSize      int     `yaml:"max_diff_batch_size"`
 	CompareUnitSize   int     `yaml:"compare_unit_size"`
+	MaxConnections    int     `yaml:"max_connections"`
 }
 
 type MTreeConfig struct {