fix(batch): require both counts unchanged before the idle flush

The wait loop in the batching threads breaks out early when a second has
passed and nothing new arrived, but the guard ORs the two "unchanged"
checks:

    if time.time() - start >= 1 and (
        len_o == len(self.__batch_objects) or len_r == len(self.__batch_references)
    ):

In an objects-only batch -- what `add_object` without `add_reference`
produces -- `len_r` and `len(self.__batch_references)` are both always 0,
so the right operand is always True and the whole guard collapses to
`if time.time() - start >= 1`. The loop then stops waiting on the first
tick past one second no matter how fast objects are still streaming in,
so the batch is flushed well short of its configured size. A
references-only batch has the mirror-image problem.

The comment above the break says "no new objects were added in the last
second", i.e. the batch is idle -- which is true only when *both* counts
are unchanged. Switch `or` to `and` at the three places that share this
loop shape (sync.py, async_.py, base.py).

Fixes #2105

Author: Anai-Guo

weaviate/collections/batch/async_.py

@@ -225,7 +225,7 @@ async def __loop(self) -> None:
                     # wait for more objects to be added up to the batch size
                     await asyncio.sleep(refresh_time)
                     if time.time() - start >= 1 and (
-                        len_o == len(self.__batch_objects) or len_r == len(self.__batch_references)
+                        len_o == len(self.__batch_objects) and len_r == len(self.__batch_references)
                     ):
                         # no new objects were added in the last second, exit the loop
                         break

weaviate/collections/batch/base.py

@@ -434,7 +434,7 @@ def __batch_send(self) -> None:
                         # shutdown was requested, exit the loop
                         break
                     if time.time() - start >= 1 and (
-                        len_o == len(self.__batch_objects) or len_r == len(self.__batch_references)
+                        len_o == len(self.__batch_objects) and len_r == len(self.__batch_references)
                     ):
                         # no new objects were added in the last second, exit the loop
                         break

weaviate/collections/batch/sync.py

@@ -177,7 +177,7 @@ def __loop(self) -> None:
                     # wait for more objects to be added up to the batch size
                     time.sleep(refresh_time)
                     if time.time() - start >= 1 and (
-                        len_o == len(self.__batch_objects) or len_r == len(self.__batch_references)
+                        len_o == len(self.__batch_objects) and len_r == len(self.__batch_references)
                     ):
                         # no new objects were added in the last second, exit the loop
                         break