[7.3.1] Development into Master

.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

CHANGES.txt

@@ -1,5 +1,8 @@
 CHANGES
 
+7.3.1 (Jul 26, 2021)
+- Updated the synchronization flow to be more reliable in the event of an edge case generating delay in cache purge propagation, keeping the SDK cache properly synced.
+
 7.3.0 (Jul 12, 2021)
 - Updated SDK telemetry storage, metrics and updater to be more effective and send less often.
 - Fixed high cpu usage when api key is wrong.

lib/splitclient-rb.rb

@@ -85,13 +85,13 @@
 require 'splitclient-rb/engine/models/label'
 require 'splitclient-rb/engine/models/treatment'
 require 'splitclient-rb/engine/auth_api_client'
+require 'splitclient-rb/engine/back_off'
 require 'splitclient-rb/engine/push_manager'
 require 'splitclient-rb/engine/sync_manager'
 require 'splitclient-rb/engine/synchronizer'
 require 'splitclient-rb/utilitites'
 
-# SSE 
-require 'splitclient-rb/sse/event_source/back_off'
+# SSE
 require 'splitclient-rb/sse/event_source/client'
 require 'splitclient-rb/sse/event_source/event_parser'
 require 'splitclient-rb/sse/event_source/event_types'

lib/splitclient-rb/cache/fetchers/segment_fetcher.rb

@@ -30,16 +30,19 @@ def call
         def fetch_segments_if_not_exists(names, cache_control_headers = false)
           names.each do |name|
             change_number = @segments_repository.get_change_number(name)
-
-            fetch_segment(name, cache_control_headers) if change_number == -1
+
+            if change_number == -1
+              fetch_options = { cache_control_headers: cache_control_headers, till: nil }
+              fetch_segment(name, fetch_options) if change_number == -1
+            end
           end
         rescue StandardError => error
           @config.log_found_exception(__method__.to_s, error)
         end
 
-        def fetch_segment(name, cache_control_headers = false)
+        def fetch_segment(name, fetch_options = { cache_control_headers: false, till: nil })
           @semaphore.synchronize do
-            segments_api.fetch_segments_by_names([name], cache_control_headers)
+            segments_api.fetch_segments_by_names([name], fetch_options)
           end
         rescue StandardError => error
           @config.log_found_exception(__method__.to_s, error)

lib/splitclient-rb/cache/fetchers/split_fetcher.rb

@@ -27,9 +27,9 @@ def call
           end
         end
 
-        def fetch_splits(cache_control_headers = false)
+        def fetch_splits(fetch_options = { cache_control_headers: false, till: nil })
           @semaphore.synchronize do
-            data = splits_since(@splits_repository.get_change_number, cache_control_headers)
+            data = splits_since(@splits_repository.get_change_number, fetch_options)
 
             data[:splits] && data[:splits].each do |split|
               add_split_unless_archived(split)
@@ -68,8 +68,8 @@ def splits_thread
           end
         end
 
-        def splits_since(since, cache_control_headers = false)
-          splits_api.since(since, cache_control_headers)
+        def splits_since(since, fetch_options = { cache_control_headers: false, till: nil })
+          splits_api.since(since, fetch_options)
         end
 
         def add_split_unless_archived(split)

lib/splitclient-rb/engine/api/segments.rb

@@ -11,13 +11,14 @@ def initialize(api_key, segments_repository, config, telemetry_runtime_producer)
         @telemetry_runtime_producer = telemetry_runtime_producer
       end
 
-      def fetch_segments_by_names(names, cache_control_headers = false)
+      def fetch_segments_by_names(names, fetch_options = { cache_control_headers: false, till: nil })
         return if names.nil? || names.empty?
 
         names.each do |name|
           since = @segments_repository.get_change_number(name)
+
           loop do
-            segment = fetch_segment_changes(name, since, cache_control_headers)
+            segment = fetch_segment_changes(name, since, fetch_options)
             @segments_repository.add_to_segment(segment)
 
             @config.split_logger.log_if_debug("Segment #{name} fetched before: #{since}, \
@@ -32,9 +33,12 @@ def fetch_segments_by_names(names, cache_control_headers = false)
 
       private
 
-      def fetch_segment_changes(name, since, cache_control_headers = false)
+      def fetch_segment_changes(name, since, fetch_options = { cache_control_headers: false, till: nil })
         start = Time.now
-        response = get_api("#{@config.base_uri}/segmentChanges/#{name}", @api_key, { since: since }, cache_control_headers)
+
+        params = { since: since }
+        params[:till] = fetch_options[:till] unless fetch_options[:till].nil?
+        response = get_api("#{@config.base_uri}/segmentChanges/#{name}", @api_key, params, fetch_options[:cache_control_headers])
 
         if response.success?
           segment = JSON.parse(response.body, symbolize_names: true)

lib/splitclient-rb/engine/api/splits.rb

@@ -10,10 +10,12 @@ def initialize(api_key, config, telemetry_runtime_producer)
         @telemetry_runtime_producer = telemetry_runtime_producer
       end
 
-      def since(since, cache_control_headers = false)
+      def since(since, fetch_options = { cache_control_headers: false, till: nil })
         start = Time.now
-
-        response = get_api("#{@config.base_uri}/splitChanges", @api_key, { since: since }, cache_control_headers)
+
+        params = { since: since }
+        params[:till] = fetch_options[:till] unless fetch_options[:till].nil?
+        response = get_api("#{@config.base_uri}/splitChanges", @api_key, params, fetch_options[:cache_control_headers])
         if response.success?
           result = splits_with_segment_names(response.body)
 

lib/splitclient-rb/engine/back_off.rb

@@ -0,0 +1,26 @@
+# frozen_string_literal: false
+
+module SplitIoClient
+  module Engine
+    BACKOFF_MAX_ALLOWED = 1.8
+    class BackOff
+      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 >= @max_allowed ? @max_allowed : interval
+      end
+
+      def reset
+        @attempt = 0
+      end
+    end
+  end
+end

lib/splitclient-rb/engine/push_manager.rb

@@ -8,7 +8,7 @@ def initialize(config, sse_handler, api_key, telemetry_runtime_producer)
         @sse_handler = sse_handler
         @auth_api_client = AuthApiClient.new(@config, telemetry_runtime_producer)
         @api_key = api_key
-        @back_off = SplitIoClient::SSE::EventSource::BackOff.new(@config.auth_retry_back_off_base, 1)
+        @back_off = Engine::BackOff.new(@config.auth_retry_back_off_base, 1)
         @telemetry_runtime_producer = telemetry_runtime_producer
       end
 

lib/splitclient-rb/engine/synchronizer.rb

@@ -6,7 +6,9 @@ class Synchronizer
       include SplitIoClient::Cache::Fetchers
       include SplitIoClient::Cache::Senders
 
-      FORCE_CACHE_CONTROL_HEADERS = true
+      ON_DEMAND_FETCH_BACKOFF_BASE_SECONDS = 10
+      ON_DEMAND_FETCH_BACKOFF_MAX_WAIT_SECONDS = 60
+      ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES = 10
 
       def initialize(
         repositories,
@@ -54,17 +56,116 @@ def stop_periodic_fetch
         @segment_fetcher.stop_segments_thread
       end
 
-      def fetch_splits
-        segment_names = @split_fetcher.fetch_splits(FORCE_CACHE_CONTROL_HEADERS)
-        @segment_fetcher.fetch_segments_if_not_exists(segment_names, FORCE_CACHE_CONTROL_HEADERS) unless segment_names.empty?
+      def fetch_splits(target_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_seconds,
+                                     false)
+
+        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,
+                                     nil,
+                                     true)
+
+        attempts = ON_DEMAND_FETCH_BACKOFF_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.") if @config.debug_enabled
+        else
+          @config.logger.debug("No changes fetched after #{attempts} attempts with CDN bypassed.") if @config.debug_enabled
+        end
+      rescue StandardError => error
+        @config.log_found_exception(__method__.to_s, error)
       end
 
-      def fetch_segment(name)
-        @segment_fetcher.fetch_segment(name, FORCE_CACHE_CONTROL_HEADERS)
+      def fetch_segment(name, target_change_number)
+        return if target_change_number <= @segments_repository.get_change_number(name).to_i
+
+        fetch_options = { cache_control_headers: true, till: nil }
+        result = attempt_segment_sync(name,
+                                      target_change_number,
+                                      fetch_options,
+                                      @config.on_demand_fetch_max_retries,
+                                      @config.on_demand_fetch_retry_delay_seconds,
+                                      false)
+
+        attempts = @config.on_demand_fetch_max_retries - result[:remaining_attempts]
+        if result[:success]
+          @config.logger.debug("Segment #{name} refresh completed in #{attempts} attempts.") if @config.debug_enabled
+
+          return
+        end
+
+        fetch_options = { cache_control_headers: true, till: target_change_number }
+        result = attempt_segment_sync(name,
+                                      target_change_number,
+                                      fetch_options,
+                                      ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES,
+                                      nil,
+                                      true)
+
+        attempts = @config.on_demand_fetch_max_retries - result[:remaining_attempts]
+        if result[:success]
+          @config.logger.debug("Segment #{name} refresh completed bypassing the CDN in #{attempts} attempts.") if @config.debug_enabled
+        else
+          @config.logger.debug("No changes fetched for segment #{name} after #{attempts} attempts with CDN bypassed.") if @config.debug_enabled
+        end
+      rescue StandardError => error
+        @config.log_found_exception(__method__.to_s, error)
       end
 
       private
 
+      def attempt_segment_sync(name, target_cn, fetch_options, max_retries, retry_delay_seconds, with_backoff)
+        remaining_attempts = max_retries
+        backoff = Engine::BackOff.new(ON_DEMAND_FETCH_BACKOFF_BASE_SECONDS, 0, ON_DEMAND_FETCH_BACKOFF_MAX_WAIT_SECONDS) if with_backoff
+
+        loop do
+          remaining_attempts -= 1
+
+          @segment_fetcher.fetch_segment(name, fetch_options)
+
+          return sync_result(true, remaining_attempts) if target_cn <= @segments_repository.get_change_number(name).to_i
+          return sync_result(false, remaining_attempts) if remaining_attempts <= 0
+
+          delay = with_backoff ? backoff.interval : retry_delay_seconds
+          sleep(delay)
+        end
+      end
+
+      def attempt_splits_sync(target_cn, fetch_options, max_retries, retry_delay_seconds, with_backoff)
+        remaining_attempts = max_retries
+        backoff = Engine::BackOff.new(ON_DEMAND_FETCH_BACKOFF_BASE_SECONDS, 0, ON_DEMAND_FETCH_BACKOFF_MAX_WAIT_SECONDS) if with_backoff
+
+        loop do
+          remaining_attempts -= 1
+
+          segment_names = @split_fetcher.fetch_splits(fetch_options)
+
+          return sync_result(true, remaining_attempts, segment_names) if target_cn <= @splits_repository.get_change_number
+          return sync_result(false, remaining_attempts, segment_names) if remaining_attempts <= 0
+
+          delay = with_backoff ? backoff.interval : retry_delay_seconds
+          sleep(delay)
+        end
+      end
+
       def fetch_segments
         @segment_fetcher.fetch_segments
       end
@@ -87,6 +188,10 @@ def impressions_count_sender
       def start_telemetry_sync_task
         Telemetry::SyncTask.new(@config, @telemetry_synchronizer).call
       end
+
+      def sync_result(success, remaining_attempts, segment_names = nil)
+        { success: success, remaining_attempts: remaining_attempts, segment_names: segment_names }
+      end
     end
   end
 end

lib/splitclient-rb/split_config.rb

@@ -113,6 +113,9 @@ def initialize(opts = {})
 
       @sdk_start_time = Time.now
 
+      @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
     end
 
@@ -278,6 +281,17 @@ def initialize(opts = {})
 
     attr_accessor :sdk_start_time
 
+    attr_accessor :on_demand_fetch_retry_delay_seconds    
+    attr_accessor :on_demand_fetch_max_retries
+
+    def self.default_on_demand_fetch_retry_delay_seconds
+      0.05
+    end
+
+    def self.default_on_demand_fetch_max_retries
+      10
+    end
+
     def self.default_impressions_mode
       :optimized
     end

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

@@ -51,11 +51,7 @@ def perform
             cn = item[:change_number]
             @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
-              @synchronizer.fetch_segment(segment_name)
-              attempt += 1
-            end
+            @synchronizer.fetch_segment(segment_name, cn)
           end
         end