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 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 759 tests  +60   4 747 ✅ +60   10m 14s ⏱️ + 1m 56s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 03df3c3. ± Comparison against base commit a853e8b.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

4 762 tests  +60   4 750 ✅ +60   14m 54s ⏱️ + 3m 2s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 03df3c3. ± Comparison against base commit a853e8b.

♻️ This comment has been updated with latest results.

[ZaneHyattAB]

/prerelease

Prerelease Job Info

This job triggers the publish workflow with default arguments to create a prerelease.

Prerelease job started... Check job output.

✅ Prerelease workflow triggered successfully.

View the publish workflow run: https://github.com/airbytehq/airbyte-python-cdk/actions/runs/36451418554

[ZaneHyattAB]

Parking this as a draft. The on-call issue that prompted it (oncall#13554) was closed on 2026-10-08. The customer is unblocked with a larger memory limit and declined to set up a test connection, so we have no way to confirm the memory fix on real data right now.

What this fixes: Zendesk's incremental tickets export returns 1000 tickets per page, which came to about 160 MB per page for this account. Parsing a page held several full copies of it in memory. On top of the record queue, that pushed the source past 4Gi, and the kernel killed it at the same cursor on every attempt.

What the PR does (all opt-in, nothing changes unless a manifest sets spool_to_disk):

• JsonItemsDecoder works with DefaultPaginator, using the rest of the document (after_url, end_of_stream, ...) captured while streaming.
• spool_to_disk downloads each page to a temp file inside the request retry, so a broken download is re-fetched with no duplicate records.
• Body-based response filters are skipped only for spooled bodies over 1 MiB.
• Gzip responses are decoded on the streaming path.
• When a page contains an integer outside int64 (which the streaming parser rejects), the spooled page is re-parsed with orjson, the same parser JsonParser uses. The output then matches the current behavior exactly.

Where it stands (head 6a82cf6):

• A local bench of one page went from a 783 MB peak to 169 MB.
• The Zendesk unit tests (231) pass against the target manifest.
• A connector test PR on a prerelease of this branch, test(source-zendesk-support): stream tickets export via CDK prerelease (airbyte-python-cdk#1176) [DO NOT MERGE] airbyte#87045, gave identical records and state to the released connector, and rollback was clean.
• Not yet shown: the memory drop at customer scale.
• The branch conflicts with main. feat: add SPLIT_REQUEST_WINDOW action for adaptive request-window splitting #1171 and fix(http): make the sqlite request cache thread-safe #1200 both touch http_client.py, so a rebase needs a careful check of the download-and-retry path.
• Because of the conflict, the pull_request CI suite hasn't run on the last commit.

To pick this back up (e.g. if another large-ticket account hits the same OOM):

1. Rebase onto main.
2. Prove the memory drop, either with a customer test connection or with a mock export serving ~160 MB pages to a source capped at 4Gi.
3. Then request review.