feat(low-code): stream paginated JSON items with opt-in spool_to_disk and retry-safe downloads

[ZaneHyattAB]

Overview

👉 TL;DR: Lets a low-code connector read very large JSON pages (hundreds of MB) without loading the whole page into memory, while keeping today's pagination, retry and error-handling behaviour. Connectors opt in per stream; nothing changes for manifests that don't. Specifically: JsonItemsDecoder now works with DefaultPaginator (the parser captures the document remainder for CursorPagination), a new opt-in JsonItemsDecoder.spool_to_disk downloads the body inside HttpClient._send_with_retry before parsing, HttpResponseFilter no longer reads the body of spooled 2xx responses larger than 1 MiB, use_cache + spool_to_disk fails fast at build time, and streamed parsers see transport-decoded bytes (raw.decode_content = True). No breaking changes: every new behaviour is gated on spool_to_disk: true or on JsonItemsDecoder + DefaultPaginator (a combination main rejects at build time) — see Compatibility below.

Requested by Zane Hyatt (on-call) for https://github.com/airbytehq/oncall/issues/13554. Draft: opened for review by the CDK owners before the downstream connector change.

Problem

A manifest-only connector using JsonDecoder on an incremental export endpoint that returns 1000 records per page (~160 MB decoded JSON, ~160 KB per record) was OOM-killed by the 4 GiB memory cgroup on every attempt (kernel: Memory cgroup out of memory: Killed process (python) anon-rss:~4.1 GB). The page is materialised several times on the way to the record queue — raw bytes, the BytesIO/.content copy, a stdlib-json tree built by a body-based error_message_contains filter, then the orjson tree — on top of the 10 000-record queue plateau (~1.6–1.9 GB at that record size). JsonItemsDecoder (streaming ijson) would avoid the spike but today (a) DefaultPaginator rejects it, (b) any body-based response filter drains the socket before the decoder runs, (c) requests_cache (auto-enabled on parent streams) buffers the whole body in memory before the decoder sees it, (d) a connection that breaks mid-body is not retried, and (e) it needs a GzipDecoder wrapper for Content-Encoding: gzip.

Changes

• JsonItemsParser.parse(data, *, on_document_remainder=None): with a callback, one ijson pass yields the items and builds the remainder document (everything outside the items array, items array left empty, nested keys preserved); without one it is main's exact ijson.items(data, prefix) path. CompositeRawDecoder(capture_document_remainder=False) only installs the callback after create_default_paginator calls enable_document_remainder_capture() on the decoder instance it shares with the record selector, so a non-paginated JsonItemsDecoder is unchanged. CompositeRawDecoder stashes the remainder on the requests.Response; PaginationDecoderDecorator yields it instead of {}, so {{ response.get("after_url") }} / {{ response.meta.after_cursor }} work unchanged. GzipParser forwards the kwarg. Factory allow-list (_is_supported_parser_for_pagination) accepts JsonItemsParser directly or Gzip-wrapped; CsvParser/JsonLineParser still rejected.
• HttpClient._send marks stream=True responses (streamed_response.mark_body_streamed; the marker is inert unless something below reads it). HttpResponseFilter skips predicate / error_message_contains / error_message body reads only for a spooled 2xx body > 1 MiB (http_codes still apply); a spooled body ≤ 1 MiB is read and the spool rewound after interpret_response; unspooled streamed responses and all non-2xx handling are byte-identical to main. create_http_requester logs a warning (Builder-visible) when a spooling decoder is combined with body-based filters, explaining the 1 MiB cut-off.
• New JsonItemsDecoder.spool_to_disk (schema + model, default false, forced off in Connector Builder mode): HttpClient._spool_response_body copies the body via iter_content into a tempfile.SpooledTemporaryFile(max_size=8 MiB) inside _send, before interpret_response. The flag travels out-of-band as a private request_kwargs key (_SPOOL_RESPONSE_KWARG, set in send_request, popped in _send before session.send), so _send_with_retry / _send keep main's exact signatures and subclass overrides keep working. ChunkedEncodingError / ConnectionError / ContentDecodingError during the copy set response = None and go through the existing fallback RETRY → DefaultBackoffException path, i.e. the connector's own backoff/max_retries. response.raw becomes SpooledResponseBody (a BufferedReader over a py3.10-safe RawIOBase adapter — fileno() would force rollover); every parser and response.content/.json() read from it. Plumbing: HttpRequester.spool_response (declared after decoder, so positional construction is unchanged), Decoder.spools_response(), CompositeRawDecoder(spool_response=), HttpClient.send_request(spool_response=) (appended keyword).
• create_http_requester raises ValueError for explicit use_cache: true + a decoder with spool_to_disk: true (requests_cache reads the whole body, leaving nothing to spool); _initialize_cache_for_parent_streams skips the auto-enable only when the (resolved) retriever decoder is a JsonItemsDecoder with spool_to_disk: true, directly or wrapped in GzipDecoder (_decoder_spools_to_disk, all four call sites incl. both StateDelegatingStream branches). Other streaming decoders keep main's caching behaviour.
• CompositeRawDecoder stream path sets response.raw.decode_content = True so a bare streamed decoder parses Content-Encoding: gzip bodies; GzipParser still sniffs magic bytes, so double-gzip payloads keep working.
• Debug logs (self._logger.debug(..., extra={...}) convention): "Spooled response body to disk" (url, status, bytes, seconds) and "Captured pagination document remainder" (keys).

Review Spotlight

Reviewers with limited time, please review first:

• http_client.py#L430-L505: spool inside the retry loop; response = None on copy failure is what makes the retry happen; _content_consumed = False is the one private requests touch.
• streamed_response.py: response markers and SpooledResponseBody.
• http_response_filter.py#L27-L34: the 1 MiB body-filter gate.
• composite_raw_decoder.py#L185-L213: single-pass items + remainder.
• concurrent_declarative_source.py _decoder_spools_to_disk / _set_cache_if_not_disabled: parent auto-cache skip for spooled decoders only.
• http_client.py _SPOOL_RESPONSE_KWARG: the out-of-band spool flag that keeps _send/_send_with_retry signatures stable.

Gating: JsonItemsDecoder.spool_to_disk (default false)

Spooling is opt-in per decoder. Without it, JsonItemsDecoder + DefaultPaginator streams straight from the socket (today's semantics for streamed decoders: a mid-body failure is not retried, records already yielded cannot be rolled back — same reason REDUCE_PAGE_SIZE refuses re-requests).

Compatibility

No breaking changes. Every new behaviour is opt-in — it requires either spool_to_disk: true or JsonItemsDecoder + DefaultPaginator (which main rejects at build time with decoder not supported for pagination, so no existing manifest has it). Audited against clean main (d8d8e6f6) with a local harness (manifests run through ConcurrentDeclarativeSource in sync and Builder mode against a stub HTTP server, plus parse benchmarks):

Surface	Change vs main
HttpClient._send / _send_with_retry signatures; subclasses overriding them (e.g. source-amazon-seller-partner's RateLimitConfigErrorHttpClient)	none — verified with the connector's exact override: 200 on both
HttpClient.send_request, CompositeRawDecoder.__init__, HttpRequester fields, Decoder ABC	additive only: appended optional keywords / defaulted method; dataclasses.fields(HttpRequester) = main + ["spool_response"]
Explicit use_cache: true + streaming decoder (Csv/Jsonl/Gzip/JsonItems)	none — builds and caches as on main (cache miss and hit both return records); ValueError only with spool_to_disk: true
Parent-stream auto-cache for Csv/Jsonl/Gzip/JsonItems parents	none — same parent call count as main; skipped only for spool_to_disk: true
Body-based response filters on unspooled streamed 2xx	none — evaluated exactly as on main (incl. the body being consumed)
Body-based filters on spooled 2xx	opt-in only: ≤ 1 MiB evaluated, > 1 MiB skipped (http_codes still apply) + build-time warning
JsonItemsDecoder without paginator	none — identical ijson.items path; parse time and peak memory match main on a 1000 × 300-field page and on a document with a 50 000-element second array (identical record hashes)
JsonItemsDecoder (+ GzipDecoder) + DefaultPaginator	opt-in only: ValueError on main → builds and paginates
PaginationDecoderDecorator for Json/Xml/Gzip(Json)/Jsonl/Csv	none
Connector Builder / test-read	none — streaming and spooling are forced off, spool_to_disk accepted and ignored; new logs are DEBUG (plus the opt-in WARNING above)
Schema / model	additive optional field on JsonItemsDecoder (default false); old manifests validate unchanged, no version bump or migration
Streamed decoders on Content-Encoding: gzip/deflate (raw.decode_content = True)	failures → successes only: every decoder × encoding cell that yields records on main yields identical records here; the 9 cells that raised on main (JsonItems/Csv on gzip/deflate transport, Gzip(*) on deflate) now parse. GzipParser sniffs magic bytes, so double-gzip payloads still work. br not measured (module absent).
Python HttpStream connectors passing stream=True	none — the response marker has no readers outside HttpResponseFilter/PaginationDecoderDecorator

Correction to an earlier claim in this description: requests_cache (1.2.1) does not drain a streamed body — _copy_body reads it and _reset_fp re-arms response.raw, so use_cache + streaming decoder returns records on main (at the cost of holding the whole body in memory). That is why the cache is now only refused for spool_to_disk, where memory is the point.

Benchmarks

Real HttpClient + shared DefaultErrorHandler (with error_message_contains) + decoder + PaginationDecoderDecorator, synthetic 1000 × 160 KB gzip page, fresh subprocess, ru_maxrss:

mode	peak RSS	wall	socket held	requests	records
JsonDecoder (today)	784 MB	1.4 s	0.9 s	1	1000
JsonItemsDecoder, spool_to_disk	169 MB	1.1 s	0.9 s	1	1000
spool_to_disk, consumer at 4 rec/s	169 MB	251.3 s	0.9 s	1	1000
spool_to_disk, connection cut at 60 %	169 MB	1.8 s	1.5 s	2	1000, 0 duplicates, pagination fields recovered

Unspooled JsonItemsDecoder with this connector's body-based error handler now behaves as on main (the filter consumes the stream → IncompleteJSONError), so spool_to_disk is the pairing to use there.

Queue plateau (not changed here): 10 000 parsed records cost ≈ 1.1–1.25 B per JSON byte, i.e. ~1.65–1.9 GB at 160 KB/record.

Risks / remaining tradeoffs

• Temp disk ≈ decoded page size per in-flight spooled request above 8 MiB (anonymous file: freed on close, generator GC, or process death; honours TMPDIR; logged per page at debug).
• A spooled 2xx body > 1 MiB is not evaluated by body-based filters (opt-in; warned at build time). Unspooled streamed decoders + body filters remain broken-as-on-main (filter consumes the stream).
• The pagination remainder materialises everything outside the items array (fine for scalar cursors; a second huge non-item array would still be built).
• response._content_consumed = False (private requests attribute, single site, unit-tested) and a hand-edited declarative_component_schema.py (+5 lines, matching the approach in feat: add REDUCE_PAGE_SIZE response action for dynamic page-size reduction #1149 because the generator's output currently fails mypy).
• +9 MB RSS on large pages from the 8 MiB in-memory spool buffer.

Follow-ups (separate PRs)

• Byte-aware bound for the concurrent record queue (Queue(maxsize=10_000) in concurrent_declarative_source.py is count-based).
• HttpComponentsResolver sets use_cache=True unconditionally; a spool_to_disk decoder there would hit the ValueError.
• spool_to_disk for JsonlDecoder/CsvDecoder/GzipDecoder.
• Downstream (airbyte monorepo, after a CDK release containing this): source-zendesk-support tickets_stream gets 8 manifest lines — decoder: {type: JsonItemsDecoder, items_path: tickets, spool_to_disk: true} and record_selector.extractor: {type: DpathExtractor, field_path: []} — plus the base-image bump; no use_cache line, no components.py, shared requester/paginator/error handler untouched. The connector's full unit suite (231 tests) passes against this branch with that manifest.

Test plan

• Compatibility regression tests: HttpClient subclass overriding _send_with_retry with main's exact signature (mirrors source-amazon-seller-partner) with and without spooling; use_cache: true + JsonlDecoder builds / + spool_to_disk raises; Csv/Jsonl/Gzip/JsonItems parents get auto-cache, spooled and Gzip(spooled) parents (incl. StateDelegatingStream branch) don't; remainder capture off by default and enabled by create_default_paginator; positional HttpRequester construction in main's order; unspooled streamed 2xx evaluates body filters; build-time warning presence/absence.
• New unit tests (~30): remainder flat/nested/Gzip-wrapped; paginator allow-list + CursorPagination end-to-end; filter marker/size semantics (skip on unspooled/large, evaluate on ≤ 1 MiB, http_codes first, non-2xx unchanged); spool: gzip decode, rewind after interpret_response, ProtocolError/DecodeError/ReadTimeoutError each → 2 sends and a good response, all five parsers × {in-memory, rolled-over}; factory guard/propagation/Builder mode; parent auto-cache skip incl. mixed StateDelegatingStream.
• poetry run pytest on unit_tests/sources/declarative/{decoders,requesters,retrievers,parsers}, unit_tests/sources/streams/http, test_concurrent_declarative_source*: 1606 passed, 2 failed, 1 skipped. The 2 failures (test_read_with_concurrent_and_synchronous_streams_with_{concurrent,sequential}_state, sqlite3 database table is locked in requests_cache) fail identically on a clean main (d8d8e6f6) checkout — pre-existing.
• poetry run ruff check ., poetry run ruff format --check ., poetry run mypy --config-file mypy.ini airbyte_cdk: clean.
• Decoder + http suites also run under Python 3.10 (SpooledTemporaryFile.readinto does not exist there): 175 passed.
• source-zendesk-support full unit_tests against this branch with the target manifest: 231 passed.

Link to Devin session: https://app.devin.ai/sessions/9fb9d3e97c71470bbbd2b6f242fb725e
Open in Devin Desktop: https://app.devin.ai/desktop/session/9fb9d3e97c71470bbbd2b6f242fb725e?variant=devin
Requested by: ZaneHyattAB

[devin-ai-integration]

I'll fix CI failures and address comments from users with write access. I'll skip comments containing "(aside)".

• Disable automatic comment, CI, and merge conflict monitoring

[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@devin/1790304599-stream-json-items-spool#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 devin/1790304599-stream-json-items-spool

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 734 tests  +60   4 722 ✅ +60   9m 54s ⏱️ +2s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 2ca9df8. ± Comparison against base commit d8d8e6f.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

4 737 tests  +60   4 725 ✅ +60   14m 33s ⏱️ +39s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 2ca9df8. ± Comparison against base commit d8d8e6f.

♻️ This comment has been updated with latest results.