add InvalidStreamException and enhance error handling in NativeIpcPeer

src/InvalidStreamException.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace StreamIpc;
+
+use RuntimeException;
+use Throwable;
+
+/**
+ * Exception thrown when a registered stream becomes invalid.
+ */
+class InvalidStreamException extends RuntimeException
+{
+    private IpcSession $session;
+
+    public function __construct(IpcSession $session, ?string $message = null, int $code = 0, ?Throwable $previous = null)
+    {
+        $this->session = $session;
+        parent::__construct($message ?? 'Invalid stream resource', $code, $previous);
+    }
+
+    public function getSession(): IpcSession
+    {
+        return $this->session;
+    }
+}

src/NativeIpcPeer.php

@@ -4,6 +4,8 @@
 namespace StreamIpc;
 
 use RuntimeException;
+use TypeError;
+use ValueError;
 use StreamIpc\Transport\MessageTransport;
 use StreamIpc\Transport\NativeMessageTransport;
 
@@ -145,9 +147,24 @@ public function tick(?float $timeout = null): void
         }
 
         $writes = $except = null;
-        if (@stream_select($reads, $writes, $except, $sec, $usec) <= 0) {
-            // no streams ready or error occurred
-            return;
+        try {
+            if (@stream_select($reads, $writes, $except, $sec, $usec) <= 0) {
+                // no streams ready or error occurred
+                return;
+            }
+        } catch (TypeError|ValueError $e) {
+            foreach ($this->readSet as $key => $stream) {
+                $test = [$stream];
+                $w = $ex = null;
+                try {
+                    stream_select($test, $w, $ex, 0, 0);
+                } catch (TypeError|ValueError) {
+                    [$session] = $this->fdMap[$key];
+                    throw new InvalidStreamException($session, null, 0, $e);
+                }
+            }
+
+            throw $e;
         }
 
         foreach ($reads as $stream) {

tests/Unit/NativeIpcPeerInvalidStreamTest.php

@@ -0,0 +1,26 @@
+<?php
+namespace StreamIpc\Tests\Unit;
+
+use PHPUnit\Framework\TestCase;
+use StreamIpc\InvalidStreamException;
+use StreamIpc\NativeIpcPeer;
+
+final class NativeIpcPeerInvalidStreamTest extends TestCase
+{
+    public function testTickThrowsInvalidStreamExceptionWithSession(): void
+    {
+        [$a, $b] = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        $peer = new NativeIpcPeer();
+        $session = $peer->createStreamSession($a, $a, $b);
+        fclose($b);
+
+        try {
+            $peer->tick();
+            $this->fail('No exception thrown');
+        } catch (InvalidStreamException $e) {
+            $this->assertSame($session, $e->getSession());
+        } finally {
+            fclose($a);
+        }
+    }
+}