diff --git a/src/orchestrations/execute_function_graph.rs b/src/orchestrations/execute_function_graph.rs index ae23ce69..508a51c7 100644 --- a/src/orchestrations/execute_function_graph.rs +++ b/src/orchestrations/execute_function_graph.rs @@ -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::(result) @@ -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, @@ -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(), diff --git a/tests/e2e/sql/03_loops.sql b/tests/e2e/sql/03_loops.sql index b62a9871..26bcb04b 100644 --- a/tests/e2e/sql/03_loops.sql +++ b/tests/e2e/sql/03_loops.sql @@ -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;