legion 0.7.0.0 → 0.8.0.0

raw patch · 17 files changed

+1388/−1390 lines, 17 filesdep −canteven-logdep −time

Dependencies removed: canteven-log, time

Files

• legion.cabal +4/−4
• src/Network/Legion/Admin.hs +10/−11
• src/Network/Legion/Application.hs +15/−10
• src/Network/Legion/ClusterState.hs +126/−152
• src/Network/Legion/Conduit.hs +15/−2
• src/Network/Legion/Distribution.hs +132/−36
• src/Network/Legion/KeySet.hs +27/−3
• src/Network/Legion/Lift.hs +76/−0
• src/Network/Legion/PartitionKey.hs +5/−1
• src/Network/Legion/PartitionState.hs +8/−178
• src/Network/Legion/PowerState.hs +86/−50
• src/Network/Legion/PowerState/Monad.hs +160/−0
• src/Network/Legion/Propagation.hs +0/−392
• src/Network/Legion/Runtime.hs +205/−220
• src/Network/Legion/Runtime/PeerMessage.hs +12/−0
• src/Network/Legion/StateMachine.hs +342/−331
• src/Network/Legion/StateMachine/Monad.hs +165/−0

@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/  name:                legion-version:             0.7.0.0+version:             0.8.0.0 synopsis:            Distributed, stateful, homogeneous microservice framework. description:         Legion is a framework for writing distributed,                      homogeneous, stateful microservices in Haskell.@@ -37,15 +37,17 @@     Network.Legion.Index     Network.Legion.KeySet     Network.Legion.LIO+    Network.Legion.Lift     Network.Legion.PartitionKey     Network.Legion.PartitionState     Network.Legion.PowerState-    Network.Legion.Propagation+    Network.Legion.PowerState.Monad     Network.Legion.Runtime     Network.Legion.Runtime.ConnectionManager     Network.Legion.Runtime.PeerMessage     Network.Legion.Settings     Network.Legion.StateMachine+    Network.Legion.StateMachine.Monad     Network.Legion.UUID     Paths_legion   -- other-extensions:@@ -58,7 +60,6 @@     binary-conduit     >= 1.2.3    && < 1.3,     bytestring         >= 0.10.4.0 && < 0.11,     canteven-http      >= 0.1.1.1  && < 0.2,-    canteven-log       >= 1.0.0.0  && < 2.1,     conduit            >= 1.2.4    && < 1.3,     conduit-extra      >= 1.1.9    && < 1.2,     containers         >= 0.5.5.1  && < 0.6,@@ -73,7 +74,6 @@     scotty-resource    >= 0.1      && < 0.3,     stm                >= 2.4.4.1  && < 2.5,     text               >= 1.2.2.0  && < 1.3,-    time               >= 1.4.2    && < 1.7,     transformers       >= 0.3.0.0  && < 0.6,     unix               >= 2.7      && < 2.8,     uuid               >= 1.3.11   && < 1.4,

@@ -19,16 +19,17 @@ import Data.Conduit (Source) import Data.Default.Class (def) import Data.Text.Encoding (encodeUtf8)-import Data.Text.Lazy (Text, pack)+import Data.Text.Lazy (Text) import Data.Version (showVersion) import Network.HTTP.Types (notFound404) import Network.Legion.Application (LegionConstraints) import Network.Legion.Conduit (chanToSource) import Network.Legion.Distribution (Peer) import Network.Legion.LIO (LIO)+import Network.Legion.Lift (lift2) import Network.Legion.PartitionKey (PartitionKey(K)) import Network.Legion.PartitionState (PartitionPowerState)-import Network.Legion.StateMachine (NodeState)+import Network.Legion.StateMachine.Monad (NodeState) import Network.Wai (Middleware, modifyResponse) import Network.Wai.Handler.Warp (HostPreference, defaultSettings, Port,   setHost, setPort)@@ -37,8 +38,8 @@ import Paths_legion (version) import Text.Read (readMaybe) import Web.Scotty.Resource.Trans (resource, get, delete)-import Web.Scotty.Trans (Options, scottyOptsT, settings, ScottyT, text,-  ActionT, param, middleware, status)+import Web.Scotty.Trans (Options, scottyOptsT, settings, ScottyT, ActionT,+  param, middleware, status, json) import qualified Data.Text as T  {- |@@ -61,14 +62,12 @@             . logExceptionsAndContinue logging            resource "/clusterstate" $-            get $ do-              val <- send chan GetState-              text (pack (show val))+            get $+              json =<< send chan GetState           resource "/propstate/:key" $             get $ do               key <- K . read <$> param "key"-              val <- send chan (GetPart key)-              text (pack (show val))+              json =<< send chan (GetPart key)           resource "/peers/:peer" $             delete $               readMaybe <$> param "peer" >>= \case@@ -84,7 +83,7 @@       :: Chan (AdminMessage e o s)       -> ((a -> LIO ()) -> AdminMessage e o s)       -> ActionT Text LIO a-    send chan msg = lift . lift $ do+    send chan msg = lift2 $ do       mvar <- newEmptyMVar       writeChan chan (msg (lift . putMVar mvar))       takeMVar mvar@@ -129,7 +128,7 @@ -} data AdminMessage e o s   = GetState (NodeState e o s -> LIO ())-  | GetPart PartitionKey (Maybe (PartitionPowerState e o s) -> LIO ())+  | GetPart PartitionKey (PartitionPowerState e o s -> LIO ())   | Eject Peer (() -> LIO ())  instance Show (AdminMessage e o s) where

@@ -8,6 +8,7 @@   Persistence(..), ) where +import Data.Aeson (ToJSON) import Data.Binary (Binary) import Data.Conduit (Source) import Data.Default.Class (Default)@@ -21,13 +22,16 @@   constraints    > (-  >   Event e o s, Default s, Binary e, Binary o, Binary s, Show e,-  >   Show o, Show s, Eq e+  >   Binary e, Binary o, Binary s, Default s, Eq e, Event e o s, Indexable s,+  >   Show e, Show o, Show s, ToJSON s   > )++  The @ToJSON s@ requirement is strictly for servicing the admin web+  endpoints. -} type LegionConstraints e o s = (-    Event e o s, Indexable s, Default s, Binary e, Binary o, Binary s,-    Show e, Show o, Show s, Eq e+    Binary e, Binary o, Binary s, Default s, Eq e, Event e o s, Indexable s,+    Show e, Show o, Show s, ToJSON s   )  @@ -40,12 +44,13 @@      getState :: PartitionKey -> IO (Maybe (PartitionPowerState e o s)),     saveState :: PartitionKey -> Maybe (PartitionPowerState e o s) -> IO (),          list :: Source IO (PartitionKey, PartitionPowerState e o s)-      {- ^-        List all the keys known to the persistence layer. It is important-        that the implementation do the right thing with regard to-        `Data.Conduit.addCleanup`, because there are cases where the-        conduit is terminated without reading the entire list.-      -}+                 {- ^+                   List all the keys known to the persistence layer. It is+                   important that the implementation do the right thing+                   with regard to `Data.Conduit.addCleanup`, because+                   there are cases where the conduit is terminated+                   without reading the entire list.+                 -}   }  

@@ -1,49 +1,54 @@ {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TupleSections #-} {- |   This module contains the data types related to the distributed cluster state. -} module Network.Legion.ClusterState (   ClusterState,   ClusterPowerState,-  ClusterPropState,-  claimParticipation,+  ClusterPowerStateT,+  RebalanceOrd,   new,-  initProp,-  getPowerState,   getPeers,-  findPartition,+  findRoute,+  findOwners,   getDistribution,   joinCluster,+  finishRebalance,   eject,-  mergeEither,-  actions,-  allParticipants,-  heartbeat,+  nextAction, ) where -import Data.Aeson (ToJSON, toJSON, object, (.=))+import Control.Exception (throw)+import Data.Aeson (ToJSON, toJSON, object, (.=), encode) import Data.Binary (Binary) import Data.Default.Class (Default(def))+import Data.Functor.Identity (runIdentity) import Data.Map (Map) import Data.Set (Set)-import Data.Time.Clock (UTCTime)+import Data.Text.Encoding (decodeUtf8) import Data.UUID (UUID)+import Data.Word (Word64) import GHC.Generics (Generic) import Network.Legion.BSockAddr (BSockAddr(BSockAddr))-import Network.Legion.Distribution (ParticipationDefaults, modify, Peer)-import Network.Legion.KeySet (KeySet, full, unions)+import Network.Legion.Distribution (ParticipationDefaults,+  Peer, rebalanceAction, RebalanceAction(NoAction)) import Network.Legion.PartitionKey (PartitionKey)-import Network.Legion.PowerState (Event(apply))-import Network.Legion.Propagation (PropState, PropPowerState)+import Network.Legion.PowerState (Event, apply, PowerState)+import Network.Legion.PowerState.Monad (PowerStateT, runPowerStateT) import Network.Socket (SockAddr)+import qualified Data.ByteString.Lazy as LBS import qualified Data.Map as Map import qualified Data.Set as Set+import qualified Data.Text as T import qualified Network.Legion.Distribution as D-import qualified Network.Legion.Propagation as P+import qualified Network.Legion.PowerState as PS+import qualified Network.Legion.PowerState.Monad as PM   {- |@@ -52,14 +57,18 @@ -} data ClusterState = ClusterState {     distribution :: ParticipationDefaults,-           peers :: Map Peer BSockAddr+           peers :: Map Peer BSockAddr,+         updates :: [ClusterChange],+    rebalanceOrd :: RebalanceOrd   }-  deriving (Show, Generic)+  deriving (Generic) instance Binary ClusterState instance Default ClusterState where   def = ClusterState {       distribution = D.empty,-             peers = Map.empty+             peers = Map.empty,+           updates = [],+      rebalanceOrd = minBound     } instance ToJSON ClusterState where   toJSON ClusterState {distribution, peers} = object [@@ -69,185 +78,150 @@           | (p, a) <- Map.toList peers         ]     ]+instance Show ClusterState where+  show = T.unpack . decodeUtf8 . LBS.toStrict . encode  -{- |-  A representation of all possible cluster states.--}-newtype ClusterPowerState = ClusterPowerState {-    unPowerState :: PropPowerState UUID ClusterState Peer Update ()-  } deriving (Show, Binary)+{- | A representation of all possible cluster states. -}+type ClusterPowerState =+  PowerState UUID ClusterState Peer Update ()  -{- |-  A reification of `PropState`, representing the propagation state of the-  cluster state.--}-newtype ClusterPropState = ClusterPropState {-    unPropState :: PropState UUID ClusterState Peer Update ()-  } deriving (Show, ToJSON)+{- | A convenient alias for the cluster power state monad transformer. -}+type ClusterPowerStateT =+  PowerStateT UUID ClusterState Peer Update ()  -{- |-  The kinds of updates that can be applied to the cluster state.--}+{- | The type of rebalancing action ordinal. -}+newtype RebalanceOrd = RebalanceOrd Word64 +  deriving (Generic, Show, Enum, Bounded, Eq, Ord)+instance Binary RebalanceOrd+++{- | The kinds of updates that can be applied to the cluster state. -} data Update-  = PeerJoined Peer BSockAddr-  | Participating Peer KeySet-  | PeerEjected Peer-  deriving (Show, Generic)+  = Change ClusterChange+  | Complete+  deriving (Show, Generic, Eq) instance Binary Update instance Event Update () ClusterState where-  apply (PeerJoined peer addr) cs@ClusterState {peers} =-    ((), cs {peers = Map.insert peer addr peers})-  apply (Participating peer ks) cs@ClusterState {distribution} =-    ((), cs {distribution = modify (Set.insert peer) ks distribution})-  apply (PeerEjected peer) cs@ClusterState {distribution, peers} =-    ((), cs {-        distribution = modify (Set.delete peer) full distribution,-        peers = Map.delete peer peers-      })+  apply update cs@ClusterState {peers, updates, distribution, rebalanceOrd} =+    ((),) . popUpdate $ case update of+      Change change -> cs {updates = updates ++ [change]}+      Complete -> cs {+          distribution =+            snd (rebalanceAction (Map.keysSet peers) distribution),+          rebalanceOrd =+            succ rebalanceOrd+        }   {- |-  Helper function, for easily claiming participation in a key set.+  Helper for 'instance Event Update () ClusterState'. Applies updates+  from the update queue until an uncompleted rebalance action prevents+  further progress, and returns the resulting cluster state. -}-claimParticipation-  :: Peer-  -> KeySet-  -> ClusterPropState-  -> ClusterPropState-claimParticipation peer ks =-  ClusterPropState-  . P.event (Participating peer ks)-  . unPropState+popUpdate :: ClusterState -> ClusterState+popUpdate cs@ClusterState {updates, distribution, peers} =+  case (updates, rebalanceAction (Map.keysSet peers) distribution) of+    (u:moreUpdates, (NoAction, _)) -> popUpdate cs {+        peers = case u of+          PeerJoined peer addr -> Map.insert peer addr peers+          PeerEjected peer -> Map.delete peer peers,+        updates = moreUpdates+      }+    _ -> cs  -{- |-  Create the cluster state appropriate for a brand-new cluster.--}-new :: UUID -> Peer -> SockAddr -> ClusterPropState-new clusterId self addy =-  claimParticipation self full-  . ClusterPropState-  . P.event (PeerJoined self (BSockAddr addy))-  $ P.new clusterId self (Set.singleton self)+{- | This type describes how a cluster topology can change. -}+data ClusterChange+  = PeerJoined Peer BSockAddr+  | PeerEjected Peer+  deriving (Show, Generic, Eq)+instance Binary ClusterChange   {- |-  Initialize a `ClusterPropState` based on the initial underlying cluster power-  state.+  Create the cluster state appropriate for a brand-new cluster. -}-initProp :: Peer -> ClusterPowerState -> ClusterPropState-initProp self = ClusterPropState . P.initProp self . unPowerState+new :: UUID -> Peer -> SockAddr -> ClusterPowerState+new clusterId self addy =+  runIdentity $ runPowerStateT self (PS.new clusterId (Set.singleton self)) (do+      PM.event (Change (PeerJoined self (BSockAddr addy)))+      PM.event Complete+      PM.acknowledge+    ) >>= \case+      Left err -> throw err+      Right ((), _, cluster, _) -> return cluster  -{- |-  Return an opaque representation of the underling power state, for transfer-  across the network, or whatever.--}-getPowerState :: ClusterPropState -> ClusterPowerState-getPowerState = ClusterPowerState . P.getPowerState . unPropState+{- | Get the cluster peers. -}+getPeers :: ClusterPowerState -> Map Peer BSockAddr+getPeers = peers . PS.projectedValue   {- |-  Get the cluster peers.+  get the cluster distribution. -}-getPeers :: ClusterPropState -> Map Peer BSockAddr-getPeers = peers . P.ask . unPropState+getDistribution :: ClusterPowerState -> ParticipationDefaults+getDistribution = distribution . PS.projectedValue   {- |-  get the cluster distribution.+  Find the nodes to which a given request might be routed. This might be+  different from `findOwners` when a cluster rebalancing is taking place. -}-getDistribution :: ClusterPropState -> ParticipationDefaults-getDistribution = distribution . P.ask . unPropState+findRoute :: PartitionKey -> ClusterPowerState -> Set Peer+findRoute key =+  D.findPartition key . distribution . PS.projectedValue   {- |-  Find the nodes that own a given partition.+  Find the nodes which own a particular partition. This is used for+  primarily for initializing a new partition, and may be different than+  `findRoute` when a cluster rebalancing is happening. -}-findPartition :: PartitionKey -> ClusterPropState -> Set Peer-findPartition key =-  D.findPartition key . distribution . P.ask . unPropState+findOwners :: PartitionKey -> ClusterPowerState -> Set Peer+findOwners key cluster =+  let ClusterState {distribution, peers} = PS.projectedValue cluster+  in+    D.findPartition+      key+      (snd (rebalanceAction (Map.keysSet peers) distribution))  -{- |-  Allow a new peer to join the cluster.--}-joinCluster-  :: Peer+{- | Allow a new peer to join the cluster. -}+joinCluster :: (Monad m)+  => Peer     {- ^ The peer that is joining. -}   -> BSockAddr     {- ^ The cluster address of the new peer. -}-  -> ClusterPropState-    {- ^ The current cluster propagation state. -}-  -> ClusterPropState-joinCluster peer addy =-  ClusterPropState-  . P.event (PeerJoined peer addy)-  . P.participate peer-  . unPropState---{- |-  Eject a peer from the cluster.--}-eject :: Peer -> ClusterPropState -> ClusterPropState-eject peer =-  ClusterPropState-  . P.event (PeerEjected peer)-  . P.disassociate peer-  . unPropState---{- |-  Merge a foreign cluster state with our own cluster state. This function-  returns the new cluster propagation state, along with a set of partition keys-  for which the default participation has changed (aka, a rebalance happened),-  indicating that some action should be taken to migrate the indicated-  partitions.--}-mergeEither-  :: Peer-  -> ClusterPowerState-  -> ClusterPropState-  -> Either String (ClusterPropState, KeySet)-mergeEither otherPeer (ClusterPowerState otherPS) (ClusterPropState prop) =-  let-    self = P.getSelf prop-    divergences = P.divergences self (P.initProp otherPeer otherPS)-    migrating = unions [-        ks-        | (_, Participating _ ks) <- Map.toList divergences-      ]-  in case P.mergeEither otherPeer otherPS prop of-    Left err -> Left err-    Right newProp -> Right (ClusterPropState newProp, migrating)+  -> ClusterPowerStateT m ()+joinCluster peer addy = do+  PM.participate peer+  PM.event (Change (PeerJoined peer addy))  -{- |-  Get the peers which require action (i.e. Send), if any, and the-  powerstate version to send to those peers, and the new propagation-  state that is applicable after those actions have been taken.--}-actions :: ClusterPropState -> (Set Peer, ClusterPowerState, ClusterPropState)-actions prop =-  let (peers, ps, newProp) = P.actions (unPropState prop)-  in (peers, ClusterPowerState ps, ClusterPropState newProp)+{- | Mark the current rebalance action as complete. -}+finishRebalance :: (Monad m) => ClusterPowerStateT m ()+finishRebalance = PM.event Complete   {- |-  Return all cluster participants.+  Eject a peer from the cluster. -}-allParticipants :: ClusterPropState -> Set Peer-allParticipants = P.allParticipants . unPropState+eject :: (Monad m) => Peer -> ClusterPowerStateT m ()+eject peer = do+  PM.event (Change (PeerEjected peer))+  PM.disassociate peer   {- |-  Move time forward for the propagation state.+  Get the current rebalance action, along with its ordinal. This is+  taken from the infimum, so it may not reflect projected changes. -}-heartbeat :: UTCTime -> ClusterPropState -> ClusterPropState-heartbeat now = ClusterPropState . P.heartbeat now . unPropState-+nextAction :: ClusterPowerState -> (RebalanceOrd, RebalanceAction)+nextAction cluster =+  let ClusterState {peers, distribution, rebalanceOrd} = PS.infimumValue cluster+  in (rebalanceOrd, fst (rebalanceAction (Map.keysSet peers) distribution)) 

@@ -4,6 +4,7 @@ module Network.Legion.Conduit (   chanToSource,   chanToSink,+  mergeE,   merge, ) where @@ -37,11 +38,23 @@   that same source, but the interleaving of items from both sources   is nondeterministic. -}-merge :: (MonadIO io) => Source IO a -> Source IO b -> Source io (Either a b)-merge left right = do+mergeE :: (MonadIO io) => Source IO a -> Source IO b -> Source io (Either a b)+mergeE left right = do   chan <- liftIO newChan   (liftIO . void . forkIO) (left $= CL.map Left $$ chanToSink chan)   (liftIO . void . forkIO) (right $= CL.map Right $$ chanToSink chan)   chanToSource chan+++{- |+  Like `mergeE`, but without `Either` in the type signature, because+  both input sources are of the same type.+-}+merge :: (MonadIO io) => Source IO a -> Source IO a -> Source io a+merge left right = mergeE left right $= CL.map unEither+  where+    unEither :: Either a a -> a+    unEither (Left a) = a+    unEither (Right a) = a  

@@ -18,19 +18,22 @@  import Prelude hiding (null) +import Control.Monad.IO.Class (MonadIO) import Data.Aeson (ToJSON, toJSON, object, (.=)) import Data.Binary (Binary) import Data.Function (on) import Data.List (sort, sortBy)-import Data.Set (Set, toList)+import Data.Map (Map)+import Data.Monoid ((<>))+import Data.Set (Set) import Data.Text (pack) import Data.UUID (UUID) import GHC.Generics (Generic) import Network.Legion.KeySet (KeySet, member, (\\), null)-import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey) import Network.Legion.UUID (getUUID) import Text.Read (readPrec)+import qualified Data.Map as Map import qualified Data.Set as Set import qualified Network.Legion.KeySet as KS @@ -110,77 +113,170 @@   {- |-  Return the best action, if any, that the indicated peer should take to-  rebalance an unbalanced distribution.+  Return the best action, if any, that should be taken to rebalance an+  unbalanced distribution, along with the resulting distribution. -} rebalanceAction-  :: Peer-  -> Set Peer+  :: Set Peer {- ^ The set of all peers in the cluster. -}   -> ParticipationDefaults-  -> Maybe RebalanceAction-rebalanceAction self allPeers (D dist) =-    rebuild-    {- TODO rebalance -}+  -> (RebalanceAction, ParticipationDefaults)+rebalanceAction allPeers distribution =+    let+      action = underServed <> overServed <> underUtilized+      newDist = case action of+        NoAction -> D dist+        Invite peer ks -> modify (Set.insert peer) ks (D dist)+        Drop peer ks -> modify (Set.delete peer) ks (D dist)+    in (action, newDist)   where-    _rebalance :: a-    _rebalance = error "rebalance undefined"-    rebuild =++    {- | Remove any defunct peers from the distribution. -}+    dist =        let-        {- |-          Figure out if there are any under-served partitions and also-          figure out if this peer is the best candidate to service-          them. "Under served" means that the partition isn't replicated-          enough times, where "enough" is the magic number 3.-        -}+        distPeers = Set.unions (snd <$> unD distribution)+        defunct = distPeers Set.\\ allPeers+      in+        unD (modify (Set.\\ defunct) KS.full distribution)+++    {- |+      Figure out if there are any under-served partitions and also figure+      out if this peer is the best candidate to service them . "Under+      served" means that the partition isn't replicated enough times,+      where "enough" is the magic number 3.+    -}+    underServed :: RebalanceAction+    underServed =+      let         underserved = [             (ks, ps)             | (ks, ps) <- dist             , Set.size ps < 3-            , not (self `Set.member` ps)           ]         mostUnderserved = sortBy (compare `on` Set.size . snd) underserved       in case mostUnderserved of-        [] -> Nothing+        [] -> NoAction         (ks, ps):_ ->           let             {- |               Any peer that is not currently servicing the keyspace               segment is a candidate.             -}-            candidateHosts = toList (allPeers Set.\\ ps)+            candidateHosts = Set.toAscList (allPeers Set.\\ ps)              {- |               The best candidate is the one that currently has the               least load.             -}-            bestHosts = sort [(weightOf p, p) | p <- candidateHosts]+            bestHosts = sort [(load p, p) | p <- candidateHosts]           in case bestHosts of-            {- we are the best host -}-            (_, candidate):_ | candidate == self -> Just (Invite ks)-            _ -> Nothing+            (currentLoad, candidate):_ ->+              {-+                Don't be too eager to take on too much additional+                load, because if we take more than our fair share, then+                the extra is just going to get rebalanced away almost+                immediately, leading to inefficiency.+              -}+              let+                additionalLoad :: KeySet+                additionalLoad = KS.take (idealLoad - currentLoad) ks+              in Invite candidate additionalLoad+            _ -> NoAction -    weightOf p = sum [KS.size ks | (ks, ps) <- dist, p `Set.member` ps]+    {- |+      Figure out if there are any partitions being over served and also+      figure out if we are the best candidate to drop them. "Over served"+      means that the partition it replicated too many times. "Too many times"+      is anything over the magic number, 3.+    -}+    overServed :: RebalanceAction+    overServed =+      let+        {- | 'over' maps peers to the set of keys that peer should drop. -}+        over :: Map Peer KeySet+        over = Map.filter (not . KS.null) . Map.fromList $ [+            (candidate, foldr KS.union KS.empty [+                ks+                | (ks, ps) <- dist+                , Set.size ps > 3+                , best:_ <- [sortBy (flip compare `on` load) (Set.toList ps)]+                , best == candidate+              ])+            | candidate <- Set.toList allPeers+          ]+      in case Map.toAscList over of+          [] -> NoAction+          (peer, ks):_ -> Drop peer ks +    {- |+      Figure out which peer is most underutilized with respect to the+      rest of the cluster and also what keys that peer should begin to+      serve to correct the underutilization.+    -}+    underUtilized :: RebalanceAction+    underUtilized =+      let+        under = sortBy (compare `on` load) [+            p+            | p <- Set.toList allPeers+            , load p + 1 < idealLoad+          ]+        over = sortBy (flip compare `on` load) [+            p+            | p <- Set.toList allPeers+            , load p > idealLoad+          ]+      in case (under, over) of+        (u:_, o:_) | u /= o ->+          {-+            Figure out which keys to take, which is a selection of+            the difference between them large enough to move the under+            utilized peer up to the ideal load.+          -}+          let+            difference = (servicedBy o \\ servicedBy u)+            keys = KS.take (idealLoad - load u) difference+          in Invite u keys +        _ -> NoAction +    {- | Figure out how much load a peer is servicing.  -}+    load :: Peer -> Integer+    load = KS.size . servicedBy +    {- | The ideal load for each peer.  -}+    idealLoad :: Integer+    idealLoad =+      let+        total = KS.size KS.full * 3+        numPeers = toInteger (Set.size allPeers)+      in (total `div` numPeers) + 1++    {- | Figure out the keyspace serviced by a peer.  -}+    servicedBy :: Peer -> KeySet+    servicedBy p = foldr KS.union KS.empty [+        ks+        | (ks, ps) <- dist+        , p `Set.member` ps+      ]++ {- | The actions that are taken in order to build a balanced cluster. -} data RebalanceAction-  = Invite KeySet+  = Invite Peer KeySet+  | Drop Peer KeySet+  | NoAction   deriving (Show, Generic) instance Binary RebalanceAction+instance Monoid RebalanceAction where+  mempty = NoAction+  mappend NoAction a = a+  mappend a _ = a   {- |   Create a new peer. -}-newPeer :: LIO Peer+newPeer :: (MonadIO m) => m Peer newPeer = Peer <$> getUUID----- {- |---   Trace helper--- -}--- t :: (Show a) => String -> a -> a--- t msg a = trace (msg ++ ": " ++ show a) a  

@@ -16,7 +16,8 @@   empty,   null,   fromRange,-  full+  full,+  minView, ) where  import Prelude hiding (take, null)@@ -24,7 +25,8 @@ import Data.Binary (Binary(put, get)) import Data.Ranged (Range(Range), RSet, rSetEmpty, Boundary(BoundaryBelow,   BoundaryAbove, BoundaryAboveAll, BoundaryBelowAll), makeRangedSet,-  rSetHas, rSetUnion, (-!-), unsafeRangedSet, rSetRanges)+  rSetHas, rSetUnion, (-!-), unsafeRangedSet, rSetRanges, rangeLower,+  rSingleton) import GHC.Generics (Generic) import Network.Legion.PartitionKey (PartitionKey(K, unKey)) @@ -34,8 +36,14 @@   semantics, but unlike `Data.Set.Set`, it performs well with dense sets   because it only stores the set of continuous ranges in memory. -}-newtype KeySet = S {unS :: RSet PartitionKey} deriving (Show, Eq)+newtype KeySet = S {unS :: RSet PartitionKey} deriving (Eq) +{- |+  Make a less cluttered 'Show' instance by removing all the newtype rapping.+-}+instance Show KeySet where+  showsPrec p = showsPrec p . rSetRanges . unS+ instance Binary KeySet where   put =       put . fmap encodeRange . rSetRanges . unS@@ -201,5 +209,21 @@       Range (BoundaryAbove a) (BoundaryAbove (fromI (toI a + n)))     takeRange n (Range (BoundaryBelow a) _) =       Range (BoundaryBelow a) (BoundaryBelow (fromI (toI a + n)))+++{- |+  Return the minimum key in the set, along with the set stripped of+  that key.+-}+minView :: KeySet -> Maybe (PartitionKey, KeySet)+minView (S rset) =+  case rSetRanges rset of+    [] -> Nothing+    r:_ ->+      case rangeLower r of+        BoundaryAbove key -> Just (succ key, S (rset -!- rSingleton (succ key)))+        BoundaryBelow key -> Just (key, S (rset -!- rSingleton key))+        BoundaryAboveAll -> Nothing+        BoundaryBelowAll -> Just (minBound, S (rset -!- rSingleton minBound))  

@@ -0,0 +1,76 @@+{- |+  This module contains some utilities for dealing with monad transformer stacks.+-}+module Network.Legion.Lift (+  lift2,+  lift3,+  lift4,+  lift5,+) where++import Control.Monad.Trans.Class (MonadTrans, lift)++{- | Lift from two levels down in a monad transformation stack. -}+lift2+  :: (+    MonadTrans a,+    MonadTrans b,+    Monad m,+    Monad (a m)+  )+  => m r+  -> b (a m) r+lift2 = lift . lift+++{- | Lift from three levels down in a monad transformation stack. -}+lift3+  :: (+    MonadTrans a,+    MonadTrans b,+    MonadTrans c,+    Monad m,+    Monad (a m),+    Monad (b (a m))+  )+  => m r+  -> c (b (a m)) r+lift3 = lift . lift . lift+++{- | Lift from four levels down in a monad transformation stack. -}+lift4+  :: (+    MonadTrans a,+    MonadTrans b,+    MonadTrans c,+    MonadTrans d,+    Monad m,+    Monad (a m),+    Monad (b (a m)),+    Monad (c (b (a m)))+  )+  => m r+  -> d (c (b (a m))) r+lift4 = lift . lift . lift . lift+++{- | Lift from five levels down in a monad transformation stack. -}+lift5+  :: (+    MonadTrans a,+    MonadTrans b,+    MonadTrans c,+    MonadTrans d,+    MonadTrans e,+    Monad m,+    Monad (a m),+    Monad (b (a m)),+    Monad (c (b (a m))),+    Monad (d (c (b (a m))))+  )+  => m r+  -> e (d (c (b (a m)))) r+lift5 = lift . lift . lift . lift . lift++

@@ -1,3 +1,4 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {- |   This module contains the PartitionKey type. -}@@ -20,7 +21,10 @@   {- | This is how partitions are identified and referenced. -}-newtype PartitionKey = K {unKey :: Word256} deriving (Eq, Ord, Show, Bounded)+newtype PartitionKey = K {+    unKey :: Word256+  }+  deriving (Eq, Ord, Show, Bounded, Enum)  instance Binary PartitionKey where   put (K (Word256 (Word128 a b) (Word128 c d))) = put (a, b, c, d)

@@ -1,194 +1,24 @@-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE MultiParamTypeClasses #-} {- |   This module contains types related to the partition state. -} module Network.Legion.PartitionState (-  PartitionPropState,   PartitionPowerState,-  ask,-  mergeEither,-  actions,-  new,-  initProp,-  participating,-  getPowerState,-  event,-  heartbeat,-  participate,-  projParticipants,-  projected,-  infimum,-  idle,+  PartitionPowerStateT, ) where -import Data.Aeson (ToJSON)-import Data.Binary (Binary)-import Data.Default.Class (Default)-import Data.Set (Set)-import Data.Time.Clock (UTCTime) import Network.Legion.Distribution (Peer) import Network.Legion.PartitionKey (PartitionKey)-import Network.Legion.PowerState (Event)-import Network.Legion.Propagation (PropState, PropPowerState)-import qualified Network.Legion.Propagation as P--{- |-  This is an opaque representation of your application's partition state.-  Internally, this represents the complete, nondeterministic set of states the-  partition can be in as a result of concurrency, eventual consistency, and all-  the other distributed systems reasons your partition state might have more-  than one value.--  You can save these guys to disk in your `Network.Legion.Persistence`-  layer by using its `Binary` instance.--}-newtype PartitionPowerState e o s = PartitionPowerState {-    unPowerState :: PropPowerState PartitionKey s Peer e o-  } deriving (Show, Binary)---{- |-  A reification of `PropState`, representing the propagation state of the-  partition state.--}-newtype PartitionPropState e o s = PartitionPropState {-    unPropState :: PropState PartitionKey s Peer e o-  } deriving (Eq, Show, ToJSON)----- {- |---   A convenient alias for the partition state infimum.--- -}--- type PartitionInfimum s = Infimum s Peer---{- |-  Get the projected partition state value.--}-ask :: (Event e o s) => PartitionPropState e o s -> s-ask = P.ask . unPropState---{- |-  Try to merge two partition states.--}-mergeEither :: (Show e, Show s, Event e o s)-  => Peer-  -> PartitionPowerState e o s-  -> PartitionPropState e o s-  -> Either String (PartitionPropState e o s)-mergeEither peer ps prop =-  PartitionPropState <$>-    P.mergeEither peer (unPowerState ps) (unPropState prop)---{- |-  Get the peers which require action (i.e. Send), if any, and the-  powerstate version to send to those peers, and the new propagation-  state that is applicable after those actions have been taken.--}-actions-  :: PartitionPropState e o s-  -> (Set Peer, PartitionPowerState e o s, PartitionPropState e o s)-actions prop =-  let (peers, ps, newProp) = P.actions (unPropState prop)-  in (peers, PartitionPowerState ps, PartitionPropState newProp)---{- |-  Create a new, default, PartitionPropState.--}-new :: (Default s)-  => PartitionKey-    {- ^ The power state origin, which is the partition key. -}-  -> Peer-    {- ^ self -}-  -> Set Peer-    {- ^ The default participation. -}-  -> PartitionPropState e o s-new key self = PartitionPropState . P.new key self---{- |-  Initialize a `PartitionPropState` based on the initial underlying-  partition power state.--}-initProp :: (Event e o s)-  => Peer-  -> PartitionPowerState e o s-  -> PartitionPropState e o s-initProp self = PartitionPropState . P.initProp self . unPowerState---{- |-  Return `True` if the local peer is participating in the partition-  power state.--}-participating :: PartitionPropState e o s -> Bool-participating = P.participating . unPropState---{- |-  Get an opaque encapsulation of the partition power state, for-  transferring accros the network or whatever.--}-getPowerState :: PartitionPropState e o s -> PartitionPowerState e o s-getPowerState = PartitionPowerState . P.getPowerState . unPropState---{- | Apply an event to the partition state.  -}-event :: (Event e o s)-  => e-  -> PartitionPropState e o s-  -> PartitionPropState e o s-event d = PartitionPropState . P.event d . unPropState---{- | Move time forward for the propagation state.  -}-heartbeat :: UTCTime -> PartitionPropState e o s -> PartitionPropState e o s-heartbeat now = PartitionPropState . P.heartbeat now . unPropState---{- |-  Allow a participant to join in the distributed nature of the power state.--}-participate :: (Event e o s)-  => Peer-  -> PartitionPropState e o s-  -> PartitionPropState e o s-participate peer = PartitionPropState . P.participate peer . unPropState---{- |-  Return the projected peers which are participating in the partition-  state.--}-projParticipants :: PartitionPropState e o s -> Set Peer-projParticipants = P.projParticipants . unPropState---{- |-  Get the projected value of a `PartitionPowerState`.--}-projected :: (Event e o s) => PartitionPowerState e o s -> s-projected = P.projected . unPowerState-+import Network.Legion.PowerState (PowerState)+import Network.Legion.PowerState.Monad (PowerStateT) -{- |-  Get the infimum value of a `PartitionPowerState`--}-infimum :: PartitionPowerState e o s -> s-infimum = P.infimum . unPowerState+{- | A representation of all possible partition states. -}+type PartitionPowerState e o s = PowerState PartitionKey s Peer e o   {- |-  Figure out if this propagation state has any work to do. Return 'True' if all-  known propagation work has been completed. The implication here is that the-  only way more work can happen is if new events are applied, either directly-  or via a merge.+  A convenient spelling for the partition-flavored power state monad+  transformer. -}-idle :: PartitionPropState e o s -> Bool-idle = P.idle . unPropState+type PartitionPowerStateT e o s = PowerStateT PartitionKey s Peer e o  

@@ -3,31 +3,40 @@ {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-}-{- | This module contains the fundamental distributed data object. -}+{- |+  This module contains the fundamental distributed data object.++  A note on terminology, "divergent" in this context referes to events+  which are not known to have been acknowledged by all participating+  peers.+-} module Network.Legion.PowerState (   PowerState,-  Infimum(..),   Event(..),   StateId,+  DifferentOrigins(..),+   new,+  event,   merge,   mergeMaybe,   mergeEither,   acknowledge,+   participate,   disassociate,+   projectedValue,   infimumValue,   infimumParticipants,   allParticipants,   projParticipants,   divergent,-  divergences,-  event, ) where  import Prelude hiding (null) +import Control.Exception (throw, Exception) import Data.Aeson (ToJSON, toJSON, object, (.=)) import Data.Binary (Binary(put, get)) import Data.Default.Class (Default(def))@@ -35,6 +44,7 @@ import Data.Map (Map, filterWithKey, unionWith, minViewWithKey, keys,   toDescList, toAscList, fromAscList) import Data.Set (Set, union, (\\), null, member)+import Data.Typeable (Typeable) import Data.Word (Word64) import GHC.Generics (Generic) import qualified Data.Map as Map@@ -53,7 +63,7 @@      events :: Map (StateId p) (Delta p e, Set p)   } deriving (Generic, Show, Eq) instance (Binary o, Binary s, Binary p, Binary e) => Binary (PowerState o s p e r)-instance (Show o, Show s, Show p, Show e) => ToJSON (PowerState o s p e r) where+instance (Show o, ToJSON s, Show p, Show e) => ToJSON (PowerState o s p e r) where   toJSON PowerState {origin, infimum, events} = object [       "origin" .= show origin,       "infimum" .= infimum,@@ -78,11 +88,11 @@   Infimum s1 _ _ == Infimum s2 _ _ = s1 == s2 instance (Ord p) => Ord (Infimum s p) where   compare (Infimum s1 _ _) (Infimum s2 _ _) = compare s1 s2-instance (Show s, Show p) => ToJSON (Infimum s p) where+instance (ToJSON s, Show p) => ToJSON (Infimum s p) where   toJSON Infimum {stateId, participants, stateValue} = object [       "stateId" .= show stateId,       "participants" .= Set.map show participants,-      "stateValue" .= show stateValue+      "stateValue" .= stateValue     ]  @@ -113,6 +123,14 @@   {- |+  This is the exception type for illegal merges. An illegal merge is+  one where the two PowerStates do not share the same origin.+-}+data DifferentOrigins o = DifferentOrigins o o deriving (Show, Typeable)+instance (Typeable o, Show o) => Exception (DifferentOrigins o)+++{- |   `Delta` is how we represent mutations to the power state. -} data Delta p e@@ -134,7 +152,8 @@   {- |-  Construct a new PowerState with the given origin and initial participants+  Construct a new PowerState with the given origin and initial+  participants. -} new :: (Default s) => o -> Set p -> PowerState o s p e r new origin participants =@@ -155,21 +174,21 @@   a lower one. This function is not total. Only `PowerState`s that originated   from the same `new` call can be merged. -}-merge :: (Eq o, Event e r s, Ord p, Show o, Show s, Show p, Show e)+merge :: (Eq o, Event e r s, Ord p, Show o, Typeable o)   => PowerState o s p e r   -> PowerState o s p e r-  -> PowerState o s p e r-merge a b = either error id (mergeEither a b)+  -> (PowerState o s p e r, Map (StateId p) r)+merge a b = either throw id (mergeEither a b)   {- |   Like `merge`, but safe. Returns `Nothing` if the two power states do   not share the same origin. -}-mergeMaybe :: (Eq o, Event e r s, Ord p, Show o, Show s, Show p, Show e)+mergeMaybe :: (Eq o, Event e r s, Ord p)   => PowerState o s p e r   -> PowerState o s p e r-  -> Maybe (PowerState o s p e r)+  -> Maybe (PowerState o s p e r, Map (StateId p) r) mergeMaybe a b = either (const Nothing) Just (mergeEither a b)  @@ -177,10 +196,10 @@   Like `mergeMaybe`, but returns a human-decipherable error message of   exactly what went wrong. -}-mergeEither :: (Eq o, Event e r s, Ord p, Show o, Show s, Show p, Show e)+mergeEither :: (Eq o, Event e r s, Ord p)   => PowerState o s p e r   -> PowerState o s p e r-  -> Either String (PowerState o s p e r)+  -> Either (DifferentOrigins o) (PowerState o s p e r, Map (StateId p) r) mergeEither (PowerState o1 i1 d1) (PowerState o2 i2 d2) | o1 == o2 =     Right . reduce . removeRenegade $ PowerState {         origin = o1,@@ -225,9 +244,8 @@      mergeAcks (e, s1) (_, s2) = (e, s1 `union` s2) -mergeEither a b = Left-  $ "PowerStates " ++ show a ++ " and " ++ show b ++ " do not share the "-  ++ "same origin, and cannot be merged."+mergeEither PowerState {origin = o1} PowerState {origin = o2} =+  Left (DifferentOrigins o1 o2)   {- |@@ -238,7 +256,7 @@ acknowledge :: (Event e r s, Ord p)   => p   -> PowerState o s p e r-  -> PowerState o s p e r+  -> (PowerState o s p e r, Map (StateId p) r) acknowledge p ps@PowerState {events} =     reduce ps {events = fmap ackOne events}   where@@ -248,11 +266,11 @@ {- |   Allow a participant to join in the distributed nature of the power state. -}-participate :: (Event e r s, Ord p)+participate :: (Ord p)   => p   -> PowerState o s p e r   -> PowerState o s p e r-participate p ps@PowerState {events} = acknowledge p $ ps {+participate p ps@PowerState {events} = ps {     events = Map.insert (nextId p ps) (Join p, Set.empty) events   } @@ -261,26 +279,30 @@   Indicate that a participant is removing itself from participating in   the distributed power state. -}-disassociate :: (Event e r s, Ord p)+disassociate :: (Ord p)   => p   -> PowerState o s p e r   -> PowerState o s p e r-disassociate p ps@PowerState {events} = acknowledge p $ ps {+disassociate p ps@PowerState {events} = ps {     events = Map.insert (nextId p ps) (UnJoin p, Set.empty) events   }   {- |   Introduce a change to the PowerState on behalf of the participant.+  Return the new powerstate along with the projected output of the event. -}-event :: (Event e r s, Ord p)+event :: (Ord p, Event e r s)   => p   -> e   -> PowerState o s p e r-  -> PowerState o s p e r-event p e ps@PowerState {events} = acknowledge p $ ps {-    events = Map.insert (nextId p ps) (Event e, Set.empty) events-  }+  -> (r, PowerState o s p e r)+event p e ps@PowerState {events} = (+    fst (apply e (projectedValue ps)),+    ps {+        events = Map.insert (nextId p ps) (Event e, Set.empty) events+      }+  )   {- |@@ -342,8 +364,8 @@  {- |   Returns the participants that we think might be diverging. In this-  context, a peer is "diverging" if there is an event that the peer has-  not acknowledged.+  context, a participant is "diverging" if there is an event that the+  participant has not acknowledged. -} divergent :: (Ord p) => PowerState o s p e r -> Set p divergent PowerState {@@ -355,7 +377,7 @@     {- |       `accum` mnemonics:         j = pro(J)ected participants-        d = (D)ivergent participants+        d = (D)iverging participants         a = peers that have (A)cknowledged an update.         p = (P)eer that is joining or unjoining     -}@@ -371,7 +393,8 @@     accum j d ((_, (UnJoin p, a)):moreDeltas) =       let         j2 = Set.delete p j-        d2 = (j2 \\ a) `union` d+        {- A participant must acknowledge its own unjoin. -}+        d2 = (j \\ a) `union` d       in         accum j2 d2 moreDeltas @@ -383,15 +406,22 @@   {- |-  Return the events that are unknown to the specified peer.+  Return all divergent events, along with the set of peers for which we+  are expecting an acknowledgement of the event. -}-divergences :: (Ord p) => p -> PowerState o s p e r -> Map (StateId p) e-divergences peer PowerState {events} =-  fromAscList [-    (sid, e)-    | (sid, (Event e, p)) <- toAscList events-    , not (peer `member` p)-  ]+_divergences :: (Ord p) => PowerState o s p e r -> Map (StateId p) (e, Set p)+_divergences PowerState {events, infimum} =+    go (participants infimum) (Map.toAscList events)+  where+    go :: (Ord p)+      => Set p+      -> [(StateId p, (Delta p e, Set p))]+      -> Map (StateId p) (e, Set p)+    go _ [] = Map.empty+    go ps ((sid, (Event e, p)):moreEvents) =+      Map.insert sid (e, ps \\ p) (go ps moreEvents)+    go ps ((_, (Join p, _)):moreEvents) = go (Set.insert p ps) moreEvents+    go ps ((_, (UnJoin p, _)):moreEvents) = go (Set.delete p ps) moreEvents   {- |@@ -399,16 +429,18 @@   has enough information to derive a new infimum value. In other words,   this is where garbage collection happens. -}-reduce :: (Event e r s, Ord p) => PowerState o s p e r -> PowerState o s p e r+reduce :: (Event e r s, Ord p)+  => PowerState o s p e r+  -> (PowerState o s p e r, Map (StateId p) r) reduce ps@PowerState {     infimum = infimum@Infimum {participants, stateValue},     events   } =     case minViewWithKey events of-      Nothing -> ps+      Nothing -> (ps, Map.empty)       Just ((sid, (update, acks)), newDeltas) ->         if not . null $ participants \\ acks-          then ps+          then (ps, Map.empty)           else case update of             Join p -> reduce ps {                 infimum = infimum {@@ -424,13 +456,17 @@                   },                 events = newDeltas               }-            Event e -> reduce ps {-                infimum = infimum {-                    stateId = sid,-                    stateValue = snd (apply e stateValue)-                  },-                events = newDeltas-              }+            Event e ->+              let+                (output, newState) = apply e stateValue+                (ps2, outputs) = reduce ps {+                    infimum = infimum {+                        stateId = sid,+                        stateValue = newState+                      },+                    events = newDeltas+                  }+              in (ps2, Map.insert sid output outputs)   {- |

@@ -0,0 +1,160 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{- |+  This module provides a monadic interface for power state manipulation,+  where the monadic context contains the current value of the power state,+  the collection of outputs for events that have reached the infimum,+  and the collection of actions that should be taken to propagate the+  powerstate to all other peers.+-}+module Network.Legion.PowerState.Monad (+  PowerStateT,+  runPowerStateT,++  PropAction(..),++  event,+  merge,+  acknowledge,++  participate,+  disassociate,+) where++import Control.Monad.Trans.Class (MonadTrans, lift)+import Control.Monad.Trans.Except (ExceptT, runExceptT, throwE)+import Control.Monad.Trans.Reader (ReaderT, runReaderT, ask)+import Control.Monad.Trans.State (StateT, runStateT, get, put, modify)+import Control.Monad.Trans.Writer (WriterT, runWriterT, tell)+import Data.Default.Class (Default, def)+import Data.Map (Map)+import Network.Legion.Lift (lift2, lift3, lift4, lift5)+import Network.Legion.PowerState (StateId, DifferentOrigins, Event, PowerState)+import qualified Network.Legion.PowerState as PS+++{- |+  Monad Transformer that manages the powerstate value, accumulated infimum+  outputs, and the actions necessary for propagation as monadic context.+-}+newtype PowerStateT o s p e r m a = PowerStateT {+    unPowerStateT ::+      StateT (PowerState o s p e r) ( {- Maintain the power state value. -}+      StateT PropAction (             {- Maintain the propagation actions. -}+      ReaderT p (                     {- Provide the 'self' value. -}+      WriterT (Map (StateId p) r) (   {- Accumulate the infimum outputs. -}+      ExceptT (DifferentOrigins o) m)))) a+  }+  deriving (Functor, Applicative, Monad)+instance (Ord p) => MonadTrans (PowerStateT o s p e r) where+  lift = PowerStateT . lift5+++{- | Run a PowerStateT monad.  -}+runPowerStateT :: (Monad m)+  => p {- ^ self -}+  -> PowerState o s p e r+  -> PowerStateT o s p e r m a+  -> m (+      Either+        (DifferentOrigins o)+        (+          a,+          PropAction,+          PowerState o s p e r,+          Map (StateId p) r+        )+    )+runPowerStateT self ps =+    (fmap . fmap) flatten+    . runExceptT+    . runWriterT+    . (`runReaderT` self)+    . (`runStateT` def)+    . (`runStateT` ps)+    . unPowerStateT+  where+    {- |+      This just converts the tuple structure of the monad transformation+      stack into the tuple structure we want to expose to the user.+    -}+    flatten (((a, ps2), prop), outputs) = (a, prop, ps2, outputs)+++{- |+  The action that needs to be taken to distribute any new information.+-}+data PropAction+  = DoNothing+  | Send+  deriving (Show, Eq)+instance Default PropAction where+  def = DoNothing+++{- | Add a user event. Return the projected output of the event. -}+event :: (Monad m, Ord p, Event e r s) => e -> PowerStateT o s p e r m r+event e = PowerStateT $ do+  self <- lift2 ask+  (r, ps) <- PS.event self e <$> get+  put ps+  return r+++{- |+  Monotonically merge the information in two power states.  The resulting+  power state may have a higher infimum value, but it will never+  have a lower one. This function is not total. Only `PowerState`s+  that originated from the same `new` call can be merged. This can+  potentially throw a 'DifferentOrigins' if the origin of @__other__@+  is not the same as the origin of the powerstate in the monadic context.+-}+merge :: (Monad m, Ord p, Eq o, Event e r s)+  => PowerState o s p e r+  -> PowerStateT o s p e r m ()+merge other = PowerStateT $ do+  ps <- get+  case PS.mergeEither other ps of+    Left err -> lift4 (throwE err)+    Right (merged, outputs) -> do+      lift3 (tell outputs)+      put merged+++{- |+  Record the fact that the participant acknowledges the information+  contained in the powerset. The implication is that the participant+  __must__ base all future operations on the result of this function.+-}+acknowledge :: (Monad m, Ord p, Event e r s, Eq e, Eq o)+  => PowerStateT o s p e r m ()+acknowledge = PowerStateT $ do+  ps <- get+  prop <- lift get+  self <- lift2 ask+  let+    (ps2, outputs) = PS.acknowledge self ps+    prop2 = if ps2 /= ps+      then Send+      else prop+  put ps2+  (lift . put) prop2+  (lift3 . tell) outputs+++{- |+  Allow a participant to join in the distributed nature of the power state.+-}+participate :: (Monad m, Ord p) => p -> PowerStateT o s p e r m ()+participate newPeer = PowerStateT $+  modify (PS.participate newPeer)+++{- |+  Indicate that a participant is removing itself from participating in+  the distributed power state.+-}+disassociate :: (Monad m, Ord p) => p -> PowerStateT o s p e r m ()+disassociate peer = PowerStateT $+  modify (PS.disassociate peer)++

@@ -1,392 +0,0 @@-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE OverloadedStrings #-}-{- |-  This module defines how to propagate a PowerState amoung its participants.--}-module Network.Legion.Propagation (-  PropState,-  PropPowerState,-  merge,-  mergeMaybe,-  mergeEither,-  heartbeat,-  event,-  actions,-  new,-  initProp,-  getPowerState,-  ask,-  participate,-  disassociate,-  getSelf,-  divergences,-  participating,-  allParticipants,-  projParticipants,-  projected,-  infimum,-  idle,-) where--import Prelude hiding (lookup)--import Data.Aeson (ToJSON, object, (.=), toJSON)-import Data.Binary (Binary)-import Data.Default.Class (Default)-import Data.Map (Map, lookup)-import Data.Maybe (fromMaybe)-import Data.Set (member, Set)-import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime)-import Data.Time.Format () -- For `instance Show UTCTime`-import Network.Legion.PowerState (PowerState, divergent, Event,-  acknowledge, projectedValue, StateId)-import qualified Data.Map as Map-import qualified Data.Set as Set-import qualified Network.Legion.PowerState as PS---{- |-  Internally, we use `Maybe UTCTime` to represent the current time, so that we-  have a convenient way to represent "now" (i.e. `Nothing`) without using `IO`.-  This type aliases gives us a convenient way to spell `Maybe UTCTime`.--}-type Time = Maybe UTCTime---{- |-  Opaque Propagation State. Values of this type encapsulate the-  current value of a power state along with state having to do with-  the distribution of that powerstate among its participants. The-  power state is not directly accessible, but rather must be accessed-  through functions provided by this module. In addition to providing-  a more coherent hierarchy of abstraction, this also helps ensure that-  the power state remains consistent with the state of its propagation-  throughout the network.--}-data PropState o s p d r = PropState {-    powerState :: PowerState o s p d r,-    peerStates :: Map p PeerStatus,-          self :: p,-           now :: Time-  } deriving (Eq, Show)-instance (Show o, Show s, Show p, Show d) => ToJSON (PropState o s p d r) where-  toJSON PropState {powerState, peerStates, self, now} = object [-      "powerState" .= powerState,-      "peerStates" .= Map.fromList [-          (show p, show s)-          | (p, s) <- Map.toList peerStates-        ],-      "self" .= show self,-      "now" .= show now-    ]---{- |-  This type is an opaque representation of the underlying power state. It-  exists because we sometimes want to pack up the power state and ship-  it over the network, but we don't want any code outside of this module-  to operate on it.--}-newtype PropPowerState o s p d r = PropPowerState {-    unPowerState :: PowerState o s p d r-  } deriving (Show, Binary)---{- |-  Retriev the current projected value of the underlying state.--}-ask :: (Event d r s) => PropState o s p d r -> s-ask = projectedValue . powerState---{- |-  Create a new propagation state based on an existing power state.--}-initProp :: (Event d r s, Ord p)-  => p-  -> PropPowerState o s p d r-  -> PropState o s p d r-initProp self ps =-  let powerState = acknowledge self (unPowerState ps)-  in PropState {-      powerState = powerState,-      peerStates = Map.fromAscList [-          (p, NeedsSendAt Nothing)-          | p <- Set.toAscList (divergent powerState)-        ],-      self,-      now = Nothing-    }---{- |-  Return an opaque representation of the power state, for transfer across-  the network, or whatever.--}-getPowerState :: PropState o s p d r -> PropPowerState o s p d r-getPowerState = PropPowerState . powerState---{- |-  The propagation state of a single remote participant.--}-data PeerStatus-  = NeedsSendAt Time-  | NeedsAck-  deriving (Show, Eq)---{- |-  Create a new propagation state.--}-new :: (Default s) => o -> p -> Set p -> PropState o s p d r-new origin self participants =-  PropState {-      powerState = PS.new origin participants,-      peerStates = Map.empty,-      self,-      now = Nothing-    }---{- |-  Like `merge`, but total. `mergeEither` returns a human readable reason why-  the foreign powerstate can't be merged in the event of an error.--}-mergeEither :: (Eq o, Ord p, Show o, Show s, Show p, Show d, Event d r s)-  => p-  -> PropPowerState o s p d r-  -> PropState o s p d r-  -> Either String (PropState o s p d r)-mergeEither source kernel (prop@PropState {powerState, peerStates, self, now}) =-  let ps = unPowerState kernel-  in case acknowledge self <$> PS.mergeEither ps powerState of-    Left err -> Left err-    Right merged -> Right prop {-        powerState = merged,--        {--          This algorithm is weaksauce. We need to find someone who knows-          a lot about gossip protocols to fix this.-        -}-        peerStates =-          Map.fromList $ [-              (p, ns)-              | p <- Set.toList (divergent merged)-              , let ns = fromMaybe (NeedsSendAt now) (lookup p peerStates)-            ]-          ++-            {--              If the source of the foreign powerstate believes we-              are divergent, then it is going to keep sending updates-              until someone clues it in. That someone is us for now.-            -}-            [(source, NeedsAck) | self `member` divergent ps]-      }---{- |-  Like `merge`, but total. `mergeMaybe` returns `Nothing` if the foreign power-  state can't be merged.--}-mergeMaybe :: (Eq o, Ord p, Show o, Show s, Show p, Show d, Event d r s)-  => p-  -> PropPowerState o s p d r-  -> PropState o s p d r-  -> Maybe (PropState o s p d r)-mergeMaybe source ps prop =-  case mergeEither source ps prop of-    Left _ -> Nothing-    Right v -> Just v---{- |-  Try to merge a foreign powerstate. The precondition is that the foreign-  powerstate shares the same origin as the local powerstate. If this-  precondition is not met, `error` will be called (making this function-  non-total). Using `mergeMaybe` or `mergeEither` is recommended.--}-merge :: (Eq o, Ord p, Show o, Show s, Show p, Show d, Event d r s)-  => p-  -> PropPowerState o s p d r-  -> PropState o s p d r-  -> PropState o s p d r-merge source ps prop =-  case mergeEither source ps prop of-    Left err -> error err-    Right v -> v---{- |-  Time moves forward.--}-heartbeat :: UTCTime -> PropState o s p d r -> PropState o s p d r-heartbeat newNow prop = prop {now = max (now prop) (Just newNow)}---{- |-  Apply an event.--}-event :: (Ord p, Event d r s)-  => d-  -> PropState o s p d r-  -> PropState o s p d r-event d prop@PropState {self, powerState, now} =-  let newPowerState = PS.event self d powerState-  in prop {-      powerState = newPowerState,-      peerStates = Map.fromAscList [-          (p, NeedsSendAt now)-          | p <- Set.toAscList (divergent newPowerState)-        ]-    }---{- |-  Get the peers which require action (i.e. Send), if any, and the-  powerstate version to send to those peers, and the new propagation-  state that is applicable after those actions have been taken.--}-actions :: (Eq p)-  => PropState o s p d r-  -> (Set p, PropPowerState o s p d r, PropState o s p d r)-actions prop@PropState {powerState, peerStates, now} =-    (outOfDatePeers, PropPowerState powerState, newPropState)-  where-    outOfDatePeers = Set.fromAscList [-        p-        | (p, status) <- Map.toAscList peerStates-        , shouldSendNow status-      ]--    shouldSendNow NeedsAck = True-    shouldSendNow (NeedsSendAt time) = now > time--    newPropState = prop {-        peerStates = Map.fromAscList [-            (p, ns)-            {- Careful, this pattern omits `NeedsAck`. This is intentional. -}-            | (p, NeedsSendAt time) <- Map.toAscList peerStates-            , let ns = NeedsSendAt (nextTime time)-          ]-      }--    nextTime :: Time -> Time-    nextTime time =-      if now > time-        then addUTCTime gracePeriod <$> now-        else time---{- |-  The grace period for receiving some response to an action.--}-gracePeriod :: NominalDiffTime-gracePeriod = oneMinute-  where-    oneMinute = 60---{- |-  Allow a participant to join in the distributed nature of the power state.--}-participate :: (Ord p, Event d r s)-  => p-  -> PropState o s p d r-  -> PropState o s p d r-participate peer prop@PropState {powerState, now} =-  let newPowerState = PS.participate peer powerState-  in prop {-      powerState = newPowerState,-      peerStates = Map.fromAscList [-          (p, NeedsSendAt now)-          | p <- Set.toAscList (divergent newPowerState)-        ]-    }---{- |-  Eject a participant from the power state.--}-disassociate :: (Ord p, Event d r s)-  => p-  -> PropState o s p d r-  -> PropState o s p d r-disassociate peer prop@PropState {powerState, now} =-  let newPowerState = PS.disassociate peer powerState-  in prop {-      powerState = newPowerState,-      peerStates = Map.fromAscList [-          (p, NeedsSendAt now)-          | p <- Set.toAscList (divergent newPowerState)-        ]-    }---{- |-  Return the events that are unknown to the specified peer.--}-divergences :: (Ord p) => p -> PropState o s p d r -> Map (StateId p) d-divergences peer = PS.divergences peer . powerState---{- |-  Return self.--}-getSelf :: PropState o s p d r -> p-getSelf = self---{- |-  Return `True` if the local peer is participating in the underlying-  power state. This will return `True` even if the peer is projected-  for removal, because until the infimum catches up to that projection,-  this peer still has an obligation to participate.--}-participating :: (Ord p) => PropState o s p d r -> Bool-participating PropState{self, powerState} =-  self `member` PS.allParticipants powerState---{- |-  Get all known participants. This includes participants that are-  projected for removal.--}-allParticipants :: (Ord p) => PropState o s p d r -> Set p-allParticipants = PS.allParticipants . powerState---{- |-  Get all of the projected participants.--}-projParticipants :: (Ord p) => PropState o s p d r -> Set p-projParticipants = PS.projParticipants . powerState---{- |-  Get the projected value of a PropPowerState.--}-projected :: (Event d r s) => PropPowerState o s p d r -> s-projected = PS.projectedValue . unPowerState---{- |-  Get the infimum value of the PropPowerState.--}-infimum :: PropPowerState o s p d r -> s-infimum = PS.infimumValue . unPowerState---{- |-  Figure out if this propagation state has any work to do. Return 'True' if all-  known propagation work has been completed. The implication here is that the-  only way more work can happen is if new events are applied, either directly-  or via a merge.--}-idle :: (Ord p) => PropState o s p d r -> Bool-idle PropState {powerState, peerStates} =-  Map.null peerStates && Set.null (divergent powerState)--

@@ -20,12 +20,13 @@ import Control.Concurrent (forkIO) import Control.Concurrent.Chan (writeChan, newChan, Chan) import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)-import Control.Monad (void, forever, join, (>=>))+import Control.Monad (void, forever, join) import Control.Monad.Catch (catchAll, try, SomeException, throwM) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Logger (logWarn, logError, logInfo, LoggingT,   MonadLoggerIO, runLoggingT, askLoggerIO, logDebug) import Control.Monad.Trans.Class (lift)+import Control.Monad.Trans.State (StateT, runStateT, get, put, modify) import Data.Binary (encode, Binary) import Data.Conduit (Source, ($$), (=$=), yield, await, awaitForever,   transPipe, ConduitM, runConduit, Sink)@@ -37,7 +38,7 @@ import GHC.Generics (Generic) import Network.Legion.Admin (runAdmin, AdminMessage(GetState, GetPart,   Eject))-import Network.Legion.Application (LegionConstraints, getState, Persistence)+import Network.Legion.Application (LegionConstraints, Persistence) import Network.Legion.BSockAddr (BSockAddr(BSockAddr)) import Network.Legion.ClusterState (ClusterPowerState) import Network.Legion.Conduit (merge, chanToSink, chanToSource)@@ -46,19 +47,23 @@ import Network.Legion.Index (IndexRecord(IndexRecord), irTag, irKey,   SearchTag(SearchTag)) import Network.Legion.LIO (LIO)+import Network.Legion.Lift (lift2,  lift3) import Network.Legion.PartitionKey (PartitionKey)+import Network.Legion.PartitionState (PartitionPowerState) import Network.Legion.Runtime.ConnectionManager (newConnectionManager,-  send, ConnectionManager, newPeers)+  ConnectionManager, newPeers) import Network.Legion.Runtime.PeerMessage (PeerMessage(PeerMessage),   PeerMessagePayload(ForwardRequest, ForwardResponse, ClusterMerge,-  PartitionMerge, Search, SearchResponse), MessageId, newSequence,-  nextMessageId)+  PartitionMerge, Search, SearchResponse, JoinNext, JoinNextResponse),+  MessageId, newSequence, nextMessageId, JoinNextResponse(Joined,+  JoinFinished)) import Network.Legion.Settings (RuntimeSettings(RuntimeSettings,   adminHost, adminPort, peerBindAddr, joinBindAddr)) import Network.Legion.StateMachine (partitionMerge, clusterMerge,-  NodeState, newNodeState, runSM, UserResponse(Forward, Respond),-  userRequest, heartbeat, rebalance, migrate, propagate, ClusterAction,-  eject, minimumCompleteServiceSet)+  newNodeState, UserResponse(Forward, Respond), userRequest, eject,+  minimumCompleteServiceSet, joinNext, joinNextResponse)+import Network.Legion.StateMachine.Monad (NodeState, runSM, ClusterAction,+  SM, popActions) import Network.Legion.UUID (getUUID) import Network.Socket (Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN),   SocketOption(ReuseAddr), SocketType(Stream), accept, bind,@@ -69,7 +74,9 @@ import qualified Data.Map as Map import qualified Data.Set as Set import qualified Network.Legion.ClusterState as C+import qualified Network.Legion.Runtime.ConnectionManager as CM import qualified Network.Legion.StateMachine as SM+import qualified Network.Legion.StateMachine.Monad as SMM   {- |@@ -106,37 +113,38 @@     peerS <- loggingC =<< startPeerListener settings     adminS <- loggingC =<< runAdmin adminPort adminHost     joinS <- loggingC (joinMsgSource settings)+    loopChan <- lift newChan      (self, nodeState, peers) <- makeNodeState settings startupMode-    cm <- newConnectionManager peers--    firstMessageId <- newSequence+    rts <- newRuntimeState self peers (writeChan loopChan)     let-      rts = RuntimeState {+      messageSource = transPipe lift (+          (joinS =$= CL.map J) `merge`+          (peerS =$= CL.map P) `merge`+          (requestSource =$= CL.map R) `merge`+          (adminS =$= CL.map A) `merge`+          chanToSource loopChan+        )+    void . runRTS persistence nodeState rts . runConduit $+      messageSource+      =$= messageSink+  where+    newRuntimeState :: (Binary e, Binary o, Binary s)+      => Peer+      -> Map Peer BSockAddr+      -> (RuntimeMessage e o s -> IO ())+      -> LoggingT IO (RuntimeState e o s)+    newRuntimeState self peers loop = do+      cm <- newConnectionManager peers+      firstMessageId <- newSequence+      return RuntimeState {           forwarded = Map.empty,           nextId = firstMessageId,           cm,           self,-          searches = Map.empty+          searches = Map.empty,+          loop         }-    runConduit $-      (joinS `merge` (peerS `merge` (requestSource `merge` adminS)))-        =$= CL.map toMessage-        =$= messageSink persistence (rts, nodeState)-  where-    toMessage-      :: Either-          (JoinRequest, JoinResponse -> LIO ())-          (Either-            (PeerMessage e o s)-            (Either-              (RequestMsg e o)-              (AdminMessage e o s)))-      -> RuntimeMessage e o s-    toMessage (Left m) = J m-    toMessage (Right (Left m)) = P m-    toMessage (Right (Right (Left m))) = R m-    toMessage (Right (Right (Right m))) = A m      {- |       Turn an LIO-based conduit into an IO-based conduit, so that it@@ -163,77 +171,48 @@   messageSink :: (LegionConstraints e o s)-  => Persistence e o s-  -> (RuntimeState e o s, NodeState e o s)-  -> Sink (RuntimeMessage e o s) LIO ()-messageSink persistence states =-    await >>= \case-      Nothing -> return ()-      Just msg -> do-        $(logDebug) . pack-          $ "Receieved: " ++ show msg-        lift . handleMessage persistence msg-          >=> lift . updatePeers persistence-          >=> lift . clusterHousekeeping persistence-          >=> messageSink persistence-          $ states+  => Sink (RuntimeMessage e o s) (RTS e o s) ()+messageSink = awaitForever (\msg -> do+    $(logDebug) . pack $ "Receieved: " ++ show msg+    lift $ do+      handleMessage msg+      updatePeers+      clusterActions+  )  -{- |-  Make sure the connection manager knows about any new peers that have-  joined the cluster.--}-updatePeers-  :: Persistence e o s-  -> (RuntimeState e o s, NodeState e o s)-  -> LIO (RuntimeState e o s, NodeState e o s)-updatePeers persistence (rts, ns) = do-  (peers, ns2) <- runSM persistence ns SM.getPeers-  newPeers (cm rts) peers-  return (rts, ns2)+{- | Make progress on outstanding cluster actions. -}+clusterActions :: RTS e o s ()+clusterActions =+    mapM_ clusterAction =<< popActions+  where+    {- |+      Actually perform a cluster action as directed by the state+      machine.+    -}+    clusterAction+      :: ClusterAction e o s+      -> RTS e o s () +    clusterAction (SMM.ClusterMerge peer ps) =+      void $ send peer (ClusterMerge ps) -{- |-  Perform any cluster management actions, and update the state-  appropriately.--}-clusterHousekeeping :: (LegionConstraints e o s)-  => Persistence e o s-  -> (RuntimeState e o s, NodeState e o s)-  -> LIO (RuntimeState e o s, NodeState e o s)-clusterHousekeeping persistence (rts, ns) = do-    (actions, ns2) <- runSM persistence ns (-        heartbeat-        >> rebalance-        >> migrate-        >> propagate-      )-    rts2 <- foldr (>=>) return (clusterAction <$> actions) rts-    return (rts2, ns2)+    clusterAction (SMM.PartitionMerge peer key ps) =+      void $ send peer (PartitionMerge key ps) +    clusterAction (SMM.PartitionJoin peer keys) =+      void $ send peer (JoinNext keys)+      {- |-  Actually perform a cluster action as directed by the state-  machine.+  Make sure the connection manager knows about any new peers that have+  joined the cluster. -}-clusterAction-  :: ClusterAction e o s-  -> RuntimeState e o s-  -> LIO (RuntimeState e o s)--clusterAction-    (SM.ClusterMerge peer ps)-    rts@RuntimeState {self, nextId, cm}-  = do-    send cm peer (PeerMessage self nextId (ClusterMerge ps))-    return rts {nextId = nextMessageId nextId}--clusterAction-    (SM.PartitionMerge peer key ps)-    rts@RuntimeState {self, nextId, cm}-  = do-    send cm peer (PeerMessage self nextId (PartitionMerge key ps))-    return rts {nextId = nextMessageId nextId}+updatePeers :: RTS e o s ()+updatePeers = do+  peers <- SM.getPeers+  RuntimeState {cm} <- lift get+  lift2 $ newPeers cm peers   {- |@@ -242,149 +221,129 @@   state and node state. -} handleMessage :: (LegionConstraints e o s)-  => Persistence e o s-  -> RuntimeMessage e o s-  -> (RuntimeState e o s, NodeState e o s)-  -> LIO (RuntimeState e o s, NodeState e o s)+  => RuntimeMessage e o s+  -> RTS e o s () +handleMessage {- Join Next Response -}+    (P (PeerMessage source _ (JoinNextResponse _messageId response)))+  =+    joinNextResponse source (toMaybe response)+  where+    toMaybe+      :: JoinNextResponse e o s+      -> Maybe (PartitionKey, PartitionPowerState e o s)+    toMaybe (Joined key partition) = Just (key, partition)+    toMaybe JoinFinished = Nothing++handleMessage {- Join Next -}+    (P (PeerMessage source messageId (JoinNext askKeys)))+  =+    joinNext source askKeys >>= \case+      Nothing -> void $+        send source (JoinNextResponse messageId JoinFinished)+      Just (gotKey, partition) -> void $+        send source (JoinNextResponse messageId (Joined gotKey partition))+ handleMessage {- Partition Merge -}-    persistence-    (P (PeerMessage source _ (PartitionMerge key ps)))-    (rts, ns)-  = do-    ((), ns2) <- runSM persistence ns (partitionMerge source key ps)-    return (rts, ns2)+    (P (PeerMessage _ _ (PartitionMerge key ps)))+  =+    partitionMerge key ps  handleMessage {- Cluster Merge -}-    persistence-    (P (PeerMessage source _ (ClusterMerge cs)))-    (rts, ns)-  = do-    ((), ns2) <- runSM persistence ns (clusterMerge source cs)-    return (rts, ns2)+    (P (PeerMessage _ _ (ClusterMerge cs)))+  =+    clusterMerge cs  handleMessage {- Forward Request -}-    persistence     (P (msg@(PeerMessage source mid (ForwardRequest key request))))-    (rts@RuntimeState {nextId, cm, self}, ns)   = do-    (output, ns2) <- runSM persistence ns (userRequest key request)+    output <- userRequest key request     case output of-      Respond response -> do-        send cm source (-            PeerMessage self nextId (ForwardResponse mid response)-          )-        return (rts {nextId = nextMessageId nextId}, ns2)-      Forward peer -> do-        send cm peer msg-        return (rts {nextId = nextMessageId nextId}, ns2)+      Respond response -> void $ send source (ForwardResponse mid response)+      Forward peer -> forward peer msg  handleMessage {- Forward Response -}-    _legionary     (msg@(P (PeerMessage _ _ (ForwardResponse mid response))))-    (rts, ns)-  =+  = do+    rts <- lift get     case lookupDelete mid (forwarded rts) of       (Nothing, fwd) -> do         $(logWarn) . pack $ "Unsolicited ForwardResponse: " ++ show msg-        return (rts {forwarded = fwd}, ns)+        (lift . put) rts {forwarded = fwd}       (Just respond, fwd) -> do-        respond response-        return (rts {forwarded = fwd}, ns)+        lift2 $ respond response+        (lift . put) rts {forwarded = fwd}  handleMessage {- User Request -}-    persistence     (R (Request key request respond))-    (rts@RuntimeState {self, cm, nextId, forwarded}, ns)   = do-    (output, ns2) <- runSM persistence ns (userRequest key request)+    output <- userRequest key request     case output of-      Respond response -> do-        lift (respond response)-        return (rts, ns2)+      Respond response -> lift3 (respond response)       Forward peer -> do-        send cm peer (-            PeerMessage self nextId (ForwardRequest key request)-          )-        return (-            rts {-              forwarded = Map.insert nextId (lift . respond) forwarded,-              nextId = nextMessageId nextId-            },-            ns2-          )+        messageId <- send peer (ForwardRequest key request)+        (lift . modify) $ \rts@RuntimeState {forwarded} -> rts {+            forwarded = Map.insert messageId (lift . respond) forwarded+          }  handleMessage {- Search Dispatch -}     {-       This is where we send out search request to all the appropriate       nodes in the cluster.     -}-    persistence     (R (SearchDispatch searchTag respond))-    (rts@RuntimeState {cm, self, searches}, ns)   =-    case Map.lookup searchTag searches of+    Map.lookup searchTag . searches <$> lift get >>= \case       Nothing -> do         {-           No identical search is currently being executed, kick off a           new one.         -}-        (mcss, ns2) <- runSM persistence ns minimumCompleteServiceSet -        rts2 <- foldr (>=>) return (sendOne <$> Set.toList mcss) rts-        return (-            rts2 {-              searches = Map.insert-                searchTag-                (mcss, Nothing, [lift . respond])-                searches-            },-            ns2-          )-      Just (peers, best, responders) ->+        mcss <- minimumCompleteServiceSet+        mapM_ sendOne (Set.toList mcss)+        rts@RuntimeState {searches} <- lift get+        (lift . put) rts {+            searches = Map.insert+              searchTag+              (mcss, Nothing, [lift . respond])+              searches+          }+      Just (peers, best, responders) -> do         {-           A search for this tag is already in progress, just add the           responder to the responder list.         -}-        return (-            rts {-              searches = Map.insert-                searchTag-                (peers, best, (lift . respond):responders)-                searches-            },-            ns-          )+        rts@RuntimeState {searches} <- lift get+        (lift . put) rts {+            searches = Map.insert+              searchTag+              (peers, best, (lift . respond):responders)+              searches+          }   where-    sendOne :: Peer -> RuntimeState e o s -> LIO (RuntimeState e o s)-    sendOne peer r@RuntimeState {nextId} = do-      send cm peer (PeerMessage self nextId (Search searchTag))-      return r {nextId = nextMessageId nextId}+    sendOne :: Peer -> RTS e o s ()+    sendOne peer =+      void $ send peer (Search searchTag)  handleMessage {- Search Execution -}     {- This is where we handle local search execution. -}-    persistence     (P (PeerMessage source _ (Search searchTag)))-    (rts@RuntimeState {nextId, cm, self}, ns)   = do-    (output, ns2) <- runSM persistence ns (SM.search searchTag) -    send cm source (PeerMessage self nextId (SearchResponse searchTag output))-    return (rts {nextId = nextMessageId nextId}, ns2)+    output <- SM.search searchTag +    void $ send source (SearchResponse searchTag output)  handleMessage {- Search Response -}     {-       This is where we gather all the responses from the various peers       to which we dispatched search requests.     -}-    _legionary     (msg@(P (PeerMessage source _ (SearchResponse searchTag response))))-    (rts@RuntimeState {searches}, ns)   =     {- TODO: see if this function can't be made more elegant. -}-    case Map.lookup searchTag searches of-      Nothing -> do+    Map.lookup searchTag . searches <$> lift get >>= \case+      Nothing ->         {- There is no search happening. -}         $(logWarn) . pack $ "Unsolicited SearchResponse: " ++ show msg-        return (rts, ns)       Just (peers, best, responders) ->         if source `Set.member` peers           then@@ -395,29 +354,24 @@                   All peers have responded, go ahead and respond to                   the client.                 -}-                mapM_ ($ bestOf best response) responders-                return (-                    rts {searches = Map.delete searchTag searches},-                    ns-                  )-              else+                lift2 $ mapM_ ($ bestOf best response) responders+                rts@RuntimeState {searches} <- lift get+                (lift . put) rts {searches = Map.delete searchTag searches}+              else do                 {- We are still waiting on some outstanding requests. -}-                return (-                    rts {-                      searches = Map.insert-                        searchTag-                        (peers2, bestOf best response, responders)-                        searches-                    },-                    ns-                  )-          else do+                rts@RuntimeState {searches} <- lift get+                (lift . put) rts {+                    searches = Map.insert+                      searchTag+                      (peers2, bestOf best response, responders)+                      searches+                  }+          else             {-               There is a search happening, but the peer that responded               is not part of it.             -}             $(logWarn) . pack $ "Unsolicited SearchResponse: " ++ show msg-            return (rts, ns)   where     {- |       Figure out which index record returned to us by the various peers@@ -431,33 +385,23 @@     bestOf a Nothing = a  handleMessage {- Join Request -}-    persistence     (J (JoinRequest addy, respond))-    (rts, ns)   = do-    ((peer, cluster), ns2) <- runSM persistence ns (SM.join addy)-    respond (JoinOk peer cluster)-    return (rts, ns2)+    (peer, cluster) <- SM.join addy+    lift2 $ respond (JoinOk peer cluster)  handleMessage {- Admin Get State -}-    _legionary     (A (GetState respond))-    (rts, ns)-  =-    respond ns >> return (rts, ns)+  = +    lift2 . respond =<< SMM.getNodeState  handleMessage {- Admin Get Partition -}-    persistence     (A (GetPart key respond))-    (rts, ns)-  = do-    respond =<< lift (getState persistence key)-    return (rts, ns)+  =+    lift2 . respond =<< SM.getPartition key  handleMessage {- Admin Eject Peer -}-    persistence     (A (Eject peer respond))-    (rts, ns)   = do     {-       TODO: we should attempt to notify the ejected peer that it has@@ -482,9 +426,8 @@       "next state id" for a peer were global across all power states       instead of local to each power state?     -}-    ((), ns2) <- runSM persistence ns (eject peer)-    respond ()-    return (rts, ns2)+    eject peer+    lift2 $ respond ()   {- | This defines the various ways a node can be spun up. -}@@ -592,9 +535,8 @@       shutdown or crash.     -}     $(logInfo) "Trying to join an existing cluster."-    (self, clusterPS) <- joinCluster (JoinRequest (BSockAddr peerBindAddr))+    (self, cluster) <- joinCluster (JoinRequest (BSockAddr peerBindAddr))     let-      cluster = C.initProp self clusterPS       nodeState = newNodeState self cluster     return (self, nodeState, C.getPeers cluster)   where@@ -814,7 +756,9 @@            cm :: ConnectionManager e o s,      searches :: Map                   SearchTag-                  (Set Peer, Maybe IndexRecord, [Maybe IndexRecord -> LIO ()])+                  (Set Peer, Maybe IndexRecord, [Maybe IndexRecord -> LIO ()]),+         loop :: RuntimeMessage e o s -> IO ()+                 {- ^ A way to send messages back into the message handler. -}   }  @@ -835,5 +779,46 @@ {- | Lookup a key from a map, and also delete the key if it exists. -} lookupDelete :: (Ord k) => k -> Map k v -> (Maybe v, Map k v) lookupDelete = Map.updateLookupWithKey (const (const Nothing))+++{- | The runtime monad.  -}+type RTS e o s =+  SM e o s (+  StateT (RuntimeState e o s)+  LIO)+++{- | Shorthand for running the RTS monad. -}+runRTS+  :: Persistence e o s+  -> NodeState e o s+  -> RuntimeState e o s+  -> RTS e o s a+  -> LIO (a, NodeState e o s, [ClusterAction e o s], RuntimeState e o s)+runRTS persistence ns rts =+    fmap flatten+    . (`runStateT` rts)+    . runSM persistence ns+  where+    flatten ((a, b, c), d) = (a, b, c, d)+++{- |+  Send a peer message in the RTS monad, automatically taking care of+  necessary state updates.+-}+send :: Peer -> PeerMessagePayload e o s -> RTS e o s MessageId+send target payload = do+  rts@RuntimeState {cm, self, nextId} <- lift get+  (lift . put) rts {nextId = nextMessageId nextId}+  lift2 $ CM.send cm target (PeerMessage self nextId payload)+  return nextId+++{- | Forward an existing message to another peer. -}+forward :: Peer -> PeerMessage e o s -> RTS e o s ()+forward target message = do+  RuntimeState {cm} <- lift get+  lift2 $ CM.send cm target message  

@@ -7,6 +7,7 @@   PeerMessage(..),   PeerMessagePayload(..),   MessageId,+  JoinNextResponse(..),   newSequence,   nextMessageId, ) where@@ -19,6 +20,7 @@ import Network.Legion.ClusterState (ClusterPowerState) import Network.Legion.Distribution (Peer) import Network.Legion.Index (SearchTag, IndexRecord)+import Network.Legion.KeySet (KeySet) import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey) import Network.Legion.PartitionState (PartitionPowerState)@@ -52,12 +54,22 @@   | ClusterMerge ClusterPowerState   | Search SearchTag   | SearchResponse SearchTag (Maybe IndexRecord)+  | JoinNext KeySet+  | JoinNextResponse MessageId (JoinNextResponse e o s)   deriving (Generic, Show) instance (Binary e, Binary o, Binary s) => Binary (PeerMessagePayload e o s)   data MessageId = M UUID Word64 deriving (Generic, Show, Eq, Ord) instance Binary MessageId+++{- | The response to a 'JoinNext' message. -}+data JoinNextResponse e o s+  = Joined PartitionKey (PartitionPowerState e o s)+  | JoinFinished+  deriving (Show, Generic)+instance (Binary e, Binary s) => Binary (JoinNextResponse e o s)   {- |

@@ -35,158 +35,107 @@ -} module Network.Legion.StateMachine(   -- * Running the state machine.-  NodeState,   newNodeState,-  SM,-  runSM,    -- * State machine inputs.   userRequest,   partitionMerge,   clusterMerge,-  migrate,-  propagate,-  rebalance,-  heartbeat,   eject,   join,   minimumCompleteServiceSet,   search, +  joinNext,+  joinNextResponse,+   -- * State machine outputs.-  ClusterAction(..),   UserResponse(..),    -- * State inspection   getPeers,+  getPartition, ) where -import Control.Monad (unless)-import Control.Monad.IO.Class (MonadIO)-import Control.Monad.Logger (MonadLogger, logWarn, logDebug, logError)-import Control.Monad.Trans.Class (lift, MonadTrans)-import Control.Monad.Trans.Reader (ReaderT, runReaderT, ask)-import Control.Monad.Trans.State (StateT, runStateT, get, put, modify)-import Data.Aeson (ToJSON, toJSON, object, (.=), encode)-import Data.ByteString.Lazy (toStrict)-import Data.Conduit (($=), ($$), Sink, transPipe, awaitForever)+import Control.Monad (void, unless)+import Control.Monad.Catch (throwM, MonadThrow)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Logger (MonadLogger, logDebug, logError,+  MonadLoggerIO, logWarn)+import Control.Monad.Trans.Class (lift)+import Data.Bool (bool)+import Data.Conduit ((=$=), runConduit, transPipe, awaitForever) import Data.Default.Class (Default) import Data.Map (Map) import Data.Maybe (fromMaybe)-import Data.Set (Set, (\\))-import Data.Text (pack, unpack)-import Data.Text.Encoding (decodeUtf8)-import Data.Time.Clock (getCurrentTime)-import Network.Legion.Application (getState, saveState, list, Persistence)+import Data.Set (Set, (\\), member)+import Data.Text (pack)+import Network.Legion.Application (getState, saveState, list) import Network.Legion.BSockAddr (BSockAddr)-import Network.Legion.ClusterState (ClusterPropState, ClusterPowerState)-import Network.Legion.Distribution (Peer, rebalanceAction, newPeer,-  RebalanceAction(Invite))+import Network.Legion.ClusterState (ClusterPowerState, ClusterPowerStateT)+import Network.Legion.Distribution (Peer, newPeer, RebalanceAction(Invite,+  Drop)) import Network.Legion.Index (IndexRecord(IndexRecord), stTag, stKey,   irTag, irKey, SearchTag(SearchTag), indexEntries, Indexable)-import Network.Legion.KeySet (KeySet, union)-import Network.Legion.LIO (LIO)+import Network.Legion.KeySet (KeySet) import Network.Legion.PartitionKey (PartitionKey)-import Network.Legion.PartitionState (PartitionPowerState, PartitionPropState)-import Network.Legion.PowerState (Event, apply)+import Network.Legion.PartitionState (PartitionPowerState, PartitionPowerStateT)+import Network.Legion.PowerState (Event)+import Network.Legion.PowerState.Monad (PropAction(Send, DoNothing))+import Network.Legion.StateMachine.Monad (SM, NodeState(NodeState),+  ClusterAction(PartitionMerge, ClusterMerge, PartitionJoin),+  self, cluster, partitions, nsIndex, getPersistence, getNodeState,+  modifyNodeState, pushActions, joins, lastRebalance) import qualified Data.Conduit.List as CL import qualified Data.Map as Map import qualified Data.Set as Set import qualified Network.Legion.ClusterState as C import qualified Network.Legion.Distribution as D import qualified Network.Legion.KeySet as KS-import qualified Network.Legion.PartitionState as P---{- |-  This is the portion of the local node state that is not persistence-  related.--}-data NodeState e o s = NodeState {-             self :: Peer,-          cluster :: ClusterPropState,-       partitions :: Map PartitionKey (PartitionPropState e o s),-        migration :: KeySet,-          nsIndex :: Set IndexRecord-  }-instance (Show e, Show s) => Show (NodeState e o s) where-  show = unpack . decodeUtf8 . toStrict . encode-{--  The ToJSON instance is mainly for debugging. The Haskell-generated 'Show'-  instance is very hard to read.--}-instance (Show e, Show s) => ToJSON (NodeState e o s) where-  toJSON (NodeState self cluster partitions migration nsIndex) =-    object [-              "self" .= show self,-           "cluster" .= cluster,-        "partitions" .= Map.mapKeys show partitions,-         "migration" .= show migration,-           "nsIndex" .= show nsIndex-      ]+import qualified Network.Legion.PowerState as PS+import qualified Network.Legion.PowerState.Monad as PM  -{- |-  Make a new node state.--}-newNodeState :: Peer -> ClusterPropState -> NodeState e o s+{- | Make a new node state. -}+newNodeState :: Peer -> ClusterPowerState -> NodeState e o s newNodeState self cluster =   NodeState {       self,       cluster,       partitions = Map.empty,-      migration = KS.empty,-      nsIndex = Set.empty+      nsIndex = Set.empty,+      joins = Map.empty,+      lastRebalance = minBound     }  -{- |-  This monad encapsulates the global state of the legion node (not-  counting the runtime stuff, like open connections and what have-  you).--  The main reason that the state is hidden behind a monad is because part-  of the sate (i.e. the partition data) lives behind 'IO'.  Therefore,-  if we want to model the global state of the node as a single unit,-  we have to do so using a monad.--}-newtype SM e o s a = SM {-    unSM :: ReaderT (Persistence e o s) (StateT (NodeState e o s) LIO) a-  }-  deriving (Functor, Applicative, Monad, MonadLogger, MonadIO)---{- |-  Run an SM action.--}-runSM-  :: Persistence e o s-  -> NodeState e o s-  -> SM e o s a-  -> LIO (a, NodeState e o s)-runSM p ns action = runStateT (runReaderT (unSM action) p) ns-- {- | Handle a user request. -}-userRequest :: (Event e o s, Default s, Indexable s)+userRequest :: (+      Default s,+      Eq e,+      Event e o s,+      Indexable s,+      MonadLoggerIO m,+      MonadThrow m,+      Show e,+      Show s+    )   => PartitionKey   -> e-  -> SM e o s (UserResponse o)-userRequest key request = SM $ do-  NodeState {self, cluster} <- lift get-  let owners = C.findPartition key cluster-  if self `Set.member` owners+  -> SM e o s m (UserResponse o)+userRequest key request = do+  NodeState {self, cluster} <- getNodeState+  let routes = C.findRoute key cluster+  if self `Set.member` routes     then do-      partition <- unSM $ getPartition key-      let-        response = fst (apply request (P.ask partition))-        partition2 = P.event request partition-      unSM $ savePartition key partition2+      (response, _) <- runPartitionPowerStateT key (+          PM.event request+        )       return (Respond response) -    else case Set.toList owners of+    else case Set.toList routes of       [] -> do-        let msg = "No owners for key: " ++ show key+        let msg = "No routes for key: " ++ show key         $(logError) . pack $ msg         error msg       peer:_ -> return (Forward peer)@@ -196,172 +145,99 @@   Handle the state transition for a partition merge event. Returns 'Left'   if there is an error, and 'Right' if everything went fine. -}-partitionMerge :: (Show e, Show s, Event e o s, Default s, Indexable s)-  => Peer-  -> PartitionKey+partitionMerge :: (+      Default s,+      Eq e,+      Event e o s,+      Indexable s,+      MonadLoggerIO m,+      MonadThrow m,+      Show e,+      Show s+    )+  => PartitionKey   -> PartitionPowerState e o s-  -> SM e o s ()-partitionMerge source key foreignPartition = do-  partition <- getPartition key-  case P.mergeEither source foreignPartition partition of-    Left err -> $(logWarn) . pack-      $ "Can't apply incomming partition merge from "-      ++ show source ++ ": " ++ show foreignPartition-      ++ ". because of: " ++ show err-    Right newPartition -> savePartition key newPartition+  -> SM e o s m ()+partitionMerge key foreignPartition =+  void $ runPartitionPowerStateT key (PM.merge foreignPartition)   {- | Handle the state transition for a cluster merge event. -}-clusterMerge-  :: Peer-  -> ClusterPowerState-  -> SM e o s ()-clusterMerge source foreignCluster = SM . lift $ do-  nodeState@NodeState {migration, cluster} <- get-  case C.mergeEither source foreignCluster cluster of-    Left err -> $(logWarn) . pack-      $ "Can't apply incomming cluster merge from "-      ++ show source ++ ": " ++ show foreignCluster-      ++ ". because of: " ++ show err-    Right (newCluster, newMigration) ->-      put nodeState {-          migration = migration `union` newMigration,-          cluster = newCluster-        }---{- |-  Migrate partitions based on new cluster state information.--  TODO: this migration algorithm is super naive. It just goes ahead-  and migrates everything in one pass, which is going to be terrible-  for performance.--  Also, it is important to remember that "migrate" in this context does-  not mean "transfer data". Rather, "migrate" means to add a participating-  peer to a partition. This will cause the data to be transfered in the-  normal course of propagation.--}-migrate :: (Default s, Event e o s, Indexable s) => SM e o s ()-migrate = do-    NodeState {migration} <- (SM . lift) get-    persistence <- SM ask-    unless (KS.null migration) $-      transPipe (SM . lift3) (list persistence)-      $= CL.filter ((`KS.member` migration) . fst)-      $$ accum-    (SM . lift) $ modify (\ns -> ns {migration = KS.empty})-  where-    accum :: (Default s, Event e o s, Indexable s)-      => Sink (PartitionKey, PartitionPowerState e o s) (SM e o s) ()-    accum = awaitForever $ \ (key, ps) -> do-      NodeState {self, cluster, partitions} <- (lift . SM . lift) get-      let-        partition = fromMaybe (P.initProp self ps) (Map.lookup key partitions)-        newPeers = C.findPartition key cluster \\ P.projParticipants partition-        newPartition = foldr P.participate partition (Set.toList newPeers)-      $(logDebug) . pack $ "Migrating: " ++ show key-      lift (savePartition key newPartition)---{- |-  Handle all cluster and partition state propagation actions, and return-  an updated node state.--}-propagate :: SM e o s [ClusterAction e o s]-propagate = SM $ do-    partitionActions <- getPartitionActions-    clusterActions <- unSM getClusterActions-    return (clusterActions ++ partitionActions)-  where-    getPartitionActions = do-      ns@NodeState {partitions} <- lift get-      let-        updates = [-            (key, newPartition, [-                PartitionMerge peer key ps-                | peer <- Set.toList peers_-              ])-            | (key, partition) <- Map.toAscList partitions-            , let (peers_, ps, newPartition) = P.actions partition-          ]-        actions = [a | (_, _, as) <- updates, a <- as]-        newPartitions = Map.fromAscList [-            (key, newPartition)-            | (key, newPartition, _) <- updates-            , not (P.idle newPartition)-          ]-      (lift . put) ns {-          partitions = newPartitions-        }-      return actions+clusterMerge :: (+      Default s,+      Eq e,+      Event e o s,+      Indexable s,+      MonadLoggerIO m,+      MonadThrow m,+      Show e,+      Show s+    )+  => ClusterPowerState+  -> SM e o s m ()+clusterMerge foreignCluster = do+  runClusterPowerStateT (PM.merge foreignCluster)+  nodeState@NodeState {lastRebalance, cluster, self} <- getNodeState+  $(logDebug) . pack+    $ "Next Rebalance: "+    ++ show (lastRebalance, C.nextAction cluster, nodeState)+  case C.nextAction cluster of+    (ord, Invite peer keys) | ord > lastRebalance && peer == self -> do+      {-+        The current action is an Invite, and this peer is the target. -    getClusterActions :: SM e o s [ClusterAction e o s]-    getClusterActions = SM $ do-      ns@NodeState {cluster} <- lift get+        Send the join request message to every peer, update lastRebalance+        so we don't repeat this on every trivial cluster merge, update+        the expected joins so we can keep track of progress, then sit+        back and wait.+      -}       let-        (peers, cs, newCluster) = C.actions cluster-        actions = [ClusterMerge peer cs | peer <- Set.toList peers]-      (lift . put) ns {-          cluster = newCluster-        }-      return actions---{- |-  Figure out if any rebalancing actions must be taken by this node, and kick-  them off if so.--}-rebalance :: SM e o s ()-rebalance = SM $ do-  ns@NodeState {self, cluster} <- lift get-  let-    allPeers = (Set.fromList . Map.keys . C.getPeers) cluster-    dist = C.getDistribution cluster-    action = rebalanceAction self allPeers dist-  $(logDebug) . pack $ "The rebalance action is: " ++ show action-  (lift . put) ns {-      cluster = case action of-        Nothing -> cluster-        Just (Invite ks) ->-          {--            This 'claimParticipation' will be enforced by the remote-            peers, because those peers will see the change in distribution-            and then perform a 'migrate'.-          -}-          C.claimParticipation self ks cluster-    }---{- | Update all of the propagation states with the current time.  -}-heartbeat :: SM e o s ()-heartbeat = SM $ do-  now <- lift3 getCurrentTime-  ns@NodeState {cluster, partitions} <- lift get-  (lift . put) ns {-      cluster = C.heartbeat now cluster,-      partitions = Map.fromAscList [-          (k, P.heartbeat now p)-          | (k, p) <- Map.toAscList partitions+        askPeers =+          Set.toList . Set.delete self . Map.keysSet . C.getPeers $ cluster+      pushActions [+          PartitionJoin p keys+          | p <- askPeers         ]-    }+      modifyNodeState (\ns -> ns {+          joins = Map.fromList [+              (p, keys)+              | p <- askPeers+            ],+          lastRebalance = ord+        })+    (ord, Drop peer keys) | ord > lastRebalance && peer == self -> do+      persistence <- getPersistence+      runConduit (+          transPipe liftIO (list persistence)+          =$= CL.map fst+          =$= CL.filter (`KS.member` keys)+          =$= awaitForever (\key ->+              lift $ runPartitionPowerStateT key (+                  PM.disassociate self+                )+            )+        )+      modifyNodeState (\ns -> ns {+          lastRebalance = ord+        })+      runClusterPowerStateT C.finishRebalance+    _ -> return ()   {- | Eject a peer from the cluster.  -}-eject :: Peer -> SM e o s ()-eject peer = SM . lift $ do-  ns@NodeState {cluster} <- get-  put ns {cluster = C.eject peer cluster}+eject :: (MonadLogger m, MonadThrow m) => Peer -> SM e o s m ()+eject peer = runClusterPowerStateT (C.eject peer)   {- | Handle a peer join request.  -}-join :: BSockAddr -> SM e o s (Peer, ClusterPowerState)-join peerAddr = SM $ do-  peer <- lift2 newPeer-  ns@NodeState {cluster} <- lift get-  let newCluster = C.joinCluster peer peerAddr cluster-  (lift . put) ns {cluster = newCluster}-  return (peer, C.getPowerState newCluster)+join :: (MonadIO m, MonadThrow m)+  => BSockAddr+  -> SM e o s m (Peer, ClusterPowerState)+join peerAddr = do+  peer <- newPeer+  void $ runClusterPowerStateT (C.joinCluster peer peerAddr)+  NodeState {cluster} <- getNodeState+  return (peer, cluster)   {- |@@ -387,9 +263,9 @@    TODO: implement fastest competitive search. -}-minimumCompleteServiceSet :: SM e o s (Set Peer)-minimumCompleteServiceSet = SM $ do-  NodeState {cluster} <- lift get+minimumCompleteServiceSet :: (Monad m) => SM e o s m (Set Peer)+minimumCompleteServiceSet = do+  NodeState {cluster} <- getNodeState   return (D.minimumCompleteServiceSet (C.getDistribution cluster))  @@ -397,25 +273,125 @@   Search the index, and return the first record that is __strictly   greater than__ the provided search tag, if such a record exists. -}-search :: SearchTag -> SM e o s (Maybe IndexRecord)-search SearchTag {stTag, stKey = Nothing} = SM $ do-  NodeState {nsIndex} <- lift get+search :: (Monad m) => SearchTag -> SM e o s m (Maybe IndexRecord)+search SearchTag {stTag, stKey = Nothing} = do+  NodeState {nsIndex} <- getNodeState   return (Set.lookupGE IndexRecord {irTag = stTag, irKey = minBound} nsIndex)-search SearchTag {stTag, stKey = Just key} = SM $ do-  NodeState {nsIndex} <- lift get+search SearchTag {stTag, stKey = Just key} = do+  NodeState {nsIndex} <- getNodeState   return (Set.lookupGT IndexRecord {irTag = stTag, irKey = key} nsIndex)   {- |-  These are the actions that a node can take which allow it to coordinate-  with other nodes. It is up to the runtime system to implement the-  actions.+  Allow a peer to participate in the replication of the partition that is+  __greater than or equal to__ the indicated partition key. Returns @Nothing@+  if there is no such partition, or @Just (key, partition)@ where @key@ is the+  partition key that was joined and @partition@ is the resulting partition+  power state. -}-data ClusterAction e o s-  = ClusterMerge Peer ClusterPowerState-  | PartitionMerge Peer PartitionKey (PartitionPowerState e o s)+joinNext :: (+      Default s,+      Eq e,+      Event e o s,+      Indexable s,+      MonadLoggerIO m,+      MonadThrow m+    )+  => Peer+  -> KeySet+  -> SM e o s m (Maybe (PartitionKey, PartitionPowerState e o s))+joinNext peer askKeys = do+  persistence <- getPersistence+  runConduit (+      transPipe liftIO (list persistence)+      =$= CL.filter ((`KS.member` askKeys) . fst)+      =$= CL.head+    ) >>= \case+      Nothing -> return Nothing+      Just (gotKey, partition) -> do+        {-+          This is very similar to the 'runPartitionPowerStateT' code,+          but there are some important differences. First, 'list' has+          already done to the trouble of fetching the partition value,+          so we don't want to have 'runPartitionPowerStateT' do it+          again. Second, and more importantly, 'runPartitionPowerStateT'+          will cause a 'PartitionMerge' message to be sent to @peer@, but+          that message would be redundant, because it contains a subset+          of the information contained within the 'JoinNextResponse'+          message that this function produces.+        -}+        NodeState {self} <- getNodeState+        PM.runPowerStateT self partition (do+            PM.participate peer+            PM.acknowledge+          ) >>= \case+            Left err -> throwM err+            Right ((), action, partition2, _infOutputs) -> do+              case action of+                Send -> pushActions [+                    PartitionMerge p gotKey partition2+                    | p <- Set.toList (PS.allParticipants partition2)+                      {-+                        Don't send a 'PartitionMerge' to @peer@. We+                        are already going to send it a more informative+                        'JoinNextResponse'+                      -}+                    , p /= peer+                    , p /= self+                  ]+                DoNothing -> return ()+              savePartition gotKey partition2+              return (Just (gotKey, partition2))  +{- | Receive the result of a JoinNext request. -}+joinNextResponse :: (+      Default s,+      Eq e,+      Event e o s,+      Indexable s,+      MonadLoggerIO m,+      MonadThrow m,+      Show e,+      Show s+    )+  => Peer+  -> Maybe (PartitionKey, PartitionPowerState e o s)+  -> SM e o s m ()+joinNextResponse peer response = do+  NodeState {cluster, lastRebalance} <- getNodeState+  if lastRebalance > fst (C.nextAction cluster)+    then+      {- We are receiving messages from an old rebalance. Log and ignore. -}+      $(logWarn) . pack+        $ "Received an old join response: "+        ++ show (peer, response, cluster, lastRebalance)+    else do+      case response of+        Just (key, partition) -> do+          partitionMerge key partition+          NodeState {joins} <- getNodeState+          case (KS.\\ KS.fromRange minBound key) <$> Map.lookup peer joins of+            Nothing ->+              {- An unexpected peer sent us this message, Ignore. TODO log. -}+              return ()+            Just needsJoinSet -> do+              unless (KS.null needsJoinSet)+                (pushActions [PartitionJoin peer needsJoinSet])+              modifyNodeState (\ns -> ns {+                  joins = Map.filter+                    (not . KS.null)+                    (Map.insert peer needsJoinSet joins)+                })+        Nothing ->+          modifyNodeState (\ns@NodeState {joins} -> ns {+              joins = Map.delete peer joins+            })+      Map.null . joins <$> getNodeState >>= bool+        (return ())+        (runClusterPowerStateT C.finishRebalance)++ {- |   The type of response to a user request, either forward to another node,   or respond directly.@@ -426,22 +402,21 @@   {- | Get the known peer data from the cluster. -}-getPeers :: SM e o s (Map Peer BSockAddr)-getPeers = SM $ C.getPeers . cluster <$> lift get+getPeers :: (Monad m) => SM e o s m (Map Peer BSockAddr)+getPeers = C.getPeers . cluster <$> getNodeState   {- | Gets a partition state. -}-getPartition :: (Default s, Event e o s)+getPartition :: (Default s, MonadIO m)   => PartitionKey-  -> SM e o s (PartitionPropState e o s)-getPartition key = SM $ do-  persistence <- ask-  NodeState {self, partitions, cluster} <- lift get+  -> SM e o s m (PartitionPowerState e o s)+getPartition key = do+  persistence <- getPersistence+  NodeState {partitions, cluster} <- getNodeState   case Map.lookup key partitions of     Nothing ->-      lift3 (getState persistence key) <&> \case-        Nothing -> P.new key self (C.findPartition key cluster)-        Just partition -> P.initProp self partition+      fromMaybe (PS.new key (C.findOwners key cluster)) <$>+        liftIO (getState persistence key)     Just partition -> return partition  @@ -449,15 +424,15 @@   Saves a partition state. This function automatically handles the cache   for active propagations, as well as reindexing of partitions. -}-savePartition :: (Default s, Event e o s, Indexable s)+savePartition :: (Default s, Event e o s, Indexable s, MonadLoggerIO m)   => PartitionKey-  -> PartitionPropState e o s-  -> SM e o s ()-savePartition key partition = SM $ do-  persistence <- ask-  oldTags <- indexEntries . P.ask <$> unSM (getPartition key)+  -> PartitionPowerState e o s+  -> SM e o s m ()+savePartition key partition = do+  persistence <- getPersistence+  oldTags <- indexEntries . PS.projectedValue <$> getPartition key   let-    currentTags = indexEntries (P.ask partition)+    currentTags = indexEntries (PS.projectedValue partition)     {- TODO: maybe use Set.mapMonotonic for performance?  -}     obsoleteRecords = Set.map (flip IndexRecord key) (oldTags \\ currentTags)     newRecords = Set.map (flip IndexRecord key) currentTags@@ -466,57 +441,93 @@     $ "Tagging " ++ show key ++ " with: "     ++ show (currentTags, obsoleteRecords, newRecords) -  ns@NodeState {partitions, nsIndex} <- lift get-  lift3 (saveState persistence key (-      if P.participating partition-        then Just (P.getPowerState partition)+  NodeState {self} <- getNodeState+  liftIO (saveState persistence key (+      if self `member` PS.allParticipants partition+        then Just partition         else Nothing     ))-  lift $ put ns {-      partitions = if P.idle partition-        then-          {--            Remove the partition from the working cache because there-            is no remaining work that needs to be done to propagage-            its changes.-          -}-          Map.delete key partitions-        else-          Map.insert key partition partitions,-      nsIndex = (nsIndex \\ obsoleteRecords) `Set.union` newRecords-    }+  modifyNodeState (\ns@NodeState {partitions, nsIndex} ->+      ns {+          partitions = if Set.null (PS.divergent partition)+            then+              {-+                Remove the partition from the working cache because there+                is no remaining work that needs to be done to propagage+                its changes.+              -}+              Map.delete key partitions+            else+              Map.insert key partition partitions,+          nsIndex = (nsIndex \\ obsoleteRecords) `Set.union` newRecords+        }+    )  -{- | Borrowed from 'lens', like @flip fmap@. -}-(<&>) :: (Functor f) => f a -> (a -> b) -> f b-(<&>) = flip fmap+-- {- |+--   Create the log message for origin conflict errors.  The reason this+--   function only creates the log message, instead of doing the logging+--   as well, is because doing the logging here would screw up the source+--   location that the template-haskell logging functions generate for us.+-- -}+-- originError :: (Show o) => DifferentOrigins o -> Text+-- originError (DifferentOrigins a b) = pack+--   $ "Tried to merge powerstates with different origins: "+--   ++ show (a, b)  -{- | Lift from two levels down in a monad transformation stack. -}-lift2-  :: (-      MonadTrans a,-      MonadTrans b,-      Monad m,-      Monad (b m)+{- | Run a partition-flavored 'PowerStateT' in the 'SM' monad. -}+runPartitionPowerStateT :: (+      Default s,+      Eq e,+      Event e o s,+      Indexable s,+      MonadLoggerIO m,+      MonadThrow m,+      Show e,+      Show s     )-  => m r-  -> a (b m) r-lift2 = lift . lift+  => PartitionKey+  -> PartitionPowerStateT e o s (SM e o s m) a+  -> SM e o s m (a, PartitionPowerState e o s)+runPartitionPowerStateT key m = do+  NodeState {self} <- getNodeState+  partition <- getPartition key+  PM.runPowerStateT self partition (m <* PM.acknowledge) >>= \case+    Left err -> throwM err+    Right (a, action, partition2, _infOutputs) -> do+      case action of+        Send -> pushActions [+            PartitionMerge p key partition2+            | p <- Set.toList (PS.allParticipants partition2)+          ]+        DoNothing -> return ()+      $(logDebug) . pack+        $ "Partition update: " ++ show partition+        ++ " --> " ++ show partition2 ++ " :: " ++ show action+      savePartition key partition2+      return (a, partition2)  -{- | Lift from three levels down in a monad transformation stack. -}-lift3-  :: (-      MonadTrans a,-      MonadTrans b,-      MonadTrans c,-      Monad m,-      Monad (c m),-      Monad (b (c m))-    )-  => m r-  -> a (b (c m)) r-lift3 = lift . lift . lift+{- |+  Run a clusterstate-flavored 'PowerStateT' in the 'SM' monad,+  automatically acknowledging the resulting power state.+-}+runClusterPowerStateT :: (MonadThrow m)+  => ClusterPowerStateT (SM e o s m) a+  -> SM e o s m a+runClusterPowerStateT m = do+  NodeState {cluster, self} <- getNodeState+  PM.runPowerStateT self cluster (m <* PM.acknowledge) >>= \case+    Left err -> throwM err+    Right (a, action, cluster2, _outputs) -> do+      case action of+        Send -> pushActions [+            ClusterMerge p cluster2+            | p <- Set.toList (PS.allParticipants cluster2)+          ]+        DoNothing -> return ()+      modifyNodeState (\ns -> ns {cluster = cluster2})+      return a  

@@ -0,0 +1,165 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-}+{- |+  This module contains the legion state machine monad and some+  primitives for manipulating the state. It is the foundation upon wish+  the 'Network.Legion.StateMachine' module is built. It is separate from+  that module because some of the primitives we export here go some small+  way to avoiding bugs that might arise if that module had direct access+  to the internals of this monad.+-}+module Network.Legion.StateMachine.Monad (+  -- * Run the monad+  runSM,++  -- * State Inspection+  getPersistence,+  getNodeState,++  -- * State Modification+  modifyNodeState,+  pushActions,+  popActions,++  -- * Other symbols+  SM,+  NodeState(..),+  ClusterAction(..),+) where++import Control.Monad.Catch (MonadThrow)+import Control.Monad.IO.Class (MonadIO)+import Control.Monad.Logger (MonadLogger)+import Control.Monad.Trans.Class (lift, MonadTrans)+import Control.Monad.Trans.Reader (ReaderT, runReaderT, ask)+import Control.Monad.Trans.State (StateT, runStateT, get, modify, put)+import Data.Aeson (ToJSON, toJSON, object, (.=), encode)+import Data.ByteString.Lazy (toStrict)+import Data.Map (Map)+import Data.Set (Set)+import Data.Text (unpack)+import Data.Text.Encoding (decodeUtf8)+import Network.Legion.Application (Persistence)+import Network.Legion.ClusterState (ClusterPowerState, RebalanceOrd)+import Network.Legion.Distribution (Peer)+import Network.Legion.Index (IndexRecord)+import Network.Legion.KeySet (KeySet)+import Network.Legion.Lift (lift2, lift3)+import Network.Legion.PartitionKey (PartitionKey)+import Network.Legion.PartitionState (PartitionPowerState)+import qualified Data.Map as Map+++{- |+  Run an SM action.+-}+runSM :: (Functor m)+  => Persistence e o s+  -> NodeState e o s+  -> SM e o s m a+  -> m (a, NodeState e o s, [ClusterAction e o s])+runSM p ns =+    fmap flatten+    . (`runStateT` [])+    . (`runStateT` ns)+    . (`runReaderT` p)+    . unSM+  where+    flatten :: ((a, b), c) -> (a, b, c)+    flatten ((a, b), c) = (a, b, c)+++{- | Get the handle to the persistence layer. -}+getPersistence :: (Monad m) => SM e o s m (Persistence e o s)+getPersistence = SM ask+++{- | Get the current node state. -}+getNodeState :: (Monad m) => SM e o s m (NodeState e o s)+getNodeState = (SM . lift) get+++{- | Update current node state. -}+modifyNodeState :: (Monad m)+  => (NodeState e o s -> NodeState e o s)+  -> SM e o s m ()+modifyNodeState = SM . lift . modify+++{- | Accumulate some cluster propagation actions. -}+pushActions :: (Monad m) => [ClusterAction e o s] -> SM e o s m ()+pushActions = SM . lift2 . modify . flip (++)+++{- | Return and reset the accumulated cluster actions. -}+popActions :: (Monad m) => SM e o s m [ClusterAction e o s]+popActions = SM . lift2 $ do+  actions <- get+  put []+  return actions+++{- |+  This monad encapsulates the global state of the legion node (not+  counting the runtime stuff, like open connections and what have+  you).++  The main reason that the state is hidden behind a monad is because part+  of the sate (i.e. the partition data) lives behind 'IO'.  Therefore,+  if we want to model the global state of the node as a single unit,+  we have to do so using a monad.+-}+newtype SM e o s m a = SM {+    unSM ::+      ReaderT (Persistence e o s) (+      StateT (NodeState e o s) (+      StateT [ClusterAction e o s]+      m)) a+  }+  deriving (Functor, Applicative, Monad, MonadLogger, MonadIO, MonadThrow)+instance MonadTrans (SM e o s) where+  lift = SM . lift3+++{- |+  This is the portion of the local node state that is not persistence+  related.+-}+data NodeState e o s = NodeState {+             self :: Peer,+          cluster :: ClusterPowerState,+       partitions :: Map PartitionKey (PartitionPowerState e o s),+          nsIndex :: Set IndexRecord,+            joins :: Map Peer KeySet,+    lastRebalance :: RebalanceOrd+  }+instance (Show e, Show s) => Show (NodeState e o s) where+  show = unpack . decodeUtf8 . toStrict . encode+{-+  The ToJSON instance is mainly for debugging. The Haskell-generated 'Show'+  instance is very hard to read.+-}+instance (Show e, Show s) => ToJSON (NodeState e o s) where+  toJSON (NodeState self_ cluster_ partitions_ nsIndex_ joins_ lastUpdate_) =+    object [+                 "self" .= show self_,+              "cluster" .= cluster_,+           "partitions" .= Map.map show (Map.mapKeys show partitions_),+              "nsIndex" .= show nsIndex_,+                "joins" .= Map.map show (Map.mapKeys show joins_),+        "lastRebalance" .= show lastUpdate_+      ]+++{- |+  These are the actions that a node can take which allow it to coordinate+  with other nodes. It is up to the runtime system to implement the+  actions.+-}+data ClusterAction e o s+  = PartitionMerge Peer PartitionKey (PartitionPowerState e o s)+  | ClusterMerge Peer ClusterPowerState+  | PartitionJoin Peer KeySet+  deriving (Show)++