Redesign Node & spawn_primary

[akichidis]

Currently when we want to spawn a new primary node we are using the spawn_primary method which is a static function that receives a bunch of inputs and returns a vector of JoinHandles of the spawned tasks. This method is suffering from a few issues

1. The input is becoming quite convoluted as more and more inputs are passed in
2. When we need to communicate information out of the Primary node, a channel needs to passed in (ex see tx_confirmation) which is used in a top to bottom approach penetrating several components until is finally used
3. Our nodes can be used with internal or external consensus. Some inputs apply only when one mode is being used (ex the tx_confirmation, execution_state) which creates unnecessary setup in the opposite case and makes things unclear on what's needed or not

Because of the above its becoming quite hard to extend and manage the primary node especially in case (2). For example, if I would like to receive the last_committed_round outside of the primary node, the way to do it , it would be to create a new channel and pass it as part of the spawn_primary continue growing the method inputs and add additional complexity by again passing down the channel until it is being used by the necessary component (ex in this case it would be the state_handler).

Proposal

To tackle the above we could probably refactor the Node and the spawn_primary method. Ideally we would like to return back something like a handle (ex like we do in Dag or in MockBlockSynchronizer). Then we can use this handle to do several stuff like:

1. subscribe (or listener) to receive supported outputs (ex committed certificates, latest committed round etc)
2. manage the node , like it shutdown
3. Query the primary node directly via the handle

We might also need to split completely the method of spawning a Primary & Worker nodes from the Node struct and have this being spawned via different ones as common functionality/concerns will not apply here.

For example on high level to get an idea this is what I am looking at if we follow an approach that still uses channels, but in a more flexible structure:

// create the channel to receive the sequenced certificates
let (tx_sequeneced_certificates, rx_sequeneced_certificates) = channel(100);

let primary = Node::spawn_primary(...inputs...);

// Subscribe to receive events for sequenced certificates to the provided channel.
// The subscribe method can receive multiple channels to listen on different events.

let id = primary.subscribe(Some(tx_sequeneced_certificates), None, None ....other optional channels).await;


// Unsubscribe if don't need to hear any more

primary.unsubscribe(id).await;


// Shutdown the node - do any background work via 
// channels etc to shutdown the node
primary.shutdown().await;

it has to be noted that the provided channels will not be passed down to the individual components, but they'll rather be stored in the Node handle. The primary will use its own channels to communicate the required info back to the Node handle.

An alternative approach like the more Java-ish listener approach (basically that's the Observer pattern we talked with @huitseeker

• or at least how I interpret it for this case) could be to not use channels but rather an object it self:

/// Implement a trait to listen on events
trait PrimarySubscriber<PublicKey> {
    
    // called when a certificate is sequenced
    fn on_sequenced(certificate: Certificate<PublicKey>);

    // called when last committed round is updated
    fn on_last_committed_round(round: Round);
}

// Assume we create a few subscribers
let sub_1 = PrimaryNodeSubscriber::new();
let sub_2 = PrimaryNodeSubscriber::new();

let primary = Node::spawn_primary(...inputs...);

primary.subscribe(sub_1);
primary.subscribe(sub_2);

// Now whenever something happens (ex a new certificate is sequenced) the subscribers are notified
// by calling their methods

Of course the challenge of the above is to make the subscribers shareable and probably even support some internal mutation, but that's a separate thing to consider.

[gdanezis]

The ideal primitive to ensure that the info channel does not block if no one is listening to ex-filtrated information might be: https://docs.rs/tokio/latest/tokio/sync/broadcast/index.html

primary.unsubscribe(id).await;

Maybe unsubscribe on drop?