feat(file-based): throttle legacy per-slice state emission by default

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
 

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

airbyte_cdk/sources/streams/checkpoint/__init__.py

@@ -2,18 +2,22 @@
 
 
 from .checkpoint_reader import (
+    DEFAULT_STATE_EMISSION_THROTTLE_SECONDS,
     CheckpointMode,
     CheckpointReader,
     CursorBasedCheckpointReader,
     FullRefreshCheckpointReader,
     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",
 ]

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

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",

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)