feat(PubSub): Streaming pull keepalive by robertvoinescu-work · Pull Request #15649 · googleapis/google-cloud-dotnet

robertvoinescu-work · 2026-06-04T22:55:22Z

b/427319804

gemini-code-assist

Code Review

This pull request introduces a protocol-compliant keep-alive ping/pong mechanism for the streaming pull client, including a timeout-based stream restart when the server becomes unresponsive. It also refactors stream disposal and updates the test suite to validate these behaviors. The review feedback highlights critical issues with the implementation: first, using a 15-second timeout for MoveNext will cause constant stream restarts during idle periods, so the timeout should be increased to 45 seconds (s_streamPingPeriod + s_streamPongPeriod); second, disposing the CancellationTokenSource while MoveNext is still active can lead to an ObjectDisposedException or race conditions; and third, the StreamTimeoutRestarts test must be updated to wait longer than the new 45-second timeout to pass successfully.

amanda-tarafa

Yep, generally good. Thanks.

amanda-tarafa · 2026-06-05T16:57:41Z

The breaking change check is a false positive, already being tracked in b/520002639. I've added the allow breaking changes label to bypass.

amanda-tarafa

Looking good, but I have a few comments.

Also, I think it'd be good to encapsulate all the ping/pong checks and times etc. on the _retryState object, so that we can use it the same or similar to how we use it for exceptions.

robertvoinescu-work · 2026-06-08T19:24:22Z

I've added refactored things a bit differently than moving things onto the RetryState, mostly because it didn't feel cohesive to include ping, pongs tracking and handling on RetryState which just handles exceptions. Instead I've created a KeepAliveMonitor to encapsulate the new logic and have this throw a DeadlineExceeded exception if we timeout. The exception will then be handled in the normal logical flow.

amanda-tarafa

Happy to chat about these.

amanda-tarafa

A little bit of pseudocode, but see if that seems like a good balance between encapsulation and knowing what's happening just by looking at the handlers.

amanda-tarafa

Looking great. Still a few changes but nothing major now.

amanda-tarafa · 2026-06-09T20:13:21Z

            }
        }

        // Message-stream has messages (or not, depending on moveNextResult)
        private void HandlePullMessageData(Task<bool> moveNextTask)
        {
+            if (!moveNextTask.IsCompleted)


We need to record the failure in _retryState as that's what increments the backoff for restarting the pull. That's why I had the idea to include what you've put in KeepAliveMonitor in RetryState instead because the lack of pong is a effectively a retriable failure.

I think we can still merge those two, and maybe call it RestartMonitor or something. But we can do that in a follow up PR, where we first clean up RetryState and then merge it with KeepAliveMonitor into RestartMonitor.

Sounds like a plan will resolve in a follow-up.

We need to record the failure in _retryState as that's what increments the backoff for restarting the pull.

This part we need to solve on this PR, otherwise we'll be retrying sooner than with the backoff and that might make the problem worse.

What we can do in a follow up is all the refactoring.

amanda-tarafa · 2026-06-09T20:17:09Z

+                _pongReceived = false;
+            }
+
+            internal void RecordPong() => _pongReceived = true;


You still want to use Interlock here and on RecordPing. The pong is recorded when the MoveNext completes and that happens in the background.

Good point - I'd confused myself after all the refactoring, done for both.

amanda-tarafa · 2026-06-09T20:21:44Z

+            public bool DisableKeepAlive { get; private set; }
+            public ServerAction WithDisableKeepAlive() { DisableKeepAlive = true; return this; }


It makes no sense to have the With to return the same instance. Either you make the type mutable, i.e. keep the DisableKeepAlive set public, or you don't, meaning you return a new instance on this method. But this is just a detour from using properties idioamatically.

Thanks for the pointer, I've reverted this change entirely based on refactor.

amanda-tarafa

These are good tests, thanks.

A couple of things, but looks great.

amanda-tarafa · 2026-06-11T05:24:50Z

            }
        }

        // Message-stream has messages (or not, depending on moveNextResult)
        private void HandlePullMessageData(Task<bool> moveNextTask)
        {
+            if (!moveNextTask.IsCompleted)


We need to record the failure in _retryState as that's what increments the backoff for restarting the pull.

This part we need to solve on this PR, otherwise we'll be retrying sooner than with the backoff and that might make the problem worse.

What we can do in a follow up is all the refactoring.

amanda-tarafa · 2026-06-11T05:29:46Z

Is it OK to change these properties based on the values in a pong? I'm guessing yes, because the next move next with real messages will have the right values, but just double checking, what do you think?

Thanks for pointing this out. There is a case where we HandleAckResponse and for that we will have incorrect _exactlyonceDelieveryEnabled field set. Fixed.

amanda-tarafa · 2026-06-11T05:37:03Z

+            /// <summary>
+            /// For testing only. Disables stream timeout enforcement when no messages are being received.
+            /// </summary>
+            internal bool DisableKeepAliveTimeout { get; set; }


Let's make this one read-only and pass the value in the constructor. And if my recommendation of configuring the pong period seems good, you'll have to rename it and change the type, of course.

Really good point, thanks Amanda. I've updated the code.

amanda-tarafa · 2026-06-11T05:41:08Z

+            internal void RecordPing()
+            {
+                long nowTicks = _clock.GetCurrentDateTimeUtc().Ticks;
+                Interlocked.Exchange(ref _lastPingTicks, nowTicks);


I think you only need interlock for the pong.

Yes you are right, the ping is grabbed synchronously on the main processing thread. Removed the interlocked.

amanda-tarafa · 2026-06-11T05:45:02Z

+    /// <summary>
+    /// For testing only. Disables stream timeout enforcement when no messages are being received.
+    /// </summary>
+    internal bool DisableKeepAliveTimeout { get; set; }


Let's keep this one read only as well, and receive it on the internal constructor, we can use that one from the tests if we have to.

And can we just configure the pong timeout instead, for tests we set to inifnity (or whatever that translates to in TimeSpan) and we don't have to add ifs on production code. If that works, rename this as PongPeriod or something like that.

That's a nice way to avoid ifs in production code 👍 . I've updated accordingly

amanda-tarafa · 2026-06-11T05:50:42Z

+                        return Task.FromResult(SubscriberClient.Reply.Ack);
+                    });
+                    // Wait for more than the timeout.
+                    // With dynamic timeout, the first stream starts at T=0.


nit: What does the word dynamic mean here?

I meant to convey that the timeout on the first was at 45 seconds instead of 30 seconds. I just removed it I think it's probably clearer without.

amanda-tarafa

LGTM!

Make sure to run all integration tests locally before merging.

product-auto-label Bot added the api: pubsub Issues related to the Pub/Sub API. label Jun 4, 2026

robertvoinescu-work force-pushed the feat/keepAlive branch from cf97acc to 8332d19 Compare June 4, 2026 22:56

robertvoinescu-work changed the title ~~feat(PubSub): Add handling for streams that silently die.~~ feat(PubSub): Streaming pull keepalive Jun 4, 2026

gemini-code-assist Bot reviewed Jun 4, 2026

View reviewed changes

amanda-tarafa reviewed Jun 4, 2026

View reviewed changes

Comment thread apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/SubscriberClientImpl.SingleChannel.cs Outdated

Comment thread apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/SubscriberClientImpl.SingleChannel.cs Outdated

robertvoinescu-work force-pushed the feat/keepAlive branch from 8332d19 to 042ee2a Compare June 5, 2026 16:37

robertvoinescu-work marked this pull request as ready for review June 5, 2026 16:53

robertvoinescu-work requested a review from a team as a code owner June 5, 2026 16:53

robertvoinescu-work requested a review from amanda-tarafa June 5, 2026 16:53

amanda-tarafa added the allow breaking changes Allows breaking changes to be merged. Use with care! label Jun 5, 2026

robertvoinescu-work force-pushed the feat/keepAlive branch from 042ee2a to 52bb6f5 Compare June 5, 2026 16:57

amanda-tarafa requested changes Jun 5, 2026

View reviewed changes

robertvoinescu-work force-pushed the feat/keepAlive branch from 52bb6f5 to 74d3477 Compare June 8, 2026 19:17

amanda-tarafa requested changes Jun 8, 2026

View reviewed changes

robertvoinescu-work force-pushed the feat/keepAlive branch from 74d3477 to baa98a3 Compare June 9, 2026 15:59

amanda-tarafa reviewed Jun 9, 2026

View reviewed changes

Comment thread apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/SubscriberClientImpl.SingleChannel.cs Outdated

robertvoinescu-work force-pushed the feat/keepAlive branch 2 times, most recently from 77d6c51 to abf1ebc Compare June 9, 2026 17:30

amanda-tarafa requested changes Jun 9, 2026

View reviewed changes

robertvoinescu-work force-pushed the feat/keepAlive branch 3 times, most recently from 7739ec0 to d3110ee Compare June 10, 2026 19:17

robertvoinescu-work requested a review from amanda-tarafa June 10, 2026 19:20

amanda-tarafa requested changes Jun 11, 2026

View reviewed changes

robertvoinescu-work force-pushed the feat/keepAlive branch from d3110ee to 7d4c98d Compare June 11, 2026 20:08

feat(PubSub): Streaming pull keepalive

974ea8e

robertvoinescu-work force-pushed the feat/keepAlive branch from 7d4c98d to 974ea8e Compare June 11, 2026 20:44

amanda-tarafa approved these changes Jun 11, 2026

View reviewed changes

		public bool DisableKeepAlive { get; private set; }
		public ServerAction WithDisableKeepAlive() { DisableKeepAlive = true; return this; }

Conversation

robertvoinescu-work commented Jun 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

Uh oh!

Uh oh!

Uh oh!

amanda-tarafa left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

amanda-tarafa commented Jun 5, 2026

Uh oh!

amanda-tarafa left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

robertvoinescu-work commented Jun 8, 2026

Uh oh!

amanda-tarafa left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

amanda-tarafa left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

amanda-tarafa left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

robertvoinescu-work Jun 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

amanda-tarafa left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

robertvoinescu-work commented Jun 4, 2026 •

edited

Loading

robertvoinescu-work Jun 9, 2026 •

edited

Loading