fix(search): detect and retry Google's HTTP-200 ErrorResponse envelope (#200)

fli/search/_wire.py

@@ -34,12 +34,15 @@
 _PREFIX = b")]}'"
 
 
-def iter_wrb_chunks(body: str | bytes) -> Iterator[Any]:
-    """Yield the inner JSON object of every ``wrb.fr`` chunk in ``body``.
-
-    Robust to single-chunk responses with no length headers (the older
-    ``GetShoppingResults`` / ``GetCalendarGraph`` shape) — those are parsed
-    by falling back to a single JSON load over the trimmed body.
+def _iter_outer(body: str | bytes) -> Iterator[Any]:
+    """Yield each top-level parsed list (one per response chunk) in ``body``.
+
+    This is the raw framing layer: it strips the JSONP prefix, walks the
+    byte-length-prefixed chunk stream and JSON-decodes each chunk's outer
+    array — but does *not* descend into ``wrb.fr`` rows. Callers that want
+    the decoded inner payloads use :func:`iter_wrb_chunks`; callers that
+    need to inspect the row shape itself (e.g. to detect Google's error
+    envelope) iterate the outer rows directly.
     """
     if isinstance(body, str):
         raw = body.encode("utf-8")
@@ -57,11 +60,9 @@ def iter_wrb_chunks(body: str | bytes) -> Iterator[Any]:
     # Fast path: no length headers (legacy single-chunk responses).
     if not (b"0" <= raw[:1] <= b"9"):
         try:
-            outer = json.loads(raw.decode("utf-8"))
+            yield json.loads(raw.decode("utf-8"))
         except (ValueError, json.JSONDecodeError, UnicodeDecodeError):
             logger.warning("Failed to decode single-chunk wrb.fr body as JSON", exc_info=True)
-            return
-        yield from _chunks_from_outer(outer)
         return
 
     cursor = 0
@@ -87,13 +88,74 @@ def iter_wrb_chunks(body: str | bytes) -> Iterator[Any]:
         payload = raw[cursor : cursor + chunk_bytes]
         cursor += chunk_bytes
         try:
-            outer = json.loads(payload.strip().decode("utf-8"))
+            yield json.loads(payload.strip().decode("utf-8"))
         except (ValueError, json.JSONDecodeError, UnicodeDecodeError):
             logger.warning("Discarding malformed wrb.fr chunk", exc_info=True)
             continue
+
+
+def iter_wrb_chunks(body: str | bytes) -> Iterator[Any]:
+    """Yield the inner JSON object of every ``wrb.fr`` chunk in ``body``.
+
+    Robust to single-chunk responses with no length headers (the older
+    ``GetShoppingResults`` / ``GetCalendarGraph`` shape) — those are parsed
+    by falling back to a single JSON load over the trimmed body.
+    """
+    for outer in _iter_outer(body):
         yield from _chunks_from_outer(outer)
 
 
+def wrb_error_code(body: str | bytes) -> int | None:
+    """Return the gRPC status code if the first ``wrb.fr`` row is an error envelope.
+
+    Google's FlightsFrontendService sometimes answers an otherwise valid
+    request with **HTTP 200** whose ``wrb.fr`` row carries no inner data
+    payload but instead a typed ``ErrorResponse`` block::
+
+        ["wrb.fr", null, null, null, null, [13, null,
+            [["type.googleapis.com/travel.frontend.flights.ErrorResponse", ...]]]]
+
+    The ``13`` is a canonical gRPC status code (``INTERNAL``). Left
+    undetected this row decodes to "no inner string" and the search
+    silently looks like an empty result — see issue #200, where a transient
+    Google-side outage made every query return zero flights with no error.
+
+    Returns the integer status code (or ``-1`` if the code field isn't an
+    int) when the first ``wrb.fr`` row is an error envelope; ``None`` when
+    the first row carries a normal data payload or no envelope is present.
+    """
+    for outer in _iter_outer(body):
+        if not isinstance(outer, list):
+            continue
+        for row in outer:
+            if not (isinstance(row, list) and len(row) >= 3 and row[0] == "wrb.fr"):
+                continue
+            # A normal data row carries the inner payload as a JSON string.
+            if isinstance(row[2], str) and row[2]:
+                return None
+            # An error envelope carries [code, null, [[type_url, ...]]] at row[5].
+            if len(row) >= 6 and isinstance(row[5], list) and row[5] and _is_error_block(row[5]):
+                code = row[5][0]
+                return code if isinstance(code, int) else -1
+            # First wrb.fr row is neither data nor a recognisable error.
+            return None
+    return None
+
+
+def _is_error_block(block: Any) -> bool:
+    """Return True when ``block`` is ``[code, null, [[type_url, ...]]]`` with an Error type."""
+    try:
+        details = block[2]
+    except (IndexError, TypeError):
+        return False
+    if not isinstance(details, list):
+        return False
+    for entry in details:
+        if isinstance(entry, list) and entry and isinstance(entry[0], str) and "Error" in entry[0]:
+            return True
+    return False
+
+
 def _chunks_from_outer(outer: Any) -> Iterator[Any]:
     """Walk a top-level chunk list and yield decoded inner-JSON payloads."""
     if not isinstance(outer, list):

fli/search/client.py

@@ -24,10 +24,12 @@
 import threading
 from typing import TYPE_CHECKING, Any
 
-from tenacity import retry, stop_after_attempt, wait_exponential
+from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential
 
 from fli.search._concurrency import TokenBucketRateLimiter
+from fli.search._wire import wrb_error_code
 from fli.search.exceptions import (
+    SearchBackendError,
     SearchClientError,
     SearchConnectionError,
     SearchHTTPError,
@@ -219,3 +221,44 @@ def get_client() -> Client:
             if client is None:
                 client = Client()
     return client
+
+
+def _is_retryable_backend_error(exc: BaseException) -> bool:
+    """Retry predicate: only transient ``ErrorResponse`` envelopes (see issue #200)."""
+    return isinstance(exc, SearchBackendError) and exc.retryable
+
+
+@retry(
+    retry=retry_if_exception(_is_retryable_backend_error),
+    stop=stop_after_attempt(4),
+    wait=wait_exponential(multiplier=0.5, max=8),
+    reraise=True,
+)
+def post_rpc(client: Client, url: str, encoded: str) -> str:
+    """POST an ``f.req`` body to a FlightsFrontendService endpoint and return the body text.
+
+    Wraps :meth:`Client.post` (which already retries network/HTTP faults)
+    with detection of Google's HTTP-200 ``ErrorResponse`` envelope: when the
+    response carries a gRPC error instead of data, raise
+    :class:`SearchBackendError` so transient codes (INTERNAL, UNAVAILABLE,
+    …) are retried with exponential backoff rather than silently decoding to
+    an empty result. This is the fix for issue #200, where a transient
+    Google-side outage made every search look like "no flights found".
+    """
+    response = client.post(
+        url=url,
+        data=f"f.req={encoded}",
+        impersonate="chrome",
+        allow_redirects=True,
+    )
+    response.raise_for_status()
+
+    code = wrb_error_code(response.text)
+    if code is not None:
+        raise SearchBackendError(
+            f"Google Flights returned a backend error (gRPC status {code}) instead of "
+            "results. This is usually a transient server-side issue — please try again "
+            "in a moment.",
+            code=code,
+        )
+    return response.text

fli/search/dates.py

@@ -17,7 +17,7 @@
 from fli.search._concurrency import parallel_map
 from fli.search._urls import with_locale_params
 from fli.search._wire import parse_first_wrb_payload
-from fli.search.client import get_client
+from fli.search.client import get_client, post_rpc
 
 logger = logging.getLogger(__name__)
 
@@ -174,15 +174,9 @@ def _search_chunk(
         encoded_filters = filters.encode()
         url = with_locale_params(self.BASE_URL, currency, language, country)
 
-        response = self.client.post(
-            url=url,
-            data=f"f.req={encoded_filters}",
-            impersonate="chrome",
-            allow_redirects=True,
-        )
-        response.raise_for_status()
+        text = post_rpc(self.client, url, encoded_filters)
 
-        data = parse_first_wrb_payload(response.text)
+        data = parse_first_wrb_payload(text)
         if data is None:
             return None
 

fli/search/exceptions.py

@@ -28,3 +28,43 @@ def __init__(self, message: str, *, status_code: int | None = None):
         """Store the HTTP status alongside the message for richer logging."""
         super().__init__(message)
         self.status_code = status_code
+
+
+class SearchBackendError(SearchClientError):
+    """Google Flights answered HTTP 200 with a typed ``ErrorResponse`` envelope.
+
+    The FlightsFrontendService intermittently returns a gRPC error (e.g.
+    ``INTERNAL`` / status 13) instead of results, while still using a 200
+    status code — so the HTTP layer can't see it. Before this was detected
+    the response decoded to an empty result and a transient Google-side
+    outage looked like "no flights found" (issue #200).
+
+    ``code`` is the canonical gRPC status from the envelope. Transient
+    server-side codes are retried by the client; codes that signal a bad
+    request (the ``NON_RETRYABLE`` set below) are surfaced immediately so
+    we don't hammer Google with a request it will always reject.
+    """
+
+    # Canonical gRPC codes that mean "the request itself is wrong" — no
+    # amount of retrying will help, so fail fast. Everything else
+    # (INTERNAL, UNAVAILABLE, unknown, ...) is treated as transient.
+    NON_RETRYABLE: frozenset[int] = frozenset(
+        {
+            3,  # INVALID_ARGUMENT
+            5,  # NOT_FOUND
+            7,  # PERMISSION_DENIED
+            9,  # FAILED_PRECONDITION
+            11,  # OUT_OF_RANGE
+            16,  # UNAUTHENTICATED
+        }
+    )
+
+    def __init__(self, message: str, *, code: int | None = None):
+        """Store the gRPC status code alongside the message."""
+        super().__init__(message)
+        self.code = code
+
+    @property
+    def retryable(self) -> bool:
+        """Whether this error is worth retrying (transient server-side fault)."""
+        return self.code not in self.NON_RETRYABLE

fli/search/flights.py

@@ -28,7 +28,7 @@
 from fli.search._urls import with_locale_params
 from fli.search._urls import with_locale_params as _with_locale_params  # noqa: F401
 from fli.search._wire import iter_wrb_chunks, parse_first_wrb_payload
-from fli.search.client import get_client
+from fli.search.client import get_client, post_rpc
 
 logger = logging.getLogger(__name__)
 
@@ -163,15 +163,9 @@ def _fetch_flights(
         encoded = filters.encode()
         url = with_locale_params(self.BASE_URL, currency, language, country)
 
-        response = self.client.post(
-            url=url,
-            data=f"f.req={encoded}",
-            impersonate="chrome",
-            allow_redirects=True,
-        )
-        response.raise_for_status()
+        text = post_rpc(self.client, url, encoded)
 
-        inner = parse_first_wrb_payload(response.text)
+        inner = parse_first_wrb_payload(text)
         if inner is None:
             return None
 
@@ -328,20 +322,14 @@ def get_booking_options(
 
         encoded = self._encode_booking_payload(token, prepared)
         url = with_locale_params(self.BOOKING_URL, currency, language, country)
-        response = self.client.post(
-            url=url,
-            data=f"f.req={encoded}",
-            impersonate="chrome",
-            allow_redirects=True,
-        )
-        response.raise_for_status()
+        text = post_rpc(self.client, url, encoded)
 
         # Booking responses are typically split into two wrb.fr chunks
         # (vendor list + price refinements). Materialise both before
         # parsing so we can parse them in parallel — each chunk is a few
         # hundred KB of pure-Python tree walking, GIL-bound but cheap to
         # overlap with the next chunk's JSON decode (which releases the GIL).
-        chunks = list(iter_wrb_chunks(response.text))
+        chunks = list(iter_wrb_chunks(text))
         if not chunks:
             return []
         parsed = parallel_map(parse_booking_chunk, chunks)

tests/search/test_client.py

@@ -2,13 +2,16 @@
 
 from __future__ import annotations
 
+import json
 from unittest.mock import MagicMock
 
 import pytest
+from tenacity import wait_none
 
 import fli.search.client as client_module
-from fli.search.client import _host_from_url, _wrap_request_error, get_client
+from fli.search.client import _host_from_url, _wrap_request_error, get_client, post_rpc
 from fli.search.exceptions import (
+    SearchBackendError,
     SearchClientError,
     SearchConnectionError,
     SearchHTTPError,
@@ -101,6 +104,85 @@ def test_empty_string_returns_empty(self):
         assert result == ""
 
 
+def _resp(text: str):
+    """Return a minimal stand-in for a curl_cffi Response."""
+    r = MagicMock()
+    r.text = text
+    r.raise_for_status = MagicMock()
+    return r
+
+
+def _ok_body():
+    return ")]}'\n\n" + json.dumps([["wrb.fr", None, json.dumps([[1, "data"]])]])
+
+
+def _error_body(code: int):
+    type_url = "type.googleapis.com/travel.frontend.flights.ErrorResponse"
+    row = ["wrb.fr", None, None, None, None, [code, None, [[type_url, [[None, [], 0]]]]]]
+    return ")]}'\n\n" + json.dumps([row])
+
+
+@pytest.fixture(autouse=True)
+def _no_backoff_sleep():
+    """Strip the exponential wait from post_rpc so retry tests run instantly."""
+    original = post_rpc.retry.wait
+    post_rpc.retry.wait = wait_none()
+    yield
+    post_rpc.retry.wait = original
+
+
+class TestPostRpcBackendErrors:
+    """post_rpc must turn Google's HTTP-200 error envelope into a retryable error (issue #200)."""
+
+    def test_returns_body_text_on_success(self):
+        c = MagicMock()
+        c.post.return_value = _resp(_ok_body())
+        assert post_rpc(c, "https://x/y", "ENC") == _ok_body()
+        assert c.post.call_count == 1
+
+    def test_retries_transient_internal_error_then_succeeds(self):
+        c = MagicMock()
+        c.post.side_effect = [_resp(_error_body(13)), _resp(_error_body(13)), _resp(_ok_body())]
+        assert post_rpc(c, "https://x/y", "ENC") == _ok_body()
+        assert c.post.call_count == 3
+
+    def test_exhausts_retries_and_raises_backend_error(self):
+        c = MagicMock()
+        c.post.return_value = _resp(_error_body(13))
+        with pytest.raises(SearchBackendError) as exc_info:
+            post_rpc(c, "https://x/y", "ENC")
+        assert exc_info.value.code == 13
+        # stop_after_attempt(4) → exactly four POSTs before giving up.
+        assert c.post.call_count == 4
+
+    def test_non_retryable_code_fails_fast(self):
+        c = MagicMock()
+        c.post.return_value = _resp(_error_body(3))  # INVALID_ARGUMENT
+        with pytest.raises(SearchBackendError) as exc_info:
+            post_rpc(c, "https://x/y", "ENC")
+        assert exc_info.value.code == 3
+        assert exc_info.value.retryable is False
+        assert c.post.call_count == 1
+
+    def test_post_receives_freq_body(self):
+        c = MagicMock()
+        c.post.return_value = _resp(_ok_body())
+        post_rpc(c, "https://x/y", "MYENCODED")
+        _, kwargs = c.post.call_args
+        assert kwargs["data"] == "f.req=MYENCODED"
+        assert kwargs["impersonate"] == "chrome"
+
+
+class TestSearchBackendErrorRetryable:
+    def test_transient_codes_are_retryable(self):
+        for code in (13, 14, 4, 8, None, -1):
+            assert SearchBackendError("x", code=code).retryable is True
+
+    def test_bad_request_codes_are_not_retryable(self):
+        for code in (3, 5, 7, 9, 11, 16):
+            assert SearchBackendError("x", code=code).retryable is False
+
+
 class TestGetClientSingleton:
     def test_returns_client_instance(self):
         from fli.search.client import Client