ConnectionManager にハイブリッド読み書きルーティングを実装 (Phase 5-1 サブ C-4)

connection(?bool $writable = null) で primary/replica を明示的にルーティング。
replica 選択フロー: dead cache フィルタ → ReplicaSelector → 接続試行 →
DO 1 ping (health_check 有効時) → 失敗時 dead cache 記録 (SQLSTATE 28000 は
pool 単位、それ以外は server 単位) → 次の replica → 全滅で primary フォール
バック → max_connection_attempts 枯渇で DatabaseConnectionException (累積
エラー詳細付き)。

null/true は primary、false は replica。null で SELECT 自動検出は行わず、
それは Builder 層の責務として Phase 5-2 以降に委ねる。read=[] 時は
connection(writable:false) も primary を返す (単一プール運用との互換)。

ReplicaSelector::pick() は non-empty-list を受け取り int を返す契約に変更
(coding-standards.md「\assert() で型保証しない」方針に準拠、設計レベルで
非空を保証)。RandomReplicaSelector の empty 早期 return も同時に削除。

Application::registerCoreBindings に ReplicaSelector (RandomReplicaSelector
singleton) と DeadReplicaCache (ApcuDeadReplicaCache::isAvailable() で
Apcu/InMemory 切替) を追加し、ConnectionManager に DI 注入する。
DeadReplicaCache の選択ロジックは createDeadReplicaCache() に抽出して
他の create*() メソッドと体裁を揃えた。

例外メッセージは「Failed to obtain a read connection (replica + primary
fallback exhausted)」として primary フォールバック失敗も含む状況を明示。
dead-cache でスキップされた replica は「host:port → skipped (dead-cache)」
形式で累積エラーに含めるため、運用診断時にスキップ理由まで一目で把握できる。
dead-cache フィルタは partitionByDeadCache() ヘルパーに切り出した。

テスト: ConnectionManagerTest に 14 ケース、ApplicationTest に 8 ケース
(singleton / default / invalid / override 反映を 2 系列ずつ) を追加。
Stub に FixedReplicaSelector / ScriptedConnectionFactory を新設してシナリオ
ベース検証を可能にした。RandomReplicaSelectorTest は non-empty-list 契約に
合わせて見直し。ping failure テストは PDO mock + \PDOException 投擲で再現
することで SQLite の DO 文挙動依存を排除しつつ、両 pool で dead 状態を
assert する。

Author: logicraft

src/Sloop/Database/ConnectionManager.php

@@ -5,9 +5,14 @@
 namespace Sloop\Database;
 
 use Sloop\Database\Config\ConnectionConfigResolver;
+use Sloop\Database\Config\PoolConfig;
+use Sloop\Database\Config\ValidatedConfig;
 use Sloop\Database\Exception\DatabaseConnectionException;
+use Sloop\Database\Exception\DatabaseException;
 use Sloop\Database\Exception\InvalidConfigException;
 use Sloop\Database\Factory\ConnectionFactory;
+use Sloop\Database\Replica\DeadReplicaCache;
+use Sloop\Database\Replica\ReplicaSelector;
 
 /**
  * Lazily creates and caches database connections from pool configurations.
@@ -18,66 +23,240 @@
  * the injected ConnectionFactory the first time they are requested and
  * cached so a single request reuses one PDO instance per pool name.
  *
- * connection() currently returns the pool's primary; replica selection and
- * the writable parameter are not yet implemented.
+ * connection() routing:
+ * - $writable === true  → primary
+ * - $writable === false → replica (dead-cache filter → ReplicaSelector → ping
+ *                                  → record on failure → next → primary fallback
+ *                                  → throw on max_connection_attempts exhaustion)
+ * - $writable === null  → primary (Builder layer detects SELECT, not the manager)
+ *
+ * Empty `read` list collapses replica routing to the primary so single-pool
+ * setups keep working without reconfiguration.
  */
 final class ConnectionManager
 {
     /**
-     * Cached Connection instances keyed by pool name.
+     * Cached primary Connection instances keyed by pool name.
      *
      * @var array<string, Connection>
      */
-    private array $connections = [];
+    private array $primaryConnections = [];
+
+    /**
+     * Cached replica Connection instances keyed by pool name.
+     *
+     * Replica selection runs at most once per pool per request: once a healthy
+     * replica is found, subsequent connection(writable: false) calls reuse it.
+     *
+     * @var array<string, Connection>
+     */
+    private array $replicaConnections = [];
 
     /**
      * Construct a new ConnectionManager.
      *
-     * @param string                              $defaultName Pool name to return from connection()
-     * @param array<string, array<string, mixed>> $configs     Pool configurations indexed by pool name
-     * @param ConnectionFactory                   $factory     Builds Connection instances from validated configs
+     * @param string                              $defaultName     Pool name to return from connection()
+     * @param array<string, array<string, mixed>> $configs         Pool configurations indexed by pool name
+     * @param ConnectionFactory                   $factory         Builds Connection instances from validated configs
+     * @param ReplicaSelector                     $replicaSelector Strategy for picking one replica from surviving candidates
+     * @param DeadReplicaCache                    $deadCache       Negative cache for replicas that recently failed to connect
      */
     public function __construct(
         private readonly string $defaultName,
         private readonly array $configs,
         private readonly ConnectionFactory $factory,
+        private readonly ReplicaSelector $replicaSelector,
+        private readonly DeadReplicaCache $deadCache,
     ) {
     }
 
     /**
-     * Return the default pool's primary connection, creating and caching it on first call.
+     * Return the default pool's connection, routing to primary or replica based on $writable.
      *
-     * @return Connection                  Lazy-created, cached Connection to the default pool's primary
+     * @param  bool|null                   $writable true → primary; false → replica with primary fallback;
+     *                                               null → primary (Builder layer is responsible for SELECT detection)
+     * @return Connection                  Lazy-created, cached Connection
      * @throws InvalidConfigException      When the default pool name is not defined or its config is malformed
-     * @throws DatabaseConnectionException When the underlying PDO connection cannot be established
+     * @throws DatabaseConnectionException When max_connection_attempts is exhausted on the replica path
      */
-    public function connection(): Connection
+    public function connection(?bool $writable = null): Connection
     {
-        if (!isset($this->connections[$this->defaultName])) {
-            $this->connections[$this->defaultName] = $this->makeConnection($this->defaultName);
+        if ($writable === false) {
+            return $this->getReplicaConnection($this->defaultName);
         }
 
-        return $this->connections[$this->defaultName];
+        return $this->getPrimaryConnection($this->defaultName);
     }
 
     /**
-     * Build a new Connection to the named pool's primary via the injected factory.
+     * Return the cached primary Connection or build it on first access.
      *
      * @param  string                      $name Pool name
      * @return Connection
-     * @throws InvalidConfigException      When the name is undefined or the config is malformed
-     * @throws DatabaseConnectionException When the underlying PDO connection cannot be established
+     * @throws InvalidConfigException      When the name is undefined or config is malformed
+     * @throws DatabaseConnectionException When the underlying PDO connection fails
      */
-    private function makeConnection(string $name): Connection
+    private function getPrimaryConnection(string $name): Connection
+    {
+        if (!isset($this->primaryConnections[$name])) {
+            $pool = $this->resolvePool($name);
+
+            $this->primaryConnections[$name] = $this->factory->make($pool->primary, $name);
+        }
+
+        return $this->primaryConnections[$name];
+    }
+
+    /**
+     * Return the cached replica Connection, or run the selection loop on first access.
+     *
+     * Selection flow:
+     * 1. Filter out replicas marked dead in the cache
+     * 2. ReplicaSelector picks one of the survivors
+     * 3. Connect via ConnectionFactory; if health_check is on, verify with Connection::ping()
+     * 4. On failure: record dead (auth → markPoolDead, otherwise markServerDead)
+     *    and remove the candidate; loop until success, candidates exhausted,
+     *    or max_connection_attempts reached
+     * 5. When all replicas fail and attempts remain, fall back to primary
+     * 6. If no healthy connection is found, throw DatabaseConnectionException
+     *    with cumulative per-attempt error details
+     *
+     * @param  string                      $name Pool name
+     * @return Connection
+     * @throws InvalidConfigException      When the name is undefined or config is malformed
+     * @throws DatabaseConnectionException When all attempts (replica + fallback) fail
+     */
+    private function getReplicaConnection(string $name): Connection
+    {
+        if (isset($this->replicaConnections[$name])) {
+            return $this->replicaConnections[$name];
+        }
+
+        $pool = $this->resolvePool($name);
+
+        if ($pool->replicas === []) {
+            // No replicas configured → primary doubles as the read endpoint.
+            return $this->getPrimaryConnection($name);
+        }
+
+        [$candidates, $errors] = $this->partitionByDeadCache($pool);
+        $attempts              = 0;
+
+        while ($candidates !== [] && $attempts < $pool->maxConnectionAttempts) {
+            $index  = $this->replicaSelector->pick($candidates);
+            $picked = $candidates[$index];
+            $attempts++;
+
+            try {
+                $connection = $this->factory->make($picked, $name);
+
+                if ($pool->healthCheck) {
+                    $connection->ping();
+                }
+
+                $this->replicaConnections[$name] = $connection;
+
+                return $connection;
+            } catch (DatabaseException $e) {
+                $errors[] = $this->formatAttemptError($picked, $e);
+                $this->recordReplicaFailure($pool, $picked, $e);
+                array_splice($candidates, $index, 1);
+            }
+        }
+
+        if ($attempts < $pool->maxConnectionAttempts) {
+            try {
+                return $this->getPrimaryConnection($name);
+            } catch (DatabaseException $e) {
+                $errors[] = $this->formatAttemptError($pool->primary, $e);
+            }
+        }
+
+        throw new DatabaseConnectionException(
+            'Failed to obtain a read connection for pool [' . $name . '] (replica + primary fallback exhausted): ' . implode(' | ', $errors),
+            $name,
+        );
+    }
+
+    /**
+     * Resolve and validate the pool config for the given name.
+     *
+     * @param  string                 $name Pool name
+     * @return PoolConfig
+     * @throws InvalidConfigException When the name is undefined or config is malformed
+     */
+    private function resolvePool(string $name): PoolConfig
     {
         if (!\array_key_exists($name, $this->configs)) {
             throw new InvalidConfigException(
                 'Database connection [' . $name . '] is not defined.',
             );
         }
 
-        $pool = ConnectionConfigResolver::validatePool($name, $this->configs[$name]);
+        return ConnectionConfigResolver::validatePool($name, $this->configs[$name]);
+    }
+
+    /**
+     * Partition replicas into alive candidates and dead-cache skip messages.
+     *
+     * Skipped entries are recorded as preformatted error strings so they can be
+     * appended to the cumulative diagnostic message that surfaces when every
+     * connection attempt (replica + primary fallback) fails.
+     *
+     * @param  PoolConfig                                       $pool Pool config (provides replica list and pool name)
+     * @return array{0: list<ValidatedConfig>, 1: list<string>}       [alive in declaration order, skip messages]
+     */
+    private function partitionByDeadCache(PoolConfig $pool): array
+    {
+        $alive   = [];
+        $skipped = [];
+        foreach ($pool->replicas as $replica) {
+            if ($this->deadCache->isDead($replica->host, $replica->port ?? 0, $pool->name)) {
+                $skipped[] = $replica->host . ':' . ($replica->port ?? '?') . ' → skipped (dead-cache)';
+
+                continue;
+            }
+            $alive[] = $replica;
+        }
+
+        return [$alive, $skipped];
+    }
+
+    /**
+     * Mark a failed replica dead in the negative cache.
+     *
+     * Auth failures (SQLSTATE 28000) are pool-specific: the credential is wrong
+     * only for this pool, so the per-pool key is used. Everything else (TCP
+     * refused, server unreachable, DO 1 failure) implies the host itself is
+     * unhealthy and goes to the shared key.
+     *
+     * @param  PoolConfig        $pool    Pool config (provides pool name and TTL)
+     * @param  ValidatedConfig   $replica The replica that just failed
+     * @param  DatabaseException $e       Failure thrown by the factory or ping()
+     * @return void
+     */
+    private function recordReplicaFailure(PoolConfig $pool, ValidatedConfig $replica, DatabaseException $e): void
+    {
+        $port = $replica->port ?? 0;
+
+        if ($e instanceof DatabaseConnectionException && $e->sqlState === '28000') {
+            $this->deadCache->markPoolDead($replica->host, $port, $pool->name, $pool->deadCacheTtlSeconds);
 
-        return $this->factory->make($pool->primary, $name);
+            return;
+        }
+
+        $this->deadCache->markServerDead($replica->host, $port, $pool->deadCacheTtlSeconds);
+    }
+
+    /**
+     * Format one failed attempt for the cumulative error message.
+     *
+     * @param  ValidatedConfig   $config Target host/port
+     * @param  DatabaseException $e      Failure
+     * @return string
+     */
+    private function formatAttemptError(ValidatedConfig $config, DatabaseException $e): string
+    {
+        return $config->host . ':' . ($config->port ?? '?') . ' → ' . $e->getMessage();
     }
 }

src/Sloop/Database/Replica/RandomReplicaSelector.php

@@ -19,15 +19,11 @@ final class RandomReplicaSelector implements ReplicaSelector
     /**
      * Pick one candidate uniformly at random and return its index.
      *
-     * @param  list<ValidatedConfig> $candidates Surviving replica configs
-     * @return int|null              Index into $candidates, or null when empty
+     * @param  non-empty-list<ValidatedConfig> $candidates Surviving replica configs (must be non-empty)
+     * @return int                                         Valid index into $candidates
      */
-    public function pick(array $candidates): ?int
+    public function pick(array $candidates): int
     {
-        if ($candidates === []) {
-            return null;
-        }
-
         return (int) array_rand($candidates);
     }
 }

src/Sloop/Database/Replica/ReplicaSelector.php

@@ -17,10 +17,14 @@
 interface ReplicaSelector
 {
     /**
-     * Pick one candidate and return its index, or null when no candidates remain.
+     * Pick one candidate and return its index.
      *
-     * @param  list<ValidatedConfig> $candidates Surviving replica configs
-     * @return int|null              Index into $candidates, or null when empty
+     * Callers must ensure $candidates is non-empty; ConnectionManager filters
+     * out dead replicas first and only invokes pick() while at least one
+     * survivor remains, so this contract is checkable at the call site.
+     *
+     * @param  non-empty-list<ValidatedConfig> $candidates Surviving replica configs (must be non-empty)
+     * @return int                                         Valid index into $candidates
      */
-    public function pick(array $candidates): ?int;
+    public function pick(array $candidates): int;
 }

src/Sloop/Foundation/Application.php

@@ -16,6 +16,11 @@
 use Sloop\Database\ConnectionManager;
 use Sloop\Database\Factory\ConnectionFactory;
 use Sloop\Database\Factory\PdoConnectionFactory;
+use Sloop\Database\Replica\ApcuDeadReplicaCache;
+use Sloop\Database\Replica\DeadReplicaCache;
+use Sloop\Database\Replica\InMemoryDeadReplicaCache;
+use Sloop\Database\Replica\RandomReplicaSelector;
+use Sloop\Database\Replica\ReplicaSelector;
 use Sloop\Http\HttpStatus;
 use Sloop\Http\Middleware\MiddlewareDispatcher;
 use Sloop\Http\Request\Request;
@@ -181,9 +186,27 @@ private function registerCoreBindings(): void
         $this->container->singleton(TraceContext::class, fn (): TraceContext => new TraceContext());
         $this->container->singleton(LogManager::class, fn (Container $container): LogManager => $this->createLogManager($container));
         $this->container->singleton(ConnectionFactory::class, fn (): ConnectionFactory => new PdoConnectionFactory());
+        $this->container->singleton(ReplicaSelector::class, fn (): ReplicaSelector => new RandomReplicaSelector());
+        $this->container->singleton(DeadReplicaCache::class, fn (): DeadReplicaCache => $this->createDeadReplicaCache());
         $this->container->singleton(ConnectionManager::class, fn (Container $container): ConnectionManager => $this->createConnectionManager($container));
     }
 
+    /**
+     * Create the DeadReplicaCache implementation appropriate to the current environment.
+     *
+     * Picks ApcuDeadReplicaCache when ext-apcu is loaded and apc.enable_cli=1
+     * (cross-request dead marks); otherwise falls back to InMemoryDeadReplicaCache
+     * (per-request dead marks, sufficient as a degraded fallback).
+     *
+     * @return DeadReplicaCache
+     */
+    private function createDeadReplicaCache(): DeadReplicaCache
+    {
+        return ApcuDeadReplicaCache::isAvailable()
+            ? new ApcuDeadReplicaCache()
+            : new InMemoryDeadReplicaCache();
+    }
+
     /**
      * Create the ApiResponseFormatter from configuration.
      *
@@ -290,9 +313,9 @@ private function resolveChannelFactory(Container $container, string $factoryClas
     /**
      * Create the ConnectionManager from configuration.
      *
-     * @param  Container          $container Container used to resolve the ConnectionFactory binding
+     * @param  Container         $container Container used to resolve the ConnectionFactory / ReplicaSelector / DeadReplicaCache bindings
      * @return ConnectionManager
-     * @throws \RuntimeException  If the ConnectionFactory binding does not implement ConnectionFactory
+     * @throws \RuntimeException If any required binding does not implement its declared interface
      */
     private function createConnectionManager(Container $container): ConnectionManager
     {
@@ -315,10 +338,26 @@ private function createConnectionManager(Container $container): ConnectionManage
             );
         }
 
+        $replicaSelector = $container->get(ReplicaSelector::class);
+        if (!$replicaSelector instanceof ReplicaSelector) {
+            throw new \RuntimeException(
+                'Container binding for ' . ReplicaSelector::class . ' must implement ReplicaSelector.',
+            );
+        }
+
+        $deadCache = $container->get(DeadReplicaCache::class);
+        if (!$deadCache instanceof DeadReplicaCache) {
+            throw new \RuntimeException(
+                'Container binding for ' . DeadReplicaCache::class . ' must implement DeadReplicaCache.',
+            );
+        }
+
         return new ConnectionManager(
             defaultName: $default,
             configs: $connections,
             factory: $factory,
+            replicaSelector: $replicaSelector,
+            deadCache: $deadCache,
         );
     }