Add implementation of returnable last_seq

ibmcloudant/features/changes_follower.py

@@ -23,7 +23,7 @@
 from datetime import datetime, timezone, timedelta
 import functools
 from queue import Queue
-from threading import Thread, Event
+from threading import Thread, Event, Lock
 
 from enum import Enum, auto
 from typing import Dict, Iterator
@@ -44,6 +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
 
 # Base delay in milliseconds between unsuccessful attempts to pull changes feed
 # in presence of transient errors
@@ -60,6 +62,14 @@ class _Mode(Enum):
     LISTEN = auto()
 
 
+class _SeqEntryType(Enum):
+    """
+    Enum for the type of a seq marker entry tracked by the changes follower.
+    """
+    ROW = auto()
+    PAGE = auto()
+
+
 class _TransientErrorSuppression(Enum):
     """
     Enums for changes follower's transient errors suppression mode.
@@ -107,6 +117,8 @@ def __init__(
         self._limit = None
         self._stop = Event()
         self.logger = logging.getLogger(__name__)
+        self._seq_markers: list = []
+        self._seq_markers_lock = Lock()
 
     @property
     def limit(self) -> int:
@@ -164,6 +176,47 @@ def __next__(self):
                 )
                 self._buffer.task_done()
 
+    def last_seq_since(self, last_persisted_seq: str) -> str:
+        """
+        Return the newest sequence ID that is safe to use as a checkpoint
+        after the given persisted sequence ID.
+
+        Walks forward through the retained seq markers from the given
+        last_persisted_seq, fast-forwarding through consecutive page entries
+        to return the furthest safe last_seq without advancing past later
+        change rows that might not yet have been processed.
+
+        Returns last_persisted_seq unchanged if not found in the markers.
+        """
+        with self._seq_markers_lock:
+            markers = list(self._seq_markers)
+        found = False
+        result = None
+        for entry in markers:
+            if found:
+                if entry['type'] == _SeqEntryType.ROW:
+                    break
+                result = entry['seq']
+            elif entry['seq'] == last_persisted_seq:
+                found = True
+                result = entry['seq']
+        return result if found else last_persisted_seq
+
+    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.
+        """
+        with self._seq_markers_lock:
+            if len(self._seq_markers) >= _SEQ_MARKERS_CAPACITY:
+                del self._seq_markers[:_SEQ_MARKERS_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})
+
     def _request_callback(self):
         while True:
             try:
@@ -179,6 +232,7 @@ def _request_callback(self):
                     self._has_next = False
                 results = result['results']
                 self.logger.debug(f'_request_callback results {results}')
+                self._update_seq_markers(results, self.since)
                 self._buffer.join()
                 if self._stop.is_set():
                     raise StopIteration
@@ -432,6 +486,40 @@ def stop(self) -> None:
         """
         self._iter.stop()
 
+    def get_last_seq_newer_than(self, last_persisted_seq: str) -> str:
+        """
+        Return the newest sequence ID that is safe to use as a checkpoint
+        after the given persisted sequence ID.
+
+        Use this after fully processing a ``ChangesResultItem`` to determine
+        whether this ``ChangesFollower`` has observed a later safe checkpoint.
+        This is useful for filtered or sparse changes feeds, where the feed
+        can advance across pages even when no additional user-processable
+        change rows are returned.
+
+        The supplied sequence ID must be the ``seq`` of a
+        ``ChangesResultItem`` that your application has fully processed and
+        already persisted. This method returns a newer sequence only when
+        doing so does not advance past later change rows that might not yet
+        have been processed by your application.
+
+        :param str last_persisted_seq: The ``seq`` of the last
+            ``ChangesResultItem`` that your application has fully processed
+            and persisted.
+        :raises ValueError: If the provided sequence ID is null or empty.
+        :return: The newest safe sequence ID to persist as
+            ``PostChangesParams.since``. Returns the supplied ID unchanged
+            if no newer safe checkpoint is available, the feed has not
+            started yet, or the supplied ID is not present in this
+            ``ChangesFollower`` instance's retained sequence history.
+        :rtype: str
+        """
+        if not last_persisted_seq:
+            raise ValueError('The provided sequence ID cannot be null or empty.')
+        if self._iter is None:
+            return last_persisted_seq
+        return self._iter.last_seq_since(last_persisted_seq)
+
     def _run(self, mode: _Mode):
         if self._iter is not None:
             raise RuntimeError('Cannot start a feed that has already started.')