Retry reads and writes when protocol errors occur

src/main/php/com/mongodb/Error.class.php

@@ -32,6 +32,19 @@ public function __construct($code, $kind, $message, $cause= null, $retried= 0) {
   /** @return int */
   public function retried() { return $this->retried; }
 
+  /**
+   * Creates a protocol error
+   *
+   * @param  com.mongodb.io.Connection $conn
+   * @param  ?lang.Throwable $cause
+   * @param  int|bool $retried
+   * @return self
+   */
+  public static function protocol($conn, $cause, $retried= 0) {
+    $kind= $conn->server['$kind'] ?? 'Unknown';
+    return new self(17, 'ProtocolError', "{$conn->address()} ({$kind})", $cause, $retried);
+  }
+
   /**
    * Creates an error from a given error document
    *

src/main/php/com/mongodb/io/Commands.class.php

@@ -1,6 +1,7 @@
 <?php namespace com\mongodb\io;
 
 use com\mongodb\Error;
+use peer\ProtocolException;
 
 /**
  * Ensures all message sent using this instance are executed against
@@ -10,18 +11,20 @@
  * @see  https://github.com/mongodb/specifications/blob/master/source/server-selection/server-selection.rst#cursors
  */
 class Commands {
-  private $proto, $conn;
-  private $retry= true;
+  private $conn, $proto, $rp, $retry;
 
   /**
    * Creates an instance using a protocol and connection instance.
    *
    * @param  com.mongodb.io.Protocol $proto
-   * @param  com.mongodb.io.Connection $conn
+   * @param  [:var] $rp
+   * @param  bool $retry
    */
-  private function __construct($proto, $conn) {
+  private function __construct($proto, $rp, $retry= true) {
+    $this->conn= $proto->establish($proto->candidates($rp), 'commands with '.$rp['mode']);
     $this->proto= $proto;
-    $this->conn= $conn;
+    $this->rp= $rp;
+    $this->retry= $retry;
   }
 
   /** @return com.mongodb.io.Connection */
@@ -30,19 +33,21 @@ public function connection() { return $this->conn; }
   /** Creates an instance for reading */
   public static function reading(Protocol $proto): self {
     $proto->nodes || $proto->connect();
-    return new self($proto, $proto->establish(
-      $proto->candidates($proto->readPreference),
-      'reading with '.$proto->readPreference['mode']
-    ));
+    return new self(
+      $proto,
+      $proto->readPreference,
+      'true' === ($proto->options()['params']['retryReads'] ?? 'true')
+    );
   }
 
   /** Creates an instance for writing */
   public static function writing(Protocol $proto): self {
     $proto->nodes || $proto->connect();
-    return new self($proto, $proto->establish(
-      [$proto->nodes['primary']],
-      'writing'
-    ));
+    return new self(
+      $proto,
+      ['mode' => 'primary'],
+      'true' === ($proto->options()['params']['retryWrites'] ?? 'true')
+    );
   }
 
   /**
@@ -73,22 +78,30 @@ public function send($options, $sections) {
       $sections+= $option->send($this->proto);
     }
 
-    $rp= $sections['$readPreference'] ?? $this->proto->readPreference;
-
     // Only retry the very first command once in this sequence!
+    $rp= $sections['$readPreference'] ?? $this->proto->readPreference;
     try {
       retry: $r= $this->conn->send(Connection::OP_MSG, "\x00\x00\x00\x00\x00", $sections, $rp);
       if (1 === (int)$r['body']['ok']) return $r;
 
       // Retry "NotWritablePrimary" errors, replacing the connection
       if ($this->retry && isset(Error::NOT_PRIMARY[$r['body']['code']])) {
         $this->proto->useCluster($this->conn->hello());
-        $this->conn= $this->proto->establish([$this->proto->nodes['primary']], 'writing');
+        $this->conn= $this->proto->establish($this->proto->candidates($this->rp), 'commands with '.$this->rp['mode']);
         $this->retry= false;
         goto retry;
       }
 
       throw Error::newInstance($r['body'], !$this->retry);
+    } catch (ProtocolException $e) {
+      if ($this->retry) {
+        $this->conn->close();
+        $this->conn= $this->proto->establish($this->proto->candidates($this->rp), 'commands with '.$this->rp['mode']);
+        $this->retry= false;
+        goto retry;
+      }
+
+      throw Error::protocol($this->conn, $e, !$this->retry);
     } finally {
       $this->retry= false;
     }

src/main/php/com/mongodb/io/Protocol.class.php

@@ -3,7 +3,7 @@
 use com\mongodb\{Authentication, NoSuitableCandidate, CannotConnect, Error};
 use io\OperationFailed;
 use lang\{IllegalStateException, IllegalArgumentException, Throwable};
-use peer\{ConnectException, Socket, SocketException};
+use peer\{ConnectException, Socket, SocketException, ProtocolException};
 use util\Objects;
 
 /**
@@ -248,11 +248,20 @@ public function read(array $options, $sections) {
     }
 
     $rp= $sections['$readPreference'] ?? $this->readPreference;
-    $conn= $this->establish($this->candidates($rp), 'reading with '.$rp['mode']);
-    $r= $conn->send(Connection::OP_MSG, "\x00\x00\x00\x00\x00", $sections, $rp);
-    if (1 === (int)$r['body']['ok']) return $r;
+    $retry= 'true' === ($this->options['params']['retryReads'] ?? 'true');
+    try {
+      retry: $conn= $this->establish($this->candidates($rp), 'reading with '.$rp['mode']);
+      $r= $conn->send(Connection::OP_MSG, "\x00\x00\x00\x00\x00", $sections, $rp);
+      if (1 === (int)$r['body']['ok']) return $r;
+    } catch (ProtocolException $e) {
+      if (!$retry) throw Error::protocol($conn, $e, !$retry);
 
-    throw Error::newInstance($r['body']);
+      $conn->close();
+      $retry= false;
+      goto retry;
+    }
+
+    throw Error::newInstance($r['body'], !$retry);
   }
 
   /**
@@ -271,18 +280,24 @@ public function write(array $options, $sections) {
     }
 
     $rp= $sections['$readPreference'] ?? $this->readPreference;
-    $retry= true;
-
-    // Use send() API to prevent using exceptions for flow control
-    retry: $conn= $this->establish([$this->nodes['primary']], 'writing');
-    $r= $conn->send(Connection::OP_MSG, "\x00\x00\x00\x00\x00", $sections, $rp);
-    if (1 === (int)$r['body']['ok']) return $r;
-
-    // Check for "NotWritablePrimary" error, which indicates our view of the cluster
-    // may be outdated, see https://github.com/xp-forge/mongodb/issues/43. Refresh
-    // view using the "hello" command, then retry the command once.
-    if ($retry && isset(Error::NOT_PRIMARY[$r['body']['code']])) {
-      $this->useCluster($conn->hello());
+    $retry= 'true' === ($this->options['params']['retryWrites'] ?? 'true');
+    try {
+      retry: $conn= $this->establish([$this->nodes['primary']], 'writing');
+      $r= $conn->send(Connection::OP_MSG, "\x00\x00\x00\x00\x00", $sections, $rp);
+      if (1 === (int)$r['body']['ok']) return $r;
+
+      // Check for "NotWritablePrimary" error, which indicates our view of the cluster
+      // may be outdated, see https://github.com/xp-forge/mongodb/issues/43. Refresh
+      // view using the "hello" command, then retry the command once.
+      if ($retry && isset(Error::NOT_PRIMARY[$r['body']['code']])) {
+        $this->useCluster($conn->hello());
+        $retry= false;
+        goto retry;
+      }
+    } catch (ProtocolException $e) {
+      if (!$retry) throw Error::protocol($conn, $e, !$retry);
+
+      $conn->close();
       $retry= false;
       goto retry;
     }

src/test/php/com/mongodb/unittest/CollectionTest.class.php

@@ -16,6 +16,12 @@ private function documents() {
     yield [[['_id' => 'one', 'name' => 'A'], ['_id' => 'two', 'name' => 'B']]];
   }
 
+  /** @return iterable */
+  private function reads() {
+    yield ['find', fn($fixture) => $fixture->find([])];
+    yield ['count', fn($fixture) => $fixture->count([])];
+  }
+
   /** @return iterable */
   private function writes() {
     yield ['update', fn($fixture) => $fixture->update('6100', ['$inc' => ['qty' => 1]])];
@@ -393,6 +399,33 @@ public function error_raised() {
     $fixture->update('6100', ['$inc' => ['qty' => 1]]);
   }
 
+  #[Test, Values(from: 'reads')]
+  public function read_eof_retried($kind, $command) {
+    $command($this->newFixture(
+      null, // throws peer.ProtocolException (Received EOF while reading)
+      $this->hello(self::$PRIMARY),
+      $this->cursor([])
+    ));
+  }
+
+  #[Test, Expect(Error::class), Values(from: 'reads')]
+  public function read_eof_not_retried_more_than_once($kind, $command) {
+    $command($this->newFixture(
+      null, // throws peer.ProtocolException (Received EOF while reading)
+      $this->hello(self::$PRIMARY),
+      null, // -"-
+    ));
+  }
+
+  #[Test, Values(from: 'writes')]
+  public function write_eof_retried($kind, $command) {
+    $command($this->newFixture(
+      null, // throws peer.ProtocolException (Received EOF while reading)
+      $this->hello(self::$PRIMARY),
+      $this->ok(['n' => 1, 'nModified' => 1])
+    ));
+  }
+
   #[Test, Values(from: 'writes')]
   public function not_writable_primary_retried($kind, $command) {
     $command($this->newFixture(