fix(grpc-web): round-2 review fixes — REST under Pyodide, OIDC refresh backoff, batch exit semantics

.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

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:

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(