2323from datetime import datetime , timezone , timedelta
2424import functools
2525from queue import Queue
26- from threading import Thread , Event
26+ from threading import Thread , Event , Lock
2727
2828from enum import Enum , auto
2929from typing import Dict , Iterator
@@ -118,6 +118,7 @@ def __init__(
118118 self ._stop = Event ()
119119 self .logger = logging .getLogger (__name__ )
120120 self ._seq_markers : list = []
121+ self ._seq_markers_lock = Lock ()
121122
122123 @property
123124 def limit (self ) -> int :
@@ -187,9 +188,11 @@ def last_seq_since(self, last_persisted_seq: str) -> str:
187188
188189 Returns last_persisted_seq unchanged if not found in the markers.
189190 """
191+ with self ._seq_markers_lock :
192+ markers = list (self ._seq_markers )
190193 found = False
191194 result = None
192- for entry in self . _seq_markers :
195+ for entry in markers :
193196 if found :
194197 if entry ['type' ] == _SeqEntryType .ROW :
195198 break
@@ -207,11 +210,12 @@ def _update_seq_markers(self, results: list, last_seq: str) -> None:
207210 a ROW entry for the last change item (if any) and a PAGE entry for
208211 the page's last_seq.
209212 """
210- if len (self ._seq_markers ) >= _SEQ_MARKERS_CAPACITY :
211- del self ._seq_markers [:_SEQ_MARKERS_EVICTION_COUNT ]
212- if len (results ) > 0 :
213- self ._seq_markers .append ({'type' : _SeqEntryType .ROW , 'seq' : results [- 1 ].get ('seq' )})
214- self ._seq_markers .append ({'type' : _SeqEntryType .PAGE , 'seq' : last_seq })
213+ with self ._seq_markers_lock :
214+ if len (self ._seq_markers ) >= _SEQ_MARKERS_CAPACITY :
215+ del self ._seq_markers [:_SEQ_MARKERS_EVICTION_COUNT ]
216+ if len (results ) > 0 :
217+ self ._seq_markers .append ({'type' : _SeqEntryType .ROW , 'seq' : results [- 1 ].get ('seq' )})
218+ self ._seq_markers .append ({'type' : _SeqEntryType .PAGE , 'seq' : last_seq })
215219
216220 def _request_callback (self ):
217221 while True :
0 commit comments