From 55b432d288f1d20ad7cbf8d793ce5117a9756be5 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 10 Jul 2026 11:47:13 +0000 Subject: [PATCH 1/3] fix: do not crash on unparseable cursor state value during concurrent cursor construction Co-Authored-By: bot_apk --- .../datetime_stream_state_converter.py | 20 ++++++--- .../test_datetime_state_converter.py | 43 +++++++++++++++++++ 2 files changed, 58 insertions(+), 5 deletions(-) diff --git a/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py b/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py index fdb5d4d774..fd91e43316 100644 --- a/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py +++ b/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py @@ -2,6 +2,7 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # +import logging from abc import abstractmethod from datetime import datetime, timedelta, timezone from typing import Any, Callable, List, MutableMapping, Optional, Tuple @@ -16,6 +17,8 @@ ) from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse +logger = logging.getLogger("airbyte") + class DateTimeStreamStateConverter(AbstractStreamStateConverter): def _from_state_message(self, value: Any) -> Any: @@ -100,11 +103,18 @@ def _get_sync_start( start: Optional[datetime], ) -> datetime: sync_start = start if start is not None else self.zero_value - prev_sync_low_water_mark = ( - self.parse_timestamp(stream_state[cursor_field.cursor_field_key]) - if cursor_field.cursor_field_key in stream_state - else None - ) + prev_sync_low_water_mark = None + if cursor_field.cursor_field_key in stream_state: + saved_cursor_value = stream_state[cursor_field.cursor_field_key] + try: + prev_sync_low_water_mark = self.parse_timestamp(saved_cursor_value) + except ValueError as exception: + logger.warning( + "Ignoring saved state for cursor field '%s': value %r could not be parsed as a datetime (%s). Falling back to the start date.", + cursor_field.cursor_field_key, + saved_cursor_value, + exception, + ) if prev_sync_low_water_mark and prev_sync_low_water_mark >= sync_start: return prev_sync_low_water_mark else: diff --git a/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py b/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py index ebcb3c20c4..6e12a3362d 100644 --- a/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py +++ b/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py @@ -165,12 +165,55 @@ def test_concurrent_stream_state_converter_is_state_message_compatible( datetime(2023, 8, 22, 5, 3, 27, tzinfo=timezone.utc), id="isomillis-converter-state-after-start-start-is-from-state", ), + pytest.param( + CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%S.%fZ"), + datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc), + {"created_at": True}, + datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc), + id="customformat-converter-unparseable-boolean-state-falls-back-to-start", + ), + pytest.param( + CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%S.%fZ"), + None, + {"created_at": True}, + CustomFormatConcurrentStreamStateConverter( + datetime_format="%Y-%m-%dT%H:%M:%S.%fZ" + ).zero_value, + id="customformat-converter-unparseable-boolean-state-no-start-falls-back-to-zero-value", + ), ], ) def test_get_sync_start(converter, start, state, expected_start): assert converter._get_sync_start(CursorField("created_at"), state, start) == expected_start +def test_get_sync_start_with_unparseable_state_value_does_not_raise(): + """Regression test for a saved cursor state value that is not a valid datetime. + + A malformed connection state (e.g. a boolean `True` reaching the datetime cursor + parser during cursor construction) must not crash the whole source. The converter + should ignore the bad value and fall back to the configured start date. + """ + converter = CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%S.%fZ") + start = datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc) + + sync_start = converter._get_sync_start(CursorField("updatedAt"), {"updatedAt": True}, start) + + assert sync_start == start + + +def test_convert_from_sequential_state_with_unparseable_state_value_does_not_raise(): + """Regression test: `convert_from_sequential_state` must not raise on a bad cursor value.""" + converter = CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%S.%fZ") + start = datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc) + + sync_start, _ = converter.convert_from_sequential_state( + CursorField("updatedAt"), {"updatedAt": True}, start + ) + + assert sync_start == start + + @pytest.mark.parametrize( "converter, start, sequential_state, expected_output_state", [ From a049329c9408b83d5d28bfbf4b29d5bd9da336c6 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 10 Jul 2026 12:12:23 +0000 Subject: [PATCH 2/3] Narrow guard to boolean cursor values to preserve retention-period fail-fast Co-Authored-By: bot_apk --- .../datetime_stream_state_converter.py | 14 +++++++++----- .../test_datetime_state_converter.py | 19 ++++++++++--------- 2 files changed, 19 insertions(+), 14 deletions(-) diff --git a/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py b/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py index fd91e43316..de827f91e5 100644 --- a/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py +++ b/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py @@ -106,15 +106,19 @@ def _get_sync_start( prev_sync_low_water_mark = None if cursor_field.cursor_field_key in stream_state: saved_cursor_value = stream_state[cursor_field.cursor_field_key] - try: - prev_sync_low_water_mark = self.parse_timestamp(saved_cursor_value) - except ValueError as exception: + # Internal state bookkeeping markers (e.g. `use_global_cursor`, + # `__ab_no_cursor_state_message`) are booleans and can end up under the + # cursor field key when a substream parent cursor is built from a malformed + # state. A boolean is never a valid datetime cursor value, so ignore it and + # fall back to the start date instead of crashing the whole source at startup. + if isinstance(saved_cursor_value, bool): logger.warning( - "Ignoring saved state for cursor field '%s': value %r could not be parsed as a datetime (%s). Falling back to the start date.", + "Ignoring saved state for cursor field '%s': value %r is not a valid datetime. Falling back to the start date.", cursor_field.cursor_field_key, saved_cursor_value, - exception, ) + else: + prev_sync_low_water_mark = self.parse_timestamp(saved_cursor_value) if prev_sync_low_water_mark and prev_sync_low_water_mark >= sync_start: return prev_sync_low_water_mark else: diff --git a/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py b/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py index 6e12a3362d..3e5575513f 100644 --- a/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py +++ b/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py @@ -170,7 +170,7 @@ def test_concurrent_stream_state_converter_is_state_message_compatible( datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc), {"created_at": True}, datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc), - id="customformat-converter-unparseable-boolean-state-falls-back-to-start", + id="customformat-converter-boolean-state-falls-back-to-start", ), pytest.param( CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%S.%fZ"), @@ -179,7 +179,7 @@ def test_concurrent_stream_state_converter_is_state_message_compatible( CustomFormatConcurrentStreamStateConverter( datetime_format="%Y-%m-%dT%H:%M:%S.%fZ" ).zero_value, - id="customformat-converter-unparseable-boolean-state-no-start-falls-back-to-zero-value", + id="customformat-converter-boolean-state-no-start-falls-back-to-zero-value", ), ], ) @@ -187,12 +187,13 @@ def test_get_sync_start(converter, start, state, expected_start): assert converter._get_sync_start(CursorField("created_at"), state, start) == expected_start -def test_get_sync_start_with_unparseable_state_value_does_not_raise(): - """Regression test for a saved cursor state value that is not a valid datetime. +def test_get_sync_start_with_boolean_state_value_does_not_raise(): + """Regression test for a boolean state bookkeeping marker reaching the cursor parser. - A malformed connection state (e.g. a boolean `True` reaching the datetime cursor - parser during cursor construction) must not crash the whole source. The converter - should ignore the bad value and fall back to the configured start date. + A malformed connection state (e.g. a boolean `True` from internal markers such as + `use_global_cursor` reaching the datetime cursor parser during cursor construction) + must not crash the whole source. The converter should ignore the bad value and fall + back to the configured start date. """ converter = CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%S.%fZ") start = datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc) @@ -202,8 +203,8 @@ def test_get_sync_start_with_unparseable_state_value_does_not_raise(): assert sync_start == start -def test_convert_from_sequential_state_with_unparseable_state_value_does_not_raise(): - """Regression test: `convert_from_sequential_state` must not raise on a bad cursor value.""" +def test_convert_from_sequential_state_with_boolean_state_value_does_not_raise(): + """Regression test: `convert_from_sequential_state` must not raise on a boolean cursor value.""" converter = CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%S.%fZ") start = datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc) From e6ff882869cf63ee2c35f2d4578cfa5dbaa1bd9b Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 20:57:19 +0000 Subject: [PATCH 3/3] fix: do not re-key __ab_no_cursor_state_message sentinel as parent cursor value Co-Authored-By: bot_apk --- .../parsers/model_to_component_factory.py | 9 +++ .../datetime_stream_state_converter.py | 11 ++-- .../test_model_to_component_factory.py | 63 +++++++++++++++++++ 3 files changed, 78 insertions(+), 5 deletions(-) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 5a75912b36..3af601f417 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -666,6 +666,7 @@ from airbyte_cdk.sources.streams.concurrent.state_converters.incrementing_count_stream_state_converter import ( IncrementingCountStreamStateConverter, ) +from airbyte_cdk.sources.streams.core import NO_CURSOR_STATE_KEY from airbyte_cdk.sources.streams.http.error_handlers.response_models import ResponseAction from airbyte_cdk.sources.types import Config from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer @@ -4164,6 +4165,14 @@ def _instantiate_parent_stream_state_manager( ) if not extracted_parent_state and not isinstance(extracted_parent_state, dict): + # The no-cursor sentinel (`{NO_CURSOR_STATE_KEY: True}`) is written by + # streams that completed without a cursor value (e.g. HubSpot's + # `associations_*` streams). It is bookkeeping, not a cursor value, so it + # must not be re-keyed as `{cursor_field: True}` — doing so feeds a boolean + # into the datetime cursor parser and crashes the whole source at startup + # (see airbytehq/oncall#13084). Start the parent cleanly instead. + if set(child_state.keys()) == {NO_CURSOR_STATE_KEY}: + return ConnectorStateManager([]) cursor_values = child_state.values() if cursor_values and len(cursor_values) == 1: incremental_sync_model: Union[ diff --git a/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py b/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py index de827f91e5..050a95b8ea 100644 --- a/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py +++ b/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py @@ -106,11 +106,12 @@ def _get_sync_start( prev_sync_low_water_mark = None if cursor_field.cursor_field_key in stream_state: saved_cursor_value = stream_state[cursor_field.cursor_field_key] - # Internal state bookkeeping markers (e.g. `use_global_cursor`, - # `__ab_no_cursor_state_message`) are booleans and can end up under the - # cursor field key when a substream parent cursor is built from a malformed - # state. A boolean is never a valid datetime cursor value, so ignore it and - # fall back to the start date instead of crashing the whole source at startup. + # Defense-in-depth guard. The primary/root fix for the boolean leaking into + # here lives in `ModelToComponentFactory._instantiate_parent_stream_state_manager`, + # which no longer re-keys the `__ab_no_cursor_state_message` sentinel as a cursor + # value. This guard is kept as a backstop: a boolean is never a valid datetime + # cursor value, so ignore it and fall back to the start date instead of crashing + # the whole source at startup (see airbytehq/oncall#13084). if isinstance(saved_cursor_value, bool): logger.warning( "Ignoring saved state for cursor field '%s': value %r is not a valid datetime. Falling back to the start date.", diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py index 5149f46d2b..111a2cf57c 100644 --- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py +++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py @@ -82,6 +82,7 @@ from airbyte_cdk.sources.declarative.models import JwtAuthenticator as JwtAuthenticatorModel from airbyte_cdk.sources.declarative.models import ListPartitionRouter as ListPartitionRouterModel from airbyte_cdk.sources.declarative.models import OAuthAuthenticator as OAuthAuthenticatorModel +from airbyte_cdk.sources.declarative.models import ParentStreamConfig as ParentStreamConfigModel from airbyte_cdk.sources.declarative.models import PropertyChunking as PropertyChunkingModel from airbyte_cdk.sources.declarative.models import RecordSelector as RecordSelectorModel from airbyte_cdk.sources.declarative.models import SimpleRetriever as SimpleRetrieverModel @@ -3923,6 +3924,68 @@ def test_create_concurrent_cursor_from_perpartition_cursor_runs_state_migrations ) +def test_instantiate_parent_stream_state_manager_ignores_no_cursor_sentinel(): + """Regression test for airbytehq/oncall#13084. + + A child stream that completed without a cursor value saves the sentinel state + `{"__ab_no_cursor_state_message": True}`. When the parent has + `incremental_dependency: true`, the factory used to re-key that boolean sentinel as + `{cursor_field: True}`, which fed a boolean into the datetime cursor parser and crashed + the whole source at startup with `ValueError: No format ... matching True`. The parent + state must instead be empty so the parent starts cleanly from its configured start date. + """ + parent_stream_config_model = ParentStreamConfigModel.parse_obj( + { + "type": "ParentStreamConfig", + "parent_key": "id", + "partition_field": "id", + "incremental_dependency": True, + "stream": { + "type": "DeclarativeStream", + "name": "parent_stream", + "primary_key": "id", + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": { + "$schema": "http://json-schema.org/draft-07/schema", + "type": "object", + "properties": {"id": {"type": "string"}}, + }, + }, + "incremental_sync": { + "type": "DatetimeBasedCursor", + "cursor_field": "updated_at", + "datetime_format": "%Y-%m-%dT%H:%M:%S.%f%z", + "start_datetime": "{{ config['start_time'] }}", + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url_base": "https://api.test.com/v3/parent", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + }, + }, + } + ) + + factory = ModelToComponentFactory() + + # Previously raised `ValueError: No format ... matching True`. + state_manager = factory._instantiate_parent_stream_state_manager( + child_state={"__ab_no_cursor_state_message": True}, + config=input_config, + model=parent_stream_config_model, + ) + + assert state_manager.get_stream_state("parent_stream", None) == {} + + def test_incrementing_count_cursor_with_partition_router_raises_error(): content = """ type: DeclarativeStream