Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions src/orchestrations/execute_function_graph.rs
Original file line number Diff line number Diff line change
Expand Up @@ -418,6 +418,12 @@ async fn execute_wait_schedule_node(
/// Sentinel key used to signal a break from within a loop
const BREAK_SENTINEL: &str = "__break__";

/// Minimum wall-clock duration that every loop iteration must take before
/// `continue_as_new` is called. If the body (plus any while-condition
/// evaluation) completes faster than this, a compensating timer makes up the
/// deficit so an empty-bodied loop can't busy-spin via continue_as_new.
const LOOP_MIN_ITER_DURATION: Duration = Duration::from_secs(1);

/// Check if a result contains a break signal
fn is_break_signal(result: &str) -> bool {
serde_json::from_str::<serde_json::Value>(result)
Expand Down Expand Up @@ -451,6 +457,11 @@ async fn execute_loop_node(
.as_ref()
.ok_or_else(|| format!("LOOP node {node_id} has no body"))?;

// Capture the iteration start time so we can rate-limit `continue_as_new`
// below. `utc_now()` is duroxide's deterministic clock (recorded in
// history and replayed verbatim), so this remains replay-safe.
let iter_started = ctx.utc_now().await.ok();

ctx.trace_info("Executing loop iteration");
let body_result = Box::pin(execute_function_node_with_vars(
ctx, graph, body_id, results, exec_ctx,
Expand Down Expand Up @@ -495,6 +506,26 @@ async fn execute_loop_node(
}

ctx.trace_info("Continuing as new for next loop iteration");

// Enforce a minimum per-iteration wall-clock duration to prevent
// busy-looping (e.g. `df.loop(df.sleep(0))`). Compute the elapsed time
// from the deterministic clock; if the iteration finished faster than
// LOOP_MIN_ITER_DURATION, schedule a timer for the deficit so the next
// continue_as_new is gated by at least that much real-clock time.
if let Some(started) = iter_started {
if let Ok(now) = ctx.utc_now().await {
let elapsed = now.duration_since(started).unwrap_or(Duration::ZERO);
if elapsed < LOOP_MIN_ITER_DURATION {
let deficit = LOOP_MIN_ITER_DURATION - elapsed;
ctx.trace_info(format!(
"Loop iteration took {elapsed:?} (< {LOOP_MIN_ITER_DURATION:?}); \
adding {deficit:?} rate-limit delay"
));
ctx.schedule_timer(deficit).await;
}
}
}

// Preserve vars in continue_as_new input
let new_input = FunctionInput {
instance_id: graph.instance_id.clone(),
Expand Down
71 changes: 71 additions & 0 deletions tests/e2e/sql/03_loops.sql
Original file line number Diff line number Diff line change
Expand Up @@ -254,5 +254,76 @@ END $$;
DROP TABLE _test_running_state;
DROP TABLE test_running_status_log;

-- === Test: zero_sleep_loop_rate_limited ===
-- Regression test for: df.loop(df.sleep(0)) busy-spin (issue #13).
-- A loop whose body contains only a zero-duration sleep must NOT spin at full
-- CPU speed. With the 1-second minimum-iteration delay enforced by the loop
-- handler, at most a handful of iterations should complete in 3 seconds.

DROP TABLE IF EXISTS test_zero_sleep_log;
CREATE TABLE test_zero_sleep_log (id SERIAL, ts TIMESTAMP DEFAULT now());

CREATE TEMP TABLE _test_zero_sleep_state AS
SELECT df.start(
df.loop(
'INSERT INTO test_zero_sleep_log DEFAULT VALUES'
~> df.sleep(0)
),
'test-loop-zero-sleep'
) AS instance_id;

DO $$
DECLARE
v_instance_id TEXT;
v_status TEXT;
v_cnt INT;
attempts INT := 0;
BEGIN
SELECT instance_id INTO v_instance_id FROM _test_zero_sleep_state;
RAISE NOTICE 'Test zero_sleep_loop_rate_limited: instance %', v_instance_id;

-- Wait until at least 1 iteration has run so the loop is clearly started.
LOOP
SELECT COUNT(*) INTO v_cnt FROM test_zero_sleep_log;
EXIT WHEN v_cnt >= 1 OR attempts > 100;
PERFORM pg_sleep(0.1);
attempts := attempts + 1;
END LOOP;

IF v_cnt < 1 THEN
RAISE EXCEPTION 'TEST FAILED [zero-sleep]: loop body never executed';
END IF;

-- Confirm the loop is still running (not failed/errored after the first iteration).
SELECT s INTO v_status FROM df.status(v_instance_id) s;
IF lower(v_status) != 'running' THEN
RAISE EXCEPTION 'TEST FAILED [zero-sleep]: expected running before observation window, got %', v_status;
END IF;

-- Let it run for ~3 more seconds and count iterations.
PERFORM pg_sleep(3);
SELECT COUNT(*) INTO v_cnt FROM test_zero_sleep_log;

-- Lower bound: the loop must have made meaningful progress.
IF v_cnt < 2 THEN
RAISE EXCEPTION 'TEST FAILED [zero-sleep]: only % iterations in ~3s; rate-limit may be too aggressive', v_cnt;
END IF;

-- With a 1-second minimum delay per continue_as_new, the loop cannot
-- complete more than ~4 iterations in 3 seconds (generous upper bound of
-- 15 to accommodate slow CI environments). Without the fix it would run
-- hundreds of times in the same window.
IF v_cnt > 15 THEN
RAISE EXCEPTION 'TEST FAILED [zero-sleep]: % iterations in ~3s (expected <= 15); minimum rate-limit may not be working', v_cnt;
END IF;

RAISE NOTICE 'PASSED: zero_sleep_loop_rate_limited - % iterations in ~3s (within expected range)', v_cnt;

PERFORM df.cancel(v_instance_id, 'Test complete');
END $$;

DROP TABLE _test_zero_sleep_state;
DROP TABLE test_zero_sleep_log;

RESET SESSION AUTHORIZATION;
SELECT 'TEST PASSED' AS result;
Loading