Backpressure

• 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

​

Reject vs drop behavior

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)

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 })

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()
});

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,
})

​

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

​

3) How RAM usage scales

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

​

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

• 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)

​

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.

• 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

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

const event = bus.emit(MyEvent({}))
console.log('pending_event_queue=', bus.pending_event_queue.length, 'event_history=', bus.event_history.size)

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}");

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