[nhobin219]

Closes #162. Closes #160: its two reads were already bounded on main by #161, #164 and #172, and this PR measures it below.

#162: connect(catch_up=True, memory_limit=)

memory_limit= (#161) reached Stream.snapshot, scan, sql and live, but not the snapshot connect(catch_up=True) opens to read a gap.

• That database sized itself at DuckDB's default: 80% of what the process may use, its cgroup's cap under one. Two catching-up consumers in one cgroup were allowed 160% of it between them.
• Passing the keyword to connect was refused from inside create_connection, because connect's **kwargs go to the websocket.

Change:

• connect(..., memory_limit: str | None = None) is carried on Catcher and given to the catch-up's _snapshot.snapshot(cache=ReadCache(memory_limit=...)). That is the setting Stream.snapshot takes, so readers asking for different limits read through different databases.
• Checked at connect. A limit DuckDB cannot read is a ValueError at the call (_published.check_memory_limit, which asks DuckDB on a throwaway in-memory database). Otherwise the DuckDB error would surface inside Catcher._open, whose handling reports any failure to open a table as a credentials problem.
• Without catch_up=True it means nothing and is ignored, as catch_up_retries is.

#160: already bounded on main, measured

#160 named two reads.

• The first recv held up to 1M rows of the gap. fix(catchup): read the gap in pages; memory_limit on snapshot readers #161 and feat(catchup): page a catch-up by a byte budget, not an offset count #164 page the gap: each page is one query over at most page_bytes of Arrow (64 MiB, litelink's row group), the first page is one row, and each later page is sized from the last.
• Connect read every row's offset. fix(published): a table's extent and count from its manifests, not a scan of every file #172 makes Table.open take the extent and row count from the manifests.

Measured on main (26e903d) with #160's reproduction, adapted to record whether the extent came from the manifests or from the full scan. The run used a local file:// table of 80,000 rows of about 0.5 KB in 120 data files, max_replay=100, under systemd-run --user --scope -p MemoryMax=1200M.

subscriber	connect	first recv	extent from
A, a 50-row gap	1.91 s, +28 MiB (first DuckDB load)	0.09 s, +2 MiB	manifests
B, the whole table (79,899 rows)	0.21 s, +4 MiB	0.01 s, +1 MiB	manifests

On 0.21.1, #160 measured case B's first recv at 4.80 s and +452 MiB, and a connect that scanned all 200,000 offsets.

The tests that hold this are already on main:

• tests/test_catchup_pages.py, for example test_connect, test_narrow_rows_take_a_probe_then_one_page and test_wide_rows_stay_within_it;
• tests/test_published.py::test_the_extent_comes_from_the_manifests_and_agrees_with_a_scan.

Test that fails without the fix: tests/test_catchup_memory_limit.py

A local stream of 2,000 published rows, served with max_replay=10, so a consumer at offset 1 catches up from the published table. A fixture records the current_setting('memory_limit') of every database the read goes through. The tests take about 4 s.

test	what it shows
test_the_catch_up_reads_through_a_database_with_the_limit	memory_limit="77MB": every database the catch-up read through reports DuckDB's rendering of 77MB
test_without_it_the_catch_up_leaves_duckdbs_default	the control: without the keyword the limit is not applied
test_a_limit_duckdb_cannot_read_is_refused_at_the_call	memory_limit="plenty" is a ValueError at connect

On unfixed main (26e903d):

FAILED test_the_catch_up_reads_through_a_database_with_the_limit
  TypeError: BaseEventLoop.create_connection() got an unexpected keyword argument 'memory_limit'
FAILED test_a_limit_duckdb_cannot_read_is_refused_at_the_call
  Failed: DID NOT RAISE ValueError
PASSED test_without_it_the_catch_up_leaves_duckdbs_default   (the control)

Falsified. Each break below made tests fail; each was restored and passes:

what I broke	test that fails
Catcher opens with ReadCache()	the limit test
connect passes memory_limit=None	the limit test
no validation at connect	the refusal test

Full suite, locally: STREAMCAST_REQUIRE_S3=1 STREAMCAST_REQUIRE_INTEROP=1 STREAMCAST_REQUIRE_POSTGRES=1 pytest gave 958 passed, 1 failed, 2 skipped.

• The failure is test_maintain.py::TestFlushing::test_a_served_row_reaches_the_published_table. It fails the same way on unmodified main on this 2-vCPU box, passes in CI, and does not touch the catch-up.
• Node is not installed here, so test_node.py is left to CI.

No critical review was run: this is keyword plumbing on the reader side, into a setting #161 already reviewed. It touches no durable state, no frames and no ordering.

Downstream

New keyword, not breaking. Under the repository's release rule a new keyword is a minor release, 0.24.0; the user decides. Consumers' <0.24 pins move to use it.

🤖 Generated with Claude Code