Pubsub: Update default settings; add maximum total lease extension (#3920)

Author: chrisdunelm

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1.Tests/SubscriberClientTest.cs

@@ -650,6 +650,34 @@ public void LeaseExtension()
             }
         }
 
+        [Fact]
+        public void LeaseMaxExtension()
+        {
+            var msgs = new[] { new[] {
+                ServerAction.Data(TimeSpan.Zero, new[] { "1" }),
+                ServerAction.Inf()
+            } };
+            using (var fake = Fake.Create(msgs, ackDeadline: TimeSpan.FromSeconds(30), ackExtendWindow: TimeSpan.FromSeconds(10)))
+            {
+                fake.Scheduler.Run(async () =>
+                {
+                    var doneTask = fake.Subscriber.StartAsync(async (msg, ct) =>
+                    {
+                        // Emulate a hanging message-processing task.
+                        await fake.TaskHelper.ConfigureAwait(fake.Scheduler.Delay(TimeSpan.FromHours(24), ct));
+                        return SubscriberClient.Reply.Ack;
+                    });
+                    await fake.TaskHelper.ConfigureAwait(fake.Scheduler.Delay(TimeSpan.FromHours(12), CancellationToken.None));
+                    await fake.TaskHelper.ConfigureAwait(fake.Subscriber.StopAsync(CancellationToken.None));
+                    await fake.TaskHelper.ConfigureAwait(doneTask);
+                    Assert.Equal(1, fake.Subscribers.Count);
+                    // Check that the lease was extended for 60 minutes only.
+                    // +1 is due to initial lease extension at time=0
+                    Assert.Equal((int)SubscriberClient.DefaultMaxTotalAckExtension.TotalSeconds / 20 + 1, fake.Subscribers[0].Extends.Count);
+                });
+            }
+        }
+
         [Fact]
         public void SlowUplinkThrottlesPull()
         {
@@ -793,6 +821,12 @@ public void ValidParameters()
                 AckExtensionWindow = TimeSpan.FromTicks(SubscriberClient.DefaultAckDeadline.Ticks / 2)
             };
             new SubscriberClientImpl(subscriptionName, clients, settingsAckExtension2, null);
+
+            var settingsMaxExtension = new SubscriberClient.Settings
+            {
+                MaxTotalAckExtension = TimeSpan.FromMinutes(20)
+            };
+            new SubscriberClientImpl(subscriptionName, clients, settingsMaxExtension, null);
         }
 
         [Fact]
@@ -842,6 +876,13 @@ public void InvalidParameters()
             };
             var ex8 = Assert.Throws<ArgumentOutOfRangeException>(() => new SubscriberClientImpl(subscriptionName, clients, settingsBadAckExtension2, null));
             Assert.Equal("AckExtensionWindow", ex8.ParamName);
+
+            var settingsBadMaxExtension = new SubscriberClient.Settings
+            {
+                MaxTotalAckExtension = TimeSpan.FromMinutes(-20)
+            };
+            var ex9 = Assert.Throws<ArgumentOutOfRangeException>(() => new SubscriberClientImpl(subscriptionName, clients, settingsBadMaxExtension, null));
+            //Assert.Equal("MaxTotalAckExtension", ex9.ParamName); There's a bug in GaxPreconditions.CheckNonNegativeDelay() which uses the wrong paramName
         }
     }
 }

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/PublisherClient.cs

@@ -154,22 +154,20 @@ internal void Validate()
             }
         }
 
-        // All defaults taken from Java (reference) implementation.
-
         /// <summary>
         /// Default <see cref="BatchingSettings"/> for <see cref="PublisherClient"/>.
         /// Default values are:
         /// <see cref="BatchingSettings.ElementCountThreshold"/> = 100;
-        /// <see cref="BatchingSettings.ByteCountThreshold"/> = 10,000;
-        /// <see cref="BatchingSettings.DelayThreshold"/> = 1 millisecond;
+        /// <see cref="BatchingSettings.ByteCountThreshold"/> = 1,000,000;
+        /// <see cref="BatchingSettings.DelayThreshold"/> = 10 milliseconds;
         /// </summary>
-        public static BatchingSettings DefaultBatchingSettings { get; } = new BatchingSettings(100L, 10_000L, TimeSpan.FromMilliseconds(1));
+        public static BatchingSettings DefaultBatchingSettings { get; } = new BatchingSettings(100L, 1_000_000L, TimeSpan.FromMilliseconds(10));
 
         /// <summary>
         /// The absolute maximum <see cref="BatchingSettings"/> supported by the service.
         /// Maximum values are:
         /// <see cref="BatchingSettings.ElementCountThreshold"/> = 1,000;
-        /// <see cref="BatchingSettings.ByteCountThreshold"/> = 9,500,000;
+        /// <see cref="BatchingSettings.ByteCountThreshold"/> = 10,000,000;
         /// </summary>
         public static BatchingSettings ApiMaxBatchingSettings { get; } = new BatchingSettings(1000L, 10_000_000L, null);
 

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/SubscriberClient.cs

@@ -78,6 +78,7 @@ internal Settings(Settings other)
                 AckDeadline = other.AckDeadline;
                 AckExtensionWindow = other.AckExtensionWindow;
                 Scheduler = other.Scheduler;
+                MaxTotalAckExtension = other.MaxTotalAckExtension;
             }
 
             /// <summary>
@@ -100,6 +101,12 @@ internal Settings(Settings other)
             /// </summary>
             public TimeSpan? AckExtensionWindow { get; set; }
 
+            /// <summary>
+            /// Maximum duration for which a message ACK deadline will be extended.
+            /// If <c>null</c>, uses the default of <see cref="DefaultMaxTotalAckExtension"/>.
+            /// </summary>
+            public TimeSpan? MaxTotalAckExtension { get; set; }
+
             /// <summary>
             /// The <see cref="IScheduler"/> used to schedule delays.
             /// If <c>null</c>, the default <see cref="SystemScheduler"/> is used.
@@ -112,6 +119,10 @@ internal void Validate()
                 GaxPreconditions.CheckArgumentRange(AckDeadline, nameof(AckDeadline), MinimumAckDeadline, MaximumAckDeadline);
                 var maxAckExtension = TimeSpan.FromTicks((AckDeadline ?? DefaultAckDeadline).Ticks / 2);
                 GaxPreconditions.CheckArgumentRange(AckExtensionWindow, nameof(AckExtensionWindow), MinimumAckExtensionWindow, maxAckExtension);
+                if (MaxTotalAckExtension is TimeSpan maxTotalAckExtension)
+                {
+                    GaxPreconditions.CheckNonNegativeDelay(maxTotalAckExtension, nameof(MaxTotalAckExtension));
+                }
             }
 
             /// <summary>
@@ -181,10 +192,10 @@ internal void Validate()
 
         /// <summary>
         /// Default <see cref="FlowControlSettings"/> for <see cref="SubscriberClient"/>.
-        /// Allows 10,000 outstanding messages; and 20Mb outstanding bytes.
+        /// Allows 1,000 outstanding messages; and 100Mb outstanding bytes.
         /// </summary>
         /// <returns>Default <see cref="FlowControlSettings"/> for <see cref="SubscriberClient"/>.</returns>
-        public static FlowControlSettings DefaultFlowControlSettings { get; } = new FlowControlSettings(10_000, 20_000_000);
+        public static FlowControlSettings DefaultFlowControlSettings { get; } = new FlowControlSettings(1_000, 100_000_000);
 
         /// <summary>
         /// The service-defined minimum message ACKnowledgement deadline of 10 seconds.
@@ -207,10 +218,15 @@ internal void Validate()
         public static TimeSpan MinimumAckExtensionWindow { get; } = TimeSpan.FromMilliseconds(50);
 
         /// <summary>
-        /// The default message ACKnowlegdment extension window of 15 seconds.
+        /// The default message ACKnowledgement extension window of 15 seconds.
         /// </summary>
         public static TimeSpan DefaultAckExtensionWindow { get; } = TimeSpan.FromSeconds(15);
 
+        /// <summary>
+        /// The default maximum total ACKnowledgement extension of 60 minutes.
+        /// </summary>
+        public static TimeSpan DefaultMaxTotalAckExtension { get; } = TimeSpan.FromMinutes(60);
+
         /// <summary>
         /// Create a <see cref="SubscriberClient"/> instance associated with the specified <see cref="SubscriptionName"/>.
         /// The default <paramref name="settings"/> and <paramref name="clientCreationSettings"/> are suitable for machines with
@@ -352,6 +368,7 @@ internal SubscriberClientImpl(SubscriptionName subscriptionName, IEnumerable<Sub
             // These values are validated in Settings.Validate() above, so no need to re-validate here.
             _modifyDeadlineSeconds = (int)((settings.AckDeadline ?? DefaultAckDeadline).TotalSeconds);
             _autoExtendInterval = TimeSpan.FromSeconds(_modifyDeadlineSeconds) - (settings.AckExtensionWindow ?? DefaultAckExtensionWindow);
+            _maxExtensionDuration = settings.MaxTotalAckExtension ?? DefaultMaxTotalAckExtension;
             _shutdown = shutdown;
             _scheduler = settings.Scheduler ?? SystemScheduler.Instance;
             _taskHelper = GaxPreconditions.CheckNotNull(taskHelper, nameof(taskHelper));
@@ -363,6 +380,7 @@ internal SubscriberClientImpl(SubscriptionName subscriptionName, IEnumerable<Sub
         private readonly SubscriberServiceApiClient[] _clients;
         private readonly Func<Task> _shutdown;
         private readonly TimeSpan _autoExtendInterval; // Interval between message lease auto-extends
+        private readonly TimeSpan _maxExtensionDuration; // Maximum duration for which a message lease will be extended.
         private readonly int _modifyDeadlineSeconds; // Value to use as new deadline when lease auto-extends
         private readonly int _maxAckExtendQueue; // Maximum count of acks/extends to push to server in a single messages
         private readonly IScheduler _scheduler;
@@ -809,6 +827,7 @@ internal SingleChannel(SubscriberClientImpl subscriber,
                 _modifyDeadlineSeconds = subscriber._modifyDeadlineSeconds;
                 _maxAckExtendQueueSize = subscriber._maxAckExtendQueue;
                 _autoExtendInterval = subscriber._autoExtendInterval;
+                _maxExtensionDuration = subscriber._maxExtensionDuration;
                 _extendQueueThrottleInterval = TimeSpan.FromTicks((long)((TimeSpan.FromSeconds(_modifyDeadlineSeconds) - _autoExtendInterval).Ticks * 0.5));
                 _maxAckExtendSendCount = Math.Max(10, subscriber._maxAckExtendQueue / 4);
                 _maxConcurrentPush = 3; // Fairly arbitrary.
@@ -829,6 +848,7 @@ internal SingleChannel(SubscriberClientImpl subscriber,
             private readonly SubscriptionName _subscriptionName;
             private readonly int _modifyDeadlineSeconds; // Seconds to add to deadling on lease extension.
             private readonly TimeSpan _autoExtendInterval; // Delay between auto-extends.
+            private readonly TimeSpan _maxExtensionDuration; // Maximum duration for which a message lease will be extended.
             private readonly TimeSpan _extendQueueThrottleInterval; // Throttle pull if items in the extend queue are older than this.
             private readonly int _maxAckExtendQueueSize; // Soft limit on push queue sizes. Used to throttle pulls.
             private readonly int _maxAckExtendSendCount; // Maximum number of ids to include in an ack/nack/extend push RPC.
@@ -1046,7 +1066,7 @@ private void HandlePullMessageData(Task<bool> moveNextTask)
                     // Get all ack-ids, used to extend leases as required.
                     var msgIds = new HashSet<string>(msgs.Select(x => x.AckId));
                     // Send an initial "lease-extension"; which starts the server timer.
-                    HandleExtendLease(msgIds);
+                    HandleExtendLease(msgIds, null);
                     // Asynchonously start message processing. Handles flow, and calls the user-supplied message handler.
                     // Uses Task.Run(), so not to clog up this "master" thread with per-message processing.
                     Task messagesTask = _taskHelper.Run(() => ProcessPullMessagesAsync(msgs, msgIds));
@@ -1099,34 +1119,115 @@ await _taskHelper.ConfigureAwait(_flow.Process(msg.CalculateSize(), msg.Message.
                 }
             }
 
-            private void HandleExtendLease(HashSet<string> msgIds)
+            private class LeaseCancellation
+            {
+                public LeaseCancellation(CancellationTokenSource softStopCts) =>
+                    _cts = CancellationTokenSource.CreateLinkedTokenSource(softStopCts.Token);
+
+                private readonly object _lock = new object();
+                private CancellationTokenSource _cts;
+
+                public CancellationToken Token
+                {
+                    get
+                    {
+                        lock (_lock)
+                        {
+                            return _cts?.Token ?? CancellationToken.None;
+                        }
+                    }
+                }
+
+                public void Dispose()
+                {
+                    lock (_lock)
+                    {
+                        _cts.Dispose();
+                        _cts = null;
+                    }
+                }
+
+                public bool IsDisposed
+                {
+                    get
+                    {
+                        lock (_lock)
+                        {
+                            return _cts == null;
+                        }
+                    }
+                }
+
+                public void Cancel()
+                {
+                    CancellationTokenSource cts2;
+                    lock (_lock)
+                    {
+                        cts2 = _cts;
+                    }
+                    // Cancel outside of lock, as continuations may be executed synchronously.
+                    cts2?.Cancel();
+                    // No need to dispose of `_cts` here, as `Dispose()` will always be called.
+                }
+            }
+
+            private void HandleExtendLease(HashSet<string> msgIds, LeaseCancellation cancellation)
             {
                 if (_softStopCts.IsCancellationRequested)
                 {
                     // No further lease extensions once stop is requested.
                     return;
                 }
-                bool anyMsgIds;
-                lock (msgIds)
+                // The first call to this method happens as soon as messages in this chunk start to be processed.
+                // This triggers the server to start its lease timer.
+                if (cancellation == null)
                 {
-                    anyMsgIds = msgIds.Count > 0;
-                    if (anyMsgIds)
+                    // Create a task to cancel lease-extension once `_maxExtensionDuration` has been reached.
+                    // This set up once for each chunk of received messages, and passed through to each future call to this method.
+                    cancellation = new LeaseCancellation(_softStopCts);
+                    Add(_scheduler.Delay(_maxExtensionDuration, cancellation.Token), Next(false, () =>
                     {
-                        lock (_lock)
+                        // This is executed when `_maxExtensionDuration` has expired, or when `cancellation` is cancelled,
+                        // Which ensures `cancellation` is aways disposed of.
+                        cancellation.Dispose();
+                        lock (msgIds)
                         {
-                            _extendQueue.Enqueue(msgIds.Select(x => new TimedId(_extendThrottleHigh + 1, x)));
+                            msgIds.Clear();
                         }
-                    }
+                    }));
                 }
-                if (anyMsgIds)
+                if (!cancellation.IsDisposed)
                 {
-                    // Ids have been added to _extendQueue, so trigger a push.
-                    _eventPush.Set();
-                    // Some ids still exist, schedule another extension.
-                    Add(_scheduler.Delay(_autoExtendInterval, _softStopCts.Token), Next(false, () => HandleExtendLease(msgIds)));
-                    // Increment _extendThrottles.
-                    _extendThrottleHigh += 1;
-                    Add(_scheduler.Delay(_extendQueueThrottleInterval, _softStopCts.Token), Next(false, () => _extendThrottleLow += 1));
+                    // If `_maxExtensionDuration` has not expired, then schedule a further lease extension.
+                    bool anyMsgIds;
+                    lock (msgIds)
+                    {
+                        anyMsgIds = msgIds.Count > 0;
+                        if (anyMsgIds)
+                        {
+                            lock (_lock)
+                            {
+                                _extendQueue.Enqueue(msgIds.Select(x => new TimedId(_extendThrottleHigh + 1, x)));
+                            }
+                        }
+                    }
+                    if (anyMsgIds)
+                    {
+                        // Ids have been added to _extendQueue, so trigger a push.
+                        _eventPush.Set();
+                        // Some ids still exist, schedule another extension.
+                        // The overall `_maxExtensionDuration` is maintained by passing through the existing `cancellation`.
+                        Add(_scheduler.Delay(_autoExtendInterval, _softStopCts.Token), Next(false, () => HandleExtendLease(msgIds, cancellation)));
+                        // Increment _extendThrottles.
+                        _extendThrottleHigh += 1;
+                        Add(_scheduler.Delay(_extendQueueThrottleInterval, _softStopCts.Token), Next(false, () => _extendThrottleLow += 1));
+                    }
+                    else
+                    {
+                        // All messages have been handled in this chunk, so cancel the max-lease-time monitoring.
+                        // This will also cause `cancellation` to be disposed in the anonymous function above.
+                        cancellation.Cancel();
+                    }
                 }
             }