From 9bfd1e1a2e4e1e1c6a09b8963b330cdad54c39bd Mon Sep 17 00:00:00 2001
From: Cees-Jan Kiewiet <github@ceesjankiewiet.nl>
Date: Sat, 27 Jun 2026 19:43:05 +0200
Subject: [PATCH] Add transaction support

---
 README.md                             |  61 ++++-
 composer.json                         |   1 +
 example/Transaction.php               |  25 ++
 src/PgAsync/Client.php                | 165 ++++++------
 src/PgAsync/Connection.php            |  66 +++++
 src/PgAsync/ConnectionPool.php        | 248 +++++++++++++++++
 src/PgAsync/Message/ReadyForQuery.php |   7 +-
 src/PgAsync/Transaction.php           | 115 ++++++++
 tests/Integration/TransactionTest.php | 374 ++++++++++++++++++++++++++
 9 files changed, 964 insertions(+), 98 deletions(-)
 create mode 100644 example/Transaction.php
 create mode 100644 src/PgAsync/ConnectionPool.php
 create mode 100644 src/PgAsync/Transaction.php
 create mode 100644 tests/Integration/TransactionTest.php

diff --git a/README.md b/README.md
index cc110d6..45a3280 100644
--- a/README.md
+++ b/README.md
@@ -114,19 +114,72 @@ With [composer](https://getcomposer.org/) install into you project with:
 Install pgasync:
 ```composer require voryx/pgasync```
 
+## Example - Transactions
+
+When using transactions with connection pooling, configure `max_connections` so normal
+queries and transactions share the pool. Transactions reserve a single connection for
+their duration; if the pool is full, both transactions and regular queries wait until a
+connection becomes available.
+
+```php
+$client = new PgAsync\Client([
+    "host"            => "127.0.0.1",
+    "port"            => "5432",
+    "user"            => "matt",
+    "database"        => "matt",
+    "max_connections" => 5,
+]);
+```
+
+Use `transaction()` for a scoped transaction that auto-commits on success and rolls back on failure:
+
+```php
+$client->transaction(function (PgAsync\Transaction $tx) {
+    return $tx->executeStatement(
+        'INSERT INTO invoices(inv_no, customer_id, amount) VALUES ($1, $2, $3)',
+        ['1234A', 1, 35.75]
+    )->concat(
+        $tx->query('SELECT SUM(amount) AS balance FROM invoices WHERE customer_id = 1')
+    );
+})->subscribe(
+    function () {
+        echo "Committed.\n";
+    },
+    function ($e) {
+        echo "Rolled back: " . $e->getMessage() . "\n";
+    }
+);
+```
+
+For manual control, use `beginTransaction()` and call `commit()` or `rollback()` yourself:
+
+```php
+$client->beginTransaction()->flatMap(function (PgAsync\Transaction $tx) {
+    return $tx->executeStatement('INSERT INTO channel(name) VALUES ($1)', ['example'])
+        ->concat($tx->query('SELECT COUNT(*) AS c FROM channel'))
+        ->concat($tx->commit());
+})->subscribe(
+    function () {
+        echo "Committed.\n";
+    },
+    function ($e) {
+        echo "Failed: " . $e->getMessage() . "\n";
+    }
+);
+```
+
+All queries inside a transaction run on a single reserved connection. Regular `$client->query()` calls may use other pooled connections and run in parallel. When every connection is reserved for an open transaction, further queries and transactions wait until one is released.
+
 ## What it can do
 - Run queries (CREATE, UPDATE, INSERT, SELECT, DELETE)
 - Queue commands
 - Return results asynchronously (using Observables - you get data one row at a time as it comes from the db server)
 - Prepared statements (as parameterized queries)
 - Connection pooling (basic pooling)
-
-## What it can't quite do yet
-- Transactions (Actually though, just grab a connection and you can run your transaction on that single connection)
+- Transactions (scoped and explicit)
 
 ## What's next
 - Add more testing
-- Transactions
 - Take over the world
 
 ## Keep in mind
diff --git a/composer.json b/composer.json
index c99517c..6c3102a 100644
--- a/composer.json
+++ b/composer.json
@@ -42,6 +42,7 @@
     "wyrihaximus/react-opportunistic-tls": "^1.0.0"
   },
   "require-dev": {
+    "phpstan/phpstan": "^1.12",
     "phpunit/phpunit": ">=8.5.23 || ^6.5.5",
     "react/dns": "^1.12.0"
   },
diff --git a/example/Transaction.php b/example/Transaction.php
new file mode 100644
index 0000000..5a7e2f5
--- /dev/null
+++ b/example/Transaction.php
@@ -0,0 +1,25 @@
+<?php
+require_once __DIR__ . '/bootstrap.php';
+
+$client = new PgAsync\Client([
+    "host"     => "127.0.0.1",
+    "port"     => "5432",
+    "user"     => "matt",
+    "database" => "matt"
+]);
+
+$client->transaction(function (PgAsync\Transaction $tx) {
+    return $tx->executeStatement(
+        'INSERT INTO channel(name, description) VALUES ($1, $2)',
+        ['Test Name', 'Inserted inside a scoped transaction']
+    )->concat(
+        $tx->query("SELECT name FROM channel WHERE name = 'Test Name'")
+    );
+})->subscribe(
+    function () {
+        echo "Transaction committed.\n";
+    },
+    function ($e) {
+        echo "Transaction failed: " . $e->getMessage() . "\n";
+    }
+);
diff --git a/src/PgAsync/Client.php b/src/PgAsync/Client.php
index 621aa37..f01e116 100644
--- a/src/PgAsync/Client.php
+++ b/src/PgAsync/Client.php
@@ -4,154 +4,143 @@
 
 use PgAsync\Message\NotificationResponse;
 use React\EventLoop\LoopInterface;
+use React\Promise\PromiseInterface;
 use React\Socket\ConnectorInterface;
 use Rx\Observable;
 use Rx\Subject\Subject;
 
 class Client
 {
-    /** @var  string */
-    protected $connectString;
-
     /** @var LoopInterface */
     protected $loop;
 
-    protected $params = [];
-
-    private $parameters = [];
-
-    /** @var Connection[] */
-    private $connections = [];
-
-    /** @var boolean */
-    private $autoDisconnect;
-
-    /** @var ConnectorInterface */
-    private $connector;
-
-    /** @var int */
-    private $maxConnections = 5;
+    /** @var ConnectionPool */
+    private $connectionPool;
 
     /** @var Subject[] */
     private $listeners = [];
 
-    /** @var Connection */
+    /** @var Connection|null */
     private $listenConnection;
 
     public function __construct(array $parameters, ?LoopInterface $loop = null, ?ConnectorInterface $connector = null)
     {
-        $this->loop      = $loop ?: \EventLoop\getLoop();
-        $this->connector = $connector;
+        $this->loop = $loop ?: \EventLoop\getLoop();
+
+        $autoDisconnect = false;
+        $maxConnections = 5;
 
         if (isset($parameters['auto_disconnect'])) {
-            $this->autoDisconnect = $parameters['auto_disconnect'];
+            $autoDisconnect = $parameters['auto_disconnect'];
         }
 
         if (isset($parameters['max_connections'])) {
             if (!is_int($parameters['max_connections'])) {
                 throw new \InvalidArgumentException('`max_connections` must an be integer greater than zero.');
             }
-            $this->maxConnections = $parameters['max_connections'];
+            $maxConnections = $parameters['max_connections'];
             unset($parameters['max_connections']);
-            if ($this->maxConnections < 1) {
+            if ($maxConnections < 1) {
                 throw new \InvalidArgumentException('`max_connections` must be greater than zero.');
             }
         }
 
-        $this->parameters = $parameters;
+        $this->connectionPool = new ConnectionPool(
+            $parameters,
+            $this->loop,
+            $connector,
+            $autoDisconnect,
+            $maxConnections
+        );
     }
 
     public function query($s)
     {
-        return Observable::defer(function () use ($s) {
-            $conn = $this->getLeastBusyConnection();
-
+        return $this->connectionPool->acquire(false)->flatMap(function (Connection $conn) use ($s) {
             return $conn->query($s);
         });
     }
 
-    public function executeStatement(string $queryString, array $parameters = [])
+    public function beginTransaction(): Observable
     {
-        return Observable::defer(function () use ($queryString, $parameters) {
-            $conn = $this->getLeastBusyConnection();
+        return $this->connectionPool->acquire(true)->flatMap(function (Connection $connection) {
+            return $connection->queryUntilReady('BEGIN')->flatMap(function ($status) use ($connection) {
+                if ($status !== 'T') {
+                    $this->connectionPool->release($connection);
 
-            return $conn->executeStatement($queryString, $parameters);
+                    return Observable::error(new \RuntimeException('Expected transaction status T after BEGIN'));
+                }
+
+                return Observable::of($this->createTransaction($connection));
+            })->catch(function (\Throwable $e) use ($connection) {
+                $this->connectionPool->release($connection);
+
+                return Observable::error($e);
+            });
         });
     }
 
-    private function getLeastBusyConnection(): Connection
+    private function createTransaction(Connection $connection): Transaction
     {
-        if (count($this->connections) === 0) {
-            // try to spin up another connection to return
-            $conn = $this->createNewConnection();
-            if ($conn === null) {
-                throw new \Exception('There are no connections. Cannot find least busy one and could not create a new one.');
-            }
-
-            return $conn;
-        }
-
-        $min = $this->connections[0];
-
-        foreach ($this->connections as $connection) {
-            // if this connection is idle - just return it
-            if ($connection->getBacklogLength() === 0 && $connection->getState() === Connection::STATE_READY) {
-                return $connection;
+        return new Transaction(
+            $connection,
+            function () use ($connection) {
+                $this->connectionPool->release($connection);
             }
+        );
+    }
 
-            if ($min->getBacklogLength() > $connection->getBacklogLength()) {
-                $min = $connection;
+    /**
+     * @param callable(Transaction): (Observable|PromiseInterface|mixed) $fn
+     */
+    public function transaction(callable $fn): Observable
+    {
+        return $this->beginTransaction()->flatMap(function (Transaction $tx) use ($fn) {
+            try {
+                $result = $fn($tx);
+            } catch (\Throwable $e) {
+                return $tx->rollback()->concat(Observable::error($e));
             }
-        }
 
-        if (count($this->connections) < $this->maxConnections) {
-            return $this->createNewConnection();
-        }
+            return $this->normalizeTransactionResult($result)
+                ->concat($tx->commit())
+                ->catch(function (\Throwable $e) use ($tx) {
+                    return $tx->rollback()->concat(Observable::error($e));
+                });
+        });
+    }
 
-        return $min;
+    public function executeStatement(string $queryString, array $parameters = [])
+    {
+        return $this->connectionPool->acquire(false)->flatMap(function (Connection $conn) use ($queryString, $parameters) {
+            return $conn->executeStatement($queryString, $parameters);
+        });
     }
 
-    public function getIdleConnection(): Connection
+    private function normalizeTransactionResult($result): Observable
     {
-        // we want to get the first available one
-        // this will keep the connections at the front the busiest
-        // and then we can add an idle timer to the connections
-        foreach ($this->connections as $connection) {
-            // need to figure out different states (in trans etc.)
-            if ($connection->getState() === Connection::STATE_READY) {
-                return $connection;
-            }
+        if ($result instanceof Observable) {
+            return $result;
         }
 
-        if (count($this->connections) >= $this->maxConnections) {
-            return null;
+        if ($result instanceof PromiseInterface) {
+            return Observable::fromPromise($result);
         }
 
-        return $this->createNewConnection();
+        return Observable::of($result);
     }
 
-    private function createNewConnection()
+    /**
+     * @return Connection|null
+     */
+    public function getIdleConnection()
     {
-        // no idle connections were found - spin up new one
-        $connection = new Connection($this->parameters, $this->loop, $this->connector);
-        if ($this->autoDisconnect) {
-            return $connection;
-        }
-
-        $this->connections[] = $connection;
-
-        $connection->on('close', function () use ($connection) {
-            $this->connections = array_values(array_filter($this->connections, function ($c) use ($connection) {
-                return $connection !== $c;
-            }));
-        });
-
-        return $connection;
+        return $this->connectionPool->getIdleConnection();
     }
 
     public function getConnectionCount(): int
     {
-        return count($this->connections);
+        return $this->connectionPool->getConnectionCount();
     }
 
     /**
@@ -162,9 +151,7 @@ public function getConnectionCount(): int
      */
     public function closeNow()
     {
-        foreach ($this->connections as $connection) {
-            $connection->disconnect();
-        }
+        $this->connectionPool->closeAll();
     }
 
     public function listen(string $channel): Observable
@@ -186,7 +173,7 @@ public function listen(string $channel): Observable
 
         $this->listeners[$channel] = Observable::defer(function () use ($channel) {
             if ($this->listenConnection === null) {
-                $this->listenConnection = $this->createNewConnection();
+                $this->listenConnection = $this->connectionPool->createConnection();
             }
 
             if ($this->listenConnection === null) {
diff --git a/src/PgAsync/Connection.php b/src/PgAsync/Connection.php
index 66cc634..b375c64 100644
--- a/src/PgAsync/Connection.php
+++ b/src/PgAsync/Connection.php
@@ -149,6 +149,9 @@ class Connection extends EventEmitter
      */
     private $backendTransactionStatus = 'UNKNOWN';
 
+    /** @var array<int, array{resolve: callable, reject: callable}> */
+    private $readyForQueryWaiters = [];
+
     /** @var  bool */
     private $auto_disconnect = false;
     private $tls = self::TLS_MODE_PREFER;
@@ -297,6 +300,65 @@ function ($a, CommandInterface $command) {
             0);
     }
 
+    public function getBackendTransactionStatus(): string
+    {
+        return $this->backendTransactionStatus;
+    }
+
+    /**
+     * @return Observable<string>
+     */
+    public function whenReadyForQuery(): Observable
+    {
+        if ($this->queryState === static::STATE_READY) {
+            return Observable::of($this->backendTransactionStatus);
+        }
+
+        return new AnonymousObservable(function (ObserverInterface $observer) {
+            $this->readyForQueryWaiters[] = [
+                'resolve' => function ($status) use ($observer) {
+                    $observer->onNext($status);
+                    $observer->onCompleted();
+                },
+                'reject' => function ($e) use ($observer) {
+                    $observer->onError($e);
+                },
+            ];
+
+            return new EmptyDisposable();
+        });
+    }
+
+    /**
+     * @return Observable<string>
+     */
+    public function queryUntilReady(string $query): Observable
+    {
+        return $this->query($query)
+            ->concat($this->whenReadyForQuery())
+            ->takeLast(1);
+    }
+
+    private function resolveReadyForQueryWaiters(string $status)
+    {
+        $waiters = $this->readyForQueryWaiters;
+        $this->readyForQueryWaiters = [];
+
+        foreach ($waiters as $waiter) {
+            $waiter['resolve']($status);
+        }
+    }
+
+    private function rejectReadyForQueryWaiters(\Throwable $e)
+    {
+        $waiters = $this->readyForQueryWaiters;
+        $this->readyForQueryWaiters = [];
+
+        foreach ($waiters as $waiter) {
+            $waiter['reject']($e);
+        }
+    }
+
     public function onData($data)
     {
         while (strlen($data) > 0) {
@@ -564,6 +626,9 @@ private function handleReadyForQuery(ReadyForQuery $message)
         $this->connStatus     = $this::CONNECTION_OK;
         $this->queryState     = $this::STATE_READY;
         $this->currentCommand = null;
+        $this->backendTransactionStatus = $message->getBackendTransactionStatus();
+        $this->resolveReadyForQueryWaiters($this->backendTransactionStatus);
+        $this->emit('ready');
         $this->processQueue();
     }
 
@@ -577,6 +642,7 @@ private function failAllCommandsWith(?\Throwable $e = null)
         $e = $e ?: new \Exception('unknown error');
 
         $this->notificationSubject->onError($e);
+        $this->rejectReadyForQueryWaiters($e);
 
         while (count($this->commandQueue) > 0) {
             $c = array_shift($this->commandQueue);
diff --git a/src/PgAsync/ConnectionPool.php b/src/PgAsync/ConnectionPool.php
new file mode 100644
index 0000000..571cd2c
--- /dev/null
+++ b/src/PgAsync/ConnectionPool.php
@@ -0,0 +1,248 @@
+<?php
+
+namespace PgAsync;
+
+use React\EventLoop\LoopInterface;
+use React\Socket\ConnectorInterface;
+use Rx\Disposable\EmptyDisposable;
+use Rx\Observable;
+use Rx\Observable\AnonymousObservable;
+use Rx\ObserverInterface;
+
+class ConnectionPool
+{
+    /** @var array */
+    private $parameters;
+
+    /** @var LoopInterface */
+    private $loop;
+
+    /** @var ConnectorInterface|null */
+    private $connector;
+
+    /** @var Connection[] */
+    private $connections = [];
+
+    /** @var Connection[] */
+    private $reservedConnections = [];
+
+    /** @var array<int, array{forTransaction: bool, resolve: callable, reject: callable}> */
+    private $connectionWaiters = [];
+
+    /** @var bool */
+    private $autoDisconnect;
+
+    /** @var int */
+    private $maxConnections;
+
+    public function __construct(
+        array $parameters,
+        LoopInterface $loop,
+        ConnectorInterface $connector = null,
+        bool $autoDisconnect = false,
+        int $maxConnections = 5
+    ) {
+        $this->parameters     = $parameters;
+        $this->loop           = $loop;
+        $this->connector      = $connector;
+        $this->autoDisconnect = $autoDisconnect;
+        $this->maxConnections = $maxConnections;
+    }
+
+    public function acquire(bool $forTransaction): Observable
+    {
+        $connection = $this->tryAcquire($forTransaction);
+        if ($connection !== null) {
+            return Observable::of($connection);
+        }
+
+        return new AnonymousObservable(function (ObserverInterface $observer) use ($forTransaction) {
+            $this->connectionWaiters[] = [
+                'forTransaction' => $forTransaction,
+                'resolve'        => function ($connection) use ($observer) {
+                    $observer->onNext($connection);
+                    $observer->onCompleted();
+                },
+                'reject'         => function ($e) use ($observer) {
+                    $observer->onError($e);
+                },
+            ];
+
+            return new EmptyDisposable();
+        });
+    }
+
+    public function release(Connection $connection)
+    {
+        unset($this->reservedConnections[spl_object_hash($connection)]);
+        $this->processConnectionWaiters();
+    }
+
+    /**
+     * @return Connection|null
+     */
+    public function getIdleConnection()
+    {
+        foreach ($this->connections as $connection) {
+            if ($this->isConnectionReserved($connection)) {
+                continue;
+            }
+
+            if ($connection->getState() === Connection::STATE_READY) {
+                return $connection;
+            }
+        }
+
+        if (count($this->connections) >= $this->maxConnections) {
+            return null;
+        }
+
+        return $this->createConnection();
+    }
+
+    public function createConnection(): Connection
+    {
+        $connection = new Connection($this->parameters, $this->loop, $this->connector);
+        if ($this->autoDisconnect) {
+            return $connection;
+        }
+
+        $this->connections[] = $connection;
+
+        $connection->on('ready', function () {
+            $this->processConnectionWaiters();
+        });
+
+        $connection->on('close', function () use ($connection) {
+            $this->release($connection);
+            $this->connections = array_values(array_filter($this->connections, function ($c) use ($connection) {
+                return $connection !== $c;
+            }));
+        });
+
+        return $connection;
+    }
+
+    public function getConnectionCount(): int
+    {
+        return count($this->connections);
+    }
+
+    public function closeAll()
+    {
+        $this->rejectConnectionWaiters(new \RuntimeException('Client closed'));
+
+        foreach ($this->connections as $connection) {
+            $connection->disconnect();
+        }
+    }
+
+    /**
+     * @return Connection|null
+     */
+    private function tryAcquire(bool $forTransaction)
+    {
+        if ($forTransaction) {
+            return $this->tryReserveConnection();
+        }
+
+        return $this->tryAcquireQueryConnection();
+    }
+
+    /**
+     * @return Connection|null
+     */
+    private function tryAcquireQueryConnection()
+    {
+        foreach ($this->connections as $connection) {
+            if ($this->isConnectionReserved($connection)) {
+                continue;
+            }
+
+            if ($connection->getState() === Connection::STATE_READY && $connection->getBacklogLength() === 0) {
+                return $connection;
+            }
+        }
+
+        if ($this->autoDisconnect || count($this->connections) < $this->maxConnections) {
+            return $this->createConnection();
+        }
+
+        $leastBusy = null;
+
+        foreach ($this->connections as $connection) {
+            if ($this->isConnectionReserved($connection)) {
+                continue;
+            }
+
+            if ($leastBusy === null || $leastBusy->getBacklogLength() > $connection->getBacklogLength()) {
+                $leastBusy = $connection;
+            }
+        }
+
+        return $leastBusy;
+    }
+
+    /**
+     * @return Connection|null
+     */
+    private function tryReserveConnection()
+    {
+        foreach ($this->connections as $connection) {
+            if ($this->isConnectionReserved($connection)) {
+                continue;
+            }
+
+            if ($connection->getState() === Connection::STATE_READY && $connection->getBacklogLength() === 0) {
+                $this->markConnectionReserved($connection);
+
+                return $connection;
+            }
+        }
+
+        if ($this->autoDisconnect || count($this->connections) < $this->maxConnections) {
+            $connection = $this->createConnection();
+            if ($connection !== null) {
+                $this->markConnectionReserved($connection);
+
+                return $connection;
+            }
+        }
+
+        return null;
+    }
+
+    private function markConnectionReserved(Connection $connection)
+    {
+        $this->reservedConnections[spl_object_hash($connection)] = $connection;
+    }
+
+    private function processConnectionWaiters()
+    {
+        while (count($this->connectionWaiters) > 0) {
+            $waiter = $this->connectionWaiters[0];
+            $connection = $this->tryAcquire($waiter['forTransaction']);
+            if ($connection === null) {
+                return;
+            }
+
+            array_shift($this->connectionWaiters);
+            $waiter['resolve']($connection);
+        }
+    }
+
+    private function rejectConnectionWaiters(\Throwable $e)
+    {
+        $waiters = $this->connectionWaiters;
+        $this->connectionWaiters = [];
+
+        foreach ($waiters as $waiter) {
+            $waiter['reject']($e);
+        }
+    }
+
+    private function isConnectionReserved(Connection $connection): bool
+    {
+        return array_key_exists(spl_object_hash($connection), $this->reservedConnections);
+    }
+}
diff --git a/src/PgAsync/Message/ReadyForQuery.php b/src/PgAsync/Message/ReadyForQuery.php
index e328df0..0acaaad 100644
--- a/src/PgAsync/Message/ReadyForQuery.php
+++ b/src/PgAsync/Message/ReadyForQuery.php
@@ -6,7 +6,7 @@ class ReadyForQuery implements ParserInterface
 {
     use ParserTrait;
 
-    private $backendTransactionStatus;
+    private $backendTransactionStatus = 'UNKNOWN';
 
     /**
      * @inheritDoc
@@ -32,10 +32,7 @@ public static function getMessageIdentifier(): string
         return 'Z';
     }
 
-    /**
-     * @return mixed
-     */
-    public function getBackendTransactionStatus()
+    public function getBackendTransactionStatus(): string
     {
         return $this->backendTransactionStatus;
     }
diff --git a/src/PgAsync/Transaction.php b/src/PgAsync/Transaction.php
new file mode 100644
index 0000000..6a30980
--- /dev/null
+++ b/src/PgAsync/Transaction.php
@@ -0,0 +1,115 @@
+<?php
+
+namespace PgAsync;
+
+use Rx\Observable;
+
+final class Transaction
+{
+    /** @var Connection */
+    private $connection;
+
+    /** @var callable */
+    private $release;
+
+    /** @var bool */
+    private $finished = false;
+
+    /**
+     * @param callable $release
+     */
+    public function __construct(Connection $connection, callable $release)
+    {
+        $this->connection = $connection;
+        $this->release    = $release;
+    }
+
+    public function query(string $queryString): Observable
+    {
+        if ($this->finished) {
+            return $this->inactiveError();
+        }
+
+        return $this->connection->query($queryString);
+    }
+
+    public function executeStatement(string $queryString, array $parameters = []): Observable
+    {
+        if ($this->finished) {
+            return $this->inactiveError();
+        }
+
+        return $this->connection->executeStatement($queryString, $parameters);
+    }
+
+    public function commit(): Observable
+    {
+        if ($this->finished) {
+            return $this->inactiveError();
+        }
+
+        return $this->finishWithStatus('COMMIT', [$this, 'validateCommitStatus']);
+    }
+
+    public function rollback(): Observable
+    {
+        if ($this->finished) {
+            return $this->inactiveError();
+        }
+
+        return $this->finishWithStatus('ROLLBACK', [$this, 'validateRollbackStatus']);
+    }
+
+    private function finishWithStatus(string $sql, callable $validate): Observable
+    {
+        return $this->connection->queryUntilReady($sql)->flatMap(function ($status) use ($validate) {
+            $error = $validate($status);
+            if ($error !== null) {
+                return Observable::error($error);
+            }
+
+            $this->markFinished();
+
+            return Observable::of($status);
+        });
+    }
+
+    /**
+     * @return \Throwable|null
+     */
+    private function validateCommitStatus(string $status)
+    {
+        if ($status === 'E') {
+            return new \RuntimeException('Cannot commit: transaction is in failed state');
+        }
+
+        if ($status !== 'I') {
+            return new \RuntimeException('Expected idle status after COMMIT');
+        }
+
+        return null;
+    }
+
+    /**
+     * @return \Throwable|null
+     */
+    private function validateRollbackStatus(string $status)
+    {
+        if ($status !== 'I') {
+            return new \RuntimeException('Expected idle status after ROLLBACK');
+        }
+
+        return null;
+    }
+
+    private function inactiveError(): Observable
+    {
+        return Observable::error(new \RuntimeException('Transaction has already been committed or rolled back'));
+    }
+
+    private function markFinished()
+    {
+        $this->finished = true;
+        ($this->release)();
+    }
+}
diff --git a/tests/Integration/TransactionTest.php b/tests/Integration/TransactionTest.php
new file mode 100644
index 0000000..9f67cb2
--- /dev/null
+++ b/tests/Integration/TransactionTest.php
@@ -0,0 +1,374 @@
+<?php
+
+namespace PgAsync\Tests\Integration;
+
+use PgAsync\Client;
+use PgAsync\Transaction;
+use Rx\Observable;
+use Rx\Observer\CallbackObserver;
+
+class TransactionTest extends TestCase
+{
+    private function client(array $parameters = []): Client
+    {
+        return self::clientFromEnv(array_merge([
+            'user'     => $this->getDbUser(),
+            'password' => $this::getDbUser(),
+            'database' => $this::getDbName(),
+        ], $parameters), $this->getLoop());
+    }
+
+    private function countThings(Client $client, string $thingType): Observable
+    {
+        return $client->executeStatement(
+            'SELECT COUNT(*) AS c FROM thing WHERE thing_type = $1',
+            [$thingType]
+        );
+    }
+
+    public function testCommitPersistsChanges()
+    {
+        $client = $this->client();
+        $thingType = 'txn_commit_' . uniqid();
+        $count = null;
+
+        $client->beginTransaction()->flatMap(function (Transaction $tx) use ($thingType) {
+            return $tx->executeStatement(
+                'INSERT INTO thing(thing_type, thing_cost) VALUES ($1, $2)',
+                [$thingType, 1.00]
+            )->concat($tx->commit());
+        })->flatMap(function () use ($client, $thingType) {
+            return $this->countThings($client, $thingType);
+        })->subscribe(new CallbackObserver(
+            function ($row) use (&$count) {
+                $count = (int) $row['c'];
+                $this->stopLoop();
+            },
+            function ($e) {
+                $this->fail('Count query failed: ' . $e->getMessage());
+                $this->stopLoop();
+            }
+        ));
+
+        $this->runLoopWithTimeout(5);
+
+        $this->assertSame(1, $count);
+
+        $client->closeNow();
+        $this->getLoop()->run();
+    }
+
+    public function testRollbackDiscardsChanges()
+    {
+        $client = $this->client();
+        $thingType = 'txn_rollback_' . uniqid();
+        $count = null;
+
+        $client->beginTransaction()->flatMap(function (Transaction $tx) use ($thingType) {
+            return $tx->executeStatement(
+                'INSERT INTO thing(thing_type, thing_cost) VALUES ($1, $2)',
+                [$thingType, 1.00]
+            )->concat($tx->rollback());
+        })->flatMap(function () use ($client, $thingType) {
+            return $this->countThings($client, $thingType);
+        })->subscribe(new CallbackObserver(
+            function ($row) use (&$count) {
+                $count = (int) $row['c'];
+                $this->stopLoop();
+            },
+            function ($e) {
+                $this->fail('Count query failed: ' . $e->getMessage());
+                $this->stopLoop();
+            }
+        ));
+
+        $this->runLoopWithTimeout(5);
+
+        $this->assertSame(0, $count);
+
+        $client->closeNow();
+        $this->getLoop()->run();
+    }
+
+    public function testExecuteStatementInsideTransaction()
+    {
+        $client = $this->client();
+        $thingType = 'txn_stmt_' . uniqid();
+        $count = null;
+
+        $client->beginTransaction()->flatMap(function (Transaction $tx) use ($thingType) {
+            return $tx->executeStatement(
+                'INSERT INTO thing(thing_type, thing_description, thing_cost) VALUES ($1, $2, $3)',
+                [$thingType, 'inside transaction', 9.99]
+            )->concat($tx->commit());
+        })->flatMap(function () use ($client, $thingType) {
+            return $this->countThings($client, $thingType);
+        })->subscribe(new CallbackObserver(
+            function ($row) use (&$count) {
+                $count = (int) $row['c'];
+                $this->stopLoop();
+            },
+            function ($e) {
+                $this->fail('Count query failed: ' . $e->getMessage());
+                $this->stopLoop();
+            }
+        ));
+
+        $this->runLoopWithTimeout(5);
+
+        $this->assertSame(1, $count);
+
+        $client->closeNow();
+        $this->getLoop()->run();
+    }
+
+    public function testConcurrentClientQueryUsesOtherConnection()
+    {
+        $client = $this->client(['max_connections' => 2]);
+        $thingType = 'txn_concurrent_' . uniqid();
+        $parallelResult = null;
+
+        $client->beginTransaction()->flatMap(function (Transaction $tx) use ($client, $thingType, &$parallelResult) {
+            return $tx->executeStatement(
+                'INSERT INTO thing(thing_type, thing_cost) VALUES ($1, $2)',
+                [$thingType, 1.00]
+            )->concat(
+                $client->executeStatement("SELECT 'parallel' AS label", [])
+                    ->doOnNext(function ($row) use (&$parallelResult) {
+                        $parallelResult = $row['label'];
+                    })
+            )->concat($tx->rollback());
+        })->subscribe(new CallbackObserver(
+            function () {
+                $this->stopLoop();
+            },
+            function ($e) {
+                $this->fail('Concurrent query failed: ' . $e->getMessage());
+                $this->stopLoop();
+            }
+        ));
+
+        $this->runLoopWithTimeout(5);
+
+        $this->assertSame('parallel', $parallelResult);
+        $this->assertGreaterThanOrEqual(2, $client->getConnectionCount());
+
+        $client->closeNow();
+        $this->getLoop()->run();
+    }
+
+    public function testFailedTransactionRequiresRollback()
+    {
+        $client = $this->client();
+        $error = null;
+        $afterRollback = null;
+        $transaction = null;
+
+        $client->beginTransaction()->flatMap(function (Transaction $tx) use (&$transaction) {
+            $transaction = $tx;
+
+            return $tx->query('SELECT 1/0');
+        })->subscribe(
+            function () {
+                $this->fail('Expected query to fail');
+            },
+            function ($e) use ($client, &$error, &$afterRollback, &$transaction) {
+                $error = $e;
+
+                $transaction->rollback()->concat(
+                    $client->executeStatement("SELECT 'ok' AS label", [])
+                        ->doOnNext(function ($row) use (&$afterRollback) {
+                            $afterRollback = $row['label'];
+                        })
+                )->subscribe(new CallbackObserver(
+                    function () {
+                        $this->stopLoop();
+                    },
+                    function ($e) {
+                        $this->fail('Post-rollback query failed: ' . $e->getMessage());
+                        $this->stopLoop();
+                    }
+                ));
+            }
+        );
+
+        $this->runLoopWithTimeout(5);
+
+        $this->assertNotNull($error);
+        $this->assertSame('ok', $afterRollback);
+
+        $client->closeNow();
+        $this->getLoop()->run();
+    }
+
+    public function testScopedTransactionAutoCommit()
+    {
+        $client = $this->client();
+        $thingType = 'txn_scoped_commit_' . uniqid();
+        $count = null;
+
+        $client->transaction(function (Transaction $tx) use ($thingType) {
+            return $tx->executeStatement(
+                'INSERT INTO thing(thing_type, thing_cost) VALUES ($1, $2)',
+                [$thingType, 2.00]
+            );
+        })->flatMap(function () use ($client, $thingType) {
+            return $this->countThings($client, $thingType);
+        })->subscribe(new CallbackObserver(
+            function ($row) use (&$count) {
+                $count = (int) $row['c'];
+                $this->stopLoop();
+            },
+            function ($e) {
+                $this->fail('Count query failed: ' . $e->getMessage());
+                $this->stopLoop();
+            }
+        ));
+
+        $this->runLoopWithTimeout(5);
+
+        $this->assertSame(1, $count);
+
+        $client->closeNow();
+        $this->getLoop()->run();
+    }
+
+    public function testScopedTransactionAutoRollback()
+    {
+        $client = $this->client();
+        $thingType = 'txn_scoped_rollback_' . uniqid();
+        $count = null;
+        $error = null;
+
+        $client->transaction(function (Transaction $tx) use ($thingType) {
+            return $tx->executeStatement(
+                'INSERT INTO thing(thing_type, thing_cost) VALUES ($1, $2)',
+                [$thingType, 2.00]
+            )->concat($tx->query('SELECT 1/0'));
+        })->subscribe(
+            function () {
+                $this->fail('Expected scoped transaction to fail');
+            },
+            function ($e) use ($client, $thingType, &$error, &$count) {
+                $error = $e;
+
+                $this->countThings($client, $thingType)->subscribe(new CallbackObserver(
+                    function ($row) use (&$count) {
+                        $count = (int) $row['c'];
+                        $this->stopLoop();
+                    },
+                    function ($e) {
+                        $this->fail('Count query failed: ' . $e->getMessage());
+                        $this->stopLoop();
+                    }
+                ));
+            }
+        );
+
+        $this->runLoopWithTimeout(5);
+
+        $this->assertNotNull($error);
+        $this->assertSame(0, $count);
+
+        $client->closeNow();
+        $this->getLoop()->run();
+    }
+
+    public function testDoubleCommitRejected()
+    {
+        $client = $this->client();
+        $error = null;
+
+        $client->beginTransaction()->flatMap(function (Transaction $tx) {
+            return $tx->commit()->concat($tx->commit());
+        })->subscribe(
+            function () {
+                $this->fail('Expected second commit to error');
+                $this->stopLoop();
+            },
+            function ($e) use (&$error) {
+                $error = $e;
+                $this->stopLoop();
+            }
+        );
+
+        $this->runLoopWithTimeout(5);
+
+        $this->assertInstanceOf(\RuntimeException::class, $error);
+        $this->assertStringContainsString('already been committed', $error->getMessage());
+
+        $client->closeNow();
+        $this->getLoop()->run();
+    }
+
+    public function testBeginTransactionWaitsForAvailableConnection()
+    {
+        $client = $this->client(['max_connections' => 1]);
+        $firstStarted = false;
+        $secondStarted = false;
+        $secondBegin = null;
+
+        $client->beginTransaction()->flatMap(function (Transaction $tx) use ($client, &$firstStarted, &$secondBegin) {
+            $firstStarted = true;
+            $secondBegin = $client->beginTransaction();
+
+            return $tx->commit();
+        })->flatMap(function () use (&$secondBegin, &$secondStarted) {
+            return $secondBegin->flatMap(function (Transaction $tx2) use (&$secondStarted) {
+                $secondStarted = true;
+
+                return $tx2->commit();
+            });
+        })->subscribe(new CallbackObserver(
+            function () {
+                $this->stopLoop();
+            },
+            function ($e) {
+                $this->fail('Unexpected error: ' . $e->getMessage());
+                $this->stopLoop();
+            }
+        ));
+
+        $this->runLoopWithTimeout(5);
+
+        $this->assertTrue($firstStarted);
+        $this->assertTrue($secondStarted);
+        $this->assertSame(1, $client->getConnectionCount());
+
+        $client->closeNow();
+        $this->getLoop()->run();
+    }
+
+    public function testQueryWaitsWhenAllConnectionsAreReservedForTransactions()
+    {
+        $client = $this->client(['max_connections' => 2]);
+        $queryResult = null;
+
+        $client->beginTransaction()->flatMap(function (Transaction $tx1) use ($client, &$queryResult) {
+            return $client->beginTransaction()->flatMap(function (Transaction $tx2) use ($client, &$queryResult, $tx1) {
+                $query = $client->executeStatement("SELECT 'queued' AS label", [])
+                    ->doOnNext(function ($row) use (&$queryResult) {
+                        $queryResult = $row['label'];
+                    });
+
+                return $tx2->rollback()->concat($query)->concat($tx1->rollback());
+            });
+        })->subscribe(new CallbackObserver(
+            function () {
+                $this->stopLoop();
+            },
+            function ($e) {
+                $this->fail('Unexpected error: ' . $e->getMessage());
+                $this->stopLoop();
+            }
+        ));
+
+        $this->runLoopWithTimeout(5);
+
+        $this->assertSame('queued', $queryResult);
+        $this->assertSame(2, $client->getConnectionCount());
+
+        $client->closeNow();
+        $this->getLoop()->run();
+    }
+}