fix: correctness and security fixes from audit

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.

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.
 
 <details>
@@ -127,7 +129,7 @@ The `--output` flag selects the sink by URL scheme: a plain path means file, `ht
 | `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` |
@@ -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`.
@@ -195,7 +197,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 | — |
@@ -216,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 | — |
@@ -253,11 +255,13 @@ 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`, ...).
 
+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:`)
@@ -283,7 +287,7 @@ Secrets can be read from env var or file; file takes precedence when set.
 | `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 | — |

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`)

docs/rest-api.md

@@ -76,15 +76,15 @@ 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.
 
 ---
 
 ### 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.
 

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,14 +170,69 @@ pub async fn get_source_config_handler(
 pub async fn get_global_config_handler(State(state): State<Arc<HealthState>>) -> 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() })),
         ),
     }
 }
 
+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<String> {
+    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<Arc<HealthState>>,
@@ -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();