grpc-js: write callback never called when a streaming call ends while the write is held by flow control #3114

const grpc = require('@grpc/grpc-js');
const id = (x) => x;
const path = '/probe/upload';
const server = new grpc.Server();
server.addService(
  { upload: { path, requestStream: true, responseStream: false, requestSerialize: id, requestDeserialize: id, responseSerialize: id, responseDeserialize: id } },
  { upload(call) { call.sendMetadata(new grpc.Metadata()); /* never reads, so flow control fills up */ } },
);
server.bindAsync('127.0.0.1:0', grpc.ServerCredentials.createInsecure(), (err, port) => {
  const client = new grpc.Client(`127.0.0.1:${port}`, grpc.credentials.createInsecure());
  const call = client.makeClientStreamRequest(path, id, id, () => {});
  call.on('error', (e) => console.log('error event:', e.code));
  call.on('status', (s) => console.log('status event:', s.code));
  call.once('metadata', async () => {
    // Write until one write's callback stays pending under backpressure.
    for (let i = 0; ; i++) {
      const done = await Promise.race([
        new Promise((resolve) => call.write(Buffer.alloc(65536), resolve)).then(() => true),
        new Promise((resolve) => setTimeout(resolve, 200, false)),
      ]);
      if (!done) { console.log(`write #${i} is pending; cancelling`); break; }
    }
    const pending = new Promise((resolve) => call.write(Buffer.alloc(1), (e) => resolve(e ? e.message : 'ok')));
    call.cancel();
    const result = await Promise.race([pending, new Promise((r) => setTimeout(r, 3000, 'still pending after 3 s'))]);
    console.log('write callback:', result);
    process.exit(0);
  });
});