fix(polling_source): replace _accumulated_stream with optimistic-lock batch list (ITL-617)

[kurodo3]

Summary

• Replaces _accumulated_stream: ArrowTableStream | None with an append-only _batches: list[ArrowTableStream] and _state_lock: threading.Lock
• All callers (sync iter_data/as_table and async async_iter_data) now use the optimistic lock protocol: snapshot cursor (brief lock) → I/O with no lock held → commit only if cursor unchanged (brief lock)
• Async loop tracks its yield position with a per-iterator local_batch_idx and drains _batches at the top of each iteration, so rows committed by a concurrent sync caller are never lost regardless of who wins the commit race
• The lock is never held across await

Closes ITL-617

Root Cause

_get_latest_stream(), called by iter_data() or as_table() while async_iter_data() was running, could advance _cursor and merge new rows into _accumulated_stream without the async loop ever emitting them — permanent silent data loss.

ITL-615 (PR #255) partially patched output_schema() and keys() by bypassing _get_latest_stream() when the cached stream was available. ITL-617 fixes the root by making all mutation of _cursor and _batches race-free.

Test Plan

• New test class TestPollingSourceSyncAccessDuringAsyncRun with two regression tests:
  • test_iter_data_and_as_table_concurrent_with_async_run_lose_no_rows: hammers iter_data() and as_table() via asyncio.to_thread while the async loop runs; asserts all 3 rows are delivered to the async iterator
  • test_output_schema_and_keys_concurrent_with_async_run_lose_no_rows: hammers output_schema() and keys() concurrently; asserts all 3 rows delivered AND exactly 3 fetches (schema calls never trigger a fetch)
• All 63 test_channels/test_polling_source.py tests pass
• Full suite: 4697 passed, 93 skipped, 2 xfailed — zero regressions

🤖 Generated with Claude Code

[codecov]

Codecov Report

❌ Patch coverage is 97.50000% with 2 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/orcapod/core/sources/polling_source.py	97.50%	2 Missing ⚠️

📢 Thoughts on this report? Let us know!

[brian-arnold]

The last committed batch is never yielded when the loop terminates.

Rows are now emitted only at this drain step, at the top of the next iteration. But three return paths sit after the commit in the same iteration — the duration check (~L906), the overrun threshold (~L894), and max_consecutive_errors (~L870). So the final fetch of any duration-limited run is committed to _batches and then dropped when the generator exits.

Repro:

fake = FakeDynamicSource(batches=[_batch(1, 10)], fetch_delay=0.05)
src = PollingSource(
    fake, tag_columns="id",
    polling_config=PollingConfig(interval=0.01, duration=0.01, max_missed_intervals=1000),
)
rows = [item async for item in src.async_iter_data()]
assert len(rows) == 1   # PR: 0, while len(src._batches) == 1

main yields 1 row here; this branch yields 0.

Why the new tests don't catch it

Triggering the bug needs a terminating return to fire in the same iteration that committed a batch. No test in the suite sets that up:

• The two new concurrency tests (test_iter_data_and_as_table_concurrent_with_async_run_lose_no_rows, test_output_schema_and_keys_concurrent_with_async_run_lose_no_rows) are the only new tests asserting a row count, and they use interval=0.02, duration=1.0 with 3 instant batches. The last batch commits at roughly t=0.06s; the duration check doesn't fire until t=1.0s. That leaves ~47 further iterations, the first of which drains batch 3. Commit and termination land in different iterations, so the gap never opens.
• The two tests that do hit a terminating return never assert on rows. test_duration_limit_terminates_source runs with batches=[] and only asserts fake.close_called. test_overrun_terminates_after_threshold collects into items but asserts only fake.close_called — it discards the row count, which is exactly the value the bug corrupts.

My repro closes the gap deliberately: fetch_delay=0.05 against duration=0.01 means the single fetch outlives the duration budget, so the commit and the duration-check return happen in one iteration and the drain never runs again.

Suggest hoisting the drain into a helper and running it once more before every return (and after the while True body), plus a row-count assertion on the two termination tests.

[kurodo3]

Fixed in commit 90c5b59 (rebased on the pull, so check the branch tip).

The three normal-termination return statements (duration check, overrun threshold, max_consecutive_errors) are now break. A single final drain block sits after the while True loop — before except asyncio.CancelledError — so it runs on every clean exit regardless of which condition fired. CancelledError and the fatal-raise paths (CursorInvalidatedError, InputValidationError) skip the final drain intentionally: a cancelled or fatally-errored consumer will not read further items.

Tests:

• Added test_last_batch_yielded_before_duration_exit — your exact repro (fetch_delay=0.05, duration=0.01), asserting len(rows) == 1.
• Strengthened test_duration_limit_terminates_source with assert len(items) == 0 (no batches in that fixture, so it acts as a negative sanity check).
• Strengthened test_overrun_terminates_after_threshold with assert len(items) > 0 (rows committed before the overrun limit must be yielded).

64/64 pass.

[eywalker]

@brian-arnold thanks for catching this! I have now addressed the regression and also added tests to catch this in the future.

[brian-arnold]

Thank you; Looks correct on my end!

[kurodo3]

Review round summary

Brian caught a real regression: any batch committed in the final iteration of async_iter_data was silently dropped if a normal-termination path (duration, overrun, max_consecutive_errors) fired before the drain step ran again at the top of the next iteration.

Changes made in response to this review:

• polling_source.py — converted the three normal-termination return statements inside while True to break, and added a single final drain block after the while True loop (before except asyncio.CancelledError). CancelledError and fatal-raise paths skip the final drain intentionally.
• test_polling_source.py — added test_last_batch_yielded_before_duration_exit (Brian's exact repro); strengthened test_duration_limit_terminates_source and test_overrun_terminates_after_threshold with row-count assertions.

64/64 tests pass.