fix(firestore): BulkWriter pop from an empty deque (#17490)

In some situations, the BulkWriter would attempt to pop a retry config
from an empty deque. This PR adds a safeguard to end immediately if the
list is empty

Fixes #16138

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: daniel-sanche

packages/google-cloud-firestore/google/cloud/firestore_v1/bulk_writer.py

@@ -508,6 +508,8 @@ def _schedule_ready_retries(self) -> None:
         )
 
         for _ in range(take_until_index):
+            if not self._retries:
+                break
             retry: OperationRetry = self._retries.popleft()
             retry.retry(self)
         return None

packages/google-cloud-firestore/tests/unit/v1/test_bulk_writer.py

@@ -759,6 +759,46 @@ def test_basebulkwriter_update_raises_with_bad_option(self):
                 option=ExistsOption(exists=True),
             )
 
+    def test_basebulkwriter_reentrancy_empty_deque_retry(self):
+        # See https://github.com/googleapis/google-cloud-python/issues/16138
+        import bisect
+        import datetime
+
+        from google.cloud.firestore_v1.bulk_writer import (
+            BulkWriterCreateOperation,
+            OperationRetry,
+        )
+
+        client = self._make_client()
+        bw = _make_no_send_bulk_writer(client)
+
+        ref1 = _get_document_reference(client, id="doc1")
+        ref2 = _get_document_reference(client, id="doc2")
+        data = {"field": "value"}
+
+        # Put ref1 in current operations to trigger a document path collision
+        # and subsequent flush/enqueue when we create ref1 again.
+        bw.create(ref1, data)
+
+        run_at = datetime.datetime.now(tz=datetime.timezone.utc) - datetime.timedelta(
+            seconds=1
+        )
+
+        retry1 = OperationRetry(
+            operation=BulkWriterCreateOperation(reference=ref1, document_data=data),
+            run_at=run_at,
+        )
+        retry2 = OperationRetry(
+            operation=BulkWriterCreateOperation(reference=ref2, document_data=data),
+            run_at=run_at,
+        )
+
+        bisect.insort(bw._retries, retry1)
+        bisect.insort(bw._retries, retry2)
+
+        # This should not raise IndexError: pop from an empty deque due to re-entrant scheduling
+        bw._schedule_ready_retries()
+
 
 class TestSyncBulkWriter(_SyncClientMixin, _BaseBulkWriterTests):
     """All BulkWriters are opaquely async, but this one simulates a BulkWriter