Fix watch operations hanging in CI by adding client-side stream timeouts

The watchPath() and watchLogsPath() methods were creating socket connections
via fopen() without any client-side timeout, relying entirely on the Kubernetes
API server to close the connection after timeoutSeconds expires.

In GitHub Actions pull_request environments, the server-side connection closure
was not working reliably, causing fgets() to block indefinitely and tests to
hang (timeout after 35+ minutes).

This fix adds stream_set_timeout() to both methods with a timeout value based
on the query parameter timeoutSeconds (default 30s) plus a 5-second buffer.
The buffer allows the server-side timeout to fire first under normal conditions
while ensuring the client always has a fallback timeout.

Additionally, the while loop now checks for stream timeout via
stream_get_meta_data() and closes the socket gracefully if a timeout occurs.

Affected tests:
- NetworkPolicyTest (previously hanging at test renoki-co#127 in CI)
- All 30+ resources with watch operations

Verified working:
- NetworkPolicyTest: passes in 10s (was 20s)
- ReplicaSetTest: passes in 15s
- DeploymentTest: passes in 50s
- All tests complete locally without hangs

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Fix watch operations hanging indefinitely by using non-blocking streams

Problem:
Watch operations (watchPath/watchLogsPath) were hanging indefinitely in CI,
causing all 24 jobs to timeout after 25-35 minutes at test renoki-co#127 (NetworkPolicyTest).

Root Cause Analysis:
1. Original code used fgets() which blocks indefinitely waiting for data
2. First fix attempt: stream_set_timeout() - Does NOT reliably interrupt fgets()
   when waiting for data that never arrives
3. Second fix attempt: stream_select() - Does NOT work with HTTP wrapper streams
   (fopen('http://...')) - PHP limitation: "Cannot cast a filtered stream"
   stream_select() only works with socket streams, not HTTP filtered streams

Solution:
Implemented non-blocking stream approach:
- Set stream_set_blocking($sock, false) to enable non-blocking mode
- Use fread() with explicit timeout checking instead of fgets()
- Buffer incoming data and process complete lines
- Sleep 100ms when no data available (usleep(100000))
- Check overall timeout (server timeoutSeconds + 5 second buffer)
- Exit cleanly on EOF or timeout

Benefits:
- Works with HTTP wrapper streams from fopen()
- Proper timeout handling prevents indefinite hangs
- Buffered line processing handles partial reads correctly
- NetworkPolicyTest now passes in 0.22s (was hanging for 27+ minutes)
- ReplicaSetTest passes with all watch operations in 3.28s

Testing:
- NetworkPolicyTest: ✔ 27 assertions, 0.221s
- ReplicaSetTest: ✔ 62 assertions, 3.282s
- All watch operations now timeout properly

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: cuppett

src/KubernetesCluster.php

@@ -255,27 +255,76 @@ protected function watchPath(string $path, Closure $callback, array $query = ['p
     {
         $resourceClass = $this->resourceClass;
         $sock = $this->createSocketConnection($this->getCallableUrl($path, $query));
-        $data = null;
 
-        while (($data = fgets($sock)) == true) {
-            $data = @json_decode($data, true);
+        if ($sock === false) {
+            return null;
+        }
+
+        // Set stream to non-blocking mode to allow timeout handling
+        stream_set_blocking($sock, false);
 
-            ['type' => $type, 'object' => $attributes] = $data;
+        // Calculate overall timeout: server timeout + buffer for network/processing
+        $timeout = ($query['timeoutSeconds'] ?? 30) + 5;
+        $endTime = time() + $timeout;
 
-            $call = call_user_func(
-                $callback,
-                $type,
-                new $resourceClass($this, $attributes)
-            );
+        $buffer = '';
 
-            if (! is_null($call)) {
+        while (time() < $endTime) {
+            // Try to read data (non-blocking)
+            $chunk = fread($sock, 8192);
+
+            if ($chunk === false) {
+                // Error occurred
                 fclose($sock);
+                return null;
+            }
+
+            if ($chunk === '') {
+                // No data available, check if stream ended
+                if (feof($sock)) {
+                    break;
+                }
+
+                // No data yet, sleep briefly and continue
+                usleep(100000); // 100ms
+                continue;
+            }
+
+            // Append chunk to buffer
+            $buffer .= $chunk;
+
+            // Process complete lines from buffer
+            while (($pos = strpos($buffer, "\n")) !== false) {
+                $line = substr($buffer, 0, $pos);
+                $buffer = substr($buffer, $pos + 1);
+
+                if (trim($line) === '') {
+                    continue;
+                }
+
+                $data = @json_decode($line, true);
+
+                if (!$data || !isset($data['type'], $data['object'])) {
+                    continue;
+                }
+
+                ['type' => $type, 'object' => $attributes] = $data;
 
-                unset($data);
+                $call = call_user_func(
+                    $callback,
+                    $type,
+                    new $resourceClass($this, $attributes)
+                );
 
-                return $call;
+                if (! is_null($call)) {
+                    fclose($sock);
+                    return $call;
+                }
             }
         }
+
+        fclose($sock);
+        return null;
     }
 
     /**
@@ -290,19 +339,59 @@ protected function watchLogsPath(string $path, Closure $callback, array $query =
     {
         $sock = $this->createSocketConnection($this->getCallableUrl($path, $query));
 
-        $data = null;
+        if ($sock === false) {
+            return null;
+        }
+
+        // Set stream to non-blocking mode to allow timeout handling
+        stream_set_blocking($sock, false);
 
-        while (($data = fgets($sock)) == true) {
-            $call = call_user_func($callback, $data);
+        // Calculate overall timeout: server timeout + buffer for network/processing
+        $timeout = ($query['timeoutSeconds'] ?? 30) + 5;
+        $endTime = time() + $timeout;
 
-            if (! is_null($call)) {
+        $buffer = '';
+
+        while (time() < $endTime) {
+            // Try to read data (non-blocking)
+            $chunk = fread($sock, 8192);
+
+            if ($chunk === false) {
+                // Error occurred
                 fclose($sock);
+                return null;
+            }
 
-                unset($data);
+            if ($chunk === '') {
+                // No data available, check if stream ended
+                if (feof($sock)) {
+                    break;
+                }
 
-                return $call;
+                // No data yet, sleep briefly and continue
+                usleep(100000); // 100ms
+                continue;
+            }
+
+            // Append chunk to buffer
+            $buffer .= $chunk;
+
+            // Process complete lines from buffer
+            while (($pos = strpos($buffer, "\n")) !== false) {
+                $line = substr($buffer, 0, $pos);
+                $buffer = substr($buffer, $pos + 1);
+
+                $call = call_user_func($callback, $line . "\n");
+
+                if (! is_null($call)) {
+                    fclose($sock);
+                    return $call;
+                }
             }
         }
+
+        fclose($sock);
+        return null;
     }
 
     /**