feat(PubSub): Nack pending messages during shutdown.

Author: robertvoinescu-work

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/SubscriberClientImpl.SingleChannel.cs

@@ -496,12 +496,28 @@ Task ProcessSuccessfulMessages(List<ReceivedMessage> messages, HashSet<string> m
             }
         }
 
-        private async Task ProcessPullMessagesAsync(List<ReceivedMessage> msgs, HashSet<string> msgIds)
+        private async Task ProcessPullMessagesAsync(List<ReceivedMessage> msgs, HashSet<string> leaseTracking)
         {
             // Running async. Common data needs locking
             for (int msgIndex = 0; msgIndex < msgs.Count; msgIndex++)
             {
-                _softStopCts.Token.ThrowIfCancellationRequested();
+                if (_softStopCts.IsCancellationRequested)
+                {
+                    var remainingAckIds = msgs.Skip(msgIndex).Select(x => x.AckId).ToList();
+                    lock (_lock)
+                    {
+                        _nackQueue.Enqueue(remainingAckIds);
+                    }
+                    lock (leaseTracking)
+                    {
+                        leaseTracking.ExceptWith(remainingAckIds);
+                    }
+                    // Ids have been added to ack/nack-queue, so trigger a push.
+                    _eventPush.Set();
+                    _softStopCts.Token.ThrowIfCancellationRequested();
+                }
+
+
                 var msg = msgs[msgIndex];
                 msgs[msgIndex] = null;
                 // Prepare to call user message handler, _flow.Process(...) enforces the user-handler concurrency constraints.
@@ -510,19 +526,25 @@ await _taskHelper.ConfigureAwait(_flow.Process(msg.CalculateSize(), _messageOrde
                     // Running async. Common data needs locking
                     lock (_lock)
                     {
-                        _softStopCts.Token.ThrowIfCancellationRequested();
                         _userHandlerInFlight += 1;
                     }
                     if (msg.DeliveryAttempt > 0)
                     {
                         msg.Message.Attributes[DeliveryAttemptAttrKey] = msg.DeliveryAttempt.ToString(CultureInfo.InvariantCulture);
                     }
                     // Call user message handler
-                    var reply = await _taskHelper.ConfigureAwaitHideErrors(() => _handler.HandleMessage(msg.Message, _hardStopCts.Token), Reply.Nack);
+                    var reply = await _taskHelper.ConfigureAwaitHideErrors(() =>
+                    {
+                        // If the subscriber shut down while waiting for flow control, skip the handler.
+                        // Throwing here triggers a Nack, releasing the message for redelivery.
+                        _softStopCts.Token.ThrowIfCancellationRequested();
+                        return _handler.HandleMessage(msg.Message, _hardStopCts.Token);
+                    }, Reply.Nack);
+
                     // Lock msgsIds, this is accessed concurrently here and in HandleExtendLease().
-                    lock (msgIds)
+                    lock (leaseTracking)
                     {
-                        msgIds.Remove(msg.AckId);
+                        leaseTracking.Remove(msg.AckId);
                     }
                     // Lock ack/nack-queues, this is accessed concurrently here and in "master" thread.
                     lock (_lock)