> ## Documentation Index
> Fetch the complete documentation index at: https://abxbus.archivebox.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Backpressure

> How emit, queueing, and history limits interact under high event volume.

Backpressure in `abxbus` is history-policy based, not queue-capacity based.

* `emit()` enqueues synchronously and returns immediately.
* Pending queues are unbounded in each runtime.
* Overload behavior is controlled by `max_history_size` + `max_history_drop`.

## 1) If I emit 1,000,000 events, will errors be raised?

### Error conditions

| Runtime    | Condition                                                                                                          | What is raised                                                |
| ---------- | ------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------- |
| Python     | `emit()` called with no running event loop                                                                         | `RuntimeError` (`emit() called but no event loop is running`) |
| Python     | `max_history_size > 0`, `max_history_drop=False`, and history already at limit                                     | `RuntimeError` (`history limit reached`)                      |
| TypeScript | `emit()` with `max_history_size > 0`, `max_history_drop=false`, and history already at limit                       | `Error` (message contains `history limit reached`)            |
| Rust       | typed `emit::<EventType>(...)` with `max_history_size > 0`, `max_history_drop=false`, and history already at limit | panic with `history limit reached`                            |
| Go         | `Emit()` with `MaxHistorySize > 0`, `MaxHistoryDrop=false`, and history already at limit                           | panic with `history limit reached`                            |
| All        | Process runs out of memory under extreme load                                                                      | Runtime/VM OOM failure (not a bus-specific exception type)    |

In normal operation, queue-capacity errors are not the backpressure mechanism.

`max_history_size=0` / `MaxHistorySize=0` is a special case in all runtimes: it does not trigger history-limit rejection, and instead keeps only in-flight visibility.

With `max_history_drop=true`, `emit()` does not reject on history size. Under sustained overload, old uncompleted entries can be dropped and a warning is logged.

### Reject vs drop behavior

<Tabs>
  <Tab title="Python">
    ```python theme={null}
    from abxbus import EventBus

    # Reject new emits once history reaches N
    reject_bus = EventBus(max_history_size=10_000, max_history_drop=False)

    # Never reject on history size; trim oldest history entries instead
    drop_bus = EventBus(max_history_size=10_000, max_history_drop=True)
    ```
  </Tab>

  <Tab title="TypeScript">
    ```ts theme={null}
    import { EventBus } from 'abxbus'

    const rejectBus = new EventBus('RejectBus', { max_history_size: 10_000, max_history_drop: false })
    const dropBus = new EventBus('DropBus', { max_history_size: 10_000, max_history_drop: true })
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    use abxbus::event_bus::{EventBus, EventBusOptions};

    let reject_bus = EventBus::new_with_options(Some("RejectBus".to_string()), EventBusOptions {
        max_history_size: Some(10_000),
        max_history_drop: false,
        ..EventBusOptions::default()
    });

    let drop_bus = EventBus::new_with_options(Some("DropBus".to_string()), EventBusOptions {
        max_history_size: Some(10_000),
        max_history_drop: true,
        ..EventBusOptions::default()
    });
    ```
  </Tab>

  <Tab title="Go">
    ```go theme={null}
    maxHistory := 10_000

    // Reject new emits once history reaches N.
    rejectBus := abxbus.NewEventBus("RejectBus", &abxbus.EventBusOptions{
    	MaxHistorySize: &maxHistory,
    	MaxHistoryDrop: false,
    })

    // Never reject on history size; trim oldest history entries instead.
    dropBus := abxbus.NewEventBus("DropBus", &abxbus.EventBusOptions{
    	MaxHistorySize: &maxHistory,
    	MaxHistoryDrop: true,
    })
    ```
  </Tab>
</Tabs>

## 2) If 1,000,000 events complete, how many are kept?

Let `N = max_history_size`.

| Setting                             | Events retained after bus becomes idle    | Notes                                                                     |
| ----------------------------------- | ----------------------------------------- | ------------------------------------------------------------------------- |
| `N = None` / `null`                 | All completed events (so up to 1,000,000) | History is unbounded.                                                     |
| `N > 0`, `max_history_drop = false` | Up to `N`                                 | New emits are rejected once history reaches `N`.                          |
| `N > 0`, `max_history_drop = true`  | Bounded to `N` at steady state            | Oldest history entries are removed first.                                 |
| `N = 0`                             | `0` completed events retained             | Only pending/in-flight visibility is kept; completed entries are dropped. |

Python nuance: in heavy bursts with `max_history_drop=True`, cleanup is amortized, so history can temporarily exceed `N` before converging back to `<= N`.

For the broader retention model, see [Event History Store](../features/event-history-store).

## 3) How RAM usage scales

At a high level, memory grows with:

* pending queue depth,
* retained history size,
* per-event handler/result payload size.

A practical model is:

`RAM ~= O(pending_event_queue) + O(event_history) + O(event_results and payloads)`

### Measured slopes from perf suites

* Python README matrix reports scenario-dependent peak RSS slopes between about `0.025kb/event` and `8.024kb/event`.
* TypeScript README matrix reports scenario/runtime-dependent peak RSS slopes between about `0.1kb/event` and `7.9kb/event`.
* TypeScript README notes those `kb/event` values are measured during active processing with history aggressively bounded (`max_history_size=1` in perf harnesses).

Use those numbers as throughput-era slope indicators, not exact long-term retention multipliers for your payloads.

Operationally:

* bounded history (`N` finite) keeps steady-state memory bounded by queue depth + `N`,
* unbounded history (`N=None/null`) makes retained RAM grow roughly linearly with total completed events.

## 4) Queue vs history lifecycle (exact behavior)

Events do not "move from queue to history." They are added to history at `emit()` time, and can exist in both structures while pending.

### Python timeline (`emit`)

1. Validate pressure policy.
2. Enqueue into `pending_event_queue`.
3. Add same event object to `event_history`.
4. Runloop dequeues event (`queue.get()`), then executes handlers.
5. Event remains in `event_history` as `pending` -> `started` -> `completed` unless trimmed/removed by history policy.

### TypeScript timeline (`emit`)

1. Validate pressure policy.
2. Add event to `event_history`.
3. Apply `trimHistory()`.
4. Push event into `pending_event_queue`.
5. Runloop shifts from queue and executes handlers.
6. Event remains in `event_history` unless trimmed/removed by policy.

### Go timeline (`Emit`)

1. Validate pressure policy against `EventHistory`.
2. Add event to `EventHistory`.
3. Apply history trimming according to `MaxHistorySize` + `MaxHistoryDrop`.
4. Push event into the pending queue.
5. Runloop dequeues the event and executes handlers.
6. Event remains in `EventHistory` unless trimmed/removed by policy.

### Rust timeline (`emit`)

1. Validate pressure policy against the runtime history store.
2. Add event to history and append its id to history order.
3. Apply history trimming according to `max_history_size` + `max_history_drop`.
4. Push event into the pending queue.
5. Runloop dequeues the event and executes handlers.
6. Event remains in history unless trimmed/removed by policy.

So yes:

* an event can be in both `pending_event_queue` and `event_history` at the same time,
* `event_history` can contain pending events (not only started/completed events).

## Observe both structures directly

<Tabs>
  <Tab title="Python">
    ```python theme={null}
    event = bus.emit(MyEvent())
    pending_count = bus.pending_event_queue.qsize() if bus.pending_event_queue else 0
    history_count = len(bus.event_history)
    print('pending_event_queue=', pending_count, 'event_history=', history_count)
    # pending_event_queue= 1 event_history= 1
    ```
  </Tab>

  <Tab title="TypeScript">
    ```ts theme={null}
    const event = bus.emit(MyEvent({}))
    console.log('pending_event_queue=', bus.pending_event_queue.length, 'event_history=', bus.event_history.size)
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    let event = bus.emit(MyEvent {
        ..Default::default()
    });
    let snapshot = bus.to_json_value();
    let pending_count = snapshot["pending_event_queue"].as_array().map(Vec::len).unwrap_or(0);
    let history_count = snapshot["event_history"].as_object().map(|items| items.len()).unwrap_or(0);
    println!("pending_event_queue={pending_count} event_history={history_count}");
    ```
  </Tab>

  <Tab title="Go">
    ```go theme={null}
    event := bus.Emit(abxbus.NewBaseEvent("MyEvent", nil))

    var state struct {
    	PendingEventQueue []string       `json:"pending_event_queue"`
    	EventHistory      map[string]any `json:"event_history"`
    }
    snapshot, err := bus.ToJSON()
    if err != nil {
    	panic(err)
    }
    if err := json.Unmarshal(snapshot, &state); err != nil {
    	panic(err)
    }

    pendingCount := len(state.PendingEventQueue)
    historyCount := len(state.EventHistory)
    fmt.Println("pending_event_queue=", pendingCount, "event_history=", historyCount)
    // pending_event_queue= 1 event_history= 1
    _ = event
    ```
  </Tab>
</Tabs>