crdt-event-fold (empty) → 1.0.0.1
raw patch · 5 files changed
+1049/−0 lines, 5 filesdep +basedep +binarydep +containerssetup-changed
Dependencies added: base, binary, containers, data-default-class, data-dword
Files
- LICENSE +19/−0
- README.md +4/−0
- Setup.hs +2/−0
- crdt-event-fold.cabal +33/−0
- src/Data/CRDT/EventFold.hs +991/−0
+ LICENSE view
@@ -0,0 +1,19 @@+Copyright 2020 Owens Murray, LLC.++Permission is hereby granted, free of charge, to any person obtaining a+copy of this software and associated documentation files (the "Software"),+to deal in the Software without restriction, including without limitation+the rights to use, copy, modify, merge, publish, distribute, sublicense,+and/or sell copies of the Software, and to permit persons to whom the+Software is furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL+THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR+OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,+ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR+OTHER DEALINGS IN THE SOFTWARE.
+ README.md view
@@ -0,0 +1,4 @@+# crdt-event-fold++A Haskell library providing garbage collected event folding CRDT.+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ crdt-event-fold.cabal view
@@ -0,0 +1,33 @@+-- Initial crdt-event-fold.cabal generated by cabal init. For further +-- documentation, see http://haskell.org/cabal/users-guide/++name: crdt-event-fold+version: 1.0.0.1+synopsis: Garbage collected event folding CRDT.+description: Garbage collected event folding CRDT. Consistently+ apply arbitrary operations to replicated data.+homepage: https://github.com/owensmurray/crdt-event-fold+license: MIT+license-file: LICENSE+author: Rick Owens+maintainer: rick@owensmurray.com+copyright: 2020 Owens Murray, LLC.+category: CRDT+build-type: Simple+extra-source-files: README.md+cabal-version: >=1.10++library+ exposed-modules: + Data.CRDT.EventFold+ -- other-modules: + -- other-extensions: + build-depends:+ base >= 4.13 && < 4.14,+ binary >= 0.8.7.0 && < 0.9,+ containers >= 0.6.2.1 && < 0.7,+ data-default-class >= 0.1.2.0 && < 0.2,+ data-dword >= 0.3.2 && < 0.4+ hs-source-dirs: src+ default-language: Haskell2010+
+ src/Data/CRDT/EventFold.hs view
@@ -0,0 +1,991 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}+{-# OPTIONS_GHC -Wmissing-deriving-strategies #-}++{- |+ Description: Garbage collected event folding CRDT.++ This module provides a CRDT data structure that collects and applies+ operations (called "events") that mutate an underlying data structure+ (like folding).++ In addition to mutating the underlying data, each operation can also+ produce an output that can be obtained by the client. The output can be+ either totally consistent across all replicas (which is slower), or it+ can be returned immediately and possibly reflect an inconsistent state.+-}+module Data.CRDT.EventFold (+ -- * Basic API+ -- ** Creating new CRDTs.+ new,++ -- ** Adding new events.+ event,++ -- ** Coordinating replica updates.+ {- |+ Functions in this section are used to help merge foreign copies of+ the CRDT, and transmit our own copy. (This library does not provide+ any kind of transport support, except that all the relevant types+ have 'Binary' instances. Actually arranging for these things to get+ shipped across a wire is left to the user.)++ In principal, the only two functions you need are 'fullMerge' and+ 'acknowledge'. You can ship the full 'EventFold' value to a remote+ participant and it can incorporate any changes using 'fullMerge',+ and vice versa. You can receive an 'EventFold' value from another+ participant and incorporate its changes locally using 'fullMerge'. You+ can then acknowledge the incorporation using 'acknowledge'.++ However, if your underlying data structure is large, it may be more+ efficient to just ship a sort of diff containing the information+ that the local participant thinks the remote participant might be+ missing. That is what 'events', 'mergeMaybe', and 'mergeEither'+ are for.++ Calling 'acknowledge' is important because that is the magic that+ allows CRDT garbage collection to happen. "CRDT garbage collection"+ means we don't store an infinite series of events that always grows+ and never shrinks. We only store the outstanding events that we+ can't prove have been seen by every participant. Events that we /can/+ prove have been seen by every participant are applied to the infimum+ (a.k.a. "base value") and the event itself is discarded.+ + -}+ fullMerge,+ acknowledge,+ events,+ mergeMaybe,+ mergeEither,+ MergeError(..),++ -- ** Participation.+ participate,+ disassociate,++ -- ** Defining your state and events.+ Event(..),+ EventResult(..),++ -- * Inspecting the 'EventFold'.+ isBlockedOnError,+ projectedValue,+ infimumValue,+ infimumId,+ infimumParticipants,+ allParticipants,+ projParticipants,+ origin,+ divergent,++ -- * Underlying Types+ EventFoldF,+ EventFold,+ EventId,+ EventPack,++) where+++import Data.Bifunctor (first)+import Data.Binary (Binary(get, put))+import Data.Default.Class (Default(def))+import Data.DoubleWord (Word128(Word128), Word256(Word256))+import Data.Functor.Identity (Identity(Identity), runIdentity)+import Data.Map (Map, keys, toAscList, toDescList, unionWith)+import Data.Maybe (catMaybes)+import Data.Set ((\\), Set, member, union)+import Data.Word (Word64)+import GHC.Generics (Generic)+import qualified Data.Map as Map+import qualified Data.Map.Merge.Lazy as Map.Merge+import qualified Data.Set as Set+++{- |+ This represents a replicated data structure into which participants can+ add 'Event's that are folded into a base 'State'. You can also think+ of the "events" as operations that mutate the base state, and the point+ of this CRDT is to coordinate the application of the operations across+ all participants so that they are applied consistently even if the+ operations themselves are not commutative, idempotent, or monotonic.+ Those properties to the CRDT by the way in which it manages the events,+ and it is therefore unnecessary that the events themselves have them.++ Variables are:++ - @o@ - Origin+ - @p@ - Participant+ - @e@ - Event+ - @f@ - The Monad in which the events live++ The "Origin" is a value that is more or less meant to identify the+ "thing" being replicated, and in particular identify the historical+ lineage of the 'EventFold'. The idea is that it is meaningless to+ try and merge two 'EventFold's that do not share a common history+ (identified by the origin value) and doing so is a programming error. It+ is only used to try and check for this type of programming error and+ throw an exception if it happens instead of producing undefined (and+ difficult to detect) behavior.+-}+data EventFoldF o p e f = EventFold {+ psOrigin :: o,+ psInfimum :: Infimum (State e) p,+ psEvents :: Map (EventId p) (f (Delta p e), Set p)+ } deriving stock (Generic)+deriving stock instance+ ( Eq (f (Delta p e))+ , Eq (Output e)+ , Eq o+ , Eq p+ , Eq e+ )+ =>+ Eq (EventFoldF o p e f)+instance+ (+ Binary (f (Delta p e)),+ Binary o,+ Binary p,+ Binary e,+ Binary (State e),+ Binary (Output e)+ )+ =>+ Binary (EventFoldF o p e f)+deriving stock instance+ ( Show (f (Delta p e))+ , Show o+ , Show p+ , Show (State e)+ )+ => Show (EventFoldF o p e f)+++type EventFold o p e = EventFoldF o p e Identity+++{- |+ `Infimum` is the infimum, or greatest lower bound, of the possible+ values of @s@.+-}+data Infimum s p = Infimum {+ eventId :: EventId p,+ participants :: Set p,+ stateValue :: s+ } deriving stock (Generic, Show)+instance (Binary s, Binary p) => Binary (Infimum s p)+instance (Eq p) => Eq (Infimum s p) where+ Infimum s1 _ _ == Infimum s2 _ _ = s1 == s2+instance (Ord p) => Ord (Infimum s p) where+ compare (Infimum s1 _ _) (Infimum s2 _ _) = compare s1 s2+++{- |+ `EventId` is a monotonically increasing, totally ordered identification+ value which allows us to lend the attribute of monotonicity to event+ application operations which would not naturally be monotonic.+-}+data EventId p+ = BottomEid+ | Eid Word256 p+ deriving stock (Generic, Eq, Ord, Show)+instance (Binary p) => Binary (EventId p) where+ put = put . toMaybe+ where+ toMaybe :: EventId p -> Maybe (Word64, Word64, Word64, Word64, p)+ toMaybe BottomEid =+ Nothing+ toMaybe (Eid (Word256 (Word128 a b) (Word128 c d)) p) =+ Just (a, b, c, d, p)+ get = do+ theThing <- get+ return $ case theThing of+ Nothing -> BottomEid+ Just (a, b, c, d, p) -> Eid (Word256 (Word128 a b) (Word128 c d)) p+instance Default (EventId p) where+ def = BottomEid+++{- |+ This is the exception type for illegal merges. These errors indicate+ a serious programming bugs.+-}+data MergeError o p e+ = DifferentOrigins o o+ {- ^+ The 'EventFold's do not have the same origin. It makes no sense+ to merge 'EventFold's that have different origins because they+ do not share a common history.+ -}+ | EventPackTooNew (EventFold o p e) (EventPack o p e)+ {- ^+ The 'EventPack''s infimum is greater than any event known to+ 'EventFold' into which it is being merged. This should be impossible+ and indicates that either the local 'EventFold' has rolled back an+ event that it had previously acknowledged, or else the source of the+ 'EventPack' moved the infimum forward without a full acknowledgement+ from all participants. Both of these conditions should be regarded+ as serious bugs.+ -}+ | EventPackTooSparse (EventFold o p e) (EventPack o p e)+ {- ^+ The 'EventPack' assumes we know about events that we do not in+ fact know about. This is only possible if we rolled back our+ copy of the state somehow and "forgot" about state that we had+ previous acknowledged, or else some other participant erroneously+ acknowledged some events on our behalf.+ -}+deriving stock instance+ ( Show (Output e)+ , Show o+ , Show p+ , Show e+ , Show (State e)+ )+ =>+ Show (MergeError o p e)+++{- | `Delta` is how we represent mutations to the event fold state. -}+data Delta p e+ = Join p+ | UnJoin p+ | Event e+ | Error (Output e) (Set p)+ deriving stock (Generic)+deriving stock instance (Eq p, Eq e, Eq (Output e)) => Eq (Delta p e)+deriving stock instance (Show p, Show e, Show (Output e)) => Show (Delta p e)+instance (Binary p, Binary e, Binary (Output e)) => Binary (Delta p e)+++{- |+ Instances of this class define the particular "events" being "folded"+ over in a distributed fashion. In addition to the event type itself,+ there are a couple of type families which define the 'State' into which+ folded events are accumulated, and the 'Output' which application of+ a particular event can generate.++ TL;DR: This is how users define their own custom operations.+-}+class Event e where+ type Output e+ type State e+ {- | Apply an event to a state value. **This function MUST be total!!!** -}+ apply :: e -> State e -> EventResult e+{- | The most trivial event type. -}+instance Event () where+ type Output () = ()+ type State () = ()+ apply () () = Pure () ()+{- | The union of two event types. -}+instance (Event a, Event b) => Event (Either a b) where+ type Output (Either a b) = Either (Output a) (Output b)+ type State (Either a b) = (State a, State b)++ apply (Left e) (a, b) = + case apply e a of+ SystemError o -> SystemError (Left o)+ Pure o s -> Pure (Left o) (s, b)+ apply (Right e) (a, b) = + case apply e b of+ SystemError o -> SystemError (Right o)+ Pure o s -> Pure (Right o) (a, s)+++{- |+ The result of applying an event.++ Morally speaking, events are always pure functions. However, mundane+ issues like finite memory constraints and finite execution time can+ cause referentially opaque behavior. In a normal Haskell program, this+ usually leads to a crash or an exception, and the crash or exception+ can itself, in a way, be thought of as being referentially transparent,+ because there is no way for it to both happen and, simultaneously,+ not happen.++ However, in our case we are replicating computations across many+ different pieces of hardware, so there most definitely is a way+ for these aberrant system failures to both happen and not happen+ simultaneously. What happens if the computation of the event runs out+ of memory on one machine, but not on another?++ There exists a strategy for dealing with these problems: if the+ computation of an event experiences a failure on every participant, then+ the event is pushed into the infimum as a failure (i.e. a no-op), but if+ any single participant successfully computes the event then all other+ participants can (somehow) request a "Full Merge" from the successful+ participant. The Full Merge will include the infimum __value__ computed+ by the successful participant, which will include the successful+ application of the problematic event. The error participants can thus+ bypass computation of the problem event altogether, and can simply+ overwrite their infimum with the infimum provided by the Full Merge.++ Doing a full merge can be much more expensive than doing a simple+ 'EventPack' merge, because it requires transmitting the full value of+ the 'EventFold' instead of just the outstanding operations.++ This type represents how computation of the event finished; with either a+ pure result, or some kind of system error.++ In general 'SystemError' is probably only ever useful for when your+ event type somehow executes untrusted code (for instance when your event+ type is a Turing-complete DSL that allows users to submit their own+ custom-programmed "events") and you want to limit the resources that can+ be consumed by such user-generated code. It is much less useful when+ you are encoding some well defined business logic directly in Haskell.+-}+data EventResult e+ = SystemError (Output e)+ | Pure (Output e) (State e)+++{- |+ Construct a new 'EventFold' with the given origin and initial+ participant.+-}+new+ :: (Default (State e), Ord p)+ => o {- ^ The "origin", iditifying the historical lineage of this CRDT. -}+ -> p {- ^ The initial participant. -}+ -> EventFold o p e+new o participant =+ EventFold {+ psOrigin = o,+ psInfimum = Infimum {+ eventId = def,+ participants = Set.singleton participant,+ stateValue = def+ },+ psEvents = mempty+ }+++{- |+ Get the outstanding events that need to be propagated to a particular+ participant.+-}+events :: (Ord p) => p -> EventFold o p e -> EventPack o p e+events peer ps =+ EventPack {+ epEvents = omitAcknowledged <$> psEvents ps,+ epOrigin = psOrigin ps,+ epInfimum = eventId (psInfimum ps)+ }+ where+ {- |+ Don't send the event data to participants which have already+ acknowledged it, saving network and cpu resources.+ -}+ omitAcknowledged (d, acks) =+ (+ case (d, peer `member` acks) of+ (Identity Error {}, _) -> Just (runIdentity d)+ (_, False) -> Just (runIdentity d)+ _ -> Nothing,+ acks+ )+++{- | A package containing events that can be merged into an event fold. -}+data EventPack o p e = EventPack {+ epEvents :: Map (EventId p) (Maybe (Delta p e), Set p),+ epOrigin :: o,+ epInfimum :: EventId p+ }+ deriving stock (Generic)+deriving stock instance (+ Show o, Show p, Show e, Show (Output e)+ ) =>+ Show (EventPack o p e)+instance (+ Binary o, Binary p, Binary e, Binary (Output e)+ ) =>+ Binary (EventPack o p e)+++{- |+ Monotonically merge the information in two 'EventFold's. The resulting+ 'EventFold' may have a higher infimum value, but it will never have+ a lower one. Only 'EventFold's that originated from the same 'new'+ call can be merged. If the origins are mismatched, then 'Nothing'+ is returned.++ Returns the new 'EventFold' value, along with the output for all of+ the events that can now be considered "fully consistent".+-}+mergeMaybe :: (Eq o, Event e, Ord p)+ => EventFold o p e+ -> EventPack o p e+ -> Maybe (EventFold o p e, Map (EventId p) (Output e))+mergeMaybe ps es = either (const Nothing) Just (mergeEither ps es)+++{- |+ Like `mergeMaybe`, but returns an error indicating exactly what+ went wrong.+-}+mergeEither :: (Eq o, Event e, Ord p)+ => EventFold o p e+ -> EventPack o p e+ -> Either+ (MergeError o p e)+ (EventFold o p e, Map (EventId p) (Output e))++mergeEither EventFold {psOrigin = o1} EventPack {epOrigin = o2} | o1 /= o2 =+ Left (DifferentOrigins o1 o2)++mergeEither ps pak | tooNew =+ Left (EventPackTooNew ps pak)+ where+ maxState =+ maximum+ . Set.insert (eventId . psInfimum $ ps)+ . Map.keysSet+ . psEvents+ $ ps++ tooNew :: Bool+ tooNew = maxState < epInfimum pak++mergeEither orig@(EventFold o infimum d1) ep@(EventPack d2 _ i2) =+ case+ reduce+ i2+ EventFold {+ psOrigin = o,+ psInfimum = infimum,+ psEvents =+ Map.Merge.merge+ (Map.Merge.mapMissing (const (first Just)))+ Map.Merge.preserveMissing+ (Map.Merge.zipWithMatched (const mergeAcks))+ (first runIdentity <$> d1)+ d2+ }+ of+ Nothing -> Left (EventPackTooSparse orig ep)+ Just ps -> Right ps+ where+ mergeAcks :: (Ord p)+ => (Delta p e, Set p)+ -> (Maybe (Delta p e), Set p)+ -> (Maybe (Delta p e), Set p)+ mergeAcks+ (Error output eacks1, acks1)+ (Just (Error _ eacks2), acks2)+ =+ (Just (Error output (eacks1 `union` eacks2)), acks1 `union` acks2)+ mergeAcks+ (Error {}, acks1)+ (d, acks2)+ =+ (d, acks1 `union` acks2)+ mergeAcks+ (d, acks1)+ (Just _, acks2)+ =+ (Just d, acks1 `union` acks2)+ mergeAcks+ (d, acks1)+ (Nothing, acks2)+ =+ (Just d, acks1 `union` acks2)+++{- |+ Like 'mergeEither', but merge a full 'EventFold' instead of just an+ event pack.++ Returns the new 'EventFold' value, along with the output for all of+ the events that can now be considered "fully consistent".+-}+fullMerge :: (Eq o, Event e, Ord p)+ => EventFold o p e+ -> EventFold o p e+ -> Either (MergeError o p e) (EventFold o p e, Map (EventId p) (Output e))+fullMerge ps (EventFold o2 i2 d2) =+ mergeEither+ ps {psInfimum = max (psInfimum ps) i2}+ EventPack {+ epOrigin = o2,+ epEvents = first (Just . runIdentity) <$> d2,+ epInfimum = eventId i2+ }+++{- |+ Record the fact that the participant acknowledges the information+ contained in the 'EventFold'. The implication is that the participant+ __must__ base all future operations on the result of this function.++ Returns the new 'EventFold' value, along with the output for all of+ the events that can now be considered "fully consistent".+-}+acknowledge :: (Event e, Ord p)+ => p+ -> EventFold o p e+ -> (EventFold o p e, Map (EventId p) (Output e))+acknowledge p ps =+ {-+ First do a normal reduction, then do a special acknowledgement of the+ reduction error, if any.+ -}+ let+ (ps2, outputs) =+ runIdentity $+ reduce+ (eventId (psInfimum ps))+ ps {psEvents = fmap ackOne (psEvents ps)}+ (ps3, outputs2) = ackErr p ps2+ in+ (ps3, outputs <> outputs2)+ where+ ackOne (e, acks) = (e, Set.insert p acks)+++{- | Acknowledge the reduction error, if one exists. -}+ackErr :: (Event e, Ord p)+ => p+ -> EventFold o p e+ -> (EventFold o p e, Map (EventId p) (Output e))+ackErr p ps =+ runIdentity $+ reduce+ (eventId (psInfimum ps))+ ps {+ psEvents =+ case Map.minViewWithKey (psEvents ps) of+ Just ((eid, (Identity (Error o eacks), acks)), deltas) ->+ Map.insert+ eid+ (Identity (Error o (Set.insert p eacks)), acks)+ deltas+ _ -> psEvents ps+ }+++{- |+ Allow a participant to join in the distributed nature of the+ 'EventFold'. Return the 'EventId' at which the participation is+ recorded, and the resulting 'EventFold'. The purpose of returning the+ state is so that it can use it to tell when the participation event+ has reached the infimum.+-}+participate :: (Ord p)+ => p+ -> p+ -> EventFold o p e+ -> (EventId p, EventFold o p e)+participate self peer ps@EventFold {psEvents} =+ let+ eid = nextId self ps+ in+ (+ eid,+ ps {+ psEvents =+ Map.insert+ eid+ (Identity (Join peer), mempty)+ psEvents+ }+ )+++{- |+ Indicate that a participant is removing itself from participating in+ the distributed 'EventFold'.+-}+disassociate :: (Ord p)+ => p+ -> p+ -> EventFold o p e+ -> EventFold o p e+disassociate self peer ps@EventFold {psEvents} =+ ps {+ psEvents =+ Map.insert+ (nextId self ps)+ (Identity (UnJoin peer), mempty)+ psEvents+ }+++{- |+ Introduce a change to the EventFold on behalf of the participant.+ Return the new 'EventFold', along with the projected output of the+ event, along with an 'EventId' which can be used to get the fully+ consistent event output at a later time.+-}+event :: (Ord p, Event e)+ => p+ -> e+ -> EventFold o p e+ -> (Output e, EventId p, EventFold o p e)+event p e ps@EventFold {psEvents} =+ let+ eid = nextId p ps+ in+ (+ case apply e (projectedValue ps) of+ Pure output _ -> output+ SystemError output -> output,+ eid,+ ps {+ psEvents =+ Map.insert+ eid+ (Identity (Event e), mempty)+ psEvents+ }+ )+++{- | Return the current projected value of the 'EventFold'. -}+projectedValue :: (Event e) => EventFold o p e -> State e+projectedValue EventFold {psInfimum = Infimum {stateValue}, psEvents} =+ foldr+ (\ e s ->+ case apply e s of+ Pure _ newState -> newState+ SystemError _ -> s+ )+ stateValue+ changes+ where+ changes = foldMap getDelta (toDescList psEvents)+ getDelta :: (EventId p, (Identity (Delta p e), Set p)) -> [e]+ getDelta (_, (Identity (Event e), _)) = [e]+ getDelta _ = mempty+++{- | Return the current infimum value of the 'EventFold'. -}+infimumValue :: EventFoldF o p e f -> State e+infimumValue EventFold {psInfimum = Infimum {stateValue}} = stateValue+++{- | Return the 'EventId' of the infimum value. -}+infimumId :: EventFoldF o p e f -> EventId p+infimumId = eventId . psInfimum+++{- |+ Gets the known participants at the infimum.+-}+infimumParticipants :: EventFoldF o p e f -> Set p+infimumParticipants EventFold {psInfimum = Infimum {participants}} =+ participants+++{- |+ Get all known participants. This includes participants that are+ projected for removal.+-}+allParticipants :: (Ord p) => EventFold o p e -> Set p+allParticipants EventFold {+ psInfimum = Infimum {participants},+ psEvents+ } =+ foldr updateParticipants participants (toDescList psEvents)+ where+ updateParticipants :: (Ord p)+ => (EventId p, (Identity (Delta p e), Set p))+ -> Set p+ -> Set p+ updateParticipants (_, (Identity (Join p), _)) = Set.insert p+ updateParticipants _ = id+++{- |+ Get all the projected participants. This does not include participants that+ are projected for removal.+-}+projParticipants :: (Ord p) => EventFold o p e -> Set p+projParticipants EventFold {+ psInfimum = Infimum {participants},+ psEvents+ } =+ foldr updateParticipants participants (toDescList psEvents)+ where+ updateParticipants :: (Ord p)+ => (EventId p, (Identity (Delta p e), Set p))+ -> Set p+ -> Set p+ updateParticipants (_, (Identity (Join p), _)) = Set.insert p+ updateParticipants (_, (Identity (UnJoin p), _)) = Set.delete p+ updateParticipants _ = id+++{- |+ Returns the participants that we think might be diverging. In+ this context, a participant is "diverging" if there is an event+ that the participant has not acknowledged but we are expecting it+ to acknowledge. Along with the participant, return the last known+ `EventId` which that participant has acknowledged.+-}+divergent :: forall o p e. (Ord p) => EventFold o p e -> Map p (EventId p)+divergent+ EventFold {+ psInfimum = Infimum {participants, eventId},+ psEvents+ }+ =+ let (byParticipant, maxEid) = eidByParticipant+ in Map.filter (< maxEid) byParticipant++ where+ eidByParticipant :: (Map p (EventId p), EventId p)+ eidByParticipant =+ foldr+ accum+ (Map.fromList [(p, eventId) | p <- Set.toList participants], eventId)+ (+ let flatten (a, (Identity b, c)) = (a, b, c)+ in (flatten <$> toAscList psEvents)+ )++ accum+ :: (EventId p, Delta p e, Set p)+ -> (Map p (EventId p), EventId p)+ -> (Map p (EventId p), EventId p)++ accum (eid, Join p, acks) (acc, maxEid) =+ (+ unionWith+ max+ (Map.insert p eid acc)+ (Map.fromList [(a, eid) | a <- Set.toList acks]),+ max maxEid eid+ )++ accum (eid, _, acks) (acc, maxEid) =+ (+ unionWith+ max+ acc+ (Map.fromList [(a, eid) | a <- Set.toList acks]),+ max maxEid eid+ )+++{- | Return the origin value of the 'EventFold'. -}+origin :: EventFoldF o p e f -> o+origin = psOrigin+++{- |+ This helper function is responsible for figuring out if the 'EventFold'+ has enough information to derive a new infimum value. In other words,+ this is where garbage collection happens.+-}+reduce+ :: forall o p e f.+ ( Event e+ , Monad f+ , Ord p+ )+ => EventId p+ {- ^+ The infimum 'EventId' as known by some node in the cluster. "Some+ node" can be different than "this node" in the case where another+ node advanced the infimum before we did (because it knew about+ our acknowledgement, but we didn't know about its acknowledgement)+ and sent us an 'EventPack' with this value of the infimum. In this+ case, this infimum value acts as a universal acknowledgement of+ all events coming before it.+ -}+ -> EventFoldF o p e f+ -> f (EventFold o p e, Map (EventId p) (Output e))+reduce+ infState+ ps@EventFold {+ psInfimum = infimum@Infimum {participants, stateValue},+ psEvents+ }+ =+ case Map.minViewWithKey psEvents of+ Nothing ->+ pure+ (+ EventFold {+ psOrigin = psOrigin ps,+ psInfimum = psInfimum ps,+ psEvents = mempty+ },+ mempty+ )+ Just ((eid, (getUpdate, acks)), newDeltas)+ | eid <= eventId infimum -> {- The event is obsolete. Ignore it. -}+ reduce infState ps {+ psEvents = newDeltas+ }+ | isRenegade eid -> {- This is a renegade event. Ignore it. -}+ reduce infState ps {+ psEvents = newDeltas+ }+ | otherwise -> do+ implicitAcks <- unjoins eid++ update <- getUpdate+ let+ {- |+ Join events must be acknowledged by the joining+ participant before moving into the infimum.+ -}+ joining =+ case update of+ Join p -> Set.singleton p+ _ -> mempty+ if+ Set.null (((participants `union` joining) \\ acks) \\ implicitAcks)+ || eid <= infState+ then+ case update of+ Join p ->+ reduce infState ps {+ psInfimum = infimum {+ eventId = eid,+ participants = Set.insert p participants+ },+ psEvents = newDeltas+ }+ UnJoin p ->+ reduce infState ps {+ psInfimum = infimum {+ eventId = eid,+ participants = Set.delete p participants+ },+ psEvents = newDeltas+ }+ Error output eacks+ | Set.null (participants \\ eacks) -> do+ (ps2, outputs) <-+ reduce infState ps {+ psInfimum = infimum {+ eventId = eid+ }+ }+ pure (ps2, Map.insert eid output outputs)+ | otherwise -> do+ events_ <- runEvents psEvents+ pure+ (+ EventFold {+ psOrigin = psOrigin ps,+ psInfimum = psInfimum ps,+ psEvents = events_+ },+ mempty+ )+ Event e ->+ case apply e stateValue of+ SystemError output -> do+ events_ <- runEvents newDeltas+ pure+ (+ EventFold {+ psOrigin = psOrigin ps,+ psInfimum = infimum,+ psEvents =+ Map.insert+ eid+ (Identity (Error output mempty), acks)+ events_+ },+ mempty+ )+ Pure output newState -> do+ (ps2, outputs) <-+ reduce infState ps {+ psInfimum = infimum {+ eventId = eid,+ stateValue = newState+ },+ psEvents = newDeltas+ }+ pure (ps2, Map.insert eid output outputs)+ else do+ events_ <- runEvents psEvents+ pure+ (+ EventFold {+ psOrigin = psOrigin ps,+ psInfimum = psInfimum ps,+ psEvents = events_+ },+ mempty+ )+ where+ {- | Unwrap the events from their monad. -}+ runEvents+ :: Map (EventId p) (f (Delta p e), Set p)+ -> f (Map (EventId p) (Identity (Delta p e), Set p))+ runEvents events_ =+ Map.fromList <$> sequence [+ do+ d <- fd+ pure (eid, (Identity d, acks))+ | (eid, (fd, acks)) <- Map.toList events_+ ]++ {- | Figure out which nodes have upcoming unjoins. -}+ unjoins+ :: EventId p+ {- ^+ The even under consideration, unjoins only after which we+ are interested.+ -}+ -> f (Set p)+ unjoins eid =+ Set.fromList+ . Map.elems+ . Map.filterWithKey (\k _ -> eid <= k)+ <$> unjoinMap++ {- | The static map of unjoins. -}+ unjoinMap :: f (Map (EventId p) p)+ unjoinMap =+ Map.fromList . catMaybes <$> sequence [+ update >>= \case+ UnJoin p -> pure (Just (eid, p))+ _ -> pure Nothing+ | (eid, (update, _acks)) <- Map.toList psEvents+ ]++ {- |+ Renegade events are events that originate from a non-participating+ peer. This might happen in a network partition situation, where+ the cluster ejected a peer that later reappears on the network,+ broadcasting updates.+ -}+ isRenegade BottomEid = False+ isRenegade (Eid _ p) = not (p `member` participants)+++{- |+ A utility function that constructs the next `EventId` on behalf of+ a participant.+-}+nextId :: (Ord p) => p -> EventFoldF o p e f -> EventId p+nextId p EventFold {psInfimum = Infimum {eventId}, psEvents} =+ case maximum (eventId:keys psEvents) of+ BottomEid -> Eid 0 p+ Eid ord _ -> Eid (succ ord) p+++{- | Return 'True' if progress on the 'EventFold' is blocked on a 'SystemError'. -}+isBlockedOnError :: EventFold o p e -> Bool+isBlockedOnError ps =+ case Map.minView (psEvents ps) of+ Just ((Identity (Error _ _), _), _) -> True+ _ -> False++