feat(cdk): Add cursor age validation to StateDelegatingStream

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3752,6 +3752,22 @@ definitions:
         title: Incremental Stream
         description: Component used to coordinate how records are extracted across stream slices and request pages when the state provided.
         "$ref": "#/definitions/DeclarativeStream"
+      api_retention_period:
+        title: API Retention Period
+        description: |
+          The data retention period of the incremental API (ISO8601 duration). If the cursor value is older than this retention period, the connector will automatically fall back to a full refresh to avoid data loss.
+          This is useful for APIs like Stripe Events API which only retain data for 30 days.
+            * **PT1H**: 1 hour
+            * **P1D**: 1 day
+            * **P1W**: 1 week
+            * **P1M**: 1 month
+            * **P1Y**: 1 year
+            * **P30D**: 30 days
+        type: string
+        examples:
+          - "P30D"
+          - "P90D"
+          - "P1Y"
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1,5 +1,3 @@
-# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
-
 # generated by datamodel-codegen:
 #   filename:  declarative_component_schema.yaml
 
@@ -2885,6 +2883,12 @@ class StateDelegatingStream(BaseModel):
         description="Component used to coordinate how records are extracted across stream slices and request pages when the state provided.",
         title="Incremental Stream",
     )
+    api_retention_period: Optional[str] = Field(
+        None,
+        description="The data retention period of the incremental API (ISO8601 duration). If the cursor value is older than this retention period, the connector will automatically fall back to a full refresh to avoid data loss.\nThis is useful for APIs like Stripe Events API which only retain data for 30 days.\n  * **PT1H**: 1 hour\n  * **P1D**: 1 day\n  * **P1W**: 1 week\n  * **P1M**: 1 month\n  * **P1Y**: 1 year\n  * **P30D**: 30 days\n",
+        examples=["P30D", "P90D", "P1Y"],
+        title="API Retention Period",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -78,6 +78,7 @@
     DynamicStreamCheckConfig,
 )
 from airbyte_cdk.sources.declarative.concurrency_level import ConcurrencyLevel
+from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser
 from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime
 from airbyte_cdk.sources.declarative.decoders import (
     Decoder,
@@ -3568,11 +3569,102 @@ def create_state_delegating_stream(
     def _get_state_delegating_stream_model(
         self, has_parent_state: bool, model: StateDelegatingStreamModel
     ) -> DeclarativeStreamModel:
-        return (
-            model.incremental_stream
-            if self._connector_state_manager.get_stream_state(model.name, None) or has_parent_state
-            else model.full_refresh_stream
+        stream_state = self._connector_state_manager.get_stream_state(model.name, None)
+
+        if not stream_state and not has_parent_state:
+            return model.full_refresh_stream
+
+        if model.api_retention_period and stream_state:
+            if self._is_cursor_older_than_retention_period(
+                stream_state, model.incremental_stream, model.api_retention_period, model.name
+            ):
+                return model.full_refresh_stream
+
+        return model.incremental_stream
+
+    def _is_cursor_older_than_retention_period(
+        self,
+        stream_state: Mapping[str, Any],
+        incremental_stream: DeclarativeStreamModel,
+        api_retention_period: str,
+        stream_name: str,
+    ) -> bool:
+        """Check if the cursor value in the state is older than the API's retention period.
+
+        If the cursor is too old, the incremental API may not have data going back that far,
+        so we should fall back to a full refresh to avoid data loss.
+
+        Returns True if the cursor is older than the retention period (should use full refresh).
+        Returns False if the cursor is within the retention period (safe to use incremental).
+        """
+        incremental_sync = incremental_stream.incremental_sync
+        if not incremental_sync:
+            return False
+
+        cursor_field = getattr(incremental_sync, "cursor_field", None)
+        if not cursor_field:
+            return False
+
+        cursor_value = stream_state.get(cursor_field)
+        if not cursor_value:
+            return False
+
+        retention_duration = parse_duration(api_retention_period)
+        retention_cutoff = datetime.datetime.now(datetime.timezone.utc) - retention_duration
+
+        cursor_datetime = self._parse_cursor_datetime(cursor_value, incremental_sync, stream_name)
+        if cursor_datetime is None:
+            return False
+
+        if cursor_datetime < retention_cutoff:
+            self._emit_warning_for_stale_cursor(
+                stream_name, cursor_value, api_retention_period, retention_cutoff
+            )
+            return True
+
+        return False
+
+    def _parse_cursor_datetime(
+        self,
+        cursor_value: str,
+        incremental_sync: Any,
+        stream_name: str,
+    ) -> Optional[datetime.datetime]:
+        """Parse the cursor value into a datetime object using the cursor's datetime formats."""
+        parser = DatetimeParser()
+
+        datetime_format = getattr(incremental_sync, "datetime_format", None)
+        cursor_datetime_formats = getattr(incremental_sync, "cursor_datetime_formats", None) or []
+
+        formats_to_try = cursor_datetime_formats + ([datetime_format] if datetime_format else [])
+
+        for fmt in formats_to_try:
+            try:
+                return parser.parse(cursor_value, fmt)
+            except (ValueError, TypeError):
+                continue
+
+        logging.warning(
+            f"Could not parse cursor value '{cursor_value}' for stream '{stream_name}' "
+            f"using formats {formats_to_try}. Skipping cursor age validation."
         )
+        return None
+
+    def _emit_warning_for_stale_cursor(
+        self,
+        stream_name: str,
+        cursor_value: str,
+        api_retention_period: str,
+        retention_cutoff: datetime.datetime,
+    ) -> None:
+        """Emit a warning message when the cursor is older than the API's retention period."""
+        warning_message = (
+            f"Stream '{stream_name}' has a cursor value '{cursor_value}' that is older than "
+            f"the API's retention period of {api_retention_period} (cutoff: {retention_cutoff.isoformat()}). "
+            f"Falling back to full refresh to avoid data loss. "
+            f"This may happen if a previous sync failed mid-way and the state was checkpointed."
+        )
+        logging.warning(warning_message)
 
     def _create_async_job_status_mapping(
         self, model: AsyncJobStatusMapModel, config: Config, **kwargs: Any

unit_tests/sources/declarative/test_state_delegating_stream.py

@@ -2,10 +2,12 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 #
 
+import copy
 import json
 from unittest.mock import MagicMock
 
 import freezegun
+import pytest
 
 from airbyte_cdk.models import (
     AirbyteStateBlob,
@@ -253,3 +255,119 @@ def test_incremental_retriever():
             {"id": 4, "name": "item_4", "updated_at": "2024-02-01"},
         ]
         assert expected_incremental == incremental_records
+
+
+def _create_manifest_with_retention_period(api_retention_period: str) -> dict:
+    """Create a manifest with api_retention_period set on the StateDelegatingStream."""
+    manifest = copy.deepcopy(_MANIFEST)
+    manifest["definitions"]["TestStream"]["api_retention_period"] = api_retention_period
+    return manifest
+
+
+@freezegun.freeze_time("2024-07-15")
+def test_cursor_age_validation_falls_back_to_full_refresh_when_cursor_too_old():
+    """Test that when cursor is older than retention period, full refresh is used."""
+    manifest = _create_manifest_with_retention_period("P7D")
+
+    with HttpMocker() as http_mocker:
+        http_mocker.get(
+            HttpRequest(url="https://api.test.com/items"),
+            HttpResponse(
+                body=json.dumps(
+                    [
+                        {"id": 1, "name": "item_1", "updated_at": "2024-07-13"},
+                        {"id": 2, "name": "item_2", "updated_at": "2024-07-14"},
+                    ]
+                )
+            ),
+        )
+
+        state = [
+            AirbyteStateMessage(
+                type=AirbyteStateType.STREAM,
+                stream=AirbyteStreamState(
+                    stream_descriptor=StreamDescriptor(name="TestStream", namespace=None),
+                    stream_state=AirbyteStateBlob(updated_at="2024-07-01"),
+                ),
+            )
+        ]
+        source = ConcurrentDeclarativeSource(
+            source_config=manifest, config=_CONFIG, catalog=None, state=state
+        )
+        configured_catalog = create_configured_catalog(source, _CONFIG)
+
+        records = get_records(source, _CONFIG, configured_catalog, state)
+        expected = [
+            {"id": 1, "name": "item_1", "updated_at": "2024-07-13"},
+            {"id": 2, "name": "item_2", "updated_at": "2024-07-14"},
+        ]
+        assert expected == records
+
+
+@freezegun.freeze_time("2024-07-15")
+def test_cursor_age_validation_uses_incremental_when_cursor_within_retention():
+    """Test that when cursor is within retention period, incremental sync is used."""
+    manifest = _create_manifest_with_retention_period("P30D")
+
+    with HttpMocker() as http_mocker:
+        http_mocker.get(
+            HttpRequest(
+                url="https://api.test.com/items_with_filtration?start=2024-07-13&end=2024-07-15"
+            ),
+            HttpResponse(
+                body=json.dumps(
+                    [
+                        {"id": 3, "name": "item_3", "updated_at": "2024-07-14"},
+                    ]
+                )
+            ),
+        )
+
+        state = [
+            AirbyteStateMessage(
+                type=AirbyteStateType.STREAM,
+                stream=AirbyteStreamState(
+                    stream_descriptor=StreamDescriptor(name="TestStream", namespace=None),
+                    stream_state=AirbyteStateBlob(updated_at="2024-07-13"),
+                ),
+            )
+        ]
+        source = ConcurrentDeclarativeSource(
+            source_config=manifest, config=_CONFIG, catalog=None, state=state
+        )
+        configured_catalog = create_configured_catalog(source, _CONFIG)
+
+        records = get_records(source, _CONFIG, configured_catalog, state)
+        expected = [
+            {"id": 3, "name": "item_3", "updated_at": "2024-07-14"},
+        ]
+        assert expected == records
+
+
+@freezegun.freeze_time("2024-07-15")
+def test_cursor_age_validation_with_1_day_retention_falls_back():
+    """Test cursor age validation with P1D retention period falls back to full refresh."""
+    manifest = _create_manifest_with_retention_period("P1D")
+
+    with HttpMocker() as http_mocker:
+        http_mocker.get(
+            HttpRequest(url="https://api.test.com/items"),
+            HttpResponse(body=json.dumps([{"id": 1, "updated_at": "2024-07-14"}])),
+        )
+
+        state = [
+            AirbyteStateMessage(
+                type=AirbyteStateType.STREAM,
+                stream=AirbyteStreamState(
+                    stream_descriptor=StreamDescriptor(name="TestStream", namespace=None),
+                    stream_state=AirbyteStateBlob(updated_at="2024-07-13"),
+                ),
+            )
+        ]
+        source = ConcurrentDeclarativeSource(
+            source_config=manifest, config=_CONFIG, catalog=None, state=state
+        )
+        configured_catalog = create_configured_catalog(source, _CONFIG)
+
+        records = get_records(source, _CONFIG, configured_catalog, state)
+        assert len(records) == 1