feat(sources): RSS provider 3/4 — freshness engine, partition-per-feed exec, table providers, registration + public egress seam

[gracexmatin]

Phase 3 of 4 stacked PRs splitting #175 (plan Tasks 11-14). Base: rss-m1-phase2; retarget after its merge.

Scope

• RssEngine: the per-feed freshness state machine (TTL hit → serve cached; expired → conditional GET; 304 → re-arm; failure → record last_error capped at 512 chars, keep the stale window). Sole production consumer of egress/fetch/cache — their dead_code allowances drop here. Feed-key discipline: every cache key originates from the engine's own subscription list, never from a predicate value.
• RssScanExec: one partition per subscription, concurrency bounded by max_concurrent, LIMIT gates further fetch launches post-acquire, scan deadline degrades a partition to zero rows + warn instead of failing the query; reset_state overridden so re-execution (e.g. RecursiveQueryExec) starts clean
• Table providers: exact pushdown + pruning for single-feed equality, IN lists, and single-column equality disjunctions — the shape DataFusion's simplifier rewrites short IN lists into, so WHERE feed IN ('a','b') prunes at every list length
• register_rss_tables: one source = one catalog (<name>.main.feeds / <name>.main.items), zero network I/O at registration. Production constructs Arc::new(AllowAll) — no destination filtering, the deliberate OSS stance.
• Public egress seam: register_rss_tables_with_policy is a public (non-test) entry point that accepts an Arc<dyn EgressPolicy>, and the egress trait/types (EgressPolicy, AllowAll, EgressReason, EgressDenied) are re-exported — so an embedder (Skardi Cloud, or an operator running Skardi as a library) can implement a destination filter and inject it in place of AllowAll. OSS still ships only AllowAll; a denying policy in-tree is #[cfg(test)]. An engine test injects a denying DenyList and asserts the refusal reaches feeds.last_error end-to-end.

Verification

cargo test -p skardi --features rss passes at this phase (1018 tests, includes non-rss crate tests); the committed tree compiles standalone.

🤖 Generated with Claude Code

[codecov]

Codecov Report

❌ Patch coverage is 97.60155% with 62 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
crates/skardi/src/sources/providers/rss/engine.rs	97.45%	29 Missing ⚠️
crates/skardi/src/sources/providers/rss/exec.rs	97.11%	16 Missing ⚠️
crates/skardi/src/sources/providers/rss/table.rs	98.23%	10 Missing ⚠️
crates/skardi/src/sources/providers/rss/mod.rs	97.04%	7 Missing ⚠️

📢 Thoughts on this report? Let us know!

[BtXin]

Requesting changes on the execution layer. The current implementation leaks secret-bearing URLs at warning level, runs attacker-controlled parsing synchronously on Tokio workers, permits stale/out-of-order concurrent cache commits, serves cached rows after an egress denial despite the design contract, and exposes a configuration-triggered semaphore panic.

[BtXin]

[P1] Redact the URL from degraded-feed warnings

This warn event includes the complete configured feed URL on every failure. The same PR documentation notes that URLs may contain private query tokens but claims they appear only at debug; in practice ordinary info/warn deployments will export them. Reqwest transport error strings can also carry the current or redirected URL into last_error. Please log only a sanitized host/path identifier here and strip userinfo, query, and fragment from transport errors before storing or logging them.

[BtXin]

[P1] Move feed parsing off Tokio worker threads

parse_feed_document synchronously performs sanitation, feed parsing, DOM construction, and HTML-to-Markdown conversion over attacker-authored input of up to the configured response cap. Because it runs inside the async serve future, several pathological feeds can occupy the runtime workers and stall unrelated server work; the surrounding timeout_at cannot interrupt synchronous CPU work once this call starts. Please move this stage to a bounded blocking/CPU executor and preserve explicit complexity/depth limits and cancellation semantics.

[BtXin]

[P1] Coalesce refreshes or generation-check cache commits

An expired snapshot is taken before the permit, and concurrent scans can independently fetch the same feed and commit in completion order. A slower response representing an older window can overwrite a newer successful fetch; a 304 can also re-arm and label whatever window is currently stored even though its validators came from the earlier snapshot. The cache can therefore regress or claim that a window was revalidated when that response never validated it. Please add per-feed singleflight, or attach a generation to the snapshot and reject/reconcile stale success, failure, and 304 commits. Add a concurrent 200/200 and 200/304 ordering test.

[BtXin]

[P2] Do not serve cached rows after an egress denial

All FetchError variants enter normal stale degradation, so FetchError::Egress serves an existing window as stale-error. The design explicitly requires a policy refusal to produce zero rows, and a dynamic policy or DNS result can deny a destination after this cache was warmed. Please special-case egress denial so the denied subscription contributes no item rows while health records the refusal, and test denial after a successful cached fetch.

[BtXin]

[P2] Cap max_concurrent before constructing the semaphore

Config validation rejects only zero, but Tokio documents that Semaphore::new panics when the permit count exceeds Semaphore::MAX_PERMITS. A valid YAML integer above that threshold therefore aborts registration instead of returning a configuration error, contrary to the no-panics production rule. Clamp the effective value to the subscription count and/or MAX_PERMITS, or reject it during validation, and add an extreme-value regression test.

[BtXin]

Sweep 2: head f656780 predates the previous review comments and has not moved since — all five prior threads are open verbatim at the same lines:

• [P1] engine.rs:581 URL in degraded-feed warn — open. degrade() still logs url = %sub.url at warn; config validation only checks scheme, so userinfo/query tokens pass and get logged whole. (The "Feed URLs are safe to log" comment predates the review — it's a rationale, not a fix.)
• [P1] engine.rs:463 synchronous parse on Tokio worker — open. parse_feed_document still runs sanitation/parse/DOM/markdown inline in the serve future, with the politeness permit held across it, and zero spawn_blocking anywhere under providers/rss/. Compounding: max_response_bytes validation rejects only 0, so the blocking work is operator-unbounded.
• [P1] engine.rs:421 concurrent-scan double-fetch — documented, not fixed. The module doc now declares "No in-flight coalescing … a documented future extension" and record_success remains last-writer-wins with no generation check. If the team decision is to defer, say so on the original thread so it's an explicit call rather than an open P1.
• [P2] engine.rs:498 egress denial serves stale rows — open, and now half-pinned. All FetchError variants including Egress still route through degrade(). The new test egress_blocked_feed_degrades_like_unreachable pins only the no-cached-window case; the contested case — denial after a window was cached → rows still served — remains reachable and untested.
• [P2] engine.rs:290 max_concurrent uncapped before Semaphore::new — open. .max(1) floors, nothing ceilings; a YAML value above Semaphore::MAX_PERMITS panics at registration. Conspicuous because the same commit does clamp MAX_TTL and MAX_SCAN_TIMEOUT.

Stack hygiene: this branch is also based on the pre-fix phase 1 — the proxy/OPML fix commits (c3a8326…649b226) are not in its history. Needs a rebase once phase 1 settles.

One more registration-time note: mod.rs register_catalog silently replaces an existing catalog — a source named datafusion passes duplicate-name validation (which only checks inter-source duplicates) and would shadow every table-level source. Pre-existing pattern shared with sqlite/open_connector, so not new debt from this PR, but worth a guard while the registration code is hot.

Verified clean this round: projection pushdown incl. empty projection, pushdown/prune consistency through the single feed_filter classifier, permit released on cancellation (RAII, tested), no std-Mutex across await in new code, cache byte+entry bounds, catalog published last, statistics correctly Absent, no .unwrap() violations.

[BtXin]

[P3] ScanShared carries LIMIT + deadline state across execute() calls of one plan object

The spent-LIMIT counter and the deadline (stamped Instant::now() at plan construction, exec.rs:207) live in the plan, not the stream. A plan object executed twice without an intervening reset_state() serves zero rows the second time, and a plan executed after scan_timeout has elapsed since planning degrades every partition to zero rows + warn. The reset_state override covers RecursiveQueryExec, and the test at exec.rs:873 pins the behavior as expected — but correctness now depends on every re-executing operator calling reset_state; a plan cached or re-collected outside that path silently shrinks. Deliberate per the docs, so P3 — but worth a doc note on the provider that this leaf is single-execution unless reset.

[gracexmatin]

Cross-referencing a phase-1 review item (PR #180, thread) whose fix lands in this engine PR.

Redirected feeds permanently lose conditional GET. CachedWindow (in the phase-2 cache) stores the validators but not the URL they belong to, and the engine caches them under the feed's original URL. For the ubiquitous http→https 301 feed, the etag/last_modified captured by the fetcher belong to the final hop B, cached under the original URL A. The next scan sends If-None-Match: etag(B) to A, A just 301s again, and the hop to B carries no validators by design — so a redirected feed re-downloads in full every scan and never 304s. Safe (a mismatched etag only ever yields a full 200), just wasteful, and it hits a very common redirect shape.

Decided fix: store the final landing URL alongside the validators and send the conditional request straight to it, so the etag and the address line up and 304 works again — paired with periodically re-probing the original URL to catch a redirect target that has drifted (301 is nominally permanent but drifts / gets misused in practice). The re-probe interval is still open; flagging here so it's designed into the engine's cache/freshness model rather than bolted on later.

[gracexmatin]

Cross-referencing a phase-1 review item (PR #180, thread): the scan deadline must wrap this fetch() call.

The fetcher grants each hop a fresh retry budget, so one fetch() is bounded worst-case at ~300s (6 hops × (3 attempts × 10s request_timeout + 2 waits × 10s MAX_RETRY_WAIT)) — 5× the default scan_timeout_seconds: 60. By design the fetcher only bounds each request/hop, never the whole fetch; the total deadline belongs to this layer.

Right now scan_timeout is stored on RssEngine with only a getter — this await is bare, so scan_timeout_seconds is still enforced by nothing. A malicious or misbehaving server can drag a single feed fetch out to ~300s via slow redirects + retries. Please wrap the whole call, e.g. tokio::time::timeout(self.scan_timeout, self.fetcher.fetch(&sub.url, validators.as_ref())) (not a per-request timeout — the whole future), cancelling the entire fetch when the deadline fires.

[gracexmatin]

Cross-reference from the phase-1 review (#180): max_concurrent was reconciled there as a bound on total fetch parallelism per source, not a per-host bound. One shared semaphore caps in-flight feed fetches (each fetch is internally sequential), so feeds sharing a host can receive up to max_concurrent concurrent requests — there is no per-host accounting. The design spec (§Fetcher and its YAML example) was corrected accordingly, and baseline host-level politeness is understood to rest on honoring Retry-After plus TTL pacing.

Two things for this phase to weigh when it lands here:

1. Wording to align. engine.rs:219 and :382 describe the semaphore as a "Politeness bound" — the same per-host framing feat(sources): RSS provider 1/4 — config, OPML, egress seam (AllowAll default), bounded fetcher #180 removed as over-promising. Suggest rewording to "total fetch-parallelism bound (not per-host)" so the engine code does not re-assert the guarantee the spec just dropped.

2. A real per-host cap is an open engine-phase decision, not promised. If we do want proactive per-host politeness, this semaphore is where it would go — but the design question has to be settled first: count by hostname vs resolved IP vs CDN? Per-hostname misses shared-CDN hosts (Substack/Feedburner/Cloudflare front thousands of feeds behind a few IPs); per-IP tangles with the egress/DNS-resolver layer and can throttle unrelated feeds sharing a CDN IP. Until that is resolved we should not re-advertise a per-host guarantee.