Postgre sql connection pool

meter-simulator/README.md

@@ -252,3 +252,66 @@ MIT License - see LICENSE file for details.
 - 📖 [Utility-Protocol Documentation](../README.md)
 - 🐛 [Issues](https://github.com/Utility-Protocol/Utility-contracts/issues)
 - 💬 [Discussions](https://github.com/Utility-Protocol/Utility-contracts/discussions)
+
+## PostgreSQL Pool Health Probe and Adaptive Sizing
+
+The simulator now includes a reusable PostgreSQL pool health probe for services that depend on database-backed ingestion, telemetry, or settlement workers. The probe is intentionally dependency-light: pass any `pg.Pool`-compatible object with a `query(sql)` method and the standard pool counters (`totalCount`, `idleCount`, and `waitingCount`).
+
+### Architecture
+
+1. **Health probe** runs `SELECT 1` (or `POSTGRES_HEALTH_PROBE_SQL`) under a strict timeout.
+2. **Metrics snapshot** captures total, idle, waiting, utilization, and rolling P99 latency.
+3. **Adaptive sizing** recommends scale-up when P99 exceeds the 100ms target, utilization is high, or clients are waiting; it recommends scale-down only when utilization is low and no clients are queued.
+4. **Cooldown guard** prevents pool-size oscillation during short-lived traffic spikes.
+5. **Monitoring contract** exposes status values of `healthy`, `degraded`, and `unhealthy` for dashboards and alert rules.
+
+### Configuration
+
+| Variable | Description | Default |
+|----------|-------------|---------|
+| `POSTGRES_HEALTH_PROBE_SQL` | SQL used for liveness checks | `SELECT 1` |
+| `POSTGRES_HEALTH_TIMEOUT_MS` | Probe timeout | `75` |
+| `POSTGRES_TARGET_P99_MS` | Critical-path latency target | `100` |
+| `POSTGRES_POOL_MIN` | Minimum recommended pool size | `2` |
+| `POSTGRES_POOL_MAX` | Maximum recommended pool size | `20` |
+| `POSTGRES_SCALE_UP_THRESHOLD` | Utilization threshold for scale up | `0.8` |
+| `POSTGRES_SCALE_DOWN_THRESHOLD` | Utilization threshold for scale down | `0.25` |
+| `POSTGRES_RESIZE_COOLDOWN_MS` | Minimum time between size changes | `30000` |
+
+### Example
+
+```js
+const { Pool } = require('pg');
+const PostgresPoolHealthProbe = require('./src/postgres-pool-health');
+const config = require('./src/config');
+
+const pool = new Pool({ max: config.postgresql.maxPoolSize });
+const probe = new PostgresPoolHealthProbe(pool, {
+  timeoutMs: config.postgresql.healthTimeoutMs,
+  targetP99Ms: config.postgresql.targetP99Ms,
+  minSize: config.postgresql.minPoolSize,
+  maxSize: config.postgresql.maxPoolSize
+});
+
+setInterval(async () => {
+  const health = await probe.check();
+  console.log(JSON.stringify(health));
+}, 10000);
+```
+
+### Monitoring and Alerting
+
+Recommended production alerts:
+
+- Page when `status="unhealthy"` for 2 consecutive probes.
+- Warn when rolling P99 probe latency exceeds `POSTGRES_TARGET_P99_MS` for 5 minutes.
+- Warn when `waitingClients > 0` for 3 minutes.
+- Warn when the adaptive recommendation stays at `maxPoolSize` for 10 minutes, which indicates database capacity or query tuning work is needed.
+
+### Runbook
+
+1. Confirm the database endpoint, credentials, and network path are healthy.
+2. Check pool metrics: utilization, waiting clients, and rolling P99 latency.
+3. If waiting clients are non-zero and database CPU/IO have headroom, apply the scale-up recommendation using a blue-green or canary rollout.
+4. If the pool is at maximum size, inspect slow queries and database saturation before increasing limits.
+5. During recovery, keep canaries below 10% traffic until `healthy` status and P99 below 100ms are stable for at least 15 minutes.

meter-simulator/src/config.js

@@ -21,6 +21,18 @@ const config = {
     qos: parseInt(process.env.MQTT_QOS) || 1
   },
 
+  // PostgreSQL pool health probe and adaptive sizing
+  postgresql: {
+    healthProbeSql: process.env.POSTGRES_HEALTH_PROBE_SQL || 'SELECT 1',
+    healthTimeoutMs: parseInt(process.env.POSTGRES_HEALTH_TIMEOUT_MS) || 75,
+    targetP99Ms: parseInt(process.env.POSTGRES_TARGET_P99_MS) || 100,
+    minPoolSize: parseInt(process.env.POSTGRES_POOL_MIN) || 2,
+    maxPoolSize: parseInt(process.env.POSTGRES_POOL_MAX) || 20,
+    scaleUpThreshold: parseFloat(process.env.POSTGRES_SCALE_UP_THRESHOLD) || 0.8,
+    scaleDownThreshold: parseFloat(process.env.POSTGRES_SCALE_DOWN_THRESHOLD) || 0.25,
+    resizeCooldownMs: parseInt(process.env.POSTGRES_RESIZE_COOLDOWN_MS) || 30000
+  },
+
   // Simulation defaults
   simulation: {
     defaultInterval: 30, // seconds

meter-simulator/src/postgres-pool-health.js

@@ -0,0 +1,137 @@
+class PostgresPoolHealthProbe {
+  constructor(pool, options = {}) {
+    if (!pool || typeof pool.query !== 'function') {
+      throw new TypeError('A PostgreSQL pool with a query(sql) method is required');
+    }
+
+    this.pool = pool;
+    this.options = {
+      probeSql: options.probeSql || 'SELECT 1',
+      timeoutMs: options.timeoutMs || 75,
+      targetP99Ms: options.targetP99Ms || 100,
+      minSize: options.minSize || 2,
+      maxSize: options.maxSize || 20,
+      scaleUpThreshold: options.scaleUpThreshold || 0.8,
+      scaleDownThreshold: options.scaleDownThreshold || 0.25,
+      coolDownMs: options.coolDownMs || 30000,
+      now: options.now || (() => Date.now())
+    };
+
+    this.latencies = [];
+    this.maxSamples = options.maxSamples || 100;
+    this.currentSize = options.currentSize || this.options.minSize;
+    this.lastResizeAt = 0;
+  }
+
+  async check() {
+    const startedAt = this.options.now();
+
+    try {
+      await this._withTimeout(this.pool.query(this.options.probeSql));
+      const latencyMs = Math.max(0, this.options.now() - startedAt);
+      this._recordLatency(latencyMs);
+
+      const metrics = this.getMetrics();
+      return {
+        status: latencyMs <= this.options.targetP99Ms ? 'healthy' : 'degraded',
+        latencyMs,
+        metrics,
+        recommendation: this.recommendSize(metrics)
+      };
+    } catch (error) {
+      return {
+        status: 'unhealthy',
+        latencyMs: Math.max(0, this.options.now() - startedAt),
+        error: error.message,
+        metrics: this.getMetrics(),
+        recommendation: this.recommendSize({ utilization: 1, p99LatencyMs: this.options.targetP99Ms + 1 })
+      };
+    }
+  }
+
+  getMetrics() {
+    return {
+      totalConnections: this._poolValue('totalCount', this.currentSize),
+      idleConnections: this._poolValue('idleCount', 0),
+      waitingClients: this._poolValue('waitingCount', 0),
+      utilization: this._utilization(),
+      p99LatencyMs: this._percentile(99)
+    };
+  }
+
+  recommendSize(metrics = this.getMetrics()) {
+    const desired = this._desiredSize(metrics);
+    return {
+      currentSize: this.currentSize,
+      desiredSize: desired,
+      action: desired > this.currentSize ? 'scale_up' : desired < this.currentSize ? 'scale_down' : 'hold'
+    };
+  }
+
+  applySizing(metrics = this.getMetrics()) {
+    const now = this.options.now();
+    if (now - this.lastResizeAt < this.options.coolDownMs) {
+      return { resized: false, reason: 'cooldown', ...this.recommendSize(metrics) };
+    }
+
+    const recommendation = this.recommendSize(metrics);
+    if (recommendation.desiredSize === this.currentSize) {
+      return { resized: false, reason: 'already_optimal', ...recommendation };
+    }
+
+    this.currentSize = recommendation.desiredSize;
+    this.lastResizeAt = now;
+    return { resized: true, reason: 'applied', ...recommendation, currentSize: this.currentSize };
+  }
+
+  _desiredSize(metrics) {
+    const highLatency = metrics.p99LatencyMs > this.options.targetP99Ms;
+    const highPressure = metrics.utilization >= this.options.scaleUpThreshold || metrics.waitingClients > 0;
+    const lowPressure = metrics.utilization <= this.options.scaleDownThreshold && metrics.waitingClients === 0;
+
+    if (highLatency || highPressure) {
+      return Math.min(this.options.maxSize, Math.max(this.currentSize + 1, Math.ceil(this.currentSize * 1.25)));
+    }
+
+    if (lowPressure) {
+      return Math.max(this.options.minSize, this.currentSize - 1);
+    }
+
+    return this.currentSize;
+  }
+
+  _utilization() {
+    const total = this._poolValue('totalCount', this.currentSize);
+    if (total <= 0) return 0;
+    const idle = this._poolValue('idleCount', 0);
+    return Math.min(1, Math.max(0, (total - idle) / total));
+  }
+
+  _poolValue(field, fallback) {
+    const value = Number(this.pool[field]);
+    return Number.isFinite(value) ? value : fallback;
+  }
+
+  _recordLatency(latencyMs) {
+    this.latencies.push(latencyMs);
+    if (this.latencies.length > this.maxSamples) this.latencies.shift();
+  }
+
+  _percentile(percentile) {
+    if (this.latencies.length === 0) return 0;
+    const sorted = [...this.latencies].sort((a, b) => a - b);
+    const index = Math.ceil((percentile / 100) * sorted.length) - 1;
+    return sorted[Math.max(0, Math.min(sorted.length - 1, index))];
+  }
+
+  _withTimeout(promise) {
+    let timer;
+    const timeout = new Promise((_, reject) => {
+      timer = setTimeout(() => reject(new Error(`PostgreSQL health probe timed out after ${this.options.timeoutMs}ms`)), this.options.timeoutMs);
+    });
+
+    return Promise.race([promise, timeout]).finally(() => clearTimeout(timer));
+  }
+}
+
+module.exports = PostgresPoolHealthProbe;

meter-simulator/tests/postgres-pool-health.test.js

@@ -0,0 +1,60 @@
+const PostgresPoolHealthProbe = require('../src/postgres-pool-health');
+
+describe('PostgresPoolHealthProbe', () => {
+  test('reports healthy status and pool metrics for fast probes', async () => {
+    let now = 1000;
+    const pool = { totalCount: 10, idleCount: 3, waitingCount: 0, query: jest.fn(async () => ({ rows: [{ '?column?': 1 }] })) };
+    const probe = new PostgresPoolHealthProbe(pool, { currentSize: 10, now: () => now });
+
+    const checkPromise = probe.check();
+    now += 20;
+    const result = await checkPromise;
+
+    expect(result.status).toBe('healthy');
+    expect(result.latencyMs).toBe(20);
+    expect(result.metrics.utilization).toBe(0.7);
+    expect(result.recommendation.action).toBe('hold');
+    expect(pool.query).toHaveBeenCalledWith('SELECT 1');
+  });
+
+  test('recommends scale up when utilization is high', () => {
+    const pool = { totalCount: 10, idleCount: 1, waitingCount: 0, query: jest.fn() };
+    const probe = new PostgresPoolHealthProbe(pool, { currentSize: 10, maxSize: 12 });
+
+    expect(probe.recommendSize()).toEqual({ currentSize: 10, desiredSize: 12, action: 'scale_up' });
+  });
+
+  test('recommends scale down when utilization is low', () => {
+    const pool = { totalCount: 10, idleCount: 9, waitingCount: 0, query: jest.fn() };
+    const probe = new PostgresPoolHealthProbe(pool, { currentSize: 10, minSize: 2 });
+
+    expect(probe.recommendSize()).toEqual({ currentSize: 10, desiredSize: 9, action: 'scale_down' });
+  });
+
+  test('returns unhealthy and scale-up recommendation on probe failure', async () => {
+    const pool = { totalCount: 5, idleCount: 0, waitingCount: 2, query: jest.fn(async () => { throw new Error('connection refused'); }) };
+    const probe = new PostgresPoolHealthProbe(pool, { currentSize: 5, maxSize: 10 });
+
+    const result = await probe.check();
+
+    expect(result.status).toBe('unhealthy');
+    expect(result.error).toBe('connection refused');
+    expect(result.recommendation.action).toBe('scale_up');
+  });
+
+  test('applies sizing changes with cooldown protection', () => {
+    let now = 60000;
+    const pool = { totalCount: 10, idleCount: 0, waitingCount: 1, query: jest.fn() };
+    const probe = new PostgresPoolHealthProbe(pool, { currentSize: 10, maxSize: 20, coolDownMs: 30000, now: () => now });
+
+    const first = probe.applySizing();
+    expect(first.resized).toBe(true);
+    expect(first.currentSize).toBe(13);
+
+    pool.totalCount = 13;
+    pool.idleCount = 0;
+    const second = probe.applySizing();
+    expect(second.resized).toBe(false);
+    expect(second.reason).toBe('cooldown');
+  });
+});