From 21cf0dc25576bc07e0a93c9784ef818b9cc5b6fe Mon Sep 17 00:00:00 2001 From: Veronika Al Raheem Date: Tue, 25 Aug 2026 12:28:35 +0200 Subject: [PATCH] feat: latestSequenceFrom for sparse changes feeds --- ibmcloudant/features/changes_follower.py | 92 ++++- test/unit/features/test_changes_follower.py | 354 ++++++++++++++++++++ 2 files changed, 445 insertions(+), 1 deletion(-) diff --git a/ibmcloudant/features/changes_follower.py b/ibmcloudant/features/changes_follower.py index 004afce2..1da8065a 100644 --- a/ibmcloudant/features/changes_follower.py +++ b/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_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 @@ -60,6 +62,14 @@ class _Mode(Enum): LISTEN = auto() +class _SeqMarkerType(Enum): + """ + Enum for the type of a seq marker entry tracked by the changes follower. + """ + ROW = auto() + LAST = 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,48 @@ 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['seqMarkerType'] == _SeqMarkerType.ROW: + break + if entry['seq'] is not None: + 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. + + 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_RETAINED_SIZE + _SEQ_EVICTION_COUNT): + del self._seq_markers[:_SEQ_EVICTION_COUNT] + if len(results) > 0: + 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: try: @@ -179,6 +233,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 +487,41 @@ def stop(self) -> None: """ self._iter.stop() + def latest_sequence_from(self, checkpoint_sequence_id: str) -> str: + """ + Returns the most recent sequence ID that is safe to use as a + checkpoint, advancing beyond the supplied checkpoint sequence ID + where possible. + + With highly filtered changes feeds, multiple pages can pass through + the follower without returning any changes. Using only the ``seq`` + of the last processed ``ChangesResultItem`` in those cases causes a + long changes feed rewind on the next run. To avoid this, call this + method after fully processing each ``ChangesResultItem`` with a + 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. + :return: The most recent safe sequence ID to use as a checkpoint, + or the supplied value if no newer sequence is available. + :rtype: str + + Throws ValueError if ``checkpoint_sequence_id`` is ``None`` or empty. + """ + if not checkpoint_sequence_id: + raise ValueError('Provided sequence ID must be a non-empty string.') + if self._iter is None: + return checkpoint_sequence_id + return self._iter.last_seq_since(checkpoint_sequence_id) + def _run(self, mode: _Mode): if self._iter is not None: raise RuntimeError('Cannot start a feed that has already started.') diff --git a/test/unit/features/test_changes_follower.py b/test/unit/features/test_changes_follower.py index e31a6d17..867a7220 100644 --- a/test/unit/features/test_changes_follower.py +++ b/test/unit/features/test_changes_follower.py @@ -20,6 +20,7 @@ import sys import timeit +import unittest import pytest import responses @@ -32,6 +33,8 @@ _BATCH_SIZE, _FOREVER, _LONGPOLL_TIMEOUT, + _SEQ_RETAINED_SIZE, + _SEQ_EVICTION_COUNT, ChangesFollower, _Mode, ) @@ -650,3 +653,354 @@ def test_retry_delay(self): self.fail("There should be no exception.") self.assertEqual(count, 0, "There should be no changes.") self.assertLessEqual(resp.call_count, 15, "Call count should not exceed limit.") + +def _seq(n): + """Build a seq string from an integer, e.g. _seq(11) -> '11-aa'.""" + return f'{n}-aa' + + +def _make_row(seq): + """Build a raw changes result item dict.""" + return {'id': 'doc', 'seq': seq, 'changes': []} + + +def _page_type(page_type, base): + """ + Factory for the 10 page types. + + Type 1: rows=[b, b+1], last_seq=b+1 (last row == last_seq, no nulls) + Type 2: rows=[b, b+1], last_seq=b+2 (last row != last_seq, no nulls) + Type 3: rows=[null, b+1], last_seq=b+1 (leading null, last row == last_seq) + Type 4: rows=[null, b+1], last_seq=b+2 (leading null, last row != last_seq) + Type 5: rows=[b, null], last_seq=b+1 (trailing null last row) + Type 6: rows=[b, null], last_seq=b+2 (trailing null last row, last_seq beyond) + Type 7: rows=[null, null], last_seq=b+1 (all nulls) + Type 8: rows=[null, null], last_seq=b+2 (all nulls, last_seq beyond) + Type 9: rows=[], last_seq=b (empty page) + Type 10: rows=[b, b+1], last_seq=None (None last_seq) + """ + if page_type == 1: + return {'results': [_make_row(_seq(base)), _make_row(_seq(base + 1))], + 'last_seq': _seq(base + 1), 'pending': 0} + elif page_type == 2: + return {'results': [_make_row(_seq(base)), _make_row(_seq(base + 1))], + 'last_seq': _seq(base + 2), 'pending': 0} + elif page_type == 3: + return {'results': [_make_row(None), _make_row(_seq(base + 1))], + 'last_seq': _seq(base + 1), 'pending': 0} + elif page_type == 4: + return {'results': [_make_row(None), _make_row(_seq(base + 1))], + 'last_seq': _seq(base + 2), 'pending': 0} + elif page_type == 5: + return {'results': [_make_row(_seq(base)), _make_row(None)], + 'last_seq': _seq(base + 1), 'pending': 0} + elif page_type == 6: + return {'results': [_make_row(_seq(base)), _make_row(None)], + 'last_seq': _seq(base + 2), 'pending': 0} + elif page_type == 7: + return {'results': [_make_row(None), _make_row(None)], + 'last_seq': _seq(base + 1), 'pending': 0} + elif page_type == 8: + return {'results': [_make_row(None), _make_row(None)], + 'last_seq': _seq(base + 2), 'pending': 0} + elif page_type == 9: + return {'results': [], 'last_seq': _seq(base), 'pending': 0} + elif page_type == 10: + return {'results': [_make_row(_seq(base)), _make_row(_seq(base + 1))], + 'last_seq': None, 'pending': 0} + else: + raise ValueError(f'Unknown page type: {page_type}') + + +def _populate_iterator(pages): + """ + Build a _ChangesFollowerIterator and populate its _seq_markers by calling + _update_seq_markers for each page. + """ + from ibmcloudant.features.changes_follower import _ChangesFollowerIterator + from threading import Lock + + iterator = _ChangesFollowerIterator.__new__(_ChangesFollowerIterator) + iterator._seq_markers = [] + iterator._seq_markers_lock = Lock() + + for page in pages: + iterator._update_seq_markers(page['results'], page['last_seq']) + + return iterator + + +def _last_seq_since(pages, query_seq): + """Populate an iterator with pages and call last_seq_since directly.""" + return _populate_iterator(pages).last_seq_since(query_seq) + + +# --------------------------------------------------------------------------- +# TestSeqMarkers — unit tests for _ChangesFollowerIterator.last_seq_since +# --------------------------------------------------------------------------- + +class TestSeqMarkers(unittest.TestCase): + + # ----------------------------------------------------------------------- + # Not-found / empty edge cases + # ----------------------------------------------------------------------- + + def test_last_seq_since_not_found(self): + """Returns the input seq unchanged when not found in markers.""" + result = _last_seq_since([_page_type(1, 10)], '999-ff') + self.assertEqual(result, '999-ff') + + def test_last_seq_since_empty_seq_markers(self): + """Returns the input seq unchanged when markers are empty.""" + result = _last_seq_since([], '1-aa') + self.assertEqual(result, '1-aa') + + # ----------------------------------------------------------------------- + # Per-page-type: single page + # ----------------------------------------------------------------------- + + def test_last_seq_since_single_page(self): + cases = [ + ('Type 1: last row seq (== last_seq)', 1, 10, _seq(11), _seq(11)), + ('Type 3: last row seq (== last_seq)', 3, 10, _seq(11), _seq(11)), + ('Type 2: last row seq -> last_seq', 2, 10, _seq(11), _seq(12)), + ('Type 2: last_seq key -> itself', 2, 10, _seq(12), _seq(12)), + ('Type 4: last row seq -> last_seq', 4, 10, _seq(11), _seq(12)), + ('Type 4: last_seq key -> itself', 4, 10, _seq(12), _seq(12)), + ('Type 5: non-stored row seq unchanged', 5, 10, _seq(10), _seq(10)), + ('Type 5: last_seq key -> itself', 5, 10, _seq(11), _seq(11)), + ('Type 6: non-stored row seq unchanged', 6, 10, _seq(10), _seq(10)), + ('Type 6: last_seq key -> itself', 6, 10, _seq(12), _seq(12)), + ('Type 7: last_seq key -> itself', 7, 10, _seq(11), _seq(11)), + ('Type 8: last_seq key -> itself', 8, 10, _seq(12), _seq(12)), + ('Type 9: last_seq key -> itself', 9, 10, _seq(10), _seq(10)), + ] + for name, page_type, base, query_seq, expected in cases: + with self.subTest(name): + result = _last_seq_since([_page_type(page_type, base)], query_seq) + self.assertEqual(result, expected) + + # ----------------------------------------------------------------------- + # Per-page-type: followed by a non-empty page (type 1 at base 20) + # Page 2 inserts ROW('21-aa') which blocks advancement. + # ----------------------------------------------------------------------- + + def test_last_seq_since_followed_by_non_empty(self): + cases = [ + ('Type 1 + non-empty: blocked by p2 ROW', 1, 10, _seq(11), _seq(11)), + ('Type 2 + non-empty: last row seq -> p1 last_seq', 2, 10, _seq(11), _seq(12)), + ('Type 2 + non-empty: last_seq key -> p1 last_seq', 2, 10, _seq(12), _seq(12)), + ('Type 3 + non-empty: blocked by p2 ROW', 3, 10, _seq(11), _seq(11)), + ('Type 4 + non-empty: last row seq -> p1 last_seq', 4, 10, _seq(11), _seq(12)), + ('Type 4 + non-empty: last_seq key -> p1 last_seq', 4, 10, _seq(12), _seq(12)), + ('Type 5 + non-empty: blocked by p2 ROW', 5, 10, _seq(11), _seq(11)), + ('Type 6 + non-empty: blocked by p2 ROW', 6, 10, _seq(12), _seq(12)), + ('Type 7 + non-empty: blocked by p2 ROW', 7, 10, _seq(11), _seq(11)), + ('Type 8 + non-empty: blocked by p2 ROW', 8, 10, _seq(12), _seq(12)), + ('Type 9 + non-empty: blocked by p2 ROW', 9, 10, _seq(10), _seq(10)), + ] + for name, page_type, base, query_seq, expected in cases: + with self.subTest(name): + result = _last_seq_since([_page_type(page_type, base), _page_type(1, 20)], query_seq) + self.assertEqual(result, expected) + + # ----------------------------------------------------------------------- + # Per-page-type: followed by an empty page (type 9 at base 20) + # Page 2 inserts only LAST('20-aa') — no ROW to block, advances to '20-aa'. + # ----------------------------------------------------------------------- + + def test_last_seq_since_followed_by_empty(self): + cases = [ + ('Type 1 + empty: advances to p2 last_seq', 1, 10, _seq(11), _seq(20)), + ('Type 2 + empty: last row seq advances to p2', 2, 10, _seq(11), _seq(20)), + ('Type 2 + empty: last_seq key advances to p2', 2, 10, _seq(12), _seq(20)), + ('Type 3 + empty: advances to p2 last_seq', 3, 10, _seq(11), _seq(20)), + ('Type 4 + empty: last row seq advances to p2', 4, 10, _seq(11), _seq(20)), + ('Type 4 + empty: last_seq key advances to p2', 4, 10, _seq(12), _seq(20)), + ('Type 5 + empty: last_seq advances to p2', 5, 10, _seq(11), _seq(20)), + ('Type 6 + empty: last_seq advances to p2', 6, 10, _seq(12), _seq(20)), + ('Type 7 + empty: last_seq advances to p2', 7, 10, _seq(11), _seq(20)), + ('Type 8 + empty: last_seq advances to p2', 8, 10, _seq(12), _seq(20)), + ('Type 9 + empty: advances to p2 last_seq', 9, 10, _seq(10), _seq(20)), + ] + for name, page_type, base, query_seq, expected in cases: + with self.subTest(name): + result = _last_seq_since([_page_type(page_type, base), _page_type(9, 20)], query_seq) + self.assertEqual(result, expected) + + # ----------------------------------------------------------------------- + # 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 LAST; N adds ROW+LAST. + # ----------------------------------------------------------------------- + + def test_last_seq_since_3_page_sequence(self): + cases = [ + ('NNN: blocked by p2 ROW -> p1 last_seq', + [1, 1, 1], [10, 20, 30], _seq(11), _seq(11)), + ('NNE: blocked by p2 ROW -> p1 last_seq', + [1, 1, 9], [10, 20, 30], _seq(11), _seq(11)), + ('NEE: advances through both empty pages', + [1, 9, 9], [10, 20, 30], _seq(11), _seq(30)), + ('NEN: advances through p2 empty, stops at p3 ROW', + [1, 9, 1], [10, 20, 30], _seq(11), _seq(20)), + ('ENN: blocked by p2 ROW -> p1 last_seq', + [9, 1, 1], [10, 20, 30], _seq(10), _seq(10)), + ('ENE: blocked by p2 ROW -> p1 last_seq', + [9, 1, 9], [10, 20, 30], _seq(10), _seq(10)), + ('EEN: advances through p2, stops at p3 ROW', + [9, 9, 1], [10, 20, 30], _seq(10), _seq(20)), + ('EEE: advances through all three empty pages', + [9, 9, 9], [10, 20, 30], _seq(10), _seq(30)), + ] + for name, types, bases, query_seq, expected in cases: + with self.subTest(name): + pages = [_page_type(t, b) for t, b in zip(types, bases)] + result = _last_seq_since(pages, query_seq) + self.assertEqual(result, expected) + + # ----------------------------------------------------------------------- + # Eviction + # 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): + 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), 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=(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 + # ----------------------------------------------------------------------- + + def test_last_seq_since_none_row_seq_does_not_raise(self): + """Types 5/6/7/8 store ROW(None). Querying a seq not in the markers must return input unchanged without raising.""" + pages = [_page_type(5, 10)] + try: + result = _last_seq_since(pages, _seq(10)) + self.assertEqual(result, _seq(10)) + except Exception as e: + self.fail(f"last_seq_since raised {type(e).__name__} unexpectedly!") + + # ----------------------------------------------------------------------- + # None last_seq page (type 10) — page entry skipped, no error + # ----------------------------------------------------------------------- + + def test_last_seq_since_none_last_seq_page_does_not_raise(self): + """Querying a page with a None last_seq does not raise.""" + pages = [_page_type(10, 10)] + try: + result = _last_seq_since(pages, _seq(11)) + self.assertEqual(result, _seq(11)) + except Exception as e: + self.fail(f"last_seq_since raised {type(e).__name__} unexpectedly!") + + def test_last_seq_since_none_last_seq_middle_page(self): + """Querying with a None last_seq page in the middle of pages.""" + pages = [_page_type(1, 10), _page_type(10, 20), _page_type(9, 30)] + result = _last_seq_since(pages, _seq(11)) + self.assertEqual(result, _seq(11)) + + def test_last_seq_since_none_last_seq_last_page(self): + """Querying with a None last_seq page at the end of pages.""" + pages = [_page_type(9, 10), _page_type(10, 20)] + result = _last_seq_since(pages, _seq(10)) + self.assertEqual(result, _seq(10)) + + def test_last_seq_since_consecutive_none_last_seq_pages(self): + """Querying with consecutive None last_seq pages.""" + pages = [ + _page_type(9, 10), + _page_type(10, 20), + _page_type(10, 30), + _page_type(9, 40), + ] + result = _last_seq_since(pages, _seq(10)) + self.assertEqual(result, _seq(10)) + + def test_last_seq_since_none_last_seq_advances_beyond(self): + """ + 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)] + result = _last_seq_since(pages, _seq(10)) + self.assertEqual(result, _seq(30)) + + +# --------------------------------------------------------------------------- +# TestLatestSequenceFrom — unit tests for ChangesFollower.latest_sequence_from +# --------------------------------------------------------------------------- + +class TestLatestSequenceFrom(ChangesFollowerBaseCase): + + def test_latest_sequence_from_with_none(self): + """Raises ValueError when passed None.""" + follower = ChangesFollower(self.client, db='db') + with self.assertRaisesRegex(ValueError, 'Provided sequence ID must be a non-empty string'): + follower.latest_sequence_from(None) + + def test_latest_sequence_from_with_empty_string(self): + """Raises ValueError when passed an empty string.""" + follower = ChangesFollower(self.client, db='db') + with self.assertRaisesRegex(ValueError, 'Provided sequence ID must be a non-empty string'): + follower.latest_sequence_from('') + + def test_latest_sequence_from_before_feed_starts(self): + """Returns the input seq unchanged when the feed has not started yet.""" + follower = ChangesFollower(self.client, db='db') + self.assertEqual(follower.latest_sequence_from('seq-a'), 'seq-a') + + @responses.activate + def test_latest_sequence_from_unknown_seq(self): + """Returns the input seq unchanged when it was never seen by this follower.""" + self.prepare_mock_changes(batches=1) + follower = ChangesFollower(self.client, db='db') + changes = follower.start_one_off() + for _ in changes: + pass + self.assertEqual(follower.latest_sequence_from('seq-unknown'), 'seq-unknown') + + @responses.activate + def test_latest_sequence_from_middle_of_batch(self): + """ + Returns the input seq unchanged when querying with a seq from the + middle of a batch — only the last item's seq is stored in seq_markers. + """ + self.prepare_mock_changes(batches=1) + follower = ChangesFollower(self.client, db='db') + changes = follower.start_one_off() + items = list(changes) + # seq-a and seq-b are middle items — not stored in seq_markers + seq_a = items[0].seq + seq_b = items[1].seq + self.assertEqual(follower.latest_sequence_from(seq_a), seq_a) + self.assertEqual(follower.latest_sequence_from(seq_b), seq_b) + + @responses.activate + def test_latest_sequence_from_end_to_end(self): + """Returns the correct last_seq through a completed stream.""" + self.prepare_mock_changes(batches=1) + follower = ChangesFollower(self.client, db='db') + changes = follower.start_one_off() + items = list(changes) + # Last item's seq should map to the page's last_seq + last_item_seq = items[-1].seq + result = follower.latest_sequence_from(last_item_seq) + # The last item seq IS the last_seq for a normal page (type 1 equivalent) + self.assertEqual(result, last_item_seq)