diff --git a/legion.cabal b/legion.cabal
--- a/legion.cabal
+++ b/legion.cabal
@@ -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,
diff --git a/src/Network/Legion/Admin.hs b/src/Network/Legion/Admin.hs
--- a/src/Network/Legion/Admin.hs
+++ b/src/Network/Legion/Admin.hs
@@ -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
diff --git a/src/Network/Legion/Application.hs b/src/Network/Legion/Application.hs
--- a/src/Network/Legion/Application.hs
+++ b/src/Network/Legion/Application.hs
@@ -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.
+                 -}
   }
 
 
diff --git a/src/Network/Legion/ClusterState.hs b/src/Network/Legion/ClusterState.hs
--- a/src/Network/Legion/ClusterState.hs
+++ b/src/Network/Legion/ClusterState.hs
@@ -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))
 
diff --git a/src/Network/Legion/Conduit.hs b/src/Network/Legion/Conduit.hs
--- a/src/Network/Legion/Conduit.hs
+++ b/src/Network/Legion/Conduit.hs
@@ -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
 
 
diff --git a/src/Network/Legion/Distribution.hs b/src/Network/Legion/Distribution.hs
--- a/src/Network/Legion/Distribution.hs
+++ b/src/Network/Legion/Distribution.hs
@@ -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
 
 
diff --git a/src/Network/Legion/KeySet.hs b/src/Network/Legion/KeySet.hs
--- a/src/Network/Legion/KeySet.hs
+++ b/src/Network/Legion/KeySet.hs
@@ -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))
 
 
diff --git a/src/Network/Legion/Lift.hs b/src/Network/Legion/Lift.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Legion/Lift.hs
@@ -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
+
+
diff --git a/src/Network/Legion/PartitionKey.hs b/src/Network/Legion/PartitionKey.hs
--- a/src/Network/Legion/PartitionKey.hs
+++ b/src/Network/Legion/PartitionKey.hs
@@ -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)
diff --git a/src/Network/Legion/PartitionState.hs b/src/Network/Legion/PartitionState.hs
--- a/src/Network/Legion/PartitionState.hs
+++ b/src/Network/Legion/PartitionState.hs
@@ -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
 
 
diff --git a/src/Network/Legion/PowerState.hs b/src/Network/Legion/PowerState.hs
--- a/src/Network/Legion/PowerState.hs
+++ b/src/Network/Legion/PowerState.hs
@@ -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)
 
 
 {- |
diff --git a/src/Network/Legion/PowerState/Monad.hs b/src/Network/Legion/PowerState/Monad.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Legion/PowerState/Monad.hs
@@ -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)
+
+
diff --git a/src/Network/Legion/Propagation.hs b/src/Network/Legion/Propagation.hs
deleted file mode 100644
--- a/src/Network/Legion/Propagation.hs
+++ /dev/null
@@ -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)
-
-
diff --git a/src/Network/Legion/Runtime.hs b/src/Network/Legion/Runtime.hs
--- a/src/Network/Legion/Runtime.hs
+++ b/src/Network/Legion/Runtime.hs
@@ -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
 
 
diff --git a/src/Network/Legion/Runtime/PeerMessage.hs b/src/Network/Legion/Runtime/PeerMessage.hs
--- a/src/Network/Legion/Runtime/PeerMessage.hs
+++ b/src/Network/Legion/Runtime/PeerMessage.hs
@@ -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)
 
 
 {- |
diff --git a/src/Network/Legion/StateMachine.hs b/src/Network/Legion/StateMachine.hs
--- a/src/Network/Legion/StateMachine.hs
+++ b/src/Network/Legion/StateMachine.hs
@@ -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
 
 
diff --git a/src/Network/Legion/StateMachine/Monad.hs b/src/Network/Legion/StateMachine/Monad.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Legion/StateMachine/Monad.hs
@@ -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)
+
+