Fix memory eventbus antipatterns causing silent drops and unbounded growth; add durable-memory engine (#188)

* Initial plan

* Fix eventbus memory engine antipatterns: worker pool buffer, drop logging, history cap, timer race

Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

* Add durable-memory engine: per-subscriber FIFO queue with backpressure, zero event loss

Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

* Address review feedback: fix MaxDurableQueueDepth validator, nolint explanation, engine comment

Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

Author: Copilot

modules/eventbus/README.md

@@ -15,7 +15,8 @@ The EventBus Module provides a publish-subscribe messaging system for Modular ap
 - **Worker Pool Management**: Configurable worker pools for async event processing
 
 ### Supported Engines
-- **Memory**: In-process event bus using Go channels (default)
+- **Memory**: In-process event bus using Go channels (default). Configurable delivery modes: `drop`, `block`, `timeout`.
+- **Durable Memory**: In-process event bus that **never drops events**. Uses a per-subscriber FIFO queue and blocks publishers (backpressure) when the queue is full. Ideal when event loss is unacceptable within a single process.
 - **Redis**: Distributed messaging using Redis pub/sub
 - **Kafka**: Enterprise messaging using Apache Kafka
 - **Kinesis**: AWS-native streaming using Amazon Kinesis
@@ -169,6 +170,7 @@ Tuning Guidance:
 - Start with `drop` in high-throughput low-criticality paths where occasional loss is acceptable.
 - Use `timeout` with a modest `publishBlockTimeout` (e.g. 5-50ms) for balanced fairness and latency in mixed-speed subscriber sets.
 - Reserve `block` for critical fan-out where all subscribers must process every event and you are comfortable applying backpressure to publishers.
+- Use `durable-memory` when you need zero event loss within a process without the operational overhead of an external broker.
 
 Example (balanced):
 ```yaml
@@ -184,6 +186,56 @@ eventbus:
         rotateSubscriberOrder: true
 ```
 
+### Durable Memory Engine (Zero Event Loss)
+
+The `durable-memory` engine is an in-process alternative to `memory` that **never drops events**. Instead of dropping events when a subscriber is busy, publishers block (backpressure) until the subscriber's queue has space. Memory usage is bounded by `maxDurableQueueDepth × number-of-subscribers`.
+
+**When to use:**
+- Event loss is unacceptable and the application runs in a single process
+- You need the simplicity of in-process delivery with stronger guarantees than `memory`+`block` mode
+- You want bounded memory without the operational overhead of Redis or Kafka
+
+**Single-engine configuration:**
+```yaml
+eventbus:
+  engine: durable-memory
+  maxEventQueueSize: 1000      # fallback queue depth if maxDurableQueueDepth is 0
+  maxDurableQueueDepth: 500    # per-subscriber queue depth (0 = use maxEventQueueSize)
+```
+
+**Multi-engine configuration (mixed with a lossy fast path):**
+```yaml
+eventbus:
+  engines:
+    - name: "fast"
+      type: "memory"
+      config:
+        workerCount: 8
+        deliveryMode: drop
+    - name: "critical"
+      type: "durable-memory"
+      config:
+        maxDurableQueueDepth: 200
+  routing:
+    - topics: ["payment.*", "order.*"]
+      engine: "critical"
+    - topics: ["*"]
+      engine: "fast"
+```
+
+**Trade-offs vs `memory` + `block`:**
+
+| | `durable-memory` | `memory` + `block` |
+|---|---|---|
+| Event loss | None | None |
+| Backpressure unit | Per-subscriber queue | Per-subscriber channel slot |
+| Queue structure | Linked list (unbounded growth before cap) | Fixed-size Go channel |
+| Memory bound | `maxDurableQueueDepth` | `defaultEventBufferSize` |
+| Async handler parallelism | Sequential per subscriber | Shared worker pool |
+
+**Important note on cross-process durability:**  
+`durable-memory` only protects against in-process loss (e.g. a slow subscriber). It does **not** survive process restarts or crashes. For durable-across-restarts guarantees, use Redis, Kafka, or Kinesis.
+
 ### Metrics Export (Prometheus & Datadog)
 
 Delivery statistics (delivered vs dropped) can be exported via the built-in Prometheus Collector or a Datadog StatsD exporter.

modules/eventbus/config.go

@@ -20,8 +20,8 @@ type EngineConfig struct {
 	Name string `json:"name" yaml:"name" validate:"required"`
 
 	// Type specifies the engine implementation to use.
-	// Supported values: "memory", "redis", "kafka", "kinesis", "custom"
-	Type string `json:"type" yaml:"type" validate:"required,oneof=memory redis kafka kinesis custom"`
+	// Supported values: "memory", "redis", "kafka", "kinesis", "nats", "custom", "durable-memory"
+	Type string `json:"type" yaml:"type" validate:"required,oneof=memory redis kafka kinesis nats custom durable-memory"`
 
 	// Config contains engine-specific configuration as a map.
 	// The structure depends on the engine type.
@@ -76,10 +76,10 @@ type EventBusConfig struct {
 	// --- Single Engine Configuration (Legacy Support) ---
 
 	// Engine specifies the event bus engine to use for single-engine mode.
-	// Supported values: "memory", "redis", "kafka", "kinesis"
+	// Supported values: "memory", "redis", "kafka", "kinesis", "nats", "custom", "durable-memory"
 	// Default: "memory"
 	// Note: This field is used only when Engines is empty (legacy mode)
-	Engine string `json:"engine,omitempty" yaml:"engine,omitempty" validate:"omitempty,oneof=memory redis kafka kinesis" env:"ENGINE"`
+	Engine string `json:"engine,omitempty" yaml:"engine,omitempty" validate:"omitempty,oneof=memory redis kafka kinesis nats custom durable-memory" env:"ENGINE"`
 
 	// MaxEventQueueSize is the maximum number of events to queue per topic.
 	// When this limit is reached, new events may be dropped or publishers
@@ -109,6 +109,13 @@ type EventBusConfig struct {
 	// PublishBlockTimeout is used when DeliveryMode == "timeout". Zero means no wait.
 	PublishBlockTimeout time.Duration `json:"publishBlockTimeout,omitempty" yaml:"publishBlockTimeout,omitempty" env:"PUBLISH_BLOCK_TIMEOUT"`
 
+	// MaxDurableQueueDepth is the per-subscriber queue depth for the "durable-memory" engine.
+	// When a subscriber's queue is full, publishers block (backpressure) until the subscriber
+	// consumes an event, ensuring zero event loss.
+	// When 0 (default), the value of MaxEventQueueSize is used.
+	// Set to a positive value to override independently of MaxEventQueueSize.
+	MaxDurableQueueDepth int `json:"maxDurableQueueDepth,omitempty" yaml:"maxDurableQueueDepth,omitempty" validate:"omitempty" env:"MAX_DURABLE_QUEUE_DEPTH"`
+
 	// RotateSubscriberOrder when true rotates the ordering of subscribers per publish
 	// to reduce starvation and provide fairer drop distribution.
 	RotateSubscriberOrder bool `json:"rotateSubscriberOrder,omitempty" yaml:"rotateSubscriberOrder,omitempty" env:"ROTATE_SUBSCRIBER_ORDER"`