# Specialized managers

EDRI ships two specialized {doc}`/manager` implementations:

* the **Scheduler** — time-based event emitter
* the **Key-Value Store** — in-memory data store + change callbacks

Both are regular Managers (they subscribe to events through the {doc}`/router`). They are documented here because they are configured and started explicitly through the `EDRI` entry point.

---

## Scheduler

The Scheduler emits any {doc}`/event` at a defined point in time, optionally on a repeating interval. It is the only built-in way to obtain time-driven events inside EDRI.

It is typically used for:

* periodic background work (cache refresh, cleanup, backups)
* one-shot timers (delayed retry, deferred notification)
* internal framework heartbeats — for example the Router’s {ref}`health-checking-router` registers a periodic job with the Scheduler

### Starting the Scheduler

```python
from edri import EDRI

edri = EDRI()
edri.start_scheduler()
edri.run()
```

Optionally, initial `Job` instances can be loaded at startup, either as a list:

```python
from datetime import datetime, timedelta
from edri import EDRI
from edri.utility.manager.scheduler import Job
from my_events import HeartBeat

edri = EDRI()
edri.start_scheduler(jobs=[
    Job(event=HeartBeat(), when=datetime.now(), repeat=timedelta(seconds=30)),
])
edri.run()
```

…or as a mapping, where the keys become the job identifiers:

```python
edri.start_scheduler(jobs={
    "heartbeat": Job(event=HeartBeat(), when=datetime.now(), repeat=timedelta(seconds=30)),
})
```

Jobs passed as a list get a generated identifier, which the application never
sees. A mapping key *is* the identifier the job is stored under, so the job can
later be rescheduled or removed by name — the `Update`, `SetUpdate` and `Cancel`
{ref}`control events <scheduler-control-events>` address jobs by that identifier.

### `Job`

`Job` is a small dataclass that describes one scheduled emission.

| Field      | Type                  | Meaning                                                                  |
| ---------- | --------------------- | ------------------------------------------------------------------------ |
| `event`    | `Event`               | event instance to emit when the job fires                                |
| `when`     | `datetime`            | first execution time (and the anchor for repetition)                     |
| `repeat`   | `timedelta \| None`   | repetition interval — if `None`, the job runs once and is then discarded |

When a job fires, the event instance stored in `Job.event` is put on the router queue. If `repeat` is set, `when` is shifted forward by `repeat`; otherwise the job is removed from the Scheduler.

(scheduler-control-events)=
### Scheduler control events

Scheduler control is also event-driven. The events live in `edri.events.edri.scheduler`.

| Event       | Purpose                                                            | Has Response |
| ----------- | ------------------------------------------------------------------ | ------------ |
| `Set`       | create a new job. Returns the assigned `identifier`                | Yes          |
| `Update`    | update fields (`event`, `when`, `repeat`) of an existing job       | Yes          |
| `SetUpdate` | create the job if missing, otherwise update it                     | Yes          |
| `Cancel`    | remove a job by `identifier`                                       | No           |

#### `Set`

Creates a new job. Requires `event` and `when`. Optional `repeat` for periodic execution. Optional `identifier` for explicit naming; if not provided, the Scheduler generates a random one and returns it in `response.identifier`.

```python
@event
class Set(Event):
    event: Event
    when: datetime
    repeat: Optional[timedelta] = None
    identifier: Optional[str] = None
    response: SetResponse  # SetResponse has identifier: str
```

#### `Update`

Updates an existing job by `identifier`. Only provided fields are updated (partial update). `identifier` is required; `event`, `when`, `repeat` are optional.

```python
@event
class Update(Scheduler):
    identifier: str
    event: Optional[Event] = None
    when: Optional[datetime] = None
    repeat: Optional[timedelta] = None
    response: UpdateResponse
```

If the target job does not exist, the response status is set to `FAILED`.

#### `SetUpdate`

Upsert: if the job exists its fields are updated, otherwise a new job is created. Unlike `Update`, `event` and `when` have **no defaults** — the constructor requires both to always be supplied (passing them explicitly as `None` is what lets the create branch detect missing data and fail with `FAILED`).

```python
@event
class SetUpdate(Event):
    event: Event
    when: datetime
    repeat: Optional[timedelta] = None
    identifier: Optional[str] = None
    response: SetUpdateResponse   # carries the generated identifier
```

Like `Set`, when `SetUpdate` creates a new job without an explicit `identifier`, the generated id is returned in `response.identifier`. When required data (`event`/`when`) is missing on a create, the response status is set to `FAILED`.

#### `Cancel`

Deletes a job by `identifier`. Silently ignores if not found.

```python
@event
class Cancel(Scheduler):
    identifier: str
```

Example: schedule a job from inside a manager.

```python
from datetime import datetime, timedelta
from edri.events.edri.scheduler import Set
from my_events import Backup

self.router_queue.put(
    Set(
        event=Backup(),
        when=datetime.now() + timedelta(hours=2),
        repeat=timedelta(days=1),
        identifier="nightly-backup",
    )
)
```

### Scheduler behavior notes

* Resolution is bounded by `SCHEDULER_TIMEOUT_MAX`. The Scheduler wakes up at least that often even when no jobs are queued.
* `repeat` is anchored to the previous `when`, not to wall-clock time at firing. If processing is delayed, the next firing keeps the original cadence rather than drifting forward.
* When the Scheduler is shut down (e.g. via `KeyboardInterrupt`), pending jobs are dropped — there is no on-disk persistence.
* **Timezone handling.** Naive `datetime` values (without `tzinfo`) are interpreted as the system's local time and converted to an aware datetime automatically. Timezone-aware values are preserved as-is. Comparisons work correctly across different timezones — a job scheduled with `when` in UTC+2 will fire at the same absolute time as one in UTC.

---

## Key-Value Store

The Key-Value Store (`Store`) is a specialized Manager that exposes a process-isolated in-memory `dict[str, Any]` through events. It also supports **change callbacks**: another component can register interest in a key and be notified through an event whenever that key is written.

Typical uses:

* sharing computed state across managers without coupling them directly
* caching expensive lookups across requests
* fan-out notifications when a value of interest changes

### Starting the Store

```python
from edri import EDRI

edri = EDRI()
edri.start_store()
edri.run()
```

Only one Store should be started per appliance — its events are unique by name and would otherwise be delivered twice.

### Optional SQLite persistence

Pass a database path to restore KV data on startup and save changes asynchronously:

```python
from edri import EDRI

edri = EDRI()
edri.start_store(path="/var/lib/edri/store.db")
edri.run()
```

```{warning}
Values are stored with Python `pickle`, and loading a pickle can execute arbitrary
code. Open only database files from a trusted source and protect the file and its
directory from untrusted writes.
```

#### Options

| Option | Default | Meaning |
| ------ | ------- | ------- |
| `path` | `None` | SQLite file (`str` or `pathlib.Path`); its directory must exist. `None` keeps the Store purely in memory and creates no queue, thread, or connection |
| `failure_policy` | `PersistenceFailurePolicy.BUFFER` | Behavior on storage failure (see Failures below); import the enum from `edri.utility.manager` |
| `startup_timeout` | `30.0` | Seconds the parent waits for restoration; `None` waits without a deadline |

`EDRI.start_store(...)` accepts all three. For direct construction use
`Store(router_queue, path=...)` from `edri.utility.manager`.

#### How it works

RAM remains the source of truth. Reads and atomic operations run in the single
Store event loop and never wait for SQLite. A dedicated writer thread commits final
values and deletes in ordered batches. Only KV data and TTL metadata are saved, not
an event history; callbacks and other runtime objects are not restored.

Each value is pickled in the event loop before it is queued, so every update is
saved as it was when the event was handled. Large values add serialization time to
the loop. Callback conditions should be free of side effects, because later
in-place changes to a value are not saved until the next Store update. Keys with
lone surrogates are supported.

#### Durability

* An event response confirms the RAM operation, not durable storage.
* A crash or machine failure can lose the unwritten tail of the queue. There is no
  per-operation `fsync`.
* Normal shutdown flushes pending changes for at most three seconds. Changes still
  pending at that point are lost.

#### Startup

Persistent `Store.start()` and `EDRI.start_store(path=...)` return after the child
process has restored the database; a running Router is not required. Failures and
timeouts raise `RuntimeError` and the child is cleaned up. Bootstrap waits for
SQLite locks for up to five seconds, independent of `startup_timeout`. The in-memory
`start_store()` still starts the process asynchronously.

Call `start_store(path=...)` before `EDRI.run()`: Managers added with
`add_component` and the Router start only in `run()`, so none of them sees an empty
Store. Call it before `start_api()`, `start_broker()` and `start_scheduler()` as
well. Those start their processes immediately, so after them the application
accepts requests or fires jobs while the Store is still restoring.

```python
edri = EDRI()
edri.start_store(path="/var/lib/edri/store.db")  # returns once data is restored
edri.start_scheduler()
edri.add_component(MyManager())
edri.start_api()
edri.run()
```

An unreadable database or pickle row aborts startup and logs the affected key; the
Store never replaces saved data with an empty one. Restore the compatible Python
classes or repair the database, then start again.

#### Failures

With `BUFFER`, the only policy, a storage failure does not interrupt the Store: it
keeps answering from RAM while the writer retries the failed batch with capped
backoff and drains the backlog in order after recovery. Status is logged on each
retry.

* The queue is unbounded, so **a prolonged outage grows memory use without limit**.
* Permanent errors, such as a read-only or corrupt database, keep later batches
  waiting behind the failed one until the problem is repaired.
* If a value cannot be pickled, the update is logged as `CRITICAL` and skipped, and
  a delete of that key is queued so an older saved value does not reappear. The
  value stays in RAM.
* If the writer thread exits unexpectedly, the next KV update or flush starts a
  replacement that resumes the pending commands first.

#### Database ownership

Only one Store may use a database file. On Linux/POSIX an advisory `flock` is held
from bootstrap until the writer stops, and a second Store (including one using a
hard link or symlink) fails to start. The kernel releases it when the process exits,
even after `SIGKILL`.

* Do not replace the file while its Store is running.
* Other SQLite clients ignore the lock and must not modify the KV data.
* Writer lock waits are limited to 0.1 seconds and contention is retried through
  `BUFFER`. The journal mode is not changed.
* Each Store keeps its own RAM state. Restarting a stopped Store process is outside
  this feature.

#### TTL

At runtime TTL uses a monotonic clock. SQLite stores absolute wall-clock expiry
times, so time spent down counts toward TTL. On startup expired rows are skipped
and each live row's remaining TTL becomes a new monotonic deadline. Explicit
deletes, lazy expiry, and `StorePurge` also delete the saved row.

Wall-clock corrections can lengthen or shorten the remaining TTL across a restart,
so persistent TTL alone is not safe for a distributed lease or leader-election
lock.

### Store control events

The events live in `edri.events.edri.store`.

| Event          | Purpose                                                                          |
| -------------- | -------------------------------------------------------------------------------- |
| `Set`          | store `value` under `name`, with an optional `ttl` (seconds). Triggers any matching `GetCallback` subscriptions |
| `Get`          | read the value of `name`. Returns `response.value` (or `None` if missing/expired) |
| `Delete`       | remove `name` from the store                                                    |
| `GetCallback`  | subscribe to writes on `name`, optionally filtered by a `condition(value)` test |
| `Increment`    | atomically add `amount` (default 1) to a counter; returns the new `response.value` |
| `CompareAndSet`| set `value` only if the current value equals `expected`; returns `response.success` |
| `SetIfAbsent`  | set `value` only if the key is absent/expired; returns `response.acquired`       |
| `StorePurge`   | sweep and drop all expired keys (fire-and-forget)                                |

Example: write a value.

```python
from edri.events.edri.store import Set

self.router_queue.put(Set(name="latest_export", value=export_path))
```

Example: read a value.

```python
from edri.events.edri.store import Get

self.router_queue.put(Get(name="latest_export"))
```

The `Get` reply comes back as a normal response event, so you consume it with a `@response` handler for `Get`:

```python
@response
def store_value(self, event: Get) -> None:
    path = event.response.value   # None if the key was missing
```

> If a **Worker** (not the Manager) emitted the `Get`, the Manager routes the reply straight back to that Worker's pipe instead of invoking the `@response` handler — the worker that asked is the one that receives the value.
>
> This works even if the event you send instead gets handled by a Worker on
> the *target* Manager too (a second worker-owning hop). `ManagerBase` tracks
> any event a Worker sends out whose response isn't filled in yet, keyed on
> `Event._id` (universal on every `Event`, unaffected by `_worker` being
> re-stamped on each hop), and forwards the reply back to the dispatching
> Worker's pipe once it arrives — no per-type registration or manual
> forwarding handler needed. See {doc}`/integrations/pydantic_ai`
> for a concrete example (`dispatch`).

Example: subscribe to changes filtered by a predicate.

```python
from edri.events.edri.store import GetCallback

self.router_queue.put(
    GetCallback(
        name="job_state",
        condition=lambda value: value == "done",
    )
)
```

When something later runs `Set(name="job_state", value="done")`, the Store re-emits a `GetCallback` event carrying the new value, which is then routed to the subscriber.

### TTL and expiry

Any write (`Set`, `Increment`, `CompareAndSet`, `SetIfAbsent`) accepts an optional `ttl` in seconds. A key with a TTL expires after that interval:

```python
self.router_queue.put(Set(name="otp:42", value="931204", ttl=300))  # forgotten in 5 min
```

Expiry is **lazy** — an expired key is dropped the next time it is read (or written). To reclaim memory held by keys that expire but are never read again, emit `StorePurge` (schedule it periodically via the [Scheduler](#scheduler) for an active sweep):

```python
from edri.events.edri.store import StorePurge
self.router_queue.put(StorePurge())
```

### Atomic operations

Because the `Store` is a single Manager that processes events **sequentially**, its read-modify-write operations are **atomic by construction — no locks required**. This makes the Store a coordination primitive for the whole appliance:

```python
from edri.events.edri.store import Increment, SetIfAbsent, CompareAndSet

# Shared counter — e.g. a rate-limit window (returns the new count in response.value)
self.router_queue.put(Increment(name="ratelimit:user42", amount=1, ttl=60))

# TTL'd lock / leader election — response.acquired is True only for the winner
self.router_queue.put(SetIfAbsent(name="lock:nightly-job", value=self.name, ttl=120))

# Optimistic concurrency — apply only if the value is unchanged (response.success)
self.router_queue.put(CompareAndSet(name="doc:7:version", expected=4, value=5))
```

Replies come back as normal response events — consume them with a `@response` handler (or, for a Worker's request, straight back to the worker's pipe), exactly like `Get` (see the caveat above about what this routing does and does not cover).

### Store behavior notes

* `Store` runs as its own process. Values cross a process boundary, so `value` must be picklable.
* The store is purely in-memory — restarting EDRI clears all keys.
* `condition` is a callable evaluated inside the Store process. Keep it side-effect free and small.
* Multiple components may subscribe to the same key; each receives its own `GetCallback` notification.

---

## See also

* {doc}`/manager` — base Manager behavior the Scheduler and Store inherit
* {doc}`/router` — how scheduler-emitted events are dispatched
* {doc}`/event` — defining the events you will schedule or store