Clamp min_results if greater than outstanding calls

Author: chriso

src/dispatch/scheduler.py

@@ -440,7 +440,7 @@ def _run(self, input: Input) -> Output:
         return Output.poll(
             state=serialized_state,
             calls=pending_calls,
-            min_results=max(1, self.poll_min_results),
+            min_results=max(1, min(state.outstanding_calls, self.poll_min_results)),
             max_results=max(1, min(state.outstanding_calls, self.poll_max_results)),
             max_wait_seconds=self.poll_max_wait_seconds,
         )

tests/dispatch/test_scheduler.py

@@ -291,16 +291,32 @@ async def main():
         output = self.start(main)
         self.assert_exit_result_error(output, ValueError, "oops")
 
-    def start(self, main: Callable, *args: Any, **kwargs: Any) -> Output:
+    def test_min_max_results_clamping(self):
+        @durable
+        async def main():
+            return await call_concurrently("a", "b", "c")
+
+        output = self.start(main, poll_min_results=1, poll_max_results=10)
+        self.assert_poll_call_functions(output, ["a", "b", "c"], min_results=1, max_results=3)
+
+        output = self.start(main, poll_min_results=1, poll_max_results=2)
+        self.assert_poll_call_functions(output, ["a", "b", "c"], min_results=1, max_results=2)
+
+        output = self.start(main, poll_min_results=10, poll_max_results=10)
+        self.assert_poll_call_functions(output, ["a", "b", "c"], min_results=3, max_results=3)
+
+    def start(self, main: Callable, *args: Any, poll_min_results=1, poll_max_results=10, poll_max_wait_seconds=None,
+              **kwargs: Any) -> Output:
         input = Input.from_input_arguments(main.__qualname__, *args, **kwargs)
-        return OneShotScheduler(main).run(input)
+        return OneShotScheduler(main, poll_min_results=poll_min_results, poll_max_results=poll_max_results,
+                                poll_max_wait_seconds=poll_max_wait_seconds).run(input)
 
     def resume(
-        self,
-        main: Callable,
-        prev_output: Output,
-        call_results: list[CallResult],
-        poll_error: Exception | None = None,
+            self,
+            main: Callable,
+            prev_output: Output,
+            call_results: list[CallResult],
+            poll_error: Exception | None = None,
     ):
         poll = self.assert_poll(prev_output)
         input = Input.from_poll_results(
@@ -330,7 +346,7 @@ def assert_exit_result_value(self, output: Output, expect: Any):
         self.assertEqual(expect, any_unpickle(result.output))
 
     def assert_exit_result_error(
-        self, output: Output, expect: type[Exception], message: str | None = None
+            self, output: Output, expect: type[Exception], message: str | None = None
     ):
         result = self.assert_exit_result(output)
         self.assertFalse(result.HasField("output"))
@@ -357,7 +373,7 @@ def assert_empty_poll(self, output: Output):
         self.assertEqual(len(poll.calls), 0)
 
     def assert_poll_call_functions(
-        self, output: Output, expect: list[str], min_results=None, max_results=None
+            self, output: Output, expect: list[str], min_results=None, max_results=None
     ):
         poll = self.assert_poll(output)
         # Note: we're not testing endpoint/input here.