diff --git a/airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py b/airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py index d1f2ca41e3..d64316a57b 100644 --- a/airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py +++ b/airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py @@ -22,6 +22,10 @@ from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager from airbyte_cdk.sources.declarative.partition_routers.partition_router import PartitionRouter from airbyte_cdk.sources.message import MessageRepository +from airbyte_cdk.sources.streams.checkpoint.checkpoint_reader import ( + DEFAULT_STATE_EMISSION_THROTTLE_SECONDS, + state_emission_is_due, +) from airbyte_cdk.sources.streams.checkpoint.per_partition_key_serializer import ( PerPartitionKeySerializer, ) @@ -285,10 +289,19 @@ def ensure_at_least_one_state_emitted(self) -> None: def _throttle_state_message(self) -> Optional[float]: """ - Throttles the state message emission to once every 600 seconds. + Throttles the state message emission to once every + DEFAULT_STATE_EMISSION_THROTTLE_SECONDS seconds. + + Shares `state_emission_is_due` with the legacy per-slice throttle in + `ThrottledCheckpointReader` so the two paths agree, including at the + boundary. This previously used `<=`, which suppressed an emission landing + exactly on the window; the shared predicate treats it as due, matching + "once every N seconds". """ current_time = time.time() - if current_time - self._last_emission_time <= 600: + if not state_emission_is_due( + self._last_emission_time, current_time, DEFAULT_STATE_EMISSION_THROTTLE_SECONDS + ): return None return current_time diff --git a/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py b/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py index 588c4a18e4..b16d5a04ff 100644 --- a/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py +++ b/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py @@ -48,6 +48,11 @@ from airbyte_cdk.sources.file_based.stream.cursor import AbstractFileBasedCursor from airbyte_cdk.sources.file_based.types import StreamSlice from airbyte_cdk.sources.streams import IncrementalMixin +from airbyte_cdk.sources.streams.checkpoint import ( + DEFAULT_STATE_EMISSION_THROTTLE_SECONDS, + CheckpointReader, + ThrottledCheckpointReader, +) from airbyte_cdk.sources.streams.core import JsonSchema from airbyte_cdk.sources.utils.record_helper import stream_data_to_airbyte_message from airbyte_cdk.utils.traced_exception import AirbyteTracedException @@ -71,6 +76,10 @@ class DefaultFileBasedStream(AbstractFileBasedStream, IncrementalMixin): preserve_directory_structure = True _file_transfer = FileTransfer() + # How often this stream may emit a per-slice state message. Overridable, but + # connectors have no principled reason to diverge from the shared default. + state_emission_throttle_seconds: Optional[float] = DEFAULT_STATE_EMISSION_THROTTLE_SECONDS + def __init__(self, **kwargs: Any): if self.FILE_TRANSFER_KW in kwargs: self.use_file_transfer = kwargs.pop(self.FILE_TRANSFER_KW, False) @@ -89,6 +98,26 @@ def state(self, value: MutableMapping[str, Any]) -> None: """State setter, accept state serialized by state getter.""" self._cursor.set_initial_state(value) + def _get_checkpoint_reader(self, *args: Any, **kwargs: Any) -> CheckpointReader: + """Wrap the reader chosen by the base class in a state-emission throttle. + + On this stream a slice is a single file and the state payload carries the + whole file-history dict, so the default emit-per-slice cadence scales + state volume with the number of files synced. Throttling here rather than + in `Stream.read()` keeps the change off the generic base class: nothing + outside file-based streams is affected. + + Wrapping the result of `super()` rather than constructing a reader + directly means we keep whatever reader the base class picks, including if + that choice changes later. + """ + reader = super()._get_checkpoint_reader(*args, **kwargs) + if self.state_emission_throttle_seconds is None: + return reader + return ThrottledCheckpointReader( + reader, throttle_seconds=self.state_emission_throttle_seconds + ) + @property # type: ignore # mypy complains wrong type, but AbstractFileBasedCursor is parent of file-based cursors def cursor(self) -> Optional[AbstractFileBasedCursor]: return self._cursor diff --git a/airbyte_cdk/sources/streams/checkpoint/__init__.py b/airbyte_cdk/sources/streams/checkpoint/__init__.py index ae4e0e46f7..d6d2cdf6c5 100644 --- a/airbyte_cdk/sources/streams/checkpoint/__init__.py +++ b/airbyte_cdk/sources/streams/checkpoint/__init__.py @@ -2,6 +2,7 @@ from .checkpoint_reader import ( + DEFAULT_STATE_EMISSION_THROTTLE_SECONDS, CheckpointMode, CheckpointReader, CursorBasedCheckpointReader, @@ -9,11 +10,14 @@ IncrementalCheckpointReader, LegacyCursorBasedCheckpointReader, ResumableFullRefreshCheckpointReader, + ThrottledCheckpointReader, + state_emission_is_due, ) from .cursor import Cursor from .resumable_full_refresh_cursor import ResumableFullRefreshCursor __all__ = [ + "DEFAULT_STATE_EMISSION_THROTTLE_SECONDS", "CheckpointMode", "CheckpointReader", "Cursor", @@ -22,5 +26,7 @@ "IncrementalCheckpointReader", "LegacyCursorBasedCheckpointReader", "ResumableFullRefreshCheckpointReader", + "ThrottledCheckpointReader", "ResumableFullRefreshCursor", + "state_emission_is_due", ] diff --git a/airbyte_cdk/sources/streams/checkpoint/checkpoint_reader.py b/airbyte_cdk/sources/streams/checkpoint/checkpoint_reader.py index 6e4ef98d77..c0f8a2128a 100644 --- a/airbyte_cdk/sources/streams/checkpoint/checkpoint_reader.py +++ b/airbyte_cdk/sources/streams/checkpoint/checkpoint_reader.py @@ -1,8 +1,9 @@ # Copyright (c) 2024 Airbyte, Inc., all rights reserved. +import time from abc import ABC, abstractmethod from enum import Enum -from typing import Any, Iterable, Mapping, Optional +from typing import Any, Callable, Iterable, Mapping, Optional from airbyte_cdk.sources.types import StreamSlice @@ -17,6 +18,44 @@ class CheckpointMode(Enum): FULL_REFRESH_COMPLETE_STATE: Mapping[str, Any] = {"__ab_full_refresh_sync_complete": True} +# How often a throttled reader may surface a checkpoint, in seconds. +# +# Not derived from any platform limit: the source heartbeat the platform enforces +# is `heartbeat-max-seconds-between-messages` (10800s / 3h), and it is reset by +# RECORD messages as well as STATE, so throttling state cannot stall a sync that +# is still producing records. 600s is the cadence already used by +# `ConcurrentPerPartitionCursor`, reused here so both paths behave the same. +DEFAULT_STATE_EMISSION_THROTTLE_SECONDS = 600.0 + + +def state_emission_is_due( + last_emitted_at: Optional[float], + now: float, + throttle_seconds: float = DEFAULT_STATE_EMISSION_THROTTLE_SECONDS, +) -> bool: + """Whether a throttled state message may be emitted at `now`. + + Shared by `ThrottledCheckpointReader` (legacy per-slice emission) and + `ConcurrentPerPartitionCursor` so the two cannot drift apart. They used to + share only the constant, which still left them disagreeing at the boundary. + + "At most once every N seconds" makes the boundary inclusive: an emission + exactly N seconds after the previous one is due. + + `last_emitted_at is None` means nothing has been emitted yet, so the first + emission is always due regardless of the clock's absolute value. + + A `throttle_seconds <= 0` never suppresses, degrading to the historical + unthrottled behaviour rather than erroring. That is checked before any + arithmetic: `ConcurrentPerPartitionCursor` measures with wall-clock + `time.time()`, which can step backwards, and a negative elapsed time would + otherwise fail the comparison and suppress an emission the caller asked to + never throttle. + """ + if last_emitted_at is None or throttle_seconds <= 0: + return True + return now - last_emitted_at >= throttle_seconds + class CheckpointReader(ABC): """ @@ -333,3 +372,75 @@ def get_checkpoint(self) -> Optional[Mapping[str, Any]]: if self._final_checkpoint: return {"__ab_no_cursor_state_message": True} return None + + +class ThrottledCheckpointReader(CheckpointReader): + """Rate-limits how often a wrapped reader surfaces a checkpoint. + + Wraps any other reader and delegates iteration to it, but suppresses + per-slice checkpoints that land inside `throttle_seconds` of the last one + surfaced. Suppression uses the documented `get_checkpoint()` contract: a + `None` return means the caller emits no state message. + + The first checkpoint of the sync always surfaces, and if the last per-slice + checkpoint was suppressed the held value is surfaced on the final + `get_checkpoint()` call after iteration ends. So the destination always sees + the latest cursor; only intermediate checkpoint granularity is reduced. + + Exists because legacy file-based streams carry a file-history dict in every + state message, so the payload grows with the sync and emitting once per file + puts GBs of un-ACKed state into the orchestrator buffer (oncall #12663, + #13210). + + Note the held checkpoint is a reference, not a snapshot: cursors that return + their live state dict (e.g. `DefaultFileBasedCursor`) keep mutating it, so + the final emit serializes state as of end-of-sync. That is what we want. + """ + + def __init__( + self, + inner: CheckpointReader, + throttle_seconds: float = DEFAULT_STATE_EMISSION_THROTTLE_SECONDS, + clock: Callable[[], float] = time.monotonic, + ): + self._inner = inner + self._throttle_seconds = throttle_seconds + # Monotonic by default: measures elapsed time, immune to wall-clock jumps. + self._clock = clock + # None means nothing has surfaced yet, so the first checkpoint always + # fires regardless of the clock's absolute value. + self._last_surfaced_at: Optional[float] = None + self._pending: Optional[Mapping[str, Any]] = None + self._finished = False + + def next(self) -> Optional[Mapping[str, Any]]: + next_slice = self._inner.next() + if next_slice is None: + self._finished = True + return next_slice + + def observe(self, new_state: Mapping[str, Any]) -> None: + self._inner.observe(new_state) + + def get_checkpoint(self) -> Optional[Mapping[str, Any]]: + checkpoint = self._inner.get_checkpoint() + + if self._finished: + # Final call, after iteration ended. Prefer whatever the wrapped + # reader wants to emit; otherwise pay out a suppressed checkpoint so + # the sync never ends on a state the destination has not seen. If + # nothing was suppressed, `_pending` is None and no duplicate final + # state is emitted. + return checkpoint if checkpoint is not None else self._pending + + if checkpoint is None: + return None + + now = self._clock() + if not state_emission_is_due(self._last_surfaced_at, now, self._throttle_seconds): + self._pending = checkpoint + return None + + self._last_surfaced_at = now + self._pending = None + return checkpoint diff --git a/unit_tests/sources/file_based/scenarios/incremental_scenarios.py b/unit_tests/sources/file_based/scenarios/incremental_scenarios.py index e34b7f4ded..e33dce2efb 100644 --- a/unit_tests/sources/file_based/scenarios/incremental_scenarios.py +++ b/unit_tests/sources/file_based/scenarios/incremental_scenarios.py @@ -1317,14 +1317,11 @@ }, "stream": "stream1", }, - { - "history": { - "old_file_same_timestamp_as_a.csv": "2023-06-06T03:54:07.000000Z", - "a.csv": "2023-06-06T03:54:07.000000Z", - "b.csv": "2023-06-07T03:54:07.000000Z", - }, - "_ab_source_file_last_modified": "2023-06-07T03:54:07.000000Z_b.csv", - }, + # No state after b.csv: file-based streams throttle per-slice state + # emission (DEFAULT_STATE_EMISSION_THROTTLE_SECONDS), so only the + # first slice and the forced final emit within the window. The + # cursor still advances through b.csv — it is carried in the final + # state below. { "data": { "col1": "val11c", diff --git a/unit_tests/sources/file_based/stream/test_state_emission_throttle.py b/unit_tests/sources/file_based/stream/test_state_emission_throttle.py new file mode 100644 index 0000000000..43376e693f --- /dev/null +++ b/unit_tests/sources/file_based/stream/test_state_emission_throttle.py @@ -0,0 +1,106 @@ +# +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +# +"""File-based streams throttle legacy per-slice state emission. + +The wiring lives in `DefaultFileBasedStream._get_checkpoint_reader()` rather than +in `Stream.read()`, so the generic base class is untouched and only file-based +streams change behaviour. Because it is a method on the stream class, connectors +that build their own `DefaultFileBasedStream` subclass from an overridden +`_make_default_stream()` without calling `super()` — source-s3 and source-gcs +both do — inherit it and cannot bypass it. +""" + +import logging +from typing import Any, Mapping, Optional +from unittest.mock import MagicMock + +from airbyte_cdk.models import SyncMode +from airbyte_cdk.sources.file_based.config.csv_format import CsvFormat +from airbyte_cdk.sources.file_based.config.file_based_stream_config import FileBasedStreamConfig +from airbyte_cdk.sources.file_based.stream import ( + DefaultFileBasedStream, + PermissionsFileBasedStream, +) +from airbyte_cdk.sources.file_based.stream.cursor import DefaultFileBasedCursor +from airbyte_cdk.sources.streams.checkpoint import ( + DEFAULT_STATE_EMISSION_THROTTLE_SECONDS, + ThrottledCheckpointReader, +) + + +def _config() -> FileBasedStreamConfig: + return FileBasedStreamConfig(name="stream1", format=CsvFormat(), globs=["*.csv"]) + + +def _stream(cls: type = DefaultFileBasedStream, **overrides: Any) -> DefaultFileBasedStream: + config = _config() + # No files: `_get_checkpoint_reader` walks `stream_slices()`, and an empty + # listing is enough to reach the reader construction we care about. + stream_reader = MagicMock() + stream_reader.get_matching_files.return_value = [] + kwargs: Mapping[str, Any] = { + "config": config, + "catalog_schema": None, + "stream_reader": stream_reader, + "availability_strategy": None, + "discovery_policy": None, + "parsers": None, + "validation_policy": None, + "errors_collector": None, + "cursor": DefaultFileBasedCursor(config), + "use_file_transfer": False, + "preserve_directory_structure": True, + **overrides, + } + return cls(**kwargs) # type: ignore[arg-type] + + +def _reader(stream: DefaultFileBasedStream) -> Any: + return stream._get_checkpoint_reader( + logger=logging.getLogger("test"), + cursor_field=None, + sync_mode=SyncMode.incremental, + stream_state={}, + ) + + +def test_checkpoint_reader_is_throttled_at_the_shared_default() -> None: + reader = _reader(_stream()) + assert isinstance(reader, ThrottledCheckpointReader) + assert reader._throttle_seconds == DEFAULT_STATE_EMISSION_THROTTLE_SECONDS + + +def test_connector_subclass_cannot_bypass_the_throttle() -> None: + """The regression this file exists for: source-s3 and source-gcs subclass this + stream and construct it themselves, so the throttle must ride on the class.""" + + class _ConnectorStream(DefaultFileBasedStream): + """Mirrors source_gcs.stream.GCSStream / source_s3 ThrottledFileBasedStream.""" + + assert isinstance(_reader(_stream(_ConnectorStream)), ThrottledCheckpointReader) + + +def test_permissions_stream_is_throttled_too() -> None: + """The permissions transfer path is on the same legacy emission path. + + Asserts the constructed reader, not just the class attribute: an override of + `_get_checkpoint_reader` on this subclass would leave the attribute intact + while removing the throttle entirely. + """ + stream = _stream(PermissionsFileBasedStream, stream_permissions_reader=MagicMock()) + assert isinstance(_reader(stream), ThrottledCheckpointReader) + assert ( + PermissionsFileBasedStream.state_emission_throttle_seconds + == DEFAULT_STATE_EMISSION_THROTTLE_SECONDS + ) + + +def test_setting_the_throttle_to_none_restores_the_unwrapped_reader() -> None: + """An escape hatch that returns the base reader untouched, so the throttle can + be turned off without a different code path.""" + + class _UnthrottledStream(DefaultFileBasedStream): + state_emission_throttle_seconds: Optional[float] = None + + assert not isinstance(_reader(_stream(_UnthrottledStream)), ThrottledCheckpointReader) diff --git a/unit_tests/sources/streams/checkpoint/test_throttled_checkpoint_reader.py b/unit_tests/sources/streams/checkpoint/test_throttled_checkpoint_reader.py new file mode 100644 index 0000000000..833a363d26 --- /dev/null +++ b/unit_tests/sources/streams/checkpoint/test_throttled_checkpoint_reader.py @@ -0,0 +1,259 @@ +# +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +# +"""`ThrottledCheckpointReader` rate-limits how often a checkpoint surfaces. + +Suppression rides on the documented `get_checkpoint()` contract — returning +`None` means the caller emits no state message — so nothing in `Stream.read()` +needs to know throttling exists. + +The reader is driven directly here rather than through a stream read, which +keeps the clock injectable and the call sequence explicit. +""" + +import itertools +from typing import Any, Iterable, List, Mapping, Optional + +import pytest + +from airbyte_cdk.sources.streams.checkpoint import ( + DEFAULT_STATE_EMISSION_THROTTLE_SECONDS, + CheckpointReader, + IncrementalCheckpointReader, + ThrottledCheckpointReader, + state_emission_is_due, +) + + +class _RecordingReader(CheckpointReader): + """Minimal inner reader: yields the given slices, echoes observed state.""" + + def __init__(self, slices: Iterable[Optional[Mapping[str, Any]]]): + self._slices = iter(slices) + self._state: Optional[Mapping[str, Any]] = None + self._exhausted = False + + def next(self) -> Optional[Mapping[str, Any]]: + try: + return next(self._slices) + except StopIteration: + # Mirrors IncrementalCheckpointReader: drop state at the end so the + # caller does not emit a duplicate final checkpoint. + self._exhausted = True + self._state = None + return None + + def observe(self, new_state: Mapping[str, Any]) -> None: + self._state = new_state + + def get_checkpoint(self) -> Optional[Mapping[str, Any]]: + return self._state + + +def _drive(reader: CheckpointReader, cursors: List[str]) -> List[Mapping[str, Any]]: + """Run the call sequence `Stream.read()` uses, collecting emitted checkpoints. + + Per slice: observe() then get_checkpoint(); then next(). After the loop, one + final get_checkpoint(). + """ + emitted = [] + next_slice = reader.next() + i = 0 + while next_slice is not None: + reader.observe({"cursor": cursors[i]}) + checkpoint = reader.get_checkpoint() + if checkpoint is not None: + emitted.append(checkpoint) + i += 1 + next_slice = reader.next() + + final = reader.get_checkpoint() + if final is not None: + emitted.append(final) + return emitted + + +CURSORS = ["a", "b", "c", "d"] +SLICES: List[Optional[Mapping[str, Any]]] = [{"s": 1}, {"s": 2}, {"s": 3}, {"s": 4}] + + +@pytest.mark.parametrize( + "throttle_seconds, tick, expected_cursors, reason", + [ + pytest.param( + 600.0, + 10, + ["a", "d"], + "cold start emits; slices 2-4 sit inside the window; final is forced", + id="suppresses_and_forces_final", + ), + pytest.param( + 15.0, + 10, + ["a", "c", "d"], + "slice 3 at t=20 is >=15s after t=0, so it emits and restarts the window", + id="re_emits_once_window_elapses", + ), + pytest.param( + 10.0, + 10, + ["a", "b", "c", "d"], + "every slice lands exactly on the boundary (delta == throttle, not <)", + id="boundary_emits_without_duplicate_final", + ), + pytest.param( + 0.0, + 10, + ["a", "b", "c", "d"], + "a non-positive window never suppresses, so behaviour matches unthrottled", + id="zero_fails_open", + ), + ], +) +def test_throttle_emission_sequence(throttle_seconds, tick, expected_cursors, reason) -> None: + clock = itertools.count(0, tick) + reader = ThrottledCheckpointReader( + _RecordingReader(SLICES), + throttle_seconds=throttle_seconds, + clock=lambda: next(clock), + ) + + emitted = _drive(reader, CURSORS) + + # Assert the whole sequence, not just the count: a count-only check also + # passes for an implementation that inverted its comparison. + assert [c["cursor"] for c in emitted] == expected_cursors, reason + + +def test_final_checkpoint_is_not_duplicated_when_nothing_was_suppressed() -> None: + """The last slice emitted, so `_pending` is empty and the final call must not + manufacture an extra state message.""" + clock = itertools.count(0, 10) + reader = ThrottledCheckpointReader( + _RecordingReader(SLICES), throttle_seconds=10.0, clock=lambda: next(clock) + ) + _drive(reader, CURSORS) + assert reader.get_checkpoint() is None + + +def test_inner_final_checkpoint_takes_precedence() -> None: + """If the wrapped reader wants to emit its own final checkpoint, that wins + over any held value — the throttle must not mask it.""" + + class _FinalEmittingReader(_RecordingReader): + def get_checkpoint(self) -> Optional[Mapping[str, Any]]: + if self._exhausted: + return {"cursor": "inner-final"} + return self._state + + clock = itertools.count(0, 10) + reader = ThrottledCheckpointReader( + _FinalEmittingReader(SLICES), throttle_seconds=600.0, clock=lambda: next(clock) + ) + emitted = _drive(reader, CURSORS) + assert [c["cursor"] for c in emitted] == ["a", "inner-final"] + + +def test_delegates_iteration_and_observation_to_the_inner_reader() -> None: + inner = IncrementalCheckpointReader(stream_state={}, stream_slices=SLICES) + reader = ThrottledCheckpointReader(inner, throttle_seconds=600.0) + + assert reader.next() == {"s": 1} + reader.observe({"cursor": "a"}) + assert inner.get_checkpoint() == {"cursor": "a"} + + +def test_default_throttle_is_the_shared_constant() -> None: + reader = ThrottledCheckpointReader(_RecordingReader(SLICES)) + assert reader._throttle_seconds == DEFAULT_STATE_EMISSION_THROTTLE_SECONDS + + +@pytest.mark.parametrize( + "last_emitted_at, now, throttle, expected, reason", + [ + pytest.param(None, 0.0, 600.0, True, "nothing emitted yet", id="cold_start"), + pytest.param(1000.0, 1599.9, 600.0, False, "inside the window", id="inside"), + pytest.param( + 1000.0, + 1600.0, + 600.0, + True, + "'once every N seconds' makes the boundary inclusive", + id="exactly_on_boundary", + ), + pytest.param(1000.0, 1600.1, 600.0, True, "past the window", id="past"), + pytest.param(1000.0, 1000.0, 0.0, True, "zero never suppresses", id="zero_fails_open"), + pytest.param(1000.0, 1000.0, -5.0, True, "negative never suppresses", id="negative"), + # Wall-clock rollback (NTP correction) makes elapsed time negative. + # `ConcurrentPerPartitionCursor` measures with `time.time()`, so this is + # reachable. A non-positive window must still never suppress, which means + # the check has to come before the arithmetic. + pytest.param( + 1000.0, 990.0, 0.0, True, "zero, clock stepped backwards", id="zero_clock_rollback" + ), + pytest.param( + 1000.0, + 990.0, + -5.0, + True, + "negative, clock stepped backwards", + id="negative_clock_rollback", + ), + pytest.param( + 1000.0, + 990.0, + 600.0, + False, + "a positive window still suppresses under rollback — the window has " + "genuinely not elapsed", + id="positive_clock_rollback", + ), + # A cold start expressed as 0.0 rather than None: ConcurrentPerPartitionCursor + # initialises `_last_emission_time = 0.0` and compares against wall-clock + # `time.time()`, so the first emission must still be due. + pytest.param(0.0, 1.7e9, 600.0, True, "epoch clock vs 0.0 sentinel", id="epoch_cold_start"), + ], +) +def test_state_emission_is_due(last_emitted_at, now, throttle, expected, reason) -> None: + assert state_emission_is_due(last_emitted_at, now, throttle) is expected, reason + + +def test_both_throttle_paths_agree_on_the_boundary(mocker) -> None: + """Regression guard for the two paths drifting apart. + + They used to share only the constant: `ConcurrentPerPartitionCursor` suppressed + at `elapsed == throttle` (`<=`) while the reader emitted. Both now route through + `state_emission_is_due`, so this asserts they agree at the exact boundary. + """ + from types import SimpleNamespace + + from airbyte_cdk.sources.declarative.incremental import concurrent_partition_cursor + from airbyte_cdk.sources.declarative.incremental.concurrent_partition_cursor import ( + ConcurrentPerPartitionCursor, + ) + + boundary = 1000.0 + DEFAULT_STATE_EMISSION_THROTTLE_SECONDS + mocker.patch.object(concurrent_partition_cursor.time, "time", return_value=boundary) + + # Concurrent path: only touches `_last_emission_time`, so a stub suffices. + concurrent_due = ( + ConcurrentPerPartitionCursor._throttle_state_message( + SimpleNamespace(_last_emission_time=1000.0) # type: ignore[arg-type] + ) + is not None + ) + + # Legacy per-slice path, driven at the same elapsed time. + reader = ThrottledCheckpointReader( + _RecordingReader(SLICES), + throttle_seconds=DEFAULT_STATE_EMISSION_THROTTLE_SECONDS, + clock=iter([1000.0, boundary]).__next__, + ) + reader.next() + reader.observe({"cursor": "a"}) + reader.get_checkpoint() # cold start, surfaces at t=1000 + reader.next() + reader.observe({"cursor": "b"}) + legacy_due = reader.get_checkpoint() is not None # at t=boundary + + assert concurrent_due is legacy_due is True