libraft (empty) → 0.1.0.0
raw patch · 31 files changed
+4817/−0 lines, 31 filesdep +QuickCheckdep +attoparsecdep +basesetup-changed
Dependencies added: QuickCheck, attoparsec, base, bytestring, cereal, concurrency, containers, dejafu, directory, exceptions, haskeline, hunit-dejafu, libraft, mtl, network, network-simple, parsec, protolude, random, repline, stm, tasty, tasty-dejafu, tasty-discover, tasty-expected-failure, tasty-hunit, tasty-quickcheck, text, time, transformers, word8
Files
- ChangeLog.md +3/−0
- LICENSE +30/−0
- README.md +464/−0
- Setup.hs +2/−0
- app/Main.hs +344/−0
- libraft.cabal +157/−0
- src/Control/Concurrent/STM/Timer.hs +75/−0
- src/Examples/Raft/FileStore.hs +101/−0
- src/Examples/Raft/Socket/Client.hs +79/−0
- src/Examples/Raft/Socket/Common.hs +30/−0
- src/Examples/Raft/Socket/Node.hs +162/−0
- src/Raft.hs +511/−0
- src/Raft/Action.hs +28/−0
- src/Raft/Candidate.hs +167/−0
- src/Raft/Client.hs +72/−0
- src/Raft/Config.hs +18/−0
- src/Raft/Event.hs +35/−0
- src/Raft/Follower.hs +161/−0
- src/Raft/Handle.hs +168/−0
- src/Raft/Leader.hs +193/−0
- src/Raft/Log.hs +128/−0
- src/Raft/Logging.hs +137/−0
- src/Raft/Monad.hs +199/−0
- src/Raft/NodeState.hs +218/−0
- src/Raft/Persistent.hs +40/−0
- src/Raft/RPC.hs +131/−0
- src/Raft/Types.hs +69/−0
- test/TestDejaFu.hs +429/−0
- test/TestDriver.hs +1/−0
- test/TestRaft.hs +573/−0
- test/TestUtils.hs +92/−0
+ ChangeLog.md view
@@ -0,0 +1,3 @@+# Changelog for raft++## Unreleased changes
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Adjoint Inc. (c) 2018++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Author name here nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,464 @@+<p align="center">+ <a href="http://www.adjoint.io"><img src="https://www.adjoint.io/assets/img/adjoint-logo@2x.png" width="250"/></a>+</p>++[](https://circleci.com/gh/adjoint-io/raft)++# Raft++Adjoint's implementation of the Raft consensus algorithm. See [original+paper](https://ramcloud.stanford.edu/wiki/download/attachments/11370504/raft.pdf)+for further details about the protocol.++# Overview++Raft proposes a strong single-leader approach to consensus. It simplifies+operations, as there are no conflicts, while being more efficient than other+leader-less approaches due to the high throughput achievable by the leader. In+this leader-driven consensus algorithm, clients must contact the leader directly+in order to communicate with the system. The system needs to have an+elected leader in order to be available.++In addition to a pure core event loop, this library uses the systematic+concurrency testing library+[dejafu](https://hackage.haskell.org/package/dejafu-1.11.0.3) to test+certain properties about streams of events throughout the system. Random thread+interleavings are generated in a raft network and realistic event+streams are delivered to each node's event queue. We test for the absence of+deadlocks and exceptions, along with checking that the convergent state of the+system matches the expected results. These concurrency tests can be found+[here](https://github.com/adjoint-io/raft/blob/master/test/TestDejaFu.hs).++## Ensuring valid transitions between node states++Each server in Raft can only be in one of these three states:++- Leader: Active node that handles all client interactions and send+ AppendEntries RPCs to all other nodes.+- Candidate: Active node that attempts to become a leader.+- Follower: Passive node that just responds to RPCs.++Temporal (e.g. ElectionTimeout) and spatial (e.g. AppendEntries or RequestVote)+events cause nodes to transition from one state to another.++```+ [0] [1] [2]+------------> Follower --------------> Candidate --------------> Leader+ ^ ^ | |+ | | [3] | |+ | |_______________________| |+ | |+ | [4] |+ |___________________________________________________|++- [0] Starts up | Recovers+- [1] Times out | Starts election+- [2] Receives votes from majority of servers and becomes leader+- [3] Discovers leader of new term | Discovers candidate with a higher term+- [4] Discovers server with higher term+```++All nodes in the Raft protocol begin in the follower state. A follower will stay+a follower unless it fails to hear from a leader or a candidate requesting a+vote within its ElectionTimeout timer. If this happens, a follower will+transition to a candidate state. These node states are illustrated in the type:++```haskell+data Mode+ = Follower+ | Candidate+ | Leader+```++The volatile state a node keeps track of may vary depending on the mode that it+is in. Using `DataKinds` and `GADTs`, we relate these specific node state+datatypes that contain the relevant data to the current node's mode with the+`NodeState` type. This way, we can enforce that the volatile state carried by a+node in mode `Follower` is indeed `FollowerState`, etc:++```haskell+-- | The volatile state of a Raft Node+data NodeState (a :: Mode) where+ NodeFollowerState :: FollowerState -> NodeState 'Follower+ NodeCandidateState :: CandidateState -> NodeState 'Candidate+ NodeLeaderState :: LeaderState -> NodeState 'Leader+```++The library's main event loop is comprised of a simple flow: Raft nodes receive+events on an STM channel, handle the event depending on the current node state,+return a list of actions to perform, and then perform those actions in the order+they were generated. The `Event` type specifies the main value to which raft+nodes react to, whereas the `Action` type specifies the action the raft node+performs as a result of the pairing of the current node state and received+event.++The Raft protocol has constraints on how nodes transition from one state to+another. For example, a follower cannot transition to a leader state+without first transitioning to a candidate state. Similarly, a leader can+never transition directly to a candidate state due to the algorithm+specification. Candidates are allowed to transition to any other node state.++To adhere to the Raft specification, we make use of some type level programming+to ensure that only valid transitions happen between node states.++```haskell+-- | All valid state transitions of a Raft node+data Transition (init :: Mode) (res :: Mode) where+ StartElection :: Transition 'Follower 'Candidate+ HigherTermFoundFollower :: Transition 'Follower 'Follower++ RestartElection :: Transition 'Candidate 'Candidate+ DiscoverLeader :: Transition 'Candidate 'Follower+ HigherTermFoundCandidate :: Transition 'Candidate 'Follower+ BecomeLeader :: Transition 'Candidate 'Leader++ SendHeartbeat :: Transition 'Leader 'Leader+ DiscoverNewLeader :: Transition 'Leader 'Follower+ HigherTermFoundLeader :: Transition 'Leader 'Follower++ Noop :: Transition init init+```++To compose the `Transition` with the resulting state from the event handler, we+use the `ResultState` datatype, existentially quantifying the result state mode:++```haskell+-- | Existential type hiding the result type of a transition, fixing the+-- result state to the state dictated by the 'Transition init res' value.+data ResultState init v where+ ResultState :: Transition init res -> NodeState res -> ResultState init+```++This datatype fixes the result state to be dependent on the transition that+occurred; as long as the allowed transitions are correctly denoted in the+`Transition` data constructors, only valid transitions can be specified by the+`ResultState`. Furthermore, `ResultState` values existentially hide the result+state types, that can be accessed via pattern matching. Thus, all event+handlers, be they RPC handlers, timeout handlers, or client request handlers,+have a type signature of the form:++```haskell+handler :: NodeState init -> ... relevant handler data ... -> ResultState init+```++Statically, the `ResultState` will enforce that invalid transitions are not made+when writing handlers for all combinations of raft node modes and events. In the+future, this approach may be extended to limit the actions a node can emit+dependent on its current mode.++## Library Architecture++Within the Raft protocol, there is a pure core that can be abstracted without+the use of global state. The protocol can be looked at simply as a series+of function calls of a function from an initial node state to a result node+state. However, sometimes these transitions have *side effects*. In this library+we have elected to separate the *pure* and **effectful** layers.++The core event handling loop is a *pure* function that, given the current node+state and a few extra bits of global state, computes a list of `Action`s for the+effectful layer to perform (updating global state, sending a message to another+node over the network, etc.).++### Pure Layer++In order to update the replicated state machine, clients contact the leader via+"client requests" containing commands to be committed to the replicated state+machine. Once a command is received, the current leader assesses whether it is+possible to commit the command to the replicated state machine.++The replicated state machine must be deterministic such that every command+committed by a leader to the state machine will eventually be replicated on+every node in the network at the same index.++As the only part of the internal event loop that needs to be specified manually,+We ask users of our library to provide an instance of the `StateMachine`+typeclass. This typeclass relates a state machine type to a command type+and a single type class function 'applyCommittedLogEntry', a pure function that+should return the result of applying the command to the initial state machine.++```haskell+class StateMachine sm v | sm -> v where+ applyCommittedLogEntry :: sm -> v -> sm+```++Everything else related to the core event handling loop is not exposed to+library users. All that needs to be specified is the type of the state machine,+the commands to update it, and how to perform those updates.++### Effectful Layers++In the protocol, there are two main components that need access to global+state and system resources. Firstly, raft nodes must maintain some persistent+state for efficient and correct recovery from network outages or partitions.+Secondly, raft nodes need to send messages to other raft nodes for the network+(the replicated state machine) to be operational.++#### Persistent State++Each node persists data to disk, including the replicated log+entries. Since persisting data is an action that programmers have many opinions+and preferences regarding, we provide two type classes that abstract the+specifics of writing log entries to disk as well as a few other small bits of+relevant data. These are separated due to the nature in which the log entries+are queried, often by specific index and without bounds. Thus, it may be+desirable to store the log entries in an efficient database. The remaining+persistent data is always read and written atomically, and has a much smaller+storage footprint.++The actions of reading or modifying existing log entries on disk is broken down+even further: we ask the user to specify how to write, delete, and read+log entries from disk. Often these types of operations can be optimized via+smarter persistent data solutions like modern SQL databases, thus we arrive at+the following level of granularity:++```haskell+-- | The type class specifying how nodes should write log entries to storage.+class Monad m => RaftWriteLog m v where+ type RaftWriteLogError m+ -- | Write the given log entries to storage+ writeLogEntries+ :: Exception (RaftWriteLogError m)+ => Entries v -> m (Either (RaftWriteLogError m) ())++-- | The type class specifying how nodes should delete log entries from storage.+class Monad m => RaftDeleteLog m v where+ type RaftDeleteLogError m+ -- | Delete log entries from a given index; e.g. 'deleteLogEntriesFrom 7'+ -- should delete every log entry+ deleteLogEntriesFrom+ :: Exception (RaftDeleteLogError m)+ => Index -> m (Either (RaftDeleteLogError m) (Maybe (Entry v)))++-- | The type class specifying how nodes should read log entries from storage.+class Monad m => RaftReadLog m v where+ type RaftReadLogError m+ -- | Read the log at a given index+ readLogEntry+ :: Exception (RaftReadLogError m)+ => Index -> m (Either (RaftReadLogError m) (Maybe (Entry v)))+ -- | Read log entries from a specific index onwards+ readLogEntriesFrom+ :: Exception (RaftReadLogError m)+ => Index -> m (Either (RaftReadLogError m) (Entries v))+ -- | Read the last log entry in the log+ readLastLogEntry+ :: Exception (RaftReadLogError m)+ => m (Either (RaftReadLogError m) (Maybe (Entry v)))+```++To read and write the `PersistentData` type (the remaining persistent data that+is not log entries), we ask the user to use the following `RaftPersist`+typeclass.++```haskell+-- | The RaftPersist type class specifies how to read and write the persistent+-- state to disk.+--+class Monad m => RaftPersist m where+ type RaftPersistError m+ readPersistentState+ :: Exception (RaftPersistError m)+ => m (Either (RaftPersistError m) PersistentState)+ writePersistentState+ :: Exception (RaftPersistError m)+ => PersistentState -> m (Either (RaftPersistError m) ())+```++### Networking++The other non-deterministic, effectful part of the protocol is the communication+between nodes over the network. It can be unreliable due to network delays,+partitions and packet loss, duplication and reordering, but the Raft consensus+algorithm was designed to achieve consensus in such harsh conditions.++The actions that must be performed in the networking layer are *sending RPCs* to+other raft nodes, *receiving RPCs* from other raft nodes, *sending client+responses* to clients who have issued requests, and *receiving client requests*+from clients wishing to update the replicated state. Depending on use of this+raft library, the two pairs are not necessary symmetric and so we do not+force the user into specifying a single way to send/receive messages to and from+raft nodes or clients.++We provide several type classes for users to specify the networking layer+themselves. The user must make sure that the `sendRPC`/`receiveRPC` and+`sendClient`/`receiveClient` pairs perform complementary actions; that an RPC+sent from one raft node to another is indeed receivable via `receiveRPC` on the+node to which it was sent:++```haskell+-- | Provide an interface for nodes to send messages to one+-- another. E.g. Control.Concurrent.Chan, Network.Socket, etc.+class RaftSendRPC m v where+ sendRPC :: NodeId -> RPCMessage v -> m ()++-- | Provide an interface for nodes to receive messages from one+-- another+class RaftRecvRPC m v where+ receiveRPC :: m (RPCMessage v)++-- | Provide an interface for Raft nodes to send messages to clients+class RaftSendClient m sm where+ sendClient :: ClientId -> ClientResponse sm -> m ()++-- | Provide an interface for Raft nodes to receive messages from clients+class RaftRecvClient m v where+ receiveClient :: m (ClientRequest v)+```++We have written a default implementation for network sockets over TCP in+[src/Examples/Raft/Socket](https://github.com/adjoint-io/raft/blob/master/src/Examples/Raft/Socket)++# Run example++We provide a complete example of the library where nodes communicate via network+sockets, and they write their logs on text files. See+[app/Main.hs](https://github.com/adjoint-io/raft/blob/master/app/Main.hs) to+have further insight.++1) Build the example executable:+```$ stack build ```++2) In separate terminals, run some raft nodes:++ The format of the cmd line invocation is:+ ``` raft-example <node-id> <peer-1-node-id> ... <peer-n-node-id> ```++ We are going to run a network of three nodes:++ - On terminal 1:+ ```$ stack exec raft-example localhost:3001 localhost:3002 localhost:3003```++ - On terminal 2:+ ```$ stack exec raft-example localhost:3002 localhost:3001 localhost:3003```++ - On terminal 3:+ ```$ stack exec raft-example localhost:3003 localhost:3001 localhost:3002```++ The first node spawned should become candidate once its election's timer+ times out and request votes to other nodes. It will then become the leader,+ once it receives a majority of votes and will broadcast messages to all+ nodes at each heartbeat.++3) Run a client:+```$ stack exec raft-example client```++ In the example provided, there are five basic operations:++ - `addNode <host:port>`: Add a nodeId to the set of nodeIds that the client+ will communicate with. Adding a single node will be sufficient, as this node+ will redirect the command to the leader in case he is not.++ - `getNodes`: Return all node ids that the client is aware of.++ - `read`: Return the state of the leader.++ - `set <var> <val>`: Set a variable to a specific value.++ - `incr <var>`: Increment the value of a variable.++ Assuming that two nodes are run as mentioned above, a valid client workflow+ would be:+ ```+ >>> addNode localhost:3001+ >>> set testVar 4+ >>> incr testVar+ >>> read+ ```++ It will return the state of the leader's state machine (and eventually the state+ of all nodes in the Raft network). In our example, it will be a map of a single+ key `testVar` of value `4`++## How to use this library++1. [Define the state+machine](https://github.com/adjoint-io/raft#define-the-state-machine)+2. [Implement the networking+layer](https://github.com/adjoint-io/raft#implement-the-networking-layer)+3. [Implement the persistent+layer](https://github.com/adjoint-io/raft#implement-the-persistent-layer)+4. [Putting it all+together](https://github.com/adjoint-io/raft#putting-it-all-together)++### Define the state machine++The only requirement for our state machine is to instantiate the `StateMachine`+type class.++```haskell+class StateMachine sm v | sm -> v where+ applyCommittedLogEntry :: sm -> v -> sm+```++In our [example](https://github.com/adjoint-io/raft/blob/master/app/Main.hs) we+use a simple map as a store whose values can only increase.++### Implement the networking layer++We leave the choice of the networking layer open to the user, as it can vary+depending on the use case (E.g. TCP/UDP/cloud-haskell/etc).++We need to specify how nodes will communicate with clients and with each other.+As described above in the [Networking+section](https://github.com/adjoint-io/raft#networking), it suffices to+implement those four type classes (`RaftSendRPC`, `RaftRecvRPC`,+`RaftSendClient`, `RaftRecvClient`).++In our example, we provide instances of nodes communicating over TCP to other+nodes+([Socket/Node.hs](https://github.com/adjoint-io/raft/blob/master/src/Examples/Raft/Socket/Node.hs))+and clients+([Socket/Client.hs](https://github.com/adjoint-io/raft/blob/master/src/Examples/Raft/Socket/Client.hs)).++Note that our datatypes will need to derive instances of `MonadThrow`,+`MonadCatch`, `MonadMask` and `MonadConc`. This allows us to test concurrent+properties of the system, using randomized thread scheduling to assert the+absence of deadlocks and exceptions.++In case of the `RaftSocketT` data type used in our example:++```haskell+deriving instance MonadConc m => MonadThrow (RaftSocketT v m)+deriving instance MonadConc m => MonadCatch (RaftSocketT v m)+deriving instance MonadConc m => MonadMask (RaftSocketT v m)+deriving instance MonadConc m => MonadConc (RaftSocketT v m)+```++### Implement the persistent layer++There are many different possibilities when it comes to persist data to disk, so+we also leave the specification open to the user.++As explained in the [Persistent+State](https://github.com/adjoint-io/raft#persistent-state) section above, we+will create instances for `RaftReadLog`, `RaftWriteLog` and+`RaftDeleteLog` to specify how we will read, write and+delete log entries, as well as `RaftPersist`.++We provide an implementation that stores persistent data on files in+[FileStore.hs](https://github.com/adjoint-io/raft/blob/master/src/Examples/Raft/FileStore.hs)++### Putting it all together++The last step is wrapping our previous data types that deal with+networking and persistent data into a single monad that also derives instances+of all the Raft type classes described (`RaftSendRPC`, `RaftRecvRPC`,+`RaftSendClient`, `RaftRecvClient`, `RaftReadLog`, `RaftWriteLog`,+`RaftDeleteLog` and `RaftPersist`).++In our example, this monad is `RaftExampleM sm v`. See+[app/Main.hs](https://github.com/adjoint-io/raft/blob/master/app/Main.hs).++Finally, we are ready to run our Raft nodes. We call the `runRaftNode` function+from the+[src/Raft.hs](https://github.com/adjoint-io/raft/blob/master/src/Raft.hs)+file, together with the function we define to run the stack of monads that+derive our Raft type classes.++# References++1. Ongaro, D., Ousterhout, J. [In Search of an Understandable Consensus+ Algorithm](https://raft.github.io/raft.pdf), 2014++2. Howard, H. [ARC: Analysis of Raft+ Consensus](https://www.cl.cam.ac.uk/techreports/UCAM-CL-TR-857.pdf) 2014
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ app/Main.hs view
@@ -0,0 +1,344 @@+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Main where++import Protolude hiding+ ( MVar, putMVar, takeMVar, newMVar, newEmptyMVar, readMVar+ , atomically, STM(..), Chan, newTVar, readTVar, writeTVar+ , newChan, writeChan, readChan+ , threadDelay, killThread, TVar(..)+ , catch, handle, takeWhile, takeWhile1, (<|>)+ , lift+ )++import Control.Concurrent.Classy hiding (catch)+import Control.Monad.Catch+import Control.Monad.Trans.Class++import Data.Sequence ((><))+import qualified Data.Map as Map+import qualified Data.List as L+import qualified Data.Set as Set+import qualified Data.Sequence as Seq+import qualified Data.Serialize as S+import qualified Network.Simple.TCP as NS+import Network.Simple.TCP+import qualified Network.Socket as N+import qualified Network.Socket.ByteString as NSB+import Numeric.Natural+import System.Console.Repline+import System.Console.Haskeline.MonadException hiding (handle)+import Text.Read hiding (lift)+import System.Random+import qualified System.Directory as Directory++import qualified Examples.Raft.Socket.Client as RS+import qualified Examples.Raft.Socket.Node as RS+import Examples.Raft.Socket.Node+import qualified Examples.Raft.Socket.Common as RS++import Examples.Raft.FileStore+import Raft++------------------------------+-- State Machine & Commands --+------------------------------++-- State machine with two basic operations: set a variable to a value and+-- increment value++type Var = ByteString++data StoreCmd+ = Set Var Natural+ | Incr Var+ deriving (Show, Generic)++instance S.Serialize StoreCmd++type Store = Map Var Natural++instance RSMP Store StoreCmd where+ data RSMPError Store StoreCmd = StoreError Text deriving (Show)+ type RSMPCtx Store StoreCmd = ()++ applyCmdRSMP _ store cmd =+ Right $ case cmd of+ Set x n -> Map.insert x n store+ Incr x -> Map.adjust succ x store++instance (sm ~ Store, v ~ StoreCmd, RSMP sm v) => RSM sm v (RaftExampleM sm v) where+ validateCmd _ = pure (Right ())+ askRSMPCtx = pure ()++--------------------+-- Raft instances --+--------------------++data NodeEnv sm = NodeEnv+ { nEnvStore :: TVar (STM IO) sm+ , nEnvNodeId :: NodeId+ }++newtype RaftExampleM sm v a = RaftExampleM { unRaftExampleM :: ReaderT (NodeEnv sm) (RaftSocketT v (RaftFileStoreT IO)) a }+ deriving (Functor, Applicative, Monad, MonadIO, MonadReader (NodeEnv sm), Alternative, MonadPlus)++deriving instance MonadThrow (RaftExampleM sm v)+deriving instance MonadCatch (RaftExampleM sm v)+deriving instance MonadMask (RaftExampleM sm v)+deriving instance MonadConc (RaftExampleM sm v)++runRaftExampleM :: NodeEnv sm -> NodeSocketEnv v -> NodeFileStoreEnv -> RaftExampleM sm v a -> IO a+runRaftExampleM nodeEnv nodeSocketEnv nodeFileStoreEnv raftExampleM =+ runReaderT (unRaftFileStoreT $+ runReaderT (unRaftSocketT $+ runReaderT (unRaftExampleM raftExampleM) nodeEnv) nodeSocketEnv)+ nodeFileStoreEnv++instance RaftSendClient (RaftExampleM Store StoreCmd) Store where+ sendClient cid msg = (RaftExampleM . lift) $ sendClient cid msg++instance RaftRecvClient (RaftExampleM Store StoreCmd) StoreCmd where+ type RaftRecvClientError (RaftExampleM Store StoreCmd) StoreCmd = Text+ receiveClient = RaftExampleM $ lift receiveClient++instance RaftSendRPC (RaftExampleM Store StoreCmd) StoreCmd where+ sendRPC nid msg = (RaftExampleM . lift) $ sendRPC nid msg++instance RaftRecvRPC (RaftExampleM Store StoreCmd) StoreCmd where+ type RaftRecvRPCError (RaftExampleM Store StoreCmd) StoreCmd = Text+ receiveRPC = RaftExampleM $ lift receiveRPC++instance RaftWriteLog (RaftExampleM Store StoreCmd) StoreCmd where+ type RaftWriteLogError (RaftExampleM Store StoreCmd) = NodeEnvError+ writeLogEntries entries = RaftExampleM $ lift $ RaftSocketT (lift $ writeLogEntries entries)++instance RaftPersist (RaftExampleM Store StoreCmd) where+ type RaftPersistError (RaftExampleM Store StoreCmd) = NodeEnvError+ writePersistentState ps = RaftExampleM $ lift $ RaftSocketT (lift $ writePersistentState ps)+ readPersistentState = RaftExampleM $ lift $ RaftSocketT (lift $ readPersistentState)++instance RaftReadLog (RaftExampleM Store StoreCmd) StoreCmd where+ type RaftReadLogError (RaftExampleM Store StoreCmd) = NodeEnvError+ readLogEntry idx = RaftExampleM $ lift $ RaftSocketT (lift $ readLogEntry idx)+ readLastLogEntry = RaftExampleM $ lift $ RaftSocketT (lift readLastLogEntry)++instance RaftDeleteLog (RaftExampleM Store StoreCmd) StoreCmd where+ type RaftDeleteLogError (RaftExampleM Store StoreCmd) = NodeEnvError+ deleteLogEntriesFrom idx = RaftExampleM $ lift $ RaftSocketT (lift $ deleteLogEntriesFrom idx)++--------------------+-- Client console --+--------------------++-- Clients interact with the nodes from a terminal:+-- Accepted operations are:+-- - addNode <host:port>+-- Add nodeId to the set of nodeIds that the client will communicate with+-- - getNodes+-- Return the node ids that the client is aware of+-- - read+-- Return the state of the leader+-- - set <var> <val>+-- Set variable to a specific value+-- - incr <var>+-- Increment the value of a variable++data ConsoleState = ConsoleState+ { csNodeIds :: NodeIds -- ^ Set of node ids that the client is aware of+ , csSocket :: Socket -- ^ Client's socket+ , csHost :: HostName -- ^ Client's host+ , csPort :: ServiceName -- ^ Client's port+ , csLeaderId :: TVar (STM IO) (Maybe LeaderId) -- ^ Node id of the leader in the Raft network+ }++newtype ConsoleT m a = ConsoleT+ { unConsoleT :: StateT ConsoleState m a+ } deriving (Functor, Applicative, Monad, MonadIO, MonadState ConsoleState)++newtype ConsoleM a = ConsoleM+ { unConsoleM :: HaskelineT (ConsoleT IO) a+ } deriving (Functor, Applicative, Monad, MonadIO, MonadState ConsoleState)++instance MonadException m => MonadException (ConsoleT m) where+ controlIO f =+ ConsoleT $ StateT $ \s ->+ controlIO $ \(RunIO run) ->+ let run' = RunIO (fmap (ConsoleT . StateT . const) . run . flip runStateT s . unConsoleT)+ in flip runStateT s . unConsoleT <$> f run'++-- | Evaluate and handle each line user inputs+handleConsoleCmd :: [Char] -> ConsoleM ()+handleConsoleCmd input = do+ nids <- gets csNodeIds+ clientSocket <- gets csSocket+ clientHost <- gets csHost+ clientPort <- gets csPort+ leaderIdT <- gets csLeaderId+ leaderIdM <- liftIO $ atomically $ readTVar leaderIdT+ let clientSocketEnv = RS.ClientSocketEnv clientPort clientHost clientSocket+ case L.words input of+ ["addNode", nid] -> modify (\st -> st { csNodeIds = Set.insert (toS nid) (csNodeIds st) })+ ["getNodes"] -> print nids+ ["read"] -> if nids == Set.empty+ then putText "Please add some nodes to query first. Eg. `addNode localhost:3001`"+ else do+ respE <- liftIO $ RS.runRaftSocketClientM clientSocketEnv $ case leaderIdM of+ Nothing -> RS.sendReadRndNode (Proxy :: Proxy StoreCmd) nids+ Just (LeaderId nid) -> RS.sendRead (Proxy :: Proxy StoreCmd) nid+ handleClientResponseE input respE leaderIdT+ ["incr", cmd] -> do+ respE <- liftIO $ RS.runRaftSocketClientM clientSocketEnv $ case leaderIdM of+ Nothing -> RS.sendWriteRndNode (Incr (toS cmd)) nids+ Just (LeaderId nid) -> RS.sendWrite (Incr (toS cmd)) nid+ handleClientResponseE input respE leaderIdT+ ["set", var, val] -> do+ respE <- liftIO $ RS.runRaftSocketClientM clientSocketEnv $ case leaderIdM of+ Nothing -> RS.sendWriteRndNode (Set (toS var) (read val)) nids+ Just (LeaderId nid) -> RS.sendWrite (Set (toS var) (read val)) nid+ handleClientResponseE input respE leaderIdT+ _ -> print "Invalid command. Press <TAB> to see valid commands"++ where+ handleClientResponseE :: [Char] -> Either [Char] (ClientResponse Store) -> TVar (STM IO) (Maybe LeaderId) -> ConsoleM ()+ handleClientResponseE input eMsgE leaderIdT =+ case eMsgE of+ Left err -> panic $ toS err+ Right (ClientRedirectResponse (ClientRedirResp leader)) ->+ case leader of+ NoLeader -> do+ putText "Sorry, the system doesn't have a leader at the moment"+ liftIO $ atomically $ writeTVar leaderIdT Nothing+ -- If the message was not sent to the leader, that node will+ -- point to the current leader+ CurrentLeader lid -> do+ putText $ "New leader found: " <> show lid+ liftIO $ atomically $ writeTVar leaderIdT (Just lid)+ handleConsoleCmd input+ Right (ClientReadResponse (ClientReadResp sm)) -> putText $ "Received sm: " <> show sm+ Right (ClientWriteResponse writeResp) -> print writeResp+++main :: IO ()+main = do+ args <- (toS <$>) <$> getArgs+ case args of+ ["client"] -> clientMainHandler+ (nid:nids) -> do+ removeExampleFiles nid+ createExampleFiles nid++ nSocketEnv <- initSocketEnv nid+ nPersistentEnv <- initRaftFileStoreEnv nid+ nEnv <- initNodeEnv nid+ runRaftExampleM nEnv nSocketEnv nPersistentEnv $ do+ let allNodeIds = Set.fromList (nid : nids)+ let nodeConfig = NodeConfig+ { configNodeId = toS nid+ , configNodeIds = allNodeIds+ , configElectionTimeout = (1500000, 3000000)+ , configHeartbeatTimeout = 200000+ }+ RaftExampleM $ lift acceptForkNode :: RaftExampleM Store StoreCmd ()+ electionTimerSeed <- liftIO randomIO+ runRaftNode nodeConfig LogStdout electionTimerSeed (mempty :: Store)+ where+ initPersistentFile :: NodeId -> IO ()+ initPersistentFile nid = do+ psPath <- persistentFile nid+ writeFile psPath (toS $ S.encode initPersistentState)++ persistentFile :: NodeId -> IO FilePath+ persistentFile nid = do+ tmpDir <- Directory.getTemporaryDirectory+ pure $ tmpDir ++ "/" ++ toS nid ++ "/" ++ "persistent"++ initLogsFile :: NodeId -> IO ()+ initLogsFile nid = do+ logsPath <- logsFile nid+ writeFile logsPath (toS $ S.encode (mempty :: Entries StoreCmd))++ logsFile :: NodeId -> IO FilePath+ logsFile nid = do+ tmpDir <- Directory.getTemporaryDirectory+ pure (tmpDir ++ "/" ++ toS nid ++ "/" ++ "logs")++ createExampleFiles :: NodeId -> IO ()+ createExampleFiles nid = void $ do+ tmpDir <- Directory.getTemporaryDirectory+ Directory.createDirectory (tmpDir ++ "/" ++ toS nid)+ initPersistentFile nid+ initLogsFile nid++ removeExampleFiles :: NodeId -> IO ()+ removeExampleFiles nid = handle (const (pure ()) :: SomeException -> IO ()) $ do+ tmpDir <- Directory.getTemporaryDirectory+ Directory.removeDirectoryRecursive (tmpDir ++ "/" ++ toS nid)+++ initNodeEnv :: NodeId -> IO (NodeEnv Store)+ initNodeEnv nid = do+ let (host, port) = RS.nidToHostPort (toS nid)+ storeTVar <- atomically (newTVar mempty)+ pure NodeEnv+ { nEnvStore = storeTVar+ , nEnvNodeId = toS host <> ":" <> toS port+ }++ initRaftFileStoreEnv :: NodeId -> IO NodeFileStoreEnv+ initRaftFileStoreEnv nid = do+ psPath <- persistentFile nid+ psLogs <- logsFile nid+ pure NodeFileStoreEnv+ { nfsPersistentState = psPath+ , nfsLogEntries = psLogs+ }++ initSocketEnv :: NodeId -> IO (NodeSocketEnv v)+ initSocketEnv nid = do+ let (host, port) = RS.nidToHostPort (toS nid)+ nodeSocket <- newSock host port+ socketPeersTVar <- atomically (newTVar mempty)+ msgQueue <- atomically newTChan+ clientReqQueue <- atomically newTChan+ pure NodeSocketEnv+ { nsPeers = socketPeersTVar+ , nsSocket = nodeSocket+ , nsMsgQueue = msgQueue+ , nsClientReqQueue = clientReqQueue+ }++ clientMainHandler :: IO ()+ clientMainHandler = do+ clientPort <- RS.getFreePort+ clientSocket <- RS.newSock "localhost" clientPort+ leaderIdT <- atomically (newTVar Nothing)+ let initClientState = ConsoleState+ { csNodeIds = mempty+ , csSocket = clientSocket+ , csHost = "localhost"+ , csPort = clientPort+ , csLeaderId = leaderIdT+ }+ runConsoleT initClientState $+ evalRepl (pure ">>> ") (unConsoleM . handleConsoleCmd) [] Nothing (Word completer) (pure ())++ runConsoleT :: Monad m => ConsoleState -> ConsoleT m a -> m a+ runConsoleT consoleState = flip evalStateT consoleState . unConsoleT++++ -- Tab Completion: return a completion for partial words entered+ completer :: Monad m => WordCompleter m+ completer n = do+ let cmds = ["addNode <host:port>", "getNodes", "incr <var>", "set <var> <val>"]+ return $ filter (isPrefixOf n) cmds
+ libraft.cabal view
@@ -0,0 +1,157 @@+-- This file has been generated from package.yaml by hpack version 0.28.2.+--+-- see: https://github.com/sol/hpack+--+-- hash: b32be6507cfec629f59764cb08eca1bc86e11f5e67036477ae0aaa54e6b1458c++name: libraft+version: 0.1.0.0+synopsis: Raft consensus algorithm+description: Please see the README on GitHub at <https://github.com/adjoint-io/raft#readme>+category: Distributed Systems+homepage: https://github.com/adjoint-io/raft#readme+bug-reports: https://github.com/adjoint-io/raft/issues+author: Adjoint Inc.+maintainer: info@adjoint.io+copyright: 2018 Adjoint Inc.+license: BSD3+license-file: LICENSE+build-type: Simple+cabal-version: >= 1.10+extra-source-files:+ ChangeLog.md+ README.md++source-repository head+ type: git+ location: https://github.com/adjoint-io/raft++library+ exposed-modules:+ Control.Concurrent.STM.Timer+ Examples.Raft.FileStore+ Examples.Raft.Socket.Client+ Examples.Raft.Socket.Common+ Examples.Raft.Socket.Node+ Raft+ Raft.Action+ Raft.Candidate+ Raft.Client+ Raft.Config+ Raft.Event+ Raft.Follower+ Raft.Handle+ Raft.Leader+ Raft.Log+ Raft.Logging+ Raft.Monad+ Raft.NodeState+ Raft.Persistent+ Raft.RPC+ Raft.Types+ other-modules:+ Paths_libraft+ hs-source-dirs:+ src+ default-extensions: NoImplicitPrelude OverloadedStrings LambdaCase+ ghc-options: -fwarn-unused-binds -fwarn-unused-imports+ build-depends:+ attoparsec+ , base >=4.7 && <5+ , bytestring+ , cereal+ , concurrency+ , containers+ , directory+ , exceptions+ , haskeline+ , mtl+ , network+ , network-simple+ , parsec+ , protolude+ , random+ , repline+ , text+ , time+ , transformers+ , word8+ default-language: Haskell2010++executable raft-example+ main-is: Main.hs+ other-modules:+ Paths_libraft+ hs-source-dirs:+ app+ default-extensions: NoImplicitPrelude OverloadedStrings LambdaCase+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends:+ attoparsec+ , base >=4.7 && <5+ , bytestring+ , cereal+ , concurrency+ , containers+ , directory+ , exceptions+ , haskeline+ , libraft+ , mtl+ , network+ , network-simple+ , parsec+ , protolude+ , random+ , repline+ , stm+ , text+ , time+ , transformers+ , word8+ default-language: Haskell2010++test-suite raft-test+ type: exitcode-stdio-1.0+ main-is: TestDriver.hs+ other-modules:+ TestDejaFu+ TestRaft+ TestUtils+ Paths_libraft+ hs-source-dirs:+ test+ default-extensions: NoImplicitPrelude OverloadedStrings LambdaCase+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends:+ QuickCheck+ , attoparsec+ , base >=4.7 && <5+ , bytestring+ , cereal+ , concurrency+ , containers+ , dejafu+ , directory+ , exceptions+ , haskeline+ , hunit-dejafu+ , libraft+ , mtl+ , network+ , network-simple+ , parsec+ , protolude+ , random+ , repline+ , tasty+ , tasty-dejafu+ , tasty-discover+ , tasty-expected-failure+ , tasty-hunit+ , tasty-quickcheck+ , text+ , time+ , transformers+ , word8+ default-language: Haskell2010
+ src/Control/Concurrent/STM/Timer.hs view
@@ -0,0 +1,75 @@++module Control.Concurrent.STM.Timer (+ Timer,+ waitTimer,+ startTimer,+ resetTimer,+ newTimer,+ newTimerRange,+) where++import Protolude hiding (STM, killThread, ThreadId, threadDelay, myThreadId, atomically)++import Control.Monad.Conc.Class+import Control.Concurrent.Classy.STM+import System.Random (StdGen, randomR, mkStdGen)++import Numeric.Natural++data Timer m = Timer+ { timerThread :: TMVar (STM m) (ThreadId m)+ , timerLock :: TMVar (STM m) ()+ , timerRange :: (Natural, Natural)+ , timerGen :: TVar (STM m) StdGen+ }++waitTimer :: MonadConc m => Timer m -> m ()+waitTimer (Timer _ lock _ _) =+ atomically $ readTMVar lock++-- | Starting a timer will only work if the timer is currently stopped+startTimer :: MonadConc m => Timer m -> m ()+startTimer timer = do+ lock <- atomically $ tryTakeTMVar (timerLock timer)+ case lock of+ Nothing -> pure ()+ Just () -> spawnTimer timer++resetTimer :: MonadConc m => Timer m -> m ()+resetTimer timer = do+ -- Kill the old timer+ mOldTid <- atomically $ tryTakeTMVar (timerThread timer)+ case mOldTid of+ Just oldTid -> killThread oldTid+ Nothing -> pure ()+ -- Spawn a new timer+ spawnTimer timer++-- | Spawn a timer thread. This function assumes that there are no other threads+-- are using the timer+spawnTimer :: MonadConc m => Timer m -> m ()+spawnTimer timer = do+ g <- atomically $ readTVar (timerGen timer)+ let (tmin, tmax) = timerRange timer+ (n, g') = randomR (toInteger tmin, toInteger tmax) g+ atomically $ writeTVar (timerGen timer) g'+ void $ fork $ do+ threadId <- myThreadId+ atomically $ do+ _ <- tryTakeTMVar (timerLock timer)+ _ <- tryTakeTMVar (timerThread timer)+ void $ putTMVar (timerThread timer) threadId+ threadDelay (fromIntegral n)+ atomically $ do+ _ <- tryTakeTMVar (timerThread timer)+ putTMVar (timerLock timer) ()++newTimer :: MonadConc m => Natural -> m (Timer m)+newTimer timeout = newTimerRange 0 (timeout, timeout)++-- | Create a new timer with the given random seed and range of timer timeouts.+newTimerRange :: MonadConc m => Int -> (Natural, Natural) -> m (Timer m)+newTimerRange seed timeoutRange = do+ (timerThread, timerLock, timerGen) <-+ atomically $ (,,) <$> newEmptyTMVar <*> newTMVar () <*> newTVar (mkStdGen seed)+ pure $ Timer timerThread timerLock timeoutRange timerGen
+ src/Examples/Raft/FileStore.hs view
@@ -0,0 +1,101 @@+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FunctionalDependencies #-}++module Examples.Raft.FileStore where++import Protolude++import Control.Concurrent.Classy hiding (catch, ThreadId)+import Control.Monad.Catch+import Control.Monad.Trans.Class++import qualified Data.ByteString as BS+import Data.Sequence ((><))+import qualified Data.Sequence as Seq+import qualified Data.Serialize as S++import Raft++newtype NodeEnvError = NodeEnvError Text+ deriving (Show)++instance Exception NodeEnvError++data NodeFileStoreEnv = NodeFileStoreEnv+ { nfsPersistentState :: FilePath+ , nfsLogEntries :: FilePath+ }++newtype RaftFileStoreT m a = RaftFileStoreT { unRaftFileStoreT :: ReaderT NodeFileStoreEnv m a }+ deriving (Functor, Applicative, Monad, MonadIO, MonadReader NodeFileStoreEnv, Alternative, MonadPlus, MonadTrans)++deriving instance MonadConc m => MonadThrow (RaftFileStoreT m)+deriving instance MonadConc m => MonadCatch (RaftFileStoreT m)+deriving instance MonadConc m => MonadMask (RaftFileStoreT m)+deriving instance MonadConc m => MonadConc (RaftFileStoreT m)++--------------------+-- Raft Instances --+--------------------++instance (MonadIO m, MonadConc m, S.Serialize v) => RaftWriteLog (RaftFileStoreT m) v where+ type RaftWriteLogError (RaftFileStoreT m) = NodeEnvError+ writeLogEntries newEntries = do+ entriesPath <- asks nfsLogEntries+ eLogEntries <- readLogEntries+ case eLogEntries of+ Left err -> panic ("writeLogEntries: " <> err)+ Right currEntries -> liftIO $ Right <$> BS.writeFile entriesPath (S.encode (currEntries >< newEntries))++instance (MonadIO m, MonadConc m) => RaftPersist (RaftFileStoreT m) where+ type RaftPersistError (RaftFileStoreT m) = NodeEnvError+ writePersistentState ps = do+ psPath <- asks nfsPersistentState+ liftIO $ Right <$> BS.writeFile psPath (S.encode ps)++ readPersistentState = do+ psPath <- asks nfsPersistentState+ fileContent <- liftIO $ BS.readFile psPath+ case S.decode fileContent of+ Left err -> panic (toS $ "readPersistentState: " ++ err)+ Right ps -> pure $ Right ps++instance (MonadIO m, MonadConc m, S.Serialize v) => RaftReadLog (RaftFileStoreT m) v where+ type RaftReadLogError (RaftFileStoreT m) = NodeEnvError+ readLogEntry (Index idx) = do+ eLogEntries <- readLogEntries+ case eLogEntries of+ Left err -> panic ("readLogEntry: " <> err)+ Right entries ->+ case entries Seq.!? fromIntegral (if idx == 0 then 0 else idx - 1) of+ Nothing -> pure (Right Nothing)+ Just e -> pure (Right (Just e))++ readLastLogEntry = do+ eLogEntries <- readLogEntries+ case eLogEntries of+ Left err -> panic (toS err)+ Right entries -> case entries of+ Seq.Empty -> pure (Right Nothing)+ (_ Seq.:|> e) -> pure (Right (Just e))++instance (MonadIO m, MonadConc m, S.Serialize v) => RaftDeleteLog (RaftFileStoreT m) v where+ type RaftDeleteLogError (RaftFileStoreT m) = NodeEnvError+ deleteLogEntriesFrom idx = do+ eLogEntries <- readLogEntries+ case eLogEntries of+ Left err -> panic ("deleteLogEntriesFrom: " <> err)+ Right (entries :: Entries v) -> pure $ const (Right DeleteSuccess) $ Seq.dropWhileR ((>= idx) . entryIndex) entries++readLogEntries :: (MonadIO m, S.Serialize v) => RaftFileStoreT m (Either Text (Entries v))+readLogEntries = liftIO . fmap (first toS . S.decode) . BS.readFile . toS =<< asks nfsLogEntries
+ src/Examples/Raft/Socket/Client.hs view
@@ -0,0 +1,79 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Examples.Raft.Socket.Client where++import Protolude++import Control.Concurrent.Classy+import qualified Data.Serialize as S+import qualified Network.Simple.TCP as N+import qualified Data.Set as Set+import qualified Data.List as L+import System.Random++import Raft.Client+import Raft.Event+import Raft.Types+import Examples.Raft.Socket.Common++data ClientSocketEnv+ = ClientSocketEnv { clientPort :: N.ServiceName+ , clientHost :: N.HostName+ , clientSocket :: N.Socket+ } deriving (Show)++newtype RaftSocketClientM a+ = RaftSocketClientM { unRaftSocketClientM :: ReaderT ClientSocketEnv IO a }+ deriving (Functor, Applicative, Monad, MonadIO, MonadReader ClientSocketEnv, Alternative, MonadPlus)++runRaftSocketClientM :: ClientSocketEnv -> RaftSocketClientM a -> IO a+runRaftSocketClientM socketEnv = flip runReaderT socketEnv . unRaftSocketClientM++-- | Randomly select a node from a set of nodes a send a message to it+selectRndNode :: NodeIds -> IO NodeId+selectRndNode nids =+ (Set.toList nids L.!!) <$> randomRIO (0, length nids - 1)++-- | Randomly read the state of a random node+sendReadRndNode :: (S.Serialize sm, S.Serialize v) => Proxy v -> NodeIds -> RaftSocketClientM (Either [Char] (ClientResponse sm))+sendReadRndNode proxyV nids =+ liftIO (selectRndNode nids) >>= sendRead proxyV++-- | Randomly write to a random node+sendWriteRndNode :: (S.Serialize v, S.Serialize sm) => v -> NodeIds -> RaftSocketClientM (Either [Char] (ClientResponse sm))+sendWriteRndNode cmd nids =+ liftIO (selectRndNode nids) >>= sendWrite cmd++-- | Request the state of a node. It blocks until the node responds+sendRead :: forall v sm. (S.Serialize sm, S.Serialize v) => Proxy v -> NodeId -> RaftSocketClientM (Either [Char] (ClientResponse sm))+sendRead _ nid = do+ socketEnv@ClientSocketEnv{..} <- ask+ let (host, port) = nidToHostPort nid+ clientId = ClientId (hostPortToNid (clientHost, clientPort))+ liftIO $ fork $ N.connect host port $ \(sock, sockAddr) -> N.send sock+ (S.encode (ClientRequestEvent (ClientRequest clientId ClientReadReq :: ClientRequest v)))+ acceptClientConnections++-- | Write to a node. It blocks until the node responds+sendWrite :: (S.Serialize v, S.Serialize sm) => v -> NodeId -> RaftSocketClientM (Either [Char] (ClientResponse sm))+sendWrite cmd nid = do+ socketEnv@ClientSocketEnv{..} <- ask+ let (host, port) = nidToHostPort nid+ clientId = ClientId (hostPortToNid (clientHost, clientPort))+ liftIO $ fork $ N.connect host port $ \(sock, sockAddr) -> N.send sock+ (S.encode (ClientRequestEvent (ClientRequest clientId (ClientWriteReq cmd))))+ acceptClientConnections++-- | Accept a connection and return the client response synchronously+acceptClientConnections :: S.Serialize sm => RaftSocketClientM (Either [Char] (ClientResponse sm))+acceptClientConnections = do+ socketEnv@ClientSocketEnv{..} <- ask+ liftIO $ N.accept clientSocket $ \(sock', sockAddr') -> do+ recvSockM <- N.recv sock' 4096+ case recvSockM of+ Nothing -> pure $ Left "Received empty data from socket"+ Just recvSock -> pure (S.decode recvSock)+
+ src/Examples/Raft/Socket/Common.hs view
@@ -0,0 +1,30 @@+module Examples.Raft.Socket.Common where++import Protolude++import qualified Data.ByteString as BS+import qualified Network.Simple.TCP as N+import qualified Network.Socket as NS+import qualified Data.Word8 as W8++import Raft.Types++-- | Convert a host and a port to a valid NodeId+hostPortToNid :: (N.HostName, N.ServiceName) -> NodeId+hostPortToNid (host, port) = toS $ host ++ ":" ++ toS port++-- | Retrieve the host and port from a valid NodeId+nidToHostPort :: NodeId -> (N.HostName, N.ServiceName)+nidToHostPort bs =+ case BS.split W8._colon bs of+ [host,port] -> (toS host, toS port)+ _ -> panic "nidToHostPort: invalid node id"++-- | Get a free port number.+getFreePort :: IO N.ServiceName+getFreePort = do+ sock <- NS.socket NS.AF_INET NS.Stream NS.defaultProtocol+ NS.bind sock (NS.SockAddrInet NS.aNY_PORT NS.iNADDR_ANY)+ port <- NS.socketPort sock+ NS.close sock+ pure $ show port
+ src/Examples/Raft/Socket/Node.hs view
@@ -0,0 +1,162 @@+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Examples.Raft.Socket.Node where++import Protolude hiding+ ( MVar, putMVar, takeMVar, newMVar, newEmptyMVar, readMVar+ , atomically, STM(..), Chan, newTVar, readTVar, writeTVar+ , newChan, writeChan, readChan+ , threadDelay, killThread, TVar(..)+ , catch, handle, takeWhile, takeWhile1, (<|>)+ )++import Control.Concurrent.Classy hiding (catch, ThreadId)+import Control.Monad.Catch+import Control.Monad.Trans.Class++import qualified Data.Map as Map+import qualified Data.Serialize as S+import qualified Network.Simple.TCP as NS+import Network.Simple.TCP++import Examples.Raft.Socket.Common++import Raft++--------------------------------------------------------------------------------+-- Network+--------------------------------------------------------------------------------++data NodeSocketEnv v = NodeSocketEnv+ { nsSocket :: Socket+ , nsPeers :: TVar (STM IO) (Map NodeId Socket)+ , nsMsgQueue :: TChan (STM IO) (RPCMessage v)+ , nsClientReqQueue :: TChan (STM IO) (ClientRequest v)+ }++newtype RaftSocketT v m a = RaftSocketT { unRaftSocketT :: ReaderT (NodeSocketEnv v) m a }+ deriving (Functor, Applicative, Monad, MonadIO, MonadReader (NodeSocketEnv v), Alternative, MonadPlus, MonadTrans)++deriving instance MonadConc m => MonadThrow (RaftSocketT v m)+deriving instance MonadConc m => MonadCatch (RaftSocketT v m)+deriving instance MonadConc m => MonadMask (RaftSocketT v m)+deriving instance MonadConc m => MonadConc (RaftSocketT v m)++--------------------+-- Raft Instances --+--------------------++instance (MonadIO m, MonadConc m, S.Serialize sm) => RaftSendClient (RaftSocketT v m) sm where+ sendClient clientId@(ClientId nid) msg = do+ let (cHost, cPort) = nidToHostPort (toS nid)+ connect cHost cPort $ \(cSock, _cSockAddr) ->+ send cSock (S.encode msg)++instance (MonadIO m, MonadConc m, S.Serialize v) => RaftRecvClient (RaftSocketT v m) v where+ type RaftRecvClientError (RaftSocketT v m) v = Text+ receiveClient = do+ cReq <- asks nsClientReqQueue+ fmap Right . liftIO . atomically $ readTChan cReq++instance (MonadIO m, MonadConc m, S.Serialize v, Show v) => RaftSendRPC (RaftSocketT v m) v where+ sendRPC nid msg = do+ tNodeSocketEnvPeers <- asks nsPeers+ nodeSocketPeers <- liftIO $ atomically $ readTVar tNodeSocketEnvPeers+ sockM <- liftIO $+ case Map.lookup nid nodeSocketPeers of+ Nothing -> handle (handleFailure tNodeSocketEnvPeers [nid] Nothing) $ do+ (sock, _) <- connectSock host port+ NS.send sock (S.encode $ RPCMessageEvent msg)+ pure $ Just sock+ Just sock -> handle (retryConnection tNodeSocketEnvPeers nid (Just sock) msg) $ do+ NS.send sock (S.encode $ RPCMessageEvent msg)+ pure $ Just sock+ liftIO $ atomically $ case sockM of+ Nothing -> pure ()+ Just sock -> writeTVar tNodeSocketEnvPeers (Map.insert nid sock nodeSocketPeers)+ where+ (host, port) = nidToHostPort nid++instance (MonadIO m, MonadConc m, Show v) => RaftRecvRPC (RaftSocketT v m) v where+ type RaftRecvRPCError (RaftSocketT v m) v = Text+ receiveRPC = do+ msgQueue <- asks nsMsgQueue+ fmap Right . liftIO . atomically $ readTChan msgQueue+++-----------+-- Utils --+-----------++-- | Handles connections failures by first trying to reconnect+retryConnection+ :: (S.Serialize v, MonadIO m, MonadConc m)+ => TVar (STM m) (Map NodeId Socket)+ -> NodeId+ -> Maybe Socket+ -> RPCMessage v+ -> SomeException+ -> m (Maybe Socket)+retryConnection tNodeSocketEnvPeers nid sockM msg e = case sockM of+ Nothing -> pure Nothing+ Just sock ->+ handle (handleFailure tNodeSocketEnvPeers [nid] Nothing) $ do+ (sock, _) <- connectSock host port+ NS.send sock (S.encode $ RPCMessageEvent msg)+ pure $ Just sock+ where+ (host, port) = nidToHostPort nid++handleFailure+ :: (MonadIO m, MonadConc m)+ => TVar (STM m) (Map NodeId Socket)+ -> [NodeId]+ -> Maybe Socket+ -> SomeException+ -> m (Maybe Socket)+handleFailure tNodeSocketEnvPeers nids sockM e = case sockM of+ Nothing -> pure Nothing+ Just sock -> do+ nodeSocketPeers <- atomically $ readTVar tNodeSocketEnvPeers+ closeSock sock+ atomically $ mapM_ (\nid -> writeTVar tNodeSocketEnvPeers (Map.delete nid nodeSocketPeers)) nids+ pure Nothing+++runRaftSocketT :: (MonadIO m, MonadConc m) => NodeSocketEnv v -> RaftSocketT v m a -> m a+runRaftSocketT nodeSocketEnv = flip runReaderT nodeSocketEnv . unRaftSocketT++-- | Recursively accept a connection.+-- It keeps trying to accept connections even when a node dies+acceptForkNode+ :: forall v m. (S.Serialize v, MonadIO m, MonadConc m)+ => RaftSocketT v m ()+acceptForkNode = do+ socketEnv@NodeSocketEnv{..} <- ask+ void $ fork $ void $ forever $ acceptFork nsSocket $ \(sock', sockAddr') ->+ forever $ do+ recvSockM <- recv sock' 4096+ case recvSockM of+ Nothing -> panic "Socket was closed on the other end"+ Just recvSock -> case ((S.decode :: ByteString -> Either [Char] (MessageEvent v)) recvSock) of+ Left err -> panic $ toS err+ Right (ClientRequestEvent req@(ClientRequest cid _)) ->+ atomically $ writeTChan nsClientReqQueue req+ Right (RPCMessageEvent msg) ->+ atomically $ writeTChan nsMsgQueue msg++newSock :: HostName -> ServiceName -> IO Socket+newSock host port = do+ (sock, _) <- bindSock (Host host) port+ listenSock sock 2048+ pure sock
+ src/Raft.hs view
@@ -0,0 +1,511 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++module Raft+ (+ -- * State machine type class+ RSMP(..)+ , RSM(..)++ -- * Networking type classes+ , RaftSendRPC(..)+ , RaftRecvRPC(..)+ , RaftSendClient(..)+ , RaftRecvClient(..)+ , RaftPersist(..)++ , EventChan++ , RaftEnv(..)+ , runRaftNode+ , runRaftT++ , handleEventLoop++ -- * Client data types+ , ClientRequest(..)+ , ClientReq(..)+ , ClientResponse(..)+ , ClientReadResp(..)+ , ClientWriteResp(..)+ , ClientRedirResp(..)++ -- * Configuration+ , NodeConfig(..)++ -- * Events+ , Event(..)+ , Timeout(..)+ , MessageEvent(..)++ -- * Log+ , Entry(..)+ , Entries+ , RaftWriteLog(..)+ , DeleteSuccess(..)+ , RaftDeleteLog(..)+ , RaftReadLog (..)+ , RaftLog+ , RaftLogError(..)+ , RaftLogExceptions(..)++ -- * Logging+ , LogDest(..)+ , Severity(..)++ -- * Raft node states+ , Mode(..)+ , RaftNodeState(..)+ , NodeState(..)+ , CurrentLeader(..)+ , FollowerState(..)+ , CandidateState(..)+ , LeaderState(..)+ , initRaftNodeState+ , isFollower+ , isCandidate+ , isLeader+ , setLastLogEntryData+ , getLastLogEntryData+ , getLastAppliedAndCommitIndex++ -- * Persistent state+ , PersistentState(..)+ , initPersistentState++ -- * Basic types+ , NodeId+ , NodeIds+ , ClientId(..)+ , LeaderId(..)+ , Term(..)+ , Index(..)+ , term0+ , index0++ -- * RPC+ , RPC(..)+ , RPCType(..)+ , RPCMessage(..)+ , AppendEntries(..)+ , AppendEntriesResponse(..)+ , RequestVote(..)+ , RequestVoteResponse(..)+ , AppendEntriesData(..)+ ) where++import Protolude hiding (STM, TChan, newTChan, readTChan, writeTChan, atomically)++import Control.Monad.Conc.Class+import Control.Concurrent.STM.Timer+import Control.Concurrent.Classy.STM.TChan+import Control.Concurrent.Classy.Async++import Control.Monad.Catch+import Control.Monad.Trans.Class++import qualified Data.Map as Map+import Data.Sequence (Seq(..), singleton)++import Raft.Action+import Raft.Client+import Raft.Config+import Raft.Event+import Raft.Handle+import Raft.Log+import Raft.Logging hiding (logInfo, logDebug, logCritical)+import Raft.Monad hiding (logInfo, logDebug)+import Raft.NodeState+import Raft.Persistent+import Raft.RPC+import Raft.Types+++type EventChan m v = TChan (STM m) (Event v)++-- | The raft server environment composed of the concurrent variables used in+-- the effectful raft layer.+data RaftEnv v m = RaftEnv+ { eventChan :: EventChan m v+ , resetElectionTimer :: m ()+ , resetHeartbeatTimer :: m ()+ , raftNodeConfig :: NodeConfig+ , raftNodeLogDest :: LogDest+ }++newtype RaftT v m a = RaftT+ { unRaftT :: ReaderT (RaftEnv v m) (StateT RaftNodeState m) a+ } deriving (Functor, Applicative, Monad, MonadReader (RaftEnv v m), MonadState RaftNodeState, Alternative, MonadPlus)++instance MonadTrans (RaftT v) where+ lift = RaftT . lift . lift++deriving instance MonadIO m => MonadIO (RaftT v m)+deriving instance MonadThrow m => MonadThrow (RaftT v m)+deriving instance MonadCatch m => MonadCatch (RaftT v m)+deriving instance MonadMask m => MonadMask (RaftT v m)+deriving instance MonadConc m => MonadConc (RaftT v m)++instance Monad m => RaftLogger (RaftT v m) where+ loggerNodeId = asks (configNodeId . raftNodeConfig)+ loggerNodeState = get++runRaftT+ :: MonadConc m+ => RaftNodeState+ -> RaftEnv v m+ -> RaftT v m ()+ -> m ()+runRaftT raftNodeState raftEnv =+ flip evalStateT raftNodeState . flip runReaderT raftEnv . unRaftT++------------------------------------------------------------------------------++logDebug :: MonadIO m => Text -> RaftT v m ()+logDebug msg = flip logDebugIO msg =<< asks raftNodeLogDest++logCritical :: MonadIO m => Text -> RaftT v m ()+logCritical msg = flip logCriticalIO msg =<< asks raftNodeLogDest++------------------------------------------------------------------------------++-- | Run timers, RPC and client request handlers and start event loop.+-- It should run forever+runRaftNode+ :: ( Show v, Show sm, Show (Action sm v)+ , MonadIO m, MonadConc m+ , RSM sm v m+ , Show (RSMPError sm v)+ , RaftSendRPC m v+ , RaftRecvRPC m v+ , RaftSendClient m sm+ , RaftRecvClient m v+ , RaftLog m v+ , RaftLogExceptions m+ , RaftPersist m+ , Exception (RaftPersistError m)+ )+ => NodeConfig -- ^ Node configuration+ -> LogDest -- ^ Logs destination+ -> Int -- ^ Timer seed+ -> sm -- ^ Initial state machine state+ -> m ()+runRaftNode nodeConfig@NodeConfig{..} logDest timerSeed initRSM = do+ eventChan <- atomically newTChan++ electionTimer <- newTimerRange timerSeed configElectionTimeout+ heartbeatTimer <- newTimer configHeartbeatTimeout++ let resetElectionTimer = resetTimer electionTimer+ resetHeartbeatTimer = resetTimer heartbeatTimer+ raftEnv = RaftEnv eventChan resetElectionTimer resetHeartbeatTimer nodeConfig logDest++ runRaftT initRaftNodeState raftEnv $ do++ -- Fork all event producers to run concurrently+ lift $ fork (electionTimeoutTimer electionTimer eventChan)+ lift $ fork (heartbeatTimeoutTimer heartbeatTimer eventChan)+ fork (rpcHandler eventChan)+ fork (clientReqHandler eventChan)++ -- Start the main event handling loop+ handleEventLoop initRSM++handleEventLoop+ :: forall sm v m.+ ( Show v, Show sm, Show (Action sm v)+ , MonadIO m, MonadConc m+ , RSM sm v m+ , Show (RSMPError sm v)+ , RaftPersist m+ , RaftSendRPC m v+ , RaftSendClient m sm+ , RaftLog m v+ , RaftLogExceptions m+ , RaftPersist m+ , Exception (RaftPersistError m)+ )+ => sm+ -> RaftT v m ()+handleEventLoop initRSM = do+ ePersistentState <- lift readPersistentState+ case ePersistentState of+ Left err -> throw err+ Right pstate -> handleEventLoop' initRSM pstate+ where+ handleEventLoop' :: sm -> PersistentState -> RaftT v m ()+ handleEventLoop' stateMachine persistentState = do+ event <- atomically . readTChan =<< asks eventChan+ loadLogEntryTermAtAePrevLogIndex event+ raftNodeState <- get+ logDebug $ "[Event]: " <> show event+ logDebug $ "[NodeState]: " <> show raftNodeState+ Right log :: Either (RaftReadLogError m) (Entries v) <- lift $ readLogEntriesFrom index0+ logDebug $ "[Log]: " <> show log+ logDebug $ "[State Machine]: " <> show stateMachine+ logDebug $ "[Persistent State]: " <> show persistentState+ -- Perform core state machine transition, handling the current event+ nodeConfig <- asks raftNodeConfig+ let transitionEnv = TransitionEnv nodeConfig stateMachine raftNodeState+ (resRaftNodeState, resPersistentState, actions, logMsgs) =+ Raft.Handle.handleEvent raftNodeState transitionEnv persistentState event+ -- Write persistent state to disk+ eRes <- lift $ writePersistentState resPersistentState+ case eRes of+ Left err -> throw err+ Right _ -> pure ()+ -- Update raft node state with the resulting node state+ put resRaftNodeState+ -- Handle logs producek by core state machine+ handleLogs logMsgs+ -- Handle actions produced by core state machine+ handleActions nodeConfig actions+ -- Apply new log entries to the state machine+ resRSM <- applyLogEntries stateMachine+ handleEventLoop' resRSM resPersistentState++ -- In the case that a node is a follower receiving an AppendEntriesRPC+ -- Event, read the log at the aePrevLogIndex+ loadLogEntryTermAtAePrevLogIndex :: Event v -> RaftT v m ()+ loadLogEntryTermAtAePrevLogIndex event =+ case event of+ MessageEvent (RPCMessageEvent (RPCMessage _ (AppendEntriesRPC ae))) -> do+ RaftNodeState rns <- get+ case rns of+ NodeFollowerState fs -> do+ eEntry <- lift $ readLogEntry (aePrevLogIndex ae)+ case eEntry of+ Left err -> throw err+ Right (mEntry :: Maybe (Entry v)) -> put $+ RaftNodeState $ NodeFollowerState fs+ { fsTermAtAEPrevIndex = entryTerm <$> mEntry }+ _ -> pure ()+ _ -> pure ()++handleActions+ :: ( Show v, Show sm, Show (Action sm v)+ , MonadIO m, MonadConc m+ , RSM sm v m+ , RaftSendRPC m v+ , RaftSendClient m sm+ , RaftLog m v+ , RaftLogExceptions m+ )+ => NodeConfig+ -> [Action sm v]+ -> RaftT v m ()+handleActions = mapM_ . handleAction++handleAction+ :: forall sm v m.+ ( Show v, Show sm, Show (Action sm v)+ , MonadIO m, MonadConc m+ , RSM sm v m+ , RaftSendRPC m v+ , RaftSendClient m sm+ , RaftLog m v+ , RaftLogExceptions m+ )+ => NodeConfig+ -> Action sm v+ -> RaftT v m ()+handleAction nodeConfig action = do+ logDebug $ "[Action]: " <> show action+ case action of+ SendRPC nid sendRpcAction -> do+ rpcMsg <- mkRPCfromSendRPCAction sendRpcAction+ lift (sendRPC nid rpcMsg)+ SendRPCs rpcMap ->+ forConcurrently_ (Map.toList rpcMap) $ \(nid, sendRpcAction) -> do+ rpcMsg <- mkRPCfromSendRPCAction sendRpcAction+ lift (sendRPC nid rpcMsg)+ BroadcastRPC nids sendRpcAction -> do+ rpcMsg <- mkRPCfromSendRPCAction sendRpcAction+ mapConcurrently_ (lift . flip sendRPC rpcMsg) nids+ RespondToClient cid cr -> lift $ sendClient cid cr+ ResetTimeoutTimer tout ->+ case tout of+ ElectionTimeout -> lift . resetElectionTimer =<< ask+ HeartbeatTimeout -> lift . resetHeartbeatTimer =<< ask+ AppendLogEntries entries -> do+ lift (updateLog entries)+ -- Update the last log entry data+ modify $ \(RaftNodeState ns) ->+ RaftNodeState (setLastLogEntryData ns entries)++ where+ mkRPCfromSendRPCAction :: SendRPCAction v -> RaftT v m (RPCMessage v)+ mkRPCfromSendRPCAction sendRPCAction = do+ RaftNodeState ns <- get+ RPCMessage (configNodeId nodeConfig) <$>+ case sendRPCAction of+ SendAppendEntriesRPC aeData -> do+ (entries, prevLogIndex, prevLogTerm, aeReadReq) <-+ case aedEntriesSpec aeData of+ FromIndex idx -> do+ eLogEntries <- lift (readLogEntriesFrom (decrIndexWithDefault0 idx))+ case eLogEntries of+ Left err -> throw err+ Right log ->+ case log of+ pe :<| entries@(e :<| _)+ | idx == 1 -> pure (log, index0, term0, Nothing)+ | otherwise -> pure (entries, entryIndex pe, entryTerm pe, Nothing)+ _ -> pure (log, index0, term0, Nothing)+ FromClientWriteReq e -> prevEntryData e+ FromNewLeader e -> prevEntryData e+ NoEntries spec -> do+ let readReq =+ case spec of+ FromClientReadReq n -> Just n+ _ -> Nothing+ (lastLogIndex, lastLogTerm) = getLastLogEntryData ns+ pure (Empty, lastLogIndex, lastLogTerm, readReq)+ let leaderId = LeaderId (configNodeId nodeConfig)+ pure . toRPC $+ AppendEntries+ { aeTerm = aedTerm aeData+ , aeLeaderId = leaderId+ , aePrevLogIndex = prevLogIndex+ , aePrevLogTerm = prevLogTerm+ , aeEntries = entries+ , aeLeaderCommit = aedLeaderCommit aeData+ , aeReadRequest = aeReadReq+ }+ SendAppendEntriesResponseRPC aer -> pure (toRPC aer)+ SendRequestVoteRPC rv -> pure (toRPC rv)+ SendRequestVoteResponseRPC rvr -> pure (toRPC rvr)++ prevEntryData e = do+ (x,y,z) <- prevEntryData' e+ pure (x,y,z,Nothing)++ prevEntryData' e+ | entryIndex e == Index 1 = pure (singleton e, index0, term0)+ | otherwise = do+ let prevLogEntryIdx = decrIndexWithDefault0 (entryIndex e)+ eLogEntry <- lift $ readLogEntry prevLogEntryIdx+ case eLogEntry of+ Left err -> throw err+ Right Nothing -> pure (singleton e, index0, term0)+ Right (Just (prevEntry :: Entry v)) ->+ pure (singleton e, entryIndex prevEntry, entryTerm prevEntry)++-- If commitIndex > lastApplied: increment lastApplied, apply+-- log[lastApplied] to state machine (Section 5.3) until the state machine+-- is up to date with all the committed log entries+applyLogEntries+ :: forall sm m v.+ ( Show sm+ , MonadConc m+ , RaftReadLog m v+ , Exception (RaftReadLogError m)+ , RSM sm v m+ , Show (RSMPError sm v)+ )+ => sm+ -> RaftT v m sm+applyLogEntries stateMachine = do+ raftNodeState@(RaftNodeState nodeState) <- get+ if commitIndex nodeState > lastApplied nodeState+ then do+ let resNodeState = incrLastApplied nodeState+ put $ RaftNodeState resNodeState+ let newLastAppliedIndex = lastApplied resNodeState+ eLogEntry <- lift $ readLogEntry newLastAppliedIndex+ case eLogEntry of+ Left err -> throw err+ Right Nothing -> panic "No log entry at 'newLastAppliedIndex'"+ Right (Just logEntry) -> do+ -- The command should be verified by the leader, thus all node+ -- attempting to apply the committed log entry should not fail when+ -- doing so; failure here means something has gone very wrong.+ eRes <- lift (applyEntryRSM stateMachine logEntry)+ case eRes of+ Left err -> panic $ "Failed to apply committed log entry: " <> show err+ Right nsm -> applyLogEntries nsm+ else pure stateMachine+ where+ incrLastApplied :: NodeState ns -> NodeState ns+ incrLastApplied nodeState =+ case nodeState of+ NodeFollowerState fs ->+ let lastApplied' = incrIndex (fsLastApplied fs)+ in NodeFollowerState $ fs { fsLastApplied = lastApplied' }+ NodeCandidateState cs ->+ let lastApplied' = incrIndex (csLastApplied cs)+ in NodeCandidateState $ cs { csLastApplied = lastApplied' }+ NodeLeaderState ls ->+ let lastApplied' = incrIndex (lsLastApplied ls)+ in NodeLeaderState $ ls { lsLastApplied = lastApplied' }++ lastApplied :: NodeState ns -> Index+ lastApplied = fst . getLastAppliedAndCommitIndex++ commitIndex :: NodeState ns -> Index+ commitIndex = snd . getLastAppliedAndCommitIndex++handleLogs+ :: (MonadIO m, MonadConc m)+ => [LogMsg]+ -> RaftT v m ()+handleLogs logs = do+ logDest <- asks raftNodeLogDest+ mapM_ (logToDest logDest) logs++------------------------------------------------------------------------------+-- Event Producers+------------------------------------------------------------------------------++-- | Producer for rpc message events+rpcHandler+ :: (MonadIO m, MonadConc m, Show v, RaftRecvRPC m v)+ => TChan (STM m) (Event v)+ -> RaftT v m ()+rpcHandler eventChan =+ forever $ do+ eRpcMsg <- lift $ Control.Monad.Catch.try receiveRPC+ case eRpcMsg of+ Left (err :: SomeException) -> logCritical (show err)+ Right (Left err) -> logCritical (show err)+ Right (Right rpcMsg) -> do+ let rpcMsgEvent = MessageEvent (RPCMessageEvent rpcMsg)+ atomically $ writeTChan eventChan rpcMsgEvent++-- | Producer for rpc message events+clientReqHandler+ :: (MonadIO m, MonadConc m, RaftRecvClient m v)+ => TChan (STM m) (Event v)+ -> RaftT v m ()+clientReqHandler eventChan =+ forever $ do+ eClientReq <- lift $ Control.Monad.Catch.try receiveClient+ case eClientReq of+ Left (err :: SomeException) -> logCritical (show err)+ Right (Left err) -> logCritical (show err)+ Right (Right clientReq) -> do+ let clientReqEvent = MessageEvent (ClientRequestEvent clientReq)+ atomically $ writeTChan eventChan clientReqEvent++-- | Producer for the election timeout event+electionTimeoutTimer :: MonadConc m => Timer m -> TChan (STM m) (Event v) -> m ()+electionTimeoutTimer timer eventChan =+ forever $ do+ startTimer timer >> waitTimer timer+ atomically $ writeTChan eventChan (TimeoutEvent ElectionTimeout)++-- | Producer for the heartbeat timeout event+heartbeatTimeoutTimer :: MonadConc m => Timer m -> TChan (STM m) (Event v) -> m ()+heartbeatTimeoutTimer timer eventChan =+ forever $ do+ startTimer timer >> waitTimer timer+ atomically $ writeTChan eventChan (TimeoutEvent HeartbeatTimeout)
+ src/Raft/Action.hs view
@@ -0,0 +1,28 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE StandaloneDeriving #-}++module Raft.Action where++import Protolude++import Raft.Client+import Raft.RPC+import Raft.Log+import Raft.Event+import Raft.Types++data Action sm v+ = SendRPC NodeId (SendRPCAction v) -- ^ Send a message to a specific node id+ | SendRPCs (Map NodeId (SendRPCAction v)) -- ^ Send a unique message to specific nodes in parallel+ | BroadcastRPC NodeIds (SendRPCAction v) -- ^ Broadcast the same message to all nodes+ | AppendLogEntries (Entries v) -- ^ Append entries to the replicated log+ | RespondToClient ClientId (ClientResponse sm) -- ^ Respond to client after a client request+ | ResetTimeoutTimer Timeout -- ^ Reset a timeout timer+ deriving (Show)++data SendRPCAction v+ = SendAppendEntriesRPC (AppendEntriesData v)+ | SendAppendEntriesResponseRPC AppendEntriesResponse+ | SendRequestVoteRPC RequestVote+ | SendRequestVoteResponseRPC RequestVoteResponse+ deriving (Show)
+ src/Raft/Candidate.hs view
@@ -0,0 +1,167 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE MonoLocalBinds #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE MultiWayIf #-}+{-# LANGUAGE TupleSections #-}++module Raft.Candidate (+ handleAppendEntries+ , handleAppendEntriesResponse+ , handleRequestVote+ , handleRequestVoteResponse+ , handleTimeout+ , handleClientRequest+) where++import Protolude++import qualified Data.Set as Set+import qualified Data.Sequence as Seq+import qualified Data.Map as Map++import Raft.NodeState+import Raft.RPC+import Raft.Client+import Raft.Event+import Raft.Action+import Raft.Persistent+import Raft.Log+import Raft.Config+import Raft.Monad+import Raft.Types++--------------------------------------------------------------------------------+-- Candidate+--------------------------------------------------------------------------------++handleAppendEntries :: RPCHandler 'Candidate sm (AppendEntries v) v+handleAppendEntries (NodeCandidateState candidateState@CandidateState{..}) sender AppendEntries {..} = do+ currentTerm <- gets currentTerm+ if currentTerm <= aeTerm+ then stepDown sender aeTerm csCommitIndex csLastApplied csLastLogEntryData+ else pure $ candidateResultState Noop candidateState++-- | Candidates should not respond to 'AppendEntriesResponse' messages.+handleAppendEntriesResponse :: RPCHandler 'Candidate sm AppendEntriesResponse v+handleAppendEntriesResponse (NodeCandidateState candidateState) _sender _appendEntriesResp =+ pure $ candidateResultState Noop candidateState++handleRequestVote :: RPCHandler 'Candidate sm RequestVote v+handleRequestVote ns@(NodeCandidateState candidateState@CandidateState{..}) sender requestVote@RequestVote{..} = do+ currentTerm <- gets currentTerm+ send sender $+ SendRequestVoteResponseRPC $+ RequestVoteResponse currentTerm False+ pure $ candidateResultState Noop candidateState++-- | Candidates should not respond to 'RequestVoteResponse' messages.+handleRequestVoteResponse+ :: forall sm v. Show v+ => RPCHandler 'Candidate sm RequestVoteResponse v+handleRequestVoteResponse (NodeCandidateState candidateState@CandidateState{..}) sender requestVoteResp@RequestVoteResponse{..} = do+ currentTerm <- gets currentTerm+ if | Set.member sender csVotes -> pure $ candidateResultState Noop candidateState+ | not rvrVoteGranted -> pure $ candidateResultState Noop candidateState+ | otherwise -> do+ let newCsVotes = Set.insert sender csVotes+ cNodeIds <- asks (configNodeIds . nodeConfig)+ if not $ hasMajority cNodeIds newCsVotes+ then do+ let newCandidateState = candidateState { csVotes = newCsVotes }+ pure $ candidateResultState Noop newCandidateState+ else leaderResultState BecomeLeader <$> becomeLeader++ where+ hasMajority :: Set a -> Set b -> Bool+ hasMajority nids votes =+ Set.size votes >= Set.size nids `div` 2 + 1++ mkNoopEntry :: TransitionM sm v (Entry v)+ mkNoopEntry = do+ let (lastLogEntryIdx, _) = csLastLogEntryData+ currTerm <- gets currentTerm+ nid <- asks (configNodeId . nodeConfig)+ pure Entry+ { entryIndex = succ lastLogEntryIdx+ , entryTerm = currTerm+ , entryValue = NoValue+ , entryIssuer = LeaderIssuer (LeaderId nid)+ }++ becomeLeader :: TransitionM sm v LeaderState+ becomeLeader = do+ currentTerm <- gets currentTerm+ resetHeartbeatTimeout+ -- In order for leaders to know which entries have been replicated or not,+ -- a "no op" log entry must be created at the start of the term. See+ -- "Client ineraction", Section 8, of https://raft.github.io/raft.pdf.+ noopEntry <- mkNoopEntry+ appendLogEntries (Seq.Empty Seq.|> noopEntry)+ broadcast $ SendAppendEntriesRPC+ AppendEntriesData+ { aedTerm = currentTerm+ , aedLeaderCommit = csCommitIndex+ , aedEntriesSpec = FromNewLeader noopEntry+ }+ cNodeIds <- asks (configNodeIds . nodeConfig)+ let lastLogEntryIdx = entryIndex noopEntry+ lastLogEntryTerm = entryTerm noopEntry+ pure LeaderState+ { lsCommitIndex = csCommitIndex+ , lsLastApplied = csLastApplied+ , lsNextIndex = Map.fromList $+ (,lastLogEntryIdx) <$> Set.toList cNodeIds+ , lsMatchIndex = Map.fromList $+ (,index0) <$> Set.toList cNodeIds+ , lsLastLogEntryData = (lastLogEntryIdx, lastLogEntryTerm, Nothing)+ , lsReadReqsHandled = 0+ , lsReadRequest = mempty+ }++handleTimeout :: TimeoutHandler 'Candidate sm v+handleTimeout (NodeCandidateState candidateState@CandidateState{..}) timeout =+ case timeout of+ HeartbeatTimeout -> pure $ candidateResultState Noop candidateState+ ElectionTimeout ->+ candidateResultState RestartElection <$>+ startElection csCommitIndex csLastApplied csLastLogEntryData++-- | When candidates handle a client request, they respond with NoLeader, as the+-- very reason they are candidate is because there is no leader. This is done+-- instead of simply not responding such that the client can know that the node+-- is live but that there is an election taking place.+handleClientRequest :: ClientReqHandler 'Candidate sm v+handleClientRequest (NodeCandidateState candidateState) (ClientRequest clientId _) = do+ redirectClientToLeader clientId NoLeader+ pure (candidateResultState Noop candidateState)++--------------------------------------------------------------------------------++stepDown+ :: NodeId+ -> Term+ -> Index+ -> Index+ -> (Index, Term)+ -> TransitionM a sm (ResultState 'Candidate)+stepDown sender term commitIndex lastApplied lastLogEntryData = do+ send sender $+ SendRequestVoteResponseRPC $+ RequestVoteResponse+ { rvrTerm = term+ , rvrVoteGranted = True+ }+ resetElectionTimeout+ pure $ ResultState DiscoverLeader $+ NodeFollowerState FollowerState+ { fsCurrentLeader = CurrentLeader (LeaderId sender)+ , fsCommitIndex = commitIndex+ , fsLastApplied = lastApplied+ , fsLastLogEntryData = lastLogEntryData+ , fsTermAtAEPrevIndex = Nothing+ }
+ src/Raft/Client.hs view
@@ -0,0 +1,72 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}++module Raft.Client where++import Protolude++import qualified Data.Serialize as S++import Raft.NodeState+import Raft.Types++-- | Interface for Raft nodes to send messages to clients+class RaftSendClient m sm where+ sendClient :: ClientId -> ClientResponse sm -> m ()++-- | Interface for Raft nodes to receive messages from clients+class Show (RaftRecvClientError m v) => RaftRecvClient m v where+ type RaftRecvClientError m v+ receiveClient :: m (Either (RaftRecvClientError m v) (ClientRequest v))++-- | Representation of a client request coupled with the client id+data ClientRequest v+ = ClientRequest ClientId (ClientReq v)+ deriving (Show, Generic)++instance S.Serialize v => S.Serialize (ClientRequest v)++-- | Representation of a client request+data ClientReq v+ = ClientReadReq -- ^ Request the latest state of the state machine+ | ClientWriteReq v -- ^ Write a command+ deriving (Show, Generic)++instance S.Serialize v => S.Serialize (ClientReq v)++-- | Representation of a client response+data ClientResponse s+ = ClientReadResponse (ClientReadResp s)+ -- ^ Respond with the latest state of the state machine.+ | ClientWriteResponse ClientWriteResp+ -- ^ Respond with the index of the entry appended to the log+ | ClientRedirectResponse ClientRedirResp+ -- ^ Respond with the node id of the current leader+ deriving (Show, Generic)++instance S.Serialize s => S.Serialize (ClientResponse s)++-- | Representation of a read response to a client+-- The `s` stands for the "current" state of the state machine+newtype ClientReadResp s+ = ClientReadResp s+ deriving (Show, Generic)++instance S.Serialize s => S.Serialize (ClientReadResp s)++-- | Representation of a write response to a client+data ClientWriteResp+ = ClientWriteResp Index+ -- ^ Index of the entry appended to the log due to the previous client request+ deriving (Show, Generic)++instance S.Serialize ClientWriteResp++-- | Representation of a redirect response to a client+data ClientRedirResp+ = ClientRedirResp CurrentLeader+ deriving (Show, Generic)++instance S.Serialize ClientRedirResp
+ src/Raft/Config.hs view
@@ -0,0 +1,18 @@+++module Raft.Config where++import Protolude++import Numeric.Natural (Natural)++import Raft.Types++-- | Configuration of a node in the cluster+data NodeConfig = NodeConfig+ { configNodeId :: NodeId -- ^ Node id of the running node+ , configNodeIds :: NodeIds -- ^ Set of all other node ids in the cluster+ , configElectionTimeout :: (Natural, Natural) -- ^ Range of times an election timeout can take+ , configHeartbeatTimeout :: Natural -- ^ Heartbeat timeout timer+ } deriving (Show)+
+ src/Raft/Event.hs view
@@ -0,0 +1,35 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE StandaloneDeriving #-}++module Raft.Event where++import Protolude++import qualified Data.Serialize as S++import Raft.Client+import Raft.RPC++-- | Representation of events a raft node can send and receive+data Event v+ = MessageEvent (MessageEvent v)+ | TimeoutEvent Timeout+ deriving (Show)++-- | Representation of timeouts+data Timeout+ = ElectionTimeout+ -- ^ Timeout after which a follower will become candidate+ | HeartbeatTimeout+ -- ^ Timeout after which a leader will send AppendEntries RPC to all peers+ deriving (Show)++-- | Representation of message events to a node+data MessageEvent v+ = RPCMessageEvent (RPCMessage v) -- ^ Incoming event from a peer+ | ClientRequestEvent (ClientRequest v) -- ^ Incoming event from a client+ deriving (Show, Generic)++instance S.Serialize v => S.Serialize (MessageEvent v)++
+ src/Raft/Follower.hs view
@@ -0,0 +1,161 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE MonoLocalBinds #-}++module Raft.Follower (+ handleAppendEntries+ , handleAppendEntriesResponse+ , handleRequestVote+ , handleRequestVoteResponse+ , handleTimeout+ , handleClientRequest+) where++import Protolude++import Data.Sequence (Seq(..))++import Raft.Action+import Raft.NodeState+import Raft.RPC+import Raft.Client+import Raft.Event+import Raft.Persistent+import Raft.Log (entryIndex)+import Raft.Monad+import Raft.Types++--------------------------------------------------------------------------------+-- Follower+--------------------------------------------------------------------------------++-- | Handle AppendEntries RPC message from Leader+-- Sections 5.2 and 5.3 of Raft Paper & Figure 2: Receiver Implementation+--+-- Note: see 'PersistentState' datatype for discussion about not keeping the+-- entire log in memory.+handleAppendEntries :: forall v sm. Show v => RPCHandler 'Follower sm (AppendEntries v) v+handleAppendEntries ns@(NodeFollowerState fs) sender AppendEntries{..} = do+ PersistentState{..} <- get+ (success, newFollowerState) <-+ if aeTerm < currentTerm+ -- 1. Reply false if term < currentTerm+ then pure (False, fs)+ else+ case fsTermAtAEPrevIndex fs of+ Nothing+ | aePrevLogIndex == index0 -> do+ appendLogEntries aeEntries+ pure (True, updateFollowerState fs)+ | otherwise -> pure (False, fs)+ Just entryAtAePrevLogIndexTerm ->+ -- 2. Reply false if log doesn't contain an entry at+ -- prevLogIndex whose term matches prevLogTerm.+ if entryAtAePrevLogIndexTerm /= aePrevLogTerm+ then pure (False, fs)+ else do+ -- 3. If an existing entry conflicts with a new one (same index+ -- but different terms), delete the existing entry and all that+ -- follow it.+ -- &+ -- 4. Append any new entries not already in the log+ -- (emits an action that accomplishes 3 & 4+ appendLogEntries aeEntries+ -- 5. If leaderCommit > commitIndex, set commitIndex =+ -- min(leaderCommit, index of last new entry)+ pure (True, updateFollowerState fs)+ send (unLeaderId aeLeaderId) $+ SendAppendEntriesResponseRPC $+ AppendEntriesResponse+ { aerTerm = currentTerm+ , aerSuccess = success+ , aerReadRequest = aeReadRequest+ }+ resetElectionTimeout+ pure (followerResultState Noop newFollowerState)+ where+ updateFollowerState :: FollowerState -> FollowerState+ updateFollowerState fs =+ if aeLeaderCommit > fsCommitIndex fs+ then updateLeader (updateCommitIndex fs)+ else updateLeader fs++ updateCommitIndex :: FollowerState -> FollowerState+ updateCommitIndex followerState =+ case aeEntries of+ Empty ->+ -- TODO move to action, as this should update to _true_ last log entry+ -- that follower has written to disk, not supposed commit index+ -- assuming leader has replicated all logs to this follower+ followerState { fsCommitIndex = aeLeaderCommit }+ _ :|> e ->+ let newCommitIndex = min aeLeaderCommit (entryIndex e)+ in followerState { fsCommitIndex = newCommitIndex }++ updateLeader :: FollowerState -> FollowerState+ updateLeader followerState = followerState { fsCurrentLeader = CurrentLeader (LeaderId sender) }++-- | Followers should not respond to 'AppendEntriesResponse' messages.+handleAppendEntriesResponse :: RPCHandler 'Follower sm AppendEntriesResponse v+handleAppendEntriesResponse (NodeFollowerState fs) _ _ =+ pure (followerResultState Noop fs)++handleRequestVote :: RPCHandler 'Follower sm RequestVote v+handleRequestVote ns@(NodeFollowerState fs) sender RequestVote{..} = do+ PersistentState{..} <- get+ let voteGranted = giveVote currentTerm votedFor+ logDebug $ "Vote granted: " <> show voteGranted+ send sender $+ SendRequestVoteResponseRPC $+ RequestVoteResponse+ { rvrTerm = currentTerm+ , rvrVoteGranted = voteGranted+ }+ when voteGranted $+ modify $ \pstate ->+ pstate { votedFor = Just sender }+ pure $ followerResultState Noop fs+ where+ giveVote term mVotedFor =+ and [ term <= rvTerm+ , validCandidateId mVotedFor+ , validCandidateLog+ ]++ validCandidateId Nothing = True+ validCandidateId (Just cid) = cid == rvCandidateId++ -- Check if the requesting candidate's log is more up to date+ -- Section 5.4.1 in Raft Paper+ validCandidateLog =+ let (lastLogEntryIdx, lastLogEntryTerm) = fsLastLogEntryData fs+ in (rvLastLogTerm > lastLogEntryTerm)+ || (rvLastLogTerm == lastLogEntryTerm && rvLastLogIndex >= lastLogEntryIdx)++-- | Followers should not respond to 'RequestVoteResponse' messages.+handleRequestVoteResponse :: RPCHandler 'Follower sm RequestVoteResponse v+handleRequestVoteResponse (NodeFollowerState fs) _ _ =+ pure (followerResultState Noop fs)++-- | Follower converts to Candidate if handling ElectionTimeout+handleTimeout :: TimeoutHandler 'Follower sm v+handleTimeout ns@(NodeFollowerState fs) timeout =+ case timeout of+ ElectionTimeout -> do+ logDebug "Follower times out. Starts election. Becomes candidate"+ candidateResultState StartElection <$>+ startElection (fsCommitIndex fs) (fsLastApplied fs) (fsLastLogEntryData fs)+ -- Follower should ignore heartbeat timeout events+ HeartbeatTimeout -> pure (followerResultState Noop fs)++-- | When a client handles a client request, it redirects the client to the+-- current leader by responding with the current leader id, if it knows of one.+handleClientRequest :: ClientReqHandler 'Follower sm v+handleClientRequest (NodeFollowerState fs) (ClientRequest clientId _)= do+ redirectClientToLeader clientId (fsCurrentLeader fs)+ pure (followerResultState Noop fs)
+ src/Raft/Handle.hs view
@@ -0,0 +1,168 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE GADTs #-}++module Raft.Handle where++import Protolude++import qualified Raft.Follower as Follower+import qualified Raft.Candidate as Candidate+import qualified Raft.Leader as Leader++import Raft.Action+import Raft.Event+import Raft.Monad+import Raft.NodeState+import Raft.Persistent+import Raft.RPC+import Raft.Logging (LogMsg)++-- | Main entry point for handling events+handleEvent+ :: forall sm v.+ (RSMP sm v, Show v)+ => RaftNodeState+ -> TransitionEnv sm+ -> PersistentState+ -> Event v+ -> (RaftNodeState, PersistentState, [Action sm v], [LogMsg])+handleEvent raftNodeState@(RaftNodeState initNodeState) transitionEnv persistentState event =+ -- Rules for all servers:+ case handleNewerRPCTerm of+ ((RaftNodeState resNodeState, logMsgs), persistentState', outputs) ->+ case handleEvent' resNodeState transitionEnv persistentState' event of+ ((ResultState _ resultState, logMsgs'), persistentState'', outputs') ->+ (RaftNodeState resultState, persistentState'', outputs <> outputs', logMsgs <> logMsgs')+ where+ handleNewerRPCTerm :: ((RaftNodeState, [LogMsg]), PersistentState, [Action sm v])+ handleNewerRPCTerm =+ case event of+ MessageEvent (RPCMessageEvent (RPCMessage _ rpc)) ->+ runTransitionM transitionEnv persistentState $ do+ -- If RPC request or response contains term T > currentTerm: set+ -- currentTerm = T, convert to follower+ currentTerm <- gets currentTerm+ if currentTerm < rpcTerm rpc+ then+ case convertToFollower initNodeState of+ ResultState _ nodeState -> do+ modify $ \pstate ->+ pstate { currentTerm = rpcTerm rpc+ , votedFor = Nothing+ }+ resetElectionTimeout+ pure (RaftNodeState nodeState)+ else pure raftNodeState+ _ -> ((raftNodeState, []), persistentState, mempty)++ convertToFollower :: forall s. NodeState s -> ResultState s+ convertToFollower nodeState =+ case nodeState of+ NodeFollowerState _ ->+ ResultState HigherTermFoundFollower nodeState+ NodeCandidateState cs ->+ ResultState HigherTermFoundCandidate $+ NodeFollowerState FollowerState+ { fsCurrentLeader = NoLeader+ , fsCommitIndex = csCommitIndex cs+ , fsLastApplied = csLastApplied cs+ , fsLastLogEntryData = csLastLogEntryData cs+ , fsTermAtAEPrevIndex = Nothing+ }+ NodeLeaderState ls ->+ ResultState HigherTermFoundLeader $+ NodeFollowerState FollowerState+ { fsCurrentLeader = NoLeader+ , fsCommitIndex = lsCommitIndex ls+ , fsLastApplied = lsLastApplied ls+ , fsLastLogEntryData =+ let (lastLogEntryIdx, lastLogEntryTerm, _) = lsLastLogEntryData ls+ in (lastLogEntryIdx, lastLogEntryTerm)+ , fsTermAtAEPrevIndex = Nothing+ }+++data RaftHandler ns sm v = RaftHandler+ { handleAppendEntries :: RPCHandler ns sm (AppendEntries v) v+ , handleAppendEntriesResponse :: RPCHandler ns sm AppendEntriesResponse v+ , handleRequestVote :: RPCHandler ns sm RequestVote v+ , handleRequestVoteResponse :: RPCHandler ns sm RequestVoteResponse v+ , handleTimeout :: TimeoutHandler ns sm v+ , handleClientRequest :: ClientReqHandler ns sm v+ }++followerRaftHandler :: Show v => RaftHandler 'Follower sm v+followerRaftHandler = RaftHandler+ { handleAppendEntries = Follower.handleAppendEntries+ , handleAppendEntriesResponse = Follower.handleAppendEntriesResponse+ , handleRequestVote = Follower.handleRequestVote+ , handleRequestVoteResponse = Follower.handleRequestVoteResponse+ , handleTimeout = Follower.handleTimeout+ , handleClientRequest = Follower.handleClientRequest+ }++candidateRaftHandler :: Show v => RaftHandler 'Candidate sm v+candidateRaftHandler = RaftHandler+ { handleAppendEntries = Candidate.handleAppendEntries+ , handleAppendEntriesResponse = Candidate.handleAppendEntriesResponse+ , handleRequestVote = Candidate.handleRequestVote+ , handleRequestVoteResponse = Candidate.handleRequestVoteResponse+ , handleTimeout = Candidate.handleTimeout+ , handleClientRequest = Candidate.handleClientRequest+ }++leaderRaftHandler :: Show v => RaftHandler 'Leader sm v+leaderRaftHandler = RaftHandler+ { handleAppendEntries = Leader.handleAppendEntries+ , handleAppendEntriesResponse = Leader.handleAppendEntriesResponse+ , handleRequestVote = Leader.handleRequestVote+ , handleRequestVoteResponse = Leader.handleRequestVoteResponse+ , handleTimeout = Leader.handleTimeout+ , handleClientRequest = Leader.handleClientRequest+ }++mkRaftHandler :: forall ns sm v. Show v => NodeState ns -> RaftHandler ns sm v+mkRaftHandler nodeState =+ case nodeState of+ NodeFollowerState _ -> followerRaftHandler+ NodeCandidateState _ -> candidateRaftHandler+ NodeLeaderState _ -> leaderRaftHandler++handleEvent'+ :: forall ns sm v.+ (RSMP sm v, Show v)+ => NodeState ns+ -> TransitionEnv sm+ -> PersistentState+ -> Event v+ -> ((ResultState ns, [LogMsg]), PersistentState, [Action sm v])+handleEvent' initNodeState transitionEnv persistentState event =+ runTransitionM transitionEnv persistentState $ do+ case event of+ MessageEvent mev ->+ case mev of+ RPCMessageEvent rpcMsg -> handleRPCMessage rpcMsg+ ClientRequestEvent cr -> do+ handleClientRequest initNodeState cr+ TimeoutEvent tout -> do+ handleTimeout initNodeState tout+ where+ RaftHandler{..} = mkRaftHandler initNodeState++ handleRPCMessage :: RPCMessage v -> TransitionM sm v (ResultState ns)+ handleRPCMessage (RPCMessage sender rpc) =+ case rpc of+ AppendEntriesRPC appendEntries ->+ handleAppendEntries initNodeState sender appendEntries+ AppendEntriesResponseRPC appendEntriesResp ->+ handleAppendEntriesResponse initNodeState sender appendEntriesResp+ RequestVoteRPC requestVote ->+ handleRequestVote initNodeState sender requestVote+ RequestVoteResponseRPC requestVoteResp ->+ handleRequestVoteResponse initNodeState sender requestVoteResp
+ src/Raft/Leader.hs view
@@ -0,0 +1,193 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE MonoLocalBinds #-}+{-# LANGUAGE RankNTypes #-}++module Raft.Leader (+ handleAppendEntries+ , handleAppendEntriesResponse+ , handleRequestVote+ , handleRequestVoteResponse+ , handleTimeout+ , handleClientRequest+) where++import Protolude++import qualified Data.Map as Map+import Data.Sequence (Seq(Empty))+import qualified Data.Set as Set+import qualified Data.Sequence as Seq++import Raft.Config (configNodeIds)+import Raft.NodeState+import Raft.RPC+import Raft.Action+import Raft.Client+import Raft.Event+import Raft.Persistent+import Raft.Log (Entry(..), EntryIssuer(..), EntryValue(..))+import Raft.Monad+import Raft.Types++--------------------------------------------------------------------------------+-- Leader+--------------------------------------------------------------------------------++-- | Leaders should not respond to 'AppendEntries' messages.+handleAppendEntries :: RPCHandler 'Leader sm (AppendEntries v) v+handleAppendEntries (NodeLeaderState ls)_ _ =+ pure (leaderResultState Noop ls)++handleAppendEntriesResponse+ :: forall sm v. Show v+ => RPCHandler 'Leader sm AppendEntriesResponse v+handleAppendEntriesResponse ns@(NodeLeaderState ls) sender appendEntriesResp+ -- If AppendEntries fails (aerSuccess == False) because of log inconsistency,+ -- decrement nextIndex and retry+ | not (aerSuccess appendEntriesResp) = do+ let newNextIndices = Map.adjust decrIndexWithDefault0 sender (lsNextIndex ls)+ newLeaderState = ls { lsNextIndex = newNextIndices }+ Just newNextIndex = Map.lookup sender newNextIndices+ aeData <- mkAppendEntriesData newLeaderState (FromIndex newNextIndex)+ send sender (SendAppendEntriesRPC aeData)+ pure (leaderResultState Noop newLeaderState)+ | otherwise = do+ let (lastLogEntryIdx,_,_) = lsLastLogEntryData ls+ newNextIndices = Map.insert sender (lastLogEntryIdx + 1) (lsNextIndex ls)+ newMatchIndices = Map.insert sender lastLogEntryIdx (lsMatchIndex ls)+ newLeaderState = ls { lsNextIndex = newNextIndices, lsMatchIndex = newMatchIndices }+ -- Increment leader commit index if now a majority of followers have+ -- replicated an entry at a given term.+ newestLeaderState <- incrCommitIndex newLeaderState+ when (lsCommitIndex newestLeaderState > lsCommitIndex newLeaderState) $ do+ let (entryIdx, _, entryIssuer) = lsLastLogEntryData newestLeaderState+ case entryIssuer of+ Nothing -> panic "No last long entry issuer"+ Just (LeaderIssuer _) -> pure ()+ Just (ClientIssuer cid) -> tellActions [RespondToClient cid (ClientWriteResponse (ClientWriteResp entryIdx))]++ case aerReadRequest appendEntriesResp of+ Nothing -> pure (leaderResultState Noop newestLeaderState)+ Just n -> handleReadReq n newestLeaderState+ where+ handleReadReq :: Int -> LeaderState -> TransitionM sm v (ResultState 'Leader)+ handleReadReq n leaderState = do+ networkSize <- Set.size <$> asks (configNodeIds . nodeConfig)+ let initReadReqs = lsReadRequest leaderState+ (mclientId, newReadReqs) =+ case Map.lookup n initReadReqs of+ Nothing -> panic "This should not happen"+ Just (cid,m)+ | isMajority (succ m) networkSize -> (Just cid, Map.delete n initReadReqs)+ | otherwise -> (Nothing, Map.adjust (second succ) n initReadReqs)+ case mclientId of+ Nothing ->+ pure $ leaderResultState Noop leaderState+ { lsReadRequest = newReadReqs+ }+ Just cid -> do+ respondClientRead cid+ pure $ leaderResultState Noop leaderState+ { lsReadReqsHandled = succ (lsReadReqsHandled leaderState)+ , lsReadRequest = newReadReqs+ }++-- | Leaders should not respond to 'RequestVote' messages.+handleRequestVote :: RPCHandler 'Leader sm RequestVote v+handleRequestVote (NodeLeaderState ls) _ _ =+ pure (leaderResultState Noop ls)++-- | Leaders should not respond to 'RequestVoteResponse' messages.+handleRequestVoteResponse :: RPCHandler 'Leader sm RequestVoteResponse v+handleRequestVoteResponse (NodeLeaderState ls) _ _ =+ pure (leaderResultState Noop ls)++handleTimeout :: Show v => TimeoutHandler 'Leader sm v+handleTimeout (NodeLeaderState ls) timeout =+ case timeout of+ -- Leader does not handle election timeouts+ ElectionTimeout -> pure (leaderResultState Noop ls)+ -- On a heartbeat timeout, broadcast append entries RPC to all peers+ HeartbeatTimeout -> do+ aeData <- mkAppendEntriesData ls (NoEntries FromHeartbeat)+ broadcast (SendAppendEntriesRPC aeData)+ pure (leaderResultState SendHeartbeat ls)++-- | The leader handles all client requests, responding with the current state+-- machine on a client read, and appending an entry to the log on a valid client+-- write.+handleClientRequest :: Show v => ClientReqHandler 'Leader sm v+handleClientRequest (NodeLeaderState ls@LeaderState{..}) (ClientRequest cid cr) = do+ case cr of+ ClientReadReq -> do+ heartbeat <- mkAppendEntriesData ls (NoEntries (FromClientReadReq lsReadReqsHandled))+ broadcast (SendAppendEntriesRPC heartbeat)+ let newLeaderState =+ ls { lsReadRequest = Map.insert lsReadReqsHandled (cid, 0) lsReadRequest+ }+ pure (leaderResultState Noop newLeaderState)+ ClientWriteReq v -> do+ newLogEntry <- mkNewLogEntry v+ appendLogEntries (Empty Seq.|> newLogEntry)+ aeData <- mkAppendEntriesData ls (FromClientWriteReq newLogEntry)+ broadcast (SendAppendEntriesRPC aeData)+ pure (leaderResultState Noop ls)+ where+ mkNewLogEntry v = do+ currentTerm <- currentTerm <$> get+ let (lastLogEntryIdx,_,_) = lsLastLogEntryData+ pure $ Entry+ { entryIndex = succ lastLogEntryIdx+ , entryTerm = currentTerm+ , entryValue = EntryValue v+ , entryIssuer = ClientIssuer cid+ }++--------------------------------------------------------------------------------++-- | If there exists an N such that N > commitIndex, a majority of+-- matchIndex[i] >= N, and log[N].term == currentTerm: set commitIndex = N+incrCommitIndex :: Show v => LeaderState -> TransitionM sm v LeaderState+incrCommitIndex leaderState@LeaderState{..} = do+ logDebug "Checking if commit index should be incremented..."+ let (_, lastLogEntryTerm, _) = lsLastLogEntryData+ currentTerm <- currentTerm <$> get+ if majorityGreaterThanN && (lastLogEntryTerm == currentTerm)+ then do+ logDebug $ "Incrementing commit index to: " <> show n+ incrCommitIndex leaderState { lsCommitIndex = n }+ else do+ logDebug "Not incrementing commit index."+ pure leaderState+ where+ n = lsCommitIndex + 1++ -- Note, the majority in this case includes the leader itself, thus when+ -- calculating the number of nodes that need a match index > N, we only need+ -- to divide the size of the map by 2 instead of also adding one.+ majorityGreaterThanN =+ isMajority (Map.size (Map.filter (>= n) lsMatchIndex) + 1)+ (Map.size lsMatchIndex)++isMajority :: Int -> Int -> Bool+isMajority n m = n >= m `div` 2 + 1++-- | Construct an AppendEntriesRPC given log entries and a leader state.+mkAppendEntriesData+ :: Show v+ => LeaderState+ -> EntriesSpec v+ -> TransitionM sm v (AppendEntriesData v)+mkAppendEntriesData ls entriesSpec = do+ currentTerm <- gets currentTerm+ pure AppendEntriesData+ { aedTerm = currentTerm+ , aedLeaderCommit = lsCommitIndex ls+ , aedEntriesSpec = entriesSpec+ }
+ src/Raft/Log.hs view
@@ -0,0 +1,128 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE GADTs #-}++module Raft.Log where++import Protolude++import Data.Serialize+import Data.Sequence (Seq(..), (|>))++import Raft.Types++data EntryIssuer+ = ClientIssuer ClientId+ | LeaderIssuer LeaderId+ deriving (Show, Generic, Serialize)++data EntryValue v+ = EntryValue v+ | NoValue -- ^ Used as a first committed entry of a new term+ deriving (Show, Generic, Serialize)++-- | Representation of an entry in the replicated log+data Entry v = Entry+ { entryIndex :: Index+ -- ^ Index of entry in the log+ , entryTerm :: Term+ -- ^ Term when entry was received by leader+ , entryValue :: EntryValue v+ -- ^ Command to update state machine+ , entryIssuer :: EntryIssuer+ -- ^ Id of the client that issued the command+ } deriving (Show, Generic, Serialize)++type Entries v = Seq (Entry v)++-- | Provides an interface for nodes to write log entries to storage.+class Monad m => RaftWriteLog m v where+ type RaftWriteLogError m+ -- | Write the given log entries to storage+ writeLogEntries+ :: Exception (RaftWriteLogError m)+ => Entries v -> m (Either (RaftWriteLogError m) ())++data DeleteSuccess v = DeleteSuccess++-- | Provides an interface for nodes to delete log entries from storage.+class Monad m => RaftDeleteLog m v where+ type RaftDeleteLogError m+ -- | Delete log entries from a given index; e.g. 'deleteLogEntriesFrom 7'+ -- should delete every log entry with an index >= 7.+ deleteLogEntriesFrom+ :: Exception (RaftDeleteLogError m)+ => Index -> m (Either (RaftDeleteLogError m) (DeleteSuccess v))++-- | Provides an interface for nodes to read log entries from storage.+class Monad m => RaftReadLog m v where+ type RaftReadLogError m+ -- | Read the log at a given index+ readLogEntry+ :: Exception (RaftReadLogError m)+ => Index -> m (Either (RaftReadLogError m) (Maybe (Entry v)))+ -- | Read log entries from a specific index onwards, including the specific+ -- index+ readLogEntriesFrom+ :: Exception (RaftReadLogError m)+ => Index -> m (Either (RaftReadLogError m) (Entries v))+ -- | Read the last log entry in the log+ readLastLogEntry+ :: Exception (RaftReadLogError m)+ => m (Either (RaftReadLogError m) (Maybe (Entry v)))++ default readLogEntriesFrom+ :: Exception (RaftReadLogError m)+ => Index+ -> m (Either (RaftReadLogError m) (Entries v))+ readLogEntriesFrom idx = do+ eLastLogEntry <- readLastLogEntry+ case eLastLogEntry of+ Left err -> pure (Left err)+ Right Nothing -> pure (Right Empty)+ Right (Just lastLogEntry)+ | entryIndex lastLogEntry < idx -> pure (Right Empty)+ | otherwise -> fmap (|> lastLogEntry) <$> go (decrIndexWithDefault0 (entryIndex lastLogEntry))+ where+ go idx'+ | idx' < idx || idx' == 0 = pure (Right Empty)+ | otherwise = do+ eLogEntry <- readLogEntry idx'+ case eLogEntry of+ Left err -> pure (Left err)+ Right Nothing -> panic "Malformed log"+ Right (Just logEntry) -> fmap (|> logEntry) <$> go (decrIndexWithDefault0 idx')++type RaftLog m v = (RaftReadLog m v, RaftWriteLog m v, RaftDeleteLog m v)+type RaftLogExceptions m = (Exception (RaftReadLogError m), Exception (RaftWriteLogError m), Exception (RaftDeleteLogError m))++-- | Representation of possible errors that come from reading, writing or+-- deleting logs from the persistent storage+data RaftLogError m+ = RaftLogReadError (RaftReadLogError m)+ | RaftLogWriteError (RaftWriteLogError m)+ | RaftLogDeleteError (RaftDeleteLogError m)++updateLog+ :: forall m v.+ ( RaftDeleteLog m v, Exception (RaftDeleteLogError m)+ , RaftWriteLog m v, Exception (RaftWriteLogError m)+ )+ => Entries v+ -> m (Either (RaftLogError m) ())+updateLog entries =+ case entries of+ Empty -> pure (Right ())+ e :<| _ -> do+ eDel <- deleteLogEntriesFrom @m @v (entryIndex e)+ case eDel of+ Left err -> pure (Left (RaftLogDeleteError err))+ Right DeleteSuccess -> first RaftLogWriteError <$> writeLogEntries entries
+ src/Raft/Logging.hs view
@@ -0,0 +1,137 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE RecordWildCards #-}++module Raft.Logging where++import Protolude++import Control.Monad.Trans.Class (MonadTrans)+import Control.Monad.State (modify')++import Data.Time++import Raft.NodeState+import Raft.Types++-- | Representation of the logs' destination+data LogDest+ = LogFile FilePath+ | LogStdout+ | NoLogs++-- | Representation of the severity of the logs+data Severity+ = Info+ | Debug+ | Critical+ deriving (Show)++data LogMsg = LogMsg+ { mTime :: Maybe UTCTime+ , severity :: Severity+ , logMsgData :: LogMsgData+ }++data LogMsgData = LogMsgData+ { logMsgNodeId :: NodeId+ , logMsgNodeState :: Mode+ , logMsg :: Text+ } deriving (Show)++logMsgToText :: LogMsg -> Text+logMsgToText (LogMsg mt s d) =+ maybe "" timeToText mt <> "(" <> show s <> ")" <> " " <> logMsgDataToText d+ where+ timeToText :: UTCTime -> Text+ timeToText t = "[" <> toS (timeToText' t) <> "]"++ timeToText' = formatTime defaultTimeLocale (iso8601DateFormat (Just "%H:%M:%S"))++logMsgDataToText :: LogMsgData -> Text+logMsgDataToText LogMsgData{..} =+ "<" <> toS logMsgNodeId <> " | " <> show logMsgNodeState <> ">: " <> logMsg++class Monad m => RaftLogger m where+ loggerNodeId :: m NodeId+ loggerNodeState :: m RaftNodeState++mkLogMsgData :: RaftLogger m => Text -> m LogMsgData+mkLogMsgData msg = do+ nid <- loggerNodeId+ ns <- nodeMode <$> loggerNodeState+ pure $ LogMsgData nid ns msg++instance RaftLogger m => RaftLogger (RaftLoggerT m) where+ loggerNodeId = lift loggerNodeId+ loggerNodeState = lift loggerNodeState++--------------------------------------------------------------------------------+-- Logging with IO+--------------------------------------------------------------------------------++logToDest :: MonadIO m => LogDest -> LogMsg -> m ()+logToDest logDest logMsg =+ case logDest of+ LogStdout -> putText (logMsgToText logMsg)+ LogFile fp -> liftIO $ appendFile fp (logMsgToText logMsg <> "\n")+ NoLogs -> pure ()++logToStdout :: MonadIO m => LogMsg -> m ()+logToStdout = logToDest LogStdout++logToFile :: MonadIO m => FilePath -> LogMsg -> m ()+logToFile fp = logToDest (LogFile fp)++logWithSeverityIO :: (RaftLogger m, MonadIO m) => Severity -> LogDest -> Text -> m ()+logWithSeverityIO s logDest msg = do+ logMsgData <- mkLogMsgData msg+ now <- liftIO getCurrentTime+ let logMsg = LogMsg (Just now) s logMsgData+ logToDest logDest logMsg++logInfoIO :: (RaftLogger m, MonadIO m) => LogDest -> Text -> m ()+logInfoIO = logWithSeverityIO Info++logDebugIO :: (RaftLogger m, MonadIO m) => LogDest -> Text -> m ()+logDebugIO = logWithSeverityIO Debug++logCriticalIO :: (RaftLogger m, MonadIO m) => LogDest -> Text -> m ()+logCriticalIO = logWithSeverityIO Critical++--------------------------------------------------------------------------------+-- Pure Logging+--------------------------------------------------------------------------------++newtype RaftLoggerT m a = RaftLoggerT {+ unRaftLoggerT :: StateT [LogMsg] m a+ } deriving (Functor, Applicative, Monad, MonadState [LogMsg], MonadTrans)++runRaftLoggerT+ :: Monad m+ => RaftLoggerT m a -- ^ The computation from which to extract the logs+ -> m (a, [LogMsg])+runRaftLoggerT = flip runStateT [] . unRaftLoggerT++type RaftLoggerM = RaftLoggerT Identity++runRaftLoggerM+ :: RaftLoggerM a+ -> (a, [LogMsg])+runRaftLoggerM = runIdentity . runRaftLoggerT++logWithSeverity :: RaftLogger m => Severity -> Text -> RaftLoggerT m ()+logWithSeverity s txt = do+ !logMsgData <- mkLogMsgData txt+ let !logMsg = LogMsg Nothing s logMsgData+ modify' (++ [logMsg])++logInfo :: RaftLogger m => Text -> RaftLoggerT m ()+logInfo = logWithSeverity Info++logDebug :: RaftLogger m => Text -> RaftLoggerT m ()+logDebug = logWithSeverity Debug++logCritical :: RaftLogger m => Text -> RaftLoggerT m ()+logCritical = logWithSeverity Critical
+ src/Raft/Monad.hs view
@@ -0,0 +1,199 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeFamilyDependencies #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE GADTs #-}++module Raft.Monad where++import Protolude hiding (pass)+import Control.Monad.RWS+import qualified Data.Set as Set++import Raft.Action+import Raft.Client+import Raft.Config+import Raft.Event+import Raft.Log+import Raft.Persistent+import Raft.NodeState+import Raft.RPC+import Raft.Types+import Raft.Logging (RaftLogger, runRaftLoggerT, RaftLoggerT(..), LogMsg)+import qualified Raft.Logging as Logging++--------------------------------------------------------------------------------+-- State Machine+--------------------------------------------------------------------------------++-- | Interface to handle commands in the underlying state machine. Functional+--dependency permitting only a single state machine command to be defined to+--update the state machine.++class RSMP sm v | sm -> v where+ data RSMPError sm v+ type RSMPCtx sm v = ctx | ctx -> sm v+ applyCmdRSMP :: RSMPCtx sm v -> sm -> v -> Either (RSMPError sm v) sm++class (Monad m, RSMP sm v) => RSM sm v m | m sm -> v where+ validateCmd :: v -> m (Either (RSMPError sm v) ())+ askRSMPCtx :: m (RSMPCtx sm v)++applyEntryRSM :: RSM sm v m => sm -> Entry v -> m (Either (RSMPError sm v) sm)+applyEntryRSM sm e =+ case entryValue e of+ NoValue -> pure (Right sm)+ EntryValue v -> do+ res <- validateCmd v+ case res of+ Left err -> pure (Left err)+ Right () -> do+ ctx <- askRSMPCtx+ pure (applyCmdRSMP ctx sm v)++--------------------------------------------------------------------------------+-- Raft Monad+--------------------------------------------------------------------------------++tellAction :: Action sm v -> TransitionM sm v ()+tellAction a = tell [a]++tellActions :: [Action sm v] -> TransitionM sm v ()+tellActions as = tell as++data TransitionEnv sm = TransitionEnv+ { nodeConfig :: NodeConfig+ , stateMachine :: sm+ , nodeState :: RaftNodeState+ }++newtype TransitionM sm v a = TransitionM+ { unTransitionM :: RaftLoggerT (RWS (TransitionEnv sm) [Action sm v] PersistentState) a+ } deriving (Functor, Applicative, Monad)++instance MonadWriter [Action sm v] (TransitionM sm v) where+ tell = TransitionM . RaftLoggerT . tell+ listen = TransitionM . RaftLoggerT . listen . unRaftLoggerT . unTransitionM+ pass = TransitionM . RaftLoggerT . pass . unRaftLoggerT . unTransitionM++instance MonadReader (TransitionEnv sm) (TransitionM sm v) where+ ask = TransitionM . RaftLoggerT $ ask+ local f = TransitionM . RaftLoggerT . local f . unRaftLoggerT . unTransitionM++instance MonadState PersistentState (TransitionM sm v) where+ get = TransitionM . RaftLoggerT $ lift get+ put = TransitionM . RaftLoggerT . lift . put++instance RaftLogger (RWS (TransitionEnv sm) [Action sm v] PersistentState) where+ loggerNodeId = configNodeId <$> asks nodeConfig+ loggerNodeState = asks nodeState++runTransitionM+ :: TransitionEnv sm+ -> PersistentState+ -> TransitionM sm v a+ -> ((a, [LogMsg]), PersistentState, [Action sm v])+runTransitionM transEnv persistentState transitionM =+ runRWS (runRaftLoggerT (unTransitionM transitionM)) transEnv persistentState++askNodeId :: TransitionM sm v NodeId+askNodeId = asks (configNodeId . nodeConfig)++--------------------------------------------------------------------------------+-- Handlers+--------------------------------------------------------------------------------++type RPCHandler ns sm r v = RPCType r v => NodeState ns -> NodeId -> r -> TransitionM sm v (ResultState ns)+type TimeoutHandler ns sm v = NodeState ns -> Timeout -> TransitionM sm v (ResultState ns)+type ClientReqHandler ns sm v = NodeState ns -> ClientRequest v -> TransitionM sm v (ResultState ns)++--------------------------------------------------------------------------------+-- RWS Helpers+--------------------------------------------------------------------------------++broadcast :: SendRPCAction v -> TransitionM sm v ()+broadcast sendRPC = do+ selfNodeId <- askNodeId+ tellAction =<<+ flip BroadcastRPC sendRPC+ <$> asks (Set.filter (selfNodeId /=) . configNodeIds . nodeConfig)++send :: NodeId -> SendRPCAction v -> TransitionM sm v ()+send nodeId sendRPC = tellAction (SendRPC nodeId sendRPC)++-- | Resets the election timeout.+resetElectionTimeout :: TransitionM sm v ()+resetElectionTimeout = tellAction (ResetTimeoutTimer ElectionTimeout)++resetHeartbeatTimeout :: TransitionM sm v ()+resetHeartbeatTimeout = tellAction (ResetTimeoutTimer HeartbeatTimeout)++redirectClientToLeader :: ClientId -> CurrentLeader -> TransitionM sm v ()+redirectClientToLeader clientId currentLeader = do+ let clientRedirResp = ClientRedirectResponse (ClientRedirResp currentLeader)+ tellAction (RespondToClient clientId clientRedirResp)++respondClientRead :: ClientId -> TransitionM sm v ()+respondClientRead clientId = do+ clientReadResp <- ClientReadResponse . ClientReadResp <$> asks stateMachine+ tellAction (RespondToClient clientId clientReadResp)++appendLogEntries :: Show v => Seq (Entry v) -> TransitionM sm v ()+appendLogEntries = tellAction . AppendLogEntries++--------------------------------------------------------------------------------++startElection+ :: Index+ -> Index+ -> (Index, Term) -- ^ Last log entry data+ -> TransitionM sm v CandidateState+startElection commitIndex lastApplied lastLogEntryData = do+ incrementTerm+ voteForSelf+ resetElectionTimeout+ broadcast =<< requestVoteMessage+ selfNodeId <- askNodeId+ -- Return new candidate state+ pure CandidateState+ { csCommitIndex = commitIndex+ , csLastApplied = lastApplied+ , csVotes = Set.singleton selfNodeId+ , csLastLogEntryData = lastLogEntryData+ }+ where+ requestVoteMessage = do+ term <- currentTerm <$> get+ selfNodeId <- askNodeId+ pure $ SendRequestVoteRPC+ RequestVote+ { rvTerm = term+ , rvCandidateId = selfNodeId+ , rvLastLogIndex = fst lastLogEntryData+ , rvLastLogTerm = snd lastLogEntryData+ }++ incrementTerm = do+ psNextTerm <- incrTerm . currentTerm <$> get+ modify $ \pstate ->+ pstate { currentTerm = psNextTerm+ , votedFor = Nothing+ }++ voteForSelf = do+ selfNodeId <- askNodeId+ modify $ \pstate ->+ pstate { votedFor = Just selfNodeId }++--------------------------------------------------------------------------------+-- Logging+--------------------------------------------------------------------------------++logInfo = TransitionM . Logging.logInfo+logDebug = TransitionM . Logging.logDebug
+ src/Raft/NodeState.hs view
@@ -0,0 +1,218 @@+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE RankNTypes #-}++module Raft.NodeState where++import Protolude++import qualified Data.Serialize as S+import Data.Sequence (Seq(..))++import Raft.Log+import Raft.Types++data Mode+ = Follower+ | Candidate+ | Leader+ deriving (Show)++-- | All valid state transitions of a Raft node+data Transition (init :: Mode) (res :: Mode) where+ StartElection :: Transition 'Follower 'Candidate+ HigherTermFoundFollower :: Transition 'Follower 'Follower++ RestartElection :: Transition 'Candidate 'Candidate+ DiscoverLeader :: Transition 'Candidate 'Follower+ HigherTermFoundCandidate :: Transition 'Candidate 'Follower+ BecomeLeader :: Transition 'Candidate 'Leader++ SendHeartbeat :: Transition 'Leader 'Leader+ DiscoverNewLeader :: Transition 'Leader 'Follower+ HigherTermFoundLeader :: Transition 'Leader 'Follower++ Noop :: Transition init init+deriving instance Show (Transition init res)++-- | Existential type hiding the result type of a transition+data ResultState init where+ ResultState :: Transition init res -> NodeState res -> ResultState init++deriving instance Show (ResultState init)++followerResultState+ :: Transition init 'Follower+ -> FollowerState+ -> ResultState init+followerResultState transition fstate =+ ResultState transition (NodeFollowerState fstate)++candidateResultState+ :: Transition init 'Candidate+ -> CandidateState+ -> ResultState init+candidateResultState transition cstate =+ ResultState transition (NodeCandidateState cstate)++leaderResultState+ :: Transition init 'Leader+ -> LeaderState+ -> ResultState init+leaderResultState transition lstate =+ ResultState transition (NodeLeaderState lstate)++-- | Existential type hiding the internal node state+data RaftNodeState where+ RaftNodeState :: { unRaftNodeState :: NodeState s } -> RaftNodeState++nodeMode :: RaftNodeState -> Mode+nodeMode (RaftNodeState rns) =+ case rns of+ NodeFollowerState _ -> Follower+ NodeCandidateState _ -> Candidate+ NodeLeaderState _ -> Leader++-- | A node in Raft begins as a follower+initRaftNodeState :: RaftNodeState+initRaftNodeState =+ RaftNodeState $+ NodeFollowerState FollowerState+ { fsCommitIndex = index0+ , fsLastApplied = index0+ , fsCurrentLeader = NoLeader+ , fsLastLogEntryData = (index0, term0)+ , fsTermAtAEPrevIndex = Nothing+ }++deriving instance Show RaftNodeState++-- | The volatile state of a Raft Node+data NodeState (a :: Mode) where+ NodeFollowerState :: FollowerState -> NodeState 'Follower+ NodeCandidateState :: CandidateState -> NodeState 'Candidate+ NodeLeaderState :: LeaderState -> NodeState 'Leader++deriving instance Show (NodeState v)++-- | Representation of the current leader in the cluster. The system is+-- considered to be unavailable if there is no leader+data CurrentLeader+ = CurrentLeader LeaderId+ | NoLeader+ deriving (Show, Eq, Generic)++instance S.Serialize CurrentLeader++data FollowerState = FollowerState+ { fsCurrentLeader :: CurrentLeader+ -- ^ Id of the current leader+ , fsCommitIndex :: Index+ -- ^ Index of highest log entry known to be committed+ , fsLastApplied :: Index+ -- ^ Index of highest log entry applied to state machine+ , fsLastLogEntryData :: (Index, Term)+ -- ^ Index and term of the last log entry in the node's log+ , fsTermAtAEPrevIndex :: Maybe Term+ -- ^ The term of the log entry specified in and AppendEntriesRPC+ } deriving (Show)++data CandidateState = CandidateState+ { csCommitIndex :: Index+ -- ^ Index of highest log entry known to be committed+ , csLastApplied :: Index+ -- ^ Index of highest log entry applied to state machine+ , csVotes :: NodeIds+ -- ^ Votes from other nodes in the raft network+ , csLastLogEntryData :: (Index, Term)+ -- ^ Index and term of the last log entry in the node's log+ } deriving (Show)++type ClientReadReqs = Map Int (ClientId, Int)++data LeaderState = LeaderState+ { lsCommitIndex :: Index+ -- ^ Index of highest log entry known to be committed+ , lsLastApplied :: Index+ -- ^ Index of highest log entry applied to state machine+ , lsNextIndex :: Map NodeId Index+ -- ^ For each server, index of the next log entry to send to that server+ , lsMatchIndex :: Map NodeId Index+ -- ^ For each server, index of highest log entry known to be replicated on server+ , lsLastLogEntryData+ :: ( Index+ , Term+ , Maybe EntryIssuer+ )+ -- ^ Index, term, and client id of the last log entry in the node's log.+ -- The only time `Maybe ClientId` will be Nothing is at the initial term.+ , lsReadReqsHandled :: Int+ -- ^ Number of read requests handled this term+ , lsReadRequest :: ClientReadReqs+ -- ^ The number of successful responses received regarding a specific read+ -- request heartbeat.+ } deriving (Show)++--------------------------------------------------------------------------------+-- Helpers+--------------------------------------------------------------------------------++-- | Update the last log entry in the node's log+setLastLogEntryData :: NodeState ns -> Entries v -> NodeState ns+setLastLogEntryData nodeState entries =+ case entries of+ Empty -> nodeState+ _ :|> e ->+ case nodeState of+ NodeFollowerState fs ->+ NodeFollowerState fs+ { fsLastLogEntryData = (entryIndex e, entryTerm e) }+ NodeCandidateState cs ->+ NodeCandidateState cs+ { csLastLogEntryData = (entryIndex e, entryTerm e) }+ NodeLeaderState ls ->+ NodeLeaderState ls+ { lsLastLogEntryData = (entryIndex e, entryTerm e, Just (entryIssuer e)) }++-- | Get the last applied index and the commit index of the last log entry in+-- the node's log+getLastLogEntryData :: NodeState ns -> (Index, Term)+getLastLogEntryData nodeState =+ case nodeState of+ NodeFollowerState fs -> fsLastLogEntryData fs+ NodeCandidateState cs -> csLastLogEntryData cs+ NodeLeaderState ls -> let (peTerm, peIndex, _) = lsLastLogEntryData ls+ in (peTerm, peIndex)++-- | Get the index of highest log entry applied to state machine and the index+-- of highest log entry known to be committed+getLastAppliedAndCommitIndex :: NodeState ns -> (Index, Index)+getLastAppliedAndCommitIndex nodeState =+ case nodeState of+ NodeFollowerState fs -> (fsLastApplied fs, fsCommitIndex fs)+ NodeCandidateState cs -> (csLastApplied cs, csCommitIndex cs)+ NodeLeaderState ls -> (lsLastApplied ls, lsCommitIndex ls)++-- | Check if node is in a follower state+isFollower :: NodeState s -> Bool+isFollower nodeState =+ case nodeState of+ NodeFollowerState _ -> True+ _ -> False++-- | Check if node is in a candidate state+isCandidate :: NodeState s -> Bool+isCandidate nodeState =+ case nodeState of+ NodeCandidateState _ -> True+ _ -> False++-- | Check if node is in a leader state+isLeader :: NodeState s -> Bool+isLeader nodeState =+ case nodeState of+ NodeLeaderState _ -> True+ _ -> False
+ src/Raft/Persistent.hs view
@@ -0,0 +1,40 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveAnyClass #-}++module Raft.Persistent where++import Protolude+import qualified Data.Serialize as S++import Raft.Types++-- | Provides an interface to read and write the persistent+-- state to disk.+class Monad m => RaftPersist m where+ type RaftPersistError m+ readPersistentState+ :: Exception (RaftPersistError m)+ => m (Either (RaftPersistError m) PersistentState)+ writePersistentState+ :: Exception (RaftPersistError m)+ => PersistentState -> m (Either (RaftPersistError m) ())++-- | Persistent state that all Raft nodes maintain, regardless of node state.+data PersistentState = PersistentState+ { currentTerm :: Term+ -- ^ Last term server has seen+ , votedFor :: Maybe NodeId+ -- ^ Candidate id that received vote in current term+ } deriving (Show, Generic, S.Serialize)++-- | A node initiates its persistent state with term 0 and with its vote blank+initPersistentState :: PersistentState+initPersistentState = PersistentState+ { currentTerm = term0+ , votedFor = Nothing+ }
+ src/Raft/RPC.hs view
@@ -0,0 +1,131 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}++module Raft.RPC where++import Protolude++import Data.Serialize++import Raft.Log+import Raft.Types++-- | Interface for nodes to send messages to one+-- another. E.g. Control.Concurrent.Chan, Network.Socket, etc.+class RaftSendRPC m v where+ sendRPC :: NodeId -> RPCMessage v -> m ()++-- | Interface for nodes to receive messages from one+-- another+class Show (RaftRecvRPCError m v) => RaftRecvRPC m v where+ type RaftRecvRPCError m v+ receiveRPC :: m (Either (RaftRecvRPCError m v) (RPCMessage v))++-- | Representation of a message sent between nodes+data RPCMessage v = RPCMessage+ { sender :: NodeId+ , rpc :: RPC v+ } deriving (Show, Generic, Serialize)++data RPC v+ = AppendEntriesRPC (AppendEntries v)+ | AppendEntriesResponseRPC AppendEntriesResponse+ | RequestVoteRPC RequestVote+ | RequestVoteResponseRPC RequestVoteResponse+ deriving (Show, Generic, Serialize)++class RPCType a v where+ toRPC :: a -> RPC v++instance RPCType (AppendEntries v) v where+ toRPC = AppendEntriesRPC++instance RPCType AppendEntriesResponse v where+ toRPC = AppendEntriesResponseRPC++instance RPCType RequestVote v where+ toRPC = RequestVoteRPC++instance RPCType RequestVoteResponse v where+ toRPC = RequestVoteResponseRPC++rpcTerm :: RPC v -> Term+rpcTerm = \case+ AppendEntriesRPC ae -> aeTerm ae+ AppendEntriesResponseRPC aer -> aerTerm aer+ RequestVoteRPC rv -> rvTerm rv+ RequestVoteResponseRPC rvr -> rvrTerm rvr++data NoEntriesSpec+ = FromInconsistency+ | FromHeartbeat+ | FromClientReadReq Int+ deriving (Show)++data EntriesSpec v+ = FromIndex Index+ | FromNewLeader (Entry v)+ | FromClientWriteReq (Entry v)+ | NoEntries NoEntriesSpec+ deriving (Show)++-- | The data used to construct an AppendEntries value, snapshotted from the+-- node state at the time the AppendEntries val should be created.+data AppendEntriesData v = AppendEntriesData+ { aedTerm :: Term+ , aedLeaderCommit :: Index+ , aedEntriesSpec :: EntriesSpec v+ } deriving (Show)++-- | Representation of a message sent from a leader to its peers+data AppendEntries v = AppendEntries+ { aeTerm :: Term+ -- ^ Leader's term+ , aeLeaderId :: LeaderId+ -- ^ Leader's identifier so that followers can redirect clients+ , aePrevLogIndex :: Index+ -- ^ Index of log entry immediately preceding new ones+ , aePrevLogTerm :: Term+ -- ^ Term of aePrevLogIndex entry+ , aeEntries :: Entries v+ -- ^ Log entries to store (empty for heartbeat)+ , aeLeaderCommit :: Index+ -- ^ Leader's commit index+ , aeReadRequest :: Maybe Int+ -- ^ which read request the message corresponds to+ } deriving (Show, Generic, Serialize)++-- | Representation of the response from a follower to an AppendEntries message+data AppendEntriesResponse = AppendEntriesResponse+ { aerTerm :: Term+ -- ^ current term for leader to update itself+ , aerSuccess :: Bool+ -- ^ true if follower contained entry matching aePrevLogIndex and aePrevLogTerm+ , aerReadRequest :: Maybe Int+ -- ^ which read request the response corresponds to+ } deriving (Show, Generic, Serialize)++-- | Representation of the message sent by candidates to their peers to request+-- their vote+data RequestVote = RequestVote+ { rvTerm :: Term+ -- ^ candidates term+ , rvCandidateId :: NodeId+ -- ^ candidate requesting vote+ , rvLastLogIndex :: Index+ -- ^ index of candidate's last log entry+ , rvLastLogTerm :: Term+ -- ^ term of candidate's last log entry+ } deriving (Show, Generic, Serialize)++-- | Representation of a response to a RequestVote message+data RequestVoteResponse = RequestVoteResponse+ { rvrTerm :: Term+ -- ^ current term for candidate to update itself+ , rvrVoteGranted :: Bool+ -- ^ true means candidate recieved vote+ } deriving (Show, Generic, Serialize)
+ src/Raft/Types.hs view
@@ -0,0 +1,69 @@+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE RankNTypes #-}++module Raft.Types where++import Protolude++import Data.Serialize+import Numeric.Natural (Natural)++--------------------------------------------------------------------------------+-- NodeIds+--------------------------------------------------------------------------------++-- | Unique identifier of a Raft node+type NodeId = ByteString+type NodeIds = Set NodeId++-- | Unique identifier of a client+newtype ClientId = ClientId NodeId+ deriving (Show, Eq, Ord, Generic, Serialize)++-- | Unique identifier of a leader+newtype LeaderId = LeaderId { unLeaderId :: NodeId }+ deriving (Show, Eq, Generic, Serialize)++----------+-- Term --+----------++-- | Representation of monotonic election terms+newtype Term = Term Natural+ deriving (Show, Eq, Ord, Enum, Generic, Serialize)++-- | Initial term. Terms start at 0+term0 :: Term+term0 = Term 0++incrTerm :: Term -> Term+incrTerm = succ++prevTerm :: Term -> Term+prevTerm (Term 0) = Term 0+prevTerm t = pred t++-----------+-- Index --+-----------++-- | Representation of monotonic indices+newtype Index = Index Natural+ deriving (Show, Eq, Ord, Enum, Num, Integral, Real, Generic, Serialize)++-- | Initial index. Indeces start at 0+index0 :: Index+index0 = Index 0++incrIndex :: Index -> Index+incrIndex = succ++-- | Decrement index.+-- If the given index is 0, return the given index+decrIndexWithDefault0 :: Index -> Index+decrIndexWithDefault0 (Index 0) = index0+decrIndexWithDefault0 i = pred i
+ test/TestDejaFu.hs view
@@ -0,0 +1,429 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE TypeFamilies #-}++module TestDejaFu where++import Protolude hiding+ (STM, TChan, newTChan, readMVar, readTChan, writeTChan, atomically, killThread, ThreadId)++import Data.Sequence (Seq(..), (><), dropWhileR, (!?))+import qualified Data.Map as Map+import qualified Data.Maybe as Maybe+import qualified Data.Serialize as S+import Numeric.Natural++import Control.Monad.Catch+import Control.Monad.Conc.Class+import Control.Concurrent.Classy.STM.TChan++import Test.DejaFu hiding (get, ThreadId)+import Test.DejaFu.Internal (Settings(..))+import Test.DejaFu.Conc hiding (ThreadId)+import Test.Tasty+import Test.Tasty.DejaFu hiding (get)++import System.Random (mkStdGen)++import TestUtils++import Raft++--------------------------------------------------------------------------------+-- Test State Machine & Commands+--------------------------------------------------------------------------------++type Var = ByteString++data StoreCmd+ = Set Var Natural+ | Incr Var+ deriving (Show, Generic)++instance S.Serialize StoreCmd++type Store = Map Var Natural++data StoreCtx = StoreCtx++instance RSMP Store StoreCmd where+ data RSMPError Store StoreCmd = StoreError Text deriving (Show)+ type RSMPCtx Store StoreCmd = StoreCtx+ applyCmdRSMP _ store cmd =+ Right $ case cmd of+ Set x n -> Map.insert x n store+ Incr x -> Map.adjust succ x store++instance RSM Store StoreCmd RaftTestM where+ validateCmd _ = pure (Right ())+ askRSMPCtx = pure StoreCtx++type TestEventChan = EventChan ConcIO StoreCmd+type TestClientRespChan = TChan (STM ConcIO) (ClientResponse Store)++-- | Node specific environment+data TestNodeEnv = TestNodeEnv+ { testNodeEventChans :: Map NodeId TestEventChan+ , testClientRespChans :: Map ClientId TestClientRespChan+ , testNodeConfig :: NodeConfig+ }++-- | Node specific state+data TestNodeState = TestNodeState+ { testNodeLog :: Entries StoreCmd+ , testNodePersistentState :: PersistentState+ }++-- | A map of node ids to their respective node data+type TestNodeStates = Map NodeId TestNodeState++newtype RaftTestM a = RaftTestM {+ unRaftTestM :: ReaderT TestNodeEnv (StateT TestNodeStates ConcIO) a+ } deriving (Functor, Applicative, Monad, MonadIO, MonadReader TestNodeEnv, MonadState TestNodeStates)++deriving instance MonadThrow RaftTestM+deriving instance MonadCatch RaftTestM+deriving instance MonadMask RaftTestM+deriving instance MonadConc RaftTestM++runRaftTestM :: TestNodeEnv -> TestNodeStates -> RaftTestM a -> ConcIO a+runRaftTestM testEnv testState =+ flip evalStateT testState . flip runReaderT testEnv . unRaftTestM++newtype RaftTestError = RaftTestError Text+ deriving (Show)++instance Exception RaftTestError+throwTestErr = throw . RaftTestError++askSelfNodeId :: RaftTestM NodeId+askSelfNodeId = asks (configNodeId . testNodeConfig)++lookupNodeEventChan :: NodeId -> RaftTestM TestEventChan+lookupNodeEventChan nid = do+ testChanMap <- asks testNodeEventChans+ case Map.lookup nid testChanMap of+ Nothing -> throwTestErr $ "Node id " <> show nid <> " does not exist in TestEnv"+ Just testChan -> pure testChan++getNodeState :: NodeId -> RaftTestM TestNodeState+getNodeState nid = do+ testState <- get+ case Map.lookup nid testState of+ Nothing -> throwTestErr $ "Node id " <> show nid <> " does not exist in TestNodeStates"+ Just testNodeState -> pure testNodeState++modifyNodeState :: NodeId -> (TestNodeState -> TestNodeState) -> RaftTestM ()+modifyNodeState nid f =+ modify $ \testState ->+ case Map.lookup nid testState of+ Nothing -> panic $ "Node id " <> show nid <> " does not exist in TestNodeStates"+ Just testNodeState -> Map.insert nid (f testNodeState) testState++instance RaftPersist RaftTestM where+ type RaftPersistError RaftTestM = RaftTestError+ writePersistentState pstate' = do+ nid <- askSelfNodeId+ fmap Right $ modify $ \testState ->+ case Map.lookup nid testState of+ Nothing -> testState+ Just testNodeState -> do+ let newTestNodeState = testNodeState { testNodePersistentState = pstate' }+ Map.insert nid newTestNodeState testState+ readPersistentState = do+ nid <- askSelfNodeId+ testState <- get+ case Map.lookup nid testState of+ Nothing -> pure $ Left (RaftTestError "Failed to find node in environment")+ Just testNodeState -> pure $ Right (testNodePersistentState testNodeState)++instance RaftSendRPC RaftTestM StoreCmd where+ sendRPC nid rpc = do+ eventChan <- lookupNodeEventChan nid+ atomically $ writeTChan eventChan (MessageEvent (RPCMessageEvent rpc))++instance RaftSendClient RaftTestM Store where+ sendClient cid cr = do+ clientRespChans <- asks testClientRespChans+ case Map.lookup cid clientRespChans of+ Nothing -> panic "Failed to find client id in environment"+ Just clientRespChan -> atomically (writeTChan clientRespChan cr)++instance RaftWriteLog RaftTestM StoreCmd where+ type RaftWriteLogError RaftTestM = RaftTestError+ writeLogEntries entries = do+ nid <- askSelfNodeId+ fmap Right $+ modifyNodeState nid $ \testNodeState ->+ let log = testNodeLog testNodeState+ in testNodeState { testNodeLog = log >< entries }++instance RaftDeleteLog RaftTestM StoreCmd where+ type RaftDeleteLogError RaftTestM = RaftTestError+ deleteLogEntriesFrom idx = do+ nid <- askSelfNodeId+ fmap (const $ Right DeleteSuccess) $+ modifyNodeState nid $ \testNodeState ->+ let log = testNodeLog testNodeState+ newLog = dropWhileR ((<=) idx . entryIndex) log+ in testNodeState { testNodeLog = newLog }++instance RaftReadLog RaftTestM StoreCmd where+ type RaftReadLogError RaftTestM = RaftTestError+ readLogEntry (Index idx)+ | idx <= 0 = pure $ Right Nothing+ | otherwise = do+ log <- fmap testNodeLog . getNodeState =<< askSelfNodeId+ case log !? fromIntegral (pred idx) of+ Nothing -> pure (Right Nothing)+ Just e+ | entryIndex e == Index idx -> pure (Right $ Just e)+ | otherwise -> pure $ Left (RaftTestError "Malformed log")+ readLastLogEntry = do+ log <- fmap testNodeLog . getNodeState =<< askSelfNodeId+ case log of+ Empty -> pure (Right Nothing)+ _ :|> lastEntry -> pure (Right (Just lastEntry))++--------------------------------------------------------------------------------++initTestChanMaps :: ConcIO (Map NodeId TestEventChan, Map ClientId TestClientRespChan)+initTestChanMaps = do+ eventChans <-+ Map.fromList . zip (toList nodeIds) <$>+ atomically (replicateM (length nodeIds) newTChan)+ clientRespChans <-+ Map.fromList . zip [client0] <$>+ atomically (replicateM 1 newTChan)+ pure (eventChans, clientRespChans)++initRaftTestEnvs+ :: Map NodeId TestEventChan+ -> Map ClientId TestClientRespChan+ -> ([TestNodeEnv], TestNodeStates)+initRaftTestEnvs eventChans clientRespChans = (testNodeEnvs, testStates)+ where+ testNodeEnvs = map (TestNodeEnv eventChans clientRespChans) testConfigs+ testStates = Map.fromList $ zip (toList nodeIds) $+ replicate (length nodeIds) (TestNodeState mempty initPersistentState)++runTestNode :: TestNodeEnv -> TestNodeStates -> ConcIO ()+runTestNode testEnv testState = do+ runRaftTestM testEnv testState $+ runRaftT initRaftNodeState raftEnv $+ handleEventLoop (mempty :: Store)+ where+ nid = configNodeId (testNodeConfig testEnv)+ Just eventChan = Map.lookup nid (testNodeEventChans testEnv)+ raftEnv = RaftEnv eventChan dummyTimer dummyTimer (testNodeConfig testEnv) NoLogs+ dummyTimer = pure ()++forkTestNodes :: [TestNodeEnv] -> TestNodeStates -> ConcIO [ThreadId ConcIO]+forkTestNodes testEnvs testStates =+ mapM (fork . flip runTestNode testStates) testEnvs++--------------------------------------------------------------------------------++type TestEventChans = Map NodeId TestEventChan+type TestClientRespChans = Map ClientId TestClientRespChan++test_concurrency :: [TestTree]+test_concurrency =+ [ testGroup "Leader Election" [ testConcurrentProps (leaderElection node0) mempty ]+ , testGroup "increment(set('x', 41)) == x := 42"+ [ testConcurrentProps incrValue (Map.fromList [("x", 42)], Index 3) ]+ , testGroup "set('x', 0) ... 10x incr(x) == x := 10"+ [ testConcurrentProps multIncrValue (Map.fromList [("x", 10)], Index 12) ]+ , testGroup "Follower redirect with no leader" [ testConcurrentProps followerRedirNoLeader NoLeader ]+ , testGroup "Follower redirect with leader" [ testConcurrentProps followerRedirLeader (CurrentLeader (LeaderId node0)) ]+ , testGroup "New leader election" [ testConcurrentProps newLeaderElection (CurrentLeader (LeaderId node1)) ]+ , testGroup "Comprehensive"+ [ testConcurrentProps comprehensive (Index 14, Map.fromList [("x", 9), ("y", 6), ("z", 42)], CurrentLeader (LeaderId node0)) ]+ ]++testConcurrentProps+ :: (Eq a, Show a)+ => (TestEventChans -> TestClientRespChans -> ConcIO a)+ -> a+ -> TestTree+testConcurrentProps test expected =+ testDejafusWithSettings settings+ [ ("No deadlocks", deadlocksNever)+ , ("No Exceptions", exceptionsNever)+ , ("Success", alwaysTrue (== Right expected))+ ] $ concurrentRaftTest test+ where+ settings = defaultSettings+ { _way = randomly (mkStdGen 42) 100+ }++ concurrentRaftTest :: (TestEventChans -> TestClientRespChans -> ConcIO a) -> ConcIO a+ concurrentRaftTest runTest =+ Control.Monad.Catch.bracket setup teardown $+ uncurry runTest . snd+ where+ setup = do+ (eventChans, clientRespChans) <- initTestChanMaps+ let (testNodeEnvs, testNodeStates) = initRaftTestEnvs eventChans clientRespChans+ tids <- forkTestNodes testNodeEnvs testNodeStates+ pure (tids, (eventChans, clientRespChans))++ teardown = mapM_ killThread . fst++leaderElection :: NodeId -> TestEventChans -> TestClientRespChans -> ConcIO Store+leaderElection nid eventChans clientRespChans = do+ atomically $ writeTChan nodeEventChan (TimeoutEvent ElectionTimeout)+ pollForReadResponse nodeEventChan client0RespChan+ where+ Just nodeEventChan = Map.lookup nid eventChans+ Just client0RespChan = Map.lookup client0 clientRespChans++incrValue :: TestEventChans -> TestClientRespChans -> ConcIO (Store, Index)+incrValue eventChans clientRespChans = do+ leaderElection node0 eventChans clientRespChans+ Right idx <- do+ syncClientWrite node0EventChan (client0, client0RespChan) (Set "x" 41)+ syncClientWrite node0EventChan (client0, client0RespChan) (Incr"x")+ store <- pollForReadResponse node0EventChan client0RespChan+ pure (store, idx)+ where+ Just node0EventChan = Map.lookup node0 eventChans+ Just client0RespChan = Map.lookup client0 clientRespChans++multIncrValue :: TestEventChans -> TestClientRespChans -> ConcIO (Store, Index)+multIncrValue eventChans clientRespChans = do+ leaderElection node0 eventChans clientRespChans+ syncClientWrite node0EventChan (client0, client0RespChan) (Set "x" 0)+ Right idx <-+ fmap (Maybe.fromJust . lastMay) $ replicateM 10 $ do+ res <- syncClientWrite node0EventChan (client0, client0RespChan) (Incr "x")+ pollForReadResponse node0EventChan client0RespChan+ pure res+ store <- pollForReadResponse node0EventChan client0RespChan+ pure (store, idx)+ where+ Just node0EventChan = Map.lookup node0 eventChans+ Just client0RespChan = Map.lookup client0 clientRespChans++leaderRedirect :: TestEventChans -> TestClientRespChans -> ConcIO CurrentLeader+leaderRedirect eventChans clientRespChans = do+ Left resp <- syncClientWrite node1EventChan (client0, client0RespChan) (Set "x" 42)+ pure resp+ where+ Just node1EventChan = Map.lookup node1 eventChans+ Just client0RespChan = Map.lookup client0 clientRespChans++followerRedirNoLeader :: TestEventChans -> TestClientRespChans -> ConcIO CurrentLeader+followerRedirNoLeader = leaderRedirect++followerRedirLeader :: TestEventChans -> TestClientRespChans -> ConcIO CurrentLeader+followerRedirLeader eventChans clientRespChans = do+ leaderElection node0 eventChans clientRespChans+ leaderRedirect eventChans clientRespChans++newLeaderElection :: TestEventChans -> TestClientRespChans -> ConcIO CurrentLeader+newLeaderElection eventChans clientRespChans = do+ leaderElection node0 eventChans clientRespChans+ leaderElection node1 eventChans clientRespChans+ leaderElection node2 eventChans clientRespChans+ leaderElection node1 eventChans clientRespChans+ atomically $ writeTChan node0EventChan $ clientReadReq client0+ ClientRedirectResponse (ClientRedirResp ldr) <- atomically $ readTChan client0RespChan+ pure ldr+ where+ Just node0EventChan = Map.lookup node0 eventChans+ Just client0RespChan = Map.lookup client0 clientRespChans++comprehensive :: TestEventChans -> TestClientRespChans -> ConcIO (Index, Store, CurrentLeader)+comprehensive eventChans clientRespChans = do+ leaderElection node0 eventChans clientRespChans+ Right idx2 <- syncClientWriteClient0 node0EventChan (Set "x" 7)+ Right idx3 <- syncClientWriteClient0 node0EventChan (Set "y" 3)+ Left (CurrentLeader _) <- syncClientWriteClient0 node1EventChan (Incr "y")+ Right _ <- syncClientRead node0EventChan (client0, client0RespChan)++ leaderElection node1 eventChans clientRespChans+ Right idx5 <- syncClientWriteClient0 node1EventChan (Incr "x")+ Right idx6 <- syncClientWriteClient0 node1EventChan (Incr "y")+ Right idx7 <- syncClientWriteClient0 node1EventChan (Set "z" 40)+ Left (CurrentLeader _) <- syncClientWriteClient0 node2EventChan (Incr "y")+ Right _ <- syncClientRead node1EventChan (client0, client0RespChan)++ leaderElection node2 eventChans clientRespChans+ Right idx9 <- syncClientWriteClient0 node2EventChan (Incr "z")+ Right idx10 <- syncClientWriteClient0 node2EventChan (Incr "x")+ Left _ <- syncClientWriteClient0 node1EventChan (Set "q" 100)+ Right idx11 <- syncClientWriteClient0 node2EventChan (Incr "y")+ Left _ <- syncClientWriteClient0 node0EventChan (Incr "z")+ Right idx12 <- syncClientWriteClient0 node2EventChan (Incr "y")+ Left (CurrentLeader _) <- syncClientWriteClient0 node0EventChan (Incr "y")+ Right _ <- syncClientRead node2EventChan (client0, client0RespChan)++ leaderElection node0 eventChans clientRespChans+ Right idx14 <- syncClientWriteClient0 node0EventChan (Incr "z")+ Left (CurrentLeader _) <- syncClientWriteClient0 node1EventChan (Incr "y")++ Right store <- syncClientRead node0EventChan (client0, client0RespChan)+ Left ldr <- syncClientRead node1EventChan (client0, client0RespChan)++ pure (idx14, store, ldr)+ where+ syncClientWriteClient0 = flip syncClientWrite (client0, client0RespChan)++ Just node0EventChan = Map.lookup node0 eventChans+ Just node1EventChan = Map.lookup node1 eventChans+ Just node2EventChan = Map.lookup node2 eventChans++ Just client0RespChan = Map.lookup client0 clientRespChans++--------------------------------------------------------------------------------+-- Helpers+--------------------------------------------------------------------------------++pollForReadResponse :: TestEventChan -> TestClientRespChan -> ConcIO Store+pollForReadResponse nodeEventChan clientRespChan = do+ -- Warning: Do not change the separate "atomically" calls, or you may+ -- introduce a deadlock+ atomically $ writeTChan nodeEventChan $ clientReadReq client0+ res <- atomically $ readTChan clientRespChan+ case res of+ ClientReadResponse (ClientReadResp res) -> pure res+ _ -> do+ liftIO $ Control.Monad.Conc.Class.threadDelay 1000+ pollForReadResponse nodeEventChan clientRespChan++syncClientRead :: TestEventChan -> (ClientId, TestClientRespChan) -> ConcIO (Either CurrentLeader Store)+syncClientRead nodeEventChan (cid, clientRespChan) = do+ atomically $ writeTChan nodeEventChan $ clientReadReq client0+ res <- atomically $ readTChan clientRespChan+ case res of+ ClientReadResponse (ClientReadResp store) -> pure $ Right store+ ClientRedirectResponse (ClientRedirResp ldr) -> pure $ Left ldr+ _ -> panic "Failed to recieve valid read response"++syncClientWrite :: TestEventChan -> (ClientId, TestClientRespChan) -> StoreCmd -> ConcIO (Either CurrentLeader Index)+syncClientWrite nodeEventChan (cid, clientRespChan) cmd = do+ atomically $ writeTChan nodeEventChan (clientWriteReq cid cmd)+ res <- atomically $ readTChan clientRespChan+ case res of+ ClientWriteResponse (ClientWriteResp idx) -> do+ heartbeat nodeEventChan+ pure $ Right idx+ ClientRedirectResponse (ClientRedirResp ldr) -> pure $ Left ldr+ _ -> panic "Failed to receive client write response..."++heartbeat :: TestEventChan -> ConcIO ()+heartbeat eventChan = atomically $ writeTChan eventChan (TimeoutEvent HeartbeatTimeout)++clientReadReq :: ClientId -> Event StoreCmd+clientReadReq cid = MessageEvent $ ClientRequestEvent $ ClientRequest cid ClientReadReq++clientWriteReq :: ClientId -> StoreCmd -> Event StoreCmd+clientWriteReq cid v = MessageEvent $ ClientRequestEvent $ ClientRequest cid $ ClientWriteReq v
+ test/TestDriver.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF tasty-discover -optF --tree-display #-}
+ test/TestRaft.hs view
@@ -0,0 +1,573 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ScopedTypeVariables #-}++module TestRaft where++import Protolude+import qualified Data.Sequence as Seq+import Data.Sequence (Seq(..), (|>))+import qualified Data.Map as Map+import qualified Data.Set as Set+import qualified Data.Serialize as S+import Numeric.Natural+import Control.Monad.Conc.Class (throw)++import qualified Test.Tasty.HUnit as HUnit++import TestUtils++import Raft hiding (sendClient)+import Raft.Logging (logMsgToText, logMsgData, logMsgNodeId, LogMsg)+import Raft.Action+import Raft.Handle+import Raft.Log+import Raft.Monad+import Raft.Types+import Raft.RPC++------------------------------+-- State Machine & Commands --+------------------------------++type Var = ByteString++data StoreCmd+ = Set Var Natural+ | Incr Var+ deriving (Show, Generic)++instance S.Serialize StoreCmd++type Store = Map Var Natural++instance RSMP Store StoreCmd where+ data RSMPError Store StoreCmd = StoreError Text deriving (Show)+ type RSMPCtx Store StoreCmd = ()+ applyCmdRSMP _ store cmd =+ Right $ case cmd of+ Set x n -> Map.insert x n store+ Incr x -> Map.adjust succ x store++testVar :: Var+testVar = "test"++testInitVal :: Natural+testInitVal = 1++testSetCmd :: StoreCmd+testSetCmd = Set testVar testInitVal++testIncrCmd :: StoreCmd+testIncrCmd = Incr testVar++--------------------+-- Scenario Monad --+--------------------++type ClientResps = Map ClientId (Seq (ClientResponse Store))++data TestState = TestState+ { testNodeIds :: NodeIds+ , testNodeLogs :: Map NodeId (Entries StoreCmd)+ , testNodeSMs :: Map NodeId Store+ , testNodeRaftStates :: Map NodeId RaftNodeState+ , testNodePersistentStates :: Map NodeId PersistentState+ , testNodeConfigs :: Map NodeId NodeConfig+ , testClientResps :: ClientResps+ } deriving (Show)++type Scenario v = StateT TestState IO v++-- | Run scenario monad with initial state+runScenario :: Scenario () -> IO ()+runScenario scenario = do+ let initPersistentState = PersistentState term0 Nothing+ let initTestState = TestState+ { testNodeIds = nodeIds+ , testNodeLogs = Map.fromList $ (, mempty) <$> Set.toList nodeIds+ , testNodeSMs = Map.fromList $ (, mempty) <$> Set.toList nodeIds+ , testNodeRaftStates = Map.fromList $ (, initRaftNodeState) <$> Set.toList nodeIds+ , testNodePersistentStates = Map.fromList $ (, initPersistentState) <$> Set.toList nodeIds+ , testNodeConfigs = Map.fromList $ zip (Set.toList nodeIds) testConfigs+ , testClientResps = Map.fromList [(client0, mempty)]+ }++ evalStateT scenario initTestState++updateStateMachine :: NodeId -> Store -> Scenario ()+updateStateMachine nodeId sm+ = modify $ \testState@TestState{..}+ -> testState+ { testNodeSMs = Map.insert nodeId sm testNodeSMs+ }++updatePersistentState :: NodeId -> PersistentState -> Scenario ()+updatePersistentState nodeId persistentState+ = modify $ \testState@TestState{..}+ -> testState+ { testNodePersistentStates = Map.insert nodeId persistentState testNodePersistentStates+ }++updateRaftNodeState :: NodeId -> RaftNodeState -> Scenario ()+updateRaftNodeState nodeId raftState+ = modify $ \testState@TestState{..}+ -> testState+ { testNodeRaftStates = Map.insert nodeId raftState testNodeRaftStates+ }++getNodeInfo :: NodeId -> Scenario (NodeConfig, Store, RaftNodeState, PersistentState)+getNodeInfo nId = do+ nodeConfigs <- gets testNodeConfigs+ nodeSMs <- gets testNodeSMs+ nodeRaftStates <- gets testNodeRaftStates+ nodePersistentStates <- gets testNodePersistentStates+ let Just nodeInfo = Map.lookup nId nodeConfigs >>= \config ->+ Map.lookup nId nodeSMs >>= \store ->+ Map.lookup nId nodeRaftStates >>= \raftState ->+ Map.lookup nId nodePersistentStates >>= \persistentState ->+ pure (config, store, raftState, persistentState)+ pure nodeInfo+++lookupClientResps :: ClientId -> ClientResps -> Seq (ClientResponse Store)+lookupClientResps clientId cResps =+ case Map.lookup clientId cResps of+ Nothing -> panic "Client id not found"+ Just resps -> resps++lookupLastClientResp :: ClientId -> ClientResps -> ClientResponse Store+lookupLastClientResp clientId cResps = r+ where+ (_ :|> r) = lookupClientResps clientId cResps++sendClient :: ClientId -> ClientResponse Store -> Scenario ()+sendClient clientId resp = do+ cResps <- gets testClientResps+ let resps = lookupClientResps clientId cResps+ modify (\st -> st { testClientResps = Map.insert clientId (resps |> resp) (testClientResps st) })++-------------------+-- Log instances --+-------------------++newtype NodeEnvError = NodeEnvError Text+ deriving (Show)++instance Exception NodeEnvError++type RTLog = ReaderT NodeId (StateT TestState IO)++instance RaftWriteLog RTLog StoreCmd where+ type RaftWriteLogError RTLog = NodeEnvError+ writeLogEntries newEntries = do+ nid <- ask+ Just log <- Map.lookup nid <$> gets testNodeLogs+ fmap Right $ modify $ \testState@TestState{..} ->+ testState { testNodeLogs = Map.insert nid (log Seq.>< newEntries) testNodeLogs }++instance RaftReadLog RTLog StoreCmd where+ type RaftReadLogError RTLog = NodeEnvError+ readLogEntry (Index idx) = do+ nid <- ask+ Just log <- Map.lookup nid <$> gets testNodeLogs+ case log Seq.!? fromIntegral (if idx == 0 then 0 else idx - 1) of+ Nothing -> pure (Right Nothing)+ Just e -> pure (Right (Just e))+ readLastLogEntry = do+ nid <- ask+ Just log <- Map.lookup nid <$> gets testNodeLogs+ case log of+ Seq.Empty -> pure (Right Nothing)+ (_ Seq.:|> e) -> pure (Right (Just e))++instance RaftDeleteLog RTLog StoreCmd where+ type RaftDeleteLogError RTLog = NodeEnvError+ deleteLogEntriesFrom idx = do+ nid <- ask+ Just log <- Map.lookup nid <$> gets testNodeLogs+ fmap (const (Right DeleteSuccess)) $ modify $ \testState@TestState{..} ->+ testState { testNodeLogs = Map.insert nid (Seq.dropWhileR ((>= idx) . entryIndex) log) testNodeLogs }++-------------------------------+-- Handle actions and events --+-------------------------------++testHandleLogs :: Maybe [NodeId] -> (Text -> IO ()) -> [LogMsg] -> Scenario ()+testHandleLogs nIdsM f logs = liftIO $+ case nIdsM of+ Nothing -> mapM_ (f . logMsgToText) logs+ Just nIds ->+ mapM_ (f . logMsgToText) $ flip filter logs $ \log ->+ logMsgNodeId (logMsgData log) `elem` nIds++testHandleActions :: NodeId -> [Action Store StoreCmd] -> Scenario ()+testHandleActions sender =+ mapM_ (testHandleAction sender)++testHandleAction :: NodeId -> Action Store StoreCmd -> Scenario ()+testHandleAction sender action = do+ case action of+ SendRPC nId rpcAction -> do+ msg <- mkRPCfromSendRPCAction sender rpcAction+ testHandleEvent nId (MessageEvent (RPCMessageEvent msg))+ SendRPCs msgs ->+ mapM_ (\(nId, rpcAction) -> do+ msg <- mkRPCfromSendRPCAction sender rpcAction+ testHandleEvent nId (MessageEvent (RPCMessageEvent msg))+ ) (Map.toList msgs)+ BroadcastRPC nIds rpcAction -> mapM_ (\nId -> do+ msg <- mkRPCfromSendRPCAction sender rpcAction+ testHandleEvent nId (MessageEvent (RPCMessageEvent msg))) nIds+ RespondToClient clientId resp -> sendClient clientId resp+ ResetTimeoutTimer _ -> noop+ AppendLogEntries entries -> do+ runReaderT (updateLog entries) sender+ modify $ \testState@TestState{..}+ -> case Map.lookup sender testNodeRaftStates of+ Nothing -> panic "No NodeState"+ Just (RaftNodeState ns) -> testState+ { testNodeRaftStates = Map.insert sender (RaftNodeState (setLastLogEntryData ns entries)) testNodeRaftStates+ }+ where+ noop = pure ()++ mkRPCfromSendRPCAction+ :: NodeId -> SendRPCAction StoreCmd -> Scenario (RPCMessage StoreCmd)+ mkRPCfromSendRPCAction nId sendRPCAction = do+ sc <- get+ (nodeConfig, _, raftState@(RaftNodeState ns), _) <- getNodeInfo nId+ RPCMessage (configNodeId nodeConfig) <$>+ case sendRPCAction of+ SendAppendEntriesRPC aeData -> do+ (entries, prevLogIndex, prevLogTerm, aeReadReq) <-+ case aedEntriesSpec aeData of+ FromIndex idx -> do+ eLogEntries <- runReaderT (readLogEntriesFrom (decrIndexWithDefault0 idx)) nId+ case eLogEntries of+ Left err -> throw err+ Right log ->+ case log of+ pe :<| entries@(e :<| _)+ | idx == 1 -> pure (log, index0, term0, Nothing)+ | otherwise -> pure (entries, entryIndex pe, entryTerm pe, Nothing)+ _ -> pure (log, index0, term0, Nothing)+ FromClientWriteReq e -> prevEntryData nId e+ FromNewLeader e -> prevEntryData nId e+ NoEntries spec -> do+ let readReq' =+ case spec of+ FromClientReadReq n -> Just n+ _ -> Nothing+ (lastLogIndex, lastLogTerm) = getLastLogEntryData ns+ pure (Empty, lastLogIndex, lastLogTerm, readReq')+ let leaderId = LeaderId (configNodeId nodeConfig)+ pure . toRPC $+ AppendEntries+ { aeTerm = aedTerm aeData+ , aeLeaderId = leaderId+ , aePrevLogIndex = prevLogIndex+ , aePrevLogTerm = prevLogTerm+ , aeEntries = entries+ , aeLeaderCommit = aedLeaderCommit aeData+ , aeReadRequest = aeReadReq+ }+ SendAppendEntriesResponseRPC aer -> do+ pure (toRPC aer)+ SendRequestVoteRPC rv -> pure (toRPC rv)+ SendRequestVoteResponseRPC rvr -> pure (toRPC rvr)++ prevEntryData nId e = do+ (x,y,z) <- prevEntryData' nId e+ pure (x,y,z,Nothing)++ prevEntryData' nId e+ | entryIndex e == Index 1 = pure (Seq.singleton e, index0, term0)+ | otherwise = do+ eLogEntry <- runReaderT (readLogEntry (decrIndexWithDefault0 (entryIndex e))) nId+ case eLogEntry of+ Left err -> throw err+ Right Nothing -> pure (Seq.singleton e, index0, term0)+ Right (Just (prevEntry :: Entry StoreCmd)) ->+ pure (Seq.singleton e, entryIndex prevEntry, entryTerm prevEntry)++testHandleEvent :: NodeId -> Event StoreCmd -> Scenario ()+testHandleEvent nodeId event = do+ (nodeConfig, sm, raftState', persistentState) <- getNodeInfo nodeId+ raftState <- loadLogEntryTermAtAePrevLogIndex raftState'+ let transitionEnv = TransitionEnv nodeConfig sm raftState+ let (newRaftState, newPersistentState, actions, logMsgs) = handleEvent raftState transitionEnv persistentState event+ updatePersistentState nodeId newPersistentState+ updateRaftNodeState nodeId newRaftState+ testHandleActions nodeId actions+ testHandleLogs Nothing (const $ pure ()) logMsgs+ applyLogEntries nodeId sm+ where+ applyLogEntries+ :: NodeId+ -> Store+ -> Scenario ()+ applyLogEntries nId stateMachine = do+ (_, _, raftNodeState@(RaftNodeState nodeState), _) <- getNodeInfo nId+ let lastAppliedIndex = lastApplied nodeState+ when (commitIndex nodeState > lastAppliedIndex) $ do+ let resNodeState = incrLastApplied nodeState+ modify $ \testState@TestState{..} -> testState {+ testNodeRaftStates = Map.insert nId (RaftNodeState resNodeState) testNodeRaftStates }+ let newLastAppliedIndex = lastApplied resNodeState+ eLogEntry <- runReaderT (readLogEntry newLastAppliedIndex) nId+ case eLogEntry of+ Left err -> throw err+ Right Nothing -> panic "No log entry at 'newLastAppliedIndex'"+ Right (Just logEntry) -> do+ case entryValue logEntry of+ NoValue -> applyLogEntries nId stateMachine+ EntryValue v -> do+ let Right newStateMachine = applyCmdRSMP () stateMachine v+ updateStateMachine nId newStateMachine+ applyLogEntries nId newStateMachine++ where+ incrLastApplied :: NodeState ns -> NodeState ns+ incrLastApplied nodeState =+ case nodeState of+ NodeFollowerState fs ->+ let lastApplied' = incrIndex (fsLastApplied fs)+ in NodeFollowerState $ fs { fsLastApplied = lastApplied' }+ NodeCandidateState cs ->+ let lastApplied' = incrIndex (csLastApplied cs)+ in NodeCandidateState $ cs { csLastApplied = lastApplied' }+ NodeLeaderState ls ->+ let lastApplied' = incrIndex (lsLastApplied ls)+ in NodeLeaderState $ ls { lsLastApplied = lastApplied' }++ lastApplied :: NodeState ns -> Index+ lastApplied = fst . getLastAppliedAndCommitIndex++ commitIndex :: NodeState ns -> Index+ commitIndex = snd . getLastAppliedAndCommitIndex++ -- In the case that a node is a follower receiving an AppendEntriesRPC+ -- Event, read the log at the aePrevLogIndex+ loadLogEntryTermAtAePrevLogIndex :: RaftNodeState -> Scenario RaftNodeState+ loadLogEntryTermAtAePrevLogIndex (RaftNodeState rns) =+ case event of+ MessageEvent (RPCMessageEvent (RPCMessage _ (AppendEntriesRPC ae))) -> do+ case rns of+ NodeFollowerState fs -> do+ eEntry <- runReaderT (readLogEntry (aePrevLogIndex ae)) nodeId+ case eEntry of+ Left err -> throw err+ Right (mEntry :: Maybe (Entry StoreCmd)) ->+ pure $ RaftNodeState $ NodeFollowerState fs+ { fsTermAtAEPrevIndex = entryTerm <$> mEntry }+ _ -> pure (RaftNodeState rns)+ _ -> pure (RaftNodeState rns)++testHeartbeat :: NodeId -> Scenario ()+testHeartbeat sender = do+ nodeRaftStates <- gets testNodeRaftStates+ nodePersistentStates <- gets testNodePersistentStates+ nIds <- gets testNodeIds+ let Just raftState = Map.lookup sender nodeRaftStates+ Just persistentState = Map.lookup sender nodePersistentStates+ unless (isRaftLeader raftState) $ panic $ toS (show sender ++ " must a be a leader to heartbeat")+ let LeaderState{..} = getInnerLeaderState raftState+ let aeData = AppendEntriesData+ { aedTerm = currentTerm persistentState+ , aedEntriesSpec = NoEntries FromHeartbeat+ , aedLeaderCommit = lsCommitIndex+ }++ -- Broadcast AppendEntriesRPC+ testHandleAction sender+ (BroadcastRPC (Set.filter (sender /=) nIds) (SendAppendEntriesRPC aeData))+ where+ getInnerLeaderState :: RaftNodeState -> LeaderState+ getInnerLeaderState nodeState = case nodeState of+ (RaftNodeState (NodeLeaderState leaderState)) -> leaderState+ _ -> panic "Node must be a leader to access its leader state"+++----------------------+-- Test raft events --+----------------------++testInitLeader :: NodeId -> Scenario ()+testInitLeader nId =+ testHandleEvent nId (TimeoutEvent ElectionTimeout)++testClientReadRequest :: NodeId -> Scenario ()+testClientReadRequest nId =+ testHandleEvent nId (MessageEvent+ (ClientRequestEvent+ (ClientRequest client0 ClientReadReq)))++testClientWriteRequest :: StoreCmd -> NodeId -> Scenario ()+testClientWriteRequest cmd nId =+ testHandleEvent nId (MessageEvent+ (ClientRequestEvent+ (ClientRequest client0 (ClientWriteReq cmd))))++----------------+-- Unit tests --+----------------++-- When the protocol starts, every node is a follower+unit_init_protocol :: IO ()+unit_init_protocol = runScenario $ do+ -- Node 0 becomes the leader+ testInitLeader node0++ raftStates <- gets testNodeRaftStates++ -- Node0 has become leader and other nodes are followers+ liftIO $ assertLeader raftStates [(node0, NoLeader), (node1, CurrentLeader (LeaderId node0)), (node2, CurrentLeader (LeaderId node0))]+ liftIO $ assertNodeState raftStates [(node0, isRaftLeader), (node1, isRaftFollower), (node2, isRaftFollower)]++unit_append_entries_client_request :: IO ()+unit_append_entries_client_request = runScenario $ do++ testInitLeader node0++ raftStates0 <- gets testNodeRaftStates+ sms0 <- gets testNodeSMs+ logs0 <- gets testNodeLogs++ liftIO $ assertPersistedLogs logs0 [(node0, 1), (node1, 1), (node2, 1)]+ liftIO $ assertCommittedLogIndex raftStates0 [(node0, Index 1), (node1, Index 0), (node2, Index 0)]+ liftIO $ assertAppliedLogIndex raftStates0 [(node0, Index 1), (node1, Index 0), (node2, Index 0)]+ liftIO $ assertSMs sms0 [(node0, mempty), (node1, mempty), (node2, mempty)]++ testClientWriteRequest testSetCmd node0++ raftStates1 <- gets testNodeRaftStates+ sms1 <- gets testNodeSMs+ logs1 <- gets testNodeLogs++ liftIO $ assertPersistedLogs logs1 [(node0, 2), (node1, 2), (node2, 2)]+ liftIO $ assertCommittedLogIndex raftStates1 [(node0, Index 2), (node1, Index 1), (node2, Index 1)]+ liftIO $ assertAppliedLogIndex raftStates1 [(node0, Index 2), (node1, Index 1), (node2, Index 1)]+ liftIO $ assertSMs sms1 [(node0, Map.fromList [(testVar, testInitVal)]), (node1, mempty), (node2, mempty)]++ ---------------------------- HEARTBEAT 1 ------------------------------+ -- After leader heartbeats, followers commit and apply leader's entries+ testHeartbeat node0++ raftStates2 <- gets testNodeRaftStates+ sms2 <- gets testNodeSMs+ logs2 <- gets testNodeLogs++ liftIO $ assertPersistedLogs logs2 [(node0, 2), (node1, 2), (node2, 2)]+ liftIO $ assertCommittedLogIndex raftStates2 [(node0, Index 2), (node1, Index 2), (node2, Index 2)]+ liftIO $ assertAppliedLogIndex raftStates2 [(node0, Index 2), (node1, Index 2), (node2, Index 2)]+ liftIO $ assertSMs sms2 [(node0, Map.fromList [(testVar, testInitVal)]), (node1, Map.fromList [(testVar, testInitVal)]), (node2, Map.fromList [(testVar, testInitVal)])]++++unit_incr_value :: IO ()+unit_incr_value = runScenario $ do+ testInitLeader node0+ testClientWriteRequest testSetCmd node0+ testClientWriteRequest testIncrCmd node0++ testHeartbeat node0++ sms <- gets testNodeSMs+ liftIO $ assertSMs sms [(node0, Map.fromList [(testVar, succ testInitVal)]), (node1, Map.fromList [(testVar, succ testInitVal)]), (node2, Map.fromList [(testVar, succ testInitVal)])]+++unit_mult_incr_value :: IO ()+unit_mult_incr_value = runScenario $ do+ testInitLeader node0+ testClientWriteRequest testSetCmd node0+ let reps = 10+ replicateM_ (fromIntegral 10) (testClientWriteRequest testIncrCmd node0)+ testHeartbeat node0++ sms <- gets testNodeSMs+ liftIO $ assertSMs sms [(node0, Map.fromList [(testVar, testInitVal + reps)]), (node1, Map.fromList [(testVar, testInitVal + reps)]), (node2, Map.fromList [(testVar, testInitVal + reps)])]++unit_client_req_no_leader :: IO ()+unit_client_req_no_leader = runScenario $ do+ testClientWriteRequest testSetCmd node1+ cResps <- gets testClientResps+ let ClientRedirectResponse (ClientRedirResp lResp) = lookupLastClientResp client0 cResps+ liftIO $ HUnit.assertBool "A follower should return a NoLeader response" (lResp == NoLeader)++unit_redirect_leader :: IO ()+unit_redirect_leader = runScenario $ do+ testInitLeader node0+ testClientWriteRequest testSetCmd node1+ cResps <- gets testClientResps+ let ClientRedirectResponse (ClientRedirResp (CurrentLeader (LeaderId lResp))) = lookupLastClientResp client0 cResps+ liftIO $ HUnit.assertBool "A follower should point to the current leader" (lResp == node0)++unit_client_read_response :: IO ()+unit_client_read_response = runScenario $ do+ testInitLeader node0+ testClientWriteRequest testSetCmd node0+ testClientReadRequest node0+ cResps <- gets testClientResps+ let ClientReadResponse (ClientReadResp store) = lookupLastClientResp client0 cResps+ liftIO $ HUnit.assertBool "A client should receive the current state of the store"+ (store == Map.fromList [(testVar, testInitVal)])++unit_client_write_response :: IO ()+unit_client_write_response = runScenario $ do+ testInitLeader node0+ testClientReadRequest node0+ testClientWriteRequest testSetCmd node0+ cResps <- gets testClientResps+ let ClientWriteResponse (ClientWriteResp idx) = lookupLastClientResp client0 cResps+ liftIO $ HUnit.assertBool "A client should receive an aknowledgement of a writing request"+ (idx == Index 2)++unit_new_leader :: IO ()+unit_new_leader = runScenario $ do+ testInitLeader node0+ testHandleEvent node1 (TimeoutEvent ElectionTimeout)+ raftStates <- gets testNodeRaftStates++ liftIO $ assertNodeState raftStates [(node0, isRaftFollower), (node1, isRaftLeader), (node2, isRaftFollower)]+ liftIO $ assertLeader raftStates [(node0, CurrentLeader (LeaderId node1)), (node1, NoLeader), (node2, CurrentLeader (LeaderId node1))]++------------------+-- Assert utils --+------------------++assertNodeState :: Map NodeId RaftNodeState -> [(NodeId, RaftNodeState -> Bool)] -> IO ()+assertNodeState raftNodeStates =+ mapM_ (\(nId, isNodeState) -> HUnit.assertBool (show nId ++ " should be in a different state")+ (maybe False isNodeState (Map.lookup nId raftNodeStates)))++assertLeader :: Map NodeId RaftNodeState -> [(NodeId, CurrentLeader)] -> IO ()+assertLeader raftNodeStates =+ mapM_ (\(nId, leader) -> HUnit.assertBool (show nId ++ " should recognize " ++ show leader ++ " as its leader")+ (maybe False ((== leader) . checkCurrentLeader) (Map.lookup nId raftNodeStates)))++assertCommittedLogIndex :: Map NodeId RaftNodeState -> [(NodeId, Index)] -> IO ()+assertCommittedLogIndex raftNodeStates =+ mapM_ (\(nId, idx) -> HUnit.assertBool (show nId ++ " should have " ++ show idx ++ " as its last committed index")+ (maybe False ((== idx) . getCommittedLogIndex) (Map.lookup nId raftNodeStates)))++assertAppliedLogIndex :: Map NodeId RaftNodeState -> [(NodeId, Index)] -> IO ()+assertAppliedLogIndex raftNodeStates =+ mapM_ (\(nId, idx) -> HUnit.assertBool (show nId ++ " should have " ++ show idx ++ " as its last applied index")+ (maybe False ((== idx) . getLastAppliedLog) (Map.lookup nId raftNodeStates)))++assertPersistedLogs :: Map NodeId (Entries v) -> [(NodeId, Int)] -> IO ()+assertPersistedLogs persistedLogs =+ mapM_ (\(nId, len) -> HUnit.assertBool (show nId ++ " should have appended " ++ show len ++ " logs")+ (maybe False ((== len) . Seq.length) (Map.lookup nId persistedLogs)))++assertSMs :: Map NodeId Store -> [(NodeId, Store)] -> IO ()+assertSMs sms =+ mapM_ (\(nId, sm) -> HUnit.assertBool (show nId ++ " state machine " ++ show sm ++ " is not valid")+ (maybe False (== sm) (Map.lookup nId sms)))
+ test/TestUtils.hs view
@@ -0,0 +1,92 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE GADTs #-}++module TestUtils where++import Protolude+import qualified Data.Set as Set+import qualified Data.Map.Merge.Lazy as Merge++import Raft++isRaftLeader :: RaftNodeState -> Bool+isRaftLeader (RaftNodeState rns) = isLeader rns++isRaftCandidate :: RaftNodeState -> Bool+isRaftCandidate (RaftNodeState rns) = isCandidate rns++isRaftFollower :: RaftNodeState -> Bool+isRaftFollower (RaftNodeState rns) = isFollower rns++checkCurrentLeader :: RaftNodeState -> CurrentLeader+checkCurrentLeader (RaftNodeState (NodeFollowerState FollowerState{..})) = fsCurrentLeader+checkCurrentLeader (RaftNodeState (NodeCandidateState _)) = NoLeader+checkCurrentLeader (RaftNodeState (NodeLeaderState _)) = NoLeader++getLastAppliedLog :: RaftNodeState -> Index+getLastAppliedLog (RaftNodeState (NodeFollowerState FollowerState{..})) = fsLastApplied+getLastAppliedLog (RaftNodeState (NodeCandidateState CandidateState{..})) = csLastApplied+getLastAppliedLog (RaftNodeState (NodeLeaderState LeaderState{..})) = lsLastApplied++getCommittedLogIndex :: RaftNodeState -> Index+getCommittedLogIndex (RaftNodeState (NodeFollowerState FollowerState{..})) = fsCommitIndex+getCommittedLogIndex (RaftNodeState (NodeCandidateState CandidateState{..})) = csCommitIndex+getCommittedLogIndex (RaftNodeState (NodeLeaderState LeaderState{..})) = lsCommitIndex++node0, node1, node2 :: NodeId+node0 = "node0"+node1 = "node1"+node2 = "node2"++client0 :: ClientId+client0 = ClientId "client0"++nodeIds :: NodeIds+nodeIds = Set.fromList [node0, node1, node2]++testConfigs :: [NodeConfig]+testConfigs = [testConfig0, testConfig1, testConfig2]++msToMicroS :: Num n => n -> n+msToMicroS = (1000 *)++pairMsToMicroS :: Num n => (n, n) -> (n, n)+pairMsToMicroS = bimap msToMicroS msToMicroS++testConfig0, testConfig1, testConfig2 :: NodeConfig+testConfig0 = NodeConfig+ { configNodeId = node0+ , configNodeIds = nodeIds+ , configElectionTimeout = pairMsToMicroS (150, 300)+ , configHeartbeatTimeout = msToMicroS 50+ }+testConfig1 = NodeConfig+ { configNodeId = node1+ , configNodeIds = nodeIds+ , configElectionTimeout = pairMsToMicroS (150, 300)+ , configHeartbeatTimeout = msToMicroS 50+ }+testConfig2 = NodeConfig+ { configNodeId = node2+ , configNodeIds = nodeIds+ , configElectionTimeout = pairMsToMicroS (150, 300)+ , configHeartbeatTimeout = msToMicroS 50+ }++-- | Zip maps using function. Throws away items left and right+zipMapWith :: Ord k => (a -> b -> c) -> Map k a -> Map k b -> Map k c+zipMapWith f = Merge.merge Merge.dropMissing Merge.dropMissing (Merge.zipWithMatched (const f))++-- | Perform an inner join on maps (hence throws away items left and right)+combine :: Ord a => Map a b -> Map a c -> Map a (b, c)+combine = zipMapWith (,)++printIfNode :: (Show nId, Eq nId) => nId -> nId -> [Char] -> IO ()+printIfNode nId nId' msg =+ when (nId == nId') $+ print $ show nId ++ " " ++ msg++printIfNodes :: (Show nId, Eq nId) => [nId] -> nId -> [Char] -> IO ()+printIfNodes nIds nId' msg =+ when (nId' `elem` nIds) $+ print $ show nId' ++ " " ++ msg