improvements in attempting splits sync

Author: Matias Melograno

splitio/sync/split.py

@@ -41,7 +41,45 @@ def __init__(self, split_api, split_storage):
                                 _ON_DEMAND_FETCH_BACKOFF_BASE,
                                 _ON_DEMAND_FETCH_BACKOFF_MAX_WAIT)
 
-    def attempt_split_sync(self, fetch_options, till=None):
+    def _fetch_until(self, fetch_options, till=None):
+        """
+        Hit endpoint, update storage and return when since==till.
+
+        :param fetch_options Fetch options for getting split definitions.
+        :type fetch_options splitio.api.FetchOptions
+
+        :param till: Passed till from Streaming.
+        :type till: int
+
+        :return: last change number
+        :rtype: int
+        """
+        while True:  # Fetch until since==till
+            change_number = self._split_storage.get_change_number()
+            if change_number is None:
+                change_number = -1
+            if till is not None and till < change_number:
+                # the passed till is less than change_number, no need to perform updates
+                return change_number
+
+            try:
+                split_changes = self._api.fetch_splits(change_number, fetch_options)
+            except APIException as exc:
+                _LOGGER.error('Exception raised while fetching splits')
+                _LOGGER.debug('Exception information: ', exc_info=True)
+                raise exc
+
+            for split in split_changes.get('splits', []):
+                if split['status'] == splits.Status.ACTIVE.value:
+                    self._split_storage.put(splits.from_raw(split))
+                else:
+                    self._split_storage.remove(split['name'])
+
+            self._split_storage.set_change_number(split_changes['till'])
+            if split_changes['till'] == split_changes['since']:
+                return split_changes['till']
+
+    def _attempt_split_sync(self, fetch_options, till=None):
         """
         Hit endpoint, update storage and return True if sync is complete.
 
@@ -58,31 +96,7 @@ def attempt_split_sync(self, fetch_options, till=None):
         remaining_attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES
         while True:
             remaining_attempts -= 1
-            while True:  # Fetch until since==till
-                change_number = self._split_storage.get_change_number()
-                if change_number is None:
-                    change_number = -1
-                if till is not None and till < change_number:
-                    # the passed till is less than change_number, no need to perform updates
-                    break
-
-                try:
-                    split_changes = self._api.fetch_splits(change_number, fetch_options)
-                except APIException as exc:
-                    _LOGGER.error('Exception raised while fetching splits')
-                    _LOGGER.debug('Exception information: ', exc_info=True)
-                    raise exc
-
-                for split in split_changes.get('splits', []):
-                    if split['status'] == splits.Status.ACTIVE.value:
-                        self._split_storage.put(splits.from_raw(split))
-                    else:
-                        self._split_storage.remove(split['name'])
-
-                self._split_storage.set_change_number(split_changes['till'])
-                if split_changes['till'] == split_changes['since']:
-                    break
-
+            change_number = self._fetch_until(fetch_options, till)
             if till is None or till <= change_number:
                 return True, remaining_attempts, change_number
             elif remaining_attempts <= 0:
@@ -98,14 +112,14 @@ def synchronize_splits(self, till=None):
         :type till: int
         """
         fetch_options = FetchOptions(True)  # Set Cache-Control to no-cache
-        successful_sync, remaining_attempts, change_number = self.attempt_split_sync(fetch_options,
-                                                                                     till)
+        successful_sync, remaining_attempts, change_number = self._attempt_split_sync(fetch_options,
+                                                                                      till)
         attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
         if successful_sync:  # succedeed sync
             _LOGGER.debug('Refresh completed in %s attempts.', attempts)
             return
         with_cdn_bypass = FetchOptions(True, change_number)  # Set flag for bypassing CDN
-        without_cdn_successful_sync, remaining_attempts, change_number = self.attempt_split_sync(with_cdn_bypass, till)
+        without_cdn_successful_sync, remaining_attempts, change_number = self._attempt_split_sync(with_cdn_bypass, till)
         without_cdn_attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
         if without_cdn_successful_sync:
             _LOGGER.debug('Refresh completed bypassing the CDN in %s attempts.',

tests/sync/test_splits_synchronizer.py

@@ -142,8 +142,10 @@ def change_number_mock():
             change_number_mock._calls += 1
             if change_number_mock._calls == 1:
                 return -1
-            elif change_number_mock._calls < 8:
+            elif change_number_mock._calls >= 2 and change_number_mock._calls <= 3:
                 return 123
+            elif change_number_mock._calls <= 7:
+                return 1234
             return 12345 # Return proper cn for CDN Bypass
         change_number_mock._calls = 0
         storage.get_change_number.side_effect = change_number_mock
@@ -199,18 +201,29 @@ def get_changes(*args, **kwargs):
                     'since': 123,
                     'till': 123
                 }
-            elif get_changes.called > 2 and get_changes.called < 5: # Simulating condition to equal since and till
+            elif get_changes.called == 3:
                 return {
                     'splits': [],
                     'since': 123,
-                    'till': 12345
+                    'till': 1234
                 }
-            else:
+            elif get_changes.called >= 4 and get_changes.called <= 6:
+                return {
+                    'splits': [],
+                    'since': 1234,
+                    'till': 1234
+                }
+            elif get_changes.called == 7:
                 return {
                     'splits': [],
-                    'since': 12345,
+                    'since': 1234,
                     'till': 12345
                 }
+            return {
+                'splits': [],
+                'since': 12345,
+                'till': 12345
+            }
         get_changes.called = 0
 
         api.fetch_splits.side_effect = get_changes
@@ -223,9 +236,8 @@ def get_changes(*args, **kwargs):
 
         split_synchronizer._backoff = Backoff(1, 0.1)
         split_synchronizer.synchronize_splits(12345)
-
-        assert mocker.call(12345, FetchOptions(True, 123)) in api.fetch_splits.mock_calls
-        assert len(api.fetch_splits.mock_calls) == 8 # 2 ok + 2 since == till + 3 backoff + 1 CDN
+        assert mocker.call(12345, FetchOptions(True, 1234)) in api.fetch_splits.mock_calls
+        assert len(api.fetch_splits.mock_calls) == 8 # 2 ok + BACKOFF(2 since==till + 2 re-attempts) + CDN(2 since==till)
 
         inserted_split = storage.put.mock_calls[0][1][0]
         assert isinstance(inserted_split, Split)