Skip to content

Reconnect after MongoDB primary changes #44

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
abnegate merged 1 commit into from
Jul 30, 2026
Merged

Reconnect after MongoDB primary changes #44

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 44 additions & 2 deletions src/Client.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,11 @@ class Client
*/
private bool $isConnected = false;

/**
* Whether the next operation must establish a fresh physical connection.
*/
private bool $reconnect = false;

/**
* Socket / receive idle timeout in seconds.
*
Expand Down Expand Up @@ -86,6 +91,8 @@ class Client
public const TRANSACTION_TIMEOUT_ERROR = 50;
public const TRANSACTION_ABORTED_ERROR = 251;

private const PRIMARY_CHANGE_ERRORS = [189, 10107, 11602, 13435, 13436];

// Transaction states
public const TRANSACTION_NONE = 'none';
public const TRANSACTION_STARTING = 'starting';
Expand Down Expand Up @@ -1689,8 +1696,10 @@ public function close(): void
/**
* Physically close a failed socket without writing session cleanup to it.
*/
private function invalidate(): void
private function invalidate(bool $preserveSessions = false): void
{
$sessions = $preserveSessions ? $this->sessions : [];

try {
if ($this->client instanceof CoroutineClient) {
@$this->client->close();
Expand All @@ -1701,6 +1710,10 @@ private function invalidate(): void
// The transport may already be closed or only partially initialized.
} finally {
$this->reset();
if ($preserveSessions) {
$this->sessions = $sessions;
$this->reconnect = true;
}
}
}

Expand All @@ -1710,6 +1723,7 @@ private function invalidate(): void
private function reset(): void
{
$this->isConnected = false;
$this->reconnect = false;
$this->sessions = [];
$this->clusterTime = null;
$this->operationTime = null;
Expand Down Expand Up @@ -1800,7 +1814,12 @@ private function parseResponse(string $response, int $responseLength): stdClass|
$writeError = $result->writeErrors[0] ?? null;

if (\is_object($writeError) && isset($writeError->errmsg, $writeError->code)) {
throw new Exception($writeError->errmsg, $writeError->code);
$code = (int)$writeError->code;
if (self::isPrimaryChange($code)) {
$this->invalidate(preserveSessions: true);
}

throw new Exception($writeError->errmsg, $code);
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}

$this->invalidate();
Expand All @@ -1811,6 +1830,10 @@ private function parseResponse(string $response, int $responseLength): stdClass|
$code = (int)($result->code ?? 0);
$name = (string)($result->codeName ?? 'MongoError');

if (self::isPrimaryChange($code)) {
$this->invalidate(preserveSessions: true);
}

throw new Exception('E' . $code . ' ' . $name . ': ' . $result->errmsg, $code);
}

Expand All @@ -1831,6 +1854,11 @@ private function parseResponse(string $response, int $responseLength): stdClass|
throw new Exception('Invalid unsuccessful response');
}

private static function isPrimaryChange(int $code): bool
{
return \in_array($code, self::PRIMARY_CHANGE_ERRORS, true);
}

/**
* Check if an exception represents a transient transaction error.
*
Expand Down Expand Up @@ -2039,6 +2067,20 @@ public function getSessionState(array $session): array
*/
private function validateConnection(): void
{
if ($this->reconnect) {
$sessions = $this->sessions;
$this->reconnect = false;

try {
$this->connect();
} finally {
$this->sessions = $sessions;
if (!$this->isConnected || !$this->client->isConnected()) {
$this->reconnect = true;
}
}
}

if (!$this->isConnected) {
throw new Exception('Client is not connected to MongoDB');
}
Expand Down
148 changes: 147 additions & 1 deletion tests/ClientTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -490,6 +490,141 @@ public function testDecodedMongoCommandErrorDoesNotCloseTransport(): void
$this->assertTrue($transport->open);
}

public function testPrimaryChangeErrorsCloseTransportAndPreserveSessions(): void
{
foreach ([189, 10107, 11602, 13435, 13436] as $code) {
$transport = new SyncTransportDouble();
$transport->receives = [[
'result' => $this->frame([
'ok' => 0.0,
'errmsg' => 'primary changed',
'code' => $code,
'codeName' => 'PrimaryChanged',
]),
'error' => 0,
]];
$client = $this->client($transport);
$this->seedState($client);
$sessions = $this->get($client, 'sessions');

$exception = $this->receiveException($client);

try {
$this->assertSame($code, $exception->getCode());
$this->assertSame([[true]], $transport->closes);
$this->assertSame($sessions, $this->get($client, 'sessions'));
$this->assertFalse($this->get($client, 'isConnected'));
$this->assertTrue($this->get($client, 'reconnect'));
$this->assertConnectionContextCleared($client);
} finally {
$this->set($client, 'sessions', []);
}
}
}

public function testPrimaryChangeWriteErrorClosesTransportAndPreservesSessions(): void
{
$transport = new SyncTransportDouble();
$transport->receives = [[
'result' => $this->frame([
'ok' => 1.0,
'writeErrors' => [[
'errmsg' => 'not primary',
'code' => 10107,
]],
]),
'error' => 0,
]];
$client = $this->client($transport);
$this->seedState($client);
$sessions = $this->get($client, 'sessions');

$exception = $this->receiveException($client);

try {
$this->assertSame(10107, $exception->getCode());
$this->assertSame([[true]], $transport->closes);
$this->assertSame($sessions, $this->get($client, 'sessions'));
$this->assertFalse($this->get($client, 'isConnected'));
$this->assertTrue($this->get($client, 'reconnect'));
$this->assertConnectionContextCleared($client);
} finally {
$this->set($client, 'sessions', []);
}
}

public function testPermanentWriteErrorDoesNotCloseTransport(): void
{
$transport = new SyncTransportDouble();
$transport->receives = [[
'result' => $this->frame([
'ok' => 1.0,
'writeErrors' => [[
'errmsg' => 'duplicate key',
'code' => 11000,
]],
]),
'error' => 0,
]];
$client = $this->client($transport);

$exception = $this->receiveException($client);

$this->assertSame(11000, $exception->getCode());
$this->assertSame([], $transport->closes);
$this->assertTrue($transport->open);
}

public function testTransactionRetriesOnAFreshTransportAfterPrimaryChange(): void
{
$transport = new SyncTransportDouble();
$transport->receives = [
[
'result' => $this->frame([
'ok' => 0.0,
'errmsg' => 'not primary',
'code' => 10107,
'codeName' => 'NotWritablePrimary',
]),
'error' => 0,
],
['result' => $this->frame(['ok' => 1.0]), 'error' => 0],
['result' => $this->frame(['ok' => 1.0]), 'error' => 0],
['result' => $this->frame(['ok' => 1.0]), 'error' => 0],
];
$client = $this->reconnectingClient($transport);
$identifier = (object) ['id' => new Binary(str_repeat("\1", 16), Binary::TYPE_GENERIC)];
$session = ['id' => $identifier, 'sessionId' => 'session'];
$this->set($client, 'isConnected', true);
$this->set($client, 'sessions', [
'session' => [
'id' => $identifier,
'state' => Client::TRANSACTION_NONE,
'txnNumber' => 0,
'lastUse' => time(),
'operationTime' => null,
'clusterTime' => null,
'options' => [],
'retryableWriteNumber' => 0,
],
]);

$result = $client->withTransaction(
$session,
fn (array $transaction): mixed => $client->query(['ping' => 1, 'session' => $transaction]),
['retryDelayMs' => 0],
);
$sessions = $this->get($client, 'sessions');

$this->assertSame(1.0, $result->ok);
$this->assertSame(Client::TRANSACTION_COMMITTED, $sessions['session']['state']);
$this->assertSame(2, $sessions['session']['txnNumber']);
$this->assertSame([[true]], $transport->closes);
$this->assertCount(1, $transport->connects);
$this->assertFalse($this->get($client, 'reconnect'));
$this->set($client, 'sessions', []);
}

private function assertStateCleared(Client $client): void
{
$this->assertFalse($this->get($client, 'isConnected'));
Expand Down Expand Up @@ -562,7 +697,18 @@ private function reconnectingClient(SwooleClient|CoroutineClient $transport): Re
private function seedState(Client $client): void
{
$this->seedConnectionContext($client);
$this->set($client, 'sessions', ['session' => ['id' => 'session']]);
$this->set($client, 'sessions', [
'session' => [
'id' => (object) ['id' => 'session'],
'state' => Client::TRANSACTION_NONE,
'txnNumber' => 0,
'lastUse' => time(),
'operationTime' => null,
'clusterTime' => null,
'options' => [],
'retryableWriteNumber' => 0,
],
]);
}

private function seedConnectionContext(Client $client): void
Expand Down
Loading