Propose stream-based Replication of Log Entries

[drmingdrmer]

Stream-based Replication of Log Entries

Rationale

The existing request-response style API offers limited parallelism and pipelining capabilities and does not allow for dynamic adjustment of request sizes.

By introducing a Stream as an argument to RaftNetwork::append_entries(), we can empower the method to replicate log entries in a more flexible and potentially optimized way.

Accordingly, the storage layer should also expose a stream-based API. We propose adding RaftLogReader::try_get_log_entries() -> impl Stream<Item=Entry>, which will enable the streaming of log entries as needed.

Proposed new API:

trait RaftLogReader<C: RaftTypeConfig> {
    async fn get_log_entries(&mut self, since: u64)
               -> impl Stream<Item=C::Entry>;
}

trait RaftNetwork<C: RaftTypeConfig> {
    async fn entries(&mut self, 
                   prev: Option<LogId>, 
                   entries: impl Stream<Item=C::Entry>) 
              -> Result<
                       impl Stream<Item=Result<Response, RPCError>, 
                       RPCError<_,_,RaftError<>>
                    >;`
}

RaftNetwork::entries() now takes a stream of log entries for transmission and returns a stream of replication responses. Should a replication attempt result in an Err, such as the absence of prev on the remote peer or the detection of a higher term vote, the caller should immediately discard the stream.

TODO:

• move trait method: from RaftLogStorage to RaftLogReader::read_vote() #1077

[github-actions]

👋 Thanks for opening this issue!

Get help or engage by:

• /help : to print help messages.
• /assignme : to assign this issue to you.

[schreter]

Nice!