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
• 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 (brianjlai's architectural review feedback):

• Simplified cursor evaluation logic — Removed the loop over cursors. Now we just pass incremental_stream.cursor directly to _is_cursor_older_than_retention_period instead of a list. The full_refresh cursor (FinalStateCursor) doesn't have a meaningful cursor datetime anyway.
• Fixed FinalStateCursor short-circuit bug — Changed FinalStateCursor.get_cursor_datetime_from_state to return None instead of now(). Previously, returning now() would short-circuit before checking the incremental cursor, potentially missing stale incremental cursors. The NO_CURSOR_STATE_KEY state format is handled separately as an early return in _is_cursor_older_than_retention_period.
• Removed hasattr check — All cursors should have get_cursor_datetime_from_state since it's in the Cursor interface. If a cursor doesn't have it, that's a bug we should surface.
• Updated FinalStateCursor docstring — Explicitly explains that full refresh streams don't track a cursor position and returns None to indicate this.
• Fixed type annotation — Changed cursor: Any to cursor: Cursor in _is_cursor_older_than_retention_period for proper type safety.

Previous changes:

• Added get_cursor_datetime_from_state to concurrent cursor hierarchy
• Added integration test for parent StateDelegatingStream with retention period
• Replaced has_parent_state boolean with actual parent state parameter
• Removed per-partition state fallback — cursor classes are solely responsible for state extraction

Review & Testing Checklist for Human

• Verify FinalStateCursor returning None is correct — The change from returning now() to None fixes the short-circuit bug, but verify no other code depends on FinalStateCursor returning a datetime value.
• Verify the simplified single-cursor design handles all state formats — The factory now only checks the incremental cursor. Verify this correctly handles: simple datetime state, per-partition state (via ConcurrentPerPartitionCursor), and FinalStateCursor state (via NO_CURSOR_STATE_KEY early return).
• Verify parent stream retention checking works — Test that when a parent stream's cursor (embedded in child's per-partition state) is older than retention, it correctly falls back to full refresh.

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) 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 session: https://app.devin.ai/sessions/5e2b4ff66c50407789e500d6e25e5d5c
Requested by: gl_anatolii.yatsuk@airbyte.io

[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 880 tests  +25   3 868 ✅ +25   6m 50s ⏱️ +23s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit acd7156. ± Comparison against base commit 15542de.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

3 883 tests  +25   3 871 ✅ +25   11m 36s ⏱️ +35s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

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

[Brian Lai (brianjlai)]

the overall idea makes sense and i like how we've moved the parsing logic into the respective concurrent cursor implementations.

I just have a few issues and concerns over how complex we've made the cursor evaluation on the model_to_component_factory so I wanna make sure my understanding is right, cleaning up the parts that feel a bit over-engineered, and confirm the case w/ the FinalStateCursor that I mention below

[Brian Lai (brianjlai)]

Not a big fan of having a custom condition here to check if the cursor has get_cursor_datetime_from_state() method because it is just making an already complicated model to component factory have even more one-off conditions.
What I think we should do is go back to the prior design where all the legacy cursors implement get_cursor_datetime_from_state() and just raise a NotImplemented error. And then we just let that bubble up to here and fail loudly. I don't think we have any flows that would result in a legacy cursor getting instantiated. But it feels like we have a flaw in our code flow logic if we have to have such an explicit field/method check. We should operate under the assumption it always exists? And if it doesn't it means we have a class where we forgot to implement it.

I see its added to our concurrent Cursor interface so that would imply that it would always exist. The exception being if we accidentally get a legacy cursor in which we should fail loudly w/ not implemented

[Anatolii Yatsuk (tolik0)]

This check is intended to ensure that, if we introduce new cursors or a custom cursor that does not inherit from the base cursor class, the system fails loudly with a clear and actionable error. The goal is to make it immediately obvious what needs to be implemented or fixed, rather than relying on implicit assumptions.

[devin-ai-integration]

Thanks for the context! This comment is on an older version of the code. In the latest version, I removed the hasattr check per brianjlai's feedback (see comment 41).

brianjlai's reasoning was that all cursors should have get_cursor_datetime_from_state since it's defined in the Cursor base class interface. If a cursor doesn't have it, that's a bug we should surface rather than handle gracefully.

However, I understand your point about defensive programming - if someone introduces a custom cursor that doesn't inherit from Cursor, we'd get an unclear AttributeError instead of a helpful message.

Options:

1. Keep current (no hasattr) - Rely on the type system (cursor: Cursor) to enforce the interface. Any AttributeError would indicate a type violation.
2. Add hasattr back - Provide a clear error message for custom cursors that don't implement the method.
3. Try/except wrapper - Catch AttributeError and raise a descriptive ValueError.

Would you and brianjlai like to align on which approach is preferred? I'm happy to implement whichever you decide on.

---

Devin session

[Brian Lai (brianjlai)]

this part confuses me a bit. We take in two cursors (the full refresh cursor and the incremental cursor), we parse the state to get the value and save it to cursor_datetime.

But then below when we compare if cursor_datetime < retention_cutoff:, we do this outside the loop. So if we get a value from the full_refresh_cursor then we break and compare. Otherwise we get the incremental_cursor and compare.

It just seems like this entire method is written in a quite unintuitive way making it hard to understand. To me, it might just be simpler to just have a simpler design:

# 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

I think this loop just complicates things quite a bit

[Brian Lai (brianjlai)]

I kind of feel like this implemention is unnecessarily complicated. We've designed _is_cursor_older_than_retention_period to be very flexible to support N number of cursors when we effectively know its gonna just be 2. Maybe in the future this changes, but it makes the code written a lot harder to understand. This is all under the hood so if we do need to change this to a list later, we should just do it then

[Brian Lai (brianjlai)]

This part didn't feel that intuitive. I think I understand the idea that by always returning now() we'll always be within the retention window for a full refresh, but we should call that out as the reason why we return now. Because in reality, a full refresh final state cursor doesn't conceptually have a state, it's empty. So we need to be explicit in the docstring about why we do this

[Brian Lai (brianjlai)]

So for the case where we have a state delegating stream where the full refresh implementation has no cursor and therefore has a 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. Am I understanding this right?

And if that is the case, because of how the code is written, my worry is that 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.

Anatolii Yatsuk (@tolik0) I might not be understand this flow right because i think it is written in a fairly overcomplicated way, but just want to check my understanding against this condition

[Anatolii Yatsuk (tolik0)]

My understanding is that FinalCursor emits the state only at the end of the sync. So, if we have a state from a full refresh, we switch to incremental. If the incremental state is outdated, we switch back to the FinalCursor to retrieve all records.

[devin-ai-integration]

Yes, your understanding is correct! The flow is:

1. State from completed full refresh (NO_CURSOR_STATE_KEY: True) → Use incremental stream (the full refresh already completed successfully)
2. Incremental state within retention period → Use incremental stream (safe to continue from where we left off)
3. Incremental state older than retention period → Fall back to full refresh (FinalStateCursor) to retrieve all records and avoid data loss

The current implementation handles this via:

• Early return False when stream_state.get(NO_CURSOR_STATE_KEY) is true (case 1)
• Checking incremental_stream.cursor.get_cursor_datetime_from_state() against the retention cutoff (cases 2 & 3)

Note: This comment is on an older version of the code. The latest version (per brianjlai's feedback) simplified the logic to only check the incremental cursor directly, since FinalStateCursor doesn't have a meaningful cursor datetime anyway.

---

Devin session

[devin-ai-integration]

Thanks for the thorough review! You've identified a real bug with the FinalStateCursor short-circuiting.

The bug: When cursors = [full_refresh_stream.cursor, incremental_stream.cursor], if the full_refresh cursor is a FinalStateCursor, it returns now() unconditionally. This causes us to break out of the loop before checking the incremental cursor, so we never detect if the incremental cursor is outside the retention window.

My plan to fix all your comments:

1. Remove the hasattr check - All cursors should have get_cursor_datetime_from_state since it's in the Cursor interface. If a cursor doesn't have it, that's a bug we should surface.

2. Fix FinalStateCursor.get_cursor_datetime_from_state - Change it to return None instead of now(). The NO_CURSOR_STATE_KEY case is already handled at the top of _is_cursor_older_than_retention_period, so FinalStateCursor doesn't need to return a value. Returning None means "I can't parse this state format", which is semantically correct.

3. Simplify the logic - Remove the loop and just check the incremental cursor directly. The full_refresh cursor (FinalStateCursor) doesn't have a meaningful cursor datetime anyway, and its state format is already handled by the NO_CURSOR_STATE_KEY check.

4. Update docstrings - Make FinalStateCursor's docstring explicit about why it returns None (full refresh doesn't have a cursor datetime).

I'll implement these changes now.

---

Devin session

[devin-ai-integration]

Thanks for the thorough review! I've addressed all your concerns in commit 1163395:

Comment 41 (hasattr check): Removed the hasattr check entirely. All cursors should have get_cursor_datetime_from_state since it's in the Cursor interface. If a cursor doesn't have it, that's a bug we should surface.

Comment 42 (loop complexity): Removed the loop entirely. Now we just pass incremental_stream.cursor directly to _is_cursor_older_than_retention_period instead of a list of cursors. The full_refresh cursor (FinalStateCursor) doesn't have a meaningful cursor datetime anyway.

Comment 43 (unnecessary list abstraction): Fixed by passing a single cursor instead of a list. The method signature is now _is_cursor_older_than_retention_period(stream_state, cursor, api_retention_period, stream_name).

Comment 44 (FinalStateCursor docstring): Updated FinalStateCursor.get_cursor_datetime_from_state to return None with an explicit docstring explaining that full refresh streams don't track a cursor position and the NO_CURSOR_STATE_KEY state format is handled separately.

Comment 45 (FinalStateCursor short-circuit bug): Fixed! You correctly identified the bug. By changing FinalStateCursor.get_cursor_datetime_from_state to return None instead of now(), and simplifying to only check the incremental cursor, we now correctly detect stale incremental cursors.

The simplified flow is now:

1. Check for NO_CURSOR_STATE_KEY (completed full refresh) → return False (use incremental)
2. Get cursor datetime from incremental cursor
3. If cursor datetime is None → return True (fall back to full refresh)
4. Compare against retention cutoff

---

Devin session

[Brian Lai (brianjlai)]

we should not use Any in favor of the correct interface class which i think is Cursor

[devin-ai-integration]

Good catch! Changed cursor: Any to cursor: Cursor in commit acd7156. The Cursor class is already imported in this file.

---

Devin session

[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