From 307ffefef637198163ff6022570018aec7c4a443 Mon Sep 17 00:00:00 2001 From: benbenbang Date: Fri, 2 Oct 2026 12:10:18 +0800 Subject: [PATCH 1/3] fix(sdk/go): add separate bulk timeout for CSV calls CSV bulk endpoints now use DefaultBulkTimeout (600 seconds) until a client sets WithTimeout, instead of sharing the 30-second default meant for regular calls. Part 0 of profile-bulk measured 31.4 MB in 202 seconds, so the shorter default could cut off legitimate bulk downloads. WithTimeout still applies to every endpoint once set, including bulk, and the client tracks whether it was configured to decide between the two defaults. The documentation now notes that CSV bulk calls default to the longer timeout and the larger body cap until those options are set. Add a deadline-recording transport and a test confirming bulk calls get the bulk timeout while others get the standard one, and that WithTimeout overrides both. --- sdk/go/client.go | 22 +++++++++++++++-- sdk/go/csv.go | 6 ++--- sdk/go/csv_test.go | 61 ++++++++++++++++++++++++++++++++++++++++++++++ sdk/go/doc.go | 4 ++- 4 files changed, 87 insertions(+), 6 deletions(-) diff --git a/sdk/go/client.go b/sdk/go/client.go index 18e0a862..a82d9536 100644 --- a/sdk/go/client.go +++ b/sdk/go/client.go @@ -20,6 +20,10 @@ const ( DefaultPathPrefix = "stable" // DefaultTimeout is the logical deadline for one call, redirects included. DefaultTimeout = 30 * time.Second + // DefaultBulkTimeout is the logical deadline for one CSV bulk call while + // WithTimeout is unset (600 s): profile-bulk part 0 measured 31.4 MB in + // 202 s on 2026-10-02 (ADR 0035). + DefaultBulkTimeout = 600 * time.Second // DefaultConnectTimeout bounds connection establishment. DefaultConnectTimeout = 10 * time.Second // DefaultMaxResponseBodyBytes is the largest body buffered per response (64 MiB). @@ -66,6 +70,7 @@ type clientConfig struct { defaultHeaders []headerPair userAgent string timeout time.Duration + timeoutSet bool connectTimeout time.Duration maxResponseBodyBytes int64 maxResponseBodyBytesSet bool @@ -116,8 +121,14 @@ func WithUserAgent(userAgent string) Option { } // WithTimeout sets one logical deadline across redirects and the body read. +// It applies to every endpoint, bulk included; until it is set, bulk +// endpoints use DefaultBulkTimeout and the others use DefaultTimeout. A +// client given through WithHTTPClient keeps its own Timeout field as well. func WithTimeout(timeout time.Duration) Option { - return func(c *clientConfig) { c.timeout = timeout } + return func(c *clientConfig) { + c.timeout = timeout + c.timeoutSet = true + } } // WithConnectTimeout bounds connection establishment. It applies only to the @@ -179,7 +190,10 @@ type Client struct { // bulkMaxResponseBodyBytes caps CSV bulk bodies: the configured limit, // or DefaultBulkMaxResponseBodyBytes when none was set. bulkMaxResponseBodyBytes int64 - redirectPolicy RedirectPolicy + // bulkTimeout bounds CSV bulk calls: the configured timeout, or + // DefaultBulkTimeout when none was set. + bulkTimeout time.Duration + redirectPolicy RedirectPolicy } // NewClient validates the options and builds a client with the direct FMP @@ -274,6 +288,10 @@ func NewClient(opts ...Option) (*Client, error) { if cfg.maxResponseBodyBytesSet { client.bulkMaxResponseBodyBytes = cfg.maxResponseBodyBytes } + client.bulkTimeout = DefaultBulkTimeout + if cfg.timeoutSet { + client.bulkTimeout = cfg.timeout + } client.bindNamespaces() return client, nil } diff --git a/sdk/go/csv.go b/sdk/go/csv.go index 07c0170c..ccb5da23 100644 --- a/sdk/go/csv.go +++ b/sdk/go/csv.go @@ -42,8 +42,8 @@ type csvColumn struct { // json-tagged member, pointer or not, must be a header column; the first // record checks it. An empty // or header-only body yields no rows, and the provider-message check of -// getJSON does not apply. The redirect, timeout, and no-retry rules of -// getJSON are unchanged; the body cap is the bulk limit. +// getJSON does not apply. The redirect and no-retry rules of getJSON are +// unchanged; the body cap is the bulk limit and the deadline the bulk timeout. func getCSV[T any](ctx context.Context, c *Client, endpointID, relativePath string, query []queryParam) ([]T, error) { if err := validateRelativePath(relativePath); err != nil { return nil, err @@ -55,7 +55,7 @@ func getCSV[T any](ctx context.Context, c *Client, endpointID, relativePath stri } target.RawQuery = rawQuery - ctx, cancel := context.WithTimeout(ctx, c.timeout) + ctx, cancel := context.WithTimeout(ctx, c.bulkTimeout) defer cancel() resp, err := c.executeRedirects(ctx, endpointID, target, c.bulkMaxResponseBodyBytes) if err != nil { diff --git a/sdk/go/csv_test.go b/sdk/go/csv_test.go index 3ebbcf33..6ac9a42b 100644 --- a/sdk/go/csv_test.go +++ b/sdk/go/csv_test.go @@ -9,7 +9,9 @@ import ( "net/http" "slices" "strconv" + "sync" "testing" + "time" ) const csvEndpoint = "csv-contract-test" @@ -220,3 +222,62 @@ func cellText(value *string) string { } return *value } + +// deadlineTransport records how much of each request's context deadline +// remains when the request leaves the client. +type deadlineTransport struct { + next http.RoundTripper + mu sync.Mutex + remaining []time.Duration +} + +func (d *deadlineTransport) RoundTrip(req *http.Request) (*http.Response, error) { + deadline, ok := req.Context().Deadline() + d.mu.Lock() + if ok { + d.remaining = append(d.remaining, time.Until(deadline)) + } else { + d.remaining = append(d.remaining, 0) + } + d.mu.Unlock() + return d.next.RoundTrip(req) +} + +// TestBulkTimeoutDefaultsUntilTheClientSetsOne mirrors +// csv_bulk_endpoints_default_to_the_bulk_timeout_until_the_client_sets_one. +func TestBulkTimeoutDefaultsUntilTheClientSetsOne(t *testing.T) { + t.Parallel() + server, _ := newServer(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/router/stable/bulk-probe" { + w.Header().Set("Content-Type", "text/csv") + _, _ = w.Write([]byte(csvProbeHeader)) + return + } + jsonHandler(`[{"symbol":"A"}]`)(w, r) + }) + cases := []struct { + name string + opts []Option + bulk, all time.Duration + }{ + {"unset", nil, DefaultBulkTimeout, DefaultTimeout}, + {"configured", []Option{WithTimeout(5 * time.Second)}, 5 * time.Second, 5 * time.Second}, + } + for _, tc := range cases { + transport := &deadlineTransport{next: server.Client().Transport} + opts := append([]Option{WithHTTPClient(&http.Client{Transport: transport})}, tc.opts...) + client := newClient(t, server, opts...) + if _, err := getCSV[csvProbeRow](context.Background(), client, csvEndpoint, "bulk-probe", nil); err != nil { + t.Fatalf("%s bulk: %v", tc.name, err) + } + if _, err := probe(t, client); err != nil { + t.Fatalf("%s json: %v", tc.name, err) + } + for i, want := range []time.Duration{tc.bulk, tc.all} { + got := transport.remaining[i] + if got > want || got < want-5*time.Second { + t.Fatalf("%s request %d deadline = %v, want about %v", tc.name, i, got, want) + } + } + } +} diff --git a/sdk/go/doc.go b/sdk/go/doc.go index 7a3751c5..5b97991f 100644 --- a/sdk/go/doc.go +++ b/sdk/go/doc.go @@ -38,7 +38,9 @@ // times; RedirectNone returns the redirect as a status error instead. One // context deadline (WithTimeout, default 30 seconds) spans the whole call, // including redirects and the body read. Response bodies are capped by -// WithMaxResponseBodyBytes (default 64 MiB). +// WithMaxResponseBodyBytes (default 64 MiB). Until those options are set, +// CSV bulk calls use DefaultBulkTimeout (600 seconds) and +// DefaultBulkMaxResponseBodyBytes (256 MiB) instead. // // WithHTTPClient injects a caller-owned *http.Client for proxies and tests. // The SDK shallow-copies it to install its redirect policy, so the caller's From 44b0d6b43abd3c4a70d86df262197bb932095501 Mon Sep 17 00:00:00 2001 From: benbenbang Date: Fri, 2 Oct 2026 12:10:52 +0800 Subject: [PATCH 2/3] fix(crates): give CSV bulk calls a longer default timeout Bulk CSV bodies such as `profile-bulk` part 0 (31.4 MB over 202 s) routinely exceeded the single 30 s transport timeout, so whole-market bulk calls failed their deadline before the body finished arriving. **Per-endpoint default timeout:** * Added `DEFAULT_BULK_TIMEOUT` (600 s) alongside `DEFAULT_TIMEOUT` (30 s), and gave `EndpointSpec` a `default_timeout` that `get_csv` sets to the bulk value. * The client timeout is now optional; while unset, each call uses its endpoint's default, and when set it still applies to every endpoint, bulk included. **Timeout plumbing:** * `PreparedRequest` now carries the resolved per-call deadline, which the reqwest executor applies, so `ReqwestExecutor::new` no longer takes a timeout. * The redirect chain threads the resolved timeout through every hop. **Tests and docs:** * Added Rust and Python tests covering the bulk default, the general default, and an explicit timeout overriding both for bulk routes. * Documented the two defaults and the override rule across libfmp and the Python binding. --- crates/fmp-py/python/fmp/_native/__init__.pyi | 3 + crates/fmp-py/src/client.rs | 3 + crates/fmp-py/tests/test_client.py | 49 ++++++++++++++++ crates/libfmp/src/client.rs | 57 +++++++++++++------ crates/libfmp/src/endpoints/bulk.rs | 9 ++- crates/libfmp/src/endpoints/mod.rs | 15 ++++- crates/libfmp/src/transport.rs | 15 ++++- crates/libfmp/tests/transport.rs | 50 +++++++++++++++- 8 files changed, 177 insertions(+), 24 deletions(-) diff --git a/crates/fmp-py/python/fmp/_native/__init__.pyi b/crates/fmp-py/python/fmp/_native/__init__.pyi index 93a96d76..2cf38466 100644 --- a/crates/fmp-py/python/fmp/_native/__init__.pyi +++ b/crates/fmp-py/python/fmp/_native/__init__.pyi @@ -172,6 +172,9 @@ class FmpClient: the default host the constructor raises `FmpConfigError` naming `FMP_API_KEY`. `timeout` and `connect_timeout` are positive finite numbers of seconds. + When `timeout` is omitted, `client.bulk` methods may take up to 600 + seconds and every other method up to 30; a given `timeout` applies to + every method, bulk included. `max_response_body_bytes` bounds each buffered response; when omitted, `client.bulk` methods buffer up to 256 MiB and every other method up to 64 MiB. Authenticated diff --git a/crates/fmp-py/src/client.rs b/crates/fmp-py/src/client.rs index 2e7fce53..f24f10ec 100644 --- a/crates/fmp-py/src/client.rs +++ b/crates/fmp-py/src/client.rs @@ -171,6 +171,9 @@ fn positive_duration(value: f64, field: &'static str) -> PyResult { /// the default host the constructor raises `FmpConfigError` naming /// `FMP_API_KEY`. /// `timeout` and `connect_timeout` are positive finite numbers of seconds. +/// When `timeout` is omitted, `client.bulk` methods may take up to 600 +/// seconds and every other method up to 30; a given `timeout` applies to +/// every method, bulk included. /// `max_response_body_bytes` bounds each buffered response; when omitted, /// `client.bulk` methods buffer up to 256 MiB and every other method up to /// 64 MiB. Authenticated diff --git a/crates/fmp-py/tests/test_client.py b/crates/fmp-py/tests/test_client.py index 4cd10a1a..5b1000de 100644 --- a/crates/fmp-py/tests/test_client.py +++ b/crates/fmp-py/tests/test_client.py @@ -1,6 +1,10 @@ """Runtime contract of ``FmpClient`` and the ``client.quote`` namespace.""" import pickle +import threading +import time +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from types import MappingProxyType, SimpleNamespace from typing import Any @@ -284,6 +288,51 @@ def test_transport_error_is_structured(errors: SimpleNamespace) -> None: assert (error.status, error.body, error.body_truncated) == (None, None, None) +class _SlowCsvHandler(BaseHTTPRequestHandler): + """Answer every request with a header-only CSV body after a short delay.""" + + def do_GET(self) -> None: + """Delay, then send the CSV header.""" + time.sleep(1.0) + body = b"symbol\n" + self.send_response(200) + self.send_header("Content-Type", "text/csv") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, _format: str, *_args: Any) -> None: + """Silence the default stderr access log.""" + + +@pytest.fixture +def slow_csv_server() -> Iterator[str]: + """A loopback server whose CSV answers take one second; yields its origin.""" + server = ThreadingHTTPServer(("127.0.0.1", 0), _SlowCsvHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}" + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def test_explicit_timeout_applies_to_bulk_methods(slow_csv_server: str, errors: SimpleNamespace) -> None: + """A given ``timeout`` bounds bulk calls too, despite their 600 s default.""" + client = FmpClient(base_url=slow_csv_server, path_prefix="", auth_mode="none", timeout=0.2) + with pytest.raises(errors.FmpTransportError) as raised: + client.bulk.company_profiles("0") + assert raised.value.endpoint == "profile-bulk" + + +def test_bulk_methods_default_to_the_long_bulk_timeout(slow_csv_server: str) -> None: + """Without ``timeout``, a slow bulk body still decodes (the 600 s bulk default).""" + client = FmpClient(base_url=slow_csv_server, path_prefix="", auth_mode="none") + assert client.bulk.company_profiles("0") == [] + + def test_status_error_redacts_configured_query_secret(fixture_server: FixtureServer, errors: SimpleNamespace) -> None: """The query credential never appears in a status error's body or repr.""" fixture_server.route(QUOTE_SHORT_PATH, b"denied?apikey=query-secret", status=401, content_type="text/plain") diff --git a/crates/libfmp/src/client.rs b/crates/libfmp/src/client.rs index dc209619..1dbfbe45 100644 --- a/crates/libfmp/src/client.rs +++ b/crates/libfmp/src/client.rs @@ -29,7 +29,19 @@ use crate::{ pub use crate::endpoints::{EndpointSpec, QueryParameters}; -const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30); +/// Default logical deadline for one call, redirects and body included (30 s). +/// +/// It applies only while the client timeout is unset; +/// [`ClientBuilder::timeout`] replaces it for every endpoint. +pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30); + +/// Default logical deadline for one CSV bulk call (600 s). +/// +/// Whole-market bulk bodies arrive slowly: `profile-bulk` part 0 measured +/// 31.4 MB in 202 s on 2026-10-02. It applies only while the client timeout +/// is unset; [`ClientBuilder::timeout`] replaces it for every endpoint, bulk +/// included (ADR 0035). +pub const DEFAULT_BULK_TIMEOUT: Duration = Duration::from_secs(600); const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); /// Default maximum bytes buffered for each HTTP response (64 MiB). /// @@ -72,7 +84,7 @@ pub struct ClientBuilder { authentication_conflict: bool, default_headers: Vec<(String, String)>, user_agent: String, - timeout: Duration, + timeout: Option, connect_timeout: Duration, max_response_body_bytes: Option, danger_allow_insecure_authentication: bool, @@ -90,7 +102,7 @@ impl Default for ClientBuilder { authentication_conflict: false, default_headers: Vec::new(), user_agent: format!("libfmp/{VERSION}"), - timeout: DEFAULT_TIMEOUT, + timeout: None, connect_timeout: DEFAULT_CONNECT_TIMEOUT, max_response_body_bytes: None, danger_allow_insecure_authentication: false, @@ -170,9 +182,12 @@ impl ClientBuilder { /// Sets one logical transport timeout across redirects and body buffering. /// /// The client enforces this deadline around custom executors as well as the - /// built-in executor; individual redirect hops do not reset it. + /// built-in executor; individual redirect hops do not reset it. Until this + /// is set, each endpoint uses its own default: [`DEFAULT_TIMEOUT`], or + /// [`DEFAULT_BULK_TIMEOUT`] for CSV bulk routes. Once set, it applies to + /// every endpoint, bulk included. pub fn timeout(mut self, timeout: Duration) -> Self { - self.timeout = timeout; + self.timeout = Some(timeout); self } @@ -270,14 +285,12 @@ impl ClientBuilder { match self.executor { Some(executor) => (executor, DeadlineBackend::Thread), None => ( - Arc::new( - ReqwestExecutor::new(self.timeout, self.connect_timeout).map_err(|_| { - Error::configuration_with_kind( - ConfigurationErrorKind::HttpClient, - "HTTP client could not be constructed", - ) - })?, - ), + Arc::new(ReqwestExecutor::new(self.connect_timeout).map_err(|_| { + Error::configuration_with_kind( + ConfigurationErrorKind::HttpClient, + "HTTP client could not be constructed", + ) + })?), DeadlineBackend::Tokio, ), }; @@ -309,7 +322,7 @@ struct ClientInner { redactor: Redactor, executor: Arc, deadline_backend: DeadlineBackend, - timeout: Duration, + timeout: Option, max_response_body_bytes: Option, redirect_policy: RedirectPolicy, } @@ -386,10 +399,11 @@ impl Client { redactor.add_secret(&SecretString::new((*value).to_owned())); } + let timeout = self.inner.timeout.unwrap_or(endpoint.default_timeout()); with_timeout( self.inner.deadline_backend, - self.inner.timeout, - self.execute_redirects(endpoint, url, headers, &redactor), + timeout, + self.execute_redirects(endpoint, url, headers, &redactor, timeout), ) .await .ok_or_else(|| Error::transport(Some(endpoint.id()), "request deadline exceeded"))? @@ -401,6 +415,7 @@ impl Client { mut url: Url, headers: HeaderMap, redactor: &Redactor, + timeout: Duration, ) -> Result { for redirect_count in 0..=MAX_REDIRECTS { apply_query_auth(&mut url, self.inner.auth.query.as_ref()); @@ -416,6 +431,7 @@ impl Client { url.clone(), headers.clone(), max_body_bytes, + timeout, )) .await .map_err(|_| Error::transport(Some(endpoint.id()), "request execution failed"))?; @@ -1052,7 +1068,7 @@ mod tests { let injected = Client::builder() .base_url("https://example.test") .executor(Arc::new( - ReqwestExecutor::new(DEFAULT_TIMEOUT, DEFAULT_CONNECT_TIMEOUT).unwrap(), + ReqwestExecutor::new(DEFAULT_CONNECT_TIMEOUT).unwrap(), )) .build() .unwrap(); @@ -1129,4 +1145,11 @@ mod tests { assert_eq!(DEFAULT_BULK_MAX_RESPONSE_BODY_BYTES, 256 * 1024 * 1024); assert_eq!(ClientBuilder::default().max_response_body_bytes, None); } + + #[test] + fn default_timeouts_are_the_documented_values() { + assert_eq!(DEFAULT_TIMEOUT, Duration::from_secs(30)); + assert_eq!(DEFAULT_BULK_TIMEOUT, Duration::from_secs(600)); + assert_eq!(ClientBuilder::default().timeout, None); + } } diff --git a/crates/libfmp/src/endpoints/bulk.rs b/crates/libfmp/src/endpoints/bulk.rs index 72e2e124..3e7624e2 100644 --- a/crates/libfmp/src/endpoints/bulk.rs +++ b/crates/libfmp/src/endpoints/bulk.rs @@ -13,9 +13,12 @@ //! client limit is unset: whole-market bodies such as `ratios-ttm-bulk` //! exceed the 64 MiB general default. A limit set with //! [`crate::client::ClientBuilder::max_response_body_bytes`] applies to bulk -//! routes too, and every bulk call shares the client's single logical -//! timeout, which [`crate::client::ClientBuilder::timeout`] can raise. The -//! Python binding exposes the same eighteen methods under `client.bulk`. +//! routes too. The same rule holds for the call deadline: while the client +//! timeout is unset, bulk calls use +//! [`crate::client::DEFAULT_BULK_TIMEOUT`] (600 s) because whole-market bodies +//! arrive slowly, and every other route keeps the 30 s default. A timeout set +//! with [`crate::client::ClientBuilder::timeout`] applies to bulk routes too. +//! The Python binding exposes the same eighteen methods under `client.bulk`. use crate::{ Client, Result, diff --git a/crates/libfmp/src/endpoints/mod.rs b/crates/libfmp/src/endpoints/mod.rs index a3a973d3..abc2f163 100644 --- a/crates/libfmp/src/endpoints/mod.rs +++ b/crates/libfmp/src/endpoints/mod.rs @@ -61,6 +61,7 @@ pub struct EndpointSpec { metadata: EndpointMetadata, max_response_body_bytes: Option, default_max_response_body_bytes: usize, + default_timeout: std::time::Duration, } impl EndpointSpec { @@ -81,6 +82,7 @@ impl EndpointSpec { metadata: EndpointMetadata::new(), max_response_body_bytes: None, default_max_response_body_bytes: crate::client::DEFAULT_MAX_RESPONSE_BODY_BYTES, + default_timeout: crate::client::DEFAULT_TIMEOUT, } } @@ -142,6 +144,14 @@ impl EndpointSpec { pub const fn default_max_response_body_bytes(&self) -> usize { self.default_max_response_body_bytes } + + /// Returns the call deadline used while the client timeout is unset. + /// + /// This is [`crate::client::DEFAULT_TIMEOUT`] except for CSV bulk + /// descriptors, which use [`crate::client::DEFAULT_BULK_TIMEOUT`]. + pub const fn default_timeout(&self) -> std::time::Duration { + self.default_timeout + } } impl EndpointSpec @@ -181,7 +191,9 @@ where /// /// Each record decodes into one `Row` by header name. Unless the endpoint /// or the client sets a limit, the body may grow to - /// [`crate::client::DEFAULT_BULK_MAX_RESPONSE_BODY_BYTES`]. + /// [`crate::client::DEFAULT_BULK_MAX_RESPONSE_BODY_BYTES`], and unless the + /// client sets a timeout, the call may take + /// [`crate::client::DEFAULT_BULK_TIMEOUT`]. pub const fn get_csv(id: &'static str, relative_path: &'static str, query: Q) -> Self { let mut spec = Self::with_response( HttpMethod::Get, @@ -191,6 +203,7 @@ where ResponseContract::csv(), ); spec.default_max_response_body_bytes = crate::client::DEFAULT_BULK_MAX_RESPONSE_BODY_BYTES; + spec.default_timeout = crate::client::DEFAULT_BULK_TIMEOUT; spec } } diff --git a/crates/libfmp/src/transport.rs b/crates/libfmp/src/transport.rs index 4e5b278e..48e98faa 100644 --- a/crates/libfmp/src/transport.rs +++ b/crates/libfmp/src/transport.rs @@ -35,6 +35,7 @@ pub struct PreparedRequest { url: Url, headers: HeaderMap, max_response_body_bytes: usize, + timeout: std::time::Duration, } impl PreparedRequest { @@ -43,12 +44,14 @@ impl PreparedRequest { url: Url, headers: HeaderMap, max_response_body_bytes: usize, + timeout: std::time::Duration, ) -> Self { Self { method, url, headers, max_response_body_bytes, + timeout, } } @@ -78,6 +81,15 @@ impl PreparedRequest { pub fn max_response_body_bytes(&self) -> usize { self.max_response_body_bytes } + + /// Returns the logical deadline of the call this request belongs to. + /// + /// The client enforces it around every executor; an executor may also + /// apply it to its own transport. It spans the whole redirect chain, so + /// it is an upper bound for this one request. + pub fn timeout(&self) -> std::time::Duration { + self.timeout + } } impl fmt::Debug for PreparedRequest { @@ -215,11 +227,9 @@ pub(crate) struct ReqwestExecutor { impl ReqwestExecutor { pub(crate) fn new( - timeout: std::time::Duration, connect_timeout: std::time::Duration, ) -> std::result::Result { let client = reqwest::Client::builder() - .timeout(timeout) .connect_timeout(connect_timeout) // Redirects are handled above the executor boundary so every auth // mode follows exactly the same same-origin rule. @@ -244,6 +254,7 @@ impl HttpExecutor for ReqwestExecutor { .client .request(request.method.into_reqwest(), request.url) .headers(request.headers) + .timeout(request.timeout) .send() .await .map_err(|_| ExecutorError::new())?; diff --git a/crates/libfmp/tests/transport.rs b/crates/libfmp/tests/transport.rs index dd740c4d..957955d3 100644 --- a/crates/libfmp/tests/transport.rs +++ b/crates/libfmp/tests/transport.rs @@ -12,7 +12,10 @@ use bytes::Bytes; use http::header::{CONTENT_TYPE, HeaderMap, HeaderValue, LOCATION}; use libfmp::{ Client, - client::{DEFAULT_BULK_MAX_RESPONSE_BODY_BYTES, DEFAULT_MAX_RESPONSE_BODY_BYTES, EndpointSpec}, + client::{ + DEFAULT_BULK_MAX_RESPONSE_BODY_BYTES, DEFAULT_BULK_TIMEOUT, + DEFAULT_MAX_RESPONSE_BODY_BYTES, DEFAULT_TIMEOUT, EndpointSpec, + }, config::{Authentication, RedirectPolicy}, endpoints::BinaryBody, error::{ConfigurationErrorKind, ErrorCategory}, @@ -753,6 +756,51 @@ async fn csv_bulk_endpoints_default_to_the_bulk_body_limit_until_the_client_sets ); } +#[tokio::test] +async fn csv_bulk_endpoints_default_to_the_bulk_timeout_until_the_client_sets_one() { + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("text/csv")); + let csv = || TransportResponse::new(200, headers.clone(), b"ok\ntrue\n".as_slice()); + let bulk = EndpointSpec::<(), Vec>::get_csv("bulk-test", "bulk-test", ()); + let executor = Arc::new(ScriptedExecutor::new(vec![ + Ok(csv()), + Ok(json_response()), + Ok(csv()), + Ok(json_response()), + ])); + let unset = Client::builder() + .base_url("https://example.test") + .executor(executor.clone()) + .build() + .unwrap(); + let configured = Client::builder() + .base_url("https://example.test") + .timeout(Duration::from_secs(5)) + .executor(executor.clone()) + .build() + .unwrap(); + + unset.execute(&bulk).await.unwrap(); + unset.execute(&endpoint()).await.unwrap(); + configured.execute(&bulk).await.unwrap(); + configured.execute(&endpoint()).await.unwrap(); + + let timeouts: Vec<_> = executor + .requests() + .iter() + .map(PreparedRequest::timeout) + .collect(); + assert_eq!( + timeouts, + [ + DEFAULT_BULK_TIMEOUT, + DEFAULT_TIMEOUT, + Duration::from_secs(5), + Duration::from_secs(5) + ] + ); +} + #[tokio::test] async fn binary_response_retains_the_transport_buffer_without_copying() { let bytes = Bytes::from_static(b"binary-response"); From 88ffac868b2bfe0e6b0a7aea9c61bf97baf312c7 Mon Sep 17 00:00:00 2001 From: benbenbang Date: Fri, 2 Oct 2026 12:12:59 +0800 Subject: [PATCH 3/3] fix(docs): document bulk call deadline to prevent mid-body timeouts Record the call deadline rule for CSV bulk descriptors in ADR 0035. The 30 s default timeout caused bulk calls to fail with a transport error mid-body (profile-bulk part 0 measured 31.4 MB in 202 s). Bulk descriptors now default to DEFAULT_BULK_TIMEOUT (600 s) while the client timeout is unset; an explicit client timeout still applies to every endpoint, bulk included. close #404 --- docs/adr/0035-bulk-csv-responses.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/docs/adr/0035-bulk-csv-responses.md b/docs/adr/0035-bulk-csv-responses.md index 4e17a9d8..b21ad949 100644 --- a/docs/adr/0035-bulk-csv-responses.md +++ b/docs/adr/0035-bulk-csv-responses.md @@ -121,6 +121,18 @@ included, and an endpoint override still wins. Bodies stay buffered, as in ADR 0028, and decode record by record from the buffer; there is no streaming API. +### Call deadline + +The call deadline follows the same rule (issue #404). Bulk bodies arrive +slowly: `profile-bulk` part 0 measured 31.4 MB in 202 s on 2026-10-02, well +past the 30 s default, so the call failed with a transport error mid-body. +CSV bulk descriptors default to `DEFAULT_BULK_TIMEOUT` (600 s, Go +`DefaultBulkTimeout`) while the client timeout is unset; every other endpoint +keeps the 30 s default. Once a caller sets the client timeout +(`ClientBuilder::timeout`, Go `WithTimeout`, Python `timeout=`), it applies to +every endpoint, bulk included. The deadline still spans the whole call, +redirects and body included, and the connect timeout is unchanged. + ## Alternatives considered - **Evidence-only `Option`s for bulk rows.** Rejected: about forty members