feat: expand sequence marker window size and add note

Author: veronika-alraheem

ibmcloudant/features/changes_follower.py

@@ -44,8 +44,8 @@
 # before the client timeout it is set to 3 seconds less.
 _LONGPOLL_TIMEOUT = _MIN_CLIENT_TIMEOUT - 3000
 _BATCH_SIZE = 10000
-_SEQ_MARKERS_CAPACITY = 200
-_SEQ_MARKERS_EVICTION_COUNT = _SEQ_MARKERS_CAPACITY // 10
+_SEQ_RETAINED_SIZE = 2048
+_SEQ_EVICTION_COUNT = _SEQ_RETAINED_SIZE // 2
 
 # Base delay in milliseconds between unsuccessful attempts to pull changes feed
 # in presence of transient errors
@@ -62,12 +62,12 @@ class _Mode(Enum):
     LISTEN = auto()
 
 
-class _SeqEntryType(Enum):
+class _SeqMarkerType(Enum):
     """
     Enum for the type of a seq marker entry tracked by the changes follower.
     """
     ROW = auto()
-    PAGE = auto()
+    LAST = auto()
 
 
 class _TransientErrorSuppression(Enum):
@@ -194,7 +194,7 @@ def last_seq_since(self, last_persisted_seq: str) -> str:
         result = None
         for entry in markers:
             if found:
-                if entry['type'] == _SeqEntryType.ROW:
+                if entry['seqMarkerType'] == _SeqMarkerType.ROW:
                     break
                 if entry['seq'] is not None:
                     result = entry['seq']
@@ -207,16 +207,16 @@ def _update_seq_markers(self, results: list, last_seq: str) -> None:
         """
         Update the seq markers list with entries from a completed page.
 
-        Evicts the oldest entries if the list is at capacity, then appends
-        a ROW entry for the last change item (if any) and a PAGE entry for
-        the page's last_seq.
+        Bulk-evicts the oldest entries if the list is at or above the eviction
+        threshold, then appends a ROW entry for the last change item (if any)
+        and a LAST entry for the page's last_seq.
         """
         with self._seq_markers_lock:
-            if len(self._seq_markers) >= _SEQ_MARKERS_CAPACITY:
-                del self._seq_markers[:_SEQ_MARKERS_EVICTION_COUNT]
+            if len(self._seq_markers) >= (_SEQ_RETAINED_SIZE + _SEQ_EVICTION_COUNT):
+                del self._seq_markers[:_SEQ_EVICTION_COUNT]
             if len(results) > 0:
-                self._seq_markers.append({'type': _SeqEntryType.ROW, 'seq': results[-1].get('seq')})
-            self._seq_markers.append({'type': _SeqEntryType.PAGE, 'seq': last_seq})
+                self._seq_markers.append({'seqMarkerType': _SeqMarkerType.ROW, 'seq': results[-1].get('seq')})
+            self._seq_markers.append({'seqMarkerType': _SeqMarkerType.LAST, 'seq': last_seq})
 
     def _request_callback(self):
         while True:
@@ -501,6 +501,12 @@ def latest_sequence_from(self, checkpoint_sequence_id: str) -> str:
         non-``None`` ``seq`` and persist the returned value to use as the
         ``since`` parameter for the next run.
 
+        Note: the follower retains a bounded window of recent sequence IDs.
+        If subsequent page fetches displace the supplied
+        ``checkpoint_sequence_id`` from that window, this method returns
+        the supplied value unchanged. Call this method and persist its result
+        regularly to obtain a fresh sequence ID.
+
         :param str checkpoint_sequence_id: The last checkpoint sequence ID
             — either the non-``None`` ``seq`` of the last ``ChangesResultItem``
             fully processed, or a value previously returned by this method.

test/unit/features/test_changes_follower.py

@@ -33,8 +33,8 @@
     _BATCH_SIZE,
     _FOREVER,
     _LONGPOLL_TIMEOUT,
-    _SEQ_MARKERS_CAPACITY,
-    _SEQ_MARKERS_EVICTION_COUNT,
+    _SEQ_RETAINED_SIZE,
+    _SEQ_EVICTION_COUNT,
     ChangesFollower,
     _Mode,
 )
@@ -806,7 +806,7 @@ def test_last_seq_since_followed_by_non_empty(self):
 
     # -----------------------------------------------------------------------
     # Per-page-type: followed by an empty page (type 9 at base 20)
-    # Page 2 inserts only PAGE('20-aa') — no ROW to block, advances to '20-aa'.
+    # Page 2 inserts only LAST('20-aa') — no ROW to block, advances to '20-aa'.
     # -----------------------------------------------------------------------
 
     def test_last_seq_since_followed_by_empty(self):
@@ -830,7 +830,7 @@ def test_last_seq_since_followed_by_empty(self):
 
     # -----------------------------------------------------------------------
     # All 8 three-page sequences of empty (E=type 9) and non-empty (N=type 1).
-    # Query from page 1's last_seq key. E adds only PAGE; N adds ROW+PAGE.
+    # Query from page 1's last_seq key. E adds only LAST; N adds ROW+LAST.
     # -----------------------------------------------------------------------
 
     def test_last_seq_since_3_page_sequence(self):
@@ -860,23 +860,27 @@ def test_last_seq_since_3_page_sequence(self):
 
     # -----------------------------------------------------------------------
     # Eviction
-    # Each non-empty page (type 2) adds 2 entries (ROW + PAGE).
-    # With CAPACITY=200 and EVICTION_COUNT=20, adding 101 pages triggers one
-    # eviction of the oldest 20 entries (first 10 pages).
+    # Each non-empty page (type 2) adds 2 entries (ROW + LAST). Adding enough
+    # pages to exceed RETAINED_SIZE + EVICTION_COUNT entries triggers one
+    # eviction of the oldest EVICTION_COUNT entries.
     # Entries for page 0 (base=0) should be gone; most recent should remain.
     # -----------------------------------------------------------------------
 
     def test_last_seq_since_eviction(self):
-        pages = [_page_type(2, i * 10) for i in range(101)]
+        entries_per_non_empty_page = 2
+        target_entries = _SEQ_RETAINED_SIZE + _SEQ_EVICTION_COUNT + entries_per_non_empty_page
+        page_count = target_entries // entries_per_non_empty_page
+        pages = [_page_type(2, i * 10) for i in range(page_count)]
         iterator = _populate_iterator(pages)
 
-        # Page 0 (base=0): row=_seq(1), page=_seq(2) — evicted
+        # Page 0 (base=0): row=_seq(1), last=_seq(2) — evicted
         self.assertEqual(iterator.last_seq_since(_seq(1)), _seq(1))
         self.assertEqual(iterator.last_seq_since(_seq(2)), _seq(2))
 
-        # Most recent page (base=1000): row=_seq(1001), page=_seq(1002) — still present
-        self.assertEqual(iterator.last_seq_since(_seq(1001)), _seq(1002))
-        self.assertEqual(iterator.last_seq_since(_seq(1002)), _seq(1002))
+        # Most recent page (base=(page_count-1)*10): row and last — still present
+        last_base = (page_count - 1) * 10
+        self.assertEqual(iterator.last_seq_since(_seq(last_base + 1)), _seq(last_base + 2))
+        self.assertEqual(iterator.last_seq_since(_seq(last_base + 2)), _seq(last_base + 2))
 
     # -----------------------------------------------------------------------
     # None seq row (seq_interval scenario) — skipped, returns input unchanged
@@ -929,9 +933,9 @@ def test_last_seq_since_consecutive_none_last_seq_pages(self):
 
     def test_last_seq_since_none_last_seq_advances_beyond(self):
         """
-        p1 (type 9, base=10): PAGE('10-aa')
-        p2 (None last_seq):   PAGE(None)     — skipped, scan continues
-        p3 (type 9, base=30): PAGE('30-aa')  — advances to here
+        p1 (type 9, base=10): LAST('10-aa')
+        p2 (None last_seq):   LAST(None)     — skipped, scan continues
+        p3 (type 9, base=30): LAST('30-aa')  — advances to here
         """
         none_page = {'results': [], 'last_seq': None, 'pending': 0}
         pages = [_page_type(9, 10), none_page, _page_type(9, 30)]