feat(pubsub): implement streaming keep-alive logic

google-cloud-pubsub/acceptance/pubsub/keepalive_acceptance_test.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 "pubsub_helper"
+
+describe Google::Cloud::PubSub, :keepalive_acceptance, :pubsub do
+  def retrieve_topic topic_name
+    topic_path = pubsub.topic_path topic_name
+    $topic_admin.get_topic(topic: topic_path) rescue $topic_admin.create_topic(name: topic_path)
+  end
+
+  def retrieve_subscription topic, subscription_name
+    subscription_path = pubsub.subscription_path subscription_name
+    $subscription_admin.get_subscription(subscription: subscription_path) \
+      rescue $subscription_admin.create_subscription(name: subscription_path, topic: topic.name)
+  end
+
+  let(:nonce) { rand 1000 }
+  let(:topic) { retrieve_topic "#{$topic_prefix}-ka-topic#{nonce}" }
+  let(:sub) { retrieve_subscription topic, "#{$topic_prefix}-ka-sub#{nonce}" }
+
+  it "maintains real gRPC keep-alive heartbeat and delivers messages from live Google Cloud Pub/Sub servers" do
+    subscriber = pubsub.subscriber sub.name
+
+    messages_received = []
+    listener = subscriber.listen streams: 1 do |msg|
+      messages_received << msg
+      msg.ack!
+    end
+
+    stream = listener.instance_variable_get(:@stream_pool).first
+    # Set a short keep-alive interval (2.0s) so multiple ping/pong cycles occur against live GCP servers
+    stream.keepalive_interval = 2.0
+    stream.pong_deadline = 10.0
+
+    listener.start
+
+    # Publish a live message to the real topic on GCP
+    publisher = pubsub.publisher topic.name
+    publish_result = nil
+    publisher.publish_async "keepalive-production-acceptance-test-#{nonce}" do |result|
+      publish_result = result
+    end
+
+    pub_retries = 0
+    until publish_result
+      fail "Live publish failed against Google Cloud Pub/Sub" if pub_retries >= 30
+      pub_retries += 1
+      sleep 0.2
+    end
+    _(publish_result).must_be :succeeded?
+
+    # Wait for the message to be received by our streaming listener from live GCP
+    sub_retries = 0
+    until messages_received.any?
+      fail "Listener did not receive published message from live Google Cloud Pub/Sub" if sub_retries >= 50
+      sub_retries += 1
+      sleep 0.2
+    end
+
+    received = messages_received.first
+    _(received.data).must_equal "keepalive-production-acceptance-test-#{nonce}"
+
+    # Allow at least 2 full keep-alive intervals (4+ seconds) to run over live production wire
+    sleep 4.5
+
+    monitor = stream.keepalive_monitor
+    _(stream.stream_opened?).must_equal true
+    _(monitor.last_ping_at).wont_be_nil
+    _(monitor.last_pong_at).wont_be_nil
+    _(monitor.last_ping_at).must_be :>, 0.0
+    _(monitor.last_pong_at).must_be :>, 0.0
+
+    listener.stop
+    listener.wait!
+  end
+
+  it "safely unpauses flow control under real network conditions against live Google Cloud Pub/Sub servers" do
+    subscriber = pubsub.subscriber sub.name
+
+    listener = subscriber.listen streams: 1 do |msg|
+      msg.ack!
+    end
+
+    stream = listener.instance_variable_get(:@stream_pool).first
+    monitor = stream.keepalive_monitor
+
+    listener.start
+
+    # Allow initial connection handshake against real GCP to establish
+    sleep 1.0
+    _(stream.stream_opened?).must_equal true
+
+    now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    # Validate unpause_streaming! atomically updates last_pong_at against real live listener thread
+    stream.send(:unpause_streaming!)
+    _(monitor.last_pong_at).must_be :>=, now
+    _(stream.stream_opened?).must_equal true
+
+    listener.stop
+    listener.wait!
+  end
+end

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

@@ -0,0 +1,137 @@
+# 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 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
+          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
+            @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
+              # 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
+          end
+
+          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)
+              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