fix(low-code): close ConcurrentCursor end-window and clamping gaps

[Baz (bazarnov)]

Overview

👉 TL;DR: Date-windowed incremental syncs stop silently skipping data in two cases. A date range that starts and ends on the same day (or other time unit) now reads that unit instead of nothing. Clamped windows (clamping DAY, WEEK or MONTH) with a step of at least one day no longer skip days or get stuck, and with a cursor_granularity they no longer read into a period that is not over yet. Specifically, ConcurrentCursor._split_per_slice_range now emits the final [end, end] window when cursor_granularity is set (skipped when is_compare_strictly is set, which the factory now passes), DayClampingStrategy and the WeekClampingStrategy floor keep already-aligned values, and clamped slicing steps from the clamped lower bound.

Changes

• _split_per_slice_range no longer returns early when lower == upper and a cursor_granularity is set. Once the start date and the clamp are applied, lower == upper on the last window emits the one-unit [upper, upper] window, as the legacy DatetimeBasedCursor does. On main that window only appeared when a lookback moved the lower bound below the end.
• ConcurrentCursor takes is_compare_strictly (default False) and keeps it in copy_without_state. When it is set, no window starts at the end, also when a lookback, the start date or the clamp moved the lower bound there. create_concurrent_cursor_from_datetime_based_cursor passes the manifest value, which also covers per-partition cursors.
• A gap between state slices whose lower bound the start date or the clamp moves onto the next slice's start is skipped, since that slice already reads it. main emitted an inverted or overlapping window there.
• DayClampingStrategy ceiling returns midnight values unchanged instead of adding a day. This guard follows the patch Varun Billuri (@varunbiluri) proposed in DayClampingStrategy ceiling skips a full day for already-midnight values (clamping: DAY loses one day per window) #1110.
• WeekClampingStrategy floor keeps a value on the target weekday on that day (at midnight) instead of moving it a week back.
• _split_per_slice_range clamps the lower bound before stepping and returns when it passes the end. A step shorter than the clamping period no longer collapses the first window (for WEEK and MONTH the step still has to be at least one day, see Risks), and when no complete period is left nothing is requested past the end.
• With a granularity, the single-window branch ends at the end instead of clamping it up into the current period. Without one, it still ends at the next period boundary.
• In the stepping loop, a boundary that clamps onto the end makes that window the last one, so the end unit is no longer dropped.
• Tests in test_cursor.py, test_clamping.py and test_model_to_component_factory.py. Two existing assertions encoded the bugs and are updated: the gap test with a granularity now also expects the [50, 50] end unit (its name now says so), and the DAY "last day of month" test uses a non-midnight input.

Review Spotlight

Reviewers with limited time, please review first:

• cursor.py#L489-L501: early returns. With a granularity, lower == upper is decided after the start date and the clamp: the end unit is read on the last window only, and never under is_compare_strictly
• cursor.py#L502-L513: single-window end
• cursor.py#L537-L540: last-window detection in the stepping loop
• clamping.py#L38-L42: DAY ceiling guard
• clamping.py#L94-L98: WEEK floor guard

Risks

• End unit re-read on later syncs (legacy parity). A stream with a non-zero cursor_granularity, no lookback, and an end that does not move between syncs (a fixed end_date, or a day-precision end such as today_utc() with several syncs a day) re-reads the end unit on every later sync once a record on that unit has moved the state to the end: one request per partition per sync, and that unit's records are emitted again. main makes no request there. The last window always starts at the state value, so every other sync already re-reads the state unit, and the legacy DatetimeBasedCursor re-reads it in this case too. The state value does not change, and append+dedup absorbs the duplicates.
  • Counted by parsing all 535 manifest.yaml files on airbyte master at fc800755: 271 connectors define an inline DatetimeBasedCursor, 73 of them with a non-zero cursor_granularity, and 69 of those set end_datetime (47 with no static lookback, 13 with a static lookback, 13 with a lookback from config; the groups overlap).
  • Likely re-readers: source-amplitude, source-app-store-connect, source-claude, source-coingecko-coins, source-dust, source-google-search-console, source-shiftbase, source-tempo, source-tvmaze-schedule, source-twilio, source-wikipedia-pageviews, source-spotify-ads, and source-tiktok-marketing daily reports only with attribution_window: 0. Streams with any lookback, or a now-based end at second precision, are unaffected in practice.
• Start equals end. Among those 69, a config with no effective lookback whose start and end resolve to the same unit now reads it instead of making no request (for example source-tiktok-marketing daily reports with attribution_window: 0).
• is_compare_strictly. The concurrent cursor ignored it on main, so start == end with a lookback sent a [start, start] request. It now skips every window that would start at the end, as the legacy cursor does. Its monorepo users only see a difference when start and end resolve to the same value, which now makes no request: source-snapchat-marketing (cursor_granularity: PT0S) and source-reprise (no granularity). Python ConcurrentCursor users (source-salesforce, ConcurrentSourceAdapter) do not set it.
• Clamping. 0 of the 535 manifest.yaml files on airbyte master at fc800755 set clamping (git grep -E '^[[:space:]]*clamping:'), and no connector code references the clamping classes, so only Builder and custom connectors that use it see these changes: days skipped at window boundaries are read; MONTH/WEEK streams with a step of at least one day that were stuck on main move again, and their first sync after upgrading can backfill several periods; with a granularity no window reaches past the end (without one, windows still end at the next period boundary, as on main); WEEK syncs on the target weekday include the last complete week; a midnight-aligned DAY state re-reads its own day, as non-clamped cursors do; a gap or a strict end window that the clamp closes makes no request.
  • Unchanged: WEEK or MONTH clamping with a step shorter than one day still produces no slices, as on main. On the boundary day their ceiling returns that day's midnight for a non-midnight value, so the first window collapses.
• Public API. ConcurrentCursor gets one keyword argument with a default. DayClampingStrategy.clamp returns midnight inputs unchanged, and WeekClampingStrategy(is_ceiling=False).clamp keeps a value on the target weekday on that day (at midnight).

Test plan

• python -m pytest unit_tests/sources/streams/concurrent/test_cursor.py unit_tests/sources/streams/concurrent/test_clamping.py unit_tests/sources/declarative/parsers/test_model_to_component_factory.py -q
  • this branch: 407 passed
  • source files from main (git checkout origin/main -- airbyte_cdk/sources/streams/concurrent/cursor.py airbyte_cdk/sources/streams/concurrent/clamping.py airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py) with this branch's tests: 15 failed, 392 passed. 10 fail on slice or clamp assertions, 5 raise TypeError for the new is_compare_strictly kwarg.
• Removing any one of the new conditions makes at least one test fail. Adding a clamp(upper) == upper condition changes no result: the lower-bound clamp already returns when clamping would move the end unit past the end.
• python -m pytest unit_tests/sources/streams unit_tests/sources/declarative/incremental unit_tests/sources/declarative/parsers unit_tests/sources/declarative/retrievers unit_tests/sources/declarative/partition_routers unit_tests/sources/declarative/stream_slicers unit_tests/legacy unit_tests/connector_builder -q: 1773 passed.
• The other unit_tests/sources/declarative folders plus unit_tests/source_declarative_manifest: 1649 passed. test_read_with_concurrent_and_synchronous_streams* was deselected because it can crash locally on macOS (Cached parent stream with more than one partition corrupts the shared sqlite request cache under concurrent reads (no such table: responses, file is not a database, no more rows available) #1150).
• Legacy DatetimeBasedCursor vs the factory-built ConcurrentCursor for start == end, with and without is_compare_strictly and a lookback: the same slices in all 4 cases. main differs in 2.
• Differential run over 360 non-clamped configs (first sync, sequential and multi-slice state, lookback, start == end, no step): 50 configs gain exactly one [end, end] window where main emits nothing, all with a granularity and no lookback; every other slice is identical.
• Sweep of 2016 clamped configs built through the factory (steps of one day or more, or no step): main has a gap in 368 DAY, 184 WEEK and 92 MONTH configs and a window past the end in 232, 168 and 190. This branch has no gap, inverted or misaligned window; the 136 remaining past-the-end cases are configs without a granularity, which keep ending at the next period boundary as on main.
• ruff format, ruff check and mypy --config-file mypy.ini on the changed files: clean.

Fixes #1198
Fixes #1110

[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@bazarnov/fix-concurrent-cursor-end-window-and-clamping#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 bazarnov/fix-concurrent-cursor-end-window-and-clamping

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 998 tests  +16   4 986 ✅ +16   11m 1s ⏱️ + 1m 28s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit b43e16b. ± Comparison against base commit d214b0c.

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

unit_tests.sources.streams.concurrent.test_cursor.ConcurrentCursorStateTest ‑ test_given_difference_between_slices_match_slice_range_and_cursor_granularity_when_generate_slices_then_create_one_slice

unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_concurrent_cursor_from_datetime_based_cursor_given_start_equals_end[reads_the_one_day]
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_concurrent_cursor_from_datetime_based_cursor_given_start_equals_end[skipped_when_compare_strictly]
unit_tests.sources.streams.concurrent.test_clamping.DayClampingStrategyTest ‑ test_given_midnight_when_clamp_then_return_same_date
unit_tests.sources.streams.concurrent.test_clamping.WeekClampingStrategyTest ‑ test_given_not_ceiling_and_same_weekday_when_clamp_then_return_same_date
unit_tests.sources.streams.concurrent.test_cursor ‑ test_given_daily_clamp_moves_lower_onto_upper_when_stream_slices_then_skip_the_window[end_unit_with_compare_strictly]
unit_tests.sources.streams.concurrent.test_cursor ‑ test_given_daily_clamp_moves_lower_onto_upper_when_stream_slices_then_skip_the_window[gap_left_empty_by_the_clamp]
unit_tests.sources.streams.concurrent.test_cursor ‑ test_given_start_equals_end_when_stream_slices_then_read_the_end_unit_if_the_window_includes_it[one_unit_window_with_granularity]
unit_tests.sources.streams.concurrent.test_cursor ‑ test_given_start_equals_end_when_stream_slices_then_read_the_end_unit_if_the_window_includes_it[skipped_when_compare_strictly]
unit_tests.sources.streams.concurrent.test_cursor ‑ test_given_start_equals_end_when_stream_slices_then_read_the_end_unit_if_the_window_includes_it[skipped_without_granularity]
unit_tests.sources.streams.concurrent.test_cursor.ClampingIntegrationTest ‑ test_given_daily_clamp_and_midnight_boundaries_when_stream_slices_then_do_not_skip_days
…

[github-actions]

PyTest Results (Full)

5 001 tests  +16   4 989 ✅ +16   12m 44s ⏱️ - 3m 20s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit b43e16b. ± Comparison against base commit d214b0c.

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

unit_tests.sources.streams.concurrent.test_cursor.ConcurrentCursorStateTest ‑ test_given_difference_between_slices_match_slice_range_and_cursor_granularity_when_generate_slices_then_create_one_slice

unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_concurrent_cursor_from_datetime_based_cursor_given_start_equals_end[reads_the_one_day]
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_concurrent_cursor_from_datetime_based_cursor_given_start_equals_end[skipped_when_compare_strictly]
unit_tests.sources.streams.concurrent.test_clamping.DayClampingStrategyTest ‑ test_given_midnight_when_clamp_then_return_same_date
unit_tests.sources.streams.concurrent.test_clamping.WeekClampingStrategyTest ‑ test_given_not_ceiling_and_same_weekday_when_clamp_then_return_same_date
unit_tests.sources.streams.concurrent.test_cursor ‑ test_given_daily_clamp_moves_lower_onto_upper_when_stream_slices_then_skip_the_window[end_unit_with_compare_strictly]
unit_tests.sources.streams.concurrent.test_cursor ‑ test_given_daily_clamp_moves_lower_onto_upper_when_stream_slices_then_skip_the_window[gap_left_empty_by_the_clamp]
unit_tests.sources.streams.concurrent.test_cursor ‑ test_given_start_equals_end_when_stream_slices_then_read_the_end_unit_if_the_window_includes_it[one_unit_window_with_granularity]
unit_tests.sources.streams.concurrent.test_cursor ‑ test_given_start_equals_end_when_stream_slices_then_read_the_end_unit_if_the_window_includes_it[skipped_when_compare_strictly]
unit_tests.sources.streams.concurrent.test_cursor ‑ test_given_start_equals_end_when_stream_slices_then_read_the_end_unit_if_the_window_includes_it[skipped_without_granularity]
unit_tests.sources.streams.concurrent.test_cursor.ClampingIntegrationTest ‑ test_given_daily_clamp_and_midnight_boundaries_when_stream_slices_then_do_not_skip_days
…