From fffd60ea2f08fa5557b2a11b9e98354fc8df751b Mon Sep 17 00:00:00 2001 From: Vaibhav Dewangan Date: Fri, 31 Jul 2026 21:52:43 +0530 Subject: [PATCH 1/2] fix(batch): set has_errors on per-object/reference errors in stream recv In _BatchBaseSync/_BatchBaseAsync.__recv, each failed object or reference from the server was wrapped in a fresh BatchObjectReturn/BatchReferenceReturn with only `errors` set. has_errors defaults to False on the dataclass and __add__ only ORs it forward (`self.has_errors or other.has_errors`), so it never became True even though `errors` was non-empty. This meant `if result.has_errors:` never fired on data.ingest() (and on batch.stream(), though failed_objects/failed_references were populated correctly there since those are tracked separately). Fixes #2107 --- mock_tests/test_batch.py | 41 ++++++++++++++++++++++++++++ weaviate/collections/batch/async_.py | 2 ++ weaviate/collections/batch/sync.py | 2 ++ 3 files changed, 45 insertions(+) diff --git a/mock_tests/test_batch.py b/mock_tests/test_batch.py index c00bd08db..f8b114534 100644 --- a/mock_tests/test_batch.py +++ b/mock_tests/test_batch.py @@ -60,3 +60,44 @@ def test_ssb_canceled_stream( for i in range(HOW_MANY): batch.add_object({"name": f"Object {i}"}) assert len(service.uuids) == HOW_MANY + + +class MockFailedObjectWeaviateService(weaviate_pb2_grpc.WeaviateServicer): + def BatchStream( + self, + request_iterator: Generator[batch_pb2.BatchStreamRequest, None, None], + context: grpc.ServicerContext, + ) -> Generator[batch_pb2.BatchStreamReply, None, None]: + yield batch_pb2.BatchStreamReply(started=batch_pb2.BatchStreamReply.Started()) + for request in request_iterator: + if request.HasField("data"): + uuids = [obj.uuid for obj in request.data.objects.values] + yield batch_pb2.BatchStreamReply( + results=batch_pb2.BatchStreamReply.Results( + errors=[ + batch_pb2.BatchStreamReply.Results.Error( + uuid=uuid, error="mock failure" + ) + for uuid in uuids + ] + ) + ) + if request.HasField("stop"): + return + + +@pytest.fixture(scope="function") +def failed_object_stream( + canceled_stream_client: weaviate.WeaviateClient, start_grpc_server: grpc.Server +): + service = MockFailedObjectWeaviateService() + weaviate_pb2_grpc.add_WeaviateServicer_to_server(service, start_grpc_server) + return canceled_stream_client.collections.use(mock_class["class"]) + + +def test_ingest_has_errors_on_failed_object( + failed_object_stream: weaviate.collections.Collection, +): + result = failed_object_stream.data.ingest([{"name": "Object 1"}]) + assert result.has_errors is True + assert len(result.errors) == 1 diff --git a/weaviate/collections/batch/async_.py b/weaviate/collections/batch/async_.py index 8b997586c..2e7aa9279 100644 --- a/weaviate/collections/batch/async_.py +++ b/weaviate/collections/batch/async_.py @@ -407,6 +407,7 @@ async def __recv(self) -> None: result_objs += BatchObjectReturn( _all_responses=[err], errors={cached.index: err}, + has_errors=True, ) failed_objs.append(err) logger.warning( @@ -428,6 +429,7 @@ async def __recv(self) -> None: ) result_refs += BatchReferenceReturn( errors={cached.index: err}, + has_errors=True, ) failed_refs.append(err) logger.warning( diff --git a/weaviate/collections/batch/sync.py b/weaviate/collections/batch/sync.py index 6627f8911..cff631e0e 100644 --- a/weaviate/collections/batch/sync.py +++ b/weaviate/collections/batch/sync.py @@ -365,6 +365,7 @@ def __recv(self) -> None: result_objs += BatchObjectReturn( _all_responses=[err], errors={cached.index: err}, + has_errors=True, ) failed_objs.append(err) logger.warning( @@ -387,6 +388,7 @@ def __recv(self) -> None: failed_refs.append(err) result_refs += BatchReferenceReturn( errors={cached.index: err}, + has_errors=True, ) logger.warning( { From 11ed12ff8fe7c0ccf598c4a908456f6e35a18eeb Mon Sep 17 00:00:00 2001 From: Ivan Despot <66276597+g-despot@users.noreply.github.com> Date: Thu, 6 Aug 2026 15:57:10 +0200 Subject: [PATCH 2/2] fix(batch): derive has_errors from errors rather than per-call-site kwargs Setting has_errors=True at each construction site leaves the flag and the errors dict as two sources of truth, which is how they drifted apart in the first place: the stream receive loop was added without the kwarg. Derive the flag in __post_init__ on BatchObjectReturn and BatchReferenceReturn so an instance cannot be constructed claiming success while carrying errors, and drop the now-redundant kwargs. Extends the tests to BatchReferenceReturn, the async ingest path and batch.stream(). --- mock_tests/test_batch.py | 77 +++++++++++++++++++++++---- test/collection/test_batch.py | 66 ++++++++++++++++++++++- weaviate/collections/batch/async_.py | 2 - weaviate/collections/batch/sync.py | 2 - weaviate/collections/classes/batch.py | 6 +++ 5 files changed, 139 insertions(+), 14 deletions(-) diff --git a/mock_tests/test_batch.py b/mock_tests/test_batch.py index f8b114534..ad7357f04 100644 --- a/mock_tests/test_batch.py +++ b/mock_tests/test_batch.py @@ -1,7 +1,8 @@ -from typing import Generator +from typing import AsyncGenerator, Generator, List import grpc import pytest +import pytest_asyncio import weaviate from weaviate.proto.v1 import batch_pb2, weaviate_pb2_grpc from .conftest import MOCK_IP, MOCK_PORT, MOCK_PORT_GRPC, mock_class, HTTPServer @@ -63,6 +64,14 @@ def test_ssb_canceled_stream( class MockFailedObjectWeaviateService(weaviate_pb2_grpc.WeaviateServicer): + """Rejects every other object, starting with the first. + + A batch of 1 object gives 1 error and 0 uuids; a batch of 4 gives 2 errors and 2 uuids. + """ + + def __init__(self) -> None: + self.seen = 0 + def BatchStream( self, request_iterator: Generator[batch_pb2.BatchStreamRequest, None, None], @@ -71,16 +80,23 @@ def BatchStream( yield batch_pb2.BatchStreamReply(started=batch_pb2.BatchStreamReply.Started()) for request in request_iterator: if request.HasField("data"): - uuids = [obj.uuid for obj in request.data.objects.values] - yield batch_pb2.BatchStreamReply( - results=batch_pb2.BatchStreamReply.Results( - errors=[ + uuids: List[str] = [] + errors: List[batch_pb2.BatchStreamReply.Results.Error] = [] + successes: List[batch_pb2.BatchStreamReply.Results.Success] = [] + for obj in request.data.objects.values: + uuids.append(obj.uuid) + if self.seen % 2 == 0: + errors.append( batch_pb2.BatchStreamReply.Results.Error( - uuid=uuid, error="mock failure" + uuid=obj.uuid, error="mock failure" ) - for uuid in uuids - ] - ) + ) + else: + successes.append(batch_pb2.BatchStreamReply.Results.Success(uuid=obj.uuid)) + self.seen += 1 + yield batch_pb2.BatchStreamReply(acks=batch_pb2.BatchStreamReply.Acks(uuids=uuids)) + yield batch_pb2.BatchStreamReply( + results=batch_pb2.BatchStreamReply.Results(errors=errors, successes=successes) ) if request.HasField("stop"): return @@ -95,9 +111,52 @@ def failed_object_stream( return canceled_stream_client.collections.use(mock_class["class"]) +@pytest_asyncio.fixture +async def failed_object_stream_async( + weaviate_mock: HTTPServer, start_grpc_server: grpc.Server +) -> AsyncGenerator[weaviate.collections.CollectionAsync, None]: + weaviate_mock.expect_request(f"/v1/schema/{mock_class['class']}").respond_with_json(mock_class) + weaviate_pb2_grpc.add_WeaviateServicer_to_server( + MockFailedObjectWeaviateService(), start_grpc_server + ) + client = weaviate.use_async_with_local(port=MOCK_PORT, host=MOCK_IP, grpc_port=MOCK_PORT_GRPC) + await client.connect() + yield client.collections.use(mock_class["class"]) + await client.close() + + def test_ingest_has_errors_on_failed_object( failed_object_stream: weaviate.collections.Collection, ): result = failed_object_stream.data.ingest([{"name": "Object 1"}]) assert result.has_errors is True assert len(result.errors) == 1 + + +def test_ssb_ingest_reports_has_errors( + failed_object_stream: weaviate.collections.Collection, +) -> None: + result = failed_object_stream.data.ingest({"name": f"Object {i}"} for i in range(4)) + assert len(result.errors) == 2 + assert len(result.uuids) == 2 + assert result.has_errors + + +@pytest.mark.asyncio +async def test_ssb_ingest_reports_has_errors_async( + failed_object_stream_async: weaviate.collections.CollectionAsync, +) -> None: + result = await failed_object_stream_async.data.ingest({"name": f"Object {i}"} for i in range(4)) + assert len(result.errors) == 2 + assert len(result.uuids) == 2 + assert result.has_errors + + +def test_ssb_stream_reports_has_errors( + failed_object_stream: weaviate.collections.Collection, +) -> None: + with failed_object_stream.batch.stream() as batch: + for i in range(4): + batch.add_object({"name": f"Object {i}"}) + assert len(failed_object_stream.batch.failed_objects) == 2 + assert failed_object_stream.batch.results.objs.has_errors diff --git a/test/collection/test_batch.py b/test/collection/test_batch.py index 0a2cda954..3837317ab 100644 --- a/test/collection/test_batch.py +++ b/test/collection/test_batch.py @@ -3,10 +3,38 @@ import pytest from weaviate.collections.batch.grpc_batch import _validate_props -from weaviate.collections.classes.batch import MAX_STORED_RESULTS, BatchObjectReturn +from weaviate.collections.classes.batch import ( + MAX_STORED_RESULTS, + BatchObject, + BatchObjectReturn, + BatchReference, + BatchReferenceReturn, + ErrorObject, + ErrorReference, +) from weaviate.exceptions import WeaviateInsertInvalidPropertyError +def _error_object(index: int) -> ErrorObject: + return ErrorObject( + message="something went wrong", + object_=BatchObject(collection="Test", properties={"name": "test"}, index=index), + ) + + +def _error_reference(index: int) -> ErrorReference: + return ErrorReference( + message="something went wrong", + reference=BatchReference( + from_object_collection="Test", + from_object_uuid=uuid.uuid4(), + from_property_name="other", + to_object_uuid=uuid.uuid4(), + index=index, + ), + ) + + def test_batch_object_return_add() -> None: lhs_uuids = [uuid.uuid4() for _ in range(MAX_STORED_RESULTS)] lhs = BatchObjectReturn( @@ -36,6 +64,42 @@ def test_batch_object_return_add() -> None: } +def test_batch_object_return_has_errors_when_constructed_with_errors() -> None: + err = _error_object(0) + result = BatchObjectReturn(_all_responses=[err], errors={0: err}) + assert result.has_errors + + +def test_batch_object_return_add_sets_has_errors() -> None: + err = _error_object(1) + result = BatchObjectReturn() + result += BatchObjectReturn(_all_responses=[uuid.uuid4()], uuids={0: uuid.uuid4()}) + result += BatchObjectReturn(_all_responses=[err], errors={1: err}) + assert result.has_errors + assert len(result.errors) == 1 + + +def test_batch_object_return_has_no_errors_when_all_succeed() -> None: + uid = uuid.uuid4() + result = BatchObjectReturn() + result += BatchObjectReturn(_all_responses=[uid], uuids={0: uid}) + assert not result.has_errors + + +def test_batch_reference_return_has_errors_when_constructed_with_errors() -> None: + err = _error_reference(0) + result = BatchReferenceReturn(errors={0: err}) + assert result.has_errors + + +def test_batch_reference_return_add_sets_has_errors() -> None: + err = _error_reference(0) + result = BatchReferenceReturn() + result += BatchReferenceReturn(errors={0: err}) + assert result.has_errors + assert len(result.errors) == 1 + + def test_validate_props_raises_for_top_level_id() -> None: with pytest.raises(WeaviateInsertInvalidPropertyError): _validate_props({"id": "abc123"}) diff --git a/weaviate/collections/batch/async_.py b/weaviate/collections/batch/async_.py index 2e7aa9279..8b997586c 100644 --- a/weaviate/collections/batch/async_.py +++ b/weaviate/collections/batch/async_.py @@ -407,7 +407,6 @@ async def __recv(self) -> None: result_objs += BatchObjectReturn( _all_responses=[err], errors={cached.index: err}, - has_errors=True, ) failed_objs.append(err) logger.warning( @@ -429,7 +428,6 @@ async def __recv(self) -> None: ) result_refs += BatchReferenceReturn( errors={cached.index: err}, - has_errors=True, ) failed_refs.append(err) logger.warning( diff --git a/weaviate/collections/batch/sync.py b/weaviate/collections/batch/sync.py index cff631e0e..6627f8911 100644 --- a/weaviate/collections/batch/sync.py +++ b/weaviate/collections/batch/sync.py @@ -365,7 +365,6 @@ def __recv(self) -> None: result_objs += BatchObjectReturn( _all_responses=[err], errors={cached.index: err}, - has_errors=True, ) failed_objs.append(err) logger.warning( @@ -388,7 +387,6 @@ def __recv(self) -> None: failed_refs.append(err) result_refs += BatchReferenceReturn( errors={cached.index: err}, - has_errors=True, ) logger.warning( { diff --git a/weaviate/collections/classes/batch.py b/weaviate/collections/classes/batch.py index eb8d0181a..b90bebfad 100644 --- a/weaviate/collections/classes/batch.py +++ b/weaviate/collections/classes/batch.py @@ -216,6 +216,9 @@ class BatchObjectReturn: uuids: Dict[int, uuid_package.UUID] = field(default_factory=dict) has_errors: bool = False + def __post_init__(self) -> None: + self.has_errors = self.has_errors or len(self.errors) > 0 + @property def all_responses(self) -> List[Union[uuid_package.UUID, ErrorObject]]: """@deprecated: A list of all the responses from the batch operation. Each response is either a `uuid_package.UUID` object or an `Error` object. @@ -284,6 +287,9 @@ class BatchReferenceReturn: errors: Dict[int, ErrorReference] = field(default_factory=dict) has_errors: bool = False + def __post_init__(self) -> None: + self.has_errors = self.has_errors or len(self.errors) > 0 + def __add__(self, other: "BatchReferenceReturn") -> "BatchReferenceReturn": self.elapsed_seconds += other.elapsed_seconds prev_max = max(self.errors.keys()) if len(self.errors) > 0 else -1