fix: reliability hardening — pool isolation, client recovery, SSRF, loop safety

[Ismaël Mejía (iemejia)]

Summary

Addresses 7 reliability and safety issues from the audit (2 Critical, 2 High, 3 Medium). All fixes compile cleanly and pass unit tests.

C4 (Critical): Isolate epoch/extension polling onto dedicated pool

Problem: The management pool (default 6 connections) was shared between epoch heartbeats and activity work. Under high concurrency (>6 simultaneous orchestrations), all connections could be consumed by status updates, starving the heartbeat loop — causing the worker to interpret a pool-timeout as "extension dropped" and spuriously shut down.

Fix: Created a separate 1-connection poll_pool used exclusively for extension-existence checks and epoch sentinel heartbeats. Activity work remains on the management pool but can no longer starve health checks.

C5 (Critical): Recoverable client on connection failure

Problem: OnceLock<Client> is permanent — once set, it can never be replaced. If connections die (PG restart, pg_terminate_backend, network partition), all df.start()/df.signal()/df.cancel() from existing sessions fail permanently.

Fix: Replaced with thread_local! { RefCell<Option<Client>> } that detects connection-level errors (broken pipe, pool timeout, connection refused) and resets the client so the next call transparently re-creates the pool.

H5 (High): Wrap std::env::set_var in unsafe block

Problem: set_var becomes unsafe in Rust 2024 edition due to potential data races.

Fix: Wrapped both call sites with unsafe {} blocks and SAFETY documentation explaining why they're sound (single-threaded tokio runtime, no concurrent env readers, called before pool creation).

H6 (High): Block CGNAT range in SSRF protection

Problem: 100.64.0.0/10 (RFC 6598 Carrier-Grade NAT) was missing from the IPv4 blocklist. In cloud environments, CGNAT addresses are sometimes used for internal metadata/service endpoints exploitable via DNS rebinding.

Fix: Added [100, b, ..] if (64..=127).contains(&b) to check_blocked_ipv4() with corresponding test (edge cases at 100.63.x and 100.128.x).

M1 (Medium): Row-count limit on $name.* expansion

Problem: expand_row_set() had no limit on rows, allowing unbounded SQL string allocation from large result sets.

Fix: Added MAX_ROWSET_EXPANSION = 10,000 — expansions exceeding this fail with a clear error suggesting pagination.

M7 (Medium): Maximum loop iteration safeguard

Problem: Infinite loops (df.loop() without break or condition) ran indefinitely with no upper bound.

Fix: Added loop_iteration counter to FunctionInput (persisted across continue_as_new generations). After 100,000 iterations (~27 hours at 1s minimum), the loop fails with a clear error.

M8 (Medium): Malformed loop condition fails instead of infinite loop

Problem: If a LOOP node's condition JSON was unparseable, if let Ok(config) silently fell through, creating an infinite loop with no exit condition.

Fix: Changed to explicit match that returns an error on parse failure, surfacing the problem immediately.

Verification

• cargo check — no errors, only pre-existing extract_host warning
• cargo test --lib types:: — 29/29 pass
• cargo test --lib ssrf:: — 37/37 pass (including new CGNAT test)
• cargo fmt — clean

[Copilot]

Pull request overview

This PR hardens pg_durable against several reliability and safety failure modes in both the background worker (duroxide runtime) and the user-session client path, plus adds guardrails around loop execution, SSRF protections, and result expansion limits.

Changes:

• Isolated extension/epoch polling onto a dedicated 1-connection poll_pool in the BGW to avoid starvation under load.
• Added safety limits/guardrails: $name.* row-set expansion cap, CGNAT SSRF blocklist, loop iteration maximum, and fail-fast behavior for malformed loop condition JSON.
• Made the per-backend Duroxide client recoverable after connection-level failures (reset-on-error so later calls can reinitialize).

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
src/worker.rs	Adds dedicated polling pool and wraps env var mutation in unsafe with a SAFETY note.
src/client.rs	Replaces permanent OnceLock<Client> with resettable per-backend thread_local client and adds connection-error detection.
src/types.rs	Adds $name.* expansion row limit and serialization support for loop iteration counter; adds unit tests.
src/ssrf.rs	Blocks RFC6598 CGNAT range and adds tests.
src/orchestrations/execute_function_graph.rs	Adds loop iteration persistence/limit and fails malformed loop config JSON instead of silently looping.
src/lib.rs	Adds pg_tests for the new regression scenarios (client error heuristic, CGNAT blocking, row-set expansion, loop_iteration serde, malformed loop config validate).
src/dsl.rs	Initializes new FunctionInput.loop_iteration field at df.start() time.

Comments suppressed due to low confidence (2)

src/worker.rs:223

• run_duroxide_runtime() closes mgmt_pool but never closes the newly added poll_pool. This can leave the dedicated polling connection open longer than necessary and makes shutdown less orderly. Consider closing both pools when run_duroxide_runtime() exits.

        run_until_extension_dropped_or_shutdown(
            &poll_pool,
            duroxide_runtime,
            EXTENSION_DROP_POLL_INTERVAL,
            SHUTDOWN_CHECK_INTERVAL,
            epoch_id.as_deref(),
        )
        .await;
    }

src/orchestrations/execute_function_graph.rs:590

• execute_loop_node() always uses FunctionInput as the continue_as_new payload. That matches ExecuteFunctionGraph::execute(), but ExecuteSubtree expects a different JSON shape (graph/node_id/results/vars/label). If a LOOP node ever runs inside a JOIN/RACE branch (i.e., inside ExecuteSubtree), continue_as_new will restart the subtree orchestration with an incompatible input and fail with "Missing graph". Either explicitly reject LOOP nodes in ExecuteSubtree with a clear error, or change the subtree orchestration input/continue_as_new strategy so loops are supported in parallel branches.

    // duroxide 0.1.1: continue_as_new returns an awaitable future - return it directly
    return ctx
        .continue_as_new(serde_json::to_string(&new_input).unwrap_or(graph.instance_id.clone()))
        .await

[Ismaël Mejía (iemejia)]

Good catch. The "before any pool or async task has started" claim was inaccurate since mgmt_pool is already created at that point. Rewrote the SAFETY comment to focus on the actual invariant: the BGW uses new_current_thread() (no additional OS threads spawned) and PostgreSQL's fork model guarantees no other threads exist in this process that could be reading the environment concurrently.

[Ismaël Mejía (iemejia)]

Agreed. Changed to pub(crate) and gated with #[cfg(any(test, feature = "pg_test"))] so it's only compiled for test builds and doesn't leak into the public API surface.

[Ismaël Mejía (iemejia)]

This PR has been split into 6 focused PRs to ease review and reduce rebase conflicts. Each PR is independent and can be reviewed/merged in any order.

PR	Title	Severity	Files
#251	fix: isolate epoch/extension polling onto dedicated pool	Critical	worker.rs
#252	fix: recoverable client on connection failure	Critical	client.rs, lib.rs
#253	fix: block CGNAT range 100.64.0.0/10 in SSRF protection	High	ssrf.rs, lib.rs
#254	fix: loop safety — max iteration guard and malformed config detection	Medium	execute_function_graph.rs, types.rs, dsl.rs, lib.rs
#255	fix: row-count limit on $name.* expansion (10,000 rows)	Medium	types.rs, lib.rs
#256	fix: wrap std::env::set_var in unsafe block for Rust 2024 edition	High	worker.rs, client.rs

Closing this PR in favor of the above.