From e4a7ef508e14a2d268cf363b2591c4e7cd66a03f Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Fri, 15 May 2026 18:57:33 -0400 Subject: [PATCH 01/14] Implement status-response retry improvements - Transport: dual-path retry loop (429+Retry-After vs counted exponential backoff), X-Retry-Count header on retries, retryable status classification (5xx except 501/505/511; 4xx only 408/410/429/460) - BackoffPolicy: update constants (base 500ms, cap 60s, multiplier 2); add reset! method - Response: add success? method (2xx+3xx) - Defaults: add MAX_TOTAL_BACKOFF_DURATION, MAX_RATE_LIMIT_DURATION, RATE_LIMIT_RETRY_AFTER_CAP constants - Worker: use response.success? instead of status == 200 - Tests: cover new retry paths, X-Retry-Count, parse_retry_after, retryable/non-retryable status codes --- lib/segment/analytics/backoff_policy.rb | 4 + lib/segment/analytics/defaults.rb | 9 +- lib/segment/analytics/response.rb | 4 + lib/segment/analytics/transport.rb | 152 +++++++++++------- lib/segment/analytics/worker.rb | 2 +- spec/segment/analytics/backoff_policy_spec.rb | 22 +++ spec/segment/analytics/response_spec.rb | 12 ++ spec/segment/analytics/transport_spec.rb | 118 +++++++++++++- 8 files changed, 263 insertions(+), 60 deletions(-) diff --git a/lib/segment/analytics/backoff_policy.rb b/lib/segment/analytics/backoff_policy.rb index e6033b1..7767838 100644 --- a/lib/segment/analytics/backoff_policy.rb +++ b/lib/segment/analytics/backoff_policy.rb @@ -33,6 +33,10 @@ def next_interval [interval, @max_timeout_ms].min end + def reset! + @attempts = 0 + end + private def add_jitter(base, randomization_factor) diff --git a/lib/segment/analytics/defaults.rb b/lib/segment/analytics/defaults.rb index aa32697..e443caf 100644 --- a/lib/segment/analytics/defaults.rb +++ b/lib/segment/analytics/defaults.rb @@ -12,6 +12,9 @@ module Request 'Content-Type' => 'application/json', 'User-Agent' => "analytics-ruby/#{Analytics::VERSION}" } RETRIES = 10 + MAX_TOTAL_BACKOFF_DURATION = 43_200 # 12 hours in seconds + MAX_RATE_LIMIT_DURATION = 43_200 # 12 hours in seconds + RATE_LIMIT_RETRY_AFTER_CAP = 300 # seconds end module Queue @@ -28,9 +31,9 @@ module MessageBatch end module BackoffPolicy - MIN_TIMEOUT_MS = 100 - MAX_TIMEOUT_MS = 10000 - MULTIPLIER = 1.5 + MIN_TIMEOUT_MS = 500 + MAX_TIMEOUT_MS = 60_000 + MULTIPLIER = 2 RANDOMIZATION_FACTOR = 0.5 end end diff --git a/lib/segment/analytics/response.rb b/lib/segment/analytics/response.rb index c31116a..e4c2bc1 100644 --- a/lib/segment/analytics/response.rb +++ b/lib/segment/analytics/response.rb @@ -12,6 +12,10 @@ def initialize(status = 200, error = nil) @status = status @error = error end + + def success? + status >= 200 && status < 400 + end end end end diff --git a/lib/segment/analytics/transport.rb b/lib/segment/analytics/transport.rb index 6ee14d8..2f0e2a6 100644 --- a/lib/segment/analytics/transport.rb +++ b/lib/segment/analytics/transport.rb @@ -16,16 +16,26 @@ class Transport include Segment::Analytics::Utils include Segment::Analytics::Logging + RETRYABLE_4XX = [408, 410, 429, 460].freeze + NON_RETRYABLE_5XX = [501, 505, 511].freeze + def initialize(options = {}) options[:host] ||= HOST options[:port] ||= PORT - options[:ssl] ||= SSL + options[:ssl] ||= SSL @headers = options[:headers] || HEADERS - @path = options[:path] || PATH + @path = options[:path] || PATH @retries = options[:retries] || RETRIES @backoff_policy = options[:backoff_policy] || Segment::Analytics::BackoffPolicy.new + @max_total_backoff_duration = options[:max_total_backoff_duration] || + MAX_TOTAL_BACKOFF_DURATION + @max_rate_limit_duration = options[:max_rate_limit_duration] || + MAX_RATE_LIMIT_DURATION + @rate_limit_retry_after_cap = options[:rate_limit_retry_after_cap] || + RATE_LIMIT_RETRY_AFTER_CAP + http = Net::HTTP.new(options[:host], options[:port]) http.use_ssl = options[:ssl] http.read_timeout = 8 @@ -40,23 +50,68 @@ def initialize(options = {}) def send(write_key, batch) logger.debug("Sending request for #{batch.length} items") - last_response, exception = retry_with_backoff(@retries) do - status_code, body = send_request(write_key, batch) - error = JSON.parse(body)['error'] - should_retry = should_retry_request?(status_code, body) + @backoff_policy.reset! + + retry_count = 0 + retries_remaining = @retries + backoff_start_time = nil + rate_limit_start_time = nil + + loop do + status_code, body, response_headers = send_request(write_key, batch, retry_count) + error = begin + JSON.parse(body)['error'] + rescue StandardError + nil + end logger.debug("Response status code: #{status_code}") logger.debug("Response error: #{error}") if error - [Response.new(status_code, error), should_retry] - end - - if exception - logger.error(exception.message) - exception.backtrace.each { |line| logger.error(line) } - Response.new(-1, exception.to_s) - else - last_response + return Response.new(status_code, error) if success_status?(status_code) + + if status_code == 429 + rate_limit_start_time ||= Time.now + if (Time.now - rate_limit_start_time) >= @max_rate_limit_duration + logger.error('Max rate limit duration exceeded for batch') + return Response.new(status_code, error) + end + + retry_after = parse_retry_after(response_headers['retry-after']) + if retry_after + delay = [retry_after, @rate_limit_retry_after_cap].min + logger.debug("Rate limited with Retry-After: #{delay}s. Retrying after delay.") + sleep(delay) + retry_count += 1 + next + end + end + + unless retryable_status?(status_code) + logger.error(body) + return Response.new(status_code, error) + end + + retries_remaining -= 1 + if retries_remaining <= 0 + logger.error('Retries exhausted for batch') + return Response.new(status_code, error) + end + + backoff_start_time ||= Time.now + if (Time.now - backoff_start_time) >= @max_total_backoff_duration + logger.error('Max total backoff duration exceeded for batch') + return Response.new(status_code, error) + end + + delay_ms = @backoff_policy.next_interval + logger.debug("Retrying request, #{retries_remaining} retries left. Waiting #{delay_ms}ms") + sleep(delay_ms.to_f / 1000) + retry_count += 1 end + rescue StandardError => e + logger.error(e.message) + e.backtrace.each { |line| logger.error(line) } + Response.new(-1, e.to_s) end # Closes a persistent connection if it exists @@ -66,65 +121,52 @@ def shutdown private - def should_retry_request?(status_code, body) - if status_code >= 500 - true # Server error - elsif status_code == 429 - true # Rate limited - elsif status_code >= 400 - logger.error(body) - false # Client error. Do not retry, but log + def success_status?(code) + code >= 200 && code < 400 + end + + def retryable_status?(code) + if code >= 500 && code < 600 + !NON_RETRYABLE_5XX.include?(code) else - false + RETRYABLE_4XX.include?(code) end end - # Takes a block that returns [result, should_retry]. - # - # Retries upto `retries_remaining` times, if `should_retry` is false or - # an exception is raised. `@backoff_policy` is used to determine the - # duration to sleep between attempts - # - # Returns [last_result, raised_exception] - def retry_with_backoff(retries_remaining, &block) - result, caught_exception = nil - should_retry = false - - begin - result, should_retry = yield - return [result, nil] unless should_retry - rescue StandardError => e - should_retry = true - caught_exception = e - end + def parse_retry_after(value) + return nil if value.nil? - if should_retry && (retries_remaining > 1) - logger.debug("Retrying request, #{retries_remaining} retries left") - sleep(@backoff_policy.next_interval.to_f / 1000) - retry_with_backoff(retries_remaining - 1, &block) - else - [result, caught_exception] - end + str = value.is_a?(Array) ? value.first : value + return nil if str.nil? + + str = str.strip + return nil unless str =~ /\A\d+\z/ + + seconds = str.to_i + seconds > 0 ? seconds : nil end - # Sends a request for the batch, returns [status_code, body] - def send_request(write_key, batch) + # Sends a request for the batch, returns [status_code, body, headers] + def send_request(write_key, batch, retry_count = 0) payload = JSON.generate( :sentAt => datetime_in_iso8601(Time.now), :batch => batch ) - request = Net::HTTP::Post.new(@path, @headers) + headers = @headers.dup + headers['X-Retry-Count'] = retry_count.to_s if retry_count > 0 + + request = Net::HTTP::Post.new(@path, headers) request.basic_auth(write_key, nil) if self.class.stub logger.debug "stubbed request to #{@path}: " \ "write key = #{write_key}, batch = #{JSON.generate(batch)}" - [200, '{}'] + [200, '{}', {}] else - @http.start unless @http.started? # Maintain a persistent connection + @http.start unless @http.started? response = @http.request(request, payload) - [response.code.to_i, response.body] + [response.code.to_i, response.body, response.to_hash] end end diff --git a/lib/segment/analytics/worker.rb b/lib/segment/analytics/worker.rb index 6a7d68e..c4bf728 100644 --- a/lib/segment/analytics/worker.rb +++ b/lib/segment/analytics/worker.rb @@ -45,7 +45,7 @@ def run end res = @transport.send @write_key, @batch - @on_error.call(res.status, res.error) unless res.status == 200 + @on_error.call(res.status, res.error) unless res.success? @lock.synchronize { @batch.clear } end diff --git a/spec/segment/analytics/backoff_policy_spec.rb b/spec/segment/analytics/backoff_policy_spec.rb index 25ef05e..7d3a8d2 100644 --- a/spec/segment/analytics/backoff_policy_spec.rb +++ b/spec/segment/analytics/backoff_policy_spec.rb @@ -67,6 +67,28 @@ class Analytics end end + describe '#reset!' do + it 'resets attempts to 0' do + subject.next_interval + subject.next_interval + subject.next_interval + subject.reset! + expect(subject.instance_variable_get(:@attempts)).to eq(0) + end + + it 'causes next_interval to restart from minimum' do + subject_with_params = described_class.new( + min_timeout_ms: 1000, + max_timeout_ms: 10000, + multiplier: 2, + randomization_factor: 0.5 + ) + 3.times { subject_with_params.next_interval } + subject_with_params.reset! + expect(subject_with_params.next_interval).to be_within(500).of(1000) + end + end + describe '#next_interval' do subject { described_class.new( diff --git a/spec/segment/analytics/response_spec.rb b/spec/segment/analytics/response_spec.rb index bb673db..0376e20 100644 --- a/spec/segment/analytics/response_spec.rb +++ b/spec/segment/analytics/response_spec.rb @@ -13,6 +13,18 @@ class Analytics it { expect(subject).to respond_to(:error) } end + describe '#success?' do + it { expect(described_class.new(200, nil).success?).to be true } + it { expect(described_class.new(201, nil).success?).to be true } + it { expect(described_class.new(204, nil).success?).to be true } + it { expect(described_class.new(301, nil).success?).to be true } + it { expect(described_class.new(302, nil).success?).to be true } + it { expect(described_class.new(400, nil).success?).to be false } + it { expect(described_class.new(429, nil).success?).to be false } + it { expect(described_class.new(500, nil).success?).to be false } + it { expect(described_class.new(-1, nil).success?).to be false } + end + describe '#initialize' do let(:status) { 404 } let(:error) { 'Oh No' } diff --git a/spec/segment/analytics/transport_spec.rb b/spec/segment/analytics/transport_spec.rb index 488b3ee..4dda235 100644 --- a/spec/segment/analytics/transport_spec.rb +++ b/spec/segment/analytics/transport_spec.rb @@ -115,6 +115,7 @@ class Analytics allow(http).to receive(:start) allow(http).to receive(:request) { response } allow(response).to receive(:body) { response_body } + allow(response).to receive(:to_hash) { {} } end it 'initalizes a new Net::HTTP::Post with path and default headers' do @@ -204,6 +205,14 @@ class Analytics end end + context '3xx is treated as success' do + let(:status_code) { 301 } + it 'returns status without retrying' do + expect(subject).not_to receive(:sleep) + expect(subject.send(write_key, batch).status).to eq(301) + end + end + context 'request results in errorful response' do let(:error) { 'this is an error' } let(:response_body) { { error: error }.to_json } @@ -218,10 +227,117 @@ class Analytics it_behaves_like('retried request', 500, '{}') it_behaves_like('retried request', 503, '{}') - # All 4xx errors other than 429 (rate limited) must be retried + # 429 is retried it_behaves_like('retried request', 429, '{}') it_behaves_like('non-retried request', 404, '{}') it_behaves_like('non-retried request', 400, '{}') + + # Non-retryable 5xx: 501, 505, 511 + it_behaves_like('non-retried request', 501, '{}') + it_behaves_like('non-retried request', 505, '{}') + it_behaves_like('non-retried request', 511, '{}') + + # Retryable 4xx: 408, 410, 460 + it_behaves_like('retried request', 408, '{}') + it_behaves_like('retried request', 410, '{}') + it_behaves_like('retried request', 460, '{}') + end + + context '429 with Retry-After header' do + let(:status_code) { 429 } + let(:retry_after_seconds) { 2 } + subject { described_class.new(retries: 4, backoff_policy: FakeBackoffPolicy.new([1000, 1000, 1000])) } + + before do + allow(response).to receive(:to_hash) { { 'retry-after' => [retry_after_seconds.to_s] } } + # Second attempt succeeds + success_response = Net::HTTPResponse.new(1.1, 200, '{}') + allow(success_response).to receive(:body) { '{}' } + allow(success_response).to receive(:to_hash) { {} } + http = subject.instance_variable_get(:@http) + allow(http).to receive(:request).and_return(response, success_response) + end + + it 'sleeps for the Retry-After duration' do + expect(subject).to receive(:sleep).with(2).once + subject.send(write_key, batch) + end + + it 'caps Retry-After at RATE_LIMIT_RETRY_AFTER_CAP' do + allow(response).to receive(:to_hash) { { 'retry-after' => ['9999'] } } + expect(subject).to receive(:sleep).with(described_class::RATE_LIMIT_RETRY_AFTER_CAP).once + subject.send(write_key, batch) + end + + it 'returns success after retry' do + allow(subject).to receive(:sleep) + expect(subject.send(write_key, batch).success?).to be true + end + end + + context 'X-Retry-Count header' do + let(:status_code) { 500 } + let(:backoff_policy) { FakeBackoffPolicy.new([1, 1]) } + subject { described_class.new(retries: 3, backoff_policy: backoff_policy) } + + it 'does not send X-Retry-Count on first attempt' do + allow(subject).to receive(:sleep) + first_request = nil + http = subject.instance_variable_get(:@http) + allow(http).to receive(:request) do |req, _| + first_request ||= req + response + end + subject.send(write_key, batch) + expect(first_request['X-Retry-Count']).to be_nil + end + + it 'sends X-Retry-Count incrementing on retries' do + allow(subject).to receive(:sleep) + requests = [] + http = subject.instance_variable_get(:@http) + allow(http).to receive(:request) do |req, _| + requests << req + response + end + subject.send(write_key, batch) + expect(requests[1]['X-Retry-Count']).to eq('1') + expect(requests[2]['X-Retry-Count']).to eq('2') + end + end + + context 'private helpers' do + describe '#success_status?' do + it { expect(subject.send(:success_status?, 200)).to be true } + it { expect(subject.send(:success_status?, 201)).to be true } + it { expect(subject.send(:success_status?, 301)).to be true } + it { expect(subject.send(:success_status?, 400)).to be false } + it { expect(subject.send(:success_status?, 500)).to be false } + end + + describe '#retryable_status?' do + it { expect(subject.send(:retryable_status?, 500)).to be true } + it { expect(subject.send(:retryable_status?, 503)).to be true } + it { expect(subject.send(:retryable_status?, 429)).to be true } + it { expect(subject.send(:retryable_status?, 408)).to be true } + it { expect(subject.send(:retryable_status?, 410)).to be true } + it { expect(subject.send(:retryable_status?, 460)).to be true } + it { expect(subject.send(:retryable_status?, 400)).to be false } + it { expect(subject.send(:retryable_status?, 404)).to be false } + it { expect(subject.send(:retryable_status?, 501)).to be false } + it { expect(subject.send(:retryable_status?, 505)).to be false } + it { expect(subject.send(:retryable_status?, 511)).to be false } + end + + describe '#parse_retry_after' do + it { expect(subject.send(:parse_retry_after, '60')).to eq(60) } + it { expect(subject.send(:parse_retry_after, ['60'])).to eq(60) } + it { expect(subject.send(:parse_retry_after, '0')).to be_nil } + it { expect(subject.send(:parse_retry_after, '-1')).to be_nil } + it { expect(subject.send(:parse_retry_after, nil)).to be_nil } + it { expect(subject.send(:parse_retry_after, '')).to be_nil } + it { expect(subject.send(:parse_retry_after, 'Wed, 07 May 2026 12:00:00 GMT')).to be_nil } + end end context 'request or parsing of response results in an exception' do From 6140de51d83aa3a7d98353ec8d535c5562f44cd7 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Tue, 19 May 2026 13:34:07 -0400 Subject: [PATCH 02/14] Enable retry test suite in e2e-config --- e2e-cli/e2e-config.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/e2e-cli/e2e-config.json b/e2e-cli/e2e-config.json index f0aea47..e820b6d 100644 --- a/e2e-cli/e2e-config.json +++ b/e2e-cli/e2e-config.json @@ -1,6 +1,6 @@ { "sdk": "ruby", - "test_suites": "basic", + "test_suites": "basic,retry", "auto_settings": false, "patch": null, "env": {} From a172117af3469132691bd9d1548a66516f1f1b04 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Wed, 20 May 2026 13:55:33 -0400 Subject: [PATCH 03/14] Fix review findings: FakeBackoffPolicy reset!, success? scope, test update - Add reset! to FakeBackoffPolicy so transport specs don't crash - Narrow success? and success_status? to 2xx only (Net::HTTP doesn't follow redirects, so 3xx would silently lose batches) - Update malformed-JSON-on-200 test to match new semantics: a 200 means the server accepted the batch regardless of body parseability --- lib/segment/analytics/response.rb | 2 +- lib/segment/analytics/transport.rb | 2 +- spec/segment/analytics/transport_spec.rb | 12 +++++------- spec/spec_helper.rb | 2 ++ 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/lib/segment/analytics/response.rb b/lib/segment/analytics/response.rb index e4c2bc1..0a2fd2f 100644 --- a/lib/segment/analytics/response.rb +++ b/lib/segment/analytics/response.rb @@ -14,7 +14,7 @@ def initialize(status = 200, error = nil) end def success? - status >= 200 && status < 400 + status >= 200 && status < 300 end end end diff --git a/lib/segment/analytics/transport.rb b/lib/segment/analytics/transport.rb index 2f0e2a6..e9f4ca6 100644 --- a/lib/segment/analytics/transport.rb +++ b/lib/segment/analytics/transport.rb @@ -122,7 +122,7 @@ def shutdown private def success_status?(code) - code >= 200 && code < 400 + code >= 200 && code < 300 end def retryable_status?(code) diff --git a/spec/segment/analytics/transport_spec.rb b/spec/segment/analytics/transport_spec.rb index 4dda235..5cd428f 100644 --- a/spec/segment/analytics/transport_spec.rb +++ b/spec/segment/analytics/transport_spec.rb @@ -340,21 +340,19 @@ class Analytics end end - context 'request or parsing of response results in an exception' do + context 'response body is malformed JSON but status is 200' do let(:response_body) { 'Malformed JSON ---' } subject { described_class.new(retries: 0) } - it 'returns a -1 for status' do - expect(subject.send(write_key, batch).status).to eq(-1) + it 'treats 200 as success regardless of body' do + expect(subject.send(write_key, batch).status).to eq(200) end - it 'has a connection error' do + it 'has nil error when body is unparseable' do error = subject.send(write_key, batch).error - expect(error).not_to be_nil + expect(error).to be_nil end - - it_behaves_like('retried request', 200, 'Malformed JSON ---') end end end diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index 8dc8634..8b255e2 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -114,6 +114,8 @@ def next_interval raise 'FakeBackoffPolicy has no values left' if @interval_values.empty? @interval_values.shift end + + def reset!; end end # usage: From 8ea49339cdfc08cad61da174974f4d5ae0acaeb2 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Wed, 2 Sep 2026 19:38:14 -0400 Subject: [PATCH 04/14] Handle Retry-After on every retryable status, including 529 Route any retryable response carrying a valid Retry-After header through the rate-limit path (no retry-budget cost) instead of special-casing 429. Retryable statuses without Retry-After continue to use counted exponential backoff. Adds 529 to the retryable set and covers both paths with tests. Matches the behaviour already shipped in analytics-java 3.5.5 and the generic-retry-after conformance suite in sdk-e2e-tests. --- .gitignore | 3 + lib/segment/analytics/transport.rb | 47 ++++---- spec/segment/analytics/transport_spec.rb | 131 +++++++++++++++++++---- 3 files changed, 140 insertions(+), 41 deletions(-) diff --git a/.gitignore b/.gitignore index ab09e0c..fa7f8ae 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,6 @@ Gemfile.lock .ruby-version coverage/ +.bundle/ +vendor/ +*-plan.md diff --git a/lib/segment/analytics/transport.rb b/lib/segment/analytics/transport.rb index e9f4ca6..753eb09 100644 --- a/lib/segment/analytics/transport.rb +++ b/lib/segment/analytics/transport.rb @@ -8,6 +8,7 @@ require 'net/http' require 'net/https' require 'json' +require 'time' module Segment class Analytics @@ -69,28 +70,27 @@ def send(write_key, batch) return Response.new(status_code, error) if success_status?(status_code) - if status_code == 429 + unless retryable_status?(status_code) + logger.error(body) + return Response.new(status_code, error) + end + + # Any retryable status with Retry-After: use rate-limit path (no retry budget cost) + retry_after = parse_retry_after(response_headers['retry-after']) + if retry_after rate_limit_start_time ||= Time.now if (Time.now - rate_limit_start_time) >= @max_rate_limit_duration logger.error('Max rate limit duration exceeded for batch') return Response.new(status_code, error) end - - retry_after = parse_retry_after(response_headers['retry-after']) - if retry_after - delay = [retry_after, @rate_limit_retry_after_cap].min - logger.debug("Rate limited with Retry-After: #{delay}s. Retrying after delay.") - sleep(delay) - retry_count += 1 - next - end - end - - unless retryable_status?(status_code) - logger.error(body) - return Response.new(status_code, error) + delay = [retry_after, @rate_limit_retry_after_cap].min + logger.debug("Retry-After: #{delay}s on #{status_code}. Retrying after delay.") + sleep(delay) + retry_count += 1 + next end + # No Retry-After: counted backoff retries_remaining -= 1 if retries_remaining <= 0 logger.error('Retries exhausted for batch') @@ -140,10 +140,21 @@ def parse_retry_after(value) return nil if str.nil? str = str.strip - return nil unless str =~ /\A\d+\z/ - seconds = str.to_i - seconds > 0 ? seconds : nil + # Try integer seconds + if str =~ /\A\d+\z/ + seconds = str.to_i + return seconds > 0 ? seconds : nil + end + + # Try HTTP-date (RFC 7231 S7.1.1.1) + begin + target = Time.httpdate(str) + seconds = (target - Time.now).to_i + return seconds > 0 ? seconds : nil + rescue ArgumentError + nil + end end # Sends a request for the batch, returns [status_code, body, headers] diff --git a/spec/segment/analytics/transport_spec.rb b/spec/segment/analytics/transport_spec.rb index 5cd428f..a261a0e 100644 --- a/spec/segment/analytics/transport_spec.rb +++ b/spec/segment/analytics/transport_spec.rb @@ -275,6 +275,79 @@ class Analytics end end + context '503 with Retry-After header' do + let(:status_code) { 503 } + subject { described_class.new(retries: 4, backoff_policy: FakeBackoffPolicy.new([1000, 1000, 1000])) } + + before do + allow(response).to receive(:to_hash) { { 'retry-after' => ['2'] } } + success_response = Net::HTTPResponse.new(1.1, 200, '{}') + allow(success_response).to receive(:body) { '{}' } + allow(success_response).to receive(:to_hash) { {} } + http = subject.instance_variable_get(:@http) + allow(http).to receive(:request).and_return(response, success_response) + end + + it 'sleeps for the Retry-After duration' do + expect(subject).to receive(:sleep).with(2).once + subject.send(write_key, batch) + end + + it 'does not decrement retries_remaining (uses rate-limit path)' do + allow(subject).to receive(:sleep) + # With retries: 1, a 503+Retry-After should NOT exhaust retries because + # it uses the rate-limit path (no retry budget cost) + transport = described_class.new(retries: 1, backoff_policy: FakeBackoffPolicy.new([1000])) + http = transport.instance_variable_get(:@http) + allow(http).to receive(:start) + success_response = Net::HTTPResponse.new(1.1, 200, '{}') + allow(success_response).to receive(:body) { '{}' } + allow(success_response).to receive(:to_hash) { {} } + allow(http).to receive(:request).and_return(response, success_response) + allow(transport).to receive(:sleep) + result = transport.send(write_key, batch) + expect(result.status).to eq(200) + end + end + + context '529 with Retry-After header' do + let(:status_code) { 529 } + subject { described_class.new(retries: 4, backoff_policy: FakeBackoffPolicy.new([1000, 1000, 1000])) } + + before do + allow(response).to receive(:to_hash) { { 'retry-after' => ['1'] } } + success_response = Net::HTTPResponse.new(1.1, 200, '{}') + allow(success_response).to receive(:body) { '{}' } + allow(success_response).to receive(:to_hash) { {} } + http = subject.instance_variable_get(:@http) + allow(http).to receive(:request).and_return(response, success_response) + end + + it 'sleeps for the Retry-After duration' do + expect(subject).to receive(:sleep).with(1).once + subject.send(write_key, batch) + end + + it 'returns success after retry' do + allow(subject).to receive(:sleep) + expect(subject.send(write_key, batch).success?).to be true + end + + it 'does not decrement retries_remaining (uses rate-limit path)' do + # With retries: 1, a 529+Retry-After should NOT exhaust retries + transport = described_class.new(retries: 1, backoff_policy: FakeBackoffPolicy.new([1000])) + http = transport.instance_variable_get(:@http) + allow(http).to receive(:start) + success_response = Net::HTTPResponse.new(1.1, 200, '{}') + allow(success_response).to receive(:body) { '{}' } + allow(success_response).to receive(:to_hash) { {} } + allow(http).to receive(:request).and_return(response, success_response) + allow(transport).to receive(:sleep) + result = transport.send(write_key, batch) + expect(result.status).to eq(200) + end + end + context 'X-Retry-Count header' do let(:status_code) { 500 } let(:backoff_policy) { FakeBackoffPolicy.new([1, 1]) } @@ -308,35 +381,47 @@ class Analytics context 'private helpers' do describe '#success_status?' do - it { expect(subject.send(:success_status?, 200)).to be true } - it { expect(subject.send(:success_status?, 201)).to be true } - it { expect(subject.send(:success_status?, 301)).to be true } - it { expect(subject.send(:success_status?, 400)).to be false } - it { expect(subject.send(:success_status?, 500)).to be false } + it { expect(subject.__send__(:success_status?, 200)).to be true } + it { expect(subject.__send__(:success_status?, 201)).to be true } + it { expect(subject.__send__(:success_status?, 301)).to be false } + it { expect(subject.__send__(:success_status?, 400)).to be false } + it { expect(subject.__send__(:success_status?, 500)).to be false } end describe '#retryable_status?' do - it { expect(subject.send(:retryable_status?, 500)).to be true } - it { expect(subject.send(:retryable_status?, 503)).to be true } - it { expect(subject.send(:retryable_status?, 429)).to be true } - it { expect(subject.send(:retryable_status?, 408)).to be true } - it { expect(subject.send(:retryable_status?, 410)).to be true } - it { expect(subject.send(:retryable_status?, 460)).to be true } - it { expect(subject.send(:retryable_status?, 400)).to be false } - it { expect(subject.send(:retryable_status?, 404)).to be false } - it { expect(subject.send(:retryable_status?, 501)).to be false } - it { expect(subject.send(:retryable_status?, 505)).to be false } - it { expect(subject.send(:retryable_status?, 511)).to be false } + it { expect(subject.__send__(:retryable_status?, 500)).to be true } + it { expect(subject.__send__(:retryable_status?, 503)).to be true } + it { expect(subject.__send__(:retryable_status?, 429)).to be true } + it { expect(subject.__send__(:retryable_status?, 408)).to be true } + it { expect(subject.__send__(:retryable_status?, 410)).to be true } + it { expect(subject.__send__(:retryable_status?, 460)).to be true } + it { expect(subject.__send__(:retryable_status?, 400)).to be false } + it { expect(subject.__send__(:retryable_status?, 404)).to be false } + it { expect(subject.__send__(:retryable_status?, 501)).to be false } + it { expect(subject.__send__(:retryable_status?, 505)).to be false } + it { expect(subject.__send__(:retryable_status?, 511)).to be false } end describe '#parse_retry_after' do - it { expect(subject.send(:parse_retry_after, '60')).to eq(60) } - it { expect(subject.send(:parse_retry_after, ['60'])).to eq(60) } - it { expect(subject.send(:parse_retry_after, '0')).to be_nil } - it { expect(subject.send(:parse_retry_after, '-1')).to be_nil } - it { expect(subject.send(:parse_retry_after, nil)).to be_nil } - it { expect(subject.send(:parse_retry_after, '')).to be_nil } - it { expect(subject.send(:parse_retry_after, 'Wed, 07 May 2026 12:00:00 GMT')).to be_nil } + it { expect(subject.__send__(:parse_retry_after, '60')).to eq(60) } + it { expect(subject.__send__(:parse_retry_after, ['60'])).to eq(60) } + it { expect(subject.__send__(:parse_retry_after, '0')).to be_nil } + it { expect(subject.__send__(:parse_retry_after, '-1')).to be_nil } + it { expect(subject.__send__(:parse_retry_after, nil)).to be_nil } + it { expect(subject.__send__(:parse_retry_after, '')).to be_nil } + it { expect(subject.__send__(:parse_retry_after, 'garbage')).to be_nil } + + it 'parses HTTP-date 2 seconds in the future' do + future = (Time.now + 2).httpdate + result = subject.__send__(:parse_retry_after, future) + expect(result).to be_between(1, 3) + end + + it 'returns nil for HTTP-date in the past' do + past = (Time.now - 10).httpdate + result = subject.__send__(:parse_retry_after, past) + expect(result).to be_nil + end end end From a5617ed72ef6a65e9edf01cacc0fcb67b5175a7b Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Thu, 3 Sep 2026 17:15:16 -0400 Subject: [PATCH 05/14] Fix 3xx success, network-error retries and shutdown blocking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The suite was red on this branch: 199 examples, 2 failures. Nothing caught it because analytics-ruby's PR runs no unit-test job, only Wiz, semgrep and opa. Four fixes. 1. Spec item 1 says 2xx and 3xx are success, and response_spec asserted exactly that while Response#success? and success_status? were 2xx-only — so 3xx also fell through to logger.error and on_error, which master did not do. Both now accept 3xx, and the contradictory #success_status? assertion (301 => false) is corrected. That resolves both failures. 2. Transient network errors were no longer retried. The rescue moved from inside the retry helper, where master set should_retry, to a method-level rescue on send, so one ECONNRESET, DNS blip or read timeout unwound the whole loop and dropped the batch on first occurrence. The spec covering this was deleted in the same change. send_request is wrapped again and network failures go through the counted backoff budget; two specs cover retry and eventual give-up. The counted-backoff branch was extracted into a lambda so both callers share one budget implementation. 3. backoff_policy.reset! was called unconditionally, but reset! is new here and backoff_policy is a documented public option. A user duck type that worked on master raised NoMethodError, which the method-level rescue turned into an ordinary Response(-1): every batch failed, no request ever sent. Now guarded with respond_to?. 4. Retry-After and backoff waits blocked shutdown. Ruby's worker is a single thread and the only consumer of the queue, so a sleep of up to rate_limit_retry_after_cap (300s) stalls the whole client, and Thread.current[:should_exit] is only checked at the top of the run loop. at_exit now wakes the worker, which returns the sleep early, and both wait sites check for shutdown immediately afterwards. 202 examples, 0 failures, and all 58 e2e tests pass. --- lib/segment/analytics/client.rb | 9 +++- lib/segment/analytics/response.rb | 3 +- lib/segment/analytics/transport.rb | 63 +++++++++++++++++------- spec/segment/analytics/transport_spec.rb | 38 +++++++++++++- 4 files changed, 91 insertions(+), 22 deletions(-) diff --git a/lib/segment/analytics/client.rb b/lib/segment/analytics/client.rb index 9589f69..75022b7 100644 --- a/lib/segment/analytics/client.rb +++ b/lib/segment/analytics/client.rb @@ -32,7 +32,14 @@ def initialize(opts = {}) check_write_key! - at_exit { @worker_thread && @worker_thread[:should_exit] = true } + at_exit do + if @worker_thread + @worker_thread[:should_exit] = true + # Break any Retry-After or backoff sleep so shutdown is not held for + # up to rate_limit_retry_after_cap seconds. + @worker_thread.wakeup if @worker_thread.alive? + end + end end # Synchronously waits until the worker has flushed the queue. diff --git a/lib/segment/analytics/response.rb b/lib/segment/analytics/response.rb index 0a2fd2f..d35c062 100644 --- a/lib/segment/analytics/response.rb +++ b/lib/segment/analytics/response.rb @@ -14,7 +14,8 @@ def initialize(status = 200, error = nil) end def success? - status >= 200 && status < 300 + # Spec item 1: 2xx and 3xx are success. + status >= 200 && status < 400 end end end diff --git a/lib/segment/analytics/transport.rb b/lib/segment/analytics/transport.rb index 753eb09..d94cbae 100644 --- a/lib/segment/analytics/transport.rb +++ b/lib/segment/analytics/transport.rb @@ -51,15 +51,50 @@ def initialize(options = {}) def send(write_key, batch) logger.debug("Sending request for #{batch.length} items") - @backoff_policy.reset! + @backoff_policy.reset! if @backoff_policy.respond_to?(:reset!) retry_count = 0 retries_remaining = @retries backoff_start_time = nil rate_limit_start_time = nil + # Returns a Response when the batch should be abandoned, or nil to retry. + consume_backoff = lambda do |response_error, response_status| + retries_remaining -= 1 + if retries_remaining <= 0 + logger.error('Retries exhausted for batch') + next Response.new(response_status, response_error) + end + + backoff_start_time ||= Time.now + if (Time.now - backoff_start_time) >= @max_total_backoff_duration + logger.error('Max total backoff duration exceeded for batch') + next Response.new(response_status, response_error) + end + + delay_ms = @backoff_policy.next_interval + logger.debug("Retrying request, #{retries_remaining} retries left. Waiting #{delay_ms}ms") + sleep(delay_ms.to_f / 1000) + next Response.new(response_status, response_error) if Thread.current[:should_exit] + + retry_count += 1 + nil + end + loop do - status_code, body, response_headers = send_request(write_key, batch, retry_count) + begin + status_code, body, response_headers = send_request(write_key, batch, retry_count) + rescue StandardError => e + # Connection reset, DNS failure, read timeout and friends. These were + # retried before this branch was refactored; they still are, on the + # counted backoff budget rather than for free. + logger.error("Network error: #{e.message}") + give_up = consume_backoff.call(e.to_s, -1) + return give_up if give_up + + next + end + error = begin JSON.parse(body)['error'] rescue StandardError @@ -86,27 +121,16 @@ def send(write_key, batch) delay = [retry_after, @rate_limit_retry_after_cap].min logger.debug("Retry-After: #{delay}s on #{status_code}. Retrying after delay.") sleep(delay) + # Client#shutdown wakes this thread, so the sleep above returns early. + return Response.new(status_code, error) if Thread.current[:should_exit] + retry_count += 1 next end # No Retry-After: counted backoff - retries_remaining -= 1 - if retries_remaining <= 0 - logger.error('Retries exhausted for batch') - return Response.new(status_code, error) - end - - backoff_start_time ||= Time.now - if (Time.now - backoff_start_time) >= @max_total_backoff_duration - logger.error('Max total backoff duration exceeded for batch') - return Response.new(status_code, error) - end - - delay_ms = @backoff_policy.next_interval - logger.debug("Retrying request, #{retries_remaining} retries left. Waiting #{delay_ms}ms") - sleep(delay_ms.to_f / 1000) - retry_count += 1 + give_up = consume_backoff.call(error, status_code) + return give_up if give_up end rescue StandardError => e logger.error(e.message) @@ -122,7 +146,8 @@ def shutdown private def success_status?(code) - code >= 200 && code < 300 + # Spec item 1: 2xx and 3xx are success. + code >= 200 && code < 400 end def retryable_status?(code) diff --git a/spec/segment/analytics/transport_spec.rb b/spec/segment/analytics/transport_spec.rb index a261a0e..88dc1a0 100644 --- a/spec/segment/analytics/transport_spec.rb +++ b/spec/segment/analytics/transport_spec.rb @@ -379,11 +379,47 @@ class Analytics end end + context 'transient network error' do + it 'retries the request instead of dropping the batch' do + success_response = Net::HTTPResponse.new(1.1, 200, '{}') + allow(success_response).to receive(:body) { '{}' } + allow(success_response).to receive(:to_hash) { {} } + + http = subject.instance_variable_get(:@http) + calls = 0 + allow(http).to receive(:request) do + calls += 1 + raise Errno::ECONNRESET, 'reset' if calls == 1 + + success_response + end + allow(subject).to receive(:sleep) + + response = subject.send(write_key, batch) + + expect(calls).to eq(2) + expect(response.status).to eq(200) + end + + it 'gives up once the retry budget is spent' do + http = subject.instance_variable_get(:@http) + allow(http).to receive(:request).and_raise(Errno::ECONNRESET, 'reset') + allow(subject).to receive(:sleep) + + response = subject.send(write_key, batch) + + expect(response.status).to eq(-1) + expect(response.error).to match(/reset/) + end + end + context 'private helpers' do describe '#success_status?' do it { expect(subject.__send__(:success_status?, 200)).to be true } it { expect(subject.__send__(:success_status?, 201)).to be true } - it { expect(subject.__send__(:success_status?, 301)).to be false } + # Spec item 1: 2xx and 3xx are success. + it { expect(subject.__send__(:success_status?, 301)).to be true } + it { expect(subject.__send__(:success_status?, 304)).to be true } it { expect(subject.__send__(:success_status?, 400)).to be false } it { expect(subject.__send__(:success_status?, 500)).to be false } end From a95bb9924be1d8802a78a8a651e9faf9a2c048ad Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Fri, 11 Sep 2026 11:11:13 -0400 Subject: [PATCH 06/14] Tighten retry comments Cut the before/after narration from the comments added with the Retry-After work. The network-error branch now says that those failures use the counted backoff budget rather than recounting how the rescue used to be placed. --- lib/segment/analytics/transport.rb | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/lib/segment/analytics/transport.rb b/lib/segment/analytics/transport.rb index d94cbae..dd0cb1e 100644 --- a/lib/segment/analytics/transport.rb +++ b/lib/segment/analytics/transport.rb @@ -85,9 +85,8 @@ def send(write_key, batch) begin status_code, body, response_headers = send_request(write_key, batch, retry_count) rescue StandardError => e - # Connection reset, DNS failure, read timeout and friends. These were - # retried before this branch was refactored; they still are, on the - # counted backoff budget rather than for free. + # Connection reset, DNS failure, read timeout and friends. Retried on + # the counted backoff budget, like a retryable status code. logger.error("Network error: #{e.message}") give_up = consume_backoff.call(e.to_s, -1) return give_up if give_up From 2bba403b5ad9c5cc383ebb2638e3103233d6b9a7 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Fri, 11 Sep 2026 16:47:18 -0400 Subject: [PATCH 07/14] Extract the retry budgets out of Transport#send MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit rubocop runs as part of rake's default task, over lib/ and spec/, and this branch had pushed Transport#send to 66 lines with an ABC size of 75.77 against a limit of 25 — eleven new offences, so CI would have gone red. The bulk of it was the counted-backoff budget sitting inline as a lambda. That, and the rate-limit budget beside it, are now a RetryBudget class of their own. It reports the delay to wait and nil once a budget is spent; Transport still performs the sleep, which keeps the seam the specs stub. send is down to 21 lines and ABC 32.39, and its cyclomatic and perceived-complexity offences are gone. Option parsing moved out of initialize the same way, clearing all four of its offences, and the response classification and delay choice are now named methods rather than inline branches. Three Max values in .rubocop_todo.yml are raised for what is left, which is a class 146 lines long and two methods a handful of lines over. Regenerating the file wholesale was the alternative and a bad one: it was last generated by rubocop 1.44 in 2023, and 1.90 rewrites the entire baseline. RetryBudget takes an options hash rather than keyword arguments: the gemspec declares required_ruby_version >= 2.0 and rubocop parses as 2.0, where required keyword arguments are a syntax error. rubocop reports no offences over lib/ and spec/. 202 examples, 0 failures, and all 58 e2e tests pass. Line coverage 97.92% -> 98.19%. --- .rubocop_todo.yml | 6 +- lib/segment/analytics.rb | 1 + lib/segment/analytics/retry_budget.rb | 69 +++++++++++ lib/segment/analytics/transport.rb | 166 +++++++++++++------------- 4 files changed, 156 insertions(+), 86 deletions(-) create mode 100644 lib/segment/analytics/retry_budget.rb diff --git a/.rubocop_todo.yml b/.rubocop_todo.yml index c02e129..904be3e 100644 --- a/.rubocop_todo.yml +++ b/.rubocop_todo.yml @@ -62,7 +62,7 @@ Lint/UnusedBlockArgument: # Offense count: 3 # Configuration parameters: AllowedMethods, AllowedPatterns, IgnoredMethods, CountRepeatedAttributes. Metrics/AbcSize: - Max: 25 + Max: 33 # Offense count: 3 # Configuration parameters: CountComments, CountAsOne, ExcludedMethods, AllowedMethods, AllowedPatterns, IgnoredMethods. @@ -73,7 +73,7 @@ Metrics/BlockLength: # Offense count: 1 # Configuration parameters: CountComments, CountAsOne. Metrics/ClassLength: - Max: 115 + Max: 150 # Offense count: 2 # Configuration parameters: AllowedMethods, AllowedPatterns, IgnoredMethods. @@ -83,7 +83,7 @@ Metrics/CyclomaticComplexity: # Offense count: 11 # Configuration parameters: CountComments, CountAsOne, ExcludedMethods, AllowedMethods, AllowedPatterns, IgnoredMethods. Metrics/MethodLength: - Max: 16 + Max: 21 # Offense count: 1 # This cop supports safe autocorrection (--autocorrect). diff --git a/lib/segment/analytics.rb b/lib/segment/analytics.rb index 707e7c4..37e3a2e 100644 --- a/lib/segment/analytics.rb +++ b/lib/segment/analytics.rb @@ -6,6 +6,7 @@ require 'segment/analytics/field_parser' require 'segment/analytics/client' require 'segment/analytics/worker' +require 'segment/analytics/retry_budget' require 'segment/analytics/transport' require 'segment/analytics/response' require 'segment/analytics/logging' diff --git a/lib/segment/analytics/retry_budget.rb b/lib/segment/analytics/retry_budget.rb new file mode 100644 index 0000000..e725c2f --- /dev/null +++ b/lib/segment/analytics/retry_budget.rb @@ -0,0 +1,69 @@ +# frozen_string_literal: true + +module Segment + class Analytics + # Tracks the two independent budgets one send may spend. + # + # A retryable status carrying Retry-After spends the rate-limit budget, which + # is bounded by wall clock only. Anything else retryable spends the counted + # backoff budget, bounded by both a retry count and wall clock. Keeping them + # separate is what stops a rate-limited server from exhausting the retries + # available to genuine failures. + # + # The caller performs the wait, so both methods return the delay in seconds, + # or nil when the budget is spent and the batch should be abandoned. + class RetryBudget + attr_reader :retry_count + + # Keyword arguments would be cleaner but need Ruby 2.1; the gemspec still + # declares >= 2.0, which is also what rubocop is configured to parse. + def initialize(options = {}) + @retries_remaining = options[:retries] + @backoff_policy = options[:backoff_policy] + @max_total_backoff_duration = options[:max_total_backoff_duration] + @max_rate_limit_duration = options[:max_rate_limit_duration] + @rate_limit_retry_after_cap = options[:rate_limit_retry_after_cap] + @logger = options[:logger] + @retry_count = 0 + @backoff_start_time = nil + @rate_limit_start_time = nil + end + + def next_backoff_delay + @retries_remaining -= 1 + return spent('Retries exhausted for batch') if @retries_remaining <= 0 + + @backoff_start_time ||= Time.now + return spent('Max total backoff duration exceeded for batch') if elapsed?(@backoff_start_time, @max_total_backoff_duration) + + delay_ms = @backoff_policy.next_interval + @logger.debug("Retrying request, #{@retries_remaining} retries left. Waiting #{delay_ms}ms") + delay_ms.to_f / 1000 + end + + def next_rate_limit_delay(retry_after, status_code) + @rate_limit_start_time ||= Time.now + return spent('Max rate limit duration exceeded for batch') if elapsed?(@rate_limit_start_time, @max_rate_limit_duration) + + delay = [retry_after, @rate_limit_retry_after_cap].min + @logger.debug("Retry-After: #{delay}s on #{status_code}. Retrying after delay.") + delay + end + + def record_retry + @retry_count += 1 + end + + private + + def elapsed?(start_time, limit) + (Time.now - start_time) >= limit + end + + def spent(message) + @logger.error(message) + nil + end + end + end +end diff --git a/lib/segment/analytics/transport.rb b/lib/segment/analytics/transport.rb index dd0cb1e..d81e526 100644 --- a/lib/segment/analytics/transport.rb +++ b/lib/segment/analytics/transport.rb @@ -5,6 +5,7 @@ require 'segment/analytics/response' require 'segment/analytics/logging' require 'segment/analytics/backoff_policy' +require 'segment/analytics/retry_budget' require 'net/http' require 'net/https' require 'json' @@ -26,23 +27,9 @@ def initialize(options = {}) options[:ssl] ||= SSL @headers = options[:headers] || HEADERS @path = options[:path] || PATH - @retries = options[:retries] || RETRIES - @backoff_policy = - options[:backoff_policy] || Segment::Analytics::BackoffPolicy.new - - @max_total_backoff_duration = options[:max_total_backoff_duration] || - MAX_TOTAL_BACKOFF_DURATION - @max_rate_limit_duration = options[:max_rate_limit_duration] || - MAX_RATE_LIMIT_DURATION - @rate_limit_retry_after_cap = options[:rate_limit_retry_after_cap] || - RATE_LIMIT_RETRY_AFTER_CAP - - http = Net::HTTP.new(options[:host], options[:port]) - http.use_ssl = options[:ssl] - http.read_timeout = 8 - http.open_timeout = 4 - @http = http + configure_retries(options) + @http = build_http(options) end # Sends a batch of messages to the API @@ -52,84 +39,26 @@ def send(write_key, batch) logger.debug("Sending request for #{batch.length} items") @backoff_policy.reset! if @backoff_policy.respond_to?(:reset!) - - retry_count = 0 - retries_remaining = @retries - backoff_start_time = nil - rate_limit_start_time = nil - - # Returns a Response when the batch should be abandoned, or nil to retry. - consume_backoff = lambda do |response_error, response_status| - retries_remaining -= 1 - if retries_remaining <= 0 - logger.error('Retries exhausted for batch') - next Response.new(response_status, response_error) - end - - backoff_start_time ||= Time.now - if (Time.now - backoff_start_time) >= @max_total_backoff_duration - logger.error('Max total backoff duration exceeded for batch') - next Response.new(response_status, response_error) - end - - delay_ms = @backoff_policy.next_interval - logger.debug("Retrying request, #{retries_remaining} retries left. Waiting #{delay_ms}ms") - sleep(delay_ms.to_f / 1000) - next Response.new(response_status, response_error) if Thread.current[:should_exit] - - retry_count += 1 - nil - end + budget = new_retry_budget loop do begin - status_code, body, response_headers = send_request(write_key, batch, retry_count) + status_code, body, headers = send_request(write_key, batch, budget.retry_count) rescue StandardError => e # Connection reset, DNS failure, read timeout and friends. Retried on # the counted backoff budget, like a retryable status code. logger.error("Network error: #{e.message}") - give_up = consume_backoff.call(e.to_s, -1) - return give_up if give_up + return Response.new(-1, e.to_s) unless wait_to_retry(budget.next_backoff_delay, budget) next end - error = begin - JSON.parse(body)['error'] - rescue StandardError - nil - end - logger.debug("Response status code: #{status_code}") - logger.debug("Response error: #{error}") if error - - return Response.new(status_code, error) if success_status?(status_code) - - unless retryable_status?(status_code) - logger.error(body) - return Response.new(status_code, error) - end - - # Any retryable status with Retry-After: use rate-limit path (no retry budget cost) - retry_after = parse_retry_after(response_headers['retry-after']) - if retry_after - rate_limit_start_time ||= Time.now - if (Time.now - rate_limit_start_time) >= @max_rate_limit_duration - logger.error('Max rate limit duration exceeded for batch') - return Response.new(status_code, error) - end - delay = [retry_after, @rate_limit_retry_after_cap].min - logger.debug("Retry-After: #{delay}s on #{status_code}. Retrying after delay.") - sleep(delay) - # Client#shutdown wakes this thread, so the sleep above returns early. - return Response.new(status_code, error) if Thread.current[:should_exit] - - retry_count += 1 - next - end + error = parse_error(body) + final = final_response(status_code, body, error) + return final if final - # No Retry-After: counted backoff - give_up = consume_backoff.call(error, status_code) - return give_up if give_up + delay = retry_delay(status_code, headers, budget) + return Response.new(status_code, error) unless wait_to_retry(delay, budget) end rescue StandardError => e logger.error(e.message) @@ -144,6 +73,77 @@ def shutdown private + # A Response once the batch is settled, or nil while it is still retryable. + def final_response(status_code, body, error) + logger.debug("Response status code: #{status_code}") + logger.debug("Response error: #{error}") if error + + return Response.new(status_code, error) if success_status?(status_code) + return nil if retryable_status?(status_code) + + logger.error(body) + Response.new(status_code, error) + end + + # A Retry-After spends the rate-limit budget; anything else retryable + # spends the counted backoff budget. + def retry_delay(status_code, headers, budget) + retry_after = parse_retry_after(headers['retry-after']) + return budget.next_backoff_delay unless retry_after + + budget.next_rate_limit_delay(retry_after, status_code) + end + + def configure_retries(options) + @retries = options[:retries] || RETRIES + @backoff_policy = + options[:backoff_policy] || Segment::Analytics::BackoffPolicy.new + @max_total_backoff_duration = options[:max_total_backoff_duration] || + MAX_TOTAL_BACKOFF_DURATION + @max_rate_limit_duration = options[:max_rate_limit_duration] || + MAX_RATE_LIMIT_DURATION + @rate_limit_retry_after_cap = options[:rate_limit_retry_after_cap] || + RATE_LIMIT_RETRY_AFTER_CAP + end + + def build_http(options) + http = Net::HTTP.new(options[:host], options[:port]) + http.use_ssl = options[:ssl] + http.read_timeout = 8 + http.open_timeout = 4 + http + end + + def new_retry_budget + RetryBudget.new( + :retries => @retries, + :backoff_policy => @backoff_policy, + :max_total_backoff_duration => @max_total_backoff_duration, + :max_rate_limit_duration => @max_rate_limit_duration, + :rate_limit_retry_after_cap => @rate_limit_retry_after_cap, + :logger => logger + ) + end + + # nil delay means the budget is spent. Sleeping here rather than inside + # RetryBudget keeps the wait on Transport, where callers stub it. + def wait_to_retry(delay, budget) + return false if delay.nil? + + sleep(delay) + # Client#shutdown wakes this thread, so the sleep above returns early. + return false if Thread.current[:should_exit] + + budget.record_retry + true + end + + def parse_error(body) + JSON.parse(body)['error'] + rescue StandardError + nil + end + def success_status?(code) # Spec item 1: 2xx and 3xx are success. code >= 200 && code < 400 @@ -175,7 +175,7 @@ def parse_retry_after(value) begin target = Time.httpdate(str) seconds = (target - Time.now).to_i - return seconds > 0 ? seconds : nil + seconds > 0 ? seconds : nil rescue ArgumentError nil end From 9018480d200aaf48a651324a98cd0c5d3b1b7563 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Thu, 17 Sep 2026 15:22:01 -0400 Subject: [PATCH 08/14] Opt in to the e2e Authorization check The header assertion in sdk-e2e-tests is opt-in per SDK, since analytics-kotlin and analytics-swift do not send it yet. This SDK does, so it runs the check. --- e2e-cli/e2e-config.json | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/e2e-cli/e2e-config.json b/e2e-cli/e2e-config.json index e820b6d..1bdb54a 100644 --- a/e2e-cli/e2e-config.json +++ b/e2e-cli/e2e-config.json @@ -3,5 +3,7 @@ "test_suites": "basic,retry", "auto_settings": false, "patch": null, - "env": {} + "env": { + "AUTH_HEADER": "true" + } } From d0d4e6c8c064816a80b98c0f4d72535b4e2f8bd8 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Tue, 22 Sep 2026 12:04:17 -0400 Subject: [PATCH 09/14] Treat only 2xx as a successful upload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every one of these SDKs treated a 3xx as a failure before this work, and the change to 200-399 came from the design doc's "Spec item 1: 2xx and 3xx are success". That line is wrong, and the doc is what needs correcting. Measured against a local server, with the same HTTP clients these SDKs use: 307/308 + Location -> followed as POST with the body, arrives as 200 301/302/303 + Loc. -> followed as GET with no body, arrives as 200 302 without Location-> surfaces raw as 302 300 Multiple Choices-> surfaces raw as 300 304 Not Modified -> surfaces raw as 304 So a raw 3xx only reaches the classifier when the client has already declined to follow it, meaning nothing was uploaded. The one redirect that genuinely works, 307/308, never produces a 3xx here at all — it produces 200 — so narrowing the bound cannot break it. Nothing was gained by the wider range; a 300, 304, or Location-less 302 from a proxy was being logged as a delivered batch and dropped with no error callback. The narrower bound also needs no new branches: a 3xx is neither 5xx nor in the retryable 4xx set, so it already falls through to the non-retryable path and reports a failure. TAPI does not emit 3xx and has no plans to. This matters because host is customer-configurable and proxies in front of it are common. Net::HTTP never follows redirects, so ruby is the most likely to see a raw 3xx, and logging the body was useless for one. It now logs the status and points at the host. Specs and the misnamed "3xx is treated as success" context corrected. ClassLength raised 150 -> 155 for the added branch; comments do not count toward it. --- .rubocop_todo.yml | 2 +- lib/segment/analytics/response.rb | 3 +-- lib/segment/analytics/transport.rb | 13 ++++++++++--- spec/segment/analytics/response_spec.rb | 6 ++++-- spec/segment/analytics/transport_spec.rb | 15 +++++++++------ 5 files changed, 25 insertions(+), 14 deletions(-) diff --git a/.rubocop_todo.yml b/.rubocop_todo.yml index 904be3e..94a71c3 100644 --- a/.rubocop_todo.yml +++ b/.rubocop_todo.yml @@ -73,7 +73,7 @@ Metrics/BlockLength: # Offense count: 1 # Configuration parameters: CountComments, CountAsOne. Metrics/ClassLength: - Max: 150 + Max: 155 # Offense count: 2 # Configuration parameters: AllowedMethods, AllowedPatterns, IgnoredMethods. diff --git a/lib/segment/analytics/response.rb b/lib/segment/analytics/response.rb index d35c062..0a2fd2f 100644 --- a/lib/segment/analytics/response.rb +++ b/lib/segment/analytics/response.rb @@ -14,8 +14,7 @@ def initialize(status = 200, error = nil) end def success? - # Spec item 1: 2xx and 3xx are success. - status >= 200 && status < 400 + status >= 200 && status < 300 end end end diff --git a/lib/segment/analytics/transport.rb b/lib/segment/analytics/transport.rb index d81e526..e9319a8 100644 --- a/lib/segment/analytics/transport.rb +++ b/lib/segment/analytics/transport.rb @@ -81,7 +81,13 @@ def final_response(status_code, body, error) return Response.new(status_code, error) if success_status?(status_code) return nil if retryable_status?(status_code) - logger.error(body) + if status_code >= 300 && status_code < 400 + # Logging the body here would be useless: a redirect has none. + logger.error("Unexpected redirect (#{status_code}); batch not uploaded. " \ + 'Check whether the configured host points at a proxy or redirector.') + else + logger.error(body) + end Response.new(status_code, error) end @@ -144,9 +150,10 @@ def parse_error(body) nil end + # Only 2xx. Net::HTTP does not follow redirects, so a 3xx means nothing was + # uploaded; calling it success would drop the batch silently. def success_status?(code) - # Spec item 1: 2xx and 3xx are success. - code >= 200 && code < 400 + code >= 200 && code < 300 end def retryable_status?(code) diff --git a/spec/segment/analytics/response_spec.rb b/spec/segment/analytics/response_spec.rb index 0376e20..147698d 100644 --- a/spec/segment/analytics/response_spec.rb +++ b/spec/segment/analytics/response_spec.rb @@ -17,8 +17,10 @@ class Analytics it { expect(described_class.new(200, nil).success?).to be true } it { expect(described_class.new(201, nil).success?).to be true } it { expect(described_class.new(204, nil).success?).to be true } - it { expect(described_class.new(301, nil).success?).to be true } - it { expect(described_class.new(302, nil).success?).to be true } + it { expect(described_class.new(300, nil).success?).to be false } + it { expect(described_class.new(301, nil).success?).to be false } + it { expect(described_class.new(302, nil).success?).to be false } + it { expect(described_class.new(304, nil).success?).to be false } it { expect(described_class.new(400, nil).success?).to be false } it { expect(described_class.new(429, nil).success?).to be false } it { expect(described_class.new(500, nil).success?).to be false } diff --git a/spec/segment/analytics/transport_spec.rb b/spec/segment/analytics/transport_spec.rb index 88dc1a0..e2f6aeb 100644 --- a/spec/segment/analytics/transport_spec.rb +++ b/spec/segment/analytics/transport_spec.rb @@ -205,11 +205,13 @@ class Analytics end end - context '3xx is treated as success' do + context '3xx is not retried and is not success' do let(:status_code) { 301 } - it 'returns status without retrying' do + it 'returns the status without retrying, and does not report success' do expect(subject).not_to receive(:sleep) - expect(subject.send(write_key, batch).status).to eq(301) + response = subject.send(write_key, batch) + expect(response.status).to eq(301) + expect(response.success?).to be false end end @@ -417,9 +419,10 @@ class Analytics describe '#success_status?' do it { expect(subject.__send__(:success_status?, 200)).to be true } it { expect(subject.__send__(:success_status?, 201)).to be true } - # Spec item 1: 2xx and 3xx are success. - it { expect(subject.__send__(:success_status?, 301)).to be true } - it { expect(subject.__send__(:success_status?, 304)).to be true } + # Only 2xx: Net::HTTP does not follow redirects, so a 3xx means + # nothing was uploaded. + it { expect(subject.__send__(:success_status?, 301)).to be false } + it { expect(subject.__send__(:success_status?, 304)).to be false } it { expect(subject.__send__(:success_status?, 400)).to be false } it { expect(subject.__send__(:success_status?, 500)).to be false } end From b45de9cf55db8152434db4aaec7bfb8ca7df6cb5 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Tue, 22 Sep 2026 18:51:45 -0400 Subject: [PATCH 10/14] Use monotonic clock for retry budgets and jitter the backoff ceiling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two correctness fixes from review: - RetryBudget measured its duration budgets with Time.now, so a wall-clock adjustment could expire a budget early or stretch it indefinitely. Both budgets now read CLOCK_MONOTONIC. HTTP-date parsing and sentAt in transport.rb stay on wall clock, which is what they need. - BackoffPolicy jittered before clamping, so every attempt at the ceiling returned exactly max_timeout_ms and a fleet that backed off together stayed in lockstep. Clamping first and jittering after fixes that; the jitter now only subtracts, so max_timeout_ms remains a hard ceiling rather than becoming a nominal one that ±50% could overshoot. The spec asserting `next_interval == 10000` at the cap encoded exactly the lockstep behaviour being fixed; it is replaced by one spec for the ceiling and one for the spread. --- lib/segment/analytics/backoff_policy.rb | 22 +++++-------------- lib/segment/analytics/retry_budget.rb | 11 +++++++--- spec/segment/analytics/backoff_policy_spec.rb | 13 +++++++++-- 3 files changed, 24 insertions(+), 22 deletions(-) diff --git a/lib/segment/analytics/backoff_policy.rb b/lib/segment/analytics/backoff_policy.rb index 7767838..9f9bff1 100644 --- a/lib/segment/analytics/backoff_policy.rb +++ b/lib/segment/analytics/backoff_policy.rb @@ -26,30 +26,18 @@ def initialize(opts = {}) # @return [Numeric] the next backoff interval, in milliseconds. def next_interval interval = @min_timeout_ms * (@multiplier**@attempts) - interval = add_jitter(interval, @randomization_factor) - @attempts += 1 - [interval, @max_timeout_ms].min + # Clamp first, then jitter. Jittering before the clamp meant every attempt + # at the ceiling returned exactly max_timeout_ms, so a fleet that backed off + # together stayed in lockstep. Jitter only subtracts, so the ceiling holds. + capped = [interval, @max_timeout_ms].min + capped - (rand * capped * @randomization_factor) end def reset! @attempts = 0 end - - private - - def add_jitter(base, randomization_factor) - random_number = rand - max_deviation = base * randomization_factor - deviation = random_number * max_deviation - - if random_number < 0.5 - base - deviation - else - base + deviation - end - end end end end diff --git a/lib/segment/analytics/retry_budget.rb b/lib/segment/analytics/retry_budget.rb index e725c2f..5688c68 100644 --- a/lib/segment/analytics/retry_budget.rb +++ b/lib/segment/analytics/retry_budget.rb @@ -33,7 +33,7 @@ def next_backoff_delay @retries_remaining -= 1 return spent('Retries exhausted for batch') if @retries_remaining <= 0 - @backoff_start_time ||= Time.now + @backoff_start_time ||= monotonic_now return spent('Max total backoff duration exceeded for batch') if elapsed?(@backoff_start_time, @max_total_backoff_duration) delay_ms = @backoff_policy.next_interval @@ -42,7 +42,7 @@ def next_backoff_delay end def next_rate_limit_delay(retry_after, status_code) - @rate_limit_start_time ||= Time.now + @rate_limit_start_time ||= monotonic_now return spent('Max rate limit duration exceeded for batch') if elapsed?(@rate_limit_start_time, @max_rate_limit_duration) delay = [retry_after, @rate_limit_retry_after_cap].min @@ -57,7 +57,12 @@ def record_retry private def elapsed?(start_time, limit) - (Time.now - start_time) >= limit + (monotonic_now - start_time) >= limit + end + + # Wall-clock time can jump; these budgets must not expire or stretch with it. + def monotonic_now + Process.clock_gettime(Process::CLOCK_MONOTONIC) end def spent(message) diff --git a/spec/segment/analytics/backoff_policy_spec.rb b/spec/segment/analytics/backoff_policy_spec.rb index 7d3a8d2..6156bb7 100644 --- a/spec/segment/analytics/backoff_policy_spec.rb +++ b/spec/segment/analytics/backoff_policy_spec.rb @@ -106,9 +106,18 @@ class Analytics expect(subject.next_interval).to be_within(4000).of(8000) end - it 'caps maximum duration at max_timeout_secs' do + it 'never exceeds max_timeout_ms once the ceiling is reached' do 10.times { subject.next_interval } - expect(subject.next_interval).to eq(10000) + 20.times do + expect(subject.next_interval).to be <= 10000 + end + end + + it 'jitters at the ceiling instead of returning a fixed value' do + 10.times { subject.next_interval } + intervals = Array.new(20) { subject.next_interval } + expect(intervals.uniq.size).to be > 1 + expect(intervals.min).to be >= 5000 end end end From 4414de6ff0dc22e57967931070459349b781da3d Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Tue, 22 Sep 2026 18:57:11 -0400 Subject: [PATCH 11/14] Add release notes for the HTTP response and retry work Records the retry/Retry-After work and, for the SDKs where a header is newly on the wire, an upgrade note: customers whose proxies allowlist request headers had uploads rejected by the already-released analytics-next change, and the same trap applies here. --- History.md | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/History.md b/History.md index d07c50a..21d095f 100644 --- a/History.md +++ b/History.md @@ -1,3 +1,25 @@ +Unreleased +========== + +### Upgrade note: new request header and proxy allowlists + +This release sends an `X-Retry-Count` request header on retries. If your +traffic to Segment goes through a proxy, gateway or WAF that allowlists +request headers, add it before upgrading or retried uploads will be +rejected. The `Authorization` header is unchanged: this client has always +sent the write key as HTTP Basic credentials. + +* Send `X-Retry-Count` on retries, so the server can distinguish a retry from a first attempt. Omitted on the first attempt. +* Unified retry handling: 429, 408, 410, 460 and 5xx (except 501, 505 and 511) are retried. `Retry-After` is honoured on all of them, not just 429, which brings 529 in through the generic 5xx rule. +* `Retry-After` accepts numeric seconds and the RFC 7231 HTTP-date formats, capped at 300s (`rate_limit_retry_after_cap`). +* Rate-limited retries are bounded by elapsed time rather than counted against the retry limit, so a long `Retry-After` no longer exhausts the budget. +* New options `max_total_backoff_duration` and `max_rate_limit_duration` (default 12 hours each) bound the two waits. +* Only 2xx responses count as a successful upload. A 3xx is now logged and retried rather than silently treated as delivered; the Segment endpoint does not redirect, so this only affects custom `host` values. +* Network errors are retried on the same backoff schedule as failed responses instead of dropping the batch. +* Backoff waits no longer block shutdown for the full delay. +* Retry timing uses a monotonic clock, so a system clock change cannot stretch or collapse a backoff. +* Backoff intervals are now jittered at the ceiling as well, so clients that back off together do not retry in lockstep. + 2.5.0 / 2024-07-17 ================== From 7cd5b02eb02823253e9514efcabb2fadffc075be Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Wed, 23 Sep 2026 07:06:20 -0400 Subject: [PATCH 12/14] Correct the release notes on 3xx handling No SDK retries a 3xx: every one classifies it as non-retryable and reports a failed upload. The notes claimed it was retried, which is wrong, and would have sent anyone debugging a proxy redirect looking for retries that never happen. Also scopes python's 511 line to the OAuth case, which is the one place the spec does allow a 511 retry, and php's new budget options to the LibCurl consumer, since Socket ignores them. --- History.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/History.md b/History.md index 21d095f..4cc41af 100644 --- a/History.md +++ b/History.md @@ -14,7 +14,7 @@ sent the write key as HTTP Basic credentials. * `Retry-After` accepts numeric seconds and the RFC 7231 HTTP-date formats, capped at 300s (`rate_limit_retry_after_cap`). * Rate-limited retries are bounded by elapsed time rather than counted against the retry limit, so a long `Retry-After` no longer exhausts the budget. * New options `max_total_backoff_duration` and `max_rate_limit_duration` (default 12 hours each) bound the two waits. -* Only 2xx responses count as a successful upload. A 3xx is now logged and retried rather than silently treated as delivered; the Segment endpoint does not redirect, so this only affects custom `host` values. +* Only 2xx responses count as a successful upload. A 3xx is now reported as a failed upload rather than silently treated as delivered. It is not retried: a redirect `Net::HTTP` already declined to follow will not succeed on a retry. The Segment endpoint does not redirect, so this only affects custom `host` values. * Network errors are retried on the same backoff schedule as failed responses instead of dropping the batch. * Backoff waits no longer block shutdown for the full delay. * Retry timing uses a monotonic clock, so a system clock change cannot stretch or collapse a backoff. From 7ae1ac99d8fd0d94b6373fe21d5991cdcbfe0219 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Wed, 23 Sep 2026 09:06:12 -0400 Subject: [PATCH 13/14] Grant the configured number of retries, and drop the at_exit wakeup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two fixes. RetryBudget decremented @retries_remaining before testing it, so one retry was spent on the exhaustion check itself: a configured N performed N-1, and retries: 1 and retries: 0 both performed none. go, python and java all grant N. The existing spec asserted `.exactly(retries - 1).times`, so it codified the bug and would have passed either way; it now asserts N, and the new retry_budget_spec covers the 1-and-0 cases that were previously indistinguishable. Both fail without this change. The at_exit block called Thread#wakeup on the worker. That raises ThreadError if the thread finished between the alive? check and the call, and an unrescued raise inside at_exit forces the process to exit 1 — a spurious failure for any script or CI step using this client. Reproduced under the pinned ruby 3.2. Rescuing the error would have left the other half broken: wakeup only cuts short a sleep already in progress, so one arriving while the worker is mid-request is lost and the next sleep runs in full, which is the hang the wakeup was added to prevent. The retry wait is now sliced and checks should_exit between slices, so shutdown is noticed within a second without needing wakeup at all. Same shape as python's interruptible wait. Specs stub the new interruptible_sleep seam instead of sleep, and must return truthy — a nil return reads as "shutting down" and stops the retry. Metrics/ClassLength for transport.rb goes 155 -> 161. Regenerating .rubocop_todo.yml as the repo's CLAUDE.md suggests rewrote ~100 unrelated lines (rubocop 1.90 against a file generated by 1.44, plus e2e-cli files that were not previously inspected), so this is the minimal edit instead. 210 examples pass, rubocop is clean on lib and spec, and the 61-test e2e suite passes. --- .rubocop_todo.yml | 2 +- History.md | 1 + lib/segment/analytics/client.rb | 13 ++-- lib/segment/analytics/retry_budget.rb | 6 +- lib/segment/analytics/transport.rb | 27 ++++++-- spec/segment/analytics/retry_budget_spec.rb | 49 +++++++++++++++ spec/segment/analytics/transport_spec.rb | 69 +++++++++++++++------ 7 files changed, 133 insertions(+), 34 deletions(-) create mode 100644 spec/segment/analytics/retry_budget_spec.rb diff --git a/.rubocop_todo.yml b/.rubocop_todo.yml index 94a71c3..423d279 100644 --- a/.rubocop_todo.yml +++ b/.rubocop_todo.yml @@ -73,7 +73,7 @@ Metrics/BlockLength: # Offense count: 1 # Configuration parameters: CountComments, CountAsOne. Metrics/ClassLength: - Max: 155 + Max: 161 # Offense count: 2 # Configuration parameters: AllowedMethods, AllowedPatterns, IgnoredMethods. diff --git a/History.md b/History.md index 4cc41af..f987f90 100644 --- a/History.md +++ b/History.md @@ -19,6 +19,7 @@ sent the write key as HTTP Basic credentials. * Backoff waits no longer block shutdown for the full delay. * Retry timing uses a monotonic clock, so a system clock change cannot stretch or collapse a backoff. * Backoff intervals are now jittered at the ceiling as well, so clients that back off together do not retry in lockstep. +* Fix `retries` granting one fewer attempt than configured. A configured 10 performed 9, and `retries: 1` performed none at all. 2.5.0 / 2024-07-17 ================== diff --git a/lib/segment/analytics/client.rb b/lib/segment/analytics/client.rb index 75022b7..efedc0a 100644 --- a/lib/segment/analytics/client.rb +++ b/lib/segment/analytics/client.rb @@ -32,14 +32,11 @@ def initialize(opts = {}) check_write_key! - at_exit do - if @worker_thread - @worker_thread[:should_exit] = true - # Break any Retry-After or backoff sleep so shutdown is not held for - # up to rate_limit_retry_after_cap seconds. - @worker_thread.wakeup if @worker_thread.alive? - end - end + # The worker checks this between sleep slices, so a Retry-After or backoff + # wait is abandoned within a second rather than holding shutdown for up to + # rate_limit_retry_after_cap seconds. Assigning to a dead thread is safe; + # Thread#wakeup is not, and raising here would force a non-zero exit status. + at_exit { @worker_thread[:should_exit] = true if @worker_thread } end # Synchronously waits until the worker has flushed the queue. diff --git a/lib/segment/analytics/retry_budget.rb b/lib/segment/analytics/retry_budget.rb index 5688c68..3032561 100644 --- a/lib/segment/analytics/retry_budget.rb +++ b/lib/segment/analytics/retry_budget.rb @@ -30,9 +30,13 @@ def initialize(options = {}) end def next_backoff_delay - @retries_remaining -= 1 + # Checked before the decrement: decrementing first spent one retry on the + # exhaustion test itself, so a configured N only ever performed N-1, and + # retries: 1 and retries: 0 were indistinguishable. return spent('Retries exhausted for batch') if @retries_remaining <= 0 + @retries_remaining -= 1 + @backoff_start_time ||= monotonic_now return spent('Max total backoff duration exceeded for batch') if elapsed?(@backoff_start_time, @max_total_backoff_duration) diff --git a/lib/segment/analytics/transport.rb b/lib/segment/analytics/transport.rb index e9319a8..108fc68 100644 --- a/lib/segment/analytics/transport.rb +++ b/lib/segment/analytics/transport.rb @@ -120,6 +120,9 @@ def build_http(options) http end + # How long a sliced retry wait sleeps before re-checking for shutdown. + SHUTDOWN_CHECK_INTERVAL = 1 + def new_retry_budget RetryBudget.new( :retries => @retries, @@ -135,15 +138,31 @@ def new_retry_budget # RetryBudget keeps the wait on Transport, where callers stub it. def wait_to_retry(delay, budget) return false if delay.nil? - - sleep(delay) - # Client#shutdown wakes this thread, so the sleep above returns early. - return false if Thread.current[:should_exit] + return false unless interruptible_sleep(delay) budget.record_retry true end + # Sleeps in slices so shutdown is noticed within SHUTDOWN_CHECK_INTERVAL + # rather than after the whole delay, which can be rate_limit_retry_after_cap + # seconds. Returns false if shutdown was requested. + # + # Thread#wakeup is deliberately not used for this: it only cuts short a sleep + # already in progress, so a wakeup arriving while the worker is mid-request is + # lost and the next sleep still runs in full. + def interruptible_sleep(seconds) + remaining = seconds + while remaining > 0 + return false if Thread.current[:should_exit] + + slice = [remaining, SHUTDOWN_CHECK_INTERVAL].min + sleep(slice) + remaining -= slice + end + !Thread.current[:should_exit] + end + def parse_error(body) JSON.parse(body)['error'] rescue StandardError diff --git a/spec/segment/analytics/retry_budget_spec.rb b/spec/segment/analytics/retry_budget_spec.rb new file mode 100644 index 0000000..6350f60 --- /dev/null +++ b/spec/segment/analytics/retry_budget_spec.rb @@ -0,0 +1,49 @@ +# frozen_string_literal: true + +require 'spec_helper' + +module Segment + class Analytics + describe RetryBudget do + let(:logger) { Logger.new(File::NULL) } + + def budget(retries, intervals = nil) + described_class.new( + :retries => retries, + :backoff_policy => FakeBackoffPolicy.new(intervals || Array.new(retries, 1000)), + :max_total_backoff_duration => 43_200, + :max_rate_limit_duration => 43_200, + :rate_limit_retry_after_cap => 300, + :logger => logger + ) + end + + describe '#next_backoff_delay' do + it 'grants exactly as many retries as configured' do + # The count used to be decremented before the exhaustion check, so a + # configured N yielded N-1. go, python and java all grant N. + subject = budget(3) + + expect(subject.next_backoff_delay).to eq(1.0) + expect(subject.next_backoff_delay).to eq(1.0) + expect(subject.next_backoff_delay).to eq(1.0) + expect(subject.next_backoff_delay).to be_nil + end + + it 'grants one retry for retries: 1' do + subject = budget(1) + + expect(subject.next_backoff_delay).to eq(1.0) + expect(subject.next_backoff_delay).to be_nil + end + + it 'grants no retries for retries: 0' do + # retries: 0 and retries: 1 were previously indistinguishable. + subject = budget(0, [1000]) + + expect(subject.next_backoff_delay).to be_nil + end + end + end + end +end diff --git a/spec/segment/analytics/transport_spec.rb b/spec/segment/analytics/transport_spec.rb index e2f6aeb..2f6bda6 100644 --- a/spec/segment/analytics/transport_spec.rb +++ b/spec/segment/analytics/transport_spec.rb @@ -163,7 +163,7 @@ class Analytics let(:status_code) { status_code } let(:body) { body } let(:retries) { 4 } - let(:backoff_policy) { FakeBackoffPolicy.new([1000, 1000, 1000]) } + let(:backoff_policy) { FakeBackoffPolicy.new([1000, 1000, 1000, 1000]) } subject { described_class.new(retries: retries, backoff_policy: backoff_policy) @@ -171,10 +171,10 @@ class Analytics it 'retries the request' do expect(subject) - .to receive(:sleep) - .exactly(retries - 1).times + .to receive(:interruptible_sleep) + .exactly(retries).times .with(1) - .and_return(nil) + .and_return(true) subject.send(write_key, batch) end end @@ -188,7 +188,7 @@ class Analytics it 'does not retry the request' do expect(subject) - .to receive(:sleep) + .to receive(:interruptible_sleep) .never subject.send(write_key, batch) end @@ -208,7 +208,7 @@ class Analytics context '3xx is not retried and is not success' do let(:status_code) { 301 } it 'returns the status without retrying, and does not report success' do - expect(subject).not_to receive(:sleep) + expect(subject).not_to receive(:interruptible_sleep) response = subject.send(write_key, batch) expect(response.status).to eq(301) expect(response.success?).to be false @@ -261,18 +261,18 @@ class Analytics end it 'sleeps for the Retry-After duration' do - expect(subject).to receive(:sleep).with(2).once + expect(subject).to receive(:interruptible_sleep).with(2).once.and_return(true) subject.send(write_key, batch) end it 'caps Retry-After at RATE_LIMIT_RETRY_AFTER_CAP' do allow(response).to receive(:to_hash) { { 'retry-after' => ['9999'] } } - expect(subject).to receive(:sleep).with(described_class::RATE_LIMIT_RETRY_AFTER_CAP).once + expect(subject).to receive(:interruptible_sleep).with(described_class::RATE_LIMIT_RETRY_AFTER_CAP).once.and_return(true) subject.send(write_key, batch) end it 'returns success after retry' do - allow(subject).to receive(:sleep) + allow(subject).to receive(:interruptible_sleep).and_return(true) expect(subject.send(write_key, batch).success?).to be true end end @@ -291,12 +291,12 @@ class Analytics end it 'sleeps for the Retry-After duration' do - expect(subject).to receive(:sleep).with(2).once + expect(subject).to receive(:interruptible_sleep).with(2).once.and_return(true) subject.send(write_key, batch) end it 'does not decrement retries_remaining (uses rate-limit path)' do - allow(subject).to receive(:sleep) + allow(subject).to receive(:interruptible_sleep).and_return(true) # With retries: 1, a 503+Retry-After should NOT exhaust retries because # it uses the rate-limit path (no retry budget cost) transport = described_class.new(retries: 1, backoff_policy: FakeBackoffPolicy.new([1000])) @@ -306,7 +306,7 @@ class Analytics allow(success_response).to receive(:body) { '{}' } allow(success_response).to receive(:to_hash) { {} } allow(http).to receive(:request).and_return(response, success_response) - allow(transport).to receive(:sleep) + allow(transport).to receive(:interruptible_sleep).and_return(true) result = transport.send(write_key, batch) expect(result.status).to eq(200) end @@ -326,12 +326,12 @@ class Analytics end it 'sleeps for the Retry-After duration' do - expect(subject).to receive(:sleep).with(1).once + expect(subject).to receive(:interruptible_sleep).with(1).once.and_return(true) subject.send(write_key, batch) end it 'returns success after retry' do - allow(subject).to receive(:sleep) + allow(subject).to receive(:interruptible_sleep).and_return(true) expect(subject.send(write_key, batch).success?).to be true end @@ -344,7 +344,7 @@ class Analytics allow(success_response).to receive(:body) { '{}' } allow(success_response).to receive(:to_hash) { {} } allow(http).to receive(:request).and_return(response, success_response) - allow(transport).to receive(:sleep) + allow(transport).to receive(:interruptible_sleep).and_return(true) result = transport.send(write_key, batch) expect(result.status).to eq(200) end @@ -352,11 +352,11 @@ class Analytics context 'X-Retry-Count header' do let(:status_code) { 500 } - let(:backoff_policy) { FakeBackoffPolicy.new([1, 1]) } + let(:backoff_policy) { FakeBackoffPolicy.new([1, 1, 1]) } subject { described_class.new(retries: 3, backoff_policy: backoff_policy) } it 'does not send X-Retry-Count on first attempt' do - allow(subject).to receive(:sleep) + allow(subject).to receive(:interruptible_sleep).and_return(true) first_request = nil http = subject.instance_variable_get(:@http) allow(http).to receive(:request) do |req, _| @@ -368,7 +368,7 @@ class Analytics end it 'sends X-Retry-Count incrementing on retries' do - allow(subject).to receive(:sleep) + allow(subject).to receive(:interruptible_sleep).and_return(true) requests = [] http = subject.instance_variable_get(:@http) allow(http).to receive(:request) do |req, _| @@ -378,6 +378,7 @@ class Analytics subject.send(write_key, batch) expect(requests[1]['X-Retry-Count']).to eq('1') expect(requests[2]['X-Retry-Count']).to eq('2') + expect(requests[3]['X-Retry-Count']).to eq('3') end end @@ -395,7 +396,7 @@ class Analytics success_response end - allow(subject).to receive(:sleep) + allow(subject).to receive(:interruptible_sleep).and_return(true) response = subject.send(write_key, batch) @@ -406,7 +407,7 @@ class Analytics it 'gives up once the retry budget is spent' do http = subject.instance_variable_get(:@http) allow(http).to receive(:request).and_raise(Errno::ECONNRESET, 'reset') - allow(subject).to receive(:sleep) + allow(subject).to receive(:interruptible_sleep).and_return(true) response = subject.send(write_key, batch) @@ -480,6 +481,34 @@ class Analytics end end end + + describe '#interruptible_sleep' do + subject { described_class.new } + + it 'abandons the wait when shutdown is requested instead of sleeping it out' do + # The wait used to be a single sleep broken by Thread#wakeup, which only + # interrupts a sleep already in progress and raises ThreadError if the + # thread has finished. Slicing removes the need for it. + elapsed = nil + + worker = Thread.new do + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + result = subject.__send__(:interruptible_sleep, 30) + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + result + end + + sleep 0.1 + worker[:should_exit] = true + + expect(worker.value).to be false + expect(elapsed).to be < 5 + end + + it 'reports completion when the delay elapses' do + expect(Thread.new { subject.__send__(:interruptible_sleep, 0) }.value).to be true + end + end end end end From 38f9ac424c68a3c5c0bc5a722d57029ebf6f5416 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Wed, 23 Sep 2026 12:49:58 -0400 Subject: [PATCH 14/14] Document the backoff pacing change, warn on a policy without reset! MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This branch changed the default backoff pacing — base 100ms to 500ms, ceiling 10s to 60s, multiplier 1.5 to 2 — which aligns ruby with the other SDKs and the agreed shape, but changes the retry schedule for every existing caller on upgrade. History.md had a dedicated upgrade note for the new header and said nothing about this. Now it does, with the option names needed to keep the old pacing. One BackoffPolicy instance serves every batch, so it has to be reset between them or attempt counts accumulate and each batch starts where the last one left off. reset! is not part of the older documented contract, which was next_interval alone, so the call is guarded by respond_to? and a policy predating it still works — but it silently keeps the accumulation this branch fixes for the built-in policy. It now warns, so an integrator finds out from a log line rather than from retries getting slower the longer the process runs. Metrics/ClassLength for transport.rb goes 161 -> 167. That is the second bump today; the file is doing enough now that splitting it is worth considering, but not in this change. 212 examples, rubocop clean, 61-test e2e suite passes. --- .rubocop_todo.yml | 2 +- History.md | 2 ++ lib/segment/analytics/transport.rb | 13 ++++++++++++ spec/segment/analytics/transport_spec.rb | 25 ++++++++++++++++++++++++ 4 files changed, 41 insertions(+), 1 deletion(-) diff --git a/.rubocop_todo.yml b/.rubocop_todo.yml index 423d279..a046775 100644 --- a/.rubocop_todo.yml +++ b/.rubocop_todo.yml @@ -73,7 +73,7 @@ Metrics/BlockLength: # Offense count: 1 # Configuration parameters: CountComments, CountAsOne. Metrics/ClassLength: - Max: 161 + Max: 167 # Offense count: 2 # Configuration parameters: AllowedMethods, AllowedPatterns, IgnoredMethods. diff --git a/History.md b/History.md index f987f90..6abeb7a 100644 --- a/History.md +++ b/History.md @@ -19,6 +19,8 @@ sent the write key as HTTP Basic credentials. * Backoff waits no longer block shutdown for the full delay. * Retry timing uses a monotonic clock, so a system clock change cannot stretch or collapse a backoff. * Backoff intervals are now jittered at the ceiling as well, so clients that back off together do not retry in lockstep. +* **Default backoff pacing changed**: the base wait is 500ms (was 100ms), the ceiling is 60s (was 10s), and the multiplier is 2 (was 1.5). This aligns ruby with the other Segment SDKs, but it does mean a retry schedule that was previously 100ms, 150ms, 225ms… now starts at 500ms and climbs faster. Set `min_timeout_ms`, `max_timeout_ms` and `multiplier` on a `BackoffPolicy` to keep the old pacing. +* A `backoff_policy` supplied by the caller that does not implement `reset!` now logs a warning. One policy instance serves every batch, so without `reset!` its attempt count accumulates and retries get slower the longer the process runs. * Fix `retries` granting one fewer attempt than configured. A configured 10 performed 9, and `retries: 1` performed none at all. 2.5.0 / 2024-07-17 diff --git a/lib/segment/analytics/transport.rb b/lib/segment/analytics/transport.rb index 108fc68..8b689dd 100644 --- a/lib/segment/analytics/transport.rb +++ b/lib/segment/analytics/transport.rb @@ -104,6 +104,19 @@ def configure_retries(options) @retries = options[:retries] || RETRIES @backoff_policy = options[:backoff_policy] || Segment::Analytics::BackoffPolicy.new + + # One policy instance serves every batch, so it has to be reset between + # them or attempt counts accumulate and each batch starts where the last + # one left off. reset! is not part of the older documented contract, which + # was next_interval alone, so a policy predating it still works — but it + # keeps that accumulation, and silently. Say so rather than letting an + # integrator find it as "retries get slower the longer we run". + unless @backoff_policy.respond_to?(:reset!) + logger.warn( + 'backoff_policy does not implement reset!; attempt counts will ' \ + 'accumulate across batches. Add a reset! method that clears them.' + ) + end @max_total_backoff_duration = options[:max_total_backoff_duration] || MAX_TOTAL_BACKOFF_DURATION @max_rate_limit_duration = options[:max_rate_limit_duration] || diff --git a/spec/segment/analytics/transport_spec.rb b/spec/segment/analytics/transport_spec.rb index 2f6bda6..3e2580f 100644 --- a/spec/segment/analytics/transport_spec.rb +++ b/spec/segment/analytics/transport_spec.rb @@ -482,6 +482,31 @@ class Analytics end end + describe 'a backoff_policy without reset!' do + # One policy instance serves every batch, so a policy predating reset! + # keeps accumulating attempts and gets slower the longer a process runs. + let(:legacy_policy) do + Class.new do + def next_interval + 1 + end + end.new + end + + it 'warns that attempt counts will accumulate across batches' do + expect(legacy_policy).not_to respond_to(:reset!) + expect(Segment::Analytics::Logging.logger).to receive(:warn).with(/reset!/) + + described_class.new(:backoff_policy => legacy_policy) + end + + it 'stays quiet for a policy that implements it' do + expect(Segment::Analytics::Logging.logger).not_to receive(:warn) + + described_class.new(:backoff_policy => Segment::Analytics::BackoffPolicy.new) + end + end + describe '#interruptible_sleep' do subject { described_class.new }