Change timeout option from absolute to relative time (#236)

* Change timeout option from absolute to relative time

* Fix failing tests

* Use clock from dependencies

Author: dfarr

.github/workflows/ci.yml

@@ -69,4 +69,4 @@ jobs:
       - name: run tests
         env:
           RESONATE_HOST: http://localhost
-        run: uv run pytest tests
+        run: uv run pytest -s

resonate/bridge.py

@@ -95,7 +95,7 @@ def run(self, conv: Convention, func: Callable, args: tuple, kwargs: dict, opts:
         promise, task = self._store.promises.create_with_task(
             id=conv.id,
             ikey=conv.idempotency_key,
-            timeout=conv.timeout,
+            timeout=int((time.time() + conv.timeout) * 1000),
             headers=conv.headers,
             data=conv.data,
             tags=conv.tags,
@@ -113,7 +113,7 @@ def run(self, conv: Convention, func: Callable, args: tuple, kwargs: dict, opts:
         elif task is not None:
             self._promise_id_to_task[promise.id] = task
             self.start_heartbeat()
-            self._cq.put_nowait((Invoke(conv.id, conv, func, args, kwargs, opts, promise), future))
+            self._cq.put_nowait((Invoke(conv.id, conv, promise.abs_timeout, func, args, kwargs, opts, promise), future))
         else:
             self._cq.put_nowait((Listen(promise.id), future))
 
@@ -123,7 +123,7 @@ def rpc(self, conv: Convention, future: Future) -> DurablePromise:
         promise = self._store.promises.create(
             id=conv.id,
             ikey=conv.idempotency_key,
-            timeout=conv.timeout,
+            timeout=int((time.time() + conv.timeout) * 1000),
             headers=conv.headers,
             data=conv.data,
             tags=conv.tags,
@@ -259,12 +259,13 @@ def _invoke(root: DurablePromise) -> Invoke:
                 root.id,
                 Base(
                     root.id,
-                    root.timeout,
+                    root.rel_timeout,
                     root.ikey_for_create,
                     root.param.headers,
                     root.param.data,
                     root.tags,
                 ),
+                root.abs_timeout,
                 func,
                 root.param.data.get("args", ()),
                 root.param.data.get("kwargs", {}),

resonate/clocks/__init__.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
 from .step import StepClock
-from .wall import WallClock
 
-__all__ = ["StepClock", "WallClock"]
+__all__ = ["StepClock"]

resonate/clocks/step.py

@@ -3,11 +3,15 @@
 
 class StepClock:
     def __init__(self) -> None:
-        self._time = 0
+        self._time = 0.0
 
-    def time(self) -> int:
-        return self._time
-
-    def set_time(self, time: int) -> None:
+    def step(self, time: float) -> None:
         assert time >= self._time, "The arrow of time only flows forward."
         self._time = time
+
+    def time(self) -> float:
+        """Return the current time in seconds."""
+        return self._time
+
+    def strftime(self, format: str, /) -> str:
+        raise NotImplementedError

resonate/clocks/wall.py

@@ -10,7 +10,7 @@
 @dataclass
 class Base:
     id: str
-    timeout: int
+    timeout: float
     idempotency_key: str | None = None
     headers: dict[str, str] | None = None
     data: Any = None
@@ -20,9 +20,9 @@ def options(
         self,
         id: str | None = None,
         idempotency_key: str | Callable[[str], str] | None = None,
-        target: str | None = None,
         tags: dict[str, str] | None = None,
-        timeout: int | None = None,
+        target: str | None = None,
+        timeout: float | None = None,
         version: int | None = None,
     ) -> Base:
         self.id = id or self.id

resonate/conventions/base.py

@@ -27,7 +27,7 @@ def data(self) -> Any:
         return None
 
     @property
-    def timeout(self) -> int:
+    def timeout(self) -> float:
         return self.opts.timeout
 
     @property
@@ -38,9 +38,9 @@ def options(
         self,
         id: str | None = None,
         idempotency_key: str | Callable[[str], str] | None = None,
-        target: str | None = None,
         tags: dict[str, str] | None = None,
-        timeout: int | None = None,
+        target: str | None = None,
+        timeout: float | None = None,
         version: int | None = None,
     ) -> Local:
         self.id = id or self.id

resonate/conventions/local.py

@@ -30,7 +30,7 @@ def data(self) -> dict[str, Any]:
         return {"func": self.name, "args": self.args, "kwargs": self.kwargs, "version": self.opts.version}
 
     @property
-    def timeout(self) -> int:
+    def timeout(self) -> float:
         return self.opts.timeout
 
     @property
@@ -41,8 +41,8 @@ def options(
         self,
         id: str | None = None,
         idempotency_key: str | Callable[[str], str] | None = None,
-        target: str | None = None,
         tags: dict[str, str] | None = None,
+        target: str | None = None,
         timeout: int | None = None,
         version: int | None = None,
     ) -> Remote:

resonate/conventions/remote.py

@@ -10,7 +10,7 @@
 @dataclass
 class Sleep:
     id: str
-    timeout: int
+    secs: float
 
     @property
     def idempotency_key(self) -> str:
@@ -24,6 +24,10 @@ def headers(self) -> dict[str, str] | None:
     def data(self) -> Any:
         return None
 
+    @property
+    def timeout(self) -> float:
+        return self.secs
+
     @property
     def tags(self) -> dict[str, str]:
         return {"resonate:timeout": "true"}
@@ -32,8 +36,8 @@ def options(
         self,
         id: str | None = None,
         idempotency_key: str | Callable[[str], str] | None = None,
-        target: str | None = None,
         tags: dict[str, str] | None = None,
+        target: str | None = None,
         timeout: int | None = None,
         version: int | None = None,
     ) -> Sleep:

resonate/conventions/sleep.py

@@ -56,6 +56,7 @@ class LFC[T](LFX[T]):
 @dataclass
 class RFX[T]:
     conv: Convention
+    opts: Options = field(default_factory=Options)  # unused for the time being
 
     @property
     def id(self) -> str:

resonate/coroutine.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
-from typing import Protocol
+from typing import Protocol, runtime_checkable
 
 
+@runtime_checkable
 class Clock(Protocol):
-    def time(self) -> int: ...
+    def time(self) -> float: ...
+    def strftime(self, format: str, /) -> str: ...

resonate/models/clock.py

@@ -24,6 +24,7 @@
 class Invoke:
     id: str
     conv: Convention
+    timeout: float  # absolute time in seconds
     func: Callable[..., Any] = field(repr=False)
     args: tuple[Any, ...] = field(default_factory=tuple)
     kwargs: dict[str, Any] = field(default_factory=dict)

resonate/models/commands.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Protocol
+from typing import Any, Protocol
 
 
 class Context(Protocol):
@@ -9,6 +9,8 @@ def id(self) -> str: ...
     @property
     def info(self) -> Info: ...
 
+    def get_dependency(self, key: str, default: Any = None) -> Any: ...
+
 
 class Info(Protocol):
     @property
@@ -18,6 +20,6 @@ def idempotency_key(self) -> str | None: ...
     @property
     def tags(self) -> dict[str, str] | None: ...
     @property
-    def timeout(self) -> int: ...
+    def timeout(self) -> float: ...
     @property
     def version(self) -> int: ...

resonate/models/context.py

@@ -16,16 +16,16 @@ def headers(self) -> dict[str, str] | None: ...
     @property
     def data(self) -> Any: ...
     @property
-    def timeout(self) -> int: ...
+    def timeout(self) -> float: ...  # relative time in seconds
     @property
     def tags(self) -> dict[str, str] | None: ...
 
     def options(
         self,
         id: str | None = None,
         idempotency_key: str | Callable[[str], str] | None = None,
-        target: str | None = None,
         tags: dict[str, str] | None = None,
-        timeout: int | None = None,
+        target: str | None = None,
+        timeout: float | None = None,
         version: int | None = None,
     ) -> Convention: ...

resonate/models/convention.py

@@ -61,6 +61,14 @@ def canceled(self) -> bool:
     def timedout(self) -> bool:
         return self.state == "REJECTED_TIMEDOUT"
 
+    @property
+    def abs_timeout(self) -> float:
+        return self.timeout / 1000
+
+    @property
+    def rel_timeout(self) -> float:
+        return (self.timeout - self.created_on) / 1000
+
     def resolve(self, data: str | None, *, ikey: str | None = None, strict: bool = False, headers: dict[str, str] | None = None) -> None:
         promise = self.store.promises.resolve(
             id=self.id,

resonate/models/durable_promise.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import sys
 from dataclasses import dataclass, field
 from inspect import isgeneratorfunction
 from typing import TYPE_CHECKING
@@ -23,7 +22,7 @@ class Options:
     retry_policy: RetryPolicy | Callable[[Callable], RetryPolicy] = lambda f: Never() if isgeneratorfunction(f) else Exponential()
     target: str = "poll://default"
     tags: dict[str, str] = field(default_factory=dict)
-    timeout: int = sys.maxsize
+    timeout: float = 31536000  # relative time in seconds, default 1 year
     version: int = 0
 
     def __post_init__(self) -> None:
@@ -44,7 +43,7 @@ def merge(
         retry_policy: RetryPolicy | Callable[[Callable], RetryPolicy] | None = None,
         target: str | None = None,
         tags: dict[str, str] | None = None,
-        timeout: int | None = None,
+        timeout: float | None = None,
         version: int | None = None,
     ) -> Options:
         if version and not (version >= 0):