fix: capture WebSocket error messages before close (#552)

* fix: capture WebSocket error messages before close

The close frame's reason field is unreliable through proxies (often
arrives as code=1006, reason=None). Fix:

Client-side (cloud_connection.py):
- Capture 'error' type messages and store in _last_close_reason
- Now when close event fires, we have the real error from the message

Server-side (fal_app.py):
- Add 100ms delay between sending error and closing connection
- Helps ensure error message is flushed before close frame

Signed-off-by: livepeer-robot <robot@livepeer.org>

* fix: don't overwrite error message reason with close frame's None

The close handler was overwriting _last_close_reason with the close
frame's reason (usually None through proxies), losing the error message
we captured earlier.

Now:
1. Error message arrives → _last_close_reason = 'Access denied: ...'
2. Close event arrives → keep existing reason, don't overwrite with None
3. UI toast shows the real error instead of 'reason: None'

Signed-off-by: livepeer-robot <robot@livepeer.org>

* Dont show previous connection errors and verify userID against API

Signed-off-by: Max Holland <max@livepeer.org>

---------

Signed-off-by: livepeer-robot <robot@livepeer.org>
Signed-off-by: Max Holland <max@livepeer.org>
Co-authored-by: livepeer-robot <robot@livepeer.org>
Co-authored-by: Max Holland <max@livepeer.org>

Author: 3 people

frontend/src/components/Header.tsx

@@ -36,29 +36,33 @@ export function Header({
   // Track the last close code we've shown a toast for to avoid duplicates
   const lastNotifiedCloseCodeRef = useRef<number | null>(null);
 
+  // Only show "connection lost" after we've seen a successful connection this session
+  const hasBeenConnectedRef = useRef(false);
+
   // Track previous connection state to detect transitions for pipeline refresh
   const prevConnectedRef = useRef(false);
 
   // Detect unexpected disconnection and show toast
   useEffect(() => {
+    if (isConnected) {
+      hasBeenConnectedRef.current = true;
+      lastNotifiedCloseCodeRef.current = null;
+    }
+
     if (
+      hasBeenConnectedRef.current &&
       lastCloseCode !== null &&
       lastCloseCode !== lastNotifiedCloseCodeRef.current
     ) {
       console.warn(
         `[Header] Cloud WebSocket closed unexpectedly (code=${lastCloseCode}, reason=${lastCloseReason})`
       );
       toast.error("Cloud connection lost", {
-        description: `WebSocket closed (code: ${lastCloseCode}${lastCloseReason ? `, reason: ${lastCloseReason}` : ""})`,
+        description: `WebSocket closed ${lastCloseReason ? `(${lastCloseReason})` : ""}`,
         duration: 10000,
       });
       lastNotifiedCloseCodeRef.current = lastCloseCode;
     }
-
-    // Reset the notified close code when connected (so we can show it again if it disconnects later)
-    if (isConnected) {
-      lastNotifiedCloseCodeRef.current = null;
-    }
   }, [lastCloseCode, lastCloseReason, isConnected]);
 
   // Refresh pipelines when cloud connection status changes

src/scope/cloud/fal_app.py

@@ -29,9 +29,30 @@ async def validate_user_access(user_id: str) -> tuple[bool, str]:
 
     Returns (is_valid, reason) tuple.
     """
+    import urllib.error
+    import urllib.request
+
     if not user_id:
         return False, "No user ID provided"
-    return True, "Access granted"
+
+    url = f"{os.getenv('DAYDREAM_API_BASE', 'https://api.daydream.live')}/v1/users/{user_id}"
+    print(f"Validating user access for {user_id} via {url}")
+
+    def fetch_user():
+        req = urllib.request.Request(url)
+        with urllib.request.urlopen(req, timeout=10) as response:
+            return json.loads(response.read().decode())
+
+    try:
+        # Run synchronous urllib in thread pool to not block event loop
+        await asyncio.get_event_loop().run_in_executor(None, fetch_user)
+        return True, "Access granted"
+    except urllib.error.HTTPError as e:
+        if e.code == 404:
+            return False, "User not found"
+        return False, f"Failed to fetch user: {e.code}"
+    except Exception as e:
+        return False, f"Error validating user: {e}"
 
 
 class KafkaPublisher:
@@ -425,9 +446,9 @@ async def check_max_duration_exceeded() -> bool:
                 )
                 await safe_send_json(
                     {
-                        "type": "closing",
-                        "reason": "max_duration",
-                        "elapsed_seconds": elapsed_seconds,
+                        "type": "error",
+                        "error": "Max duration exceeded",
+                        "code": "MAX_DURATION_EXCEEDED",
                     }
                 )
                 return True
@@ -728,10 +749,12 @@ async def handle_message(payload: dict) -> dict | None:
                     await safe_send_json(
                         {
                             "type": "error",
-                            "error": f"Access denied: {reason}",
+                            "error": "Access denied",
                             "code": "ACCESS_DENIED",
                         }
                     )
+                    # Small delay to let error message reach client before close
+                    # (close frame often gets lost through proxies)
                     await ws.close(code=4003, reason="Access denied")
                     return None
 

src/scope/server/cloud_connection.py

@@ -289,24 +289,30 @@ async def _receive_loop(self) -> None:
                     # Get close code and reason from the WebSocket
                     close_code = self.ws.close_code if self.ws else None
                     # Try to get close reason from message data or ws object
-                    close_reason = None
+                    close_frame_reason = None
                     if msg.data:
-                        close_reason = str(msg.data)
+                        close_frame_reason = str(msg.data)
                     elif msg.extra:
-                        close_reason = str(msg.extra)
+                        close_frame_reason = str(msg.extra)
 
-                    # Store for status reporting to frontend
+                    # Store close code
                     self._last_close_code = close_code
-                    self._last_close_reason = close_reason
+                    # Only update reason if we don't already have one from an error message
+                    # (error messages sent before close are more reliable than close frame reason)
+                    if close_frame_reason and not self._last_close_reason:
+                        self._last_close_reason = close_frame_reason
+
+                    # Use the captured reason (from error message or close frame)
+                    effective_reason = self._last_close_reason
 
                     logger.warning(
-                        f"Cloud WebSocket closed (code={close_code}, reason={close_reason})"
+                        f"Cloud WebSocket closed (code={close_code}, reason={effective_reason})"
                     )
 
                     # Publish error if this was an unexpected close
                     if not self._stop_event.is_set():
                         self._publish_cloud_error(
-                            f"Cloud WebSocket closed unexpectedly (code={close_code}, reason={close_reason})",
+                            f"Cloud WebSocket closed unexpectedly (code={close_code}, reason={effective_reason})",
                             "CloudWebSocketClosed",
                             error_type="cloud_websocket_closed",
                             extra_error_fields={"close_code": close_code},
@@ -332,16 +338,26 @@ async def _receive_loop(self) -> None:
 
     async def _handle_message(self, data: dict) -> None:
         """Handle incoming WebSocket message."""
-        request_id = data.get("request_id")
+        msg_type = data.get("type")
+
+        # Capture error messages (often sent right before connection close)
+        # The close frame's reason field is unreliable through proxies, so we
+        # store the error from the message to have it when the close event fires
+        if msg_type == "error":
+            error_msg = data.get("error", "Unknown error")
+            error_code = data.get("code")
+            logger.warning(f"Cloud error message: {error_msg} (code={error_code})")
+            self._last_close_reason = error_msg
+            return
 
+        request_id = data.get("request_id")
         if request_id and request_id in self._pending_requests:
             # This is a response to a pending request
             future = self._pending_requests.pop(request_id)
             if not future.done():
                 future.set_result(data)
         else:
             # Unsolicited message (e.g., notifications)
-            msg_type = data.get("type")
             logger.debug(f"Received unsolicited message: {msg_type}")
 
     async def send_and_wait(