cache: stop a caller's read deadline from killing the shared connection

.vale/styles/config/vocabularies/Utopia/accept.txt

@@ -66,6 +66,8 @@ dotenv
 DuckDuckGo
 Eldad
 enqueue(?:s|d)?
+dequeue(?:s|d)?
+misroute(?:s|d)?
 enum(?:s)?
 etcd
 failover

packages/cache/docs/multiplexing.md

@@ -44,28 +44,58 @@ Construct once at worker start, share across requests, `disconnect()` on
 
 ```php
 new Multiplexing(
-    host:        'redis',
-    port:        6379,
-    timeout:     1.0,    // connect timeout (s)
-    readTimeout: 0.25,   // per-command response timeout (s)
-    auth:        null,   // string password, [user, password], or null
-    dbIndex:     0,
+    host:            'redis',
+    port:            6379,
+    timeout:         1.0,    // connect timeout (s)
+    readTimeout:     0.25,   // per-call response deadline (s)
+    auth:            null,   // string password, [user, password], or null
+    dbIndex:         0,
+    livenessTimeout: 5.0,    // reader silence before the connection is dead (s)
 );
 ```
 
-`readTimeout` defaults to **250 ms**. A timeout fails the command *and tears
-down the connection* — every other in-flight command on it also fails with
-`ConnectionException`, and the next command reconnects. This is intentional:
-caches should fail fast and let callers fall through to the source of truth.
-Per-request resync would add significant complexity for little gain.
+The two timeouts answer different questions, and the difference matters on a
+shared connection.
+
+`readTimeout` defaults to **250 ms** and is a deadline for *one call*: how long
+this caller waits for its own reply before giving up with a `TimeoutException`.
+Caches should fail fast and let callers fall through to the source of truth. It
+expires that call only — the connection is left alone, because a caller running
+out of patience is not evidence that the socket is broken.
+
+The abandoned call keeps its slot on the pending queue. That queue's order is
+what pairs replies with callers, so dropping the slot would hand this call's
+reply to whoever asked next, and every reply after it to the wrong caller.
+Leaving it means the reader dequeues it in order and pushes the late reply into
+a channel nobody is reading. No resync is needed, and no reply is misrouted.
+
+`livenessTimeout` defaults to **5 s** and is a verdict on the *connection*: if
+the reader has made no progress at all while callers are still waiting, the
+socket is treated as dead. It is torn down, every pending caller fails with
+`IdleConnectionException`, and the next call rebuilds it. This is what catches a
+connection whose packets are dropped rather than refused, where no close ever
+arrives — the reader blocks in `recv()` with no deadline of its own, so without
+this check its callers would wait forever. Keep it well above `readTimeout`: a server that is merely busy looks
+exactly like one that is gone until enough time has passed.
 
 ## Errors
 
 - `\RedisException` — Redis-side error (`WRONGTYPE`, `NOAUTH`, …). Connection
   is fine; the command was wrong. Not retried.
 - `Utopia\Cache\Adapter\Redis\ConnectionException` — transport failure
-  (timeout, socket closed, send failed). Connection has been discarded.
-  Retried if `setMaxRetries(n)` was called.
+  (socket closed, send failed, frame failed to parse). Connection has been
+  discarded and the call is retried once on a rebuilt connection.
+- `Utopia\Cache\Adapter\Redis\TimeoutException` — this call's `readTimeout`
+  expired. The connection is still in use by everyone else. Not retried:
+  resending would put a second copy of the command on a server that is already
+  answering too slowly.
+- `Utopia\Cache\Adapter\Redis\IdleConnectionException` — the reader went
+  quiet for `livenessTimeout` with replies outstanding, so the connection was
+  declared dead and discarded. Not retried: a fresh socket to a server that has
+  answered nothing for seconds will not answer this attempt either.
+
+Both of the latter extend `ConnectionException`, so existing `catch` sites keep
+working. Match them ahead of the parent to tell "slow" apart from "broken".
 
 ## Telemetry
 

packages/cache/src/Cache/Adapter/Redis/ConnectionContext.php

@@ -8,11 +8,39 @@
 
 class ConnectionContext
 {
+    /**
+     * When the reader last made progress on this connection, as a monotonic
+     * timestamp.
+     *
+     * A caller's expired deadline does not say whether the connection is alive —
+     * only that this caller waited long enough. The reader is the only party that
+     * observes the socket, so its last progress is what separates "the server is
+     * slow" from "the server is gone". Written by the reader on every dispatched
+     * frame and read by {@see Multiplexing::awaitResponse()}.
+     */
+    public float $lastProgressAt;
+
     /**
      * @param  SplQueue<\Swoole\Coroutine\Channel<mixed>>  $pending
      */
     public function __construct(
         public Client $client,
         public SplQueue $pending,
-    ) {}
+        ?float $lastProgressAt = null,
+    ) {
+        $this->lastProgressAt = $lastProgressAt ?? hrtime(true) / 1e9;
+    }
+
+    /**
+     * Seconds since the reader last dispatched a frame on this connection.
+     */
+    public function idleFor(): float
+    {
+        return (hrtime(true) / 1e9) - $this->lastProgressAt;
+    }
+
+    public function recordProgress(): void
+    {
+        $this->lastProgressAt = hrtime(true) / 1e9;
+    }
 }

packages/cache/src/Cache/Adapter/Redis/IdleConnectionException.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Cache\Adapter\Redis;
+
+/**
+ * Thrown by Multiplexing when the reader made no progress for longer than
+ * livenessTimeout while callers were still waiting, so the connection was
+ * declared dead and torn down.
+ *
+ * This is a connection verdict, unlike {@see TimeoutException} — but resending
+ * is still not the recovery. The connection has already been replaced, and a
+ * fresh socket to a server that has answered nothing for seconds will not answer
+ * this attempt either. Surfacing the failure lets the caller's own breaker see
+ * it; the next call gets the rebuilt connection.
+ */
+final class IdleConnectionException extends ConnectionException {}

packages/cache/src/Cache/Adapter/Redis/Multiplexing.php

@@ -39,8 +39,20 @@ class Multiplexing extends Leasable implements Adapter, TelemetryFeature
 
     /**
      * @param  float  $timeout connect timeout in seconds
-     * @param  float  $readTimeout read timeout in seconds — caches should
-     *                             fail fast, default 0.25s
+     * @param  float  $readTimeout per-call read deadline in seconds — how long
+     *                             one caller waits for its own reply before
+     *                             giving up. Caches should fail fast, default
+     *                             0.25s. Expiring fails that call only; it is
+     *                             not a verdict on the connection.
+     * @param  float  $livenessTimeout how long the reader may make no progress
+     *                                 at all, with callers still waiting, before
+     *                                 the connection is declared dead and torn
+     *                                 down. This is the verdict that fails every
+     *                                 pending caller, so it wants to be far
+     *                                 larger than $readTimeout: a socket that has
+     *                                 gone silent is indistinguishable from a
+     *                                 server that is merely busy until enough
+     *                                 time has passed. Default 5s.
      * @param  string|array<string>|null  $auth password or [username, password]
      */
     public function __construct(
@@ -50,13 +62,17 @@ public function __construct(
         private readonly float $readTimeout = 0.25,
         private readonly string|array|null $auth = null,
         private readonly int $dbIndex = 0,
+        private readonly float $livenessTimeout = 5.0,
     ) {
         if ($this->timeout <= 0) {
             throw new \InvalidArgumentException('timeout must be greater than 0');
         }
         if ($this->readTimeout <= 0) {
             throw new \InvalidArgumentException('readTimeout must be greater than 0');
         }
+        if ($this->livenessTimeout < $this->readTimeout) {
+            throw new \InvalidArgumentException('livenessTimeout must be greater than or equal to readTimeout');
+        }
         $this->sendLock = new Lock();
         $this->setTelemetry(new NoTelemetry());
 
@@ -227,16 +243,27 @@ public function getName(?string $key = null): string
 
     /**
      * Send a Redis command and block the calling coroutine until the response arrives.
-     * On a connection error, transparently reconnects once and retries — multiplexed
-     * connections drop all in-flight callers when they fail, so the only sensible
-     * recovery is to rebuild the connection before failing the call.
+     * On a connection error, transparently reconnects once and retries — a failed
+     * connection drops every in-flight caller, so the only sensible recovery is to
+     * rebuild it before failing the call.
+     *
+     * An expired read deadline is not retried. It means the server was slow, and the
+     * connection is still good; reconnecting and resending would put a second copy of
+     * the same command on an already-struggling server, once per caller.
      *
      * @param  array<int|string>  $args
      */
     private function command(array $args): mixed
     {
         try {
             return $this->dispatch($args);
+        } catch (TimeoutException|IdleConnectionException $unretryable) {
+            // Caught ahead of their parent to opt out of the retry below. Neither
+            // is fixed by resending: the server is slow, or it has answered
+            // nothing for seconds and the connection has just been replaced.
+            // Resending would add a second copy of this command, per caller, to a
+            // server that is already failing to keep up.
+            throw $unretryable;
         } catch (ConnectionException) {
             $this->ensureConnected();
 
@@ -300,10 +327,38 @@ private function awaitResponse(ConnectionContext $context, Channel $response): m
     {
         $result = $response->pop($this->readTimeout);
         if ($result === false && $response->errCode !== 0) {
-            $error = new ConnectionException('Timed out waiting for Redis response');
-            $this->teardownIfCurrent($context, $error);
+            $idleFor = $context->idleFor();
+
+            // The reader has gone quiet for longer than a busy server explains,
+            // and callers are still queued: treat the connection as dead so the
+            // next caller rebuilds it. Without this a socket that never returns
+            // from recv() — blackholed rather than reset — would strand every
+            // caller forever, because the reader has no deadline of its own.
+            if ($idleFor >= $this->livenessTimeout) {
+                $error = new IdleConnectionException(\sprintf(
+                    'Redis connection idle for %.3fs with responses outstanding',
+                    $idleFor,
+                ));
+                $this->teardownIfCurrent($context, $error);
+
+                throw $error;
+            }
 
-            throw $error;
+            // Otherwise the server is just slower than this caller was willing to
+            // wait. Fail this call and leave the connection alone.
+            //
+            // The response Channel deliberately stays on $pending. The FIFO
+            // invariant is that the queue's order matches the order of frames on
+            // the wire, so removing this slot would make the reader hand this
+            // call's reply to the next caller — and every reply after it to the
+            // wrong caller. Leaving it means the reader dequeues it in order and
+            // pushes the late reply into a Channel nobody is reading; capacity is
+            // 1, so the push cannot block, and the Channel is then collected. The
+            // pending-depth counter is decremented by the reader as usual.
+            throw new TimeoutException(\sprintf(
+                'Timed out waiting for Redis response after %.3fs',
+                $this->readTimeout,
+            ));
         }
 
         return Client::unwrap($result);
@@ -445,6 +500,11 @@ private function readerLoop(ConnectionContext $context): void
                 $waiting = $context->pending->isEmpty() ? null : $context->pending->dequeue();
                 if ($waiting instanceof Channel) {
                     $this->getPendingDepth()->add(-1);
+                    $context->recordProgress();
+                    // May be a reply whose caller already gave up on its own
+                    // deadline. The slot is still dequeued in order so the frames
+                    // behind it stay aligned, and the push cannot block on a
+                    // capacity-1 Channel, so an abandoned reply is simply dropped.
                     $waiting->push($value);
                 } else {
                     // Should never happen given the send-lock invariant. Log
@@ -458,6 +518,12 @@ private function readerLoop(ConnectionContext $context): void
             }
 
             $chunk = $context->client->recv(-1);
+            if (\is_string($chunk) && $chunk !== '') {
+                // Bytes arriving is progress even before they complete a frame,
+                // so a large reply streaming in slowly is not mistaken for a dead
+                // connection.
+                $context->recordProgress();
+            }
             if ($chunk === false || $chunk === '') {
                 $this->teardownIfCurrent($context, new ConnectionException('Redis connection closed'));
 

packages/cache/src/Cache/Adapter/Redis/TimeoutException.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Cache\Adapter\Redis;
+
+/**
+ * Thrown by Multiplexing when a caller's own read deadline expires before its
+ * response arrived.
+ *
+ * Distinct from its parent on purpose. A {@see ConnectionException} is a verdict
+ * about the *connection* — the socket is gone, a frame failed to parse, a send
+ * was truncated — and the only recovery is to rebuild it. This is a verdict
+ * about one *call*: the server was slower than this caller was willing to wait,
+ * which says nothing about whether the connection is healthy. Conflating the two
+ * is expensive on a multiplexed connection, because tearing it down fails every
+ * other caller queued behind the slow one.
+ *
+ * Extends ConnectionException so existing `catch` sites keep working; callers
+ * that need to tell "slow" from "broken" apart match this type first.
+ */
+final class TimeoutException extends ConnectionException {}