Specialized managers¶

EDRI ships two specialized 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 Router). They are documented here because they are configured and started explicitly through the EDRI entry point.

Scheduler¶

The Scheduler emits any 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 Health checking (Router) registers a periodic job with the Scheduler

Starting the Scheduler¶

from edri import EDRI

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

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

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:

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 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 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.

@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.

@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).

@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.

@event
class Cancel(Scheduler):
    identifier: str

Example: schedule a job from inside a manager.

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¶

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:

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.

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.

from edri.events.edri.store import Set

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

Example: read a value.

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:

@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 Pydantic AI (edri.integrations.pydantic_ai) for a concrete example (dispatch).

Example: subscribe to changes filtered by a predicate.

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:

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 for an active sweep):

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:

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.

Specialized managers¶

Scheduler¶

Starting the Scheduler¶

Job¶

Scheduler control events¶

Set¶

Update¶

SetUpdate¶

Cancel¶

Scheduler behavior notes¶

Key-Value Store¶

Starting the Store¶

Optional SQLite persistence¶

Options¶

How it works¶

Durability¶

Startup¶

Failures¶

Database ownership¶

TTL¶

Store control events¶

TTL and expiry¶

Atomic operations¶

Store behavior notes¶

See also¶

`Job`¶

`Set`¶

`Update`¶

`SetUpdate`¶

`Cancel`¶