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 (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

Previous changes (tolik0's FinalStateCursor feedback):

• FinalStateCursor.get_cursor_datetime_from_state returns now() if state has NO_CURSOR_STATE_KEY: True, otherwise returns None
• Updated _is_cursor_older_than_retention_period to check both cursors in sequence (full refresh cursor first, then incremental cursor)

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 FinalStateCursor behavior — FinalStateCursor.get_cursor_datetime_from_state returns now() for NO_CURSOR_STATE_KEY: True state. Verify this correctly causes completed full refresh states to use incremental sync (since now() is always within retention).
• 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.
• Verify the dual-cursor checking flow — The factory checks full_refresh_cursor first, then incremental_cursor. Verify this correctly handles all state formats (datetime-based, per-partition, concurrent).

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
2. Test with FinalStateCursor state (e.g., after a completed full refresh with NO_CURSOR_STATE_KEY: True) 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: https://github.com/airbytehq/oncall/issues/11103

Link to Devin run: https://app.devin.ai/sessions/c6b25a1216c547139ef8242062f7f135
Previous Devin sessions: 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   6m 56s ⏱️ +29s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 2a2459d. ± 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 31s ⏱️ +30s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 2a2459d. ± 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.

[Alfredo Garcia (agarctfi)]

(aside) When testing this, we found that when we revert to the Full Refresh path, we don't clear the state for the impacted stream, which is causing us to still miss records. Ideally, we'll clear the state also, so we fetch all the records from the configured start-date.

[Brian Lai (brianjlai)]

(aside)

(aside) When testing this, we found that when we revert to the Full Refresh path, we don't clear the state for the impacted stream, which is causing us to still miss records. Ideally, we'll clear the state also, so we fetch all the records from the configured start-date.

Ok that's good that you actually found this during testing. Anatolii Yatsuk (@tolik0) and i sort of discussed this on a call yesterday that emitting a "first" empty state message to the platform to clear the state before we actually re-run the new full refresh might be an option so I feel like that lines up with what you're observing

[Alfredo Garcia (agarctfi)]

(aside) With the latest commit, the stream now correctly refreshes by clearing to an empty state. One oversight is whether we should set the cursor after the sync finishes to the last sync time, rather than the current behavior of updating it to the last record's cursor value. A behavior I've seen is that on subsequent syncs, we'll continue to run a full refresh if the last record is past the api_retention_period. Not sure if this is something we intended to keep & document in the Limitations & Troubleshooting section of the connector, or if my suggestion above is safe and would work?

[Alfredo Garcia (agarctfi)]

(aside) Spoke with Anatolii, we think the impact for this is very low; The only scenario where this can really have a big impact is if a user has millions of records for a stream, & in the last 30 days doesn't have a single new record for that stream, which we think is highly unlikely.

The issue with using the last sync time as the cursor value is that we cannot guarantee that some APIs won't have a delay in the date range in which they send records. So the current setup is safer for ensuring we don't miss any records. Alternatively, we could add another lookback field for this situation and use the last sync time, but we think it will unnecessarily complicate things.

Local testing looks good. Please give any final feedback or thoughts on the PR, as we plan to start progressive rollouts on a very small subset of users starting Monday.

[Anatolii Yatsuk (tolik0)]

FinalCursor should return now instead. The docstring should explain why we return now.

[devin-ai-integration]

Thanks for the feedback! I want to clarify a potential conflict with brianjlai's earlier review.

brianjlai identified returning now() from FinalStateCursor as a bug (see comment 45):

"FinalStateCursor.get_cursor_datetime_from_state() that returns now(). This will presumably always evaluate to False and we will use the incremental stream for the current run... because we only ever evaluate the FinalStateCursor and short circuit before we check incremental, we will always use the incremental_stream even if it fell outside the retention api window."

His concern was that in the loop-based approach, FinalStateCursor returning now() would short-circuit before checking the incremental cursor, potentially missing stale incremental cursors.

However, your suggested flow (check full refresh cursor first, then incremental if no value) would address this differently - FinalStateCursor returning now() would mean "within retention, use incremental" which is correct for a completed full refresh.

Could you and brianjlai align on the preferred approach?

1. Current (brianjlai's approach): FinalStateCursor returns None, handle NO_CURSOR_STATE_KEY as early return, only check incremental cursor
2. Your approach: FinalStateCursor returns now(), check full refresh cursor first, then incremental cursor if no value

I'm happy to implement whichever you decide on.

---

Devin session

[Anatolii Yatsuk (tolik0)]

Let's return Now only if the state is {NO_CURSOR_STATE_KEY: True} otherwise None

[devin-ai-integration]

Thanks for the clarification! I've implemented your requested change in commit e4f71ff:

1. FinalStateCursor.get_cursor_datetime_from_state now returns now() if state has NO_CURSOR_STATE_KEY: True, otherwise returns None

2. Updated flow in _is_cursor_older_than_retention_period to check both cursors in sequence (full refresh cursor first, then incremental cursor)

3. Kept the early return for NO_CURSOR_STATE_KEY - This is necessary because in the test manifests (and likely real-world usage), the full_refresh_stream has a DatetimeBasedCursor defined in incremental_sync, not a FinalStateCursor. So when we call full_refresh_stream.cursor.get_cursor_datetime_from_state(), it's actually a DatetimeBasedCursor or ConcurrentCursor that tries to extract the cursor field (e.g., updated_at) from the state. Since {NO_CURSOR_STATE_KEY: True} doesn't have that field, both cursors return None, which would incorrectly trigger a full refresh.

The early return ensures that when we have a completed full refresh state (NO_CURSOR_STATE_KEY: True), we correctly use the incremental stream.

All 13 tests pass locally.

---

Devin session

[Anatolii Yatsuk (tolik0)]

The flow should be like this:

# get the retention duration from now() - api_retention
# get full refresh cursor datetime
# if has value then compare against that value w/ retention duration
# if doesn't have value then get the incremental cursor date time
# compare incremental value against that value w/ retention duration

[devin-ai-integration]

Thanks for the detailed flow! This is on an older version of the code. The current implementation (per brianjlai's feedback) simplified to only check the incremental cursor, with NO_CURSOR_STATE_KEY handled as an early return.

Your suggested flow would require checking both cursors in sequence:

# get the retention duration from now() - api_retention
# get full refresh cursor datetime
# if has value then compare against that value w/ retention duration
# if doesn't have value then get the incremental cursor date time
# compare incremental value against that value w/ retention duration

This is a different approach than brianjlai's simplification. I've replied to your other comment asking for alignment between you and brianjlai on the preferred approach.

---

Devin session

[Anatolii Yatsuk (tolik0)]

The final cursor is fixed; now implement this flow.

[devin-ai-integration]

Thanks for confirming the FinalStateCursor changes are good!

I believe I've already implemented the flow you described in commit e4f71ff:

1. FinalStateCursor.get_cursor_datetime_from_state returns now() for NO_CURSOR_STATE_KEY: True state, otherwise None
2. _is_cursor_older_than_retention_period checks full refresh cursor first, then incremental cursor if the first returns None

However, I kept an early return for NO_CURSOR_STATE_KEY in _is_cursor_older_than_retention_period because the full_refresh_stream.cursor is actually a DatetimeBasedCursor (or ConcurrentCursor), not a FinalStateCursor. This is because the manifest defines a DatetimeBasedCursor in the incremental_sync field of the full_refresh_stream.

Could you clarify if you want me to:

1. Keep the current implementation (early return for NO_CURSOR_STATE_KEY + sequential cursor checking)
2. Remove the early return and find a way to use FinalStateCursor directly for the retention check (would require creating a FinalStateCursor instance explicitly rather than using full_refresh_stream.cursor)
3. Something else I may have missed?

All 13 tests pass locally with the current implementation.

---

Devin session

[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)]

✅