> ## Documentation Index
> Fetch the complete documentation index at: https://usefused.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Task and cron storage

> Use PostgreSQL, Redis, or a custom database for Harnest-owned durable workers and schedules.

Use PostgreSQL, Redis, or your own storage without changing your Tasks. Harnest runs the workers, retries, and schedules; your provider persists their state.

Queued tasks require `@lifecycle.storage.tasks`. Add `@lifecycle.storage.cron` to the same factory for recurring work. There is no default database.

## Choose and register a provider

`harnest_postgres` and `harnest_redis` ship with the Harnest runtime, not as separate PyPI distributions. Managed ADK and LangGraph environments include both drivers.

From your agent folder, synchronize the matching runtime:

```bash theme={null}
harnest env sync .
```

For direct Python development from source, install the driver you need:

<CodeGroup>
  ```bash PostgreSQL theme={null}
  python -m pip install '/path/to/harnest[postgres]'
  ```

  ```bash Redis theme={null}
  python -m pip install '/path/to/harnest[redis]'
  ```
</CodeGroup>

Use `[postgres,redis]` to install both drivers. Imports do not open connections.

Register one shared factory in `lifecycle/storage.py`, replacing the corresponding existing storage factories:

<CodeGroup>
  ```python PostgreSQL theme={null}
  import os

  from harnest import lifecycle
  from harnest_postgres import PostgresStore


  @lifecycle.storage.sessions
  @lifecycle.storage.checkpoints
  @lifecycle.storage.tasks
  @lifecycle.storage.cron
  def storage() -> PostgresStore:
      """Share one managed PostgreSQL pool across durable storage roles."""
      return PostgresStore(os.environ["DATABASE_URL"])
  ```

  ```python Redis theme={null}
  import os

  from harnest import lifecycle
  from harnest_redis import RedisStore


  @lifecycle.storage.sessions
  @lifecycle.storage.checkpoints
  @lifecycle.storage.tasks
  @lifecycle.storage.cron
  def storage() -> RedisStore:
      """Use Redis for conversation state, tasks, and recurring schedules."""
      return RedisStore(os.environ["REDIS_URL"])
  ```
</CodeGroup>

| Role                     | Configuration                                                                                             |
| ------------------------ | --------------------------------------------------------------------------------------------------------- |
| Tasks and cron           | Must share one instance so scheduling and enqueueing are atomic. Omit the cron decorator for tasks alone. |
| Sessions and checkpoints | Can share this provider or use a separate one.                                                            |
| Local tests              | Use `harnest.task.MemoryTaskStore`; jobs disappear on process exit.                                       |

Harnest calls the factory, starts the provider, and closes it once. Remove unused decorators; keep any file that still provides sessions or checkpoints.

```bash theme={null}
harnest test .
harnest serve .
```

PostgreSQL creates Harnest-owned task and cron tables. For Redis, configure persistence, replication, backups, and no eviction; a cache-only deployment is not durable. Task records do not inherit session/checkpoint TTLs.

## Share connection configuration

Start your database service before the agent. Workers run inside Harnest; no separate queue engine is needed.

For reusable configuration, move the factory into `lib/`. Use this instead of the inline task/cron factory above:

<CodeGroup>
  ```python lib/task_storage.py theme={null}
  import os

  from harnest_redis import RedisStore


  def create_store() -> RedisStore:
      """Read deployment configuration without opening a connection at import time."""
      return RedisStore(os.environ["TASK_STORAGE_URL"])
  ```

  ```python lifecycle/task_storage.py theme={null}
  from harnest import lifecycle
  from harnest.lib.task_storage import create_store


  @lifecycle.storage.tasks
  @lifecycle.storage.cron
  def task_storage():
      """Share one managed connection across tasks and recurring schedules."""
      return create_store()
  ```
</CodeGroup>

For PostgreSQL, substitute `harnest_postgres.PostgresStore` and a PostgreSQL URL. Declare custom adapter dependencies in your agent's `pyproject.toml`.

From your agent folder, point the factory at your database and start the service:

```bash theme={null}
export TASK_STORAGE_URL='redis://127.0.0.1:6379/0'
harnest env sync .
harnest test .
harnest serve .
```

Match environment-variable names to your factory and inject production credentials through deployment secrets. Replicas must share the same application identity and persistent provider.

## Write a custom database adapter

Implement the public structural contracts; subclassing a Harnest store is not required:

```python theme={null}
from harnest.task import TaskRecord, TaskStore, TaskStoreConflictError
from harnest.cron import CronRecord, CronStore, CronStoreConflictError
```

| Contract    | Required operations                                                                                                                 |
| ----------- | ----------------------------------------------------------------------------------------------------------------------------------- |
| `TaskStore` | `start`, `close`, `enqueue_task`, `get_task`, `claim_tasks`, `renew_task_lease`, `finish_task`, `cancel_task`                       |
| `CronStore` | `start`, `close`, `create_cron`, `get_cron`, `list_crons`, `update_cron`, `delete_cron`, `list_due_crons`, `commit_cron_occurrence` |

Follow the protocol signatures and method docstrings. Records use UTC epoch seconds, JSON-safe payloads, and detached snapshots.

| Guarantee    | Provider responsibility                                                                                                                     |
| ------------ | ------------------------------------------------------------------------------------------------------------------------------------------- |
| Enqueue      | Persist payload and job atomically. Scope idempotency to application, user, task, and key; retain a fingerprint after payload cleanup.      |
| Leases       | Claim atomically with a fresh attempt token. Fence renewals and results with the matching, unexpired token.                                 |
| Retries      | Persist timing and attempt counts; terminalize exhausted crashed attempts.                                                                  |
| Cancellation | Invalidate leases atomically; scrub arguments, invocation snapshots, and permissions at terminal transitions.                               |
| Isolation    | Apply owner predicates, ordering, and bounded pagination in the database. `get_task(user_id=None)` is for trusted workers, not Agent Tools. |
| Cron         | Revision-check edits; atomically commit each occurrence and advance its cursor through the same provider.                                   |

Register the adapter with `@lifecycle.storage.tasks` and, when supported, `@lifecycle.storage.cron`.

## Test a custom adapter

Put this in your adapter package's tests and run it against an isolated real database, not only a mock:

```python theme={null}
import unittest

from harnest.testing import TaskStoreConformanceMixin
from my_storage import MyStore


class MyStoreContractTests(TaskStoreConformanceMixin,
                           unittest.IsolatedAsyncioTestCase):
    async def make_store(self) -> MyStore:
        """Return a fresh facade over an isolated test database."""
        return MyStore(...)
```

```bash theme={null}
python -m unittest discover -s tests -v
```

The suite checks isolation, concurrent claims, leases, deduplication, revisions, and atomic scheduling. Add database-specific crash, rollback, connection-loss, and restart tests. Verify deployment persistence separately.

## Execution and recovery guarantees

Delivery is **at least once**, not exactly-once external effects. A worker may crash after sending an email but before committing its result. Keep side effects idempotent. Cancellation prevents later state commits and requests cooperative execution cancellation; it cannot undo an effect or forcibly stop a synchronous function already running in a thread.

The provider runtime polls for work, renews attempt leases, and periodically reconciles retained results with durable continuation storage. Use a persistent Harnest-owned checkpointer for `handle.result()` to resume across restarts; persistent Task storage alone does not make a memory checkpointer durable. Queue ordering is best-effort, not strict FIFO across replicas and recovery.

Cron cursors survive restart. Missed occurrences catch up in bounded batches; there is no skip/coalesce misfire option in this version. Cancelling a schedule before its atomic occurrence commit prevents that enqueue; already committed jobs continue. Removed static declarations are paused at the next deployment startup, while a retargeted declaration gets a new identity. Do not run old and new conflicting static declarations concurrently during a rolling deployment.

## Move an existing application safely

The Procrastinate backend and automatic PostgreSQL fallback have been removed. If you used them, drain work on your currently deployed version **before upgrading**. The new runtime cannot process the old queue tables.

1. Back up your database and inventory recurring schedules for each owner.
2. Stop new submissions and pause recurring schedules on the old runtime.
3. Drain or explicitly cancel queued/running Tasks and resolve their waiting
   continuations before stopping the old workers.
4. Configure the new provider, preserving session/checkpoint storage when it
   contains history you need. Synchronize the runtime and recompile your agent.
   Start the new deployment and recreate dynamic
   schedules under their original user scopes. Static declarations reconcile
   automatically.
5. Verify task execution, recovery, and owner isolation before reopening traffic.

There is no automatic copy of old queue rows, schedule IDs, results, or continuation references. Keep the old database until its retention and recovery needs are satisfied. Likewise, drain work before renaming/removing a Task or changing its queue: workers subscribe to currently compiled queues, not every historical queue in the database.