ron-rdt 0.10 → 0.11
raw patch · 19 files changed
+2339/−940 lines, 19 filesdep +bytestringdep +directorydep +filelockdep −integer-gmpdep −transformersdep ~Diffdep ~base
Dependencies added: bytestring, directory, filelock, filepath, generic-lens, lens, network-info, stm, tf-random
Dependencies removed: integer-gmp, transformers
Dependency ranges changed: Diff, base
Files
- CHANGELOG.md +21/−1
- LICENSE +1/−1
- lib/RON/Data.hs +120/−92
- lib/RON/Data/CT.hs +341/−0
- lib/RON/Data/GTree.hs +53/−0
- lib/RON/Data/Internal.hs +202/−160
- lib/RON/Data/LWW.hs +81/−55
- lib/RON/Data/ORSet.hs +277/−204
- lib/RON/Data/RGA.hs +371/−347
- lib/RON/Data/Time.hs +14/−12
- lib/RON/Data/VersionVector.hs +108/−22
- lib/RON/Experimental/Data.hs +129/−0
- lib/RON/Experimental/Data/ORSet.hs +145/−0
- lib/RON/Experimental/Data/ORSet/Type.hs +16/−0
- lib/RON/Store.hs +68/−0
- lib/RON/Store/Class.hs +27/−0
- lib/RON/Store/FS.hs +225/−0
- lib/RON/Store/Test.hs +76/−0
- ron-rdt.cabal +64/−46
CHANGELOG.md view
@@ -7,6 +7,24 @@ ## [Unreleased] +## [0.12] - 2026-02-27+### Changed+- Use sequential ids on RGA block insertion and deletions (#15).++### Fixed+- Use strict patch file reading to save on file numbers when reading too many+ objects.++## [0.11] - 2021-06-15+### Added+- `VersionVector`: operator (·≻)+- Experimental `GTree` -- grow-only tree+- `CausalTree`++### Fixed+- Adapt to changes in RON.Prelude+- `advanceToObject`+ ## [0.10] - 2019-10-08 ### Added - `RGA.toList`@@ -203,7 +221,9 @@ - RON-Schema - RON-Schema TemplateHaskell code generator -[Unreleased]: https://github.com/ff-notes/ron/compare/ron-rdt-0.10...HEAD+[Unreleased]: https://github.com/ff-notes/ron/compare/ron-rdt-0.12...HEAD+[0.12]: https://github.com/ff-notes/ron/compare/ron-rdt-0.11...ron-rdt-0.12+[0.11]: https://github.com/ff-notes/ron/compare/ron-rdt-0.10...ron-rdt-0.11 [0.10]: https://github.com/ff-notes/ron/compare/ron-rdt-0.9...ron-rdt-0.10 [0.9]: https://github.com/ff-notes/ron/compare/ron-rdt-0.8...ron-rdt-0.9 [0.8]: https://github.com/ff-notes/ron/compare/ron-rdt-0.7...ron-rdt-0.8
LICENSE view
@@ -1,6 +1,6 @@ BSD 3-Clause License -Copyright (c) 2018, Yuriy Syrovetskiy+Copyright (c) 2018-2019, Yuriy Syrovetskiy All rights reserved. Redistribution and use in source and binary forms, with or without
lib/RON/Data.hs view
@@ -1,8 +1,9 @@ {-# LANGUAGE AllowAmbiguousTypes #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ImportQualifiedPost #-} {-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiWayIf #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}@@ -25,6 +26,7 @@ reduceWireFrame, stateFromWireChunk, stateToWireChunk,+ -- * Object-state monad ObjectStateT, MonadObjectState,@@ -38,92 +40,111 @@ runObjectState_, ) where -import RON.Prelude+import RON.Prelude -import qualified Data.List.NonEmpty as NonEmpty-import Data.Map.Strict ((!?))-import qualified Data.Map.Strict as Map+import Data.List.NonEmpty qualified as NonEmpty+import Data.Map.Strict ((!?))+import Data.Map.Strict qualified as Map -import RON.Data.Internal-import RON.Data.LWW (LwwRep)-import RON.Data.ORSet (ORSetRep)-import RON.Data.RGA (RgaRep)-import RON.Data.VersionVector (VersionVector)-import RON.Error (MonadE, throwErrorString)-import RON.Types (ClosedOp (..),- ObjectFrame (ObjectFrame, frame, uuid),- ObjectRef (ObjectRef), Op (..), StateChunk (..),- StateFrame, UUID, WireChunk (Closed, Query, Value),- WireFrame, WireReducedChunk (..),- WireStateChunk (WireStateChunk, stateBody, stateType))-import RON.UUID (pattern Zero)+import RON.Data.Internal+import RON.Data.LWW (LwwRep)+import RON.Data.ORSet (ORSetRep)+import RON.Data.RGA (RgaRep)+import RON.Data.VersionVector (VersionVector)+import RON.Error (MonadE, throwErrorString)+import RON.Types (+ ClosedOp (..),+ ObjectFrame (ObjectFrame, frame, uuid),+ ObjectRef (ObjectRef),+ Op (..),+ StateChunk (..),+ StateFrame,+ UUID,+ WireChunk (Closed, Query, Value),+ WireFrame,+ WireReducedChunk (..),+ WireStateChunk (WireStateChunk, stateBody, stateType),+ )+import RON.UUID (pattern Zero) reducers :: Map UUID Reducer-reducers = Map.fromList- [ mkReducer @LwwRep- , mkReducer @RgaRep- , mkReducer @ORSetRep- , mkReducer @VersionVector- ]+reducers =+ Map.fromList+ [ mkReducer @LwwRep+ , mkReducer @RgaRep+ , mkReducer @ORSetRep+ , mkReducer @VersionVector+ ] reduceWireFrame :: WireFrame -> WireFrame-reduceWireFrame chunks = values' ++ queries where- chunkTypeAndObject = opTypeAndObject . \case- Closed op -> op- Value WireReducedChunk{wrcHeader = op} -> op- Query WireReducedChunk{wrcHeader = op} -> op+reduceWireFrame chunks = values' ++ queries+ where+ chunkTypeAndObject =+ opTypeAndObject . \case+ Closed op -> op+ Value WireReducedChunk{wrcHeader = op} -> op+ Query WireReducedChunk{wrcHeader = op} -> op opTypeAndObject ClosedOp{..} = (reducerId, objectId) (queries, values) = partition isQuery chunks values' = fold $- Map.mapWithKey reduceWireFrameByType $- NonEmpty.fromList <$>- Map.fromListWith (++)- [(chunkTypeAndObject value, [value]) | value <- values]+ Map.mapWithKey reduceWireFrameByType $+ NonEmpty.fromList+ <$> Map.fromListWith+ (++)+ [(chunkTypeAndObject value, [value]) | value <- values] reduceWireFrameByType :: (UUID, UUID) -> NonEmpty WireChunk -> [WireChunk] reduceWireFrameByType (typ, obj) = case reducers !? typ of Nothing ->- toList -- TODO(2018-11-08, cblp, #27) use default reducer+ toList -- TODO(2018-11-08, cblp, #27) use default reducer Just Reducer{wireReducer} -> wireReducer obj isQuery :: WireChunk -> Bool isQuery = \case Query _ -> True- _ -> False+ _ -> False -mkReducer :: forall a . Reducible a => (UUID, Reducer)+mkReducer :: forall a. (Reducible a) => (UUID, Reducer) mkReducer = ( reducibleOpType @a , Reducer- {wireReducer = mkWireReducer @a, stateReducer = unsafeReduceState @a}+ { wireReducer = mkWireReducer @a+ , stateReducer = unsafeReduceState @a+ } ) where- -- | Reduce states using type @rep@ without checking 'stateType'- unsafeReduceState- :: forall rep- . Reducible rep => WireStateChunk -> WireStateChunk -> WireStateChunk+ -- \| Reduce states using type @rep@ without checking 'stateType'+ unsafeReduceState ::+ forall rep.+ (Reducible rep) =>+ WireStateChunk ->+ WireStateChunk ->+ WireStateChunk unsafeReduceState- WireStateChunk{stateBody = c1} WireStateChunk{stateBody = c2} =- wireStateChunk @rep $ reduceState (StateChunk c1) (StateChunk c2)+ WireStateChunk{stateBody = c1}+ WireStateChunk{stateBody = c2} =+ wireStateChunk @rep $ reduceState (StateChunk c1) (StateChunk c2) -mkWireReducer :: forall a . Reducible a => WireReducer-mkWireReducer obj chunks = chunks' <> leftovers where- chunks'- = maybeToList stateChunk'- ++ map (Value . wrapRChunk) unappliedPatches- ++ map (Closed . wrapOp) unappliedOps+mkWireReducer :: forall a. (Reducible a) => WireReducer+mkWireReducer obj chunks = chunks' <> leftovers+ where+ chunks' =+ maybeToList stateChunk'+ ++ map (Value . wrapRChunk) unappliedPatches+ ++ map (Closed . wrapOp) unappliedOps mStates = nonEmpty states (stateChunk', (unappliedPatches, unappliedOps)) = case mStates of Nothing -> (Nothing, reduceUnappliedPatches @a (patches, closedOps))- Just nStates -> let- nState = sconcat $ fmap snd nStates- (reducedState, unapplied') =- applyPatches nState (patches, closedOps)- reducedStateBody = stateToChunk @a reducedState- rc = ReducedChunk{rcRef = Zero, rcBody = reducedStateBody}- in- (Just $ Value $ wrapRChunk rc, reduceUnappliedPatches @a unapplied')+ Just nStates ->+ let nState = sconcat $ fmap snd nStates+ (reducedState, unapplied') =+ applyPatches nState (patches, closedOps)+ reducedStateBody = stateToChunk @a reducedState+ rc = ReducedChunk{rcRef = Zero, rcBody = reducedStateBody}+ in ( Just $ Value $ wrapRChunk rc+ , reduceUnappliedPatches @a unapplied'+ ) typ = reducibleOpType @a wrapOp = ClosedOp typ obj (states, patches, closedOps, leftovers) = foldMap load chunks@@ -134,16 +155,18 @@ pure ([], [], [op], []) Value WireReducedChunk{wrcHeader, wrcBody} -> do guardSameObject wrcHeader- let ref = refId $ op wrcHeader+ let ref = wrcHeader.op.refId case ref of- Zero -> -- state+ Zero ->+ -- state pure- ( [(opId $ op wrcHeader, stateFromChunk wrcBody)]+ ( [(wrcHeader.op.opId, stateFromChunk wrcBody)] , [] , [] , [] )- _ -> -- patch+ _ ->+ -- patch pure ( [] , [ReducedChunk{rcRef = ref, rcBody = wrcBody}]@@ -153,69 +176,74 @@ _ -> Nothing guardSameObject ClosedOp{reducerId, objectId} = guard $ reducerId == typ && objectId == obj- wrapRChunk ReducedChunk{rcRef, rcBody} = WireReducedChunk- { wrcHeader = wrapOp Op{opId = Zero, refId = rcRef, payload = []}- , wrcBody = rcBody- }+ wrapRChunk ReducedChunk{rcRef, rcBody} =+ WireReducedChunk+ { wrcHeader = wrapOp Op{opId = Zero, refId = rcRef, payload = []}+ , wrcBody = rcBody+ } -reduceStateFrame :: MonadE m => StateFrame -> StateFrame -> m StateFrame+reduceStateFrame :: (MonadE m) => StateFrame -> StateFrame -> m StateFrame reduceStateFrame s1 s2 =- (`execStateT` s1) . (`Map.traverseWithKey` s2) $ \oid chunk -> let- WireStateChunk{stateType} = chunk+ (`execStateT` s1) . (`Map.traverseWithKey` s2) $ \oid chunk ->+ let+ WireStateChunk{stateType} = chunk in- case reducers !? stateType of- Just Reducer{stateReducer} ->- modify' $ Map.insertWith stateReducer oid chunk- Nothing ->- throwErrorString $- "Cannot reduce StateFrame of unknown type " ++ show stateType+ case reducers !? stateType of+ Just Reducer{stateReducer} ->+ modify' $ Map.insertWith stateReducer oid chunk+ Nothing ->+ throwErrorString $+ "Cannot reduce StateFrame of unknown type "+ ++ show stateType -unsafeReduceObject- :: MonadE m => ObjectFrame a -> StateFrame -> m (ObjectFrame a)+unsafeReduceObject ::+ (MonadE m) => ObjectFrame a -> StateFrame -> m (ObjectFrame a) unsafeReduceObject obj@ObjectFrame{frame = s1} s2 = do frame' <- reduceStateFrame s1 s2 pure obj{frame = frame'} -- | Reduce object with frame from another version of the same object.-reduceObject :: MonadE m => ObjectFrame a -> ObjectFrame a -> m (ObjectFrame a)+reduceObject ::+ (MonadE m) => ObjectFrame a -> ObjectFrame a -> m (ObjectFrame a) reduceObject o1@ObjectFrame{uuid = id1} ObjectFrame{uuid = id2, frame = frame2} | id1 == id2 = unsafeReduceObject o1 frame2- | otherwise = throwErrorString $ "Object ids differ: " ++ show (id1, id2)+ | otherwise = throwErrorString $ "Object ids differ: " ++ show (id1, id2) newtype MaxOnFst a b = MaxOnFst (a, b) -instance Ord a => Semigroup (MaxOnFst a b) where+instance (Ord a) => Semigroup (MaxOnFst a b) where mof1@(MaxOnFst (a1, _)) <> mof2@(MaxOnFst (a2, _))- | a1 < a2 = mof2+ | a1 < a2 = mof2 | otherwise = mof1 -- | Run ObjectFrame action-execObjectState- :: Monad m => ObjectFrame b -> ObjectStateT b m a -> m (ObjectFrame b)+execObjectState ::+ (Monad m) => ObjectFrame b -> ObjectStateT b m a -> m (ObjectFrame b) execObjectState state@ObjectFrame{uuid, frame} action = do frame' <- execStateT (runReaderT action $ ObjectRef uuid) frame pure state{frame = frame'} -- | Run ObjectFrame action, starting with an empty frame-execObjectState_ :: Monad m => StateT StateFrame m a -> m StateFrame+execObjectState_ :: (Monad m) => StateT StateFrame m a -> m StateFrame execObjectState_ action = execStateT action mempty -- | Run ObjectFrame action-runObjectState- :: Functor m- => ObjectFrame b- -> ObjectStateT b m a- -> m (a, ObjectFrame b)+runObjectState ::+ (Functor m) =>+ ObjectFrame b ->+ ObjectStateT b m a ->+ m (a, ObjectFrame b) runObjectState state@ObjectFrame{uuid, frame} action = runStateT (runReaderT action $ ObjectRef uuid) frame- <&> \(a, frame') -> (a, state{frame = frame'})+ <&> \(a, frame') -> (a, state{frame = frame'}) -- | Run ObjectFrame action, starting with an empty frame runObjectState_ :: StateT StateFrame m a -> m (a, StateFrame) runObjectState_ action = runStateT action mempty -- | Create new 'ObjectFrame' with an action-newObjectFrameWith- :: Functor m => StateT StateFrame m (ObjectRef a) -> m (ObjectFrame a)+newObjectFrameWith ::+ (Functor m) => StateT StateFrame m (ObjectRef a) -> m (ObjectFrame a) newObjectFrameWith action =- runObjectState_ action <&> \(ObjectRef uuid, frame) -> ObjectFrame{uuid, frame}+ runObjectState_ action+ <&> \(ObjectRef uuid, frame) -> ObjectFrame{uuid, frame}
+ lib/RON/Data/CT.hs view
@@ -0,0 +1,341 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ImportQualifiedPost #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}++-- | Causal Tree (CT)+module RON.Data.CT (+ CT (..),+ CTRep,+ CTString,+ edit,+ editText,+ getAliveWithIndices,+ aliveWithIndices,+ getList,+ getText,+ insert,+ insertAtBegin,+ insertText,+ insertTextAtBegin,+ newFromList,+ newFromText,+ remove,+ ctType,+ RON.Data.CT.toList,+ toText,+)+where++import RON.Prelude++import Control.Monad.State (state)+import Data.Algorithm.Diff (PolyDiff (Both, First, Second), getGroupedDiffBy)+import Data.Foldable qualified as Foldable+import Data.Map.Strict ((!?))+import Data.Map.Strict qualified as Map+import Data.Set qualified as Set+import Data.Text qualified as Text++import RON.Data.Internal (+ MonadObjectState,+ Reducible,+ Rep,+ Replicated (encoding),+ ReplicatedAsObject,+ ReplicatedAsPayload,+ fromRon,+ getObjectStateChunk,+ modifyObjectStateChunk_,+ newObject,+ newRon,+ objectEncoding,+ readObject,+ reducibleOpType,+ stateFromChunk,+ stateToChunk,+ toPayload,+ )+import RON.Error (MonadE, errorContext)+import RON.Event (ReplicaClock, getEventUuid, getEventUuids)+import RON.Semilattice (Semilattice)+import RON.Types (+ ObjectRef (ObjectRef),+ Op (Op, opId, payload, refId),+ Payload,+ StateChunk (StateChunk),+ StateFrame,+ UUID,+ WireStateChunk (WireStateChunk, stateBody, stateType),+ )+import RON.UUID (zero)+import RON.UUID qualified as UUID+import RON.Util.Word (ls60)++data CTRepItem = CTRepItem+ { op :: Op+ , next :: Maybe UUID+ -- ^ Nothing is the end+ , isDeleted :: Bool+ -- ^ is tombstone or has tombstone among children+ }+ deriving (Eq, Show)++-- | Untyped Causal Tree+data CTRep = CTRep+ { ops :: Map UUID CTRepItem+ -- ^ indexed by opId+ , start :: Maybe UUID+ }+ deriving stock (Eq, Show)+ deriving anyclass (Semilattice)++instance Semigroup CTRep where+ c <> d = CTRep{ops = c.ops <> d.ops, start = min c.start d.start}++instance Monoid CTRep where+ mempty = CTRep{ops = mempty, start = Nothing}++instance Reducible CTRep where+ reducibleOpType = ctType+ stateFromChunk = ctRepFromChunk+ stateToChunk CTRep{ops} = map (.op) $ Foldable.toList ops++ctRepFromChunk :: [Op] -> CTRep+ctRepFromChunk ops = CTRep{ops = items, start}+ where+ opIndex = Map.fromList [(op.opId, op) | op <- ops]++ -- from parent to children+ childrenIndex =+ Map.fromListWith (<>) [(op.refId, Set.singleton op.opId) | op <- ops]++ roots = childrenIndex !? zero ?: Set.empty++ traversed = traverseFromNodes roots+ where+ traverseFromNodes = foldMap traverseFromNode . reverse . Foldable.toList+ traverseFromNode p =+ p : traverseFromNodes (childrenIndex !? p ?: Set.empty)++ items =+ Map.fromList+ [ (opId, CTRepItem{op, next, isDeleted})+ | (opId, next) <-+ zip traversed (map Just (drop 1 traversed) ++ [Nothing])+ , let op =+ opIndex+ !? opId+ ?: error+ ( "CTRep: op "+ <> show opId+ <> " not found in "+ <> show opIndex+ )+ , let children = childrenIndex !? opId ?: Set.empty+ , let isDeleted =+ null op.payload+ || any+ ( \c ->+ maybe False (\d -> null d.payload) $+ opIndex !? c+ )+ children+ ]++ start = Set.lookupMax roots++-- | Name-UUID to use as CT type marker.+ctType :: UUID+ctType = $(UUID.liftName "ct")++-- | Typed CT+newtype CT a = CT [a]+ deriving (Eq, Show)++instance (Replicated a) => Replicated (CT a) where encoding = objectEncoding++instance (Replicated a) => ReplicatedAsObject (CT a) where+ type Rep (CT a) = CTRep++ newObject (CT items) = do+ oid <- getEventUuid+ opIds <- getEventUuids $ ls60 $ genericLength items+ ops <-+ for (zip3 items opIds (zero : opIds)) \(item, opId, refId) -> do+ payload <- newRon item+ pure Op{opId, refId, payload}+ modify' $ Map.insert oid $ wireStateChunk ops+ pure $ ObjectRef oid++ readObject = do+ StateChunk stateBody <- getObjectStateChunk+ fmap CT $+ traverse (fromRon . snd) $+ aliveWithIndices $+ stateFromChunk stateBody++{- | Replace content of the CT through introducing changes detected by+'getGroupedDiffBy'.+-}+edit ::+ ( MonadE m+ , MonadObjectState (CT a) m+ , ReplicaClock m+ , ReplicatedAsPayload a+ ) =>+ [a] ->+ m ()+edit newItems =+ modifyObjectStateChunk_ \(StateChunk stateBody) -> do+ let oldItems' = aliveWithIndices $ stateFromChunk stateBody+ -- TODO(2019-04-17, #59, cblp) replace 'toPayload' with 'newRon' and+ -- relax constraint on 'a' from 'ReplicatedAsPayload' to+ -- 'Replicated'+ let newItems' = map toPayload newItems+ let diff =+ getGroupedDiffBy+ (\(_, p1) p2 -> p1 == p2)+ oldItems'+ newItems'+ newOps <-+ (`evalStateT` zero {- prevous op id -}) $+ fold <$> for diff \case+ First removed -> do+ opIds <- getEventUuids $ genericLength removed+ for (zip removed opIds) \((refId, _), opId) -> do+ put opId+ pure Op{opId, refId, payload = []}+ Both xs _ -> do+ for_ (lastMay xs) \(opId, _payload) -> put opId+ pure []+ Second added -> do+ opIds <- getEventUuids $ genericLength added+ for (zip added opIds) \(item, opId) -> do+ refId <- state (,opId)+ pure Op{opId, refId, payload = item}+ pure $ StateChunk $ stateBody <> newOps++-- | Speciaization of 'edit' for 'Text'+editText ::+ (MonadE m, MonadObjectState CTString m, ReplicaClock m) => Text -> m ()+editText = edit . Text.unpack++{- | Speciaization of 'CT' to 'Char'.+This is the recommended way to store a string.+-}+type CTString = CT Char++-- | Create a CT from a list+newFromList ::+ (MonadState StateFrame m, ReplicaClock m, Replicated a) =>+ [a] ->+ m (ObjectRef (CT a))+newFromList = newObject . CT++-- | Create a 'CTString' from a text+newFromText ::+ (MonadState StateFrame m, ReplicaClock m) =>+ Text ->+ m (ObjectRef CTString)+newFromText = newFromList . Text.unpack++-- | Read elements from CT+getList :: (MonadE m, MonadObjectState (CT a) m, Replicated a) => m [a]+getList = coerce <$> readObject++-- | Read characters from 'CTString'+getText :: (MonadE m, MonadObjectState CTString m) => m Text+getText = Text.pack <$> getList++{- | Insert a sequence of elements after the specified position.+Position is identified by 'UUID'. 'Nothing' means the beginning.+-}+insert ::+ (MonadE m, MonadObjectState (CT a) m, ReplicaClock m, Replicated a) =>+ [a] ->+ -- | position+ UUID ->+ m ()+insert [] _ = pure ()+insert items startRef =+ modifyObjectStateChunk_ \(StateChunk stateBody) -> do+ opIds <- getEventUuids $ ls60 $ genericLength items+ ops <-+ for (zip3 items opIds (startRef : opIds)) \(item, opId, refId) -> do+ payload <- newRon item+ pure Op{opId, refId, payload}+ pure $ StateChunk $ stateBody <> ops++insertAtBegin ::+ (MonadE m, MonadObjectState (CT a) m, ReplicaClock m, Replicated a) =>+ [a] ->+ m ()+insertAtBegin items = insert items zero++{- | Insert a text after the specified position.+Position is identified by 'UUID'. 'zero' means the beginning.+-}+insertText ::+ (MonadE m, MonadObjectState CTString m, ReplicaClock m) =>+ Text ->+ -- | position+ UUID ->+ m ()+insertText = insert . Text.unpack++insertTextAtBegin ::+ (MonadE m, MonadObjectState CTString m, ReplicaClock m) => Text -> m ()+insertTextAtBegin text = insertText text zero++-- | Record a removal of a specific item+remove ::+ (MonadE m, MonadObjectState (CT a) m, ReplicaClock m) =>+ -- | reference to the item to remove+ UUID ->+ m ()+remove refId =+ errorContext "CT.remove" . errorContext ("refId = " <> show refId) $+ modifyObjectStateChunk_ \(StateChunk stateBody) -> do+ opId <- getEventUuid+ let op = Op{opId, refId, payload = []}+ pure $ StateChunk $ stateBody <> [op]++wireStateChunk :: [Op] -> WireStateChunk+wireStateChunk stateBody = WireStateChunk{stateType = ctType, stateBody}++toList :: CT a -> [a]+toList (CT xs) = xs++toText :: CTString -> Text+toText (CT s) = Text.pack s++aliveWithIndices :: CTRep -> [(UUID, Payload)]+aliveWithIndices CTRep{ops, start} = go start+ where+ go = \case+ Nothing -> []+ Just p ->+ case ops !? p of+ Nothing -> []+ Just CTRepItem{op, next, isDeleted} ->+ [(op.opId, op.payload) | not isDeleted] ++ go next++getAliveWithIndices ::+ (MonadE m, MonadObjectState (CT a) m) =>+ m [(UUID, Payload)]+getAliveWithIndices = do+ StateChunk stateBody <- getObjectStateChunk+ pure $ aliveWithIndices $ stateFromChunk stateBody
+ lib/RON/Data/GTree.hs view
@@ -0,0 +1,53 @@+{-# LANGUAGE DisambiguateRecordFields #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ImportQualifiedPost #-}+{-# LANGUAGE NamedFieldPuns #-}++-- | Experimental!+module RON.Data.GTree (GTree, insert, loadForest) where++import RON.Prelude++import Data.HashMap.Strict qualified as HashMap+import Data.HashSet (HashSet)+import Data.HashSet qualified as HashSet+import Data.Tree (Forest, Tree (Node))++import RON.Error (MonadE)+import RON.Event (ReplicaClock, advanceToUuid, getEventUuid)+import RON.Store (MonadStore, appendPatch, loadSubObjectLog)+import RON.Types (Op (..), OpenFrame, UUID)+import RON.Types.Experimental (Patch (..), Ref (..))++-- | Grow-only tree+data GTree++type GTreeState = HashMap UUID (HashSet Op)++readFrame :: UUID -> OpenFrame -> GTreeState+readFrame object ops =+ HashMap.fromListWith+ (<>)+ [ (refId, HashSet.singleton op)+ | op@Op{opId, refId} <- ops+ , opId /= object+ ]++toForest :: UUID -> GTreeState -> Forest Op+toForest object children = go object+ where+ go i =+ [ Node op $ go opId+ | op@Op{opId} <- toList $ HashMap.lookupDefault mempty i children+ ]++loadForest :: (MonadE m, MonadStore m) => Ref GTree -> m (Forest Op)+loadForest ref@(Ref oid _) =+ toForest oid . readFrame oid <$> loadSubObjectLog ref mempty++insert :: (MonadStore m, ReplicaClock m) => Ref GTree -> UUID -> m ()+insert (Ref object prefix) parent = do+ advanceToUuid object+ opId <- getEventUuid+ appendPatch+ Patch{object, log = Op{opId, refId = parent, payload = prefix} :| []}
lib/RON/Data/Internal.hs view
@@ -1,8 +1,12 @@ {-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BlockArguments #-} {-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE DisambiguateRecordFields #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ImportQualifiedPost #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE Rank2Types #-}@@ -51,40 +55,50 @@ MonadObjectState, ) where -import RON.Prelude+import RON.Prelude -import Data.List (minimum)-import qualified Data.Map.Strict as Map-import qualified Data.Text as Text+import Data.List (minimum)+import Data.Map.Strict qualified as Map+import Data.Text qualified as Text -import RON.Error (Error (Error), MonadE, correct, errorContext,- liftMaybe)-import RON.Event (ReplicaClock, advanceToUuid)-import RON.Semilattice (BoundedSemilattice)-import RON.Types (Atom (AInteger, AString, AUuid),- ObjectFrame (ObjectFrame, frame, uuid),- ObjectRef (ObjectRef),- Op (Op, opId, payload, refId), Payload,- StateChunk (StateChunk), StateFrame, UUID (UUID),- WireChunk,- WireStateChunk (WireStateChunk, stateBody, stateType))+import RON.Error (+ Error (Error),+ MonadE,+ correct,+ errorContext,+ liftMaybe,+ )+import RON.Event (Event (..), ReplicaClock, advance, decodeEvent, timeValue)+import RON.Semilattice (BoundedSemilattice)+import RON.Types (+ Atom (AInteger, AString, AUuid),+ ObjectFrame (ObjectFrame, frame, uuid),+ ObjectRef (ObjectRef),+ Op (Op, opId, payload, refId),+ Payload,+ StateChunk (StateChunk),+ StateFrame,+ UUID (UUID),+ WireChunk,+ WireStateChunk (WireStateChunk, stateBody, stateType),+ ) -- | Reduce all chunks of specific type and object in the frame type WireReducer = UUID -> NonEmpty WireChunk -> [WireChunk] data Reducer = Reducer- { wireReducer :: WireReducer+ { wireReducer :: WireReducer , stateReducer :: WireStateChunk -> WireStateChunk -> WireStateChunk } -- | Unapplied patches and raw ops type Unapplied = ([ReducedChunk], [Op]) --- | Untyped-reducible types.--- Untyped means if this type is a container then the types of data contained in--- it is not considered.-class (Eq a, BoundedSemilattice a) => Reducible a where-+{- | Untyped-reducible types.+Untyped means if this type is a container then the types of data contained in+it is not considered.+-}+class (BoundedSemilattice a, Eq a) => Reducible a where -- | UUID of the type reducibleOpType :: UUID @@ -97,57 +111,57 @@ -- | Merge a state with patches and raw ops applyPatches :: a -> Unapplied -> (a, Unapplied) applyPatches a (patches, ops) =- ( a <> foldMap (patchValue . patchFromChunk) patches- <> foldMap (patchValue . patchFromRawOp) ops+ ( a+ <> foldMap ((.patchValue) . patchFromChunk) patches+ <> foldMap ((.patchValue) . patchFromRawOp) ops , mempty ) -- | Merge patches and raw ops into bigger patches or throw obsolete ops reduceUnappliedPatches :: Unapplied -> Unapplied reduceUnappliedPatches (patches, ops) =- ( maybeToList .- fmap (patchToChunk @a . sconcat) .- nonEmpty $- map patchFromChunk patches <> map patchFromRawOp ops+ ( maybeToList+ . fmap (patchToChunk @a . sconcat)+ . nonEmpty+ $ map patchFromChunk patches <> map patchFromRawOp ops , [] ) -stateToWireChunk :: forall rep . Reducible rep => rep -> WireStateChunk+stateToWireChunk :: forall rep. (Reducible rep) => rep -> WireStateChunk stateToWireChunk rep = wireStateChunk @rep $ StateChunk $ stateToChunk @rep rep -stateFromWireChunk- :: forall a m . (MonadE m, Reducible a) => WireStateChunk -> m a+stateFromWireChunk ::+ forall a m. (MonadE m, Reducible a) => WireStateChunk -> m a stateFromWireChunk WireStateChunk{stateType, stateBody} = do unless (stateType == reducibleOpType @a) $ throwError $- Error "bad type"- [ Error ("expected " <> show (reducibleOpType @a)) []- , Error ("got " <> show stateType) []- ]+ Error+ "bad type"+ [ Error ("expected " <> show (reducibleOpType @a)) []+ , Error ("got " <> show stateType) []+ ] pure $ stateFromChunk stateBody data ReducedChunk = ReducedChunk- { rcRef :: UUID- , rcBody :: [Op]+ { rcRef :: UUID+ , rcBody :: [Op] } deriving (Show) -data Patch a = Patch{patchRef :: UUID, patchValue :: a}+data Patch a = Patch {patchRef :: UUID, patchValue :: a} -instance Semigroup a => Semigroup (Patch a) where+instance (Semigroup a) => Semigroup (Patch a) where Patch ref1 a1 <> Patch ref2 a2 = Patch (min ref1 ref2) (a1 <> a2) -patchFromRawOp :: Reducible a => Op -> Patch a-patchFromRawOp op@Op{opId} = Patch- { patchRef = opId- , patchValue = stateFromChunk [op]- }+patchFromRawOp :: (Reducible a) => Op -> Patch a+patchFromRawOp op@Op{opId} =+ Patch{patchRef = opId, patchValue = stateFromChunk [op]} -patchFromChunk :: Reducible a => ReducedChunk -> Patch a+patchFromChunk :: (Reducible a) => ReducedChunk -> Patch a patchFromChunk ReducedChunk{rcRef, rcBody} = Patch{patchRef = rcRef, patchValue = stateFromChunk rcBody} -patchToChunk :: Reducible a => Patch a -> ReducedChunk+patchToChunk :: (Reducible a) => Patch a -> ReducedChunk patchToChunk Patch{patchRef, patchValue} = ReducedChunk{rcRef = patchRef, rcBody = stateBody} where@@ -155,54 +169,63 @@ -- | Base class for typed encoding class Replicated a where- -- | Instances SHOULD implement 'encoding' either as 'objectEncoding' or as- -- 'payloadEncoding'+ {- | Instances SHOULD implement 'encoding' either as 'objectEncoding' or as+ 'payloadEncoding'+ -} encoding :: Encoding a data Encoding a = Encoding- { encodingNewRon- :: forall m- . (ReplicaClock m, MonadState StateFrame m) => a -> m Payload- , encodingFromRon- :: forall m . (MonadE m, MonadState StateFrame m) => Payload -> m a+ { encodingNewRon ::+ forall m.+ (MonadState StateFrame m, ReplicaClock m) =>+ a ->+ m Payload+ , encodingFromRon ::+ forall m.+ (MonadE m, MonadState StateFrame m) =>+ Payload ->+ m a } -- | Encode typed data to a payload with possible addition objects-newRon- :: (Replicated a, ReplicaClock m, MonadState StateFrame m) => a -> m Payload-newRon = encodingNewRon encoding+newRon ::+ (MonadState StateFrame m, ReplicaClock m, Replicated a) => a -> m Payload+newRon = encodingNewRon where Encoding{encodingNewRon} = encoding --- | Decode typed data from a payload.--- The implementation may use other objects in the frame to resolve references.-fromRon :: (MonadE m, Replicated a, MonadState StateFrame m) => Payload -> m a-fromRon = encodingFromRon encoding+{- | Decode typed data from a payload.+The implementation may use other objects in the frame to resolve references.+-}+fromRon :: (MonadE m, MonadState StateFrame m, Replicated a) => Payload -> m a+fromRon = encodingFromRon where Encoding{encodingFromRon} = encoding -- | Standard implementation of 'Replicated' for 'ReplicatedAsObject' types.-objectEncoding :: ReplicatedAsObject a => Encoding a-objectEncoding = Encoding- { encodingNewRon = \a -> do- ObjectRef uuid <- newObject a- pure [AUuid uuid]- , encodingFromRon = objectFromRon $ runReaderT readObject- }+objectEncoding :: (ReplicatedAsObject a) => Encoding a+objectEncoding =+ Encoding+ { encodingNewRon = \a -> do+ ObjectRef uuid <- newObject a+ pure [AUuid uuid]+ , encodingFromRon = objectFromRon $ runReaderT readObject+ } -- | Standard implementation of 'Replicated' for 'ReplicatedAsPayload' types.-payloadEncoding :: ReplicatedAsPayload a => Encoding a-payloadEncoding = Encoding- { encodingNewRon = pure . toPayload- , encodingFromRon = fromPayload- }+payloadEncoding :: (ReplicatedAsPayload a) => Encoding a+payloadEncoding =+ Encoding+ { encodingNewRon = pure . toPayload+ , encodingFromRon = fromPayload+ } --- | Instances of this class are encoded as payload only.------ Law: @'encoding' == 'payloadEncoding'@-class Replicated a => ReplicatedAsPayload a where+{- | Instances of this class are encoded as payload only. +Law: @'encoding' == 'payloadEncoding'@+-}+class (Replicated a) => ReplicatedAsPayload a where -- | Encode data toPayload :: a -> Payload -- | Decode data- fromPayload :: MonadE m => Payload -> m a+ fromPayload :: (MonadE m) => Payload -> m a instance Replicated Int64 where encoding = payloadEncoding @@ -210,7 +233,7 @@ toPayload int = [AInteger int] fromPayload atoms = errorContext "Integer" $ case atoms of [AInteger int] -> pure int- _ -> throwError "Expected Integer"+ _ -> throwError "Expected Integer" instance Replicated UUID where encoding = payloadEncoding @@ -218,7 +241,7 @@ toPayload u = [AUuid u] fromPayload atoms = errorContext "UUID" $ case atoms of [AUuid u] -> pure u- _ -> throwError "Expected UUID"+ _ -> throwError "Expected UUID" instance Replicated Text where encoding = payloadEncoding @@ -226,7 +249,7 @@ toPayload t = [AString t] fromPayload atoms = errorContext "String" $ case atoms of [AString t] -> pure t- _ -> throwError "Expected String"+ _ -> throwError "Expected String" instance Replicated Char where encoding = payloadEncoding @@ -236,38 +259,38 @@ [AString (Text.uncons -> Just (c, ""))] -> pure c _ -> throwError "Expected one-character string" --- | Instances of this class are encoded as objects.--- An enclosing object's payload will be filled with this object's id.------ Law: @'encoding' == 'objectEncoding'@-class (Reducible (Rep a), Replicated a) => ReplicatedAsObject a where+{- | Instances of this class are encoded as objects.+An enclosing object's payload will be filled with this object's id. +Law: @'encoding' == 'objectEncoding'@+-}+class (Reducible (Rep a), Replicated a) => ReplicatedAsObject a where -- | Untyped RON-RDT representation type Rep a -- | Encode data. Write frame and return id.- newObject- :: (ReplicaClock m, MonadState StateFrame m) => a -> m (ObjectRef a)+ newObject ::+ (MonadState StateFrame m, ReplicaClock m) => a -> m (ObjectRef a) -- | Decode data readObject :: (MonadE m, MonadObjectState a m) => m a -objectFromRon :: MonadE m => (ObjectRef a -> m a) -> Payload -> m a+objectFromRon :: (MonadE m) => (ObjectRef a -> m a) -> Payload -> m a objectFromRon handler atoms = errorContext "objectFromRon" $ case atoms of [AUuid uuid] -> handler $ ObjectRef uuid- _ -> throwError "Expected object UUID"+ _ -> throwError "Expected object UUID" -- | Create new 'ObjectFrame' from a value-newObjectFrame- :: (ReplicatedAsObject a, ReplicaClock m) => a -> m (ObjectFrame a)+newObjectFrame ::+ (ReplicaClock m, ReplicatedAsObject a) => a -> m (ObjectFrame a) newObjectFrame a = do (ObjectRef uuid, frame) <- runStateT (newObject a) mempty- pure $ ObjectFrame{uuid, frame}+ pure ObjectFrame{uuid, frame} -getObjectStateChunk- :: forall a m . (MonadE m, MonadObjectState a m) => m (StateChunk (Rep a))+getObjectStateChunk ::+ forall a m. (MonadE m, MonadObjectState a m) => m (StateChunk (Rep a)) getObjectStateChunk = do ObjectRef uuid <- ask frame <- get@@ -275,58 +298,67 @@ liftMaybe "no such object in chunk" $ Map.lookup uuid frame unless (stateType == reducibleOpType @(Rep a)) $ throwError $- Error "bad type"- [ Error ("expected " <> show (reducibleOpType @(Rep a))) []- , Error ("got " <> show stateType) []- ]+ Error+ "bad type"+ [ Error ("expected " <> show (reducibleOpType @(Rep a))) []+ , Error ("got " <> show stateType) []+ ] pure $ StateChunk stateBody -modifyObjectStateChunk- :: forall a b m- . (MonadObjectState a m, ReplicaClock m, MonadE m)- => (StateChunk (Rep a) -> m (b, StateChunk (Rep a))) -> m b+modifyObjectStateChunk ::+ forall a b m.+ (MonadE m, MonadObjectState a m, ReplicaClock m) =>+ (StateChunk (Rep a) -> m (b, StateChunk (Rep a))) ->+ m b modifyObjectStateChunk f = do advanceToObject ObjectRef uuid <- ask chunk <- getObjectStateChunk (a, StateChunk chunk') <- f chunk modify' $- Map.insert uuid $- WireStateChunk{stateType = reducibleOpType @(Rep a), stateBody = chunk'}+ Map.insert+ uuid+ WireStateChunk+ { stateType = reducibleOpType @(Rep a)+ , stateBody = chunk'+ } pure a -modifyObjectStateChunk_- :: (MonadObjectState a m, ReplicaClock m, MonadE m)- => (StateChunk (Rep a) -> m (StateChunk (Rep a))) -> m ()-modifyObjectStateChunk_ f = modifyObjectStateChunk $ \chunk -> do- chunk' <- f chunk- pure ((), chunk')+modifyObjectStateChunk_ ::+ (MonadE m, MonadObjectState a m, ReplicaClock m) =>+ (StateChunk (Rep a) -> m (StateChunk (Rep a))) ->+ m ()+modifyObjectStateChunk_ f =+ modifyObjectStateChunk \chunk -> do+ chunk' <- f chunk+ pure ((), chunk') eqRef :: ObjectRef a -> Payload -> Bool-eqRef (ObjectRef uuid) atoms = case atoms of+eqRef (ObjectRef uuid) = \case [AUuid ref] -> uuid == ref- _ -> False+ _ -> False -eqPayload :: ReplicatedAsPayload a => a -> Payload -> Bool-eqPayload a atoms = toPayload a == atoms+eqPayload :: (ReplicatedAsPayload a) => a -> Payload -> Bool+eqPayload a = (toPayload a ==) pattern ATrue :: Atom-pattern ATrue = AUuid (UUID 0xe36e69000000000 0) -- true+pattern ATrue = AUuid (UUID 0xe36e69000000000 0) -- true pattern AFalse :: Atom-pattern AFalse = AUuid (UUID 0xaa5c37a40000000 0) -- false+pattern AFalse = AUuid (UUID 0xaa5c37a40000000 0) -- false instance Replicated Bool where encoding = payloadEncoding instance ReplicatedAsPayload Bool where toPayload b- | b = [ATrue]+ | b = [ATrue] | otherwise = [AFalse] - fromPayload = errorContext "Bool" . \case- [ATrue] -> pure True- [AFalse] -> pure False- _ -> throwError "Expected single UUID `true` or `false`"+ fromPayload =+ errorContext "Bool" . \case+ [ATrue] -> pure True+ [AFalse] -> pure False+ _ -> throwError "Expected single UUID `true` or `false`" type ObjectStateT b m a = ReaderT (ObjectRef b) (StateT StateFrame m) a @@ -335,66 +367,75 @@ advanceToObject :: (MonadE m, MonadObjectState a m, ReplicaClock m) => m () advanceToObject = do- ObjectRef uuid <- ask- StateChunk chunk <- getObjectStateChunk- advanceToUuid $- maximumDef- uuid- [ max opId $ maximumDef refId $ mapMaybe atomAsUuid payload- | Op{opId, refId, payload} <- chunk- ]+ allObjectTimes <- getAllObjectTimes+ for_ (maximumMay allObjectTimes) advance where- -- | TODO(2019-07-26, cblp) Use lens- atomAsUuid = \case- AUuid u -> Just u- _ -> Nothing+ getAllObjectTimes =+ map (\Event{time} -> timeValue time) . mapMaybe decodeEvent+ <$> getAllObjectUuids -reduceState- :: forall rep- . Reducible rep => StateChunk rep -> StateChunk rep -> StateChunk rep+ getAllObjectUuids = do+ ObjectRef objectId <- ask+ StateChunk chunk <- getObjectStateChunk+ pure $ objectId : foldMap allOpUuids chunk++ allOpUuids Op{opId, refId, payload} =+ opId : refId : [u | AUuid u <- payload]++reduceState ::+ forall rep.+ (Reducible rep) =>+ StateChunk rep ->+ StateChunk rep ->+ StateChunk rep reduceState (StateChunk s1) (StateChunk s2) = StateChunk $ stateToChunk @rep $ stateFromChunk s1 <> stateFromChunk s2 -reduceObjectStates- :: forall a m- . (MonadE m, MonadState StateFrame m, ReplicatedAsObject a)- => NonEmpty (ObjectRef a) -> m (ObjectRef a)+reduceObjectStates ::+ forall a m.+ (MonadE m, MonadState StateFrame m, ReplicatedAsObject a) =>+ NonEmpty (ObjectRef a) ->+ m (ObjectRef a) reduceObjectStates (obj :| objs) = do c :| cs <- for (obj :| objs) $ runReaderT getObjectStateChunk let chunk = foldl' (reduceState @(Rep a)) c cs- oid = minimum [i | ObjectRef i <- obj:objs]+ oid = minimum [i | ObjectRef i <- obj : objs] modify' $ Map.insert oid $ wireStateChunk chunk pure $ ObjectRef oid -rconcat- :: forall a m- . (MonadE m, MonadState StateFrame m, ReplicatedAsObject a)- => NonEmpty UUID -> m a+rconcat ::+ forall a m.+ (MonadE m, MonadState StateFrame m, ReplicatedAsObject a) =>+ NonEmpty UUID ->+ m a rconcat uuids =- errorContext "rconcat" $ do+ errorContext "rconcat" do ObjectRef ref <- reduceObjectStates @a $ ObjectRef <$> uuids fromRon [AUuid ref] -tryFromRon- :: (MonadE m, MonadState StateFrame m, Replicated a)- => Payload -> m (Maybe a)+tryFromRon ::+ (MonadE m, MonadState StateFrame m, Replicated a) =>+ Payload ->+ m (Maybe a) tryFromRon = correct Nothing . fmap Just . fromRon -tryOptionFromRon- :: (MonadE m, MonadState StateFrame m, Replicated a)- => Payload -> m (Maybe a)+tryOptionFromRon ::+ (MonadE m, MonadState StateFrame m, Replicated a) =>+ Payload ->+ m (Maybe a) tryOptionFromRon payload = case payload of Some : payload' -> correct Nothing $ (Just <$> fromRon payload)- `catchError` \e1 ->- (Just <$> fromRon payload')- `catchError` \e2 ->- throwError $ Error "tryOptionFromRon" [e1, e2]+ `catchError` \e1 ->+ (Just <$> fromRon payload')+ `catchError` \e2 ->+ throwError $ Error "tryOptionFromRon" [e1, e2] _ -> tryFromRon payload --- | TODO(2019-08-06, cblp) Remove a year after release--- (the release is planned on 2019-08; removal is planned on 2020-08)+{- | TODO(2019-08-06, cblp) Remove a year after release+(the release is planned on 2019-08; removal is planned on 2020-08)+-} pattern Some :: Atom pattern Some = AUuid (UUID 0xdf3c69000000000 0) {-# DEPRECATED Some "Will be removed soon" #-}@@ -404,17 +445,18 @@ StateChunk chunk <- getObjectStateChunk pure $ stateFromChunk chunk -wireStateChunk :: forall rep . Reducible rep => StateChunk rep -> WireStateChunk+wireStateChunk ::+ forall rep. (Reducible rep) => StateChunk rep -> WireStateChunk wireStateChunk (StateChunk stateBody) = WireStateChunk{stateType = reducibleOpType @rep, stateBody} -- | Run ObjectFrame action-evalObjectState :: Monad m => ObjectFrame b -> ObjectStateT b m a -> m a+evalObjectState :: (Monad m) => ObjectFrame b -> ObjectStateT b m a -> m a evalObjectState ObjectFrame{uuid, frame} action = evalStateT (runReaderT action $ ObjectRef uuid) frame -- | Run ObjectFrame action, starting with an empty frame-evalObjectState_ :: Monad m => StateT StateFrame m a -> m a+evalObjectState_ :: (Monad m) => StateT StateFrame m a -> m a evalObjectState_ action = evalStateT action mempty instance Replicated (ObjectRef a) where
lib/RON/Data/LWW.hs view
@@ -1,7 +1,9 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ImportQualifiedPost #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE RecordWildCards #-}@@ -9,6 +11,7 @@ {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TypeApplications #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-} -- | LWW-per-field RDT module RON.Data.LWW (@@ -21,29 +24,43 @@ zoomField, ) where -import RON.Prelude+import RON.Prelude -import qualified Data.Map.Strict as Map+import Data.Map.Strict qualified as Map -import RON.Data.Internal (MonadObjectState, ObjectStateT, Reducible,- Rep, Replicated, ReplicatedAsObject,- getObjectStateChunk,- modifyObjectStateChunk_, newRon,- reducibleOpType, stateFromChunk,- stateToChunk, tryOptionFromRon)-import RON.Error (MonadE, errorContext)-import RON.Event (ReplicaClock, getEventUuid)-import RON.Semilattice (Semilattice)-import RON.Types (Atom (AUuid), ObjectRef (ObjectRef),- Op (Op, opId, payload, refId),- StateChunk (StateChunk), StateFrame, UUID,- WireStateChunk (WireStateChunk, stateBody, stateType))-import RON.Util (Instance (Instance))-import qualified RON.UUID as UUID+import RON.Data.Internal (+ MonadObjectState,+ ObjectStateT,+ Reducible,+ Rep,+ Replicated,+ ReplicatedAsObject,+ getObjectStateChunk,+ modifyObjectStateChunk_,+ newRon,+ reducibleOpType,+ stateFromChunk,+ stateToChunk,+ tryOptionFromRon,+ )+import RON.Error (MonadE, errorContext)+import RON.Event (ReplicaClock, getEventUuid)+import RON.Semilattice (Semilattice)+import RON.Types (+ Atom (AUuid),+ ObjectRef (..),+ Op (..),+ StateChunk (..),+ StateFrame,+ UUID,+ WireStateChunk (..),+ )+import RON.UUID qualified as UUID+import RON.Util (Instance (Instance)) -- | Last-Write-Wins: select an op with latter event lww :: Op -> Op -> Op-lww = maxOn opId+lww = maxOn (.opId) -- | Untyped LWW. Implementation: a map from 'opRef' to the original op. newtype LwwRep = LwwRep (Map UUID Op)@@ -53,9 +70,10 @@ LwwRep fields1 <> LwwRep fields2 = LwwRep $ Map.unionWith lww fields1 fields2 --- | Laws:--- 1. Idempotent because 'Map.unionWith' is idempotent.--- 2. Commutative because 'lww' is commutative.+{- | Laws:+1. Idempotent because 'Map.unionWith' is idempotent.+2. Commutative because 'lww' is commutative.+-} instance Semilattice LwwRep instance Reducible LwwRep where@@ -74,74 +92,82 @@ lwwType = $(UUID.liftName "lww") -- | Create an LWW object from a list of named fields.-newStruct- :: (MonadState StateFrame m, ReplicaClock m)- => [(UUID, Maybe (Instance Replicated))] -> m UUID+newStruct ::+ (MonadState StateFrame m, ReplicaClock m) =>+ [(UUID, Maybe (Instance Replicated))] ->+ m UUID newStruct fields = do event <- getEventUuid stateBody <- for fields $ \(name, mvalue) -> do payload <- case mvalue of Just (Instance value) -> newRon value- Nothing -> pure []+ Nothing -> pure [] pure $ Op event name payload modify' $ Map.insert event $ wireStateChunk stateBody pure event -- | Decode field value-viewField- :: (Replicated a, MonadE m, MonadState StateFrame m)- => UUID -- ^ Field name- -> StateChunk LwwRep -- ^ LWW object chunk- -> m (Maybe a)+viewField ::+ (MonadE m, MonadState StateFrame m, Replicated a) =>+ -- | Field name+ UUID ->+ -- | LWW object chunk+ StateChunk LwwRep ->+ m (Maybe a) viewField field (StateChunk ops) = errorContext "LWW.viewField" $- maybe (pure Nothing) (tryOptionFromRon . payload) $- maximumMayOn opId $- filter (\Op{refId} -> refId == field) ops+ maybe (pure Nothing) (tryOptionFromRon . (.payload)) $+ maximumMayOn (.opId) $+ filter (\Op{refId} -> refId == field) ops -- | Read field value-readField- :: ( MonadE m- , MonadObjectState struct m- , Rep struct ~ LwwRep- , Replicated field- )- => UUID -- ^ Field name- -> m (Maybe field)+readField ::+ ( MonadE m+ , MonadObjectState struct m+ , Rep struct ~ LwwRep+ , Replicated field+ ) =>+ -- | Field name+ UUID ->+ m (Maybe field) readField field = do stateChunk <- getObjectStateChunk viewField field stateChunk -- | Assign a value to a field-assignField- :: (Replicated a, ReplicaClock m, MonadE m, MonadObjectState struct m)- => UUID -- ^ Field name- -> Maybe a -- ^ Value- -> m ()+assignField ::+ (MonadE m, MonadObjectState struct m, ReplicaClock m, Replicated a) =>+ -- | Field name+ UUID ->+ -- | Value+ Maybe a ->+ m () assignField field mvalue = modifyObjectStateChunk_ $ \(StateChunk ops) -> do let chunk = filter (\Op{refId} -> refId /= field) ops event <- getEventUuid p <- maybe (pure []) newRon mvalue let newOp = Op event field p- pure $ StateChunk $ sortOn refId $ newOp : chunk+ pure $ StateChunk $ sortOn (.refId) $ newOp : chunk -- | Pseudo-lens to an object inside a specified field-zoomField- :: (MonadE m, ReplicatedAsObject struct)- => UUID -- ^ Field name- -> ObjectStateT field m a -- ^ Inner object modifier- -> ObjectStateT struct m a+zoomField ::+ (MonadE m, ReplicatedAsObject struct) =>+ -- | Field name+ UUID ->+ -- | Inner object modifier+ ObjectStateT field m a ->+ ObjectStateT struct m a zoomField field innerModifier = errorContext ("LWW.zoomField" <> show field) $ do StateChunk ops <- getObjectStateChunk let fieldOps = filter (\Op{refId} -> refId == field) ops Op{payload} <- case fieldOps of- [] -> throwError "empty chunk"+ [] -> throwError "empty chunk" [op] -> pure op- _ -> throwError "unreduced state"+ _ -> throwError "unreduced state" fieldObjectId <- errorContext "inner object" $ case payload of [AUuid oid] -> pure oid- _ -> throwError "Expected object UUID"+ _ -> throwError "Expected object UUID" lift $ runReaderT innerModifier $ ObjectRef fieldObjectId
lib/RON/Data/ORSet.hs view
@@ -1,7 +1,9 @@ {-# LANGUAGE ApplicativeDo #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ImportQualifiedPost #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE ScopedTypeVariables #-}@@ -22,7 +24,9 @@ removeObjectIf, removeRef, removeValue,+ setType, zoomItem,+ -- * struct_set addFieldValue, assignField,@@ -38,47 +42,68 @@ zoomFieldObject, ) where -import RON.Prelude+import RON.Prelude -import qualified Data.Map.Strict as Map+import Data.Map.Strict qualified as Map -import RON.Data.Internal (MonadObjectState, ObjectStateT, Reducible,- Rep, Replicated (encoding),- ReplicatedAsObject,- ReplicatedAsPayload (fromPayload),- eqPayload, eqRef, evalObjectState, fromRon,- getObjectState, getObjectStateChunk,- modifyObjectStateChunk_, newObject, newRon,- objectEncoding, rconcat, readObject,- reduceObjectStates, reducibleOpType,- stateFromChunk, stateToChunk)-import RON.Error (MonadE, errorContext, throwErrorText)-import RON.Event (ReplicaClock, getEventUuid)-import RON.Semilattice (Semilattice)-import RON.Types (Atom (AUuid),- ObjectFrame (ObjectFrame, frame, uuid),- ObjectRef (ObjectRef),- Op (Op, opId, payload, refId), Payload,- StateChunk (StateChunk), StateFrame, UUID,- WireStateChunk (WireStateChunk, stateBody, stateType))-import RON.Util (Instance (Instance))-import RON.UUID (pattern Zero)-import qualified RON.UUID as UUID+import RON.Data.Internal (+ MonadObjectState,+ ObjectStateT,+ Reducible,+ Rep,+ Replicated (encoding),+ ReplicatedAsObject,+ ReplicatedAsPayload (fromPayload),+ eqPayload,+ eqRef,+ evalObjectState,+ fromRon,+ getObjectState,+ getObjectStateChunk,+ modifyObjectStateChunk_,+ newObject,+ newRon,+ objectEncoding,+ rconcat,+ readObject,+ reduceObjectStates,+ reducibleOpType,+ stateFromChunk,+ stateToChunk,+ )+import RON.Error (MonadE, errorContext, throwErrorText)+import RON.Event (ReplicaClock, getEventUuid)+import RON.Semilattice (Semilattice)+import RON.Types (+ Atom (AUuid),+ ObjectFrame (..),+ ObjectRef (..),+ Op (..),+ Payload,+ StateChunk (..),+ StateFrame,+ UUID,+ WireStateChunk (..),+ )+import RON.UUID (pattern Zero)+import RON.UUID qualified as UUID+import RON.Util (Instance (..)) --- | Untyped OR-Set.--- Implementation: a map from the itemKey to the original op.--- Each time a value is added, a new item=op is created.--- Deletion of a value replaces all its known items with tombstone ops.+{- | Untyped OR-Set.+Implementation: a map from the itemKey to the original op.+Each time a value is added, a new item=op is created.+Deletion of a value replaces all its known items with tombstone ops.+-} newtype ORSetRep = ORSetRep (Map UUID Op) deriving (Eq, Show) itemKey :: Op -> UUID itemKey Op{opId, refId} = case refId of- Zero -> opId -- alive- _ -> refId -- tombstone+ Zero -> opId -- alive+ _ -> refId -- tombstone preferTombstone :: Op -> Op -> Op-preferTombstone = maxOn refId+preferTombstone = maxOn (.refId) instance Semigroup ORSetRep where ORSetRep set1 <> ORSetRep set2 =@@ -87,18 +112,20 @@ instance Monoid ORSetRep where mempty = ORSetRep mempty --- | Laws:--- 1. Idempotent because 'Map.unionWith' is idempotent.--- 2. Commutative because 'preferTombstone' is commutative.+{- | Laws:+1. Idempotent because 'Map.unionWith' is idempotent.+2. Commutative because 'preferTombstone' is commutative.+-} instance Semilattice ORSetRep instance Reducible ORSetRep where reducibleOpType = setType - stateFromChunk ops = ORSetRep $- Map.fromListWith preferTombstone [(itemKey op, op) | op <- ops]+ stateFromChunk ops =+ ORSetRep $+ Map.fromListWith preferTombstone [(itemKey op, op) | op <- ops] - stateToChunk (ORSetRep set) = sortOn opId $ Map.elems set+ stateToChunk (ORSetRep set) = sortOn (.opId) $ Map.elems set wireStateChunk :: [Op] -> WireStateChunk wireStateChunk stateBody = WireStateChunk{stateType = setType, stateBody}@@ -107,20 +134,22 @@ setType :: UUID setType = $(UUID.liftName "set") --- | Type-directing wrapper for typed OR-Set.--- 'Eq' instance is purely technical, it doesn't use 'Ord', nor 'Hashable',--- so its result may be confusing.+{- | Type-directing wrapper for typed OR-Set.+'Eq' instance is purely technical, it doesn't use 'Ord', nor 'Hashable',+so its result may be confusing.+-} newtype ORSet a = ORSet [a] deriving (Eq, Show)+ {- TODO(2019-07-24, cblp, #102) data family ORSet c a where newtype instance ORSet Ord a = ORSetOrd (Set a) newtype instance ORSet Hashable a = ORSetHash (HashSet a) -} -instance Replicated a => Replicated (ORSet a) where+instance (Replicated a) => Replicated (ORSet a) where encoding = objectEncoding -instance Replicated a => ReplicatedAsObject (ORSet a) where+instance (Replicated a) => ReplicatedAsObject (ORSet a) where type Rep (ORSet a) = ORSetRep newObject (ORSet items) = do@@ -138,7 +167,7 @@ Zero -> do item <- fromRon payload pure $ Just item- _ -> pure Nothing+ _ -> pure Nothing pure . ORSet $ catMaybes mItems -- | XXX Internal. Common implementation of 'addValue' and 'addRef'.@@ -149,37 +178,40 @@ pure $ StateChunk $ stateBody ++ [Op event Zero payload] -- | Encode a value and add a it to the OR-Set-addValue- :: (Replicated a, ReplicaClock m, MonadE m, MonadObjectState (ORSet a) m)- => a -> m ()+addValue ::+ (MonadE m, MonadObjectState (ORSet a) m, ReplicaClock m, Replicated a) =>+ a ->+ m () addValue item = do payload <- newRon item commonAdd payload -addRef- :: (ReplicaClock m, MonadE m, MonadObjectState (ORSet a) m)- => ObjectRef a -> m ()+addRef ::+ (MonadE m, MonadObjectState (ORSet a) m, ReplicaClock m) =>+ ObjectRef a ->+ m () addRef (ObjectRef itemUuid) = commonAdd [AUuid itemUuid] -- | XXX Internal. Common implementation of 'removeValue' and 'removeRef'.-commonRemove- :: (MonadE m, ReplicaClock m, MonadObjectState (ORSet a) m)- => (Payload -> m Bool) -> m ()+commonRemove ::+ (MonadE m, MonadObjectState (ORSet a) m, ReplicaClock m) =>+ (Payload -> m Bool) ->+ m () commonRemove isTarget = modifyObjectStateChunk_ $ \(StateChunk chunk) -> do let state0@(ORSetRep opMap) = stateFromChunk chunk targetEvents <-- fmap catMaybes- $ for (toList opMap) $ \op@Op{refId, payload} ->- case refId of- Zero -> do- t <- isTarget payload- pure $ if t then Just op else Nothing- _ -> pure Nothing- StateChunk <$>- case targetEvents of+ fmap catMaybes $+ for (toList opMap) $ \op@Op{refId, payload} ->+ case refId of+ Zero -> do+ t <- isTarget payload+ pure $ if t then Just op else Nothing+ _ -> pure Nothing+ StateChunk+ <$> case targetEvents of [] -> pure chunk- _ -> do+ _ -> do tombstone <- getEventUuid let patch = [ op{opId = tombstone, refId = observed}@@ -188,9 +220,10 @@ let state' = state0 <> stateFromChunk patch pure $ stateToChunk state' -removeObjectIf- :: (MonadE m, ReplicaClock m, MonadObjectState (ORSet a) m)- => ObjectStateT a m Bool -> m ()+removeObjectIf ::+ (MonadE m, MonadObjectState (ORSet a) m, ReplicaClock m) =>+ ObjectStateT a m Bool ->+ m () removeObjectIf isTarget = commonRemove $ \case AUuid uuid' : _ -> do frame <- get@@ -198,53 +231,60 @@ _ -> pure False -- | Remove an atomic value from the OR-Set-removeValue- :: ( ReplicatedAsPayload a- , MonadE m, ReplicaClock m, MonadObjectState (ORSet a) m- )- => a -> m ()+removeValue ::+ ( MonadE m+ , MonadObjectState (ORSet a) m+ , ReplicaClock m+ , ReplicatedAsPayload a+ ) =>+ a ->+ m () removeValue v = commonRemove $ pure . eqPayload v -- | Remove from a field all values that equal to the specified value.-removeFieldValue- :: ( MonadE m- , MonadObjectState struct m- , ReplicaClock m- , ReplicatedAsPayload a- )- => UUID -- ^ Field name- -> a -- ^ Value- -> m ()+removeFieldValue ::+ ( MonadE m+ , MonadObjectState struct m+ , ReplicaClock m+ , ReplicatedAsPayload a+ ) =>+ -- | Field name+ UUID ->+ -- | Value+ a ->+ m () removeFieldValue field value = removeFieldValueIfP field $ pure . eqPayload value -- | Remove from a field all values that obey the predicate.-removeFieldValueIf- :: ( MonadE m- , MonadObjectState struct m- , ReplicaClock m- , ReplicatedAsPayload a- )- => UUID -- ^ Field name- -> (a -> m Bool)- -> m ()+removeFieldValueIf ::+ ( MonadE m+ , MonadObjectState struct m+ , ReplicaClock m+ , ReplicatedAsPayload a+ ) =>+ -- | Field name+ UUID ->+ (a -> m Bool) ->+ m () removeFieldValueIf field isTarget = removeFieldValueIfP field $ \valuePayload -> do value <- fromPayload valuePayload isTarget value -- | Remove from a field all payloads that obey the predicate.-removeFieldValueIfP- :: (MonadE m, MonadObjectState struct m, ReplicaClock m)- => UUID -- ^ Field name- -> (Payload -> m Bool)- -> m ()+removeFieldValueIfP ::+ (MonadE m, MonadObjectState struct m, ReplicaClock m) =>+ -- | Field name+ UUID ->+ (Payload -> m Bool) ->+ m () removeFieldValueIfP field isTarget = modifyObjectStateChunk_ $ \(StateChunk stateBody) -> do (observedOps, stateBody1) <- partitionM isFieldAliveAndSameValue stateBody removeOps <- for observedOps $ \op@Op{opId = observedEvent} -> do- tombstone <- getEventUuid -- TODO(2019-07-10, cblp) sequential+ tombstone <- getEventUuid -- TODO(2019-07-10, cblp) sequential pure op{opId = tombstone, refId = observedEvent} let stateBody2 = stateBody1 ++ removeOps pure $ StateChunk stateBody2@@ -255,20 +295,24 @@ _ -> pure False -- | Remove an object reference from the OR-Set-removeRef- :: (MonadE m, ReplicaClock m, MonadObjectState (ORSet a) m)- => ObjectRef a -> m ()+removeRef ::+ (MonadE m, MonadObjectState (ORSet a) m, ReplicaClock m) =>+ ObjectRef a ->+ m () removeRef r = commonRemove $ pure . eqRef r -- | Reference to an item inside an 'ORSet'. newtype ORSetItem a = ORSetItem UUID deriving (Show) --- | Go from modification of the whole set to the modification of an item--- object.-zoomItem- :: MonadE m- => ORSetItem item -> ObjectStateT item m a -> ObjectStateT (ORSet item) m a+{- | Go from modification of the whole set to the modification of an item+object.+-}+zoomItem ::+ (MonadE m) =>+ ORSetItem item ->+ ObjectStateT item m a ->+ ObjectStateT (ORSet item) m a zoomItem (ORSetItem key) innerModifier = do ORSetRep opMap <- getObjectState itemValueRef <- case Map.lookup key opMap of@@ -281,66 +325,74 @@ lift $ runReaderT innerModifier $ ObjectRef itemValueRef -- | Find any alive item. If no alive item found, return 'Nothing'.-findAnyAlive- :: (MonadE m, MonadObjectState (ORSet item) m) => m (Maybe (ORSetItem item))+findAnyAlive ::+ (MonadE m, MonadObjectState (ORSet item) m) => m (Maybe (ORSetItem item)) findAnyAlive = do ORSetRep opMap <- getObjectState let aliveItems = [op | op@Op{refId = Zero} <- toList opMap] pure $ case listToMaybe aliveItems of- Nothing -> Nothing+ Nothing -> Nothing Just Op{opId} -> Just $ ORSetItem opId -- | Find any alive item. If no alive item found, report an error.-findAnyAlive'- :: (MonadE m, MonadObjectState (ORSet item) m) => m (ORSetItem item)+findAnyAlive' ::+ (MonadE m, MonadObjectState (ORSet item) m) => m (ORSetItem item) findAnyAlive' = do mx <- findAnyAlive case mx of- Just x -> pure x+ Just x -> pure x Nothing -> throwErrorText "empty set" type ORSetMap k v = ORSet (k, v) --- | Assign a value to a field, deleting all previous (observed) values.--- Assignment of 'Nothing' just deletes all values.-assignField- :: (Replicated a, ReplicaClock m, MonadE m, MonadObjectState struct m)- => UUID -- ^ Field name- -> Maybe a -- ^ Value- -> m ()+{- | Assign a value to a field, deleting all previous (observed) values.+Assignment of 'Nothing' just deletes all values.+-}+assignField ::+ (MonadE m, MonadObjectState struct m, ReplicaClock m, Replicated a) =>+ -- | Field name+ UUID ->+ -- | Value+ Maybe a ->+ m () assignField field mvalue = modifyObjectStateChunk_ $ \(StateChunk stateBody) -> do addOp <- case mvalue of Just value -> do event <- getEventUuid valuePayload <- newRon value- pure $ Just Op- { opId = event- , refId = Zero- , payload = AUuid field : valuePayload- }+ pure $+ Just+ Op+ { opId = event+ , refId = Zero+ , payload = AUuid field : valuePayload+ } Nothing -> pure Nothing let (observedOps, stateBody1) = partition (isAliveField field) stateBody removeOps <- for observedOps $ \op@Op{opId = observedEvent} -> do- tombstone <- getEventUuid -- TODO(2019-07-10, cblp) sequential+ tombstone <- getEventUuid -- TODO(2019-07-10, cblp) sequential pure op{opId = tombstone, refId = observedEvent} let stateBody2 = stateBody1 ++ toList addOp ++ removeOps pure $ StateChunk stateBody2 -addFieldValue- :: (Replicated a, ReplicaClock m, MonadE m, MonadObjectState struct m)- => UUID -- ^ Field name- -> a -- ^ Value- -> m ()+addFieldValue ::+ (MonadE m, MonadObjectState struct m, ReplicaClock m, Replicated a) =>+ -- | Field name+ UUID ->+ -- | Value+ a ->+ m () addFieldValue field value = modifyObjectStateChunk_ $ \(StateChunk stateBody) -> do event <- getEventUuid valuePayload <- newRon value- let addOp = Op- { opId = event- , refId = Zero- , payload = AUuid field : valuePayload- }+ let addOp =+ Op+ { opId = event+ , refId = Zero+ , payload = AUuid field : valuePayload+ } pure $ StateChunk $ stateBody ++ [addOp] isAliveField :: UUID -> Op -> Bool@@ -348,20 +400,26 @@ Op{refId = Zero, payload = AUuid field' : _} -> field == field' _ -> False -filterAliveFieldPayloads- :: UUID -- ^ field- -> [Op] -- ^ state body- -> [Payload] -- ^ value payloads+filterAliveFieldPayloads ::+ -- | field+ UUID ->+ -- | state body+ [Op] ->+ -- | value payloads+ [Payload] filterAliveFieldPayloads field ops = [ valuePayload | Op{refId = Zero, payload = AUuid field' : valuePayload} <- ops , field' == field ] -filterAliveFieldIdsAndPayloads- :: UUID -- ^ field- -> [Op] -- ^ state body- -> [(UUID, Payload)] -- ^ op ids and value payloads+filterAliveFieldIdsAndPayloads ::+ -- | field+ UUID ->+ -- | state body+ [Op] ->+ -- | op ids and value payloads+ [(UUID, Payload)] filterAliveFieldIdsAndPayloads field ops = [ (opId, valuePayload) | Op{opId, refId = Zero, payload = AUuid field' : valuePayload} <- ops@@ -369,95 +427,108 @@ ] -- | Find object field, merge all versions, return 'Nothing' if no versions-getFieldObject- :: (MonadE m, MonadObjectState struct m, ReplicatedAsObject a)- => UUID -- ^ Field name- -> m (Maybe (ObjectRef a))+getFieldObject ::+ (MonadE m, MonadObjectState struct m, ReplicatedAsObject a) =>+ -- | Field name+ UUID ->+ m (Maybe (ObjectRef a)) getFieldObject field = errorContext "ORSet.getFieldObject" $ do StateChunk ops <- getObjectStateChunk let payloads = filterAliveFieldPayloads field ops refs = [ref | AUuid ref : _ <- payloads] case refs of- [] -> pure Nothing- p:ps -> fmap Just $ reduceObjectStates $ fmap ObjectRef $ p :| ps+ [] -> pure Nothing+ p : ps -> fmap Just $ reduceObjectStates $ fmap ObjectRef $ p :| ps -- | Decode field value, merge all versions, return 'Nothing' if no versions-viewField- :: (MonadE m, MonadState StateFrame m, ReplicatedAsObject a)- => UUID -- ^ Field name- -> StateChunk ORSetRep -- ^ ORSet object chunk- -> m (Maybe a)+viewField ::+ (MonadE m, MonadState StateFrame m, ReplicatedAsObject a) =>+ -- | Field name+ UUID ->+ -- | ORSet object chunk+ StateChunk ORSetRep ->+ m (Maybe a) viewField field (StateChunk stateBody) = errorContext "ORSet.viewField" $ do let payloads = filterAliveFieldPayloads field stateBody refs = [ref | AUuid ref : _ <- payloads] case refs of- [] -> pure Nothing- p:ps -> fmap Just . rconcat $ p :| ps+ [] -> pure Nothing+ p : ps -> fmap Just . rconcat $ p :| ps -- | Decode field value, keep last version only-viewFieldLWW- :: (MonadE m, MonadState StateFrame m, Replicated a)- => UUID -- ^ Field name- -> StateChunk ORSetRep -- ^ ORSet object chunk- -> m (Maybe a)+viewFieldLWW ::+ (MonadE m, MonadState StateFrame m, Replicated a) =>+ -- | Field name+ UUID ->+ -- | ORSet object chunk+ StateChunk ORSetRep ->+ m (Maybe a) viewFieldLWW field (StateChunk stateBody) = errorContext "ORSet.viewFieldLWW" $ do- let mPayload =- fmap snd . maximumMayOn fst $- filterAliveFieldIdsAndPayloads field stateBody- case mPayload of- Nothing -> pure Nothing- Just [] -> pure Nothing- Just payload -> Just <$> fromRon payload+ let mPayload =+ fmap snd . maximumMayOn fst $+ filterAliveFieldIdsAndPayloads field stateBody+ case mPayload of+ Nothing -> pure Nothing+ Just [] -> pure Nothing+ Just payload -> Just <$> fromRon payload -- | Decode field value, keep max value only, only for Integer and Float-viewFieldMax- :: (MonadE m, Ord a, ReplicatedAsPayload a)- => UUID -- ^ Field name- -> StateChunk ORSetRep -- ^ ORSet object chunk- -> m (Maybe a)+viewFieldMax ::+ (MonadE m, Ord a, ReplicatedAsPayload a) =>+ -- | Field name+ UUID ->+ -- | ORSet object chunk+ StateChunk ORSetRep ->+ m (Maybe a) viewFieldMax field (StateChunk stateBody) = errorContext "ORSet.viewFieldMax" $- fmap maximumMay $- traverse fromPayload $- filterAliveFieldPayloads field stateBody+ fmap maximumMay $+ traverse fromPayload $+ filterAliveFieldPayloads field stateBody -- | Decode field value, keep min value only, only for Integer and Float-viewFieldMin- :: (MonadE m, Ord a, ReplicatedAsPayload a)- => UUID -- ^ Field name- -> StateChunk ORSetRep -- ^ ORSet object chunk- -> m (Maybe a)+viewFieldMin ::+ (MonadE m, Ord a, ReplicatedAsPayload a) =>+ -- | Field name+ UUID ->+ -- | ORSet object chunk+ StateChunk ORSetRep ->+ m (Maybe a) viewFieldMin field (StateChunk stateBody) = errorContext "ORSet.viewFieldMin" $- fmap minimumMay $- traverse fromPayload $- filterAliveFieldPayloads field stateBody+ fmap minimumMay $+ traverse fromPayload $+ filterAliveFieldPayloads field stateBody -- | Decode field value, keep all versions-viewFieldSet- :: (MonadE m, MonadState StateFrame m, Replicated a)- => UUID -- ^ Field name- -> StateChunk ORSetRep -- ^ ORSet object chunk- -> m [a]+viewFieldSet ::+ (MonadE m, MonadState StateFrame m, Replicated a) =>+ -- | Field name+ UUID ->+ -- | ORSet object chunk+ StateChunk ORSetRep ->+ m [a] viewFieldSet field (StateChunk stateBody) = errorContext "ORSet.viewFieldSet" $- traverse fromRon $- filterAliveFieldPayloads field stateBody+ traverse fromRon $+ filterAliveFieldPayloads field stateBody -- | Pseudo-lens to an object inside a specified field-zoomFieldObject- :: forall a field m struct- . ( MonadE m- , ReplicaClock m- , ReplicatedAsObject field- , ReplicatedAsObject struct- )- => UUID -- ^ Field name- -> ObjectStateT field m a -- ^ Inner object modifier- -> ObjectStateT struct m a+zoomFieldObject ::+ forall a field m struct.+ ( MonadE m+ , ReplicaClock m+ , ReplicatedAsObject field+ , ReplicatedAsObject struct+ ) =>+ -- | Field name+ UUID ->+ -- | Inner object modifier+ ObjectStateT field m a ->+ ObjectStateT struct m a zoomFieldObject field innerModifier = errorContext ("ORSet.zoomFieldObject " <> show field) $ do (StateChunk stateBody) <- getObjectStateChunk@@ -466,14 +537,16 @@ | AUuid objectId : _ <- filterAliveFieldPayloads field stateBody ] object <- case objectIds of- [] -> ObjectRef <$> getEventUuid -- create empty object- oid:oids -> reduceObjectStates @field $ fmap ObjectRef $ oid :| oids+ [] -> ObjectRef <$> getEventUuid -- create empty object+ oid : oids ->+ reduceObjectStates @field $ fmap ObjectRef $ oid :| oids lift $ runReaderT innerModifier object -- | Create an ORSet object from a list of named fields.-newStruct- :: (MonadState StateFrame m, ReplicaClock m)- => [(UUID, Instance Replicated)] -> m UUID+newStruct ::+ (MonadState StateFrame m, ReplicaClock m) =>+ [(UUID, Instance Replicated)] ->+ m UUID newStruct fields = do objectId <- getEventUuid stateBody <-@@ -484,10 +557,10 @@ modify' $ Map.insert objectId $ wireStateChunk stateBody pure objectId -partitionM :: Applicative m => (a -> m Bool) -> [a] -> m ([a], [a])+partitionM :: (Applicative m) => (a -> m Bool) -> [a] -> m ([a], [a]) partitionM f = \case [] -> pure ([], [])- x:xs -> do+ x : xs -> do res <- f x (as, bs) <- partitionM f xs pure ([x | res] ++ as, [x | not res] ++ bs)
lib/RON/Data/RGA.hs view
@@ -1,7 +1,10 @@+{-# LANGUAGE BlockArguments #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ImportQualifiedPost #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE RecordWildCards #-}@@ -11,8 +14,8 @@ {-# LANGUAGE TypeFamilies #-} -- | Replicated Growable Array (RGA)-module RON.Data.RGA- ( RGA (..),+module RON.Data.RGA (+ RGA (..), RgaRep, RgaString, edit,@@ -32,16 +35,18 @@ rgaType, RON.Data.RGA.toList, toText,- )+) where -import Data.Algorithm.Diff (Diff (Both, First, Second), getGroupedDiffBy)-import qualified Data.HashMap.Strict as HashMap+import Data.Algorithm.Diff (PolyDiff (Both, First, Second), getGroupedDiffBy)+import Data.Bifunctor (second)+import Data.HashMap.Strict qualified as HashMap import Data.Map.Strict ((!?))-import qualified Data.Map.Strict as Map-import qualified Data.Text as Text-import RON.Data.Internal- ( MonadObjectState,+import Data.Map.Strict qualified as Map+import Data.Text qualified as Text++import RON.Data.Internal (+ MonadObjectState, ReducedChunk (ReducedChunk, rcBody, rcRef), Reducible, Rep,@@ -62,293 +67,303 @@ stateFromChunk, stateToChunk, toPayload,- )+ ) import RON.Error (MonadE, errorContext, throwErrorText) import RON.Event (ReplicaClock, getEventUuid, getEventUuids) import RON.Prelude import RON.Semilattice (Semilattice)-import RON.Types- ( ObjectRef (ObjectRef),+import RON.Types (+ ObjectRef (ObjectRef), Op (Op, opId, payload, refId), StateChunk (StateChunk), StateFrame, UUID, WireStateChunk (WireStateChunk, stateBody, stateType),- )-import RON.UUID (uuidVersion, pattern Zero)-import qualified RON.UUID as UUID+ )+import RON.UUID (pattern Zero)+import RON.UUID qualified as UUID import RON.Util.Word (ls60, pattern B11) {-# ANN module ("HLint: ignore Reduce duplication" :: String) #-} --- | opId = vertex id--- refId:--- 0 = value is alive,--- _ = tombstone event, value is backup for undo--- payload: the value+{- | opId = vertex id+ refId:+ 0 = value is alive,+ _ = tombstone event, value is backup for undo+ payload: the value+-} type Vertex = Op data VertexListItem- = VertexListItem- { itemValue :: Vertex,- itemNext :: Maybe UUID- }- deriving (Eq, Show)+ = VertexListItem+ { itemValue :: Vertex+ , itemNext :: Maybe UUID+ }+ deriving (Eq, Show) data VertexList- = VertexList- { listHead :: UUID,- listItems :: HashMap UUID VertexListItem- }- deriving (Eq, Show)+ = VertexList+ { listHead :: UUID+ , listItems :: HashMap UUID VertexListItem+ }+ deriving (Eq, Show) instance Semigroup VertexList where- (<>) = merge+ (<>) = merge +instance Semilattice VertexList+ vertexListToOps :: VertexList -> [Vertex]-vertexListToOps v@VertexList {..} = go listHead listItems+vertexListToOps v@VertexList{..} = go listHead listItems where go root items =- let VertexListItem {..} =- HashMap.lookupDefault- ( error $ unlines- $ ["Cannot find vertex id", show root, "in array"]- ++ map show (HashMap.toList items)- ++ ["Original array is", show v]- )- root- items- rest = case itemNext of- Just next -> go next (HashMap.delete root items)- Nothing -> []- in itemValue : rest+ let VertexListItem{..} =+ HashMap.lookupDefault+ ( error $+ unlines $+ ["Cannot find vertex id", show root, "in array"]+ ++ map show (HashMap.toList items)+ ++ ["Original array is", show v]+ )+ root+ items+ rest = case itemNext of+ Just next -> go next (HashMap.delete root items)+ Nothing -> []+ in itemValue : rest vertexListFromOps :: [Vertex] -> Maybe VertexList vertexListFromOps = foldr go mempty where- go v@Op {opId} vlist =- Just $ VertexList {listHead = opId, listItems = vlist'}+ go v@Op{opId} vlist = Just VertexList{listHead = opId, listItems = vlist'} where- item itemNext = VertexListItem {itemValue = v, itemNext}+ item itemNext = VertexListItem{itemValue = v, itemNext} vlist' = case vlist of- Nothing -> HashMap.singleton opId (item Nothing)- Just VertexList {listHead, listItems} ->- HashMap.insert opId (item $ Just listHead) listItems+ Nothing -> HashMap.singleton opId (item Nothing)+ Just VertexList{listHead, listItems} ->+ HashMap.insert opId (item $ Just listHead) listItems -- | Untyped RGA newtype RgaRep = RgaRep (Maybe VertexList)- deriving (Eq, Monoid, Semigroup, Semilattice, Show)+ deriving (Eq, Monoid, Semigroup, Semilattice, Show) data PatchSet- = PatchSet- { psPatches :: Map UUID VertexList,- -- ^ the key is the parent event, the value is a non-empty VertexList- psRemovals :: Map UUID UUID- -- ^ the key is the target event, the value is the tombstone event- }- deriving (Eq, Show)+ = PatchSet+ { psPatches :: Map UUID VertexList+ -- ^ the key is the parent event, the value is a non-empty VertexList+ , psRemovals :: Map UUID UUID+ -- ^ the key is the target event, the value is the tombstone event+ }+ deriving (Eq, Show) instance Semigroup PatchSet where- rga1 <> rga2 = reapplyPatchSet $ preMerge rga1 rga2+ rga1 <> rga2 = reapplyPatchSet $ preMerge rga1 rga2 preMerge :: PatchSet -> PatchSet -> PatchSet-preMerge (PatchSet p1 r1) (PatchSet p2 r2) = PatchSet- { psPatches = Map.unionWith (<>) p1 p2,- psRemovals = Map.unionWith max r1 r2- }+preMerge (PatchSet p1 r1) (PatchSet p2 r2) =+ PatchSet+ { psPatches = Map.unionWith (<>) p1 p2+ , psRemovals = Map.unionWith max r1 r2+ } instance Monoid PatchSet where- mempty = PatchSet {psPatches = mempty, psRemovals = mempty}+ mempty = PatchSet{psPatches = mempty, psRemovals = mempty} patchSetFromRawOp :: Op -> PatchSet-patchSetFromRawOp op@Op {opId, refId, payload} = case payload of- [] ->- -- remove op- mempty {psRemovals = Map.singleton refId opId}- _ : _ ->- -- append op- mempty- { psPatches = Map.singleton- refId- VertexList- { listHead = opId,- listItems = HashMap.singleton- opId- VertexListItem- { itemValue = op {refId = Zero},- itemNext = Nothing- }+patchSetFromRawOp op@Op{opId, refId, payload} = case payload of+ [] ->+ -- remove op+ mempty{psRemovals = Map.singleton refId opId}+ _ : _ ->+ -- append op+ mempty+ { psPatches =+ Map.singleton+ refId+ VertexList+ { listHead = opId+ , listItems =+ HashMap.singleton+ opId+ VertexListItem+ { itemValue = op{refId = Zero}+ , itemNext = Nothing+ } }- }+ } patchSetFromChunk :: ReducedChunk -> PatchSet-patchSetFromChunk ReducedChunk {rcRef, rcBody} =- case uuidVersion $ UUID.split rcRef of- B11 ->- -- derived event -- rm-patch compatibility- foldMap patchSetFromRawOp rcBody- _ ->- -- patch- case vertexListFromOps rcBody of- Just patch -> mempty {psPatches = Map.singleton rcRef patch}- Nothing -> mempty+patchSetFromChunk ReducedChunk{rcRef, rcBody} =+ case (UUID.split rcRef).uuidVersion of+ B11 ->+ -- derived event -- rm-patch compatibility+ foldMap patchSetFromRawOp rcBody+ _ ->+ -- patch+ case vertexListFromOps rcBody of+ Just patch -> mempty{psPatches = Map.singleton rcRef patch}+ Nothing -> mempty instance Reducible RgaRep where-- reducibleOpType = rgaType+ reducibleOpType = rgaType - stateFromChunk = RgaRep . vertexListFromOps+ stateFromChunk = RgaRep . vertexListFromOps - stateToChunk (RgaRep rga) = maybe [] vertexListToOps rga+ stateToChunk (RgaRep rga) = maybe [] vertexListToOps rga - applyPatches rga (patches, ops) =- bimap id patchSetToChunks . reapplyPatchSetToState rga- $ foldMap patchSetFromChunk patches <> foldMap patchSetFromRawOp ops+ applyPatches rga (patches, ops) =+ second patchSetToChunks . reapplyPatchSetToState rga $+ foldMap patchSetFromChunk patches <> foldMap patchSetFromRawOp ops - reduceUnappliedPatches (patches, ops) =- patchSetToChunks . reapplyPatchSet- $ foldMap patchSetFromChunk patches <> foldMap patchSetFromRawOp ops+ reduceUnappliedPatches (patches, ops) =+ patchSetToChunks . reapplyPatchSet $+ foldMap patchSetFromChunk patches <> foldMap patchSetFromRawOp ops patchSetToChunks :: PatchSet -> Unapplied-patchSetToChunks PatchSet {psPatches, psRemovals} =- ( [ ReducedChunk {..}- | (rcRef, vertices) <- Map.assocs psPatches,- let rcBody = vertexListToOps vertices- ],- [ Op {opId = tombstone, refId, payload = []}+patchSetToChunks PatchSet{psPatches, psRemovals} =+ ( [ ReducedChunk{..}+ | (rcRef, vertices) <- Map.assocs psPatches+ , let rcBody = vertexListToOps vertices+ ]+ , [ Op{opId = tombstone, refId, payload = []} | (refId, tombstone) <- Map.assocs psRemovals- ]- )+ ]+ ) reapplyPatchSet :: PatchSet -> PatchSet reapplyPatchSet ps =- continue ps [reapplyPatchesToOtherPatches, reapplyRemovalsToPatches]+ continue ps [reapplyPatchesToOtherPatches, reapplyRemovalsToPatches] reapplyPatchSetToState :: RgaRep -> PatchSet -> (RgaRep, PatchSet) reapplyPatchSetToState rga ps =- continue (rga, ps) [reapplyPatchesToState, reapplyRemovalsToState]+ continue (rga, ps) [reapplyPatchesToState, reapplyRemovalsToState] continue :: x -> [x -> Maybe x] -> x continue x fs = case asum $ map ($ x) fs of- Nothing -> x- Just x' -> continue x' fs+ Nothing -> x+ Just x' -> continue x' fs reapplyPatchesToState :: (RgaRep, PatchSet) -> Maybe (RgaRep, PatchSet)-reapplyPatchesToState (RgaRep rstate, ps@PatchSet {..}) = case rstate of- Just VertexList {listHead = targetHead, listItems = targetItems} ->- asum- [ do- targetItems' <- applyPatch parent patch targetItems- pure- ( RgaRep . Just $ VertexList targetHead targetItems',- ps {psPatches = Map.delete parent psPatches}- )- | (parent, patch) <- Map.assocs psPatches- ]- Nothing -> do- -- rstate is empty => only virtual 0 node exists- -- => we can apply only 0 patch- patch <- psPatches !? Zero- pure (RgaRep $ Just patch, ps {psPatches = Map.delete Zero psPatches})+reapplyPatchesToState (RgaRep rstate, ps@PatchSet{..}) = case rstate of+ Just VertexList{listHead = targetHead, listItems = targetItems} ->+ asum+ [ do+ targetItems' <- applyPatch parent patch targetItems+ pure+ ( RgaRep . Just $ VertexList targetHead targetItems'+ , ps{psPatches = Map.delete parent psPatches}+ )+ | (parent, patch) <- Map.assocs psPatches+ ]+ Nothing -> do+ -- rstate is empty => only virtual 0 node exists+ -- => we can apply only 0 patch+ patch <- psPatches !? Zero+ pure (RgaRep $ Just patch, ps{psPatches = Map.delete Zero psPatches}) reapplyPatchesToOtherPatches :: PatchSet -> Maybe PatchSet-reapplyPatchesToOtherPatches ps@PatchSet {..} =- asum- [ do- targetItems' <- applyPatch parent patch targetItems- pure- ps- { psPatches =- Map.insert targetParent (VertexList targetHead targetItems')- $ Map.delete parent psPatches- }- | (parent, patch) <- Map.assocs psPatches,- (targetParent, targetPatch) <- Map.assocs psPatches,- parent /= targetParent,- let VertexList targetHead targetItems = targetPatch- ]+reapplyPatchesToOtherPatches ps@PatchSet{..} =+ asum+ [ do+ targetItems' <- applyPatch parent patch targetItems+ pure+ ps+ { psPatches =+ Map.insert+ targetParent+ (VertexList targetHead targetItems')+ . Map.delete parent+ $ psPatches+ }+ | (parent, patch) <- Map.assocs psPatches+ , (targetParent, targetPatch) <- Map.assocs psPatches+ , parent /= targetParent+ , let VertexList targetHead targetItems = targetPatch+ ] -applyPatch- :: UUID- -> VertexList- -> HashMap UUID VertexListItem- -> Maybe (HashMap UUID VertexListItem)+applyPatch ::+ UUID ->+ VertexList ->+ HashMap UUID VertexListItem ->+ Maybe (HashMap UUID VertexListItem) applyPatch parent patch targetItems = case parent of- Zero ->- error "chunk with zero ref must be considered a state, not a patch"- _ -> do- item@VertexListItem {itemNext} <- HashMap.lookup parent targetItems- let VertexList next' newItems = case itemNext of- Nothing -> patch- Just next -> VertexList next targetItems <> patch- let item' = item {itemNext = Just next'}- pure $ HashMap.insert parent item' targetItems <> newItems+ Zero ->+ error "chunk with zero ref must be considered a state, not a patch"+ _ -> do+ item@VertexListItem{itemNext} <- HashMap.lookup parent targetItems+ let VertexList next' newItems = case itemNext of+ Nothing -> patch+ Just next -> VertexList next targetItems <> patch+ let item' = item{itemNext = Just next'}+ pure $ HashMap.insert parent item' targetItems <> newItems reapplyRemovalsToState :: (RgaRep, PatchSet) -> Maybe (RgaRep, PatchSet)-reapplyRemovalsToState (RgaRep rstate, ps@PatchSet {..}) = do- VertexList {listHead = targetHead, listItems = targetItems} <- rstate- asum- [ do- targetItems' <- applyRemoval parent tombstone targetItems- pure- ( RgaRep . Just $ VertexList targetHead targetItems',- ps {psRemovals = Map.delete parent psRemovals}- )- | (parent, tombstone) <- Map.assocs psRemovals- ]+reapplyRemovalsToState (RgaRep rstate, ps@PatchSet{..}) = do+ VertexList{listHead = targetHead, listItems = targetItems} <- rstate+ asum+ [ do+ targetItems' <- applyRemoval parent tombstone targetItems+ pure+ ( RgaRep . Just $ VertexList targetHead targetItems'+ , ps{psRemovals = Map.delete parent psRemovals}+ )+ | (parent, tombstone) <- Map.assocs psRemovals+ ] reapplyRemovalsToPatches :: PatchSet -> Maybe PatchSet-reapplyRemovalsToPatches PatchSet {..} =- asum- [ do- targetItems' <- applyRemoval parent tombstone targetItems- pure PatchSet- { psRemovals = Map.delete parent psRemovals,- psPatches =- Map.insert- targetParent- (VertexList targetHead targetItems')- psPatches- }- | (parent, tombstone) <- Map.assocs psRemovals,- (targetParent, targetPatch) <- Map.assocs psPatches,- let VertexList targetHead targetItems = targetPatch- ]+reapplyRemovalsToPatches PatchSet{..} =+ asum+ [ do+ targetItems' <- applyRemoval parent tombstone targetItems+ pure+ PatchSet+ { psRemovals = Map.delete parent psRemovals+ , psPatches =+ Map.insert+ targetParent+ (VertexList targetHead targetItems')+ psPatches+ }+ | (parent, tombstone) <- Map.assocs psRemovals+ , (targetParent, targetPatch) <- Map.assocs psPatches+ , let VertexList targetHead targetItems = targetPatch+ ] -applyRemoval- :: UUID- -> UUID- -> HashMap UUID VertexListItem- -> Maybe (HashMap UUID VertexListItem)+applyRemoval ::+ UUID ->+ UUID ->+ HashMap UUID VertexListItem ->+ Maybe (HashMap UUID VertexListItem) applyRemoval parent tombstone targetItems = do- item@VertexListItem {itemValue = v@Op {refId}} <-- HashMap.lookup parent targetItems- let item' = item {itemValue = v {refId = max refId tombstone}}- pure $ HashMap.insert parent item' targetItems+ item@VertexListItem{itemValue = v@Op{refId}} <-+ HashMap.lookup parent targetItems+ let item' = item{itemValue = v{refId = max refId tombstone}}+ pure $ HashMap.insert parent item' targetItems merge :: VertexList -> VertexList -> VertexList merge v1 v2 =- fromMaybe (error "merge of non-empty lists cannot be empty")- $ vertexListFromOps- $ (merge' `on` vertexListToOps) v1 v2+ fromMaybe (error "merge of non-empty lists cannot be empty") $+ vertexListFromOps $+ (merge' `on` vertexListToOps) v1 v2 merge' :: [Vertex] -> [Vertex] -> [Vertex] merge' [] vs2 = vs2 merge' vs1 [] = vs1 merge' w1@(v1 : vs1) w2@(v2 : vs2) =- case compare e1 e2 of- LT -> v2 : merge' w1 vs2- GT -> v1 : merge' vs1 w2- EQ -> mergeVertices : merge' vs1 vs2+ case compare e1 e2 of+ LT -> v2 : merge' w1 vs2+ GT -> v1 : merge' vs1 w2+ EQ -> mergeVertices : merge' vs1 vs2 where- Op {opId = e1, refId = tombstone1, payload = p1} = v1- Op {opId = e2, refId = tombstone2, payload = p2} = v2+ Op{opId = e1, refId = tombstone1, payload = p1} = v1+ Op{opId = e2, refId = tombstone2, payload = p2} = v2 -- priority of deletion- mergeVertices = Op- { opId = e1,- refId = max tombstone1 tombstone2,- payload = maxOn length p1 p2- }+ mergeVertices =+ Op+ { opId = e1+ , refId = max tombstone1 tombstone2+ , payload = maxOn length p1 p2+ } -- | Name-UUID to use as RGA type marker. rgaType :: UUID@@ -356,199 +371,208 @@ -- | Typed RGA newtype RGA a = RGA [a]- deriving (Eq, Show)--instance Replicated a => Replicated (RGA a) where encoding = objectEncoding+ deriving (Eq, Show) -instance Replicated a => ReplicatedAsObject (RGA a) where+instance (Replicated a) => Replicated (RGA a) where encoding = objectEncoding - type Rep (RGA a) = RgaRep+instance (Replicated a) => ReplicatedAsObject (RGA a) where+ type Rep (RGA a) = RgaRep - newObject (RGA items) = do- oid <- getEventUuid- vertexIds <- getEventUuids $ ls60 $ genericLength items- ops <-- for (zip items vertexIds) $ \(item, vertexId) -> do- payload <- newRon item- pure $ Op vertexId Zero payload- modify' $ Map.insert oid $ wireStateChunk ops- pure $ ObjectRef oid+ newObject (RGA items) = do+ oid <- getEventUuid+ vertexIds <- getEventUuids $ ls60 $ genericLength items+ ops <-+ for (zip items vertexIds) $ \(item, vertexId) -> do+ payload <- newRon item+ pure $ Op vertexId Zero payload+ modify' $ Map.insert oid $ wireStateChunk ops+ pure $ ObjectRef oid - readObject = do- StateChunk stateBody <- getObjectStateChunk- mItems <-- for stateBody $ \Op {refId, payload} -> case refId of- Zero -> Just <$> fromRon payload- _ -> pure Nothing- pure . RGA $ catMaybes mItems+ readObject = do+ StateChunk stateBody <- getObjectStateChunk+ mItems <-+ for stateBody $ \Op{refId, payload} -> case refId of+ Zero -> Just <$> fromRon payload+ _ -> pure Nothing+ pure . RGA $ catMaybes mItems --- | Replace content of the RGA throug introducing changes detected by--- 'getGroupedDiffBy'.-edit- :: ( ReplicatedAsPayload a,- ReplicaClock m,- MonadE m,- MonadObjectState (RGA a) m- )- => [a]- -> m ()+{- | Replace content of the RGA throug introducing changes detected by+'getGroupedDiffBy'.+-}+edit ::+ ( MonadE m+ , MonadObjectState (RGA a) m+ , ReplicaClock m+ , ReplicatedAsPayload a+ ) =>+ [a] ->+ m () edit newItems =- modifyObjectStateChunk_ $ \chunk@(StateChunk stateBody) -> do- let newItems' = [Op Zero Zero $ toPayload item | item <- newItems]- -- TODO(2019-04-17, #59, cblp) replace 'toPayload' with 'newRon' and- -- relax constraint on 'a' from 'ReplicatedAsPayload' to- -- 'Replicated'- let diff = getGroupedDiffBy eqAliveOnPayload stateBody newItems'- (stateBody', Last lastEvent) <-- runWriterT . fmap fold . for diff $ \case- First removed -> for removed $ \case- op@Op {refId = Zero} -> do- -- not deleted yet- -- TODO(2018-11-03, #15, cblp) get sequential ids- tombstone <- getEventUuid- tell . Last $ Just tombstone- pure op {refId = tombstone}- op ->- -- deleted already- pure op- Both v _ -> pure v- Second added -> for added $ \op -> do- -- TODO(2018-11-03, #15, cblp) get sequential ids- opId <- getEventUuid- tell . Last $ Just opId- pure op {opId}- pure $ case lastEvent of- Nothing -> chunk- Just _ -> StateChunk stateBody'+ modifyObjectStateChunk_ $ \chunk@(StateChunk stateBody) -> do+ let newItems' = [Op Zero Zero $ toPayload item | item <- newItems]+ -- TODO(2019-04-17, #59, cblp) replace 'toPayload' with 'newRon' and+ -- relax constraint on 'a' from 'ReplicatedAsPayload' to+ -- 'Replicated'+ let diff = getGroupedDiffBy eqAliveOnPayload stateBody newItems'+ (stateBody', Last lastEvent) <-+ runWriterT $+ fold <$> for diff \case+ First removed -> do+ tombstones <- getEventUuids $ genericLength removed+ for (zip removed tombstones) \case+ (op@Op{refId = Zero}, tombstone) -> do+ -- not deleted yet+ tell . Last $ Just tombstone+ pure op{refId = tombstone}+ (op, _) ->+ -- deleted already+ pure op+ Both v _ -> pure v+ Second added -> do+ opIds <- getEventUuids $ genericLength added+ for (zip added opIds) \(op, opId) -> do+ tell . Last $ Just opId+ pure op{opId}+ pure $ case lastEvent of+ Nothing -> chunk+ Just _ -> StateChunk stateBody' where eqAliveOnPayload- Op {refId = Zero, payload = p1}- Op {refId = Zero, payload = p2} =- p1 == p2+ Op{refId = Zero, payload = p1}+ Op{refId = Zero, payload = p2} =+ p1 == p2 eqAliveOnPayload _ _ = False -- | Speciaization of 'edit' for 'Text'-editText- :: (ReplicaClock m, MonadE m, MonadObjectState RgaString m) => Text -> m ()+editText ::+ (MonadE m, MonadObjectState RgaString m, ReplicaClock m) => Text -> m () editText = edit . Text.unpack --- | Speciaization of 'RGA' to 'Char'.--- This is the recommended way to store a string.+{- | Speciaization of 'RGA' to 'Char'.+This is the recommended way to store a string.+-} type RgaString = RGA Char -- | Create an RGA from a list-newFromList- :: (Replicated a, MonadState StateFrame m, ReplicaClock m)- => [a]- -> m (ObjectRef (RGA a))+newFromList ::+ (MonadState StateFrame m, ReplicaClock m, Replicated a) =>+ [a] ->+ m (ObjectRef (RGA a)) newFromList = newObject . RGA -- | Create an 'RgaString' from a text-newFromText- :: (MonadState StateFrame m, ReplicaClock m)- => Text- -> m (ObjectRef RgaString)+newFromText ::+ (MonadState StateFrame m, ReplicaClock m) =>+ Text ->+ m (ObjectRef RgaString) newFromText = newFromList . Text.unpack getAliveIndices :: (MonadE m, MonadObjectState (RGA a) m) => m [UUID] getAliveIndices = do- StateChunk stateBody <- getObjectStateChunk- pure [opId | Op {opId, refId = Zero} <- stateBody]+ StateChunk stateBody <- getObjectStateChunk+ pure [opId | Op{opId, refId = Zero} <- stateBody] -- | Read elements from RGA-getList :: (Replicated a, MonadE m, MonadObjectState (RGA a) m) => m [a]+getList :: (MonadE m, MonadObjectState (RGA a) m, Replicated a) => m [a] getList = coerce <$> readObject -- | Read characters from 'RgaString' getText :: (MonadE m, MonadObjectState RgaString m) => m Text getText = Text.pack <$> getList --- | Insert a sequence of elements after the specified position.--- Position is identified by 'UUID'. 'Nothing' means the beginning.-insert- :: (Replicated a, MonadE m, MonadObjectState (RGA a) m, ReplicaClock m)- => [a]- -> Maybe UUID -- ^ position- -> m ()+{- | Insert a sequence of elements after the specified position.+Position is identified by 'UUID'. 'Nothing' means the beginning.+-}+insert ::+ (MonadE m, MonadObjectState (RGA a) m, ReplicaClock m, Replicated a) =>+ [a] ->+ -- | position+ Maybe UUID ->+ m () insert [] _ = pure () insert items mPosition =- modifyObjectStateChunk_ $ \(StateChunk stateBody) -> do- vertexIds <- getEventUuids $ ls60 $ genericLength items- ops <-- for (zip items vertexIds) $ \(item, vertexId) -> do- payload <- newRon item- pure $ Op vertexId Zero payload- stateBody' <-- case mPosition of- Nothing -> pure $ ops <> stateBody- Just position -> findAndInsertAfter position ops stateBody- pure $ StateChunk stateBody'+ modifyObjectStateChunk_ $ \(StateChunk stateBody) -> do+ vertexIds <- getEventUuids $ ls60 $ genericLength items+ ops <-+ for (zip items vertexIds) $ \(item, vertexId) -> do+ payload <- newRon item+ pure $ Op vertexId Zero payload+ stateBody' <-+ case mPosition of+ Nothing -> pure $ ops <> stateBody+ Just position -> findAndInsertAfter position ops stateBody+ pure $ StateChunk stateBody' where findAndInsertAfter pos newOps = go where go = \case- [] -> throwErrorText "Position not found"- op@Op {opId} : ops- | opId == pos -> pure $ op : newOps ++ ops- | otherwise -> (op :) <$> go ops+ [] -> throwErrorText "Position not found"+ op@Op{opId} : ops+ | opId == pos -> pure $ op : newOps ++ ops+ | otherwise -> (op :) <$> go ops -insertAtBegin- :: (Replicated a, MonadE m, MonadObjectState (RGA a) m, ReplicaClock m)- => [a]- -> m ()+insertAtBegin ::+ (MonadE m, MonadObjectState (RGA a) m, ReplicaClock m, Replicated a) =>+ [a] ->+ m () insertAtBegin items = insert items Nothing -insertAfter- :: (Replicated a, MonadE m, MonadObjectState (RGA a) m, ReplicaClock m)- => [a]- -> UUID -- ^ position- -> m ()+insertAfter ::+ (MonadE m, MonadObjectState (RGA a) m, ReplicaClock m, Replicated a) =>+ [a] ->+ -- | position+ UUID ->+ m () insertAfter items = insert items . Just --- | Insert a text after the specified position.--- Position is identified by 'UUID'. 'Nothing' means the beginning.-insertText- :: (ReplicaClock m, MonadE m, MonadObjectState RgaString m)- => Text- -> Maybe UUID -- ^ position- -> m ()+{- | Insert a text after the specified position.+Position is identified by 'UUID'. 'Nothing' means the beginning.+-}+insertText ::+ (MonadE m, MonadObjectState RgaString m, ReplicaClock m) =>+ Text ->+ -- | position+ Maybe UUID ->+ m () insertText = insert . Text.unpack -insertTextAtBegin- :: (ReplicaClock m, MonadE m, MonadObjectState RgaString m) => Text -> m ()+insertTextAtBegin ::+ (MonadE m, MonadObjectState RgaString m, ReplicaClock m) => Text -> m () insertTextAtBegin text = insertText text Nothing -insertTextAfter- :: (ReplicaClock m, MonadE m, MonadObjectState RgaString m)- => Text- -> UUID -- ^ position- -> m ()+insertTextAfter ::+ (MonadE m, MonadObjectState RgaString m, ReplicaClock m) =>+ Text ->+ -- | position+ UUID ->+ m () insertTextAfter text = insertText text . Just -- | Record a removal of a specific item-remove- :: (MonadE m, MonadObjectState (RGA a) m, ReplicaClock m)- => UUID -- ^ position- -> m ()+remove ::+ (MonadE m, MonadObjectState (RGA a) m, ReplicaClock m) =>+ -- | position+ UUID ->+ m () remove position =- errorContext "RGA.remove"- $ errorContext ("position = " <> show position)- $ modifyObjectStateChunk_- $ \(StateChunk stateBody) -> do- event <- getEventUuid- stateBody' <- findAndTombstone event stateBody- pure $ StateChunk stateBody'+ errorContext "RGA.remove" $+ errorContext ("position = " <> show position) $+ modifyObjectStateChunk_ $+ \(StateChunk stateBody) -> do+ event <- getEventUuid+ stateBody' <- findAndTombstone event stateBody+ pure $ StateChunk stateBody' where findAndTombstone event = go where go = \case- [] -> throwErrorText "Position not found"- op@Op {opId} : ops- | opId == position -> pure $ op {refId = event} : ops- | otherwise -> (op :) <$> go ops+ [] -> throwErrorText "Position not found"+ op@Op{opId} : ops+ | opId == position -> pure $ op{refId = event} : ops+ | otherwise -> (op :) <$> go ops wireStateChunk :: [Op] -> WireStateChunk-wireStateChunk stateBody = WireStateChunk {stateType = rgaType, stateBody}+wireStateChunk stateBody = WireStateChunk{stateType = rgaType, stateBody} toList :: RGA a -> [a] toList (RGA xs) = xs
lib/RON/Data/Time.hs view
@@ -1,28 +1,30 @@ {-# OPTIONS -Wno-orphans #-}- {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} -- | 'Day' instances module RON.Data.Time (Day) where -import RON.Prelude+import RON.Prelude -import Data.Time (Day, fromGregorian, toGregorian)+import Data.Time (Day, fromGregorian, toGregorian) -import RON.Data (Replicated (..), ReplicatedAsPayload (..),- payloadEncoding)-import RON.Types (Atom (..))+import RON.Data (+ Replicated (..),+ ReplicatedAsPayload (..),+ payloadEncoding,+ )+import RON.Types (Atom (..)) instance Replicated Day where encoding = payloadEncoding instance ReplicatedAsPayload Day where- toPayload- = (\(y, m, d) ->- map AInteger [fromIntegral y, fromIntegral m, fromIntegral d])- . toGregorian+ toPayload x =+ let (y, m, d) = toGregorian x+ in map AInteger [fromIntegral y, fromIntegral m, fromIntegral d] fromPayload = \case- [AInteger y, AInteger m, AInteger d] -> pure $- fromGregorian (fromIntegral y) (fromIntegral m) (fromIntegral d)+ [AInteger y, AInteger m, AInteger d] ->+ pure $+ fromGregorian (fromIntegral y) (fromIntegral m) (fromIntegral d) _ -> throwError "bad Day"
lib/RON/Data/VersionVector.hs view
@@ -1,4 +1,6 @@+{-# LANGUAGE ImportQualifiedPost #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE TemplateHaskell #-}@@ -8,36 +10,64 @@ -- | Version Vector module RON.Data.VersionVector ( VersionVector,+ VV (..),+ lookup,+ mkVV,+ serializeVV,+ (·≻),+ (·≼), ) where -import RON.Prelude--import Data.Hashable (hashWithSalt)-import qualified Data.Map.Strict as Map+import RON.Prelude hiding (lookup) -import RON.Data.Internal (Reducible, Rep, Replicated (encoding),- ReplicatedAsObject, getObjectState,- newObject, objectEncoding, readObject,- reducibleOpType, stateFromChunk,- stateToChunk)-import RON.Event (getEventUuid)-import RON.Semilattice (Semilattice)-import RON.Types (ObjectRef (ObjectRef), Op (Op, opId), UUID (UUID), WireStateChunk (WireStateChunk, stateBody, stateType))-import qualified RON.UUID as UUID+import Data.Hashable (hashWithSalt)+import Data.Map.Strict qualified as Map -type Origin = Word64+import RON.Data.Internal (+ Reducible,+ Rep,+ Replicated (encoding),+ ReplicatedAsObject,+ getObjectState,+ newObject,+ objectEncoding,+ readObject,+ reducibleOpType,+ stateFromChunk,+ stateToChunk,+ )+import RON.Event (+ Event (Event),+ Replica,+ Time,+ encodeEvent,+ getEventUuid,+ unsafeDecodeEvent,+ )+import RON.Event qualified+import RON.Semilattice (Semilattice, (≼))+import RON.Text.Serialize (serializePayload)+import RON.Types (+ Atom (AUuid),+ ObjectRef (ObjectRef),+ Op (Op, opId),+ UUID (UUID),+ WireStateChunk (WireStateChunk, stateBody, stateType),+ )+import RON.UUID qualified as UUID opTime :: Op -> Word64 opTime Op{opId = UUID time _} = time -opOrigin :: Op -> Word64-opOrigin Op{opId = UUID _ origin} = origin+-- | Assume opId is event+opReplica :: Op -> Replica+opReplica Op{opId} = (unsafeDecodeEvent opId).replica latter :: Op -> Op -> Op latter = maxOn opTime -- | Version Vector type. May be used both in typed and untyped contexts.-newtype VersionVector = VersionVector (Map Origin Op)+newtype VersionVector = VersionVector (Map Replica Op) deriving (Eq, Show) instance Hashable VersionVector where@@ -49,16 +79,30 @@ instance Monoid VersionVector where mempty = VersionVector mempty --- | Laws:--- 1. Idempotent because 'Map.unionWith' is idempotent.--- 2. Commutative because 'latter' is commutative.-instance Semilattice VersionVector+{- | Laws:+1. Idempotent because 'Map.unionWith' is idempotent.+2. Commutative because 'latter' is commutative.+-}+instance Semilattice VersionVector where+ VersionVector a ≼ VersionVector b =+ case leftJoin (Map.assocs a) (Map.assocs b) of+ Nothing -> False+ Just joined -> and [opTime opA <= opTime opB | (opA, opB) <- joined]+ where+ -- leftJoin :: Ord k => [(k, v1)] -> [(k, v2)] -> Maybe [(v1, v2)]+ leftJoin [] _ = Just []+ leftJoin (_ : _) [] = Nothing+ leftJoin ((xk, xv) : xs') ((yk, yv) : ys') =+ case compare xk yk of+ LT -> Nothing -- x must present but doesn't+ EQ -> ((xv, yv) :) <$> leftJoin xs' ys' -- intersection+ GT -> leftJoin xs' ys' -- skip extra y instance Reducible VersionVector where reducibleOpType = vvType stateFromChunk ops =- VersionVector $ Map.fromListWith latter [(opOrigin op, op) | op <- ops]+ VersionVector $ Map.fromListWith latter [(opReplica op, op) | op <- ops] stateToChunk (VersionVector vv) = Map.elems vv @@ -82,3 +126,45 @@ pure $ ObjectRef oid readObject = getObjectState++lookup :: Replica -> VersionVector -> Maybe Op+lookup replica (VersionVector vv) = Map.lookup replica vv++-- | Simplified version vector+newtype VV = VV (Map Replica Time)+ deriving (Show)++instance Semigroup VV where+ VV a <> VV b = VV $ Map.unionWith max a b++instance Monoid VV where+ mempty = VV Map.empty++-- | Assuming all UUIDs are events.+mkVV :: [UUID] -> VV+mkVV uuids =+ VV $+ Map.fromList+ [ (replica, time)+ | uuid <- uuids+ , let Event{replica, time} = unsafeDecodeEvent uuid+ ]++unVV :: VV -> [UUID]+unVV (VV vv) =+ [encodeEvent Event{replica, time} | (replica, time) <- Map.assocs vv]++(·≻) :: UUID -> VV -> Bool+uuid ·≻ VV vv =+ maybe True (time >) $ Map.lookup replica vv+ where+ Event{replica, time} = unsafeDecodeEvent uuid++(·≼) :: UUID -> VV -> Bool+uuid ·≼ VV vv =+ maybe False (time <=) $ Map.lookup replica vv+ where+ Event{replica, time} = unsafeDecodeEvent uuid++serializeVV :: VV -> ByteStringL+serializeVV = serializePayload . map AUuid . unVV
+ lib/RON/Experimental/Data.hs view
@@ -0,0 +1,129 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE ImportQualifiedPost #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}++module RON.Experimental.Data (+ AsAtom (..),+ AsAtoms (..),+ BaseType (..),+ ReplicatedObject (..),+ baseTypeId,+) where++import RON.Prelude++import RON.Error (MonadE, throwErrorText)+import RON.Event (ReplicaClock)+import RON.Store.Class (MonadStore)+import RON.Types (+ Atom (AString, AUuid),+ ObjectRef (..),+ OpenFrame,+ UUID,+ )+import RON.Types.Experimental (Ref (..))+import RON.UUID qualified as UUID++-- | Basic RON type.+data BaseType = LWW | Set++baseTypeId :: BaseType -> UUID+baseTypeId LWW = $(UUID.liftName "lww")+baseTypeId Set = $(UUID.liftName "set")++-- | Any type that may be encoded as a RON object in whole.+class ReplicatedObject a where+ -- | RON representation type+ baseType :: BaseType+ baseType = Set++ encodeObject ::+ (MonadStore m, ReplicaClock m) =>+ -- | Object id+ UUID ->+ a ->+ m ()++ decodeObject ::+ (MonadE m) =>+ -- | Object id+ UUID ->+ -- | Object ops+ OpenFrame ->+ m a++class AsAtom a where+ toAtom :: a -> Atom+ fromAtom :: (MonadE m) => Atom -> m a++instance AsAtom Atom where+ toAtom = id+ fromAtom = pure++instance AsAtom UUID where+ toAtom = AUuid++ fromAtom = \case+ AUuid u -> pure u+ a -> throwErrorText $ "Expected UUID atom, got " <> show a++instance AsAtom Text where+ toAtom = AString++ fromAtom = \case+ AString t -> pure t+ a -> throwErrorText $ "Expected string atom, got " <> show a++class AsAtoms a where+ toAtoms :: a -> [Atom]+ fromAtoms :: (MonadE m) => [Atom] -> m a++instance AsAtoms [Atom] where+ toAtoms = id+ fromAtoms = pure++instance AsAtoms Atom where+ toAtoms a = [a]++ fromAtoms = \case+ [a] -> pure a+ as -> throwErrorText $ "Expected 1 atom, got " <> show (length as)++instance AsAtoms UUID where+ toAtoms u = [AUuid u]++ fromAtoms as = do+ a <- fromAtoms as+ case a of+ AUuid u -> pure u+ _ -> throwErrorText $ "Expected UUID, got " <> show a++instance AsAtoms Text where+ toAtoms a = [toAtom a]+ fromAtoms = fromAtoms @Atom >=> fromAtom @Text++instance AsAtoms (ObjectRef a) where+ toAtoms (ObjectRef uuid) = toAtoms uuid+ fromAtoms = fmap ObjectRef . fromAtoms++instance (AsAtom head, AsAtoms tail) => AsAtoms (head, tail) where+ toAtoms (head, tail) = toAtom head : toAtoms tail++ fromAtoms = \case+ [] -> throwErrorText "Expected some atoms, got none"+ head : tail -> (,) <$> fromAtom head <*> fromAtoms tail++instance AsAtoms (Ref a) where+ toAtoms Ref{object, path} = AUuid object : path++ fromAtoms = \case+ [] -> throwErrorText "Expected some atoms, got none"+ AUuid object : path -> pure Ref{object, path}+ a : _ -> throwErrorText $ "Expected UUID, got " <> show a
+ lib/RON/Experimental/Data/ORSet.hs view
@@ -0,0 +1,145 @@+{-# LANGUAGE DisambiguateRecordFields #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ImportQualifiedPost #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ViewPatterns #-}+{-# OPTIONS_GHC -Wno-orphans #-}++module RON.Experimental.Data.ORSet (+ ORSet,+ ORMap,+ add,+ add_,+ decode,+ empty,+ getDecode,+ lookupLww,+ lookupLwwThrow,+ lookupLwwDecode,+ lookupLwwDecodeThrow,+ lookupSet,+) where++import RON.Prelude++import Data.Map.Strict qualified as Map+import Data.Text.Lazy qualified as TextL+import Data.Text.Lazy.Encoding qualified as TextL++import Data.List (stripPrefix)+import RON.Error (MonadE, liftMaybe)+import RON.Event (ReplicaClock, advanceToUuid, getEventUuid)+import RON.Experimental.Data (+ AsAtom,+ AsAtoms,+ fromAtoms,+ toAtom,+ toAtoms,+ )+import RON.Store.Class (MonadStore, appendPatch, loadWholeObjectLog)+import RON.Text.Serialize (serializeAtom)+import RON.Types (Op (..), OpenFrame, Payload, UUID)+import RON.Types.Experimental (Patch (..), Ref (..))++import RON.Experimental.Data.ORSet.Type (ORMap, ORSet (..))++decode :: (Applicative f) => UUID -> OpenFrame -> f (ORSet a)+decode objectId ops =+ pure+ . ORSet+ $ Map.fromListWith+ (maxOn fst)+ [ (itemId, (opId, payload))+ | Op{opId, refId, payload} <- ops+ , opId /= objectId+ , let+ itemId =+ case payload of+ [] -> refId -- tombstone+ _ : _ -> opId -- add+ ]++-- | Add value to the set. Return the reference to the set item.+add ::+ (AsAtoms item, MonadStore m, ReplicaClock m) =>+ Ref (ORSet item) ->+ item ->+ m UUID+add (Ref object path) value = do+ advanceToUuid object+ opId <- getEventUuid+ appendPatch+ Patch+ { object+ , log =+ Op{opId, refId = object, payload = path ++ toAtoms value} :| []+ }+ pure opId++{- |+ Add value to the set or map.++ @add_ :: Ref (ORSet a) -> a -> m ()@+ @add_ :: Ref (ORMap k v) -> (k, v) -> m ()@+-}+add_ ::+ (AsAtoms item, MonadStore m, ReplicaClock m) =>+ Ref (ORSet item) ->+ item ->+ m ()+add_ ref = void . add ref++-- | Get items from database and decode+getDecode :: (AsAtoms a, MonadE m, MonadStore m) => Ref (ORSet a) -> m [a]+getDecode (Ref object pre) = do+ -- TODO loadObjectLog object (PayloadPrefix pre)+ ops <- loadWholeObjectLog object mempty+ let alivePayloads =+ filter (not . null) $+ map snd $+ toList $+ Map.fromListWith+ (maxOn fst)+ [ (itemId, (opId, payload'))+ | Op{opId, refId, payload} <- ops+ , (itemId, payload') <-+ case payload of+ [] -> pure (refId, payload)+ (stripPrefix pre -> Just suf) ->+ pure (opId, suf)+ _ -> []+ ]+ traverse fromAtoms alivePayloads++lookupLww :: (AsAtom k) => k -> ORMap k v -> Maybe Payload+lookupLww key (ORSet s) =+ snd+ <$> maximumMayOn+ fst+ [(item, value) | (item, k : value) <- Map.elems s, k == toAtom key]++-- | Like 'lookupLww' but also decode payload.+lookupLwwDecode ::+ (AsAtom k, AsAtoms v, MonadE m) => k -> ORMap k v -> m (Maybe v)+lookupLwwDecode key = traverse fromAtoms . lookupLww key++lookupLwwThrow :: (AsAtom k, MonadE m) => k -> ORMap k v -> m Payload+lookupLwwThrow key obj =+ liftMaybe ("key " <> showAtom key <> " must present") $ lookupLww key obj+ where+ showAtom = TextL.toStrict . TextL.decodeUtf8 . serializeAtom . toAtom++-- | Like 'lookupLwwDecode' but assert that key exists.+lookupLwwDecodeThrow :: (AsAtom k, AsAtoms v, MonadE m) => k -> ORMap k v -> m v+lookupLwwDecodeThrow key = lookupLwwThrow key >=> fromAtoms++empty :: ORSet a+empty = ORSet Map.empty++lookupSet :: (AsAtom k, AsAtoms v, MonadE m) => k -> ORMap k v -> m [v]+lookupSet key (ORSet s) =+ traverse+ fromAtoms+ [value | (_item, k : value) <- Map.elems s, k == toAtom key]
+ lib/RON/Experimental/Data/ORSet/Type.hs view
@@ -0,0 +1,16 @@+module RON.Experimental.Data.ORSet.Type where++import RON.Prelude++import RON.Types (Payload, UUID)++{- | Observed-Remove Set.+Implementation: a map from the itemId to the original op.+Each time a value is added, a new item=op is created.+Deletion of a value replaces all its known items with tombstone ops.+Tombstone is an op with empty payload (even without prefix) referencing item.+-}+newtype ORSet a = ORSet (Map UUID (UUID, Payload))+ deriving (Eq, Show)++type ORMap k v = ORSet (k, v)
+ lib/RON/Store.hs view
@@ -0,0 +1,68 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++module RON.Store (+ MonadStore (..),+ loadSubObjectLog,+ newObject,+ readObject,+) where++import RON.Prelude++import Data.List (stripPrefix)+import RON.Data.VersionVector (VV)+import RON.Error (MonadE, errorContext)+import RON.Event (ReplicaClock, getEventUuid)+import RON.Experimental.Data (+ ReplicatedObject,+ baseType,+ baseTypeId,+ decodeObject,+ encodeObject,+ )+import RON.Store.Class (MonadStore (..))+import RON.Types (Op (..))+import RON.Types.Experimental (Patch (..), Ref (..))++newObject ::+ forall a m.+ (MonadStore m, ReplicaClock m, ReplicatedObject a) =>+ a ->+ m (Ref a)+newObject a = do+ objectId <- getEventUuid+ let typeId = baseTypeId $ baseType @a+ let initOp = Op{opId = objectId, refId = typeId, payload = []}+ appendPatch $ Patch objectId $ initOp :| []+ encodeObject objectId a+ pure $ Ref objectId []++-- | Nothing if object doesn't exist in the replica.+readObject ::+ (MonadE m, MonadStore m, ReplicatedObject a, Typeable a) =>+ Ref a ->+ m (Maybe a)+readObject object@(Ref objectId _) =+ errorContext ("readObject " <> show object) do+ ops <- loadSubObjectLog object mempty+ case ops of+ [] -> pure Nothing+ _ : _ -> fmap Just $ decodeObject objectId $ sortOn (.opId) ops++loadSubObjectLog ::+ (MonadE m, MonadStore m, Typeable a) => Ref a -> VV -> m [Op]+loadSubObjectLog object@(Ref objectId path) version =+ errorContext ("loadSubObjectLog " <> show object) do+ ops <- loadWholeObjectLog objectId version+ pure+ [ op{payload = payload'}+ | op@Op{opId, payload} <- ops+ , opId /= objectId+ , Just payload' <- [stripPrefix path payload]+ ]
+ lib/RON/Store/Class.hs view
@@ -0,0 +1,27 @@+module RON.Store.Class (MonadStore (..)) where++import RON.Prelude++import RON.Data.VersionVector (VV)+import RON.Types (Op, UUID)+import RON.Types.Experimental (Patch)++class (Monad m) => MonadStore m where+ -- | Get list of all object ids in the database.+ listObjects :: m [UUID]++ -- | Append a sequence of operations to an object.+ appendPatch :: Patch -> m ()++ {- | Get all RON-object logs split by replicas.+ Replicas order is not guaranteed.+ Implementation SHOULD return object creation op.+ -}+ loadWholeObjectLog ::+ -- | Object id+ UUID ->+ -- | Base version. To get object logs from the beginning, pass 'mempty'.+ VV ->+ m [Op]++-- TODO loadObjectLog filtered in DB
+ lib/RON/Store/FS.hs view
@@ -0,0 +1,225 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}++module RON.Store.FS (+ module RON.Store,+ Handle,+ Store,+ debugDump,+ fetchUpdates,+ newHandle,+ newHandleWithReplica,+ runStore,+) where++import RON.Prelude++import Control.Concurrent (MVar, newMVar, withMVar)+import Control.Concurrent.STM (+ TChan,+ atomically,+ dupTChan,+ newBroadcastTChanIO,+ writeTChan,+ )+import Control.Exception (throwIO)+import Data.Bits (shiftL)+import Data.ByteString qualified as BS+import Data.ByteString.Lazy qualified as BSL+import Data.ByteString.Lazy.Char8 qualified as BSLC+import Data.Foldable (find)+import Data.IORef (newIORef)+import Network.Info (MAC (MAC), getNetworkInterfaces, mac)+import System.Directory (+ createDirectoryIfMissing,+ doesDirectoryExist,+ listDirectory,+ makeAbsolute,+ )+import System.FileLock (SharedExclusive (Exclusive), tryLockFile)+import System.FilePath ((</>))+import System.Random.TF (newTFGen)+import System.Random.TF.Instances (random)++import RON.Data.VersionVector (VV, (·≼))+import RON.Epoch (EpochClock, getCurrentEpochTime, runEpochClock)+import RON.Error (+ Error (..),+ MonadE,+ errorContext,+ liftEitherString,+ throwErrorText,+ tryIO,+ )+import RON.Event (+ OriginVariety (ApplicationSpecific),+ Replica,+ ReplicaClock,+ getEventUuid,+ mkReplica,+ )+import RON.Store (MonadStore (..))+import RON.Text.Parse (parseOpenFrame)+import RON.Text.Serialize.Experimental (serializeOpenFrame)+import RON.Types (Op (..), UUID)+import RON.Types.Experimental (Patch (..))+import RON.UUID qualified as UUID+import RON.Util.Word (Word60, leastSignificant60)++-- | Store handle (uses the “Handle pattern”).+data Handle = Handle+ { clock :: IORef Word60+ , dataDir :: FilePath+ , onObjectChanged :: TChan UUID+ {- ^ A channel of changes in the database.+ This is a broadcast channel, so you MUST NOT read from it directly,+ call 'fetchUpdates' to read from derived channel instead.+ -}+ , opLock :: MVar ()+ , replica :: Replica+ }++newtype Store a = Store (ExceptT Error (ReaderT Handle EpochClock) a)+ deriving+ (Applicative, Functor, Monad, MonadError Error, MonadIO, ReplicaClock)++instance MonadStore Store where+ listObjects =+ errorContext "Store.listObjects" do+ Handle{dataDir} <- Store ask+ objectDirs <-+ tryIO do+ exists <- doesDirectoryExist dataDir+ if exists then listDirectoryDirs dataDir else pure []+ traverse uuidFromFileName objectDirs++ appendPatch = appendPatchFS++ loadWholeObjectLog = loadWholeObjectLogFS++askObjectLogsDir :: (MonadReader Handle m) => UUID -> m FilePath+askObjectLogsDir objectId = do+ Handle{dataDir} <- ask+ pure $ dataDir </> uuidToFileName objectId </> "log"++getObjectPatches :: UUID -> Store [FilePath]+getObjectPatches objectId = do+ objectLogsDir <- Store $ askObjectLogsDir objectId+ objectExists <- tryIO $ doesDirectoryExist objectLogsDir+ if objectExists then tryIO $ listDirectory objectLogsDir else pure []++loadWholeObjectLogFS :: UUID -> VV -> Store [Op]+loadWholeObjectLogFS objectId version = do+ objectLogsDir <- Store $ askObjectLogsDir objectId+ patchNames <- getObjectPatches objectId+ fold . catMaybes <$> for patchNames \patchName -> do+ patchTimestamp <- uuidFromFileName patchName+ if patchTimestamp ·≼ version then+ pure Nothing+ else do+ let patchFile = objectLogsDir </> patchName+ patchContent <- tryIO $ BS.readFile patchFile+ patch <-+ liftEitherString $ parseOpenFrame $ BSL.fromStrict patchContent+ pure $ Just patch++appendPatchFS :: Patch -> Store ()+appendPatchFS Patch{object, log} = do+ Handle{dataDir, onObjectChanged} <- Store ask+ let objectLogsDir = dataDir </> uuidToFileName object </> "log"+ tryIO $ createDirectoryIfMissing True objectLogsDir+ patchVersion <- getEventUuid+ let patchFile = objectLogsDir </> uuidToFileName patchVersion+ tryIO $ BSL.writeFile patchFile $ serializeOpenFrame $ toList log+ tryIO $ atomically $ writeTChan onObjectChanged object++-- | Run a 'Store' action+runStore :: Handle -> Store a -> IO a+runStore h@Handle{replica, clock, opLock} (Store action) = do+ res <-+ withMVar opLock \_ ->+ runEpochClock replica clock $ (`runReaderT` h) $ runExceptT action+ either throwIO pure res++{- | Create new storage handle.+Uses MAC address for replica id or generates a random one.+-}+newHandle :: FilePath -> IO (Maybe Handle)+newHandle dataDir = do+ macAddress <- getMacAddress+ replicaId <-+ case macAddress of+ Just macAddress' -> pure macAddress'+ Nothing -> fst . random <$> newTFGen+ newHandleWithReplica dataDir $ leastSignificant60 replicaId++newHandleWithReplica :: FilePath -> Word60 -> IO (Maybe Handle)+newHandleWithReplica dataDir' replicaId = do+ dataDir <- makeAbsolute dataDir'+ let appLockFile = dataDir </> "applock"+ mLock <- tryLockFile appLockFile Exclusive+ case mLock of+ Nothing -> pure Nothing+ Just _ -> do+ time <- getCurrentEpochTime+ clock <- newIORef time+ onObjectChanged <- newBroadcastTChanIO+ opLock <- newMVar ()+ let replica = mkReplica ApplicationSpecific replicaId+ pure $ Just Handle{..}++getMacAddress :: IO (Maybe Word64)+getMacAddress =+ do+ macAddress <- getMac+ pure $ decodeMac <$> macAddress+ where+ getMac = find (/= minBound) . map mac <$> getNetworkInterfaces+ decodeMac (MAC b5 b4 b3 b2 b1 b0) =+ (fromIntegral b5 `shiftL` 40)+ + (fromIntegral b4 `shiftL` 32)+ + (fromIntegral b3 `shiftL` 24)+ + (fromIntegral b2 `shiftL` 16)+ + (fromIntegral b1 `shiftL` 8)+ + fromIntegral b0++uuidFromFileName :: (MonadE m) => FilePath -> m UUID+uuidFromFileName name =+ maybe+ ( throwErrorText $+ "UUID.decodeBase32: file name "+ <> show name+ <> " is not a valid UUID"+ )+ pure+ (UUID.decodeBase32 name)++uuidToFileName :: UUID -> FilePath+uuidToFileName = UUID.encodeBase32++debugDump :: FilePath -> IO ()+debugDump dataDir = do+ objectDirs <- do+ exists <- doesDirectoryExist dataDir+ if exists then listDirectory dataDir else pure []+ for_ (sort objectDirs) \objectDir -> do+ isDir <- doesDirectoryExist $ dataDir </> objectDir+ when isDir do+ let logsDir = dataDir </> objectDir </> "log"+ logs <- listDirectory logsDir+ for_ (sort logs) \logName -> do+ let logPath = logsDir </> logName+ BSL.putStr =<< BSL.readFile logPath+ BSLC.putStrLn ""++fetchUpdates :: Handle -> IO (TChan UUID)+fetchUpdates Handle{onObjectChanged} = atomically $ dupTChan onObjectChanged++listDirectoryDirs :: FilePath -> IO [FilePath]+listDirectoryDirs dir =+ listDirectory dir >>= filterM (\name -> doesDirectoryExist (dir </> name))
+ lib/RON/Store/Test.hs view
@@ -0,0 +1,76 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DuplicateRecordFields #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedLabels #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++module RON.Store.Test (emptyDB, runStoreSim) where++import RON.Prelude++import Control.Lens (at, non, (<>=))+import Data.Generics.Labels ()+import Data.Map.Strict ((!?))+import Data.Map.Strict qualified as Map+import Data.Sequence qualified as Seq++import RON.Data.VersionVector ((·≼))+import RON.Error (Error (..), liftMaybe)+import RON.Event (+ OriginVariety (ApplicationSpecific),+ Replica,+ ReplicaClock,+ mkReplica,+ )+import RON.Event.Simulation (+ ReplicaSimT,+ runNetworkSimT,+ runReplicaSimT,+ )+import RON.Store (MonadStore (..))+import RON.Types (Op (..), UUID)+import RON.Types.Experimental (Patch (..))++newtype Object = Object {logs :: Map Replica (Seq Op)}+ deriving (Eq, Generic, Show)++type TestDB = Map UUID Object++emptyDB :: TestDB+emptyDB = Map.empty++newtype StoreSim a = StoreSim (StateT TestDB (ReplicaSimT (Either Error)) a)+ deriving (Applicative, Functor, Monad, MonadError Error, ReplicaClock)++runStoreSim :: TestDB -> StoreSim a -> Either Error (a, TestDB)+runStoreSim db (StoreSim action) =+ runNetworkSimT $ runReplicaSimT thisReplicaId $ runStateT action db++thisReplicaId :: Replica+thisReplicaId = mkReplica ApplicationSpecific 2020++instance MonadStore StoreSim where+ listObjects = StoreSim $ gets Map.keys++ appendPatch Patch{object, log} =+ StoreSim $ atObject . #logs . atReplica <>= Seq.fromList (toList log)+ where+ atObject = at object . non emptyObject+ atReplica = at thisReplicaId . non Seq.empty++ loadWholeObjectLog objectId version = do+ db <- StoreSim get+ Object{logs} <- liftMaybe "object not found" $ db !? objectId+ pure $+ fold+ [ filter (not . isKnown) $ toList @Seq replicaLog+ | replicaLog <- Map.elems logs+ ]+ where+ isKnown Op{opId} = opId ·≼ version++emptyObject :: Object+emptyObject = Object{logs = Map.empty}
ron-rdt.cabal view
@@ -1,53 +1,71 @@-cabal-version: 2.2+cabal-version: 2.2+name: ron-rdt+version: 0.11+bug-reports: https://github.com/ff-notes/ron/issues+category: Distributed Systems, Protocol, Database+copyright: 2018-2019, 2025-2026 Yuriy Syrovetskiy+homepage: https://github.com/ff-notes/ron+license: BSD-3-Clause+license-file: LICENSE+maintainer: Yuriy Syrovetskiy <haskell@cblp.su>+synopsis: Replicated Data Types (RON-RDT)+description:+ Replicated Object Notation (RON), data types (RDT), and RON-Schema+ .+ Examples: https://github.com/ff-notes/ron/tree/master/examples -name: ron-rdt-version: 0.10+build-type: Simple+extra-source-files: CHANGELOG.md -bug-reports: https://github.com/ff-notes/ron/issues-category: Distributed Systems, Protocol, Database-copyright: 2018-2019 Yuriy Syrovetskiy-homepage: https://github.com/ff-notes/ron-license: BSD-3-Clause-license-file: LICENSE-maintainer: Yuriy Syrovetskiy <haskell@cblp.su>-synopsis: Replicated Data Types (RON-RDT)+common language+ build-depends: base >=4.10 && <4.21+ default-extensions:+ NoFieldSelectors+ NoImplicitPrelude+ StrictData -description:- Replicated Object Notation (RON), data types (RDT), and RON-Schema- .- Examples: https://github.com/ff-notes/ron/tree/master/examples+ default-language: GHC2021 -build-type: Simple+library+ import: language -extra-source-files:- CHANGELOG.md+ -- global+ build-depends:+ , bytestring+ , containers+ , Diff+ , directory+ , filelock+ , filepath+ , generic-lens+ , hashable+ , lens+ , mtl+ , network-info+ , stm+ , text+ , tf-random+ , time+ , unordered-containers -common language- build-depends: base >= 4.10 && < 4.13, integer-gmp- default-extensions: MonadFailDesugaring NoImplicitPrelude StrictData- default-language: Haskell2010+ -- project+ build-depends: ron >=0.9+ exposed-modules:+ RON.Data+ RON.Data.CT+ RON.Data.GTree+ RON.Data.LWW+ RON.Data.ORSet+ RON.Data.RGA+ RON.Data.Time+ RON.Data.VersionVector+ RON.Experimental.Data+ RON.Experimental.Data.ORSet+ RON.Experimental.Data.ORSet.Type+ RON.Store+ RON.Store.Class+ RON.Store.FS+ RON.Store.Test -library- import: language- build-depends:- -- global- containers,- Diff < 0.4,- hashable,- mtl,- text,- time,- transformers,- unordered-containers,- -- project- ron >= 0.9,- exposed-modules:- RON.Data- RON.Data.LWW- RON.Data.ORSet- RON.Data.RGA- RON.Data.Time- RON.Data.VersionVector- other-modules:- RON.Data.Internal- hs-source-dirs: lib+ other-modules: RON.Data.Internal+ hs-source-dirs: lib