From 1994ef448c1a4f35cb1a80a41bb4a76a90c69237 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Mon, 26 Oct 2026 14:16:26 +0100 Subject: [PATCH 01/11] fix(api): redact expanded secrets in config endpoints --- docs/rest-api.md | 4 +-- src/api.rs | 89 ++++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 89 insertions(+), 4 deletions(-) diff --git a/docs/rest-api.md b/docs/rest-api.md index 36eab74..de8ea08 100644 --- a/docs/rest-api.md +++ b/docs/rest-api.md @@ -76,7 +76,7 @@ Returns `404` if the source is not in config. If no state store is configured, ` ### GET /api/v1/sources/:id/config -Config for a single source (URL, schedule, auth, pagination, resilience, etc.). Same shape as one entry under `sources` in `helr.yaml`. Auth fields expose env var names (e.g. `token_env`) and paths, not secret values. +Config for a single source (URL, schedule, auth, pagination, resilience, etc.). Same shape as one entry under `sources` in `helr.yaml`. Auth fields expose env var names (e.g. `token_env`) and paths, not secret values. Because `${VAR}` placeholders are expanded when config loads, values that commonly carry secrets are replaced with `[redacted]`: every `headers`, `query_params`, and `body` value, and the password and query values of any URL. **Response:** `200 OK` with source config JSON, or `404 Not Found` if the source id is not in config. @@ -84,7 +84,7 @@ Config for a single source (URL, schedule, auth, pagination, resilience, etc.). ### GET /api/v1/config -Global config (log_level, log_format, state, health, metrics, backpressure, degradation, reload, bulkhead, load_shedding, hooks, audit, etc.). Same shape as `global` in `helr.yaml`. +Global config (log_level, log_format, state, health, metrics, backpressure, degradation, reload, bulkhead, load_shedding, hooks, audit, etc.). Same shape as `global` in `helr.yaml`, with the same redaction as the source config endpoint (for example `output.http.headers` and the password in `state.url`). **Response:** `200 OK` with global config JSON. diff --git a/src/api.rs b/src/api.rs index f60a796..de00e54 100644 --- a/src/api.rs +++ b/src/api.rs @@ -155,7 +155,10 @@ pub async fn get_source_config_handler( } }; match serde_json::to_value(source) { - Ok(v) => (StatusCode::OK, Json(v)), + Ok(mut v) => { + redact_secrets(&mut v); + (StatusCode::OK, Json(v)) + } Err(e) => ( StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() })), @@ -167,7 +170,10 @@ pub async fn get_source_config_handler( pub async fn get_global_config_handler(State(state): State>) -> impl IntoResponse { let config = state.config.read().await; match serde_json::to_value(&config.global) { - Ok(v) => (StatusCode::OK, Json(v)), + Ok(mut v) => { + redact_secrets(&mut v); + (StatusCode::OK, Json(v)) + } Err(e) => ( StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() })), @@ -175,6 +181,58 @@ pub async fn get_global_config_handler(State(state): State>) -> } } +const REDACTED: &str = "[redacted]"; + +/// Config is served after `${VAR}` expansion, so values that commonly carry secrets are masked: +/// header, query param, and body values, plus URL passwords and query values. +fn redact_secrets(value: &mut serde_json::Value) { + match value { + serde_json::Value::Object(map) => { + for (key, v) in map.iter_mut() { + match key.as_str() { + "headers" | "query_params" => { + if let Some(inner) = v.as_object_mut() { + for inner_v in inner.values_mut() { + *inner_v = serde_json::Value::from(REDACTED); + } + } + } + "body" if !v.is_null() => *v = serde_json::Value::from(REDACTED), + _ => redact_secrets(v), + } + } + } + serde_json::Value::Array(arr) => arr.iter_mut().for_each(redact_secrets), + serde_json::Value::String(s) => { + if let Some(redacted) = redact_url(s) { + *s = redacted; + } + } + _ => {} + } +} + +fn redact_url(s: &str) -> Option { + if !s.contains("://") { + return None; + } + let mut url = reqwest::Url::parse(s).ok()?; + if url.password().is_none() && url.query().is_none() { + return None; + } + if url.password().is_some() { + let _ = url.set_password(Some(REDACTED)); + } + if url.query().is_some() { + let pairs: Vec<(String, String)> = url + .query_pairs() + .map(|(k, _)| (k.into_owned(), REDACTED.to_string())) + .collect(); + url.query_pairs_mut().clear().extend_pairs(pairs); + } + Some(url.to_string()) +} + /// POST /api/v1/sources/:id/poll — run one poll tick for the given source. pub async fn trigger_poll_handler( State(state): State>, @@ -423,6 +481,33 @@ sources: assert_eq!(json["url"], "https://example.com/a"); } + #[test] + fn redact_secrets_masks_expanded_values() { + let mut v = serde_json::json!({ + "url": "https://example.com/logs?token=abc&limit=10", + "headers": { "Authorization": "Bearer abc" }, + "query_params": { "api_key": "abc" }, + "body": { "secret": "abc" }, + "state": { "backend": "postgres", "url": "postgres://helr:pw@db:5432/helr" }, + "output": { "http": { "headers": { "X-Api-Key": "abc" }, "batch_size": 100 } }, + "log_level": "info" + }); + redact_secrets(&mut v); + let s = v.to_string(); + assert!(!s.contains("abc"), "{s}"); + assert!(!s.contains(":pw@"), "{s}"); + assert_eq!(v["headers"]["Authorization"], REDACTED); + assert_eq!(v["output"]["http"]["batch_size"], 100); + assert_eq!(v["log_level"], "info"); + assert!(v["url"].as_str().unwrap().contains("limit=")); + assert!( + v["state"]["url"] + .as_str() + .unwrap() + .starts_with("postgres://helr:") + ); + } + #[tokio::test] async fn get_source_config_not_found_returns_404() { let state = api_state_no_poll_deps(); From f3d1e835ea8c3dfbd2906292a49cee5f1b50a906 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Mon, 26 Oct 2026 14:18:11 +0100 Subject: [PATCH 02/11] fix(hooks): let getNextPage override buildRequest on later pages Also treat an empty parseResponse array as an empty page instead of falling back to default parsing, and send buildRequest query params when the hook returns only a query. --- docs/hooks.md | 4 +-- src/hooks.rs | 8 +++-- src/poll/hooks.rs | 45 +++++++++++++------------ tests/integration.rs | 79 ++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 109 insertions(+), 27 deletions(-) diff --git a/docs/hooks.md b/docs/hooks.md index 67c8c1c..5b132b4 100644 --- a/docs/hooks.md +++ b/docs/hooks.md @@ -60,8 +60,8 @@ Each hook is a JavaScript function in your script. You can define one or more; u |------|-------------|--------------| | **getAuth(ctx)** | Before buildRequest; optional | `null` or `{ headers?, cookie?, body?, query? }` — see [Auth hook](#auth-hook-getauth) | | **buildRequest(ctx)** | Before each HTTP request | `{ url?, headers?, query?, body? }` or `null` to use default | -| **parseResponse(ctx, response)** | After each response | Array of `{ ts, source, event, meta? }` | -| **getNextPage(ctx, request, response)** | After parsing; decide next page | `{ url?, body? }` or `null` when no more pages | +| **parseResponse(ctx, response)** | After each response | Array of `{ ts, source, event, meta? }`. An empty array means the page has no events. Return anything other than an array to fall back to default parsing. | +| **getNextPage(ctx, request, response)** | After parsing; decide next page | `{ url?, body? }` or `null` when no more pages. On the following pages, the returned `url` and `body` take precedence over those from `buildRequest`. | | **commitState(ctx, events)** | After a successful poll tick | Object of key-value pairs to write to the state store | ### Context (`ctx`) diff --git a/src/hooks.rs b/src/hooks.rs index 1845922..de48c08 100644 --- a/src/hooks.rs +++ b/src/hooks.rs @@ -301,12 +301,13 @@ pub async fn call_build_request( } /// Call parseResponse(ctx, response). Returns array of { ts, source, event, meta? }. +/// `None` when the hook is not defined or does not return an array; `Some(vec![])` is an empty page. pub async fn call_parse_response( script: &str, ctx: &HookContext, response: &HookResponse, hooks_config: &HooksConfig, -) -> anyhow::Result> { +) -> anyhow::Result>> { let timeout = Duration::from_secs(hooks_config.timeout_secs); let ctx_json = serde_json::to_value(ctx).context("serialize hook ctx")?; let resp_json = serde_json::to_value(response).context("serialize hook response")?; @@ -320,7 +321,7 @@ pub async fn call_parse_response( .await?; let arr = match result.and_then(|v| v.as_array().cloned()) { Some(a) => a, - None => return Ok(vec![]), + None => return Ok(None), }; let mut events = Vec::new(); for item in arr { @@ -351,7 +352,7 @@ pub async fn call_parse_response( meta, }); } - Ok(events) + Ok(Some(events)) } /// Call getNextPage(ctx, request, response). Returns null or { url?, body? }. request is the request that was sent (url, body). @@ -515,6 +516,7 @@ mod tests { let cfg = default_hooks_config(); let events = call_parse_response(script, &ctx, &response, &cfg) .await + .unwrap() .unwrap(); assert_eq!(events.len(), 1); assert_eq!(events[0].ts, "2024-01-01T00:00:00Z"); diff --git a/src/poll/hooks.rs b/src/poll/hooks.rs index 8619f33..4b99213 100644 --- a/src/poll/hooks.rs +++ b/src/poll/hooks.rs @@ -46,12 +46,15 @@ pub(super) async fn poll_with_hooks( ) -> anyhow::Result<()> { let client = build_client(source.resilience.as_ref())?; let max_pages = 100u32; - let mut url = store + let url = store .get(source_id, "next_url") .await? .filter(|s| !s.is_empty()) .unwrap_or_else(|| source.url.clone()); - let mut body_override: Option = source.body.clone(); + let body_override: Option = source.body.clone(); + // Set by getNextPage; takes precedence over buildRequest on later pages. + let mut next_url: Option = None; + let mut next_body: Option = None; let mut all_events: Vec = Vec::new(); let path_for_emit = url.clone(); let label = effective_source_label(source, source_id); @@ -142,22 +145,21 @@ pub(super) async fn poll_with_hooks( }; let build_result = call_build_request(script, &ctx, hooks_config).await?; - let (final_url, final_body) = if let Some(ref br) = build_result { - let u = br.url.as_deref().unwrap_or(url.as_str()); - let b = br.body.clone().or_else(|| body_override.clone()); - (u.to_string(), b) - } else { - (url.clone(), body_override.clone()) - }; + let final_url = next_url + .clone() + .or_else(|| build_result.as_ref().and_then(|br| br.url.clone())) + .unwrap_or_else(|| url.clone()); + let paged_body = next_body + .clone() + .or_else(|| build_result.as_ref().and_then(|br| br.body.clone())); + let final_body = paged_body.clone().or_else(|| body_override.clone()); let auth_has_concrete = auth_result.as_ref().is_some_and(|ar| { ar.headers.is_some() || ar.cookie.is_some() || ar.body.is_some() || ar.query.is_some() }); - let response = if build_result.as_ref().and_then(|r| r.url.as_ref()).is_some() - || build_result - .as_ref() - .and_then(|r| r.headers.as_ref()) - .is_some() + let response = if build_result + .as_ref() + .is_some_and(|r| r.url.is_some() || r.headers.is_some() || r.query.is_some()) || auth_has_concrete { let method = source.method; @@ -226,11 +228,10 @@ pub(super) async fn poll_with_hooks( } } } - let body_to_send = build_result - .as_ref() - .and_then(|br| br.body.clone()) + let body_to_send = paged_body + .clone() .or_else(|| auth_result.as_ref().and_then(|ar| ar.body.clone())) - .or_else(|| final_body.clone()) + .or_else(|| body_override.clone()) .unwrap_or_else(|| serde_json::Value::Object(serde_json::Map::new())); if matches!(method, HttpMethod::Post) { req = req.json(&body_to_send); @@ -274,8 +275,8 @@ pub(super) async fn poll_with_hooks( let events: Vec = match call_parse_response(script, &ctx, &hook_response, hooks_config).await? { - ev if !ev.is_empty() => ev, - _ => { + Some(ev) => ev, + None => { let parsed = parse_events_from_body_for_source(&body_bytes, source)?; parsed .into_iter() @@ -342,10 +343,10 @@ pub(super) async fn poll_with_hooks( match next { Some(n) if n.url.is_some() || n.body.is_some() => { if let Some(u) = n.url { - url = u; + next_url = Some(u); } if let Some(b) = n.body { - body_override = Some(b); + next_body = Some(b); } } _ => break, diff --git a/tests/integration.rs b/tests/integration.rs index fd8dc4d..7177732 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -1243,6 +1243,85 @@ sources: "#); } +/// Hook pagination: getNextPage body wins over buildRequest on later pages, and an empty parseResponse page emits nothing. +#[cfg(feature = "hooks")] +#[tokio::test] +async fn integration_hooks_get_next_page_overrides_build_request() { + use wiremock::matchers::body_partial_json; + let server = MockServer::start().await; + for (skip, ids) in [(0, vec!["a", "b"]), (2, vec!["c", "d"]), (4, vec![])] { + let edges: Vec<_> = ids.iter().map(|id| json!({"node": {"id": id}})).collect(); + Mock::given(method("POST")) + .and(body_partial_json(json!({"skip": skip}))) + .respond_with( + ResponseTemplate::new(200).set_body_json(json!({"data": {"edges": edges}})), + ) + .mount(&server) + .await; + } + + let config_dir = std::env::temp_dir().join("hel_integration_hooks_next_page"); + let _ = std::fs::create_dir_all(&config_dir); + std::fs::write( + config_dir.join("paged.js"), + r#" +function buildRequest(ctx) { + return { body: { skip: parseInt(ctx.state.skip || "0", 10) } }; +} +function parseResponse(ctx, response) { + return response.body.data.edges.map(function (e) { return { ts: "", event: e.node }; }); +} +function getNextPage(ctx, request, response) { + if (response.body.data.edges.length < 2) return null; + return { body: { skip: request.body.skip + 2 } }; +} +function commitState(ctx, events) { + return { skip: String(parseInt(ctx.state.skip || "0", 10) + events.length) }; +} +"#, + ) + .expect("write hook"); + let config_path = config_dir.join("helr.yaml"); + let yaml = format!( + r#" +global: + log_level: error + state: + backend: memory + hooks: + enabled: true + path: "{}" +sources: + paged: + url: "{}/graphql" + method: post + hooks: + script: "paged.js" + resilience: + timeout_secs: 5 +"#, + config_dir.display(), + server.uri() + ); + std::fs::write(&config_path, yaml).expect("write config"); + + let output = run_hel(&["run", "--once"], config_path.to_str().unwrap()); + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8_lossy(&output.stdout); + let ids: Vec = stdout + .lines() + .map(|l| { + let v: serde_json::Value = serde_json::from_str(l).expect("ndjson"); + v["event"]["id"].as_str().unwrap_or("").to_string() + }) + .collect(); + assert_eq!(ids, ["a", "b", "c", "d"]); +} + /// Page/offset pagination: two pages then empty. #[tokio::test] async fn integration_page_offset_pagination_two_pages() { From ad2034679b74dc3904cb87911e6323749b111b61 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Mon, 26 Oct 2026 14:21:13 +0100 Subject: [PATCH 03/11] fix(scheduler): poll each source on its own interval The scheduler slept for the smallest interval across all sources and then polled every source, so a 3600s source ran every 60s next to a 60s one. Track a next-due time per source and run only due sources. Each tick now runs on a config snapshot instead of holding the config read lock, so reload no longer waits for an in-flight poll. --- README.md | 2 +- src/main.rs | 104 ++++++++++++++++++++++++++++++++++++++++++---------- 2 files changed, 86 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index f94fc60..7cca725 100644 --- a/README.md +++ b/README.md @@ -195,7 +195,7 @@ Metrics: `helr_events_dropped_total{source, reason="backpressure"|"max_queue_age | `body` | Request body for POST (JSON). Cursor merged in when using cursor pagination. | object/array | — | | `source_label_key` | Override producer label key for this source | string | — (use global) | | `source_label_value` | Override producer label value for this source | string | source id | -| `schedule.interval_secs` | Poll interval in seconds | number | `60` | +| `schedule.interval_secs` | Poll interval in seconds for this source, measured from the end of its previous poll | number | `60` | | `schedule.jitter_secs` | Random jitter added to interval (seconds) | number | — | | `auth` | Auth config; see Auth types below | object | — | | `pagination` | Pagination config; see Pagination types below | object | — | diff --git a/src/main.rs b/src/main.rs index 43f8312..05df77b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1003,12 +1003,32 @@ async fn run_collector( const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30); + let mut next_due: HashMap = HashMap::new(); + { + let g = config_arc.read().await; + let now = Instant::now(); + for (id, source) in &g.sources { + next_due.insert( + id.clone(), + now + source_delay(source, &g, state_store_fallback_active), + ); + } + } let mut tick = 0u64; 'run: loop { let (delay, dump_enabled) = { let g = config_arc.read().await; + next_due.retain(|id, _| g.sources.contains_key(id)); + let now = Instant::now(); + let earliest = g + .sources + .keys() + .filter(|id| source_filter.is_none_or(|f| f == id.as_str())) + .map(|id| next_due.get(id).copied().unwrap_or(now)) + .min() + .unwrap_or(now + Duration::from_secs(60)); ( - next_delay(&g, state_store_fallback_active), + earliest.saturating_duration_since(now), g.global.dump_on_sigusr1.is_some(), ) }; @@ -1058,8 +1078,19 @@ async fn run_collector( _ = tokio::time::sleep(delay) => {} } - let config_guard = config_arc.read().await; - let config_ref = &*config_guard; + // shortcut: due sources run together, so a slow source delays the next tick of the others; move to per-source tasks if that matters. + let due_config = { + let g = config_arc.read().await; + let now = Instant::now(); + let mut c = g.clone(); + c.sources + .retain(|id, _| next_due.get(id).is_none_or(|due| *due <= now)); + c + }; + if due_config.sources.is_empty() { + continue 'run; + } + let config_ref = &due_config; let store_ref = store.clone(); let source_filter_ref = source_filter; let circuit_store_ref = circuit_store.clone(); @@ -1110,6 +1141,13 @@ async fn run_collector( } } } + let now = Instant::now(); + for (id, source) in &due_config.sources { + next_due.insert( + id.clone(), + now + source_delay(source, &due_config, state_store_fallback_active), + ); + } } if let Err(e) = event_sink.flush() { @@ -1181,21 +1219,15 @@ async fn shutdown_signal() { } } -/// Interval + jitter: min interval across sources, max jitter; delay = interval ± jitter, at least 1s. +/// Delay until a source's next poll: its interval ± its jitter, at least 1s. /// When state_store_fallback_active and degradation.reduced_frequency_multiplier is set, multiplies delay by that factor. -fn next_delay(config: &Config, state_store_fallback_active: bool) -> Duration { - let interval_secs = config - .sources - .values() - .map(|s| s.schedule.interval_secs) - .min() - .unwrap_or(60); - let jitter_secs = config - .sources - .values() - .filter_map(|s| s.schedule.jitter_secs) - .max() - .unwrap_or(0); +fn source_delay( + source: &config::SourceConfig, + config: &Config, + state_store_fallback_active: bool, +) -> Duration { + let interval_secs = source.schedule.interval_secs; + let jitter_secs = source.schedule.jitter_secs.unwrap_or(0); let delta = if jitter_secs > 0 { rand::random_range(-(jitter_secs as i64)..=(jitter_secs as i64)) } else { @@ -1236,6 +1268,39 @@ mod tests { assert_eq!(out, "not json\n"); } + #[test] + fn test_source_delay_uses_each_source_interval() { + let dir = std::env::temp_dir().join("hel_source_delay_test"); + let _ = std::fs::create_dir_all(&dir); + let path = dir.join("helr.yaml"); + std::fs::write( + &path, + r#" +global: {} +sources: + fast: + url: "https://example.com/fast" + schedule: + interval_secs: 60 + slow: + url: "https://example.com/slow" + schedule: + interval_secs: 3600 +"#, + ) + .unwrap(); + let config = Config::load(&path).unwrap(); + let _ = std::fs::remove_file(&path); + assert_eq!( + source_delay(&config.sources["fast"], &config, false).as_secs(), + 60 + ); + assert_eq!( + source_delay(&config.sources["slow"], &config, false).as_secs(), + 3600 + ); + } + #[test] fn test_next_delay_reduced_frequency_when_fallback_active() { let dir = std::env::temp_dir().join("hel_next_delay_test"); @@ -1262,8 +1327,9 @@ sources: .unwrap(); let config = Config::load(&path).unwrap(); let _ = std::fs::remove_file(&path); - let delay_fallback = next_delay(&config, true); - let delay_normal = next_delay(&config, false); + let s1 = &config.sources["s1"]; + let delay_fallback = source_delay(s1, &config, true); + let delay_normal = source_delay(s1, &config, false); assert_eq!( delay_normal.as_secs(), 60, From 9ddf2763f6460d0eb62de09e64b8d61c22e5ae73 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Mon, 26 Oct 2026 14:22:30 +0100 Subject: [PATCH 04/11] fix(poll): resume page and offset pagination from stored state Both strategies stored a watermark but never sent it, so every poll restarted from the first page and re-emitted the same events. Send the watermark, incremental_from, or from value, plus query_params, on every page. --- README.md | 2 + src/poll/page_offset.rs | 22 ++--------- tests/integration.rs | 82 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 88 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 7cca725..2a04e7a 100644 --- a/README.md +++ b/README.md @@ -258,6 +258,8 @@ Secrets can be read from env var or file; file takes precedence when set. - **page_offset:** Query params `page_param` (1-based page) and `limit_param` (page size); `limit` is the value. - **offset:** True offset-based pagination: `offset_param` starts at 0 and increments by `limit` each page (e.g. `offset=0&limit=100`, `offset=100&limit=100`, ...). +For `page_offset` and `offset`, the stored watermark (`state.watermark_param`), `incremental_from`, or `from`, plus `query_params`, are sent on every page so each poll pages through the same filtered result set. + --- #### Resilience (`resilience:`) diff --git a/src/poll/page_offset.rs b/src/poll/page_offset.rs index f1c57c2..9e1afbb 100644 --- a/src/poll/page_offset.rs +++ b/src/poll/page_offset.rs @@ -39,7 +39,7 @@ pub(super) async fn poll_page_offset_pagination( ) -> anyhow::Result<()> { use reqwest::Url; let start = Instant::now(); - let base_url = source.url.as_str(); + let base_url = url_with_first_request_params(&store, source_id, source, &source.url).await?; let mut total_events = 0u64; let mut incremental_max_ts: Option = None; let mut watermark_max_ts: Option = None; @@ -55,18 +55,11 @@ pub(super) async fn poll_page_offset_pagination( tracing::debug!(source = %source_id, delay_secs = secs, "delay between pages"); tokio::time::sleep(Duration::from_secs(secs)).await; } - let mut u = Url::parse(base_url).context("page/offset base url")?; + let mut u = Url::parse(&base_url).context("page/offset base url")?; u.query_pairs_mut() .append_pair(page_param, &page.to_string()); u.query_pairs_mut() .append_pair(limit_param, &limit.to_string()); - if page == 1 - && let Some(ref params) = source.query_params - { - for (k, v) in params { - u.query_pairs_mut().append_pair(k, &v.to_param_value()); - } - } let url = u.to_string(); if let Some(cb) = source .resilience @@ -406,7 +399,7 @@ pub(super) async fn poll_offset_pagination( ) -> anyhow::Result<()> { use reqwest::Url; let start = Instant::now(); - let base_url = source.url.as_str(); + let base_url = url_with_first_request_params(&store, source_id, source, &source.url).await?; let mut total_events = 0u64; let mut incremental_max_ts: Option = None; let mut watermark_max_ts: Option = None; @@ -423,18 +416,11 @@ pub(super) async fn poll_offset_pagination( tracing::debug!(source = %source_id, delay_secs = secs, "delay between pages"); tokio::time::sleep(Duration::from_secs(secs)).await; } - let mut u = Url::parse(base_url).context("offset base url")?; + let mut u = Url::parse(&base_url).context("offset base url")?; u.query_pairs_mut() .append_pair(offset_param, &offset.to_string()); u.query_pairs_mut() .append_pair(limit_param, &limit.to_string()); - if page == 1 - && let Some(ref params) = source.query_params - { - for (k, v) in params { - u.query_pairs_mut().append_pair(k, &v.to_param_value()); - } - } let url = u.to_string(); if let Some(cb) = source .resilience diff --git a/tests/integration.rs b/tests/integration.rs index 7177732..e936fb4 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -1379,6 +1379,88 @@ sources: "#); } +/// Page/offset pagination resumes from the stored watermark and sends it on every page. +#[tokio::test] +async fn integration_page_offset_resumes_from_watermark() { + use wiremock::matchers::{query_param, query_param_is_missing}; + let server = MockServer::start().await; + let pages = [ + ( + None, + "1", + json!([{"id": "a", "ts": "100"}, {"id": "b", "ts": "200"}]), + ), + (None, "2", json!([])), + ( + Some("200"), + "1", + json!([{"id": "c", "ts": "300"}, {"id": "d", "ts": "400"}]), + ), + (Some("200"), "2", json!([{"id": "e", "ts": "500"}])), + (Some("200"), "3", json!([])), + ]; + for (since, page, body) in pages { + let mock = Mock::given(method("GET")).and(query_param("page", page)); + let mock = match since { + Some(s) => mock.and(query_param("since", s)), + None => mock.and(query_param_is_missing("since")), + }; + mock.respond_with(ResponseTemplate::new(200).set_body_json(body)) + .mount(&server) + .await; + } + + let config_dir = std::env::temp_dir().join("hel_integration_page_offset_watermark"); + let _ = std::fs::create_dir_all(&config_dir); + let config_path = config_dir.join("helr.yaml"); + let state_path = config_dir.join("state.db"); + let _ = std::fs::remove_file(&state_path); + let yaml = format!( + r#" +global: + log_level: error + state: + backend: sqlite + path: "{}" +sources: + page-source: + url: "{}/" + pagination: + strategy: page_offset + page_param: page + limit_param: limit + limit: 2 + state: + watermark_field: "ts" + watermark_param: "since" + resilience: + timeout_secs: 5 +"#, + state_path.to_str().unwrap().replace('\\', "/"), + server.uri() + ); + std::fs::write(&config_path, yaml).expect("write config"); + + let ids = |output: std::process::Output| -> Vec { + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8_lossy(&output.stdout) + .lines() + .map(|l| { + let v: serde_json::Value = serde_json::from_str(l).expect("ndjson"); + v["event"]["id"].as_str().unwrap_or("").to_string() + }) + .collect() + }; + let first = ids(run_hel(&["run", "--once"], config_path.to_str().unwrap())); + assert_eq!(first, ["a", "b"]); + let second = ids(run_hel(&["run", "--once"], config_path.to_str().unwrap())); + assert_eq!(second, ["c", "d", "e"]); +} + /// True offset pagination: offset=0 then offset=2 (limit=2), stop when empty. #[tokio::test] async fn integration_offset_pagination_two_pages() { From 0a084dd9e7f41405872b4faaa3f331dd29de4e73 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Mon, 26 Oct 2026 14:27:12 +0100 Subject: [PATCH 05/11] fix(output): do not checkpoint past undelivered HTTP output The HTTP sink queued lines and dropped a batch after a few retries or on any non-5xx response, while the poller had already saved the cursor, so a short collector outage lost events for good. Retry transient failures until delivery, make every later flush fail once a batch is dropped, and flush the output before each checkpoint. Watermark and incremental_from write errors are now propagated, and the per-page Link next_url write honors on_state_write_error. --- README.md | 4 +- src/main.rs | 13 +++- src/output.rs | 165 ++++++++++++++++++++++++++++------------ src/poll/cursor.rs | 9 ++- src/poll/helpers.rs | 19 ++++- src/poll/hooks.rs | 1 + src/poll/link_header.rs | 19 ++++- src/poll/page_offset.rs | 12 ++- src/poll/single_page.rs | 6 +- tests/integration.rs | 64 ++++++++++++++++ 10 files changed, 246 insertions(+), 66 deletions(-) diff --git a/README.md b/README.md index 2a04e7a..b267612 100644 --- a/README.md +++ b/README.md @@ -112,6 +112,8 @@ Configuration is merged in this order (later overrides earlier): The `--output` flag selects the sink by URL scheme: a plain path means file, `http://` or `https://` means HTTP POST, and `nats://` means NATS publish. When omitted, output goes to stdout. The `global.output` config section provides optional tuning for HTTP (batch size, headers, retries) and NATS (subject override). NATS output requires `--features nats`. +Before a source saves its cursor, watermark, or other checkpoint, Helr waits for the output to accept the events emitted so far. If the HTTP sink drops a batch after a non-retryable response (for example 400 or 401), checkpoints stay paused and the affected sources report errors until Helr restarts, so the events are fetched again instead of lost. + **Broken pipe (SIGPIPE):** When stdout is a pipe and the consumer (e.g. Alloy, `helr run | alloy ...`) exits, writes return EPIPE. Helr treats this as **fatal**: the error is logged, `helr_output_errors_total` is incremented, and the process exits with a non-zero code so an orchestrator can restart. Keep the downstream process running, or use file output (`--output /path`) and have the collector tail the file instead.
@@ -141,7 +143,7 @@ The `--output` flag selects the sink by URL scheme: a plain path means file, `ht | `output.http.batch_timeout_ms` | Max milliseconds to wait for a full batch before flushing early | number | `500` | | `output.http.headers` | Extra HTTP headers (e.g. `Authorization: "Bearer ..."`) | map | — | | `output.http.timeout_secs` | HTTP request timeout (seconds) | number | `30` | -| `output.http.max_retries` | Max retries on transient failure (5xx, timeout) with exponential backoff | number | `3` | +| `output.http.max_retries` | Retries on transient failure (5xx, 408, 429, timeout, connect) before each further failed attempt is logged as an error. Delivery keeps retrying with backoff capped at 30s, so polling pauses instead of dropping events. | number | `3` | | `output.nats.subject` | Override the NATS subject parsed from the `--output nats://` URL | string | — | When `api.enabled` is true, GET `/healthz` returns full JSON (version, uptime, per-source status, circuit state, last_error). GET `/readyz` and `/startupz` return version, uptime, and their flags only (no per-source detail). **Readyz semantics:** `/readyz` returns 200 only when (1) output path is writable (or stdout), (2) state store is connected (e.g. SQLite reachable), and (3) at least one source is healthy (circuit not open). The JSON includes `ready`, `output_writable`, `state_store_connected`, and `at_least_one_source_healthy` so you can see which condition failed. When graceful degradation is used (state store fallback to memory), the JSON includes `state_store_fallback_active: true`. diff --git a/src/main.rs b/src/main.rs index 05df77b..6791b64 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1150,8 +1150,17 @@ async fn run_collector( } } - if let Err(e) = event_sink.flush() { - tracing::warn!("flush output: {}", e); + let sink = event_sink.clone(); + match tokio::time::timeout( + SHUTDOWN_TIMEOUT, + tokio::task::spawn_blocking(move || sink.flush()), + ) + .await + { + Ok(Ok(Ok(()))) => {} + Ok(Ok(Err(e))) => tracing::warn!("flush output: {}", e), + Ok(Err(e)) => tracing::warn!("flush output task: {}", e), + Err(_) => tracing::warn!("output did not flush within shutdown timeout"), } tracing::info!("graceful shutdown complete"); Ok(()) diff --git a/src/output.rs b/src/output.rs index ead2d4c..0a659ee 100644 --- a/src/output.rs +++ b/src/output.rs @@ -619,7 +619,7 @@ pub struct HttpSink { enum HttpSinkMsg { Line(String), - Flush(tokio::sync::oneshot::Sender<()>), + Flush(tokio::sync::oneshot::Sender>), } impl HttpSink { @@ -674,7 +674,8 @@ async fn http_sink_worker( max_retries: u32, ) { let mut buf: Vec = Vec::with_capacity(batch_size); - let mut flush_waiters: Vec> = Vec::new(); + // Set once a batch is dropped; every later flush fails so no checkpoint advances past lost events. + let mut dropped: Option = None; let sleep = tokio::time::sleep(batch_timeout); tokio::pin!(sleep); let mut timer_active = false; @@ -691,45 +692,59 @@ async fn http_sink_worker( timer_active = true; } if buf.len() >= batch_size { - http_sink_flush_batch(&client, &url, &mut buf, max_retries).await; + http_sink_send(&client, &url, &mut buf, max_retries, &mut dropped).await; timer_active = false; - for w in flush_waiters.drain(..) { let _ = w.send(()); } } } Some(HttpSinkMsg::Flush(waiter)) => { - if !buf.is_empty() { - http_sink_flush_batch(&client, &url, &mut buf, max_retries).await; - timer_active = false; - } - let _ = waiter.send(()); - for w in flush_waiters.drain(..) { let _ = w.send(()); } + http_sink_send(&client, &url, &mut buf, max_retries, &mut dropped).await; + timer_active = false; + let _ = waiter.send(dropped.clone().map_or(Ok(()), Err)); } None => { - if !buf.is_empty() { - http_sink_flush_batch(&client, &url, &mut buf, max_retries).await; - } - for w in flush_waiters.drain(..) { let _ = w.send(()); } + http_sink_send(&client, &url, &mut buf, max_retries, &mut dropped).await; return; } } } () = &mut sleep, if timer_active => { - if !buf.is_empty() { - http_sink_flush_batch(&client, &url, &mut buf, max_retries).await; - } + http_sink_send(&client, &url, &mut buf, max_retries, &mut dropped).await; timer_active = false; - for w in flush_waiters.drain(..) { let _ = w.send(()); } } } } } -async fn http_sink_flush_batch( +async fn http_sink_send( client: &reqwest::Client, url: &str, buf: &mut Vec, max_retries: u32, + dropped: &mut Option, ) { + if buf.is_empty() { + return; + } + if let Err(e) = http_sink_flush_batch(client, url, buf, max_retries).await { + dropped.get_or_insert(e); + } +} + +fn http_sink_retryable_status(status: reqwest::StatusCode) -> bool { + status.is_server_error() + || status == reqwest::StatusCode::TOO_MANY_REQUESTS + || status == reqwest::StatusCode::REQUEST_TIMEOUT +} + +/// Send one batch. Transient failures (5xx, 408, 429, timeout, connect) are retried until delivery, +/// with backoff capped at 30s, so a downstream outage pauses polling instead of losing events. +/// Returns `Err` only when the batch was dropped after a non-retryable failure. +async fn http_sink_flush_batch( + client: &reqwest::Client, + url: &str, + buf: &mut Vec, + max_retries: u32, +) -> Result<(), String> { let mut body = String::with_capacity(buf.iter().map(|l| l.len() + 1).sum()); for line in buf.iter() { body.push_str(line); @@ -740,43 +755,46 @@ async fn http_sink_flush_batch( let mut attempt = 0u32; loop { - match client.post(url).body(body.clone()).send().await { + let transient = match client.post(url).body(body.clone()).send().await { Ok(resp) if resp.status().is_success() => { tracing::debug!(lines = count, status = %resp.status(), "http sink: batch sent"); - return; - } - Ok(resp) if resp.status().is_server_error() && attempt < max_retries => { - attempt += 1; - let delay = std::time::Duration::from_millis(100 * 2u64.pow(attempt)); - tracing::warn!( - status = %resp.status(), attempt, max_retries, - "http sink: retryable error, backing off {delay:?}", - ); - tokio::time::sleep(delay).await; + return Ok(()); } + Ok(resp) if http_sink_retryable_status(resp.status()) => resp.status().to_string(), Ok(resp) => { tracing::error!( status = %resp.status(), lines = count, "http sink: non-retryable response, dropping batch", ); metrics::record_output_error("http_sink"); - return; - } - Err(e) if (e.is_timeout() || e.is_connect()) && attempt < max_retries => { - attempt += 1; - let delay = std::time::Duration::from_millis(100 * 2u64.pow(attempt)); - tracing::warn!( - error = %e, attempt, max_retries, - "http sink: transient error, backing off {delay:?}", - ); - tokio::time::sleep(delay).await; + return Err(format!( + "dropped {count} lines after non-retryable response {}", + resp.status() + )); } + Err(e) if e.is_timeout() || e.is_connect() => e.to_string(), Err(e) => { tracing::error!(error = %e, lines = count, "http sink: send failed, dropping batch"); metrics::record_output_error("http_sink"); - return; + return Err(format!("dropped {count} lines after send error: {e}")); } + }; + attempt = attempt.saturating_add(1); + let delay = std::time::Duration::from_millis(100u64 << attempt.min(9)) + .min(std::time::Duration::from_secs(30)); + if attempt > max_retries { + tracing::error!( + error = %transient, attempt, lines = count, + "http sink: delivery still failing, retrying in {delay:?}", + ); + metrics::record_output_error("http_sink"); + } else { + tracing::warn!( + error = %transient, attempt, max_retries, + "http sink: transient error, backing off {delay:?}", + ); } + tokio::time::sleep(delay).await; } } @@ -793,13 +811,10 @@ impl EventSink for HttpSink { .send(HttpSinkMsg::Flush(tx)) .map_err(|_| anyhow::anyhow!("http sink worker closed"))?; let rt = self.rt.clone(); - std::thread::scope(|s| { - s.spawn(move || { - rt.block_on(rx).ok(); - }) - .join() - .map_err(|_| anyhow::anyhow!("http sink flush thread panicked")) - }) + std::thread::scope(|s| s.spawn(move || rt.block_on(rx)).join()) + .map_err(|_| anyhow::anyhow!("http sink flush thread panicked"))? + .map_err(|_| anyhow::anyhow!("http sink worker closed"))? + .map_err(|e| anyhow::anyhow!("http sink {e}; checkpoints stay paused until restart")) } } @@ -1296,4 +1311,58 @@ mod tests { // First 2 calls return 503, third succeeds => 3 total calls. assert_eq!(call_count.load(Ordering::SeqCst), 3); } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn http_sink_keeps_retrying_past_max_retries() { + use wiremock::matchers::method; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(503)) + .up_to_n_times(3) + .mount(&mock_server) + .await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&mock_server) + .await; + + let cfg = crate::config::HttpOutputConfig { + max_retries: 0, + ..Default::default() + }; + let sink = HttpSink::new(mock_server.uri(), &cfg).unwrap(); + sink.write_line(r#"{"msg":"kept"}"#).unwrap(); + sink.flush().unwrap(); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn http_sink_flush_fails_after_dropped_batch() { + use wiremock::matchers::method; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(401)) + .up_to_n_times(1) + .mount(&mock_server) + .await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&mock_server) + .await; + + let cfg = crate::config::HttpOutputConfig::default(); + let sink = HttpSink::new(mock_server.uri(), &cfg).unwrap(); + sink.write_line(r#"{"msg":"lost"}"#).unwrap(); + let err = sink.flush().unwrap_err().to_string(); + assert!(err.contains("401"), "{err}"); + sink.write_line(r#"{"msg":"later"}"#).unwrap(); + assert!( + sink.flush().is_err(), + "flush must keep failing after a drop" + ); + } } diff --git a/src/poll/cursor.rs b/src/poll/cursor.rs index 51a0ec5..ac1722a 100644 --- a/src/poll/cursor.rs +++ b/src/poll/cursor.rs @@ -496,6 +496,9 @@ pub(super) async fn poll_cursor_pagination( } metrics::record_events(source_id, emitted_count); let checkpoint_per_page = source.checkpoint != Some(CheckpointTiming::EndOfTick); + if checkpoint_per_page { + flush_before_checkpoint(&event_sink).await?; + } match next_cursor { Some(c) => { if checkpoint_per_page { @@ -526,12 +529,14 @@ pub(super) async fn poll_cursor_pagination( } } } + flush_before_checkpoint(&event_sink).await?; if source.checkpoint == Some(CheckpointTiming::EndOfTick) && let Some(ref v) = pending_cursor { store_set_or_skip(&store, source_id, source, global, "cursor", v).await?; } - store_incremental_from_after_poll(&store, source_id, source, global, incremental_max_ts).await; - store_watermark_after_poll(&store, source_id, source, global, watermark_max_ts).await; + store_incremental_from_after_poll(&store, source_id, source, global, incremental_max_ts) + .await?; + store_watermark_after_poll(&store, source_id, source, global, watermark_max_ts).await?; Ok(()) } diff --git a/src/poll/helpers.rs b/src/poll/helpers.rs index 305f373..bb34e1a 100644 --- a/src/poll/helpers.rs +++ b/src/poll/helpers.rs @@ -140,6 +140,15 @@ pub(crate) fn emit_event_line( }) } +/// Wait until emitted events are accepted by the output, so a checkpoint never advances past undelivered events. +pub(crate) async fn flush_before_checkpoint(event_sink: &Arc) -> anyhow::Result<()> { + let sink = event_sink.clone(); + tokio::task::spawn_blocking(move || sink.flush()) + .await + .context("output flush task")? + .context("output flush before checkpoint") +} + /// Set state key; on error, fail or skip checkpoint per source config or global degradation.emit_without_checkpoint. pub(crate) async fn store_set_or_skip( store: &Arc, @@ -298,13 +307,14 @@ pub(crate) async fn store_incremental_from_after_poll( source: &SourceConfig, global: &GlobalConfig, max_ts: Option, -) { +) -> anyhow::Result<()> { if let Some(ref inc) = source.incremental_from { let val = max_ts.unwrap_or_default(); if !val.is_empty() { - let _ = store_set_or_skip(store, source_id, source, global, &inc.state_key, &val).await; + store_set_or_skip(store, source_id, source, global, &inc.state_key, &val).await?; } } + Ok(()) } /// Store per-source watermark state after poll when state.watermark_field is configured. @@ -314,13 +324,14 @@ pub(crate) async fn store_watermark_after_poll( source: &SourceConfig, global: &GlobalConfig, max_ts: Option, -) { +) -> anyhow::Result<()> { if let Some(key) = watermark_state_key(source) { let val = max_ts.unwrap_or_default(); if !val.is_empty() { - let _ = store_set_or_skip(store, source_id, source, global, key, &val).await; + store_set_or_skip(store, source_id, source, global, key, &val).await?; } } + Ok(()) } pub(crate) fn status_class(status: u16) -> &'static str { diff --git a/src/poll/hooks.rs b/src/poll/hooks.rs index 4b99213..e5ff8cf 100644 --- a/src/poll/hooks.rs +++ b/src/poll/hooks.rs @@ -377,6 +377,7 @@ pub(super) async fn poll_with_hooks( }, headers: source.headers.clone(), }; + flush_before_checkpoint(&event_sink).await?; let to_commit = call_commit_state(script, &commit_ctx, &all_events, hooks_config).await?; for (key, value) in to_commit { store_set_or_skip(&store, source_id, source, global, &key, &value).await?; diff --git a/src/poll/link_header.rs b/src/poll/link_header.rs index e5ad5de..67b0c7c 100644 --- a/src/poll/link_header.rs +++ b/src/poll/link_header.rs @@ -363,10 +363,21 @@ pub(super) async fn poll_link_header( let checkpoint_per_page = source.checkpoint != Some(CheckpointTiming::EndOfTick); let hit_max_bytes = max_bytes.is_some() && total_bytes > max_bytes.unwrap(); + if checkpoint_per_page { + flush_before_checkpoint(&event_sink).await?; + } if let Some(next) = next_url { let absolute = base_url.join(&next).context("resolve next URL")?; if checkpoint_per_page { - store.set(source_id, "next_url", absolute.as_str()).await?; + store_set_or_skip( + &store, + source_id, + source, + global, + "next_url", + absolute.as_str(), + ) + .await?; } pending_next_url = Some(absolute.to_string()); if hit_max_bytes { @@ -401,12 +412,14 @@ pub(super) async fn poll_link_header( break; } } + flush_before_checkpoint(&event_sink).await?; if source.checkpoint == Some(CheckpointTiming::EndOfTick) && let Some(ref v) = pending_next_url { store_set_or_skip(&store, source_id, source, global, "next_url", v).await?; } - store_incremental_from_after_poll(&store, source_id, source, global, incremental_max_ts).await; - store_watermark_after_poll(&store, source_id, source, global, watermark_max_ts).await; + store_incremental_from_after_poll(&store, source_id, source, global, incremental_max_ts) + .await?; + store_watermark_after_poll(&store, source_id, source, global, watermark_max_ts).await?; Ok(()) } diff --git a/src/poll/page_offset.rs b/src/poll/page_offset.rs index 9e1afbb..ac82476 100644 --- a/src/poll/page_offset.rs +++ b/src/poll/page_offset.rs @@ -370,9 +370,11 @@ pub(super) async fn poll_page_offset_pagination( tracing::warn!(source = %source_id, "reached max_pages {}", max_pages); } } + flush_before_checkpoint(&event_sink).await?; store_set_or_skip(&store, source_id, source, global, "next_url", "").await?; - store_incremental_from_after_poll(&store, source_id, source, global, incremental_max_ts).await; - store_watermark_after_poll(&store, source_id, source, global, watermark_max_ts).await; + store_incremental_from_after_poll(&store, source_id, source, global, incremental_max_ts) + .await?; + store_watermark_after_poll(&store, source_id, source, global, watermark_max_ts).await?; Ok(()) } @@ -731,8 +733,10 @@ pub(super) async fn poll_offset_pagination( tracing::warn!(source = %source_id, "reached max_pages {}", max_pages); } } + flush_before_checkpoint(&event_sink).await?; store_set_or_skip(&store, source_id, source, global, "next_url", "").await?; - store_incremental_from_after_poll(&store, source_id, source, global, incremental_max_ts).await; - store_watermark_after_poll(&store, source_id, source, global, watermark_max_ts).await; + store_incremental_from_after_poll(&store, source_id, source, global, incremental_max_ts) + .await?; + store_watermark_after_poll(&store, source_id, source, global, watermark_max_ts).await?; Ok(()) } diff --git a/src/poll/single_page.rs b/src/poll/single_page.rs index 454b425..800bd32 100644 --- a/src/poll/single_page.rs +++ b/src/poll/single_page.rs @@ -336,9 +336,11 @@ pub(super) async fn poll_single_page( } metrics::record_events(source_id, emitted_count); + flush_before_checkpoint(&event_sink).await?; store_set_or_skip(&store, source_id, source, global, "next_url", "").await?; - store_incremental_from_after_poll(&store, source_id, source, global, incremental_max_ts).await; - store_watermark_after_poll(&store, source_id, source, global, watermark_max_ts).await; + store_incremental_from_after_poll(&store, source_id, source, global, incremental_max_ts) + .await?; + store_watermark_after_poll(&store, source_id, source, global, watermark_max_ts).await?; tracing::info!( source = %source_id, events = emitted_count, diff --git a/tests/integration.rs b/tests/integration.rs index e936fb4..c4e55ec 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -1322,6 +1322,70 @@ sources: assert_eq!(ids, ["a", "b", "c", "d"]); } +/// A rejected HTTP output batch fails the tick and leaves the source checkpoint where it was. +#[tokio::test] +async fn integration_http_output_failure_does_not_checkpoint() { + let source = MockServer::start().await; + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "items": [{"id": "x", "ts": "100"}] + }))) + .mount(&source) + .await; + let sink = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(401)) + .mount(&sink) + .await; + + let config_dir = std::env::temp_dir().join("hel_integration_http_output_failure"); + let _ = std::fs::create_dir_all(&config_dir); + let config_path = config_dir.join("helr.yaml"); + let state_path = config_dir.join("state.db"); + let _ = std::fs::remove_file(&state_path); + let yaml = format!( + r#" +global: + log_level: error + state: + backend: sqlite + path: "{}" +sources: + wm-source: + url: "{}/" + state: + watermark_field: "ts" + watermark_param: "since" + resilience: + timeout_secs: 5 +"#, + state_path.to_str().unwrap().replace('\\', "/"), + source.uri() + ); + std::fs::write(&config_path, yaml).expect("write config"); + + let config = config_path.to_str().unwrap(); + let out = std::process::Command::new(hel_bin()) + .args(["run", "--config", config, "--once", "--output", &sink.uri()]) + .output() + .expect("run helr"); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!( + stderr.contains("401"), + "expected output error in logs: {stderr}" + ); + + let show = std::process::Command::new(hel_bin()) + .args(["state", "--config", config, "show", "wm-source"]) + .output() + .expect("run helr state show"); + let stdout = String::from_utf8_lossy(&show.stdout); + assert!( + !stdout.contains("100"), + "watermark must not advance: {stdout}" + ); +} + /// Page/offset pagination: two pages then empty. #[tokio::test] async fn integration_page_offset_pagination_two_pages() { From 8f5510822fbb903b4ef57444a7331080e4251ba2 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Mon, 26 Oct 2026 14:28:10 +0100 Subject: [PATCH 06/11] fix(pagination): keep Link next URLs on the source origin Refuse to follow a next URL on another origin, since every request carries the source's auth. Parse Link headers by segments so commas inside URIs and multi-value rel attributes work. --- README.md | 2 +- src/pagination.rs | 33 +++++++++++++++++++-------- src/poll/link_header.rs | 9 ++++++++ tests/integration.rs | 49 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 83 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index b267612..b80112c 100644 --- a/README.md +++ b/README.md @@ -255,7 +255,7 @@ Secrets can be read from env var or file; file takes precedence when set. | `page_offset` | `page_param`, `limit_param`, `limit` | `max_pages` | — | | `offset` | `offset_param`, `limit_param`, `limit` | `max_pages` | — | -- **link_header:** Next URL from `Link` header (e.g. `rel="next"`). +- **link_header:** Next URL from `Link` header (e.g. `rel="next"`). A next URL on a different origin (scheme, host, port) than the source `url` fails the poll instead of receiving the source's credentials. - **cursor:** Cursor from response JSON at `cursor_path`; sent as query param `cursor_param` (GET) or merged into body (POST). - **page_offset:** Query params `page_param` (1-based page) and `limit_param` (page size); `limit` is the value. - **offset:** True offset-based pagination: `offset_param` starts at 0 and increments by `limit` each page (e.g. `offset=0&limit=100`, `offset=100&limit=100`, ...). diff --git a/src/pagination.rs b/src/pagination.rs index 5dcd261..57e8f51 100644 --- a/src/pagination.rs +++ b/src/pagination.rs @@ -19,20 +19,24 @@ pub fn next_link_from_headers(headers: &HeaderMap, rel: &str) -> Option } /// Parse a single Link header value string. Format: ; rel="next"[, ; rel="prev"] +/// URIs may contain commas, so links are split on `<...>` rather than on `,`. fn parse_link_header(s: &str, want_rel: &str) -> Option { - for part in s.split(',') { - let part = part.trim(); - let (uri, rest) = part.strip_prefix('<')?.split_once('>')?; - let rest = rest.trim().trim_start_matches(';').trim(); - for param in rest.split(';') { - let param = param.trim(); - if let Some(rel_val) = param.strip_prefix("rel=") { - let rel_val = rel_val.trim().trim_matches('"').trim(); - if rel_val.eq_ignore_ascii_case(want_rel) { + let mut rest = s; + while let Some(start) = rest.find('<') { + let (uri, tail) = rest[start + 1..].split_once('>')?; + let params_end = tail.find('<').unwrap_or(tail.len()); + for param in tail[..params_end].split([';', ',']) { + if let Some(rel_val) = param.trim().strip_prefix("rel=") { + let rel_val = rel_val.trim().trim_matches('"'); + if rel_val + .split_whitespace() + .any(|r| r.eq_ignore_ascii_case(want_rel)) + { return Some(uri.trim().to_string()); } } } + rest = &tail[params_end..]; } None } @@ -68,6 +72,17 @@ mod tests { ); } + #[test] + fn parse_next_with_comma_in_uri_and_multi_value_rel() { + let h = header_map( + r#"; rel="prev", ; rel="next last""#, + ); + assert_eq!( + next_link_from_headers(&h, "next"), + Some("https://api.example.com/logs?ids=3,4&after=z".into()) + ); + } + #[test] fn no_next_returns_none() { let h = header_map(r#"; rel="self""#); diff --git a/src/poll/link_header.rs b/src/poll/link_header.rs index 67b0c7c..439413f 100644 --- a/src/poll/link_header.rs +++ b/src/poll/link_header.rs @@ -368,6 +368,15 @@ pub(super) async fn poll_link_header( } if let Some(next) = next_url { let absolute = base_url.join(&next).context("resolve next URL")?; + let source_origin = reqwest::Url::parse(&source.url) + .context("parse source url")? + .origin(); + if absolute.origin() != source_origin { + anyhow::bail!( + "refusing to follow Link next URL to a different origin than the source url: {}", + absolute.origin().ascii_serialization() + ); + } if checkpoint_per_page { store_set_or_skip( &store, diff --git a/tests/integration.rs b/tests/integration.rs index c4e55ec..0bb08d9 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -1642,6 +1642,55 @@ sources: ); } +/// A Link next URL on another origin is not followed, so source credentials stay on the source host. +#[tokio::test] +async fn integration_link_header_refuses_cross_origin_next() { + let other = MockServer::start().await; + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!([]))) + .expect(0) + .mount(&other) + .await; + let server = MockServer::start().await; + let next_link = format!(r#"<{}/steal>; rel="next""#, other.uri()); + Mock::given(method("GET")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("Link", next_link.as_str()) + .set_body_json(json!([{"id": "m1"}])), + ) + .mount(&server) + .await; + + let config_dir = std::env::temp_dir().join("hel_integration_link_cross_origin"); + let _ = std::fs::create_dir_all(&config_dir); + let config_path = config_dir.join("helr.yaml"); + let yaml = format!( + r#" +global: + log_level: error + state: + backend: memory +sources: + link-source: + url: "{}/" + headers: + Authorization: "Bearer secret" + pagination: + strategy: link_header + rel: next + resilience: + timeout_secs: 5 +"#, + server.uri() + ); + std::fs::write(&config_path, yaml).expect("write config"); + + let output = run_hel(&["run", "--once"], config_path.to_str().unwrap()); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("different origin"), "stderr: {stderr}"); +} + /// Recorded/fixture: wiremock returns Okta-shaped JSON; parser accepts it. #[tokio::test] async fn integration_fixture_okta_shaped_events() { From 6184f7201ee3dc29df1ff5f963a9a036357cb471 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Mon, 26 Oct 2026 14:29:04 +0100 Subject: [PATCH 07/11] fix(auth): use a random DPoP jti and cap token expires_in DPoP proofs used source_id plus the iat second as jti, and the resource-request variant appended Instant::now().elapsed(), which is always about zero, so proofs within one second reused the same jti and servers that track jti reject the replay. A huge expires_in from the token server overflowed Instant and panicked. --- src/dpop.rs | 12 ++++++++++++ src/oauth2.rs | 29 ++++++++++++++++++++++++----- src/retry.rs | 7 +------ 3 files changed, 37 insertions(+), 11 deletions(-) diff --git a/src/dpop.rs b/src/dpop.rs index fa762c9..eda5996 100644 --- a/src/dpop.rs +++ b/src/dpop.rs @@ -54,6 +54,11 @@ fn uri_without_query_and_fragment(url: &str) -> anyhow::Result { Ok(u.to_string()) } +/// Random proof id: servers reject a reused `jti`, and several proofs can share the same `iat` second. +pub fn new_jti() -> String { + format!("{:032x}", rand::random::()) +} + /// Build DPoP proof JWT (RFC 9449): htm, htu (no query/fragment), iat, jti; optional nonce and ath when provided. pub fn build_dpop_proof( method: &str, @@ -112,6 +117,13 @@ pub fn build_dpop_proof( mod tests { use super::*; + #[test] + fn new_jti_is_unique_per_call() { + let a = new_jti(); + assert_eq!(a.len(), 32); + assert_ne!(a, new_jti()); + } + #[test] fn test_dpop_proof_shape() { let key = RsaPrivateKey::new(&mut rsa::rand_core::OsRng, 2048).unwrap(); diff --git a/src/oauth2.rs b/src/oauth2.rs index 4299aff..0839d4f 100644 --- a/src/oauth2.rs +++ b/src/oauth2.rs @@ -155,7 +155,7 @@ pub async fn get_oauth_token( .duration_since(UNIX_EPOCH) .context("system time")? .as_secs(); - let jti = format!("{}-{}", source_id, iat); + let jti = crate::dpop::new_jti(); let proof = build_dpop_proof("POST", token_url.as_str(), &key, &jti, iat, None, None)?; token_req = token_req.header("DPoP", proof); } @@ -191,7 +191,7 @@ pub async fn get_oauth_token( .duration_since(UNIX_EPOCH) .context("system time")? .as_secs(); - let jti = format!("{}-{}", source_id, iat); + let jti = crate::dpop::new_jti(); let proof = build_dpop_proof( "POST", token_url.as_str(), @@ -242,7 +242,7 @@ pub async fn get_oauth_token( .get("expires_in") .and_then(|v| v.as_u64()) .unwrap_or(3600); - let expires_at = now + Duration::from_secs(expires_in); + let expires_at = token_expires_at(now, expires_in); { let mut g = cache.write().await; g.insert(source_id.to_string(), (access_token.clone(), expires_at)); @@ -272,7 +272,7 @@ pub async fn get_oauth_token( .get("expires_in") .and_then(|v| v.as_u64()) .unwrap_or(3600); - let expires_at = now + Duration::from_secs(expires_in); + let expires_at = token_expires_at(now, expires_in); { let mut g = cache.write().await; @@ -282,6 +282,12 @@ pub async fn get_oauth_token( Ok(access_token) } +/// Cap `expires_in` from the token server so a huge value cannot overflow `Instant`. +fn token_expires_at(now: Instant, expires_in: u64) -> Instant { + const MAX_TOKEN_LIFETIME_SECS: u64 = 365 * 24 * 3600; + now + Duration::from_secs(expires_in.min(MAX_TOKEN_LIFETIME_SECS)) +} + /// Build client_assertion JWT for private_key_jwt (RFC 7523): iss/sub=client_id, aud=token_url, RS256. fn build_client_assertion( client_id: &str, @@ -442,7 +448,7 @@ pub async fn get_google_sa_token( .get("expires_in") .and_then(|v| v.as_u64()) .unwrap_or(3600); - let expires_at = now + Duration::from_secs(expires_in); + let expires_at = token_expires_at(now, expires_in); { let mut g = cache.write().await; g.insert(source_id.to_string(), (access_token.clone(), expires_at)); @@ -450,3 +456,16 @@ pub async fn get_google_sa_token( debug!(source = %source_id, expires_in, "google service account token obtained"); Ok(access_token) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn token_expires_at_caps_huge_expires_in() { + let now = Instant::now(); + let at = token_expires_at(now, u64::MAX); + assert_eq!(at - now, Duration::from_secs(365 * 24 * 3600)); + assert_eq!(token_expires_at(now, 60) - now, Duration::from_secs(60)); + } +} diff --git a/src/retry.rs b/src/retry.rs index dbf93ea..e33a070 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -197,12 +197,7 @@ async fn dpop_proof_for_request( .duration_since(UNIX_EPOCH) .context("system time")? .as_secs(); - let jti = format!( - "{}-{}-{}", - source_id, - iat, - std::time::Instant::now().elapsed().as_nanos() - ); + let jti = crate::dpop::new_jti(); let proof = build_dpop_proof(method, url, &key, &jti, iat, nonce, access_token)?; Ok(Some(proof)) } From 57d9439791e3b0a385b5170270431b2a02aa6a74 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Mon, 26 Oct 2026 14:30:58 +0100 Subject: [PATCH 08/11] fix(resilience): correct circuit failure rate, fractional rate limits, and backoff math The circuit failure rate divided consecutive failures by every request since startup, so it almost never opened on mixed failures; evaluate it over windows of minimum_requests. A rate limit of 0.5 rps was rounded up to 1 rps. Retry backoff panicked on a negative or NaN multiplier, and jitter could exceed max_backoff_secs. --- README.md | 2 +- src/circuit.rs | 66 +++++++++++++++++++++++++++++++++++++++++-------- src/poll/mod.rs | 34 +++++++++++++++++++------ src/retry.rs | 56 +++++++++++++++++++++++++++++++---------- 4 files changed, 126 insertions(+), 32 deletions(-) diff --git a/README.md b/README.md index b80112c..f1ac5bd 100644 --- a/README.md +++ b/README.md @@ -287,7 +287,7 @@ For `page_offset` and `offset`, the stored watermark (`state.watermark_param`), | `circuit_breaker.half_open_timeout_secs` | Seconds before half-open probe | number | `60` | | `circuit_breaker.reset_timeout_secs` | Max time in open state (open duration = min(half_open_timeout_secs, reset_timeout_secs)) | number | — | | `circuit_breaker.failure_rate_threshold` | Optional: open when failure rate ≥ this (0.0–1.0); requires `minimum_requests` | number | — | -| `circuit_breaker.minimum_requests` | Minimum requests before evaluating `failure_rate_threshold` | number | — | +| `circuit_breaker.minimum_requests` | Window size for `failure_rate_threshold`: the rate is evaluated over each consecutive window of this many requests | number | — | | `rate_limit.respect_headers` | Use Retry-After or reset header on 429 (see `headers.reset_header`) | boolean | `true` | | `rate_limit.page_delay_secs` | Delay between pagination requests (seconds) | number | — | | `rate_limit.headers` | Header names for limit/remaining/reset (when API uses different names) | object | — | diff --git a/src/circuit.rs b/src/circuit.rs index e72685c..74e97ef 100644 --- a/src/circuit.rs +++ b/src/circuit.rs @@ -11,10 +11,12 @@ use tracing::{info, warn}; #[derive(Debug, Clone)] pub enum CircuitState { - /// Closed: normal operation. failures/requests used for count and rate thresholds. + /// Closed: normal operation. `failures` is consecutive (count threshold); `requests` and + /// `window_failures` cover the current window of `minimum_requests` (rate threshold). Closed { failures: u32, requests: u32, + window_failures: u32, }, Open { open_until: Instant, @@ -63,6 +65,7 @@ pub async fn allow_request( Some(CircuitState::Closed { failures: 0, requests: 0, + window_failures: 0, }), ), Some(CircuitState::Closed { .. }) => (true, None), @@ -102,6 +105,7 @@ pub async fn record_result( let state = g.get(source_id).cloned().unwrap_or(CircuitState::Closed { failures: 0, requests: 0, + window_failures: 0, }); let now = Instant::now(); let open_duration_secs = config @@ -109,16 +113,23 @@ pub async fn record_result( .map(|r| std::cmp::min(config.half_open_timeout_secs, r)) .unwrap_or(config.half_open_timeout_secs); let new_state = match state { - CircuitState::Closed { failures, requests } => { + CircuitState::Closed { + failures, + requests, + window_failures, + } => { let requests = requests + 1; let failures = if success { 0 } else { failures + 1 }; - let open_on_count = failures >= config.failure_threshold; - let open_on_rate = config.minimum_requests.is_some() - && config.failure_rate_threshold.is_some() - && requests >= config.minimum_requests.unwrap_or(0) - && (failures as f64 / requests as f64) - >= config.failure_rate_threshold.unwrap_or(0.0); - if !success && (open_on_count || open_on_rate) { + let window_failures = window_failures + u32::from(!success); + let open_on_count = !success && failures >= config.failure_threshold; + let window_full = config + .minimum_requests + .is_some_and(|min| requests >= min.max(1)); + let open_on_rate = window_full + && config + .failure_rate_threshold + .is_some_and(|t| (window_failures as f64 / requests as f64) >= t); + if open_on_count || open_on_rate { let open_until = now + std::time::Duration::from_secs(open_duration_secs); warn!( source = %source_id, @@ -128,8 +139,18 @@ pub async fn record_result( "circuit opened" ); CircuitState::Open { open_until } + } else if window_full { + CircuitState::Closed { + failures, + requests: 0, + window_failures: 0, + } } else { - CircuitState::Closed { failures, requests } + CircuitState::Closed { + failures, + requests, + window_failures, + } } } CircuitState::Open { open_until } => CircuitState::Open { open_until }, @@ -141,6 +162,7 @@ pub async fn record_result( CircuitState::Closed { failures: 0, requests: 0, + window_failures: 0, } } else { CircuitState::HalfOpen { successes: s } @@ -264,6 +286,30 @@ mod tests { assert!(err.open_until > Instant::now()); } + #[tokio::test] + async fn test_circuit_opens_on_interleaved_failure_rate_after_long_run() { + let store = new_circuit_store(); + let config = CircuitBreakerConfig { + enabled: true, + failure_threshold: 100, + success_threshold: 1, + half_open_timeout_secs: 60, + reset_timeout_secs: None, + failure_rate_threshold: Some(0.5), + minimum_requests: Some(10), + }; + for _ in 0..1000 { + allow_request(&store, "s1", &config).await.unwrap(); + record_result(&store, "s1", &config, true).await; + } + for i in 0..10 { + allow_request(&store, "s1", &config).await.unwrap(); + record_result(&store, "s1", &config, i % 2 == 0).await; + } + let err = allow_request(&store, "s1", &config).await.unwrap_err(); + assert!(err.open_until > Instant::now()); + } + #[tokio::test] async fn test_circuit_does_not_open_on_rate_before_minimum_requests() { let store = new_circuit_store(); diff --git a/src/poll/mod.rs b/src/poll/mod.rs index 2c64963..11e0cd6 100644 --- a/src/poll/mod.rs +++ b/src/poll/mod.rs @@ -45,6 +45,16 @@ static HOOK_AUTH_CACHE: std::sync::LazyLock = /// Run one poll tick for all sources (or only those matching source_filter). /// Sources are polled concurrently (one task per source). +/// One request every 1/rps seconds, so fractional rates like 0.5 are honored instead of rounded up. +fn rate_limit_quota(rps: f64, burst_size: Option) -> Option { + if !(rps > 0.0 && rps.is_finite()) { + return None; + } + let period = Duration::try_from_secs_f64(1.0 / rps).ok()?; + let burst = burst_size.unwrap_or((rps.ceil() as u32).max(1)).max(1); + Some(Quota::with_period(period)?.allow_burst(NonZeroU32::new(burst)?)) +} + #[allow(clippy::too_many_arguments)] pub async fn run_one_tick( config: &Config, @@ -114,14 +124,7 @@ pub async fn run_one_tick( .as_ref() .and_then(|r| r.rate_limit.as_ref()) .and_then(|r| { - let rps = r - .max_requests_per_second - .filter(|&x| x > 0.0 && x.is_finite())?; - let rps_u = (rps.ceil() as u32).max(1); - let rps_nz = NonZeroU32::new(rps_u)?; - let burst = r.burst_size.unwrap_or(rps_u).max(1); - let burst_nz = NonZeroU32::new(burst).unwrap_or(NonZeroU32::MIN); - let quota = Quota::per_second(rps_nz).allow_burst(burst_nz); + let quota = rate_limit_quota(r.max_requests_per_second?, r.burst_size)?; Some(Arc::new(RateLimiter::direct(quota))) }); let poll_tick_secs = source @@ -387,6 +390,21 @@ mod tests { use super::parse::*; use crate::config::{GlobalConfig, SourceConfig}; + #[test] + fn rate_limit_quota_honors_fractional_rps() { + let q = super::rate_limit_quota(0.5, None).unwrap(); + assert_eq!(q.replenish_interval(), std::time::Duration::from_secs(2)); + assert_eq!(q.burst_size().get(), 1); + let q = super::rate_limit_quota(10.0, Some(3)).unwrap(); + assert_eq!( + q.replenish_interval(), + std::time::Duration::from_millis(100) + ); + assert_eq!(q.burst_size().get(), 3); + assert!(super::rate_limit_quota(0.0, None).is_none()); + assert!(super::rate_limit_quota(f64::NAN, None).is_none()); + } + #[test] fn test_parse_events_from_value_top_level_array() { let v = serde_json::json!([{"id": 1}, {"id": 2}]); diff --git a/src/retry.rs b/src/retry.rs index e33a070..5fffaeb 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -33,22 +33,30 @@ pub fn is_retryable_status_with_codes( } /// Compute backoff duration for the given attempt (0 = first retry). Applies jitter when retry.jitter is set. +/// The result never exceeds max_backoff_secs, and odd configs (negative or NaN multiplier) cannot panic. fn backoff_duration(retry: &RetryConfig, attempt: u32) -> Duration { - let secs = (retry.initial_backoff_secs as f64) * retry.multiplier.powi(attempt as i32); - let capped = retry + let max_secs = retry .max_backoff_secs - .map(|max| secs.min(max as f64)) - .unwrap_or(secs); - let base_secs = capped.min(u64::MAX as f64); - let delay_secs = if let Some(j) = retry.jitter { - let j = j.clamp(0.0, 1.0); - let mut rng = rand::rng(); - let factor = 1.0 + rng.random_range(-j..=j); - (base_secs * factor).max(0.0) - } else { - base_secs + .map(|m| m as f64) + .unwrap_or(u64::MAX as f64); + let clamp = |s: f64| { + if s.is_nan() { + max_secs + } else { + s.clamp(0.0, max_secs) + } + }; + let base_secs = + clamp((retry.initial_backoff_secs as f64) * retry.multiplier.powi(attempt as i32)); + let delay_secs = match retry.jitter { + Some(j) if j.is_finite() && j > 0.0 => { + let j = j.min(1.0); + let factor = 1.0 + rand::rng().random_range(-j..=j); + clamp(base_secs * factor) + } + _ => base_secs, }; - Duration::from_secs_f64(delay_secs) + Duration::try_from_secs_f64(delay_secs).unwrap_or(Duration::MAX) } /// Parse Retry-After (delta-seconds or HTTP-date) and optionally reset header (e.g. X-RateLimit-Reset). @@ -481,6 +489,28 @@ mod tests { assert_eq!(backoff_duration(&retry, 10), Duration::from_secs(30)); // capped } + #[test] + fn test_backoff_duration_odd_configs_do_not_panic() { + let base = RetryConfig { + max_attempts: 3, + initial_backoff_secs: 10, + max_backoff_secs: Some(10), + multiplier: -2.0, + jitter: Some(1.0), + retryable_status_codes: None, + }; + for attempt in 0..4 { + assert!(backoff_duration(&base, attempt) <= Duration::from_secs(10)); + } + let nan = RetryConfig { + multiplier: f64::NAN, + max_backoff_secs: None, + jitter: None, + ..base + }; + let _ = backoff_duration(&nan, 2); + } + #[test] fn test_retry_after_delta_seconds() { let mut headers = HeaderMap::new(); From 17e850ae78307b0fb49e6213c8c3061dc580d0d8 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Mon, 26 Oct 2026 14:32:40 +0100 Subject: [PATCH 09/11] fix(output): keep truncated lines valid NDJSON within max_line_bytes Truncation cut the serialized line and appended "...", producing invalid JSON, and counted characters instead of bytes, so multibyte events still exceeded the limit. Replace event with the longest prefix of its JSON text that fits and mark meta.truncated. --- README.md | 2 +- src/event.rs | 4 ++++ src/poll/helpers.rs | 43 ++++++++++++++++++++++++++++++++----------- src/poll/mod.rs | 41 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 78 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index f1ac5bd..a2753c2 100644 --- a/README.md +++ b/README.md @@ -218,7 +218,7 @@ Metrics: `helr_events_dropped_total{source, reason="backpressure"|"max_queue_age | `max_response_bytes` | Fail if a single response body exceeds this (bytes) | number | — | | `on_invalid_utf8` | When response body is not valid UTF-8 | `replace`, `escape`, `fail` | — | | `max_line_bytes` | Max size of one emitted NDJSON line (bytes) | number | — | -| `max_line_bytes_behavior` | When a line exceeds `max_line_bytes` | `truncate`, `skip`, `fail` | — | +| `max_line_bytes_behavior` | When a line exceeds `max_line_bytes`. `truncate` keeps the line valid JSON: `event` becomes a string holding the start of the event's JSON, and `meta.truncated` is `true`. | `truncate`, `skip`, `fail` | — | | `checkpoint` | When to write state | `end_of_tick`, `per_page` | — | | `on_state_write_error` | When state store write fails (e.g. disk full) | `fail`, `skip_checkpoint` | `fail` | | `incremental_from` | Store latest event timestamp in state and send as query param on first request (e.g. Slack `oldest`); see below | object | — | diff --git a/src/event.rs b/src/event.rs index e577114..c9fee51 100644 --- a/src/event.rs +++ b/src/event.rs @@ -25,6 +25,9 @@ pub struct EventMeta { pub request_id: Option, #[serde(skip_serializing_if = "Option::is_none")] pub id: Option, + /// Set when `event` was replaced by a truncated JSON string to fit `max_line_bytes`. + #[serde(skip_serializing_if = "Option::is_none")] + pub truncated: Option, } impl EmittedEvent { @@ -38,6 +41,7 @@ impl EmittedEvent { cursor: None, request_id: None, id: None, + truncated: None, }, } } diff --git a/src/poll/helpers.rs b/src/poll/helpers.rs index bb34e1a..f3e5cdb 100644 --- a/src/poll/helpers.rs +++ b/src/poll/helpers.rs @@ -97,17 +97,7 @@ pub(crate) fn emit_event_line( .unwrap_or(MaxEventBytesBehavior::Fail) { MaxEventBytesBehavior::Truncate => { - let max_usize = max as usize; - let truncated: String = if max_usize >= 3 { - format!( - "{}...", - line.chars() - .take(max_usize.saturating_sub(3)) - .collect::() - ) - } else { - line.chars().take(max_usize).collect() - }; + let truncated = truncate_event_line(emitted, label_key, max as usize)?; event_sink .write_line_from_source(Some(source_id), &truncated) .inspect_err(|_e| { @@ -140,6 +130,37 @@ pub(crate) fn emit_event_line( }) } +/// Replace `event` with a prefix of its JSON text so the line stays valid NDJSON within `max` bytes. +fn truncate_event_line( + emitted: &EmittedEvent, + label_key: &str, + max: usize, +) -> anyhow::Result { + let text = emitted.event.to_string(); + let mut stub = emitted.clone(); + stub.meta.truncated = Some(true); + // Escaping makes output length non-linear in the prefix, so binary search the longest fit. + let (mut lo, mut hi) = (0usize, text.len().min(max)); + let mut best = None; + while lo <= hi { + let mut mid = lo + (hi - lo) / 2; + while !text.is_char_boundary(mid) { + mid -= 1; + } + stub.event = serde_json::Value::String(text[..mid].to_string()); + let line = stub.to_ndjson_line_with_label_key(label_key)?; + if line.len() <= max { + best = Some(line); + lo = lo + (hi - lo) / 2 + 1; + } else if mid == 0 { + break; + } else { + hi = mid - 1; + } + } + best.ok_or_else(|| anyhow::anyhow!("max_line_bytes {} is smaller than the event envelope", max)) +} + /// Wait until emitted events are accepted by the output, so a checkpoint never advances past undelivered events. pub(crate) async fn flush_before_checkpoint(event_sink: &Arc) -> anyhow::Result<()> { let sink = event_sink.clone(); diff --git a/src/poll/mod.rs b/src/poll/mod.rs index 11e0cd6..b9124be 100644 --- a/src/poll/mod.rs +++ b/src/poll/mod.rs @@ -390,6 +390,47 @@ mod tests { use super::parse::*; use crate::config::{GlobalConfig, SourceConfig}; + #[test] + fn truncated_line_is_valid_json_within_max_bytes() { + use crate::config::MaxEventBytesBehavior; + use crate::output::EventSink; + use std::sync::{Arc, Mutex}; + + struct Capture(Mutex>); + impl EventSink for Capture { + fn write_line(&self, line: &str) -> anyhow::Result<()> { + self.0.lock().unwrap().push(line.to_string()); + Ok(()) + } + } + let source: SourceConfig = serde_yaml_ng::from_str( + "url: https://example.com\nmax_line_bytes: 200\nmax_line_bytes_behavior: truncate\n", + ) + .unwrap(); + assert_eq!( + source.max_line_bytes_behavior, + Some(MaxEventBytesBehavior::Truncate) + ); + let global: GlobalConfig = serde_yaml_ng::from_str("{}").unwrap(); + let capture = Arc::new(Capture(Mutex::new(Vec::new()))); + let sink: Arc = capture.clone(); + let event = crate::event::EmittedEvent::new( + "2024-01-01T00:00:00Z".into(), + "s".into(), + "/".into(), + serde_json::json!({ "msg": "é\"".repeat(200) }), + ); + emit_event_line(&global, "s", &source, &sink, &event).unwrap(); + let line = capture.0.lock().unwrap()[0].clone(); + assert!((196..=200).contains(&line.len()), "{} bytes", line.len()); + let v: serde_json::Value = serde_json::from_str(&line).unwrap(); + assert_eq!(v["meta"]["truncated"], true); + assert!( + v["event"].as_str().unwrap().starts_with("{\"msg\":\"é"), + "{line}" + ); + } + #[test] fn rate_limit_quota_honors_fractional_rps() { let q = super::rate_limit_quota(0.5, None).unwrap(); From 16f43cfcd5e0076d8a71075f8112053ccd1c31a3 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Mon, 26 Oct 2026 14:33:45 +0100 Subject: [PATCH 10/11] fix(dedupe): forget a failed poll's IDs so the refetch is emitted Dedupe recorded each ID before the event was delivered. When the poll then failed (output flush, write error, timeout) the next tick refetched the same events from the unchanged checkpoint and dedupe dropped them as duplicates. IDs added during a poll are now kept only if it succeeds. --- src/dedupe.rs | 33 +++++++++++++++++++++++++++++++++ src/poll/mod.rs | 9 ++++++--- 2 files changed, 39 insertions(+), 3 deletions(-) diff --git a/src/dedupe.rs b/src/dedupe.rs index 398a2e5..71223df 100644 --- a/src/dedupe.rs +++ b/src/dedupe.rs @@ -9,6 +9,8 @@ pub struct LruDedupe { capacity: u64, order: VecDeque, seen: HashSet, + /// IDs added during the current poll; forgotten again if the poll fails before its checkpoint. + pending: Vec, } impl LruDedupe { @@ -17,6 +19,7 @@ impl LruDedupe { capacity: capacity.max(1), order: VecDeque::new(), seen: HashSet::new(), + pending: Vec::new(), } } @@ -29,6 +32,7 @@ impl LruDedupe { return true; // duplicate } self.seen.insert(id.clone()); + self.pending.push(id.clone()); self.order.push_back(id); while self.order.len() as u64 > self.capacity { if let Some(evicted) = self.order.pop_front() { @@ -37,6 +41,17 @@ impl LruDedupe { } false // new } + + /// Keep (`ok`) or forget the IDs added since the last call. + pub fn finish_poll(&mut self, ok: bool) { + let pending = std::mem::take(&mut self.pending); + if ok || pending.is_empty() { + return; + } + let pending: HashSet = pending.into_iter().collect(); + self.seen.retain(|id| !pending.contains(id)); + self.order.retain(|id| !pending.contains(id)); + } } /// Store of per-source LRU dedupes. Shared across poll ticks. @@ -57,6 +72,13 @@ pub async fn seen_and_add(store: &DedupeStore, source_id: &str, id: String, capa dedupe.seen_and_add(id) } +/// Commit this poll's IDs when it succeeded; roll them back when it failed so a refetch emits them again. +pub async fn finish_poll(store: &DedupeStore, source_id: &str, ok: bool) { + if let Some(d) = store.write().await.get_mut(source_id) { + d.finish_poll(ok); + } +} + #[cfg(test)] mod tests { use super::*; @@ -95,6 +117,17 @@ mod tests { assert!(!d.seen_and_add("a".to_string())); } + #[test] + fn lru_dedupe_failed_poll_forgets_its_ids() { + let mut d = LruDedupe::new(10); + assert!(!d.seen_and_add("kept".to_string())); + d.finish_poll(true); + assert!(!d.seen_and_add("retried".to_string())); + d.finish_poll(false); + assert!(d.seen_and_add("kept".to_string())); + assert!(!d.seen_and_add("retried".to_string())); + } + #[tokio::test] async fn store_seen_and_add_per_source() { let store = new_dedupe_store(); diff --git a/src/poll/mod.rs b/src/poll/mod.rs index b9124be..24efcfe 100644 --- a/src/poll/mod.rs +++ b/src/poll/mod.rs @@ -14,7 +14,7 @@ mod streaming; use crate::circuit::CircuitStore; use crate::client::build_client; use crate::config::{Config, GlobalConfig, PaginationConfig, SourceConfig}; -use crate::dedupe::DedupeStore; +use crate::dedupe::{self, DedupeStore}; use crate::dpop::DPoPKeyCache; use crate::metrics; use crate::oauth2::OAuth2TokenCache; @@ -132,6 +132,7 @@ pub async fn run_one_tick( .as_ref() .and_then(|r| r.timeouts.as_ref()) .and_then(|t| t.poll_tick_secs); + let dedupe_for_finish = dedupe_store.clone(); let h = tokio::spawn(async move { let _source_permit = match &global_sources_semaphore_clone { Some(s) => Some( @@ -155,14 +156,16 @@ pub async fn run_one_tick( rate_limiter, request_semaphore, ); - match poll_tick_secs { + let result = match poll_tick_secs { Some(secs) => match tokio::time::timeout(Duration::from_secs(secs), poll_fut).await { Ok(inner) => inner, Err(_) => Err(anyhow::anyhow!("poll tick timed out after {}s", secs)), }, None => poll_fut.await, - } + }; + dedupe::finish_poll(&dedupe_for_finish, &source_id_key, result.is_ok()).await; + result }); handles.push((source_id, h)); } From e70abd5e97c56e0e5952b1e36349a45a4824b289 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Mon, 26 Oct 2026 14:38:33 +0100 Subject: [PATCH 11/11] fix(state): use verified TLS for Postgres and reconnect after drops The Postgres state store connected with NoTls, so sslmode=require URLs (managed Postgres) could not connect and prefer sent credentials in plaintext. Connect through rustls with the platform verifier, matching the HTTP client. A dropped connection also failed every later state call until restart; reconnect when the client reports it closed. --- Cargo.lock | 71 +++++++++++++++++++++++++++++++++++++++++++++++++++- Cargo.toml | 3 +++ README.md | 2 +- src/state.rs | 65 ++++++++++++++++++++++++++++++++++------------- 4 files changed, 122 insertions(+), 19 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 12c707e..8500a75 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1033,10 +1033,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" dependencies = [ "const-oid 0.9.6", + "der_derive", + "flagset", "pem-rfc7468", "zeroize", ] +[[package]] +name = "der_derive" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8034092389675178f570469e6c3b0465d3d30b4505c294a6550db47f3c17ad18" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "deranged" version = "0.5.9" @@ -1317,6 +1330,12 @@ version = "0.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" +[[package]] +name = "flagset" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7ac824320a75a52197e8f2d787f6a38b6718bb6897a35142d749af3c0e8f4fe" + [[package]] name = "float16" version = "0.1.7" @@ -1651,6 +1670,8 @@ dependencies = [ "reqwest 0.13.5", "rsa", "rusqlite", + "rustls", + "rustls-platform-verifier", "serde", "serde_json", "serde_yaml_ng", @@ -1660,6 +1681,7 @@ dependencies = [ "testcontainers", "tokio", "tokio-postgres", + "tokio-postgres-rustls", "tokio-util", "tracing", "tracing-subscriber", @@ -4135,7 +4157,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", "windows-sys 0.61.2", @@ -4276,6 +4298,27 @@ version = "1.13.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fd3ca314f692efd6c868f8408f53fe444634a845f96c028b97d35f6a1f79f0ee" +[[package]] +name = "tls_codec" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0de2e01245e2bb89d6f05801c564fa27624dbd7b1846859876c7dad82e90bf6b" +dependencies = [ + "tls_codec_derive", + "zeroize", +] + +[[package]] +name = "tls_codec_derive" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d2e76690929402faae40aebdda620a2c0e25dd6d3b9afe48867dfd95991f4bd" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "tokio" version = "1.53.3" @@ -4340,6 +4383,20 @@ dependencies = [ "whoami", ] +[[package]] +name = "tokio-postgres-rustls" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c2ad44aa0ae96db89c4742212ed41645b2f597311ff6e1945542a4d9fadc2fb" +dependencies = [ + "rustls", + "sha2 0.11.1", + "tokio", + "tokio-postgres", + "tokio-rustls", + "x509-cert", +] + [[package]] name = "tokio-rustls" version = "0.26.6" @@ -5198,6 +5255,18 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" +[[package]] +name = "x509-cert" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1301e935010a701ae5f8655edc0ad17c44bad3ac5ce8c39185f75453b720ae94" +dependencies = [ + "const-oid 0.9.6", + "der", + "spki", + "tls_codec", +] + [[package]] name = "xattr" version = "1.6.1" diff --git a/Cargo.toml b/Cargo.toml index 0d5b4c8..eccb128 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,6 +34,9 @@ sysinfo = "0.38" governor = "0.10" redis = { version = "1.0", features = ["tokio-comp", "connection-manager"] } tokio-postgres = "0.7" +tokio-postgres-rustls = "0.14" +rustls-platform-verifier = "0.7" +rustls = { version = "0.23", default-features = false, features = ["aws_lc_rs", "std", "tls12"] } tokio-util = { version = "0.7", features = ["io", "io-util"], optional = true } # Optional NATS output sink. diff --git a/README.md b/README.md index a2753c2..b64f218 100644 --- a/README.md +++ b/README.md @@ -129,7 +129,7 @@ Before a source saves its cursor, watermark, or other checkpoint, Helr waits for | `source_label_value` | Value for producer label in Helr's own logs | string | — (effective: `helr`) | | `state.backend` | State store backend | `sqlite`, `memory`, `redis`, `postgres` | — | | `state.path` | Path to state file (SQLite) | string | `./helr-state.db` (when backend is sqlite) | -| `state.url` | Connection URL for Redis (`redis://...`) or Postgres (`postgres://...`) | string | — (required when backend is redis or postgres) | +| `state.url` | Connection URL for Redis (`redis://...`) or Postgres (`postgres://...`). For Postgres, `sslmode` is `prefer` by default (TLS when the server offers it) or `require`; the server certificate is verified against the system trust store, so use `sslmode=disable` for a local server with a self-signed certificate. Helr reconnects if the Postgres connection drops. | string | — (required when backend is redis or postgres) | | `api.enabled` | Enable API and health HTTP server | boolean | `false` | | `api.address` | API/health server bind address | string | `0.0.0.0` | | `api.port` | API/health server port | number | `8080` | diff --git a/src/state.rs b/src/state.rs index 7c415b9..99085b4 100644 --- a/src/state.rs +++ b/src/state.rs @@ -5,6 +5,7 @@ //! **Single-writer assumption:** Only one Helr process should use a given state store (e.g. one SQLite file). //! For multi-instance deployments use Redis or Postgres state backend. +use anyhow::Context; use async_trait::async_trait; use std::collections::HashMap; use std::path::Path; @@ -309,21 +310,15 @@ impl StateStore for RedisStateStore { /// Postgres-backed state store. Same schema as SQLite: (source_id, key, value, updated_at). pub struct PostgresStateStore { - client: tokio_postgres::Client, - _connection: tokio::task::JoinHandle<()>, + url: String, + client: tokio::sync::Mutex>, } impl PostgresStateStore { /// Connect to Postgres at url (e.g. `postgres://user:pass@localhost/dbname`). Creates helr_state table if missing. + /// TLS follows the URL's `sslmode` (default `prefer`) and verifies certificates against the system trust store. pub async fn connect(url: &str) -> anyhow::Result { - let (client, connection) = tokio_postgres::connect(url, tokio_postgres::NoTls) - .await - .map_err(|e| anyhow::anyhow!("postgres connect: {}", e))?; - let _connection = tokio::spawn(async move { - if let Err(e) = connection.await { - tracing::error!("postgres connection error: {}", e); - } - }); + let client = Self::open(url).await?; client .batch_execute( r#" @@ -339,17 +334,49 @@ impl PostgresStateStore { .await .map_err(|e| anyhow::anyhow!("postgres init: {}", e))?; Ok(Self { - client, - _connection, + url: url.to_string(), + client: tokio::sync::Mutex::new(Arc::new(client)), }) } + + async fn open(url: &str) -> anyhow::Result { + use rustls_platform_verifier::BuilderVerifierExt; + let tls_config = rustls::ClientConfig::builder_with_provider(Arc::new( + rustls::crypto::aws_lc_rs::default_provider(), + )) + .with_safe_default_protocol_versions() + .and_then(|b| b.with_platform_verifier()) + .map_err(|e| anyhow::anyhow!("postgres tls config: {}", e))? + .with_no_client_auth(); + let tls = tokio_postgres_rustls::MakeRustlsConnect::new(tls_config); + let (client, connection) = tokio_postgres::connect(url, tls) + .await + .context("postgres connect")?; + tokio::spawn(async move { + if let Err(e) = connection.await { + tracing::error!("postgres connection error: {}", e); + } + }); + Ok(client) + } + + /// Current client, reconnecting first if the connection closed (server restart, failover, idle timeout). + async fn client(&self) -> anyhow::Result> { + let mut guard = self.client.lock().await; + if guard.is_closed() { + tracing::warn!("postgres state connection closed, reconnecting"); + *guard = Arc::new(Self::open(&self.url).await?); + } + Ok(guard.clone()) + } } #[async_trait] impl StateStore for PostgresStateStore { async fn get(&self, source_id: &str, key: &str) -> anyhow::Result> { let row: Option = self - .client + .client() + .await? .query_opt( "SELECT value FROM helr_state WHERE source_id = $1 AND key = $2", &[&source_id, &key], @@ -364,7 +391,8 @@ impl StateStore for PostgresStateStore { .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_secs() as i64; - self.client + self.client() + .await? .execute( "INSERT INTO helr_state (source_id, key, value, updated_at) VALUES ($1, $2, $3, $4) ON CONFLICT (source_id, key) DO UPDATE SET value = $3, updated_at = $4", @@ -377,7 +405,8 @@ impl StateStore for PostgresStateStore { async fn list_keys(&self, source_id: &str) -> anyhow::Result> { let rows = self - .client + .client() + .await? .query( "SELECT key FROM helr_state WHERE source_id = $1 ORDER BY key", &[&source_id], @@ -389,7 +418,8 @@ impl StateStore for PostgresStateStore { async fn list_sources(&self) -> anyhow::Result> { let rows = self - .client + .client() + .await? .query( "SELECT DISTINCT source_id FROM helr_state ORDER BY source_id", &[], @@ -400,7 +430,8 @@ impl StateStore for PostgresStateStore { } async fn clear_source(&self, source_id: &str) -> anyhow::Result<()> { - self.client + self.client() + .await? .execute("DELETE FROM helr_state WHERE source_id = $1", &[&source_id]) .await .map_err(|e| anyhow::anyhow!("postgres: {}", e))?;