Repository navigation
Batch: re-added rate-limited objects stay in results.objs.all_responses as errors #2179
Description
A rate-limited batch emits Dep020 in code that never reads all_responses, because the re-add in _BatchBase.__send_batch reads the deprecated property (weaviate/collections/batch/base.py:720). That line also filters by list position, while errors and uuids filter by the batch-wide index of the object. From the second chunk on the two differ, and the ErrorObject of the re-added object stays in _all_responses after its retry succeeds.
Mock gRPC server at 38f30c1 with the mock_tests/conftest.py fixtures, 4 objects through fixed_size(batch_size=2, concurrent_requests=1), the first object of the second chunk rate-limited once. The test never reads all_responses:
$ pytest mock_tests/test_probe_readd_dep020.py -q -s
Dep020 during batch=1 errors=0 len(_all_responses)=5 stale=1
mock_tests/test_probe_readd_dep020.py
import warnings
import weaviate
from weaviate.collections.classes.batch import ErrorObject
from weaviate.proto.v1 import batch_pb2, weaviate_pb2_grpc
from .conftest import MOCK_IP, MOCK_PORT, MOCK_PORT_GRPC, mock_class
class Svc(weaviate_pb2_grpc.WeaviateServicer):
calls = 0
def BatchObjects(self, request, context):
self.calls += 1
if self.calls == 2: # second chunk, once: rate-limit its first object
return batch_pb2.BatchObjectsReply(errors=[{"index": 0, "error": "failed with status: 503 error"}])
return batch_pb2.BatchObjectsReply()
def test_readd(weaviate_mock, start_grpc_server):
weaviate_mock.expect_request(f"/v1/schema/{mock_class['class']}").respond_with_json(mock_class)
weaviate_pb2_grpc.add_WeaviateServicer_to_server(Svc(), start_grpc_server)
client = weaviate.connect_to_local(port=MOCK_PORT, host=MOCK_IP, grpc_port=MOCK_PORT_GRPC)
try:
col = client.collections.use(mock_class["class"])
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
with col.batch.fixed_size(batch_size=2, concurrent_requests=1) as b:
for i in range(4):
b.add_object({"name": f"o{i}"})
r = col.batch.results.objs
dep = [x for x in w if "all_responses" in str(x.message)]
stale = [x for x in r._all_responses if isinstance(x, ErrorObject)]
print(f"\nDep020 during batch={len(dep)} errors={len(r.errors)} "
f"len(_all_responses)={len(r._all_responses)} stale={len(stale)}")
finally:
client.close()Expected: no warning and 4 entries. Reading _all_responses and filtering on err.object_.index fixes both. #2166 already edits this function.
@tsmith023, you added Dep020. Would you rather take that one-line fix, or remove all_responses now that its Q4 2024 date has passed?
Activity
fix(batch): filter re-added errors by uuid and avoid the deprecated a…
test(batch): re-added errors are removed and no Dep020 fires (weaviat…