splits logic implementation

Author: sanzmauro

lib/splitclient-rb/engine/synchronizer.rb

@@ -6,6 +6,10 @@ 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_MAX_RETRIES = 10
+
       def initialize(
         repositories,
         api_key,
@@ -25,6 +29,7 @@ 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
@@ -52,10 +57,40 @@ def stop_periodic_fetch
         @segment_fetcher.stop_segments_thread
       end
 
-      def fetch_splits
+      def fetch_splits(target_change_number)
+        return if target_change_number <= @splits_repository.get_change_number
+
         fetch_options = { cache_control_headers: true, till: nil }
-        segment_names = @split_fetcher.fetch_splits(fetch_options)
-        @segment_fetcher.fetch_segments_if_not_exists(segment_names, true) unless segment_names.empty?
+
+        result = attempt_splits_sync(target_change_number,
+                                     fetch_options,
+                                     @config.on_demand_fetch_max_retries,
+                                     @config.on_demand_fetch_retry_delay_ms)
+
+        attempts = @config.on_demand_fetch_max_retries - result[:remaining_attempts]
+        if result[:success]
+          @segment_fetcher.fetch_segments_if_not_exists(result[:segment_names], true) unless result[:segment_names].empty?
+          @config.logger.debug("Refresh completed in #{attempts} attempts.") if @config.debug_enabled
+
+          return
+        end
+
+        fetch_options[:till] = 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)
+
+        attempts = @config.on_demand_fetch_max_retries - result[:remaining_attempts]
+
+        if result[:success]
+          @segment_fetcher.fetch_segments_if_not_exists(result[:segment_names], true) unless result[:segment_names].empty?
+          @config.logger.debug("Refresh completed bypassing the CDN in #{attempts} attempts.")
+        else
+          @config.logger.debug("No changes fetched after #{attempts} attempts with CDN bypassed.")
+        end
+      rescue StandardError => error
+        @config.log_found_exception(__method__.to_s, error)
       end
 
       def fetch_segment(name)
@@ -65,6 +100,21 @@ def fetch_segment(name)
 
       private
 
+      def attempt_splits_sync(target_cn, fetch_options, max_retries, retry_delay_ms)
+        remaining_attempts = max_retries
+
+        loop do
+          remaining_attempts -= 1
+
+          segment_names = @split_fetcher.fetch_splits(fetch_options)
+
+          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)
+        end
+      end
+
       def fetch_segments
         @segment_fetcher.fetch_segments
       end
@@ -87,6 +137,10 @@ def impressions_count_sender
       def start_telemetry_sync_task
         Telemetry::SyncTask.new(@config, @telemetry_synchronizer).call
       end
+
+      def split_sync_result(success, remaining_attempts, segment_names)
+        { success: success, remaining_attempts: remaining_attempts, segment_names: segment_names }
+      end
     end
   end
 end

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

@@ -3,8 +3,6 @@
 module SplitIoClient
   module SSE
     module Workers
-      MAX_RETRIES_ALLOWED = 10
-
       class SplitsWorker
         def initialize(synchronizer, config, splits_repository)
           @synchronizer = synchronizer
@@ -62,12 +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}")
-
-            attempt = 0
-            while change_number > @splits_repository.get_change_number.to_i && attempt <= Workers::MAX_RETRIES_ALLOWED
-              @synchronizer.fetch_splits
-              attempt += 1
-            end
+            @synchronizer.fetch_splits
           end
         end