splits retries implementation

Author: sanzmauro

.rubocop.yml

@@ -31,6 +31,7 @@ Metrics/LineLength:
     - spec/engine/sync_manager_spec.rb
     - spec/engine/auth_api_client_spec.rb
     - spec/telemetry/synchronizer_spec.rb
+    - spec/splitclient/split_config_spec.rb
 
 Style/BracesAroundHashParameters:
   Exclude:
@@ -62,3 +63,4 @@ AllCops:
     - lib/splitclient-rb/engine/models/**/*
     - lib/splitclient-rb/engine/parser/**/*
     - spec/telemetry/synchronizer_spec.rb
+    - lib/splitclient-rb/engine/synchronizer.rb

lib/splitclient-rb/engine/synchronizer.rb

@@ -6,8 +6,8 @@ class Synchronizer
       include SplitIoClient::Cache::Fetchers
       include SplitIoClient::Cache::Senders
 
-      ON_DEMAND_FETCH_BACKOFF_BASE_MS = 10_000
-      ON_DEMAND_FETCH_BACKOFF_MAX_WAIT_MS = 60_000
+      ON_DEMAND_FETCH_BACKOFF_BASE_SECONDS = 10
+      ON_DEMAND_FETCH_BACKOFF_MAX_WAIT_SECONDS = 60
       ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES = 10
 
       def initialize(
@@ -29,7 +29,6 @@ def initialize(
         @impressions_api = SplitIoClient::Api::Impressions.new(@api_key, @config, params[:telemetry_runtime_producer])
         @impression_counter = params[:imp_counter]
         @telemetry_synchronizer = params[:telemetry_synchronizer]
-        @backoff = new SSE::EventSource::BackOff.new()
       end
 
       def sync_all
@@ -58,14 +57,15 @@ def stop_periodic_fetch
       end
 
       def fetch_splits(target_change_number)
-        return if target_change_number <= @splits_repository.get_change_number
+        return if target_change_number <= @splits_repository.get_change_number.to_i
 
         fetch_options = { cache_control_headers: true, till: nil }
 
         result = attempt_splits_sync(target_change_number,
                                      fetch_options,
                                      @config.on_demand_fetch_max_retries,
-                                     @config.on_demand_fetch_retry_delay_ms)
+                                     @config.on_demand_fetch_retry_delay_seconds,
+                                     false)
 
         attempts = @config.on_demand_fetch_max_retries - result[:remaining_attempts]
         if result[:success]
@@ -79,7 +79,8 @@ def fetch_splits(target_change_number)
         result = attempt_splits_sync(target_change_number,
                                      fetch_options,
                                      ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES,
-                                     @config.on_demand_fetch_retry_delay_ms)
+                                     nil,
+                                     true)
 
         attempts = @config.on_demand_fetch_max_retries - result[:remaining_attempts]
 
@@ -100,8 +101,9 @@ def fetch_segment(name)
 
       private
 
-      def attempt_splits_sync(target_cn, fetch_options, max_retries, retry_delay_ms)
+      def attempt_splits_sync(target_cn, fetch_options, max_retries, retry_delay_seconds, with_backoff)
         remaining_attempts = max_retries
+        backoff = SSE::EventSource::BackOff.new(ON_DEMAND_FETCH_BACKOFF_BASE_SECONDS, 0, ON_DEMAND_FETCH_BACKOFF_MAX_WAIT_SECONDS) if with_backoff
 
         loop do
           remaining_attempts -= 1
@@ -111,7 +113,8 @@ def attempt_splits_sync(target_cn, fetch_options, max_retries, retry_delay_ms)
           return split_sync_result(true, remaining_attempts, segment_names) if target_cn <= @splits_repository.get_change_number
           return split_sync_result(false, remaining_attempts, segment_names) if remaining_attempts <= 0
 
-          sleep(retry_delay_ms)
+          delay = with_backoff ? backoff.interval : retry_delay_seconds
+          sleep(delay)
         end
       end
 

lib/splitclient-rb/split_config.rb

@@ -113,7 +113,7 @@ def initialize(opts = {})
 
       @sdk_start_time = Time.now
 
-      @on_demand_fetch_retry_delay_ms = SplitConfig.default_on_demand_fetch_retry_delay_ms
+      @on_demand_fetch_retry_delay_seconds = SplitConfig.default_on_demand_fetch_retry_delay_seconds
       @on_demand_fetch_max_retries = SplitConfig.default_on_demand_fetch_max_retries
 
       startup_log
@@ -281,11 +281,11 @@ def initialize(opts = {})
 
     attr_accessor :sdk_start_time
 
-    attr_accessor :on_demand_fetch_retry_delay_ms    
+    attr_accessor :on_demand_fetch_retry_delay_seconds    
     attr_accessor :on_demand_fetch_max_retries
 
-    def self.default_on_demand_fetch_retry_delay_ms
-      50
+    def self.default_on_demand_fetch_retry_delay_seconds
+      0.05
     end
 
     def self.default_on_demand_fetch_max_retries

lib/splitclient-rb/sse/event_source/back_off.rb

@@ -3,17 +3,20 @@
 module SplitIoClient
   module SSE
     module EventSource
+      BACKOFF_MAX_ALLOWED = 1.8
       class BackOff
-        def initialize(back_off_base, attempt = 0)
+        def initialize(back_off_base, attempt = 0, max_allowed = BACKOFF_MAX_ALLOWED)
           @attempt = attempt
           @back_off_base = back_off_base
+          @max_allowed = max_allowed
         end
 
         def interval
+          interval = 0
           interval = (@back_off_base * (2**@attempt)) if @attempt.positive?
           @attempt += 1
 
-          interval || 0
+          interval >= @max_allowed ? @max_allowed : interval
         end
 
         def reset

lib/splitclient-rb/sse/workers/segments_worker.rb

@@ -3,6 +3,7 @@
 module SplitIoClient
   module SSE
     module Workers
+      MAX_RETRIES_ALLOWED = 10
       class SegmentsWorker
         def initialize(synchronizer, config, segments_repository)
           @synchronizer = synchronizer
@@ -52,7 +53,7 @@ def perform
             @config.logger.debug("SegmentsWorker change_number dequeue #{segment_name}, #{cn}")
 
             attempt = 0
-            while cn > @segments_repository.get_change_number(segment_name).to_i && attempt <= Workers::MAX_RETRIES_ALLOWED
+            while cn > @segments_repository.get_change_number(segment_name).to_i && attempt <= MAX_RETRIES_ALLOWED
               @synchronizer.fetch_segment(segment_name)
               attempt += 1
             end

lib/splitclient-rb/sse/workers/splits_worker.rb

@@ -60,7 +60,7 @@ def kill_split(change_number, split_name, default_treatment)
         def perform
           while (change_number = @queue.pop)
             @config.logger.debug("SplitsWorker change_number dequeue #{change_number}")
-            @synchronizer.fetch_splits
+            @synchronizer.fetch_splits(change_number)
           end
         end
 

spec/engine/synchronizer_spec.rb

@@ -92,10 +92,43 @@
     mock_segment_changes('segment1', segment1, '-1')
     mock_segment_changes('segment1', segment1, '1470947453877')
 
-    synchronizer.fetch_splits
+    synchronizer.fetch_splits(0)
     expect(a_request(:get, 'https://sdk.split.io/api/splitChanges?since=-1')).to have_been_made.once
   end
 
+  it 'fetch_splits - ' do
+    sync = subject.new(repositories, api_key, config, sdk_blocker, parameters)
+    stub_request(:get, 'https://sdk.split.io/api/splitChanges?since=-1')
+      .to_return(status: 200, body:
+    '{
+      "splits": [],
+      "since": -1,
+      "till": 1506703262918
+    }')
+
+    stub_request(:get, 'https://sdk.split.io/api/splitChanges?since=1506703262918')
+      .to_return(status: 200, body:
+    '{
+      "splits": [],
+      "since": 1506703262918,
+      "till": 1506703262918
+    }')
+
+    stub_request(:get, 'https://sdk.split.io/api/splitChanges?since=1506703262918&till=1506703262920')
+      .to_return(status: 200, body:
+    '{
+      "splits": [],
+      "since": 1506703262918,
+      "till": 1506703262921
+    }')
+
+    sync.fetch_splits(1_506_703_262_920)
+
+    expect(a_request(:get, 'https://sdk.split.io/api/splitChanges?since=-1')).to have_been_made.once
+    expect(a_request(:get, 'https://sdk.split.io/api/splitChanges?since=1506703262918')).to have_been_made.times(9)
+    expect(a_request(:get, 'https://sdk.split.io/api/splitChanges?since=1506703262918&till=1506703262920')).to have_been_made.once
+  end
+
   it 'fetch_segment' do
     mock_segment_changes('segment3', segment3, '-1')
 

spec/splitclient/split_config_spec.rb

@@ -31,7 +31,7 @@
       expect(configs.ip_addresses_enabled).to eq default_ip
       expect(configs.machine_name).to eq SplitIoClient::SplitConfig.machine_hostname(default_ip, nil, :redis)
       expect(configs.machine_ip).to eq SplitIoClient::SplitConfig.machine_ip(default_ip, nil, :redis)
-      expect(configs.on_demand_fetch_retry_delay_ms).to eq SplitIoClient::SplitConfig.default_on_demand_fetch_retry_delay_ms
+      expect(configs.on_demand_fetch_retry_delay_seconds).to eq SplitIoClient::SplitConfig.default_on_demand_fetch_retry_delay_seconds
       expect(configs.on_demand_fetch_max_retries).to eq SplitIoClient::SplitConfig.default_on_demand_fetch_max_retries
     end
 

spec/sse/event_source/back_off_spec.rb

@@ -9,7 +9,7 @@
   let(:log) { StringIO.new }
 
   it 'get intervals and reset attemps' do
-    back_off = subject.new(1)
+    back_off = subject.new(1, 0, 5)
 
     firts_interval = back_off.interval
     expect(firts_interval).to eq(0)
@@ -27,7 +27,7 @@
 
   it 'with custom config' do
     streaming_reconnect_back_off_base = 5
-    back_off = subject.new(streaming_reconnect_back_off_base)
+    back_off = subject.new(streaming_reconnect_back_off_base, 0, 30)
 
     firts_interval = back_off.interval
     expect(firts_interval).to eq(0)
@@ -42,4 +42,34 @@
     reset_interval = back_off.interval
     expect(reset_interval).to eq(0)
   end
+
+  it 'with max' do
+    streaming_reconnect_back_off_base = 5
+    back_off = subject.new(streaming_reconnect_back_off_base, 0, 30)
+
+    firts_interval = back_off.interval
+    expect(firts_interval).to eq(0)
+
+    second_interval = back_off.interval
+    expect(second_interval).to eq(10)
+
+    third_interval = back_off.interval
+    expect(third_interval).to eq(20)
+
+    third_interval = back_off.interval
+    expect(third_interval).to eq(30)
+
+    third_interval = back_off.interval
+    expect(third_interval).to eq(30)
+
+    third_interval = back_off.interval
+    expect(third_interval).to eq(30)
+
+    third_interval = back_off.interval
+    expect(third_interval).to eq(30)
+
+    back_off.reset
+    reset_interval = back_off.interval
+    expect(reset_interval).to eq(0)
+  end
 end

spec/sse/sse_handler_spec.rb

@@ -62,7 +62,7 @@
     mock_segment_changes('segment2', segment2, '1470947453878')
     mock_segment_changes('segment3', segment3, '-1')
 
-    synchronizer.fetch_splits
+    synchronizer.fetch_splits(0)
   end
 
   context 'SPLIT UPDATE event' do