fix(cdk): emit interim heartbeat state for global-cursor substreams with empty parent state

[devin-ai-integration]

Summary

A global_substream_cursor substream that walks a large parent with mostly-empty children can emit no RECORD and no STATE for the entire walk, tripping the Airbyte source heartbeat (heartbeat_timeout, threshold 86400s). This happened on a customer's source-stripe initial sync (see airbytehq/oncall#12977) for the customers child streams bank_accounts and customer_balance_transactions.

Root cause: ConcurrentPerPartitionCursor._emit_state_message suppressed the throttled checkpoint STATE whenever _parent_state was empty:

if throttle:
    current_time = self._throttle_state_message()   # 600s throttle
    if current_time is None:
        return
    self._last_emission_time = current_time
    if self._use_global_cursor and not self._parent_state:   # <-- empty-parent guard
        return                                               #     withholds the ONLY keepalive

For a substream without incremental_dependency, _parent_state stays empty for the whole walk, so every interim (per-close_partition) checkpoint is suppressed — the single mechanism that could keep the heartbeat alive is withheld exactly when it is needed. This PR removes that guard so the existing 600s-throttled checkpoint STATE flows even when parent state is empty.

The interim STATE is inert on resume: on the global path, close_partition never advances _global_cursor (only ensure_at_least_one_state_emitted does, at the end of the sync — concurrent_partition_cursor.py:281), so an interim checkpoint carries the stable/unadvanced cursor. Resume from it re-reads rather than skipping → no data loss, no re-read regression. The 600s throttle is unchanged. The guard only ever applied to the global-cursor path (_use_global_cursor and ...), so per-partition state behavior is untouched.

The fix

             self._last_emission_time = current_time
-            # Skip state emit for global cursor if parent state is empty
-            if self._use_global_cursor and not self._parent_state:
-                return

Diagnosis

Emission of interim STATE is driven entirely by close_partition (which calls _emit_state_message(throttle=True)). Two distinct failure modes were investigated for the two failing streams:

• bank_accounts (SubstreamPartitionRouter, global_substream_cursor: true, no incremental_dependency): close_partition fires repeatedly during the walk, but _parent_state is empty → every throttled checkpoint hit the guard → 24h silence. This PR fixes it directly.

• customer_balance_transactions (incremental_dependency: true): SubstreamPartitionRouter.get_stream_state() returns {parent_name: parent.cursor.state} — a dict that always contains the parent key, hence truthy — so the guard was already bypassed for this stream. It is therefore not the cause of its silence. Its production symptom (0 records / 0 STATE for 24h) is a separate root cause: a stall in the parent-generation / parent-read path. Because emission is gated on close_partition, if the parent read stalls, no partition closes and no state-emission change (including this one) can keep it alive. This aligns with the earlier concurrent-source RCA (synchronous parent reads holding partition-generator slots + unbounded queue.get()) and is out of scope for this minimal fix — flagged for a follow-up.

Fixture validation

A large-parent fixture (3000 parents over 30 slow pages, empty children) exercises the real cursor. The 600s throttle is compressed to 1s purely to simulate the many 600s windows that elapse over a 24h walk (production throttle unchanged). "STATE messages (output)" is the number that reached stdout:

stream	incremental_dependency	mode	STATE messages	outcome
bank_accounts	false	fixed (this PR)	12	keepalive flows across the walk
bank_accounts	false	pre-fix guard	1	11 interim checkpoints suppressed → silence
customer_balance_transactions	true	fixed (this PR)	12	flows
customer_balance_transactions	true	pre-fix guard	12	guard already bypassed → not the cause
customer_balance_transactions + 6s parent stall	true	fixed	12, but a 6.0s gap with zero STATE	proves a parent stall silences the stream regardless of the guard

The stall row demonstrates the separate customer_balance_transactions root cause: during the injected 6s parent-read stall, close_partition does not fire, so the max gap between STATE messages equals the stall duration — no state-emission change can bridge it.

Testing

• test_global_cursor_emits_interim_state_even_when_parent_state_empty (parametrized over empty and non-empty parent state): the throttled checkpoint STATE is emitted in both cases for a global-cursor stream.
• Updated test_incremental_parent_state_no_incremental_dependency expectation from 1 → 2 state messages for the resume case: the previously-suppressed interim keepalive is now emitted; the final state is unchanged (interim carries the same inert global cursor).
• Resume safety (test_global_cursor_substream_resume_after_interruption_skips_no_records): interrupt the walk at every interim checkpoint, resume from each, and assert before ∪ after == full expected set (deduped) so both skips and extras are caught. Non-vacuous: restoring the guard emits only 1 STATE and the test fails.
• Resume hardening (OC-12977):
  • No behavior change on the non-empty-parent shape (test_guard_removal_is_a_noop_for_non_empty_parent_state): for a parent with incremental_dependency: true, the emitted records and the full STATE sequence are byte-for-byte identical guard-removed vs. guard-restored — bounding this shared-CDK change's blast radius to the empty-parent case.
  • Data-last fixture (test_global_cursor_substream_resume_with_data_bearing_partition_last_skips_no_records): mirrors the real failure — many empty parents precede the single data-bearing partition, so resume is exercised from checkpoints emitted during the silent stretch before any record exists.
  • Determinism: both resume tests pin default_concurrency=1 and assert the RECORD/STATE interleaving is identical across repeated runs, so reconstructing "records committed before the kill" can't become a concurrent-ordering flake.
  • Tail contribution: assert at least one early checkpoint's resume returns records not already emitted before the kill, so the test can't pass trivially if all records preceded the first checkpoint.
• State size stays bounded (Serhii's orchestrator-overload risk): persisted state remains a single global-cursor value (64 B empty / 110 B with an advancing cursor), byte-identical at 5 vs. 3000 empty parents and vs. the pre-fix baseline; no per-partition states map accumulates and the 600s throttle holds emission to a small constant count regardless of parent count.
• Full file unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py: 34 passed. ruff format, ruff check, mypy clean.

Context: airbytehq/oncall#12977

Link to Devin session: https://app.devin.ai/sessions/36bd90d019394ef59de523d61c4abc46

[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, 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/1783450708-heartbeat-guard-relax#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/1783450708-heartbeat-guard-relax

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)

4 134 tests  +11   4 122 ✅ +11   8m 55s ⏱️ + 1m 19s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit d66b623. ± Comparison against base commit 53785b6.

This pull request removes 1 and adds 12 tests. Note that renamed tests count towards both.

unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_incremental_parent_state_no_incremental_dependency[test_incremental_parent_state_with-manifest1-mock_requests1-expected_records1-initial_state1-expected_state1-1]

unit_tests.sources.declarative.extractors.test_response_to_file_extractor.ResponseToFileExtractorTest ‑ test_empty_fields_are_converted_to_none_when_preserving
unit_tests.sources.declarative.extractors.test_response_to_file_extractor.ResponseToFileExtractorTest ‑ test_na_string_values_are_converted_to_none_by_default
unit_tests.sources.declarative.extractors.test_response_to_file_extractor.ResponseToFileExtractorTest ‑ test_na_string_values_are_preserved_when_enabled
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_global_cursor_emits_interim_state_even_when_parent_state_empty[empty_parent_state]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_global_cursor_emits_interim_state_even_when_parent_state_empty[non_empty_parent_state]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_global_cursor_substream_resume_after_interruption_skips_no_records
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_global_cursor_substream_resume_with_data_bearing_partition_last_skips_no_records
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_guard_removal_is_a_noop_for_non_empty_parent_state
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_incremental_parent_state_no_incremental_dependency[test_incremental_parent_state_with-manifest1-mock_requests1-expected_records1-initial_state1-expected_state1-2]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_persisted_state_has_no_per_partition_entries_even_when_cursor_advances
…

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

4 137 tests   4 125 ✅  10m 40s ⏱️
    1 suites     12 💤
    1 files        0 ❌

Results for commit d66b623.

♻️ This comment has been updated with latest results.

[Anatolii Yatsuk (tolik0)]

/prerelease

Prerelease Job Info

This job triggers the publish workflow with default arguments to create a prerelease.

Prerelease job started... Check job output.

✅ Prerelease workflow triggered successfully.

View the publish workflow run: https://github.com/airbytehq/airbyte-python-cdk/actions/runs/28958762520

[ZaneHyattAB]

/prerelease

Prerelease Job Info

This job triggers the publish workflow with default arguments to create a prerelease.

Prerelease job started... Check job output.

✅ Prerelease workflow triggered successfully.

View the publish workflow run: https://github.com/airbytehq/airbyte-python-cdk/actions/runs/28969579675