# Architecture
This document describes the internal module structure, crate dependencies, and data flow of `event_base`.
---
## Workspace structure
`event_base` is a Cargo workspace with 8 crates:
```
event_base/ # Umbrella crate — re-exports all public APIs
├── event_base_core/ # Core types, traits, routing, workers, WAL trait
├── event_base_wal/ # WAL implementations (MemoryWal, PersistentWal)
├── event_base_queue/ # Queue implementations (Flume, MPMC)
├── event_base_middleware/ # Built-in middleware (Logger)
├── event_base_macro_attr/ # #[handler] procedural macro
├── event_base_macro_func/ # send_msg! and start_system! macros
├── event_base_audit/ # Audit logging subsystem
├── event_base_grpc/ # gRPC management API (optional)
└── event_base_test/ # Integration tests and benchmarks
```
---
## Crate dependency graph
```
┌─────────────────────┐
│ event_base │ (umbrella)
│ (feature gates) │
└──────┬──────┬────────┘
│ │
┌───────────────────┘ └───────────────────┐
▼ ▼
┌──────────────────────┐ ┌──────────────────────────┐
│ event_base_core │◄─── depends ────┤ event_base_macro_attr │
│ │ │ (#[handler]) │
│ • EMessage │ └──────────────────────────┘
│ • EHandler / Ack │
│ • Middleware / Next │ ┌──────────────────────────┐
│ • TopicRouter │◄─── depends ────┤ event_base_macro_func │
│ • ConsumerRouter │ │ (send_msg!,start_system!)│
│ • Worker │ └──────────────────────────┘
│ • WorkerRegistry │
│ • Shutdown strategies│ ┌──────────────────────────┐
│ • Wal trait │◄─── impl ────────┤ event_base_wal │
│ • AuditManager │ │ • MemoryWal │
│ • TraceLayer │ │ • PersistentWal │
│ • Metrics │ └──────────────────────────┘
│ • SystemHandlers │
│ • Error types │ ┌──────────────────────────┐
└──────────────────────┘◄─── impl ────────┤ event_base_queue │
│ │ • flume (MemoryQueue) │
│ ┌───────────────────────┐ │ • mpmc (Multi-prod/cons)│
├──┤ event_base_middleware │ └──────────────────────────┘
│ │ • LoggerMiddleware │
│ └───────────────────────┘ ┌──────────────────────────┐
│ │ event_base_audit │
├──────────────────────────────────┤ • Audit writers │
│ └──────────────────────────┘
│ ┌───────────────────────┐
└──┤ event_base_grpc │
│ • Management API │
└───────────────────────┘
```
---
## Module breakdown (`event_base_core`)
`event_base_core` is the central crate. Its modules:
| `message` | `EMessage`, `MessageTopic`, `MessagePayload`, `MessageMetadata`, `DeliveryMode` |
| `handler` | `EHandler` trait, `Ack` enum |
| `middleware` | `Middleware` trait, `Next`, `Pipeline` |
| `topic` | `TopicRouter` — message entry point, WAL append, delayed delivery |
| `queues` | `EProducer`/`EConsumer` traits, `QueueFactory`, `ConsumerFactory`, `ConsumerRouter` |
| `worker` | `Worker` — message processing loop, audit, WAL sync |
| `worker_registry` | `WorkerRegistry` — worker tracking, heartbeats, persistence |
| `registry` | `HANDLER_REGISTRY` — compile-time handler registration via `linkme` |
| `shutdown` | `ShutdownSender`/`ShutdownReceiver`, 7 shutdown strategies |
| `wal` | `Wal` trait, `WalRecord`, `WalRecordState`, codec, `WalClient` |
| `dead_letter` | `DeadLetterMessage`, `DeadReason` |
| `audit` | `AuditManager`, `AuditRecord`, `AuditWriter` trait |
| `trace` | `TraceRecord`, `TraceLevel` |
| `trace_layer` | `TraceLayer` — `tracing::Layer` implementation that emits trace messages |
| `metrics` | `MetricsManager`, `MetricsStore`, `NodeMetrics` |
| `system_handlers` | Built-in handlers for system topics (audit, trace, shutdown, discovery, etc.) |
| `constant` | System topic name constants (e.g., `_system.audit`) |
| `error` | Unified `CoreError` type, sub-module errors |
| `traits` | Re-exports of key traits |
---
## Data flow diagrams
### Message send flow
```
┌─────────┐ send_msg! / TopicRouter::send()
│ Sender │──────────────────────────────────────►┌──────────────┐
└─────────┘ │ TopicRouter │
│ │
│ 1. WAL.append│──►┌────────┐
│ (Pending) │ │ WAL │
│ │ └────────┘
│ 2. If delayed│──►┌────────┐
│ → schedule│ │Schedule│
│ │ └────────┘
│ 3. producer │
│ .send() │──►┌──────────────┐
└──────────────┘ │ Queue (flume)│
└──────────────┘
```
### Message processing flow
```
┌──────────────────┐ claim() ┌──────────────────┐
│ Main Consumer │──────────────►│ ConsumerRouter │
│ (dispatch loop) │ │ │
└──────────────────┘ │ 1. Select worker │
│ (by topic) │
│ 2. Forward msg │
│ 3. Ack claim │
└────────┬─────────┘
│
worker.producer.send()
│
▼
┌──────────────────┐
│ Worker │
│ │
│ 1. receive() │
│ 2. status=Working│
│ 3. WAL: mark │
│ Processing │
│ 4. Pipeline.run()│
│ ├─ Middleware1 │
│ ├─ Middleware2 │
│ └─ Handler │──► Ack
│ 5. WAL: mark │
│ Complete/Fail │
│ 6. Send audit │──► _system.audit
│ 7. status=Idle │
└──────────────────┘
```
### Crash recovery flow
```
┌─────────────┐ startup ┌──────────────────┐
│ System │───────────────►│ TopicRouter │
│ Startup │ │ .replay() │
└─────────────┘ │ │
│ 1. Load Pending │──►┌────────┐
│ from WAL │ │ WAL │
│ │ └────────┘
│ 2. For each msg: │
│ ├─ Future? │──► Re-schedule
│ └─ Now? │──► Re-send via
│ │ TopicRouter
│ └──────────► Queue
└──────────────────┘
```
### System topic interaction (distributed)
```
┌──────────┐ ┌──────────┐
│ Host │ │ Worker │
└────┬─────┘ └────┬─────┘
│ │
│◄──── worker_discovery ────────┤ Worker registers itself
│ │
│◄──── worker_heartbeat ────────┤ Periodic heartbeat
│ │
│◄──── wal_sync ────────────────┤ WAL state updates
│ │
│◄──── audit ───────────────────┤ Audit log events
│◄──── trace ───────────────────┤ Tracing spans
│ │
│──── shutdown ────────────────►│ Shutdown command
│◄──── shutdown_ack ────────────┤ Shutdown confirmation
│ │
│──── topic_sync ──────────────►│ Topic configuration
│◄──── topic_discovery ─────────┤ Worker's topic list
```
---
---
## ConsumerRouter batch dispatch
`ConsumerRouter` uses **batch claiming** to reduce lock contention in the dispatch loop. Instead of claiming one message at a time, it collects up to `batch_size` messages in a single `claim_batch()` call, dispatches them all, then acknowledges the batch in one lock acquisition.
```
loop {
consumer.lock() // one lock per batch
batch = consumer.claim_batch(N) // one pending-lock per batch
consumer.unlock()
for msg in batch { // dispatch (lock-free)
worker = select_idle_worker(topic)
worker.producer.send(msg.clone())
}
consumer.lock() // one lock for all acks
for msg in batch { consumer.ack(id) }
consumer.unlock()
}
```
### Configurable batch size
The batch size is set at initialization and defaults to 64:
```rust
ConsumerRouter::init(consumer, factory, None)?; // default 64
ConsumerRouter::init(consumer, factory, Some(128))?; // custom
```
### Batch-friendly trait method
Each queue backend should override `EConsumer::claim_batch()` for optimal performance. The optimized version acquires the internal pending-map lock once for the entire batch:
```rust
// Default (calls claim() in a loop — one pending-lock per message)
async fn claim_batch(&mut self, max: usize) -> Result<Vec<ClaimedMessage>, CoreError>
// Optimized (flume, mpmc, crossfire — single pending-lock + try_recv loop)
```
The default implementation on the trait falls back to calling `claim()` in a loop, which is correct but slower.
---
## System message template caching
`Worker` and `WalClient` cache `EMessage` templates for system topics (`_system.audit`, `_system.wal_sync`) to avoid re-creating fixed fields (topic, delivery mode, `to_worker`, default metadata) on every message. Each `process_msg` call sends 4 system messages (2 audit + 2 WAL sync); the templates reduce this overhead by cloning a pre-built message and only replacing the payload + UUID.
---
## Global singletons
The system uses several global singletons (initialized once at startup):
| `TOPIC_ROUTER` | `OnceLock<Arc<TopicRouter>>` | `TopicRouter::init()` |
| `CONSUMER_ROUTER` | `OnceLock<Arc<ConsumerRouter>>` | `ConsumerRouter::init()` |
| `WORKER_REGISTRY` | `OnceLock<Arc<WorkerRegistry>>` | `WorkerRegistry::init()` |
| `AUDIT_MANAGER` | `OnceLock<Arc<AuditManager>>` | `AuditManager::init()` |
| `NODE_NAME` | `OnceLock<Arc<String>>` | `set_node_name()` |
| `NODE_TYPE` | `RwLock<Option<Arc<NodeType>>>` | `set_node_type()` |
---
## Error handling
All errors are unified under `CoreError`:
```
CoreError
├── Queue(QueueError)
├── Wal(WalError)
├── SerdeSerialize(serde_json::Error)
├── Middleware(MiddlewareError)
├── Audit(AuditError)
├── Shutdown(ShutdownError)
├── Serialize(SerializeError)
├── Topic(TopicError)
├── Handler(HandlerError)
├── IoError(std::io::Error)
├── QueueSendError(String)
├── InvalidParameter(String)
├── AlreadyInitialized
├── NotFound(String)
├── Timeout
├── SystemShutdown
├── ErrorTime
└── Unknown(String)
```
---
## Feature flags
The umbrella crate (`event_base`) uses feature flags to conditionally include sub-crates:
```toml
[features]
default = ["memory", "macro"]
full = ["gRPC", "middleware", "macro", "memory", "persistent", "audit"]
gRPC = [] # Includes event_base_grpc
middleware = [] # Includes event_base_middleware
audit = [] # Includes event_base_audit
memory = [] # Includes event_base_wal::memory + event_base_queue::flume
persistent = [] # Includes event_base_wal::persistent
macro = [] # Includes event_base_macro_attr + event_base_macro_func
```
---
## Key design decisions
1. **WAL-first** — Messages are persisted before enqueuing. This guarantees at-least-once delivery semantics.
2. **Compile-time registration** — The `#[handler]` macro uses `linkme` distributed slices to collect all handlers at compile time, eliminating runtime registration boilerplate.
3. **Singleton architecture** — Core components (TopicRouter, ConsumerRouter, WorkerRegistry) are global singletons. This simplifies the API but limits multi-instance deployments to separate processes.
4. **Trait-based pluggability** — `Wal`, `QueueFactory`, `EProducer`/`EConsumer`, `Middleware`, and `AuditWriter` are all traits, allowing custom implementations.
5. **System topics** — Internal communication uses the same message infrastructure as user messages, ensuring consistent reliability and observability.
---
## Next steps
- [Quick Start](../guide/quick-start.md) — Build your first event_base application
- [Core Concepts](../guide/core-concepts.md) — Deep dive into the event model