[ZaneHyattAB]

Summary

A declarative substream with incremental_dependency: true can persist a parent state that already covers a parent record whose child was never read. This happens when a parent slice request fails after the previous parent slice was closed. The skipped child is never read again, because the resumed sync starts the parent after that record.

This PR changes one line: ConcurrentPerPartitionCursor.ensure_at_least_one_state_emitted now refreshes _parent_state from the partition router only when the stream completed.

Requested by ZaneHyattAB.

Why a record is lost

1. In SubstreamPartitionRouter.stream_slices, the router calls parent_stream.cursor.close_partition(partition) on the last parent record of a slice, before it yields that record's child slice.
2. ConcurrentPerPartitionCursor.stream_slices consumes the router through iterate_with_last_flag_and_state, which calls next() on the router before yielding the current child slice. When the next parent slice request raises, the previous slice's last child slice is held as current and is never yielded.
3. On stream completion, ensure_at_least_one_state_emitted sees every generated partition closed and ran self._parent_state = self._partition_router.get_stream_state() (line 298 on main). That parent state includes the closed parent slice. It overwrote the low-water mark that _check_and_update_parent_state keeps from closed child partitions, and the overwritten value was emitted.

When the parent API returns newest first (as Stripe does), the lost record is the oldest record of the slice before the failing one. A resume with the emitted state starts at the newest record of that slice, so the lost child is never requested.

Fix

         if all_partitions_closed:
             if self._synced_some_data and stream_completed:
                 self._global_cursor = self._new_global_cursor
                 self._lookback_window = self._timer.finish()
-            self._parent_state = self._partition_router.get_stream_state()
+            if stream_completed:
+                self._parent_state = self._partition_router.get_stream_state()

stream_completed is all_partitions_closed and self._finished_generating_slices. _finished_generating_slices is set only after the router loop finishes, so syncs that complete are unchanged:

• Nested parent cursors are still refreshed. The router calls ensure_at_least_one_state_emitted on a parent cursor only at its last slice, which is after that parent's generator is exhausted.
• Parent slices with no records are covered by the tests below.

When generation stops early, the emitted parent state is the one recorded for closed child partitions.

The same gate also stops one case of a progress-free state from being emitted: a global-cursor stream whose first parent request fails. On main that case emits:

{"use_global_cursor": true, "lookback_window": 0, "parent_state": {"posts": {"updated_at": "2024-01-01T00:00:00Z"}}}

This PR also changes that case: the progress-free state is no longer emitted. For a plain substream on main the state was harmless: an attempt resumed from it requests the same parent windows as a fresh attempt, starting at the start date (checked by a test run). It is the progress-free shape #1254 meant to suppress, though, because a StateDelegatingStream treats any non-empty state as "already synced once".

Tests

All the new tests are in unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py. They read with ConcurrentDeclarativeSource and requests_mock. The mocked parent returns records newest first and filters them by the start_time/end_time request parameters. The failure-path tests run a failing attempt, then resume from its last emitted state with the failure removed, and assert that the lost record's child is requested.

Test	main	this PR
test_given_failed_next_parent_request_when_resume_then_last_child_of_previous_parent_slice_is_read[global]	fail	pass
test_given_failed_next_parent_request_when_resume_then_last_child_of_previous_parent_slice_is_read[per-partition]	fail	pass
test_given_empty_parent_window_and_failed_next_parent_request_when_resume_then_child_is_read	fail	pass
test_given_no_state_and_failed_next_parent_request_when_resume_then_child_is_read	fail	pass
test_given_global_cursor_and_failed_first_parent_request_when_read_then_no_state_emitted	fail	pass
test_given_nested_parents_and_failed_middle_parent_request_when_resume_then_no_regression	pass	pass
test_given_empty_parent_window_when_sync_succeeds_then_parent_state_is_newest_record	pass	pass
test_given_successful_sync_when_read_then_parent_and_child_state_advance_to_newest_record[global]	pass	pass
test_given_successful_sync_when_read_then_parent_and_child_state_advance_to_newest_record[per-partition]	pass	pass

The nested test (posts → comments → votes) passes on main as well, so it doesn't reproduce the bug for nested parents. It's kept as a no-regression test. In it, post 1's day-2 comments request fails after its day-1 comments were returned newest first. The middle cursor stays at its input value because its partition for post 1 never finishes, so the resume re-reads day 1 and recovers the lost vote.

The existing #1254 tests (test_given_failed_*_when_ensure_state_emitted_*) pass unchanged.

Local runs on this branch:

• pytest unit_tests/sources/declarative/incremental/ unit_tests/sources/declarative/partition_routers/ unit_tests/sources/declarative/parsers/ unit_tests/sources/streams/concurrent/: 990 passed. After the latest test-only commit, incremental/ passes 186 and partition_routers/ passes 82.
• ruff check ., ruff format --check . and mypy --config-file mypy.ini airbyte_cdk: clean.

Exposure

incremental_dependency: true appears 107 times across 39 connector manifests in airbytehq/airbyte. The counts come from a grep of airbyte-integrations/connectors/**/*.yaml. The most affected are source-fastly and source-zendesk-support (10 each), source-track-pms (8), source-retailexpress-by-maropost (6), and source-dolibarr and source-stripe (5 each).

Each of them can lose one parent record's children per sync in which a parent request fails. The loss is permanent when the parent cursor is a creation time. A parent cursored on an update time recovers the record on its next update.

Risk

The emitted parent state can now only lag behind the parent router, never lead it. The worst case is that a resumed sync re-reads parent records, and their children, from the last parent state that is fully processed.

Syncs that complete emit the same state as before.

Link to Devin session: https://app.devin.ai/sessions/74504a9af7b44c0ea9c97b52349f1b12
Open in Devin Desktop: https://app.devin.ai/desktop/session/74504a9af7b44c0ea9c97b52349f1b12?variant=devin

[devin-ai-integration]

I'll fix CI failures and address comments from users with write access. I'll skip comments containing "(aside)".

• Disable automatic comment, CI, and merge conflict monitoring

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@devin/1793381933-cdk-parent-state-gate#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch devin/1793381933-cdk-parent-state-gate

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Fast)

6 261 tests  +38   6 239 ✅ +38   6m 36s ⏱️ + 1m 12s
    1 suites ± 0      22 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 0d8c920. ± Comparison against base commit abb5be8.

This pull request removes 3 and adds 41 tests. Note that renamed tests count towards both.

unit_tests.sources.streams.http.requests_native_auth.test_requests_native_auth.TestOauth2Authenticator ‑ test_refresh_access_token_surfaces_provider_error_code[invalid_client-AADSTS7000218: The request body must contain the following parameter: 'client_assertion' or 'client_secret'.\r\nCorrelation ID: 11111111-2222-3333-4444-555555555555-AADSTS7000218]
unit_tests.sources.streams.http.requests_native_auth.test_requests_native_auth.TestOauth2Authenticator ‑ test_refresh_access_token_surfaces_provider_error_code[invalid_grant-AADSTS50076: Due to a configuration change made by your administrator, or because you moved to a new location, you must use multi-factor authentication to access the resource.\r\nCorrelation ID: 11111111-2222-3333-4444-555555555555-AADSTS50076]
unit_tests.sources.streams.http.requests_native_auth.test_requests_native_auth.TestOauth2Authenticator ‑ test_refresh_access_token_surfaces_provider_error_code[invalid_grant-AADSTS50173: The provided grant has expired due to it being revoked, a fresh auth token is needed. The user might have changed or reset their password.\r\nTrace ID: 00000000-0000-0000-0000-000000000000\r\nCorrelation ID: 11111111-2222-3333-4444-555555555555-AADSTS50173]

unit_tests.sources.declarative.auth.test_oauth.TestOauth2Authenticator ‑ test_refresh_access_token_rejected_uses_configured_message
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_empty_parent_window_and_failed_next_parent_request_when_resume_then_child_is_read
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_empty_parent_window_when_sync_succeeds_then_parent_state_is_newest_record
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_next_parent_request_when_resume_then_last_child_of_previous_parent_slice_is_read[False]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_next_parent_request_when_resume_then_last_child_of_previous_parent_slice_is_read[True]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_global_cursor_and_failed_first_parent_request_when_read_then_no_state_emitted
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_nested_parents_and_failed_middle_parent_request_when_resume_then_no_regression
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_no_state_and_failed_next_parent_request_when_resume_then_child_is_read
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_successful_sync_when_read_then_parent_and_child_state_advance_to_newest_record[False]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_successful_sync_when_read_then_parent_and_child_state_advance_to_newest_record[True]
…

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

6 264 tests  +38   6 242 ✅ +38   12m 0s ⏱️ +23s
    1 suites ± 0      22 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 0d8c920. ± Comparison against base commit abb5be8.

This pull request removes 3 and adds 41 tests. Note that renamed tests count towards both.

unit_tests.sources.streams.http.requests_native_auth.test_requests_native_auth.TestOauth2Authenticator ‑ test_refresh_access_token_surfaces_provider_error_code[invalid_client-AADSTS7000218: The request body must contain the following parameter: 'client_assertion' or 'client_secret'.\r\nCorrelation ID: 11111111-2222-3333-4444-555555555555-AADSTS7000218]
unit_tests.sources.streams.http.requests_native_auth.test_requests_native_auth.TestOauth2Authenticator ‑ test_refresh_access_token_surfaces_provider_error_code[invalid_grant-AADSTS50076: Due to a configuration change made by your administrator, or because you moved to a new location, you must use multi-factor authentication to access the resource.\r\nCorrelation ID: 11111111-2222-3333-4444-555555555555-AADSTS50076]
unit_tests.sources.streams.http.requests_native_auth.test_requests_native_auth.TestOauth2Authenticator ‑ test_refresh_access_token_surfaces_provider_error_code[invalid_grant-AADSTS50173: The provided grant has expired due to it being revoked, a fresh auth token is needed. The user might have changed or reset their password.\r\nTrace ID: 00000000-0000-0000-0000-000000000000\r\nCorrelation ID: 11111111-2222-3333-4444-555555555555-AADSTS50173]

unit_tests.sources.declarative.auth.test_oauth.TestOauth2Authenticator ‑ test_refresh_access_token_rejected_uses_configured_message
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_empty_parent_window_and_failed_next_parent_request_when_resume_then_child_is_read
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_empty_parent_window_when_sync_succeeds_then_parent_state_is_newest_record
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_next_parent_request_when_resume_then_last_child_of_previous_parent_slice_is_read[False]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_next_parent_request_when_resume_then_last_child_of_previous_parent_slice_is_read[True]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_global_cursor_and_failed_first_parent_request_when_read_then_no_state_emitted
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_nested_parents_and_failed_middle_parent_request_when_resume_then_no_regression
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_no_state_and_failed_next_parent_request_when_resume_then_child_is_read
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_successful_sync_when_read_then_parent_and_child_state_advance_to_newest_record[False]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_successful_sync_when_read_then_parent_and_child_state_advance_to_newest_record[True]
…

♻️ This comment has been updated with latest results.

[Anatolii Yatsuk (tolik0)]

Verified locally: the 5 tests listed as failing on main fail with main's cursor and pass at 0d8c920 (46/46 in the file); completed syncs are unchanged since _finished_generating_slices is set before the generation sentinel. A probe also confirms the gate covers GroupingPartitionRouter, where main lost up to group_size − 1 children per failure.