diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -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 diff --git a/lib/RON/Data.hs b/lib/RON/Data.hs --- a/lib/RON/Data.hs +++ b/lib/RON/Data.hs @@ -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} diff --git a/lib/RON/Data/CT.hs b/lib/RON/Data/CT.hs new file mode 100644 --- /dev/null +++ b/lib/RON/Data/CT.hs @@ -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 diff --git a/lib/RON/Data/GTree.hs b/lib/RON/Data/GTree.hs new file mode 100644 --- /dev/null +++ b/lib/RON/Data/GTree.hs @@ -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} :| []} diff --git a/lib/RON/Data/Internal.hs b/lib/RON/Data/Internal.hs --- a/lib/RON/Data/Internal.hs +++ b/lib/RON/Data/Internal.hs @@ -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 diff --git a/lib/RON/Data/LWW.hs b/lib/RON/Data/LWW.hs --- a/lib/RON/Data/LWW.hs +++ b/lib/RON/Data/LWW.hs @@ -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 diff --git a/lib/RON/Data/ORSet.hs b/lib/RON/Data/ORSet.hs --- a/lib/RON/Data/ORSet.hs +++ b/lib/RON/Data/ORSet.hs @@ -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) diff --git a/lib/RON/Data/RGA.hs b/lib/RON/Data/RGA.hs --- a/lib/RON/Data/RGA.hs +++ b/lib/RON/Data/RGA.hs @@ -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 diff --git a/lib/RON/Data/Time.hs b/lib/RON/Data/Time.hs --- a/lib/RON/Data/Time.hs +++ b/lib/RON/Data/Time.hs @@ -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" diff --git a/lib/RON/Data/VersionVector.hs b/lib/RON/Data/VersionVector.hs --- a/lib/RON/Data/VersionVector.hs +++ b/lib/RON/Data/VersionVector.hs @@ -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 diff --git a/lib/RON/Experimental/Data.hs b/lib/RON/Experimental/Data.hs new file mode 100644 --- /dev/null +++ b/lib/RON/Experimental/Data.hs @@ -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 diff --git a/lib/RON/Experimental/Data/ORSet.hs b/lib/RON/Experimental/Data/ORSet.hs new file mode 100644 --- /dev/null +++ b/lib/RON/Experimental/Data/ORSet.hs @@ -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] diff --git a/lib/RON/Experimental/Data/ORSet/Type.hs b/lib/RON/Experimental/Data/ORSet/Type.hs new file mode 100644 --- /dev/null +++ b/lib/RON/Experimental/Data/ORSet/Type.hs @@ -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) diff --git a/lib/RON/Store.hs b/lib/RON/Store.hs new file mode 100644 --- /dev/null +++ b/lib/RON/Store.hs @@ -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] + ] diff --git a/lib/RON/Store/Class.hs b/lib/RON/Store/Class.hs new file mode 100644 --- /dev/null +++ b/lib/RON/Store/Class.hs @@ -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 diff --git a/lib/RON/Store/FS.hs b/lib/RON/Store/FS.hs new file mode 100644 --- /dev/null +++ b/lib/RON/Store/FS.hs @@ -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)) diff --git a/lib/RON/Store/Test.hs b/lib/RON/Store/Test.hs new file mode 100644 --- /dev/null +++ b/lib/RON/Store/Test.hs @@ -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} diff --git a/ron-rdt.cabal b/ron-rdt.cabal --- a/ron-rdt.cabal +++ b/ron-rdt.cabal @@ -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 +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 -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