diff --git a/ibmcloudant/features/changes_follower.py b/ibmcloudant/features/changes_follower.py index 004afce2..6ba7bbea 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_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.') diff --git a/test/unit/features/test_changes_follower.py b/test/unit/features/test_changes_follower.py index e31a6d17..f5f30ac5 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_MARKERS_CAPACITY, + _SEQ_MARKERS_EVICTION_COUNT, ChangesFollower, _Mode, ) @@ -650,3 +653,286 @@ 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 9 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) + """ + 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} + 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 PAGE('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 PAGE; N adds ROW+PAGE. + # ----------------------------------------------------------------------- + + 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 + PAGE). + # With CAPACITY=200 and EVICTION_COUNT=20, adding 101 pages triggers one + # eviction of the oldest 20 entries (first 10 pages). + # 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)] + iterator = _populate_iterator(pages) + + # Page 0 (base=0): row=_seq(1), page=_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)) + + +# --------------------------------------------------------------------------- +# TestGetLastSeqNewerThan — unit tests for ChangesFollower.get_last_seq_newer_than +# --------------------------------------------------------------------------- + +class TestGetLastSeqNewerThan(ChangesFollowerBaseCase): + + def test_get_last_seq_newer_than_with_none(self): + """Raises ValueError when passed None.""" + follower = ChangesFollower(self.client, db='db') + with self.assertRaisesRegex(ValueError, 'The provided sequence ID cannot be null or empty'): + follower.get_last_seq_newer_than(None) + + def test_get_last_seq_newer_than_with_empty_string(self): + """Raises ValueError when passed an empty string.""" + follower = ChangesFollower(self.client, db='db') + with self.assertRaisesRegex(ValueError, 'The provided sequence ID cannot be null or empty'): + follower.get_last_seq_newer_than('') + + def test_get_last_seq_newer_than_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.get_last_seq_newer_than('seq-a'), 'seq-a') + + @responses.activate + def test_get_last_seq_newer_than_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.get_last_seq_newer_than('seq-unknown'), 'seq-unknown') + + @responses.activate + def test_get_last_seq_newer_than_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.get_last_seq_newer_than(seq_a), seq_a) + self.assertEqual(follower.get_last_seq_newer_than(seq_b), seq_b) + + @responses.activate + def test_get_last_seq_newer_than_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.get_last_seq_newer_than(last_item_seq) + # The last item seq IS the last_seq for a normal page (type 1 equivalent) + self.assertEqual(result, last_item_seq)