feat(pubsub): implement streaming keep-alive logic

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

@@ -49,6 +49,35 @@ 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
+
+          # 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
@@ -68,24 +97,33 @@ def initialize subscriber
 
             @callback_thread_pool = Concurrent::ThreadPoolExecutor.new max_threads: @subscriber.callback_threads
 
-            @stream_keepalive_task = Concurrent::TimerTask.new(
-              execution_interval: 30
-            ) 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}"
-                end
-                push Google::Cloud::PubSub::V1::StreamingPullRequest.new
-              end
-            end.execute
+            @keepalive_interval = DEFAULT_KEEPALIVE_INTERVAL
+            @pong_deadline = DEFAULT_PONG_DEADLINE
+            @last_ping_at = nil
+            @last_pong_at = nil
+            @stream_opened = false
+            @reconnect_delay = nil
+
+            @stream_keepalive_task = nil
+            @pong_monitor_task = nil
           end
 
           def start
             synchronize do
               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
 
               start_streaming!
             end
@@ -108,6 +146,11 @@ def stop
               @stopped = true
               @pause_cond.broadcast
 
+              @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
               # is stopped and perform a noop.
@@ -219,6 +262,15 @@ class RestartStream < StandardError; end
 
           # rubocop:disable all
 
+          def backoff_and_wait!
+            @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
+              @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 +297,27 @@ 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
             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
 
+            # 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
+              @last_pong_at = now
+              @stream_opened = true
+            end
+
             loop do
               synchronize do
                 if @paused && !@stopped
@@ -264,8 +332,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,18 +387,21 @@ 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
               "error on stream for subscription #{@subscriber.subscription_name}: #{e.inspect}"
             end
             @subscriber.error! e
 
+            backoff_and_wait!
             retry
           end
 
@@ -416,22 +496,68 @@ def pause_streaming?
             @inventory.full?
           end
 
-          def unpause_streaming!
-            return unless unpause_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
+            end
+          end
 
-            @paused = nil
-            subscriber.service.logger.log :info, "subscriber-flow-control" do
-              "subscriber for #{@subscriber.subscription_name} is unblocking client-side flow control"
+          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
+          end
+
+          def unpause_streaming!
+            synchronize do
+              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"
+              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?
             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
@@ -443,6 +569,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
 

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
       )]
     ]
 

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 "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 "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? || false}"
+    _(keepalive_task.nil? || !keepalive_task.running?).must_equal true
+  end
+end

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
       )]
     ]