feat(mdds)!: shard streaming pulls; size shards from request shape#1215
Open
userFRM wants to merge 2 commits into
Open
feat(mdds)!: shard streaming pulls; size shards from request shape#1215userFRM wants to merge 2 commits into
userFRM wants to merge 2 commits into
Conversation
The dedicated streaming builders (*_stream) now route through the bulk-fetch sharder like the buffered path; previously they always ran a single stream regardless of bulk_fetch policy or tier concurrency. Shard sizing no longer runs a density probe. plan_query decides shardability from the request shape and cuts the requested time or date range into equal concurrent bands, removing the OHLC sizing probe and its per-endpoint rows_per_event multiplier; that multiplier mis-sized quote chains badly enough that large quote pulls never sharded. BREAKING CHANGE: MarketDataClient::bulk_fetch_plan is now fn -> Option<ShardPlan> (was async fn -> Result<Option<ShardPlan>, Error>).
…eaming chunk delivery A terminal band failure on a chunk-streaming sharded pull no longer cancels the sibling bands: they drain to completion, and the call returns the new Error::PartialShardFetch naming the failed band window(s) so the caller can re-pull exactly the missing slices instead of restarting the whole pull. A pull where no chunk reached the handler still fails wholesale with the underlying error, a band that fails before delivering anything still retries transparently, and the call deadline still cancels every band at once. The buffered path's per-band from-scratch re-fetch (attempt-local collection, dedup-free replay) is documented and pinned by tests. The new variant maps to StreamError / THETADATADX_ERR_STREAM across the Python, TypeScript, and C surfaces. The four dedicated *_stream builders now drain through MarketDataClient::deliver_chunk_slices, the same primitive the generated .stream(handler) methods use, so both public surfaces of one endpoint agree on the no-replay delivered gating (marked only when rows reach the handler, keeping a recoverable transient after empty keepalives retryable) and stop at the first decode failure; the divergent for_each_chunk_body template is deleted. The planner now logs at debug why a query did not shard (unlisted endpoint, no cut axis, width under two, malformed window, provably small grid, narrow window), and each band future runs inside a shard_band tracing span carrying its window plus a per-band completion line with rows and duration, so concurrent bands stay distinguishable in the log stream. The shardable-endpoint set is now a const roster asserted equal to the registry's intraday-history endpoints (history* subcategory carrying start_time/end_time filters), so a new intraday history endpoint cannot silently never-shard. A new integration suite drives sharded pulls end-to-end through a per-band scripted mock, buffered and streaming: fan-out shape, band-order merge, NotFound folding, buffered band re-fetch, pre-delivery streaming retry, and the streaming partial-fetch error.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Bulk-fetch sharding, made generic, resilient, and reachable from the streaming path. Two commits:
*_stream()) now shard concurrently like the buffered path — they previously always ran a single stream regardless ofbulk_fetchor tier concurrency.plan_querydecides shardability from the request shape and cuts the requested time/date range into equal concurrent bands. The OHLC sizing probe and its per-endpointrows_per_eventmultiplier are deleted.Sizing: the probe is gone
The old sizer fired a separate OHLC query to estimate a pull by counting trades and multiplying by a hardcoded per-endpoint constant (roughly 15 rows per trade for quote endpoints). Real NBBO density is two orders of magnitude higher, so large quote chains fell under the shard threshold and ran single-stream. Equal-time banding needs no estimate and no per-endpoint tuning; the concurrency win dominates the residual imbalance from equal-time versus equal-work cuts.
Resilience: a band failure bounds its blast radius
Previously one band's transport error cancelled the siblings and lost the entire pull.
.await): a band's rows are held internally until the merge, so a failed band is re-fetched from scratch within the existing retry budget. Transient transport errors recover transparently, no duplicates..stream): sibling bands drain to completion instead of being cancelled, and the returnedError::PartialShardFetch { failed }names the exact failed band windows, so a caller re-pulls only those slices instead of the whole result. A band that fails before delivering any chunk still retries transparently.Delivery unification
The
*_stream()builders used a codegen template that marked an empty keepalive chunk as "delivered," diverging from the macro path'sdeliver_chunk_sliceson both retry budget and empty-query status. The template is deleted; both the sharded per-band arm and the single-stream arm now calldeliver_chunk_slices, so the two public surfaces of an endpoint behave identically.Generic and machine-enforced
No per-endpoint tuning remains. The shardable-endpoint set is now checked against the endpoint registry (intraday history = a
historysubcategory carryingstart_time/end_time), so a newly added history endpoint cannot silently never-shard.Breaking / API changes
bulk_fetch_planis nowfn(&self, endpoint, query) -> Option<ShardPlan>(wasasync fn -> Result<Option<ShardPlan>, Error>). It is pure computation on the request shape, so there is nothing to await and no error to surface. Rust-only; exposed in no binding.Error::PartialShardFetchvariant (additive), mapped toStreamError/THETADATADX_ERR_STREAMin every binding beside the existingPartialReconnect.Behavior notes (intentional)
NotFound, so wall-clock is neutral.bulk_fetch = "off"bypasses sharding entirely.Deadline
The 300s request timeout is a default. Large pulls should raise it via
with_deadline/timeout_ms(Duration::ZEROdisables it). Unchanged here.Verification
cargo clippy --workspace --all-targets --locked -- -D warnings,cargo test --workspace+ doctests,cargo fmt --check— cleantests/sharded_fanout.rs, 7 cases): fan-out issues N band requests, merge/interleave order, NotFound folding, buffered re-fetch recovery, streaming partial-error naming the failed windowNote:
check_c_abi_completenessflags 12 FFI config accessors (bulk_fetch/shard_concurrency/wait_mode/ window sizes) declared in the header but unimplemented inthetadatadx-ffi/src. This is pre-existing onmain(shipped in 0.2.0) and unrelated to this change.