fix(search): decode wrb.fr chunks by JSON grammar, and surface Google's error envelopes

fli/search/__init__.py

@@ -1,5 +1,6 @@
 from .dates import DatePrice, SearchDates
 from .exceptions import (
+    SearchBackendError,
     SearchClientError,
     SearchConnectionError,
     SearchHTTPError,
@@ -15,4 +16,5 @@
     "SearchTimeoutError",
     "SearchConnectionError",
     "SearchHTTPError",
+    "SearchBackendError",
 ]

fli/search/_wire.py

@@ -3,20 +3,24 @@
 The Service returns JSONP-flavoured responses of the form::
 
     )]}'\n\n
-    <chunk1_byte_len>\n
+    <chunk1_len>\n
     [["wrb.fr", null, "<inner JSON string>"]]
-    <chunk2_byte_len>\n
+    <chunk2_len>\n
     [["wrb.fr", null, "<inner JSON string>"]]
     ...
 
 `GetShoppingResults` and `GetCalendarGraph` happen to emit a single chunk so
 the legacy parsers in this package could get away with `lstrip(")]}'")`.
 `GetBookingResults` emits two chunks, so we need a proper multi-chunk reader.
 
-Important quirk: the length headers count UTF-8 **bytes**, not Python string
-characters. When the response contains any non-ASCII characters (which it
-sometimes does — airport names, airline names) the offsets diverge, so the
-reader must operate over the byte representation of the body.
+Important quirk: the length headers are **not** a dependable frame
+delimiter. They count the chunk plus its two surrounding newlines, but in
+characters rather than UTF-8 bytes, so any response carrying non-ASCII text
+(accented airport or airline names) desynchronises a byte-oriented reader —
+and an ASCII response hides the difference entirely. Rather than encode a
+guess about Google's convention, this reader ignores the announced length
+and lets the JSON grammar delimit each chunk, which is correct either way.
+The headers are still used to re-synchronise after a malformed chunk.
 
 This module centralises that reader and exposes :func:`iter_wrb_chunks` which
 yields the decoded inner JSON of each ``wrb.fr`` chunk.
@@ -26,76 +30,181 @@
 
 import json
 import logging
+import re
 from collections.abc import Iterator
-from typing import Any
+from typing import Any, NamedTuple
+
+from fli.search.exceptions import SearchBackendError
 
 logger = logging.getLogger(__name__)
 
-_PREFIX = b")]}'"
+_PREFIX = ")]}'"
+
+# Framing noise between two chunks: the length header and the newlines
+# around it. Skipped wholesale — the header's value is never trusted.
+_FRAMING_CHARS = "0123456789 \t\r\n"
+
+# A chunk boundary in the raw stream: newline, decimal length header,
+# newline, then the "[" that opens the next chunk. Literal newlines are
+# escaped inside JSON strings, so this can never match within a payload.
+_CHUNK_BOUNDARY = re.compile(r"\n\d+\n(?=\[)")
+
+# Google reports rejected requests with gRPC's canonical status codes.
+_STATUS_NAMES = {
+    1: "CANCELLED",
+    2: "UNKNOWN",
+    3: "INVALID_ARGUMENT",
+    4: "DEADLINE_EXCEEDED",
+    5: "NOT_FOUND",
+    6: "ALREADY_EXISTS",
+    7: "PERMISSION_DENIED",
+    8: "RESOURCE_EXHAUSTED",
+    9: "FAILED_PRECONDITION",
+    10: "ABORTED",
+    11: "OUT_OF_RANGE",
+    12: "UNIMPLEMENTED",
+    13: "INTERNAL",
+    14: "UNAVAILABLE",
+    15: "DATA_LOSS",
+    16: "UNAUTHENTICATED",
+}
+
+# Error details are echoed into the exception message, so cap them.
+_MAX_DETAIL_CHARS = 200
+
+
+class _BackendStatus(NamedTuple):
+    """A rejection reported by an error-envelope ``wrb.fr`` row."""
+
+    code: int
+    detail: str | None
 
 
 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.
-    """
-    if isinstance(body, str):
-        raw = body.encode("utf-8")
-    else:
-        raw = body
+    ``GetShoppingResults`` / ``GetCalendarGraph`` shape) — those parse the
+    same way, since chunk boundaries are derived from the JSON itself.
 
-    raw = raw.lstrip()
-    if raw.startswith(_PREFIX):
-        raw = raw[len(_PREFIX) :]
-    raw = raw.lstrip()
+    A response may mix payload chunks and an error row. Raising the moment
+    the error row is read would make the outcome depend on how far the
+    caller drains the generator: a caller taking only the first chunk would
+    never see an error that trails it, while a caller draining fully would
+    lose every chunk it had already accumulated to the exception. So an
+    error is recorded and only raised once the body is exhausted without a
+    single chunk — whatever Google did send is always delivered.
 
-    if not raw:
-        return
-
-    # 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"))
-        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
+    Raises:
+        SearchBackendError: If Google answered with an error envelope and no
+            usable chunk at all. See :func:`_error_status`.
 
+    """
+    # ``errors="replace"`` keeps a corrupted transfer from raising here;
+    # the resulting chunk simply fails to parse and is reported below.
+    text = body.decode("utf-8", errors="replace") if isinstance(body, bytes) else body
+
+    text = text.lstrip()
+    if text.startswith(_PREFIX):
+        text = text[len(_PREFIX) :]
+    text = text.lstrip()
+
+    decoder = json.JSONDecoder()
+    errors: list[_BackendStatus] = []
+    yielded = 0
     cursor = 0
-    while cursor < len(raw):
-        # Read the decimal length prefix terminated by \n.
-        end = raw.find(b"\n", cursor)
-        if end == -1:
+    while cursor < len(text):
+        while cursor < len(text) and text[cursor] in _FRAMING_CHARS:
+            cursor += 1
+        if cursor >= len(text):
             break
         try:
-            length = int(raw[cursor:end])
+            outer, cursor = decoder.raw_decode(text, cursor)
         except ValueError:
-            logger.warning(
-                "Malformed length header at offset %d; truncating chunk stream",
-                cursor,
-            )
-            break
-        # Google's length header counts the leading newline after the header
-        # AND the trailing newline that separates this chunk from the next.
-        # We've already consumed the leading newline (it terminated the header),
-        # so we read `length - 1` bytes which gives JSON + trailing \n.
-        cursor = end + 1
-        chunk_bytes = max(length - 1, 0)
-        payload = raw[cursor : cursor + chunk_bytes]
-        cursor += chunk_bytes
-        try:
-            outer = json.loads(payload.strip().decode("utf-8"))
-        except (ValueError, json.JSONDecodeError, UnicodeDecodeError):
             logger.warning("Discarding malformed wrb.fr chunk", exc_info=True)
+            boundary = _CHUNK_BOUNDARY.search(text, cursor)
+            if boundary is None:
+                break
+            cursor = boundary.end()
             continue
-        yield from _chunks_from_outer(outer)
-
+        for chunk in _chunks_from_outer(outer, errors):
+            yielded += 1
+            yield chunk
 
-def _chunks_from_outer(outer: Any) -> Iterator[Any]:
-    """Walk a top-level chunk list and yield decoded inner-JSON payloads."""
+    if not errors:
+        return
+    if not yielded:
+        raise _backend_error(errors[0])
+    logger.warning(
+        "Google Flights reported error %d (%s) alongside %d usable chunk(s); "
+        "keeping the partial payload",
+        errors[0].code,
+        errors[0].detail or _STATUS_NAMES.get(errors[0].code, "unknown"),
+        yielded,
+    )
+
+
+def _error_status(row: list[Any]) -> _BackendStatus | None:
+    """Return the status of an error-envelope ``wrb.fr`` row, if any.
+
+    Google reports a rejected request with ``HTTP 200`` and a payload-less
+    row of the shape ``["wrb.fr", null, null, null, null, [code]]``, e.g.
+    ``3`` (``INVALID_ARGUMENT``) for a payload it cannot decode or ``13``
+    (``INTERNAL``) for a request it declines to serve. The status may carry
+    a message and a ``google.rpc``-style detail block after the code::
+
+        [13, null, [["type.googleapis.com/…ErrorResponse", [[…, "<req-id>"], 0]]]]
+
+    Only a strictly positive integer code counts as a rejection: ``0`` is
+    gRPC's ``OK`` and would otherwise raise a spurious "error 0", and
+    ``bool`` is excluded because it is a subclass of ``int``.
+    """
+    if len(row) < 6:
+        return None
+    status = row[5]
+    if not isinstance(status, list) or not status:
+        return None
+    code = status[0]
+    if isinstance(code, bool) or not isinstance(code, int) or code <= 0:
+        return None
+    return _BackendStatus(code, _error_detail(status))
+
+
+def _error_detail(status: list[Any]) -> str | None:
+    """Summarise the message and detail block trailing a status code."""
+    parts: list[str] = []
+    message = status[1] if len(status) > 1 else None
+    if isinstance(message, str) and message:
+        parts.append(message)
+    details = status[2] if len(status) > 2 else None
+    if details:
+        parts.append(_compact(details))
+    return "; ".join(parts) or None
+
+
+def _compact(value: Any) -> str:
+    """Render a decoded JSON value as a compact, length-capped string."""
+    try:
+        text = json.dumps(value, separators=(",", ":"), ensure_ascii=False)
+    except (TypeError, ValueError):  # pragma: no cover - values come from json.loads
+        text = repr(value)
+    if len(text) > _MAX_DETAIL_CHARS:
+        text = text[: _MAX_DETAIL_CHARS - 1] + "…"
+    return text
+
+
+def _backend_error(status: _BackendStatus) -> SearchBackendError:
+    """Build the exception describing a backend rejection."""
+    name = _STATUS_NAMES.get(status.code)
+    code_text = f"{status.code} ({name})" if name else str(status.code)
+    message = f"Google Flights returned error {code_text} instead of results"
+    if status.detail:
+        message = f"{message}: {status.detail}"
+    return SearchBackendError(message, error_code=status.code, error_detail=status.detail)
+
+
+def _chunks_from_outer(outer: Any, errors: list[_BackendStatus]) -> Iterator[Any]:
+    """Yield a top-level chunk list's payloads, recording rejections into ``errors``."""
     if not isinstance(outer, list):
         return
     for row in outer:
@@ -105,6 +214,9 @@ def _chunks_from_outer(outer: Any) -> Iterator[Any]:
             continue
         inner = row[2]
         if not isinstance(inner, str) or not inner:
+            status = _error_status(row)
+            if status is not None:
+                errors.append(status)
             continue
         try:
             yield json.loads(inner)
@@ -114,7 +226,16 @@ def _chunks_from_outer(outer: Any) -> Iterator[Any]:
 
 
 def parse_first_wrb_payload(body: str | bytes) -> Any:
-    """Return the inner JSON of the first ``wrb.fr`` chunk, or None."""
+    """Return the inner JSON of the first ``wrb.fr`` chunk, or None.
+
+    An error row trailing a usable chunk never raises: the first chunk is
+    returned and the generator is abandoned. See :func:`iter_wrb_chunks`.
+
+    Raises:
+        SearchBackendError: If Google answered with an error envelope and no
+            usable chunk at all.
+
+    """
     for chunk in iter_wrb_chunks(body):
         return chunk
     return None

fli/search/exceptions.py

@@ -28,3 +28,32 @@ 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 an error envelope, not results.
+
+    A rejected request comes back as ``HTTP 200`` carrying a payload-less
+    ``wrb.fr`` row of the shape ``["wrb.fr", null, null, null, null,
+    [code]]``. Without this error the response is indistinguishable from
+    "this route genuinely has no flights".
+
+    Richer rejections carry a status message and/or a ``google.rpc``-style
+    detail block after the code, which is preserved verbatim (truncated) in
+    ``error_detail`` — the code alone is often too coarse to debug with. An
+    ``INTERNAL`` (13), for instance, can mean either "Google declined to
+    serve this" or "a required request header was missing", and only the
+    detail block tells the two apart.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        error_code: int | None = None,
+        error_detail: str | None = None,
+    ):
+        """Store the backend status code and detail alongside the message."""
+        super().__init__(message)
+        self.error_code = error_code
+        self.error_detail = error_detail