Fix database lease ownership, transaction cleanup and read extensions

docs/todo.md

@@ -51,7 +51,7 @@
 
 ## HTTP Server
 
-- Integrate a server-level HTTP/2 stream-cancel event once Swoole exposes it in a supported release, and raise the `ext-swoole` constraint to that release when integrating it. The current `RST_STREAM` path removes the stream without notifying PHP, so an AI response producer waiting on a silent provider cannot stop immediately when its caller cancels that stream. Use the event's connection and stream identities with Hypervel's active-response map to cancel only the affected opted-in producer. Preserve sibling streams, application close callbacks and terminable work. Whole-connection closes continue using existing onClose handling; no native per-response subscription is needed. Keep the AI package usable on existing supported Swoole releases with connection-close handling where available, failed-write cleanup and provider timeouts; document the remaining HTTP/2 limitation. This enhancement is not an AI-port release prerequisite. Use the actual released event name and arguments rather than polling or anticipating an unreleased API.
+- Integrate a server-level HTTP/2 stream-cancel event once Swoole exposes it in a supported release, and raise the `ext-swoole` constraint to that release when integrating it. The current `RST_STREAM` path removes the stream without notifying PHP, so an AI response producer waiting on a silent provider cannot stop immediately when its caller cancels that stream. Add stream identity to Hypervel's active-response registrations when integrating the released event, then use its connection and stream identities to cancel only the affected opted-in producer. Preserve sibling streams, application close callbacks and terminable work. Whole-connection closes continue using existing onClose handling; no native per-response subscription is needed. Keep the AI package usable on existing supported Swoole releases with connection-close handling where available, failed-write cleanup and provider timeouts; document the remaining HTTP/2 limitation. This enhancement is not an AI-port release prerequisite. Use the actual released event name and arguments rather than polling or anticipating an unreleased API.
 - Require a Swoole release that resets signal-listener state in forked server workers before releasing Hypervel 0.4. In Swoole 6.2.3 and earlier, a worker forked after the manager calls `Process::signal()` inherits the listener count, so `Coroutine\System::waitSignal()` fails in it. Hypervel's SIGINT shutdown handling registers a manager callback in both server modes, so after a reload, `max_request` recycling or a crash restart, replacement workers stop receiving configured signal handlers and Artisan traps. Once a fixed release is verified, raise the `ext-swoole` constraint and remove the version skip from `ShutdownOnInterruptListenerTest::testReplacementWorkersKeepTheirSignalHandlers()`.
 - Remove trailer-stream one-chunk lookahead once the minimum supported Swoole release includes [swoole-src#6124](https://github.com/swoole/swoole-src/pull/6124). Current releases send an empty `END_STREAM` DATA frame before trailer HEADERS when `end()` receives no body after `write()`, so `ResponseBridge` retains the final chunk for `end($chunk)` and delays delivery by one chunk. Once fixed, raise the `ext-swoole` constraint, write every chunk immediately, emit trailers, call bare `end()`, invert the deterministic bridge ordering tests, and add real gRPC incremental-delivery coverage.
 

src/database/src/Connection.php

@@ -1101,7 +1101,8 @@ public function hasPinnedSession(): bool
     {
         return $this->sessionPinDepth > 0
             || $this->transactions > 0
-            || $this->foreignKeyConstraintSuppressionDepth > 0;
+            || $this->foreignKeyConstraintSuppressionDepth > 0
+            || $this->hasPhysicalTransaction();
     }
 
     /**
@@ -1614,6 +1615,16 @@ private function throwUnsupportedTransactionException(): never
      */
     abstract public function inTransaction(): bool;
 
+    /**
+     * Determine whether any owned driver resource has an active transaction.
+     *
+     * @internal
+     */
+    public function hasPhysicalTransaction(): bool
+    {
+        return $this->inTransaction();
+    }
+
     /**
      * Set the transaction manager instance on the connection.
      */

src/database/src/ConnectionResolver.php

@@ -159,7 +159,7 @@ public function connection(UnitEnum|string|null $name = null): ConnectionInterfa
             CoroutineContext::forget($leaseContextKey);
             unset($this->nonCoroutineConnections[$connectionOwnerName]);
 
-            $this->discardFailedConnection($pooledConnection, $exception);
+            $this->discardFailedConnection($owner ?? $pooledConnection, $exception);
 
             throw $exception;
         }
@@ -252,12 +252,12 @@ protected function getLeaseContextKey(string $name): string
     }
 
     /**
-     * Discard a failed connection while preserving cancellation precedence.
+     * Discard a failed connection owner while preserving cancellation precedence.
      */
-    protected function discardFailedConnection(PooledConnection $pooledConnection, Throwable $exception): void
+    protected function discardFailedConnection(ConnectionLease|PooledConnection $owner, Throwable $exception): void
     {
         try {
-            $pooledConnection->discard();
+            $owner->discard();
         } catch (CanceledException $cancellation) {
             if (! $exception instanceof CanceledException) {
                 throw $cancellation;

src/database/src/PdoConnection.php

@@ -767,18 +767,30 @@ protected function forgetDriverResources(): void
     protected function disconnectDriverResources(): void
     {
         $pdo = $this->getRawPdo();
+        $readPdo = $this->getRawReadPdo();
         $exception = null;
 
         try {
-            if ($pdo instanceof PDO && $pdo->inTransaction()) {
-                $pdo->rollBack();
-                $this->invalidateSessionState($pdo);
-            }
-        } catch (Throwable $throwable) {
-            $this->markSessionStateUnknown($pdo);
+            foreach ($readPdo === $pdo ? [$pdo] : [$pdo, $readPdo] as $handle) {
+                if (! $handle instanceof PDO) {
+                    continue;
+                }
 
-            if (! $this->causedByLostConnection($throwable)) {
-                $exception = $throwable;
+                try {
+                    if ($handle->inTransaction()) {
+                        $handle->rollBack();
+                        $this->invalidateSessionState($handle);
+                    }
+                } catch (Throwable $throwable) {
+                    $this->markSessionStateUnknown($handle);
+
+                    if (! $this->causedByLostConnection($throwable)
+                        && ($exception === null
+                            || ($throwable instanceof CanceledException && ! $exception instanceof CanceledException))
+                    ) {
+                        $exception = $throwable;
+                    }
+                }
             }
         } finally {
             $this->forgetDriverResources();
@@ -880,6 +892,19 @@ public function inTransaction(): bool
         return $this->pdo instanceof PDO && $this->pdo->inTransaction();
     }
 
+    /**
+     * Determine whether either open PDO handle has an active transaction.
+     *
+     * @internal
+     */
+    public function hasPhysicalTransaction(): bool
+    {
+        return $this->inTransaction()
+            || ($this->readPdo instanceof PDO
+                && $this->readPdo !== $this->pdo
+                && $this->readPdo->inTransaction());
+    }
+
     /**
      * Run the statement to start a new transaction.
      */

src/database/src/Pool/ConnectionLease.php

@@ -8,6 +8,7 @@
 use Hypervel\Context\NonCopyableContext;
 use Hypervel\Database\Connectors\ConnectionFactory;
 use Hypervel\Database\PdoConnection;
+use LogicException;
 use PDO;
 use Swoole\Coroutine\CanceledException;
 use Throwable;
@@ -21,6 +22,8 @@ class ConnectionLease implements NonCopyableContext
 
     protected ?PooledConnection $pooledConnection;
 
+    protected bool $ended = false;
+
     /** @var Closure(): PDO */
     protected readonly Closure $pdoResolver;
 
@@ -57,6 +60,10 @@ public function __construct(
     protected function resolvePdo(bool $read = false): PDO
     {
         if ($this->pooledConnection === null) {
+            if ($this->ended) {
+                throw new LogicException('This database connection is no longer available because the coroutine or task that resolved it has finished or failed to set it up. Resolve the connection where you use it.');
+            }
+
             /** @var PooledConnection $pooledConnection */
             $pooledConnection = $this->pool->borrow();
             $this->pooledConnection = $pooledConnection;
@@ -113,23 +120,25 @@ public function reconnect(): PdoConnection
     public function releaseIfIdle(): void
     {
         if (! $this->connection->hasPinnedSession()) {
-            $this->release();
+            $this->pooledConnection?->release();
         }
     }
 
     /**
-     * Settle the currently held physical session.
+     * End logical ownership and release the currently held physical session.
      */
     public function release(): void
     {
+        $this->ended = true;
         $this->pooledConnection?->release();
     }
 
     /**
-     * Discard the currently held physical session.
+     * End logical ownership and discard the currently held physical session.
      */
     public function discard(): void
     {
+        $this->ended = true;
         $this->pooledConnection?->discard();
     }
 
@@ -148,7 +157,7 @@ public function detach(): void
     protected function discardAfterFailure(Throwable $exception): void
     {
         try {
-            $this->discard();
+            $this->pooledConnection?->discard();
         } catch (CanceledException $cancellation) {
             if (! $exception instanceof CanceledException) {
                 throw $cancellation;

src/database/src/Pool/DatabasePool.php

@@ -72,8 +72,7 @@ public function __construct(Container $container, string $name)
         }
 
         $this->config = $config;
-        $this->usesSessionLeases = $factory->getExtension($config, $connectionName->base) === null
-            && $this->hasConsistentLogicalIdentity($factory, $connectionName, $config);
+        $this->usesSessionLeases = $this->supportsSessionLeases($factory, $connectionName, $config);
 
         $poolOptions = Arr::except(
             Arr::get($poolConfig, 'pool', []),
@@ -131,23 +130,33 @@ public function usesSessionLeases(): bool
     }
 
     /**
-     * Determine whether selectable endpoints share the same logical database identity.
+     * Determine whether endpoints can share a logical connection without bypassing extensions.
      */
-    protected function hasConsistentLogicalIdentity(ConnectionFactory $factory, ConnectionName $name, array $config): bool
+    protected function supportsSessionLeases(ConnectionFactory $factory, ConnectionName $name, array $config): bool
     {
+        if ($factory->getExtension($config, $name->base) !== null) {
+            return false;
+        }
+
         $role = $name->isRead() && $factory->hasReadConfig($config) ? 'read' : 'write';
 
-        if (! isset($config[$role][0])) {
+        if (! isset($config[$role])) {
             return true;
         }
 
+        $records = isset($config[$role][0]) ? $config[$role] : [$config[$role]];
         $identity = null;
 
-        foreach ($config[$role] as $record) {
+        foreach ($records as $record) {
             $candidate = array_replace($config, [$role => $record]);
             $endpoint = $role === 'read'
                 ? $factory->configForRead($candidate)
                 : $factory->configForWrite($candidate);
+
+            if ($role === 'read' && $factory->getExtension($endpoint, $name->base) !== null) {
+                return false;
+            }
+
             $candidateIdentity = [$endpoint['driver'], $endpoint['database'], $endpoint['prefix']];
 
             if ($identity !== null && $identity !== $candidateIdentity) {

src/database/src/Pool/PooledConnection.php

@@ -480,8 +480,25 @@ public function release(): void
             $this->lease = null;
         }
 
+        $discard = false;
+
+        try {
+            // Callbacks may leave a raw transaction outside the framework counters.
+            if ($this->connection?->hasPhysicalTransaction()) {
+                $discard = true;
+                $this->logger->error('Database transaction was not committed or rolled back before release.');
+            }
+        } catch (CanceledException $transactionCancellation) {
+            $discard = true;
+            $cancellationFailure ??= $transactionCancellation;
+        } catch (Throwable $exception) {
+            $discard = true;
+            $ordinaryFailure ??= $exception;
+        }
+
         try {
-            if ($cancellationFailure === null
+            if (! $discard
+                && $cancellationFailure === null
                 && $this->connection !== null
                 && ! $this->connection->isReusable()
             ) {
@@ -494,10 +511,14 @@ public function release(): void
             $ordinaryFailure ??= $exception;
         }
 
-        $this->availableForReuse = true;
+        $this->availableForReuse = ! $discard;
 
         try {
-            $this->pool->release($this);
+            if ($discard) {
+                $this->pool->discard($this);
+            } else {
+                $this->pool->release($this);
+            }
         } catch (CanceledException $releaseCancellation) {
             $cancellationFailure ??= $releaseCancellation;
         } catch (Throwable $exception) {