feat(pubsub): implement streaming keep-alive logic by torreypayne · Pull Request #34653 · googleapis/google-cloud-ruby

torreypayne · 2026-06-22T23:32:24Z

Overview

Implements proactive streaming keep-alive logic and connection health monitoring in Google::Cloud::PubSub::MessageListener::Stream, mirroring the design implemented in the .NET Pub/Sub client (dotnet#15649).

Long-running bi-directional gRPC streaming pull connections (StreamingPull) can experience silent TCP drops, intermediary network timeouts, or read deadlocks during periods of low message volume. This change introduces background timer tasks to push regular keep-alive requests and actively monitor server Pong timestamps.

Key Changes

Protocol Version Initialization: Explicitly initializes protocol_version = 1 on the initial StreamingPullRequest protobuf to enable bi-directional stream keep-alive support.
Unconditional Keep-Alive Pings: Configures a background timer task (@stream_keepalive_task) to dispatch empty StreamingPullRequest pings at regular intervals (default 30 seconds), regardless of current lease inventory volume.
Pong Monitoring & Automatic Reconnection: Introduces @pong_monitor_task to inspect timestamps (@last_ping_at, @last_pong_at). If a keep-alive response is overdue by more than pong_deadline seconds (default 15 seconds), the monitor raises RestartStream to safely recycle the connection and back off.
Concurrency Timestamp Guard: Guards ping timestamp updates (@last_ping_at = now if @last_pong_at >= @last_ping_at) to ensure consecutive un-ponged pings cannot overwrite the timestamp of an overdue request.

Testing & Validation

Unit Test Suite (keepalive_test.rb): Added targeted unit test coverage asserting protocol version flags, timer intervals, deadline timeouts, and non-disruptive Pong handling.
Resiliency & Robustness Suite: Validated against live GCP test instances (helical-zone-771) across simulated TCP socket hangs, sub-millisecond deadline starvation, and post-recovery downstream message delivery.

Fixes b/427319802

robertvoinescu-work

LGTM from a functional perspective. Just a few minor comments.

- Wrap unpause_streaming! in synchronize block and reset @last_pong_at to prevent false-positive stream restarts upon unblocking flow control. - Extract send_keepalive_ping! and check_liveness! helper methods on Stream and re-create TimerTask instances on start/stop to support clean restarts and deterministic unit testing. - Overhaul keepalive_test.rb with synchronous unit test coverage for timer helpers, unpause race condition, and TimerTask re-creation, while increasing CI polling timeouts to prevent flakiness.

…g real-time threads - Remove listener.start from synchronous unit tests (test_0003, test_0004, test_0005) so they execute on standalone Stream instances without spawning background threads or real-time TimerTask loops. - Isolate ENV['PUBSUB_TEST_PONG_DEADLINE'] = '0.05' from global before block so short 50ms timeouts only apply during asynchronous integration tests, eliminating flakiness on CI in slow build environments (Ubuntu Ruby 3.2 and Windows Ruby 4.0).

…talize comment

… concurrent keep-alive logic - Add concise reasoning comment for @pong_monitor_task execution interval (1/5th keepalive, floored at 10ms). - Set @stream_opened = false under synchronization inside backoff_and_wait! before condition wait to prevent @pong_monitor_task from interrupting retry delays and causing double-backoffs. - Restore multi-line comment above conditional in check_liveness! and sprinkle structured play-by-play comments across concurrent helper methods (send_keepalive_ping!, check_liveness!, unpause_streaming!). - Update keepalive_test.rb integration test to configure attr_accessors instead of removed test ENV variables.

…ing sentinel on stream restart - Enrich check_liveness! error logging with elapsed seconds since last pong, current inventory count, and flow control pause state. - Push closing sentinel to @request_queue before raising RestartStream to guarantee both write-side and read-side gRPC C-core streams unblock cleanly during TCP blackholes.

…guard in background_run - Explain why @stream_opened is set to false prior to calling streaming_pull (to temporarily disable liveness monitoring during gRPC handshake). - Document why ping/pong timestamps are re-initialized to monotonic now alongside setting @stream_opened = true after streaming_pull returns (to give the new stream a fresh deadline window and prevent false-positive disconnect detections from stale timestamps).

…ream-closing sentinel - Add descriptive comment above @request_queue&.push self on line 505 explaining how pushing self terminates EnumeratorQueue#each to cleanly send HTTP/2 END_STREAM. - Extract numeric keep-alive intervals, monitor polling ratios, reconnect backoff delays, and unpause thresholds into well-named constants on Google::Cloud::PubSub::MessageListener::Stream.

@quartzmo

…_wait! - Add unit test suggested by @quartzmo asserting @stream_opened is set to false inside backoff_and_wait! prior to @pause_cond.wait, guaranteeing the liveness checker cannot interrupt exponential backoff sleep.

…ream - Extract stream health monitoring, keep-alive timers, monotonic timestamps (@last_ping_at, @last_pong_at), and liveness checking into a cohesive collaborator class Google::Cloud::PubSub::MessageListener::KeepaliveMonitor. - Keep Stream focused entirely on primary message delivery, inventory lease renewal, and flow control. - Delegate keepalive_interval and pong_deadline accessors from Stream to KeepaliveMonitor for 100% backwards compatibility.

…eepaliveMonitor and Stream - Restore line-by-line attr_accessor and attr_reader declarations with descriptive comments on KeepaliveMonitor. - Restore original comments above keepalive_interval and pong_deadline forwarding accessors on Stream.

…onitor - Restore comment blocks documenting DEFAULT_INTERVAL, DEFAULT_DEADLINE, MONITOR_DIVISOR, and MIN_MONITOR_INTERVAL. - Restore concurrency and flow control rationale comments inside send_ping! and check_liveness!.

- Validate happy path effectiveness: continuous multi-interval ping/pong cycling and unpause timestamp atomicity. - Validate resiliency & failure recovery: silent network black hole recovery via stream-closing sentinel + RestartStream, and backoff sleep isolation during transient GRPC::Unavailable errors.

…flag - Allow standard 'bundle exec rake test' commands to automatically enable SimpleCov code coverage when COVERAGE=true is passed.

…r and Stream - Add acceptance/pubsub/keepalive_acceptance_test.rb running against real Google Cloud Pub/Sub live production instances. - Validate real gRPC wire keep-alive ping/pong heartbeat, live message delivery/ACK, and flow-control unpause resiliency.

…est/helper.rb - Remove SimpleCov environment trigger from test/helper.rb so coverage configuration can be handled in a separate PR.

aandreassa

Implementation looks great, just a couple of Ruby things & docs to think about!

aandreassa · 2026-07-15T18:31:06Z

+    listener.wait!
+
+    elapsed = attempts[1] - attempts[0]
+    puts "\n[b/528401453 Test] Elapsed delay between attempts: #{elapsed.round(3)}s"


Can we remove the outputs in this file (or capture it if needed)? It is better to prevent this test from polluting relevant output logs.

aandreassa · 2026-07-15T18:44:36Z

+          end
+
+          # keepalive_interval used to send pings and maintain a healthy stream
+          def keepalive_interval


Do we need these? Maybe we should keep it minimal and consistent.
You are only using it on tests, and you already call stream.keepalive_monitor.last_ping_at, for example.

aandreassa · 2026-07-15T18:55:34Z

+          MIN_MONITOR_INTERVAL = 0.01
+
+          # keepalive_interval used to send pings and maintain a healthy stream
+          attr_accessor :interval


Can we pls have YARD docstrings for these attr_accessors?

rg -U '^\s*#[^\n]*\n\s*attr_accessor' for examples.

Alternatively, do we need these if they are for testing only? After modernizing in 3.0, we try to keep this library pretty lightweight.

Take a look and LMK if I should do another pass

aandreassa · 2026-07-15T19:01:52Z

            self
          end

+          def stream_opened?


I find this confusing. stream_opened as in opened in the past and its done, or currently open? You should try using present tense for predicates.

updated to present tense (stream_open); should I add is_... to make it more verb-y?

aandreassa · 2026-07-15T19:12:41Z

+      class MessageListener
+        ##
+        # @private
+        # Monitors gRPC streaming connection health via periodic bi-directional keep-alive pings and pongs.


Can you please document what this class does at a high level (here or initialize)? I'd love to see more YARD docstrings overall.

aandreassa · 2026-07-15T19:24:27Z

+          # background timer task for monitoring stream liveness
+          attr_reader :monitor_task
+
+          def initialize stream, interval: DEFAULT_INTERVAL, deadline: DEFAULT_DEADLINE


You could do something like (pls double check convention)

# Initializes the KeepaliveMonitor. # # @param [Stream] stream The parent streaming pull connection to monitor. # @param [Float] interval The interval in seconds between keep-alive pings. # @param [Float] deadline The deadline in seconds for receiving a pong.

and document others in-line below.

Implemented your suggestion as-is; LMK what you think

torreypayne · 2026-07-16T17:55:15Z

        # Monitors gRPC streaming connection health via periodic bi-directional keep-alive pings and pongs.
+        #
+        # Tracks ping and pong timestamps over the active gRPC stream to detect silent connection drops or freezes,
+        # and triggers a stream restart when server responses exceed the configured pong deadline.


@aandreassa LMK if this additional context is useful/sufficient to address your comment on L23

@Private

…tream - Rename stream_opened? / @stream_opened to present tense stream_open? / @stream_open - Remove redundant forwarding accessors (keepalive_interval, pong_deadline, send_keepalive_ping!, and check_liveness!) from Stream - Add comprehensive YARD docstrings (with @Private for internal accessors) for KeepaliveMonitor and Stream - Extract PROTOCOL_VERSION = 1 into a formal Stream constant - Remove diagnostic puts outputs from bug_regression_test.rb

torreypayne force-pushed the pubsub-streaming-keepalive branch 2 times, most recently from cf5df9b to b1acc8a Compare June 23, 2026 04:52

torreypayne marked this pull request as ready for review June 23, 2026 15:54

torreypayne requested review from a team and yoshi-approver as code owners June 23, 2026 15:54

torreypayne requested review from amanda-tarafa and robertvoinescu-work June 23, 2026 16:26

torreypayne force-pushed the pubsub-streaming-keepalive branch 2 times, most recently from 94e9f14 to ad59cd2 Compare June 23, 2026 17:29

robertvoinescu-work reviewed Jun 24, 2026

View reviewed changes

Comment thread google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb Outdated

Comment thread google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb Outdated

torreypayne force-pushed the pubsub-streaming-keepalive branch 2 times, most recently from ca925d3 to e53a9bd Compare June 26, 2026 20:41

This comment was marked as resolved.

Sign in to view

torreypayne force-pushed the pubsub-streaming-keepalive branch from 26bcbb1 to 24055fd Compare June 30, 2026 15:53

torreypayne added kokoro:run Add this label to force Kokoro to re-run the tests. kokoro:force-run Add this label to force Kokoro to re-run the tests. labels Jun 30, 2026

yoshi-kokoro removed kokoro:run Add this label to force Kokoro to re-run the tests. kokoro:force-run Add this label to force Kokoro to re-run the tests. labels Jun 30, 2026

This comment was marked as resolved.

Sign in to view

aandreassa requested changes Jul 1, 2026

View reviewed changes

quartzmo requested changes Jul 1, 2026

View reviewed changes

Comment thread google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb Outdated

Comment thread google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb

yoshi-automation and others added 6 commits July 6, 2026 16:16

feat(pubsub): implement streaming keep-alive logic

f46e80d

fix(pubsub): remove test-only env vars, remove buganizer IDs and capi…

a7165c9

…talize comment

capitalize comment

0ad78cb

torreypayne force-pushed the pubsub-streaming-keepalive branch from 009d0ec to 9461a56 Compare July 6, 2026 23:17

torreypayne force-pushed the pubsub-streaming-keepalive branch from 2fa07a4 to cca4878 Compare July 7, 2026 15:36

torreypayne requested review from aandreassa and quartzmo July 7, 2026 15:41

torreypayne added 3 commits July 7, 2026 16:58

quartzmo approved these changes Jul 7, 2026

View reviewed changes

torreypayne commented Jul 7, 2026

View reviewed changes

Comment thread google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb Outdated

torreypayne added 2 commits July 8, 2026 16:31

This comment was marked as resolved.

Sign in to view

quartzmo approved these changes Jul 8, 2026

View reviewed changes

torreypayne added 4 commits July 9, 2026 20:09

test(pubsub): enable SimpleCov startup via COVERAGE=true environment …

431c646

…flag - Allow standard 'bundle exec rake test' commands to automatically enable SimpleCov code coverage when COVERAGE=true is passed.

test(pubsub): revert SimpleCov COVERAGE=true environment trigger in t…

f0e26a6

…est/helper.rb - Remove SimpleCov environment trigger from test/helper.rb so coverage configuration can be handled in a separate PR.

aandreassa requested changes Jul 15, 2026

View reviewed changes

torreypayne commented Jul 16, 2026

View reviewed changes

torreypayne force-pushed the pubsub-streaming-keepalive branch from 9ee8d60 to 43562dd Compare July 16, 2026 18:00

torreypayne force-pushed the pubsub-streaming-keepalive branch from 43562dd to f4927d5 Compare July 16, 2026 18:43

Uh oh!

Conversation

torreypayne commented Jun 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Overview

Key Changes

Testing & Validation

Uh oh!

robertvoinescu-work left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

This comment was marked as resolved.

This comment was marked as resolved.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

This comment was marked as resolved.

Uh oh!

aandreassa 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

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!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

torreypayne Jul 16, 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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

torreypayne commented Jun 22, 2026 •

edited

Loading

torreypayne Jul 16, 2026 •

edited

Loading