pubsub: messages buffered on the stream are lost until their ack deadline when a MessageStream closes #7412

When a MessageStream shuts down (via its ShutdownToken, after a permanent error, or by being dropped), MessageStreamImpl::close() drops the gRPC StreamingPull stream. Any responses the server already sent, but that the client has not read from the stream yet, are discarded with it.

Those messages never reach lease management: they are not leased, not yielded to the application, and not nacked. The service treats them as delivered, so it only redelivers them once their ack deadline expires. With the default 60 second deadline, a message published around the time a subscriber shuts down can stall for a full minute before another subscriber receives it.

This happens with both ShutdownBehavior::NackImmediately and ShutdownBehavior::WaitForProcessing, since both close the stream the same way.

How we observed it

In an application's end-to-end tests against the Pub/Sub emulator, with per-message receipt logs: a message published while one subscriber was shutting down was never returned by MessageStream::next() in any process, and was then redelivered exactly one ack deadline (60 s) later. The application never saw the message, so no ack, nack, or handler drop was involved.

Expected behavior

Messages the client already holds in its stream buffers when it closes the stream should be nacked (ModifyAckDeadline with 0 seconds), so the service can redeliver them right away, the same as messages that were already under lease management at shutdown.

Notes

The Go and Java clients cancel the stream at shutdown without draining it either, so they have the same gap. Messages still in flight on the network when the stream closes can't be recovered by the client; this is only about responses already buffered on the client side.

I have a fix ready and will send a PR that links to this issue.