[alkatrivedi]

batchWriteAtLeastOnce previously failed immediately on any stream error including transient UNAVAILABLE errors and treated streams that closed prematurely as complete. This PR introduce automatic retries and resumes interrupted BatchWrite streams for only the remaining unacknowledged mutation groups, while translating response indexes back to their original positions.

[gemini-code-assist]

Code Review

This pull request implements a resilient batch write stream mechanism (batchWriteStream) for Cloud Spanner, allowing batchWriteAtLeastOnce to transparently retry and resume on transient failures (such as UNAVAILABLE or premature stream termination) for only the unacknowledged mutation groups while correctly translating response indexes. It also updates metrics tracing and request stream coordination to support deferred operation completion. The feedback recommends avoiding the override of the private _destroy method on instantiated Node.js streams in both batch-write-stream.ts and database.ts, suggesting instead the use of standard public stream events like 'close' and 'error' for cleanup.

[olavloite]

Can we consider adding these tests as well:

• Trailing gRPC error after all mutation groups are acknowledged (remaining === 0)

  • Scenario: Attempt 1 streams responses acknowledging all mutation groups (0..N-1), but the underlying gRPC stream terminates with an error (e.g. UNAVAILABLE or connection reset) rather than a clean EOF / end.
  • Expected behavior: The stream should finalize with grpc.status.OK and emit end on userStream without error or attempting another retry (batch-write-stream.ts:L335-L340).

• Total timeout / deadline exceeded during retries (DeadlineError)

  • Scenario: An attempt fails with UNAVAILABLE, but elapsed + delayMs >= timeout (or elapsed time already exceeds timeout).
  • Expected behavior: The client should short-circuit the backoff timer, finalize with grpc.status.DEADLINE_EXCEEDED, and destroy userStream with a DeadlineError wrapping the underlying error (batch-write-stream.ts:L222-L227). Currently, no tests assert DeadlineError.

• Fatal error after a partial retry

  • Scenario: Attempt 1 acknowledges group 0, but fails with UNAVAILABLE before acknowledging group 1. Attempt 2 is issued with [group1], but fails with a non-retryable error (e.g. PERMISSION_DENIED).
  • Expected behavior: userStream emits the PERMISSION_DENIED error, final status is PERMISSION_DENIED, and no further retry is attempted. (Currently, every retry test in test/batch-write-stream.ts eventually succeeds; testing eventual failure after partial progress is missing).

• Exhaustion of maxRetries

  • Scenario: Configure backoffSettings.maxRetries: 2 and fail consecutive attempts with UNAVAILABLE.
  • Expected behavior: Retries cease after the 2nd retry attempt, and the user stream emits the final UNAVAILABLE error (batch-write-stream.ts:L216).

• Corrupted RetryInfo metadata decode fallback

  • Scenario: Unit test getNextRetryDelayMs with malformed binary payload in google.rpc.retryinfo-bin.
  • Expected behavior: getNextRetryDelayMs gracefully catches the decode exception and falls back to jittered exponential backoff rather than throwing an unhandled exception (batch-write-stream.ts:L73-L75).

[olavloite]

nit: add space with INTERNAL

[olavloite]

nit: This constant seems to be duplicated throughout the client in multiple places. Can we rather export it from one of the 'central' util files like src/proto.ts or src/common.tsx and import it here?

[olavloite]

nit: Can we move this function to common.ts? In follow-up PRs, we can update other places that implement similar functionality (but in a less correct way) to reuse this method.

[olavloite]

nit: I had to read over this a few times to understand why exactly we are doing this here. Maybe rewrite to this to make it easier to understand:

const allowsNetworkRetries = retryableCodes.includes(grpc.status.UNAVAILABLE);
return retryableCodes.includes(err.code) || (allowsNetworkRetries && isRetryableInternalError(err));

[olavloite]

Can we split this function into smaller parts? It is almost 300 lines long, and therefore hard to read.

[olavloite]

Could we for example extract this into a separate method?

[olavloite]

I think that this is only needed for a test case, and will never be used in real life, right? If so, can we get rid of this by modifying the test case to send a real object into the stream?

[olavloite]

I think that this is only used to handle specific test cases. Can we get rid of it?

[olavloite]

Can we try to split up this method also? It is relatively long and heavily indented, making it very hard to read and understand what is going on.

[olavloite]

I assume that deferOperationCompletion was added here for multi-attempt streaming RPCs like BatchWrite, we don't want RequestStreamCoordinator to prematurely finalize the operation metric when Attempt 1 fails with UNAVAILABLE, because the overall operation is still ongoing and will be retried.

However, adding deferOperationCompletion?: boolean to the generic RequestConfig feels like a leaky abstraction:

1. It leaks an internal coordination detail of the OpenTelemetry metrics subsystem into the generic gRPC request configuration interface (RequestConfig), even though it is only used in this one place.
2. In database.ts, it leads to a non-obvious side-effect dance: Attempt 1 passes metricsTracer: undefined, waits for requestStream to mutate config.metricsTracer in-place, and then captures it for Attempt 2+. The only reason Attempt 1 doesn't finalize the operation metric is because deferOperationCompletion: true is passed alongside metricsTracer: undefined.

Proposed Alternative: Ownership by Injection

We can drop deferOperationCompletion completely by establishing a clean ownership rule:

If the caller provides config.metricsTracer, the caller owns the operation lifecycle.

Instead of letting requestStream lazily create the tracer during Attempt 1:

1. database.ts initializes metricsTracer upfront at operation start (via this._getSpanner()._initMetricsTracer(...)) and passes it into config.metricsTracer for all attempts (1, 2, ... N).
2. In requestStream, we check if config.metricsTracer !== undefined. If so, we pass isExternalTracer = true to RequestStreamCoordinator.
3. In RequestStreamCoordinator.prototype.completeOperation:

completeOperation(): void {
  if (this.operationCompleted || this.isExternalTracer) {
    return;
  }
  this.operationCompleted = true;
  this.metricsTracer?.recordOperationCompletion();
}

This achieves the exact same multi-attempt behavior without needing deferOperationCompletion on RequestConfig, eliminates the in-place mutation of config, and establishes a clean pattern that streaming queries/reads (partial-result-stream.ts) can also adopt in the future when retrying with resume tokens.