From 0e2c06c9c06379f7372ebe9b15ec9f67400d1409 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 7 Jul 2026 18:58:35 +0000 Subject: [PATCH 1/4] fix(cdk): emit interim heartbeat state for global-cursor substreams with empty parent state During a long parent walk with mostly-empty children, a global_substream_cursor substream without incremental_dependency emitted no RECORD and no STATE for the whole walk, tripping the Airbyte source heartbeat (86400s). ConcurrentPerPartitionCursor._emit_state_message suppressed the throttled checkpoint STATE whenever the parent state was empty, so the only keepalive signal was withheld exactly when it was needed. Remove the empty-parent guard so the existing 600s-throttled checkpoint STATE is emitted even when parent state is empty. The interim state carries the stable (unadvanced) global cursor, so it is inert on resume (no data loss, no re-read regression). The 600s throttle is unchanged. Co-Authored-By: bot_apk --- .../concurrent_partition_cursor.py | 3 -- .../test_concurrent_perpartitioncursor.py | 47 ++++++++++++++++++- 2 files changed, 46 insertions(+), 4 deletions(-) diff --git a/airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py b/airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py index d1f2ca41e3..e6c69a296d 100644 --- a/airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py +++ b/airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py @@ -298,9 +298,6 @@ def _emit_state_message(self, throttle: bool = True) -> None: if current_time is None: return 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 self._connector_state_manager.update_state_for_stream( self._stream_name, diff --git a/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py b/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py index 28b9b84600..6ccc4cde81 100644 --- a/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py +++ b/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py @@ -1021,7 +1021,11 @@ def _run_read( "parent_state": {}, "state": {"created_at": VOTE_100_CREATED_AT}, }, - 1, + # Two state messages: one interim heartbeat-keepalive checkpoint emitted + # during the walk (previously suppressed by the empty-parent guard) plus + # the final state. The interim checkpoint carries the same inert global + # cursor, so the final state is unchanged. + 2, ), ], ) @@ -3609,6 +3613,47 @@ def test_state_throttling(mocker): mock_repo.emit_message.assert_called_once() +@pytest.mark.parametrize( + "parent_state", + [ + pytest.param({}, id="empty_parent_state"), + pytest.param({"parent": {"updated_at": "2024-01-01"}}, id="non_empty_parent_state"), + ], +) +def test_global_cursor_emits_interim_state_even_when_parent_state_empty(mocker, parent_state): + """ + Regression test for OC-12977: a global_substream_cursor stream walking a large + parent with mostly-empty children must still emit its throttled checkpoint STATE + to keep the Airbyte source heartbeat alive, even when the parent state is empty + (no ``incremental_dependency``). The previous empty-parent guard suppressed this + interim emission, causing >24h silence and a heartbeat timeout. + """ + cursor = ConcurrentPerPartitionCursor( + cursor_factory=MagicMock(), + partition_router=MagicMock(), + stream_name="test_stream", + stream_namespace=None, + stream_state={}, + message_repository=MagicMock(), + connector_state_manager=MagicMock(), + connector_state_converter=MagicMock(), + cursor_field=MagicMock(), + ) + + cursor._use_global_cursor = True + cursor._parent_state = parent_state + cursor._last_emission_time = 0 + + mock_time = mocker.patch("time.time") + # Exceed the 600s throttle so emission is not throttled. + mock_time.return_value = 700 + + cursor._emit_state_message() + + cursor._connector_state_manager.update_state_for_stream.assert_called_once() + cursor._message_repository.emit_message.assert_called_once() + + def test_given_no_partitions_processed_when_close_partition_then_no_state_update(): mock_cursor = MagicMock() # No slices for no partitions From cf525515dff35eebdf81ed91fe5fa7a4d1676b8e Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 7 Jul 2026 21:46:03 +0000 Subject: [PATCH 2/4] test(cdk): resume global-cursor substream after mid-walk interruption without skipping records Co-Authored-By: bot_apk --- .../test_concurrent_perpartitioncursor.py | 217 ++++++++++++++++++ 1 file changed, 217 insertions(+) diff --git a/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py b/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py index 6ccc4cde81..13065a2cf1 100644 --- a/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py +++ b/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py @@ -1056,6 +1056,223 @@ def test_incremental_parent_state_no_incremental_dependency( ) +# Fresh initial sync of the ``post_comment_votes`` global-cursor substream whose parent +# (``post_comments``) has no ``incremental_dependency`` -- i.e. the ``bank_accounts`` shape +# from OC-12977. Every child request filters on the config ``start_date`` because the global +# cursor is empty/unadvanced during the walk, so the exact same URLs answer both the initial +# read and any resume-from-interim-checkpoint read. +GLOBAL_CURSOR_RESUME_MOCK_REQUESTS = [ + # Posts page 1 + ( + f"https://api.example.com/community/posts?per_page=100&start_time={START_DATE}", + { + "posts": [ + {"id": 1, "updated_at": POST_1_UPDATED_AT}, + {"id": 2, "updated_at": POST_2_UPDATED_AT}, + ], + "next_page": f"https://api.example.com/community/posts?per_page=100&start_time={START_DATE}&page=2", + }, + ), + # Posts page 2 + ( + f"https://api.example.com/community/posts?per_page=100&start_time={START_DATE}&page=2", + {"posts": [{"id": 3, "updated_at": POST_3_UPDATED_AT}]}, + ), + # Comments for post 1 page 1 + ( + "https://api.example.com/community/posts/1/comments?per_page=100", + { + "comments": [ + {"id": 9, "post_id": 1, "updated_at": COMMENT_9_OLDEST}, + {"id": 10, "post_id": 1, "updated_at": COMMENT_10_UPDATED_AT}, + {"id": 11, "post_id": 1, "updated_at": COMMENT_11_UPDATED_AT}, + ], + "next_page": "https://api.example.com/community/posts/1/comments?per_page=100&page=2", + }, + ), + # Comments for post 1 page 2 + ( + "https://api.example.com/community/posts/1/comments?per_page=100&page=2", + {"comments": [{"id": 12, "post_id": 1, "updated_at": COMMENT_12_UPDATED_AT}]}, + ), + # Votes for comment 10 page 1 + ( + f"https://api.example.com/community/posts/1/comments/10/votes?per_page=100&start_time={START_DATE}", + { + "votes": [{"id": 100, "comment_id": 10, "created_at": VOTE_100_CREATED_AT}], + "next_page": f"https://api.example.com/community/posts/1/comments/10/votes?per_page=100&page=2&start_time={START_DATE}", + }, + ), + # Votes for comment 10 page 2 + ( + f"https://api.example.com/community/posts/1/comments/10/votes?per_page=100&page=2&start_time={START_DATE}", + {"votes": [{"id": 101, "comment_id": 10, "created_at": VOTE_101_CREATED_AT}]}, + ), + # Votes for comment 11 + ( + f"https://api.example.com/community/posts/1/comments/11/votes?per_page=100&start_time={START_DATE}", + {"votes": [{"id": 111, "comment_id": 11, "created_at": VOTE_111_CREATED_AT}]}, + ), + # Votes for comment 12 (empty child partition) + ( + f"https://api.example.com/community/posts/1/comments/12/votes?per_page=100&start_time={START_DATE}", + {"votes": []}, + ), + # Comments for post 2 page 1 + ( + "https://api.example.com/community/posts/2/comments?per_page=100", + { + "comments": [{"id": 20, "post_id": 2, "updated_at": COMMENT_20_UPDATED_AT}], + "next_page": "https://api.example.com/community/posts/2/comments?per_page=100&page=2", + }, + ), + # Comments for post 2 page 2 + ( + "https://api.example.com/community/posts/2/comments?per_page=100&page=2", + {"comments": [{"id": 21, "post_id": 2, "updated_at": COMMENT_21_UPDATED_AT}]}, + ), + # Votes for comment 20 + ( + f"https://api.example.com/community/posts/2/comments/20/votes?per_page=100&start_time={START_DATE}", + {"votes": [{"id": 200, "comment_id": 20, "created_at": VOTE_200_CREATED_AT}]}, + ), + # Votes for comment 21 + ( + f"https://api.example.com/community/posts/2/comments/21/votes?per_page=100&start_time={START_DATE}", + {"votes": [{"id": 210, "comment_id": 21, "created_at": VOTE_210_CREATED_AT}]}, + ), + # Comments for post 3 + ( + "https://api.example.com/community/posts/3/comments?per_page=100", + {"comments": [{"id": 30, "post_id": 3, "updated_at": COMMENT_30_UPDATED_AT}]}, + ), + # Votes for comment 30 + ( + f"https://api.example.com/community/posts/3/comments/30/votes?per_page=100&start_time={START_DATE}", + {"votes": [{"id": 300, "comment_id": 30, "created_at": VOTE_300_CREATED_AT_TIMESTAMP}]}, + ), +] + +GLOBAL_CURSOR_RESUME_EXPECTED_RECORDS = [ + { + "comment_id": 10, + "comment_updated_at": COMMENT_10_UPDATED_AT, + "created_at": VOTE_100_CREATED_AT, + "id": 100, + }, + { + "comment_id": 10, + "comment_updated_at": COMMENT_10_UPDATED_AT, + "created_at": VOTE_101_CREATED_AT, + "id": 101, + }, + { + "comment_id": 11, + "comment_updated_at": COMMENT_11_UPDATED_AT, + "created_at": VOTE_111_CREATED_AT, + "id": 111, + }, + { + "comment_id": 20, + "comment_updated_at": COMMENT_20_UPDATED_AT, + "created_at": VOTE_200_CREATED_AT, + "id": 200, + }, + { + "comment_id": 21, + "comment_updated_at": COMMENT_21_UPDATED_AT, + "created_at": VOTE_210_CREATED_AT, + "id": 210, + }, + { + "comment_id": 30, + "comment_updated_at": COMMENT_30_UPDATED_AT, + "created_at": str(VOTE_300_CREATED_AT_TIMESTAMP), + "id": 300, + }, +] + + +def test_global_cursor_substream_resume_after_interruption_skips_no_records(): + """Interrupt a global-cursor substream mid-walk, resume from every interim checkpoint, and prove no records are skipped. + + Regression test for OC-12977. A ``global_substream_cursor`` substream whose parent has no + ``incremental_dependency`` (empty parent state -- the ``bank_accounts`` shape) now emits its + throttled checkpoint STATE during the walk because the empty-parent guard was removed. This + test proves those interim checkpoints are *safe to resume from*: for each interim STATE the + cursor emits, we simulate the source being killed right after it (keeping only the records + emitted before that STATE), then start a fresh read seeded with that STATE and assert the + union of pre-kill records and resume records covers the full expected set with nothing + skipped. Because the interim global cursor is inert until the sync finishes, resuming + re-reads rather than skips -- which is exactly the no-data-loss guarantee we want. + """ + # Disable the 600s throttle so an interim checkpoint STATE is emitted at every close_partition, + # giving us many mid-walk "kill" points to resume from. + with patch.object( + ConcurrentPerPartitionCursor, "_throttle_state_message", return_value=9999999.0 + ): + with requests_mock.Mocker() as m: + for url, response in GLOBAL_CURSOR_RESUME_MOCK_REQUESTS: + m.get(url, json=response) + + # Fresh initial sync (no incoming state), interrupted implicitly by inspecting + # every interim checkpoint it emits. + output = _run_read( + SUBSTREAM_MANIFEST_WITH_GLOBAL_CURSOR_AND_NO_DEPENDENCY, + CONFIG, + STREAM_NAME, + None, + ) + + # Sanity: the uninterrupted run yields the full expected record set. + assert sorted([r.record.data for r in output.records], key=lambda x: x["id"]) == sorted( + GLOBAL_CURSOR_RESUME_EXPECTED_RECORDS, key=lambda x: x["id"] + ) + + # The fix must produce at least one interim checkpoint STATE during the walk + # (before the final state) -- otherwise there is nothing to resume from. + assert len(output.state_messages) >= 2, ( + "Expected at least one interim checkpoint STATE plus the final STATE; " + f"got {len(output.state_messages)} state message(s)." + ) + + # Map each interim STATE to the records emitted before it (what the destination + # would have committed had the source been killed right after that checkpoint). + cumulative_records = [] + interim_checkpoints = [] + for message in output.records_and_state_messages: + if message.type.value == "RECORD": + cumulative_records.append(message.record.data) + elif message.type.value == "STATE": + interim_checkpoints.append((message.state, cumulative_records.copy())) + + expected_deduped = list( + {orjson.dumps(r): r for r in GLOBAL_CURSOR_RESUME_EXPECTED_RECORDS}.values() + ) + + # Resume from every checkpoint except the final one (the final state is the + # completed sync, not an interruption). + for state, records_before_kill in interim_checkpoints[:-1]: + resume_output = _run_read( + SUBSTREAM_MANIFEST_WITH_GLOBAL_CURSOR_AND_NO_DEPENDENCY, + CONFIG, + STREAM_NAME, + [state], + ) + records_after_resume = [r.record.data for r in resume_output.records] + + combined = records_before_kill + records_after_resume + combined_deduped = list({orjson.dumps(r): r for r in combined}.values()) + + assert sorted(combined_deduped, key=lambda x: x["id"]) == sorted( + expected_deduped, key=lambda x: x["id"] + ), ( + "Records were skipped when resuming from interim checkpoint " + f"{state.state.stream.stream_state.__dict__}. " + f"Expected {expected_deduped}, got {combined_deduped}." + ) + + def run_incremental_parent_state_test( manifest, mock_requests, From d54aa34f4c7e8315b97ee8ba7200da648639312f Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 8 Jul 2026 18:07:35 +0000 Subject: [PATCH 3/4] test(concurrent-cursor): verify empty-parent guard removal keeps state size bounded Add two tests for OC-12977 proving the guard removal does not balloon persisted connection state as the number of empty parents grows (Serhii's orchestrator-overload risk): state stays a single bounded global-cursor value with no per-partition map, byte-for-byte identical vs the pre-fix baseline, and the 600s throttle keeps emission count flat regardless of parent count. Co-Authored-By: bot_apk --- .../test_concurrent_perpartitioncursor.py | 215 ++++++++++++++++++ 1 file changed, 215 insertions(+) diff --git a/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py b/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py index 13065a2cf1..72ed0f2875 100644 --- a/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py +++ b/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py @@ -1,5 +1,7 @@ # Copyright (c) 2024 Airbyte, Inc., all rights reserved. import copy +import re +from contextlib import nullcontext from copy import deepcopy from datetime import datetime, timedelta from typing import Any, List, Mapping, MutableMapping, Optional, Union @@ -1273,6 +1275,219 @@ def test_global_cursor_substream_resume_after_interruption_skips_no_records(): ) +# OC-12977 / Serhii's orchestrator-overload risk: removing the empty-parent guard makes a +# global_substream_cursor substream emit interim checkpoint STATE during a long empty-parent +# walk. The constants + helpers below build that substream (the bank_accounts shape) walking an +# arbitrary number of empty parents (comments whose votes are all empty) so we can measure the +# persisted state size as the empty-parent count grows. +MANY_EMPTY_PARENTS_UPDATED_AT = "2024-06-01T00:00:00Z" # >= START_DATE so no parent is filtered out + + +def _emit_state_message_with_guard(self, throttle: bool = True) -> None: + """Pre-fix ``_emit_state_message`` with the empty-parent guard restored. + + Used to reproduce the pre-fix baseline so we can compare persisted state size before vs. + after the guard removal on the exact same fixture. + """ + if throttle: + current_time = self._throttle_state_message() + if current_time is None: + return + self._last_emission_time = current_time + # The guard that PR #1068 removes: + if self._use_global_cursor and not self._parent_state: + return + + self._connector_state_manager.update_state_for_stream( + self._stream_name, + self._stream_namespace, + self.state, + ) + state_message = self._connector_state_manager.create_state_message( + self._stream_name, self._stream_namespace + ) + self._message_repository.emit_message(state_message) + + +def _run_global_cursor_over_empty_parents( + num_empty_parents, num_non_empty_parents=0, reintroduce_guard=False +): + """Run the global-cursor substream over ``num_empty_parents`` empty parents. + + One post fans out to ``num_non_empty_parents + num_empty_parents`` comments (the parents). + The first ``num_non_empty_parents`` comments each return a single vote (advancing the global + cursor and populating the in-memory per-partition map), and the remaining ``num_empty_parents`` + comments return empty votes. The only thing that could grow with the empty-parent count is the + persisted state, which is exactly what we measure. When ``reintroduce_guard`` is True the + pre-fix empty-parent guard is patched back in to produce the baseline for comparison. + """ + posts_response = {"posts": [{"id": 1, "updated_at": MANY_EMPTY_PARENTS_UPDATED_AT}]} + non_empty_ids = [1000 + i for i in range(num_non_empty_parents)] + empty_ids = [1000 + num_non_empty_parents + i for i in range(num_empty_parents)] + comments_response = { + "comments": [ + {"id": cid, "post_id": 1, "updated_at": MANY_EMPTY_PARENTS_UPDATED_AT} + for cid in non_empty_ids + empty_ids + ] + } + + guard_ctx = ( + patch.object( + ConcurrentPerPartitionCursor, + "_emit_state_message", + _emit_state_message_with_guard, + ) + if reintroduce_guard + else nullcontext() + ) + with guard_ctx, requests_mock.Mocker() as m: + # Every empty parent's votes call (one per comment) answers empty by default. + m.get(re.compile(r"/votes"), json={"votes": []}) + # The non-empty parents each return a single vote (more specific match wins). + for idx, cid in enumerate(non_empty_ids): + m.get( + f"https://api.example.com/community/posts/1/comments/{cid}/votes?per_page=100&start_time={START_DATE}", + json={ + "votes": [ + { + "id": 900000 + idx, + "comment_id": cid, + "created_at": MANY_EMPTY_PARENTS_UPDATED_AT, + } + ] + }, + ) + m.get( + f"https://api.example.com/community/posts?per_page=100&start_time={START_DATE}", + json=posts_response, + ) + m.get( + "https://api.example.com/community/posts/1/comments?per_page=100", + json=comments_response, + ) + return _run_read( + SUBSTREAM_MANIFEST_WITH_GLOBAL_CURSOR_AND_NO_DEPENDENCY, + CONFIG, + STREAM_NAME, + None, + ) + + +def _persisted_state_size(state_message) -> int: + return len(orjson.dumps(state_message.state.stream.stream_state.__dict__)) + + +def test_removing_empty_parent_guard_keeps_state_size_flat_as_empty_parents_grow(): + """OC-12977: prove removing the empty-parent guard does not balloon persisted state. + + Serhii's flagged risk is that emitting interim checkpoint STATE for a ``global_substream_cursor`` + substream with empty parents could grow the persisted connection state per empty parent and + overload the orchestrator. This test walks the same substream over a handful vs. thousands of + empty parents and asserts: + + 1. the persisted state carries a single bounded global-cursor value with **no** per-partition + ``states`` map -- so there is no per-parent growth vector at all; + 2. the persisted state is byte-for-byte identical at low vs. high empty-parent counts; + 3. the persisted state is byte-for-byte identical to the pre-fix (guard-restored) baseline; + 4. the 600s throttle keeps the emitted-state *count* a small constant, independent of the + parent count (we re-emit the same small state at most ~once/600s, not once per parent); and + 5. the in-memory per-partition retention cap is unchanged (it bounds memory, not state). + """ + low, high = 5, 3000 + + low_output = _run_global_cursor_over_empty_parents(low) + high_output = _run_global_cursor_over_empty_parents(high) + + # Pure empty-parent walk: no records emitted for either count. + assert len(low_output.records) == 0 + assert len(high_output.records) == 0 + + low_final = low_output.state_messages[-1].state.stream.stream_state.__dict__ + high_final = high_output.state_messages[-1].state.stream.stream_state.__dict__ + + # (1) Global mode never serializes a per-partition state map -> no per-parent growth vector. + assert "states" not in low_final + assert "states" not in high_final + + # (2) Persisted state is byte-for-byte identical at low vs. high empty-parent counts. + assert orjson.dumps(low_final) == orjson.dumps(high_final) + low_size = len(orjson.dumps(low_final)) + high_size = len(orjson.dumps(high_final)) + assert low_size == high_size, ( + f"Persisted state grew with empty-parent count: {low_size}B at {low} parents " + f"vs {high_size}B at {high} parents." + ) + # Bounded: a single small global-cursor value, not kilobytes-per-parent. + assert high_size < 500, f"Persisted state unexpectedly large: {high_size}B" + # Every emitted state message (interim + final) is the same bounded size. + assert max(_persisted_state_size(m) for m in high_output.state_messages) < 500 + + # (3) Unchanged vs. the pre-fix (guard-restored) baseline on the same fixture. + baseline_output = _run_global_cursor_over_empty_parents(high, reintroduce_guard=True) + baseline_final = baseline_output.state_messages[-1].state.stream.stream_state.__dict__ + assert orjson.dumps(high_final) == orjson.dumps(baseline_final), ( + "Persisted final state differs from the pre-fix baseline; guard removal must not " + f"change state content. after={high_final} baseline={baseline_final}" + ) + + # (4) The 600s throttle keeps the emitted-state count a small constant, not per-parent. + assert len(high_output.state_messages) <= 3, ( + "Expected the 600s throttle to bound state emissions to a small constant, got " + f"{len(high_output.state_messages)} state messages for {high} empty parents." + ) + assert len(high_output.state_messages) == len(low_output.state_messages), ( + "State-emission count scaled with empty-parent count: " + f"{len(low_output.state_messages)} at {low} vs {len(high_output.state_messages)} at {high}." + ) + + # (5) The in-memory per-partition retention cap is unchanged (bounds memory, not state). + assert ConcurrentPerPartitionCursor.DEFAULT_MAX_PARTITIONS_NUMBER == 25_000 + assert ConcurrentPerPartitionCursor.SWITCH_TO_GLOBAL_LIMIT == 10_000 + + +def test_persisted_state_has_no_per_partition_entries_even_when_cursor_advances(): + """The per-partition state map must not leak into persisted state, even with real records. + + A skeptic might argue the flat state size above is only because a fully-empty walk produces + no cursor value. This test mixes a few non-empty parents (which advance the global cursor and + populate the in-memory per-partition map) among many empty parents, then proves the persisted + state still carries **no** ``states`` per-partition list and stays byte-for-byte identical as + the empty-parent count grows from a handful to thousands. This is the core guarantee behind the + orchestrator-overload risk: a ``global_substream_cursor`` stream serializes a single global + cursor value, never one entry per parent. + """ + low_output = _run_global_cursor_over_empty_parents(5, num_non_empty_parents=3) + high_output = _run_global_cursor_over_empty_parents(3000, num_non_empty_parents=3) + + # The non-empty parents produced records, so the global cursor actually advanced. + assert len(low_output.records) == 3 + assert len(high_output.records) == 3 + + low_final = low_output.state_messages[-1].state.stream.stream_state.__dict__ + high_final = high_output.state_messages[-1].state.stream.stream_state.__dict__ + + # A real global cursor value is present now... + assert "state" in high_final + # ...but there is still no per-partition state map, regardless of the empty-parent count. + assert "states" not in low_final + assert "states" not in high_final + + # The persisted state is identical at 5 vs. 3000 empty parents apart from ``lookback_window``, + # which is a small integer derived from the sync's wall-clock duration (not the parent count). + # Everything else -- the single global cursor value and the empty parent_state -- is unchanged. + def _without_lookback(state): + return {k: v for k, v in state.items() if k != "lookback_window"} + + assert _without_lookback(low_final) == _without_lookback(high_final), ( + f"Persisted state (excluding lookback_window) changed with empty-parent count: " + f"{low_final} vs {high_final}" + ) + # And the total serialized size stays flat and bounded regardless of the empty-parent count + # (the only possible variation is a digit or two in the wall-clock-derived lookback_window). + assert abs(len(orjson.dumps(low_final)) - len(orjson.dumps(high_final))) <= 3 + assert len(orjson.dumps(high_final)) < 500 + + def run_incremental_parent_state_test( manifest, mock_requests, From d66b623973419ea248064bbfb8ef85418099fe4f Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 8 Jul 2026 18:43:46 +0000 Subject: [PATCH 4/4] test(concurrent-cursor): harden global-cursor resume test (OC-12977) Adds four hardening checks around the empty-parent guard removal: - no-op test for the non-empty-parent-state (incremental_dependency) shape: records + STATE byte-identical guard-removed vs guard-restored, bounding the shared-CDK change's blast radius to the empty-parent case; - data-last fixture: 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: pin default_concurrency=1 and assert RECORD/STATE interleaving is identical across repeated runs (no concurrent-ordering flake); - assert resume meaningfully contributes the tail (records not already emitted before the checkpoint), so the test can't pass trivially. Co-Authored-By: bot_apk --- .../test_concurrent_perpartitioncursor.py | 268 ++++++++++++++++-- 1 file changed, 238 insertions(+), 30 deletions(-) diff --git a/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py b/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py index 72ed0f2875..353d331f04 100644 --- a/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py +++ b/unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py @@ -325,6 +325,29 @@ "global_substream_cursor" ] = True +# Single-worker (serial) variants used by the resume/no-op tests below. Pinning +# `default_concurrency` to 1 makes the RECORD<->STATE interleaving deterministic so the resume +# tests can reconstruct "records committed before the kill" without becoming a concurrent-ordering +# flake. It is hardcoded rather than driven by `config['num_workers']` because the test spec sets +# `additionalProperties: False` and rejects an extra `num_workers` config field. +SUBSTREAM_MANIFEST_GLOBAL_CURSOR_NO_DEPENDENCY_SERIAL = deepcopy( + SUBSTREAM_MANIFEST_WITH_GLOBAL_CURSOR_AND_NO_DEPENDENCY +) +SUBSTREAM_MANIFEST_GLOBAL_CURSOR_NO_DEPENDENCY_SERIAL["concurrency_level"][ + "default_concurrency" +] = 1 + +# Same global cursor, but the parent keeps `incremental_dependency: true` (the +# `customer_balance_transactions` shape) so `_parent_state` is non-empty during the walk and the +# empty-parent guard never fired -- used to prove the guard removal is a no-op there. +SUBSTREAM_MANIFEST_GLOBAL_CURSOR_WITH_DEPENDENCY_SERIAL = deepcopy(SUBSTREAM_MANIFEST) +SUBSTREAM_MANIFEST_GLOBAL_CURSOR_WITH_DEPENDENCY_SERIAL["definitions"]["cursor_incremental_sync"][ + "global_substream_cursor" +] = True +SUBSTREAM_MANIFEST_GLOBAL_CURSOR_WITH_DEPENDENCY_SERIAL["concurrency_level"][ + "default_concurrency" +] = 1 + import orjson import requests_mock @@ -1195,40 +1218,131 @@ def test_incremental_parent_state_no_incremental_dependency( ] -def test_global_cursor_substream_resume_after_interruption_skips_no_records(): - """Interrupt a global-cursor substream mid-walk, resume from every interim checkpoint, and prove no records are skipped. +# Data-last variant of the resume fixture that mirrors the real OC-12977 failure more faithfully: +# a run of empty parents (comments with no votes) *precedes* the single data-bearing parent +# (comment 105). Interim checkpoint STATE is therefore emitted repeatedly during the empty stretch, +# before any record exists -- exactly the checkpoints we most need to prove are safe to resume from. +GLOBAL_CURSOR_DATA_LAST_MOCK_REQUESTS = [ + ( + f"https://api.example.com/community/posts?per_page=100&start_time={START_DATE}", + {"posts": [{"id": 1, "updated_at": POST_1_UPDATED_AT}]}, + ), + ( + "https://api.example.com/community/posts/1/comments?per_page=100", + { + "comments": [ + {"id": 101, "post_id": 1, "updated_at": "2024-01-20T00:00:00Z"}, + {"id": 102, "post_id": 1, "updated_at": "2024-01-19T00:00:00Z"}, + {"id": 103, "post_id": 1, "updated_at": "2024-01-18T00:00:00Z"}, + {"id": 104, "post_id": 1, "updated_at": "2024-01-17T00:00:00Z"}, + {"id": 105, "post_id": 1, "updated_at": "2024-01-25T00:00:00Z"}, + ] + }, + ), + # The first four parents are empty (the long silent stretch). + ( + f"https://api.example.com/community/posts/1/comments/101/votes?per_page=100&start_time={START_DATE}", + {"votes": []}, + ), + ( + f"https://api.example.com/community/posts/1/comments/102/votes?per_page=100&start_time={START_DATE}", + {"votes": []}, + ), + ( + f"https://api.example.com/community/posts/1/comments/103/votes?per_page=100&start_time={START_DATE}", + {"votes": []}, + ), + ( + f"https://api.example.com/community/posts/1/comments/104/votes?per_page=100&start_time={START_DATE}", + {"votes": []}, + ), + # The data-bearing parent comes LAST. + ( + f"https://api.example.com/community/posts/1/comments/105/votes?per_page=100&start_time={START_DATE}", + { + "votes": [ + {"id": 500, "comment_id": 105, "created_at": "2024-01-15T00:00:00Z"}, + {"id": 501, "comment_id": 105, "created_at": "2024-01-16T00:00:00Z"}, + ] + }, + ), +] - Regression test for OC-12977. A ``global_substream_cursor`` substream whose parent has no - ``incremental_dependency`` (empty parent state -- the ``bank_accounts`` shape) now emits its - throttled checkpoint STATE during the walk because the empty-parent guard was removed. This - test proves those interim checkpoints are *safe to resume from*: for each interim STATE the - cursor emits, we simulate the source being killed right after it (keeping only the records - emitted before that STATE), then start a fresh read seeded with that STATE and assert the - union of pre-kill records and resume records covers the full expected set with nothing - skipped. Because the interim global cursor is inert until the sync finishes, resuming - re-reads rather than skips -- which is exactly the no-data-loss guarantee we want. +GLOBAL_CURSOR_DATA_LAST_EXPECTED_RECORDS = [ + { + "comment_id": 105, + "comment_updated_at": "2024-01-25T00:00:00Z", + "created_at": "2024-01-15T00:00:00Z", + "id": 500, + }, + { + "comment_id": 105, + "comment_updated_at": "2024-01-25T00:00:00Z", + "created_at": "2024-01-16T00:00:00Z", + "id": 501, + }, +] + + +def _record_state_sequence(output: EntrypointOutput) -> List[tuple]: + """Return the ordered `(kind, record-id)` sequence of RECORD/STATE messages. + + Used to assert the RECORD<->STATE interleaving is deterministic across repeated runs so the + resume reconstruction below cannot become a concurrent-ordering flake. + """ + sequence: List[tuple] = [] + for message in output.records_and_state_messages: + if message.type.value == "RECORD": + sequence.append(("RECORD", message.record.data["id"])) + elif message.type.value == "STATE": + sequence.append(("STATE", None)) + return sequence + + +def _assert_resume_from_every_interim_checkpoint_skips_no_records( + manifest: Mapping[str, Any], + mock_requests: List[tuple], + expected_records: List[Mapping[str, Any]], +) -> None: + """Prove every interim checkpoint of a global-cursor substream is safe to resume from. + + Runs `manifest` to completion with the 600s throttle disabled (so an interim checkpoint STATE + is emitted at every `close_partition`), then for each interim STATE simulates a kill right after + it -- keeping only the records emitted before that STATE -- and a fresh resume seeded with it. + Asserts: + + 1. the RECORD<->STATE interleaving is deterministic across repeated runs (the caller pins + `default_concurrency` to 1, so this cannot become a concurrent-ordering flake); + 2. the union of pre-kill and resumed records equals the full expected set with nothing skipped; + and + 3. resume meaningfully contributes the tail -- for at least one checkpoint that precedes some + data, the resumed read returns records not already emitted before the kill, so the test + cannot pass trivially if all records preceded the first checkpoint. """ - # Disable the 600s throttle so an interim checkpoint STATE is emitted at every close_partition, - # giving us many mid-walk "kill" points to resume from. with patch.object( ConcurrentPerPartitionCursor, "_throttle_state_message", return_value=9999999.0 ): with requests_mock.Mocker() as m: - for url, response in GLOBAL_CURSOR_RESUME_MOCK_REQUESTS: + for url, response in mock_requests: m.get(url, json=response) # Fresh initial sync (no incoming state), interrupted implicitly by inspecting # every interim checkpoint it emits. - output = _run_read( - SUBSTREAM_MANIFEST_WITH_GLOBAL_CURSOR_AND_NO_DEPENDENCY, - CONFIG, - STREAM_NAME, - None, + output = _run_read(manifest, CONFIG, STREAM_NAME, None) + + # (1) Determinism: the same fixture must produce byte-identical record/state + # interleaving on every run, otherwise reconstructing "records committed before the + # kill" would be flaky under the concurrent engine. + rerun = _run_read(manifest, CONFIG, STREAM_NAME, None) + assert _record_state_sequence(output) == _record_state_sequence(rerun), ( + "RECORD/STATE interleaving is not deterministic across runs; the resume test would " + "be a concurrent-ordering flake." ) + assert [r.record.data for r in output.records] == [r.record.data for r in rerun.records] # Sanity: the uninterrupted run yields the full expected record set. assert sorted([r.record.data for r in output.records], key=lambda x: x["id"]) == sorted( - GLOBAL_CURSOR_RESUME_EXPECTED_RECORDS, key=lambda x: x["id"] + expected_records, key=lambda x: x["id"] ) # The fix must produce at least one interim checkpoint STATE during the walk @@ -1240,7 +1354,7 @@ def test_global_cursor_substream_resume_after_interruption_skips_no_records(): # Map each interim STATE to the records emitted before it (what the destination # would have committed had the source been killed right after that checkpoint). - cumulative_records = [] + cumulative_records: List[Mapping[str, Any]] = [] interim_checkpoints = [] for message in output.records_and_state_messages: if message.type.value == "RECORD": @@ -1248,19 +1362,13 @@ def test_global_cursor_substream_resume_after_interruption_skips_no_records(): elif message.type.value == "STATE": interim_checkpoints.append((message.state, cumulative_records.copy())) - expected_deduped = list( - {orjson.dumps(r): r for r in GLOBAL_CURSOR_RESUME_EXPECTED_RECORDS}.values() - ) + expected_deduped = list({orjson.dumps(r): r for r in expected_records}.values()) # Resume from every checkpoint except the final one (the final state is the # completed sync, not an interruption). + resume_contributed_new_tail = False for state, records_before_kill in interim_checkpoints[:-1]: - resume_output = _run_read( - SUBSTREAM_MANIFEST_WITH_GLOBAL_CURSOR_AND_NO_DEPENDENCY, - CONFIG, - STREAM_NAME, - [state], - ) + resume_output = _run_read(manifest, CONFIG, STREAM_NAME, [state]) records_after_resume = [r.record.data for r in resume_output.records] combined = records_before_kill + records_after_resume @@ -1274,6 +1382,106 @@ def test_global_cursor_substream_resume_after_interruption_skips_no_records(): f"Expected {expected_deduped}, got {combined_deduped}." ) + # (3) The resume must actually supply records that were not already committed + # before the kill, for at least one checkpoint that precedes some data. + before_keys = {orjson.dumps(r) for r in records_before_kill} + new_from_resume = [ + r for r in records_after_resume if orjson.dumps(r) not in before_keys + ] + if ( + len(records_before_kill) < len(expected_deduped) + and records_after_resume + and new_from_resume + ): + resume_contributed_new_tail = True + + assert resume_contributed_new_tail, ( + "No interim checkpoint's resume contributed records beyond those already emitted " + "before the kill; the test could pass trivially if all records preceded the first " + "checkpoint." + ) + + +def test_global_cursor_substream_resume_after_interruption_skips_no_records(): + """Interrupt a global-cursor substream mid-walk, resume from every interim checkpoint, and prove no records are skipped. + + Regression test for OC-12977. A ``global_substream_cursor`` substream whose parent has no + ``incremental_dependency`` (empty parent state -- the ``bank_accounts`` shape) now emits its + throttled checkpoint STATE during the walk because the empty-parent guard was removed. This + test proves those interim checkpoints are *safe to resume from*: for each interim STATE the + cursor emits, we simulate the source being killed right after it (keeping only the records + emitted before that STATE), then start a fresh read seeded with that STATE and assert the + union of pre-kill records and resume records covers the full expected set with nothing + skipped. Because the interim global cursor is inert until the sync finishes, resuming + re-reads rather than skips -- which is exactly the no-data-loss guarantee we want. + """ + _assert_resume_from_every_interim_checkpoint_skips_no_records( + SUBSTREAM_MANIFEST_GLOBAL_CURSOR_NO_DEPENDENCY_SERIAL, + GLOBAL_CURSOR_RESUME_MOCK_REQUESTS, + GLOBAL_CURSOR_RESUME_EXPECTED_RECORDS, + ) + + +def test_global_cursor_substream_resume_with_data_bearing_partition_last_skips_no_records(): + """Resume is safe even when the data-bearing partition comes after a long empty-parent stretch. + + This mirrors the customer failure more faithfully than the front-loaded fixture: the source + emits several interim checkpoint STATE messages while walking empty parents *before* any record + exists, then the last parent finally yields the data. Resuming from any of those early + (zero-record) checkpoints must still recover the full tail with nothing skipped -- proving the + dangerous case (checkpoints emitted during the silent stretch that precedes real data) is safe. + """ + _assert_resume_from_every_interim_checkpoint_skips_no_records( + SUBSTREAM_MANIFEST_GLOBAL_CURSOR_NO_DEPENDENCY_SERIAL, + GLOBAL_CURSOR_DATA_LAST_MOCK_REQUESTS, + GLOBAL_CURSOR_DATA_LAST_EXPECTED_RECORDS, + ) + + +def test_guard_removal_is_a_noop_for_non_empty_parent_state(): + """The guard removal is a complete no-op for the non-empty-parent-state (incremental_dependency) shape. + + The empty-parent guard only ever fired when ``_use_global_cursor and not _parent_state``. For a + substream whose parent has ``incremental_dependency: true`` (the ``customer_balance_transactions`` + shape), ``_parent_state`` is non-empty during the walk, so the guard was already bypassed. This + test proves removing it changes nothing there: on the same fixture, the emitted records and the + full sequence of STATE messages are byte-for-byte identical with the guard removed (current + code) vs. restored (pre-fix). This bounds the blast radius of the shared-CDK change to the + empty-parent case, which is the main concern for a change in shared CDK code. + """ + with patch.object( + ConcurrentPerPartitionCursor, "_throttle_state_message", return_value=9999999.0 + ): + with requests_mock.Mocker() as m: + for url, response in GLOBAL_CURSOR_RESUME_MOCK_REQUESTS: + m.get(url, json=response) + + after_output = _run_read( + SUBSTREAM_MANIFEST_GLOBAL_CURSOR_WITH_DEPENDENCY_SERIAL, CONFIG, STREAM_NAME, None + ) + with patch.object( + ConcurrentPerPartitionCursor, + "_emit_state_message", + _emit_state_message_with_guard, + ): + before_output = _run_read( + SUBSTREAM_MANIFEST_GLOBAL_CURSOR_WITH_DEPENDENCY_SERIAL, + CONFIG, + STREAM_NAME, + None, + ) + + assert [r.record.data for r in after_output.records] == [ + r.record.data for r in before_output.records + ], "Guard removal changed emitted records for the non-empty-parent-state shape." + + after_states = [m.state.stream.stream_state.__dict__ for m in after_output.state_messages] + before_states = [m.state.stream.stream_state.__dict__ for m in before_output.state_messages] + assert after_states == before_states, ( + "Guard removal changed emitted STATE for the non-empty-parent-state shape; the change must " + "be a no-op when the parent has incremental_dependency (non-empty parent state)." + ) + # OC-12977 / Serhii's orchestrator-overload risk: removing the empty-parent guard makes a # global_substream_cursor substream emit interim checkpoint STATE during a long empty-parent