Connection に query_timeout_ms の lazy apply を追加 (Phase 5-1 サブ D-2)

logicraft · logicraft · commit 74bb6da7709a · 2026-05-04T17:03:19.000+09:00
- Pool config に `query_timeout_ms` を追加 (PoolConfig.queryTimeoutMs + Resolver バリデーション)
- Connection が初回クエリ前に dialect 検出後 SET SESSION を 1 回発行
  - MySQL:   SET SESSION max_execution_time = N (ms 単位)
  - MariaDB: SET SESSION max_statement_time = N/1000 (3 桁小数の秒単位)
- ConnectionManager の primary/replica 経路で applyQueryTimeout を呼び、
  probeReplicas は対象外（短命 probe で SET SESSION の round trip を発生させない）
- SET SESSION 失敗時は logger があれば error ログを出力して伝搬
diff --git a/src/Sloop/Database/Config/ConnectionConfigResolver.php b/src/Sloop/Database/Config/ConnectionConfigResolver.php
@@ -63,6 +63,7 @@ final class ConnectionConfigResolver
         'log_bindings',
         'log_all_queries',
         'slow_query_threshold_ms',
+        'query_timeout_ms',
     ];
 
     /**
@@ -197,6 +198,7 @@ public static function validatePool(string $name, array $config): PoolConfig
         $logBindings     = self::extractOptionalBool($name, $config, 'log_bindings') ?? self::DEFAULT_LOG_BINDINGS;
         $logAllQueries   = self::extractOptionalBool($name, $config, 'log_all_queries') ?? self::DEFAULT_LOG_ALL_QUERIES;
         $slowThresholdMs = self::extractOptionalPositiveInt($name, $config, 'slow_query_threshold_ms');
+        $queryTimeoutMs  = self::extractOptionalPositiveInt($name, $config, 'query_timeout_ms');
 
         return new PoolConfig(
             name:                  $name,
@@ -209,6 +211,7 @@ public static function validatePool(string $name, array $config): PoolConfig
             logBindings:           $logBindings,
             logAllQueries:         $logAllQueries,
             slowQueryThresholdMs:  $slowThresholdMs,
+            queryTimeoutMs:        $queryTimeoutMs,
         );
     }
 
diff --git a/src/Sloop/Database/Config/PoolConfig.php b/src/Sloop/Database/Config/PoolConfig.php
@@ -28,6 +28,10 @@
      * @param bool                  $logBindings           Whether prepared-statement bindings appear in log context
      * @param bool                  $logAllQueries         Whether every query is logged at `debug` level
      * @param int|null              $slowQueryThresholdMs  Threshold (ms) above which SELECT queries log at `warning`
+     * @param int|null              $queryTimeoutMs        Per-session query timeout in ms; null disables it.
+     *                                                     Connection issues a single SET SESSION (max_execution_time
+     *                                                     for MySQL, max_statement_time for MariaDB) lazily on the
+     *                                                     first query after dialect detection
      */
     public function __construct(
         public string $name,
@@ -40,6 +44,7 @@ public function __construct(
         public bool $logBindings,
         public bool $logAllQueries,
         public ?int $slowQueryThresholdMs,
+        public ?int $queryTimeoutMs,
     ) {
     }
 }
diff --git a/src/Sloop/Database/Connection.php b/src/Sloop/Database/Connection.php
@@ -71,6 +71,24 @@ final class Connection
      */
     private LoggingOptions $loggingOptions;
 
+    /**
+     * Per-session query timeout in milliseconds; null disables it. Set via setQueryTimeoutMs().
+     *
+     * @var int|null
+     */
+    private ?int $queryTimeoutMs = null;
+
+    /**
+     * Whether the configured timeout has already been issued to the server on this connection.
+     *
+     * Toggled true after a successful SET SESSION (max_execution_time on MySQL,
+     * max_statement_time on MariaDB) so subsequent queries skip the redundant
+     * statement. Reset to false when setQueryTimeoutMs() is called again.
+     *
+     * @var bool
+     */
+    private bool $queryTimeoutApplied = false;
+
     /**
      * Construct with an already-configured PDO.
      *
@@ -102,6 +120,28 @@ public function setLogger(LoggerInterface $logger, LoggingOptions $options): voi
         $this->loggingOptions = $options;
     }
 
+    /**
+     * Set the per-session query timeout in milliseconds (null disables it).
+     *
+     * Stored on the Connection but not issued to the server until the first
+     * query() or statement() call: dialect detection (`SELECT VERSION()`)
+     * has to run first to choose between MySQL `max_execution_time` and
+     * MariaDB `max_statement_time`. Calling this resets the "applied" flag
+     * so a fresh value will take effect on the next query — but a previously
+     * applied value remains in effect on the server session until the next
+     * SET SESSION fires (passing null does not retroactively clear it).
+     *
+     * Intended for one-time configuration before the first query runs.
+     *
+     * @param  int|null $ms Timeout in milliseconds (positive int) or null to disable
+     * @return void
+     */
+    public function setQueryTimeoutMs(?int $ms): void
+    {
+        $this->queryTimeoutMs      = $ms;
+        $this->queryTimeoutApplied = false;
+    }
+
     /**
      * Open a new Connection from a DSN, applying sloop's PDO defaults internally.
      *
@@ -150,6 +190,8 @@ public static function open(
      */
     public function query(string $sql, array $bindings = []): Result
     {
+        $this->applyQueryTimeoutIfPending();
+
         $startTime = $this->shouldMeasureElapsed() ? microtime(true) : null;
 
         try {
@@ -187,6 +229,8 @@ public function query(string $sql, array $bindings = []): Result
      */
     public function statement(string $sql, array $bindings = []): int
     {
+        $this->applyQueryTimeoutIfPending();
+
         $startTime = $this->shouldMeasureElapsed() ? microtime(true) : null;
 
         try {
@@ -486,6 +530,49 @@ private function execSimple(string $sql): void
         }
     }
 
+    /**
+     * Issue the configured per-session query timeout to the server on the first call.
+     *
+     * The timeout statement differs between server flavors (MySQL uses
+     * `max_execution_time` in milliseconds and applies only to SELECT;
+     * MariaDB uses `max_statement_time` in fractional seconds and applies
+     * to every statement), so dialect detection has to run before the
+     * SET SESSION can be assembled. The result is cached in
+     * $queryTimeoutApplied so subsequent queries do not re-issue it.
+     *
+     * Failures here are logged at `error` level when a logger is present
+     * and then re-thrown so the caller learns the timeout was not applied.
+     *
+     * @return void
+     * @throws DatabaseException When dialect detection or the SET SESSION fails
+     */
+    private function applyQueryTimeoutIfPending(): void
+    {
+        if ($this->queryTimeoutMs === null || $this->queryTimeoutApplied) {
+            return;
+        }
+
+        try {
+            $sql = match ($this->dialect()) {
+                Dialect::MySQL   => 'SET SESSION max_execution_time = ' . $this->queryTimeoutMs,
+                Dialect::MariaDB => 'SET SESSION max_statement_time = ' . \sprintf('%.3F', $this->queryTimeoutMs / 1000),
+            };
+            $this->execSimple($sql);
+            $this->queryTimeoutApplied = true;
+        } catch (DatabaseException $e) {
+            $this->logger?->error(
+                'failed to apply query timeout: ' . $e->getMessage(),
+                [
+                    'sqlstate'        => $e->sqlState,
+                    'driver_code'     => $e->driverCode,
+                    'connection_name' => $this->connectionName,
+                ],
+            );
+
+            throw $e;
+        }
+    }
+
     /**
      * Prepare and execute $sql with $bindings, wrapping any PDOException.
      *
diff --git a/src/Sloop/Database/ConnectionManager.php b/src/Sloop/Database/ConnectionManager.php
@@ -123,6 +123,7 @@ private function getPrimaryConnection(string $name): Connection
 
             $connection = $this->factory->make($pool->primary, $name);
             $this->applyLogger($connection, $pool);
+            $this->applyQueryTimeout($connection, $pool);
             $this->primaryConnections[$name] = $connection;
         }
 
@@ -172,6 +173,7 @@ private function getReplicaConnection(string $name): Connection
             try {
                 $connection = $this->factory->make($picked, $name);
                 $this->applyLogger($connection, $pool);
+                $this->applyQueryTimeout($connection, $pool);
 
                 if ($pool->healthCheck) {
                     $connection->ping();
@@ -268,6 +270,27 @@ private function applyLogger(Connection $connection, PoolConfig $pool): void
         );
     }
 
+    /**
+     * Forward the pool's query timeout into the Connection so it can lazy-apply on the first query.
+     *
+     * Skipped when the pool does not configure a timeout. Probe connections
+     * built by probeReplicas() are intentionally bypassed: they only run a
+     * `DO 1` ping and would just burn an extra round trip on a dialect probe
+     * plus SET SESSION before being discarded.
+     *
+     * @param  Connection $connection Newly built Connection that has not yet executed a query
+     * @param  PoolConfig $pool       Pool config supplying queryTimeoutMs
+     * @return void
+     */
+    private function applyQueryTimeout(Connection $connection, PoolConfig $pool): void
+    {
+        if ($pool->queryTimeoutMs === null) {
+            return;
+        }
+
+        $connection->setQueryTimeoutMs($pool->queryTimeoutMs);
+    }
+
     /**
      * Resolve and validate the pool config for the given name.
      *
diff --git a/tests/Unit/Database/Config/ConnectionConfigResolverTest.php b/tests/Unit/Database/Config/ConnectionConfigResolverTest.php
@@ -560,12 +560,14 @@ public function testValidatePoolAcceptsAllPoolBehaviorKeys(): void
             'dead_cache_ttl_seconds'  => 60,
             'replica_selector'        => 'random',
             'max_connection_attempts' => 6,
+            'query_timeout_ms'        => 5000,
         ]);
 
         $this->assertFalse($pool->healthCheck);
         $this->assertSame(60, $pool->deadCacheTtlSeconds);
         $this->assertSame('random', $pool->replicaSelector);
         $this->assertSame(6, $pool->maxConnectionAttempts);
+        $this->assertSame(5000, $pool->queryTimeoutMs);
     }
 
     public function testValidatePoolDefaultsMaxConnectionAttemptsToReplicaCountPlusOne(): void
@@ -588,15 +590,15 @@ public function testValidatePoolRejectsUnknownPoolKey(): void
     {
         try {
             ConnectionConfigResolver::validatePool('mydb', [
-                'driver'           => 'mysql',
-                'host'             => 'primary.example.com',
-                'database'         => 'app',
-                'query_timeout_ms' => 5000,
+                'driver'     => 'mysql',
+                'host'       => 'primary.example.com',
+                'database'   => 'app',
+                'persistent' => true,
             ]);
             $this->fail('Expected InvalidConfigException');
         } catch (InvalidConfigException $e) {
             $this->assertSame(
-                'Connection [mydb]: unsupported config key "query_timeout_ms".',
+                'Connection [mydb]: unsupported config key "persistent".',
                 $e->getMessage(),
             );
         }
@@ -886,6 +888,7 @@ public function testValidatePoolDefaultsLoggingKeysWhenOmitted(): void
         $this->assertTrue($pool->logBindings);
         $this->assertFalse($pool->logAllQueries);
         $this->assertNull($pool->slowQueryThresholdMs);
+        $this->assertNull($pool->queryTimeoutMs);
     }
 
     public function testValidatePoolAcceptsExplicitLogBindings(): void
@@ -1042,4 +1045,92 @@ public function testValidatePoolAcceptsExplicitNullMaxConnectionAttempts(): void
 
         $this->assertSame(3, $pool->maxConnectionAttempts);
     }
+
+    /**
+     * @return array<string, array{0: int|null, 1: int|null}>
+     */
+    public static function validQueryTimeoutMsProvider(): array
+    {
+        return [
+            'positive int'  => [5000, 5000],
+            'minimum 1'     => [1,    1],
+            'explicit null' => [null, null],
+        ];
+    }
+
+    #[DataProvider('validQueryTimeoutMsProvider')]
+    public function testValidatePoolAcceptsValidQueryTimeoutMs(?int $input, ?int $expected): void
+    {
+        // Valid inputs: positive int (typical), 1 (boundary inclusion guarding
+        // a `> 1` regression), and explicit null (the "off" sentinel shared
+        // with slow_query_threshold_ms / dead_cache_ttl_seconds).
+        $pool = ConnectionConfigResolver::validatePool('mydb', [
+            'driver'           => 'mysql',
+            'host'             => 'primary.example.com',
+            'database'         => 'app',
+            'query_timeout_ms' => $input,
+        ]);
+
+        $this->assertSame($expected, $pool->queryTimeoutMs);
+    }
+
+    /**
+     * @return array<string, array{0: mixed, 1: string}>
+     */
+    public static function invalidQueryTimeoutMsProvider(): array
+    {
+        return [
+            'non-int string' => ['5000', 'must be an integer'],
+            'float'          => [1.5,    'must be an integer'],
+            'bool true'      => [true,   'must be an integer'],
+            'zero'           => [0,      'must be >= 1, got 0'],
+            'negative'       => [-1,     'must be >= 1, got -1'],
+        ];
+    }
+
+    #[DataProvider('invalidQueryTimeoutMsProvider')]
+    public function testValidatePoolRejectsInvalidQueryTimeoutMs(mixed $value, string $expectedSuffix): void
+    {
+        // The two error-message templates correspond to the two rejection
+        // branches in extractOptionalPositiveInt: type check (`!is_int`) and
+        // range check (`< 1`). Bool/float are exercised explicitly so silent
+        // coercion regressions are caught.
+        try {
+            ConnectionConfigResolver::validatePool('mydb', [
+                'driver'           => 'mysql',
+                'host'             => 'primary.example.com',
+                'database'         => 'app',
+                'query_timeout_ms' => $value,
+            ]);
+            $this->fail('Expected InvalidConfigException');
+        } catch (InvalidConfigException $e) {
+            $this->assertSame(
+                'Connection [mydb]: config key "query_timeout_ms" ' . $expectedSuffix . '.',
+                $e->getMessage(),
+            );
+        }
+    }
+
+    public function testValidatePoolRejectsQueryTimeoutMsInsideReplica(): void
+    {
+        // query_timeout_ms is pool-level: it must be set on the pool itself,
+        // never inside a read[] entry. Same protection as the health_check
+        // regression in testValidatePoolRejectsPoolOnlyKeyInsideReplica.
+        try {
+            ConnectionConfigResolver::validatePool('mydb', [
+                'driver'   => 'mysql',
+                'host'     => 'primary.example.com',
+                'database' => 'app',
+                'read'     => [
+                    ['host' => 'replica-1.example.com', 'query_timeout_ms' => 5000],
+                ],
+            ]);
+            $this->fail('Expected InvalidConfigException');
+        } catch (InvalidConfigException $e) {
+            $this->assertSame(
+                'Connection [mydb]: "read[0]" has unsupported key "query_timeout_ms". Pool-level keys must be set on the pool itself, not inside read[].',
+                $e->getMessage(),
+            );
+        }
+    }
 }
diff --git a/tests/Unit/Database/Config/PoolConfigTest.php b/tests/Unit/Database/Config/PoolConfigTest.php
@@ -27,6 +27,7 @@ public function testStoresAllFields(): void
             logBindings: false,
             logAllQueries: true,
             slowQueryThresholdMs: 200,
+            queryTimeoutMs: 5000,
         );
 
         $this->assertSame('mydb', $pool->name);
@@ -39,6 +40,7 @@ public function testStoresAllFields(): void
         $this->assertFalse($pool->logBindings);
         $this->assertTrue($pool->logAllQueries);
         $this->assertSame(200, $pool->slowQueryThresholdMs);
+        $this->assertSame(5000, $pool->queryTimeoutMs);
     }
 
     public function testStoresEmptyReplicas(): void
@@ -54,6 +56,7 @@ public function testStoresEmptyReplicas(): void
             logBindings: true,
             logAllQueries: false,
             slowQueryThresholdMs: null,
+            queryTimeoutMs: null,
         );
 
         $this->assertSame([], $pool->replicas);
@@ -63,6 +66,7 @@ public function testStoresEmptyReplicas(): void
         $this->assertTrue($pool->logBindings);
         $this->assertFalse($pool->logAllQueries);
         $this->assertNull($pool->slowQueryThresholdMs);
+        $this->assertNull($pool->queryTimeoutMs);
     }
 
     public function testStoresSingleReplica(): void
@@ -81,6 +85,7 @@ public function testStoresSingleReplica(): void
             logBindings: true,
             logAllQueries: false,
             slowQueryThresholdMs: null,
+            queryTimeoutMs: null,
         );
 
         $this->assertCount(1, $pool->replicas);
diff --git a/tests/Unit/Database/ConnectionManagerTest.php b/tests/Unit/Database/ConnectionManagerTest.php
diff --git a/tests/Unit/Database/ConnectionTest.php b/tests/Unit/Database/ConnectionTest.php
