faucet_source_xml/
stream.rs

1//! XML stream executor.
2
3use crate::config::{XmlAuth, XmlPagination, XmlStreamConfig};
4use crate::convert;
5use async_trait::async_trait;
6use faucet_core::util::{self, DEFAULT_ERROR_BODY_MAX_LEN};
7use faucet_core::{AuthSpec, Credential, FaucetError, SharedAuthProvider};
8use faucet_core::{Stream, StreamPage};
9use reqwest::Client;
10use serde_json::Value;
11use std::collections::HashMap;
12use std::pin::Pin;
13use std::time::Duration;
14
15/// Content fingerprint of a fetched page, used as a pagination loop guard: a
16/// server that ignores the page/offset parameter (or clamps to the last page)
17/// returns the same non-empty page on every request, which would otherwise loop
18/// forever. Stopping when two consecutive pages fingerprint identically mirrors
19/// the REST source's body-fingerprint guard (audit #146 H4/H5).
20fn page_fingerprint(records: &[Value]) -> u64 {
21    use std::hash::{Hash, Hasher};
22    // `serde_json::Value` is not `Hash`; hash its canonical string form.
23    let mut hasher = std::collections::hash_map::DefaultHasher::new();
24    records.len().hash(&mut hasher);
25    for r in records {
26        r.to_string().hash(&mut hasher);
27    }
28    hasher.finish()
29}
30
31/// Retries on transient (5xx / connection) failures before giving up.
32const RETRY_MAX_ATTEMPTS: u32 = 3;
33/// Base exponential-backoff delay between retries.
34const RETRY_BASE_BACKOFF: Duration = Duration::from_millis(500);
35
36/// A configured XML API source that handles pagination and extraction.
37pub struct XmlStream {
38    config: XmlStreamConfig,
39    client: Client,
40    /// Optional shared auth provider. When present it takes precedence over
41    /// inline auth, so several sources can share one token with single-flight
42    /// refresh. Used by the CLI to resolve `auth: { ref }`, and by library
43    /// callers who construct one provider and inject it into many sources.
44    auth_provider: Option<SharedAuthProvider>,
45    /// Retry policy for transient request failures. Defaulted in `new()` to
46    /// reproduce the legacy `RETRY_MAX_ATTEMPTS` / `RETRY_BASE_BACKOFF`
47    /// constants; overridable via [`with_retry_policy`](Self::with_retry_policy).
48    retry_policy: faucet_core::RetryPolicy,
49}
50
51/// Map a [`Credential`] from a shared provider onto the XML [`XmlAuth`]
52/// representation so the existing header-application path can be reused.
53fn credential_to_auth(cred: Credential) -> XmlAuth {
54    match cred {
55        Credential::Bearer(token) => XmlAuth::Bearer { token },
56        Credential::Token(token) => XmlAuth::Custom {
57            headers: std::iter::once(("Authorization".to_string(), token)).collect(),
58        },
59        Credential::Basic { username, password } => XmlAuth::Basic { username, password },
60        Credential::Header { name, value } => XmlAuth::Custom {
61            headers: std::iter::once((name, value)).collect(),
62        },
63    }
64}
65
66impl XmlStream {
67    /// Create a new XML stream from the given configuration.
68    pub fn new(config: XmlStreamConfig) -> Self {
69        Self {
70            config,
71            client: Client::new(),
72            auth_provider: None,
73            // Reproduce the legacy `execute_with_retry(RETRY_MAX_ATTEMPTS,
74            // RETRY_BASE_BACKOFF, …)` behavior exactly: `max_retries` is
75            // retries-after-first, so `max_attempts = RETRY_MAX_ATTEMPTS + 1`.
76            retry_policy: faucet_core::RetryPolicy {
77                max_attempts: RETRY_MAX_ATTEMPTS + 1,
78                backoff: faucet_core::BackoffKind::Exponential,
79                base: RETRY_BASE_BACKOFF,
80                max: Duration::from_secs(60),
81                jitter: true,
82                retry_on: faucet_core::RetryClassSet::default(),
83            },
84        }
85    }
86
87    /// Attach a custom [`RetryPolicy`](faucet_core::RetryPolicy) for transient
88    /// request failures, replacing the default derived from
89    /// `RETRY_MAX_ATTEMPTS` / `RETRY_BASE_BACKOFF`. Used by the CLI to inject a
90    /// pipeline-level `resilience:` policy into the source.
91    pub fn with_retry_policy(mut self, policy: faucet_core::RetryPolicy) -> Self {
92        self.retry_policy = policy;
93        self
94    }
95
96    /// Attach a shared [`AuthProvider`](faucet_core::AuthProvider). When set,
97    /// the provider supplies the credential for every request (taking precedence
98    /// over inline auth), so several sources can share one token with
99    /// single-flight refresh. Used by the CLI to resolve `auth: { ref }`, and by
100    /// library callers who construct one provider and inject it into many
101    /// sources.
102    pub fn with_auth_provider(mut self, provider: SharedAuthProvider) -> Self {
103        self.auth_provider = Some(provider);
104        self
105    }
106
107    /// Fetch all records across all pages.
108    pub async fn fetch_all(&self) -> Result<Vec<Value>, FaucetError> {
109        self.fetch_all_with_context(&HashMap::new()).await
110    }
111
112    /// Fetch all records, substituting parent context into path, query_params, and body.
113    async fn fetch_all_with_context(
114        &self,
115        context: &HashMap<String, serde_json::Value>,
116    ) -> Result<Vec<Value>, FaucetError> {
117        let mut all_records = Vec::new();
118        let mut pages_fetched = 0usize;
119        let mut offset = 0usize;
120        let mut page_number = None;
121        let mut prev_fingerprint: Option<u64> = None;
122
123        // Initialize pagination state.
124        if let Some(XmlPagination::PageNumber { start_page, .. }) = &self.config.pagination {
125            page_number = Some(*start_page);
126        }
127
128        loop {
129            if let Some(max) = self.config.max_pages
130                && pages_fetched >= max
131            {
132                tracing::warn!("max pages ({max}) reached");
133                break;
134            }
135
136            let mut params = self.config.query_params.clone();
137            self.apply_pagination_params(&mut params, page_number, offset);
138
139            let xml_text = self.execute_request(&params, context).await?;
140            let json = convert::xml_to_json(&xml_text)?;
141
142            let records = match &self.config.records_element_path {
143                Some(path) => convert::extract_at_path(&json, path),
144                None => vec![json],
145            };
146
147            let record_count = records.len();
148            let fingerprint = page_fingerprint(&records);
149            all_records.extend(records);
150            pages_fetched += 1;
151
152            // Loop guard: a server that ignores the page/offset parameter (or
153            // clamps to the last page) returns the same non-empty page forever.
154            // Stop when two consecutive pages are identical (audit #146 H4/H5).
155            if record_count > 0 && prev_fingerprint == Some(fingerprint) {
156                tracing::warn!(
157                    "XML pagination returned an identical page; stopping to avoid an infinite loop"
158                );
159                break;
160            }
161            prev_fingerprint = Some(fingerprint);
162
163            // Advance pagination or stop.
164            match &self.config.pagination {
165                Some(XmlPagination::PageNumber { page_size, .. }) => {
166                    if record_count == 0 {
167                        break;
168                    }
169                    // Stop if page_size is set and we got fewer records than the page size.
170                    if let Some(size) = page_size
171                        && record_count < *size
172                    {
173                        break;
174                    }
175                    page_number = page_number.map(|p| p + 1);
176                }
177                Some(XmlPagination::Offset { limit, .. }) => {
178                    if record_count < *limit {
179                        break;
180                    }
181                    offset += record_count;
182                }
183                None => break,
184            }
185        }
186
187        tracing::info!(
188            records = all_records.len(),
189            pages = pages_fetched,
190            "XML fetch complete"
191        );
192        Ok(all_records)
193    }
194
195    fn apply_pagination_params(
196        &self,
197        params: &mut HashMap<String, String>,
198        page_number: Option<usize>,
199        offset: usize,
200    ) {
201        match &self.config.pagination {
202            Some(XmlPagination::PageNumber {
203                param_name,
204                page_size,
205                page_size_param,
206                ..
207            }) => {
208                if let Some(page) = page_number {
209                    params.insert(param_name.clone(), page.to_string());
210                }
211                if let (Some(size), Some(param)) = (page_size, page_size_param) {
212                    params.insert(param.clone(), size.to_string());
213                }
214            }
215            Some(XmlPagination::Offset {
216                offset_param,
217                limit_param,
218                limit,
219            }) => {
220                params.insert(offset_param.clone(), offset.to_string());
221                params.insert(limit_param.clone(), limit.to_string());
222            }
223            None => {}
224        }
225    }
226
227    async fn execute_request(
228        &self,
229        params: &HashMap<String, String>,
230        context: &HashMap<String, serde_json::Value>,
231    ) -> Result<String, FaucetError> {
232        let path = if context.is_empty() {
233            self.config.path.clone()
234        } else {
235            faucet_core::util::substitute_context(&self.config.path, context)
236        };
237
238        let url = format!("{}/{}", self.config.base_url, path.trim_start_matches('/'));
239
240        // Substitute context into query parameter values.
241        let resolved_params: HashMap<String, String> = if context.is_empty() {
242            params.clone()
243        } else {
244            params
245                .iter()
246                .map(|(k, v)| (k.clone(), faucet_core::util::substitute_context(v, context)))
247                .collect()
248        };
249
250        let mut req = self
251            .client
252            .request(self.config.method.clone(), &url)
253            .headers(self.config.headers.clone())
254            .query(&resolved_params);
255
256        // Resolve credentials to concrete auth. A shared auth provider
257        // (from `auth: { ref }` or injected by a library caller) takes
258        // precedence; otherwise inline auth is used.
259        let effective_auth: XmlAuth = if let Some(provider) = &self.auth_provider {
260            credential_to_auth(provider.credential().await?)
261        } else {
262            match &self.config.auth {
263                AuthSpec::Inline(a) => a.clone(),
264                AuthSpec::Reference(r) => {
265                    return Err(FaucetError::Auth(format!(
266                        "auth references provider '{}' but no provider was supplied; \
267                         set one via the CLI `auth:` catalog or `with_auth_provider`",
268                        r.name
269                    )));
270                }
271            }
272        };
273
274        // Apply auth.
275        match &effective_auth {
276            XmlAuth::None => {}
277            XmlAuth::Bearer { token } => {
278                req = req.bearer_auth(token);
279            }
280            XmlAuth::Basic { username, password } => {
281                req = req.basic_auth(username, Some(password));
282            }
283            XmlAuth::Custom { headers } => {
284                let mut hm = reqwest::header::HeaderMap::new();
285                for (name, value) in headers {
286                    let n =
287                        reqwest::header::HeaderName::from_bytes(name.as_bytes()).map_err(|e| {
288                            FaucetError::Auth(format!("invalid custom header name {name:?}: {e}"))
289                        })?;
290                    let v = reqwest::header::HeaderValue::from_str(value).map_err(|e| {
291                        FaucetError::Auth(format!("invalid custom header value for {name:?}: {e}"))
292                    })?;
293                    hm.insert(n, v);
294                }
295                req = req.headers(hm);
296            }
297        }
298
299        // Set body for POST requests (SOAP), with context substitution.
300        if let Some(body) = &self.config.body {
301            let resolved_body = if context.is_empty() {
302                body.clone()
303            } else {
304                faucet_core::util::substitute_context(body, context)
305            };
306            req = req
307                .header("Content-Type", "text/xml; charset=utf-8")
308                .body(resolved_body);
309        }
310
311        // Retry transient failures (5xx / connection resets) with jittered
312        // backoff, matching the REST source's reliability layer (#78/#16).
313        // The request body is a String, so `try_clone` always succeeds.
314        faucet_core::execute_with_policy(&self.retry_policy, None, || {
315            let attempt = req.try_clone();
316            async move {
317                let req = attempt.ok_or_else(|| {
318                    FaucetError::Source("xml: request is not cloneable for retry".into())
319                })?;
320                let resp = req.send().await.map_err(FaucetError::Http)?;
321                let resp = util::check_http_response(resp, DEFAULT_ERROR_BODY_MAX_LEN).await?;
322                resp.text().await.map_err(FaucetError::Http)
323            }
324        })
325        .await
326    }
327}
328
329#[async_trait]
330impl faucet_core::Source for XmlStream {
331    async fn fetch_with_context(
332        &self,
333        context: &std::collections::HashMap<String, serde_json::Value>,
334    ) -> Result<Vec<Value>, FaucetError> {
335        self.fetch_all_with_context(context).await
336    }
337
338    /// Stream records from the XML response without materialising the whole
339    /// document tree. The event-driven parser only builds JSON values for
340    /// elements matching [`XmlStreamConfig::records_element_path`]; other
341    /// elements are observed and discarded, so client-side memory is bounded
342    /// at `O(batch_size * record_size)` regardless of how large the document
343    /// is.
344    ///
345    /// Records are accumulated into a buffer of
346    /// [`XmlStreamConfig::batch_size`] entries and yielded as a
347    /// [`StreamPage`] once the buffer is full. The trailing partial buffer
348    /// (if any) is emitted after the parser hits EOF and all pagination
349    /// rounds drain.
350    ///
351    /// The trait-level `batch_size` argument is intentionally ignored in
352    /// favour of the config field — the config is the user-facing knob the
353    /// README documents, and routing the pipeline-supplied hint through it
354    /// would silently override an explicit config value. `batch_size = 0`
355    /// drains every page into a single emitted page.
356    ///
357    /// Bookmarks are always `None` — the XML source has no
358    /// incremental-replication mode today; pagination only walks the
359    /// API's own page-number / offset cursor.
360    fn stream_pages<'a>(
361        &'a self,
362        context: &'a HashMap<String, Value>,
363        _batch_size: usize,
364    ) -> Pin<Box<dyn Stream<Item = Result<StreamPage, FaucetError>> + Send + 'a>> {
365        let batch_size = self.config.batch_size;
366        let owned_context = context.clone();
367
368        Box::pin(async_stream::try_stream! {
369            let chunk = if batch_size == 0 { usize::MAX } else { batch_size };
370            let initial_capacity = if batch_size == 0 { 1024 } else { batch_size };
371            let mut buffer: Vec<Value> = Vec::with_capacity(initial_capacity);
372            let mut total = 0usize;
373            let mut pages_fetched = 0usize;
374            let mut offset = 0usize;
375            let mut page_number = None;
376            let mut prev_fingerprint: Option<u64> = None;
377
378            if let Some(XmlPagination::PageNumber { start_page, .. }) =
379                &self.config.pagination
380            {
381                page_number = Some(*start_page);
382            }
383
384            loop {
385                if let Some(max) = self.config.max_pages
386                    && pages_fetched >= max
387                {
388                    tracing::warn!("max pages ({max}) reached");
389                    break;
390                }
391
392                let mut params = self.config.query_params.clone();
393                self.apply_pagination_params(&mut params, page_number, offset);
394
395                let xml_text = self.execute_request(&params, &owned_context).await?;
396
397                // Event-driven extraction: only the matched subtree is
398                // ever materialised. The closure pushes into the local
399                // buffer; once it crosses `chunk`, the surrounding loop
400                // can flush, but we can't `yield` from inside the closure,
401                // so we collect this HTTP page's records into a scratch
402                // Vec and then iterate them after.
403                let mut page_records: Vec<Value> = Vec::new();
404                convert::stream_extract(
405                    &xml_text,
406                    self.config.records_element_path.as_deref(),
407                    |rec| page_records.push(rec),
408                )?;
409
410                let record_count = page_records.len();
411                let fingerprint = page_fingerprint(&page_records);
412
413                for rec in page_records.drain(..) {
414                    buffer.push(rec);
415                    if buffer.len() >= chunk {
416                        let flush = std::mem::replace(&mut buffer, Vec::with_capacity(initial_capacity));
417                        total += flush.len();
418                        yield StreamPage { records: flush, bookmark: None };
419                    }
420                }
421                pages_fetched += 1;
422
423                // Loop guard: stop when two consecutive pages are identical — a
424                // server ignoring the page/offset parameter (or clamping to the
425                // last page) returns the same non-empty page forever (#146 H4/H5).
426                if record_count > 0 && prev_fingerprint == Some(fingerprint) {
427                    tracing::warn!(
428                        "XML pagination returned an identical page; stopping to avoid an infinite loop"
429                    );
430                    break;
431                }
432                prev_fingerprint = Some(fingerprint);
433
434                // Advance pagination using the same rules as
435                // `fetch_all_with_context`.
436                match &self.config.pagination {
437                    Some(XmlPagination::PageNumber { page_size, .. }) => {
438                        if record_count == 0 {
439                            break;
440                        }
441                        if let Some(size) = page_size
442                            && record_count < *size
443                        {
444                            break;
445                        }
446                        page_number = page_number.map(|p| p + 1);
447                    }
448                    Some(XmlPagination::Offset { limit, .. }) => {
449                        if record_count < *limit {
450                            break;
451                        }
452                        offset += record_count;
453                    }
454                    None => break,
455                }
456            }
457
458            if !buffer.is_empty() {
459                total += buffer.len();
460                yield StreamPage { records: buffer, bookmark: None };
461            }
462
463            tracing::info!(
464                records = total,
465                pages = pages_fetched,
466                batch_size,
467                "XML source stream complete",
468            );
469        })
470    }
471
472    fn config_schema(&self) -> serde_json::Value {
473        serde_json::to_value(faucet_core::schema_for!(XmlStreamConfig))
474            .expect("schema serialization")
475    }
476
477    fn dataset_uri(&self) -> String {
478        format!(
479            "{}{}",
480            faucet_core::redact_uri_credentials(&self.config.base_url),
481            self.config.path
482        )
483    }
484}
485
486#[cfg(test)]
487mod tests {
488    use super::*;
489    use faucet_core::Source;
490
491    #[test]
492    fn dataset_uri_combines_base_and_path() {
493        let source = XmlStream::new(XmlStreamConfig::new(
494            "https://soap.example.com",
495            "/api/v1/service",
496        ));
497        assert_eq!(
498            source.dataset_uri(),
499            "https://soap.example.com/api/v1/service"
500        );
501    }
502
503    #[test]
504    fn dataset_uri_redacts_credentials() {
505        let source = XmlStream::new(XmlStreamConfig::new(
506            "https://user:pass@soap.example.com",
507            "/svc",
508        ));
509        assert_eq!(source.dataset_uri(), "https://soap.example.com/svc");
510    }
511
512    #[test]
513    fn default_retry_policy_reproduces_legacy_constants() {
514        let source = XmlStream::new(XmlStreamConfig::new("https://soap.example.com", "/svc"));
515        assert_eq!(source.retry_policy.max_attempts, RETRY_MAX_ATTEMPTS + 1);
516        assert_eq!(source.retry_policy.base, RETRY_BASE_BACKOFF);
517    }
518
519    #[test]
520    fn with_retry_policy_overrides_the_default() {
521        let policy = faucet_core::RetryPolicy {
522            max_attempts: 9,
523            base: Duration::from_secs(7),
524            ..faucet_core::RetryPolicy::default()
525        };
526        let source = XmlStream::new(XmlStreamConfig::new("https://soap.example.com", "/svc"))
527            .with_retry_policy(policy);
528        assert_eq!(source.retry_policy.max_attempts, 9);
529        assert_eq!(source.retry_policy.base, Duration::from_secs(7));
530    }
531}
faucet_source_xml/stream.rs

faucet_source_xml/
stream.rs
