feat: add one-off scheduled tasks

.github/workflows/ci.yml

@@ -20,7 +20,7 @@ jobs:
       - uses: astral-sh/setup-uv@v4
 
       - name: Install dependencies
-        run: uv pip install --system -e ".[redis,dev]" ruff mypy
+        run: uv pip install --system -e ".[redis,encryption,scheduler,dev]" "ruff>=0.15.9,<0.16" mypy
 
       - name: Lint
         run: |

README.md

@@ -98,6 +98,7 @@ The route signature does not change. Tasks that fail are retried. If the server
 - Priority queues: `priority=` on `@task_manager.task()` or `add_task()`, higher-priority tasks run first, equal-priority tasks are FIFO
 - Eager dispatch: `eager=True` starts a task immediately via `asyncio.create_task` before the HTTP response is sent
 - Scheduled tasks: `@task_manager.schedule(every=)` and `cron=` with distributed lock for multi-instance
+- One-off scheduled tasks: `schedule_once(run_at=, run_key=)` runs a task once at a runtime-computed time, persisted across restarts, reschedulable and cancellable by key
 - Zero-migration injection: keep your existing `BackgroundTasks` annotations
 - Both sync and async task functions supported
 
@@ -430,6 +431,64 @@ Cron expressions require `pip install "fastapi-taskflow[scheduler]"`. Interval-b
 
 In multi-instance deployments, a distributed lock ensures only one instance fires each scheduled entry per interval.
 
+## One-off scheduled tasks
+
+`@task_manager.schedule()` fixes a cadence at import time. When the run time is only known at runtime and computed from your own data, use `schedule_once()` instead. It runs a single invocation at an exact future timestamp.
+
+```python
+from datetime import datetime, timedelta, timezone
+
+task_manager = TaskManager(snapshot_db="tasks.db")
+
+
+@task_manager.task()
+async def settle_auction(listing_id: str) -> None:
+    ...
+
+
+@app.post("/listings/{listing_id}/close-at")
+async def set_close_time(listing_id: str, closes_at: datetime):
+    await task_manager.schedule_once(
+        settle_auction,
+        listing_id,
+        run_at=closes_at,
+        run_key=f"auction-close:{listing_id}",
+    )
+    return {"scheduled": True}
+```
+
+`run_key` is the identity of the pending firing. Calling `schedule_once()` again with the same key replaces the entry rather than creating a second one, which is how you move a deadline:
+
+```python
+# The close time moved. This replaces the pending firing, it does not add one.
+await task_manager.schedule_once(
+    settle_auction, listing_id,
+    run_at=new_closes_at,
+    run_key=f"auction-close:{listing_id}",
+)
+
+# No longer needed.
+await task_manager.cancel_scheduled(f"auction-close:{listing_id}")
+
+# Inspect what is still pending.
+pending = await task_manager.list_scheduled()
+```
+
+Pending firings are written to the configured backend, so they survive a restart and are re-armed automatically on the next startup. When a firing comes due, exactly one instance claims it with an atomic delete, so a multi-instance deployment runs it once and only once.
+
+`run_key` is a separate namespace from `idempotency_key`. `run_key` identifies a pending schedule and is meant to be replaced or cancelled. `idempotency_key` guards against duplicate execution and is permanent once recorded. You can pass both:
+
+```python
+await task_manager.schedule_once(
+    settle_auction, listing_id,
+    run_at=closes_at,
+    run_key=f"auction-close:{listing_id}",
+    idempotency_key=f"settled:{listing_id}",
+)
+```
+
+A backend is required, since the whole point is that the firing outlives the process. `SqliteBackend`, `PostgresBackend`, `MySQLBackend`, and `RedisBackend` all support it. Calling `schedule_once()` without a backend, or with a custom backend that does not implement the storage methods, raises `RuntimeError` at the call site rather than dropping the task silently.
+
 ## Custom dashboard title
 
 Replace the "fastapi-taskflow" label in the dashboard header and login page with your own app name:

docs/api/scheduled-tasks.md

@@ -1,8 +1,8 @@
 # Scheduled Tasks API
 
-This page documents the data types involved in periodic task scheduling. For the decorator that creates scheduled tasks, see [`schedule()` in the TaskManager API](task-manager.md#schedule).
+This page documents the data types involved in task scheduling, both recurring and one-off. For the decorator that creates recurring scheduled tasks, see [`schedule()` in the TaskManager API](task-manager.md#schedule).
 
-> **Guide:** [Scheduled Tasks](../guide/scheduled-tasks.md) covers interval vs. cron scheduling, timezone configuration, and multi-instance deployments.
+> **Guide:** [Scheduled Tasks](../guide/scheduled-tasks.md) covers interval vs. cron scheduling, timezone configuration, and multi-instance deployments. [One-Off Scheduled Tasks](../guide/one-off-tasks.md) covers running a task once at a runtime-computed time.
 
 ---
 
@@ -48,9 +48,114 @@ async def morning_report() -> None:
 
 ---
 
+## `task_manager.schedule_once()`
+
+Coroutine that schedules a single run of `func` at an exact future time. Unlike `@schedule()`, which fixes a cadence at import time, this takes a timestamp computed at runtime.
+
+```python
+await task_manager.schedule_once(
+    settle_auction,
+    listing_id,
+    run_at=closes_at,
+    run_key=f"auction-close:{listing_id}",
+)
+```
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `func` | `Callable` | required | The task function. Must be registered with `@task_manager.task()` or `@task_manager.schedule()` so it can be resolved by name when the entry fires. |
+| `*args` | `Any` | | Positional arguments forwarded to `func`. |
+| `run_at` | `datetime` | required | When to run. Naive datetimes are treated as UTC. A past time fires on the next scheduler tick. |
+| `run_key` | `str` | required | Identity of this pending firing. Scheduling again with the same key replaces the entry. |
+| `idempotency_key` | `str \| None` | `None` | Forwarded onto the task record when the entry fires. Guards duplicate execution, independent of `run_key`. |
+| `tags` | `dict[str, str] \| None` | `None` | Key/value labels attached to the task when it fires. |
+| `priority` | `int \| None` | `None` | Execution priority for the firing. |
+| `queue` | `str \| None` | `None` | Named queue to route the firing into. |
+| `**kwargs` | `Any` | | Keyword arguments forwarded to `func`. |
+
+Raises `RuntimeError` if no snapshot backend is configured, or if the configured backend does not support one-off schedules. Raises `TaskArgumentError` if `func` uses `executor='process'` and any argument is not picklable.
+
+Note that `run_at`, `run_key`, `idempotency_key`, `tags`, `priority`, and `queue` are reserved names. A task function parameter with one of those names must be passed positionally or bound with `functools.partial`.
+
+---
+
+## `task_manager.cancel_scheduled()`
+
+Coroutine that removes a pending one-off firing.
+
+```python
+removed = await task_manager.cancel_scheduled(f"auction-close:{listing_id}")
+```
+
+| Parameter | Type | Description |
+|-----------|------|-------------|
+| `run_key` | `str` | The key passed to `schedule_once()`. |
+
+Returns `True` if a pending entry was removed, `False` if none existed (already fired, already cancelled, or never scheduled).
+
+The backend row is deleted before the in-memory copy is cleared, so a cancel racing a restart cannot re-arm the cancelled task.
+
+---
+
+## `task_manager.list_scheduled()`
+
+Coroutine returning pending one-off entries.
+
+```python
+pending = await task_manager.list_scheduled()
+```
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `before` | `datetime \| None` | one year out | Upper bound on `fire_at`. Pass a nearer bound when the pending set is large. |
+
+Returns a `list[ScheduledOnce]`.
+
+---
+
+## `ScheduledOnce`
+
+Represents one pending one-off firing. Created by `schedule_once()` and persisted to the backend. This is not a task invocation: no `task_id` exists until the entry fires, at which point a normal `TaskRecord` is created and the entry is deleted.
+
+```python
+from fastapi_taskflow import ScheduledOnce
+```
+
+```python
+@dataclass
+class ScheduledOnce:
+    run_key:           str
+    func_name:         str
+    fire_at:           datetime
+    args:              tuple
+    kwargs:            dict
+    encrypted_payload: bytes | None
+    queue:             str
+    priority:          int | None
+    idempotency_key:   str | None
+    tags:              dict[str, str]
+    created_at:        datetime
+```
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `run_key` | `str` | Identity of the pending firing. Primary key in the backend. |
+| `func_name` | `str` | Name of the registered function to run. |
+| `fire_at` | `datetime` | UTC time at which the task should run. |
+| `args` | `tuple` | Positional arguments. Empty when `encrypted_payload` is set. |
+| `kwargs` | `dict` | Keyword arguments. Empty when `encrypted_payload` is set. |
+| `encrypted_payload` | `bytes \| None` | Fernet-encrypted `(args, kwargs)` when `encrypt_args_key` is configured. |
+| `queue` | `str` | Named queue the firing is routed into. |
+| `priority` | `int \| None` | Priority to enqueue the firing at. |
+| `idempotency_key` | `str \| None` | Forwarded onto the `TaskRecord` at fire time. |
+| `tags` | `dict[str, str]` | Labels forwarded onto the `TaskRecord` at fire time. |
+| `created_at` | `datetime` | When the entry was scheduled. |
+
+---
+
 ## `ScheduledEntry`
 
-`ScheduledEntry` represents one registered periodic task. Created by `@task_manager.schedule()` and stored in `PeriodicScheduler`. You never instantiate this directly.
+`ScheduledEntry` represents one entry in the scheduler's heap. Created by `@task_manager.schedule()` for recurring tasks, or built from a `ScheduledOnce` when a one-off is armed. You never instantiate this directly.
 
 ```python
 from fastapi_taskflow.periodic import ScheduledEntry
@@ -59,24 +164,30 @@ from fastapi_taskflow.periodic import ScheduledEntry
 ```python
 @dataclass
 class ScheduledEntry:
-    func:          Callable
-    config:        TaskConfig
-    every:         float | None
-    cron:          str | None
+    func:           Callable
+    config:         TaskConfig
+    every:          float | None
+    cron:           str | None
     run_on_startup: bool
-    timezone:      str
-    next_run:      datetime
+    timezone:       str
+    next_run:       datetime
+    once:           ScheduledOnce | None
+    cancelled:      bool
 ```
 
 | Field | Type | Description |
 |-------|------|-------------|
 | `func` | `Callable` | The task function, already registered in the task registry. |
 | `config` | `TaskConfig` | Execution settings (retries, delay, backoff). |
-| `every` | `float \| None` | Interval in seconds between runs. `None` when `cron` is used. |
-| `cron` | `str \| None` | Five-field cron expression. `None` when `every` is used. |
+| `every` | `float \| None` | Interval in seconds between runs. `None` when `cron` is used or for one-offs. |
+| `cron` | `str \| None` | Five-field cron expression. `None` when `every` is used or for one-offs. |
 | `run_on_startup` | `bool` | Whether to fire on the first tick immediately after startup. |
 | `timezone` | `str` | IANA timezone name used when evaluating the cron expression. `"UTC"` by default. |
 | `next_run` | `datetime` | UTC time of the next scheduled execution. Updated after each firing. |
+| `once` | `ScheduledOnce \| None` | Set for one-off entries. `None` for recurring entries. |
+| `cancelled` | `bool` | Tombstone flag. Cancelling or replacing a one-off marks the heap entry dead, since `heapq` cannot remove an arbitrary element. |
+
+The `recurring` property returns `True` when `once` is `None`. Calling `compute_next()` on a one-off entry raises `ValueError`, since a one-off has no next run.
 
 ---
 

docs/changelog.md

@@ -1,5 +1,39 @@
 # Changelog
 
+## v0.12.0
+
+Adds one-off scheduled tasks: run a task once at a timestamp computed at runtime, rather than on a cadence fixed at import time.
+
+### One-off scheduled tasks
+
+- `await task_manager.schedule_once(func, *args, run_at=, run_key=, **kwargs)` schedules a single run at an exact future time. Also accepts `tags`, `priority`, `queue`, and `idempotency_key`.
+- `await task_manager.cancel_scheduled(run_key)` cancels a pending firing. `await task_manager.list_scheduled(before=None)` lists them.
+- Scheduling again with an existing `run_key` replaces the pending entry rather than adding a second one. `run_key` is a separate namespace from `idempotency_key`.
+- Pending firings persist to the backend, survive restarts, and are claimed atomically so exactly one instance fires each one.
+- Requires a backend. `schedule_once()` raises `RuntimeError` if none is configured or the backend does not support one-off schedules.
+- Added `ScheduledOnce` to the public API.
+
+### Backends
+
+- Added optional `save_scheduled`, `load_due`, `delete_scheduled`, and `claim_scheduled` to `SnapshotBackend`, plus a `supports_scheduled_once` flag. Implemented on all four built-in backends.
+- New `task_scheduled_once` table on the SQL backends, sorted set on Redis. Created automatically on first connection. No manual migration.
+
+### Scheduler
+
+- One-off entries are held in a horizon window instead of all being loaded at once. Tunable via `horizon` and `refill_interval` on `PeriodicScheduler`, defaulting to 300s and 150s.
+- `ScheduledEntry` gained `once` and `cancelled` fields and a `recurring` property. `compute_next()` raises `ValueError` on a one-off entry.
+
+### Dashboard
+
+- One-off entries are not listed individually in the Schedules tab, which now reports an armed count. Fired one-offs appear in the task list as normal with `source="scheduled"`.
+
+### Upgrade notes
+
+- Upgrading from 0.11.0 requires no action. Schema changes are additive.
+- Downgrading past 0.12.0 leaves pending entries in `task_scheduled_once` unfired, with no error.
+
+---
+
 ## v0.11.0
 
 - Timestamps (`created_at`, `start_time`, `end_time`, audit log, task logs) are now always serialized with an explicit UTC offset. The dashboard displays and filters times in the viewer's local timezone instead of misreading them as local time.

docs/guide/backends.md

@@ -72,14 +72,53 @@ Two additional methods have sensible defaults that you can override for better p
 !!! tip
     See the [API reference](../api/task-admin.md) for the full method signatures, including `claim_pending` and `acquire_schedule_lock`, which are relevant for multi-instance deployments.
 
+## Supporting one-off schedules
+
+[One-off scheduled tasks](one-off-tasks.md) need four more methods. They are optional: a backend that does not implement them simply cannot be used with `schedule_once()`, and the error surfaces at the call site rather than silently dropping the firing.
+
+To opt in, set `supports_scheduled_once = True` and implement all four:
+
+```python
+from fastapi_taskflow.models import ScheduledOnce
+
+
+class MyBackend(SnapshotBackend):
+    supports_scheduled_once = True
+
+    async def save_scheduled(self, entry: ScheduledOnce) -> None:
+        # Upsert on entry.run_key. Scheduling again with an existing key must
+        # replace the entry, not create a second firing.
+        ...
+
+    async def load_due(self, before: datetime) -> list[ScheduledOnce]:
+        # Return pending entries with fire_at at or before `before`.
+        # Index fire_at: this runs on every refill tick.
+        ...
+
+    async def delete_scheduled(self, run_key: str) -> bool:
+        # Remove the entry. Returns True if one was removed.
+        ...
+
+    async def claim_scheduled(self, run_key: str) -> bool:
+        # Atomically remove the entry and report whether THIS caller removed
+        # it. Must be a single atomic operation, since it is what guarantees
+        # exactly one instance fires the task.
+        ...
+```
+
+`claim_scheduled` is the important one. In every built-in backend it is a single delete that reports whether it removed a row, which is what makes the firing exactly-once across instances. Implementing it as a read followed by a separate delete opens a window where two instances both see the entry and both fire it.
+
+Deleting a `run_key` that does not exist is not an error. Both `delete_scheduled` and `claim_scheduled` return `False` in that case.
+
 ## Storage separation
 
-The ABC deliberately separates two concerns:
+The ABC deliberately separates three concerns:
 
 - **History** (`save` / `load`): completed tasks kept for observability and the dashboard.
 - **Requeue** (`save_pending` / `load_pending` / `clear_pending`): unfinished tasks saved at shutdown for re-execution on the next startup.
+- **One-off schedules** (`save_scheduled` / `load_due` / `delete_scheduled` / `claim_scheduled`): firings scheduled for a future time, which have not run yet and may never run if cancelled.
 
-Keep these in separate tables or key namespaces so they never mix.
+Keep these in separate tables or key namespaces so they never mix. In particular, do not store one-off schedules in the requeue namespace: requeue is loaded and dispatched immediately on startup, which would fire every pending schedule at once.
 
 ## Built-in backends