proposal/parachain: Use event driven model for Overseer instead of direct-channel communication

[kanishkatn]

Use event driven model for Overseer instead of direct-channel communication

Related to #3418

TL;DR:

I propose switching from direct channel communication to an EventBus system to simplify communication, enhance system performance, and make it easier to maintain and scale.

This change will allow subsystems to interact without direct dependencies, handle multiple events efficiently, and improve testing and lifecycle management. A demo app illustrating these benefits is available for review.

Summary

This proposal suggests a strategic shift from our current direct channel communication system to an EventBus-driven approach. The goal is to enhance the overall robustness, maintainability, and scalability of the system by adopting an event-driven architecture.

Background

Our existing system relies on direct channel communication between subsystems, which may lead to complexities as we add more subsystems. This model necessitates detailed knowledge of the system's internals, leading to a tightly coupled architecture that is difficult to maintain and extend.

Proposal

I propose we adopt an EventBus system, a proven method that facilitates loose coupling and an event-driven architecture. The EventBus model enables subsystems to publish events or subscribe to them without being aware of other subsystems, streamlining communication and reducing dependencies.

Key Benefits

1. Decoupled Communication

Existing: Subsystems communicate via direct channels which requires both the sender and receiver to be aware of the channel's existence.

// Existing: Subsystem must send messages directly to a specific channel.
subsystemChannel <- msg

Event Bus: Subsystems publish and subscribe to events, allowing communication without knowing about each other's channels.

// Event Bus: Subsystem publishes an event, decoupling from direct channel references.
t.eventBus.Publish("event:subsystemStarted", eventData)

2. Scalability

Existing: Adding new subsystems can require adding new message cases to the overseer's select statement, potentially leading to a complex and unmanageable switch-case structure.

Event Bus: New subsystems can be added with minimal changes to the overseer, as they simply subscribe to events they're interested in.

// Event Bus: New subsystems can be easily integrated by subscribing to topics.
eventBus.Subscribe("topic:newSubsystem", newSubsystem.EventHandler)

3. Improved Testability

Existing: Testing requires careful construction and management of channels, and synchronizing the sends and receives within the test cases.

// Existing: Direct interaction with channels can make isolated testing complex.
require.NoError(t, sendTestMessage(overseerChannel, testMsg))

Event Bus: The event bus interface can be mocked or replaced with a testing implementation, allowing for isolated and controlled tests.

// Event Bus: Testing with a mock event bus can isolate tests from real event bus behavior.
mockEventBus.Publish("test:subsystemEvent", mockEvent)

4. Asynchronous and Concurrent Processing

Existing: The overseer processes messages synchronously, limiting concurrency and potentially causing bottlenecks. It could lead to a need of a priority channel leading to further complexity.

Event Bus: The event bus can handle messages asynchronously, allowing multiple messages to be processed in parallel.

// Event Bus: Enable concurrent message processing without changing the overseer's core logic.
t.eventBus.SubscribeAsync("topic:handleConcurrently", messageHandler, false)

5. Simplified Lifecycle Management

Existing: The overseer is responsible for managing the start and stop signals for each subsystem, which can complicate its logic.

// Existing: The overseer sends stop signals directly to subsystems.
for _, sub := range o.subsystems {
    close(sub)
}

Event Bus: Lifecycle events can be published on the event bus, simplifying the overseer's role in starting and stopping subsystems.

// Event Bus: Lifecycle management is handled through event publication.
t.eventBus.Publish("lifecycle:subsystemStop", StopSignal)

In summary, the event bus approach abstracts the communication layer between subsystems, promoting loose coupling and enhancing the system's ability to grow and adapt without significant refactoring. Meanwhile, the existing direct channel communication requires more intimate knowledge of the system's internals, which can become cumbersome as the system scales.

Comparison of Use Cases: Direct Channel Communication vs. EventBus

Sending Messages Between Subsystems

Direct Channel Communication:
Currently, when one subsystem wants to send a message to another, it has to go through the overseer, which acts as a middleman. This process can be cumbersome and slow.

// Existing: Subsystem A sends to Overseer, which then sends to Subsystem B.
subsystemAChannel <- messageForOverseer // A sends to Overseer
msg := <-overseerChannel // Overseer receives
subsystemBChannel <- msg // Overseer sends to B

EventBus System:
With EventBus, subsystems can publish messages to a topic that other subsystems are subscribed to, removing the need for an intermediary and streamlining the process.

// Event Bus: Subsystem A publishes directly to a topic that B is subscribed to.
eventBus.Publish("topicForB", messageFromA) // A publishes
// Subsystem B has a handler for "topicForB" and processes the message.

Sending Messages from Overseer to Subsystems

Direct Channel Communication:
The overseer directly sends messages to each subsystem through their respective channels. This requires the overseer to manage multiple channels and can complicate message broadcasting.

// Existing: Overseer sends messages to each subsystem channel.
subsystemAChannel <- messageForA
subsystemBChannel <- messageForB
// ...and so on for each subsystem.

EventBus System:
The overseer publishes a message to a specific topic, and all subscribed subsystems receive it. This allows for a one-to-many communication pattern that's easier to manage.

// Event Bus: Overseer publishes to a topic, and all interested subsystems receive it.
eventBus.Publish("overseerUpdate", updateMessage)

Broadcasting Global Updates

Direct Channel Communication:
Broadcasting a global update requires the overseer to send the message through each subsystem's channel, which is inefficient and error-prone.

// Existing: Overseer must send the update individually.
for _, channel := range subsystemChannels {
    channel <- globalUpdateMessage
}

EventBus System:
A global update can be published once to a general topic, and all subsystems interested in global updates can react accordingly.

// Event Bus: One publish reaches all subscribers.
eventBus.Publish("globalUpdate", globalUpdateMessage)

Error Handling

Direct Channel Communication:
Error messages need to be passed back to the overseer through specific channels, which may lead to missed or unhandled errors if not correctly managed.

// Existing: Error handling requires careful channel management.
subsystemChannel <- errorOccurred
// Overseer needs to check each channel for errors.

EventBus System:
Subsystem can publish errors to a common 'error' topic, allowing for centralized error handling without channel clutter.

// Event Bus: Centralized error handling through a common topic.
eventBus.Publish("error", errorOccurred)

By implementing an EventBus, we not only improve the communication flow within our system but also enhance our ability to manage global updates and errors efficiently. This transition facilitates a more responsive, reliable, and maintainable architecture.

Demonstration and References

To better illustrate the benefits and functionality of the EventBus system, a demo application has been developed. This application serves as a practical example of the EventBus in action, showcasing the ease of adding wrappers, new subsystems and the streamlined message handling process:

• Overseer Demo Application

[P1sar]

This approach looks interesting to me. Currently my only concern is that for this EventBus there is no kinda maintained library, just copied part of code, so we a) It does not have big history of usage by different clients b) we would have maintain it by ourself

[kanishkatn]

The library in the demo app is just a potential library that we could use. Happy to share more about it if we want to consider that!

[kishansagathiya]

Messages sent from one subsystem to other are also often request.
Say, I send Validate this candidate to backing subsystem and backing subsystem sends me a Valid/Invalid response, because we are both subscribed to the respective topics. How do I figure out which response belongs to which request?

Say I add some id to link req and res. I need additional logic to store and keep all req and res, because I might be getting responses in random order (because the other subsytems could finish different responses at different times)

Currently, in overseer, I can add a res channel in the req and get response in it, which is very convenient.

[kanishkatn]

Maybe I didn't understand the usecase correctly. Could you explain it in some other way if possible?

[kishansagathiya]

// subsystem A send a request to subsystem B
var res chan any
subSystemA.Send(ValidateCandidate{resChan: res})

// Along with ValidateCandidate message with we are sending a `res` channel. response will be sent on this res channel. message that we `ValidateCandidate` sent is a request. We expect a response from subsystem B, which could be either yes, this candidate is valid, or no this candidate is invalid.

// things get blocked until I get a response

response := <- res
// Whatever I need to do with the response

This is what we are able to do in overseer right now.

In the proposed solution, subsystem A could be getting multiple messages of the type of responses. How do I know which response belongs to which request?

In your proposal (or event bus), how easy is it to do the behaviour that I have shown in the code.

[kishansagathiya]

If it is still unclear, try asking for specific parts of the description that is unclear?

[kanishkatn]

Thank you for sharing. It could be replaced by this code!

res, ok := subsystemA.subsystemLibrary.SubsystemBServices().ValidateCandidate(candidate)

Reference:

https://github.com/kanishkatn/overseer#subsystem-library-subsystemlibrary
https://github.com/kanishkatn/overseer/blob/main/subsystem_library.go
https://github.com/kanishkatn/overseer/blob/main/subsystem_library_test.go#L42

[kishansagathiya]

Ok. That was helpful.