diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,4 @@ +# crdt-event-fold + +A Haskell library providing garbage collected event folding CRDT. + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/crdt-event-fold.cabal b/crdt-event-fold.cabal new file mode 100644 --- /dev/null +++ b/crdt-event-fold.cabal @@ -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 + diff --git a/src/Data/CRDT/EventFold.hs b/src/Data/CRDT/EventFold.hs new file mode 100644 --- /dev/null +++ b/src/Data/CRDT/EventFold.hs @@ -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 + +