diff --git a/cabal.project b/cabal.project index 74a929fb48..0599430dd7 100644 --- a/cabal.project +++ b/cabal.project @@ -16,6 +16,7 @@ packages: doc plutus-playground-server plutus-script-utils plutus-use-cases + plutus-streaming quickcheck-dynamic web-ghc diff --git a/nix/pkgs/haskell/haskell.nix b/nix/pkgs/haskell/haskell.nix index 86216faa38..1ff309612c 100644 --- a/nix/pkgs/haskell/haskell.nix +++ b/nix/pkgs/haskell/haskell.nix @@ -60,6 +60,7 @@ let plutus-pab-executables.package.buildable = false; plutus-playground-server.package.buildable = false; # Would also require libpq plutus-script-utils.package.buildable = false; + plutus-streaming.package.buildable = false; plutus-tx-plugin.package.buildable = false; plutus-use-cases.package.buildable = false; plutus-example.package.buildable = false; diff --git a/plutus-streaming/README.md b/plutus-streaming/README.md new file mode 100644 index 0000000000..fab406ce3f --- /dev/null +++ b/plutus-streaming/README.md @@ -0,0 +1,42 @@ +# Plutus Streaming + +plutus-streaming is a simple library that wraps cardano-api giving a +streaming interface to the chain-sync protocol. This means you can use this +library to stream blocks from a locally running node. + +The blocks are presented as received from the chain-sync protocol so you +will still need to use function from `Cardano.Api` (in cardano-node) or +`Plutus.Ledger`(in plutus-ledger) to extract information from blocks. + +# Example applications + +## Example 1 + +This application simply streams to stdout events in JSON format. + +## Example 2 + +This application uses plutus-ledger to extract transactions (transaction +id, ins and outs) and datums. Then it prints them to stdout in JSON format. + +## Example 3 + +This application uses folds the stream into a UTxoState as defined in +plutus-chain-index-core. + +# Running the example applications + +You can run the example applications with cabal. Remember you need to have +a local node running and reachable through the provided socket. + +From Genesis + +``` +$ cabal run -- plutus-streaming-example-1 --socket-path /tmp/node.socket --mainnet +``` + +Passing a starting point + +``` +$ cabal run -- plutus-streaming-example-1 --socket-path /tmp/node.socket --slot-no 53427524 --block-hash 5e2bde4e504a9888a4f218dafc79a7619083f97d48684fcdba9dc78190df8f99 +``` diff --git a/plutus-streaming/examples/Common.hs b/plutus-streaming/examples/Common.hs new file mode 100644 index 0000000000..fda74c185f --- /dev/null +++ b/plutus-streaming/examples/Common.hs @@ -0,0 +1,77 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE RankNTypes #-} +module Common where + +import Cardano.Api qualified +import Data.Aeson qualified +import Data.ByteString.Lazy.Char8 qualified +import Options.Applicative (Alternative ((<|>)), Parser, auto, execParser, flag', help, helper, info, long, metavar, + option, str, strOption, (<**>)) +import Orphans () +import Streaming.Prelude qualified as S + +-- +-- Options parsing +-- + +data Options = Options + { optionsSocketPath :: String, + optionsNetworkId :: Cardano.Api.NetworkId, + optionsChainPoint :: Cardano.Api.ChainPoint + } + deriving (Show) + +optionsParser :: Parser Options +optionsParser = + Options + <$> strOption (long "socket-path" <> help "Node socket path") + <*> networkIdParser + <*> chainPointParser + +networkIdParser :: Parser Cardano.Api.NetworkId +networkIdParser = + pMainnet' <|> fmap Cardano.Api.Testnet testnetMagicParser + where + pMainnet' :: Parser Cardano.Api.NetworkId + pMainnet' = + flag' + Cardano.Api.Mainnet + ( long "mainnet" + <> help "Use the mainnet magic id." + ) + +testnetMagicParser :: Parser Cardano.Api.NetworkMagic +testnetMagicParser = + Cardano.Api.NetworkMagic + <$> option + auto + ( long "testnet-magic" + <> metavar "NATURAL" + <> help "Specify a testnet magic id." + ) + +chainPointParser :: Parser Cardano.Api.ChainPoint +chainPointParser = + pure Cardano.Api.ChainPointAtGenesis + <|> ( Cardano.Api.ChainPoint + <$> option (Cardano.Api.SlotNo <$> auto) (long "slot-no" <> metavar "SLOT-NO") + <*> option str (long "block-hash" <> metavar "BLOCK-HASH") + ) + +parseOptions :: IO Options +parseOptions = execParser $ info (optionsParser <**> helper) mempty + +printJson :: Data.Aeson.ToJSON a => S.Stream (S.Of a) IO r -> IO r +printJson = S.mapM_ Data.ByteString.Lazy.Char8.putStrLn . S.map Data.Aeson.encode + +-- https://github.com/input-output-hk/cardano-node/pull/3665 +workaround :: + (Cardano.Api.IsCardanoEra era => Cardano.Api.EraInMode era Cardano.Api.CardanoMode -> a) -> + Cardano.Api.EraInMode era Cardano.Api.CardanoMode -> + a +workaround k Cardano.Api.ByronEraInCardanoMode = k Cardano.Api.ByronEraInCardanoMode +workaround k Cardano.Api.ShelleyEraInCardanoMode = k Cardano.Api.ShelleyEraInCardanoMode +workaround k Cardano.Api.AllegraEraInCardanoMode = k Cardano.Api.AllegraEraInCardanoMode +workaround k Cardano.Api.MaryEraInCardanoMode = k Cardano.Api.MaryEraInCardanoMode +workaround k Cardano.Api.AlonzoEraInCardanoMode = k Cardano.Api.AlonzoEraInCardanoMode + diff --git a/plutus-streaming/examples/Example1.hs b/plutus-streaming/examples/Example1.hs new file mode 100644 index 0000000000..0a107eb4ad --- /dev/null +++ b/plutus-streaming/examples/Example1.hs @@ -0,0 +1,27 @@ +module Main where + +import Cardano.Api qualified +import Common (Options (Options, optionsChainPoint, optionsNetworkId, optionsSocketPath), parseOptions) +import Data.Aeson.Text qualified as Aeson +import Data.Text.Lazy qualified as TL +import Orphans () +import Plutus.Streaming (ChainSyncEvent (RollBackward, RollForward), withChainSyncEventStream) +import Streaming.Prelude qualified as S + +-- +-- Main +-- + +main :: IO () +main = do + Options {optionsSocketPath, optionsNetworkId, optionsChainPoint} <- parseOptions + + withChainSyncEventStream optionsSocketPath optionsNetworkId optionsChainPoint $ + S.stdoutLn + . S.map + ( \case + RollForward (Cardano.Api.BlockInMode (Cardano.Api.Block header _txs) _era) _ct -> + "RollForward, header: " <> TL.unpack (Aeson.encodeToLazyText header) + RollBackward cp _ct -> + "RollBackward, point: " <> TL.unpack (Aeson.encodeToLazyText cp) + ) diff --git a/plutus-streaming/examples/Example2.hs b/plutus-streaming/examples/Example2.hs new file mode 100644 index 0000000000..7360687c64 --- /dev/null +++ b/plutus-streaming/examples/Example2.hs @@ -0,0 +1,56 @@ +module Main where + +import Cardano.Api qualified +import Common (Options (Options, optionsChainPoint, optionsNetworkId, optionsSocketPath), parseOptions, printJson, + workaround) +import Data.Foldable (toList) +import Ledger qualified +import Orphans () +import Plutus.Script.Utils.V1.Scripts qualified +import Plutus.Streaming (withChainSyncEventStream) +import Streaming.Prelude qualified as S + +-- +-- Accessory functions +-- + +transactions :: + Cardano.Api.BlockInMode Cardano.Api.CardanoMode -> + [Ledger.CardanoTx] +transactions (Cardano.Api.BlockInMode (Cardano.Api.Block _ txs) eim) = + map (\tx -> Ledger.CardanoApiTx (workaround (Ledger.SomeTx tx) eim)) txs + +txInsAndOuts :: + Ledger.CardanoTx -> + (Ledger.TxId, [Ledger.TxOutRef], [Ledger.TxOutRef]) +txInsAndOuts tx = (txId, inRefs, outRefs) + where + txId = Ledger.getCardanoTxId tx + inRefs = map Ledger.txInRef $ toList $ Ledger.getCardanoTxInputs tx + outRefs = map snd $ Ledger.getCardanoTxOutRefs tx + +datums :: + Ledger.CardanoTx -> + [(Plutus.Script.Utils.V1.Scripts.DatumHash, Plutus.Script.Utils.V1.Scripts.Datum)] +datums tx = do + let txIns = toList $ Ledger.getCardanoTxInputs tx + (Ledger.TxIn _ (Just (Ledger.ConsumeScriptAddress _validator _redeemer datum))) <- txIns + pure (Plutus.Script.Utils.V1.Scripts.datumHash datum, datum) + +-- +-- Main +-- + +main :: IO () +main = do + Options {optionsSocketPath, optionsNetworkId, optionsChainPoint} <- parseOptions + + withChainSyncEventStream optionsSocketPath optionsNetworkId optionsChainPoint $ + printJson + . S.map -- Each ChainSyncEvent + ( fmap -- Inside the payload of RollForward events + ( fmap -- Each transaction + (\tx -> (txInsAndOuts tx, datums tx)) + . transactions + ) + ) diff --git a/plutus-streaming/examples/Example3.hs b/plutus-streaming/examples/Example3.hs new file mode 100644 index 0000000000..f9bcb73e0f --- /dev/null +++ b/plutus-streaming/examples/Example3.hs @@ -0,0 +1,53 @@ +module Main where + +import Cardano.Api qualified +import Common (Options (Options, optionsChainPoint, optionsNetworkId, optionsSocketPath), parseOptions) +import Plutus.ChainIndex (TxUtxoBalance) +import Plutus.ChainIndex.Compatibility qualified as CI +import Plutus.ChainIndex.TxUtxoBalance qualified as TxUtxoBalance +import Plutus.ChainIndex.UtxoState (UtxoIndex, UtxoState) +import Plutus.ChainIndex.UtxoState qualified as UtxoState +import Plutus.Streaming (ChainSyncEvent (RollBackward, RollForward), withChainSyncEventStream) +import Streaming (Of, Stream) +import Streaming.Prelude qualified as S + +utxoState :: + Monad m => + Stream (Of (ChainSyncEvent (Cardano.Api.BlockInMode Cardano.Api.CardanoMode))) m r -> + Stream (Of (ChainSyncEvent (Cardano.Api.BlockInMode Cardano.Api.CardanoMode), UtxoState TxUtxoBalance)) m r +utxoState = + S.scanned step initial projection + where + step index (RollForward block _) = + case CI.fromCardanoBlock block of + Left err -> error ("FromCardanoError: " <> show err) + Right txs -> + let tip = CI.tipFromCardanoBlock block + balance = TxUtxoBalance.fromBlock tip txs + in case UtxoState.insert balance index of + Left err -> + error (show err) + Right (UtxoState.InsertUtxoSuccess newIndex _insertPosition) -> + newIndex + step index (RollBackward cardanoPoint _) = + let point = CI.fromCardanoPoint cardanoPoint + in case TxUtxoBalance.rollback point index of + Left err -> error (show err) + Right (UtxoState.RollbackResult _newTip rolledBackIndex) -> + rolledBackIndex + + initial :: UtxoIndex TxUtxoBalance + initial = mempty + + projection = UtxoState.utxoState + +-- +-- Main +-- + +main :: IO () +main = do + Options {optionsSocketPath, optionsNetworkId, optionsChainPoint} <- parseOptions + + withChainSyncEventStream optionsSocketPath optionsNetworkId optionsChainPoint $ + S.print . utxoState diff --git a/plutus-streaming/examples/Orphans.hs b/plutus-streaming/examples/Orphans.hs new file mode 100644 index 0000000000..119bdd54e8 --- /dev/null +++ b/plutus-streaming/examples/Orphans.hs @@ -0,0 +1,38 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# OPTIONS_GHC -Wno-orphans #-} + +module Orphans where + +import Cardano.Api (BlockHeader (BlockHeader), BlockNo, ChainPoint (ChainPoint, ChainPointAtGenesis), + HasTypeProxy (proxyToAsType), Hash, SerialiseAsRawBytes (deserialiseFromRawBytes), ToJSON) +import Data.ByteString.Base16 qualified as Base16 +import Data.ByteString.Char8 qualified as C8 +import Data.Proxy (Proxy (Proxy)) +import Data.String (IsString (fromString)) +import GHC.Generics (Generic) +import Plutus.Streaming (ChainSyncEvent) + +-- https://github.com/input-output-hk/cardano-node/pull/3608 +instance IsString (Hash BlockHeader) where + fromString = either error id . deserialiseFromRawBytesBase16 . C8.pack + where + deserialiseFromRawBytesBase16 str = + case Base16.decode str of + Right raw -> case deserialiseFromRawBytes ttoken raw of + Just x -> Right x + Nothing -> Left ("cannot deserialise " ++ show str) + Left msg -> Left ("invalid hex " ++ show str ++ ", " ++ msg) + where + ttoken = proxyToAsType (Proxy :: Proxy a) + +deriving instance Generic ChainPoint + +instance ToJSON ChainPoint + +instance ToJSON BlockNo + +deriving instance Generic BlockHeader + +instance ToJSON BlockHeader + +instance ToJSON a => ToJSON (ChainSyncEvent a) diff --git a/plutus-streaming/plutus-streaming.cabal b/plutus-streaming/plutus-streaming.cabal new file mode 100644 index 0000000000..52ac8d8922 --- /dev/null +++ b/plutus-streaming/plutus-streaming.cabal @@ -0,0 +1,93 @@ +cabal-version: 2.4 +name: plutus-streaming +version: 0.1.0.0 +author: Andrea Bedini +maintainer: andrea.bedini@iohk.io +extra-source-files: CHANGELOG.md + +common lang + default-language: Haskell2010 + default-extensions: + DeriveFoldable + DeriveFunctor + DeriveGeneric + DeriveLift + DeriveTraversable + ExplicitForAll + GeneralizedNewtypeDeriving + ImportQualifiedPost + LambdaCase + NamedFieldPuns + ScopedTypeVariables + StandaloneDeriving + ghc-options: + -Wall + -Widentities + -Wincomplete-record-updates + -Wincomplete-uni-patterns + -Wmissing-import-lists + -Wnoncanonical-monad-instances + -Wredundant-constraints + -Wunused-packages + +library + import: lang + hs-source-dirs: src + exposed-modules: + Plutus.Streaming + build-depends: + base >=4.9 && <5, + async, + cardano-api, + ouroboros-network, + stm, + streaming, + +executable plutus-streaming-example-1 + import: lang + hs-source-dirs: examples + main-is: Example1.hs + other-modules: Common, Orphans + build-depends: + plutus-streaming, + base >=4.9 && <5, + aeson, + base16-bytestring, + bytestring, + cardano-api, + optparse-applicative, + streaming, + text + +executable plutus-streaming-example-2 + import: lang + hs-source-dirs: examples + main-is: Example2.hs + other-modules: Common, Orphans + build-depends: + plutus-streaming, + base >=4.9 && <5, + aeson, + base16-bytestring, + bytestring, + cardano-api, + optparse-applicative, + plutus-ledger, + plutus-script-utils, + streaming, + +executable plutus-streaming-example-3 + import: lang + hs-source-dirs: examples + main-is: Example3.hs + other-modules: Common, Orphans + build-depends: + plutus-streaming, + base >=4.9 && <5, + aeson, + base16-bytestring, + bytestring, + cardano-api, + optparse-applicative, + plutus-chain-index-core, + streaming, diff --git a/plutus-streaming/src/Plutus/Streaming.hs b/plutus-streaming/src/Plutus/Streaming.hs new file mode 100644 index 0000000000..6acaa6eaa7 --- /dev/null +++ b/plutus-streaming/src/Plutus/Streaming.hs @@ -0,0 +1,129 @@ +module Plutus.Streaming + ( withChainSyncEventStream, + ChainSyncEvent (..), + ChainSyncEventException (..), + ) +where + +import Cardano.Api (BlockInMode, CardanoMode, ChainPoint, ChainSyncClient (ChainSyncClient), ChainTip, + ConsensusModeParams (CardanoModeParams), EpochSlots (EpochSlots), + LocalChainSyncClient (LocalChainSyncClient), + LocalNodeClientProtocols (LocalNodeClientProtocols, localChainSyncClient, localStateQueryClient, localTxSubmissionClient), + LocalNodeConnectInfo (LocalNodeConnectInfo, localConsensusModeParams, localNodeNetworkId, localNodeSocketPath), + NetworkId, connectToLocalNode) +import Cardano.Api.ChainSync.Client (ClientStIdle (SendMsgFindIntersect, SendMsgRequestNext), + ClientStIntersect (ClientStIntersect, recvMsgIntersectFound, recvMsgIntersectNotFound), + ClientStNext (ClientStNext, recvMsgRollBackward, recvMsgRollForward)) +import Control.Concurrent.Async (link, withAsync) +import Control.Concurrent.STM (TChan, atomically, dupTChan, newBroadcastTChanIO, readTChan, writeTChan) +import Control.Exception (Exception, throw) +import GHC.Generics (Generic) +import Streaming (Of, Stream) +import Streaming.Prelude qualified as S + +data ChainSyncEvent a + = RollForward a ChainTip + | RollBackward ChainPoint ChainTip + deriving (Show, Functor, Generic) + +data ChainSyncEventException + = NoIntersectionFound + deriving (Show) + +instance Exception ChainSyncEventException + +-- | `withChainSyncEventStream` uses the chain-sync mini-protocol to +-- connect to a locally running node and fetch blocks from the given +-- starting point. +withChainSyncEventStream :: + -- | Path to the node socket + FilePath -> + NetworkId -> + -- | The point on the chain to start streaming from + ChainPoint -> + -- | Stream consumer + (Stream (Of (ChainSyncEvent (BlockInMode CardanoMode))) IO r -> IO b) -> + IO b +withChainSyncEventStream socketPath networkId point consumer = do + -- The chain-sync client runs in a different thread and it will send us + -- block through this channel. + + -- By using newBroadcastTChan, messages can be garbage collected after + -- clients have seen them, preventing pile up. The only way to read a + -- broadcast channel is to duplicate it with dupTChan. See note at + -- `newBroadcastTChan`. + chan <- newBroadcastTChanIO + readerChannel <- atomically $ dupTChan chan + + let client = chainSyncStreamingClient point chan + + localNodeClientProtocols = + LocalNodeClientProtocols + { localChainSyncClient = LocalChainSyncClient client, + localTxSubmissionClient = Nothing, + localStateQueryClient = Nothing + } + + connectInfo = + LocalNodeConnectInfo + { localConsensusModeParams = CardanoModeParams epochSlots, + localNodeNetworkId = networkId, + localNodeSocketPath = socketPath + } + + -- FIXME this comes from the config file but Cardano.Api does not expose readNetworkConfig! + epochSlots = EpochSlots 40 + + clientThread = do + connectToLocalNode connectInfo localNodeClientProtocols + -- the only reason connectToLocalNode can terminate successfully is if it + -- doesn't find an intersection, we report that case to the + -- consumer as an exception + throw NoIntersectionFound + + withAsync clientThread $ \a -> do + -- Make sure all exceptions in the client thread are passed to the consumer thread + link a + -- Run the consumer + consumer $ S.repeatM $ atomically (readTChan readerChannel) + +-- | `chainSyncStreamingClient` is the client that connects to a local node +-- and runs the chain-sync mini-protocol. This client is fire-and-forget +-- and does not require any control. +-- +-- Blocks obtained from the chain-sync mini-protocol are passed to a +-- consumer through a channel. +-- +-- If the starting point is such that an intersection cannot be found, this +-- client will throw a NoIntersectionFound exception. +chainSyncStreamingClient :: + ChainPoint -> + TChan (ChainSyncEvent e) -> + ChainSyncClient e ChainPoint ChainTip IO () +chainSyncStreamingClient point chan = + ChainSyncClient $ pure $ SendMsgFindIntersect [point] onIntersect + where + onIntersect = + ClientStIntersect + { recvMsgIntersectFound = \_ _ -> + ChainSyncClient sendRequestNext, + recvMsgIntersectNotFound = \_ -> + ChainSyncClient $ + -- There is nothing we can do here + throw NoIntersectionFound + } + + sendRequestNext = + pure $ SendMsgRequestNext onNext (pure onNext) + where + onNext = + ClientStNext + { recvMsgRollForward = \bim ct -> + ChainSyncClient $ do + atomically $ writeTChan chan (RollForward bim ct) + sendRequestNext, + recvMsgRollBackward = \cp ct -> + ChainSyncClient $ do + atomically $ writeTChan chan (RollBackward cp ct) + sendRequestNext + }