marko/database-readwrite

Read/write connection splitting with replica routing for Marko database connections. Wraps any existing marko/database driver connection — all write operations and transactions route to the primary, all reads route to one of your configured replicas. Uses the decorator pattern: ReadWriteConnection implements ConnectionInterface and TransactionInterface so the rest of the application code is unchanged.

Installation

composer require marko/database-readwrite

This automatically installs marko/database (the interface package) as a dependency.

Configuration

Set driver to readwrite in your config/database.php, then declare a connections block with your write primary and one or more read replicas:

config/database.php

<?php

declare(strict_types=1);

use Marko\Config\Env;

return [
    'driver' => 'readwrite',
    'connections' => [
        'write' => [
            'driver'   => 'pgsql',
            'host'     => Env::string('DB_WRITE_HOST', 'localhost'),
            'port'     => Env::int('DB_WRITE_PORT', 5432, min: 1, max: 65535),
            'database' => Env::string('DB_DATABASE', 'marko'),
            'username' => Env::string('DB_USERNAME', 'postgres'),
            'password' => Env::string('DB_PASSWORD', ''),
        ],
        'read' => [
            [
                'driver'   => 'pgsql',
                'host'     => Env::string('DB_READ_HOST_1', 'replica-1'),
                'port'     => Env::int('DB_READ_PORT_1', 5432, min: 1, max: 65535),
                'database' => Env::string('DB_DATABASE', 'marko'),
                'username' => Env::string('DB_USERNAME', 'postgres'),
                'password' => Env::string('DB_PASSWORD', ''),
            ],
            [
                'driver'   => 'pgsql',
                'host'     => Env::string('DB_READ_HOST_2', 'replica-2'),
                'port'     => Env::int('DB_READ_PORT_2', 5432, min: 1, max: 65535),
                'database' => Env::string('DB_DATABASE', 'marko'),
                'username' => Env::string('DB_USERNAME', 'postgres'),
                'password' => Env::string('DB_PASSWORD', ''),
            ],
        ],
        'read_strategy' => 'random',  // 'random' (default) or 'weighted'
    ],
];

Config Schema

Key	Type	Required	Description
connections.write	array	Yes	Single write (primary) connection config
connections.read	array[]	Yes	One or more replica connection configs
connections.read_strategy	string	No	Replica selection strategy: random (default) or weighted
read[n].weight	int	No	Required when read_strategy is weighted; positive integer
timezone	string	No	Top-level database.timezone (default UTC)

Each connection config inside write and read[] follows the same structure as a standalone driver config (e.g., marko/database-pgsql). The time zone is the exception: every node’s session is pinned to the top-level timezone (the database session time zone), so the primary and the replicas agree. A timezone inside a node config is ignored.

Replica Strategies

Random (default)

Selects a replica uniformly at random on each read query. Use when all replicas have similar capacity:

config/database.php

'connections' => [
    // ...
    'read_strategy' => 'random',
],

Weighted

Routes a proportional share of read traffic to each replica based on its weight. Use when replicas have different hardware or capacity:

config/database.php

'connections' => [
    'write' => [ /* primary config */ ],
    'read' => [
        [
            'driver' => 'pgsql',
            'host'   => 'replica-1',
            // ...
            'weight' => 3,  // receives 3/4 of reads
        ],
        [
            'driver' => 'pgsql',
            'host'   => 'replica-2',
            // ...
            'weight' => 1,  // receives 1/4 of reads
        ],
    ],
    'read_strategy' => 'weighted',
],

Weights are positive integers. The probability of selecting a replica equals its weight divided by the total of all weights. Each weight must be a positive integer or it is rejected at config parse time.

Note

WeightedReplicaSelector calculates probabilities based on the original replica list. When a replica fails and is removed during a request, the weights for the remaining replicas are not rebalanced. This is a known v1 limitation.

Sticky Writes

After any write operation or transaction, subsequent reads within the same request are routed to the write connection rather than a replica. This prevents stale reads caused by replication lag.

The sticky flag is set by:

• execute() — any INSERT, UPDATE, DELETE, or DDL statement
• query() with a statement that may write — one whose first keyword is INSERT, UPDATE, DELETE, WITH, MERGE, REPLACE or CALL — such as the INSERT ... RETURNING a repository runs to read a database-generated key back, so a following find() sees the new row
• prepare() with one of those statements
• beginTransaction() — entering a transaction
• transaction(callable $callback, int $attempts = 1, int|Closure|null $backoff = null) — the entire callback, every retried attempt included, runs on the primary. When the callback wrote anything through the connection, the flag stays set so later reads see the write; when it only read, the flag goes back to what it was before the call. A nested transaction() leaves the outer transaction’s reads on the primary either way.

Leading whitespace and any number of leading comments (-- ..., # ..., /* ... */) are skipped before the first keyword is read, so a comment cannot hide a write. A WITH query is always treated as a write, because a CTE can end in INSERT, UPDATE, DELETE or MERGE: a read-only CTE is served by the primary, which is the safe side of the trade.

use Marko\Database\Connection\ConnectionInterface;

class OrderService
{
    public function __construct(
        private ConnectionInterface $connection,
    ) {}

    public function placeOrder(array $data): int
    {
        // Writes set the sticky flag
        $orderId = $this->connection->execute(
            'INSERT INTO orders (user_id, total) VALUES (?, ?)',
            [$data['user_id'], $data['total']],
        );

        // This query routes to the write connection (sticky), not a replica,
        // so it sees the order that was just inserted
        return $this->connection->query(
            'SELECT id FROM orders WHERE id = ?',
            [$orderId],
        )[0]['id'];
    }
}

Row-locking reads (lockForUpdate(), sharedLock()) must run inside a transaction, so they always reach the primary. upsert() is an INSERT and is routed to the primary too.

The sticky flag persists until resetStickyState() is called. In a PHP-FPM application this happens automatically because each request runs in a fresh process. In long-running processes you must call it manually (see Long-Running Processes).

Reads That Must Hit the Primary

Each request starts non-sticky, so its first reads go to a replica. A read whose answer must never be stale (a session lookup after logout, a “token already used” check, a permission revoked a moment ago) runs inside onPrimary(), which sends every read in the callback to the primary:

use Marko\Database\Connection\ConnectionInterface;
use Marko\Database\Connection\PrimaryReadInterface;

$rows = $connection instanceof PrimaryReadInterface
    ? $connection->onPrimary(fn (): array => $connection->query($sql, $bindings))
    : $connection->query($sql, $bindings);

onPrimary() comes from Marko\Database\Connection\PrimaryReadInterface in marko/database. A connection without replicas does not implement it, so code that may run on either checks for the interface and runs the read directly otherwise. Like transaction(), it returns what the callback returns, and afterwards reads go back to replicas unless the callback wrote something.

marko/session-database reads and validates session rows this way, so a logged-out or regenerated session ID cannot be revived from a lagging replica.

prepare() Policy

prepare() always routes to the write connection. Prepared statements are typically used for bulk writes or repeated mutation patterns, so routing them to the primary is the safe default. Preparing a statement that may write also sets the sticky flag, since it runs on the primary once executed. There are no production callers of prepare() in the current Marko core, so this policy has no performance impact on stock setups.

Single-Request Fallback

When a read query fails on a replica with a PDOException or MarkoException, ReadWriteConnection removes that replica from the candidate pool and retries the query on the next available replica. This continues until:

• A replica responds successfully, or
• All replicas are exhausted, at which point ReadException is thrown

ReadException: All replicas failed to execute the query: <replica-1 message>; <replica-2 message>
Context: While attempting to route a read query to an available replica
Suggestion: Check that at least one replica is reachable and accepting connections

Fallback is per-request only. The replica pool is rebuilt fresh on the next request (PHP-FPM) or the next call to resetStickyState() (long-running processes).

Caution

Sticky writes (via execute() or beginTransaction()) bypass all replicas entirely --- there is no fallback path. If the write connection is unavailable the underlying driver exception is propagated directly.

Long-Running Processes

In PHP-FPM the sticky flag is cleared automatically at the end of each request because each request is a new process. In a queue worker or other long-running process the sticky flag persists for the lifetime of the process. Call resetStickyState() between jobs to restore replica routing:

ReadWriteConnection also implements Marko\Core\Contracts\ResettableInterface, so a worker that resets every registered ResettableInterface implementation between requests will clear the sticky flag automatically via reset(). Calling resetStickyState() directly remains supported for callers that don’t go through the contract.

Beyond clearing the sticky flag, reset() also rolls back any transaction left open by a request that called beginTransaction() directly and then threw before commit()/rollback(). When the write connection is itself resettable (the MySQL and PostgreSQL drivers are), reset() delegates to it, which rolls back every nested level and drops pending after-commit callbacks. Otherwise it calls rollback() once per open level, innermost first. Without this, the underlying write connection stays mid-transaction on the pooled connection, and the next request’s writes would silently land inside the previous request’s abandoned transaction. The rollback only runs when a transaction is actually open; if the rollback itself throws, the sticky flag is still cleared before the exception propagates, so the connection is never left permanently sticky even when a reset only partially succeeds.

use Marko\Database\ReadWrite\Connection\ReadWriteConnection;
use Marko\Database\Connection\ConnectionInterface;

class JobWorker
{
    public function __construct(
        private ConnectionInterface $connection,
    ) {}

    public function run(Job $job): void
    {
        try {
            $job->handle();
        } finally {
            // Always reset between jobs — even if the job threw
            if ($this->connection instanceof ReadWriteConnection) {
                $this->connection->resetStickyState();
            }
        }
    }
}

Multi-Driver Limitation

ReadWriteConnection binds ConnectionInterface as a container instance. This is the same interface binding used by all other database drivers, so only one driver can be active at a time. You cannot, for example, have a PostgreSQL read/write split alongside a MySQL connection in the same container.

This constraint is inherited from the single-binding design of ConnectionInterface in marko/database and applies equally to marko/database-pgsql and marko/database-mysql.

Customization

Override ReadWriteConnection using the Preferences pattern. Define a Preference in your module’s module.php to substitute your own implementation wherever ReadWriteConnection is resolved:

module.php

<?php

declare(strict_types=1);

use Marko\Database\ReadWrite\Connection\ReadWriteConnection;
use Acme\Database\ReadWrite\Connection\CustomReadWriteConnection;

return [
    'preferences' => [
        ReadWriteConnection::class => CustomReadWriteConnection::class,
    ],
];

Your CustomReadWriteConnection must extend ReadWriteConnection (or independently implement ConnectionInterface and TransactionInterface).

API Reference

ReadWriteConnection

Implements ConnectionInterface, TransactionInterface, PendingAfterCommitInterface, PrimaryReadInterface, and ResettableInterface. Routes reads to replicas and writes to the primary.

Method	Routes To	Description
query(string $sql, array $bindings = []): array	Read (replica or write if sticky); write for a statement that may write (sets sticky)	Execute a SELECT and return all rows
execute(string $sql, array $bindings = []): int	Write (sets sticky)	Execute a write statement; returns affected row count
prepare(string $sql): StatementInterface	Write (always; sets sticky for a statement that may write)	Prepare a statement for repeated execution
lastInsertId(): int	Write	Get the last auto-increment ID
connect(): void	Write	Establish the write connection
disconnect(): void	Write	Close the write connection
isConnected(): bool	Write	Check if the write connection is open
beginTransaction(): void	Write (sets sticky)	Start a transaction on the write connection
commit(): void	Write	Commit the current transaction
rollback(): void	Write	Roll back the current transaction
inTransaction(): bool	Write	Check if a transaction is active
transactionLevel(): int	Write	Number of open transaction levels (savepoints included)
transaction(callable $callback, int $attempts = 1, int|Closure|null $backoff = null): mixed	Write (sets sticky; kept if the callback wrote)	Run a callback inside an auto-managed transaction (a savepoint when nested); $attempts and $backoff are passed to the write connection, which waits between attempts and retries the outermost transaction on a deadlock or serialization failure. The sticky flag is set for the callback duration; afterwards it stays set if the callback wrote anything, otherwise it is restored to its previous value
onPrimary(callable $callback): mixed	Write (sets sticky; kept if the callback wrote)	PrimaryReadInterface method: run the callback with every read on the primary and return its result, for reads that must not see replica lag
afterCommit(callable $callback): void	Write	Run the callback after the write connection’s outermost commit
afterRollback(callable $callback): void	Write	Run the callback if its transaction level rolls back
runPendingAfterCommitCallbacks(): void	Write	Run the write connection’s queued afterCommit() callbacks without committing (for test helpers such as RefreshDatabase); throws TransactionException when the write connection does not implement PendingAfterCommitInterface
driverName(): string	Write (delegates)	Return the write connection’s driver name (e.g. 'mysql', 'pgsql')
supportsReturning(): bool	Write (delegates)	Whether the write connection supports INSERT ... RETURNING
quoteIdentifier(string $identifier): string	Write (delegates)	Quote a table or column name with the write connection’s quoting rule
resetStickyState(): void	—	Clear the sticky flag; subsequent reads route to replicas again
reset(): void	—	ResettableInterface contract method; rolls back every open transaction level (delegating to a resettable write connection) and clears the sticky flag

ReadException

Thrown when all replicas fail during a single read query.

Factory	Description
ReadException::allReplicasFailed(array $messages): self	Builds the exception with a semicolon-joined summary of each replica’s error message

ReadWriteConnectionConfig

Parses and validates the connections block from config/database.php.

Factory	Throws	Description
ReadWriteConnectionConfig::fromArray(array $config): self	ReadWriteConfigException	Validates presence of write, non-empty read[], and valid read_strategy

RandomReplicaSelector

Selects a replica uniformly at random. Default strategy.

WeightedReplicaSelector

Selects a replica proportionally to its configured weight. Constructed with an array of integer weights parallel to the replica array.