fix(concurrent cursor): Ensure than when start and state are provided, sequential state value… (#52)

Co-authored-by: octavia-squidington-iii <[email protected]>

Author: maxi297

airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py

@@ -82,7 +82,11 @@ def convert_from_sequential_state(
         # The start and end are the same to avoid confusion as to whether the records for this slice
         # were actually synced
         slices = [
-            {self.START_KEY: start if start is not None else sync_start, self.END_KEY: sync_start}
+            {
+                self.START_KEY: start if start is not None else sync_start,
+                self.END_KEY: sync_start,
+                self.MOST_RECENT_RECORD_KEY: sync_start,
+            }
         ]
 
         return sync_start, {

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -3017,6 +3017,7 @@ def test_create_concurrent_cursor_from_datetime_based_cursor_all_fields(
                 {
                     "start": pendulum.parse(config["start_time"]),
                     "end": pendulum.parse(stream_state["updated_at"]),
+                    "most_recent_cursor_value": pendulum.parse(stream_state["updated_at"]),
                 },
             ],
             "state_type": "date-range",
@@ -3028,6 +3029,7 @@ def test_create_concurrent_cursor_from_datetime_based_cursor_all_fields(
                 {
                     "start": pendulum.parse(config["start_time"]),
                     "end": pendulum.parse(config["start_time"]),
+                    "most_recent_cursor_value": pendulum.parse(config["start_time"]),
                 },
             ],
             "state_type": "date-range",
@@ -3261,6 +3263,7 @@ def test_create_concurrent_cursor_uses_min_max_datetime_format_if_defined():
             {
                 "start": expected_start,
                 "end": expected_start,
+                "most_recent_cursor_value": expected_start,
             },
         ],
         "state_type": "date-range",

unit_tests/sources/streams/concurrent/test_cursor.py

@@ -64,7 +64,9 @@ def setUp(self) -> None:
         self._message_repository = Mock(spec=MessageRepository)
         self._state_manager = Mock(spec=ConnectorStateManager)
 
-    def _cursor_with_slice_boundary_fields(self, is_sequential_state=True) -> ConcurrentCursor:
+    def _cursor_with_slice_boundary_fields(
+        self, is_sequential_state: bool = True
+    ) -> ConcurrentCursor:
         return ConcurrentCursor(
             _A_STREAM_NAME,
             _A_STREAM_NAMESPACE,
@@ -128,7 +130,10 @@ def test_given_state_not_sequential_when_close_partition_then_emit_state(self) -
             _A_STREAM_NAME,
             _A_STREAM_NAMESPACE,
             {
-                "slices": [{"end": 0, "start": 0}, {"end": 30, "start": 12}],
+                "slices": [
+                    {"end": 0, "most_recent_cursor_value": 0, "start": 0},
+                    {"end": 30, "start": 12},
+                ],
                 "state_type": "date-range",
             },
         )
@@ -720,6 +725,39 @@ def test_given_overflowing_slice_gap_when_generate_slices_then_cap_upper_bound_t
             (datetime.fromtimestamp(0, timezone.utc), datetime.fromtimestamp(10, timezone.utc))
         ]
 
+    @freezegun.freeze_time(time_to_freeze=datetime.fromtimestamp(50, timezone.utc))
+    def test_given_initial_state_is_sequential_and_start_provided_when_generate_slices_then_state_emitted_is_initial_state(
+        self,
+    ) -> None:
+        cursor = ConcurrentCursor(
+            _A_STREAM_NAME,
+            _A_STREAM_NAMESPACE,
+            {_A_CURSOR_FIELD_KEY: 10},
+            self._message_repository,
+            self._state_manager,
+            EpochValueConcurrentStreamStateConverter(is_sequential_state=True),
+            CursorField(_A_CURSOR_FIELD_KEY),
+            _SLICE_BOUNDARY_FIELDS,
+            datetime.fromtimestamp(0, timezone.utc),
+            EpochValueConcurrentStreamStateConverter.get_end_provider(),
+            _NO_LOOKBACK_WINDOW,
+        )
+
+        # simulate the case where at least the first slice fails but others succeed
+        cursor.close_partition(
+            _partition(
+                {_LOWER_SLICE_BOUNDARY_FIELD: 40, _UPPER_SLICE_BOUNDARY_FIELD: 50},
+            )
+        )
+
+        self._state_manager.update_state_for_stream.assert_called_once_with(
+            _A_STREAM_NAME,
+            _A_STREAM_NAMESPACE,
+            {
+                _A_CURSOR_FIELD_KEY: 10
+            },  # State message is updated to the legacy format before being emitted
+        )
+
 
 @freezegun.freeze_time(time_to_freeze=datetime(2024, 4, 1, 0, 0, 0, 0, tzinfo=timezone.utc))
 @pytest.mark.parametrize(