The API of Leader Based Journal

[w41ter]

This is part of thread #251, and focus on the design of API of leader based Journal.

In #229 , @huachaohuang comment:

We can also further divide the interfaces into two levels. The first level provides some basic interfaces. The second level extends the basic interfaces with leader-based features.

I have try to prototype the interfaces of leader based journal with the implementation of Kernel::apply_update (you can found it in here):

/// An interface to manipulate a journal.
#[async_trait]
pub trait StreamManipulator {
    type StreamLister: BatchResultStream<Elem = String, Error = Error>;

    /// Lists streams.
    async fn list_streams(&self) -> Result<Self::StreamLister>;

    /// Creates a stream.
    ///
    /// # Errors
    ///
    /// Returns `Error::AlreadyExists` if the stream already exists.
    async fn create_stream(&self, name: &str) -> Result<()>;

    /// Deletes a stream.
    ///
    /// Using a deleted stream is an undefined behavior.
    ///
    /// # Errors
    ///
    /// Returns `Error::NotFound` if the stream doesn't exist.
    async fn delete_stream(&self, name: &str) -> Result<()>;
}

#[async_trait]
pub trait Journal: Send + Sync + 'static {
    type StreamReader: StreamReader;
    type StreamWriter: StreamWriter;

    /// Returns a stream reader.
    async fn new_stream_reader(&self, name: &str) -> Result<Self::StreamReader>;

    /// Returns a stream writer.
    async fn new_stream_writer(&self, name: &str) -> Result<Self::StreamWriter>;
}

pub enum RoleState<W>
where
    W: StreamWriter,
{
    Leader(W),
    Follower,
}

#[async_trait]
pub trait LeaderBasedJournal: StreamManipulator + Sync + Send + 'static {
    type StreamWriter: StreamWriter + Send;
    type StreamReader: StreamReader;

    type Role;
    type Peer;
    type StateStream: Stream<Item = RoleState<Self::StreamWriter>> + Unpin + Send;

    fn role(&self, name: &str) -> Self::Role;

    fn leader(&self, name: &str) -> Option<Self::Peer>;

    async fn transfer(&mut self, name: &str, peer: Self::Peer) -> Result<()>;

    async fn observe_state(&self, name: &str) -> Self::StateStream;

    /// Returns a stream reader.
    async fn new_stream_reader(&self, name: &str) -> Result<Self::StreamReader>;
}

I create a new trait named LeaderBasedJournal and add some methods to it:

• role: returns the role of current journal
• leader: returns the leader peer of current journal's replication group
• transfer: tranfer leadership to the specified peer
• observe_state: observe the changes of role

All these methods need to specify stream name. Once this peer become the leader of a stream, it will open a StreamWriter for appending.

So far, this design is still a draft, and everyone is welcome to discuss it.

[huachaohuang]

The design looks a bit complicated to me. Can you describe the expected way to use a LeaderBasedJournal? For example, if I want to use it as a WAL, what control flow do I need to use these APIs properly?

[w41ter]

Here is an exmaple about how to use these API properly: https://gist.github.com/w41ter-l/e7091bb2173700285c52ddbe63630451.

[huachaohuang]

Thanks. Two questions so far:

• Why not let the leader create a stream writer when it becomes a leader instead of returning a stream writer in RoleState? In this case, you can keep the basic Journal APIs consistent and let LeaderBasedJournal extend the basic Journal.
• What's the semantic of writing to a stream writer when the writer loses its leadership?

[huachaohuang]

BTW, when an instance becomes a leader, it may need to replay some un-synchronized events before serving.

[w41ter]

Yes, but the replaying progress should be hide in Journal implementation.

[w41ter]

Two questions so far

I'll write about this in more detail later.

[huachaohuang]

Is it a good idea to model the API with leader leases? The spanner paper provided some details in "Appendix A".

[w41ter]

I'll study it first and come back to fill in the details later.

[w41ter]

This thread is mainly used to answer the problems that API design of Leader based Journal has faced and the solutions given.

The current trait definations of Journal and related traits:

#[async_trait]
pub trait StreamReader {
    /// Seeks to the given sequence.
    async fn seek(&mut self, sequence: Sequence) -> Result<()>;

    /// Returns the next event.
    async fn next(&mut self) -> Result<Option<Vec<u8>>>;
}

#[async_trait]
pub trait StreamWriter {
    /// Appends an event.
    async fn append(&mut self, event: Vec<u8>) -> Result<Sequence>;

    /// Truncates events up to a sequence (exclusive).
    async fn truncate(&mut self, sequence: Sequence) -> Result<()>;
}

#[async_trait]
pub trait Journal: Send + Sync + 'static {
    type StreamLister: BatchResultStream<Elem = String, Error = Error>;
    type StreamReader: StreamReader;
    type StreamWriter: StreamWriter;

    /// Lists streams.
    async fn list_streams(&self) -> Result<Self::StreamLister>;

    /// Creates a stream.
    ///
    /// # Errors
    ///
    /// Returns `Error::AlreadyExists` if the stream already exists.
    async fn create_stream(&self, name: &str) -> Result<()>;

    /// Deletes a stream.
    ///
    /// Using a deleted stream is an undefined behavior.
    ///
    /// # Errors
    ///
    /// Returns `Error::NotFound` if the stream doesn't exist.
    async fn delete_stream(&self, name: &str) -> Result<()>;

    /// Returns a stream reader.
    async fn new_stream_reader(&self, name: &str) -> Result<Self::StreamReader>;

    /// Returns a stream writer.
    async fn new_stream_writer(&self, name: &str) -> Result<Self::StreamWriter>;
}

When the Journal interfaces is used to adapt Leader Based WAL, the below problems will encountered:

1. Semantic
   1. Does the StreamWriter returned by Journal::new_stream_writer support concurrent calls?
   2. If StreamWriter doesn't support concurrent call, and the leader might call Journal::new_stream_writer multiple times, how to deal with this behaviour?
   3. What happen if a follower calls Journal::new_stream_writer?
   4. How to deal with that leader becomes follower before calling Journal::new_stream_wrier?
2. Role, State transform
   1. How do journal users monitor role transform?
   2. How do journal users get the internal state, such as current role, leader address, replication group, sync point, epoch or term, etc?
   3. What happen if the leader becomes to follower but some requests have submitted to StreamWriter?
   4. What to do if a leader has transformed to follower but StreamWriter::append is called?
3. Consistency
   1. What level of consistency does the StreamReader need to support (eg, a Follower read events from StreamReader)? And how to reflect it on API?
   2. StreamReader lacks a mechanism to wait for new writes

In addition, another important question is whether it should keep consistent with the existing Journal design? If so, how should it be designed?

The API design proposal I haved post in early thread focus on problems 1 and 2. It assumes that StreamWriter doesn't support concurrent calls and avoid problems 1.2-1.4 by removing new_stream_writer from LeaderBasedJournal trait. A series of methods add to LeaderBasedJournal to solve problems 2.1-2.2, including the alternative method of new_stream_writer. The only way to obtain a StreamWriter is calling new method LeaderBasedJournal::observe_state and waiting it becomes leader. Once a leader becomes to follower, the opened StreamWriter will returns a error like JournalError::Sealed for problems 2.3-2.4.

cc @huachaohuang

[huachaohuang]

Thanks for your explanation. I see there are a lot of things to be considered here. I will take some time to think about them.

In addition, another important question is whether it should keep consistent with the existing Journal design?

You can bring a brand new design if it is more reasonable than the existing one.

[tisonkun]

@w41ter-l what about the LeaderBasedJournal be a struct implement Journal trait and take another Journal impl as the inner?

In this way, the LeaderBasedJournal serves as a decorator dealing with replication stuff while all real data r/w ops delegated. Also, it keep the Journal trait clean and doesn't bothered by consensus things.

[w41ter]

I found that if I change the semantics of Journal::new_stream_writer and StreamWriter, the defination of LeaderBasedJournal and Journal can keep consistency.

Before start discussion, I want change the word LeaderBasedJournal to SingleWriteJournal to make it intutively.

These considrations are:

1. Journal::new_stream_writer could be called multiple times and all of the returned StreamWriter is valied
2. For SingleWriteJournal, the returned StreamWriter will return a error if the stream isn't leader.

After that, the API changed to:

pub enum RoleState {
    Leader,
    Follower,
}

pub trait SingleWriteJournal : Journal {
    type Role;
    type Peer;
    type StateStream: Stream<Item = RoleState>;

    fn state(&self, name: &str) -> (Self::Role, Option<Self::Peer>);

    async fn transfer(&mut self, name: &str, peer: Self::Peer) -> Result<()>;

    async fn observe_state(&self, name: &str) -> Self::StateStream;
}

An Engine could open both StreamWriter and StreamReader once it starts servering, then call observe_state to observe the role change. Once it becomes leader, it accepts requests and persist updates to journal by invoking StreamWriter::append.

[w41ter]

@tisonkun I think this defination could keep the Journal trait clean, and doesn't bothered by consensus things.