This is an implementation of the Raft protocol made from scratch.
This branch tries to follow the implementation approach explained by Martin Kleppmann in its distributed systems lectures and the original paper.
This is a second attempt of implementation. The first on the main branch, used from the beginning a flat hierarchy of classes to distinguish the different roles a Node can have in the Raft state machine. However, when try the implementation had too much boilerplate code due to the hierarch.
This implementation will be more "spaghetti" as there will be conditionals to check each role. Although from the very beginning it is clear it is not the ideal solution, it will be easier to contemplate the algorithm as a whole without abstractions and with explicit role separation (several conditionals in a place instead of responsibilities in different classes and files). Also, it will give an easy understanding of what is shared between all the roles.
After a better completion of the algorithm a refactor to a more organised sophisticated approach can be done.
The ergonomics of the previous architecture to simulate time and the network are useful and will be adapted to the new approach. Everything new will be inside the statemachine package until the refactor.
Nodes start as a Follower.When the system is just initialised, a Follower may be promoted to Candidate in one of the two following scenarios:
When a Follower have not yet received any messages and the ElectionTimeOut is reached. Or when the period that takes two different messages from the Leader to arrive at the Follower is bigger than the HeartBeatTimeOut. This is a sign that indicates the Leader might have crashed.
Each Follower is instantiated with a random ElectionTimeOut to avoid two nodes starting an election at the same time.
After a promotion from Follower to Candidate, the Candidate must start an election. Another scenario where the Candidate must start an election is when after having started a previous election, it is not promoted to Leader before the ElectionTimeOut has been reached.To start an election, the Candidate increments its term, votes to itself, and request votes to all its peers.
The Candidate will keep the "timestamp" of when the term starts . That way if we subtract from the current time and the result is bigger or equals to the ElectionTimeOut, it timed out and a new election is needed.
Left for later to focus on happy path scenarios and as it can be simulated removing the Node from the network. When receiving a Request for Vote (and probably any other message), if the Term in the message is higher than the Term in the receiving Node, it must update its term to the new value. This signals that successful elections have happened and the receiving Node is outdated. This also means the Node have to be demoted to Follower and have the vote decision cleared. Note that even in this case a Node is still going to take decision on how to vote. A Node votes for a Candidate if three conditions are met.1 - The Candidate Term (that comes in the Vote Request) is the same as its own Term. If they are not, it means that the one of the Nodes have participated in other elections.
2 - The Node have yet not voted for another Candidate. Remember that when a new Election happens a new Candidate will send a Request For Vote with a higher term and the Node will have its state adjusted to reflect the new election (see the description of the beginning of this session). If a Node has voted for itself before on the same Election, it can repeat that vote.
3 - The Log of the candidate is more up to date than the current Node. That is verified checking if the Candidate Log Term is higher than the Last Log Term of the current node. Or if the Logs are on the same term, then the Candidate's log must be bigger than the current Node's log.
If one of the previous conditions are not met then the Node should respond with a negative vote.
When a Vote is received, if the Term of the vote is higher than the Term in the receiving Node, the current Node is outdated. The Node's Term has to be updated as the Term from the Vote, the Node must be Demoted to Follower, its Vote Record (votedFor) deleted, as it voted for itself previously. Finally, the election timeout must be canceled. For a Node to accept a Vote it needs to be a Candidate, the Term for the Vote be the same as its Term and the Vote to have been granted as positive (agrees = true).If all the conditions are met, then the Vote should be stored in Votes Received. Note that Votes Received is a set, so this operation is idempotent.
And if the candidate has the quorum (more than 50% of the participants), the Node can be Promoted to Leader.
During the Promotion step Kleppmann is using a flag (currentLeader) to sign that. This implementation accounts for that in different ways, checking the role or even other mechanisms when refactoring for a interfaces hierarch.
The Election timeout has to be canceled and for each of the other Nodes the log must be replicated.
The Log Replication will be fully implemented in the next steps. At this point, the Leader will only have the state that controls what to send to each of the other Nodes initialized.The Number of Logs Records that the Leader already sent to the other Nodes must be initialized as the Log Length, this assumes that the other Nodes have already received all entries the Leader has. If that is not true, the Log Replication process will adjust the information: (sentLength[follower] = log.length)
The Number of Log Records that the other Node has already acknowledged. It should assume nothing was acknowledged yet and as the Leader receives the acknowledgements, this number will increase: (ackedLength[followe] = 0)
When receiving a Command from a Client, if the Node is not a Leader, it should redirect the receiving messages to the Leader via FIFO (First In, First Out).I'm not sure if the Candidate also should do the same. I will assume it will.
When the Leader receives the Client Command, does not matter if via another Node or straight from the Client, it must append the Command (Message) to its Log, the Log must contain the Current Term of the Leader. The Command coming from the Clients should not be aware of the Leader Term, thus a new type might be needed.
TODO: When adding the new type, it would still be considered a Message, to be handled sequentially. A type outside the Message interface would have to be handled before or after the other messages arrivals, because it is not part of the same collection.
The Leader must set the ackLength of itself to the size of its Log: ackLength[log.size], this signals that the Node (itself) has acked the Log until that point. Then the Leader must replicate its Log to all its Peers.
TODO: When a non Leader receives a Client Command, it must forward to the Leader. The implementation still do not have a mechanism that changes the state of the Followers to know the last elected Leader.
To prevent a new Election to start when the Leader is still alive, the Leader sends Heartbeats. The Heartbeat timeout must be calculated to be of a shorter time than the Election Timeout added of the time the Heartbeat takes to arrive at the other Nodes.The Heartbeats implementation is the same as the Log Replication and do not have its own Message type as in the previous implementation.
The Leader replicates the Log Entries it believes it have not yet sent to the Followers. This is done via the 'prefixLen', that is the index of the last Entry sent to a particular Follower: sentLength[ follower ].It 'slices' its log from that the prefixLen until the end of the Log. It sends to the Follower the missing Entries and with that the Term of the last not sent Entry. This is similar to the Last Term of the Log, but considering only the sent messages to the particular Follower. If there are no Entries, the Term is 0.
The Node receiving the Log replication must do the usual Term check: if the Term in the Message is higher than its current Term it should update its Term, reset the Vote record and cancel the Election timer.If the Term in the message is the SAME as the Node Term, then the Node must set the Current Leader record to the Message Sender and Demote itself to Follower. It seems quite common that a Node will receive an AppendEntries (Log Replication) message with the same Term as its own, and it seems odd that the Node should perform the two previously mentioned steps at each Message. The explanation is that this might happen when the Node is a Candidate and receives a message from the Leader in the same Election Term. However, Klepmann does not add a contidional based on the Node Role. This implementation will perform the Demotion and Current Leader record set, independently of the Node Role.
In order to accept the Log Entries the receiving Node must verify if its Log is in accordance with the sent Entries. This is similar to the checks performed in the Log during the Leader election but slightly different.On the Leader Election we split the Log in a Prefix, that is what the Leader believe it has already sent to the Follower. For that, the receiving Node Log must be at least as long as the prefix sent in the Message, meaning the receiving Node must have a Log Size bigger or equal to the prefixLen sent in the Message.
If it is not, it means there is a gap where the receiving Node does not have the Entries the Leader belive it has.
The second requirement is that the prefixLen is 0, meaning there was nothing on the Leader to be sent. Or the Last Log Entry Term of the prefix is the same as the prefixTerm (prefixTerm is the corresponding Log Entry Term in the Leader). Raft has an invariant where if there are two Nodes in the system and if the Log Entries at the same Index have the same Term, then the protocol guarantees that the Log up to that index is identical.
To efficiently check if two Nodes have identical Logs up to a certain index, it is just matter of checking if the Term of the Entries of the same index are the same. That is what the previous conditions are guaranteeing.
The ACK sent in the Response, in Kleppmann's implementation, is calculated in the Follower receiving the Log Entries. However, the Leader knows the prefixLen and the suffix length, so a positive or negative Response would be sufficient and the Leader could calculate that value on its side. Klepmman's implementation will be followed here as the consequences of that change are unknown at the moment. The first check to be done when Appending Entries to the Followers Log is to check if the Follower already have some of the Entries sent in the Message. The check is done verifying if the size of the Log is greater than the prefixLen and also if the Message contains Entries to be appended (the Number of Entries to append is greater than 0).If the Follower has some of the Entries in the Message then the Log should be truncated, removing the Entries that are out of sync and keeping the synched ones.
To remove the out of sync entries we need to find the index of Last comparable Log Entry between the Follower and the Leader. For that we compare which is smaller, the Log in the Follower or the prefixLen + the suffixLen. The last expression is a reflection of the part of the Log in the Leader that it believe the Follower must have. Then we subtract one from this index.
If the Terms on the two corresponding Entries are different in the Follower and the Leader on the calculated index (followerLog[index].ter != suffix[index - prefixLen]) it means the Logs are not in sync and the then the Log must be truncated until prefixLen - 1, this position is where all the new entries should be appended.
With the out of sync entries removed, it is time to append the new Entries. For that the prefixLen and the suffix Len must be greater than the Follower's Log size. Could be that the entries to be appended are already contained in the Followers Log, and they should not be appended again. To accomplish that the first suffix log index to be appended should be the Followers Log size - prefixLen, until the end of the suffix (suffix - 1). To visualise this case, see images/07_appending_the_new_entries.jpg. The appended new entries are not yet commited in the Follower. The entries to be commited is a decision only the Leader can take. And it signaled to the Followers through the 'commitLength' contained in the AppendEntries Message, if leader commitLength is greater than the commitLength in the Follower, then the Entries in the Follower Log, from the index pointed by commitLength until leader commitLength - 1 should be delivered to the application and the commitLength in the Follower should be updated to the leader commitLength value. When receiving a Log Acknowledgment the Leader must perform the usual Term checks. If the Term from the Message is greater than its Term, then it should be demoted to Follower and set its Current term as the Message Term. Kleppmann's implementation also resets the Vote Record and cancel the Election timeout.For the Term check, Kleppmann does not check the Role, this implies that the Leader can send an AppendEntries and before the Acknowledgement arrives, it can have changed to another Role. In this case, it still has to update its Term but discard the Acknowledgement.
In the case where the Leader has the same Term as the Message, it has to check if the Acknowledgement was successful and that the Number of Entries acknowledged (ack) is at least the same as the number of entries previously acknowledged by the Follower, then the Leader can record the Followers acknowledgement setting the sentLength and ackLength to the ack number from the message and Commit its Log Entries.
The reason for the Ack check, is that if a message arrives with an Ack smaller than the previously acknowledged by the Follower, it means the message is out of order (outdated, delayed) and other ack messages already acked the Log.
If the Acknowledgement was not successful, could be there was a GAP in the Logs, for example, the Leader assumed the Follower has entries that it does not have (Follower lost part of the Logs and does not have logs synched with the entire prefixLen set on the Leader).
For that we need to check if the Leader had previously attempted to send entries to the Follower (sentLength[follower] > 0) and try to replicate the log again, without sending the last Log Entry (sentLength[follower] = sentLength[follower] - 1). On the replication log this means shrinking the prefix by one and sending one more entry in the suffix.
Note that this might trigger many messages between the Leader and the Follower until the logs gets synchronised. In terms of networking, one network call will trigger another (nested) call, that can trigger another call and so on, until the log gets synched.
The ackedLength is where the Leader records the number of Log Entries the Followers have confirmed they have received. If the 'commitLength' in the Leader is smaller than its Log size, the Log entry in the 'commitLength' index can be applied to the application and commited (commitLength + 1) if the Log has been acked by the quorum of the cluster, and the process repeated while 'commitLogLength' < log.size or until an entry without quorum is found. As there is no point in continuing check as subsequent entries will also not have been acked by the Followers and can't be commited.