diff --git a/fli/search/_wire.py b/fli/search/_wire.py index ee889b32..e57cd7ab 100644 --- a/fli/search/_wire.py +++ b/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): diff --git a/fli/search/client.py b/fli/search/client.py index 78852835..18cbdbd6 100644 --- a/fli/search/client.py +++ b/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 diff --git a/fli/search/dates.py b/fli/search/dates.py index b4396199..e13ff054 100644 --- a/fli/search/dates.py +++ b/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 diff --git a/fli/search/exceptions.py b/fli/search/exceptions.py index ced740a8..fa8763e4 100644 --- a/fli/search/exceptions.py +++ b/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 diff --git a/fli/search/flights.py b/fli/search/flights.py index 62c159a2..ca72bec8 100644 --- a/fli/search/flights.py +++ b/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) diff --git a/tests/search/test_client.py b/tests/search/test_client.py index b05174e8..69cf2250 100644 --- a/tests/search/test_client.py +++ b/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 diff --git a/tests/search/test_wire.py b/tests/search/test_wire.py index 05fa22e9..cedc412f 100644 --- a/tests/search/test_wire.py +++ b/tests/search/test_wire.py @@ -2,7 +2,7 @@ import json -from fli.search._wire import iter_wrb_chunks, parse_first_wrb_payload +from fli.search._wire import iter_wrb_chunks, parse_first_wrb_payload, wrb_error_code def _single_chunk(payload): @@ -129,6 +129,49 @@ def test_multiple_wrb_chunks_all_yielded_with_mixed_rows(self): assert chunks == [[10], [20], [30]] +def _error_envelope(code=13, type_url="type.googleapis.com/travel.frontend.flights.ErrorResponse"): + """Build a real-shape HTTP-200 ErrorResponse body (see issue #200).""" + row = ["wrb.fr", None, None, None, None, [code, None, [[type_url, [[None, [], 0]]]]]] + return ")]}'\n\n" + json.dumps([row]) + + +class TestWrbErrorCode: + def test_detects_internal_error_envelope(self): + assert wrb_error_code(_error_envelope(13)) == 13 + + def test_detects_unavailable_error_envelope(self): + assert wrb_error_code(_error_envelope(14)) == 14 + + def test_returns_none_for_normal_data_chunk(self): + body = _single_chunk([1, "alpha", [2, 3]]) + assert wrb_error_code(body) is None + + def test_returns_none_for_empty_body(self): + assert wrb_error_code("") is None + + def test_returns_none_when_first_row_is_data_even_if_later_error(self): + # A data row first means a successful response; don't misread a + # trailing diagnostic row as an error. + good = json.dumps([7]) + body = ")]}'\n\n" + json.dumps([["wrb.fr", None, good]]) + assert wrb_error_code(body) is None + + def test_non_int_code_returns_sentinel(self): + assert wrb_error_code(_error_envelope("oops")) == -1 + + def test_error_block_without_error_type_is_ignored(self): + body = _error_envelope(13, type_url="type.googleapis.com/travel.frontend.flights.Results") + assert wrb_error_code(body) is None + + def test_bytes_input_works(self): + body = _error_envelope(13) + assert wrb_error_code(body.encode("utf-8")) == 13 + + def test_error_envelope_yields_no_data_chunks(self): + # The envelope must not leak through as a phantom data chunk. + assert list(iter_wrb_chunks(_error_envelope(13))) == [] + + class TestParseFirstWrbPayloadEdgeCases: def test_returns_none_for_only_non_wrb_rows(self): body = ")]}'\n\n" + json.dumps([["di", 44], ["af.httprm", 43, "x"]])