New features: scheduling and throttling.

Author: codeSamuraii

README.md

@@ -81,6 +81,8 @@ After setup, tasks are signed automatically. No client-side code changes. See [S
 | **Package auto-install** | Workers `pip install` missing packages before execution |
 | **Async-native** | `.run()`, `.start()`, `.map()`, `asyncio.gather` |
 | **Retry & timeout** | `@trace(timeout=30, retries=3)` with exponential backoff |
+| **Scheduling** | `.run_in(delay)`, `.run_at(datetime)`, `.run_every(freq)` with cancellation |
+| **Throttling** | `@trace(throttle=timedelta(hours=24)/50)` — rate-limit executions |
 | **Progress & cancellation** | `pyfuse.progress(3, 10)` inside tasks; `await future.cancel()` on client |
 | **Heartbeat & stall detection** | Workers heartbeat; clients raise `TaskStalled` on silence |
 | **Content-hash caching** | Same code = cache hit, regardless of client |
@@ -105,7 +107,7 @@ pyfuse worker --backend local://localhost:9748 --tmp
 pyfuse run examples/remote_execution.py
 ```
 
-[`remote_execution.py`](examples/remote_execution.py) · [`async_execution.py`](examples/async_execution.py) · [`package_installation.py`](examples/package_installation.py) · [`progress_reporting.py`](examples/progress_reporting.py) · [`cancellation.py`](examples/cancellation.py) · [`large_module.py`](examples/large_module.py)
+[`remote_execution.py`](examples/remote_execution.py) · [`async_execution.py`](examples/async_execution.py) · [`package_installation.py`](examples/package_installation.py) · [`progress_reporting.py`](examples/progress_reporting.py) · [`cancellation.py`](examples/cancellation.py) · [`scheduling.py`](examples/scheduling.py) · [`throttling_and_retry.py`](examples/throttling_and_retry.py) · [`large_module.py`](examples/large_module.py)
 
 ## License
 

docs/QUICK_START.md

@@ -58,6 +58,39 @@ def flaky_task(url: str) -> str: ...
 
 Retries use exponential backoff (1s, 2s, 4s).
 
+## Scheduling
+
+Execute tasks on a delay, at a specific time, or on a recurring schedule:
+
+```python
+from datetime import datetime, timedelta
+
+# Run after a delay
+result = await func.run_in(timedelta(minutes=5), *args)
+
+# Run at a specific time
+result = await func.run_at(datetime(2026, 4, 21, 9, 0), *args)
+
+# Recurring execution (every hour)
+schedule = await func.run_every(timedelta(hours=1), *args)
+await schedule.cancel()  # stop the schedule
+```
+
+`start_at` and `start_in` return a `Result` handle (like `.start()`).
+
+## Throttling
+
+Rate-limit how often a function can be executed:
+
+```python
+from datetime import timedelta
+
+@trace(throttle=timedelta(hours=24) / 50)  # ~29 min cooldown
+def expensive_api_call(query: str) -> str: ...
+```
+
+If a task arrives during the cooldown window, the worker returns a `ThrottleError` immediately (no retry). The cooldown is only recorded after a successful execution.
+
 ## Third-party packages
 
 Workers auto-install missing packages. When the import name differs from the pip package:

docs/TECHNICAL_OVERVIEW.md

@@ -59,6 +59,7 @@ pyfuse/
 # Decorator
 @trace                                    # capture function for remote execution
 @trace(timeout=30, retries=3)             # with execution options
+@trace(throttle=timedelta(hours=24)/50)   # rate-limit: 50 calls per day
 
 # Remote execution (all async)
 pyfuse.connect("redis://localhost:6379")  # configure backend (sync)
@@ -67,6 +68,12 @@ await func.run(*args)                     # submit + await result
 future = await func.start(*args)          # submit, returns Result handle
 results = await func.map([(a1, b1), ...]) # batch submit + await all
 
+# Scheduling
+await func.run_in(timedelta(minutes=5), *args)       # execute after delay
+await func.run_at(datetime(2026, 1, 1), *args)       # execute at specific time
+schedule = await func.run_every(timedelta(hours=1), *args)  # recurring
+await schedule.cancel()                               # stop recurring
+
 # Result handling
 result = await future                     # await result value
 result = await future.result(timeout=10)  # with timeout and stall detection
@@ -101,13 +108,17 @@ pyfuse.get_graph().to_mermaid(func)       # -> Mermaid diagram string
 ### Worker side: `await serve()` / `pyfuse worker`
 
 1. **Listen** -- `async for task_json in backend.listen()` yields tasks as they arrive.
-2. **Cancellation check** -- If `await backend.is_cancelled(task_id)` returns ``True``, skip execution and log the cancellation.
-3. **Deserialize** -- Parse the JSON graph into a `Store`.
+2. **Scheduling wait** -- If the task has a `scheduled_at` timestamp in the future, `await asyncio.sleep(delay)` until that time.
+3. **Cancellation check** -- If `await backend.is_cancelled(task_id)` returns ``True``, skip execution and log the cancellation.
+4. **Throttle check** -- If the task has a `throttle` value and the cooldown hasn't elapsed, return a `"throttled"` result immediately.
+5. **Deserialize** -- Parse the JSON graph into a `Store`.
 4. **Cache check** -- Compute a subgraph key (SHA-256 of all reachable content hashes). If cached, skip to step 7.
 5. **Install dependencies** -- Extract third-party imports, install missing packages via `asyncio.create_subprocess_exec`.
 6. **Reconstruct** -- Produce a self-contained Python script from the store. `compile()` and `exec()` it into a fresh namespace.
-7. **Execute** -- Call the function with the provided arguments. Async functions are awaited directly; sync functions run in an executor with explicit context propagation (for progress reporting). Apply retry/timeout policies via `asyncio.wait_for`.
-8. **Send result** -- Wrap the return value (or exception traceback) in a `ResultEnvelope` and send it back. If cancelled during execution, skip result delivery.
+9. **Execute** -- Call the function with the provided arguments. Async functions are awaited directly; sync functions run in an executor with explicit context propagation (for progress reporting). Apply retry/timeout policies via `asyncio.wait_for`.
+10. **Send result** -- Wrap the return value (or exception traceback) in a `ResultEnvelope` and send it back. If cancelled during execution, skip result delivery.
+11. **Record throttle** -- If the task has a `throttle` value and execution succeeded, record a cooldown in the backend.
+12. **Re-enqueue recurring** -- If the task has a `recur_interval` and its schedule hasn't been cancelled, submit a new task instance with `scheduled_at = now + interval`.
 
 ### Client side: `await future` / `await future.result()`
 
@@ -117,7 +128,7 @@ The `Result` object supports two waiting strategies:
 
 **With stall detection** (default, `stall_timeout=10.0`) -- An async polling loop calls `try_get_result()` and checks heartbeats every second. If the heartbeat hasn't changed for longer than `stall_timeout`, `TaskStalled` is raised.
 
-**Unwrap** -- If status is `"ok"`, return the value. If `"error"`, raise `RemoteError` with the remote traceback.
+**Unwrap** -- If status is `"ok"`, return the value. If `"error"`, raise `RemoteError` with the remote traceback. If `"throttled"`, raise `ThrottleError`.
 
 ## Transport backends
 
@@ -137,6 +148,10 @@ The `Backend` ABC defines an async transport interface:
 | `async is_cancelled(task_id)` | Check cancellation flag (default returns `False`) |
 | `async send_progress(task_id, json)` | Store latest progress data (no-op default) |
 | `async get_progress(task_id)` | Get latest progress JSON (default returns `None`) |
+| `async cancel_schedule(schedule_id)` | Mark recurring schedule cancelled (no-op default) |
+| `async is_schedule_cancelled(schedule_id)` | Check schedule cancellation (default `False`) |
+| `async check_throttle(function_name)` | Check if function is rate-limited (default `True`) |
+| `async record_throttle(fn, seconds)` | Start cooldown after execution (no-op default) |
 | `async notify_result(task_id)` | Push notification that result is ready (no-op default) |
 | `async subscribe_results()` | Async iterator yielding task_ids on result arrival |
 | `async close()` | Release resources |
@@ -151,6 +166,8 @@ Uses `redis.asyncio.Redis` with `RPUSH`/`BLPOP` patterns. Keys:
 - `pyfuse:heartbeat:{task_id}` -- worker heartbeat timestamp (TTL: 30s)
 - `pyfuse:cancel:{task_id}` -- cancellation flag (TTL: 3600s)
 - `pyfuse:progress:{task_id}` -- latest progress JSON (TTL: 300s)
+- `pyfuse:schedule:{schedule_id}` -- schedule cancellation flag (TTL: 30 days)
+- `pyfuse:throttle:{function_name}` -- throttle cooldown (TTL: throttle seconds)
 - `pyfuse:notify` -- Pub/Sub channel for result notifications
 
 Result notifications use Redis Pub/Sub (`PUBLISH`/`SUBSCRIBE`). Batch heartbeat fetching uses `MGET` for efficiency.

examples/scheduling.py

@@ -0,0 +1,61 @@
+"""Task scheduling with pyfuse.
+
+Demonstrates:
+- run_in(delay)   — execute after a delay
+- run_at(datetime) — execute at a specific time
+- run_every(freq)  — recurring execution with cancellation
+
+Usage:
+    # Terminal 1 -- start a worker
+    pyfuse worker --backend redis://localhost:6379 --tmp
+
+    # Terminal 2 -- run this script
+    pyfuse run examples/scheduling.py
+"""
+
+import asyncio
+from datetime import datetime, timedelta
+
+import pyfuse
+from pyfuse import trace
+
+pyfuse.connect("local://localhost:9748")
+
+
+@trace
+def greet(name: str) -> str:
+    return f"Hello, {name}! (executed at {datetime.now():%H:%M:%S})"
+
+
+@trace
+def tick(n: int) -> str:
+    return f"Tick #{n} at {datetime.now():%H:%M:%S}"
+
+
+async def main() -> None:
+    print(f"Now: {datetime.now():%H:%M:%S}\n")
+
+    # 1. run_in — execute after a 2-second delay
+    print("— run_in(2 seconds) —")
+    result = await greet.run_in(timedelta(seconds=2), "Alice")
+    print(result)
+
+    # 2. run_at — execute at a specific time (3 seconds from now)
+    print("\n— run_at(now + 3s) —")
+    target = datetime.now() + timedelta(seconds=3)
+    print(f"Scheduled for: {target:%H:%M:%S}")
+    result = await greet.run_at(target, "Bob")
+    print(result)
+
+    # 3. run_every — recurring execution every 2 seconds
+    print("\n— run_every(2 seconds) for ~6 seconds —")
+    schedule = await tick.run_every(timedelta(seconds=2), 1)
+    print(f"Schedule started: {schedule.schedule_id}")
+    await asyncio.sleep(6)
+
+    # Cancel the recurring schedule
+    await schedule.cancel()
+    print(f"Schedule cancelled: {schedule.schedule_id}")
+
+
+asyncio.run(main())

examples/throttling_and_retry.py

@@ -0,0 +1,70 @@
+"""Throttling and retry with pyfuse.
+
+Demonstrates:
+- @trace(throttle=timedelta) — rate-limit executions
+- @trace(retries=N, retry_delay=T) — retry with exponential backoff
+- Combining throttle + retry
+
+Usage:
+    # Terminal 1 -- start a worker
+    pyfuse worker --backend redis://localhost:6379 --tmp
+
+    # Terminal 2 -- run this script
+    pyfuse run examples/throttling_and_retry.py
+"""
+
+import asyncio
+import random
+from datetime import timedelta
+
+import pyfuse
+from pyfuse import trace, ThrottleError
+
+pyfuse.connect("local://localhost:9748")
+
+
+# --- Throttling: at most once every 5 seconds ---
+
+@trace(throttle=timedelta(seconds=5))
+def expensive_query(query: str) -> str:
+    return f"Result for '{query}'"
+
+
+# --- Retry: 3 attempts with exponential backoff ---
+
+@trace(retries=3, retry_delay=0.5)
+def flaky_operation(x: int) -> int:
+    if random.random() < 0.5:
+        raise RuntimeError("Random failure!")
+    return x * 2
+
+
+async def main() -> None:
+    # 1. Throttling demo
+    print("— Throttle demo (5s cooldown) —")
+    result = await expensive_query.run("first call")
+    print(f"  Call 1: {result}")
+
+    try:
+        result = await expensive_query.run("second call (too soon)")
+        print(f"  Call 2: {result}")
+    except ThrottleError:
+        print("  Call 2: ThrottleError — rate limited!")
+
+    # Wait for cooldown
+    print("  Waiting 5s for cooldown...")
+    await asyncio.sleep(5)
+
+    result = await expensive_query.run("third call (after cooldown)")
+    print(f"  Call 3: {result}")
+
+    # 2. Retry demo
+    print("\n— Retry demo (3 retries, 0.5s base delay) —")
+    try:
+        result = await flaky_operation.run(21)
+        print(f"  Result: {result}")
+    except Exception as exc:
+        print(f"  Failed after retries: {exc}")
+
+
+asyncio.run(main())

pyfuse/__init__.py

@@ -14,6 +14,7 @@
     TaskCancelled,
     SignatureError,
     DependencyError,
+    ThrottleError,
 )
 from pyfuse.core.models import ImportInfo, FunctionNode
 from pyfuse.graph.graph import Graph
@@ -50,6 +51,7 @@
 from pyfuse.worker.worker import Worker
 from pyfuse.worker.worker import execute as execute
 from pyfuse.worker.sandbox import DockerSandbox
+from pyfuse.worker.schedule import ScheduleHandle
 from pyfuse.graph.decorator import trace
 from pyfuse.worker.backends.base import Backend
 
@@ -125,6 +127,7 @@ def pack(func: Callable[..., object], *args: Any, **kwargs: Any) -> Task:
     "RemoteError",
     "TaskStalled",
     "TaskCancelled",
+    "ThrottleError",
     "SignatureError",
     "PairingError",
     # Graph
@@ -154,6 +157,8 @@ def pack(func: Callable[..., object], *args: Any, **kwargs: Any) -> Task:
     "Task",
     "Worker",
     "Backend",
+    # Scheduling
+    "ScheduleHandle",
     # Sandbox
     "DockerSandbox",
 ]

pyfuse/core/errors.py

@@ -46,6 +46,10 @@ class PairingError(Error):
     """Raised when the PIN-based pairing protocol fails."""
 
 
+class ThrottleError(Error):
+    """Raised when a task is rejected due to rate limiting."""
+
+
 # ---------------------------------------------------------------------------
 # Custom excepthook: suppress client traceback for RemoteError
 # ---------------------------------------------------------------------------

pyfuse/core/task.py

@@ -102,6 +102,10 @@ class Task:
     timeout: float | None = None
     retries: int = 0
     retry_delay: float = 1.0
+    scheduled_at: float | None = None
+    recur_interval: float | None = None
+    schedule_id: str | None = None
+    throttle: float | None = None
     signature: str | None = None
 
     # -- Serialization -------------------------------------------------------
@@ -121,6 +125,14 @@ def _to_dict(self) -> dict[str, Any]:
             d["retries"] = self.retries
         if self.retry_delay != 1.0:
             d["retry_delay"] = self.retry_delay
+        if self.scheduled_at is not None:
+            d["scheduled_at"] = self.scheduled_at
+        if self.recur_interval is not None:
+            d["recur_interval"] = self.recur_interval
+        if self.schedule_id is not None:
+            d["schedule_id"] = self.schedule_id
+        if self.throttle is not None:
+            d["throttle"] = self.throttle
         return d
 
     def to_json(self, *, signing_key: bytes | None = None) -> str:
@@ -185,5 +197,9 @@ def from_json(
             timeout=data.get("timeout"),
             retries=data.get("retries", 0),
             retry_delay=data.get("retry_delay", 1.0),
+            scheduled_at=data.get("scheduled_at"),
+            recur_interval=data.get("recur_interval"),
+            schedule_id=data.get("schedule_id"),
+            throttle=data.get("throttle"),
             signature=sig or None,  # normalise empty string to None
         )