Use exact microsecond ceiling for durable call_at deadlines

Convert loop deadlines with exact rational arithmetic so durable timers never fire before the requested loop time, and add exact-oracle regression tests.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: andystaples

durabletask/internal/asyncio_loop.py

@@ -12,12 +12,16 @@
 from collections import deque
 from collections.abc import Callable, Coroutine, Generator
 from datetime import datetime, timedelta
+from fractions import Fraction
 from typing import Any, TypeVar, cast
 
 from durabletask import task
 
 T = TypeVar("T")
 
+_EPOCH = datetime(1970, 1, 1)
+_MICROSECOND = timedelta(microseconds=1)
+
 
 class _CompletionObserver(task.CompositeTask[None]):
     def __init__(self, durable: task.Task[Any], callback: Callable[[], None]) -> None:
@@ -133,7 +137,7 @@ def __init__(
 
     def time(self) -> float:
         # A fixed UTC epoch avoids local-time and platform timestamp differences.
-        return (self.ctx.current_utc_datetime - datetime(1970, 1, 1)).total_seconds()
+        return (self.ctx.current_utc_datetime - _EPOCH).total_seconds()
 
     def get_debug(self) -> bool:
         return False
@@ -205,14 +209,16 @@ def call_at(
         if not math.isfinite(when):
             raise ValueError("Durable deadlines must be finite.")
         handle = _DurableTimerHandle(when, callback, args, self, context)
-        if when <= self.time():
+        # Exact rational ceiling: float round-trips can emit a deadline that
+        # precedes the requested one by a fraction of a microsecond.
+        deadline_us = math.ceil(Fraction(when) * 1_000_000)
+        now_us = (self.ctx.current_utc_datetime - _EPOCH) // _MICROSECOND
+        if when <= self.time() or deadline_us <= now_us:
             self._ready.append(handle)
         else:
             if self._closing:
                 raise RuntimeError("Dispatch cleanup cannot schedule durable timers.")
-            deadline = datetime(1970, 1, 1) + timedelta(seconds=when)
-            if (deadline - datetime(1970, 1, 1)).total_seconds() < when:
-                deadline += timedelta(microseconds=1)
+            deadline = _EPOCH + timedelta(microseconds=deadline_us)
             handle.durable_timer = self.ctx.create_timer(deadline)
             self._timers[handle] = _CompletionObserver(
                 handle.durable_timer,

tests/durabletask/test_asyncio_orchestrators.py

@@ -12,6 +12,7 @@
 from contextvars import ContextVar
 from dataclasses import dataclass
 from datetime import timedelta
+from decimal import ROUND_CEILING, Decimal, localcontext
 from functools import partial
 from typing import Any
 from weakref import ReferenceType, ref
@@ -399,6 +400,65 @@ async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
     assert (deadline - datetime(1970, 1, 1)).total_seconds() >= observed[0]
 
 
+@pytest.mark.parametrize("offset", [0.000003, 0.0000003, 0.000001, 0.0000013, 1.0, 2.25])
+def test_call_at_deadline_is_exact_microsecond_ceiling(offset: float) -> None:
+    requested: list[float] = []
+
+    async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
+        loop = asyncio.get_running_loop()
+        when = loop.time() + offset
+        requested.append(when)
+        fired = loop.create_future()
+        loop.call_at(when, fired.set_result, None)
+        await fired
+
+    actions = _execute(orchestrator, [], _start()).actions
+    assert len(actions) == 1
+    fire_at = actions[0].createTimer.fireAt
+    assert fire_at.nanos % 1000 == 0
+    emitted_us = fire_at.seconds * 1_000_000 + fire_at.nanos // 1000
+    # Decimal(float) is exact, so this oracle shares no float arithmetic with the loop.
+    with localcontext() as exact:
+        exact.prec = 400
+        requested_seconds = Decimal(requested[0])
+        oracle_us = int((requested_seconds * 1_000_000).to_integral_value(rounding=ROUND_CEILING))
+        assert emitted_us == oracle_us
+        assert Decimal(emitted_us) / 1_000_000 >= requested_seconds
+
+
+def test_call_at_known_float_deadline_rounds_up_to_next_microsecond() -> None:
+    async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
+        loop = asyncio.get_running_loop()
+        fired = loop.create_future()
+        loop.call_at(loop.time() + 0.000003, fired.set_result, None)
+        await fired
+
+    actions = _execute(orchestrator, [], _start()).actions
+    assert actions[0].createTimer.fireAt.ToDatetime() == NOW + timedelta(microseconds=4)
+
+
+@pytest.mark.parametrize("offset", [0.0, -1.0])
+def test_call_at_due_deadline_runs_without_durable_timer(offset: float) -> None:
+    async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str:
+        loop = asyncio.get_running_loop()
+        fired = loop.create_future()
+        loop.call_at(loop.time() + offset, fired.set_result, "fired")
+        return await fired
+
+    result = _execute(orchestrator, [], _start())
+    assert json.loads(_completion(result).result.value) == "fired"
+
+
+@pytest.mark.parametrize("when", [1e20, float(sys.float_info.max)])
+def test_call_at_out_of_range_deadline_fails(when: float) -> None:
+    async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
+        asyncio.get_running_loop().call_at(when, lambda: None)
+
+    complete = _completion(_execute(orchestrator, [], _start()))
+    assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
+    assert complete.failureDetails.errorType.endswith("OverflowError")
+
+
 def test_async_sleep_uses_shared_long_timer_chunking() -> None:
     async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str:
         await asyncio.sleep(7 * 86400)