feat(cdk): Add cursor age validation to StateDelegatingStream

[devin-ai-integration]

Summary

Adds an optional api_retention_period field to StateDelegatingStream that validates whether a cursor is within an API's data retention window before using incremental sync. When the cursor is older than the retention period, the connector automatically falls back to full refresh to avoid data loss.

This addresses the issue where APIs like Stripe Events only retain data for 30 days - if a sync fails mid-way and resumes after the retention window, incremental sync would miss data.

Key changes:

• Added api_retention_period field to StateDelegatingStream schema (ISO8601 duration format: P30D, P1D, PT1H, etc.)
• Added get_cursor_datetime_from_state to concurrent cursor hierarchy (Cursor base, ConcurrentCursor, ConcurrentPerPartitionCursor, FinalStateCursor)
• Raises ValueError at model level when IncrementingCountCursor is used with api_retention_period
• Emits warning log when falling back to full refresh due to stale cursor
• Clears stream state and emits an empty state message to the platform when falling back to full refresh, so the platform does not retain stale state across sync attempts
• No breaking changes — the field is optional with no default

Example usage:

type: StateDelegatingStream
name: events
api_retention_period: P30D  # 30 days
full_refresh_stream: ...
incremental_stream: ...

Updates since last revision

Latest changes (fix temporal ordering of state clearing — per Alfredo Garcia (@agarctfi)'s review):

Previously, full_refresh_stream was constructed before clearing state, so its cursor inherited the old stale cursor position — making the "full refresh" fallback start from the stale cursor instead of start_date. This fix restructures create_state_delegating_stream so that:

1. Only the incremental_stream cursor is used for the retention check (no need to construct full_refresh_stream first)
2. State is cleared before constructing full_refresh_stream, so its cursor starts from start_date
3. NO_CURSOR_STATE_KEY is checked directly on the raw state dict — if present, the previous sync was a completed full refresh and the stream uses incremental (not full refresh again)
4. _is_cursor_older_than_retention_period simplified: removed full_refresh_cursor parameter, only uses incremental_cursor

Previous changes (removed early return for NO_CURSOR_STATE_KEY — per tolik0's feedback):

• Removed the explicit early return for NO_CURSOR_STATE_KEY in _is_cursor_older_than_retention_period
• FinalStateCursor.get_cursor_datetime_from_state now handles this case by returning now() for NO_CURSOR_STATE_KEY: True state, which is always within any retention period
• This simplifies the code by letting the cursor class handle its own state format

Earlier changes (state clearing on fallback to full refresh — per brianjlai):

• When cursor age validation triggers a fallback to full refresh, the stream's state is now cleared via ConnectorStateManager.update_state_for_stream and an empty state message is emitted through the message repository.

Review & Testing Checklist for Human

• Verify the temporal ordering fix — Confirm that full_refresh_stream is constructed after clearing state (line 3605-3610 in factory), so its cursor starts from start_date instead of the old stale cursor position. This is the core bug fix.
• Verify NO_CURSOR_STATE_KEY behavior — When state contains {NO_CURSOR_STATE_KEY: True} (completed full refresh), verify the stream correctly uses incremental sync instead of falling back to full refresh again (test: test_no_cursor_state_key_uses_incremental_not_full_refresh).
• Verify parent stream edge case — When a StateDelegatingStream is used as a parent stream (not in the configured catalog), verify it does not emit state messages that would cause "Stream not found" errors in the destination (test: test_unconfigured_parent_stream_does_not_emit_state_on_retention_fallback).
• Verify state-clearing message ordering — The empty state message is emitted during stream creation (in the factory) before the read starts. Verify that the message repository correctly delivers this empty state message to the platform before any record messages.

Recommended test plan:

1. Apply this change to the Stripe connector's events stream with api_retention_period: P30D, set a cursor state older than 30 days, and verify it falls back to full refresh with the expected warning message and that the entity endpoint is called with created[gte]=<start_date> (not the old cursor)
2. Test with NO_CURSOR_STATE_KEY: True state (e.g., after a completed full refresh) to verify it correctly uses incremental sync without falling back to full refresh
3. Test with IncrementingCountCursor to confirm it raises ValueError during discover() with a clear error message

Notes

Fixes: airbytehq/oncall#11103

Link to Devin run: https://app.devin.ai/sessions/782b85a317204e3c833c3ecc3bc02f1e
Previous Devin sessions: https://app.devin.ai/sessions/c6b25a1216c547139ef8242062f7f135, https://app.devin.ai/sessions/443d1d1a2f524eb1bc378aa48f24aa9c, https://app.devin.ai/sessions/5e2b4ff66c50407789e500d6e25e5d5c
Requested by: Alfredo Garcia (@agarctfi)

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment and CI monitoring

[github-actions]

PyTest Results (Fast)

3 882 tests  +27   3 870 ✅ +27   7m 5s ⏱️ +38s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 5c6c88b. ± Comparison against base commit 15542de.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

3 885 tests  +27   3 873 ✅ +27   11m 27s ⏱️ +26s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 5c6c88b. ± Comparison against base commit 15542de.

♻️ This comment has been updated with latest results.

[Alfredo Garcia (agarctfi)]

/autofix

Auto-Fix Job Info

This job attempts to auto-fix any linting or formating issues. If any fixes are made,
those changes will be automatically committed and pushed back to the PR.

Note: This job can only be run by maintainers. On PRs from forks, this command requires
that the PR author has enabled the Allow edits from maintainers option.

PR auto-fix job started... Check job output.

✅ Changes applied successfully.

[Copilot]

Pull request overview

This pull request adds cursor age validation to StateDelegatingStream to automatically fall back to full refresh when a cursor is older than an API's data retention period. This prevents data loss scenarios where APIs (like Stripe Events) only retain data for a limited time window (e.g., 30 days), and a sync resumes after that window has passed.

Changes:

• Added optional api_retention_period field to StateDelegatingStream schema (ISO8601 duration format)
• Implemented cursor age validation logic that compares cursor datetime against retention cutoff
• Added warning log emission when falling back to full refresh due to stale cursor

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.

File	Description
airbyte_cdk/sources/declarative/declarative_component_schema.yaml	Added api_retention_period field definition with ISO8601 duration examples
airbyte_cdk/sources/declarative/models/declarative_component_schema.py	Auto-generated model updates reflecting the new field in StateDelegatingStream
airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py	Core implementation of cursor age validation with three new helper methods
unit_tests/sources/declarative/test_state_delegating_stream.py	Added three test cases covering cursor age validation scenarios (too old, within retention, edge case)

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Anatolii Yatsuk (tolik0)]

Do we need this check, since it is already covered by the final cursor?

[devin-ai-integration]

Good question! Yes, this check is still necessary because in practice, full_refresh_stream.cursor is a DatetimeBasedCursor (created from the manifest's full_refresh_stream definition), not a FinalStateCursor.

When we call full_refresh_cursor.get_cursor_datetime_from_state(stream_state) with state {NO_CURSOR_STATE_KEY: True}:

• DatetimeBasedCursor tries to extract the cursor field from the state, but since the state only has NO_CURSOR_STATE_KEY, it returns None
• Then incremental_cursor.get_cursor_datetime_from_state() (also DatetimeBasedCursor) returns None for the same reason
• If both cursors return None, we fall back to full refresh (return True), which is incorrect for NO_CURSOR_STATE_KEY state

The FinalStateCursor.get_cursor_datetime_from_state implementation exists for completeness in the concurrent cursor hierarchy, but it's not used in this specific flow because the declarative stream factory creates DatetimeBasedCursor instances, not FinalStateCursor instances.

If you'd like to remove this early return, we'd need to ensure one of the cursors can correctly parse NO_CURSOR_STATE_KEY state. One option would be to add NO_CURSOR_STATE_KEY handling to DatetimeBasedCursor.get_cursor_datetime_from_state, but that feels like it's mixing concerns.

Would you prefer to:

1. Keep the early return as-is (current approach)
2. Move the NO_CURSOR_STATE_KEY handling into DatetimeBasedCursor
3. Some other approach?

---

Devin session

[Anatolii Yatsuk (tolik0)]

But if the state is NO_CURSOR_STATE_KEY: True - it is produced by the finalcursor

[devin-ai-integration]

You're right - the {NO_CURSOR_STATE_KEY: True} state is produced by FinalStateCursor.

The issue is that in the current flow, full_refresh_stream.cursor is created from the manifest's full_refresh_stream definition, which uses DatetimeBasedCursor (not FinalStateCursor). So when we call full_refresh_cursor.get_cursor_datetime_from_state(), we're calling DatetimeBasedCursor's method, not FinalStateCursor's method.

Would you like me to:

1. Create a FinalStateCursor instance specifically for checking NO_CURSOR_STATE_KEY state, then call its get_cursor_datetime_from_state method (which returns now() and would correctly indicate the cursor is "current")?
2. Or change the flow to use FinalStateCursor as the first cursor to check instead of full_refresh_stream.cursor?

Either approach would let us remove the early return and rely on FinalStateCursor.get_cursor_datetime_from_state to handle NO_CURSOR_STATE_KEY state.

---

Devin session

[Anatolii Yatsuk (tolik0)]

How is the finalcursor created to be used during the sync?

[devin-ai-integration]

FinalStateCursor is created in two places:

1. In create_simple_retriever (line 3301-3302): When no cursor is provided to the retriever (i.e., the stream doesn't have an incremental_sync defined):

   if cursor is None:
       cursor = FinalStateCursor(name, None, self._message_repository)

2. In _build_concurrent_cursor (line 2289): When there's no incremental_sync defined for concurrent streams.

In the context of StateDelegatingStream:

When we call self._create_component_from_model(model.full_refresh_stream, ...), it creates a DefaultStream. The DefaultStream.cursor property returns the cursor from its retriever.

• If full_refresh_stream doesn't have an incremental_sync defined → the retriever gets a FinalStateCursor → state produced is {NO_CURSOR_STATE_KEY: True}
• If full_refresh_stream does have an incremental_sync defined → the retriever gets a DatetimeBasedCursor → state produced is {"cursor_field": "value"}

The early return for NO_CURSOR_STATE_KEY handles the first case where the full_refresh_stream used FinalStateCursor and produced {NO_CURSOR_STATE_KEY: True} state. In this case, full_refresh_stream.cursor (which is now a DatetimeBasedCursor if the incremental_stream has one) can't parse that state, so we need the explicit check.

Would you like me to change the approach to explicitly create a FinalStateCursor instance and use it to check NO_CURSOR_STATE_KEY state instead of the early return?

---

Devin session

[Anatolii Yatsuk (tolik0)]

So, finalcursor will be created if the stream does not define incremental sync at all. Why do we need to create it explicitly?

[devin-ai-integration]

You're right - if the full_refresh_stream doesn't define incremental_sync, then full_refresh_stream.cursor is already a FinalStateCursor, and it will handle NO_CURSOR_STATE_KEY correctly by returning now().

The early return is a defensive check for an edge case where:

1. The state {NO_CURSOR_STATE_KEY: True} was produced by a previous sync (when full_refresh_stream used FinalStateCursor)
2. But the manifest was later updated to add incremental_sync to full_refresh_stream
3. Now full_refresh_stream.cursor is a DatetimeBasedCursor which can't parse NO_CURSOR_STATE_KEY state

If you think this edge case is unlikely and we should remove the early return to simplify the code, I can do that. The code would rely on FinalStateCursor.get_cursor_datetime_from_state to handle NO_CURSOR_STATE_KEY state, which works correctly when full_refresh_stream doesn't have incremental_sync.

Should I remove the early return?

---

Devin session

[Anatolii Yatsuk (tolik0)]

Yes, remove it

[devin-ai-integration]

Done! Removed the early return for NO_CURSOR_STATE_KEY in commit 1dcc8ab. The FinalStateCursor.get_cursor_datetime_from_state method now handles this case by returning now(), which is always within any retention period.

---

Devin session

[Brian Lai (brianjlai)]

✅