RFC: Local event bus

# libp2p local event bus

> Authors: @raulk, @gpestana
> Revision: r0, 2019-05-31

## Context

A libp2p host comprises a number of components. Some are compulsory like the swarm, peerstore, transports, etc. while others are optional like the DHT, autonat, autorelay, etc.

In most cases, events are happening inside those components that other components may be interested in knowing. Examples:

* autonat discovers new self addresses that other elements are interested in (e.g. identify push).
* autonat flips the NAT status of a peer (e.g. private to public).
* identification finishes and we have enumerated the peer’s protocols (identify).
* we receive and respond to a DHT query; we store a new inbound DHT entry.
* component lifecycle events: start and stop.

We are lacking a local event bus where we components can emit events, and observers can subscribe and react to those events. We currently depend on hard timing assumptions to wire things together, resulting in a brittle overall system.

An added corollary is that creating certain dynamic protocols and behaviours is impossible nowadays without forking. Conversely, with a reactive, event-drivel solution, we can enable use cases that extend the base functionality of libp2p to be deployed as an add-on rather than a fork.

## Technical proposal

An asynchronous event bus seems like the right fit for our requirements. The `EventBus` is a top-level object owned by the `Host`:

```
Host -------------------------- (message bus) x
| | |
CompA CompB CompC
```

The proposed interface and object model is:

```go
// extends the existing Host type.
type Host interface {
EventBus() EventBus
}

type SubOpts struct {
// BufferSize is the size of the channel where events are delivered.
BufferSize int

// FailIfNoEmitters fails the registration if, after system initialisation,
// no emitters have been registered for one or more of the requested event types.
FailIfNoEmitters bool
}

type EventBus interface {
// RegisterSubscriber registers a new subscriber for the listed event types.
// If successful, it returns a Subscription. Before starting to consume events,
// the caller must check whether the subscription was initialised successfully by
// receiving once from the InitResult() channel.
RegisterSubscriber(opts SubOpts, evtTypes interface{}...) (Subscription, error)

// RegisterEmitter registers an Emitter for the specified event types.
RegisterEmitter(evtTypes interface{}...) (Emitter, error)
}

type Subscription interface {
io.Closer

InitResult() <-chan error
Events() <-chan interface{}
}

type Emitter interface {
io.Closer

// Emit broadcasts an event to registered subscribers.
// It panics if the emitter has not declared handling for this event type.
Emit(evt interface{})
}
```

Each event is an instance of a typed struct, passed-by-value for safety, with a simple payload containing only primitive data types, pointers to the originator object (e.g. Conn, Stream, etc.), and basic libp2p types like `peer.ID`, `protocol.ID`, etc. We want to avoid shared state and lean towards immutability; each subscriber will receive a copy of the event.

These event payloads will live in go-libp2p-core, namespaced by the owning abstraction. Struct names generally follow the convention: `Evt[Entity (noun)][Event (verb past tense / gerund)]`. The gerund form of the verb (-ing) is allowed to express that a process is in progress.

```
go-libp2p-core
|
|_ events (p)
| |_ network (p)
| | |_ EvtNetworkStarted (s) // component lifecycle event
| | |_ EvtNetworkStopped (s) // component lifecycle event
| | |_ EvtConnEstablishing (s)
| | |_ EvtConnEstablished (s)
| | |_ EvtConnDisconnected (s)
| | |_ EvtStreamOpened (s)
| | |_ EvtStreamClosed (s)
| | |_ ...
| |
| |_ mux
| | |_ EvtStreamProtocolAgreed (s)
| | |_ ...
| |
| |_ routing
| | |_ EvtProviderRecordPublishing (s)
| | |_ EvtProviderRecordPublished (s)
| | |_ EvtProviderRecordQuerying (s)
| | |_ EvtProviderRecordQueried (s)
| | |_ EvtContentQuerying (s)
| | |_ EvtContentQueried (s)
| | |_ EvtContentStoring (s)
| | |_ EvtContentStored (s)
| | |_ EvtIncomingContentQueryReplied (s)
| | |_ EvtIncomingContentStored (s)
| | |_ ...
| |
| |_...
|
|_ ...

legend: (p)ackage, (s)truct.
```

Registration methods take zero-value structs, and use reflection to keep a registry of enlisted emitters & subscribers, and the events they handle.

_NOTE: we considered generic dictionary objects for event payloads along with string-based topics, but quickly discarded it because it's an unsafe and uncertain model._

### Event delivery / concurrency model

We considered several designs for event delivery, with implications for the concurrency model and goroutine count.

1. **callback-based delivery, no channels:** even though it might simplify the solution, this is synchronous in nature, and if a callback blocks, the event loop is doomed. We could execute callbacks in goroutines, probably with a shared worker pool. Unfortunately this breaks the isolation property (e.g. if a subscriber is slow, avoid affecting other subscribers), unless we shard worker pools by subscriber, but this is already unnecessarily complex.
2. **channel-based delivery, with a fixed channel size:** single goroutine to receive events, one dispatch goroutine per subscriber; if the subscriber is slow, only that goroutine blocks, affecting only that subscriber.
3. **channel-based delivery, with a subscriber-defined channel size:** same as (2), but allows the subscriber to define the buffer length based on the amount of work it performs per event.

We propose to adopt (3).

### Ordering of events

We did consider introducing a subscription mode that strongly guarantees the ordering of events, but this is too complex for a first iteration of the event bus, and at this time the benefit is not clear.

### Handling backpressure

This is an open point. What happens if the subscriber becomes slow and the event queues are backlogged? Should we drop events? Doing so may cause inconsistency. We definitely must not slow down the producers, so backpressure must be somehow absorbed.

One option is to allow subscribers to specify the backpressure policy they want via the subscription options. Possibilities are: `DROP`, `KILL`, `NOTIFY`:

* `DROP`: drop events silently -- used when doing so would not cause inconsistency.
* `KILL`: kill the subscription by erroring it immediately.
* `NOTIFY`: do not kill the subscription, but notify a user-provided callback synchronously. The subscriber can use this notification to stop gracefully and restart from a known safe state.

`NOTIFY` should probably be the default.

## Standardisation

Since the local event bus has no implications on the wire, we don't consider it material for imminent standardisation in https://github.com/libp2p/specs.

However, if we deem this abstraction valuable enough to issue a RECOMMENDATION for all libp2p implementations to adopt it, then we can consider speccing it out in a language-agnostic way.