# Architecture Overview

The **Event Driven Routing Infrastructure (EDRI)** is a framework designed to facilitate the development of applications that efficiently utilize parallelism across multiple threads, processes, or even distributed systems. The architecture is built around the concepts of **Events** and a **routing infrastructure** to manage communication and task distribution.

---

## Design philosophy

EDRI was designed around three core principles:

1. **Events as universal primitives** — All communication happens through typed Event objects. Components never call each other directly.
2. **Dispatch by type** — The Router delivers events based on their Python type, not their content. This keeps routing simple and predictable.
3. **Explicit concurrency** — Managers are single-threaded by default. Concurrency must be explicitly delegated to Workers, making parallelism a conscious design decision.

---

## Process model

EDRI uses **multiprocessing** as its primary isolation mechanism:

```mermaid
flowchart TD
    subgraph Main["Main Process (Router)"]
        direction TB
        RQ["Router Queue<br/>+ Health Checker"]
    end
    subgraph M1["Manager Process"]
        direction TB
        W1["Worker (Thread)"]
    end
    subgraph M2["Manager Process"]
        direction TB
        W2["Worker (Process)"]
    end
    subgraph API["API Listener (ASGI Server)"]
        direction TB
        L["Listener (request parsing)"]
    end
    subgraph BRK["Broker Process"]
        direction TB
        BR["Broker (Middleware)"]
    end
    Main -->|multiprocessing.Connection| M1
    Main -->|multiprocessing.Connection| M2
    API -->|multiprocessing.Queue| BRK
    BRK -->|multiprocessing.Queue| Main
```

**Key points:**

* The **Router** runs in the main process.
* Each **Manager** runs in its own child process (started via `multiprocessing.Process`). `Manager.__init__()` is called during construction in the parent process; the child process begins in `Manager.run()`.
* **Workers** run inside their Manager's process — either as threads (`WorkerThread`) or subprocesses (`WorkerProcess`).
* The **API Listener** and the **Broker** each run in their own child process. The Listener parses requests; the Broker runs middleware before forwarding events to the Router.
* The **Connector** also runs as its own child process with sender/receiver threads inside.
* All inter-process communication happens through `multiprocessing.Queue` or `multiprocessing.Connection` (pipes), with events serialized via `pickle`.

---

## Core components

| # | Component | Process model | Role |
|---|-----------|---------------|------|
| 1 | **Event** | Data (serializable) | Universal message type |
| 2 | **Router** | Main process | Central dispatcher |
| 3 | **Manager** | Child process | Event handler, task distributor |
| 4 | **Worker** | Thread or subprocess | Task executor |
| 5 | **API (Listener)** | Child process | External protocol adapter |
| 6 | **Broker** | Child process | Middleware pipeline |
| 7 | **Connector** | Child process | Router ↔ Switch bridge |
| 8 | **Switch** | Standalone process | Cross-node router hub |
| 9 | **Scheduler** | Manager (child process) | Time-based event emitter |
| 10 | **Key-Value Store** | Manager (child process) | In-memory shared storage |

---

## Component interaction diagram

```mermaid
graph TB
    SW((Switch))
    subgraph EDRI 1
        direction TB
        CN1((Connector A))
        R1((Router))
        M1((Manager A))
        M2((Manager B))
        BR1((Broker))
        L1((HTTP))
        SC1((Scheduler))
        KV1((Key-Value Store))

        R1 -- IPC <--> M1
        R1 -- IPC <--> M2
        M1 -- IPC <--> W1A((Worker 1))
        M1 -- IPC <--> W2A((Worker 2))
        M2 -- IPC <--> W1B((Worker 1))
        M2 -- IPC <--> W2B((Worker 2))
        R1 -- IPC <--> BR1
        R1 -- IPC <--> SC1
        R1 -- IPC <--> KV1
        BR1 -- IPC <--> L1
    end

    subgraph EDRI 2
        direction TB
        CN2((Connector B))
        R2((Router))
        M3((Manager C))
        M4((Manager D))
        BR2((Broker))
        L2((HTTP))

        R2 -- IPC <--> M3
        R2 -- IPC <--> M4
        M3 -- IPC <--> W1C((Worker 1))
        M3 -- IPC <--> W2C((Worker 2))
        M4 -- IPC <--> W1D((Worker 1))
        M4 -- IPC <--> W2D((Worker 2))
        R2 -- IPC <--> BR2
        BR2 -- IPC <--> L2
    end

    SW -- TCP/IP <--> CN1
    SW -- TCP/IP <--> CN2
    CN1 -- IPC <--> R1
    CN2 -- IPC <--> R2
```

---

## Event flow (detailed)

### Local flow (no Switch)

```mermaid
flowchart TD
    P["Producer<br/>(Manager / API / Worker)"] -->|"put(Event) on Router Queue"| R[Router]
    R -->|"by Event type<br/>to subscribers"| M[Manager]
    M -->|"execute @request<br/>or @response"| H{Handler}
    H -. "optional: workers.start(Event, WorkerClass)" .-> W["Worker.do()<br/>event_send(Event)<br/>resolve_callback_worker()"]
    H -. "optional: router_queue.put(new_event)" .-> R
    H -->|"returns → response auto-sent"| R
```

### Distributed flow (with Switch)

```mermaid
flowchart TD
    MA["Manager A<br/>(Node 1)<br/>shareable Event"] --> RA[Router A]
    RA --> CA[Connector A]
    CA -->|TCP/IP| SW[Switch]
    SW -->|"looks up destination router UUID"| CB[Connector B]
    CB -->|TCP/IP| RB[Router B]
    RB --> MB["Manager B (Node 2)"]
```

---

## Manager lifecycle: construction vs. execution

Manager construction (`Manager.__init__`) happens in the **parent (main) process**
when `EDRI.add_component()` or `Manager()` is called. The actual event loop
starts in a **child process** after `Manager.start()` is called. This distinction
matters for what resources are safe to create where:

| Operation | Process | When |
|-----------|---------|------|
| `Manager.__init__()` | Parent | During `EDRI.add_component()` |
| `Router` construction | Parent | During `EDRI.run()` |
| `Router.run()` | Parent | Main event loop |
| `Manager.start()` → child process | Parent | During `EDRI.run()` |
| `Manager._prepare_resolvers()` | Child | Start of `Manager.run()` |
| `Manager._subscribe()` | Child | After resolver preparation |
| `Manager.after_start()` | Child | After subscription |
| `Manager.run_resolver()` | Child | Event processing loop |
| `Manager.quit()` | Parent | During shutdown (signals child) |

### Safe resource creation

> **Pickle warning:** `Manager` extends `multiprocessing.Process`. When running under the **spawn** or **forkserver** start methods (defaults on macOS, Windows, and Linux since Python 3.14, and also used by `WorkerProcess` with a custom Python environment), the Manager object constructed in the parent process is **pickled and sent to the child**. Any object created in `__init__` must therefore be pickle-safe or the child will fail to unpickle it. With **fork** (the default on Linux for Python ≤ 3.13), the child inherits a copy of the parent's memory, so pickle is not involved — but file descriptors, locks, and network connections are still shared in unsafe ways. The safe rule: create heavyweight resources (DB pools, sockets, threads) in `after_start()`, not in `__init__`.

Since ``multiprocessing`` uses different start methods depending on platform
and Python version, EDRI does not enforce a specific one. Your application
code must be compatible with **fork**, **forkserver**, and **spawn**:

| Platform / Python | Default start method | Process object | Child `__init__` |
|-------------------|---------------------|----------------|-----------------|
| Linux, Python ≤ 3.13 | ``fork`` | Inherited via copy-on-write — no serialisation | Not called. Child begins at ``run()`` |
| Linux, Python 3.14+ | ``forkserver`` | Serialised (pickled) to the server process, then forked | Not called. Child begins at ``run()`` |
| macOS | ``spawn`` | Serialised (pickled), unpickled in a fresh interpreter | Not called. Child begins at ``run()`` |
| Windows | ``spawn`` | Serialised (pickled), unpickled in a fresh interpreter | Not called. Child begins at ``run()`` |

Certain resources must not be created until the child process is running:

| Resource | Safe in `__init__` (parent) | Safe in `after_start()` (child) |
|----------|:---------------------------:|:-------------------------------:|
| Plain configuration | Yes | Not needed |
| Logger | Verify | Verify |
| DB connection | Typically **no** | Yes |
| Connection pool | Typically **no** | Yes |
| Open socket | Typically **no** | Yes |
| Thread | **No** | Yes |
| Async client | Typically **no** | Yes |
| Pickle-safe dataclass | Yes | Not needed |



## Lifecycle of a Manager process (diagram)

```mermaid
flowchart TD
    Start["EDRI.run() called"] --> Router["Router(queue, components)<br/>→ HealthChecker created"]
    Router --> CStart["for each component:<br/>component.start(router_queue)<br/>→ multiprocessing.Process.start()"]
    subgraph parent["in main process"]
        direction TB
        Init["Manager.__init__()<br/>(called when EDRI.add_component() or<br/>Manager() is constructed)"]
        CStart --> RouterLoop["Router.run()<br/>read queue<br/>dispatch + health checks"]
    end
    subgraph child["in child process (Manager.run())"]
        direction TB
        Resolvers["_prepare_resolvers()"] --> Static["_subscribe_static_resolvers()"]
        Static --> Sub2["_subscribe()<br/>pipe creation + Subscribe event<br/>→ Router registers subscription"]
        Sub2 --> Prepare["_prepare()"]
        Prepare --> After["after_start() hook<br/>e.g. create Cache"]
        After --> Undelivered["_resolve_undelivered_events()"]
        Undelivered --> Loop["run_resolver()<br/>read pipes + resolve"]
    end
    CStart -.-> Resolvers
    RouterLoop -->|KeyboardInterrupt / SIGTERM| Q1["Router.quit()"]
    Q1 --> Q2["each component.quit()"]
    Q2 --> Q3["Connector.quit() (if started)"]
    Q3 --> Q4["Watcher.quit() (if started)"]
    Q2 --> Q5["terminate_processes()<br/>→ SIGTERM to all children"]
```

---

## Data flow for an API request

```mermaid
sequenceDiagram
    participant C as Client
    participant L as API Listener
    participant B as Broker
    participant R as Router
    participant M as Manager
    C->>L: HTTP request
    Note over L: parse method/path/headers/body<br/>match URL to @api(...) Event<br/>extract fields, run validators (inject)
    L->>B: Event (Broker Queue)
    Note over B: middleware.process_request(...)
    B->>R: Event (Router Queue)
    R->>M: dispatch to subscribed Manager
    Note over M: execute @request handler<br/>(optionally start Worker)
    M->>R: response
    R->>B: deliver response
    Note over B: middleware.process_response(...)
    B->>L: response
    Note over L: serialize (JSON / HTML / WS)
    L->>C: HTTP response
```

---

## Component responsibilities

### Router

* Maintains a mapping of `Event type → set[Connection]` (multiprocessing pipes to subscribed Managers).
* Reads events from the main router queue.
* Dispatches each event to all subscribed managers.
* If an event is shareable, also sends it through the Connector to the Switch.
* Runs a HealthChecker that periodically verifies Manager processes are alive.
* Maintains an event cache to survive Switch outages.

### Manager

* Subscribes to event types via `@request` and `@response` decorators.
* Runs in its own process with its own input queue.
* Selects the correct handler method based on event type.
* May delegate work to Workers (thread or process).
* **Sole mediator of Worker traffic.** Workers have no direct connection to the Router; every event a Worker emits is read off its pipe by the Manager's resolver loop and may be forwarded to the router queue.
* Because all Worker traffic passes through it, the Manager can inspect, tag, modify, drop, or reroute any event before it reaches the rest of the system.
* May emit new events back to the Router.
* May override `after_start()` for one-time initialization.
* Responses are sent automatically when the handler returns (if `event.response` was modified).

### Worker

* `WorkerThread` — runs in the Manager process (shared memory, GIL-bound).
* `WorkerProcess` — runs in a subprocess (isolated memory, true parallelism).
* Receives an event and executes `do()`.
* Must call `self.event_send(event)` explicitly to send results back.
* `WorkerQuit` is emitted automatically when `do()` finishes.

### API Listener

* Implements the ASGI protocol.
* Supports multiple server backends (uvicorn, hypercorn, daphne).
* Parses HTTP requests, WebSocket messages, and serves static assets.
* Constructs API Event instances from incoming data.
* Runs field validators (`inject(...)`).
* Serializes responses (JSON, HTML, or WebSocket frames).

### Broker

* Sits between the Listener and the Router in a separate child process.
* Runs the middleware pipeline (`process_request` / `process_response`) on every event.
* Manages active WebSocket client subscriptions.
* Forwards events that pass middleware to the Router Queue.
* Receives responses from the Router and routes them back to the originating Listener socket.

### Switch

* Standalone TCP/IP server.
* Maintains a dict of `UUID → Connection` for connected EDRI instances.
* Transparently forwards serialized events between EDRI instances.
* Preserves queues during disconnection (no event loss on reconnect).
* Purely in-memory — no persistence.

---

## Summary

EDRI's architecture separates **routing** (Router), **business logic** (Manager), **execution** (Worker), and **external communication** (API/Connector). This separation allows each concern to scale independently:

* Add more API nodes behind a load balancer for HTTP throughput.
* Add more Manager processes for parallel event handling.
* Add more Workers inside a Manager for concurrent task execution.
* Connect multiple EDRI instances through the Switch for geographic distribution.

All communication stays event-driven, making the system loosely coupled and extensible.

## Deployment roles and scaling

| Role | Replicatable | Effect of multiple replicas | Needs leader election |
|------|:------------:|----------------------------|:--------------------:|
| API + Broker | Yes | Better HTTP throughput behind a load balancer | No |
| Stateless Manager | No (see note) | Duplicate event handling — each replica runs every handler | No |
| Stateful / side-effect Manager | No | Duplicate side effects — **not safe** without external coordination | Yes |
| Scheduler | No | Duplicate scheduled events — use at most one replica | Yes |
| Store (key-value) | No | Inconsistent state — single replica only | N/A |
| Switch | No | Single-instance; in-memory only | N/A |

> **Stateless Manager note:** The Router fans out to **all** subscribers of a given event type. Two identical Managers will both receive and handle every matching event. This is safe only for idempotent handlers. For competing-consumer semantics, use Workers inside a single Manager.

(singleton-operations)=
### Singleton operations

EDRI has no built-in leader election. For operations that must run exactly once (scheduled jobs, migrations, cleanup, payments, notifications):

| Approach | How |
|----------|-----|
| Run on a single replica | Use `replicas: 1` in Kubernetes, or run a single EDRI instance |
| External lock | Use Redis lock, PostgreSQL advisory lock, or similar |
| Scheduler on dedicated node | Run the Scheduler Manager only on one EDRI instance; other instances are workers only |
| Idempotent handlers | Design handlers to tolerate duplicate execution (e.g., use `INSERT ... ON CONFLICT DO NOTHING`)

## See also

* {doc}`/event` — defining events and responses
* {doc}`/router` — routing, subscriptions, caching
* {doc}`/manager` — manager lifecycle, workers, caching
* {doc}`/worker` — thread vs process workers
* {doc}`/connector` — connecting to the Switch
* {doc}`/switch` — the distributed router hub
* {doc}`/api` — REST, WebSocket, HTML API layer