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            pages_fetched += 1;
150
151            // Loop guard: a server that ignores the page/offset parameter (or
152            // clamps to the last page) returns the same non-empty page forever.
153            // Stop when two consecutive pages are identical (audit #146 H4/H5) —
154            // and do it BEFORE appending, so the duplicate page's records are
155            // never emitted to the sink a second time (audit #321 M4).
156            if record_count > 0 && prev_fingerprint == Some(fingerprint) {
157                tracing::warn!(
158                    "XML pagination returned an identical page; stopping to avoid an infinite loop"
159                );
160                break;
161            }
162            prev_fingerprint = Some(fingerprint);
163            all_records.extend(records);
164
165            // Advance pagination or stop.
166            match &self.config.pagination {
167                Some(XmlPagination::PageNumber { page_size, .. }) => {
168                    if record_count == 0 {
169                        break;
170                    }
171                    // Stop if page_size is set and we got fewer records than the page size.
172                    if let Some(size) = page_size
173                        && record_count < *size
174                    {
175                        break;
176                    }
177                    page_number = page_number.map(|p| p + 1);
178                }
179                Some(XmlPagination::Offset { limit, .. }) => {
180                    if record_count < *limit {
181                        break;
182                    }
183                    offset += record_count;
184                }
185                None => break,
186            }
187        }
188
189        tracing::info!(
190            records = all_records.len(),
191            pages = pages_fetched,
192            "XML fetch complete"
193        );
194        Ok(all_records)
195    }
196
197    fn apply_pagination_params(
198        &self,
199        params: &mut HashMap<String, String>,
200        page_number: Option<usize>,
201        offset: usize,
202    ) {
203        match &self.config.pagination {
204            Some(XmlPagination::PageNumber {
205                param_name,
206                page_size,
207                page_size_param,
208                ..
209            }) => {
210                if let Some(page) = page_number {
211                    params.insert(param_name.clone(), page.to_string());
212                }
213                if let (Some(size), Some(param)) = (page_size, page_size_param) {
214                    params.insert(param.clone(), size.to_string());
215                }
216            }
217            Some(XmlPagination::Offset {
218                offset_param,
219                limit_param,
220                limit,
221            }) => {
222                params.insert(offset_param.clone(), offset.to_string());
223                params.insert(limit_param.clone(), limit.to_string());
224            }
225            None => {}
226        }
227    }
228
229    async fn execute_request(
230        &self,
231        params: &HashMap<String, String>,
232        context: &HashMap<String, serde_json::Value>,
233    ) -> Result<String, FaucetError> {
234        let path = if context.is_empty() {
235            self.config.path.clone()
236        } else {
237            faucet_core::util::substitute_context(&self.config.path, context)
238        };
239
240        let url = format!("{}/{}", self.config.base_url, path.trim_start_matches('/'));
241
242        // Substitute context into query parameter values.
243        let resolved_params: HashMap<String, String> = if context.is_empty() {
244            params.clone()
245        } else {
246            params
247                .iter()
248                .map(|(k, v)| (k.clone(), faucet_core::util::substitute_context(v, context)))
249                .collect()
250        };
251
252        let mut req = self
253            .client
254            .request(self.config.method.clone(), &url)
255            .headers(self.config.headers.clone())
256            .query(&resolved_params);
257
258        // Resolve credentials to concrete auth. A shared auth provider
259        // (from `auth: { ref }` or injected by a library caller) takes
260        // precedence; otherwise inline auth is used.
261        let effective_auth: XmlAuth = if let Some(provider) = &self.auth_provider {
262            credential_to_auth(provider.credential().await?)
263        } else {
264            match &self.config.auth {
265                AuthSpec::Inline(a) => a.clone(),
266                AuthSpec::Reference(r) => {
267                    return Err(FaucetError::Auth(format!(
268                        "auth references provider '{}' but no provider was supplied; \
269                         set one via the CLI `auth:` catalog or `with_auth_provider`",
270                        r.name
271                    )));
272                }
273            }
274        };
275
276        // Apply auth.
277        match &effective_auth {
278            XmlAuth::None => {}
279            XmlAuth::Bearer { token } => {
280                req = req.bearer_auth(token);
281            }
282            XmlAuth::Basic { username, password } => {
283                req = req.basic_auth(username, Some(password));
284            }
285            XmlAuth::Custom { headers } => {
286                let mut hm = reqwest::header::HeaderMap::new();
287                for (name, value) in headers {
288                    let n =
289                        reqwest::header::HeaderName::from_bytes(name.as_bytes()).map_err(|e| {
290                            FaucetError::Auth(format!("invalid custom header name {name:?}: {e}"))
291                        })?;
292                    let v = reqwest::header::HeaderValue::from_str(value).map_err(|e| {
293                        FaucetError::Auth(format!("invalid custom header value for {name:?}: {e}"))
294                    })?;
295                    hm.insert(n, v);
296                }
297                req = req.headers(hm);
298            }
299        }
300
301        // Set body for POST requests (SOAP), with context substitution.
302        if let Some(body) = &self.config.body {
303            let resolved_body = if context.is_empty() {
304                body.clone()
305            } else {
306                faucet_core::util::substitute_context(body, context)
307            };
308            req = req
309                .header("Content-Type", "text/xml; charset=utf-8")
310                .body(resolved_body);
311        }
312
313        // Retry transient failures (5xx / connection resets) with jittered
314        // backoff, matching the REST source's reliability layer (#78/#16).
315        // The request body is a String, so `try_clone` always succeeds.
316        faucet_core::execute_with_policy(&self.retry_policy, None, || {
317            let attempt = req.try_clone();
318            async move {
319                let req = attempt.ok_or_else(|| {
320                    FaucetError::Source("xml: request is not cloneable for retry".into())
321                })?;
322                let resp = req.send().await.map_err(FaucetError::Http)?;
323                let resp = util::check_http_response(resp, DEFAULT_ERROR_BODY_MAX_LEN).await?;
324                resp.text().await.map_err(FaucetError::Http)
325            }
326        })
327        .await
328    }
329}
330
331#[async_trait]
332impl faucet_core::Source for XmlStream {
333    async fn fetch_with_context(
334        &self,
335        context: &std::collections::HashMap<String, serde_json::Value>,
336    ) -> Result<Vec<Value>, FaucetError> {
337        self.fetch_all_with_context(context).await
338    }
339
340    /// Stream records from the XML response without materialising the whole
341    /// document tree. The event-driven parser only builds JSON values for
342    /// elements matching [`XmlStreamConfig::records_element_path`]; other
343    /// elements are observed and discarded, so client-side memory is bounded
344    /// at `O(batch_size * record_size)` regardless of how large the document
345    /// is.
346    ///
347    /// Records are accumulated into a buffer of
348    /// [`XmlStreamConfig::batch_size`] entries and yielded as a
349    /// [`StreamPage`] once the buffer is full. The trailing partial buffer
350    /// (if any) is emitted after the parser hits EOF and all pagination
351    /// rounds drain.
352    ///
353    /// The trait-level `batch_size` argument is intentionally ignored in
354    /// favour of the config field — the config is the user-facing knob the
355    /// README documents, and routing the pipeline-supplied hint through it
356    /// would silently override an explicit config value. `batch_size = 0`
357    /// drains every page into a single emitted page.
358    ///
359    /// Bookmarks are always `None` — the XML source has no
360    /// incremental-replication mode today; pagination only walks the
361    /// API's own page-number / offset cursor.
362    fn stream_pages<'a>(
363        &'a self,
364        context: &'a HashMap<String, Value>,
365        _batch_size: usize,
366    ) -> Pin<Box<dyn Stream<Item = Result<StreamPage, FaucetError>> + Send + 'a>> {
367        let batch_size = self.config.batch_size;
368        let owned_context = context.clone();
369
370        Box::pin(async_stream::try_stream! {
371            let chunk = if batch_size == 0 { usize::MAX } else { batch_size };
372            let initial_capacity = if batch_size == 0 { 1024 } else { batch_size };
373            let mut buffer: Vec<Value> = Vec::with_capacity(initial_capacity);
374            let mut total = 0usize;
375            let mut pages_fetched = 0usize;
376            let mut offset = 0usize;
377            let mut page_number = None;
378            let mut prev_fingerprint: Option<u64> = None;
379
380            if let Some(XmlPagination::PageNumber { start_page, .. }) =
381                &self.config.pagination
382            {
383                page_number = Some(*start_page);
384            }
385
386            loop {
387                if let Some(max) = self.config.max_pages
388                    && pages_fetched >= max
389                {
390                    tracing::warn!("max pages ({max}) reached");
391                    break;
392                }
393
394                let mut params = self.config.query_params.clone();
395                self.apply_pagination_params(&mut params, page_number, offset);
396
397                let xml_text = self.execute_request(&params, &owned_context).await?;
398
399                // Event-driven extraction: only the matched subtree is
400                // ever materialised. The closure pushes into the local
401                // buffer; once it crosses `chunk`, the surrounding loop
402                // can flush, but we can't `yield` from inside the closure,
403                // so we collect this HTTP page's records into a scratch
404                // Vec and then iterate them after.
405                let mut page_records: Vec<Value> = Vec::new();
406                convert::stream_extract(
407                    &xml_text,
408                    self.config.records_element_path.as_deref(),
409                    |rec| page_records.push(rec),
410                )?;
411
412                let record_count = page_records.len();
413                let fingerprint = page_fingerprint(&page_records);
414                pages_fetched += 1;
415
416                // Loop guard: stop when two consecutive pages are identical — a
417                // server ignoring the page/offset parameter (or clamping to the
418                // last page) returns the same non-empty page forever (#146 H4/H5).
419                // Check BEFORE buffering/yielding so the duplicate page's records
420                // are not emitted to the sink a second time (audit #321 M4).
421                if record_count > 0 && prev_fingerprint == Some(fingerprint) {
422                    tracing::warn!(
423                        "XML pagination returned an identical page; stopping to avoid an infinite loop"
424                    );
425                    break;
426                }
427                prev_fingerprint = Some(fingerprint);
428
429                for rec in page_records.drain(..) {
430                    buffer.push(rec);
431                    if buffer.len() >= chunk {
432                        let flush = std::mem::replace(&mut buffer, Vec::with_capacity(initial_capacity));
433                        total += flush.len();
434                        yield StreamPage { records: flush, bookmark: None };
435                    }
436                }
437
438                // Advance pagination using the same rules as
439                // `fetch_all_with_context`.
440                match &self.config.pagination {
441                    Some(XmlPagination::PageNumber { page_size, .. }) => {
442                        if record_count == 0 {
443                            break;
444                        }
445                        if let Some(size) = page_size
446                            && record_count < *size
447                        {
448                            break;
449                        }
450                        page_number = page_number.map(|p| p + 1);
451                    }
452                    Some(XmlPagination::Offset { limit, .. }) => {
453                        if record_count < *limit {
454                            break;
455                        }
456                        offset += record_count;
457                    }
458                    None => break,
459                }
460            }
461
462            if !buffer.is_empty() {
463                total += buffer.len();
464                yield StreamPage { records: buffer, bookmark: None };
465            }
466
467            tracing::info!(
468                records = total,
469                pages = pages_fetched,
470                batch_size,
471                "XML source stream complete",
472            );
473        })
474    }
475
476    fn config_schema(&self) -> serde_json::Value {
477        serde_json::to_value(faucet_core::schema_for!(XmlStreamConfig))
478            .expect("schema serialization")
479    }
480
481    fn dataset_uri(&self) -> String {
482        format!(
483            "{}{}",
484            faucet_core::redact_uri_credentials(&self.config.base_url),
485            self.config.path
486        )
487    }
488}
489
490#[cfg(test)]
491mod tests {
492    use super::*;
493    use faucet_core::Source;
494
495    #[test]
496    fn dataset_uri_combines_base_and_path() {
497        let source = XmlStream::new(XmlStreamConfig::new(
498            "https://soap.example.com",
499            "/api/v1/service",
500        ));
501        assert_eq!(
502            source.dataset_uri(),
503            "https://soap.example.com/api/v1/service"
504        );
505    }
506
507    #[test]
508    fn dataset_uri_redacts_credentials() {
509        let source = XmlStream::new(XmlStreamConfig::new(
510            "https://user:pass@soap.example.com",
511            "/svc",
512        ));
513        assert_eq!(source.dataset_uri(), "https://soap.example.com/svc");
514    }
515
516    #[test]
517    fn default_retry_policy_reproduces_legacy_constants() {
518        let source = XmlStream::new(XmlStreamConfig::new("https://soap.example.com", "/svc"));
519        assert_eq!(source.retry_policy.max_attempts, RETRY_MAX_ATTEMPTS + 1);
520        assert_eq!(source.retry_policy.base, RETRY_BASE_BACKOFF);
521    }
522
523    #[test]
524    fn with_retry_policy_overrides_the_default() {
525        let policy = faucet_core::RetryPolicy {
526            max_attempts: 9,
527            base: Duration::from_secs(7),
528            ..faucet_core::RetryPolicy::default()
529        };
530        let source = XmlStream::new(XmlStreamConfig::new("https://soap.example.com", "/svc"))
531            .with_retry_policy(policy);
532        assert_eq!(source.retry_policy.max_attempts, 9);
533        assert_eq!(source.retry_policy.base, Duration::from_secs(7));
534    }
535}
faucet_source_xml/stream.rs

faucet_source_xml/
stream.rs
