fix(firestore): bubble up errors in BulkWriter (#14481)

This PR addresses an issue where the Firestore `BulkWriter` could
silently fail to persist documents without notifying the caller. This
behavior was primarily observed when the `BulkWriter`'s context was
canceled or when internal limits were reached.

#### Root Causes
1. **Dropped Batches on Cancellation:** In the background `send`
function, if the context was canceled, the current batch of jobs was
dropped immediately without notifying the individual job result channels
.
2. **Ignored Queuing Errors:** The `write` function and the retry logic
ignored errors returned by the underlying bundler when attempting to
queue a write .
3. **Ignored Rate Limit Errors:** Context errors occurring during the
rate limiter's wait period were not checked or propagated .

#### Changes
- Updated the internal `write` method and its public callers (`Create`,
`Delete`, `Set`, and `Update`) to return errors from the rate limiter
and the bundler.
- Modified the `send` function to iterate through and notify all jobs in
a batch with a context error if the transmission is aborted due to
cancellation.
- Added error checking to the retry logic within `send` to ensure
failures to re-queue a job are surfaced.

These changes ensure that any failure to queue or send a document is
explicitly reported through the `BulkWriterJob` results or as an
immediate return value, preventing data loss.

#11422.

---------

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

Author: bhshkh

firestore/bulkwriter.go

@@ -186,7 +186,10 @@ func (bw *BulkWriter) Create(doc *DocumentRef, datum interface{}) (*BulkWriterJo
 		return nil, fmt.Errorf("firestore: too many document writes sent to bulkwriter")
 	}
 
-	j := bw.write(w[0])
+	j, err := bw.write(w[0])
+	if err != nil {
+		return nil, err
+	}
 	return j, nil
 }
 
@@ -209,7 +212,10 @@ func (bw *BulkWriter) Delete(doc *DocumentRef, preconds ...Precondition) (*BulkW
 		return nil, fmt.Errorf("firestore: too many document writes sent to bulkwriter")
 	}
 
-	j := bw.write(w[0])
+	j, err := bw.write(w[0])
+	if err != nil {
+		return nil, err
+	}
 	return j, nil
 }
 
@@ -232,7 +238,10 @@ func (bw *BulkWriter) Set(doc *DocumentRef, datum interface{}, opts ...SetOption
 		return nil, fmt.Errorf("firestore: too many writes sent to bulkwriter")
 	}
 
-	j := bw.write(w[0])
+	j, err := bw.write(w[0])
+	if err != nil {
+		return nil, err
+	}
 	return j, nil
 }
 
@@ -255,7 +264,10 @@ func (bw *BulkWriter) Update(doc *DocumentRef, updates []Update, preconds ...Pre
 		return nil, fmt.Errorf("firestore: too many writes sent to bulkwriter")
 	}
 
-	j := bw.write(w[0])
+	j, err := bw.write(w[0])
+	if err != nil {
+		return nil, err
+	}
 	return j, nil
 }
 
@@ -284,20 +296,23 @@ func (bw *BulkWriter) checkWriteConditions(doc *DocumentRef) error {
 }
 
 // write packages up write requests into bulkWriterJob objects.
-func (bw *BulkWriter) write(w *pb.Write) *BulkWriterJob {
+func (bw *BulkWriter) write(w *pb.Write) (*BulkWriterJob, error) {
 
 	j := &BulkWriterJob{
 		resultChan: make(chan bulkWriterResult, 1),
 		write:      w,
 		ctx:        bw.ctx,
 	}
 
-	bw.limiter.Wait(bw.ctx)
-	// ignore operation size constraints and related errors; can't be inferred at compile time
-	// Bundler is set to accept an unlimited amount of bytes
-	_ = bw.bundler.Add(j, 0)
+	if err := bw.limiter.Wait(bw.ctx); err != nil {
+		return nil, err
+	}
+	err := bw.bundler.Add(j, 0)
+	if err != nil {
+		return nil, err
+	}
 
-	return j
+	return j, nil
 }
 
 // send transmits writes to the service and matches response results to job channels.
@@ -321,6 +336,9 @@ func (bw *BulkWriter) send(i interface{}) {
 
 	select {
 	case <-bw.ctx.Done():
+		for _, j := range bwj {
+			j.setError(bw.ctx.Err())
+		}
 		return
 	default:
 		resp, err := bw.vc.BatchWrite(bw.ctx, bwr)
@@ -342,9 +360,10 @@ func (bw *BulkWriter) send(i interface{}) {
 				// Do we need separate retry bundler?
 				_, isRetryable := batchWriteRetryCodes[codes.Code(s.Code)]
 				if j.attempts < maxRetryAttempts && isRetryable {
-					// ignore operation size constraints and related errors; job size can't be inferred at compile time
-					// Bundler is set to accept an unlimited amount of bytes
-					_ = bw.bundler.Add(j, 0)
+					err := bw.bundler.Add(j, 0)
+					if err != nil {
+						j.setError(fmt.Errorf("firestore: bulk write retry failed %w original error %v", err, status.Error(codes.Code(s.Code), s.Message)))
+					}
 				} else {
 					j.setError(status.Error(codes.Code(s.Code), s.Message))
 				}

firestore/bulkwriter_test.go

@@ -245,6 +245,15 @@ func TestBulkWriterErrors(t *testing.T) {
 				return b.Delete(c.Doc("C/b"))
 			},
 		},
+		{
+			name: "cannot write with cancelled context",
+			test: func(ignored *BulkWriter) (*BulkWriterJob, error) {
+				ctx, cancel := context.WithCancel(context.Background())
+				bw := c.BulkWriter(ctx)
+				cancel()
+				return bw.Delete(c.Doc("C/c"))
+			},
+		},
 	}
 
 	for _, tc := range tcs {
@@ -479,3 +488,28 @@ func TestBulkWriterConcurrent(t *testing.T) {
 		}
 	}
 }
+
+func TestBulkWriterSendCancelled(t *testing.T) {
+	c, _, cleanup := newMock(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	bw := c.BulkWriter(ctx)
+
+	j, err := bw.Create(c.Doc("C/a"), testData)
+	if err != nil {
+		t.Fatalf("Create failed: %v", err)
+	}
+
+	cancel() // Cancel context before Flush/send can complete successfully
+
+	bw.Flush()
+
+	_, err = j.Results()
+	if err == nil {
+		t.Fatal("wanted error, got nil")
+	}
+	if err != context.Canceled {
+		t.Errorf("want context.Canceled, got %v", err)
+	}
+}