feat(pubsub): add HedgingOptions and fixed-point HedgingTokenBucket

google-cloud-pubsub/lib/google/cloud/pubsub/async_publisher/hedging_options.rb

@@ -0,0 +1,174 @@
+# 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.
+
+
+module Google
+  module Cloud
+    module PubSub
+      class AsyncPublisher
+        ##
+        # @private
+        #
+        # Configuration options and validation invariants for Publish Hedging.
+        #
+        class HedgingOptions
+          DEFAULT_ENABLED = true
+          DEFAULT_HEDGE_DELAY = 1.0
+          DEFAULT_MAX_TOKENS = 50
+          DEFAULT_REFILL_RATIO = 0.1
+
+          MIN_HEDGE_DELAY = 0.1
+          MAX_HEDGE_DELAY = 10.0
+
+          MIN_MAX_TOKENS = 1
+          MAX_MAX_TOKENS = 250
+
+          MIN_REFILL_RATIO = 0.001
+          MAX_REFILL_RATIO = 0.2
+
+          attr_reader :hedge_delay
+          attr_reader :max_tokens
+          attr_reader :refill_ratio
+
+          def initialize enabled: DEFAULT_ENABLED,
+                         hedge_delay: DEFAULT_HEDGE_DELAY,
+                         max_tokens: DEFAULT_MAX_TOKENS,
+                         refill_ratio: DEFAULT_REFILL_RATIO
+            @enabled = enabled.nil? ? DEFAULT_ENABLED : validate_boolean!(enabled)
+            @hedge_delay = hedge_delay.nil? ? DEFAULT_HEDGE_DELAY : hedge_delay
+            @max_tokens = max_tokens.nil? ? DEFAULT_MAX_TOKENS : max_tokens
+            @refill_ratio = refill_ratio.nil? ? DEFAULT_REFILL_RATIO : refill_ratio
+
+            validate_bounds! if @enabled
+          end
+
+          def enabled?
+            @enabled
+          end
+
+          ##
+          # Resolves and validates hedging options from a Hash, HedgingOptions, or nil.
+          # Returns nil if hedging is not configured or explicitly disabled.
+          #
+          # @param [Hash, HedgingOptions, Boolean, nil] options
+          # @param [Boolean] message_ordering Whether message ordering is enabled.
+          # @param [Numeric, nil] timeout Effective publish RPC timeout in seconds.
+          # @return [HedgingOptions, nil]
+          #
+          def self.from options, message_ordering: false, timeout: nil
+            return nil if options.nil? || options == false
+
+            resolved = case options
+                       when HedgingOptions
+                         options
+                       when true
+                         new
+                       when Hash
+                         from_hash options
+                       else
+                         raise ArgumentError, "hedging must be a Hash or HedgingOptions"
+                       end
+
+            return nil unless resolved.enabled?
+
+            resolved.validate_message_ordering! message_ordering
+            resolved.validate_timeout! timeout
+            resolved
+          end
+
+          def validate_message_ordering! message_ordering
+            return unless @enabled && message_ordering
+
+            raise ArgumentError, "Publish hedging and message ordering cannot be enabled at the same time."
+          end
+
+          def validate_timeout! timeout
+            return unless @enabled && timeout.is_a?(Numeric)
+            return if @hedge_delay < timeout
+
+            raise ArgumentError,
+                  "hedge_delay (#{@hedge_delay}s) must be strictly less than publish timeout (#{timeout}s)"
+          end
+
+          def self.from_hash hash
+            new(
+              enabled: fetch_opt(hash, :enabled, DEFAULT_ENABLED),
+              hedge_delay: fetch_opt(hash, :hedge_delay),
+              max_tokens: fetch_opt(hash, :max_tokens),
+              refill_ratio: fetch_opt(hash, :refill_ratio)
+            )
+          end
+          private_class_method :from_hash
+
+          def self.fetch_opt hash, key, default = nil
+            if hash.key? key
+              hash[key]
+            elsif hash.key? key.to_s
+              hash[key.to_s]
+            else
+              default
+            end
+          end
+          private_class_method :fetch_opt
+
+          private
+
+          def validate_boolean! value
+            return value if value == true || value == false
+
+            raise ArgumentError, "enabled must be a boolean"
+          end
+
+          def validate_bounds!
+            validate_hedge_delay!
+            validate_max_tokens!
+            validate_refill_ratio!
+          end
+
+          def validate_hedge_delay!
+            unless @hedge_delay.is_a?(Numeric) && !nan_or_infinite?(@hedge_delay) &&
+                   @hedge_delay >= MIN_HEDGE_DELAY && @hedge_delay <= MAX_HEDGE_DELAY
+              raise ArgumentError,
+                    "hedge_delay must be between #{MIN_HEDGE_DELAY} and #{MAX_HEDGE_DELAY} seconds"
+            end
+            @hedge_delay = @hedge_delay.to_f
+          end
+
+          def validate_max_tokens!
+            unless @max_tokens.is_a?(Integer) &&
+                   @max_tokens >= MIN_MAX_TOKENS && @max_tokens <= MAX_MAX_TOKENS
+              raise ArgumentError,
+                    "max_tokens must be an Integer between #{MIN_MAX_TOKENS} and #{MAX_MAX_TOKENS}"
+            end
+          end
+
+          def validate_refill_ratio!
+            unless @refill_ratio.is_a?(Numeric) && !nan_or_infinite?(@refill_ratio) &&
+                   @refill_ratio >= MIN_REFILL_RATIO && @refill_ratio <= MAX_REFILL_RATIO
+              raise ArgumentError,
+                    "refill_ratio must be between #{MIN_REFILL_RATIO} and #{MAX_REFILL_RATIO}"
+            end
+            @refill_ratio = @refill_ratio.to_f
+          end
+
+          def nan_or_infinite? num
+            num.respond_to?(:nan?) && (num.nan? || num.infinite?)
+          end
+        end
+      end
+    end
+
+    Pubsub = PubSub unless const_defined? :Pubsub
+  end
+end

google-cloud-pubsub/lib/google/cloud/pubsub/async_publisher/hedging_token_bucket.rb

@@ -0,0 +1,102 @@
+# 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 "google/cloud/pubsub/async_publisher/hedging_options"
+
+module Google
+  module Cloud
+    module PubSub
+      class AsyncPublisher
+        ##
+        # @private
+        #
+        # Thread-safe client-side token bucket rate limiter for Publish Hedging.
+        # Uses fixed-point integer arithmetic scaled by 1,000 to avoid floating-point
+        # rounding drift on fractional refill ratios down to 0.001.
+        #
+        class HedgingTokenBucket
+          HEDGE_TOKEN_SCALE = 1000
+
+          attr_reader :options
+          attr_reader :hedge_delay
+          attr_reader :max_tokens
+          attr_reader :refill_ratio
+
+          def initialize options = {}
+            @options = options.is_a?(HedgingOptions) ? options : HedgingOptions.from(options)
+            raise ArgumentError, "HedgingOptions must be enabled" if @options.nil?
+
+            @hedge_delay = @options.hedge_delay
+            @max_tokens = @options.max_tokens
+            @refill_ratio = @options.refill_ratio
+
+            @scaled_tokens = 0
+            @scaled_max_tokens = @max_tokens * HEDGE_TOKEN_SCALE
+            @scaled_refill_amount = (@refill_ratio * HEDGE_TOKEN_SCALE).round
+            @mutex = Mutex.new
+          end
+
+          ##
+          # Attempts to deduct 1.0 full token (1,000 scaled units) from the bucket.
+          #
+          # @return [Boolean] true if a token was acquired, false otherwise.
+          #
+          def try_acquire_hedge_token
+            @mutex.synchronize do
+              return false if @scaled_tokens < HEDGE_TOKEN_SCALE
+
+              @scaled_tokens -= HEDGE_TOKEN_SCALE
+              true
+            end
+          end
+
+          ##
+          # Credits refill_ratio tokens to the bucket, capped at max_tokens.
+          #
+          # @return [Float] Updated logical token count.
+          #
+          def refill
+            @mutex.synchronize do
+              @scaled_tokens = [@scaled_tokens + @scaled_refill_amount, @scaled_max_tokens].min
+              @scaled_tokens.to_f / HEDGE_TOKEN_SCALE
+            end
+          end
+
+          ##
+          # Returns the current logical token balance.
+          #
+          # @return [Float]
+          #
+          def tokens
+            @mutex.synchronize do
+              @scaled_tokens.to_f / HEDGE_TOKEN_SCALE
+            end
+          end
+
+          ##
+          # Returns the raw fixed-point scaled token balance (1,000 = 1.0 token).
+          #
+          # @return [Integer]
+          #
+          def scaled_tokens
+            @mutex.synchronize { @scaled_tokens }
+          end
+        end
+      end
+    end
+
+    Pubsub = PubSub unless const_defined? :Pubsub
+  end
+end