From a344c8965c7c170ac81c990c332e660a2fb6aed1 Mon Sep 17 00:00:00 2001 From: Ivan Despot <66276597+g-despot@users.noreply.github.com> Date: Mon, 17 Aug 2026 20:33:51 +0200 Subject: [PATCH 1/6] fix(web): always run REST over the package fetch transport under Emscripten Pyodide's bundled httpx transport dereferences Response.body unconditionally, which is null for HEAD and 204 responses, so data.exists/update/delete_by_id and tenants.exists crashed with a raw AttributeError; it also does not enforce sub-5s REST timeouts end to end. Install the package transport whenever the platform is Emscripten instead of deferring, build the response as a stream so httpx stamps .elapsed (reference_add_many read it), take the request deadline from the read timeout only, and round the AbortSignal up with an int32 cap. The import hook now raises the install hint only when the companion itself is missing and chains any other ImportError, so a broken companion surfaces its own error. The Pyodide e2e gains a transport self-check plus HEAD/204 and reference steps so CI pins the fix. --- ci/pyodide-e2e/e2e.py | 55 +++++- .../src/weaviate_client_web/_httpx_fetch.py | 76 ++++---- packages/web/tests/test_httpx_fetch.py | 163 ++++++++++++++---- packages/web/tests/test_single_import.py | 52 +++++- weaviate/__init__.py | 8 +- weaviate/proto/v1/__init__.py | 13 +- 6 files changed, 275 insertions(+), 92 deletions(-) diff --git a/ci/pyodide-e2e/e2e.py b/ci/pyodide-e2e/e2e.py index 0ebba1391..ae92e40f5 100644 --- a/ci/pyodide-e2e/e2e.py +++ b/ci/pyodide-e2e/e2e.py @@ -10,14 +10,17 @@ """ import os +import uuid import warnings import weaviate_client_web # bootstraps the grpc shim + fetch transport under Emscripten import grpc +import httpx import weaviate import weaviate.classes as wvc -from weaviate.classes.config import DataType, Property +from weaviate.classes.config import DataType, Property, ReferenceProperty +from weaviate.classes.data import DataReference from weaviate.classes.query import Filter from weaviate.classes.tenants import Tenant from weaviate.exceptions import WeaviateBatchStreamError, WeaviateQueryError @@ -38,6 +41,13 @@ async def main() -> None: assert getattr(grpc, "__weaviate_client_web_shim__", False), ( "sys.modules['grpc'] is not the shim" ) + # REST must run through the package's own fetch transport, not Pyodide's bundled + # httpx transport (which cannot read the null body of HEAD / 204 responses). + assert weaviate_client_web.is_fetch_transport_installed(), "fetch transport not installed" + assert getattr( + httpx.AsyncHTTPTransport.handle_async_request, "__weaviate_fetch_shim__", False + ), "httpx.AsyncHTTPTransport is not the package's fetch transport" + ok("self-check: package fetch transport is the active httpx transport") host = os.environ.get("WEAVIATE_HOST", "localhost") port = int(os.environ.get("WEAVIATE_PORT", "8090")) @@ -66,6 +76,7 @@ async def main() -> None: Property(name="title", data_type=DataType.TEXT), Property(name="idx", data_type=DataType.INT), ], + references=[ReferenceProperty(name="related", target_collection=COLL)], ) ok("collections.create") @@ -97,6 +108,40 @@ async def main() -> None: assert idx.minimum == 0 and idx.maximum == 49, agg.properties ok("aggregate count=50 min=0 max=49") + # REST calls answered without a body (HEAD 204/404, PATCH/DELETE 204) and the + # batch-references path, which reads httpx's response.elapsed. + first, second, last = ret.uuids[0], ret.uuids[1], ret.uuids[49] + assert await coll.data.exists(first) is True + assert await coll.data.exists(uuid.uuid4()) is False + ok("data.exists (HEAD 204 / 404) = True / False") + + await coll.data.update(uuid=first, properties={"title": "article 0 (updated)"}) + obj = await coll.query.fetch_object_by_id(first) + assert obj is not None and obj.properties["title"] == "article 0 (updated)", obj + ok("data.update (PATCH 204) -> fetch_object_by_id sees the update") + + refs = await coll.data.reference_add_many( + [ + DataReference( + from_property="related", from_uuid=ret.uuids[i], to_uuid=ret.uuids[i + 1] + ) + for i in range(5) + ] + ) + assert not refs.has_errors, f"reference_add_many errors: {refs.errors}" + assert refs.elapsed_seconds >= 0, refs + ok("data.reference_add_many (REST /batch/references) = 5") + + await coll.data.reference_delete(from_uuid=first, from_property="related", to=second) + ok("data.reference_delete (DELETE 204)") + + assert await coll.data.delete_by_id(last) is True + assert await coll.data.exists(last) is False + # deleting a missing object answers 204 or 404 depending on the server topology; + # either way it is a body-less response the transport must handle + assert isinstance(await coll.data.delete_by_id(last), bool) + ok("data.delete_by_id (DELETE 204; repeat -> 204/404) = True, then bool") + await client.collections.create( MT_COLL, vector_config=wvc.config.Configure.Vectors.self_provided(), @@ -109,6 +154,10 @@ async def main() -> None: assert set(tenants.keys()) == {"t1", "t2"}, f"TenantsGet: {set(tenants.keys())}" ok("multi-tenant create + TenantsGet = {t1, t2}") + assert await mt.tenants.exists("t1") is True + assert await mt.tenants.exists("t404") is False + ok("tenants.exists (HEAD 200 / 404) = True / False") + t1 = mt.with_tenant("t1") ret = await t1.data.insert_many([{"title": f"tenant doc {i}"} for i in range(10)]) assert not ret.has_errors and len(ret.uuids) == 10, f"tenant insert_many: {ret.errors}" @@ -126,8 +175,8 @@ async def main() -> None: await client.collections.get("DoesNotExistXyz").query.fetch_objects(limit=1) raise AssertionError("expected WeaviateQueryError for nonexistent collection") except WeaviateQueryError as e: - assert "DoesNotExistXyz" in str(e) or "not" in str(e).lower(), str(e) - ok("error mapping: nonexistent collection -> WeaviateQueryError") + assert "DoesNotExistXyz" in str(e), str(e) + ok("error mapping: nonexistent collection -> WeaviateQueryError names the collection") try: async with client.batch.stream() as batch: diff --git a/packages/web/src/weaviate_client_web/_httpx_fetch.py b/packages/web/src/weaviate_client_web/_httpx_fetch.py index f09d6bfbe..e3f0ddf4a 100644 --- a/packages/web/src/weaviate_client_web/_httpx_fetch.py +++ b/packages/web/src/weaviate_client_web/_httpx_fetch.py @@ -11,11 +11,11 @@ philosophy as the grpc shim in ``_shim.py``. Responses are fully buffered, which matches how the base client consumes them (JSON bodies, no streaming). -NOTE: Pyodide >= 0.27 distributes a patched httpx whose ``AsyncHTTPTransport`` already -routes through JS ``fetch`` natively (``httpx/_transports/jsfetch.py``) with streaming -support and a proper connect/read timeout split. When that build is detected, installing -is skipped — overwriting it would replace a better implementation. This transport is the -fallback for environments where httpx resolved from PyPI (httpcore + raw sockets). +It installs under Emscripten even when Pyodide's bundled httpx carries its own JS-fetch +transport (``httpx/_transports/jsfetch.py``): that transport reads ``Response.body`` +unconditionally, which is ``null`` for HEAD requests and 204 responses (``data.exists``, +``data.delete_by_id``, ``tenants.exists``, …), and it does not enforce the per-request +read timeout end-to-end. Known divergences from native httpx (acceptable for the weaviate client's usage): - the browser's fetch follows redirects internally, so httpx never sees a 3xx; @@ -23,7 +23,7 @@ - responses are fully buffered (no streaming). """ -import importlib.util +import math import sys from typing import Callable, Dict, Optional @@ -56,6 +56,10 @@ _TIMEOUT_HINTS = ("timeout", "timed out", "abort") +# JS timers take a signed 32-bit millisecond delay; anything larger overflows and fires +# immediately, so a huge timeout would abort every request at once. +_MAX_ABORT_SIGNAL_MS = 2**31 - 1 + async def _read_request_body(request: httpx.Request) -> bytes: try: @@ -65,18 +69,25 @@ async def _read_request_body(request: httpx.Request) -> bytes: def _pick_timeout(request: httpx.Request) -> Optional[float]: - """Pick the effective deadline from httpx's timeout extension. + """Pick the request deadline from httpx's timeout extension: the ``read`` value only. - httpx populates ``extensions['timeout']`` with connect/read/write/pool values; the - read timeout is what the weaviate client configures per request. ``get(...) or`` - chains would silently skip an explicit 0, so check for None instead. + The base client sets ``read`` per request and passes ``read=None`` for "no deadline" + while still carrying a ``pool`` value; ``connect`` is not separable under fetch and a + pool-acquire timeout has no meaning there, so neither may stand in for ``read``. """ timeouts = request.extensions.get("timeout") or {} - for key in ("read", "connect", "pool"): - value = timeouts.get(key) - if value is not None: - return value - return None + return timeouts.get("read") + + +def _abort_signal_ms(timeout: Optional[float]) -> Optional[int]: + """Milliseconds for ``AbortSignal.timeout``; ``None`` means no client-side deadline. + + Zero, negative and non-finite timeouts all mean "no deadline". Rounded up so a + sub-millisecond timeout never becomes an immediate abort. + """ + if timeout is None or not math.isfinite(timeout) or timeout <= 0: + return None + return min(math.ceil(timeout * 1000), _MAX_ABORT_SIGNAL_MS) def _map_fetch_error( @@ -121,19 +132,21 @@ async def _fetch_handle_async_request( # fetch rejects GET/HEAD requests that carry a body kwargs["body"] = body - timeout = _pick_timeout(request) deadline_set = False - if timeout is not None and timeout > 0: + deadline_ms = _abort_signal_ms(_pick_timeout(request)) + if deadline_ms is not None: try: from js import AbortSignal # type: ignore[import-not-found] - kwargs["signal"] = AbortSignal.timeout(int(timeout * 1000)) + kwargs["signal"] = AbortSignal.timeout(deadline_ms) deadline_set = True except Exception: # pragma: no cover - AbortSignal.timeout availability varies pass try: response = await pyfetch(str(request.url), method=request.method, headers=headers, **kwargs) + # A body-less response (HEAD, 204) reads as b"": fetch resolves a null body to + # an empty ArrayBuffer. data = await response.bytes() except OSError as e: # incl. pyodide.http.AbortError raise _map_fetch_error(e, request, deadline_set) from e @@ -146,10 +159,12 @@ async def _fetch_handle_async_request( } except Exception: # pragma: no cover - header shape varies across Pyodide versions resp_headers = {} + # Hand httpx an unread stream, as its own transports do: the client reads it and + # only then stamps ``response.elapsed``, which the batch-references path relies on. return httpx.Response( status_code=int(response.status), headers=resp_headers, - content=data, + stream=httpx.ByteStream(data), request=request, ) @@ -158,34 +173,17 @@ async def _fetch_handle_async_request( _fetch_handle_async_request.__weaviate_fetch_shim__ = True # type: ignore[attr-defined] -def _platform_httpx_has_fetch_support() -> bool: - """True when the running httpx is Pyodide's distributed build. - - That build replaces the httpcore transport with a native JS-fetch one - (httpx/_transports/jsfetch.py), so the weaviate REST path already works without - this shim — and works better (streaming, connect/read timeout split). - """ - try: - return importlib.util.find_spec("httpx._transports.jsfetch") is not None - except (ImportError, ValueError): # pragma: no cover - exotic import states - return False - - def install_fetch_transport(force: bool = False) -> None: """Patch ``httpx.AsyncHTTPTransport`` to send requests through ``fetch``. Installs only under Emscripten unless ``force=True`` (CPython testing, where a - ``pyodide`` stub must be importable), and is skipped when httpx itself already has - fetch support (Pyodide's distributed build). Idempotent. + ``pyodide`` stub must be importable). Idempotent. """ global _installed, _original_handle_async_request if _installed: return - if not force: - if sys.platform != "emscripten": - return - if _platform_httpx_has_fetch_support(): - return + if not force and sys.platform != "emscripten": + return # Fail fast: the handler imports pyfetch per request, so a missing pyodide module # would otherwise surface as a confusing ModuleNotFoundError on the first request. from pyodide.http import pyfetch # type: ignore[import-not-found] # noqa: F401 diff --git a/packages/web/tests/test_httpx_fetch.py b/packages/web/tests/test_httpx_fetch.py index 119c4bcf5..63480311e 100644 --- a/packages/web/tests/test_httpx_fetch.py +++ b/packages/web/tests/test_httpx_fetch.py @@ -18,19 +18,26 @@ import httpx import pytest -from weaviate_client_web._httpx_fetch import _fetch_handle_async_request +from weaviate_client_web._httpx_fetch import ( + _MAX_ABORT_SIGNAL_MS, + _abort_signal_ms, + _fetch_handle_async_request, +) _SRC = str(pathlib.Path(__file__).resolve().parents[1] / "src") class FakeFetchResponse: - def __init__(self, status: int = 200, headers: Optional[Any] = None, body: bytes = b""): + def __init__( + self, status: int = 200, headers: Optional[Any] = None, body: Optional[bytes] = b"" + ): self.status = status self.headers: Any = headers or {} self._body = body async def bytes(self) -> bytes: # noqa: A003 - mirrors pyodide's FetchResponse API - return self._body + # a null JS body (HEAD, 204) resolves to an empty ArrayBuffer, i.e. b"" + return b"" if self._body is None else self._body class FakePyfetch: @@ -55,10 +62,31 @@ def fake_pyfetch(monkeypatch) -> FakePyfetch: return fetch -def _handle(request: httpx.Request) -> httpx.Response: +async def _handle_async(request: httpx.Request) -> httpx.Response: # self is unused by the handler implementation; a bare transport instance suffices transport = httpx.AsyncHTTPTransport.__new__(httpx.AsyncHTTPTransport) - return asyncio.run(_fetch_handle_async_request(transport, request)) + response = await _fetch_handle_async_request(transport, request) + await response.aread() # httpx.AsyncClient reads non-streamed responses the same way + return response + + +def _handle(request: httpx.Request) -> httpx.Response: + return asyncio.run(_handle_async(request)) + + +class _FetchTransport(httpx.AsyncBaseTransport): + """Route an ``httpx.AsyncClient`` through the handler without patching httpx globally.""" + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + return await _fetch_handle_async_request(self, request) # type: ignore[arg-type] + + +def _via_client(method: str, url: str, **kwargs: Any) -> httpx.Response: + async def main() -> httpx.Response: + async with httpx.AsyncClient(transport=_FetchTransport()) as client: + return await client.request(method, url, **kwargs) + + return asyncio.run(main()) def test_basic_get_round_trip(fake_pyfetch): @@ -82,6 +110,44 @@ def test_response_has_request_attached_for_raise_for_status(fake_pyfetch): response.raise_for_status() +@pytest.mark.parametrize("status", [204, 404]) +def test_head_response_without_body_yields_empty_content(fake_pyfetch, status): + # data.exists() / tenants.exists() are HEAD requests answered 204/404 with a null + # body; the transport must hand httpx an empty response, not fail on the missing body + fake_pyfetch.response = FakeFetchResponse(status=status, body=None) + response = _handle(httpx.Request("HEAD", "http://h:8080/v1/objects/A/uuid")) + assert response.status_code == status + assert response.content == b"" + assert "body" not in fake_pyfetch.calls[0] + + +def test_delete_204_without_body_yields_empty_content(fake_pyfetch): + # data.delete_by_id() / reference_delete() are answered 204 with a null body + fake_pyfetch.response = FakeFetchResponse(status=204, body=None) + response = _handle(httpx.Request("DELETE", "http://h:8080/v1/objects/A/uuid")) + assert response.status_code == 204 + assert response.content == b"" + + +def test_body_less_response_through_async_client(fake_pyfetch): + # the full httpx.AsyncClient path (stream wrapping + read) on a body-less response + fake_pyfetch.response = FakeFetchResponse(status=204, body=None) + response = _via_client("HEAD", "http://h:8080/v1/objects/A/uuid") + assert response.status_code == 204 + assert response.content == b"" + + +def test_response_through_async_client_exposes_elapsed_and_content(fake_pyfetch): + # the batch-references path reads ``res.elapsed``, which httpx only sets after it + # has read/closed a stream-backed response; a pre-loaded body never gets one + payload = b'[{"result": {"status": "SUCCESS"}}]' + fake_pyfetch.response = FakeFetchResponse(status=200, body=payload) + response = _via_client("POST", "http://h:8080/v1/batch/references", content=b"[]") + assert response.content == payload + assert response.json() == [{"result": {"status": "SUCCESS"}}] + assert response.elapsed.total_seconds() >= 0 + + def test_fetch_managed_request_headers_stripped(fake_pyfetch): request = httpx.Request( "POST", @@ -184,11 +250,22 @@ def test_read_timeout_maps_to_abort_signal_ms(fake_pyfetch, fake_abort_signal): assert fake_pyfetch.calls[0]["signal"] == "signal-30000" -def test_timeout_falls_back_to_connect_then_pool(fake_pyfetch, fake_abort_signal): +def test_read_none_means_no_deadline_even_with_pool_and_connect_set( + fake_pyfetch, fake_abort_signal +): + # what the base client hands over for a non-finite request timeout: read=None with the + # session pool timeout still set; falling back to pool/connect would abort a long + # insert after 5 s + _handle(_request_with_timeout({"connect": None, "read": None, "write": None, "pool": 5})) _handle(_request_with_timeout({"connect": 2.0, "read": None, "write": None, "pool": 9.0})) - _handle(_request_with_timeout({"connect": None, "read": None, "write": None, "pool": 9.0})) - assert fake_abort_signal.timeouts == [2000, 9000] - assert [c["signal"] for c in fake_pyfetch.calls] == ["signal-2000", "signal-9000"] + assert fake_abort_signal.timeouts == [] + assert all("signal" not in c for c in fake_pyfetch.calls) + + +def test_read_timeout_alone_sets_the_deadline(fake_pyfetch, fake_abort_signal): + _handle(_request_with_timeout({"connect": None, "read": 7, "write": None, "pool": 5})) + assert fake_abort_signal.timeouts == [7000] + assert fake_pyfetch.calls[0]["signal"] == "signal-7000" def test_no_timeout_extension_sends_no_signal(fake_pyfetch, fake_abort_signal): @@ -215,6 +292,40 @@ def test_zero_timeout_means_no_deadline(fake_pyfetch, fake_abort_signal): assert "signal" not in fake_pyfetch.calls[0] +@pytest.mark.parametrize( + "timeout,expected_ms", + [ + (None, None), + (0, None), + (-1, None), + (float("inf"), None), + (float("nan"), None), + (0.0001, 1), # rounds up: never an immediate AbortSignal.timeout(0) + (30.0, 30_000), + (1e8, _MAX_ABORT_SIGNAL_MS), + (1e10, _MAX_ABORT_SIGNAL_MS), + ], +) +def test_abort_signal_ms_bounds(timeout, expected_ms): + assert _abort_signal_ms(timeout) == expected_ms + + +def test_infinite_timeout_sends_no_signal(fake_pyfetch, fake_abort_signal): + # an inf read deadline reaching the transport: no signal, not an OverflowError + _handle( + _request_with_timeout({"connect": None, "read": float("inf"), "write": None, "pool": 5}) + ) + assert fake_abort_signal.timeouts == [] + assert "signal" not in fake_pyfetch.calls[0] + + +def test_huge_timeout_is_capped_to_int32_ms(fake_pyfetch, fake_abort_signal): + # setTimeout delays above 2^31-1 ms overflow and fire at once, aborting the request + _handle(_request_with_timeout({"connect": None, "read": 1e10, "write": None, "pool": None})) + assert fake_abort_signal.timeouts == [_MAX_ABORT_SIGNAL_MS] + assert fake_pyfetch.calls[0]["signal"] == f"signal-{_MAX_ABORT_SIGNAL_MS}" + + class RaisingPyfetch: def __init__(self, exc: BaseException): self.exc = exc @@ -286,20 +397,6 @@ def test_crlf_in_header_value_rejected(fake_pyfetch): assert fake_pyfetch.calls == [] -def test_platform_jsfetch_detection(monkeypatch): - import importlib.machinery - - from weaviate_client_web._httpx_fetch import _platform_httpx_has_fetch_support - - # the dev environment runs PyPI httpx (httpcore-based): no jsfetch transport - assert _platform_httpx_has_fetch_support() is False - - fake = types.ModuleType("httpx._transports.jsfetch") - fake.__spec__ = importlib.machinery.ModuleSpec("httpx._transports.jsfetch", loader=None) - monkeypatch.setitem(sys.modules, "httpx._transports.jsfetch", fake) - assert _platform_httpx_has_fetch_support() is True - - # --------------------------------------------------------------------------- # Install semantics: these patch httpx.AsyncHTTPTransport globally, so each # scenario runs in a fresh subprocess (same pattern as test_shim_install.py). @@ -476,11 +573,12 @@ def test_force_install_without_pyodide_fails_fast(): assert "OK" in result.stdout -def test_emscripten_with_platform_jsfetch_skips_install(): - # on Pyodide's distributed httpx (jsfetch transport built in), the shim must NOT - # overwrite the platform implementation +def test_emscripten_installs_even_when_platform_httpx_has_jsfetch(): + # Pyodide's bundled httpx ships a jsfetch transport that crashes on body-less + # responses (HEAD / 204); the shim must take over regardless of the httpx build result = _run( - """ + prelude=_FAKE_PYODIDE_PRELUDE, + body=""" import importlib.machinery, sys, types sys.platform = "emscripten" @@ -493,11 +591,14 @@ def test_emscripten_with_platform_jsfetch_skips_install(): import httpx before = httpx.AsyncHTTPTransport.handle_async_request from weaviate_client_web import install_fetch_transport, is_fetch_transport_installed - install_fetch_transport() # no force: platform transport must win - assert not is_fetch_transport_installed() - assert httpx.AsyncHTTPTransport.handle_async_request is before + install_fetch_transport() # no force: the platform alone must trigger it + assert is_fetch_transport_installed() + assert httpx.AsyncHTTPTransport.handle_async_request is not before + assert getattr( + httpx.AsyncHTTPTransport.handle_async_request, "__weaviate_fetch_shim__", False + ) is True print("OK") - """ + """, ) assert result.returncode == 0, result.stderr assert "OK" in result.stdout diff --git a/packages/web/tests/test_single_import.py b/packages/web/tests/test_single_import.py index aff239c0d..562f31e0a 100644 --- a/packages/web/tests/test_single_import.py +++ b/packages/web/tests/test_single_import.py @@ -23,6 +23,19 @@ sysconfig.get_config_vars() """ +# The companion's bootstrap installs the fetch transport under Emscripten and fails fast +# if pyodide.http cannot be imported, so a faked platform needs a stand-in module. +_FAKE_PYODIDE = """ +import types + +_pyodide = types.ModuleType("pyodide") +_http = types.ModuleType("pyodide.http") +_http.pyfetch = None +_pyodide.http = _http +sys.modules["pyodide"] = _pyodide +sys.modules["pyodide.http"] = _http +""" + def _run( body: str, *, prelude: str = "", path_entry: str = _SRC, no_site: bool = False @@ -36,22 +49,16 @@ def _run( def test_bare_import_weaviate_installs_shim_under_emscripten(): result = _run( - prelude=_PRIME_SYSCONFIG, + prelude=_PRIME_SYSCONFIG + _FAKE_PYODIDE, body=""" - import importlib.machinery, types - sys.platform = "emscripten" - # Pretend httpx is Pyodide's jsfetch build so the companion's bootstrap skips - # the fetch-transport install (there is no pyodide module on CPython). - fake = types.ModuleType("httpx._transports.jsfetch") - fake.__spec__ = importlib.machinery.ModuleSpec("httpx._transports.jsfetch", loader=None) - sys.modules["httpx._transports.jsfetch"] = fake import weaviate # the ONLY weaviate-side import: must bootstrap the companion assert "weaviate_client_web" in sys.modules, "hook did not import the companion" import weaviate_client_web assert weaviate_client_web.is_installed() + assert weaviate_client_web.is_fetch_transport_installed() import grpc assert getattr(grpc, "__weaviate_client_web_shim__", False) is True print("OK") @@ -101,3 +108,32 @@ def test_bare_import_with_grpc_present_falls_through_silently(): ) assert result.returncode == 0, result.stderr assert "OK" in result.stdout + + +def test_bare_import_with_broken_companion_surfaces_its_own_error(tmp_path): + # An INSTALLED companion whose import fails (here: a missing dependency of its own) + # must raise that error, not the "install weaviate-client-web" hint — the hint would + # send the user to reinstall a package that is already there. + fake_pkg = tmp_path / "weaviate_client_web" + fake_pkg.mkdir() + (fake_pkg / "__init__.py").write_text( + "raise ModuleNotFoundError(\"No module named 'anyio'\", name='anyio')\n" + ) + result = _run( + prelude=_PRIME_SYSCONFIG, + body=""" + sys.platform = "emscripten" + try: + import weaviate + except ImportError as e: + assert e.name == "anyio", (e.name, str(e)) + assert "anyio" in str(e), str(e) + assert "weaviate-client-web" not in str(e), str(e) + print("OK") + else: + raise AssertionError("expected the companion's own ImportError to surface") + """, + path_entry=str(tmp_path), + ) + assert result.returncode == 0, result.stderr + assert "OK" in result.stdout diff --git a/weaviate/__init__.py b/weaviate/__init__.py index 49c1dbadf..20ccff232 100644 --- a/weaviate/__init__.py +++ b/weaviate/__init__.py @@ -7,9 +7,13 @@ if sys.platform == "emscripten": try: import weaviate_client_web # noqa: F401 - except ImportError: + except ImportError as exc: from importlib.util import find_spec + # Only an absent companion earns the install hint; a companion that is present + # but fails to import (a broken dependency of its own) must surface that error. + if not (isinstance(exc, ModuleNotFoundError) and exc.name == "weaviate_client_web"): + raise if find_spec("grpc") is None: raise ImportError( "weaviate requires the weaviate-client-web package under " @@ -17,7 +21,7 @@ "weaviate-client-web provides the grpc-web (fetch) transport in its " "place. Install it (e.g. micropip.install('weaviate-client-web')) and " "import weaviate again." - ) from None + ) from exc import os from importlib.metadata import PackageNotFoundError, version diff --git a/weaviate/proto/v1/__init__.py b/weaviate/proto/v1/__init__.py index 1ad304222..62b0910f4 100644 --- a/weaviate/proto/v1/__init__.py +++ b/weaviate/proto/v1/__init__.py @@ -16,15 +16,10 @@ from weaviate.exceptions import WeaviateProtobufIncompatibility -# Fallback grpcio version used only when grpcio is not installed as a distribution. -# This happens under Pyodide/Emscripten, where grpcio has no wheel and is excluded -# via the `sys_platform != "emscripten"` marker in setup.cfg; the grpc module itself -# is provided there by a pure-Python shim (see the weaviate-client-web package). -# On every normal install grpcio's metadata is present and the real version is used, so -# this branch is not taken. Restricted to grpcio AND to Emscripten, so that a broken or -# partial grpcio install on a normal platform (metadata missing) still surfaces as -# PackageNotFoundError instead of silently selecting a fallback stub, and so a genuinely -# missing protobuf (required and pure-Python under Pyodide) is never masked. +# grpcio version assumed under Pyodide/Emscripten, where grpcio has no wheel (excluded by +# the `sys_platform != "emscripten"` marker in setup.cfg) and the grpc module is the +# weaviate-client-web shim. Restricted to grpcio AND Emscripten so a broken grpcio install +# elsewhere still raises PackageNotFoundError, and a missing protobuf is never masked. _GRPCIO_FALLBACK_VERSION = "1.72.1" def get_version(pkg: str) -> version.Version: From e9c81bec39582b52f8de17812197c20d2463e19b Mon Sep 17 00:00:00 2001 From: Ivan Despot <66276597+g-despot@users.noreply.github.com> Date: Mon, 17 Aug 2026 20:34:15 +0200 Subject: [PATCH 2/6] fix(web): harden grpc-web framing, deadline encoding and diagnostics Non-finite deadlines send no grpc-timeout header and no client-side wait; finite ones escalate m -> S -> M to stay within the spec's 8 digits and never use H, which Weaviate's transcoder rejects above 8H (values past ~190 years mean no deadline). Trailer blocks accept LF-only lines and non-ASCII keys, an unknown frame flag or a message after the trailer is a framing error, and a unary response carrying more than one message frame is INTERNAL rather than silently truncated. Metadata with CR/LF/NUL is rejected before any I/O. Diagnostics: an HTTP 405 and a missing grpc_path_prefix under Emscripten both name the fix, and a truncated grpc-web body is reported as truncated instead of as a wrong path. Stale cross-references in comments trimmed. --- .../web/src/weaviate_client_web/_channel.py | 127 ++++++++++--- .../web/src/weaviate_client_web/_framing.py | 51 +++-- packages/web/src/weaviate_client_web/_shim.py | 24 +-- packages/web/tests/test_framing.py | 42 ++++- packages/web/tests/test_transport.py | 174 +++++++++++++++++- 5 files changed, 357 insertions(+), 61 deletions(-) diff --git a/packages/web/src/weaviate_client_web/_channel.py b/packages/web/src/weaviate_client_web/_channel.py index c3ba77837..d6dd4b7cc 100644 --- a/packages/web/src/weaviate_client_web/_channel.py +++ b/packages/web/src/weaviate_client_web/_channel.py @@ -15,10 +15,11 @@ import asyncio import base64 import math +import sys import urllib.parse from typing import Any, Callable, Dict, List, Optional -from ._framing import encode_message, split_response +from ._framing import TruncatedFrameError, UnknownFrameFlagError, encode_message, split_response from ._sender import Sender, pyfetch_sender from ._shim import AioChannel, AioRpcError, StatusCode, status_from_int @@ -36,14 +37,26 @@ def get_sender() -> Sender: return _default_sender -def _encode_timeout(seconds: float) -> str: - """Encode a timeout as a grpc-timeout header value (````).""" - # Round up so we never advertise a shorter deadline than requested (which would risk - # premature server-side cancellation); grpc-timeout takes a positive integer + unit. - millis = max(1, math.ceil(seconds * 1000)) - if millis < 100_000_000: - return f"{millis}m" - return f"{max(1, math.ceil(seconds))}S" +# grpc-timeout is at most 8 digits plus a unit; anything longer is rejected by the server. +_GRPC_TIMEOUT_MAX = 100_000_000 + + +def _encode_timeout(seconds: Optional[float]) -> Optional[str]: + """Encode a timeout as a grpc-timeout header value; ``None`` means no deadline. + + ``None``, non-finite values and anything beyond 99,999,999 minutes (~190 years) carry + no deadline. Rounds up so we never advertise a shorter deadline than requested (which + would risk premature server-side cancellation), moving to a coarser unit + (m -> S -> M) to stay within 8 digits. Hours are never emitted: grpc-web transcoders + (vanguard) reject any H value above 8H with HTTP 400. + """ + if seconds is None or not math.isfinite(seconds): + return None + for amount, unit in ((seconds * 1000, "m"), (seconds, "S"), (seconds / 60, "M")): + value = max(1, math.ceil(amount)) + if value < _GRPC_TIMEOUT_MAX: + return f"{value}{unit}" + return None def _fold_metadata(headers: Dict[str, str], metadata: Any) -> None: @@ -57,9 +70,13 @@ def _fold_metadata(headers: Dict[str, str], metadata: Any) -> None: name = key.lower() if name.endswith("-bin"): raw = value if isinstance(value, (bytes, bytearray)) else str(value).encode() - headers[name] = base64.b64encode(raw).decode("ascii") + text = base64.b64encode(raw).decode("ascii") else: - headers[name] = value if isinstance(value, str) else str(value) + text = value if isinstance(value, str) else str(value) + # This path bypasses h11/grpcio's header validation, so keep their defence here. + if any(c in name or c in text for c in ("\r", "\n", "\0")): + raise ValueError(f"Illegal character in gRPC metadata {name!r}") + headers[name] = text def _header_lookup(headers: Dict[str, str], name: str) -> Optional[str]: @@ -106,8 +123,8 @@ async def __call__( class _UnsupportedStreamMultiCallable: """Placeholder for ``stream_stream`` (bidirectional streaming). - Calling it raises immediately, before the ``async for`` in ``connect/v4.py:1243`` - begins iterating. + Calling it raises immediately, before the ``async for`` in ``connect/v4.py`` begins + iterating. """ def __init__(self, path: str) -> None: @@ -181,16 +198,20 @@ async def _unary( "x-user-agent": "weaviate-client-web", } _fold_metadata(headers, metadata) - if timeout is not None: - headers["grpc-timeout"] = _encode_timeout(timeout) + grpc_timeout = _encode_timeout(timeout) + if grpc_timeout is None: + timeout = None # None / non-finite: no deadline, server- or client-side + else: + headers["grpc-timeout"] = grpc_timeout url = self._base_url + self._path_prefix + path framed = encode_message(payload) # Send. Enforce a client-side deadline (the grpc-timeout header is server-side # only; pyfetch ignores its timeout arg, so without this a stalled request could - # hang forever). Any transport/parse failure is surfaced as AioRpcError so the - # client only ever sees gRPC error types (never a bare ValueError/TimeoutError). + # hang forever). Any transport/parse failure is surfaced as AioRpcError; the only + # non-gRPC error a caller can see is the ValueError from metadata validation + # above, raised before any I/O (as native grpcio does). try: send = self._sender(url, headers, framed, timeout) if timeout is not None: @@ -208,10 +229,10 @@ async def _unary( # str() of transport errors can be empty (e.g. httpx.ConnectError) — always # include the exception type so failures stay diagnosable detail = f"{type(exc).__name__}: {exc}" if str(exc) else repr(exc) - raise AioRpcError( - code=StatusCode.UNAVAILABLE, - details=f"grpc-web transport error for {path}: {detail}", - ) from exc + details = f"grpc-web transport error for {path}: {detail}" + if not self._path_prefix and sys.platform == "emscripten": + details += " " + _no_path_prefix_hint() + raise AioRpcError(code=StatusCode.UNAVAILABLE, details=details) from exc try: return self._handle_response(status, resp_headers, body, deserialize, url) @@ -231,11 +252,9 @@ def _handle_response( deserialize: Callable[[bytes], Any], url: str = "", ) -> Any: - # Frame-parse defensively, and never let a parse failure decide the outcome: - # every real error response carries a body that is NOT grpc-web framing - # (Weaviate's 404 JSON, an nginx error page), so reading '{' or '<' as a frame - # flag would report "malformed grpc-web response" and throw away the HTTP - # status, the URL and the server's own explanation. + # A frame-parse failure must never decide the outcome by itself: real error + # responses carry non-grpc-web bodies (Weaviate's 404 JSON, an nginx page), and + # the HTTP status, URL and the server's own text must survive into the error. messages: List[bytes] = [] trailers: Dict[str, str] = {} frame_error: Optional[BaseException] = None @@ -257,7 +276,7 @@ def _handle_response( # No grpc-status anywhere AND either a non-200 or a body that is not # grpc-web framing: a gRPC service did not answer this request at all. if http_status != 200 or frame_error is not None: - raise _non_grpc_web_error(http_status, url, body, frame_error) + raise _frame_error_to_rpc(http_status, url, body, frame_error) if messages: # Every grpc-web unary response must carry a grpc-status (trailer frame # or header); a proxy that drops the trailer must not read as success. @@ -274,7 +293,12 @@ def _handle_response( if frame_error is not None: # grpc-status said OK but the body will not parse — report what actually # came back rather than a bare "no message frame". - raise _non_grpc_web_error(http_status, url, body, frame_error) + raise _frame_error_to_rpc(http_status, url, body, frame_error) + if len(messages) > 1: + raise AioRpcError( + code=StatusCode.INTERNAL, + details=f"unary grpc-web response carried {len(messages)} message frames", + ) if not messages: details = "grpc-web response contained no message frame" if raw_status is None: @@ -309,22 +333,53 @@ def _body_excerpt(body: bytes, limit: int = _BODY_EXCERPT_LIMIT) -> str: return text + ("..." if len(body) > limit else "") +def _no_path_prefix_hint() -> str: + # Lazy import: this module is imported while ``weaviate/__init__`` is still + # bootstrapping the shim under Emscripten. + from weaviate.exceptions import GRPC_WEB_MIN_SERVER_VERSION, GRPC_WEB_SERVER_PATH_PREFIX + + return ( + "(no grpc_path_prefix set — for Weaviate >= " + f"{GRPC_WEB_MIN_SERVER_VERSION} use use_async_with_custom(..., " + f"grpc_path_prefix='{GRPC_WEB_SERVER_PATH_PREFIX}') on the REST port, or point " + "grpc_host/grpc_port at a grpc-web transcoder)" + ) + + +def _frame_error_to_rpc( + http_status: int, url: str, body: bytes, frame_error: Optional[BaseException] +) -> AioRpcError: + """Choose the error for a body that did not parse as grpc-web frames.""" + if http_status != 200 or isinstance(frame_error, (UnknownFrameFlagError, TruncatedFrameError)): + return _non_grpc_web_error(http_status, url, body, frame_error) + # Well-formed grpc-web up to the point of failure: a grpc-web endpoint answered but + # broke the protocol (compressed frame, message after trailer, …). + return AioRpcError( + code=StatusCode.INTERNAL, + details=f"malformed grpc-web response from {url or ''}: {frame_error}", + ) + + def _non_grpc_web_error( http_status: int, url: str, body: bytes, frame_error: Optional[BaseException] = None, ) -> AioRpcError: - """Build the error for a response that is not a grpc-web response at all. + """Build the error for a response that is not a usable grpc-web response. Details always begin with ``HTTP `` and carry the request URL plus a body excerpt (``weaviate/connect`` matches on that prefix). The status alone rarely separates "endpoint missing" from "proxy misconfigured"; the server's own body text usually does. """ + truncated = isinstance(frame_error, TruncatedFrameError) what = "not a grpc-web response" if http_status == 200 and frame_error is not None: - what = f"the body is not grpc-web framing ({frame_error})" + if truncated: + what = f"the grpc-web body is truncated ({frame_error})" + else: + what = f"the body is not grpc-web framing ({frame_error})" parts = [f"HTTP {http_status} from {url or ''}: {what}."] if http_status == 404: @@ -336,8 +391,20 @@ def _non_grpc_web_error( "the configured grpc-web path prefix is wrong for the proxy in front of it. " "Weaviate's native prefix is '/v1/grpc-web'." ) + elif http_status == 405: + # A 405 can only come from an existing HTTP route: the prefix points at one. + parts.append( + "An HTTP route answered instead of the grpc-web endpoint (method not " + "allowed): the configured grpc-web path prefix is wrong. Weaviate's native " + "prefix is '/v1/grpc-web'." + ) elif http_status in (502, 503, 504): parts.append("Weaviate or the proxy in front of it is unavailable.") + elif http_status == 200 and truncated: + parts.append( + "The response was cut short — a proxy or browser buffering limit, or the " + "connection dropped mid-response." + ) elif http_status == 200: parts.append( "Something other than a grpc-web endpoint answered — typically a proxy " diff --git a/packages/web/src/weaviate_client_web/_framing.py b/packages/web/src/weaviate_client_web/_framing.py index 4449d265b..7e6a3dfc8 100644 --- a/packages/web/src/weaviate_client_web/_framing.py +++ b/packages/web/src/weaviate_client_web/_framing.py @@ -19,9 +19,22 @@ _FLAG_TRAILER = 0x80 _FLAG_COMPRESSED = 0x01 +_KNOWN_FLAGS = _FLAG_TRAILER | _FLAG_COMPRESSED _HEADER = struct.Struct(">BI") # 1 flag byte + 4-byte big-endian length +class FrameError(ValueError): + """The body is not a well-formed grpc-web response.""" + + +class UnknownFrameFlagError(FrameError): + """A flag byte outside the grpc-web set: the body is not grpc-web framing (JSON, HTML, …).""" + + +class TruncatedFrameError(FrameError): + """The body ends before the length its frame header announces.""" + + def encode_message(payload: bytes) -> bytes: """Frame a single (uncompressed) protobuf payload for sending.""" return _HEADER.pack(0x00, len(payload)) + payload @@ -30,31 +43,41 @@ def encode_message(payload: bytes) -> bytes: def iter_frames(buf: bytes) -> Iterator[Tuple[int, bytes]]: """Yield ``(flag, payload)`` for each frame in a grpc-web response body.""" off, n = 0, len(buf) - while off + 5 <= n: - flag, length = _HEADER.unpack_from(buf, off) + while off < n: + # Validate the flag before the length so a text body ('{', '<') is reported as + # non-grpc-web rather than as a truncated frame with a garbage length. + flag = buf[off] + if flag & ~_KNOWN_FLAGS: + raise UnknownFrameFlagError(f"unknown grpc-web frame flag 0x{flag:02x} at byte {off}") + if off + 5 > n: + raise TruncatedFrameError(f"truncated grpc-web frame header at byte {off}") + _, length = _HEADER.unpack_from(buf, off) off += 5 if off + length > n: - raise ValueError("truncated grpc-web frame") + raise TruncatedFrameError( + f"truncated grpc-web frame: header announces {length} bytes, {n - off} remain" + ) yield flag, buf[off : off + length] off += length - if off != n: - raise ValueError("trailing bytes after final grpc-web frame") def parse_trailers(raw: bytes) -> Dict[str, str]: """Parse a trailer frame payload into a lower-cased header dict. - Names are ASCII by spec, but values are decoded leniently: a proxy that does not - percent-encode ``grpc-message``, or a server error quoting a UTF-8 collection / - tenant / property name, puts raw non-ASCII bytes in the trailer. Failing here would - throw away the ``grpc-status`` that came with it. + Decoded leniently on both sides of the colon: a proxy that does not percent-encode + ``grpc-message``, or a server error quoting a UTF-8 collection / tenant / property + name, puts raw non-ASCII bytes in the trailer, and one odd key must not discard the + ``grpc-status`` travelling with it. Lines are CRLF-terminated by spec; bare LF is + accepted. """ out: Dict[str, str] = {} - for line in raw.split(b"\r\n"): + for line in raw.split(b"\n"): + line = line.rstrip(b"\r") if not line: continue key, _, value = line.partition(b":") - out[key.strip().decode("ascii").lower()] = value.strip().decode("utf-8", "replace") + name = key.strip().decode("utf-8", "replace").lower() + out[name] = value.strip().decode("utf-8", "replace") return out @@ -62,13 +85,17 @@ def split_response(body: bytes) -> Tuple[List[bytes], Dict[str, str]]: """Split a grpc-web response body into message payloads and trailers.""" messages: List[bytes] = [] trailers: Dict[str, str] = {} + seen_trailer = False for flag, payload in iter_frames(body): if flag & _FLAG_TRAILER: trailers.update(parse_trailers(payload)) + seen_trailer = True elif flag & _FLAG_COMPRESSED: - raise ValueError( + raise FrameError( "compressed grpc-web message frames are not supported by this transport" ) + elif seen_trailer: + raise FrameError("message frame after the trailer frame") else: messages.append(payload) return messages, trailers diff --git a/packages/web/src/weaviate_client_web/_shim.py b/packages/web/src/weaviate_client_web/_shim.py index dbf0d79c6..b04b8837a 100644 --- a/packages/web/src/weaviate_client_web/_shim.py +++ b/packages/web/src/weaviate_client_web/_shim.py @@ -8,14 +8,14 @@ 1. **Import surface** — every ``import grpc`` / ``from grpc(.aio) import ...`` executed while ``weaviate`` and its generated ``*_pb2_grpc`` stubs are imported - (``weaviate/config.py:4-5``, ``exceptions.py:7-8``, ``retry.py:5-6``, - ``connect/base.py:5-8``, ``connect/v4.py:24,29-32``, and the v6300 stub's - ``grpc.__version__`` / ``grpc._utilities.first_version_is_lower`` version gate). + (``weaviate/config.py``, ``exceptions.py``, ``retry.py``, ``connect/base.py``, + ``connect/v4.py``, and the v6300 stub's ``grpc.__version__`` / + ``grpc._utilities.first_version_is_lower`` version gate). 2. **Runtime type contract** — :class:`AioChannel` becomes ``grpc.aio.Channel`` so the real grpc-web channel (which subclasses it) passes the - ``isinstance(..., grpc.aio.Channel)`` assertions in ``connect/v4.py`` (lines 722, - 1241); :class:`AioRpcError` is the error the client catches and inspects via - ``.code()`` / ``.details()`` (``exceptions.py:62-76``, ``retry.py:30-31``). + ``isinstance(..., grpc.aio.Channel)`` assertions in ``connect/v4.py``; + :class:`AioRpcError` is the error the client catches and inspects via ``.code()`` / + ``.details()`` (``exceptions.py``, ``retry.py``). """ import enum @@ -34,8 +34,8 @@ class StatusCode(enum.Enum): """Mirror of ``grpc.StatusCode``. ``value`` is the canonical ``(int, str)`` tuple, matching grpcio so ``code.value[0]`` - / ``code.value[1]`` (``exceptions.py:63,66``) and ``code.name`` - (``connect/v4.py:1189``) behave identically. + / ``code.value[1]`` (``exceptions.py``) and ``code.name`` (``connect/v4.py``) behave + identically. """ OK = (0, "ok") @@ -122,11 +122,11 @@ def debug_error_string(self) -> Optional[str]: class StreamStreamCall: - """Stand-in for ``grpc.aio.StreamStreamCall`` (imported as a type at ``v4.py:31``).""" + """Stand-in for ``grpc.aio.StreamStreamCall`` (imported as a type by ``connect/v4.py``).""" class ChannelCredentials: - """Stand-in for ``grpc.ChannelCredentials`` (imported by ``config.py:4``).""" + """Stand-in for ``grpc.ChannelCredentials`` (imported by ``config.py``).""" def ssl_channel_credentials(*_args: Any, **_kwargs: Any) -> ChannelCredentials: @@ -152,8 +152,8 @@ class AioChannel: def first_version_is_lower(_version: str, _other: str) -> bool: """Stand-in for ``grpc._utilities.first_version_is_lower``. - Returning ``False`` makes the v6300 stub's version gate - (``weaviate_pb2_grpc.py:17-29``) pass. + Returning ``False`` makes the v6300 stub's import-time version gate + (``weaviate_pb2_grpc.py``) pass. """ return False diff --git a/packages/web/tests/test_framing.py b/packages/web/tests/test_framing.py index 5bcefe354..fd18a35a3 100644 --- a/packages/web/tests/test_framing.py +++ b/packages/web/tests/test_framing.py @@ -3,6 +3,9 @@ import pytest from weaviate_client_web._framing import ( + FrameError, + TruncatedFrameError, + UnknownFrameFlagError, encode_message, iter_frames, parse_trailers, @@ -29,12 +32,20 @@ def test_split_response_message_and_trailer(): def test_split_response_multiple_messages(): + # the splitter returns every message frame; whether more than one is acceptable is + # the channel's decision (a unary RPC rejects it) body = _frame(b"a") + _frame(b"bb") + _frame(b"grpc-status:0\r\n", 0x80) messages, trailers = split_response(body) assert messages == [b"a", b"bb"] assert trailers["grpc-status"] == "0" +def test_split_response_message_after_trailer_raises(): + body = _frame(b"a") + _frame(b"grpc-status:0\r\n", 0x80) + _frame(b"late") + with pytest.raises(FrameError, match="after the trailer"): + split_response(body) + + def test_split_response_trailers_only(): body = _frame(b"grpc-status:7\r\ngrpc-message:denied\r\n", 0x80) messages, trailers = split_response(body) @@ -70,13 +81,40 @@ def test_split_response_survives_non_ascii_trailer(): assert trailers["grpc-status"] == "7" +def test_parse_trailers_accepts_lf_only_lines(): + parsed = parse_trailers(b"grpc-status:0\ngrpc-message:ok\n") + assert parsed == {"grpc-status": "0", "grpc-message": "ok"} + + +def test_parse_trailers_keeps_status_when_a_key_is_not_ascii(): + # one odd key from a proxy must not throw away the whole block + parsed = parse_trailers("x-caf\u00e9:1\r\ngrpc-status:0\r\n".encode("utf-8")) + assert parsed["grpc-status"] == "0" + assert parsed["x-caf\u00e9"] == "1" + + def test_truncated_frame_raises(): framed = encode_message(b"hello")[:-2] - with pytest.raises(ValueError): + with pytest.raises(TruncatedFrameError): list(iter_frames(framed)) + with pytest.raises(TruncatedFrameError): + list(iter_frames(b"\x00\x00\x00")) # shorter than one frame header + + +@pytest.mark.parametrize("first_byte", [b"{", b"<", b"\x02", b"\x40", b"\xff"]) +def test_unknown_frame_flag_raises(first_byte): + # a JSON / HTML body, or a flag bit this transport does not know + body = first_byte + b"\x00\x00\x00\x01x" + with pytest.raises(UnknownFrameFlagError, match="unknown grpc-web frame flag"): + list(iter_frames(body)) + + +def test_frame_errors_are_value_errors(): + assert issubclass(TruncatedFrameError, ValueError) + assert issubclass(UnknownFrameFlagError, ValueError) def test_compressed_message_frame_rejected(): body = _frame(b"x", 0x01) - with pytest.raises(ValueError): + with pytest.raises(FrameError, match="compressed"): split_response(body) diff --git a/packages/web/tests/test_transport.py b/packages/web/tests/test_transport.py index 3c7d30dcc..02871fa69 100644 --- a/packages/web/tests/test_transport.py +++ b/packages/web/tests/test_transport.py @@ -7,11 +7,17 @@ import asyncio import struct +import sys from typing import Dict, List, Optional, Tuple import pytest -from weaviate_client_web._channel import GrpcWebChannel, set_sender +from weaviate_client_web._channel import ( + GrpcWebChannel, + _body_excerpt, + _encode_timeout, + set_sender, +) from weaviate_client_web._shim import AioChannel, AioRpcError, StatusCode @@ -80,7 +86,7 @@ def test_health_call_without_metadata(): sender = FakeSender(body=_ok_response(b"pong")) channel = _channel(sender) mc = channel.unary_unary("/grpc.health.v1.Health/Check", lambda x: x, lambda b: b) - # mirrors connect/v4.py:316 — request + timeout, no metadata + # mirrors the health check in connect/v4.py — request + timeout, no metadata assert asyncio.run(mc(b"ping", timeout=2)) == b"pong" @@ -118,9 +124,8 @@ def test_trailers_only_status_in_http_headers(): # --- non-grpc-web responses ------------------------------------------------------- # # Every real error response carries a body, and none of them is grpc-web framing. The -# bodies below are verbatim shapes seen in the wild; an empty error body (the old -# fixture) is the one shape no server or proxy produces, which is why these tests used -# to pass against a code path that never ran. +# bodies below are verbatim shapes seen in the wild (an empty error body is the one +# shape no server or proxy produces). # Weaviate's own 404, verbatim from a 1.39.0 server asked for the wrong prefix. WEAVIATE_404_JSON = ( @@ -192,6 +197,61 @@ def test_nginx_404_html_is_reported_as_an_http_404(): assert "malformed grpc-web response" not in err.details() +def test_405_names_the_wrong_prefix(): + # a 405 can only come from an existing HTTP route (measured live: a prefix pointing + # at /v1/objects answers "method POST is not allowed"), so the prefix is wrong + err = _details_of(405, b'{"code":405,"message":"method POST is not allowed, but [GET] are"}') + assert err.details().startswith("HTTP 405 ") + assert "path prefix" in err.details() + assert "/v1/grpc-web" in err.details() + assert "method POST is not allowed" in err.details() + + +def test_truncated_grpc_web_body_is_reported_as_truncated_not_as_wrong_prefix(): + # a valid frame header whose payload was cut short: the endpoint IS grpc-web, so the + # SPA / path-prefix hint would send the user the wrong way + body = _ok_response(b"reply-bytes")[:-6] + err = _details_of(200, body) + assert err.code() is StatusCode.INTERNAL + assert "truncated" in err.details() + assert "cut short" in err.details() + assert "single-page-app" not in err.details() + assert "path prefix" not in err.details() + + +def test_spa_html_body_is_not_reported_as_truncated(): + # text bodies decode to an unknown flag byte and a garbage length; they must read as + # "not grpc-web framing", never as a truncated grpc-web body + err = _details_of(200, SPA_INDEX_HTML) + assert "truncated" not in err.details() + assert "not grpc-web framing" in err.details() + + +def test_message_frame_after_trailer_is_internal(): + body = _frame(b"a") + _frame(b"grpc-status:0\r\n", 0x80) + _frame(b"late") + err = _details_of(200, body) + assert err.code() is StatusCode.INTERNAL + assert "malformed grpc-web response" in err.details() + assert "after the trailer" in err.details() + assert "single-page-app" not in err.details() + + +def test_multiple_message_frames_in_unary_response_is_internal(): + # a unary RPC has exactly one message; silently taking the first would hide a proxy + # or server that streams several + body = _frame(b"a") + _frame(b"bb") + _frame(b"grpc-status:0\r\n", 0x80) + err = _details_of(200, body) + assert err.code() is StatusCode.INTERNAL + assert "2 message frames" in err.details() + + +def test_error_status_wins_over_multiple_message_frames(): + body = _frame(b"a") + _frame(b"bb") + _frame(b"grpc-status:5\r\ngrpc-message:gone\r\n", 0x80) + err = _details_of(200, body) + assert err.code() is StatusCode.NOT_FOUND + assert err.details() == "gone" + + def test_spa_fallback_html_200_is_distinguishable_from_a_404(): # An HTTP 200 serving index.html is the other half of a wrong path prefix: the app's # catch-all route answers instead of Weaviate. It must not read as malformed framing. @@ -396,6 +456,110 @@ def test_grpc_timeout_header_rounds_up(): assert sender.calls[0][1]["grpc-timeout"] == "124m" +def test_body_excerpt_empty_and_non_printable(): + assert _body_excerpt(b"") == "" + assert _body_excerpt(b"\x00\x01\x02\x7f") == "<4 non-printable bytes>" + assert _body_excerpt(b"ok\x00\x01") == "ok" + + +@pytest.mark.parametrize( + "seconds,expected", + [ + (None, None), + (float("inf"), None), + (float("nan"), None), + (0, "1m"), + (0.1234, "124m"), + (5, "5000m"), + (99_999, "99999000m"), + (100_000, "100000S"), # 1e8 ms would be 9 digits + (1e8, "1666667M"), # 1e8 s would be 9 digits + (1e9, "16666667M"), + (5_999_999_940, "99999999M"), # the largest deadline that still fits in minutes + (1e10, None), # would need hours, which transcoders reject above 8H: no deadline + (1e15, None), + ], +) +def test_encode_timeout_stays_within_eight_digits(seconds, expected): + encoded = _encode_timeout(seconds) + assert encoded == expected + if encoded is not None: + assert len(encoded) <= 9 # 8 digits + unit + assert not encoded.endswith("H") + + +def test_infinite_timeout_sends_no_deadline(): + # Timeout(query=inf): neither a grpc-timeout header nor a client-side wait + sender = FakeSender(body=_ok_response(b"x")) + channel = _channel(sender) + mc = channel.unary_unary("/svc/M", lambda x: x, lambda b: b) + assert asyncio.run(mc(b"q", timeout=float("inf"))) == b"x" + _, headers, _, timeout = sender.calls[0] + assert "grpc-timeout" not in headers + assert timeout is None + + +def test_huge_timeout_uses_minutes_then_no_deadline(): + # 1e8 s in milliseconds is 12 digits; the server rejects more than 8 ("timeout is + # too long", HTTP 400) and transcoders reject hour values above 8H, so past the + # minute range the request carries no deadline at all + sender = FakeSender(body=_ok_response(b"x")) + channel = _channel(sender) + mc = channel.unary_unary("/svc/M", lambda x: x, lambda b: b) + asyncio.run(mc(b"q", timeout=1e8)) + asyncio.run(mc(b"q", timeout=1e9)) + asyncio.run(mc(b"q", timeout=1e10)) + assert sender.calls[0][1]["grpc-timeout"] == "1666667M" + assert sender.calls[1][1]["grpc-timeout"] == "16666667M" + assert "grpc-timeout" not in sender.calls[2][1] + assert sender.calls[2][3] is None # no client-side wait either + + +@pytest.mark.parametrize("bad", ["val\r\nx-injected: evil", "val\nx", "v\0"]) +def test_crlf_in_metadata_rejected(bad): + sender = FakeSender(body=_ok_response(b"x")) + channel = _channel(sender) + mc = channel.unary_unary("/svc/M", lambda x: x, lambda b: b) + with pytest.raises(ValueError, match="Illegal character"): + asyncio.run(mc(b"q", metadata=[("x-key", bad)])) + with pytest.raises(ValueError, match="Illegal character"): + asyncio.run(mc(b"q", metadata=[("x-key\r\n", "v")])) + assert sender.calls == [] + + +def _unavailable_details(monkeypatch, path_prefix, platform): + async def boom(url, headers, body, timeout): + raise ConnectionError("Failed to fetch") + + monkeypatch.setattr(sys, "platform", platform) + channel = GrpcWebChannel("h:50051", secure=False, sender=boom, path_prefix=path_prefix) + mc = channel.unary_unary("/svc/M", lambda x: x, lambda b: b) + with pytest.raises(AioRpcError) as excinfo: + asyncio.run(mc(b"q")) + assert excinfo.value.code() is StatusCode.UNAVAILABLE + return excinfo.value.details() + + +def test_unavailable_without_path_prefix_under_emscripten_hints_at_grpc_path_prefix(monkeypatch): + # use_async_with_local() under Pyodide builds a prefix-less channel aimed at the + # native gRPC port, which fetch cannot reach; the error must say what to do instead + details = _unavailable_details(monkeypatch, path_prefix="", platform="emscripten") + assert "grpc_path_prefix='/v1/grpc-web'" in details + assert "1.38.3" in details + assert "use_async_with_custom" in details + + +def test_unavailable_with_path_prefix_has_no_prefix_hint(monkeypatch): + details = _unavailable_details(monkeypatch, path_prefix="/v1/grpc-web", platform="emscripten") + assert "no grpc_path_prefix" not in details + + +def test_unavailable_without_path_prefix_off_emscripten_has_no_prefix_hint(monkeypatch): + # on CPython an empty prefix against a transcoder is the normal configuration + details = _unavailable_details(monkeypatch, path_prefix="", platform="linux") + assert "no grpc_path_prefix" not in details + + def test_close_is_awaitable_noop(): channel = _channel(FakeSender()) assert asyncio.run(channel.close()) is None From 1ffa2a0500c35c7b6bfa0d2161149f659e22c9ef Mon Sep 17 00:00:00 2001 From: Ivan Despot <66276597+g-despot@users.noreply.github.com> Date: Mon, 17 Aug 2026 20:34:35 +0200 Subject: [PATCH 3/6] docs(web): describe the core-native grpc-web setup first use_async_with_custom(..., grpc_path_prefix="/v1/grpc-web") on the REST port is the primary example; use_async_with_local/weaviate_cloud take no prefix and only work with a transcoder at the gRPC host:port root. Drop the skip_init_checks recommendation (the health check runs over grpc-web), list every header the client sends in the CORS section, state the tested Pyodide version, and correct the message-size and REST-transport wording. --- packages/web/README.md | 77 ++++++++++++++----- .../web/src/weaviate_client_web/__init__.py | 16 +++- 2 files changed, 69 insertions(+), 24 deletions(-) diff --git a/packages/web/README.md b/packages/web/README.md index 800f476ae..e1aa554f1 100644 --- a/packages/web/README.md +++ b/packages/web/README.md @@ -8,6 +8,10 @@ WASM workers) where there is no socket and no `grpcio` wheel. It is built from the same repository as `weaviate-client` and reuses its generated protobuf stubs — it does **not** fork code generation. +Requires Weaviate ≥ 1.38.3 (the first release to serve grpc-web natively) or a grpc-web +transcoder in front of an older server. Pyodide ≥ 0.27 recommended; verified on +Pyodide 314.0.4 (CPython 3.14). + ## How it works Under Pyodide there is no `grpcio` Emscripten wheel, and `import weaviate` hard-imports @@ -23,31 +27,55 @@ Under Pyodide there is no `grpcio` Emscripten wheel, and `import weaviate` hard- (`grpc.__version__` / `grpc._utilities.first_version_is_lower`). The `GrpcWebChannel` frames unary RPCs as grpc-web (a 5-byte header + protobuf payload) -and POSTs them via `pyodide.http.pyfetch` to a server fronted by a grpc-web transcoder -(e.g. Envoy or [connectrpc/vanguard](https://github.com/connectrpc/vanguard-go)). Call -metadata (API key / OIDC bearer) is folded into `fetch` headers. - -For REST, Pyodide ≥ 0.27 distributes a patched httpx that already routes through the -browser's `fetch` natively — when that build is detected the package leaves it alone. -Only when httpx resolved from PyPI (httpcore + raw sockets, which cannot work under -WASM) does the package patch `httpx.AsyncHTTPTransport` with its own pyfetch-based -transport. +and POSTs them via `pyodide.http.pyfetch`. Call metadata (API key / OIDC bearer) is +folded into `fetch` headers. Two deployments work: + +1. **Core-native grpc-web (Weaviate ≥ 1.38.3).** Weaviate serves grpc-web on its REST + port under `/v1/grpc-web`, so gRPC and REST share one host, port and TLS setup and + no proxy is needed. Select it with `grpc_path_prefix="/v1/grpc-web"` and point + `grpc_host`/`grpc_port` at the REST endpoint. +2. **A grpc-web transcoder** (e.g. Envoy or + [connectrpc/vanguard](https://github.com/connectrpc/vanguard-go)) listening at the + root of `grpc_host:grpc_port` in front of Weaviate's native gRPC port. This is what + `use_async_with_local()` / `use_async_with_weaviate_cloud()` need, because those + helpers have no `grpc_path_prefix` parameter. + +For REST (`is_ready`, collection config, `/batch/references`, …) the package patches +`httpx.AsyncHTTPTransport` with its own `pyfetch`-based transport. It does so even on +Pyodide builds whose bundled httpx has a JS-fetch transport of its own: that transport +cannot read the null body of HEAD requests and 204 responses (`data.exists`, +`data.delete_by_id`, `tenants.exists`, …) and does not enforce the client's per-request +timeouts. ## Usage With this package installed, a plain `import weaviate` is all you need — under Emscripten the base client imports `weaviate_client_web` itself before anything else, -which installs the shim (and raises a clear error if the package is missing): +which installs the shim and the fetch transport (and raises a clear error if the package +is missing). Against Weaviate ≥ 1.38.3: ```python import weaviate # bootstraps weaviate_client_web automatically under Emscripten -client = weaviate.use_async_with_local(skip_init_checks=True) -await client.connect() +client = weaviate.use_async_with_custom( + http_host="localhost", + http_port=8080, + http_secure=False, + grpc_host="localhost", # same host as REST + grpc_port=8080, # same port as REST: grpc-web rides on the REST listener + grpc_secure=False, + grpc_path_prefix="/v1/grpc-web", +) +await client.connect() # runs the gRPC health check over grpc-web collection = client.collections.get("Article") await collection.query.near_text("hello", limit=3) ``` +`use_async_with_local()` and `use_async_with_weaviate_cloud()` have no +`grpc_path_prefix`, so under Pyodide they only work when a grpc-web transcoder answers at +the root of `grpc_host:grpc_port`. Pass `headers={...}` / `auth_credentials=...` to +`use_async_with_custom` as usual for API keys, OIDC or WCD. + Importing the companion explicitly first also works and remains the explicit form: ```python @@ -60,8 +88,8 @@ import weaviate | Feature | Kind | Status | |----------------------------------------------------------|-----------------|--------| | Search, Aggregate, TenantsGet, BatchObjects, BatchDelete | unary gRPC | ✅ works over grpc-web | -| Health check (`/grpc.health.v1.Health/Check`) | unary gRPC | ✅ (recommend `skip_init_checks=True` + REST `/.well-known/ready`) | -| REST (`is_ready`, config, `/batch/references`, …) | REST | ✅ via fetch (Pyodide's httpx build, or this package's fallback transport) | +| Health check (`/grpc.health.v1.Health/Check`) | unary gRPC | ✅ runs on `connect()` over grpc-web | +| REST (`is_ready`, config, `/batch/references`, …) | REST | ✅ via the package's own fetch transport | | API-key auth (`Auth.api_key`) | header | ✅ | | OIDC auth (`client_credentials` / `client_password` / `bearer_token`) | REST | ✅ token fetch + asyncio-task refresh (no threads) | | Bulk insert: `collection.data.insert_many()` | unary gRPC | ✅ the supported bulk path under WASM | @@ -70,7 +98,7 @@ import weaviate | Embedded Weaviate (`use_async_with_embedded`) | subprocess | ❌ raises "not supported under WebAssembly/Pyodide" | | Synchronous client | — | ❌ async-only under WASM | | Weaviate Agents: `AsyncQueryAgent` `run/ask/search` | REST | ✅ via fetch | -| Weaviate Agents: `ask_stream` / `research_stream` (SSE) | REST streaming | ⚠️ degraded under the fallback transport: fully buffered, events arrive only when the run completes (and long runs can hit the request timeout) | +| Weaviate Agents: `ask_stream` / `research_stream` (SSE) | REST streaming | ⚠️ degraded: the fetch transport buffers the whole response, so events arrive only when the run completes (and long runs can hit the request timeout) | | Weaviate Agents: sync `QueryAgent`, `TransformationAgent`, `PersonalizationAgent` | REST sync | ❌ no async flavour exists | ## Configuration not honored in the browser @@ -82,17 +110,22 @@ under WASM: cannot proxy fetch requests per-client), - connection-pool sizing and `session_pool_max_retries`, - `GrpcConfig.credentials` (custom CA bundles — the browser's trust store decides TLS), -- `GrpcConfig.channel_options`, including `grpc.max_send/receive_message_length` - (only `grpc-web.path_prefix` is consumed), +- `GrpcConfig.channel_options`, including `grpc.max_send_message_length` / + `grpc.max_receive_message_length` (only `grpc-web.path_prefix` is consumed). The + practical message-size ceiling is the server's `grpcMaxMessageSize` (reported by + `/v1/meta`); exceeding it surfaces as `RESOURCE_EXHAUSTED`, - `Proxies.grpc` / `GRPC_PROXY`. ## CORS requirements (browsers) -Cross-origin browser deployments must configure the grpc-web transcoder / REST endpoint -with CORS, or failures become hard to diagnose: +Weaviate ≥ 1.38.3 serves the CORS headers below for its `/v1/grpc-web` endpoint itself, +with no configuration. Its request-header list is a **closed allowlist**: custom +`headers={...}` that are not on it fail the browser's preflight. Cross-origin +deployments that go through a grpc-web transcoder or a proxy must configure CORS there: -- allow the request headers the client sends: `authorization`, `content-type`, - `x-grpc-web`, and any custom headers; +- allow every request header the client sends: `content-type`, `x-grpc-web`, + `x-user-agent`, `grpc-timeout`, `x-weaviate-client`, `authorization` (when auth is + used) and `x-weaviate-cluster-url` (Weaviate Cloud); - expose the grpc-web status headers on responses: `Access-Control-Expose-Headers: grpc-status, grpc-message` — without this, trailers-only error responses (e.g. a bad API key) are reported as @@ -106,3 +139,5 @@ with CORS, or failures become hard to diagnose: interpreter (run it in a fresh process, before importing `weaviate`). Inject a sender with `weaviate_client_web.set_sender(...)` (e.g. `make_httpx_sender()`) to exercise the transport against an Envoy/vanguard transcoder without a browser. +`install_fetch_transport(force=True)` likewise patches httpx on CPython, given an +importable `pyodide.http` stand-in. diff --git a/packages/web/src/weaviate_client_web/__init__.py b/packages/web/src/weaviate_client_web/__init__.py index f0a481d19..24820d41d 100644 --- a/packages/web/src/weaviate_client_web/__init__.py +++ b/packages/web/src/weaviate_client_web/__init__.py @@ -3,16 +3,26 @@ Under Pyodide/Emscripten there is no ``grpcio`` wheel. Importing this package installs a pure-Python ``grpc`` shim into ``sys.modules`` (and forces the pure-Python protobuf runtime) so that the subsequent ``import weaviate`` succeeds and its async gRPC data path -runs over grpc-web (``fetch``) instead of HTTP/2 sockets. +runs over grpc-web (``fetch``) instead of HTTP/2 sockets; REST runs through the package's +own ``fetch``-based httpx transport. -Usage under Pyodide (with this package installed, a bare ``import weaviate`` suffices — +Usage under Pyodide against Weaviate >= 1.38.3, which serves grpc-web on its REST port +under ``/v1/grpc-web`` (with this package installed, a bare ``import weaviate`` suffices — the base client imports this package itself under Emscripten before anything else):: import weaviate - client = weaviate.use_async_with_local(skip_init_checks=True) + client = weaviate.use_async_with_custom( + http_host="localhost", http_port=8080, http_secure=False, + grpc_host="localhost", grpc_port=8080, grpc_secure=False, + grpc_path_prefix="/v1/grpc-web", + ) await client.connect() +``use_async_with_local`` / ``use_async_with_weaviate_cloud`` have no ``grpc_path_prefix`` +and therefore only work with a grpc-web transcoder (Envoy, vanguard) at the root of +``grpc_host:grpc_port``. + An explicit ``import weaviate_client_web`` before ``import weaviate`` also works and remains the explicit form. The shim is installed automatically only under Emscripten, so importing this package on a normal CPython install never clobbers a real, working From b950e39ae6b0fdf0d47a9b1c0720d4c3fe597b17 Mon Sep 17 00:00:00 2001 From: Ivan Despot <66276597+g-despot@users.noreply.github.com> Date: Mon, 17 Aug 2026 20:34:54 +0200 Subject: [PATCH 4/6] fix(connect): back off token refresh, await the refresher on close, sanitise timeouts A permanent OAuth failure (expired or revoked refresh token) made the async refresher POST to the IdP every second; the sync thread died silently on the same error. Both colours now catch any exception and retry with a capped exponential backoff (1s doubling to 60s, reset on success); Con001 says so. The async close() awaits the cancelled task, and the sync thread waits on the Event it was started with, so close() ends it promptly and a reconnect leaves no second refresher. A test drives a real refresher death through connect() to pin the done-callback wiring. Non-finite timeouts mean "no deadline" for every REST and gRPC hand-off, the Emscripten connect-timeout arm is gone (the package transport reads only the read timeout), the grpc-web prefix guards run at construction with a message that explains the CPython testing mode, and a RuntimeError is rewritten to WeaviateClosedClientError only when the http client is actually closed. Native connection errors keep the observed gRPC status line; the duplicated no-prefix hint lives in the transport only. --- mock_tests/test_auth.py | 220 ++++++++++++++++++++++++++---- test/test_connection.py | 90 ++++++++++++ test/test_connection_params.py | 48 +++++++ test/test_wasm_compat.py | 81 ++++++++--- weaviate/connect/base.py | 42 +++--- weaviate/connect/helpers.py | 21 +-- weaviate/connect/v4.py | 241 +++++++++++++++++++++------------ weaviate/warnings.py | 10 +- 8 files changed, 591 insertions(+), 162 deletions(-) create mode 100644 test/test_connection.py diff --git a/mock_tests/test_auth.py b/mock_tests/test_auth.py index 95d64858e..cbd5e1f4a 100644 --- a/mock_tests/test_auth.py +++ b/mock_tests/test_auth.py @@ -1,8 +1,9 @@ import asyncio import json +import threading import time import warnings -from typing import Union +from typing import List, Union import grpc import pytest @@ -12,7 +13,7 @@ import weaviate from mock_tests.conftest import CLIENT_ID, MOCK_IP, MOCK_PORT, MOCK_PORT_GRPC from weaviate.connect.v4 import _ConnectionBase -from weaviate.exceptions import MissingScopeException +from weaviate.exceptions import MissingScopeException, UnexpectedStatusCodeError ACCESS_TOKEN = "HELLO!IamAnAccessToken" CLIENT_SECRET = "SomeSecret.DontTell" @@ -123,27 +124,48 @@ def handler(request: Request) -> Response: assert token_requests > first # a fresh token was fetched with the credentials -@pytest.mark.asyncio -async def test_token_refresh_survives_non_http_error_async( - weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server, recwarn -) -> None: - """A refresh failure that is NOT an httpx.HTTPError must warn and keep the refresher alive. +def _reject_refreshes(weaviate_auth_mock: HTTPServer) -> List[float]: + """Make the IdP reject every refresh (400 invalid_grant); returns the hit timestamps.""" + hits: List[float] = [] - An IdP rejecting the refresh (400 invalid_grant) makes authlib raise OAuthError. When - only HTTPError was caught, the refresh task died silently — nothing awaits it, so not - even asyncio's 'exception was never retrieved' fired — and every later request 401ed - with nothing pointing at the token refresh. - """ - weaviate_auth_mock.expect_request("/auth").respond_with_response( - Response( + def handler(request: Request) -> Response: + hits.append(time.monotonic()) + return Response( json.dumps({"error": "invalid_grant", "error_description": "refresh token expired"}), status=400, content_type="application/json", ) - ) + + weaviate_auth_mock.expect_request("/auth").respond_with_handler(handler) weaviate_auth_mock.expect_request( "/v1/schema", headers={"Authorization": "Bearer " + ACCESS_TOKEN} ).respond_with_json({"classes": []}) + return hits + + +def _assert_backed_off(hits: List[float], recwarn) -> None: + # attempts at ~1s, ~2s, ~4s after connect: a fixed 1s retry would have made 5 in 5s + assert 2 <= len(hits) <= 3 + gaps = [b - a for a, b in zip(hits, hits[1:])] + assert all(later > earlier * 1.5 for earlier, later in zip(gaps, gaps[1:])) + failed = [w for w in recwarn if str(w.message).startswith("Con001")] + assert len(failed) == len(hits) # one warning per failed attempt + assert "invalid_grant" in str(failed[0].message) + assert "retrying in 1s" in str(failed[0].message) + assert "retrying in 2s" in str(failed[1].message) + assert "unstable internet" not in str(failed[0].message) + + +@pytest.mark.asyncio +async def test_token_refresh_backs_off_on_persistent_failure_async( + weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server, recwarn +) -> None: + """A permanently failing refresh must neither kill the refresher nor hot-loop the IdP. + + A 400 invalid_grant makes authlib raise OAuthError: warn, back off exponentially, + stay alive. + """ + hits = _reject_refreshes(weaviate_auth_mock) async with weaviate.use_async_with_local( host=MOCK_IP, @@ -157,23 +179,132 @@ async def test_token_refresh_survives_non_http_error_async( ) as client: task = getattr(client._connection, "_ConnectionBase__token_refresh_task") # noqa: B009 assert task is not None - await asyncio.sleep(2.5) # long enough for at least two failed attempts - assert not task.done() # the refresher survived the failure + await asyncio.sleep(5) + assert not task.done() # the refresher survived the failures await client.collections.list_all() # ... and the client still works - failed = [w for w in recwarn if str(w.message).startswith("Con001")] - assert len(failed) >= 1 - assert "invalid_grant" in str(failed[0].message) + _assert_backed_off(hits, recwarn) # the task ended by close()'s cancellation, not by dying on the exception assert [w for w in recwarn if str(w.message).startswith("Con003")] == [] +def test_token_refresh_backs_off_on_persistent_failure( + weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server, recwarn +) -> None: + """Sync colour of the test above. + + The daemon thread used to die silently on anything but an httpx.HTTPError; now it + warns and backs off like the async task. + """ + hits = _reject_refreshes(weaviate_auth_mock) + + threads_before = set(threading.enumerate()) + with weaviate.connect_to_local( + host=MOCK_IP, + port=MOCK_PORT, + grpc_port=MOCK_PORT_GRPC, + auth_credentials=weaviate.auth.AuthBearerToken( + ACCESS_TOKEN, refresh_token=REFRESH_TOKEN, expires_in=1 + ), + ) as client: + refreshers = [ + t for t in set(threading.enumerate()) - threads_before if t.name == "TokenRefresh" + ] + assert len(refreshers) == 1 + time.sleep(5) + assert refreshers[0].is_alive() # survived the failures + client.collections.list_all() + + _assert_backed_off(hits, recwarn) + refreshers[0].join(timeout=2) + assert not refreshers[0].is_alive() # close() stops the daemon thread promptly + + +class _Boom(BaseException): + """Escapes the refresh loop's `except Exception` like a real BaseException would.""" + + +@pytest.mark.asyncio +async def test_token_refresh_death_is_surfaced_through_real_connect( + weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server, recwarn +) -> None: + """A refresher started by connect() that dies outside the loop body must warn (Con003). + + Pins the done-callback wiring in _create_background_token_refresh, which the + callback-only unit test below cannot see. + """ + weaviate_auth_mock.expect_request( + "/v1/schema", headers={"Authorization": "Bearer " + ACCESS_TOKEN} + ).respond_with_json({"classes": []}) + + async def boom(*args, **kwargs): + raise _Boom("boom") + + async with weaviate.use_async_with_local( + host=MOCK_IP, + port=MOCK_PORT, + grpc_port=MOCK_PORT_GRPC, + auth_credentials=weaviate.auth.AuthBearerToken( + ACCESS_TOKEN, refresh_token=REFRESH_TOKEN, expires_in=1 + ), + ) as client: + task = getattr(client._connection, "_ConnectionBase__token_refresh_task") # noqa: B009 + assert task is not None + client._connection._client.refresh_token = boom # type: ignore[union-attr] + await asyncio.sleep(2) # the first refresh is due after ~1s + assert task.done() and not task.cancelled() + + stopped = [w for w in recwarn if str(w.message).startswith("Con003")] + assert len(stopped) == 1 + assert "_Boom" in str(stopped[0].message) + + +@pytest.mark.asyncio +async def test_failed_connect_cancels_the_refresher_and_close_is_safe_after( + weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server +) -> None: + """A failed connect() must not leave the refresher running; close() after it is safe. + + The OIDC step starts the refresher before /v1/meta is checked, and close() afterwards + (even twice) must neither hang nor raise. + """ + weaviate_auth_mock.expect_request("/auth").respond_with_json( + {"access_token": ACCESS_TOKEN, "expires_in": 500, "refresh_token": REFRESH_TOKEN} + ) + # oneshot handlers take precedence over the fixture's permanent /v1/meta handler + weaviate_auth_mock.expect_oneshot_request("/v1/meta").respond_with_response( + Response(status=500) + ) + + tasks_before = asyncio.all_tasks() + client = weaviate.use_async_with_local( + host=MOCK_IP, + port=MOCK_PORT, + grpc_port=MOCK_PORT_GRPC, + auth_credentials=weaviate.auth.AuthBearerToken( + ACCESS_TOKEN, refresh_token=REFRESH_TOKEN, expires_in=500 + ), + ) + with pytest.raises(UnexpectedStatusCodeError): + await client.connect() + + refresh_tasks = [ + t for t in asyncio.all_tasks() - tasks_before if "token_refresh" in repr(t.get_coro()) + ] + assert len(refresh_tasks) == 1 # it was started ... + await asyncio.sleep(0) + assert refresh_tasks[0].cancelled() # ... and cancelled by the failed connect + + await asyncio.wait_for(client.close(), timeout=2) + await asyncio.wait_for(client.close(), timeout=2) + + @pytest.mark.asyncio async def test_token_refresh_death_outside_loop_body_is_surfaced() -> None: """A refresher that dies where the loop body cannot catch it must still warn. - Nothing awaits the task and _cancel_background_token_refresh drops the reference, so - the done-callback is the only thing that can observe such a death. + Nothing awaits the task while it runs (close() only gathers it, exceptions included), + so the done-callback is the only thing that can observe such a death. """ on_done = getattr(_ConnectionBase, "_ConnectionBase__warn_if_token_refresh_died") # noqa: B009 @@ -347,11 +478,52 @@ async def test_async_auth_starts_no_threads( t for t in asyncio.all_tasks() - tasks_before if "token_refresh" in repr(t.get_coro()) ] assert len(refresh_tasks) == 1 # the refresher runs as an asyncio task instead - # ... and close() must cancel it, not leak it (one wait for the cancellation to land) - await asyncio.wait(refresh_tasks, timeout=1) + # ... and close() must cancel it AND await it: done as soon as close() returns assert refresh_tasks[0].done() +def test_sync_reconnect_leaves_exactly_one_refresher_thread( + weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server +) -> None: + """close() must end the daemon thread promptly, and a later connect() must not revive it. + + The thread used to re-read the connection's shutdown event on every loop, so after + close()+connect() it picked up the NEW (unset) event and kept refreshing next to the + new thread — two refreshers per client. + """ + weaviate_auth_mock.expect_request("/auth").respond_with_json( + {"access_token": ACCESS_TOKEN, "expires_in": 500, "refresh_token": REFRESH_TOKEN} + ) + weaviate_auth_mock.expect_request( + "/v1/schema", headers={"Authorization": "Bearer " + ACCESS_TOKEN} + ).respond_with_json({"classes": []}) + + def refreshers() -> List[threading.Thread]: + return [t for t in set(threading.enumerate()) - threads_before if t.name == "TokenRefresh"] + + threads_before = set(threading.enumerate()) + client = weaviate.connect_to_local( + host=MOCK_IP, + port=MOCK_PORT, + grpc_port=MOCK_PORT_GRPC, + auth_credentials=weaviate.auth.AuthBearerToken( + ACCESS_TOKEN, refresh_token=REFRESH_TOKEN, expires_in=500 + ), + ) + (first,) = refreshers() + client.close() + first.join(timeout=2) + assert not first.is_alive() # not asleep until the next (470s away) wake-up + + client.connect() + client.collections.list_all() + alive = [t for t in refreshers() if t.is_alive()] + assert len(alive) == 1 and alive[0] is not first + client.close() + alive[0].join(timeout=2) + assert not alive[0].is_alive() + + def test_refresh_of_refresh(weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server) -> None: """Test that refresh tokens are used to get a new refresh token token.""" weaviate_auth_mock.expect_request( diff --git a/test/test_connection.py b/test/test_connection.py new file mode 100644 index 000000000..28540f162 --- /dev/null +++ b/test/test_connection.py @@ -0,0 +1,90 @@ +"""Unit tests for weaviate.connect.v4 connection-level behaviour that needs no server.""" + +import asyncio +import inspect +import threading +import time + +import grpc +import pytest +from grpc.aio import AioRpcError, Metadata + +from weaviate.config import ConnectionConfig +from weaviate.config import Timeout as TimeoutConfig +from weaviate.connect.base import ConnectionParams +from weaviate.connect.v4 import ConnectionAsync, ConnectionSync, _ConnectionBase +from weaviate.exceptions import WeaviateBatchError +from weaviate.proto.v1 import batch_pb2 + + +def test_connection_sync_init_mirrors_the_base_signature() -> None: + # explicit, typed parameters (no *args/**kwargs pass-through) + assert ( + inspect.signature(ConnectionSync.__init__).parameters + == inspect.signature(_ConnectionBase.__init__).parameters + ) + + +def _connection_async() -> ConnectionAsync: + return ConnectionAsync( + connection_params=ConnectionParams.from_url("http://localhost:8080", 50051), + auth_client_secret=None, + timeout_config=TimeoutConfig(), + proxies=None, + trust_env=False, + additional_headers=None, + connection_config=ConnectionConfig(), + ) + + +def test_async_batch_objects_error_carries_the_grpc_details_only() -> None: + # like WeaviateQueryError: the message is the server's details, not the whole + # '' repr + class FailingStub: + async def BatchObjects(self, request, metadata=None, timeout=None): + raise AioRpcError( + grpc.StatusCode.INVALID_ARGUMENT, Metadata(), Metadata(), details="bad object" + ) + + conn = _connection_async() + conn._connected = True + conn._grpc_stub = FailingStub() # type: ignore[assignment] + + with pytest.raises(WeaviateBatchError) as excinfo: + asyncio.run( + conn.grpc_batch_objects(batch_pb2.BatchObjectsRequest(), timeout=1, max_retries=0) + ) + assert excinfo.value.message == "bad object" + assert "AioRpcError" not in str(excinfo.value) + + +def test_cancel_refresher_owned_by_another_loop_is_scheduled_not_awaited() -> None: + # close() from a thread/loop other than the refresher's must not try to await the + # task (which would hang); it schedules the cancel on the owning loop instead + loop = asyncio.new_event_loop() + started = threading.Event() + holder: dict = {} + + async def forever() -> None: + started.set() + await asyncio.sleep(3600) + + def run() -> None: + holder["task"] = loop.create_task(forever()) + loop.run_forever() + + thread = threading.Thread(target=run, daemon=True) + thread.start() + assert started.wait(timeout=2) + conn = _connection_async() + conn._ConnectionBase__token_refresh_task = holder["task"] # type: ignore[attr-defined] + try: + assert conn._cancel_background_token_refresh() is None # nothing to await here + deadline = time.monotonic() + 2 + while not holder["task"].done() and time.monotonic() < deadline: + time.sleep(0.01) + assert holder["task"].cancelled() + finally: + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=2) + loop.close() diff --git a/test/test_connection_params.py b/test/test_connection_params.py index 89b9a4cf4..a0dc1c7d4 100644 --- a/test/test_connection_params.py +++ b/test/test_connection_params.py @@ -171,3 +171,51 @@ def fake_insecure_channel(target, options=None, **kwargs): option_keys = [key for key, _ in captured["options"]] assert "grpc-web.path_prefix" not in option_keys + + +def test_connect_to_custom_treats_root_prefix_as_native_grpc(monkeypatch) -> None: + # "/" normalizes to "" (native gRPC), so the async-only guard must not fire on it + import weaviate + import weaviate.connect.helpers as helpers + + monkeypatch.setattr(helpers, "__connect", lambda client: client) # module-level: not mangled + client = weaviate.connect_to_custom( + http_host="localhost", + http_port=8080, + http_secure=False, + grpc_host="localhost", + grpc_port=50051, + grpc_secure=False, + grpc_path_prefix="/", + ) + assert client._connection._connection_params._grpc_web_path_prefix == "" + + +def test_async_client_construction_rejects_prefix_without_shim(monkeypatch) -> None: + # fail at construction with actionable text, not deep inside connect() after the + # OIDC and /v1/meta round trips already succeeded + from weaviate import WeaviateAsyncClient + + monkeypatch.delattr(base_mod.grpc, "__weaviate_client_web_shim__", raising=False) + with pytest.raises(WeaviateInvalidInputError) as excinfo: + WeaviateAsyncClient(_grpc_web_params()) + msg = str(excinfo.value) + assert "weaviate-client-web" in msg + assert "install(force=True)" in msg + assert "set_sender(make_httpx_sender())" in msg + assert "import weaviate" in msg # ... and how Pyodide differs + + +def test_async_client_construction_allows_prefix_with_shim(monkeypatch) -> None: + from weaviate import WeaviateAsyncClient + + monkeypatch.setattr(base_mod.grpc, "__weaviate_client_web_shim__", True, raising=False) + client = WeaviateAsyncClient(_grpc_web_params()) + assert client._connection._connection_params._grpc_web_path_prefix == "/grpc-web" + + +def test_sync_client_construction_rejects_grpc_web_prefix() -> None: + from weaviate import WeaviateClient + + with pytest.raises(WeaviateInvalidInputError, match="async"): + WeaviateClient(_grpc_web_params()) diff --git a/test/test_wasm_compat.py b/test/test_wasm_compat.py index f568d08e9..5ac1d58fc 100644 --- a/test/test_wasm_compat.py +++ b/test/test_wasm_compat.py @@ -6,8 +6,10 @@ """ import sys +from typing import Optional import grpc +import httpx import pytest from grpc.aio import AioRpcError, Metadata from httpx import ConnectError, ConnectTimeout, PoolTimeout, ReadTimeout, WriteTimeout @@ -16,7 +18,7 @@ from weaviate.config import ConnectionConfig from weaviate.config import Timeout as TimeoutConfig from weaviate.connect.base import ConnectionParams -from weaviate.connect.v4 import _ConnectionBase, _exc_detail +from weaviate.connect.v4 import _ConnectionBase, _deadline, _exc_detail from weaviate.embedded import _EmbeddedBase from weaviate.exceptions import ( WeaviateClosedClientError, @@ -58,25 +60,33 @@ def test_async_client_construction_allowed_under_emscripten(monkeypatch) -> None assert client is not None -def _handle_exceptions(e: Exception, error_msg: str = "") -> None: +def _handle_exceptions( + e: Exception, error_msg: str = "", client: Optional[httpx.Client] = None +) -> None: conn = object.__new__(_ConnectionBase) - # keep the bare instance's __del__ quiet (it checks these for unclosed connections) - conn._client = None + # __del__ checks these for unclosed connections; None also means "no client at all" + conn._client = client conn._grpc_channel = None getattr(conn, "_ConnectionBase__handle_exceptions")(e, error_msg) # noqa: B009 -def test_httpx_closed_client_runtime_error_maps_to_closed_client() -> None: - # the exact message httpx raises for a closed AsyncClient/Client +def test_runtime_error_from_a_closed_client_maps_to_closed_client() -> None: + # httpx raises a bare RuntimeError('Cannot send a request, as the client has been + # closed.'); the client's state, not the message text, is what makes it 'closed' + closed = httpx.Client() + closed.close() with pytest.raises(WeaviateClosedClientError): _handle_exceptions(RuntimeError("Cannot send a request, as the client has been closed.")) + with pytest.raises(WeaviateClosedClientError): + _handle_exceptions(RuntimeError("some other wording"), client=closed) def test_unrelated_runtime_error_is_not_rewritten_as_closed_client() -> None: # Emscripten's canonical thread failure must propagate as-is, not as a misleading # 'client is closed - run client.connect()' - with pytest.raises(RuntimeError, match="can't start new thread"): - _handle_exceptions(RuntimeError("can't start new thread")) + with httpx.Client() as open_client: + with pytest.raises(RuntimeError, match="can't start new thread"): + _handle_exceptions(RuntimeError("can't start new thread"), client=open_client) def test_connect_error_message_includes_exception_type() -> None: @@ -206,21 +216,52 @@ def test_rest_timeouts_are_capped_at_five_seconds_on_native_platforms() -> None: assert insert.read == 90 -def test_rest_connect_timeout_follows_the_request_timeout_under_emscripten(monkeypatch) -> None: - # under Pyodide the connect timeout bounds the entire fetch promise, so a fixed 5s - # silently caps every request at ~5s of wall clock +def test_rest_deadline_under_emscripten_is_the_read_timeout(monkeypatch) -> None: + # the fetch transport (weaviate-client-web) reads only `read` as the whole-request + # deadline; connect/write/pool are never consulted there, so nothing platform-specific + # is computed: same httpx.Timeout as on CPython, finite -> value, non-finite -> None monkeypatch.setattr(sys, "platform", "emscripten") - conn = _connection() - insert = _get_timeout(conn, "POST") - assert insert.connect == 90 and insert.write == 90 - assert insert.read == 90 - - query = _get_timeout(conn, "GET") - assert query.connect == 30 + insert = _get_timeout(_connection(), "POST") + assert insert.read == 90 and insert.connect == 5.0 + query = _get_timeout(_connection(), "GET") assert query.read == 30 - # a configured timeout below the httpx default must not make connecting stricter short = _get_timeout(_connection(insert=1, query=1), "POST") - assert short.connect == 5.0 assert short.read == 1 + unbounded = _get_timeout(_connection(insert=float("inf")), "POST") + assert unbounded.read is None + + +@pytest.mark.parametrize("value", [float("inf"), float("nan")]) +def test_deadline_maps_non_finite_to_none(value: float) -> None: + # Timeout(query=float('inf')) passes pydantic's ge=0; handed on as-is it overflows in + # the fetch layer under Pyodide (int(inf * 1000)). None is "no deadline" for both + # httpx and grpc. + assert _deadline(value) is None + assert _deadline(None) is None + assert _deadline(0) == 0 + assert _deadline(2.5) == 2.5 + + +def test_infinite_rest_timeouts_become_no_read_timeout() -> None: + conn = _connection(insert=float("inf"), query=float("inf")) + insert = _get_timeout(conn, "POST") + assert insert.read is None + assert insert.connect == 5.0 # the httpx default is kept on native platforms + query = _get_timeout(conn, "GET") + assert query.read is None + + +def test_deadlines_view_sanitises_every_timeout() -> None: + conn = _connection() + conn.timeout_config = TimeoutConfig( + query=float("inf"), insert=90, init=float("inf"), stream=float("inf") + ) + deadlines = conn._deadlines + assert deadlines.query is None + assert deadlines.insert == 90 + assert deadlines.init is None # Timeout(init=inf) used to end in an OverflowError + assert deadlines.stream is None + # the user's own config object is left untouched + assert conn.timeout_config.init == float("inf") diff --git a/weaviate/connect/base.py b/weaviate/connect/base.py index fd5bed8e5..4fc093023 100644 --- a/weaviate/connect/base.py +++ b/weaviate/connect/base.py @@ -149,6 +149,29 @@ def _grpc_web_path_prefix(self) -> str: cleaned = (self.grpc_path_prefix or "").strip("/") return f"/{cleaned}" if cleaned else "" + def _check_grpc_web_usable(self, is_async: bool) -> None: + """Fail fast on a grpc-web prefix this process cannot honour; a no-op for native gRPC. + + A native grpcio channel would silently ignore the ``grpc-web.path_prefix`` option + and route over native gRPC, so the shim (which consumes it) must be in place. + """ + if self._grpc_web_path_prefix == "": + return + if not is_async: + raise WeaviateInvalidInputError( + "grpc_path_prefix (grpc-web) is only supported for async clients; " + "use use_async_with_custom(...) / WeaviateAsyncClient" + ) + if not _grpc_web_shim_active(): + raise WeaviateInvalidInputError( + "grpc_path_prefix enables grpc-web, which requires the " + "'weaviate-client-web' package (it installs a grpc shim before " + "'import weaviate'); it is not active in this environment. Under Pyodide a " + "plain `import weaviate` activates it; on CPython call " + "weaviate_client_web.install(force=True) and set_sender(make_httpx_sender()) " + "before importing weaviate (intended for integration testing)." + ) + def _grpc_channel( self, proxies: Dict[str, str], @@ -172,24 +195,9 @@ def _grpc_channel( if grpc_config is not None and grpc_config.channel_options is not None: options.extend(grpc_config.channel_options) - # grpc-web mode (prefix set): only valid for an async client, and only when the - # weaviate-client-web shim has replaced the grpc module (it consumes the - # grpc-web.path_prefix option). Fail fast otherwise — a native grpcio channel - # would silently ignore the option and route over native gRPC, a confusing - # misconfiguration. Nothing is added for native gRPC, so its channel options stay - # byte-for-byte unchanged. + # nothing is added for native gRPC, so its channel options stay byte-for-byte unchanged if (prefix := self._grpc_web_path_prefix) != "": - if not is_async: - raise WeaviateInvalidInputError( - "grpc_path_prefix (grpc-web) is only supported for async clients; " - "use use_async_with_custom(...) / WeaviateAsyncClient" - ) - if not _grpc_web_shim_active(): - raise WeaviateInvalidInputError( - "grpc_path_prefix enables grpc-web, which requires the " - "'weaviate-client-web' package (it installs a grpc shim before " - "'import weaviate'); it is not active in this environment" - ) + self._check_grpc_web_usable(is_async) options.append(("grpc-web.path_prefix", prefix)) if is_async: diff --git a/weaviate/connect/helpers.py b/weaviate/connect/helpers.py index a2d87afdb..28153ddc1 100644 --- a/weaviate/connect/helpers.py +++ b/weaviate/connect/helpers.py @@ -351,22 +351,23 @@ def connect_to_custom( True >>> # The connection is automatically closed when the context is exited. """ - if grpc_path_prefix: + connection_params = ConnectionParams.from_params( + http_host=http_host, + http_port=http_port, + http_secure=http_secure, + grpc_host=grpc_host, + grpc_port=grpc_port, + grpc_secure=grpc_secure, + grpc_path_prefix=grpc_path_prefix, + ) + if connection_params._grpc_web_path_prefix: # normalized: "/" is native gRPC raise WeaviateInvalidInputError( "grpc_path_prefix enables grpc-web, which is async-only; use " "use_async_with_custom(...) instead of connect_to_custom(...)" ) return __connect( WeaviateClient( - ConnectionParams.from_params( - http_host=http_host, - http_port=http_port, - http_secure=http_secure, - grpc_host=grpc_host, - grpc_port=grpc_port, - grpc_secure=grpc_secure, - grpc_path_prefix=grpc_path_prefix, - ), + connection_params, auth_client_secret=__parse_auth_credentials(auth_credentials), additional_headers=headers, additional_config=additional_config, diff --git a/weaviate/connect/v4.py b/weaviate/connect/v4.py index 852da8dd9..e0c4a7fca 100644 --- a/weaviate/connect/v4.py +++ b/weaviate/connect/v4.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import math import sys import time from copy import copy @@ -111,6 +112,44 @@ PERMISSION_DENIED = "PERMISSION_DENIED" +# ceiling (seconds) for the exponential backoff between failed token refresh attempts +TOKEN_REFRESH_BACKOFF_CAP = 60 + + +def _token_refresh_backoff(consecutive_failures: int) -> int: + """Seconds to wait before the next refresh attempt: 1, 2, 4, ... up to the cap.""" + return min(2 ** min(consecutive_failures - 1, 16), TOKEN_REFRESH_BACKOFF_CAP) + + +def _deadline(timeout: Union[int, float, None]) -> Union[int, float, None]: + """A timeout as handed to httpx/grpc: non-finite (e.g. ``float('inf')``) means no deadline. + + Passed on as-is, ``inf`` overflows in the fetch layer under Pyodide; None is "no + timeout" for httpx and "no deadline" for grpc alike. + """ + if timeout is None or not math.isfinite(timeout): + return None + return timeout + + +@dataclass(frozen=True) +class _Deadlines: + """The user's timeout config as handed to httpx/grpc (each value through ``_deadline``).""" + + query: Union[int, float, None] + insert: Union[int, float, None] + init: Union[int, float, None] + stream: Union[int, float, None] + + @classmethod + def of(cls, config: TimeoutConfig) -> "_Deadlines": + return cls( + query=_deadline(config.query), + insert=_deadline(config.insert), + init=_deadline(config.init), + stream=_deadline(config.stream), + ) + def _exc_detail(e: BaseException) -> str: """Format an exception for user-facing messages. @@ -158,6 +197,8 @@ def __init__( self._connection_params = connection_params self._grpc_stub: Optional[weaviate_pb2_grpc.WeaviateStub] = None self._grpc_channel: Union[AsyncChannel, SyncChannel, None] = None + # a grpc-web prefix this process cannot honour fails here, not deep inside connect() + connection_params._check_grpc_web_usable(is_async=not isinstance(self, ConnectionSync)) self.timeout_config = timeout_config self.__connection_config = connection_config self.__trust_env = trust_env @@ -328,7 +369,7 @@ def _ping_grpc(self, colour: executor.Colour) -> Union[None, Awaitable[None]]: "/grpc.health.v1.Health/Check", request_serializer=health_weaviate_pb2.WeaviateHealthCheckRequest.SerializeToString, response_deserializer=health_weaviate_pb2.WeaviateHealthCheckResponse.FromString, - )(health_weaviate_pb2.WeaviateHealthCheckRequest(), timeout=self.timeout_config.init) + )(health_weaviate_pb2.WeaviateHealthCheckRequest(), timeout=self._deadlines.init) if colour == "async": async def execute(): @@ -388,6 +429,10 @@ def server_version(self) -> str: """Version of the weaviate instance.""" return str(self._weaviate_version) + @property + def _deadlines(self) -> _Deadlines: + return _Deadlines.of(self.timeout_config) + def get_proxies(self) -> Dict[str, str]: return self._proxies @@ -432,7 +477,7 @@ def _open_connections_rest( async def get_oidc() -> None: async with self._make_client("async") as client: try: - response = await client.get(oidc_url, timeout=self.timeout_config.init) + response = await client.get(oidc_url, timeout=self._deadlines.init) except Exception as e: raise WeaviateConnectionError( f"Error: {_exc_detail(e)}. \nIs Weaviate running and reachable at {self.url}?" @@ -447,7 +492,7 @@ async def get_oidc() -> None: with self._make_client("sync") as client: try: - response = client.get(oidc_url, timeout=self.timeout_config.init) + response = client.get(oidc_url, timeout=self._deadlines.init) except Exception as e: raise WeaviateConnectionError( f"Error: {_exc_detail(e)}. \nIs Weaviate running and reachable at {self.url}?" @@ -557,15 +602,17 @@ def _create_background_token_refresh(self, _auth: Optional[_Auth] = None) -> Non if "refresh_token" not in self._client.token and _auth is None: return - # a previous connect() may have left a refresher behind (e.g. a retry after a - # partially failed connect); stop it before replacing the shutdown event, or it - # would keep refreshing concurrently forever + # stop the refresher a previous connect() may have left behind (e.g. a retry + # after a partially failed connect) self._cancel_background_token_refresh() expires_in: int = self._client.token.get( "expires_in", 60 ) # use 1minute as token lifetime if not supplied - self._shutdown_background_event = Event() + # captured by the refresher below, so that a later close()+connect() (which + # replaces the attribute) still stops THIS refresher and not the new one + shutdown = Event() + self._shutdown_background_event = shutdown if isinstance(self._client, AsyncOAuth2Client): try: @@ -575,7 +622,9 @@ def _create_background_token_refresh(self, _auth: Optional[_Auth] = None) -> Non if loop is not None: # The async colour refreshes on its own already-running loop: threads # cannot start under WASM/Pyodide and are unnecessary here anyway. - task = loop.create_task(self.__periodic_token_refresh_async(expires_in, _auth)) + task = loop.create_task( + self.__periodic_token_refresh_async(expires_in, _auth, shutdown) + ) task.add_done_callback(self.__warn_if_token_refresh_died) self.__token_refresh_task = task return @@ -617,12 +666,9 @@ def update_refresh_time() -> int: return self._client.token.get("expires_in", 60) - 30 def periodic_refresh_token(refresh_time: int, _auth: Optional[_Auth]) -> None: - while ( - self._shutdown_background_event is not None - and not self._shutdown_background_event.is_set() - ): - # use refresh token when available - time.sleep(max(refresh_time, 1)) + failures = 0 + # Event.wait instead of time.sleep so close() ends the thread promptly + while not shutdown.wait(timeout=max(refresh_time, 1)): try: if self._client is None: continue @@ -636,10 +682,12 @@ def periodic_refresh_token(refresh_time: int, _auth: Optional[_Auth]) -> None: # saved credentials refresh_session() refresh_time = update_refresh_time() - except HTTPError as exc: - # retry again after one second, might be an unstable connection - refresh_time = 1 - _Warnings.token_refresh_failed(exc) + failures = 0 + except Exception as exc: + # any failure (not only transport errors) keeps the refresher alive + failures += 1 + refresh_time = _token_refresh_backoff(failures) + _Warnings.token_refresh_failed(exc, refresh_time, failures) demon = Thread( target=periodic_refresh_token, @@ -649,25 +697,38 @@ def periodic_refresh_token(refresh_time: int, _auth: Optional[_Auth]) -> None: ) demon.start() - def _cancel_background_token_refresh(self) -> None: + def _cancel_background_token_refresh(self) -> Optional["asyncio.Task[None]"]: """Stop the token refresher, whichever form it took. - Sets the shutdown event (observed by the sync colour's daemon thread at its next - wake-up) and cancels the async colour's refresh task immediately. + Sets the shutdown event (ends the sync colour's daemon thread) and cancels the + async colour's refresh task. Returns that task when it belongs to the running + loop so the caller can await its wind-down; a task on another loop is only asked + to cancel (thread-safely) and never awaited, which would hang. """ if self._shutdown_background_event is not None: self._shutdown_background_event.set() - if self.__token_refresh_task is not None: - self.__token_refresh_task.cancel() - self.__token_refresh_task = None + task, self.__token_refresh_task = self.__token_refresh_task, None + if task is None: + return None + try: + running: Optional[asyncio.AbstractEventLoop] = asyncio.get_running_loop() + except RuntimeError: + running = None + loop = task.get_loop() + if loop is running: + task.cancel() + return task + if not loop.is_closed(): + loop.call_soon_threadsafe(task.cancel) + return None @staticmethod def __warn_if_token_refresh_died(task: "asyncio.Task[None]") -> None: """Surface a refresher death that happened outside the loop body's own handler. - Nothing ever awaits this task and ``_cancel_background_token_refresh`` drops the - reference, so without this callback not even asyncio's "exception was never - retrieved" message fires — the client just silently stops refreshing. + Nothing awaits this task while it runs (close() only gathers it with + return_exceptions=True), so without this callback not even asyncio's "exception + was never retrieved" message fires — the client just silently stops refreshing. """ if task.cancelled(): return @@ -676,17 +737,14 @@ def __warn_if_token_refresh_died(task: "asyncio.Task[None]") -> None: _Warnings.token_refresh_stopped(exc) async def __periodic_token_refresh_async( - self, refresh_time: int, _auth: Optional[_Auth] + self, refresh_time: int, _auth: Optional[_Auth], shutdown: Event ) -> None: """Thread-free equivalent of ``periodic_refresh_token`` for the async colour. Cancelled by ``close('async')``. """ - while ( - self._shutdown_background_event is not None - and not self._shutdown_background_event.is_set() - ): - # use refresh token when available + failures = 0 + while not shutdown.is_set(): await asyncio.sleep(max(refresh_time, 1)) try: client = self._client @@ -701,17 +759,16 @@ async def __periodic_token_refresh_async( new_session = await _Auth.aresult(_auth.get_auth_session()) client.token = await new_session.fetch_token() refresh_time = client.token.get("expires_in", 60) - 30 + failures = 0 except asyncio.CancelledError: # close() cancels this task — cancellation must never be swallowed raise except Exception as exc: - # retry again after one second, might be an unstable connection. - # Catch broadly: an authlib OAuthError (e.g. invalid_grant), a KeyError on - # a missing token_endpoint or a malformed IdP body would otherwise kill the - # task with nobody watching, and every later request would 401 with nothing - # pointing at the token refresh. - refresh_time = 1 - _Warnings.token_refresh_failed(exc) + # any failure (an authlib OAuthError such as invalid_grant, a KeyError on a + # malformed IdP body, ...) keeps the refresher alive + failures += 1 + refresh_time = _token_refresh_backoff(failures) + _Warnings.token_refresh_failed(exc, refresh_time, failures) def __get_latest_headers(self) -> Dict[str, str]: if "authorization" in self._headers: @@ -740,42 +797,32 @@ def __get_timeout( They specify the times depending on how they expect Weaviate to behave. For example, a query might take longer than an insert or vice versa but, in either case, the user only cares about how long it takes for a response to be received. - The one exception is Emscripten/Pyodide, where the connect timeout bounds the whole fetch - promise rather than connection setup, so it has to follow the request timeout (see below). + Under Emscripten/Pyodide the fetch transport (weaviate-client-web) reads only `read`, + so it is the whole-request deadline there; a non-finite value means no deadline. https://www.python-httpx.org/advanced/timeouts/ """ + deadlines = self._deadlines timeout = None if method == "DELETE" or method == "PATCH" or method == "PUT": - timeout = self.timeout_config.insert + timeout = deadlines.insert elif method == "GET" or method == "HEAD": - timeout = self.timeout_config.query + timeout = deadlines.query elif method == "POST" and is_gql_query: - timeout = self.timeout_config.query + timeout = deadlines.query elif method == "POST" and not is_gql_query: - timeout = self.timeout_config.insert - - connect: float = 5.0 - if sys.platform == "emscripten" and timeout is not None: - # Under Pyodide the REST transport is httpx's AsyncJavascriptFetchTransport, - # where the connect timeout bounds the WHOLE fetch promise (upload + response) - # rather than connection setup only. A fixed 5s there silently caps every - # request at ~5s of wall clock and ignores the configured insert/query - # timeouts. Never below 5s, and untouched off Emscripten, so native platforms - # keep the httpx default for connect/write. - connect = max(connect, float(timeout)) + timeout = deadlines.insert + return Timeout( - timeout=connect, + timeout=5.0, read=timeout, pool=self.__connection_config.session_pool_timeout, ) def __handle_exceptions(self, e: Exception, error_msg: str) -> None: - # httpx raises a bare RuntimeError('Cannot send a request, as the client has been - # closed.'); match its message so unrelated RuntimeErrors (e.g. Emscripten's - # "can't start new thread") are not rewritten into a misleading 'client is - # closed' error. - if isinstance(e, RuntimeError) and "client has been closed" in str(e): + # httpx raises a bare RuntimeError for a closed client; only rewrite when the + # client really is closed, so unrelated RuntimeErrors keep their meaning + if isinstance(e, RuntimeError) and (self._client is None or self._client.is_closed): raise WeaviateClosedClientError() from e if isinstance(e, ConnectError): raise WeaviateConnectionError(self.__error_msg_with_detail(error_msg, e)) from e @@ -848,10 +895,14 @@ def exc(e: Exception) -> None: def close(self, colour: executor.Colour) -> executor.Result[None]: if self.embedded_db is not None: self.embedded_db.stop() - self._cancel_background_token_refresh() + refresh_task = self._cancel_background_token_refresh() if colour == "async": async def execute() -> None: + if refresh_task is not None: + # let the cancellation land before the client goes away; the task's + # own outcome (cancelled, or dead earlier) is not close()'s error + await asyncio.gather(refresh_task, return_exceptions=True) if self._client is not None: assert isinstance(self._client, AsyncClient) await self._client.aclose() @@ -889,7 +940,7 @@ def resp(res: Response) -> None: async def _execute() -> None: try: async with AsyncClient() as client: - res = await client.get(PYPI_PACKAGE_URL, timeout=self.timeout_config.init) + res = await client.get(PYPI_PACKAGE_URL, timeout=self._deadlines.init) return resp(res) except (RequestError, OSError): # ignore any errors related to requests, it is a best-effort warning. @@ -901,7 +952,7 @@ async def _execute() -> None: try: with Client() as client: - res = client.get(PYPI_PACKAGE_URL, timeout=self.timeout_config.init) + res = client.get(PYPI_PACKAGE_URL, timeout=self._deadlines.init) return resp(res) except (RequestError, OSError): pass # ignore any errors related to requests, it is a best-effort warning @@ -1039,13 +1090,22 @@ def resp(res: Response) -> Optional[Dict[str, Any]]: class ConnectionSync(_ConnectionBase): """Connection class used to communicate to a weaviate instance.""" - def __init__(self, *args: Any, **kwargs: Any) -> None: + def __init__( + self, + connection_params: ConnectionParams, + auth_client_secret: Optional[AuthCredentials], + timeout_config: TimeoutConfig, + proxies: Union[str, Proxies, None], + trust_env: bool, + additional_headers: Optional[Dict[str, Any]], + connection_config: ConnectionConfig, + embedded_db: Optional[EmbeddedV4] = None, + skip_init_checks: bool = False, + grpc_config: Optional[GrpcConfig] = None, + ) -> None: if sys.platform == "emscripten": - # Fail at construction with the async-only message; otherwise the first - # REST call surfaces an opaque ConnectError long before the grpc-web - # shim's own sync guard is reached (wording mirrors the shim's - # _ASYNC_ONLY_MESSAGE). Pre-set the attributes __del__ reads so the - # never-initialized instance is collected quietly. + # fail at construction, before the first REST call surfaces an opaque + # ConnectError; pre-set what __del__ reads so the instance is collected quietly self._client = None self._grpc_channel = None raise WeaviateStartUpError( @@ -1054,7 +1114,18 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: "use_async_with_weaviate_cloud / use_async_with_custom, or " "WeaviateAsyncClient) instead." ) - super().__init__(*args, **kwargs) + super().__init__( + connection_params=connection_params, + auth_client_secret=auth_client_secret, + timeout_config=timeout_config, + proxies=proxies, + trust_env=trust_env, + additional_headers=additional_headers, + connection_config=connection_config, + embedded_db=embedded_db, + skip_init_checks=skip_init_checks, + grpc_config=grpc_config, + ) def connect(self, force: bool = False) -> None: if self._connected and not force: @@ -1137,7 +1208,7 @@ def grpc_search(self, request: search_get_pb2.SearchRequest) -> search_get_pb2.S self.grpc_stub.Search, request, metadata=self.grpc_headers(), - timeout=self.timeout_config.query, + timeout=self._deadlines.query, ) return cast(search_get_pb2.SearchReply, res) except RpcError as e: @@ -1162,7 +1233,7 @@ def grpc_batch_objects( f=self.grpc_stub.BatchObjects, request=request, metadata=self.grpc_headers(), - timeout=timeout, + timeout=_deadline(timeout), ) res = cast(batch_pb2.BatchObjectsReply, res) @@ -1184,7 +1255,7 @@ def grpc_batch_stream( assert self.grpc_stub is not None for msg in self.grpc_stub.BatchStream( request_iterator=requests, - timeout=self.timeout_config.stream, + timeout=self._deadlines.stream, metadata=self.grpc_headers(), ): yield msg @@ -1206,7 +1277,7 @@ def grpc_batch_delete( self.grpc_stub.BatchDelete( request, metadata=self.grpc_headers(), - timeout=self.timeout_config.insert, + timeout=self._deadlines.insert, ), ) except RpcError as e: @@ -1226,7 +1297,7 @@ def grpc_tenants_get( self.grpc_stub.TenantsGet, request, metadata=self.grpc_headers(), - timeout=self.timeout_config.query, + timeout=self._deadlines.query, ) except RpcError as e: error = cast(Call, e) @@ -1247,7 +1318,7 @@ def grpc_aggregate( self.grpc_stub.Aggregate, request, metadata=self.grpc_headers(), - timeout=self.timeout_config.query, + timeout=self._deadlines.query, ) return cast(aggregate_pb2.AggregateReply, res) except RpcError as e: @@ -1348,7 +1419,7 @@ async def grpc_search( self.grpc_stub.Search, request, metadata=self.grpc_headers(), - timeout=self.timeout_config.query, + timeout=self._deadlines.query, ) return cast(search_get_pb2.SearchReply, res) except AioRpcError as e: @@ -1372,7 +1443,7 @@ async def grpc_batch_objects( f=self.grpc_stub.BatchObjects, request=request, metadata=self.grpc_headers(), - timeout=timeout, + timeout=_deadline(timeout), ) res = cast(batch_pb2.BatchObjectsReply, res) @@ -1383,7 +1454,7 @@ async def grpc_batch_objects( except AioRpcError as e: if e.code().name == PERMISSION_DENIED: raise InsufficientPermissionsError(e) - raise WeaviateBatchError(str(e)) from e + raise WeaviateBatchError(str(e.details())) from e async def grpc_batch_delete( self, request: batch_delete_pb2.BatchDeleteRequest @@ -1393,7 +1464,7 @@ async def grpc_batch_delete( return await self.grpc_stub.BatchDelete( request, metadata=self.grpc_headers(), - timeout=self.timeout_config.insert, + timeout=self._deadlines.insert, ) except AioRpcError as e: if e.code().name == PERMISSION_DENIED: @@ -1412,7 +1483,7 @@ async def grpc_batch_stream( response_deserializer=batch_pb2.BatchStreamReply.FromString, )( request_iterator=requests, - timeout=self.timeout_config.stream, + timeout=self._deadlines.stream, metadata=self.grpc_headers(), ): yield msg @@ -1467,7 +1538,7 @@ async def grpc_tenants_get( self.grpc_stub.TenantsGet, request, metadata=self.grpc_headers(), - timeout=self.timeout_config.query, + timeout=self._deadlines.query, ) except AioRpcError as e: if e.code().name == PERMISSION_DENIED: @@ -1487,7 +1558,7 @@ async def grpc_aggregate( self.grpc_stub.Aggregate, request, metadata=self.grpc_headers(), - timeout=self.timeout_config.query, + timeout=self._deadlines.query, ) return cast(aggregate_pb2.AggregateReply, res) except AioRpcError as e: diff --git a/weaviate/warnings.py b/weaviate/warnings.py index 053bac634..a1dd3180b 100644 --- a/weaviate/warnings.py +++ b/weaviate/warnings.py @@ -67,13 +67,11 @@ def auth_cannot_parse_oidc_config(url: str) -> None: warnings.warn(message=msg, category=UserWarning, stacklevel=1) @staticmethod - def token_refresh_failed(exc: Exception) -> None: + def token_refresh_failed(exc: Exception, retry_in: float, failures: int) -> None: + detail = f"{type(exc).__name__}: {exc}" if str(exc) else repr(exc) warnings.warn( - message=f"""Con001: Could not reach token issuer for the periodic refresh. This client will automatically - retry to refresh. If the retry does not succeed, the client will become unauthenticated. - - The cause might be an unstable internet connection or a problem with your authentication provider. - Exception: {exc} + message=f"""Con001: Token refresh failed ({detail}); retrying in {retry_in}s (consecutive failures: {failures}). + The client will become unauthenticated once the current token expires if the refresh keeps failing. """, category=UserWarning, stacklevel=1, From c099eb3cdd738e4b3d81bfed7f81d1c9255485d2 Mon Sep 17 00:00:00 2001 From: Ivan Despot <66276597+g-despot@users.noreply.github.com> Date: Mon, 17 Aug 2026 20:35:17 +0200 Subject: [PATCH 5/6] fix(batch): keep the caller's exception on stream exit; sync/async parity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Leaving `with`/`async with client.batch.stream()` no longer replaces the exception raised in the block (CancelledError included) with the background failure — the caller's wins and the background one is logged, unless it is the same exception already propagating. Both colours raise WeaviateBatchStreamError when the workers failed or the stream ended with data still queued, with a message that says how much was unsent; the sync colour previously swallowed this. _BatchStreamShutdownError joins the taxonomy, the shutdown wait tolerates an infinite insert timeout, and WeaviateBatchError carries the gRPC details only. --- test/test_batch_async.py | 179 +++++++++++++++++++- test/test_batch_sync.py | 164 ++++++++++++++++++ weaviate/collections/batch/async_.py | 40 ++--- weaviate/collections/batch/batch_wrapper.py | 29 +++- weaviate/collections/batch/client.py | 10 ++ weaviate/collections/batch/collection.py | 10 ++ weaviate/collections/batch/sync.py | 21 ++- weaviate/exceptions.py | 7 +- 8 files changed, 432 insertions(+), 28 deletions(-) create mode 100644 test/test_batch_sync.py diff --git a/test/test_batch_async.py b/test/test_batch_async.py index 5ad611b94..cc8a9a3f2 100644 --- a/test/test_batch_async.py +++ b/test/test_batch_async.py @@ -7,13 +7,16 @@ """ import asyncio +import logging +from typing import Optional import grpc import pytest from weaviate.collections.batch.async_ import _BatchBaseAsync, _BgTasks from weaviate.collections.batch.base import _BatchDataWrapper -from weaviate.exceptions import WeaviateBatchStreamError +from weaviate.collections.batch.batch_wrapper import _ContextManagerAsync +from weaviate.exceptions import WeaviateBatchStreamError, _BatchStreamShutdownError class _NotAnException(BaseException): @@ -155,3 +158,177 @@ class FakeConnection: with pytest.raises(RuntimeError, match="boom"): asyncio.run(batch._wait()) assert backup.failed_objects == ["sentinel-failure"] + + +async def _finished_task() -> "asyncio.Task[None]": + """A background task that returned normally (e.g. the server closed the stream).""" + task = asyncio.get_running_loop().create_task(asyncio.sleep(0)) + await task + return task + + +class _FakeTimeouts: + insert = 1 + + +class _FakeConnection: + timeout_config = _FakeTimeouts() + + +def test_wait_names_unsent_data_when_the_tasks_ended_cleanly() -> None: + # both tasks returned normally (server closed the stream) with data still queued: + # that is not "the background tasks died unexpectedly", it is an early end of the + # stream — say what was left behind + async def run() -> None: + batch = _bare_batch( + bg_exception=None, + bg_tasks=_BgTasks(recv=await _finished_task(), loop=await _finished_task()), + connection=_FakeConnection(), + results_for_wrapper=_BatchDataWrapper(), + results_for_wrapper_backup=_BatchDataWrapper(), + batch_objects=[object(), object()], + batch_references=[object()], + ) + await batch._wait() + + with pytest.raises( + WeaviateBatchStreamError, match="ended with 2 objects and 1 references unsent" + ): + asyncio.run(run()) + + +def test_check_alive_after_a_clean_end_says_the_stream_ended() -> None: + async def run() -> None: + batch = _bare_batch( + bg_exception=None, + bg_tasks=_BgTasks(recv=await _finished_task(), loop=await _finished_task()), + ) + getattr(batch, "_BatchBaseAsync__check_bg_tasks_alive")() # noqa: B009 + + with pytest.raises(WeaviateBatchStreamError, match="stream has ended"): + asyncio.run(run()) + + +def test_put_gives_up_when_the_tasks_are_dead() -> None: + # a full queue with a dead receiver never drains: __put must return False instead of + # retrying (and recursing) once per second forever + async def run() -> bool: + reqs: asyncio.Queue = asyncio.Queue(maxsize=1) + await reqs.put(object()) # full + batch = _bare_batch( + reqs=reqs, + bg_exception=None, # nothing recorded ... + shutdown_loop=asyncio.Event(), # ... and no shutdown either + bg_tasks=_BgTasks(recv=await _dead_task("cancel"), loop=await _finished_task()), + ) + put = getattr(batch, "_BatchBaseAsync__put") # noqa: B009 + return await asyncio.wait_for(put(object()), timeout=5) + + assert asyncio.run(run()) is False + + +def test_batch_stream_shutdown_error_is_in_the_taxonomy() -> None: + # raised on gRPC ABORTED and can surface from a clean `async with` exit + assert issubclass(_BatchStreamShutdownError, WeaviateBatchStreamError) + assert isinstance(_BatchStreamShutdownError(), Exception) + + +class _FakeBatch: + """Stands in for _BatchBaseAsync behind the context manager.""" + + def __init__(self, wait_error: Optional[BaseException] = None) -> None: + self.wait_error = wait_error + self.shutdown_called = False + self.wait_called = False + + async def _start(self) -> None: + pass + + async def _shutdown(self) -> None: + self.shutdown_called = True + + async def _wait(self) -> None: + self.wait_called = True + if self.wait_error is not None: + raise self.wait_error + + +def test_aexit_raises_a_background_failure_on_a_clean_block() -> None: + fake = _FakeBatch(WeaviateBatchStreamError("bg died")) + + async def run() -> None: + async with _ContextManagerAsync(fake): # type: ignore[arg-type] + pass + + with pytest.raises(WeaviateBatchStreamError, match="bg died"): + asyncio.run(run()) + + +def test_aexit_keeps_the_users_exception_over_a_background_failure(caplog) -> None: + # the block's own exception must not be REPLACED by the background failure (which + # used to demote it to __context__); the failure is logged instead + fake = _FakeBatch(WeaviateBatchStreamError("bg died")) + + async def run() -> None: + async with _ContextManagerAsync(fake): # type: ignore[arg-type] + raise ValueError("user code") + + with caplog.at_level(logging.WARNING, logger="weaviate-client"): + with pytest.raises(ValueError, match="user code"): + asyncio.run(run()) + assert fake.shutdown_called and fake.wait_called # still drained/awaited + assert "bg died" in caplog.text + + +def test_aexit_never_swallows_cancellation() -> None: + # a CancelledError leaving the block used to be replaced by the background failure, + # i.e. the cancellation was swallowed + fake = _FakeBatch(WeaviateBatchStreamError("bg died")) + + async def run() -> None: + async with _ContextManagerAsync(fake): # type: ignore[arg-type] + raise asyncio.CancelledError() + + with pytest.raises(asyncio.CancelledError): + asyncio.run(run()) + assert fake.wait_called + + +def test_wait_with_an_infinite_insert_timeout_does_not_raise() -> None: + # Timeout(insert=inf) means "no deadline": the shutdown wait must not turn it into an + # error (the sync colour's Thread.join(inf) overflows; keep both colours consistent) + from weaviate.config import Timeout as TimeoutConfig + + class InfiniteInsert: + timeout_config = TimeoutConfig(insert=float("inf")) + + async def run() -> None: + batch = _bare_batch( + bg_exception=None, + bg_tasks=_BgTasks(recv=await _finished_task(), loop=await _finished_task()), + connection=InfiniteInsert(), + results_for_wrapper=_BatchDataWrapper(), + results_for_wrapper_backup=_BatchDataWrapper(), + batch_objects=[], + batch_references=[], + ) + await batch._wait() + + asyncio.run(run()) + + +def test_aexit_does_not_log_the_exception_that_is_already_propagating(caplog) -> None: + # flush()/add_object raised __bg_exception inside the block; _wait() re-raises the + # SAME object on exit — that is not a second failure to log + err = WeaviateBatchStreamError("bg died") + fake = _FakeBatch(err) + + async def run() -> None: + async with _ContextManagerAsync(fake): # type: ignore[arg-type] + raise err + + with caplog.at_level(logging.WARNING, logger="weaviate-client"): + with pytest.raises(WeaviateBatchStreamError, match="bg died"): + asyncio.run(run()) + assert fake.wait_called + assert "batch stream failed" not in caplog.text diff --git a/test/test_batch_sync.py b/test/test_batch_sync.py new file mode 100644 index 000000000..5656e7388 --- /dev/null +++ b/test/test_batch_sync.py @@ -0,0 +1,164 @@ +"""Unit tests for the sync batch-stream failure handling, mirroring test_batch_async.py. + +_wait() surfaces a background failure (or data left unsent) instead of returning as if +the batch had succeeded, while an exception leaving the `with` block still wins. +""" + +import logging +import threading +import time +from typing import Optional + +import pytest + +from weaviate.collections.batch.base import _BatchDataWrapper, _BgThreads +from weaviate.collections.batch.batch_wrapper import _ContextManagerSync +from weaviate.collections.batch.sync import _BatchBaseSync +from weaviate.config import Timeout as TimeoutConfig +from weaviate.exceptions import WeaviateBatchStreamError + + +def _bare_batch(**mangled) -> _BatchBaseSync: + batch = object.__new__(_BatchBaseSync) + for name, value in mangled.items(): + setattr(batch, f"_BatchBaseSync__{name}", value) + return batch + + +class _FakeThreads: + def __init__(self, alive: bool = False) -> None: + self.alive = alive + + def join(self, timeout=None) -> None: + return None + + def is_alive(self) -> bool: + return self.alive + + +class _FakeTimeouts: + insert = 1 + + +class _FakeConnection: + timeout_config = _FakeTimeouts() + + +def _batch_for_wait(**mangled) -> _BatchBaseSync: + defaults = { + "bg_exception": None, + "bg_threads": _FakeThreads(alive=False), + "connection": _FakeConnection(), + "results_for_wrapper": _BatchDataWrapper(), + "results_for_wrapper_backup": _BatchDataWrapper(), + "batch_objects": [], + "batch_references": [], + } + defaults.update(mangled) + return _bare_batch(**defaults) + + +def test_wait_returns_quietly_when_everything_was_sent() -> None: + _batch_for_wait()._wait() + + +def test_wait_raises_the_background_exception_and_keeps_partial_results() -> None: + # like the async colour: a background failure must not come back as a success + partial = _BatchDataWrapper() + partial.failed_objects = ["sentinel-failure"] # type: ignore[list-item] + backup = _BatchDataWrapper() + batch = _batch_for_wait( + bg_exception=RuntimeError("boom"), + results_for_wrapper=partial, + results_for_wrapper_backup=backup, + ) + + with pytest.raises(RuntimeError, match="boom"): + batch._wait() + assert backup.failed_objects == ["sentinel-failure"] + + +def test_wait_names_unsent_data_when_the_threads_are_gone() -> None: + batch = _batch_for_wait(batch_objects=[object(), object()], batch_references=[object()]) + with pytest.raises( + WeaviateBatchStreamError, match="ended with 2 objects and 1 references unsent" + ): + batch._wait() + + +def test_check_alive_raises_inside_the_taxonomy() -> None: + # used to be a bare Exception("Batch thread died unexpectedly") + batch = _bare_batch(bg_exception=None, bg_threads=_FakeThreads(alive=False)) + with pytest.raises(WeaviateBatchStreamError, match="stream has ended"): + getattr(batch, "_BatchBaseSync__check_bg_threads_alive")() # noqa: B009 + + +class _FakeBatch: + def __init__(self, wait_error: Optional[BaseException] = None) -> None: + self.wait_error = wait_error + self.shutdown_called = False + self.wait_called = False + + def _start(self) -> None: + pass + + def _shutdown(self) -> None: + self.shutdown_called = True + + def _wait(self) -> None: + self.wait_called = True + if self.wait_error is not None: + raise self.wait_error + + +def test_exit_raises_a_background_failure_on_a_clean_block() -> None: + fake = _FakeBatch(WeaviateBatchStreamError("bg died")) + with pytest.raises(WeaviateBatchStreamError, match="bg died"): + with _ContextManagerSync(fake): # type: ignore[type-var] + pass + + +def test_exit_keeps_the_users_exception_over_a_background_failure(caplog) -> None: + fake = _FakeBatch(WeaviateBatchStreamError("bg died")) + with caplog.at_level(logging.WARNING, logger="weaviate-client"): + with pytest.raises(ValueError, match="user code"): + with _ContextManagerSync(fake): # type: ignore[type-var] + raise ValueError("user code") + assert fake.shutdown_called and fake.wait_called + assert "bg died" in caplog.text + + +def test_exit_never_swallows_a_base_exception() -> None: + fake = _FakeBatch(WeaviateBatchStreamError("bg died")) + with pytest.raises(KeyboardInterrupt): + with _ContextManagerSync(fake): # type: ignore[type-var] + raise KeyboardInterrupt() + assert fake.wait_called + + +def test_wait_with_an_infinite_insert_timeout_does_not_overflow_join() -> None: + # Timeout(insert=inf) is accepted (it means "no deadline"), but Thread.join(inf) + # raises OverflowError — the shutdown wait must become "no timeout" instead + class InfiniteInsert: + timeout_config = TimeoutConfig(insert=float("inf")) + + threads = _BgThreads( + loop=threading.Thread(target=lambda: None), recv=threading.Thread(target=lambda: None) + ) + threads.start_recv() + threads.start_loop() + time.sleep(0.1) # let both finish; is_alive()/join() are deliberately not called yet + _batch_for_wait(bg_threads=threads, connection=InfiniteInsert())._wait() + + +def test_exit_does_not_log_the_exception_that_is_already_propagating(caplog) -> None: + # flush()/add_object raised __bg_exception inside the block; _wait() re-raises the + # SAME object on exit — that is not a second failure to log + err = WeaviateBatchStreamError("bg died") + fake = _FakeBatch(err) + with caplog.at_level(logging.WARNING, logger="weaviate-client"): + with pytest.raises(WeaviateBatchStreamError, match="bg died"): + with _ContextManagerSync(fake): # type: ignore[type-var] + raise err + assert fake.wait_called + assert "batch stream failed" not in caplog.text diff --git a/weaviate/collections/batch/async_.py b/weaviate/collections/batch/async_.py index 21310ab94..d09bbf723 100644 --- a/weaviate/collections/batch/async_.py +++ b/weaviate/collections/batch/async_.py @@ -39,7 +39,7 @@ from weaviate.collections.classes.types import WeaviateProperties from weaviate.connect.base import _grpc_web_shim_active from weaviate.connect.executor import aresult -from weaviate.connect.v4 import ConnectionAsync +from weaviate.connect.v4 import ConnectionAsync, _deadline from weaviate.exceptions import ( WeaviateBatchFailedToReestablishStreamError, WeaviateBatchStreamError, @@ -194,7 +194,8 @@ async def recv_wrapper() -> None: async def _wait(self) -> None: assert self.__bg_tasks is not None # this is how long an insert will take to timeout for, so we wait at most this time +5s for the batch to finish after shutdown is initiated, in case the server never hangs up - shutdown_timeout = self.__connection.timeout_config.insert + 5 + insert = _deadline(self.__connection.timeout_config.insert) + shutdown_timeout = None if insert is None else insert + 5 try: await self.__bg_tasks.gather(timeout=shutdown_timeout) except asyncio.TimeoutError as e: @@ -217,13 +218,14 @@ async def _wait(self) -> None: # surface background-task failures instead of returning partial results # as if the batch had succeeded raise self.__bg_exception - if ( - len(self.__batch_objects) > 0 or len(self.__batch_references) > 0 - ) and not self.__all_tasks_alive(): - # the tasks are gone with data still queued and nothing recorded in - # __bg_exception (a BaseException escaping loop_wrapper/recv_wrapper): the - # batch did NOT complete, so do not return as if it had - raise self.__bg_task_death_cause() + n_objs, n_refs = len(self.__batch_objects), len(self.__batch_references) + if n_objs + n_refs > 0 and not self.__all_tasks_alive(): + # the tasks are gone with data still queued: the batch did NOT complete, so + # do not return as if it had — whether a task died (a BaseException escaping + # loop_wrapper/recv_wrapper) or the server ended the stream early + raise self.__bg_task_death_cause() or WeaviateBatchStreamError( + f"batch stream ended with {n_objs} objects and {n_refs} references unsent" + ) async def _shutdown(self) -> None: self.__is_stopped.set() @@ -560,12 +562,8 @@ async def flush(self) -> None: """Flush the batch queue and wait for all requests to be finished.""" # bg thread is sending objs+refs automatically, so simply wait for everything to be done while len(self.__batch_objects) > 0 or len(self.__batch_references) > 0: - # a dead task is the condition to check, not __bg_exception: loop_wrapper / - # recv_wrapper only catch Exception, so a BaseException (notably the - # asyncio.CancelledError grpc.aio raises on a cancelled streaming call) ends - # the task with __bg_exception unset. Nothing then drains the queues, so - # waiting any longer would hang forever. Mirrors the sync colour's - # __check_bg_threads_alive(). + # a dead task (not just __bg_exception, which a BaseException such as grpc.aio's + # CancelledError leaves unset) means nothing drains the queues: raise, don't hang self.__check_bg_tasks_alive() await asyncio.sleep(0.01) @@ -664,10 +662,14 @@ def __check_bg_tasks_alive(self) -> None: if self.__all_tasks_alive(): return - raise self.__bg_exception or self.__bg_task_death_cause() + raise ( + self.__bg_exception + or self.__bg_task_death_cause() + or WeaviateBatchStreamError("the batch stream has ended") + ) - def __bg_task_death_cause(self) -> Exception: - """Explain a background task that ended without setting __bg_exception. + def __bg_task_death_cause(self) -> Optional[Exception]: + """Explain a background task that died without setting __bg_exception; None if none did. loop_wrapper/recv_wrapper only catch Exception, so a BaseException — notably the asyncio.CancelledError grpc.aio raises when a streaming call is cancelled — ends @@ -689,4 +691,4 @@ def __bg_task_death_cause(self) -> Exception: return exc if exc is not None: return WeaviateBatchStreamError(f"the background {name} task died with {exc!r}") - return WeaviateBatchStreamError("the background tasks died unexpectedly") + return None diff --git a/weaviate/collections/batch/batch_wrapper.py b/weaviate/collections/batch/batch_wrapper.py index a3a3598d6..eb1353dd7 100644 --- a/weaviate/collections/batch/batch_wrapper.py +++ b/weaviate/collections/batch/batch_wrapper.py @@ -508,7 +508,18 @@ def __init__(self, current_batch: T): def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: self.__current_batch._shutdown() - self.__current_batch._wait() + if exc_type is None: + self.__current_batch._wait() + return + # the exception leaving the block wins; a background failure is only logged + # (unless it IS the one in flight, re-raised by _wait after flush/add already did) + try: + self.__current_batch._wait() + except Exception as e: + if e is not exc_val: + logger.warning( + f"batch stream failed while the block raised {exc_type.__name__}: {e}" + ) def __enter__(self) -> P: self.__current_batch._start() @@ -521,7 +532,21 @@ def __init__(self, current_batch: _BatchBaseAsync): async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: await self.__current_batch._shutdown() - await self.__current_batch._wait() + if exc_type is None: + await self.__current_batch._wait() + return + # the exception leaving the block wins (never replace a CancelledError or other + # BaseException); a background failure is only logged (unless it IS the one in + # flight, re-raised by _wait after flush/add already did) + try: + await self.__current_batch._wait() + except asyncio.CancelledError: + raise + except Exception as e: + if e is not exc_val: + logger.warning( + f"batch stream failed while the block raised {exc_type.__name__}: {e}" + ) async def __aenter__(self) -> Q: await self.__current_batch._start() diff --git a/weaviate/collections/batch/client.py b/weaviate/collections/batch/client.py index d9a2524d7..b95238e03 100644 --- a/weaviate/collections/batch/client.py +++ b/weaviate/collections/batch/client.py @@ -285,6 +285,11 @@ def stream( Args: concurrency: The number of concurrent streams to use when sending batches. If not provided, the default will be one. consistency_level: The consistency level to be used when inserting data. If not provided, the default value is `None`. + + Raises: + WeaviateBatchStreamError: On exit, if the background stream failed or ended with objects or references + still unsent. An exception raised inside the block is propagated unchanged; a background failure is + then only logged. """ if self._connection._weaviate_version.is_lower_than(1, 36, 0): raise WeaviateUnsupportedFeatureError( @@ -344,6 +349,11 @@ def stream( Args: concurrency: The number of concurrent streams to use when sending batches. If not provided, the default will be one. consistency_level: The consistency level to be used when inserting data. If not provided, the default value is `None`. + + Raises: + WeaviateBatchStreamError: On exit, if the background stream failed or ended with objects or references + still unsent. An exception raised inside the block is propagated unchanged; a background failure is + then only logged. """ if self._connection._weaviate_version.is_lower_than(1, 36, 0): raise WeaviateUnsupportedFeatureError( diff --git a/weaviate/collections/batch/collection.py b/weaviate/collections/batch/collection.py index 415b2df49..565412563 100644 --- a/weaviate/collections/batch/collection.py +++ b/weaviate/collections/batch/collection.py @@ -310,6 +310,11 @@ def stream( Args: concurrency: The number of concurrent requests when sending batches. This controls the number of concurrent requests made to Weaviate. If not provided, the default value is 1. + + Raises: + WeaviateBatchStreamError: On exit, if the background stream failed or ended with objects or references + still unsent. An exception raised inside the block is propagated unchanged; a background failure is + then only logged. """ if self._connection._weaviate_version.is_lower_than(1, 36, 0): raise WeaviateUnsupportedFeatureError( @@ -370,6 +375,11 @@ def stream( Args: concurrency: The number of concurrent requests when sending batches. This controls the number of concurrent requests made to Weaviate. If not provided, the default value is 1. + + Raises: + WeaviateBatchStreamError: On exit, if the background stream failed or ended with objects or references + still unsent. An exception raised inside the block is propagated unchanged; a background failure is + then only logged. """ if self._connection._weaviate_version.is_lower_than(1, 36, 0): raise WeaviateUnsupportedFeatureError( diff --git a/weaviate/collections/batch/sync.py b/weaviate/collections/batch/sync.py index 6cf8c1edc..40e53b6c0 100644 --- a/weaviate/collections/batch/sync.py +++ b/weaviate/collections/batch/sync.py @@ -35,7 +35,7 @@ ) from weaviate.collections.classes.types import WeaviateProperties from weaviate.connect.executor import result -from weaviate.connect.v4 import ConnectionSync +from weaviate.connect.v4 import ConnectionSync, _deadline from weaviate.exceptions import ( WeaviateBatchFailedToReestablishStreamError, WeaviateBatchStreamError, @@ -134,7 +134,8 @@ def _start(self) -> None: def _wait(self) -> None: # this is how long an insert will take to timeout for, so we wait at most this time +5s for the batch to finish after shutdown is initiated, in case the server never hangs up - shutdown_timeout = self.__connection.timeout_config.insert + 5 + insert = _deadline(self.__connection.timeout_config.insert) + shutdown_timeout = None if insert is None else insert + 5 # inf overflows Thread.join try: self.__bg_threads.join(shutdown_timeout) except TimeoutError as e: @@ -142,7 +143,8 @@ def _wait(self) -> None: "Background batch threads did not terminate after forced shutdown." ) from e - # copy the results to the public results + # copy the results to the public results — even on failure, so the user can + # still inspect batch.results / batch.failed_objects after catching self.__results_for_wrapper_backup.results = self.__results_for_wrapper.results self.__results_for_wrapper_backup.failed_objects = self.__results_for_wrapper.failed_objects self.__results_for_wrapper_backup.failed_references = ( @@ -152,6 +154,17 @@ def _wait(self) -> None: self.__results_for_wrapper.imported_shards ) + if self.__bg_exception is not None: + # surface background-thread failures instead of returning partial results + # as if the batch had succeeded + raise self.__bg_exception + n_objs, n_refs = len(self.__batch_objects), len(self.__batch_references) + if n_objs + n_refs > 0 and not self.__all_threads_alive(): + # the threads are gone with data still queued: the batch did NOT complete + raise WeaviateBatchStreamError( + f"batch stream ended with {n_objs} objects and {n_refs} references unsent" + ) + def _shutdown(self) -> None: # Shutdown the current batch and wait for all requests to be finished self.__is_stopped.set() @@ -644,4 +657,4 @@ def __check_bg_threads_alive(self) -> None: if self.__all_threads_alive(): return - raise self.__bg_exception or Exception("Batch thread died unexpectedly") + raise self.__bg_exception or WeaviateBatchStreamError("the batch stream has ended") diff --git a/weaviate/exceptions.py b/weaviate/exceptions.py index 2024a1258..3f0873a32 100644 --- a/weaviate/exceptions.py +++ b/weaviate/exceptions.py @@ -473,5 +473,8 @@ def __init__(self, pb: version.Version, grpc: version.Version) -> None: ) -class _BatchStreamShutdownError(Exception): - """Internal exception to signal that the batch stream was shutdown.""" +class _BatchStreamShutdownError(WeaviateBatchStreamError): + """Internal exception to signal that the batch stream was shutdown (gRPC ABORTED).""" + + def __init__(self, message: str = "the server aborted the batch stream") -> None: + super().__init__(message) From 8a18975e852100b7de732f4e5dd0a6227a638efd Mon Sep 17 00:00:00 2001 From: Ivan Despot <66276597+g-despot@users.noreply.github.com> Date: Mon, 17 Aug 2026 20:35:32 +0200 Subject: [PATCH 6/6] ci: type-check and flake8 packages/web pyright now includes packages/web/src and the lint job's flake8 run covers the package alongside ruff, so the companion is held to the same bar as the client. --- .github/workflows/main.yaml | 2 +- pyrightconfig.json | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml index 240aaa417..163c70afa 100644 --- a/.github/workflows/main.yaml +++ b/.github/workflows/main.yaml @@ -49,7 +49,7 @@ jobs: - name: "Ruff format" run: ruff format --diff weaviate test mock_tests integration packages/web - name: "Flake 8" - run: flake8 weaviate test mock_tests integration + run: flake8 weaviate test mock_tests integration packages/web - name: "Check release for pypi" run: | python -m build diff --git a/pyrightconfig.json b/pyrightconfig.json index 396d62cfd..61eb3eaa3 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -1,6 +1,6 @@ { "include": [ - "weaviate", "integration" + "weaviate", "integration", "packages/web/src" ], "exclude": [