-
Notifications
You must be signed in to change notification settings - Fork 86
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1351 from SundaeSwap-finance/SB-1352-persistence-…
…types EventSource and EventSink abstractions for Hydra Extensibility
- Loading branch information
Showing
25 changed files
with
914 additions
and
376 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| module Hydra.Environment where | ||
|
|
||
| import Hydra.Prelude | ||
|
|
||
| import Hydra.ContestationPeriod (ContestationPeriod) | ||
| import Hydra.Crypto (HydraKey, SigningKey) | ||
| import Hydra.OnChainId (OnChainId) | ||
| import Hydra.Party (Party, deriveParty) | ||
|
|
||
| data Environment = Environment | ||
| { party :: Party | ||
| -- ^ This is the p_i from the paper | ||
| , -- XXX: In the long run we would not want to keep the signing key in memory, | ||
| -- i.e. have an 'Effect' for signing or so. | ||
| signingKey :: SigningKey HydraKey | ||
| , otherParties :: [Party] | ||
| , -- XXX: Improve naming | ||
| participants :: [OnChainId] | ||
| , contestationPeriod :: ContestationPeriod | ||
| } | ||
| deriving stock (Show, Eq) | ||
|
|
||
| instance Arbitrary Environment where | ||
| arbitrary = do | ||
| signingKey <- arbitrary | ||
| otherParties <- arbitrary | ||
| participants <- arbitrary | ||
| contestationPeriod <- arbitrary | ||
| pure $ | ||
| Environment | ||
| { signingKey | ||
| , party = deriveParty signingKey | ||
| , otherParties | ||
| , contestationPeriod | ||
| , participants | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| -- | This module defines the types and functions for creating 'EventSource' and | ||
| -- 'EventSink' instances and is intended to be used as an extension point. | ||
| -- | ||
| -- A single 'EventSource' and zero or more 'EventSink' handles are used by the | ||
| -- main 'HydraNode' handle to load and send out events. | ||
| -- | ||
| -- See 'Hydra.Events.FileBased' for an example implementation and | ||
| -- 'Hydra.Events.FileBasedSpec' for the corresponding test suite. | ||
| -- | ||
| -- Custom implementations should be located under Hydra.Events to avoid | ||
| -- conflicts. | ||
| module Hydra.Events where | ||
|
|
||
| import Hydra.Prelude | ||
|
|
||
| import Hydra.Chain (IsChainState) | ||
| import Hydra.HeadLogic.Outcome (StateChanged) | ||
|
|
||
| type EventId = Word64 | ||
|
|
||
| class HasEventId a where | ||
| getEventId :: a -> EventId | ||
|
|
||
| instance HasEventId (EventId, a) where | ||
| getEventId = fst | ||
|
|
||
| newtype EventSource e m = EventSource | ||
| { getEvents :: HasEventId e => m [e] | ||
| -- ^ Retrieve all events from the event source. | ||
| } | ||
|
|
||
| newtype EventSink e m = EventSink | ||
| { putEvent :: HasEventId e => e -> m () | ||
| -- ^ Send a single event to the event sink. | ||
| } | ||
|
|
||
| -- | Put a list of events to a list of event sinks in a round-robin fashion. | ||
| putEventsToSinks :: (Monad m, HasEventId e) => [EventSink e m] -> [e] -> m () | ||
| putEventsToSinks sinks events = | ||
| forM_ events $ \event -> | ||
| forM_ sinks $ \sink -> | ||
| putEvent sink event | ||
|
|
||
| -- * State change events as used by Hydra.Node | ||
|
|
||
| -- | A state change event with an event id that is the common entity to be | ||
| -- loaded from an 'EventSource' and sent to 'EventSink's. | ||
| data StateEvent tx = StateEvent | ||
| { eventId :: EventId | ||
| , stateChanged :: StateChanged tx | ||
| } | ||
| deriving (Generic) | ||
|
|
||
| instance HasEventId (StateEvent tx) where | ||
| getEventId = eventId | ||
|
|
||
| deriving instance IsChainState tx => Show (StateEvent tx) | ||
| deriving instance IsChainState tx => Eq (StateEvent tx) | ||
| deriving instance IsChainState tx => ToJSON (StateEvent tx) | ||
| deriving instance IsChainState tx => FromJSON (StateEvent tx) | ||
|
|
||
| instance IsChainState tx => Arbitrary (StateEvent tx) where | ||
| arbitrary = genericArbitrary | ||
| shrink = genericShrink | ||
|
|
||
| genStateEvent :: StateChanged tx -> Gen (StateEvent tx) | ||
| genStateEvent sc = StateEvent <$> arbitrary <*> pure sc |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| -- | A file-based event source and sink using JSON encoding. | ||
| -- | ||
| -- This serves as an example of how to create an 'EventSource' and 'EventSink'. | ||
| module Hydra.Events.FileBased where | ||
|
|
||
| import Hydra.Prelude | ||
|
|
||
| import Control.Concurrent.Class.MonadSTM (newTVarIO, writeTVar) | ||
| import Hydra.Chain (IsChainState) | ||
| import Hydra.Events (EventSink (..), EventSource (..), StateEvent (..)) | ||
| import Hydra.HeadLogic.Outcome (StateChanged) | ||
| import Hydra.Persistence (PersistenceIncremental (..)) | ||
|
|
||
| -- | A basic file based event source and sink defined using an | ||
| -- 'PersistenceIncremental' handle. | ||
| -- | ||
| -- The complexity in this implementation mostly stems from the fact that we want | ||
| -- to be backward-compatible with the old, plain format of storing | ||
| -- 'StateChanged' items directly to disk using 'PersistenceIncremental'. | ||
| -- | ||
| -- If any 'Legacy StateChanged' items are discovered, a running index is used | ||
| -- for the 'eventId', while the 'New StateEvent' values are just stored as is. | ||
| -- | ||
| -- A new implementation for an 'EventSource' with a compatible 'EventSink' could | ||
| -- be defined more generically with constraints: | ||
| -- | ||
| -- (ToJSON e, FromJSON e, HasEventId) e => (EventSource e m, EventSink e m) | ||
| eventPairFromPersistenceIncremental :: | ||
| (IsChainState tx, MonadSTM m) => | ||
| PersistenceIncremental (PersistedStateChange tx) m -> | ||
| m (EventSource (StateEvent tx) m, EventSink (StateEvent tx) m) | ||
| eventPairFromPersistenceIncremental PersistenceIncremental{append, loadAll} = do | ||
| eventIdV <- newTVarIO Nothing | ||
| let | ||
| getLastSeenEventId = readTVar eventIdV | ||
|
|
||
| setLastSeenEventId StateEvent{eventId} = do | ||
| writeTVar eventIdV (Just eventId) | ||
|
|
||
| getNextEventId = | ||
| maybe 0 (+ 1) <$> readTVar eventIdV | ||
|
|
||
| -- Keep track of the last seen event id when loading | ||
| getEvents = do | ||
| items <- loadAll | ||
| atomically . forM items $ \i -> do | ||
| event <- case i of | ||
| New e -> pure e | ||
| Legacy sc -> do | ||
| eventId <- getNextEventId | ||
| pure $ StateEvent eventId sc | ||
|
|
||
| setLastSeenEventId event | ||
| pure event | ||
|
|
||
| -- Filter events that are already stored | ||
| putEvent e@StateEvent{eventId} = do | ||
| atomically getLastSeenEventId >>= \case | ||
| Nothing -> store e | ||
| Just lastSeenEventId | ||
| | eventId > lastSeenEventId -> store e | ||
| | otherwise -> pure () | ||
|
|
||
| store e = do | ||
| append (New e) | ||
| atomically $ setLastSeenEventId e | ||
|
|
||
| pure (EventSource{getEvents}, EventSink{putEvent}) | ||
|
|
||
| -- | Internal data type used by 'createJSONFileEventSourceAndSink' to be | ||
| -- compatible with plain usage of 'PersistenceIncrementa' using plain | ||
| -- 'StateChanged' items to the new 'StateEvent' persisted items. | ||
| data PersistedStateChange tx | ||
| = Legacy (StateChanged tx) | ||
| | New (StateEvent tx) | ||
| deriving stock (Generic, Show, Eq) | ||
|
|
||
| instance IsChainState tx => ToJSON (PersistedStateChange tx) where | ||
| toJSON = \case | ||
| Legacy sc -> toJSON sc | ||
| New e -> toJSON e | ||
|
|
||
| instance IsChainState tx => FromJSON (PersistedStateChange tx) where | ||
| parseJSON v = | ||
| New <$> parseJSON v | ||
| <|> Legacy <$> parseJSON v |
Oops, something went wrong.