From f46e80dd4659d4d04379b8afc40c98e36234e07c Mon Sep 17 00:00:00 2001
From: Yoshi Automation Bot <yoshi-automation@google.com>
Date: Mon, 22 Jun 2026 23:30:07 +0000
Subject: [PATCH 01/13] feat(pubsub): implement streaming keep-alive logic

---
 .../cloud/pubsub/message_listener/stream.rb   |  84 +++++++++-
 .../message_listener/acknowledge_test.rb      |   6 +-
 .../message_listener/bug_regression_test.rb   |  73 +++++++++
 .../pubsub/message_listener/inventory_test.rb |   6 +-
 .../pubsub/message_listener/keepalive_test.rb | 146 ++++++++++++++++++
 .../modify_ack_deadline_test.rb               |   6 +-
 .../pubsub/message_listener/nack_test.rb      |   6 +-
 7 files changed, 311 insertions(+), 16 deletions(-)
 create mode 100644 google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb
 create mode 100644 google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb

diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
index fa563dd17f57..61cb481d051c 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
@@ -48,6 +48,7 @@ class Stream
           ##
           # @private exactly_once_delivery_enabled.
           attr_reader :exactly_once_delivery_enabled
+          attr_accessor :keepalive_interval, :pong_deadline
 
           ##
           # @private Create an empty Subscriber::Stream object.
@@ -68,17 +69,49 @@ def initialize subscriber
 
             @callback_thread_pool = Concurrent::ThreadPoolExecutor.new max_threads: @subscriber.callback_threads
 
+            @keepalive_interval = Float(ENV["PUBSUB_TEST_KEEPALIVE_INTERVAL"] || 30)
+            @pong_deadline = Float(ENV["PUBSUB_TEST_PONG_DEADLINE"] || 15)
+            @last_ping_at = nil
+            @last_pong_at = nil
+            @stream_opened = false
+            @reconnect_delay = nil
+
             @stream_keepalive_task = Concurrent::TimerTask.new(
-              execution_interval: 30
+              execution_interval: @keepalive_interval
             ) do
-              # push empty request every 30 seconds to keep stream alive
-              unless inventory.empty?
-                subscriber.service.logger.log :info, "subscriber-streams" do
-                  "sending keepAlive to stream for subscription #{@subscriber.subscription_name}"
+              synchronize do
+                # @request_queue feeds client requests (initial pull request and keep-alive pings) into gRPC.
+                # Note: ACKs are sent via unary RPCs (TimedUnaryBuffer), not over this stream.
+                # Check that @request_queue is initialized (not nil) before pushing unconditional keep-alive pings.
+                if @stream_opened && !@stopped && @request_queue
+                  subscriber.service.logger.log :info, "subscriber-streams" do
+                    "sending keepAlive to stream for subscription #{@subscriber.subscription_name}"
+                  end
+                  @last_ping_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) if @last_pong_at >= @last_ping_at
+                  push Google::Cloud::PubSub::V1::StreamingPullRequest.new
                 end
-                push Google::Cloud::PubSub::V1::StreamingPullRequest.new
               end
-            end.execute
+            end
+
+            @pong_monitor_task = Concurrent::TimerTask.new(
+              execution_interval: [@keepalive_interval / 5.0, 0.01].max
+            ) do
+              synchronize do
+                # Do not check pong deadline if @paused (client flow control inventory full).
+                # When @paused, background_run waits on condition variable and stops calling enum.next,
+                # so incoming server pongs sit buffered in gRPC and @last_pong_at stays un-updated.
+                if @stream_opened && @last_ping_at && @last_pong_at && !@stopped && !@paused
+                  now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+                  if now - @last_ping_at >= @pong_deadline && @last_pong_at < @last_ping_at
+                    subscriber.service.logger.log :error, "subscriber-streams" do
+                      "Keep-alive pong not received within #{@pong_deadline}s; restarting stream."
+                    end
+                    @stream_opened = false
+                    @background_thread&.raise RestartStream
+                  end
+                end
+              end
+            end
           end
 
           def start
@@ -86,6 +119,8 @@ def start
               break if @background_thread
 
               @inventory.start
+              @stream_keepalive_task.execute
+              @pong_monitor_task.execute
 
               start_streaming!
             end
@@ -108,6 +143,9 @@ def stop
               @stopped = true
               @pause_cond.broadcast
 
+              @stream_keepalive_task.shutdown
+              @pong_monitor_task.shutdown
+
               # Now that the reception thread is stopped, immediately stop the
               # callback thread pool. All queued callbacks will see the stream
               # is stopped and perform a noop.
@@ -219,6 +257,13 @@ class RestartStream < StandardError; end
 
           # rubocop:disable all
 
+          def backoff_and_wait!
+            @reconnect_delay = @reconnect_delay ? [@reconnect_delay * 1.5, 60.0].min : 1.0
+            synchronize do
+              @pause_cond.wait(@reconnect_delay + rand(0.0..0.5)) unless @stopped
+            end
+          end
+
           def background_run
             synchronize do
               # Don't allow a stream to restart if already stopped
@@ -245,11 +290,21 @@ def background_run
 
             # Call the StreamingPull API to get the response enumerator
             options = { :"metadata" => { :"x-goog-request-params" =>  @subscriber.subscription_name } }
+            synchronize do
+              @stream_opened = false
+            end
             enum = @subscriber.service.streaming_pull @request_queue.each, options
             subscriber.service.logger.log :info, "subscriber-streams" do
               "rpc: streamingPull, subscription: #{@subscriber.subscription_name}, stream opened"
             end
 
+            synchronize do
+              now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+              @last_ping_at = now
+              @last_pong_at = now
+              @stream_opened = true
+            end
+
             loop do
               synchronize do
                 if @paused && !@stopped
@@ -264,8 +319,17 @@ def background_run
               begin
                 # Cannot synchronize the enumerator, causes deadlock
                 response = enum.next
-                new_exactly_once_delivery_enabled = response&.subscription_properties&.exactly_once_delivery_enabled
+                synchronize do
+                  @last_pong_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+                  # Reset backoff delay only after successfully reading a frame from enum.next.
+                  # If the connection drops immediately upon reading, @reconnect_delay is preserved.
+                  @reconnect_delay = nil
+                end
                 received_messages = response.received_messages
+                # Skip processing properties and inventory on Pong frames (empty received_messages).
+                # Subscription properties on keep-alive Pongs are not valid.
+                next if received_messages.empty?
+                new_exactly_once_delivery_enabled = response&.subscription_properties&.exactly_once_delivery_enabled
 
                 # Use synchronize so changes happen atomically
                 synchronize do
@@ -310,11 +374,13 @@ def background_run
               "#{status_code}; will be retried."
             end
             # Restart the stream with an incremental back for a retriable error.
+            backoff_and_wait!
             retry
           rescue RestartStream
             subscriber.service.logger.log :info, "subscriber-streams" do
               "Subscriber stream for subscription #{@subscriber.subscription_name} has ended; will be retried."
             end
+            backoff_and_wait!
             retry
           rescue StandardError => e
             subscriber.service.logger.log :error, "subscriber-streams" do
@@ -322,6 +388,7 @@ def background_run
             end
             @subscriber.error! e
 
+            backoff_and_wait!
             retry
           end
 
@@ -443,6 +510,7 @@ def initial_input_request
               req.client_id = @subscriber.service.client_id
               req.max_outstanding_messages = @inventory.limit
               req.max_outstanding_bytes = @inventory.bytesize
+              req.protocol_version = 1
             end
           end
 
diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/acknowledge_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/acknowledge_test.rb
index 05b8eeb4ccb8..4f282be9dac1 100644
--- a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/acknowledge_test.rb
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/acknowledge_test.rb
@@ -71,7 +71,8 @@
         subscription: sub_path,
         stream_ack_deadline_seconds: 60,
         max_outstanding_messages: 1000,
-        max_outstanding_bytes: 100 * 1000 * 1000
+        max_outstanding_bytes: 100 * 1000 * 1000,
+        protocol_version: 1
       )]
     ]
 
@@ -132,7 +133,8 @@
         subscription: sub_path,
         stream_ack_deadline_seconds: 60,
         max_outstanding_messages: 1000,
-        max_outstanding_bytes: 100 * 1000 * 1000
+        max_outstanding_bytes: 100 * 1000 * 1000,
+        protocol_version: 1
       )]
     ]
 
diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb
new file mode 100644
index 000000000000..e6fecbfa56b3
--- /dev/null
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb
@@ -0,0 +1,73 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+require "helper"
+
+describe Google::Cloud::PubSub::MessageListener, :bug_regression, :mock_pubsub do
+  let(:topic_name) { "topic-name-goes-here" }
+  let(:sub_name) { "subscription-name-goes-here" }
+  let(:sub_hash) { subscription_hash topic_name, sub_name }
+  let(:sub_grpc) { Google::Cloud::PubSub::V1::Subscription.new(sub_hash) }
+  let(:subscriber) { Google::Cloud::PubSub::Subscriber.from_grpc sub_grpc, pubsub.service }
+
+  it "b/528401453: waits for exponential backoff before retrying on GRPC::Unavailable" do
+    attempts = []
+    pull_res = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: []
+    response_groups = [[GRPC::Unavailable.new("simulated disconnect")], [pull_res]]
+    stub = StreamingPullStub.new response_groups
+    def stub.streaming_pull_internal request, options = nil
+      @attempts ||= []
+      @attempts << Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      super
+    end
+    stub.instance_variable_set(:@attempts, attempts)
+    subscriber.service.mocked_subscription_admin = stub
+
+    listener = subscriber.listen streams: 1 do |msg|
+    end
+    listener.start
+
+    retries = 0
+    until attempts.count >= 2
+      fail "stream did not retry" if retries > 200
+      retries += 1
+      sleep 0.05
+    end
+
+    listener.stop
+    listener.wait!
+
+    elapsed = attempts[1] - attempts[0]
+    puts "\n[b/528401453 Test] Elapsed delay between attempts: #{elapsed.round(3)}s"
+    _(elapsed).must_be :>=, 1.0
+  end
+
+  it "b/528404815: shuts down keepalive TimerTask when stream is stopped" do
+    pull_res = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: []
+    stub = StreamingPullStub.new [[pull_res]]
+    subscriber.service.mocked_subscription_admin = stub
+
+    listener = subscriber.listen streams: 1 do |msg|
+    end
+    listener.start
+    sleep 0.1
+    listener.stop
+    listener.wait!
+
+    stream = listener.instance_variable_get(:@stream_pool).first
+    keepalive_task = stream.instance_variable_get(:@stream_keepalive_task)
+    puts "\n[b/528404815 Test] Keepalive task running state after stop: #{keepalive_task.running?}"
+    _(keepalive_task.running?).must_equal false
+  end
+end
diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/inventory_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/inventory_test.rb
index 2d16c67b0364..bdb3a202c879 100644
--- a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/inventory_test.rb
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/inventory_test.rb
@@ -76,7 +76,8 @@
         subscription: sub_path,
         stream_ack_deadline_seconds: 60,
         max_outstanding_messages: 1000,
-        max_outstanding_bytes: 100 * 1000 * 1000
+        max_outstanding_bytes: 100 * 1000 * 1000,
+        protocol_version: 1
       )]
     ]
 
@@ -141,7 +142,8 @@
         subscription: sub_path,
         stream_ack_deadline_seconds: 60,
         max_outstanding_messages: 1000,
-        max_outstanding_bytes: 100 * 1000 * 1000
+        max_outstanding_bytes: 100 * 1000 * 1000,
+        protocol_version: 1
       )]
     ]
 
diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
new file mode 100644
index 000000000000..ccdef5193081
--- /dev/null
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
@@ -0,0 +1,146 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+require "helper"
+
+describe Google::Cloud::PubSub::MessageListener, :keepalive, :mock_pubsub do
+  let(:topic_name) { "topic-name-goes-here" }
+  let(:sub_name) { "subscription-name-goes-here" }
+  let(:sub_hash) { subscription_hash topic_name, sub_name }
+  let(:sub_grpc) { Google::Cloud::PubSub::V1::Subscription.new(sub_hash) }
+  let(:subscriber) { Google::Cloud::PubSub::Subscriber.from_grpc sub_grpc, pubsub.service }
+  let(:rec_msg1_grpc) { Google::Cloud::PubSub::V1::ReceivedMessage.new \
+                          rec_message_hash("rec_message1-msg-goes-here", 1111) }
+
+  before do
+    ENV["PUBSUB_TEST_KEEPALIVE_INTERVAL"] = "0.05"
+    ENV["PUBSUB_TEST_PONG_DEADLINE"] = "0.05"
+  end
+
+  after do
+    ENV.delete "PUBSUB_TEST_KEEPALIVE_INTERVAL"
+    ENV.delete "PUBSUB_TEST_PONG_DEADLINE"
+  end
+
+  it "sends protocol_version = 1 in initial streaming pull request" do
+    pull_res1 = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: [rec_msg1_grpc]
+    stub = StreamingPullStub.new [[pull_res1]]
+    subscriber.service.mocked_subscription_admin = stub
+
+    called = false
+    listener = subscriber.listen streams: 1 do |msg|
+      called = true
+    end
+    listener.start
+
+    listener_retries = 0
+    until called
+      fail "callback was not called" if listener_retries > 100
+      listener_retries += 1
+      sleep 0.01
+    end
+
+    listener.stop
+    listener.wait!
+
+    initial_req = stub.requests.first.to_a.first
+    _(initial_req.protocol_version).must_equal 1
+  end
+
+  it "sends keep-alive pings periodically even when inventory is empty" do
+    q = StreamingPullStub::RaisableEnumeratorQueue.new
+    stub = StreamingPullStub.new [[]]
+    def stub.streaming_pull_internal req, opt = nil
+      @requests << req
+      @my_q.each
+    end
+    stub.instance_variable_set(:@my_q, q)
+    subscriber.service.mocked_subscription_admin = stub
+
+    listener = subscriber.listen streams: 1 do |msg|
+    end
+    listener.start
+
+    pong_thread = Thread.new do
+      10.times do
+        sleep 0.02
+        q.push Google::Cloud::PubSub::V1::StreamingPullResponse.new(received_messages: [])
+      end
+    end
+
+    sleep 0.18
+    pong_thread.join
+
+    listener.stop
+    listener.wait!
+
+    reqs = stub.requests.first.to_a
+    _(reqs.count).must_be :>=, 2
+  end
+
+  it "restarts stream when keep-alive pong deadline is exceeded" do
+    pull_res2 = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: [rec_msg1_grpc]
+    stub = StreamingPullStub.new [[], [pull_res2]]
+    subscriber.service.mocked_subscription_admin = stub
+
+    called = false
+    listener = subscriber.listen streams: 1 do |msg|
+      called = true
+    end
+    listener.start
+
+    listener_retries = 0
+    until called
+      fail "stream did not restart and deliver message" if listener_retries > 200
+      listener_retries += 1
+      sleep 0.01
+    end
+
+    listener.stop
+    listener.wait!
+
+    _(stub.requests.count).must_equal 2
+  end
+
+  it "does not restart stream when actively receiving keep-alive pongs" do
+    q = StreamingPullStub::RaisableEnumeratorQueue.new
+    stub = StreamingPullStub.new [[]]
+    def stub.streaming_pull_internal req, opt = nil
+      @requests << req
+      @my_q.each
+    end
+    stub.instance_variable_set(:@my_q, q)
+    subscriber.service.mocked_subscription_admin = stub
+
+    listener = subscriber.listen streams: 1 do |msg|
+    end
+    listener.start
+
+    pong_sender = Thread.new do
+      8.times do
+        sleep 0.02
+        empty_pong = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: []
+        q.push empty_pong
+      end
+    end
+
+    sleep 0.15
+    pong_sender.join
+
+    listener.stop
+    listener.wait!
+
+    _(stub.requests.count).must_equal 1
+  end
+end
diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/modify_ack_deadline_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/modify_ack_deadline_test.rb
index 190bbe8d6a78..acd1476f81cc 100644
--- a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/modify_ack_deadline_test.rb
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/modify_ack_deadline_test.rb
@@ -70,7 +70,8 @@
         subscription: sub_path,
         stream_ack_deadline_seconds: 60,
         max_outstanding_messages: 1000,
-        max_outstanding_bytes: 100 * 1000 * 1000
+        max_outstanding_bytes: 100 * 1000 * 1000,
+        protocol_version: 1
       )]
     ]
 
@@ -126,7 +127,8 @@
         subscription: sub_path,
         stream_ack_deadline_seconds: 60,
         max_outstanding_messages: 1000,
-        max_outstanding_bytes: 100 * 1000 * 1000
+        max_outstanding_bytes: 100 * 1000 * 1000,
+        protocol_version: 1
       )]
     ]
 
diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/nack_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/nack_test.rb
index 5b0744ef97de..9f6200a0126a 100644
--- a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/nack_test.rb
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/nack_test.rb
@@ -70,7 +70,8 @@
         subscription: sub_path,
         stream_ack_deadline_seconds: 60,
         max_outstanding_messages: 1000,
-        max_outstanding_bytes: 100 * 1000 * 1000
+        max_outstanding_bytes: 100 * 1000 * 1000,
+        protocol_version: 1
       )]
     ]
 
@@ -126,7 +127,8 @@
         subscription: sub_path,
         stream_ack_deadline_seconds: 60,
         max_outstanding_messages: 1000,
-        max_outstanding_bytes: 100 * 1000 * 1000
+        max_outstanding_bytes: 100 * 1000 * 1000,
+        protocol_version: 1
       )]
     ]
 

From 6b07f4486da72fef0333dce201a7a2e959f22039 Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Mon, 6 Jul 2026 16:16:23 -0700
Subject: [PATCH 02/13] fix(pubsub): remediate keep-alive unpause race and
 timer testability

- 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.
---
 .../cloud/pubsub/message_listener/stream.rb   |  93 ++++++------
 .../message_listener/bug_regression_test.rb   |   4 +-
 .../pubsub/message_listener/keepalive_test.rb | 135 ++++++++++++------
 3 files changed, 142 insertions(+), 90 deletions(-)

diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
index 61cb481d051c..e6e242dd4f7a 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
@@ -76,42 +76,8 @@ def initialize subscriber
             @stream_opened = false
             @reconnect_delay = nil
 
-            @stream_keepalive_task = Concurrent::TimerTask.new(
-              execution_interval: @keepalive_interval
-            ) do
-              synchronize do
-                # @request_queue feeds client requests (initial pull request and keep-alive pings) into gRPC.
-                # Note: ACKs are sent via unary RPCs (TimedUnaryBuffer), not over this stream.
-                # Check that @request_queue is initialized (not nil) before pushing unconditional keep-alive pings.
-                if @stream_opened && !@stopped && @request_queue
-                  subscriber.service.logger.log :info, "subscriber-streams" do
-                    "sending keepAlive to stream for subscription #{@subscriber.subscription_name}"
-                  end
-                  @last_ping_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) if @last_pong_at >= @last_ping_at
-                  push Google::Cloud::PubSub::V1::StreamingPullRequest.new
-                end
-              end
-            end
-
-            @pong_monitor_task = Concurrent::TimerTask.new(
-              execution_interval: [@keepalive_interval / 5.0, 0.01].max
-            ) do
-              synchronize do
-                # Do not check pong deadline if @paused (client flow control inventory full).
-                # When @paused, background_run waits on condition variable and stops calling enum.next,
-                # so incoming server pongs sit buffered in gRPC and @last_pong_at stays un-updated.
-                if @stream_opened && @last_ping_at && @last_pong_at && !@stopped && !@paused
-                  now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
-                  if now - @last_ping_at >= @pong_deadline && @last_pong_at < @last_ping_at
-                    subscriber.service.logger.log :error, "subscriber-streams" do
-                      "Keep-alive pong not received within #{@pong_deadline}s; restarting stream."
-                    end
-                    @stream_opened = false
-                    @background_thread&.raise RestartStream
-                  end
-                end
-              end
-            end
+            @stream_keepalive_task = nil
+            @pong_monitor_task = nil
           end
 
           def start
@@ -119,7 +85,14 @@ def start
               break if @background_thread
 
               @inventory.start
+              @stream_keepalive_task = Concurrent::TimerTask.new(
+                execution_interval: @keepalive_interval
+              ) { send_keepalive_ping! }
               @stream_keepalive_task.execute
+
+              @pong_monitor_task = Concurrent::TimerTask.new(
+                execution_interval: [@keepalive_interval / 5.0, 0.01].max
+              ) { check_liveness! }
               @pong_monitor_task.execute
 
               start_streaming!
@@ -143,8 +116,10 @@ def stop
               @stopped = true
               @pause_cond.broadcast
 
-              @stream_keepalive_task.shutdown
-              @pong_monitor_task.shutdown
+              @stream_keepalive_task&.shutdown
+              @stream_keepalive_task = nil
+              @pong_monitor_task&.shutdown
+              @pong_monitor_task = nil
 
               # Now that the reception thread is stopped, immediately stop the
               # callback thread pool. All queued callbacks will see the stream
@@ -483,15 +458,45 @@ def pause_streaming?
             @inventory.full?
           end
 
+          def send_keepalive_ping!
+            synchronize do
+              if @stream_opened && !@stopped && @request_queue
+                subscriber.service.logger.log :info, "subscriber-streams" do
+                  "sending keepAlive to stream for subscription #{@subscriber.subscription_name}"
+                end
+                @last_ping_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) if @last_pong_at >= @last_ping_at
+                push Google::Cloud::PubSub::V1::StreamingPullRequest.new
+              end
+            end
+          end
+
+          def check_liveness!
+            synchronize do
+              if @stream_opened && @last_ping_at && @last_pong_at && !@stopped && !@paused
+                now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+                if now - @last_ping_at >= @pong_deadline && @last_pong_at < @last_ping_at
+                  subscriber.service.logger.log :error, "subscriber-streams" do
+                    "Keep-alive pong not received within #{@pong_deadline}s; restarting stream."
+                  end
+                  @stream_opened = false
+                  @background_thread&.raise RestartStream
+                end
+              end
+            end
+          end
+
           def unpause_streaming!
-            return unless unpause_streaming?
+            synchronize do
+              return unless unpause_streaming?
 
-            @paused = nil
-            subscriber.service.logger.log :info, "subscriber-flow-control" do
-              "subscriber for #{@subscriber.subscription_name} is unblocking client-side flow control"
+              @paused = nil
+              @last_pong_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+              subscriber.service.logger.log :info, "subscriber-flow-control" do
+                "subscriber for #{@subscriber.subscription_name} is unblocking client-side flow control"
+              end
+              # signal to the background thread that we are unpaused
+              @pause_cond.broadcast
             end
-            # signal to the background thread that we are unpaused
-            @pause_cond.broadcast
           end
 
           def unpause_streaming?
diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb
index e6fecbfa56b3..cda585f176e1 100644
--- a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb
@@ -67,7 +67,7 @@ def stub.streaming_pull_internal request, options = nil
 
     stream = listener.instance_variable_get(:@stream_pool).first
     keepalive_task = stream.instance_variable_get(:@stream_keepalive_task)
-    puts "\n[b/528404815 Test] Keepalive task running state after stop: #{keepalive_task.running?}"
-    _(keepalive_task.running?).must_equal false
+    puts "\n[b/528404815 Test] Keepalive task running state after stop: #{keepalive_task&.running? || false}"
+    _(keepalive_task.nil? || !keepalive_task.running?).must_equal true
   end
 end
diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
index ccdef5193081..667d03704261 100644
--- a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
@@ -58,29 +58,40 @@
     _(initial_req.protocol_version).must_equal 1
   end
 
-  it "sends keep-alive pings periodically even when inventory is empty" do
-    q = StreamingPullStub::RaisableEnumeratorQueue.new
-    stub = StreamingPullStub.new [[]]
-    def stub.streaming_pull_internal req, opt = nil
-      @requests << req
-      @my_q.each
-    end
-    stub.instance_variable_set(:@my_q, q)
+  it "restarts stream when keep-alive pong deadline is exceeded" do
+    pull_res2 = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: [rec_msg1_grpc]
+    stub = StreamingPullStub.new [[], [pull_res2]]
     subscriber.service.mocked_subscription_admin = stub
 
+    called = false
     listener = subscriber.listen streams: 1 do |msg|
+      called = true
     end
     listener.start
 
-    pong_thread = Thread.new do
-      10.times do
-        sleep 0.02
-        q.push Google::Cloud::PubSub::V1::StreamingPullResponse.new(received_messages: [])
-      end
+    listener_retries = 0
+    until called
+      fail "stream did not restart and deliver message" if listener_retries > 500
+      listener_retries += 1
+      sleep 0.01
     end
 
-    sleep 0.18
-    pong_thread.join
+    listener.stop
+    listener.wait!
+
+    _(stub.requests.count).must_equal 2
+  end
+
+  it "sends keep-alive ping synchronously when stream is open and queue exists" do
+    stub = StreamingPullStub.new [[]]
+    subscriber.service.mocked_subscription_admin = stub
+
+    listener = subscriber.listen streams: 1 do |msg|
+    end
+    listener.start
+
+    stream = listener.instance_variable_get(:@stream_pool).first
+    stream.send(:send_keepalive_ping!)
 
     listener.stop
     listener.wait!
@@ -89,58 +100,94 @@ def stub.streaming_pull_internal req, opt = nil
     _(reqs.count).must_be :>=, 2
   end
 
-  it "restarts stream when keep-alive pong deadline is exceeded" do
-    pull_res2 = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: [rec_msg1_grpc]
-    stub = StreamingPullStub.new [[], [pull_res2]]
+  it "does not restart stream when check_liveness! runs under active pongs" do
+    stub = StreamingPullStub.new [[]]
     subscriber.service.mocked_subscription_admin = stub
 
-    called = false
     listener = subscriber.listen streams: 1 do |msg|
-      called = true
     end
     listener.start
 
-    listener_retries = 0
-    until called
-      fail "stream did not restart and deliver message" if listener_retries > 200
-      listener_retries += 1
-      sleep 0.01
-    end
+    stream = listener.instance_variable_get(:@stream_pool).first
+    stream.instance_variable_set :@last_ping_at, Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    stream.instance_variable_set :@last_pong_at, Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+    stream.send(:check_liveness!)
+    _(stream.instance_variable_get(:@stream_opened)).must_equal true
 
     listener.stop
     listener.wait!
-
-    _(stub.requests.count).must_equal 2
   end
 
-  it "does not restart stream when actively receiving keep-alive pongs" do
-    q = StreamingPullStub::RaisableEnumeratorQueue.new
+  it "does not trigger false restarts on unpausing even if pause exceeded deadline" do
     stub = StreamingPullStub.new [[]]
-    def stub.streaming_pull_internal req, opt = nil
-      @requests << req
-      @my_q.each
-    end
-    stub.instance_variable_set(:@my_q, q)
     subscriber.service.mocked_subscription_admin = stub
 
     listener = subscriber.listen streams: 1 do |msg|
     end
     listener.start
 
-    pong_sender = Thread.new do
-      8.times do
-        sleep 0.02
-        empty_pong = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: []
-        q.push empty_pong
-      end
+    stream = listener.instance_variable_get(:@stream_pool).first
+    stream.instance_variable_set :@stream_opened, true
+    stream.instance_variable_set :@last_ping_at, Process.clock_gettime(Process::CLOCK_MONOTONIC) - 30.0
+    stream.instance_variable_set :@last_pong_at, Process.clock_gettime(Process::CLOCK_MONOTONIC) - 35.0
+    stream.instance_variable_set :@paused, true
+    stream.instance_variable_set :@stopped, false
+
+    # 1. When paused, liveness checker should skip verification.
+    stream.send(:check_liveness!)
+    _(stream.instance_variable_get(:@stream_opened)).must_equal true
+
+    # 2. Unpausing must reset @last_pong_at to prevent immediate restart.
+    stream.send(:unpause_streaming!)
+
+    # 3. Direct liveness check immediately after unpausing (simulates the monitor thread racing the background thread).
+    # It should NOT close the stream because our reset made @last_pong_at newer than @last_ping_at.
+    stream.send(:check_liveness!)
+    _(stream.instance_variable_get(:@stream_opened)).must_equal true
+
+    listener.stop
+    listener.wait!
+  end
+
+  it "re-creates and executes timer tasks if stopped and restarted" do
+    stub = StreamingPullStub.new [[]]
+    subscriber.service.mocked_subscription_admin = stub
+
+    listener = subscriber.listen streams: 1 do |msg|
     end
+    listener.start
+
+    stream = listener.instance_variable_get(:@stream_pool).first
+
+    # 1. Timers should be active on a started stream
+    refute_nil stream.instance_variable_get(:@stream_keepalive_task)
+    refute_nil stream.instance_variable_get(:@pong_monitor_task)
 
-    sleep 0.15
-    pong_sender.join
+    first_keepalive = stream.instance_variable_get(:@stream_keepalive_task)
+    first_monitor = stream.instance_variable_get(:@pong_monitor_task)
 
+    # 2. Stopping the stream must shutdown and nilify the timers
     listener.stop
     listener.wait!
+    assert_nil stream.instance_variable_get(:@stream_keepalive_task)
+    assert_nil stream.instance_variable_get(:@pong_monitor_task)
+    assert first_keepalive.shutdown?
+    assert first_monitor.shutdown?
+
+    # 3. Simulating a restart should create completely new timer instances
+    stream.instance_variable_set :@background_thread, nil
+    stream.instance_variable_set :@stopped, false
+    stream.start
+
+    new_keepalive = stream.instance_variable_get(:@stream_keepalive_task)
+    new_monitor = stream.instance_variable_get(:@pong_monitor_task)
+
+    refute_nil new_keepalive
+    refute_nil new_monitor
+    refute_equal first_keepalive, new_keepalive
+    refute_equal first_monitor, new_monitor
 
-    _(stub.requests.count).must_equal 1
+    stream.stop
   end
 end

From 12191564cb66f071cebf7ee1aa0bbdc7321635f0 Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Mon, 6 Jul 2026 16:16:43 -0700
Subject: [PATCH 03/13] fix(pubsub): make keep-alive unit tests deterministic
 without starting 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).
---
 .../pubsub/message_listener/keepalive_test.rb | 98 +++++++++----------
 1 file changed, 47 insertions(+), 51 deletions(-)

diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
index 667d03704261..4f7108f6e4c7 100644
--- a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
@@ -23,16 +23,6 @@
   let(:rec_msg1_grpc) { Google::Cloud::PubSub::V1::ReceivedMessage.new \
                           rec_message_hash("rec_message1-msg-goes-here", 1111) }
 
-  before do
-    ENV["PUBSUB_TEST_KEEPALIVE_INTERVAL"] = "0.05"
-    ENV["PUBSUB_TEST_PONG_DEADLINE"] = "0.05"
-  end
-
-  after do
-    ENV.delete "PUBSUB_TEST_KEEPALIVE_INTERVAL"
-    ENV.delete "PUBSUB_TEST_PONG_DEADLINE"
-  end
-
   it "sends protocol_version = 1 in initial streaming pull request" do
     pull_res1 = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: [rec_msg1_grpc]
     stub = StreamingPullStub.new [[pull_res1]]
@@ -59,27 +49,34 @@
   end
 
   it "restarts stream when keep-alive pong deadline is exceeded" do
-    pull_res2 = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: [rec_msg1_grpc]
-    stub = StreamingPullStub.new [[], [pull_res2]]
-    subscriber.service.mocked_subscription_admin = stub
-
-    called = false
-    listener = subscriber.listen streams: 1 do |msg|
-      called = true
-    end
-    listener.start
-
-    listener_retries = 0
-    until called
-      fail "stream did not restart and deliver message" if listener_retries > 500
-      listener_retries += 1
-      sleep 0.01
+    ENV["PUBSUB_TEST_KEEPALIVE_INTERVAL"] = "0.05"
+    ENV["PUBSUB_TEST_PONG_DEADLINE"] = "0.05"
+    begin
+      pull_res2 = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: [rec_msg1_grpc]
+      stub = StreamingPullStub.new [[], [pull_res2]]
+      subscriber.service.mocked_subscription_admin = stub
+
+      called = false
+      listener = subscriber.listen streams: 1 do |msg|
+        called = true
+      end
+      listener.start
+
+      listener_retries = 0
+      until called
+        fail "stream did not restart and deliver message" if listener_retries > 500
+        listener_retries += 1
+        sleep 0.01
+      end
+
+      listener.stop
+      listener.wait!
+
+      _(stub.requests.count).must_equal 2
+    ensure
+      ENV.delete "PUBSUB_TEST_KEEPALIVE_INTERVAL"
+      ENV.delete "PUBSUB_TEST_PONG_DEADLINE"
     end
-
-    listener.stop
-    listener.wait!
-
-    _(stub.requests.count).must_equal 2
   end
 
   it "sends keep-alive ping synchronously when stream is open and queue exists" do
@@ -88,16 +85,17 @@
 
     listener = subscriber.listen streams: 1 do |msg|
     end
-    listener.start
-
     stream = listener.instance_variable_get(:@stream_pool).first
-    stream.send(:send_keepalive_ping!)
 
-    listener.stop
-    listener.wait!
+    queue = Google::Cloud::PubSub::MessageListener::EnumeratorQueue.new stream
+    stream.instance_variable_set :@request_queue, queue
+    stream.instance_variable_set :@stream_opened, true
+    stream.instance_variable_set :@stopped, false
+    stream.instance_variable_set :@last_ping_at, 0.0
+    stream.instance_variable_set :@last_pong_at, 1.0
 
-    reqs = stub.requests.first.to_a
-    _(reqs.count).must_be :>=, 2
+    stream.send(:send_keepalive_ping!)
+    _(stream.instance_variable_get(:@last_ping_at)).must_be :>, 0.0
   end
 
   it "does not restart stream when check_liveness! runs under active pongs" do
@@ -106,17 +104,17 @@
 
     listener = subscriber.listen streams: 1 do |msg|
     end
-    listener.start
-
     stream = listener.instance_variable_get(:@stream_pool).first
-    stream.instance_variable_set :@last_ping_at, Process.clock_gettime(Process::CLOCK_MONOTONIC)
-    stream.instance_variable_set :@last_pong_at, Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+    now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    stream.instance_variable_set :@stream_opened, true
+    stream.instance_variable_set :@stopped, false
+    stream.instance_variable_set :@paused, false
+    stream.instance_variable_set :@last_ping_at, now - 5.0
+    stream.instance_variable_set :@last_pong_at, now - 1.0
 
     stream.send(:check_liveness!)
     _(stream.instance_variable_get(:@stream_opened)).must_equal true
-
-    listener.stop
-    listener.wait!
   end
 
   it "does not trigger false restarts on unpausing even if pause exceeded deadline" do
@@ -125,12 +123,12 @@
 
     listener = subscriber.listen streams: 1 do |msg|
     end
-    listener.start
-
     stream = listener.instance_variable_get(:@stream_pool).first
+
+    now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
     stream.instance_variable_set :@stream_opened, true
-    stream.instance_variable_set :@last_ping_at, Process.clock_gettime(Process::CLOCK_MONOTONIC) - 30.0
-    stream.instance_variable_set :@last_pong_at, Process.clock_gettime(Process::CLOCK_MONOTONIC) - 35.0
+    stream.instance_variable_set :@last_ping_at, now - 30.0
+    stream.instance_variable_set :@last_pong_at, now - 35.0
     stream.instance_variable_set :@paused, true
     stream.instance_variable_set :@stopped, false
 
@@ -140,14 +138,12 @@
 
     # 2. Unpausing must reset @last_pong_at to prevent immediate restart.
     stream.send(:unpause_streaming!)
+    _(stream.instance_variable_get(:@last_pong_at)).must_be :>=, now
 
     # 3. Direct liveness check immediately after unpausing (simulates the monitor thread racing the background thread).
     # It should NOT close the stream because our reset made @last_pong_at newer than @last_ping_at.
     stream.send(:check_liveness!)
     _(stream.instance_variable_get(:@stream_opened)).must_equal true
-
-    listener.stop
-    listener.wait!
   end
 
   it "re-creates and executes timer tasks if stopped and restarted" do

From a7165c9ab762019ed3b8ce2a899dc5845500aa2b Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Mon, 6 Jul 2026 16:17:05 -0700
Subject: [PATCH 04/13] fix(pubsub): remove test-only env vars, remove
 buganizer IDs and capitalize comment

---
 .../lib/google/cloud/pubsub/message_listener/stream.rb | 10 +++++++---
 .../pubsub/message_listener/bug_regression_test.rb     |  4 ++--
 2 files changed, 9 insertions(+), 5 deletions(-)

diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
index e6e242dd4f7a..4ff00f6d8a41 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
@@ -48,7 +48,11 @@ class Stream
           ##
           # @private exactly_once_delivery_enabled.
           attr_reader :exactly_once_delivery_enabled
-          attr_accessor :keepalive_interval, :pong_deadline
+
+          # keepalive_interval used to send pings and maintain a healthy stream
+          attr_accessor :keepalive_interval
+          # the interval used to validate whether we received a pong back from the stream
+          attr_accessor :pong_deadline
 
           ##
           # @private Create an empty Subscriber::Stream object.
@@ -69,8 +73,8 @@ def initialize subscriber
 
             @callback_thread_pool = Concurrent::ThreadPoolExecutor.new max_threads: @subscriber.callback_threads
 
-            @keepalive_interval = Float(ENV["PUBSUB_TEST_KEEPALIVE_INTERVAL"] || 30)
-            @pong_deadline = Float(ENV["PUBSUB_TEST_PONG_DEADLINE"] || 15)
+            @keepalive_interval = 30
+            @pong_deadline = 15
             @last_ping_at = nil
             @last_pong_at = nil
             @stream_opened = false
diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb
index cda585f176e1..ff16430f96b8 100644
--- a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/bug_regression_test.rb
@@ -21,7 +21,7 @@
   let(:sub_grpc) { Google::Cloud::PubSub::V1::Subscription.new(sub_hash) }
   let(:subscriber) { Google::Cloud::PubSub::Subscriber.from_grpc sub_grpc, pubsub.service }
 
-  it "b/528401453: waits for exponential backoff before retrying on GRPC::Unavailable" do
+  it "waits for exponential backoff before retrying on GRPC::Unavailable" do
     attempts = []
     pull_res = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: []
     response_groups = [[GRPC::Unavailable.new("simulated disconnect")], [pull_res]]
@@ -53,7 +53,7 @@ def stub.streaming_pull_internal request, options = nil
     _(elapsed).must_be :>=, 1.0
   end
 
-  it "b/528404815: shuts down keepalive TimerTask when stream is stopped" do
+  it "shuts down keepalive TimerTask when stream is stopped" do
     pull_res = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: []
     stub = StreamingPullStub.new [[pull_res]]
     subscriber.service.mocked_subscription_admin = stub

From 0ad78cbffc03868840440838536480ea183991d2 Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Mon, 6 Jul 2026 16:17:15 -0700
Subject: [PATCH 05/13] capitalize comment

---
 .../lib/google/cloud/pubsub/message_listener/stream.rb          | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
index 4ff00f6d8a41..84aec9140543 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
@@ -498,7 +498,7 @@ def unpause_streaming!
               subscriber.service.logger.log :info, "subscriber-flow-control" do
                 "subscriber for #{@subscriber.subscription_name} is unblocking client-side flow control"
               end
-              # signal to the background thread that we are unpaused
+              # Signal to the background thread that we are unpaused
               @pause_cond.broadcast
             end
           end

From 9461a5694b8abfa70f33e817f3405938681241e9 Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Mon, 6 Jul 2026 16:15:07 -0700
Subject: [PATCH 06/13] fix(pubsub): disable liveness check during retry
 backoff and document 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.
---
 .../cloud/pubsub/message_listener/stream.rb   | 16 ++++++
 .../pubsub/message_listener/keepalive_test.rb | 50 +++++++++----------
 2 files changed, 39 insertions(+), 27 deletions(-)

diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
index 84aec9140543..8229b5f61a68 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
@@ -94,6 +94,8 @@ def start
               ) { send_keepalive_ping! }
               @stream_keepalive_task.execute
 
+              # Check liveness at 1/5th of keepalive interval to detect timeouts promptly without waiting a full cycle.
+              # Floor interval at 0.01s (10ms) to prevent thread spinning and CPU exhaustion during sub-second tests.
               @pong_monitor_task = Concurrent::TimerTask.new(
                 execution_interval: [@keepalive_interval / 5.0, 0.01].max
               ) { check_liveness! }
@@ -239,6 +241,8 @@ class RestartStream < StandardError; end
           def backoff_and_wait!
             @reconnect_delay = @reconnect_delay ? [@reconnect_delay * 1.5, 60.0].min : 1.0
             synchronize do
+              # Disable liveness checker during backoff sleep to prevent monitor task from interrupting and double-backing off
+              @stream_opened = false
               @pause_cond.wait(@reconnect_delay + rand(0.0..0.5)) unless @stopped
             end
           end
@@ -464,10 +468,15 @@ def pause_streaming?
 
           def send_keepalive_ping!
             synchronize do
+              # Push unconditional keep-alive pings only when the stream is actively open and @request_queue is initialized.
+              # Note: ACKs are sent via unary RPCs (TimedUnaryBuffer), not over this stream.
               if @stream_opened && !@stopped && @request_queue
                 subscriber.service.logger.log :info, "subscriber-streams" do
                   "sending keepAlive to stream for subscription #{@subscriber.subscription_name}"
                 end
+                # Only advance @last_ping_at if the previous ping was successfully ponged.
+                # If a pong is outstanding (@last_pong_at < @last_ping_at), freezing @last_ping_at preserves the start time
+                # of the un-ponged cycle so check_liveness! can accurately evaluate the elapsed deadline.
                 @last_ping_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) if @last_pong_at >= @last_ping_at
                 push Google::Cloud::PubSub::V1::StreamingPullRequest.new
               end
@@ -476,8 +485,12 @@ def send_keepalive_ping!
 
           def check_liveness!
             synchronize do
+              # Do not check pong deadline if @paused (client flow control inventory full).
+              # When @paused, background_run waits on condition variable and stops calling enum.next,
+              # so incoming server pongs sit buffered in gRPC and @last_pong_at stays un-updated.
               if @stream_opened && @last_ping_at && @last_pong_at && !@stopped && !@paused
                 now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+                # If elapsed time since ping exceeds deadline and no pong arrived after the ping, restart stream.
                 if now - @last_ping_at >= @pong_deadline && @last_pong_at < @last_ping_at
                   subscriber.service.logger.log :error, "subscriber-streams" do
                     "Keep-alive pong not received within #{@pong_deadline}s; restarting stream."
@@ -494,6 +507,9 @@ def unpause_streaming!
               return unless unpause_streaming?
 
               @paused = nil
+              # Reset @last_pong_at when unpausing flow control. While paused, incoming server pongs sit buffered in gRPC,
+              # leaving @last_pong_at stale. Updating timestamp guarantees the monitor thread will not trigger an immediate
+              # false-positive restart while the reader thread wakes up to drain buffered frames.
               @last_pong_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
               subscriber.service.logger.log :info, "subscriber-flow-control" do
                 "subscriber for #{@subscriber.subscription_name} is unblocking client-side flow control"
diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
index 4f7108f6e4c7..137ebc0f0b8b 100644
--- a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
@@ -49,34 +49,30 @@
   end
 
   it "restarts stream when keep-alive pong deadline is exceeded" do
-    ENV["PUBSUB_TEST_KEEPALIVE_INTERVAL"] = "0.05"
-    ENV["PUBSUB_TEST_PONG_DEADLINE"] = "0.05"
-    begin
-      pull_res2 = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: [rec_msg1_grpc]
-      stub = StreamingPullStub.new [[], [pull_res2]]
-      subscriber.service.mocked_subscription_admin = stub
-
-      called = false
-      listener = subscriber.listen streams: 1 do |msg|
-        called = true
-      end
-      listener.start
-
-      listener_retries = 0
-      until called
-        fail "stream did not restart and deliver message" if listener_retries > 500
-        listener_retries += 1
-        sleep 0.01
-      end
-
-      listener.stop
-      listener.wait!
-
-      _(stub.requests.count).must_equal 2
-    ensure
-      ENV.delete "PUBSUB_TEST_KEEPALIVE_INTERVAL"
-      ENV.delete "PUBSUB_TEST_PONG_DEADLINE"
+    pull_res2 = Google::Cloud::PubSub::V1::StreamingPullResponse.new received_messages: [rec_msg1_grpc]
+    stub = StreamingPullStub.new [[], [pull_res2]]
+    subscriber.service.mocked_subscription_admin = stub
+
+    called = false
+    listener = subscriber.listen streams: 1 do |msg|
+      called = true
     end
+    stream = listener.instance_variable_get(:@stream_pool).first
+    stream.keepalive_interval = 0.05
+    stream.pong_deadline = 0.05
+    listener.start
+
+    listener_retries = 0
+    until called
+      fail "stream did not restart and deliver message" if listener_retries > 500
+      listener_retries += 1
+      sleep 0.01
+    end
+
+    listener.stop
+    listener.wait!
+
+    _(stub.requests.count).must_equal 2
   end
 
   it "sends keep-alive ping synchronously when stream is open and queue exists" do

From cca4878b91d04f18f387aea156ef8408ef5b58db Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Tue, 7 Jul 2026 15:32:32 +0000
Subject: [PATCH 07/13] feat(pubsub): enrich keep-alive timeout diagnostic logs
 and push closing 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.
---
 .../lib/google/cloud/pubsub/message_listener/stream.rb       | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
index 8229b5f61a68..f1ddae0642fa 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
@@ -490,12 +490,13 @@ def check_liveness!
               # so incoming server pongs sit buffered in gRPC and @last_pong_at stays un-updated.
               if @stream_opened && @last_ping_at && @last_pong_at && !@stopped && !@paused
                 now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
-                # If elapsed time since ping exceeds deadline and no pong arrived after the ping, restart stream.
                 if now - @last_ping_at >= @pong_deadline && @last_pong_at < @last_ping_at
+                  elapsed_pong = (now - @last_pong_at).round(2)
                   subscriber.service.logger.log :error, "subscriber-streams" do
-                    "Keep-alive pong not received within #{@pong_deadline}s; restarting stream."
+                    "Keep-alive pong not received within #{@pong_deadline}s (last pong #{elapsed_pong}s ago, inventory: #{@inventory.count}, paused: #{@paused.inspect}); restarting stream."
                   end
                   @stream_opened = false
+                  @request_queue&.push self
                   @background_thread&.raise RestartStream
                 end
               end

From 82b508d023ce1ca629f46474bdae5fe63e43d7e2 Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Tue, 7 Jul 2026 16:58:43 +0000
Subject: [PATCH 08/13] docs(pubsub): document timestamp re-initialization and
 stream_opened 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).
---
 .../lib/google/cloud/pubsub/message_listener/stream.rb      | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
index f1ddae0642fa..508658f138ca 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
@@ -273,6 +273,8 @@ def background_run
 
             # Call the StreamingPull API to get the response enumerator
             options = { :"metadata" => { :"x-goog-request-params" =>  @subscriber.subscription_name } }
+            # Temporarily disable the liveness monitor while establishing the gRPC connection
+            # to prevent check_liveness! from evaluating stale timestamps during the handshake.
             synchronize do
               @stream_opened = false
             end
@@ -281,6 +283,10 @@ def background_run
               "rpc: streamingPull, subscription: #{@subscriber.subscription_name}, stream opened"
             end
 
+            # Once the stream handshake completes, initialize ping/pong timestamps to monotonic now
+            # and mark @stream_opened = true under synchronization.
+            # This ensures @pong_monitor_task evaluates liveness against a fresh window and prevents
+            # false-positive disconnect detections from stale timestamps prior to reconnecting.
             synchronize do
               now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
               @last_ping_at = now

From c707629f7d5341ff8a26108b5d15db4d6d446546 Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Tue, 7 Jul 2026 20:38:42 +0000
Subject: [PATCH 09/13] refactor(pubsub): extract numeric constants on Stream
 and document stream-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.
---
 .../cloud/pubsub/message_listener/stream.rb   | 37 ++++++++++++++++---
 1 file changed, 32 insertions(+), 5 deletions(-)

diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
index 508658f138ca..7beff5b8706a 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
@@ -54,6 +54,30 @@ class Stream
           # the interval used to validate whether we received a pong back from the stream
           attr_accessor :pong_deadline
 
+          # Default interval in seconds between keep-alive ping requests.
+          DEFAULT_KEEPALIVE_INTERVAL = 30.0
+
+          # Default deadline in seconds to wait for a keep-alive pong response before restarting the stream.
+          DEFAULT_PONG_DEADLINE = 15.0
+
+          # Divisor applied to keep-alive interval to calculate monitor polling interval (1/5th of interval).
+          PONG_MONITOR_INTERVAL_DIVISOR = 5.0
+
+          # Minimum floor in seconds (10ms) for the monitor polling interval to prevent CPU spinning.
+          PONG_MONITOR_MIN_INTERVAL = 0.01
+
+          # Initial backoff delay in seconds when reconnecting after a transient stream disconnection.
+          INITIAL_RECONNECT_DELAY = 1.0
+
+          # Maximum backoff delay in seconds for stream reconnection attempts.
+          MAX_RECONNECT_DELAY = 60.0
+
+          # Exponential backoff multiplier applied to reconnect delay on successive retry attempts.
+          RECONNECT_BACKOFF_MULTIPLIER = 1.5
+
+          # Inventory capacity ratio (80%) below which a flow-control paused stream will unpause.
+          UNPAUSE_INVENTORY_RATIO = 0.8
+
           ##
           # @private Create an empty Subscriber::Stream object.
           def initialize subscriber
@@ -73,8 +97,8 @@ def initialize subscriber
 
             @callback_thread_pool = Concurrent::ThreadPoolExecutor.new max_threads: @subscriber.callback_threads
 
-            @keepalive_interval = 30
-            @pong_deadline = 15
+            @keepalive_interval = DEFAULT_KEEPALIVE_INTERVAL
+            @pong_deadline = DEFAULT_PONG_DEADLINE
             @last_ping_at = nil
             @last_pong_at = nil
             @stream_opened = false
@@ -97,7 +121,7 @@ def start
               # Check liveness at 1/5th of keepalive interval to detect timeouts promptly without waiting a full cycle.
               # Floor interval at 0.01s (10ms) to prevent thread spinning and CPU exhaustion during sub-second tests.
               @pong_monitor_task = Concurrent::TimerTask.new(
-                execution_interval: [@keepalive_interval / 5.0, 0.01].max
+                execution_interval: [@keepalive_interval / PONG_MONITOR_INTERVAL_DIVISOR, PONG_MONITOR_MIN_INTERVAL].max
               ) { check_liveness! }
               @pong_monitor_task.execute
 
@@ -239,7 +263,7 @@ class RestartStream < StandardError; end
           # rubocop:disable all
 
           def backoff_and_wait!
-            @reconnect_delay = @reconnect_delay ? [@reconnect_delay * 1.5, 60.0].min : 1.0
+            @reconnect_delay = @reconnect_delay ? [@reconnect_delay * RECONNECT_BACKOFF_MULTIPLIER, MAX_RECONNECT_DELAY].min : INITIAL_RECONNECT_DELAY
             synchronize do
               # Disable liveness checker during backoff sleep to prevent monitor task from interrupting and double-backing off
               @stream_opened = false
@@ -502,6 +526,9 @@ def check_liveness!
                     "Keep-alive pong not received within #{@pong_deadline}s (last pong #{elapsed_pong}s ago, inventory: #{@inventory.count}, paused: #{@paused.inspect}); restarting stream."
                   end
                   @stream_opened = false
+                  # Push self as a stream-closing sentinel to @request_queue.
+                  # When EnumeratorQueue#each pops the sentinel object, it terminates the request enumerator,
+                  # cleanly sending an HTTP/2 END_STREAM flag and unblocking the write-side gRPC C-core pipeline.
                   @request_queue&.push self
                   @background_thread&.raise RestartStream
                 end
@@ -530,7 +557,7 @@ def unpause_streaming?
             return false if @stopped
             return false if @paused.nil?
 
-            @inventory.count < @inventory.limit * 0.8
+            @inventory.count < @inventory.limit * UNPAUSE_INVENTORY_RATIO
           end
 
           def initial_input_request

From 4b5d1be8d8403dc144bf57faed5bdbdb4307def0 Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Tue, 7 Jul 2026 23:10:46 +0000
Subject: [PATCH 10/13] test(pubsub): verify stream_opened is set to false
 during backoff_and_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.
---
 .../pubsub/message_listener/keepalive_test.rb | 23 +++++++++++++++++++
 1 file changed, 23 insertions(+)

diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
index 137ebc0f0b8b..e65919c6fd55 100644
--- a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
@@ -182,4 +182,27 @@
 
     stream.stop
   end
+
+  it "sets stream_opened to false during backoff to prevent monitor interruption" do
+    stub = StreamingPullStub.new [[]]
+    subscriber.service.mocked_subscription_admin = stub
+
+    listener = subscriber.listen streams: 1 do |msg|
+    end
+    stream = listener.instance_variable_get(:@stream_pool).first
+    stream.instance_variable_set :@stream_opened, true
+    stream.instance_variable_set :@stopped, false
+
+    wait_called = false
+    pause_cond = stream.instance_variable_get(:@pause_cond)
+
+    pause_cond.stub :wait, ->(_timeout) {
+      wait_called = true
+      _(stream.instance_variable_get(:@stream_opened)).must_equal false
+    } do
+      stream.send :backoff_and_wait!
+    end
+
+    assert wait_called
+  end
 end

From 9e62a046d24fddcca862533f963ab225b257d353 Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Wed, 8 Jul 2026 16:31:39 +0000
Subject: [PATCH 11/13] refactor(pubsub): extract KeepaliveMonitor collaborator
 class from Stream

- 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.
---
 .../message_listener/keepalive_monitor.rb     | 114 +++++++++++++
 .../cloud/pubsub/message_listener/stream.rb   | 158 +++++++++---------
 .../pubsub/message_listener/keepalive_test.rb |  38 ++---
 3 files changed, 212 insertions(+), 98 deletions(-)
 create mode 100644 google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb

diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb
new file mode 100644
index 000000000000..535620e75921
--- /dev/null
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb
@@ -0,0 +1,114 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+require "concurrent"
+
+module Google
+  module Cloud
+    module PubSub
+      class MessageListener
+        ##
+        # @private
+        # Monitors gRPC streaming connection health via periodic bi-directional keep-alive pings and pongs.
+        class KeepaliveMonitor
+          DEFAULT_INTERVAL = 30.0
+          DEFAULT_DEADLINE = 15.0
+          MONITOR_DIVISOR = 5.0
+          MIN_MONITOR_INTERVAL = 0.01
+
+          attr_accessor :interval, :deadline, :last_ping_at, :last_pong_at
+          attr_reader :ping_task, :monitor_task
+
+          def initialize stream, interval: DEFAULT_INTERVAL, deadline: DEFAULT_DEADLINE
+            @stream       = stream
+            @interval     = interval
+            @deadline     = deadline
+            @last_ping_at = nil
+            @last_pong_at = nil
+            @ping_task    = nil
+            @monitor_task = nil
+          end
+
+          def start
+            @stream.synchronize do
+              return self if @ping_task
+
+              @ping_task = Concurrent::TimerTask.new(execution_interval: @interval) do
+                send_ping!
+              end
+              @ping_task.execute
+
+              @monitor_task = Concurrent::TimerTask.new(
+                execution_interval: [@interval / MONITOR_DIVISOR, MIN_MONITOR_INTERVAL].max
+              ) do
+                check_liveness!
+              end
+              @monitor_task.execute
+            end
+            self
+          end
+
+          def stop
+            @stream.synchronize do
+              @ping_task&.shutdown
+              @ping_task = nil
+              @monitor_task&.shutdown
+              @monitor_task = nil
+            end
+            self
+          end
+
+          def record_handshake!
+            @stream.synchronize do
+              now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+              @last_ping_at = now
+              @last_pong_at = now
+            end
+          end
+
+          def record_pong!
+            @stream.synchronize do
+              @last_pong_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+            end
+          end
+
+          def send_ping!
+            @stream.synchronize do
+              return unless @stream.stream_opened? && !@stream.stopped? && @stream.request_queue_active?
+
+              @stream.log_info "sending keepAlive to stream for subscription #{@stream.subscriber.subscription_name}"
+              @last_ping_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) if @last_pong_at >= @last_ping_at
+              @stream.send_ping_request!
+            end
+          end
+
+          def check_liveness!
+            @stream.synchronize do
+              return unless @stream.stream_opened? && @last_ping_at && @last_pong_at && !@stream.stopped? && !@stream.paused?
+
+              now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+              if now - @last_ping_at >= @deadline && @last_pong_at < @last_ping_at
+                elapsed_pong = (now - @last_pong_at).round(2)
+                @stream.log_error "Keep-alive pong not received within #{@deadline}s (last pong #{elapsed_pong}s ago); restarting stream."
+                @stream.restart_stream_for_timeout!
+              end
+            end
+          end
+        end
+      end
+    end
+
+    Pubsub = PubSub unless const_defined? :Pubsub
+  end
+end
diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
index 7beff5b8706a..9aee6e761fb4 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
@@ -15,6 +15,7 @@
 require "google/cloud/pubsub/message_listener/sequencer"
 require "google/cloud/pubsub/message_listener/enumerator_queue"
 require "google/cloud/pubsub/message_listener/inventory"
+require "google/cloud/pubsub/message_listener/keepalive_monitor"
 require "google/cloud/pubsub/service"
 require "google/cloud/errors"
 require "monitor"
@@ -49,22 +50,9 @@ class Stream
           # @private exactly_once_delivery_enabled.
           attr_reader :exactly_once_delivery_enabled
 
-          # keepalive_interval used to send pings and maintain a healthy stream
-          attr_accessor :keepalive_interval
-          # the interval used to validate whether we received a pong back from the stream
-          attr_accessor :pong_deadline
-
-          # Default interval in seconds between keep-alive ping requests.
-          DEFAULT_KEEPALIVE_INTERVAL = 30.0
-
-          # Default deadline in seconds to wait for a keep-alive pong response before restarting the stream.
-          DEFAULT_PONG_DEADLINE = 15.0
-
-          # Divisor applied to keep-alive interval to calculate monitor polling interval (1/5th of interval).
-          PONG_MONITOR_INTERVAL_DIVISOR = 5.0
-
-          # Minimum floor in seconds (10ms) for the monitor polling interval to prevent CPU spinning.
-          PONG_MONITOR_MIN_INTERVAL = 0.01
+          ##
+          # @private KeepaliveMonitor.
+          attr_reader :keepalive_monitor
 
           # Initial backoff delay in seconds when reconnecting after a transient stream disconnection.
           INITIAL_RECONNECT_DELAY = 1.0
@@ -97,15 +85,25 @@ def initialize subscriber
 
             @callback_thread_pool = Concurrent::ThreadPoolExecutor.new max_threads: @subscriber.callback_threads
 
-            @keepalive_interval = DEFAULT_KEEPALIVE_INTERVAL
-            @pong_deadline = DEFAULT_PONG_DEADLINE
-            @last_ping_at = nil
-            @last_pong_at = nil
+            @keepalive_monitor = KeepaliveMonitor.new self
             @stream_opened = false
             @reconnect_delay = nil
+          end
+
+          def keepalive_interval
+            @keepalive_monitor.interval
+          end
+
+          def keepalive_interval= val
+            @keepalive_monitor.interval = val
+          end
+
+          def pong_deadline
+            @keepalive_monitor.deadline
+          end
 
-            @stream_keepalive_task = nil
-            @pong_monitor_task = nil
+          def pong_deadline= val
+            @keepalive_monitor.deadline = val
           end
 
           def start
@@ -113,17 +111,7 @@ def start
               break if @background_thread
 
               @inventory.start
-              @stream_keepalive_task = Concurrent::TimerTask.new(
-                execution_interval: @keepalive_interval
-              ) { send_keepalive_ping! }
-              @stream_keepalive_task.execute
-
-              # Check liveness at 1/5th of keepalive interval to detect timeouts promptly without waiting a full cycle.
-              # Floor interval at 0.01s (10ms) to prevent thread spinning and CPU exhaustion during sub-second tests.
-              @pong_monitor_task = Concurrent::TimerTask.new(
-                execution_interval: [@keepalive_interval / PONG_MONITOR_INTERVAL_DIVISOR, PONG_MONITOR_MIN_INTERVAL].max
-              ) { check_liveness! }
-              @pong_monitor_task.execute
+              @keepalive_monitor.start
 
               start_streaming!
             end
@@ -146,10 +134,7 @@ def stop
               @stopped = true
               @pause_cond.broadcast
 
-              @stream_keepalive_task&.shutdown
-              @stream_keepalive_task = nil
-              @pong_monitor_task&.shutdown
-              @pong_monitor_task = nil
+              @keepalive_monitor.stop
 
               # Now that the reception thread is stopped, immediately stop the
               # callback thread pool. All queued callbacks will see the stream
@@ -182,6 +167,55 @@ def wait! timeout = nil
             self
           end
 
+          def stream_opened?
+            @stream_opened
+          end
+
+          def stopped?
+            @stopped
+          end
+
+          def paused?
+            @paused
+          end
+
+          def request_queue_active?
+            !@request_queue.nil?
+          end
+
+          def send_ping_request!
+            push Google::Cloud::PubSub::V1::StreamingPullRequest.new
+          end
+
+          def restart_stream_for_timeout!
+            @stream_opened = false
+            # Push self as a stream-closing sentinel to @request_queue.
+            # When EnumeratorQueue#each pops the sentinel object, it terminates the request enumerator,
+            # cleanly sending an HTTP/2 END_STREAM flag and unblocking the write-side gRPC C-core pipeline.
+            @request_queue&.push self
+            @background_thread&.raise RestartStream
+          end
+
+          def log_info msg
+            subscriber.service.logger.log :info, "subscriber-streams" do
+              msg
+            end
+          end
+
+          def log_error msg
+            subscriber.service.logger.log :error, "subscriber-streams" do
+              msg
+            end
+          end
+
+          def stream_keepalive_task
+            @keepalive_monitor.ping_task
+          end
+
+          def pong_monitor_task
+            @keepalive_monitor.monitor_task
+          end
+
           ##
           # @private
           def acknowledge *messages, &callback
@@ -309,12 +343,10 @@ def background_run
 
             # Once the stream handshake completes, initialize ping/pong timestamps to monotonic now
             # and mark @stream_opened = true under synchronization.
-            # This ensures @pong_monitor_task evaluates liveness against a fresh window and prevents
+            # This ensures @keepalive_monitor evaluates liveness against a fresh window and prevents
             # false-positive disconnect detections from stale timestamps prior to reconnecting.
             synchronize do
-              now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
-              @last_ping_at = now
-              @last_pong_at = now
+              @keepalive_monitor.record_handshake!
               @stream_opened = true
             end
 
@@ -333,7 +365,7 @@ def background_run
                 # Cannot synchronize the enumerator, causes deadlock
                 response = enum.next
                 synchronize do
-                  @last_pong_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+                  @keepalive_monitor.record_pong!
                   # Reset backoff delay only after successfully reading a frame from enum.next.
                   # If the connection drops immediately upon reading, @reconnect_delay is preserved.
                   @reconnect_delay = nil
@@ -497,43 +529,11 @@ def pause_streaming?
           end
 
           def send_keepalive_ping!
-            synchronize do
-              # Push unconditional keep-alive pings only when the stream is actively open and @request_queue is initialized.
-              # Note: ACKs are sent via unary RPCs (TimedUnaryBuffer), not over this stream.
-              if @stream_opened && !@stopped && @request_queue
-                subscriber.service.logger.log :info, "subscriber-streams" do
-                  "sending keepAlive to stream for subscription #{@subscriber.subscription_name}"
-                end
-                # Only advance @last_ping_at if the previous ping was successfully ponged.
-                # If a pong is outstanding (@last_pong_at < @last_ping_at), freezing @last_ping_at preserves the start time
-                # of the un-ponged cycle so check_liveness! can accurately evaluate the elapsed deadline.
-                @last_ping_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) if @last_pong_at >= @last_ping_at
-                push Google::Cloud::PubSub::V1::StreamingPullRequest.new
-              end
-            end
+            @keepalive_monitor.send_ping!
           end
 
           def check_liveness!
-            synchronize do
-              # Do not check pong deadline if @paused (client flow control inventory full).
-              # When @paused, background_run waits on condition variable and stops calling enum.next,
-              # so incoming server pongs sit buffered in gRPC and @last_pong_at stays un-updated.
-              if @stream_opened && @last_ping_at && @last_pong_at && !@stopped && !@paused
-                now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
-                if now - @last_ping_at >= @pong_deadline && @last_pong_at < @last_ping_at
-                  elapsed_pong = (now - @last_pong_at).round(2)
-                  subscriber.service.logger.log :error, "subscriber-streams" do
-                    "Keep-alive pong not received within #{@pong_deadline}s (last pong #{elapsed_pong}s ago, inventory: #{@inventory.count}, paused: #{@paused.inspect}); restarting stream."
-                  end
-                  @stream_opened = false
-                  # Push self as a stream-closing sentinel to @request_queue.
-                  # When EnumeratorQueue#each pops the sentinel object, it terminates the request enumerator,
-                  # cleanly sending an HTTP/2 END_STREAM flag and unblocking the write-side gRPC C-core pipeline.
-                  @request_queue&.push self
-                  @background_thread&.raise RestartStream
-                end
-              end
-            end
+            @keepalive_monitor.check_liveness!
           end
 
           def unpause_streaming!
@@ -541,10 +541,10 @@ def unpause_streaming!
               return unless unpause_streaming?
 
               @paused = nil
-              # Reset @last_pong_at when unpausing flow control. While paused, incoming server pongs sit buffered in gRPC,
-              # leaving @last_pong_at stale. Updating timestamp guarantees the monitor thread will not trigger an immediate
+              # Record pong when unpausing flow control. While paused, incoming server pongs sit buffered in gRPC,
+              # leaving last_pong_at stale. Updating timestamp guarantees the monitor thread will not trigger an immediate
               # false-positive restart while the reader thread wakes up to drain buffered frames.
-              @last_pong_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+              @keepalive_monitor.record_pong!
               subscriber.service.logger.log :info, "subscriber-flow-control" do
                 "subscriber for #{@subscriber.subscription_name} is unblocking client-side flow control"
               end
diff --git a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
index e65919c6fd55..9dbfc253ad3e 100644
--- a/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
+++ b/google-cloud-pubsub/test/google/cloud/pubsub/message_listener/keepalive_test.rb
@@ -87,11 +87,11 @@
     stream.instance_variable_set :@request_queue, queue
     stream.instance_variable_set :@stream_opened, true
     stream.instance_variable_set :@stopped, false
-    stream.instance_variable_set :@last_ping_at, 0.0
-    stream.instance_variable_set :@last_pong_at, 1.0
+    stream.keepalive_monitor.last_ping_at = 0.0
+    stream.keepalive_monitor.last_pong_at = 1.0
 
     stream.send(:send_keepalive_ping!)
-    _(stream.instance_variable_get(:@last_ping_at)).must_be :>, 0.0
+    _(stream.keepalive_monitor.last_ping_at).must_be :>, 0.0
   end
 
   it "does not restart stream when check_liveness! runs under active pongs" do
@@ -106,11 +106,11 @@
     stream.instance_variable_set :@stream_opened, true
     stream.instance_variable_set :@stopped, false
     stream.instance_variable_set :@paused, false
-    stream.instance_variable_set :@last_ping_at, now - 5.0
-    stream.instance_variable_set :@last_pong_at, now - 1.0
+    stream.keepalive_monitor.last_ping_at = now - 5.0
+    stream.keepalive_monitor.last_pong_at = now - 1.0
 
     stream.send(:check_liveness!)
-    _(stream.instance_variable_get(:@stream_opened)).must_equal true
+    _(stream.stream_opened?).must_equal true
   end
 
   it "does not trigger false restarts on unpausing even if pause exceeded deadline" do
@@ -123,23 +123,23 @@
 
     now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
     stream.instance_variable_set :@stream_opened, true
-    stream.instance_variable_set :@last_ping_at, now - 30.0
-    stream.instance_variable_set :@last_pong_at, now - 35.0
+    stream.keepalive_monitor.last_ping_at = now - 30.0
+    stream.keepalive_monitor.last_pong_at = now - 35.0
     stream.instance_variable_set :@paused, true
     stream.instance_variable_set :@stopped, false
 
     # 1. When paused, liveness checker should skip verification.
     stream.send(:check_liveness!)
-    _(stream.instance_variable_get(:@stream_opened)).must_equal true
+    _(stream.stream_opened?).must_equal true
 
     # 2. Unpausing must reset @last_pong_at to prevent immediate restart.
     stream.send(:unpause_streaming!)
-    _(stream.instance_variable_get(:@last_pong_at)).must_be :>=, now
+    _(stream.keepalive_monitor.last_pong_at).must_be :>=, now
 
     # 3. Direct liveness check immediately after unpausing (simulates the monitor thread racing the background thread).
     # It should NOT close the stream because our reset made @last_pong_at newer than @last_ping_at.
     stream.send(:check_liveness!)
-    _(stream.instance_variable_get(:@stream_opened)).must_equal true
+    _(stream.stream_opened?).must_equal true
   end
 
   it "re-creates and executes timer tasks if stopped and restarted" do
@@ -153,17 +153,17 @@
     stream = listener.instance_variable_get(:@stream_pool).first
 
     # 1. Timers should be active on a started stream
-    refute_nil stream.instance_variable_get(:@stream_keepalive_task)
-    refute_nil stream.instance_variable_get(:@pong_monitor_task)
+    refute_nil stream.stream_keepalive_task
+    refute_nil stream.pong_monitor_task
 
-    first_keepalive = stream.instance_variable_get(:@stream_keepalive_task)
-    first_monitor = stream.instance_variable_get(:@pong_monitor_task)
+    first_keepalive = stream.stream_keepalive_task
+    first_monitor = stream.pong_monitor_task
 
     # 2. Stopping the stream must shutdown and nilify the timers
     listener.stop
     listener.wait!
-    assert_nil stream.instance_variable_get(:@stream_keepalive_task)
-    assert_nil stream.instance_variable_get(:@pong_monitor_task)
+    assert_nil stream.stream_keepalive_task
+    assert_nil stream.pong_monitor_task
     assert first_keepalive.shutdown?
     assert first_monitor.shutdown?
 
@@ -172,8 +172,8 @@
     stream.instance_variable_set :@stopped, false
     stream.start
 
-    new_keepalive = stream.instance_variable_get(:@stream_keepalive_task)
-    new_monitor = stream.instance_variable_get(:@pong_monitor_task)
+    new_keepalive = stream.stream_keepalive_task
+    new_monitor = stream.pong_monitor_task
 
     refute_nil new_keepalive
     refute_nil new_monitor

From 2dc081a7d66f3ad33adcffa331df66a3eafd3e58 Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Wed, 8 Jul 2026 16:42:37 +0000
Subject: [PATCH 12/13] docs(pubsub): restore line-by-line attr definitions and
 comments on KeepaliveMonitor 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.
---
 .../pubsub/message_listener/keepalive_monitor.rb  | 15 +++++++++++++--
 .../cloud/pubsub/message_listener/stream.rb       |  2 ++
 2 files changed, 15 insertions(+), 2 deletions(-)

diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb
index 535620e75921..2e7558ded535 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb
@@ -27,8 +27,19 @@ class KeepaliveMonitor
           MONITOR_DIVISOR = 5.0
           MIN_MONITOR_INTERVAL = 0.01
 
-          attr_accessor :interval, :deadline, :last_ping_at, :last_pong_at
-          attr_reader :ping_task, :monitor_task
+          # keepalive_interval used to send pings and maintain a healthy stream
+          attr_accessor :interval
+          # the interval used to validate whether we received a pong back from the stream
+          attr_accessor :deadline
+          # timestamp of the last ping sent
+          attr_accessor :last_ping_at
+          # timestamp of the last pong received
+          attr_accessor :last_pong_at
+
+          # background timer task for sending keepalive pings
+          attr_reader :ping_task
+          # background timer task for monitoring stream liveness
+          attr_reader :monitor_task
 
           def initialize stream, interval: DEFAULT_INTERVAL, deadline: DEFAULT_DEADLINE
             @stream       = stream
diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
index 9aee6e761fb4..bd5a12e6d7d2 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/stream.rb
@@ -90,6 +90,7 @@ def initialize subscriber
             @reconnect_delay = nil
           end
 
+          # keepalive_interval used to send pings and maintain a healthy stream
           def keepalive_interval
             @keepalive_monitor.interval
           end
@@ -98,6 +99,7 @@ def keepalive_interval= val
             @keepalive_monitor.interval = val
           end
 
+          # the interval used to validate whether we received a pong back from the stream
           def pong_deadline
             @keepalive_monitor.deadline
           end

From 7e22f63c2043c537f226700855e9cbe08664cebe Mon Sep 17 00:00:00 2001
From: Torrey Payne <11740989+torreypayne@users.noreply.github.com>
Date: Wed, 8 Jul 2026 17:47:46 +0000
Subject: [PATCH 13/13] docs(pubsub): restore concurrency and constant comments
 in KeepaliveMonitor

- 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!.
---
 .../pubsub/message_listener/keepalive_monitor.rb     | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb
index 2e7558ded535..27bfce2e8573 100644
--- a/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb
+++ b/google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/keepalive_monitor.rb
@@ -22,9 +22,13 @@ class MessageListener
         # @private
         # Monitors gRPC streaming connection health via periodic bi-directional keep-alive pings and pongs.
         class KeepaliveMonitor
+          # Default interval in seconds between keep-alive ping requests.
           DEFAULT_INTERVAL = 30.0
+          # Default deadline in seconds to receive a keep-alive pong response.
           DEFAULT_DEADLINE = 15.0
+          # Divisor applied to keep-alive interval to calculate monitor polling interval (1/5th of interval).
           MONITOR_DIVISOR = 5.0
+          # Minimum floor in seconds (10ms) for the monitor polling interval to prevent CPU spinning.
           MIN_MONITOR_INTERVAL = 0.01
 
           # keepalive_interval used to send pings and maintain a healthy stream
@@ -96,9 +100,14 @@ def record_pong!
 
           def send_ping!
             @stream.synchronize do
+              # Push unconditional keep-alive pings only when the stream is actively open and request queue is active.
+              # Note: ACKs are sent via unary RPCs (TimedUnaryBuffer), not over this stream.
               return unless @stream.stream_opened? && !@stream.stopped? && @stream.request_queue_active?
 
               @stream.log_info "sending keepAlive to stream for subscription #{@stream.subscriber.subscription_name}"
+              # Only advance @last_ping_at if the previous ping was successfully ponged.
+              # If a pong is outstanding (@last_pong_at < @last_ping_at), freezing @last_ping_at preserves the start time
+              # of the un-ponged cycle so check_liveness! can accurately evaluate the elapsed deadline.
               @last_ping_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) if @last_pong_at >= @last_ping_at
               @stream.send_ping_request!
             end
@@ -106,6 +115,9 @@ def send_ping!
 
           def check_liveness!
             @stream.synchronize do
+              # Do not check pong deadline if paused (client flow control inventory full).
+              # When paused, background_run waits on condition variable and stops calling enum.next,
+              # so incoming server pongs sit buffered in gRPC and last_pong_at stays un-updated.
               return unless @stream.stream_opened? && @last_ping_at && @last_pong_at && !@stream.stopped? && !@stream.paused?
 
               now = Process.clock_gettime(Process::CLOCK_MONOTONIC)