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 fdb5d4d774..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 @@ -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,23 @@ 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] + # 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.", + cursor_field.cursor_field_key, + saved_cursor_value, + ) + 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/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 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..3e5575513f 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,56 @@ 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-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-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_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` 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) + + sync_start = converter._get_sync_start(CursorField("updatedAt"), {"updatedAt": True}, start) + + assert sync_start == start + + +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) + + 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", [