libraft 0.1.1.0 → 0.2.0.0
raw patch · 33 files changed
+2945/−1590 lines, 33 filesdep +atomic-writedep +base16-bytestringdep +cryptohash-sha256dep ~concurrencydep ~dejafu
Dependencies added: atomic-write, base16-bytestring, cryptohash-sha256, file-embed, lifted-base, monad-control, postgresql-simple, process, quickcheck-state-machine, transformers-base, tree-diff
Dependency ranges changed: concurrency, dejafu
Files
- ChangeLog.md +34/−1
- README.md +82/−28
- app/Main.hs +181/−190
- libraft.cabal +42/−10
- src/Control/Concurrent/STM/Timer.hs +66/−47
- src/Examples/Raft/FileStore.hs +0/−102
- src/Examples/Raft/FileStore/Log.hs +100/−0
- src/Examples/Raft/FileStore/Persistent.hs +81/−0
- src/Examples/Raft/Socket/Client.hs +117/−53
- src/Examples/Raft/Socket/Common.hs +25/−2
- src/Examples/Raft/Socket/Node.hs +32/−82
- src/Raft.hs +178/−80
- src/Raft/Action.hs +2/−1
- src/Raft/Candidate.hs +10/−37
- src/Raft/Client.hs +466/−16
- src/Raft/Config.hs +3/−0
- src/Raft/Event.hs +3/−3
- src/Raft/Follower.hs +12/−15
- src/Raft/Handle.hs +24/−20
- src/Raft/Leader.hs +69/−46
- src/Raft/Log.hs +228/−15
- src/Raft/Log/PostgreSQL.hs +292/−0
- src/Raft/Logging.hs +73/−43
- src/Raft/Monad.hs +45/−52
- src/Raft/NodeState.hs +105/−61
- src/Raft/Persistent.hs +4/−1
- src/Raft/RPC.hs +2/−2
- src/Raft/StateMachine.hs +52/−0
- src/Raft/Types.hs +54/−4
- test/QuickCheckStateMachine.hs +396/−0
- test/TestDejaFu.hs +157/−100
- test/TestRaft.hs +0/−573
- test/TestUtils.hs +10/−6
ChangeLog.md view
@@ -1,3 +1,36 @@ # Changelog for raft -## Unreleased changes+## 0.2.0.0++- Feature: Client requests are now cached by the current leader such that duplicate+ client requests are not serviced+- Bug Fix: Fix issue in concurrent timer not resetting properly when the timeout+ was set to values under 500ms+- Feature: Client read requests can now query entries by index or range of+ indices+- Improvement: Raft nodes now only write to disk if the state changes during+ handling of events+- Feature: Users can now specify what severity of log messages should be logged+- Bug Fix: Receiving all data from a socket in the RaftSocketT monad+ transformer RaftRecvX typeclass instances inducing a deadlock has been+ rewritten+- Feature: A PostgreSQL backend has been added for storage of log entries++## 0.1.2.0++- Added a heartbeat broadcast by leader on read requests from client to ensure+ that the node is leader before responding to client+- Each log entry is now linked by sha256 hash to the log entry immediately+ preceding it+- Fixed bug where the last log entry was not being read from disk before+ starting the main event loop+- Added quickcheck-state-machine tests for black-box testing of the raft-example+ client program++## 0.1.1.0++- Fixed MonadFail constraints for GHC 8.6++## 0.1.0.0++- Initial release.
README.md view
@@ -77,11 +77,11 @@ 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 volatile state of a Raft node+data NodeState (a :: Mode) v where+ NodeFollowerState :: FollowerState v -> NodeState 'Follower v+ NodeCandidateState :: CandidateState v -> NodeState 'Candidate v+ NodeLeaderState :: LeaderState v -> NodeState 'Leader v ``` The library's main event loop is comprised of a simple flow: Raft nodes receive@@ -112,6 +112,7 @@ HigherTermFoundCandidate :: Transition 'Candidate 'Follower BecomeLeader :: Transition 'Candidate 'Leader + HandleClientReq :: Transition 'Leader 'Leader SendHeartbeat :: Transition 'Leader 'Leader DiscoverNewLeader :: Transition 'Leader 'Follower HigherTermFoundLeader :: Transition 'Leader 'Follower@@ -171,14 +172,16 @@ 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`+We ask users of our library to provide an instance of the state machine `RaftStateMachinePure` typeclass. This typeclass relates a state machine type to a command type-and a single type class function 'applyCommittedLogEntry', a pure function that+and a single type class function `applyCmdRaftStateMachinePure`, 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+class RaftStateMachinePure sm v | sm -> v where+ data RaftStateMachinePureError sm v+ type RaftStateMachinePureCtx sm v = ctx | ctx -> sm v+ applyCmdRaftStateMachinePure :: RaftStateMachinePureCtx sm v -> sm -> v -> Either (RaftStateMachinePureError sm v) sm ``` Everything else related to the core event handling loop is not exposed to@@ -253,7 +256,6 @@ ```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@@ -286,23 +288,25 @@ node to which it was sent: ```haskell--- | Provide an interface for nodes to send messages to one+-- | 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+-- | Interface for nodes to receive messages from one -- another class RaftRecvRPC m v where- receiveRPC :: m (RPCMessage v)+ type RaftRecvRPCError m v+ receiveRPC :: m (Either (RaftRecvRPCError m v) (RPCMessage v)) --- | Provide an interface for Raft nodes to send messages to clients-class RaftSendClient m sm where- sendClient :: ClientId -> ClientResponse sm -> m ()+-- | Interface for Raft nodes to send messages to clients+class RaftSendClient m sm v where+ sendClient :: ClientId -> ClientResponse sm v -> m () --- | Provide an interface for Raft nodes to receive messages from clients+-- | Interface for Raft nodes to receive messages from clients class RaftRecvClient m v where- receiveClient :: m (ClientRequest v)+ type RaftRecvClientError m v+ receiveClient :: m (Either (RaftRecvClientError m v) (ClientRequest v)) ``` We have written a default implementation for network sockets over TCP in@@ -321,24 +325,44 @@ 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> ```+ ``` + $ raft-example node <fresh/existing> <file/postgres> <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``` + ```$ stack exec raft-example node fresh file localhost:3001 localhost:3002 localhost:3003```+ - On terminal 2:- ```$ stack exec raft-example localhost:3002 localhost:3001 localhost:3003``` + ```$ stack exec raft-example node fresh file localhost:3002 localhost:3001 localhost:3003```+ - On terminal 3:- ```$ stack exec raft-example localhost:3003 localhost:3001 localhost:3002``` + ```$ stack exec raft-example node fresh file 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. + **Note:** If you want to run a raft example node with _existing_ persistent data,+ pass the `existing` command line option to the `raft-example` program instead+ of `fresh`:+ + ```$ stack exec raft-example node existing ...```+++ **Note:** The example also runs using a PostgreSQL database as long as a+ user 'libraft_test' with password 'libraft_test' exists in your local+ postgresql installation. For ease of experimentation, if such a user does+ not exist, create it like so:++ ```$ sudo -su postgres psql -U postgres -c "CREATE USER libraft_test WITH CREATEDB PASSWORD 'libraft_test';"```+ 3) Run a client: ```$ stack exec raft-example client``` @@ -382,12 +406,18 @@ ### Define the state machine -The only requirement for our state machine is to instantiate the `StateMachine`+The only requirement for our state machine is to instantiate the state machine `RaftStateMachinePure` type class. ```haskell-class StateMachine sm v | sm -> v where- applyCommittedLogEntry :: sm -> v -> sm+-- | 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 RaftStateMachinePure sm v | sm -> v where+ data RaftStateMachinePureError sm v+ type RaftStateMachinePureCtx sm v = ctx | ctx -> sm v+ rsmTransition :: RaftStateMachinePureCtx sm v -> sm -> v -> Either (RaftStateMachinePureError sm v) sm+ ``` In our [example](https://github.com/adjoint-io/raft/blob/master/app/Main.hs) we@@ -433,11 +463,22 @@ 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`.+delete log entries, as well as `RaftPersist`. There are actually several data+that must be stored on disk; 1) the data the raft paper calls "persistent data"+and 2) the log entries of the node. -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)+We provide an implementation that stores the persistent data in a file in+[src/Examples/Raft/FileStore/Persistent.hs](https://github.com/adjoint-io/raft/blob/master/src/Examples/Raft/FileStore/Persistent.hs) +An example of storing log entries in a single file (for ease of implementation+in lieu of good performance for reads/writes) can be found in+[src/Examples/Raft/FileStore/Log.hs](https://github.com/adjoint-io/raft/blob/master/src/Examples/Raft/FileStore/Log.hs)++Lastly, a more "production ready" example of log entry storage using a+PostgreSQL database can be found in [src/Raft/Log/PostgreSQL.hs](https://github.com/adjoint-io/raft/blob/master/src/Raft/Log/PostgreSQL.hs).+This implementation is used in our `quickcheck-state-machine` model testing+module and is thus thoroughly tested. + ### Putting it all together The last step is wrapping our previous data types that deal with@@ -454,6 +495,19 @@ [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.++# Test suite dependencies++The test suite depends on libfiu (commonly installed with package `fiu-utils`), +which it uses to simulate network failures. In addition, the test suite also+depends on libpq-dev and postgresql. Furthermore, in order to successfully run+the model tests (which will run autmatically when executing `stack test`, but+fail immediately with "Failed to spawn node), you will have to run the following+command:++```+$ sudo -su postgres psql -U postgres -c "CREATE USER libraft_test WITH CREATEDB PASSWORD 'libraft_test';" +``` # References
app/Main.hs view
@@ -25,30 +25,31 @@ 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 Raft+import Raft.Config+import Raft.Log+import Raft.Log.PostgreSQL+import Raft.Client++import Database.PostgreSQL.Simple+ 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+import Examples.Raft.FileStore.Log+import Examples.Raft.FileStore.Persistent ------------------------------ -- State Machine & Commands --@@ -68,18 +69,18 @@ type Store = Map Var Natural -instance RSMP Store StoreCmd where- data RSMPError Store StoreCmd = StoreError Text deriving (Show)- type RSMPCtx Store StoreCmd = ()+instance RaftStateMachinePure Store StoreCmd where+ data RaftStateMachinePureError Store StoreCmd = StoreError Text deriving (Show)+ type RaftStateMachinePureCtx Store StoreCmd = () - applyCmdRSMP _ store cmd =+ rsmTransition _ 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+instance (Monad m, sm ~ Store, v ~ StoreCmd, RaftStateMachinePure sm v) => RaftStateMachine (RaftExampleM m sm v) sm v where validateCmd _ = pure (Right ())- askRSMPCtx = pure ()+ askRaftStateMachinePureCtx = pure () -------------------- -- Raft instances --@@ -90,53 +91,73 @@ , nEnvNodeId :: NodeId } -newtype RaftExampleM sm v a = RaftExampleM { unRaftExampleM :: ReaderT (NodeEnv sm) (RaftSocketT v (RaftFileStoreT IO)) a }- deriving (Functor, Applicative, Monad, MonadIO, MonadFail, MonadReader (NodeEnv sm), Alternative, MonadPlus)+newtype RaftExampleM m sm v a = RaftExampleM {+ unRaftExampleM :: ReaderT (NodeEnv sm) (RaftSocketT v (RaftPersistFileStoreT m)) a+ } -deriving instance MonadThrow (RaftExampleM sm v)-deriving instance MonadCatch (RaftExampleM sm v)-deriving instance MonadMask (RaftExampleM sm v)-deriving instance MonadConc (RaftExampleM sm v)+deriving instance Functor m => Functor (RaftExampleM m sm v)+deriving instance Applicative m => Applicative (RaftExampleM m sm v)+deriving instance Monad m => Monad (RaftExampleM m sm v)+deriving instance MonadIO m => MonadIO (RaftExampleM m sm v)+deriving instance MonadFail m => MonadFail (RaftExampleM m sm v)+deriving instance Monad m => MonadReader (NodeEnv sm) (RaftExampleM m sm v)+deriving instance Alternative m => Alternative (RaftExampleM m sm v)+deriving instance MonadPlus m => MonadPlus (RaftExampleM m 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+deriving instance MonadThrow m => MonadThrow (RaftExampleM m sm v)+deriving instance MonadCatch m => MonadCatch (RaftExampleM m sm v)+deriving instance MonadMask m => MonadMask (RaftExampleM m sm v)+deriving instance MonadConc m => MonadConc (RaftExampleM m sm v) -instance RaftSendClient (RaftExampleM Store StoreCmd) Store where+runRaftExampleM+ :: (MonadIO m, MonadConc m)+ => NodeEnv sm+ -> NodeSocketEnv v+ -> RaftPersistFile+ -> RaftExampleM m sm v a+ -> m a+runRaftExampleM nodeEnv nodeSocketEnv raftPersistFile raftExampleM =+ flip runReaderT raftPersistFile . unRaftPersistFileStoreT $+ flip runReaderT nodeSocketEnv . unRaftSocketT $+ flip runReaderT nodeEnv $ unRaftExampleM raftExampleM++instance (MonadIO m, MonadConc m) => RaftSendClient (RaftExampleM m Store StoreCmd) Store StoreCmd where sendClient cid msg = (RaftExampleM . lift) $ sendClient cid msg -instance RaftRecvClient (RaftExampleM Store StoreCmd) StoreCmd where- type RaftRecvClientError (RaftExampleM Store StoreCmd) StoreCmd = Text+instance (MonadIO m, MonadConc m) => RaftRecvClient (RaftExampleM m Store StoreCmd) StoreCmd where+ type RaftRecvClientError (RaftExampleM m Store StoreCmd) StoreCmd = Text receiveClient = RaftExampleM $ lift receiveClient -instance RaftSendRPC (RaftExampleM Store StoreCmd) StoreCmd where+instance (MonadIO m, MonadConc m) => RaftSendRPC (RaftExampleM m 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+instance (MonadIO m, MonadConc m) => RaftRecvRPC (RaftExampleM m Store StoreCmd) StoreCmd where+ type RaftRecvRPCError (RaftExampleM m 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 (MonadIO m, MonadConc m) => RaftPersist (RaftExampleM m Store StoreCmd) where+ type RaftPersistError (RaftExampleM m Store StoreCmd) = RaftPersistFileStoreError+ initializePersistentState = RaftExampleM $ lift $ lift initializePersistentState+ writePersistentState ps = RaftExampleM $ lift $ lift $ writePersistentState ps+ readPersistentState = RaftExampleM $ lift $ lift $ readPersistentState -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 (MonadConc m, RaftInitLog m StoreCmd) => RaftInitLog (RaftExampleM m Store StoreCmd) StoreCmd where+ type RaftInitLogError (RaftExampleM m Store StoreCmd) = RaftInitLogError m+ initializeLog p = RaftExampleM $ lift $ lift $ lift $ initializeLog p -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 RaftWriteLog m StoreCmd => RaftWriteLog (RaftExampleM m Store StoreCmd) StoreCmd where+ type RaftWriteLogError (RaftExampleM m Store StoreCmd) = RaftWriteLogError m+ writeLogEntries entries = RaftExampleM $ lift $ lift $ lift $ writeLogEntries entries -instance RaftDeleteLog (RaftExampleM Store StoreCmd) StoreCmd where- type RaftDeleteLogError (RaftExampleM Store StoreCmd) = NodeEnvError- deleteLogEntriesFrom idx = RaftExampleM $ lift $ RaftSocketT (lift $ deleteLogEntriesFrom idx)+instance RaftReadLog m StoreCmd => RaftReadLog (RaftExampleM m Store StoreCmd) StoreCmd where+ type RaftReadLogError (RaftExampleM m Store StoreCmd) = RaftReadLogError m+ readLogEntry idx = RaftExampleM $ lift $ lift $ lift $ readLogEntry idx+ readLastLogEntry = RaftExampleM $ lift $ lift $ lift $ readLastLogEntry +instance RaftDeleteLog m StoreCmd => RaftDeleteLog (RaftExampleM m Store StoreCmd) StoreCmd where+ type RaftDeleteLogError (RaftExampleM m Store StoreCmd) = RaftDeleteLogError m+ deleteLogEntriesFrom idx = RaftExampleM $ lift $ lift $ lift $ deleteLogEntriesFrom idx+ -------------------- -- Client console -- --------------------@@ -154,138 +175,126 @@ -- - 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)+ { unConsoleM :: HaskelineT (RS.RaftSocketClientM Store StoreCmd) a+ } deriving (Functor, Applicative, Monad, MonadIO) -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'+liftRSCM = ConsoleM . lift -- | 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+ nids <- liftRSCM clientGetNodes 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+ ["addNode", nid] -> liftRSCM $ clientAddNode (toS nid)+ ["getNodes"] -> print =<< liftRSCM clientGetNodes+ ["read"] ->+ ifNodesAdded nids $+ handleResponse =<< liftRSCM (RS.socketClientRead ClientReadStateMachine)+ ["read", n] ->+ ifNodesAdded nids $+ handleResponse =<< liftRSCM (RS.socketClientRead (ClientReadEntries (ByIndex (Index (read n)))))+ ["read", "[", low, high, "]" ] ->+ ifNodesAdded nids $ do+ let byInterval = ByIndices $ IndexInterval (Just (Index (read low))) (Just (Index (read high)))+ handleResponse =<< liftRSCM (RS.socketClientRead (ClientReadEntries byInterval))+ ["incr", cmd] ->+ ifNodesAdded nids $+ handleResponse =<< liftRSCM (RS.socketClientWrite (Incr (toS cmd)))+ ["set", var, val] ->+ ifNodesAdded nids $+ handleResponse =<< liftRSCM (RS.socketClientWrite (Set (toS var) (read val))) _ -> 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+ ifNodesAdded nids m+ | nids == Set.empty =+ putText "Please add some nodes to query first. Eg. `addNode localhost:3001`"+ | otherwise = m + handleResponse :: Show a => Either Text a -> ConsoleM ()+ handleResponse res = do+ case res of+ Left err -> liftIO $ putText err+ Right resp -> liftIO $ putText (show resp) +data LogStorage = FileStore | PostgreSQL [Char]+ main :: IO () main = do args <- (toS <$>) <$> getArgs case args of- ["client"] -> clientMainHandler- (nid:nids) -> do- removeExampleFiles nid- createExampleFiles nid-+ ["client"] -> clientMain+ ("node":"fresh":"file":nid:nids) -> initNode New FileStore (nid:nids)+ ("node":"existing":"file":nid:nids) -> initNode Existing FileStore (nid:nids)+ ("node":"fresh":"postgres":nm:nid:nids) -> initNode New (PostgreSQL $ toS nm) (nid:nids)+ ("node":"existing":"postgres":nm:nid:nids) -> initNode Existing (PostgreSQL $ toS nm) (nid:nids)+ where+ initNode storageState storageType (nid:nids) = do+ nodeDir <- mkExampleDir nid+ case storageState of+ New -> cleanStorage nodeDir storageType+ Existing -> pure () 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)+ nPersistFile <- RaftPersistFile <$> persistentFilePath nid+ case storageType of+ FileStore -> do+ nLogsFile <- RaftLogFile <$> logsFilePath nid+ runRaftLogFileStoreT nLogsFile $+ runRaftNode' nSocketEnv nEnv nPersistFile+ PostgreSQL dbName -> do+ let pgConnInfo = raftDatabaseConnInfo "libraft_test" "libraft_test" dbName+ runRaftPostgresM pgConnInfo $+ runRaftNode' nSocketEnv nEnv nPersistFile+ where+ runRaftNode'+ :: ( MonadIO m, MonadConc m, MonadFail m+ , RaftInitLog m StoreCmd, RaftReadLog m StoreCmd, RaftWriteLog m StoreCmd+ , RaftDeleteLog m StoreCmd, Exception (RaftInitLogError m), Exception (RaftReadLogError m)+ , Exception (RaftWriteLogError m), Exception (RaftDeleteLogError m), Typeable m+ )+ => NodeSocketEnv StoreCmd+ -> NodeEnv Store+ -> RaftPersistFile+ -> m ()+ runRaftNode' nSocketEnv nEnv nPersistFile =+ runRaftExampleM nEnv nSocketEnv nPersistFile $ do+ let allNodeIds = Set.fromList (nid : nids)+ let (host, port) = RS.nidToHostPort (toS nid)+ let nodeConfig = NodeConfig+ { configNodeId = toS nid+ , configNodeIds = allNodeIds+ -- These are recommended timeouts from the original+ -- raft paper and the ARC report.+ , configElectionTimeout = (150000, 300000)+ , configHeartbeatTimeout = 50000+ , configStorageState = storageState+ }+ fork $ RaftExampleM $ lift (acceptConnections host port)+ electionTimerSeed <- liftIO randomIO+ runRaftNode nodeConfig (LogCtx LogStdout Debug) electionTimerSeed (mempty :: Store) - persistentFile :: NodeId -> IO FilePath- persistentFile nid = do+ cleanStorage :: FilePath -> LogStorage -> IO ()+ cleanStorage nodeDir ls = do+ case ls of+ PostgreSQL dbName -> do+ Control.Monad.Catch.bracket (connect initConnInfo) close $ \conn ->+ void $ deleteDB (raftDatabaseName dbName) conn+ _ -> pure ()+ Directory.removePathForcibly nodeDir+ Directory.createDirectoryIfMissing False nodeDir++ persistentFilePath :: NodeId -> IO FilePath+ persistentFilePath 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+ logsFilePath :: NodeId -> IO FilePath+ logsFilePath 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)@@ -295,48 +304,30 @@ , 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- }+ { nsMsgQueue = msgQueue+ , nsClientReqQueue = clientReqQueue+ } - clientMainHandler :: IO ()- clientMainHandler = do+ mkExampleDir :: NodeId -> IO FilePath+ mkExampleDir nid = do+ tmpDir <- Directory.getTemporaryDirectory+ let nodeDir = tmpDir ++ "/" ++ toS nid+ pure nodeDir++ clientMain :: IO ()+ clientMain = do+ let clientHost = "localhost" 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 $+ let clientId = ClientId $ RS.hostPortToNid (clientHost, clientPort)+ clientRespChan <- RS.newClientRespChan+ RS.runRaftSocketClientM clientId mempty clientRespChan $ do+ fork (lift (RS.clientResponseServer clientHost clientPort)) 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
libraft.cabal view
@@ -2,11 +2,13 @@ -- -- see: https://github.com/sol/hpack ----- hash: 0e9cecc332c8c5b872a193fa8e1bcf55c994bce6217b0d17f13894079c45af32+-- hash: 2d373c5064dc13cca463f24496ffaca3207f2659d824261e3338e264da11247e name: libraft-version: 0.1.1.0+version: 0.2.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.@@ -27,7 +29,8 @@ library exposed-modules: Control.Concurrent.STM.Timer- Examples.Raft.FileStore+ Examples.Raft.FileStore.Log+ Examples.Raft.FileStore.Persistent Examples.Raft.Socket.Client Examples.Raft.Socket.Common Examples.Raft.Socket.Node@@ -41,11 +44,13 @@ Raft.Handle Raft.Leader Raft.Log+ Raft.Log.PostgreSQL Raft.Logging Raft.Monad Raft.NodeState Raft.Persistent Raft.RPC+ Raft.StateMachine Raft.Types other-modules: Paths_libraft@@ -54,25 +59,33 @@ default-extensions: NoImplicitPrelude OverloadedStrings LambdaCase ghc-options: -fwarn-unused-binds -fwarn-unused-imports build-depends:- attoparsec+ atomic-write+ , attoparsec , base >=4.7 && <5+ , base16-bytestring , bytestring , cereal- , concurrency+ , concurrency >=1.3.0.0 && <1.7.0.0 , containers+ , cryptohash-sha256 , directory , exceptions+ , file-embed , haskeline+ , lifted-base+ , monad-control , mtl , network , network-simple , parsec+ , postgresql-simple , protolude , random , repline , text , time , transformers+ , transformers-base , word8 default-language: Haskell2010 @@ -85,20 +98,27 @@ default-extensions: NoImplicitPrelude OverloadedStrings LambdaCase ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends:- attoparsec+ atomic-write+ , attoparsec , base >=4.7 && <5+ , base16-bytestring , bytestring , cereal- , concurrency+ , concurrency >=1.3.0.0 && <1.7.0.0 , containers+ , cryptohash-sha256 , directory , exceptions+ , file-embed , haskeline , libraft+ , lifted-base+ , monad-control , mtl , network , network-simple , parsec+ , postgresql-simple , protolude , random , repline@@ -106,6 +126,7 @@ , text , time , transformers+ , transformers-base , word8 default-language: Haskell2010 @@ -113,8 +134,8 @@ type: exitcode-stdio-1.0 main-is: TestDriver.hs other-modules:+ QuickCheckStateMachine TestDejaFu- TestRaft TestUtils Paths_libraft hs-source-dirs:@@ -123,23 +144,32 @@ ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: QuickCheck+ , atomic-write , attoparsec , base >=4.7 && <5+ , base16-bytestring , bytestring , cereal- , concurrency+ , concurrency >=1.3.0.0 && <1.7.0.0 , containers- , dejafu+ , cryptohash-sha256+ , dejafu >=1.0.0.0 && <2.0.0.0 , directory , exceptions+ , file-embed , haskeline , hunit-dejafu , libraft+ , lifted-base+ , monad-control , mtl , network , network-simple , parsec+ , postgresql-simple+ , process >=1.6.3.0 && <1.7.0.0 , protolude+ , quickcheck-state-machine , random , repline , tasty@@ -151,5 +181,7 @@ , text , time , transformers+ , transformers-base+ , tree-diff , word8 default-language: Haskell2010
src/Control/Concurrent/STM/Timer.hs view
@@ -1,75 +1,94 @@ module Control.Concurrent.STM.Timer ( Timer,- waitTimer,- startTimer,- resetTimer, newTimer, newTimerRange,+ startTimer,+ resetTimer,+ waitTimer, ) where -import Protolude hiding (STM, killThread, ThreadId, threadDelay, myThreadId, atomically)+import Protolude hiding (wait, async, withAsync, cancel, Async, STM, killThread, ThreadId, threadDelay, myThreadId, atomically) import Control.Monad.Conc.Class import Control.Concurrent.Classy.STM+import Control.Concurrent.Classy.Async import System.Random (StdGen, randomR, mkStdGen) import Numeric.Natural data Timer m = Timer- { timerThread :: TMVar (STM m) (ThreadId m)+ { timerAsync :: TMVar (STM m) (Async m ())+ -- ^ The async computation of the timer , timerLock :: TMVar (STM m) ()- , timerRange :: (Natural, Natural)+ -- ^ When the TMVar is empty, the timer is being used , timerGen :: TVar (STM m) StdGen+ , timerRange :: (Natural, Natural)+ , timerAction :: m () } -waitTimer :: MonadConc m => Timer m -> m ()-waitTimer (Timer _ lock _ _) =- atomically $ readTMVar lock+-- | Create a new timer with the supplied timer action and timer length,+newTimer :: MonadConc m => m () -> Natural -> m (Timer m)+newTimer action timeout = newTimerRange action 0 (timeout, timeout) --- | Starting a timer will only work if the timer is currently stopped-startTimer :: MonadConc m => Timer m -> m ()+-- | Create a new timer with the supplied timer action, random seed, and range+-- from which the the timer will choose a random timer length at each+-- start or reset.+newTimerRange :: MonadConc m => m () -> Int -> (Natural, Natural) -> m (Timer m)+newTimerRange action seed timeoutRange = do+ (timerAsync, timerLock, timerGen) <-+ atomically $ (,,) <$> newEmptyTMVar <*> newTMVar () <*> newTVar (mkStdGen seed)+ pure $ Timer timerAsync timerLock timerGen timeoutRange action++--------------------------------------------------------------------------------++-- | Start the timer. If the timer is already running, the timer is not started.+-- Returns True if the timer was succesfully started.+startTimer :: MonadConc m => Timer m -> m Bool startTimer timer = do- lock <- atomically $ tryTakeTMVar (timerLock timer)- case lock of- Nothing -> pure ()- Just () -> spawnTimer timer+ mlock <- atomically $ tryTakeTMVar (timerLock timer)+ case mlock of+ Nothing -> pure False+ Just () -> resetTimer timer >> pure True +-- | Resets the timer with a new random timeout. 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++ -- Check if a timer is already running. If it is, asynchronously kill the+ -- thread.+ mta <- atomically $ tryTakeTMVar (timerAsync timer)+ case mta of Nothing -> pure ()- -- Spawn a new timer- spawnTimer timer+ Just ta -> void $ async (uninterruptibleCancel ta) --- | 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) ()+ -- Fork a new async computation that waits the specified (random) amount of+ -- time, performs the timer action, and then puts the lock back signaling the+ -- timer finishing.+ ta <- async $ do+ threadDelay =<< randomDelay timer+ timerAction timer+ success <- atomically $ tryPutTMVar (timerLock timer) ()+ when (not success) $+ panic "[Failed Invariant]: Putting the timer lock back should succeed" -newTimer :: MonadConc m => Natural -> m (Timer m)-newTimer timeout = newTimerRange 0 (timeout, timeout)+ -- Check that putting the new async succeeded. If it did not, there is a race+ -- condition and the newly created async should be canceled. Warning: This may+ -- not work for _very_ short timers.+ success <- atomically $ tryPutTMVar (timerAsync timer) ta+ when (not success) $+ void $ async (uninterruptibleCancel ta) --- | 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+-- | Wait for a timer to complete+waitTimer :: MonadConc m => Timer m -> m ()+waitTimer timer = atomically $ readTMVar (timerLock timer)++--------------------------------------------------------------------------------++randomDelay :: MonadConc m => Timer m -> m Int+randomDelay timer = atomically $ do+ g <- readTVar (timerGen timer)+ let (tmin, tmax) = timerRange timer+ (n, g') = randomR (toInteger tmin, toInteger tmax) g+ writeTVar (timerGen timer) g'+ pure (fromIntegral n)
− src/Examples/Raft/FileStore.hs
@@ -1,102 +0,0 @@-{-# 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.Fail-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, MonadFail, 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/FileStore/Log.hs view
@@ -0,0 +1,100 @@+{-# 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.FileStore.Log where++import Protolude hiding (try)++import Control.Concurrent.Classy hiding (catch, ThreadId)+import Control.Monad.Fail+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 qualified System.AtomicWrite.Writer.ByteString as AW++import Raft.Log+import Raft.Types++newtype RaftLogFileStoreError = RaftLogFileStoreError Text+ deriving (Show)++newtype RaftLogFile = RaftLogFile { unRaftLogFile :: FilePath }++instance Exception RaftLogFileStoreError++newtype RaftLogFileStoreT m a = RaftLogFileStoreT { unRaftLogFileStoreT :: ReaderT RaftLogFile m a }+ deriving (Functor, Applicative, Monad, MonadIO, MonadFail, MonadReader RaftLogFile, Alternative, MonadPlus, MonadTrans)++deriving instance MonadThrow m => MonadThrow (RaftLogFileStoreT m)+deriving instance MonadCatch m => MonadCatch (RaftLogFileStoreT m)+deriving instance MonadMask m => MonadMask (RaftLogFileStoreT m)+deriving instance MonadConc m => MonadConc (RaftLogFileStoreT m)++runRaftLogFileStoreT :: RaftLogFile -> RaftLogFileStoreT m a -> m a+runRaftLogFileStoreT rlogFile = flip runReaderT rlogFile . unRaftLogFileStoreT++instance (S.Serialize v, MonadIO m, MonadConc m) => RaftInitLog (RaftLogFileStoreT m) v where+ type RaftInitLogError (RaftLogFileStoreT m) = RaftLogFileStoreError+ initializeLog _ = do+ RaftLogFile logFile <- ask+ eRes <- liftIO $ try (BS.writeFile logFile (S.encode (mempty :: Entries v)))+ case eRes of+ Left (e :: SomeException) -> pure $ Left (RaftLogFileStoreError (show e))+ Right _ -> pure $ Right ()++instance (MonadIO m, MonadConc m, S.Serialize v) => RaftWriteLog (RaftLogFileStoreT m) v where+ type RaftWriteLogError (RaftLogFileStoreT m) = RaftLogFileStoreError+ writeLogEntries newEntries = do+ RaftLogFile logFile <- ask+ eLogEntries <- readLogEntries+ case eLogEntries of+ Left err -> panic ("writeLogEntries: " <> err)+ Right currEntries -> liftIO $ Right <$> AW.atomicWriteFile logFile (S.encode (currEntries >< newEntries))++instance (MonadIO m, MonadConc m, S.Serialize v) => RaftReadLog (RaftLogFileStoreT m) v where+ type RaftReadLogError (RaftLogFileStoreT m) = RaftLogFileStoreError+ 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 ("readLastLogEntry: " <> 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 (RaftLogFileStoreT m) v where+ type RaftDeleteLogError (RaftLogFileStoreT m) = RaftLogFileStoreError+ deleteLogEntriesFrom idx = do+ eLogEntries <- readLogEntries+ case eLogEntries of+ Left err -> panic ("deleteLogEntriesFrom: " <> err)+ Right (entries :: Entries v) -> do+ let newLogEntries = Seq.dropWhileR ((>= idx) . entryIndex) entries+ RaftLogFile logFile <- ask+ liftIO $ AW.atomicWriteFile logFile (S.encode newLogEntries)+ pure (Right DeleteSuccess)++readLogEntries :: (MonadIO m, S.Serialize v) => RaftLogFileStoreT m (Either Text (Entries v))+readLogEntries = liftIO . fmap (first toS . S.decode) . BS.readFile . unRaftLogFile =<< ask
+ src/Examples/Raft/FileStore/Persistent.hs view
@@ -0,0 +1,81 @@+{-# 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.FileStore.Persistent where++import Protolude hiding (try)++import Control.Concurrent.Classy hiding (catch, ThreadId)+import Control.Monad.Fail+import Control.Monad.Catch+import Control.Monad.Trans.Class++import qualified Data.ByteString as BS+import qualified Data.Serialize as S++import System.Directory (doesFileExist)+import qualified System.AtomicWrite.Writer.ByteString as AW++import Raft.Persistent++newtype RaftPersistFileStoreError = RaftPersistFileStoreError Text+ deriving (Show)++newtype RaftPersistFile = RaftPersistFile FilePath++instance Exception RaftPersistFileStoreError++newtype RaftPersistFileStoreT m a = RaftPersistFileStoreT { unRaftPersistFileStoreT :: ReaderT RaftPersistFile m a }+ deriving (Functor, Applicative, Monad, MonadIO, MonadFail, MonadReader RaftPersistFile, Alternative, MonadPlus, MonadTrans)++deriving instance MonadThrow m => MonadThrow (RaftPersistFileStoreT m)+deriving instance MonadCatch m => MonadCatch (RaftPersistFileStoreT m)+deriving instance MonadMask m => MonadMask (RaftPersistFileStoreT m)+deriving instance MonadConc m => MonadConc (RaftPersistFileStoreT m)++--------------------------------------------------------------------------------+-- Raft Instances+--------------------------------------------------------------------------------++-- A RaftPersist instance for the RaftPersistFileStoreT monad.+--+-- Warning: `atomicWriteFile` causes large pauses on the order of hundreds of+-- milliseconds. This can cause leadership to change since the node become+-- unresponsive for much longer than the recommended election timeouts.+instance (MonadIO m, MonadConc m) => RaftPersist (RaftPersistFileStoreT m) where+ type RaftPersistError (RaftPersistFileStoreT m) = RaftPersistFileStoreError++ initializePersistentState = do+ RaftPersistFile psFile <- ask+ fileExists <- liftIO $ doesFileExist psFile+ if fileExists+ then pure $ Left (RaftPersistFileStoreError ("Persistent file " <> toS psFile <> " already exists!"))+ else do+ eRes <-+ liftIO . try $+ AW.atomicWriteFile psFile (toS (S.encode initPersistentState))+ case eRes of+ Left (err :: SomeException) -> do+ let errPrefix = "Failed initializing persistent storage: "+ pure (Left (RaftPersistFileStoreError (errPrefix <> show err)))+ Right _ -> pure (Right ())++ writePersistentState ps = do+ RaftPersistFile psFile <- ask+ liftIO $ Right <$> AW.atomicWriteFile psFile (S.encode ps)++ readPersistentState = do+ RaftPersistFile psFile <- ask+ fileContent <- liftIO $ BS.readFile psFile+ case S.decode fileContent of+ Left err -> panic (toS $ "readPersistentState: " ++ err)+ Right ps -> pure $ Right ps
src/Examples/Raft/Socket/Client.hs view
@@ -2,16 +2,26 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE TypeFamilies #-} module Examples.Raft.Socket.Client where -import Protolude+import Protolude hiding (STM, atomically, try, ThreadId, newTChan) -import Control.Concurrent.Classy+import Control.Concurrent.Classy hiding (catch)+import Control.Monad.Base+import Control.Monad.Catch+import Control.Monad.Trans.Class+import Control.Monad.Trans.Control+ 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@@ -19,61 +29,115 @@ import Raft.Types import Examples.Raft.Socket.Common -data ClientSocketEnv- = ClientSocketEnv { clientPort :: N.ServiceName- , clientHost :: N.HostName- , clientSocket :: N.Socket- } deriving (Show)+import System.Timeout.Lifted (timeout)+import System.Console.Haskeline.MonadException (MonadException(..), RunIO(..)) -newtype RaftSocketClientM a- = RaftSocketClientM { unRaftSocketClientM :: ReaderT ClientSocketEnv IO a }- deriving (Functor, Applicative, Monad, MonadIO, MonadReader ClientSocketEnv, Alternative, MonadPlus)+newtype ClientRespChan s v+ = ClientRespChan { clientRespChan :: TChan (STM IO) (ClientResponse s v) } -runRaftSocketClientM :: ClientSocketEnv -> RaftSocketClientM a -> IO a-runRaftSocketClientM socketEnv = flip runReaderT socketEnv . unRaftSocketClientM+newClientRespChan :: IO (ClientRespChan s v)+newClientRespChan = ClientRespChan <$> atomically newTChan --- | 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)+newtype RaftClientRespChanT s v m a+ = RaftClientRespChanT { unRaftClientRespChanT :: ReaderT (ClientRespChan s v) m a }+ deriving newtype (Functor, Applicative, Monad, MonadIO, MonadReader (ClientRespChan s v), Alternative, MonadPlus) --- | 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+instance MonadTrans (RaftClientRespChanT s v) where+ lift = RaftClientRespChanT . lift --- | 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+deriving instance MonadBase IO m => MonadBase IO (RaftClientRespChanT s v m) --- | 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+instance MonadTransControl (RaftClientRespChanT s v) where+ type StT (RaftClientRespChanT s v) a = StT (ReaderT (ClientRespChan s v)) a+ liftWith = defaultLiftWith RaftClientRespChanT unRaftClientRespChanT+ restoreT = defaultRestoreT RaftClientRespChanT --- | 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+instance MonadBaseControl IO m => MonadBaseControl IO (RaftClientRespChanT s v m) where+ type StM (RaftClientRespChanT s v m) a = ComposeSt (RaftClientRespChanT s v) m a+ liftBaseWith = defaultLiftBaseWith+ restoreM = defaultRestoreM --- | 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)+deriving instance MonadSTM m => MonadSTM (RaftClientRespChanT s v m)+deriving instance MonadThrow m => MonadThrow (RaftClientRespChanT s v m)+deriving instance MonadCatch m => MonadCatch (RaftClientRespChanT s v m)+deriving instance MonadMask m => MonadMask (RaftClientRespChanT s v m)+deriving instance MonadConc m => MonadConc (RaftClientRespChanT s v m) +-- This annoying instance is because of the Haskeline library, letting us use a+-- custom monad transformer stack as the base monad of 'InputT'. IMO it should+-- be automatically derivable. Why does haskeline use a custom exception+-- monad... ?+instance MonadException m => MonadException (RaftClientRespChanT s v m) where+ controlIO f =+ RaftClientRespChanT $ ReaderT $ \r ->+ controlIO $ \(RunIO run) ->+ let run' = RunIO (fmap (RaftClientRespChanT . ReaderT . const) . run . flip runReaderT r . unRaftClientRespChanT)+ in fmap (flip runReaderT r . unRaftClientRespChanT) $ f run'++instance (S.Serialize v, MonadIO m) => RaftClientSend (RaftClientRespChanT s v m) v where+ type RaftClientSendError (RaftClientRespChanT s v m) v = Text+ raftClientSend nid creq = do+ let (host,port) = nidToHostPort nid+ mRes <-+ liftIO $ timeout 100000 $ try $ do+ -- Warning: blocks if socket is allocated by OS, even though the socket+ -- may have been closed by the running process+ N.connect host port $ \(sock, sockAddr) ->+ N.send sock (S.encode (ClientRequestEvent creq))+ let errPrefix = "Failed to send client request: "+ case mRes of+ Nothing -> pure (Left (errPrefix <> "'connect' timed out"))+ Just (Left (err :: SomeException)) -> pure $ Left (errPrefix <> show err)+ Just (Right _) -> pure (Right ())++instance (S.Serialize s, S.Serialize v, MonadIO m) => RaftClientRecv (RaftClientRespChanT s v m) s v where+ type RaftClientRecvError (RaftClientRespChanT s v m) s = Text+ raftClientRecv = do+ respChan <- asks clientRespChan+ fmap Right . liftIO . atomically $ readTChan respChan++--------------------------------------------------------------------------------++type RaftSocketClientM s v = RaftClientT s v (RaftClientRespChanT s v IO)++runRaftSocketClientM+ :: ClientId+ -> Set NodeId+ -> ClientRespChan s v+ -> RaftSocketClientM s v a+ -> IO a+runRaftSocketClientM cid nids respChan rscm = do+ raftClientState <- initRaftClientState nids <$> liftIO newStdGen+ let raftClientEnv = RaftClientEnv cid+ flip runReaderT respChan+ . unRaftClientRespChanT+ . runRaftClientT raftClientEnv raftClientState+ $ rscm++clientResponseServer+ :: forall s v m. (S.Serialize s, S.Serialize v, MonadIO m, MonadConc m)+ => N.HostName+ -> N.ServiceName+ -> RaftClientRespChanT s v m ()+clientResponseServer host port = do+ respChan <- asks clientRespChan+ N.serve (N.Host host) port $ \(sock, _) -> do+ mBytes <- N.recv sock (4 * 4096)+ case mBytes of+ Nothing -> putText "Socket was closed on the other end"+ Just bytes -> case S.decode bytes of+ Left err -> putText $ "Failed to decode message: " <> toS err+ Right cresp -> atomically $ writeTChan respChan cresp++-- | Send a client read request using the example socket interface of RaftSocketClientM+socketClientRead+ :: (S.Serialize s, S.Serialize v, Show s, Show v, Show (RaftClientError s v (RaftSocketClientM s v)))+ => ClientReadReq+ -> RaftSocketClientM s v (Either Text (ClientReadResp s v))+socketClientRead rr = first show <$> retryOnRedirect (clientReadTimeout 1000000 rr)++socketClientWrite+ :: (S.Serialize s, S.Serialize v, Show s, Show v, Show (RaftClientError s v (RaftSocketClientM s v)))+ => v+ -> RaftSocketClientM s v (Either Text ClientWriteResp)+socketClientWrite v = first show <$> retryOnRedirect (clientWriteTimeout 1000000 v)
src/Examples/Raft/Socket/Common.hs view
@@ -3,16 +3,22 @@ import Protolude import qualified Data.ByteString as BS+import qualified Data.Serialize as S+import qualified Data.Serialize as Get+import qualified Data.Word8 as W8+ 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+hostPortToNid = toS . hostPortToNidBS +hostPortToNidBS :: (N.HostName, N.ServiceName) -> ByteString+hostPortToNidBS (host, port) = toS $ host ++ ":" ++ toS port+ -- | Retrieve the host and port from a valid NodeId nidToHostPort :: NodeId -> (N.HostName, N.ServiceName) nidToHostPort bs =@@ -28,3 +34,20 @@ port <- NS.socketPort sock NS.close sock pure $ show port++-- | Receive bytes on a socket until an entire message can be decoded.+-- This function fixes the deserialization of the bytes sent on the socket to+-- the implementation of the Serialize typeclass.+recvSerialized :: S.Serialize a => N.Socket -> IO (Maybe a)+recvSerialized sock = do+ result <- go $ Get.runGetPartial S.get+ case result of+ Get.Fail {} -> pure Nothing+ Get.Done val _ -> pure . pure $ val+ Get.Partial {} -> pure Nothing+ where+ go getPartial = do+ bytes <- fromMaybe BS.empty <$> N.recv sock 4096+ case getPartial bytes of+ Get.Partial getNextPartial -> go getNextPartial+ x -> pure x
src/Examples/Raft/Socket/Node.hs view
@@ -25,7 +25,6 @@ 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@@ -39,29 +38,31 @@ -------------------------------------------------------------------------------- data NodeSocketEnv v = NodeSocketEnv- { nsSocket :: Socket- , nsPeers :: TVar (STM IO) (Map NodeId Socket)- , nsMsgQueue :: TChan (STM IO) (RPCMessage v)+ { 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, MonadFail, 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 MonadThrow m => MonadThrow (RaftSocketT v m)+deriving instance MonadCatch m => MonadCatch (RaftSocketT v m)+deriving instance MonadMask 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+instance (MonadIO m, MonadConc m, S.Serialize sm, S.Serialize v) => RaftSendClient (RaftSocketT v m) sm v where sendClient clientId@(ClientId nid) msg = do let (cHost, cPort) = nidToHostPort (toS nid)- connect cHost cPort $ \(cSock, _cSockAddr) ->- send cSock (S.encode msg)+ eRes <- Control.Monad.Catch.try $+ connect cHost cPort $ \(cSock, _cSockAddr) ->+ send cSock (S.encode msg)+ case eRes of+ Left (err :: SomeException) -> putText ("Failed to send Client: " <> show err)+ Right _ -> pure () instance (MonadIO m, MonadConc m, S.Serialize v) => RaftRecvClient (RaftSocketT v m) v where type RaftRecvClientError (RaftSocketT v m) v = Text@@ -71,20 +72,12 @@ 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)+ eRes <- Control.Monad.Catch.try $+ connect host port $ \(sock,_) -> do+ NS.send sock (S.encode $ RPCMessageEvent msg)+ case eRes of+ Left (err :: SomeException) -> putText ("Failed to send RPC: " <> show err)+ Right _ -> pure () where (host, port) = nidToHostPort nid @@ -94,70 +87,27 @@ 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+acceptConnections :: forall v m. (S.Serialize v, MonadIO m, MonadConc m)- => RaftSocketT v m ()-acceptForkNode = do+ => HostName+ -> ServiceName+ -> RaftSocketT v m ()+acceptConnections host port = 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 _)) ->+ serve (Host host) port $ \(sock, _) -> do+ mVal <- recvSerialized sock+ case mVal of+ Nothing -> putText "Socket was closed on the other end"+ Just val ->+ case val of+ ClientRequestEvent req -> atomically $ writeTChan nsClientReqQueue req- Right (RPCMessageEvent msg) ->+ 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+newSock host port =+ listen (Host host) port (pure . fst)
src/Raft.hs view
@@ -13,8 +13,8 @@ module Raft ( -- * State machine type class- RSMP(..)- , RSM(..)+ RaftStateMachinePure(..)+ , RaftStateMachine(..) -- * Networking type classes , RaftSendRPC(..)@@ -59,6 +59,7 @@ , RaftLogExceptions(..) -- * Logging+ , LogCtx(..) , LogDest(..) , Severity(..) @@ -74,8 +75,8 @@ , isFollower , isCandidate , isLeader- , setLastLogEntryData- , getLastLogEntryData+ , setLastLogEntry+ , getLastLogEntry , getLastAppliedAndCommitIndex -- * Persistent state@@ -103,7 +104,7 @@ , AppendEntriesData(..) ) where -import Protolude hiding (STM, TChan, newTChan, readTChan, writeTChan, atomically)+import Protolude hiding (STM, TChan, newTChan, readBoundedChan, writeBoundedChan, atomically) import Control.Monad.Conc.Class import Control.Concurrent.STM.Timer@@ -115,7 +116,9 @@ import Control.Monad.Trans.Class import qualified Data.Map as Map+import Data.Serialize (Serialize) import Data.Sequence (Seq(..), singleton)+import Data.Time.Clock.System (getSystemTime) import Raft.Action import Raft.Client@@ -123,11 +126,12 @@ import Raft.Event import Raft.Handle import Raft.Log-import Raft.Logging hiding (logInfo, logDebug, logCritical)+import Raft.Logging hiding (logInfo, logDebug, logCritical, logAndPanic) import Raft.Monad hiding (logInfo, logDebug) import Raft.NodeState import Raft.Persistent import Raft.RPC+import Raft.StateMachine import Raft.Types @@ -140,12 +144,12 @@ , resetElectionTimer :: m () , resetHeartbeatTimer :: m () , raftNodeConfig :: NodeConfig- , raftNodeLogDest :: LogDest+ , raftNodeLogCtx :: LogCtx } 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, MonadFail, Alternative, MonadPlus)+ { unRaftT :: ReaderT (RaftEnv v m) (StateT (RaftNodeState v) m) a+ } deriving (Functor, Applicative, Monad, MonadReader (RaftEnv v m), MonadState (RaftNodeState v), MonadFail, Alternative, MonadPlus) instance MonadTrans (RaftT v) where lift = RaftT . lift . lift@@ -156,13 +160,12 @@ 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+instance Monad m => RaftLogger v (RaftT v m) where+ loggerCtx = (,) <$> asks (configNodeId . raftNodeConfig) <*> get runRaftT :: MonadConc m- => RaftNodeState+ => RaftNodeState v -> RaftEnv v m -> RaftT v m () -> m ()@@ -172,23 +175,26 @@ ------------------------------------------------------------------------------ logDebug :: MonadIO m => Text -> RaftT v m ()-logDebug msg = flip logDebugIO msg =<< asks raftNodeLogDest+logDebug msg = flip logDebugIO msg =<< asks raftNodeLogCtx logCritical :: MonadIO m => Text -> RaftT v m ()-logCritical msg = flip logCriticalIO msg =<< asks raftNodeLogDest+logCritical msg = flip logCriticalIO msg =<< asks raftNodeLogCtx +logAndPanic :: MonadIO m => Text -> RaftT v m a+logAndPanic msg = flip logAndPanicIO msg =<< asks raftNodeLogCtx+ ------------------------------------------------------------------------------ -- | 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, MonadFail m- , RSM sm v m- , Show (RSMPError sm v)+ :: forall m sm v.+ ( Show v, Show sm, Serialize v, Show (Action sm v), Show (RaftLogError m), Show (RaftStateMachinePureError sm v)+ , Typeable m, MonadIO m, MonadConc m, MonadFail m+ , RaftStateMachine m sm v , RaftSendRPC m v , RaftRecvRPC m v- , RaftSendClient m sm+ , RaftSendClient m sm v , RaftRecvClient m v , RaftLog m v , RaftLogExceptions m@@ -196,40 +202,60 @@ , Exception (RaftPersistError m) ) => NodeConfig -- ^ Node configuration- -> LogDest -- ^ Logs destination+ -> LogCtx -- ^ Logs destination -> Int -- ^ Timer seed -> sm -- ^ Initial state machine state -> m ()-runRaftNode nodeConfig@NodeConfig{..} logDest timerSeed initRSM = do+runRaftNode nodeConfig@NodeConfig{..} logCtx timerSeed initRaftStateMachine = do+ -- Initialize the persistent state and logs storage if specified+ initializeStorage+ eventChan <- atomically newTChan - electionTimer <- newTimerRange timerSeed configElectionTimeout- heartbeatTimer <- newTimer configHeartbeatTimeout+ electionTimer <-+ newTimerRange (writeTimeoutEvent eventChan ElectionTimeout) timerSeed configElectionTimeout - let resetElectionTimer = resetTimer electionTimer- resetHeartbeatTimer = resetTimer heartbeatTimer- raftEnv = RaftEnv eventChan resetElectionTimer resetHeartbeatTimer nodeConfig logDest+ heartbeatTimer <-+ newTimer (writeTimeoutEvent eventChan HeartbeatTimeout) configHeartbeatTimeout + let resetElectionTimer = void $ resetTimer electionTimer+ resetHeartbeatTimer = void $ resetTimer heartbeatTimer+ raftEnv = RaftEnv eventChan resetElectionTimer resetHeartbeatTimer nodeConfig logCtx+ runRaftT initRaftNodeState raftEnv $ do -- Fork all event producers to run concurrently- lift $ fork (electionTimeoutTimer electionTimer eventChan)- lift $ fork (heartbeatTimeoutTimer heartbeatTimer eventChan)+ lift $ fork (electionTimeoutTimer electionTimer)+ lift $ fork (heartbeatTimeoutTimer heartbeatTimer) fork (rpcHandler eventChan) fork (clientReqHandler eventChan) -- Start the main event handling loop- handleEventLoop initRSM+ handleEventLoop initRaftStateMachine + where+ initializeStorage =+ case configStorageState of+ New -> do+ ipsRes <- initializePersistentState+ case ipsRes of+ Left err -> throw err+ Right _ -> do+ ilRes <- initializeLog (Proxy :: Proxy v)+ case ilRes of+ Left err -> throw err+ Right _ -> pure ()+ Existing -> pure ()+ handleEventLoop :: forall sm v m.- ( Show v, Show sm, Show (Action sm v)+ ( Show v, Serialize v, Show sm, Show (Action sm v), Show (RaftLogError m), Typeable m , MonadIO m, MonadConc m, MonadFail m- , RSM sm v m- , Show (RSMPError sm v)+ , RaftStateMachine m sm v+ , Show (RaftStateMachinePureError sm v) , RaftPersist m , RaftSendRPC m v- , RaftSendClient m sm+ , RaftSendClient m sm v , RaftLog m v , RaftLogExceptions m , RaftPersist m@@ -237,11 +263,12 @@ ) => sm -> RaftT v m ()-handleEventLoop initRSM = do+handleEventLoop initRaftStateMachine = do+ setInitLastLogEntry ePersistentState <- lift readPersistentState case ePersistentState of Left err -> throw err- Right pstate -> handleEventLoop' initRSM pstate+ Right pstate -> handleEventLoop' initRaftStateMachine pstate where handleEventLoop' :: sm -> PersistentState -> RaftT v m () handleEventLoop' stateMachine persistentState = do@@ -250,29 +277,34 @@ 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 ()++ -- Write persistent state to disk.+ --+ -- Checking equality of Term + NodeId (what PersistentState is comprised of)+ -- is very cheap, but writing to disk is not necessarily cheap.+ when (resPersistentState /= persistentState) $ do+ 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+ -- Handle logs produced 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+ resRaftStateMachine <- applyLogEntries stateMachine+ handleEventLoop' resRaftStateMachine resPersistentState -- In the case that a node is a follower receiving an AppendEntriesRPC -- Event, read the log at the aePrevLogIndex@@ -292,12 +324,23 @@ _ -> pure () _ -> pure () + -- Load the last log entry from a existing log+ setInitLastLogEntry :: RaftT v m ()+ setInitLastLogEntry = do+ RaftNodeState rns <- get+ eLogEntry <- lift readLastLogEntry+ case eLogEntry of+ Left err -> throw err+ Right Nothing -> pure ()+ Right (Just e) ->+ put (RaftNodeState (setLastLogEntry rns (singleton e)))+ handleActions- :: ( Show v, Show sm, Show (Action sm v)+ :: ( Show v, Show sm, Show (Action sm v), Show (RaftLogError m), Typeable m , MonadIO m, MonadConc m- , RSM sm v m+ , RaftStateMachine m sm v , RaftSendRPC m v- , RaftSendClient m sm+ , RaftSendClient m sm v , RaftLog m v , RaftLogExceptions m )@@ -308,11 +351,11 @@ handleAction :: forall sm v m.- ( Show v, Show sm, Show (Action sm v)+ ( Show v, Show sm, Show (Action sm v), Show (RaftLogError m), Typeable m , MonadIO m, MonadConc m- , RSM sm v m+ , RaftStateMachine m sm v , RaftSendRPC m v- , RaftSendClient m sm+ , RaftSendClient m sm v , RaftLog m v , RaftLogExceptions m )@@ -332,18 +375,63 @@ BroadcastRPC nids sendRpcAction -> do rpcMsg <- mkRPCfromSendRPCAction sendRpcAction mapConcurrently_ (lift . flip sendRPC rpcMsg) nids- RespondToClient cid cr -> lift $ sendClient cid cr- ResetTimeoutTimer tout ->+ RespondToClient cid cr -> do+ clientResp <- mkClientResp cr+ void . fork . lift $ sendClient cid clientResp+ ResetTimeoutTimer tout -> do 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)-+ eRes <- lift (updateLog entries)+ case eRes of+ Left err -> logAndPanic (show err)+ Right _ -> do+ -- Update the last log entry data+ modify $ \(RaftNodeState ns) ->+ RaftNodeState (setLastLogEntry ns entries)+ UpdateClientReqCacheFrom idx -> do+ RaftNodeState ns <- get+ case ns of+ NodeLeaderState ls@LeaderState{..} -> do+ eentries <- lift (readLogEntriesFrom idx)+ case eentries of+ Left err -> throw err+ Right (entries :: Entries v) -> do+ let committedClientReqs = clientReqData entries+ when (Map.size committedClientReqs > 0) $ do+ mapM_ respondClientWrite (Map.toList committedClientReqs)+ let creqMap = Map.map (second Just) committedClientReqs+ put $ RaftNodeState $ NodeLeaderState+ ls { lsClientReqCache = creqMap `Map.union` lsClientReqCache }+ _ -> logAndPanic "Only the leader should update the client request cache..." where+ respondClientWrite :: (ClientId, (SerialNum, Index)) -> RaftT v m ()+ respondClientWrite (cid, (sn,idx)) =+ handleAction nodeConfig $+ RespondToClient cid (ClientWriteRespSpec idx sn :: ClientRespSpec sm)++ mkClientResp :: ClientRespSpec sm -> RaftT v m (ClientResponse sm v)+ mkClientResp crs =+ case crs of+ ClientReadRespSpec crrs ->+ ClientReadResponse <$>+ case crrs of+ ClientReadRespSpecEntries res -> do+ eRes <- lift (readEntries res)+ case eRes of+ Left err -> throw err+ Right res ->+ case res of+ OneEntry e -> pure (ClientReadRespEntry e)+ ManyEntries es -> pure (ClientReadRespEntries es)+ ClientReadRespSpecStateMachine sm ->+ pure (ClientReadRespStateMachine sm)+ ClientWriteRespSpec idx sn ->+ pure (ClientWriteResponse (ClientWriteResp idx sn))+ ClientRedirRespSpec cl ->+ pure (ClientRedirectResponse (ClientRedirResp cl))+ mkRPCfromSendRPCAction :: SendRPCAction v -> RaftT v m (RPCMessage v) mkRPCfromSendRPCAction sendRPCAction = do RaftNodeState ns <- get@@ -369,7 +457,7 @@ case spec of FromClientReadReq n -> Just n _ -> Nothing- (lastLogIndex, lastLogTerm) = getLastLogEntryData ns+ (lastLogIndex, lastLogTerm) = lastLogEntryIndexAndTerm (getLastLogEntry ns) pure (Empty, lastLogIndex, lastLogTerm, readReq) let leaderId = LeaderId (configNodeId nodeConfig) pure . toRPC $@@ -407,11 +495,12 @@ applyLogEntries :: forall sm m v. ( Show sm+ , MonadIO m , MonadConc m , RaftReadLog m v , Exception (RaftReadLogError m)- , RSM sm v m- , Show (RSMPError sm v)+ , RaftStateMachine m sm v+ , Show (RaftStateMachinePureError sm v) ) => sm -> RaftT v m sm@@ -425,18 +514,18 @@ eLogEntry <- lift $ readLogEntry newLastAppliedIndex case eLogEntry of Left err -> throw err- Right Nothing -> panic "No log entry at 'newLastAppliedIndex'"+ Right Nothing -> logAndPanic $ "No log entry at 'newLastAppliedIndex := " <> show 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)+ eRes <- lift (applyLogEntry stateMachine logEntry) case eRes of- Left err -> panic $ "Failed to apply committed log entry: " <> show err+ Left err -> logAndPanic $ "Failed to apply committed log entry: " <> show err Right nsm -> applyLogEntries nsm else pure stateMachine where- incrLastApplied :: NodeState ns -> NodeState ns+ incrLastApplied :: NodeState ns v -> NodeState ns v incrLastApplied nodeState = case nodeState of NodeFollowerState fs ->@@ -449,10 +538,10 @@ let lastApplied' = incrIndex (lsLastApplied ls) in NodeLeaderState $ ls { lsLastApplied = lastApplied' } - lastApplied :: NodeState ns -> Index+ lastApplied :: NodeState ns v -> Index lastApplied = fst . getLastAppliedAndCommitIndex - commitIndex :: NodeState ns -> Index+ commitIndex :: NodeState ns v -> Index commitIndex = snd . getLastAppliedAndCommitIndex handleLogs@@ -460,8 +549,8 @@ => [LogMsg] -> RaftT v m () handleLogs logs = do- logDest <- asks raftNodeLogDest- mapM_ (logToDest logDest) logs+ logCtx <- asks raftNodeLogCtx+ mapM_ (logToDest logCtx) logs ------------------------------------------------------------------------------ -- Event Producers@@ -470,7 +559,7 @@ -- | Producer for rpc message events rpcHandler :: (MonadIO m, MonadConc m, Show v, RaftRecvRPC m v)- => TChan (STM m) (Event v)+ => EventChan m v -> RaftT v m () rpcHandler eventChan = forever $ do@@ -485,7 +574,7 @@ -- | Producer for rpc message events clientReqHandler :: (MonadIO m, MonadConc m, RaftRecvClient m v)- => TChan (STM m) (Event v)+ => EventChan m v -> RaftT v m () clientReqHandler eventChan = forever $ do@@ -498,15 +587,24 @@ atomically $ writeTChan eventChan clientReqEvent -- | Producer for the election timeout event-electionTimeoutTimer :: MonadConc m => Timer m -> TChan (STM m) (Event v) -> m ()-electionTimeoutTimer timer eventChan =+electionTimeoutTimer :: (MonadIO m, MonadConc m) => Timer m -> m ()+electionTimeoutTimer timer = forever $ do- startTimer timer >> waitTimer timer- atomically $ writeTChan eventChan (TimeoutEvent ElectionTimeout)+ success <- startTimer timer+ when (not success) $+ panic "[Failed invariant]: Election timeout timer failed to start."+ waitTimer timer -- | Producer for the heartbeat timeout event-heartbeatTimeoutTimer :: MonadConc m => Timer m -> TChan (STM m) (Event v) -> m ()-heartbeatTimeoutTimer timer eventChan =+heartbeatTimeoutTimer :: (MonadIO m, MonadConc m) => Timer m -> m ()+heartbeatTimeoutTimer timer = forever $ do- startTimer timer >> waitTimer timer- atomically $ writeTChan eventChan (TimeoutEvent HeartbeatTimeout)+ success <- startTimer timer+ when (not success) $+ panic "[Failed invariant]: Heartbeat timeout timer failed to start."+ waitTimer timer++writeTimeoutEvent :: (MonadIO m , MonadConc m) => EventChan m v -> Timeout -> m ()+writeTimeoutEvent eventChan timeout = do+ now <- liftIO getSystemTime+ atomically $ writeTChan eventChan (TimeoutEvent now timeout)
src/Raft/Action.hs view
@@ -16,8 +16,9 @@ | 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+ | RespondToClient ClientId (ClientRespSpec sm) -- ^ Respond to client after a client request | ResetTimeoutTimer Timeout -- ^ Reset a timeout timer+ | UpdateClientReqCacheFrom Index -- ^ Update the client request cache from the given index onward deriving (Show) data SendRPCAction v
src/Raft/Candidate.hs view
@@ -20,6 +20,7 @@ import Protolude +import qualified Data.Serialize as S import qualified Data.Set as Set import qualified Data.Sequence as Seq import qualified Data.Map as Map@@ -41,10 +42,7 @@ 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+ pure $ candidateResultState Noop candidateState -- | Candidates should not respond to 'AppendEntriesResponse' messages. handleAppendEntriesResponse :: RPCHandler 'Candidate sm AppendEntriesResponse v@@ -61,7 +59,7 @@ -- | Candidates should not respond to 'RequestVoteResponse' messages. handleRequestVoteResponse- :: forall sm v. Show v+ :: forall sm v. (Show v, S.Serialize v) => RPCHandler 'Candidate sm RequestVoteResponse v handleRequestVoteResponse (NodeCandidateState candidateState@CandidateState{..}) sender requestVoteResp@RequestVoteResponse{..} = do currentTerm <- gets currentTerm@@ -83,7 +81,7 @@ mkNoopEntry :: TransitionM sm v (Entry v) mkNoopEntry = do- let (lastLogEntryIdx, _) = csLastLogEntryData+ let lastLogEntryIdx = lastLogEntryIndex csLastLogEntry currTerm <- gets currentTerm nid <- asks (configNodeId . nodeConfig) pure Entry@@ -91,12 +89,12 @@ , entryTerm = currTerm , entryValue = NoValue , entryIssuer = LeaderIssuer (LeaderId nid)+ , entryPrevHash = hashLastLogEntry csLastLogEntry } - becomeLeader :: TransitionM sm v LeaderState+ becomeLeader :: TransitionM sm v (LeaderState v) 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.@@ -108,9 +106,9 @@ , aedLeaderCommit = csCommitIndex , aedEntriesSpec = FromNewLeader noopEntry }+ resetHeartbeatTimeout cNodeIds <- asks (configNodeIds . nodeConfig) let lastLogEntryIdx = entryIndex noopEntry- lastLogEntryTerm = entryTerm noopEntry pure LeaderState { lsCommitIndex = csCommitIndex , lsLastApplied = csLastApplied@@ -118,9 +116,10 @@ (,lastLogEntryIdx) <$> Set.toList cNodeIds , lsMatchIndex = Map.fromList $ (,index0) <$> Set.toList cNodeIds- , lsLastLogEntryData = (lastLogEntryIdx, lastLogEntryTerm, Nothing)+ , lsLastLogEntry = csLastLogEntry , lsReadReqsHandled = 0 , lsReadRequest = mempty+ , lsClientReqCache = csClientReqCache } handleTimeout :: TimeoutHandler 'Candidate sm v@@ -129,7 +128,7 @@ HeartbeatTimeout -> pure $ candidateResultState Noop candidateState ElectionTimeout -> candidateResultState RestartElection <$>- startElection csCommitIndex csLastApplied csLastLogEntryData+ startElection csCommitIndex csLastApplied csLastLogEntry csClientReqCache -- | 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@@ -139,29 +138,3 @@ 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
@@ -1,26 +1,113 @@ {-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE UndecidableInstances #-} -module Raft.Client where+module Raft.Client+(+ -- ** Raft Interface+ RaftSendClient(..)+, RaftRecvClient(..) -import Protolude+, SerialNum(..) +-- ** Client Requests+, ClientRequest(..)+, ClientReq(..)+, ClientReadReq(..)+, ReadEntriesSpec(..)++-- ** Client Responses+, ClientResponse(..)++, ClientRespSpec(..)+, ClientReadRespSpec(..)++, ClientReadResp(..)+, ClientWriteResp(..)+, ClientRedirResp(..)+++-- ** Client Interface+, RaftClientSend(..)+, RaftClientRecv(..)++, RaftClientState(..)+, RaftClientEnv(..)+, initRaftClientState++, RaftClientT+, runRaftClientT++, RaftClientError(..)+, clientRead+, clientReadFrom+, clientReadTimeout++, clientWrite+, clientWriteTo+, clientWriteTimeout++, retryOnRedirect++, clientAddNode+, clientGetNodes++) where++import Protolude hiding (threadDelay, STM, )++import Control.Concurrent.Classy+import Control.Monad.Base+import Control.Monad.Catch+import Control.Monad.Fail+import Control.Monad.Trans.Class+import Control.Monad.Trans.Control++import qualified Data.Set as Set import qualified Data.Serialize as S -import Raft.NodeState+import System.Random+import System.Console.Haskeline.MonadException (MonadException(..), RunIO(..))+import System.Timeout.Lifted (timeout)++import Raft.Log (Entry, Entries, ReadEntriesSpec) import Raft.Types +--------------------------------------------------------------------------------+-- Raft Interface+--------------------------------------------------------------------------------++{- This is the interface with which Raft nodes interact with client programs -}+ -- | Interface for Raft nodes to send messages to clients-class RaftSendClient m sm where- sendClient :: ClientId -> ClientResponse sm -> m ()+--+-- TODO It would be really nice if 'RSMP' was a superclass, but currently this+-- can't happen because of cyclic imports.+class RaftSendClient m sm v where+ sendClient :: ClientId -> ClientResponse sm v -> 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)) +--------------------------------------------------------------------------------+-- Client Requests+--------------------------------------------------------------------------------+ -- | Representation of a client request coupled with the client id data ClientRequest v = ClientRequest ClientId (ClientReq v)@@ -30,15 +117,38 @@ -- | Representation of a client request data ClientReq v- = ClientReadReq -- ^ Request the latest state of the state machine- | ClientWriteReq v -- ^ Write a command+ = ClientReadReq ClientReadReq -- ^ Request the latest state of the state machine+ | ClientWriteReq SerialNum 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)+data ClientReadReq+ = ClientReadEntries ReadEntriesSpec+ | ClientReadStateMachine+ deriving (Show, Generic, S.Serialize)++--------------------------------------------------------------------------------+-- Client Responses+--------------------------------------------------------------------------------++-- | Specification for the data inside a ClientResponse+data ClientRespSpec sm+ = ClientReadRespSpec (ClientReadRespSpec sm)+ | ClientWriteRespSpec Index SerialNum+ | ClientRedirRespSpec CurrentLeader+ deriving (Show, Generic, S.Serialize)++data ClientReadRespSpec sm+ = ClientReadRespSpecEntries ReadEntriesSpec+ | ClientReadRespSpecStateMachine sm+ deriving (Show, Generic, S.Serialize)++--------------------------------------------------------------------------------++-- | The datatype sent back to the client as an actual response+data ClientResponse sm v+ = ClientReadResponse (ClientReadResp sm v) -- ^ Respond with the latest state of the state machine. | ClientWriteResponse ClientWriteResp -- ^ Respond with the index of the entry appended to the log@@ -46,19 +156,20 @@ -- ^ Respond with the node id of the current leader deriving (Show, Generic) -instance S.Serialize s => S.Serialize (ClientResponse s)+instance (S.Serialize sm, S.Serialize v) => S.Serialize (ClientResponse sm v) -- | Representation of a read response to a client--- The `s` stands for the "current" state of the state machine-newtype ClientReadResp s- = ClientReadResp s+data ClientReadResp sm v+ = ClientReadRespStateMachine sm+ | ClientReadRespEntry (Entry v)+ | ClientReadRespEntries (Entries v) deriving (Show, Generic) -instance S.Serialize s => S.Serialize (ClientReadResp s)+instance (S.Serialize sm, S.Serialize v) => S.Serialize (ClientReadResp sm v) -- | Representation of a write response to a client data ClientWriteResp- = ClientWriteResp Index+ = ClientWriteResp Index SerialNum -- ^ Index of the entry appended to the log due to the previous client request deriving (Show, Generic) @@ -70,3 +181,342 @@ deriving (Show, Generic) instance S.Serialize ClientRedirResp++--------------------------------------------------------------------------------+-- Client Interface+--------------------------------------------------------------------------------++{- This is the interface with which clients interact with Raft nodes -}++class Monad m => RaftClientSend m v where+ type RaftClientSendError m v+ raftClientSend :: NodeId -> ClientRequest v -> m (Either (RaftClientSendError m v) ())++class Monad m => RaftClientRecv m sm v | m sm -> v where+ type RaftClientRecvError m sm+ raftClientRecv :: m (Either (RaftClientRecvError m sm) (ClientResponse sm v))++-- | Each client may have at most one command outstanding at a time and+-- commands must be dispatched in serial number order.+data RaftClientState = RaftClientState+ { raftClientCurrentLeader :: CurrentLeader+ , raftClientSerialNum :: SerialNum+ , raftClientRaftNodes :: Set NodeId+ , raftClientRandomGen :: StdGen+ }++data RaftClientEnv = RaftClientEnv+ { raftClientId :: ClientId+ }++initRaftClientState :: Set NodeId -> StdGen -> RaftClientState+initRaftClientState = RaftClientState NoLeader 0++newtype RaftClientT s v m a = RaftClientT+ { unRaftClientT :: ReaderT RaftClientEnv (StateT RaftClientState m) a+ } deriving newtype (Functor, Applicative, Monad, MonadIO, MonadState RaftClientState, MonadReader RaftClientEnv, MonadFail, Alternative, MonadPlus)++deriving newtype instance MonadThrow m => MonadThrow (RaftClientT s v m)+deriving newtype instance MonadCatch m => MonadCatch (RaftClientT s v m)+deriving newtype instance MonadMask m => MonadMask (RaftClientT s v m)+deriving newtype instance MonadSTM m => MonadSTM (RaftClientT s v m)+deriving newtype instance MonadConc m => MonadConc (RaftClientT s v m)++instance MonadTrans (RaftClientT s v) where+ lift = RaftClientT . lift . lift++deriving newtype instance MonadBase IO m => MonadBase IO (RaftClientT s v m)++instance MonadTransControl (RaftClientT s v) where+ type StT (RaftClientT s v) a = StT (ReaderT RaftClientEnv) (StT (StateT RaftClientState) a)+ liftWith = defaultLiftWith2 RaftClientT unRaftClientT+ restoreT = defaultRestoreT2 RaftClientT++instance (MonadBaseControl IO m) => MonadBaseControl IO (RaftClientT s v m) where+ type StM (RaftClientT s v m) a = ComposeSt (RaftClientT s v) m a+ liftBaseWith = defaultLiftBaseWith+ restoreM = defaultRestoreM++-- This annoying instance is because of the Haskeline library, letting us use a+-- custom monad transformer stack as the base monad of 'InputT'. IMO it should+-- be automatically derivable. Why does haskeline use a custom exception+-- monad... ?+instance MonadException m => MonadException (RaftClientT s v m) where+ controlIO f =+ RaftClientT $ ReaderT $ \r -> StateT $ \s ->+ controlIO $ \(RunIO run) ->+ let run' = RunIO (fmap (RaftClientT . ReaderT . const . StateT . const) . run . flip runStateT s . flip runReaderT r . unRaftClientT)+ in fmap (flip runStateT s . flip runReaderT r . unRaftClientT) $ f run'++instance RaftClientSend m v => RaftClientSend (RaftClientT s v m) v where+ type RaftClientSendError (RaftClientT s v m) v = RaftClientSendError m v+ raftClientSend nid creq = lift (raftClientSend nid creq)++instance RaftClientRecv m s v => RaftClientRecv (RaftClientT s v m) s v where+ type RaftClientRecvError (RaftClientT s v m) s = RaftClientRecvError m s+ raftClientRecv = lift raftClientRecv++runRaftClientT :: Monad m => RaftClientEnv -> RaftClientState -> RaftClientT s v m a -> m a+runRaftClientT raftClientEnv raftClientState =+ flip evalStateT raftClientState . flip runReaderT raftClientEnv . unRaftClientT++--------------------------------------------------------------------------------++data RaftClientError s v m where+ RaftClientSendError :: RaftClientSendError m v -> RaftClientError s v m+ RaftClientRecvError :: RaftClientRecvError m s -> RaftClientError s v m+ RaftClientTimeout :: Text -> RaftClientError s v m+ RaftClientUnexpectedReadResp :: ClientReadResp s v -> RaftClientError s v m+ RaftClientUnexpectedWriteResp :: ClientWriteResp -> RaftClientError s v m+ RaftClientUnexpectedRedirect :: ClientRedirResp -> RaftClientError s v m++deriving instance (Show s, Show v, Show (RaftClientSendError m v), Show (RaftClientRecvError m s)) => Show (RaftClientError s v m)++-- | Send a read request to the curent leader and wait for a response+clientRead+ :: (Show (RaftClientSendError m v), RaftClientSend m v, RaftClientRecv m s v)+ => ClientReadReq+ -> RaftClientT s v m (Either (RaftClientError s v m) (ClientReadResp s v))+clientRead crr = do+ eSend <- clientSendRead crr+ case eSend of+ Left err -> pure (Left (RaftClientSendError err))+ Right _ -> clientRecvRead++-- | Send a read request to a specific raft node, regardless of leader, and wait+-- for a response.+clientReadFrom+ :: (RaftClientSend m v, RaftClientRecv m s v)+ => NodeId+ -> ClientReadReq+ -> RaftClientT s v m (Either (RaftClientError s v m) (ClientReadResp s v))+clientReadFrom nid crr = do+ eSend <- clientSendReadTo nid crr+ case eSend of+ Left err -> pure (Left (RaftClientSendError err))+ Right _ -> clientRecvRead++-- | 'clientRead' but with a timeout+clientReadTimeout+ :: (MonadBaseControl IO m, Show (RaftClientSendError m v), RaftClientSend m v, RaftClientRecv m s v)+ => Int+ -> ClientReadReq+ -> RaftClientT s v m (Either (RaftClientError s v m) (ClientReadResp s v))+clientReadTimeout t = clientTimeout "clientRead" t . clientRead++-- | Send a write request to the current leader and wait for a response+clientWrite+ :: (Show (RaftClientSendError m v), RaftClientSend m v, RaftClientRecv m s v)+ => v+ -> RaftClientT s v m (Either (RaftClientError s v m) ClientWriteResp)+clientWrite cmd = do+ eSend <- clientSendWrite cmd+ case eSend of+ Left err -> pure (Left (RaftClientSendError err))+ Right _ -> clientRecvWrite++-- | Send a read request to a specific raft node, regardless of leader, and wait+-- for a response.+clientWriteTo+ :: (RaftClientSend m v, RaftClientRecv m s v)+ => NodeId+ -> v+ -> RaftClientT s v m (Either (RaftClientError s v m) ClientWriteResp)+clientWriteTo nid cmd = do+ eSend <- clientSendWriteTo nid cmd+ case eSend of+ Left err -> pure (Left (RaftClientSendError err))+ Right _ -> clientRecvWrite++clientWriteTimeout+ :: (MonadBaseControl IO m, Show (RaftClientSendError m v), RaftClientSend m v, RaftClientRecv m s v)+ => Int+ -> v+ -> RaftClientT s v m (Either (RaftClientError s v m) ClientWriteResp)+clientWriteTimeout t cmd = clientTimeout "clientWrite" t (clientWrite cmd)++clientTimeout+ :: (MonadBaseControl IO m, RaftClientSend m v, RaftClientRecv m s v)+ => Text+ -> Int+ -> RaftClientT s v m (Either (RaftClientError s v m) r)+ -> RaftClientT s v m (Either (RaftClientError s v m) r)+clientTimeout fnm t r = do+ mRes <- timeout t r+ case mRes of+ Nothing -> pure (Left (RaftClientTimeout fnm))+ Just (Left err) -> pure (Left err)+ Just (Right cresp) -> pure (Right cresp)++-- | Given a blocking client send/receive, retry if the received value is not+-- expected+retryOnRedirect+ :: MonadConc m+ => RaftClientT s v m (Either (RaftClientError s v m) r)+ -> RaftClientT s v m (Either (RaftClientError s v m) r)+retryOnRedirect action = do+ eRes <- action+ case eRes of+ Left (RaftClientUnexpectedRedirect _) -> do+ threadDelay 10000+ retryOnRedirect action+ Left err -> pure (Left err)+ Right resp -> pure (Right resp)++--------------------------------------------------------------------------------+-- Helpers+--------------------------------------------------------------------------------++-- | Send a read request to the current leader. Nonblocking.+clientSendRead+ :: (Show (RaftClientSendError m v), RaftClientSend m v)+ => ClientReadReq+ -> RaftClientT s v m (Either (RaftClientSendError m v) ())+clientSendRead crr =+ asks raftClientId >>= \cid ->+ clientSend (ClientRequest cid (ClientReadReq crr))++clientSendReadTo+ :: RaftClientSend m v+ => NodeId+ -> ClientReadReq+ -> RaftClientT s v m (Either (RaftClientSendError m v) ())+clientSendReadTo nid crr =+ asks raftClientId >>= \cid ->+ clientSendTo nid (ClientRequest cid (ClientReadReq crr))++-- | Send a write request to the current leader. Nonblocking.+clientSendWrite+ :: (Show (RaftClientSendError m v), RaftClientSend m v)+ => v+ -> RaftClientT s v m (Either (RaftClientSendError m v) ())+clientSendWrite v = do+ asks raftClientId >>= \cid -> gets raftClientSerialNum >>= \sn ->+ clientSend (ClientRequest cid (ClientWriteReq sn v))++-- | Send a write request to a specific raft node, ignoring the current+-- leader. This function is used in testing.+clientSendWriteTo+ :: RaftClientSend m v+ => NodeId+ -> v+ -> RaftClientT s v m (Either (RaftClientSendError m v) ())+clientSendWriteTo nid v =+ asks raftClientId >>= \cid -> gets raftClientSerialNum >>= \sn ->+ clientSendTo nid (ClientRequest cid (ClientWriteReq sn v))++-- | Send a request to the current leader. Nonblocking.+clientSend+ :: (Show (RaftClientSendError m v), RaftClientSend m v)+ => ClientRequest v+ -> RaftClientT s v m (Either (RaftClientSendError m v) ())+clientSend creq = do+ currLeader <- gets raftClientCurrentLeader+ case currLeader of+ NoLeader -> clientSendRandom creq+ CurrentLeader (LeaderId nid) -> do+ eRes <- raftClientSend nid creq+ case eRes of+ Left err -> clientSendRandom creq+ Right resp -> pure (Right resp)++-- | Send a request to a specific raft node, ignoring the current leader.+-- This function is used in testing.+clientSendTo+ :: RaftClientSend m v+ => NodeId+ -> ClientRequest v+ -> RaftClientT s v m (Either (RaftClientSendError m v) ())+clientSendTo nid creq = raftClientSend nid creq++-- | Send a request to a random node.+-- This function is used if there is no leader.+clientSendRandom+ :: RaftClientSend m v+ => ClientRequest v+ -> RaftClientT s v m (Either (RaftClientSendError m v) ())+clientSendRandom creq = do+ raftNodes <- gets raftClientRaftNodes+ randomGen <- gets raftClientRandomGen+ let (idx, newRandomGen) = randomR (0, length raftNodes - 1) randomGen+ case atMay (toList raftNodes) idx of+ Nothing -> panic "No raft nodes known by client"+ Just nid -> do+ modify $ \s -> s { raftClientRandomGen = newRandomGen }+ raftClientSend nid creq++--------------------------------------------------------------------------------++-- | Waits for a write response on the client socket+-- Warning: This function discards unexpected read and redirect responses+clientRecvWrite+ :: (RaftClientSend m v, RaftClientRecv m s v)+ => RaftClientT s v m (Either (RaftClientError s v m) ClientWriteResp)+clientRecvWrite = do+ eRes <- clientRecv+ case eRes of+ Left err -> pure (Left (RaftClientRecvError err))+ Right cresp ->+ case cresp of+ ClientRedirectResponse crr -> pure (Left (RaftClientUnexpectedRedirect crr))+ ClientReadResponse crr -> pure (Left (RaftClientUnexpectedReadResp crr))+ ClientWriteResponse cwr -> pure (Right cwr)++-- | Waits for a read response on the client socket+-- Warning: This function discards unexpected write and redirect responses+clientRecvRead+ :: (RaftClientSend m v, RaftClientRecv m s v)+ => RaftClientT s v m (Either (RaftClientError s v m) (ClientReadResp s v))+clientRecvRead = do+ eRes <- clientRecv+ case eRes of+ Left err -> pure (Left (RaftClientRecvError err))+ Right cresp ->+ case cresp of+ ClientRedirectResponse crr -> pure (Left (RaftClientUnexpectedRedirect crr))+ ClientWriteResponse cwr -> pure (Left (RaftClientUnexpectedWriteResp cwr))+ ClientReadResponse crr -> pure (Right crr)++-- | Wait for a response from the current leader.+-- This function handles leader changes and write request serial numbers.+clientRecv+ :: RaftClientRecv m s v+ => RaftClientT s v m (Either (RaftClientRecvError m s) (ClientResponse s v))+clientRecv = do+ ecresp <- raftClientRecv+ case ecresp of+ Left err -> pure (Left err)+ Right cresp ->+ case cresp of+ ClientWriteResponse (ClientWriteResp _ (SerialNum n)) -> do+ SerialNum m <- gets raftClientSerialNum+ if m == n+ then do+ modify $ \s -> s+ { raftClientSerialNum = SerialNum (succ m) }+ pure (Right cresp)+ else+ -- Here, we ignore the response if we are receiving a response+ -- regarding a previously committed write request. This regularly+ -- happens when a write request is submitted, committed, and+ -- responded to, but the leader subsequently dies before letting all+ -- other nodes know that the write request has been committed.+ if n < m+ then clientRecv+ else do+ let errMsg = "Received invalid serial number response: Expected " <> show m <> " but got " <> show n+ panic $ errMsg+ ClientRedirectResponse (ClientRedirResp currLdr) -> do+ modify $ \s -> s+ { raftClientCurrentLeader = currLdr }+ pure (Right cresp)+ _ -> pure (Right cresp)++--------------------------------------------------------------------------------++clientAddNode :: Monad m => NodeId -> RaftClientT s v m ()+clientAddNode nid = modify $ \s ->+ s { raftClientRaftNodes = Set.insert nid (raftClientRaftNodes s) }++clientGetNodes :: Monad m => RaftClientT s v m (Set NodeId)+clientGetNodes = gets raftClientRaftNodes
src/Raft/Config.hs view
@@ -14,5 +14,8 @@ , 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+ , configStorageState :: StorageState -- ^ Create a fresh DB or read from existing } deriving (Show) +data StorageState = New | Existing+ deriving Show
src/Raft/Event.hs view
@@ -10,10 +10,12 @@ import Raft.Client import Raft.RPC +import Data.Time.Clock.System+ -- | Representation of events a raft node can send and receive data Event v = MessageEvent (MessageEvent v)- | TimeoutEvent Timeout+ | TimeoutEvent SystemTime Timeout deriving (Show) -- | Representation of timeouts@@ -31,5 +33,3 @@ deriving (Show, Generic) instance S.Serialize v => S.Serialize (MessageEvent v)--
src/Raft/Follower.hs view
@@ -69,6 +69,7 @@ -- 5. If leaderCommit > commitIndex, set commitIndex = -- min(leaderCommit, index of last new entry) pure (True, updateFollowerState fs)+ when success resetElectionTimeout send (unLeaderId aeLeaderId) $ SendAppendEntriesResponseRPC $ AppendEntriesResponse@@ -76,28 +77,24 @@ , aerSuccess = success , aerReadRequest = aeReadRequest }- resetElectionTimeout pure (followerResultState Noop newFollowerState) where- updateFollowerState :: FollowerState -> FollowerState+ updateFollowerState :: FollowerState v -> FollowerState v updateFollowerState fs = if aeLeaderCommit > fsCommitIndex fs then updateLeader (updateCommitIndex fs) else updateLeader fs - updateCommitIndex :: FollowerState -> FollowerState+ updateCommitIndex :: FollowerState v -> FollowerState v 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 v -> FollowerState v updateLeader followerState = followerState { fsCurrentLeader = CurrentLeader (LeaderId sender) } -- | Followers should not respond to 'AppendEntriesResponse' messages.@@ -109,16 +106,16 @@ handleRequestVote ns@(NodeFollowerState fs) sender RequestVote{..} = do PersistentState{..} <- get let voteGranted = giveVote currentTerm votedFor- logDebug $ "Vote granted: " <> show voteGranted+ when voteGranted $ do+ modify $ \pstate ->+ pstate { votedFor = Just sender }+ resetElectionTimeout send sender $ SendRequestVoteResponseRPC $ RequestVoteResponse { rvrTerm = currentTerm , rvrVoteGranted = voteGranted }- when voteGranted $- modify $ \pstate ->- pstate { votedFor = Just sender } pure $ followerResultState Noop fs where giveVote term mVotedFor =@@ -133,9 +130,9 @@ -- 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)+ let (lastEntryIdx, lastEntryTerm) = lastLogEntryIndexAndTerm (fsLastLogEntry fs)+ in (rvLastLogTerm > lastEntryTerm)+ || (rvLastLogTerm == lastEntryTerm && rvLastLogIndex >= lastEntryIdx) -- | Followers should not respond to 'RequestVoteResponse' messages. handleRequestVoteResponse :: RPCHandler 'Follower sm RequestVoteResponse v@@ -149,7 +146,7 @@ ElectionTimeout -> do logDebug "Follower times out. Starts election. Becomes candidate" candidateResultState StartElection <$>- startElection (fsCommitIndex fs) (fsLastApplied fs) (fsLastLogEntryData fs)+ startElection (fsCommitIndex fs) (fsLastApplied fs) (fsLastLogEntry fs) (fsClientReqCache fs) -- Follower should ignore heartbeat timeout events HeartbeatTimeout -> pure (followerResultState Noop fs)
src/Raft/Handle.hs view
@@ -11,6 +11,8 @@ import Protolude +import Data.Serialize (Serialize)+ import qualified Raft.Follower as Follower import qualified Raft.Candidate as Candidate import qualified Raft.Leader as Leader@@ -21,17 +23,19 @@ import Raft.NodeState import Raft.Persistent import Raft.RPC+import Raft.StateMachine+import Raft.Types import Raft.Logging (LogMsg) -- | Main entry point for handling events handleEvent :: forall sm v.- (RSMP sm v, Show v)- => RaftNodeState- -> TransitionEnv sm+ (RaftStateMachinePure sm v, Show v, Serialize v)+ => RaftNodeState v+ -> TransitionEnv sm v -> PersistentState -> Event v- -> (RaftNodeState, PersistentState, [Action sm v], [LogMsg])+ -> (RaftNodeState v, PersistentState, [Action sm v], [LogMsg]) handleEvent raftNodeState@(RaftNodeState initNodeState) transitionEnv persistentState event = -- Rules for all servers: case handleNewerRPCTerm of@@ -40,7 +44,7 @@ ((ResultState _ resultState, logMsgs'), persistentState'', outputs') -> (RaftNodeState resultState, persistentState'', outputs <> outputs', logMsgs <> logMsgs') where- handleNewerRPCTerm :: ((RaftNodeState, [LogMsg]), PersistentState, [Action sm v])+ handleNewerRPCTerm :: ((RaftNodeState v, [LogMsg]), PersistentState, [Action sm v]) handleNewerRPCTerm = case event of MessageEvent (RPCMessageEvent (RPCMessage _ rpc)) ->@@ -61,7 +65,7 @@ else pure raftNodeState _ -> ((raftNodeState, []), persistentState, mempty) - convertToFollower :: forall s. NodeState s -> ResultState s+ convertToFollower :: forall s. NodeState s v -> ResultState s v convertToFollower nodeState = case nodeState of NodeFollowerState _ ->@@ -72,8 +76,9 @@ { fsCurrentLeader = NoLeader , fsCommitIndex = csCommitIndex cs , fsLastApplied = csLastApplied cs- , fsLastLogEntryData = csLastLogEntryData cs+ , fsLastLogEntry = csLastLogEntry cs , fsTermAtAEPrevIndex = Nothing+ , fsClientReqCache = csClientReqCache cs } NodeLeaderState ls -> ResultState HigherTermFoundLeader $@@ -81,10 +86,9 @@ { fsCurrentLeader = NoLeader , fsCommitIndex = lsCommitIndex ls , fsLastApplied = lsLastApplied ls- , fsLastLogEntryData =- let (lastLogEntryIdx, lastLogEntryTerm, _) = lsLastLogEntryData ls- in (lastLogEntryIdx, lastLogEntryTerm)+ , fsLastLogEntry = lsLastLogEntry ls , fsTermAtAEPrevIndex = Nothing+ , fsClientReqCache = lsClientReqCache ls } @@ -97,7 +101,7 @@ , handleClientRequest :: ClientReqHandler ns sm v } -followerRaftHandler :: Show v => RaftHandler 'Follower sm v+followerRaftHandler :: (Show v, Serialize v) => RaftHandler 'Follower sm v followerRaftHandler = RaftHandler { handleAppendEntries = Follower.handleAppendEntries , handleAppendEntriesResponse = Follower.handleAppendEntriesResponse@@ -107,7 +111,7 @@ , handleClientRequest = Follower.handleClientRequest } -candidateRaftHandler :: Show v => RaftHandler 'Candidate sm v+candidateRaftHandler :: (Show v, Serialize v) => RaftHandler 'Candidate sm v candidateRaftHandler = RaftHandler { handleAppendEntries = Candidate.handleAppendEntries , handleAppendEntriesResponse = Candidate.handleAppendEntriesResponse@@ -117,7 +121,7 @@ , handleClientRequest = Candidate.handleClientRequest } -leaderRaftHandler :: Show v => RaftHandler 'Leader sm v+leaderRaftHandler :: (Show v, Serialize v) => RaftHandler 'Leader sm v leaderRaftHandler = RaftHandler { handleAppendEntries = Leader.handleAppendEntries , handleAppendEntriesResponse = Leader.handleAppendEntriesResponse@@ -127,7 +131,7 @@ , handleClientRequest = Leader.handleClientRequest } -mkRaftHandler :: forall ns sm v. Show v => NodeState ns -> RaftHandler ns sm v+mkRaftHandler :: forall ns sm v. (Show v, Serialize v) => NodeState ns v -> RaftHandler ns sm v mkRaftHandler nodeState = case nodeState of NodeFollowerState _ -> followerRaftHandler@@ -136,12 +140,12 @@ handleEvent' :: forall ns sm v.- (RSMP sm v, Show v)- => NodeState ns- -> TransitionEnv sm+ (RaftStateMachinePure sm v, Show v, Serialize v)+ => NodeState ns v+ -> TransitionEnv sm v -> PersistentState -> Event v- -> ((ResultState ns, [LogMsg]), PersistentState, [Action sm v])+ -> ((ResultState ns v, [LogMsg]), PersistentState, [Action sm v]) handleEvent' initNodeState transitionEnv persistentState event = runTransitionM transitionEnv persistentState $ do case event of@@ -150,12 +154,12 @@ RPCMessageEvent rpcMsg -> handleRPCMessage rpcMsg ClientRequestEvent cr -> do handleClientRequest initNodeState cr- TimeoutEvent tout -> do+ TimeoutEvent _ tout -> do handleTimeout initNodeState tout where RaftHandler{..} = mkRaftHandler initNodeState - handleRPCMessage :: RPCMessage v -> TransitionM sm v (ResultState ns)+ handleRPCMessage :: RPCMessage v -> TransitionM sm v (ResultState ns v) handleRPCMessage (RPCMessage sender rpc) = case rpc of AppendEntriesRPC appendEntries ->
src/Raft/Leader.hs view
@@ -1,6 +1,7 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TupleSections #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE DataKinds #-}@@ -18,6 +19,7 @@ import Protolude import qualified Data.Map as Map+import Data.Serialize (Serialize) import Data.Sequence (Seq(Empty)) import qualified Data.Set as Set import qualified Data.Sequence as Seq@@ -42,9 +44,7 @@ handleAppendEntries (NodeLeaderState ls)_ _ = pure (leaderResultState Noop ls) -handleAppendEntriesResponse- :: forall sm v. Show v- => RPCHandler 'Leader sm AppendEntriesResponse v+handleAppendEntriesResponse :: forall sm 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@@ -56,41 +56,37 @@ 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+ Nothing -> leaderResultState Noop <$> do+ let lastLogEntryIdx = lastLogEntryIndex (lsLastLogEntry ls)+ newNextIndices = Map.insert sender (lastLogEntryIdx + 1) (lsNextIndex ls)+ newMatchIndices = Map.insert sender lastLogEntryIdx (lsMatchIndex ls)+ lsUpdatedIndices = ls { lsNextIndex = newNextIndices, lsMatchIndex = newMatchIndices }+ -- Increment leader commit index if now a majority of followers have+ -- replicated an entry at a given term.+ lsUpdatedCommitIdx <- incrCommitIndex lsUpdatedIndices+ when (lsCommitIndex lsUpdatedCommitIdx > lsCommitIndex lsUpdatedIndices) $+ updateClientReqCacheFromIdx (lsCommitIndex lsUpdatedIndices)+ pure lsUpdatedCommitIdx+ Just n -> handleReadReq n ls where- handleReadReq :: Int -> LeaderState -> TransitionM sm v (ResultState 'Leader)+ handleReadReq :: Int -> LeaderState v -> TransitionM sm v (ResultState 'Leader v) handleReadReq n leaderState = do networkSize <- Set.size <$> asks (configNodeIds . nodeConfig) let initReadReqs = lsReadRequest leaderState- (mclientId, newReadReqs) =+ (mCreqData, 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)+ Nothing -> (Nothing, initReadReqs)+ Just (creqData, m)+ | isMajority (succ m) networkSize -> (Just creqData, Map.delete n initReadReqs) | otherwise -> (Nothing, Map.adjust (second succ) n initReadReqs)- case mclientId of+ case mCreqData of Nothing -> pure $ leaderResultState Noop leaderState { lsReadRequest = newReadReqs }- Just cid -> do- respondClientRead cid+ Just (ClientReadReqData cid res) -> do+ respondClientRead cid res pure $ leaderResultState Noop leaderState { lsReadReqsHandled = succ (lsReadReqsHandled leaderState) , lsReadRequest = newReadReqs@@ -120,43 +116,70 @@ -- | 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 :: (Show v, Serialize v) => ClientReqHandler 'Leader sm v handleClientRequest (NodeLeaderState ls@LeaderState{..}) (ClientRequest cid cr) = do case cr of- ClientReadReq -> do+ ClientReadReq crr -> do heartbeat <- mkAppendEntriesData ls (NoEntries (FromClientReadReq lsReadReqsHandled)) broadcast (SendAppendEntriesRPC heartbeat)- let newLeaderState =- ls { lsReadRequest = Map.insert lsReadReqsHandled (cid, 0) lsReadRequest+ let clientReqData = ClientReadReqData cid crr+ newLeaderState =+ ls { lsReadRequest = Map.insert lsReadReqsHandled (clientReqData, 1) 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)+ pure (leaderResultState HandleClientReq newLeaderState)+ ClientWriteReq newSerial v ->+ leaderResultState HandleClientReq <$> handleClientWriteReq newSerial v where- mkNewLogEntry v = do+ mkNewLogEntry v sn = do currentTerm <- currentTerm <$> get- let (lastLogEntryIdx,_,_) = lsLastLogEntryData+ let lastLogEntryIdx = lastLogEntryIndex lsLastLogEntry pure $ Entry { entryIndex = succ lastLogEntryIdx , entryTerm = currentTerm , entryValue = EntryValue v- , entryIssuer = ClientIssuer cid+ , entryIssuer = ClientIssuer cid sn+ , entryPrevHash = hashLastLogEntry lsLastLogEntry } + handleClientWriteReq newSerial v =+ case Map.lookup cid lsClientReqCache of+ -- This is important case #1+ Nothing -> do+ let lsClientReqCache' = Map.insert cid (newSerial, Nothing) lsClientReqCache+ handleNewEntry+ pure ls { lsClientReqCache = lsClientReqCache' }+ Just (currSerial, mResp)+ | newSerial < currSerial -> do+ let debugMsg s1 s2 = "Ignoring serial number " <> s1 <> ", current serial is " <> s2+ logDebug $ debugMsg (show newSerial) (show currSerial)+ pure ls+ | currSerial == newSerial -> do+ case mResp of+ Nothing -> logDebug $ "Serial " <> show currSerial <> " already exists. Ignoring repeat request."+ Just idx -> respondClientWrite cid idx newSerial+ pure ls+ -- This is important case #2, where newSerial > currSerial+ | otherwise -> do+ let lsClientReqCache' = Map.insert cid (newSerial, Nothing) lsClientReqCache+ handleNewEntry+ pure ls { lsClientReqCache = lsClientReqCache' }+ where+ handleNewEntry = do+ newLogEntry <- mkNewLogEntry v newSerial+ appendLogEntries (Empty Seq.|> newLogEntry)+ aeData <- mkAppendEntriesData ls (FromClientWriteReq newLogEntry)+ broadcast (SendAppendEntriesRPC aeData)+ -------------------------------------------------------------------------------- -- | 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 :: Show v => LeaderState v -> TransitionM sm v (LeaderState v) incrCommitIndex leaderState@LeaderState{..} = do logDebug "Checking if commit index should be incremented..."- let (_, lastLogEntryTerm, _) = lsLastLogEntryData+ let lastEntryTerm = lastLogEntryTerm lsLastLogEntry currentTerm <- currentTerm <$> get- if majorityGreaterThanN && (lastLogEntryTerm == currentTerm)+ if majorityGreaterThanN && (lastEntryTerm == currentTerm) then do logDebug $ "Incrementing commit index to: " <> show n incrCommitIndex leaderState { lsCommitIndex = n }@@ -179,8 +202,8 @@ -- | Construct an AppendEntriesRPC given log entries and a leader state. mkAppendEntriesData :: Show v- => LeaderState- -> EntriesSpec v+ => LeaderState v+ -> AppendEntriesSpec v -> TransitionM sm v (AppendEntriesData v) mkAppendEntriesData ls entriesSpec = do currentTerm <- gets currentTerm
src/Raft/Log.hs view
@@ -8,27 +8,75 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE DefaultSignatures #-} {-# LANGUAGE TypeApplications #-}+{-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE GADTs #-} -module Raft.Log where+module Raft.Log ( + EntryIssuer(..),+ EntryValue(..),++ EntryHash,+ genesisHash,+ hashEntry,++ Entry(..),+ Entries,++ RaftInitLog(..),+ ReadEntriesSpec(..),+ ReadEntriesRes(..),+ IndexInterval(..),+ RaftReadLog(..),+ RaftWriteLog(..),+ RaftDeleteLog(..),+ DeleteSuccess(..),++ RaftLog(..),+ RaftLogError,+ RaftLogExceptions,++ updateLog,+ clientReqData,+ readEntries,++) where+ import Protolude +import qualified Crypto.Hash.SHA256 as SHA256++import qualified Data.ByteString as BS+import qualified Data.ByteString.Base16 as BS16+import qualified Data.Map as Map import Data.Serialize-import Data.Sequence (Seq(..), (|>))+import Data.Sequence (Seq(..), (|>), foldlWithIndex)+import qualified Data.Sequence as Seq +import Database.PostgreSQL.Simple.ToField (Action(..), ToField(..))+import Database.PostgreSQL.Simple.FromField (FromField(..), returnError, ResultError(..))+ import Raft.Types data EntryIssuer- = ClientIssuer ClientId+ = ClientIssuer ClientId SerialNum | LeaderIssuer LeaderId- deriving (Show, Generic, Serialize)+ deriving (Show, Read, Eq, Generic, Serialize) data EntryValue v = EntryValue v | NoValue -- ^ Used as a first committed entry of a new term- deriving (Show, Generic, Serialize)+ deriving (Show, Eq, Generic, Serialize) +newtype EntryHash = EntryHash ByteString+ deriving (Show, Eq, Ord, Generic, Serialize)++genesisHash :: EntryHash+genesisHash = EntryHash $ BS16.encode $ BS.replicate 32 0++hashEntry :: Serialize v => Entry v -> EntryHash+hashEntry = EntryHash . BS16.encode . SHA256.hash . encode+ -- | Representation of an entry in the replicated log data Entry v = Entry { entryIndex :: Index@@ -39,12 +87,70 @@ -- ^ Command to update state machine , entryIssuer :: EntryIssuer -- ^ Id of the client that issued the command- } deriving (Show, Generic, Serialize)+ , entryPrevHash :: EntryHash+ } deriving (Show, Eq, Generic, Serialize) type Entries v = Seq (Entry v) +data InvalidLog+ = InvalidIndex { expectedIndex :: Index, actualIndex :: Index }+ | InvalidPrevHash { expectedHash :: EntryHash, actualHash :: EntryHash }+ deriving (Show)++-- | For debugging & testing purposes+validateLog :: (Serialize v) => Entries v -> Either InvalidLog ()+validateLog es =+ case es of+ Empty -> Right ()+ e :<| _ ->+ second (const ()) $+ foldlWithIndex accValidateEntry (Right Nothing) es+ where+ accValidateEntry (Left err) _ _ = Left err+ accValidateEntry (Right mPrevEntry) idx e = validateEntry mPrevEntry idx e++ validateEntry mPrevEntry expectedIdx currEntry = do+ case mPrevEntry of+ Nothing -> validatePrevHash genesisHash currEntryPrevHash+ Just prevEntry -> validatePrevHash (hashEntry prevEntry) currEntryPrevHash+ validateIndex expectedEntryIdx currEntryIdx+ pure (Just currEntry)+ where+ currEntryIdx = entryIndex currEntry+ expectedEntryIdx = Index (fromIntegral expectedIdx + 1)++ currEntryPrevHash = entryPrevHash currEntry++ validateIndex :: Index -> Index -> Either InvalidLog ()+ validateIndex expectedIndex currIndex+ | expectedIndex /= currIndex =+ Left (InvalidIndex expectedIndex currIndex)+ | otherwise = Right ()++ validatePrevHash :: EntryHash -> EntryHash -> Either InvalidLog ()+ validatePrevHash expectedHash currHash+ | expectedHash /= currHash =+ Left (InvalidPrevHash expectedHash currHash)+ | otherwise = Right ()++clientReqData :: Entries v -> Map ClientId (SerialNum, Index)+clientReqData = go mempty+ where+ go acc es =+ case es of+ Empty -> acc+ e :<| rest ->+ case entryIssuer e of+ LeaderIssuer _ -> go acc rest+ ClientIssuer cid sn -> go (Map.insert cid (sn, entryIndex e) acc) rest++-- | Provides an interface to initialize a fresh log entry storage+class RaftInitLog m v where+ type RaftInitLogError m+ initializeLog :: Proxy v -> m (Either (RaftInitLogError m) ())+ -- | Provides an interface for nodes to write log entries to storage.-class Monad m => RaftWriteLog m v where+class (Show (RaftWriteLogError m), Monad m) => RaftWriteLog m v where type RaftWriteLogError m -- | Write the given log entries to storage writeLogEntries@@ -54,7 +160,7 @@ data DeleteSuccess v = DeleteSuccess -- | Provides an interface for nodes to delete log entries from storage.-class Monad m => RaftDeleteLog m v where+class (Show (RaftDeleteLogError m), 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.@@ -63,7 +169,7 @@ => 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+class (Show (RaftReadLogError m), Monad m) => RaftReadLog m v where type RaftReadLogError m -- | Read the log at a given index readLogEntry@@ -101,16 +207,19 @@ 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))+type RaftLog m v = (RaftInitLog m v, RaftReadLog m v, RaftWriteLog m v, RaftDeleteLog m v)+type RaftLogExceptions m = (Exception (RaftInitLogError 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)+data RaftLogError m where+ RaftLogInitError :: Show (RaftInitLogError m) => RaftInitLogError m -> RaftLogError m+ RaftLogReadError :: Show (RaftReadLogError m) => RaftReadLogError m -> RaftLogError m+ RaftLogWriteError :: Show (RaftWriteLogError m) => RaftWriteLogError m -> RaftLogError m+ RaftLogDeleteError :: Show (RaftDeleteLogError m) => RaftDeleteLogError m -> RaftLogError m +deriving instance Show (RaftLogError m)+ updateLog :: forall m v. ( RaftDeleteLog m v, Exception (RaftDeleteLogError m)@@ -126,3 +235,107 @@ case eDel of Left err -> pure (Left (RaftLogDeleteError err)) Right DeleteSuccess -> first RaftLogWriteError <$> writeLogEntries entries++--------------------------------------------------------------------------------+-- Reading entries by <X>+--+-- Note:+--+-- This would be best left up to the implementor, instead of+-- 'readEntriesFrom', it would be best to force the programmer to specify a+-- `readEntriesBetween` function that reads entries from disk between a+-- specified interval. For efficiency and ease of use, it might be best to+-- provide an instance of a typeclass from a popular haskell SQL lib for the+-- 'Entry v' type so that reading entries by index _or_ hash is efficient out+-- of the box.+--+--------------------------------------------------------------------------------++data IndexInterval = IndexInterval (Maybe Index) (Maybe Index)+ deriving (Show, Generic, Serialize)++data ReadEntriesSpec+ = ByIndex Index+ | ByIndices IndexInterval+ deriving (Show, Generic, Serialize)++data ReadEntriesError m where+ EntryDoesNotExist :: Either EntryHash Index -> ReadEntriesError m+ InvalidIntervalSpecified :: (Index, Index) -> ReadEntriesError m+ ReadEntriesError :: Exception (RaftReadLogError m) => RaftReadLogError m -> ReadEntriesError m++deriving instance Show (ReadEntriesError m)+deriving instance Typeable m => Exception (ReadEntriesError m)++-- | The result of reading one or more+data ReadEntriesRes v+ = OneEntry (Entry v)+ | ManyEntries (Entries v)++readEntries+ :: forall m v. (RaftReadLog m v, Exception (RaftReadLogError m))+ => ReadEntriesSpec+ -> m (Either (ReadEntriesError m) (ReadEntriesRes v))+readEntries res =+ case res of+ ByIndex idx -> do+ res <- readLogEntry idx+ case res of+ Left err -> pure (Left (ReadEntriesError err))+ Right Nothing -> pure (Left (EntryDoesNotExist (Right idx)))+ Right (Just e) -> pure (Right (OneEntry e))+ ByIndices interval -> fmap ManyEntries <$> readEntriesByIndices interval++-- | Read entries from the log between two indices+readEntriesByIndices+ :: forall m v. (RaftReadLog m v, Exception (RaftReadLogError m))+ => IndexInterval+ -> m (Either (ReadEntriesError m) (Entries v))+readEntriesByIndices (IndexInterval l h) =+ case (l,h) of+ (Nothing, Nothing) ->+ first ReadEntriesError <$> readLogEntriesFrom (Index 0)+ (Nothing, Just hidx) ->+ bimap ReadEntriesError (Seq.takeWhileL ((<= hidx) . entryIndex))+ <$> readLogEntriesFrom (Index 0)+ (Just lidx, Nothing) ->+ first ReadEntriesError <$> readLogEntriesFrom lidx+ (Just lidx, Just hidx)+ | lidx >= hidx ->+ pure (Left (InvalidIntervalSpecified (lidx, hidx)))+ | otherwise ->+ bimap ReadEntriesError (Seq.takeWhileL ((<= hidx) . entryIndex))+ <$> readLogEntriesFrom lidx++--------------------------------------------------------------------------------+-- PostgreSQL Utils+--------------------------------------------------------------------------------++instance Serialize v => ToField (EntryValue v) where+ toField = EscapeByteA . encode++instance (Typeable v, Serialize v) => FromField (EntryValue v) where+ fromField f mdata = do+ bs <- fromField f mdata+ case decode <$> bs of+ Nothing -> returnError UnexpectedNull f ""+ Just (Left err) -> returnError ConversionFailed f err+ Just (Right entry) -> return entry++-- TODO don't use show/read+instance ToField EntryIssuer where+ toField entryIssuer = Escape (show entryIssuer)++-- TODO don't use show/read+instance FromField EntryIssuer where+ fromField f mdata = do+ case readEither . toS <$> mdata of+ Nothing -> returnError UnexpectedNull f ""+ Just (Left err) -> returnError ConversionFailed f err+ Just (Right entryIssuer) -> return entryIssuer++instance ToField EntryHash where+ toField (EntryHash hbs) = toField hbs++instance FromField EntryHash where+ fromField f = fmap EntryHash . fromField f
+ src/Raft/Log/PostgreSQL.hs view
@@ -0,0 +1,292 @@+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE TemplateHaskell #-}++module Raft.Log.PostgreSQL (+ RaftPostgresT(..),+ runRaftPostgresT,+ runRaftPostgresM,++ raftDatabaseName,+ raftDatabaseConnInfo,+ initConnInfo,++ setupDB,+ deleteDB++) where++import Protolude hiding (atomically, try, bracket, catches, tryReadTMVar, newEmptyTMVar, STM, Handler)++import Control.Concurrent.Classy (STM, MonadConc, atomically)+import Control.Concurrent.Classy.STM.TMVar+import Control.Monad.Catch+import Control.Monad.Fail+import Control.Monad.Trans.Class++import Data.FileEmbed+import qualified Data.Sequence as Seq+import Data.Serialize (Serialize)++import Database.PostgreSQL.Simple+import Database.PostgreSQL.Simple.Types (Identifier(..))++import Raft.Client+import Raft.RPC+import Raft.Log+import Raft.StateMachine+import Raft.Persistent+import Raft.Types++data RaftPostgresEnv = RaftPostgresEnv+ { raftPostgresConnInfo :: ConnectInfo+ , raftPostgresConn :: TMVar (STM IO) Connection+ }++-- | A single threaded PostgreSQL storage monad transformer+newtype RaftPostgresT m a = RaftPostgresT { unRaftPostgresT :: ReaderT RaftPostgresEnv m a }+ deriving newtype (Functor, Applicative, Monad, MonadIO, MonadFail, MonadReader RaftPostgresEnv, Alternative, MonadPlus, MonadTrans)++deriving newtype instance MonadThrow m => MonadThrow (RaftPostgresT m)+deriving newtype instance MonadCatch m => MonadCatch (RaftPostgresT m)+deriving newtype instance MonadMask m => MonadMask (RaftPostgresT m)+deriving newtype instance MonadConc m => MonadConc (RaftPostgresT m)++runRaftPostgresT :: MonadIO m => ConnectInfo -> RaftPostgresT m a -> m a+runRaftPostgresT connInfo m = do+ connTMVar <- liftIO $ atomically newEmptyTMVar+ let postgresEnv = RaftPostgresEnv connInfo connTMVar+ runReaderT (unRaftPostgresT m) postgresEnv++type RaftPostgresM = RaftPostgresT IO++runRaftPostgresM :: ConnectInfo -> RaftPostgresM a -> IO a+runRaftPostgresM = runRaftPostgresT++data RaftPostgresError+ = RaftPostgresError PGError+ | RaftPostgresFailedToConnect PGError+ | RaftPostgresSerializeError Text+ deriving (Show)++instance Exception (RaftPostgresError)++-- | Helper function for RaftXLog typeclasses such that we can easily catch all+-- @PGError@s and connect to the DB if we haven't already+withRaftPostgresConn+ :: (MonadIO m, MonadConc m)+ => (Connection -> IO (Either RaftPostgresError a))+ -> RaftPostgresT m (Either RaftPostgresError a)+withRaftPostgresConn f = do+ mconn <- liftIO . atomically . tryReadTMVar =<< asks raftPostgresConn+ eConn <-+ case mconn of+ -- If there is no existing connection, attempt to connect and store the+ -- connection in the TMVar+ Nothing -> do+ eConn <- liftIO . tryPG . connect =<< asks raftPostgresConnInfo+ case eConn of+ Left err -> pure (Left (RaftPostgresFailedToConnect err))+ Right conn -> do+ connTMVar <- asks raftPostgresConn+ liftIO . atomically $ putTMVar connTMVar conn+ pure (Right conn)+ Just conn -> pure (Right conn)+ case eConn of+ Left err -> pure (Left err)+ Right conn -> do+ eRes <- liftIO (tryPG (f conn))+ case eRes of+ Left err -> pure (Left (RaftPostgresError err))+ Right (Left err) -> pure (Left err)+ Right (Right a) -> pure (Right a)++instance (MonadIO m) => RaftInitLog (RaftPostgresT m) v where+ type RaftInitLogError (RaftPostgresT m) = RaftPostgresError+ initializeLog _ = do+ eConn <- liftIO . setupDB =<< asks raftPostgresConnInfo+ case eConn of+ Left err -> pure (Left (RaftPostgresError err))+ Right conn -> do+ connTMVar <- asks raftPostgresConn+ fmap Right . liftIO . atomically $ putTMVar connTMVar conn++instance (Typeable v, Serialize v, MonadIO m, MonadConc m) => RaftReadLog (RaftPostgresT m) v where+ type RaftReadLogError (RaftPostgresT m) = RaftPostgresError+ readLogEntry idx =+ withRaftPostgresConn $ \conn ->+ Right . fmap rowTypeToEntry . listToMaybe <$>+ query conn "select * from entries where entryIndex = ?" (Only idx)++ readLogEntriesFrom idx =+ withRaftPostgresConn $ \conn ->+ Right . Seq.fromList . map rowTypeToEntry <$>+ query conn "select * from entries where entryIndex >= ?" (Only idx)++ readLastLogEntry =+ withRaftPostgresConn $ \conn ->+ Right . fmap rowTypeToEntry . listToMaybe <$>+ query_ conn "SELECT * FROM entries ORDER BY entryIndex DESC LIMIT 1"++instance (Serialize v, MonadIO m, MonadConc m) => RaftWriteLog (RaftPostgresT m) v where+ type RaftWriteLogError (RaftPostgresT m) = RaftPostgresError+ writeLogEntries entries =+ withRaftPostgresConn $ \conn ->+ fmap Right . void $+ executeMany conn "INSERT INTO entries VALUES (?,?,?,?,?,?)" (map entryToRowType (toList entries))++instance (Serialize v, MonadIO m, MonadConc m) => RaftDeleteLog (RaftPostgresT m) v where+ type RaftDeleteLogError (RaftPostgresT m) = RaftPostgresError+ deleteLogEntriesFrom idx =+ withRaftPostgresConn $ \conn ->+ fmap (const $ Right DeleteSuccess) . void $+ execute conn "DELETE FROM entries WHERE entryIndex >= ?" (Only idx)++--------------------------------------------------------------------------------+-- Inherited Raft instances+--------------------------------------------------------------------------------++instance RaftPersist m => RaftPersist (RaftPostgresT m) where+ type RaftPersistError (RaftPostgresT m) = RaftPersistError m+ initializePersistentState = lift initializePersistentState+ readPersistentState = lift readPersistentState+ writePersistentState = lift . writePersistentState++instance (Monad m, RaftSendRPC m v) => RaftSendRPC (RaftPostgresT m) v where+ sendRPC nid msg = lift (sendRPC nid msg)++instance (Monad m, RaftRecvRPC m v) => RaftRecvRPC (RaftPostgresT m) v where+ type RaftRecvRPCError (RaftPostgresT m) v = RaftRecvRPCError m v+ receiveRPC = lift receiveRPC++instance (Monad m, RaftSendClient m sm v) => RaftSendClient (RaftPostgresT m) sm v where+ sendClient clientId = lift . sendClient clientId++instance (Monad m, RaftRecvClient m v) => RaftRecvClient (RaftPostgresT m) v where+ type RaftRecvClientError (RaftPostgresT m) v = RaftRecvClientError m v+ receiveClient = lift receiveClient++instance RaftStateMachine m sm v => RaftStateMachine (RaftPostgresT m) sm v where+ validateCmd = lift . validateCmd+ askRaftStateMachinePureCtx = lift askRaftStateMachinePureCtx++--------------------------------------------------------------------------------++data EntryRow v = EntryRow+ { entryRowIndex :: Index+ , entryRowTerm :: Term+ , entryRowValueHash :: EntryHash+ , entryRowValue :: EntryValue v+ , entryRowIssuer :: EntryIssuer+ , entryRowPrevHash :: EntryHash+ } deriving (Show, Generic, ToRow, FromRow)++entryToRowType :: Serialize v => Entry v -> EntryRow v+entryToRowType entry@Entry{..} =+ EntryRow+ { entryRowIndex = entryIndex+ , entryRowTerm = entryTerm+ , entryRowValueHash = hashEntry entry+ , entryRowValue = entryValue+ , entryRowIssuer = entryIssuer+ , entryRowPrevHash = entryPrevHash+ }++rowTypeToEntry :: Serialize v => EntryRow v -> Entry v+rowTypeToEntry EntryRow{..} = Entry+ { entryIndex = entryRowIndex+ , entryTerm = entryRowTerm+ , entryValue = entryRowValue+ , entryIssuer = entryRowIssuer+ , entryPrevHash = entryRowPrevHash+ }++--------------------------------------------------------------------------------+-- Create Database for Entry Values+--------------------------------------------------------------------------------++data PGError+ = PGSqlError SqlError+ | PGFormatError FormatError+ | PGQueryError QueryError+ | PGResultError ResultError+ | PGUnexpectedError Text+ deriving (Show)++raftDatabasePrefix :: [Char]+raftDatabasePrefix = "libraft_"++raftDatabaseName :: [Char] -> [Char]+raftDatabaseName suffix = raftDatabasePrefix ++ suffix++raftDatabaseConnInfo :: [Char] -> [Char] -> [Char] -> ConnectInfo+raftDatabaseConnInfo usrnm pswd dbSuffix =+ defaultConnectInfo {+ connectUser = usrnm+ , connectPassword = pswd+ , connectDatabase = raftDatabaseName dbSuffix+ }++initConnInfo :: ConnectInfo+initConnInfo = defaultConnectInfo+ { connectDatabase = "postgres"+ , connectUser = "libraft_test"+ , connectPassword = "libraft_test"+ }++tryPG :: IO a -> IO (Either PGError a)+tryPG action =+ catches (Right <$> action)+ [ catchSqlError, catchFmtError, catchQueryError, catchResultError, catchAllError ]+ where+ handler :: Exception e => (e -> IO a) -> Handler IO a+ handler = Handler++ catchSqlError = handler (pure . Left . PGSqlError)+ catchFmtError = handler (pure . Left . PGFormatError)+ catchQueryError = handler (pure . Left . PGQueryError)+ catchResultError = handler (pure . Left . PGResultError)+ catchAllError = handler (pure . Left . PGUnexpectedError . (show :: SomeException -> Text))++createDB :: [Char] -> Connection -> IO Int64+createDB dbName conn =+ execute conn "CREATE DATABASE ?" (Only $ Identifier (toS dbName))++deleteDB :: [Char] -> Connection -> IO Int64+deleteDB dbName conn =+ execute conn "DROP DATABASE IF EXISTS ?" (Only $ Identifier (toS dbName))++-- | Load the raft schema at compile time using TH+raftSchema :: IsString a => a+raftSchema = $(makeRelativeToProject "postgres/entries.sql" >>= embedStringFile)++-- | Execute DB schema to build the raft database+createEntriesTable :: Connection -> IO Int64+createEntriesTable conn = execute_ conn raftSchema++-- | Create the libraft database to store all log entries+setupDB :: ConnectInfo -> IO (Either PGError Connection)+setupDB connInfo = tryPG $ do+ -- Create the database with the "postgres" user+ bracket (connect initConnInfo) close $ \conn ->+ createDB dbName conn+ -- Connect to the DB with the provided connection info+ conn <- connect connInfo+ -- Create the log entries table with the resulting connection+ createEntriesTable conn+ pure conn+ where++ dbName = connectDatabase connInfo
src/Raft/Logging.hs view
@@ -1,6 +1,10 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE RecordWildCards #-} module Raft.Logging where@@ -11,25 +15,33 @@ import Control.Monad.State (modify') import Data.Time+import Data.Time.Clock.System import Raft.NodeState import Raft.Types +-- | Representation of the logs' context+data LogCtx+ = LogCtx+ { logCtxDest :: LogDest+ , logCtxSeverity :: Severity+ }+ | NoLogs+ -- | Representation of the logs' destination data LogDest = LogFile FilePath | LogStdout- | NoLogs -- | Representation of the severity of the logs data Severity- = Info- | Debug+ = Debug+ | Info | Critical- deriving (Show)+ deriving (Show, Eq, Ord) data LogMsg = LogMsg- { mTime :: Maybe UTCTime+ { mTime :: Maybe SystemTime , severity :: Severity , logMsgData :: LogMsgData }@@ -44,8 +56,10 @@ logMsgToText (LogMsg mt s d) = maybe "" timeToText mt <> "(" <> show s <> ")" <> " " <> logMsgDataToText d where- timeToText :: UTCTime -> Text- timeToText t = "[" <> toS (timeToText' t) <> "]"+ timeToText :: SystemTime -> Text+ timeToText sysTime = "[" <> toS (timeToText' (systemToUTCTime sysTime)) <> ":" <> show ms <> "]"+ where+ ms = (systemNanoseconds sysTime) `div` 1000000 timeToText' = formatTime defaultTimeLocale (iso8601DateFormat (Just "%H:%M:%S")) @@ -53,85 +67,101 @@ logMsgDataToText LogMsgData{..} = "<" <> toS logMsgNodeId <> " | " <> show logMsgNodeState <> ">: " <> logMsg -class Monad m => RaftLogger m where- loggerNodeId :: m NodeId- loggerNodeState :: m RaftNodeState+class Monad m => RaftLogger v m | m -> v where+ loggerCtx :: m (NodeId, RaftNodeState v) -mkLogMsgData :: RaftLogger m => Text -> m LogMsgData+mkLogMsgData :: RaftLogger v m => Text -> m (LogMsgData) mkLogMsgData msg = do- nid <- loggerNodeId- ns <- nodeMode <$> loggerNodeState- pure $ LogMsgData nid ns msg+ (nid, nodeState) <- loggerCtx+ let mode = nodeMode nodeState+ pure $ LogMsgData nid mode msg -instance RaftLogger m => RaftLogger (RaftLoggerT m) where- loggerNodeId = lift loggerNodeId- loggerNodeState = lift loggerNodeState+instance RaftLogger v m => RaftLogger v (RaftLoggerT v m) where+ loggerCtx = lift loggerCtx -------------------------------------------------------------------------------- -- 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 ()+logToDest :: MonadIO m => LogCtx -> LogMsg -> m ()+logToDest LogCtx{..} logMsg =+ case logCtxDest of+ LogStdout -> if severity logMsg >= logCtxSeverity+ then liftIO $ putText (logMsgToText logMsg)+ else pure ()+ LogFile fp -> if severity logMsg >= logCtxSeverity+ then liftIO $ appendFile fp (logMsgToText logMsg <> "\n")+ else pure ()+logToDest NoLogs _ = pure () -logToStdout :: MonadIO m => LogMsg -> m ()-logToStdout = logToDest LogStdout+logToStdout :: MonadIO m => Severity -> LogMsg -> m ()+logToStdout s = logToDest $ LogCtx LogStdout s -logToFile :: MonadIO m => FilePath -> LogMsg -> m ()-logToFile fp = logToDest (LogFile fp)+logToFile :: MonadIO m => FilePath -> Severity -> LogMsg -> m ()+logToFile fp s = logToDest $ LogCtx (LogFile fp) s -logWithSeverityIO :: (RaftLogger m, MonadIO m) => Severity -> LogDest -> Text -> m ()-logWithSeverityIO s logDest msg = do+logWithSeverityIO :: forall m v. (RaftLogger v m, MonadIO m) => Severity -> LogCtx -> Text -> m ()+logWithSeverityIO s logCtx msg = do logMsgData <- mkLogMsgData msg- now <- liftIO getCurrentTime- let logMsg = LogMsg (Just now) s logMsgData- logToDest logDest logMsg+ sysTime <- liftIO getSystemTime+ let logMsg = LogMsg (Just sysTime) s logMsgData+ logToDest logCtx logMsg -logInfoIO :: (RaftLogger m, MonadIO m) => LogDest -> Text -> m ()+logInfoIO :: (RaftLogger v m, MonadIO m) => LogCtx -> Text -> m () logInfoIO = logWithSeverityIO Info -logDebugIO :: (RaftLogger m, MonadIO m) => LogDest -> Text -> m ()+logDebugIO :: (RaftLogger v m, MonadIO m) => LogCtx -> Text -> m () logDebugIO = logWithSeverityIO Debug -logCriticalIO :: (RaftLogger m, MonadIO m) => LogDest -> Text -> m ()+logCriticalIO :: (RaftLogger v m, MonadIO m) => LogCtx -> Text -> m () logCriticalIO = logWithSeverityIO Critical -------------------------------------------------------------------------------- -- Pure Logging -------------------------------------------------------------------------------- -newtype RaftLoggerT m a = RaftLoggerT {+newtype RaftLoggerT v 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+ => RaftLoggerT v m a -- ^ The computation from which to extract the logs -> m (a, [LogMsg]) runRaftLoggerT = flip runStateT [] . unRaftLoggerT -type RaftLoggerM = RaftLoggerT Identity+type RaftLoggerM v = RaftLoggerT v Identity runRaftLoggerM- :: RaftLoggerM a+ :: RaftLoggerM v a -> (a, [LogMsg]) runRaftLoggerM = runIdentity . runRaftLoggerT -logWithSeverity :: RaftLogger m => Severity -> Text -> RaftLoggerT m ()+logWithSeverity :: RaftLogger v m => Severity -> Text -> RaftLoggerT v m () logWithSeverity s txt = do !logMsgData <- mkLogMsgData txt let !logMsg = LogMsg Nothing s logMsgData modify' (++ [logMsg]) -logInfo :: RaftLogger m => Text -> RaftLoggerT m ()+logInfo :: RaftLogger v m => Text -> RaftLoggerT v m () logInfo = logWithSeverity Info -logDebug :: RaftLogger m => Text -> RaftLoggerT m ()+logDebug :: RaftLogger v m => Text -> RaftLoggerT v m () logDebug = logWithSeverity Debug -logCritical :: RaftLogger m => Text -> RaftLoggerT m ()+logCritical :: RaftLogger v m => Text -> RaftLoggerT v m () logCritical = logWithSeverity Critical++--------------------------------------------------------------------------------+-- Panic after logging+--------------------------------------------------------------------------------++logAndPanic :: RaftLogger v m => Text -> m a+logAndPanic msg = do+ runRaftLoggerT $ logCritical msg+ panic ("logAndPanic: " <> msg)++logAndPanicIO :: (RaftLogger v m, MonadIO m) => LogCtx -> Text -> m a+logAndPanicIO logCtx msg = do+ logCriticalIO logCtx msg+ panic ("logAndPanicIO: " <> msg)
src/Raft/Monad.hs view
@@ -13,7 +13,10 @@ module Raft.Monad where import Protolude hiding (pass)++import Control.Arrow ((&&&)) import Control.Monad.RWS+ import qualified Data.Set as Set import Raft.Action@@ -28,37 +31,9 @@ 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+-- Raft Transition Monad -------------------------------------------------------------------------------- tellAction :: Action sm v -> TransitionM sm v ()@@ -67,14 +42,14 @@ tellActions :: [Action sm v] -> TransitionM sm v () tellActions as = tell as -data TransitionEnv sm = TransitionEnv+data TransitionEnv sm v = TransitionEnv { nodeConfig :: NodeConfig , stateMachine :: sm- , nodeState :: RaftNodeState+ , nodeState :: RaftNodeState v } newtype TransitionM sm v a = TransitionM- { unTransitionM :: RaftLoggerT (RWS (TransitionEnv sm) [Action sm v] PersistentState) a+ { unTransitionM :: RaftLoggerT v (RWS (TransitionEnv sm v) [Action sm v] PersistentState) a } deriving (Functor, Applicative, Monad) instance MonadWriter [Action sm v] (TransitionM sm v) where@@ -82,7 +57,7 @@ listen = TransitionM . RaftLoggerT . listen . unRaftLoggerT . unTransitionM pass = TransitionM . RaftLoggerT . pass . unRaftLoggerT . unTransitionM -instance MonadReader (TransitionEnv sm) (TransitionM sm v) where+instance MonadReader (TransitionEnv sm v) (TransitionM sm v) where ask = TransitionM . RaftLoggerT $ ask local f = TransitionM . RaftLoggerT . local f . unRaftLoggerT . unTransitionM @@ -90,12 +65,11 @@ 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+instance RaftLogger v (RWS (TransitionEnv sm v) [Action sm v] PersistentState) where+ loggerCtx = asks ((configNodeId . nodeConfig) &&& nodeState) runTransitionM- :: TransitionEnv sm+ :: TransitionEnv sm v -> PersistentState -> TransitionM sm v a -> ((a, [LogMsg]), PersistentState, [Action sm v])@@ -109,9 +83,9 @@ -- 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)+type RPCHandler ns sm r v = (RPCType r v, Show v) => NodeState ns v -> NodeId -> r -> TransitionM sm v (ResultState ns v)+type TimeoutHandler ns sm v = Show v => NodeState ns v -> Timeout -> TransitionM sm v (ResultState ns v)+type ClientReqHandler ns sm v = Show v => NodeState ns v -> ClientRequest v -> TransitionM sm v (ResultState ns v) -------------------------------------------------------------------------------- -- RWS Helpers@@ -136,25 +110,43 @@ redirectClientToLeader :: ClientId -> CurrentLeader -> TransitionM sm v () redirectClientToLeader clientId currentLeader = do- let clientRedirResp = ClientRedirectResponse (ClientRedirResp currentLeader)- tellAction (RespondToClient clientId clientRedirResp)+ let clientRedirRespSpec = ClientRedirRespSpec currentLeader+ tellAction (RespondToClient clientId clientRedirRespSpec) -respondClientRead :: ClientId -> TransitionM sm v ()-respondClientRead clientId = do- clientReadResp <- ClientReadResponse . ClientReadResp <$> asks stateMachine- tellAction (RespondToClient clientId clientReadResp)+respondClientRead :: ClientId -> ClientReadReq -> TransitionM sm v ()+respondClientRead clientId readReq = do+ readReqData <-+ case readReq of+ ClientReadEntries res -> pure (ClientReadRespSpecEntries res)+ ClientReadStateMachine -> do+ sm <- asks stateMachine+ pure (ClientReadRespSpecStateMachine sm)+ tellAction . RespondToClient clientId . ClientReadRespSpec $ readReqData ++respondClientWrite :: ClientId -> Index -> SerialNum -> TransitionM sm v ()+respondClientWrite cid entryIdx sn =+ tellAction (RespondToClient cid (ClientWriteRespSpec entryIdx sn))++respondClientRedir :: ClientId -> CurrentLeader -> TransitionM sm v ()+respondClientRedir cid cl =+ tellAction (RespondToClient cid (ClientRedirRespSpec cl))+ appendLogEntries :: Show v => Seq (Entry v) -> TransitionM sm v () appendLogEntries = tellAction . AppendLogEntries +updateClientReqCacheFromIdx :: Index -> TransitionM sm v ()+updateClientReqCacheFromIdx = tellAction . UpdateClientReqCacheFrom+ -------------------------------------------------------------------------------- startElection :: Index -> Index- -> (Index, Term) -- ^ Last log entry data- -> TransitionM sm v CandidateState-startElection commitIndex lastApplied lastLogEntryData = do+ -> LastLogEntry v+ -> ClientWriteReqCache+ -> TransitionM sm v (CandidateState v)+startElection commitIndex lastApplied lastLogEntry clientReqCache = do incrementTerm voteForSelf resetElectionTimeout@@ -165,7 +157,8 @@ { csCommitIndex = commitIndex , csLastApplied = lastApplied , csVotes = Set.singleton selfNodeId- , csLastLogEntryData = lastLogEntryData+ , csLastLogEntry = lastLogEntry+ , csClientReqCache = clientReqCache } where requestVoteMessage = do@@ -175,8 +168,8 @@ RequestVote { rvTerm = term , rvCandidateId = selfNodeId- , rvLastLogIndex = fst lastLogEntryData- , rvLastLogTerm = snd lastLogEntryData+ , rvLastLogIndex = lastLogEntryIndex lastLogEntry+ , rvLastLogTerm = lastLogEntryTerm lastLogEntry } incrementTerm = do
src/Raft/NodeState.hs view
@@ -4,6 +4,7 @@ {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE StrictData #-} module Raft.NodeState where @@ -12,7 +13,9 @@ import qualified Data.Serialize as S import Data.Sequence (Seq(..)) +import Raft.Client import Raft.Log+import Raft.Client (SerialNum) import Raft.Types data Mode@@ -31,6 +34,7 @@ HigherTermFoundCandidate :: Transition 'Candidate 'Follower BecomeLeader :: Transition 'Candidate 'Leader + HandleClientReq :: Transition 'Leader 'Leader SendHeartbeat :: Transition 'Leader 'Leader DiscoverNewLeader :: Transition 'Leader 'Follower HigherTermFoundLeader :: Transition 'Leader 'Follower@@ -39,37 +43,46 @@ 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+data ResultState init v where+ ResultState+ :: Show v+ => Transition init res+ -> NodeState res v+ -> ResultState init v -deriving instance Show (ResultState init)+deriving instance Show v => Show (ResultState init v) followerResultState- :: Transition init 'Follower- -> FollowerState- -> ResultState init+ :: Show v+ => Transition init 'Follower+ -> FollowerState v+ -> ResultState init v followerResultState transition fstate = ResultState transition (NodeFollowerState fstate) candidateResultState- :: Transition init 'Candidate- -> CandidateState- -> ResultState init+ :: Show v+ => Transition init 'Candidate+ -> CandidateState v+ -> ResultState init v candidateResultState transition cstate = ResultState transition (NodeCandidateState cstate) leaderResultState- :: Transition init 'Leader- -> LeaderState- -> ResultState init+ :: Show v+ => Transition init 'Leader+ -> LeaderState v+ -> ResultState init v leaderResultState transition lstate = ResultState transition (NodeLeaderState lstate) -- | Existential type hiding the internal node state-data RaftNodeState where- RaftNodeState :: { unRaftNodeState :: NodeState s } -> RaftNodeState+data RaftNodeState v where+ RaftNodeState :: { unRaftNodeState :: NodeState s v } -> RaftNodeState v -nodeMode :: RaftNodeState -> Mode+deriving instance Show v => Show (RaftNodeState v)++nodeMode :: RaftNodeState v -> Mode nodeMode (RaftNodeState rns) = case rns of NodeFollowerState _ -> Follower@@ -77,63 +90,99 @@ NodeLeaderState _ -> Leader -- | A node in Raft begins as a follower-initRaftNodeState :: RaftNodeState+initRaftNodeState :: RaftNodeState v initRaftNodeState = RaftNodeState $ NodeFollowerState FollowerState { fsCommitIndex = index0 , fsLastApplied = index0 , fsCurrentLeader = NoLeader- , fsLastLogEntryData = (index0, term0)+ , fsLastLogEntry = NoLogEntries , fsTermAtAEPrevIndex = Nothing+ , fsClientReqCache = mempty } -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+data NodeState (a :: Mode) v where+ NodeFollowerState :: FollowerState v -> NodeState 'Follower v+ NodeCandidateState :: CandidateState v -> NodeState 'Candidate v+ NodeLeaderState :: LeaderState v -> NodeState 'Leader v -deriving instance Show (NodeState v)+deriving instance Show v => Show (NodeState s 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)+data LastLogEntry v+ = LastLogEntry (Entry v)+ | NoLogEntries+ deriving (Show) -instance S.Serialize CurrentLeader+hashLastLogEntry :: S.Serialize v => LastLogEntry v -> EntryHash+hashLastLogEntry = \case+ LastLogEntry e -> hashEntry e+ NoLogEntries -> genesisHash -data FollowerState = FollowerState+lastLogEntryIndex :: LastLogEntry v -> Index+lastLogEntryIndex = \case+ LastLogEntry e -> entryIndex e+ NoLogEntries -> index0++lastLogEntryTerm :: LastLogEntry v -> Term+lastLogEntryTerm = \case+ LastLogEntry e -> entryTerm e+ NoLogEntries -> term0++lastLogEntryIndexAndTerm :: LastLogEntry v -> (Index, Term)+lastLogEntryIndexAndTerm lle = (lastLogEntryIndex lle, lastLogEntryTerm lle)++lastLogEntryIssuer :: LastLogEntry v -> Maybe EntryIssuer+lastLogEntryIssuer = \case+ LastLogEntry e -> Just (entryIssuer e)+ NoLogEntries -> Nothing++data FollowerState v = 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)+ , fsLastLogEntry :: LastLogEntry v -- ^ 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+ , fsClientReqCache :: ClientWriteReqCache+ -- ^ The client write request cache, growing linearly with the number of+ -- clients } deriving (Show) -data CandidateState = CandidateState+data CandidateState v = 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)+ , csLastLogEntry :: LastLogEntry v -- ^ Index and term of the last log entry in the node's log+ , csClientReqCache :: ClientWriteReqCache+ -- ^ The client write request cache, growing linearly with the number of+ -- clients } deriving (Show) -type ClientReadReqs = Map Int (ClientId, Int)+data ClientReadReqData = ClientReadReqData+ { crrdClientId :: ClientId+ , crrdReadReq :: ClientReadReq+ } deriving (Show) -data LeaderState = LeaderState+-- | The type mapping the number of the read request serviced to the id of the+-- client that issued it and the number of success responses from followers+-- confirming the leadership of the current leader+type ClientReadReqs = Map Int (ClientReadReqData, Int)++-- | The type mapping client ids to the serial number of their latest write+-- requests and the index of the entry if it has been replicated.+type ClientWriteReqCache = Map ClientId (SerialNum, Maybe Index)++data LeaderState v = LeaderState { lsCommitIndex :: Index -- ^ Index of highest log entry known to be committed , lsLastApplied :: Index@@ -142,11 +191,7 @@ -- ^ 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- )+ , lsLastLogEntry :: LastLogEntry v -- ^ 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@@ -154,6 +199,8 @@ , lsReadRequest :: ClientReadReqs -- ^ The number of successful responses received regarding a specific read -- request heartbeat.+ , lsClientReqCache :: ClientWriteReqCache+ -- ^ The cache of client write requests received by the leader } deriving (Show) --------------------------------------------------------------------------------@@ -161,35 +208,32 @@ -------------------------------------------------------------------------------- -- | Update the last log entry in the node's log-setLastLogEntryData :: NodeState ns -> Entries v -> NodeState ns-setLastLogEntryData nodeState entries =+setLastLogEntry :: NodeState s v -> Entries v -> NodeState s v+setLastLogEntry nodeState entries = case entries of Empty -> nodeState- _ :|> e ->+ _ :|> e -> do+ let lastLogEntry = LastLogEntry e case nodeState of NodeFollowerState fs ->- NodeFollowerState fs- { fsLastLogEntryData = (entryIndex e, entryTerm e) }+ NodeFollowerState fs { fsLastLogEntry = lastLogEntry } NodeCandidateState cs ->- NodeCandidateState cs- { csLastLogEntryData = (entryIndex e, entryTerm e) }+ NodeCandidateState cs { csLastLogEntry = lastLogEntry } NodeLeaderState ls ->- NodeLeaderState ls- { lsLastLogEntryData = (entryIndex e, entryTerm e, Just (entryIssuer e)) }+ NodeLeaderState ls { lsLastLogEntry = lastLogEntry } -- | 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 =+getLastLogEntry :: NodeState ns v -> LastLogEntry v+getLastLogEntry nodeState = case nodeState of- NodeFollowerState fs -> fsLastLogEntryData fs- NodeCandidateState cs -> csLastLogEntryData cs- NodeLeaderState ls -> let (peTerm, peIndex, _) = lsLastLogEntryData ls- in (peTerm, peIndex)+ NodeFollowerState fs -> fsLastLogEntry fs+ NodeCandidateState cs -> csLastLogEntry cs+ NodeLeaderState ls -> lsLastLogEntry ls -- | 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 ns v -> (Index, Index) getLastAppliedAndCommitIndex nodeState = case nodeState of NodeFollowerState fs -> (fsLastApplied fs, fsCommitIndex fs)@@ -197,21 +241,21 @@ NodeLeaderState ls -> (lsLastApplied ls, lsCommitIndex ls) -- | Check if node is in a follower state-isFollower :: NodeState s -> Bool+isFollower :: NodeState s v -> Bool isFollower nodeState = case nodeState of NodeFollowerState _ -> True _ -> False -- | Check if node is in a candidate state-isCandidate :: NodeState s -> Bool+isCandidate :: NodeState s v -> Bool isCandidate nodeState = case nodeState of NodeCandidateState _ -> True _ -> False -- | Check if node is in a leader state-isLeader :: NodeState s -> Bool+isLeader :: NodeState s v -> Bool isLeader nodeState = case nodeState of NodeLeaderState _ -> True
src/Raft/Persistent.hs view
@@ -17,6 +17,9 @@ -- state to disk. class Monad m => RaftPersist m where type RaftPersistError m+ initializePersistentState+ :: Exception (RaftPersistError m)+ => m (Either (RaftPersistError m) ()) readPersistentState :: Exception (RaftPersistError m) => m (Either (RaftPersistError m) PersistentState)@@ -30,7 +33,7 @@ -- ^ Last term server has seen , votedFor :: Maybe NodeId -- ^ Candidate id that received vote in current term- } deriving (Show, Generic, S.Serialize)+ } deriving (Show, Eq, Generic, S.Serialize) -- | A node initiates its persistent state with term 0 and with its vote blank initPersistentState :: PersistentState
src/Raft/RPC.hs view
@@ -66,7 +66,7 @@ | FromClientReadReq Int deriving (Show) -data EntriesSpec v+data AppendEntriesSpec v = FromIndex Index | FromNewLeader (Entry v) | FromClientWriteReq (Entry v)@@ -78,7 +78,7 @@ data AppendEntriesData v = AppendEntriesData { aedTerm :: Term , aedLeaderCommit :: Index- , aedEntriesSpec :: EntriesSpec v+ , aedEntriesSpec :: AppendEntriesSpec v } deriving (Show) -- | Representation of a message sent from a leader to its peers
+ src/Raft/StateMachine.hs view
@@ -0,0 +1,52 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeFamilyDependencies #-}++module Raft.StateMachine (+ RaftStateMachinePure(..),+ RaftStateMachinePureError(..),+ RaftStateMachine(..),++ applyLogEntry+) where++import Protolude++import Raft.Log (Entry(..), EntryValue(..))++--------------------------------------------------------------------------------+-- 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 RaftStateMachinePure sm v | sm -> v where+ data RaftStateMachinePureError sm v+ type RaftStateMachinePureCtx sm v = ctx | ctx -> sm v+ rsmTransition+ :: RaftStateMachinePureCtx sm v+ -> sm+ -> v+ -> Either (RaftStateMachinePureError sm v) sm++class (Monad m, RaftStateMachinePure sm v) => RaftStateMachine m sm v | m sm -> v where+ validateCmd :: v -> m (Either (RaftStateMachinePureError sm v) ())+ askRaftStateMachinePureCtx :: m (RaftStateMachinePureCtx sm v)++applyLogEntry+ :: RaftStateMachine m sm v+ => sm+ -> Entry v+ -> m (Either (RaftStateMachinePureError sm v) sm)+applyLogEntry 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 <- askRaftStateMachinePureCtx+ pure (rsmTransition ctx sm v)
src/Raft/Types.hs view
@@ -1,6 +1,8 @@ {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE KindSignatures #-} {-# LANGUAGE RankNTypes #-}@@ -12,6 +14,9 @@ import Data.Serialize import Numeric.Natural (Natural) +import Database.PostgreSQL.Simple.ToField+import Database.PostgreSQL.Simple.FromField+ -------------------------------------------------------------------------------- -- NodeIds --------------------------------------------------------------------------------@@ -22,20 +27,46 @@ -- | Unique identifier of a client newtype ClientId = ClientId NodeId- deriving (Show, Eq, Ord, Generic, Serialize)+ deriving stock (Show, Read, Eq, Ord, Generic)+ deriving newtype Serialize -- | Unique identifier of a leader newtype LeaderId = LeaderId { unLeaderId :: NodeId }- deriving (Show, Eq, Generic, Serialize)+ deriving stock (Show, Read, Eq, Generic)+ deriving newtype Serialize +-- | 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 stock (Show, Eq, Generic)+ deriving anyclass (Serialize)+ ---------- -- Term -- ---------- -- | Representation of monotonic election terms newtype Term = Term Natural- deriving (Show, Eq, Ord, Enum, Generic, Serialize)+ deriving stock (Generic)+ deriving newtype (Show, Eq, Ord, Enum, Serialize) +instance ToField Term where+ toField (Term n) = toField (fromIntegral n :: Int)++instance FromField Term where+ fromField f mdata =+ case (intToNatural <=< readEither) . toS <$> mdata of+ Nothing -> returnError UnexpectedNull f ""+ Just (Left err) -> returnError ConversionFailed f err+ Just (Right nat) -> return (Term nat)++intToNatural :: Int -> Either [Char] Natural+intToNatural n+ | n < 0 = Left "Natural numbers must be >= 0"+ | otherwise = Right (fromIntegral n)+ -- | Initial term. Terms start at 0 term0 :: Term term0 = Term 0@@ -53,8 +84,19 @@ -- | Representation of monotonic indices newtype Index = Index Natural- deriving (Show, Eq, Ord, Enum, Num, Integral, Real, Generic, Serialize)+ deriving stock (Show, Generic)+ deriving newtype (Eq, Ord, Enum, Num, Integral, Real, Serialize) +instance ToField Index where+ toField (Index n) = toField (fromIntegral n :: Int)++instance FromField Index where+ fromField f mdata =+ case (intToNatural <=< readEither) . toS <$> mdata of+ Nothing -> returnError UnexpectedNull f ""+ Just (Left err) -> returnError ConversionFailed f err+ Just (Right nat) -> return (Index nat)+ -- | Initial index. Indeces start at 0 index0 :: Index index0 = Index 0@@ -67,3 +109,11 @@ decrIndexWithDefault0 :: Index -> Index decrIndexWithDefault0 (Index 0) = index0 decrIndexWithDefault0 i = pred i++-----------------------------------+-- Client Request Serial Numbers --+-----------------------------------++newtype SerialNum = SerialNum Natural+ deriving stock (Show, Read, Generic)+ deriving newtype (Eq, Ord, Enum, Num, Serialize)
+ test/QuickCheckStateMachine.hs view
@@ -0,0 +1,396 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE StandaloneDeriving #-}++module QuickCheckStateMachine where++import Control.Concurrent (threadDelay)+import Control.Exception (bracket)+import Control.Monad.IO.Class (liftIO)+import Data.Bifunctor (bimap)+import Data.Char (isDigit)+import Data.List (isInfixOf, (\\))+import Data.Maybe (isJust, isNothing)+import qualified Data.Set as Set+import Data.TreeDiff (ToExpr)+import GHC.Generics (Generic, Generic1)+import Prelude hiding (notElem)+import System.Directory (removePathForcibly)+import System.IO (BufferMode (NoBuffering),+ Handle, IOMode (WriteMode),+ hClose, hGetLine, hPutStrLn,+ hPutStrLn, hSetBuffering,+ openFile)+import System.Process (ProcessHandle, StdStream (CreatePipe, UseHandle),+ callCommand, createProcess_,+ getPid, getProcessExitCode, waitForProcess,+ proc, std_err, std_in, std_out,+ terminateProcess)+import System.Timeout (timeout)+import Test.QuickCheck (Gen, Property, arbitrary,+ elements, frequency,+ noShrinking, shrink,+ verboseCheck, withMaxSuccess,+ (===))+import Test.QuickCheck.Monadic (monadicIO)+import Test.StateMachine (Concrete, GenSym, Logic (..),+ Opaque (..), Reason (Ok),+ Reference, StateMachine (..),+ Symbolic, forAllCommands, notElem,+ genSym, opaque, prettyCommands,+ reference, runCommands, (.&&),+ (.//), (.<), (.==), (.>=))+import Test.StateMachine.Types (Command(..), Commands (..), Reference(..), Symbolic(..), Var(..))+import qualified Test.StateMachine.Types.Rank2 as Rank2+import Text.Read (readEither)++import qualified Data.Functor.Classes+import Debug.Trace (trace)++------------------------------------------------------------------------++type Port = Int++data Persistence = Fresh | Existing+ deriving (Show)++data ClientHandleRefs (r :: * -> *) = ClientHandleRefs+ { client_hin :: Reference (Opaque Handle) r+ , client_hout :: Reference (Opaque Handle) r+ } deriving (Show, Generic, Generic1, Rank2.Functor, Rank2.Foldable, Rank2.Traversable)++deriving instance ToExpr (ClientHandleRefs Concrete)++type ProcessHandleRef r = Reference (Opaque ProcessHandle) r++data Action (r :: * -> *)+ = SpawnNode Port Persistence+ | SpawnClient Port+ | KillNode (Port, ProcessHandleRef r)+ | Set (ClientHandleRefs r) Integer+ | Read (ClientHandleRefs r)+ | Incr (ClientHandleRefs r)+ | BreakConnection (Port, ProcessHandleRef r)+ | FixConnection (Port, ProcessHandleRef r)+ deriving (Show, Generic1, Rank2.Functor, Rank2.Foldable, Rank2.Traversable)++data Response (r :: * -> *)+ = SpawnedNode (ProcessHandleRef r)+ | SpawnedClient (ClientHandleRefs r) (ProcessHandleRef r)+ | SpawnFailed Port+ | BrokeConnection+ | FixedConnection+ | Ack+ | Timeout+ | Value (Either String Integer)+ deriving (Show, Generic1, Rank2.Foldable)++data Model (r :: * -> *) = Model+ { nodes :: [(Port, ProcessHandleRef r)]+ , client :: Maybe (ClientHandleRefs r, ProcessHandleRef r)+ , started :: Bool+ , value :: Maybe Integer+ , isolated :: [(Port, ProcessHandleRef r)]+ }+ deriving (Show, Generic)++deriving instance ToExpr (Model Concrete)++initModel :: Model r+initModel = Model [] Nothing False Nothing []++transition :: Data.Functor.Classes.Show1 r => Model r -> Action r -> Response r -> Model r+transition Model {..} act resp = case (act, resp) of+ (SpawnNode port _, SpawnedNode ph) ->+ let newNodes = nodes ++ [(port, ph)]+ in if length newNodes == 3+ then Model { nodes = newNodes, started = True, .. }+ else Model { nodes = newNodes, .. }+ (SpawnNode {}, SpawnFailed _) -> Model {..}+ (SpawnClient {}, SpawnedClient chrs cph) -> Model { client = Just (chrs, cph), .. }+ (SpawnClient {}, SpawnFailed _) -> Model {..}+ (KillNode (port, _ph), Ack) -> Model { nodes = filter ((/= port) . fst) nodes, isolated = filter ((/= port) . fst) isolated, .. }+ (Set _ i, Ack) -> Model { value = Just i, .. }+ (Read {}, Value _i) -> Model {..}+ (Incr {}, Ack) -> Model { value = succ <$> value, ..}+ (BreakConnection node, BrokeConnection) -> Model { isolated = node:isolated, .. }+ (FixConnection (port,_), FixedConnection) -> Model { isolated = filter ((/= port) . fst) isolated, .. }+ (Read {}, Timeout) -> Model {..}+ (Set {}, Timeout) -> Model {..}+ (Incr {}, Timeout) -> Model {..}+ unaccounted -> trace (show unaccounted) $ error "transition"++-- TODO I don't think the precondition is being checked...+precondition :: Model Symbolic -> Action Symbolic -> Logic+precondition Model {..} act = case act of+ SpawnNode {} -> length nodes .< 3+ SpawnClient {} -> length nodes .== 3 .&& Boolean (isNothing client)+ KillNode {} -> length nodes .== 3+ Set _ i -> length nodes .== 3 .&& i .>= 0+ Read _ -> length nodes .== 3 .&& Boolean (isJust value)+ Incr _ -> length nodes .== 3 .&& Boolean (isJust value)+ BreakConnection (port, _) -> length nodes .== 3 .&& port `notElem` map fst isolated+ FixConnection {} -> length nodes .== 3 .&& Boolean (not (null isolated))++postcondition :: Model Concrete -> Action Concrete -> Response Concrete -> Logic+postcondition Model {..} act resp = case (act, resp) of+ (Read _, Value (Right i)) -> Just i .== value+ (Read _, Value (Left e)) -> Bot .// e+ (SpawnNode {}, SpawnedNode {}) -> Top+ (SpawnNode {}, SpawnFailed {}) -> Bot .// "SpawnFailed - Node"+ (SpawnClient {}, SpawnedClient {}) -> Top+ (SpawnClient {}, SpawnFailed {}) -> Bot .// "SpawnFailed - Client"+ (KillNode {}, Ack) -> Top+ (Set {}, Ack) -> Top+ (Incr {}, Ack) -> Top+ (BreakConnection {}, BrokeConnection) -> Top+ (FixConnection {}, FixedConnection) -> Top+ (Read _, Timeout) -> Boolean (not (null isolated)) .// "Read timeout"+ (Set {}, Timeout) -> Boolean (not (null isolated)) .// "Set timeout"+ (Incr {}, Timeout) -> Boolean (not (null isolated)) .// "Incr timeout"+ (_, _) -> Bot .// "postcondition"++command :: Handle -> ClientHandleRefs Concrete -> String -> IO (Int, Maybe String)+command h chs@ClientHandleRefs{..} cmd = go 3 0+ where+ go 0 unex = pure (unex, Nothing)+ go n unex = do+ eRes <- do+ hPutStrLn h cmd+ hPutStrLn (opaque client_hin) cmd+ mresp <- getResponse (opaque client_hout)+ case mresp of+ Nothing -> do+ hPutStrLn h "'getResponse' timed out"+ pure (Just (unex, Nothing))+ Just resp ->+ if "Timeout" `isInfixOf` resp+ then do+ hPutStrLn h ("Command timed out, retrying: " ++ show resp)+ pure (Just (unex, Nothing))+ else if "Unexpected" `isInfixOf` resp+ then do+ hPutStrLn h ("Unexpected read/write response, retrying: " ++ show resp)+ pure (Just (unex + 1, Nothing))+ else do+ hPutStrLn h ("got response `" ++ resp ++ "'")+ pure (Just (unex, Just resp))+ case eRes of+ Nothing -> pure (unex, Nothing)+ -- Recurse if command successful+ Just (unex, Nothing) -> threadDelay 1000000 >> go (n-1) unex+ Just (unex, Just resp) -> pure (unex, Just resp)++ getResponse :: Handle -> IO (Maybe String)+ getResponse hout = do+ mline <- timeout 10000000 (hGetLine hout)+ case mline of+ Nothing -> return Nothing+ Just line+ | "New leader found" `isInfixOf` line -> getResponse hout+ | otherwise -> return (Just line)++semantics :: Handle -> Action Concrete -> IO (Response Concrete)+semantics h (SpawnNode port1 p) = do+ hPutStrLn h ("Spawning node on port " ++ show port1)+ removePathForcibly ("/tmp/raft-log-" ++ show port1 ++ ".txt")+ h' <- openFile ("/tmp/raft-log-" ++ show port1 ++ ".txt") WriteMode+ let port2, port3 :: Int+ (port2, port3) = case port1 of+ 3000 -> (3001, 3002)+ 3001 -> (3000, 3002)+ 3002 -> (3000, 3001)+ _ -> error "semantics: invalid port1"+ let persistence Fresh = "fresh"+ persistence Existing = "existing"+ (_, _, _, ph) <- createProcess_ "raft node"+ (proc "fiu-run" [ "-x", "stack", "exec", "raft-example", "node"+ , persistence p, "postgres", show port1+ , "localhost:" ++ show port1+ , "localhost:" ++ show port2+ , "localhost:" ++ show port3+ ])+ { std_out = UseHandle h'+ , std_err = UseHandle h'+ }+ threadDelay 1500000+ mec <- getProcessExitCode ph+ case mec of+ Nothing -> return (SpawnedNode (reference (Opaque ph)))+ Just ec -> do+ hPutStrLn h (show ec)+ return (SpawnFailed port1)++semantics h (SpawnClient cport) = do+ hPutStrLn h "Spawning client"+ (Just hin, Just hout, _, ph) <- createProcess_ "raft client"+ (proc "stack" [ "exec", "raft-example", "client" ])+ { std_out = CreatePipe+ , std_in = CreatePipe+ , std_err = CreatePipe+ }++ threadDelay 1000000+ mec <- getProcessExitCode ph+ case mec of+ Just ec -> do+ hPutStrLn h (show ec)+ return (SpawnFailed cport)+ Nothing -> do+ threadDelay 100000+ hSetBuffering hin NoBuffering+ hSetBuffering hout NoBuffering+ hPutStrLn hin "addNode localhost:3000"+ hPutStrLn hin "addNode localhost:3001"+ hPutStrLn hin "addNode localhost:3002"+ let refClient_hin = reference (Opaque hin)+ refClient_hout = reference (Opaque hout)+ clientHandleRefs = ClientHandleRefs refClient_hin refClient_hout+ return (SpawnedClient clientHandleRefs (reference (Opaque ph)))++semantics h (KillNode (_port, ph)) = do+ hPutStrLn h $ "Attempting to kill node on port " ++ show _port ++ "..."+ terminateProcess (opaque ph)+ waitForProcess (opaque ph)+ hPutStrLn h $ "Successfully killed node on port " ++ show _port+ return Ack+semantics h (Set chs i) = do+ (_, mresp) <- command h chs ("set x " ++ show i)+ case mresp of+ Nothing -> return Timeout+ Just _resp -> return Ack+semantics h (Read chs) = do+ (ue, mresp) <- command h chs "read"+ case mresp of+ Nothing -> return Timeout+ Just resp -> do+ let parse = readEither+ . takeWhile isDigit+ . drop 1+ . snd+ . break (== ',')+ return (Value (bimap (++ (": " ++ resp)) id (parse resp)))+semantics h (Incr chs) = do+ (_, mresp) <- command h chs "incr x"+ case mresp of+ Nothing -> return Timeout+ Just resp -> return Ack+semantics h (BreakConnection (port, ph)) = do+ hPutStrLn h ("Break connection, port: " ++ show port)+ Just pid <- getPid (opaque ph)+ callCommand ("fiu-ctrl -c \"enable name=posix/io/net/send\" " ++ show pid)+ callCommand ("fiu-ctrl -c \"enable name=posix/io/net/recv\" " ++ show pid)+ threadDelay 2000000+ return BrokeConnection+semantics h (FixConnection (port, ph)) = do+ hPutStrLn h ("Fix connection, port: " ++ show port)+ Just pid <- getPid (opaque ph)+ callCommand ("fiu-ctrl -c \"disable name=posix/io/net/send\" " ++ show pid)+ callCommand ("fiu-ctrl -c \"disable name=posix/io/net/recv\" " ++ show pid)+ threadDelay 2000000+ return FixedConnection++generator :: Model Symbolic -> Gen (Action Symbolic)+generator Model {..}+ | length nodes < 3 =+ if started+ then flip SpawnNode Existing <$> elements ([3000..3002] \\ map fst nodes)+ else flip SpawnNode Fresh <$> elements ([3000..3002] \\ map fst nodes)+ | otherwise =+ case client of+ Nothing -> SpawnClient <$> elements [3003..3010]+ Just (chs, _) ->+ case value of+ Nothing -> Set chs <$> arbitrary+ Just _+ | length isolated <= 1 -> frequency $+ [ (1, Set chs <$> arbitrary)+ , (5, pure (Incr chs))+ , (3, pure (Read chs))+ , (1, KillNode <$> elements nodes)+ , (1, BreakConnection <$> elements nodes)+ ] ++ case isolated of+ [] -> []+ _ -> [(1, FixConnection <$> elements isolated)]+ | otherwise ->+ let notIsolated = filter (\(p,_) -> not (p `elem` map fst isolated)) nodes+ in frequency $+ [ (1, KillNode <$> elements nodes)+ ] ++ case notIsolated of+ [] -> []+ _ -> [(1, BreakConnection <$> elements notIsolated )]+ ++ case isolated of+ [] -> []+ _ -> [(1, FixConnection <$> elements isolated)]++shrinker :: Action Symbolic -> [Action Symbolic]+shrinker (Set cph i) = [ Set cph i' | i' <- shrink i ]+shrinker _ = []++mock :: Model Symbolic -> Action Symbolic -> GenSym (Response Symbolic)+mock _m SpawnNode {} = SpawnedNode <$> genSym+mock _m SpawnClient {} = SpawnedClient <$> (ClientHandleRefs <$> genSym <*> genSym) <*> genSym+mock _m KillNode {} = pure Ack+mock _m Set {} = pure Ack+mock _m Read {} = pure (Value (Right 0))+mock _m Incr {} = pure Ack+mock _m BreakConnection {} = pure BrokeConnection+mock _m FixConnection {} = pure FixedConnection++setup :: IO Handle+setup = do+ removePathForcibly "/tmp/raft-log.txt"+ h <- openFile "/tmp/raft-log.txt" WriteMode+ hSetBuffering h NoBuffering+ return h++sm :: Handle -> StateMachine Model Action IO Response+sm h = StateMachine initModel transition precondition postcondition+ Nothing generator Nothing shrinker (semantics h) mock++prop_sequential :: Property+prop_sequential = withMaxSuccess 10 $ noShrinking $+ forAllCommands (sm undefined) (Just 20) $ \cmds -> monadicIO $ do+ h <- liftIO setup+ let sm' = sm h+ (hist, model, res) <- runCommands sm' cmds+ prettyCommands sm' hist (res === Ok)+ liftIO (hClose h)+ -- Terminate node processes+ liftIO (mapM_ (terminateProcess . opaque . snd) (nodes model))+ -- Terminate client process+ case client model of+ Nothing -> pure ()+ Just (ClientHandleRefs chin chout, cph) -> do+ liftIO (terminateProcess (opaque cph))+ liftIO (hClose (opaque chin) >> hClose (opaque chout))++------------------------------------------------------------------------++runMany :: Commands Action -> Handle -> Property+runMany cmds log = monadicIO $ do+ (hist, model, res) <- runCommands (sm log) cmds+ prettyCommands (sm log) hist (res === Ok)+ -- Terminate node processes+ liftIO (mapM_ (terminateProcess . opaque . snd) (nodes model))+ -- Terminate client process+ case client model of+ Nothing -> pure ()+ Just (ClientHandleRefs chin chout, cph) -> do+ liftIO (terminateProcess (opaque cph))+ liftIO (hClose (opaque chin) >> hClose (opaque chout))++-------------------------------------------------------------------------------++-- unit_exampleUnit :: IO ()+-- unit_exampleUnit = bracket setup hClose (verboseCheck . exampleUnit)+--+-- exampleUnit :: Handle -> Property+-- exampleUnit = runMany cmds+-- where+-- cmds = Commands+-- [ -- Commands go here...+-- ]
test/TestDejaFu.hs view
@@ -7,6 +7,7 @@ {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-} module TestDejaFu where@@ -31,12 +32,16 @@ import Test.Tasty import Test.Tasty.DejaFu hiding (get) -import System.Random (mkStdGen)+import System.Random (mkStdGen, newStdGen) import TestUtils import Raft+import Raft.Log+import Raft.Client +import Data.Time.Clock.System (getSystemTime)+ -------------------------------------------------------------------------------- -- Test State Machine & Commands --------------------------------------------------------------------------------@@ -54,25 +59,28 @@ data StoreCtx = StoreCtx -instance RSMP Store StoreCmd where- data RSMPError Store StoreCmd = StoreError Text deriving (Show)- type RSMPCtx Store StoreCmd = StoreCtx- applyCmdRSMP _ store cmd =+instance RaftStateMachinePure Store StoreCmd where+ data RaftStateMachinePureError Store StoreCmd = StoreError Text deriving (Show)+ type RaftStateMachinePureCtx Store StoreCmd = StoreCtx+ rsmTransition _ 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+instance RaftStateMachine RaftTestM Store StoreCmd where validateCmd _ = pure (Right ())- askRSMPCtx = pure StoreCtx+ askRaftStateMachinePureCtx = pure StoreCtx type TestEventChan = EventChan ConcIO StoreCmd-type TestClientRespChan = TChan (STM ConcIO) (ClientResponse Store)+type TestEventChans = Map NodeId TestEventChan +type TestClientRespChan = TChan (STM ConcIO) (ClientResponse Store StoreCmd)+type TestClientRespChans = Map ClientId TestClientRespChan+ -- | Node specific environment data TestNodeEnv = TestNodeEnv- { testNodeEventChans :: Map NodeId TestEventChan- , testClientRespChans :: Map ClientId TestClientRespChan+ { testNodeEventChans :: TestEventChans+ , testClientRespChans :: TestClientRespChans , testNodeConfig :: NodeConfig } @@ -130,6 +138,7 @@ instance RaftPersist RaftTestM where type RaftPersistError RaftTestM = RaftTestError+ initializePersistentState = pure (Right ()) writePersistentState pstate' = do nid <- askSelfNodeId fmap Right $ modify $ \testState ->@@ -150,13 +159,18 @@ eventChan <- lookupNodeEventChan nid atomically $ writeTChan eventChan (MessageEvent (RPCMessageEvent rpc)) -instance RaftSendClient RaftTestM Store where+instance RaftSendClient RaftTestM Store StoreCmd 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 RaftInitLog RaftTestM StoreCmd where+ type RaftInitLogError RaftTestM = RaftTestError+ -- No log initialization needs to be done here, everything is in memory.+ initializeLog _ = pure (Right ())+ instance RaftWriteLog RaftTestM StoreCmd where type RaftWriteLogError RaftTestM = RaftTestError writeLogEntries entries = do@@ -195,6 +209,43 @@ -------------------------------------------------------------------------------- +data TestClientEnv = TestClientEnv+ { testClientEnvRespChan :: TestClientRespChan+ , testClientEnvNodeEventChans :: TestEventChans+ }++type RaftTestClientM' = ReaderT TestClientEnv ConcIO+type RaftTestClientM = RaftClientT Store StoreCmd RaftTestClientM'++instance RaftClientSend RaftTestClientM' StoreCmd where+ type RaftClientSendError RaftTestClientM' StoreCmd = ()+ raftClientSend nid creq = do+ Just nodeEventChan <- asks (Map.lookup nid . testClientEnvNodeEventChans)+ lift $ atomically $ writeTChan nodeEventChan (MessageEvent (ClientRequestEvent creq))+ pure (Right ())++instance RaftClientRecv RaftTestClientM' Store StoreCmd where+ type RaftClientRecvError RaftTestClientM' Store = ()+ raftClientRecv = do+ clientRespChan <- asks testClientEnvRespChan+ fmap Right $ lift $ atomically $ readTChan clientRespChan++runRaftTestClientM+ :: ClientId+ -> TestClientRespChan+ -> TestEventChans+ -> RaftTestClientM a+ -> ConcIO a+runRaftTestClientM cid chan chans rtcm = do+ raftClientState <- initRaftClientState mempty <$> liftIO newStdGen+ let raftClientEnv = RaftClientEnv cid+ testClientEnv = TestClientEnv chan chans+ in flip runReaderT testClientEnv+ . runRaftClientT raftClientEnv raftClientState { raftClientRaftNodes = Map.keysSet chans }+ $ rtcm++--------------------------------------------------------------------------------+ initTestChanMaps :: ConcIO (Map NodeId TestEventChan, Map ClientId TestClientRespChan) initTestChanMaps = do eventChans <-@@ -232,9 +283,6 @@ -------------------------------------------------------------------------------- -type TestEventChans = Map NodeId TestEventChan-type TestClientRespChans = Map ClientId TestClientRespChan- test_concurrency :: [TestTree] test_concurrency = [ testGroup "Leader Election" [ testConcurrentProps (leaderElection node0) mempty ]@@ -279,46 +327,51 @@ teardown = mapM_ killThread . fst leaderElection :: NodeId -> TestEventChans -> TestClientRespChans -> ConcIO Store-leaderElection nid eventChans clientRespChans = do- atomically $ writeTChan nodeEventChan (TimeoutEvent ElectionTimeout)- pollForReadResponse nodeEventChan client0RespChan+leaderElection nid eventChans clientRespChans =+ runRaftTestClientM client0 client0RespChan eventChans $+ leaderElection' nid eventChans where+ Just client0RespChan = Map.lookup client0 clientRespChans++leaderElection' :: NodeId -> TestEventChans -> RaftTestClientM Store+leaderElection' nid eventChans = do+ sysTime <- liftIO getSystemTime+ lift $ lift $ atomically $ writeTChan nodeEventChan (TimeoutEvent sysTime ElectionTimeout)+ pollForReadResponse nid+ 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)+ runRaftTestClientM client0 client0RespChan eventChans $ do+ Right idx <- do+ syncClientWrite node0 (Set "x" 41)+ syncClientWrite node0 (Incr"x")+ store <- pollForReadResponse node0+ 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)+ leaderElection node0 eventChans clientRespChans+ runRaftTestClientM client0 client0RespChan eventChans $ do+ syncClientWrite node0 (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+ fmap (Maybe.fromJust . lastMay) $+ replicateM 10 $ syncClientWrite node0 (Incr "x")+ store <- pollForReadResponse node0 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)+leaderRedirect eventChans clientRespChans =+ runRaftTestClientM client0 client0RespChan eventChans $ do+ Left resp <- syncClientWrite node1 (Set "x" 42) pure resp where- Just node1EventChan = Map.lookup node1 eventChans Just client0RespChan = Map.lookup client0 clientRespChans followerRedirNoLeader :: TestEventChans -> TestClientRespChans -> ConcIO CurrentLeader@@ -335,96 +388,100 @@ 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+ runRaftTestClientM client0 client0RespChan eventChans $ do+ Left ldr <- syncClientRead node0+ 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)+comprehensive eventChans clientRespChans =+ runRaftTestClientM client0 client0RespChan eventChans $ do+ leaderElection'' node0+ Right idx2 <- syncClientWrite node0 (Set "x" 7)+ Right idx3 <- syncClientWrite node0 (Set "y" 3)+ Left (CurrentLeader _) <- syncClientWrite node1 (Incr "y")+ Right _ <- syncClientRead node0 - 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'' node1+ Right idx5 <- syncClientWrite node1 (Incr "x")+ Right idx6 <- syncClientWrite node1 (Incr "y")+ Right idx7 <- syncClientWrite node1 (Set "z" 40)+ Left (CurrentLeader _) <- syncClientWrite node2 (Incr "y")+ Right _ <- syncClientRead node1 - 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'' node2+ Right idx9 <- syncClientWrite node2 (Incr "z")+ Right idx10 <- syncClientWrite node2 (Incr "x")+ Left _ <- syncClientWrite node1 (Set "q" 100)+ Right idx11 <- syncClientWrite node2 (Incr "y")+ Left _ <- syncClientWrite node0 (Incr "z")+ Right idx12 <- syncClientWrite node2 (Incr "y")+ Left (CurrentLeader _) <- syncClientWrite node0 (Incr "y")+ Right _ <- syncClientRead node2 - leaderElection node0 eventChans clientRespChans- Right idx14 <- syncClientWriteClient0 node0EventChan (Incr "z")- Left (CurrentLeader _) <- syncClientWriteClient0 node1EventChan (Incr "y")+ leaderElection'' node0+ Right idx14 <- syncClientWrite node0 (Incr "z")+ Left (CurrentLeader _) <- syncClientWrite node1 (Incr "y") - Right store <- syncClientRead node0EventChan (client0, client0RespChan)- Left ldr <- syncClientRead node1EventChan (client0, client0RespChan)+ Right store <- syncClientRead node0+ Left ldr <- syncClientRead node1 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-+ leaderElection'' nid = leaderElection' nid 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+-- | This function can be safely "run" without worry about impacting the client+-- SerialNum of the client requests.+--+-- Warning: If read requests start to include serial numbers, this function will+-- no longer be safe to `runRaftTestClientM` on.+pollForReadResponse :: NodeId -> RaftTestClientM Store+pollForReadResponse nid = do+ eRes <- clientReadFrom nid ClientReadStateMachine+ case eRes of+ -- TODO Handle other cases of 'ClientReadResp'+ Right (ClientReadRespStateMachine res) -> pure res _ -> do- liftIO $ Control.Monad.Conc.Class.threadDelay 1000- pollForReadResponse nodeEventChan clientRespChan+ liftIO $ Control.Monad.Conc.Class.threadDelay 10000+ pollForReadResponse nid -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+syncClientRead :: NodeId -> RaftTestClientM (Either CurrentLeader Store)+syncClientRead nid = do+ eRes <- clientReadFrom nid ClientReadStateMachine+ case eRes of+ -- TODO Handle other cases of 'ClientReadResp'+ Right (ClientReadRespStateMachine store) -> pure $ Right store+ Left (RaftClientUnexpectedRedirect (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+syncClientWrite+ :: NodeId+ -> StoreCmd+ -> RaftTestClientM (Either CurrentLeader Index)+syncClientWrite nid cmd = do+ eRes <- clientWriteTo nid cmd+ case eRes of+ Right (ClientWriteResp idx sn) -> do+ Just nodeEventChan <- lift (asks (Map.lookup nid . testClientEnvNodeEventChans)) pure $ Right idx- ClientRedirectResponse (ClientRedirResp ldr) -> pure $ Left ldr+ Left (RaftClientUnexpectedRedirect (ClientRedirResp ldr)) -> pure $ Left ldr _ -> panic "Failed to receive client write response..." heartbeat :: TestEventChan -> ConcIO ()-heartbeat eventChan = atomically $ writeTChan eventChan (TimeoutEvent HeartbeatTimeout)+heartbeat eventChan = do+ sysTime <- liftIO getSystemTime+ atomically $ writeTChan eventChan (TimeoutEvent sysTime HeartbeatTimeout) clientReadReq :: ClientId -> Event StoreCmd-clientReadReq cid = MessageEvent $ ClientRequestEvent $ ClientRequest cid ClientReadReq+clientReadReq cid = MessageEvent $ ClientRequestEvent $ ClientRequest cid (ClientReadReq ClientReadStateMachine) -clientWriteReq :: ClientId -> StoreCmd -> Event StoreCmd-clientWriteReq cid v = MessageEvent $ ClientRequestEvent $ ClientRequest cid $ ClientWriteReq v+clientReadRespChan :: RaftTestClientM (ClientResponse Store StoreCmd)+clientReadRespChan = do+ clientRespChan <- lift (asks testClientEnvRespChan)+ lift $ lift $ atomically $ readTChan clientRespChan
− test/TestRaft.hs
@@ -1,573 +0,0 @@-{-# 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
@@ -8,27 +8,28 @@ import qualified Data.Map.Merge.Lazy as Merge import Raft+import Raft.Config -isRaftLeader :: RaftNodeState -> Bool+isRaftLeader :: RaftNodeState v -> Bool isRaftLeader (RaftNodeState rns) = isLeader rns -isRaftCandidate :: RaftNodeState -> Bool+isRaftCandidate :: RaftNodeState v -> Bool isRaftCandidate (RaftNodeState rns) = isCandidate rns -isRaftFollower :: RaftNodeState -> Bool+isRaftFollower :: RaftNodeState v -> Bool isRaftFollower (RaftNodeState rns) = isFollower rns -checkCurrentLeader :: RaftNodeState -> CurrentLeader+checkCurrentLeader :: RaftNodeState v -> CurrentLeader checkCurrentLeader (RaftNodeState (NodeFollowerState FollowerState{..})) = fsCurrentLeader checkCurrentLeader (RaftNodeState (NodeCandidateState _)) = NoLeader checkCurrentLeader (RaftNodeState (NodeLeaderState _)) = NoLeader -getLastAppliedLog :: RaftNodeState -> Index+getLastAppliedLog :: RaftNodeState v -> Index getLastAppliedLog (RaftNodeState (NodeFollowerState FollowerState{..})) = fsLastApplied getLastAppliedLog (RaftNodeState (NodeCandidateState CandidateState{..})) = csLastApplied getLastAppliedLog (RaftNodeState (NodeLeaderState LeaderState{..})) = lsLastApplied -getCommittedLogIndex :: RaftNodeState -> Index+getCommittedLogIndex :: RaftNodeState v -> Index getCommittedLogIndex (RaftNodeState (NodeFollowerState FollowerState{..})) = fsCommitIndex getCommittedLogIndex (RaftNodeState (NodeCandidateState CandidateState{..})) = csCommitIndex getCommittedLogIndex (RaftNodeState (NodeLeaderState LeaderState{..})) = lsCommitIndex@@ -59,18 +60,21 @@ , configNodeIds = nodeIds , configElectionTimeout = pairMsToMicroS (150, 300) , configHeartbeatTimeout = msToMicroS 50+ , configStorageState = New } testConfig1 = NodeConfig { configNodeId = node1 , configNodeIds = nodeIds , configElectionTimeout = pairMsToMicroS (150, 300) , configHeartbeatTimeout = msToMicroS 50+ , configStorageState = New } testConfig2 = NodeConfig { configNodeId = node2 , configNodeIds = nodeIds , configElectionTimeout = pairMsToMicroS (150, 300) , configHeartbeatTimeout = msToMicroS 50+ , configStorageState = New } -- | Zip maps using function. Throws away items left and right