企业级 AI Agent 数据库连接池治理与慢查询自适应熔断架构实战：破解连接泄漏雪崩、长推理挂起与全表扫描锁库难题

AI Agent 数据库访问生命周期对比图解：

【错误模式：跨阶段持久持有（严重违背架构原则，极易引发池耗尽）】
用户问题输入
  │
  ├─> 向连接池借出连接 [CONN_ACQUIRED] ──┐ 持续占有底层 Socket，阻断其他任务！
  ├─> 调用大模型生成 SQL (耗时 3~10秒)   │
  ├─> 执行数据库查询 (耗时 50ms)         │ 持有期间遭遇网络超时或异常，
  ├─> 调用大模型反思总结 (耗时 5~15秒)   │ 连接永久泄漏成为僵尸 Socket！
  └─> 归还连接 [CONN_RELEASED] ──────────┘ (总持有时间超过 20 秒)

【生产级正确模式：短租约与快借快还（百信企业级治理架构）】
用户问题输入
  │
  ├─> 纯内存环境：调用大模型生成待查 SQL 与意图 (零连接占用)
  ├─> 纯内存环境：AST 语法解析与只读安全性静态校验 (零连接占用)
  ├─> 进入严格短租约作用域 [ENTER_LEASE_SCOPE]:
  │    ├─> 从只读连接池快速借出 [CONN_ACQUIRED] (租约看门狗 Watchdog 计时开始: 上限 3000ms)
  │    ├─> 会话级安全参数注入: statement_timeout = 2000ms, work_mem = 64MB
  │    ├─> 执行 SQL 查询并将结果集映射为轻量 JSON 对象
  │    └─> 立即在 finally 块中归还连接 [CONN_RELEASED] (物理持有耗时仅 15~40ms)
  ├─> 租约彻底销毁，连接回归公共池供其它并发任务复用
  │
  └─> 纯内存环境：大模型基于查询结果进行多模态总结与格式化输出

两阶段前置防慢查询执行门禁流转图：

大模型生成待执行 SQL
       │
       ▼
【第一阶段：AST 语法树与模式静态校验 (AST Static Audit)】
 ├─ 校验一：禁止任何非 SELECT 语句 (UPDATE/INSERT/DELETE/ALTER/TRUNCATE)
 ├─ 校验二：禁止包含系统保留函数 (pg_sleep, sys_eval, load_file)
 ├─ 校验三：检查 WHERE 子句完整性 (禁止无条件全量查询)
 └─ 校验四：检查多表 JOIN 是否存在缺失连接谓词的隐式笛卡尔积
       │ ──[静态规则违背]──> 直接阻断并返回语法诊断错误给 Agent
       ▼ 通过静态审查
【第二阶段：只读事务内 EXPLAIN 预检与代价评估 (Cost Gatekeeper)】
 ├─ 在临时只读事务中执行: EXPLAIN (FORMAT JSON) <Target_SQL>
 ├─ 解析执行计划 JSON 树:
 │   ├─ 检查总执行代价 Total Cost (例如上限阈值 50,000)
 │   ├─ 检查预估返回行数 Plan Rows (例如上限阈值 10,000)
 │   └─ 遍历执行节点类型: 是否在超过 10 万行的大表上触发 Seq Scan (全表扫描)
 │
 ├───[代价超标 / 发现高危全表扫描]───┐
 │                                   ▼
 │                   生成结构化反思 Prompt 阻断并回传给 Agent:
 │                   "执行计划代价超标 (Cost: 185000 > 50000)，在 orders 表上
 │                    触发全表扫描，请针对 create_time 建立过滤条件或优化索引匹配。"
 │                                   │
 │                                   ▼ (Agent 触发自愈重规划，修正 SQL)
 ▼ 预检通过 (Cost < 50,000 且无危险全表扫描)
【正式执行阶段：向只读从库发送物理查询并收集结果集】

production-agent-db-governance-gateway.ts（企业级 AI Agent 数据库连接治理网关）

/**
 * 企业级 AI Agent 数据库连接池治理与慢查询自适应熔断网关
 * 严格遵守全中文注释规范与企业高可用架构标准
 */

export interface QueryOptions {
  /** 单次查询超时硬上限 (毫秒)，默认 3000ms */
  statementTimeoutMs?: number;
  /** 单次查询最大允许返回行数限制，默认 500 行 */
  maxRows?: number;
  /** EXPLAIN 代价门禁预检的最大允许 Total Cost，默认 50000 */
  maxAllowedCost?: number;
  /** 关联的任务上下文 ID，用于全链路审计追踪 */
  traceId: string;
}

export interface QueryResult<T = unknown> {
  rows: T[];
  rowCount: number;
  executionDurationMs: number;
  costEstimate: number;
  truncated: boolean;
}

/**
 * 熔断器状态定义
 */
export enum CircuitState {
  CLOSED = 'CLOSED',       // 正常闭合，全量放行
  OPEN = 'OPEN',           // 熔断开启，短路阻断
  HALF_OPEN = 'HALF_OPEN'  // 半开试探，少量放行
}

/**
 * 模拟底层物理数据库连接接口
 */
export interface IDbConnection {
  backendPid: number;
  query<T = unknown>(sql: string, params?: unknown[]): Promise<{ rows: T[]; rowCount: number }>;
  release(): Promise<void>;
}

/**
 * 模拟只读数据库连接池接口
 */
export interface IDbPool {
  acquire(): Promise<IDbConnection>;
  getActiveCount(): number;
  getIdleCount(): number;
  getWaitingCount(): number;
}

/**
 * 基于 RAII 原则的显式短租约守卫对象
 * 实现 AsyncDisposable 规范，确保连接确定性即时归还
 */
export class DbConnectionLeaseGuard implements AsyncDisposable {
  private isReleased = false;

  constructor(
    public readonly connection: IDbConnection,
    private readonly pool: IDbPool,
    private readonly leaseTimeoutMs: number = 5000
  ) {
    // 启动看门狗定时器，防止超长租约泄漏
    setTimeout(() => {
      if (!this.isReleased) {
        console.warn(`[租约看门狗警告] 检测到超长连接租约挂起，PID: ${connection.backendPid}，触发自动告警`);
      }
    }, this.leaseTimeoutMs).unref();
  }

  /**
   * 离开作用域时由运行时自动调用的析构回收逻辑
   */
  async [Symbol.asyncDispose](): Promise<void> {
    if (!this.isReleased) {
      this.isReleased = true;
      try {
        await this.connection.release();
      } catch (err) {
        console.error(`[连接归还失败] 回收连接 PID: ${this.connection.backendPid} 异常:`, err);
      }
    }
  }
}

/**
 * 滑动窗口自适应慢查询熔断器
 */
export class AdaptiveQueryCircuitBreaker {
  private state: CircuitState = CircuitState.CLOSED;
  private recentResults: boolean[] = []; // true 代表成功，false 代表慢查询/超时失败
  private readonly windowSize: number = 50;
  private readonly failureThreshold: number = 0.3; // 失败率超过 30% 熔断
  private readonly coolingPeriodMs: number = 30000; // 冷却 30 秒
  private lastStateChangeTime: number = Date.now();
  private halfOpenPassCount: number = 0;
  private readonly requiredHalfOpenPasses: number = 3;

  public getState(): CircuitState {
    const now = Date.now();
    if (this.state === CircuitState.OPEN && now - this.lastStateChangeTime > this.coolingPeriodMs) {
      this.state = CircuitState.HALF_OPEN;
      this.halfOpenPassCount = 0;
      this.lastStateChangeTime = now;
      console.log(`[熔断器状态迁移] 冷却期结束，熔断器进入半开状态 (HALF_OPEN)，放行探针测试`);
    }
    return this.state;
  }

  public recordResult(success: boolean): void {
    const currentState = this.getState();
    if (currentState === CircuitState.HALF_OPEN) {
      if (success) {
        this.halfOpenPassCount++;
        if (this.halfOpenPassCount >= this.requiredHalfOpenPasses) {
          this.state = CircuitState.CLOSED;
          this.recentResults = [];
          this.lastStateChangeTime = Date.now();
          console.log(`[熔断器状态自愈] 连续 ${this.requiredHalfOpenPasses} 次探针执行成功，熔断器恢复为闭合正常状态 (CLOSED)`);
        }
      } else {
        this.state = CircuitState.OPEN;
        this.lastStateChangeTime = Date.now();
        console.warn(`[熔断器再次触发] 半开探针查询失败，重新进入熔断断开状态 (OPEN)`);
      }
      return;
    }

    this.recentResults.push(success);
    if (this.recentResults.length > this.windowSize) {
      this.recentResults.shift();
    }

    if (this.recentResults.length >= 20) {
      const failureCount = this.recentResults.filter(r => !r).length;
      const failureRate = failureCount / this.recentResults.length;
      if (failureRate >= this.failureThreshold) {
        this.state = CircuitState.OPEN;
        this.lastStateChangeTime = Date.now();
        console.error(`[触发自适应熔断] 最近窗口内慢查询失败率达 ${(failureRate * 100).toFixed(1)}%，熔断器强制开启 (OPEN)，开始短路保护`);
      }
    }
  }
}

/**
 * 百信企业级 AI Agent 数据库连接治理网关核心类
 */
export class AgentDatabaseGateway {
  private readonly circuitBreaker = new AdaptiveQueryCircuitBreaker();

  constructor(
    private readonly readReplicaPool: IDbPool,
    private readonly adminPool: IDbPool
  ) {}

  /**
   * 执行经由治理网关防护的 Agent 即席查询
   */
  public async executeAgentQuery<T = unknown>(
    rawSql: string,
    options: QueryOptions
  ): Promise<QueryResult<T>> {
    const startTime = Date.now();
    const timeoutMs = options.statementTimeoutMs ?? 3000;
    const maxRows = options.maxRows ?? 500;
    const maxCost = options.maxAllowedCost ?? 50000;

    // 1. 检查熔断器状态，实施短路保护
    if (this.circuitBreaker.getState() === CircuitState.OPEN) {
      throw new Error(`[慢查询熔断告警] 底层数据库处于自适应熔断保护期中，拒绝执行即席查询，请稍后重试或切换降级策略。`);
    }

    // 2. 第一阶段：AST 语法与只读性静态前置审查
    this.auditSqlSecurity(rawSql);

    // 3. 规范化并注入 LIMIT 截断防爆
    const securedSql = this.injectLimitIfMissing(rawSql, maxRows);

    // 4. 获取短租约连接并在作用域结束时自动安全归还
    let connectionPid = 0;
    try {
      // 借出只读物理连接
      const rawConn = await this.readReplicaPool.acquire();
      connectionPid = rawConn.backendPid;

      // 使用异步资源管理守卫包装连接 (在离开函数或抛出错误时确定性归还)
      await using lease = new DbConnectionLeaseGuard(rawConn, this.readReplicaPool, timeoutMs + 1000);

      // 5. 第二阶段：EXPLAIN 预检代价评估门禁
      const estimatedCost = await this.evaluateExplainCost(lease.connection, securedSql);
      if (estimatedCost > maxCost) {
        throw new Error(`[执行代价超标拦截] 预估执行代价 Total Cost: ${estimatedCost} 超过安全阈值 ${maxCost}，存在严重全表扫描风险，已安全拦截。建议增加精确索引过滤条件。`);
      }

      // 6. 注入会话级隔离与硬超时限制
      await lease.connection.query(`SET LOCAL statement_timeout = '${timeoutMs}ms'`);
      await lease.connection.query(`SET LOCAL work_mem = '64MB'`);

      // 7. 物理执行查询并设置应用层超时竞态
      const queryPromise = lease.connection.query<T>(securedSql);
      const timerPromise = new Promise<never>((_, reject) => {
        setTimeout(() => reject(new Error(`[查询执行超时] 执行超过设定的 ${timeoutMs}ms 硬上限`)), timeoutMs);
      });

      const result = await Promise.race([queryPromise, timerPromise]);
      const duration = Date.now() - startTime;

      // 标记熔断器成功样本
      this.circuitBreaker.recordResult(true);

      const isTruncated = result.rows.length >= maxRows;
      return {
        rows: result.rows,
        rowCount: result.rows.length,
        executionDurationMs: duration,
        costEstimate: estimatedCost,
        truncated: isTruncated
      };
    } catch (error: any) {
      // 记录熔断器失败样本
      this.circuitBreaker.recordResult(false);

      // 若发生超时，立即向数据库内核发起主动取消，销毁孤儿后台进程
      if (connectionPid > 0 && error.message?.includes('超时')) {
        this.cancelBackendProcess(connectionPid).catch(cancelErr => {
          console.error(`[主动取消失败] 取消 PID: ${connectionPid} 时报错:`, cancelErr);
        });
      }

      throw error;
    }
  }

  /**
   * AST 与敏感关键字前置静态审查
   */
  private auditSqlSecurity(sql: string): void {
    const trimmed = sql.trim().toLowerCase();
    // 强制只允许 SELECT 语句
    if (!trimmed.startsWith('select') && !trimmed.startsWith('with')) {
      throw new Error(`[越权语句拦截] AI Agent 仅被授予只读查询权限，禁止执行任何 DDL 或 DML 修改指令。`);
    }

    // 高危危险关键字拦截
    const blackList = ['pg_sleep', 'sys_eval', 'information_schema', 'pg_shadow', 'into outfile', 'bulk insert'];
    for (const kw of blackList) {
      if (trimmed.includes(kw)) {
        throw new Error(`[高危函数拦截] 包含未授权的系统探测关键字: "${kw}"，已被安全拦截。`);
      }
    }
  }

  /**
   * 自动追加 LIMIT 截断防爆
   */
  private injectLimitIfMissing(sql: string, maxRows: number): string {
    const trimmed = sql.trim();
    if (!/limit\s+\d+/i.test(trimmed)) {
      return `${trimmed} LIMIT ${maxRows}`;
    }
    return trimmed;
  }

  /**
   * EXPLAIN 预检获取执行计划代价值
   */
  private async evaluateExplainCost(conn: IDbConnection, sql: string): Promise<number> {
    try {
      const explainResult = await conn.query<any>(`EXPLAIN (FORMAT JSON) ${sql}`);
      const planJson = explainResult.rows[0];
      const rootPlan = Array.isArray(planJson) ? planJson[0]?.Plan : planJson?.Plan;
      return rootPlan?.['Total Cost'] ?? 0;
    } catch (err) {
      console.warn(`[EXPLAIN 预检跳过] 无法生成执行计划，按保守策略放行:`, err);
      return 0;
    }
  }

  /**
   * 双向主动切断数据库内核孤儿慢查询进程
   */
  private async cancelBackendProcess(pid: number): Promise<void> {
    console.warn(`[双向主动切断] 正在通过管理通道向数据库内核发送取消指令，终止孤儿 PID: ${pid}`);
    try {
      const adminConn = await this.adminPool.acquire();
      try {
        await adminConn.query(`SELECT pg_cancel_backend($1)`, [pid]);
        console.log(`[取消指令送达] 数据库内核已确认取消 PID: ${pid}`);
      } finally {
        await adminConn.release();
      }
    } catch (err) {
      console.error(`[执行取消异常] 向管理通道发送 pg_cancel_backend 失败:`, err);
    }
  }
}