fix(microservices): keep grpc response errors out of the request stream

[drakeo338]

PR Checklist

Please check if your PR fulfills the following requirements:

• The commit message follows our guidelines: https://github.com/nestjs/nest/blob/master/CONTRIBUTING.md
• Tests for the changes have been added (for bug fixes / features)
• Docs have been added / updated (for bug fixes / features)

PR Type

What kind of change does this PR introduce?

• Bugfix
• Feature
• Code style update (formatting, local variables)
• Refactoring (no functional changes, no api changes)
• Build related changes
• CI related changes
• Other... Please describe:

What is the current behavior?

Issue Number: #17971

In a gRPC streaming handler, when the server emits a response error on the call, the same call is the emitter the request stream listens on, so the request subject also receives the server's own response error.

What is the new behavior?

Fixes #17971. The server marks calls on which it is emitting a response error, and the request-side error listener ignores errors emitted that way. Genuine client stream errors are still forwarded. A regression test is added.

Does this PR introduce a breaking change?

• Yes
• No

Other information

Docs box left unticked: no docs change needed.

[micalevisk]

Thanks for the fix. I reproduced #17971 against a real @grpc/grpc-js bidi call: on master the request subscriber receives the response error (and when it has no error callback, rxjs rethrows it as an uncaught exception). On this branch it completes, and the client still gets INVALID_ARGUMENT. The trigger is #16401, first released in 11.1.15, which made the non-cancelled branch of this listener reachable.

The approach works: listeners run synchronously inside emit, so the marker covers exactly the server's own emission. The new regression spec fails without the change, and the unit and gRPC e2e suites pass.

Two test additions are needed before this goes in; details inline. Both snippets were run against this branch and against master. Happy to approve once they are in.

[micalevisk]

This call site is not covered. Putting call.emit('error', error) back here leaves the whole spec file green (74/74), so the bug can return for a response that errors while values are still waiting for drain. Please add a spec for it. This one passes on this branch and fails with this line reverted (it needs concat from rxjs):

it('should not error the request stream when the response errors after draining', async () => {
  const call = createCall();
  // Backpressure: the first write asks the server to wait for "drain"
  call.write.mockReturnValueOnce(false);
  const responseError = new Error('response failed');
  const requestErrors: unknown[] = [];
  const serverErrors: unknown[] = [];
  call.on('error', e => serverErrors.push(e));

  const handler = (stream: any) => {
    stream.subscribe({ error: (e: unknown) => requestErrors.push(e) });
    stream.drainBuffer?.();
    return concat(
      of('a', 'b'),
      throwError(() => responseError),
    );
  };

  const result = server.createRequestStreamMethod(handler, true)(
    call as any,
    vi.fn(),
  );
  await tick();
  // The error waits for the buffered value to be written
  expect(serverErrors).toEqual([]);
  call.emit('drain');
  await result;

  expect(call.write.mock.calls).toEqual([['a'], ['b']]);
  expect(serverErrors).toEqual([responseError]);
  expect(requestErrors).toEqual([]);
});

[micalevisk]

This drives a bare EventEmitter, so it shows that no error is forwarded but not what a real call does next: grpc-js emits cancelled right after the status is sent, which completes the request stream. Please also add an e2e case next to the other gRPC edge cases, as c70e76d and #17946 did. On master this fails with expected [ 'error' ] to deeply equal [ 'complete' ].

integration/microservices/src/grpc-edge/edge.proto

rpc EchoReject (stream Num) returns (stream Num);

integration/microservices/src/grpc-edge/edge.controller.ts (plus Subject from rxjs)

readonly echoRejectRequestEvents: string[] = [];

@GrpcStreamMethod('Edge')
echoReject(messages: Observable<unknown>) {
  const response = new Subject<unknown>();
  messages.subscribe({
    next: () =>
      response.error(
        new RpcException({
          code: GrpcStatus.INVALID_ARGUMENT,
          message: 'echo rejected',
        }),
      ),
    error: () => this.echoRejectRequestEvents.push('error'),
    complete: () => this.echoRejectRequestEvents.push('complete'),
  });
  return response;
}

integration/microservices/e2e/edge-cases-grpc.spec.ts

it('completes the request stream of a bidi call whose response errors', async () => {
  const call = client.echoReject({ deadline: deadline() });
  const err = await new Promise<any>(resolve => {
    call.on('error', resolve);
    call.on('data', () => undefined);
    call.write({ value: 1 });
  });

  expect(err.code).toBe(GRPC.status.INVALID_ARGUMENT);
  expect(err.details).toBe('echo rejected');
  expect(app.get(EdgeController).echoRejectRequestEvents).toEqual([
    'complete',
  ]);
});

[drakeo338]

Both tests are added as suggested: the unit spec fails with the drain call site reverted to call.emit('error', error), and the e2e case fails on master with expected [ 'error' ] to deeply equal [ 'complete' ].