haskoin-store 0.31.0 → 0.32.0
raw patch · 12 files changed
+1896/−1791 lines, 12 filesdep ~haskoin-nodedep ~haskoin-storedep ~haskoin-store-data
Dependency ranges changed: haskoin-node, haskoin-store, haskoin-store-data
Files
- LICENSE +19/−0
- UNLICENSE +0/−24
- app/Main.hs +57/−36
- haskoin-store.cabal +15/−27
- src/Haskoin/Store/BlockStore.hs +974/−826
- src/Haskoin/Store/Cache.hs +17/−10
- src/Haskoin/Store/Common.hs +4/−6
- src/Haskoin/Store/Database/Writer.hs +105/−172
- src/Haskoin/Store/Logic.hs +402/−428
- src/Haskoin/Store/Manager.hs +170/−127
- src/Haskoin/Store/Web.hs +59/−52
- test/Haskoin/StoreSpec.hs +74/−83
+ LICENSE view
@@ -0,0 +1,19 @@+Copyright 2020 Haskoin Developers++Permission is hereby granted, free of charge, to any person obtaining a copy of+this software and associated documentation files (the "Software"), to deal in+the Software without restriction, including without limitation the rights to+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of+the Software, and to permit persons to whom the Software is furnished to do so,+subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.+
− UNLICENSE
@@ -1,24 +0,0 @@-This is free and unencumbered software released into the public domain.--Anyone is free to copy, modify, publish, use, compile, sell, or-distribute this software, either in source code form or as a compiled-binary, for any purpose, commercial or non-commercial, and by any-means.--In jurisdictions that recognize copyright laws, the author or authors-of this software dedicate any and all copyright interest in the-software to the public domain. We make this dedication for the benefit-of the public at large and to the detriment of our heirs and-successors. We intend this dedication to be an overt act of-relinquishment in perpetuity of all present and future rights to this-software under copyright law.--THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,-EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF-MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.-IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR-OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,-ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR-OTHER DEALINGS IN THE SOFTWARE.--For more information, please refer to <http://unlicense.org/>
app/Main.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE ApplicativeDo #-}+{-# LANGUAGE CPP #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-}@@ -15,7 +16,6 @@ import Data.List (intercalate) import Data.Maybe (fromMaybe) import Data.String.Conversions (cs)-import Data.Version (showVersion) import Haskoin (Network (..), allNets, bch, bchRegTest, bchTest, btc, btcRegTest, btcTest, eitherToMaybe)@@ -28,7 +28,6 @@ help, helper, info, long, many, metavar, option, progDesc, short, showDefault, strOption, switch, value)-import Paths_haskoin_store as P import System.Exit (exitSuccess) import System.FilePath ((</>)) import System.IO.Unsafe (unsafePerformIO)@@ -38,6 +37,13 @@ getAppUserDataDirectory) import UnliftIO.Environment (lookupEnv) +version :: String+#ifdef CURRENT_PACKAGE_VERSION+version = CURRENT_PACKAGE_VERSION+#else+version = "Unavailable"+#endif+ data Config = Config { configDir :: !FilePath , configHost :: !String@@ -56,7 +62,8 @@ , configRedisMax :: !Integer , configWipeMempool :: !Bool , configPeerTimeout :: !Int- , configPeerTooOld :: !Int+ , configPeerMaxLife :: !Int+ , configMaxPeers :: !Int } instance Default Config where@@ -77,7 +84,8 @@ , configRedisMax = defRedisMax , configWipeMempool = defWipeMempool , configPeerTimeout = defPeerTimeout- , configPeerTooOld = defPeerTooOld+ , configPeerMaxLife = defPeerMaxLife+ , configMaxPeers = defMaxPeers } defEnv :: MonadIO m => String -> a -> (String -> Maybe a) -> m a@@ -85,6 +93,11 @@ ms <- lookupEnv e return $ fromMaybe d $ p =<< ms +defMaxPeers :: Int+defMaxPeers = unsafePerformIO $+ defEnv "MAX_PEERS" 20 readMaybe+{-# NOINLINE defMaxPeers #-}+ defDirectory :: FilePath defDirectory = unsafePerformIO $ do d <- getAppUserDataDirectory "haskoin-store"@@ -182,10 +195,10 @@ defEnv "PEER_TIMEOUT" 120 readMaybe {-# NOINLINE defPeerTimeout #-} -defPeerTooOld :: Int-defPeerTooOld = unsafePerformIO $- defEnv "PEER_TOO_OLD" (48 * 3600) readMaybe-{-# NOINLINE defPeerTooOld #-}+defPeerMaxLife :: Int+defPeerMaxLife = unsafePerformIO $+ defEnv "PEER_MAX_LIFE" (48 * 3600) readMaybe+{-# NOINLINE defPeerMaxLife #-} netNames :: String netNames = intercalate "|" (map getNetworkName allNets)@@ -206,7 +219,7 @@ config = do configDir <- strOption $- metavar "WORKDIR"+ metavar "PATH" <> long "dir" <> short 'd' <> help "Data directory"@@ -248,6 +261,12 @@ <> long "peer" <> short 'p' <> help "Network peer (as many as required)"+ configMaxPeers <-+ option auto $+ metavar "INT"+ <> long "max-peers"+ <> showDefault+ <> value (configMaxPeers def) configVersion <- switch $ long "version"@@ -263,74 +282,74 @@ <> help "HTTP request logging" maxLimitCount <- option auto $- metavar "MAXLIMIT"+ metavar "INT" <> long "max-limit"- <> help "Hard limit for simple listings (0 = ∞)"+ <> help "Hard limit for simple listings (0 = inf)" <> showDefault <> value (maxLimitCount (configWebLimits def)) maxLimitFull <- option auto $- metavar "MAXLIMITFULL"+ metavar "INT" <> long "max-full"- <> help "Hard limit for full listings (0 = ∞)"+ <> help "Hard limit for full listings (0 = inf)" <> showDefault <> value (maxLimitFull (configWebLimits def)) maxLimitOffset <- option auto $- metavar "MAXOFFSET"+ metavar "INT" <> long "max-offset"- <> help "Hard limit for offsets (0 = ∞)"+ <> help "Hard limit for offsets (0 = inf)" <> showDefault <> value (maxLimitOffset (configWebLimits def)) maxLimitDefault <- option auto $- metavar "LIMITDEFAULT"+ metavar "INT" <> long "def-limit"- <> help "Soft default limit (0 = ∞)"+ <> help "Soft default limit (0 = inf)" <> showDefault <> value (maxLimitDefault (configWebLimits def)) maxLimitGap <- option auto $- metavar "MAXGAP"+ metavar "INT" <> long "max-gap" <> help "Max gap for xpub queries" <> showDefault <> value (maxLimitGap (configWebLimits def)) maxLimitInitialGap <- option auto $- metavar "INITGAP"+ metavar "INT" <> long "init-gap" <> help "Max gap for empty xpub" <> showDefault <> value (maxLimitInitialGap (configWebLimits def)) blockTimeout <- option auto $- metavar "BLOCKSECONDS"+ metavar "SECONDS" <> long "block-timeout"- <> help "Last block mined health timeout (0 = ∞)"+ <> help "Last block mined health timeout (0 = inf)" <> showDefault <> value (blockTimeout (configWebTimeouts def)) txTimeout <- option auto $- metavar "TXSECONDS"+ metavar "SECONDS" <> long "tx-timeout"- <> help "Last tx recived health timeout (0 = ∞)"+ <> help "Last tx recived health timeout (0 = inf)" <> showDefault <> value (txTimeout (configWebTimeouts def)) configPeerTimeout <- option auto $- metavar "TIMEOUT"+ metavar "SECONDS" <> long "peer-timeout" <> help "Unresponsive peer timeout" <> showDefault <> value (configPeerTimeout def)- configPeerTooOld <-+ configPeerMaxLife <- option auto $- metavar "TIMEOUT"- <> long "peer-old"+ metavar "SECONDS"+ <> long "peer-max-life" <> help "Disconnect peers older than this" <> showDefault- <> value (configPeerTooOld def)+ <> value (configPeerMaxLife def) configRedis <- flag (configRedis def) True $ long "cache"@@ -343,14 +362,14 @@ <> value (configRedisURL def) configRedisMin <- option auto $- metavar "MINADDRS"+ metavar "INT" <> long "cache-min" <> help "Minimum used xpub addresses to cache" <> showDefault <> value (configRedisMin def) configRedisMax <- option auto $- metavar "MAXKEYS"+ metavar "INT" <> long "cache-keys" <> help "Maximum number of keys in Redis xpub cache" <> showDefault@@ -394,7 +413,7 @@ main = do conf <- execParser opts when (configVersion conf) $ do- putStrLn $ showVersion P.version+ putStrLn version exitSuccess if null (configPeers conf) && not (configDiscover conf) then run conf {configDiscover = True}@@ -405,7 +424,7 @@ fullDesc <> progDesc "Bitcoin (BCH & BTC) block chain index with HTTP API" <> Options.Applicative.header- ("haskoin-store version " <> showVersion P.version)+ ("haskoin-store version " <> version) run :: Config -> IO () run Config { configHost = host@@ -424,7 +443,8 @@ , configRedisMax = redismax , configWipeMempool = wipemempool , configPeerTimeout = peertimeout- , configPeerTooOld = peerold+ , configPeerMaxLife = peerlife+ , configMaxPeers = maxpeers } = runStderrLoggingT . filterLogger l $ do $(logInfoS) "Main" $@@ -432,7 +452,7 @@ createDirectoryIfMissing True wd let scfg = StoreConfig- { storeConfMaxPeers = 20+ { storeConfMaxPeers = maxpeers , storeConfInitPeers = map (second (fromMaybe (getDefaultPort net))) peers , storeConfDiscover = disc@@ -447,8 +467,8 @@ , storeConfCacheMin = cachemin , storeConfMaxKeys = redismax , storeConfWipeMempool = wipemempool- , storeConfPeerTimeout = peertimeout- , storeConfPeerTooOld = peerold+ , storeConfPeerTimeout = fromIntegral peertimeout+ , storeConfPeerMaxLife = fromIntegral peerlife , storeConfConnect = withConnection } withStore scfg $ \st ->@@ -460,6 +480,7 @@ , webMaxLimits = limits , webReqLog = reqlog , webWebTimeouts = tos+ , webVersion = version } where l _ lvl
haskoin-store.cabal view
@@ -1,25 +1,22 @@-cabal-version: 1.12+cabal-version: 2.0 -- This file has been generated from package.yaml by hpack version 0.33.0. -- -- see: https://github.com/sol/hpack ----- hash: c5fef1d2f34e119fa142b47d873bda6c5c99c325a1e861107d9ca8ce81746b1d+-- hash: aea38ad5b8a4407c2f0cc77fb2c28eeb48b90238ea2bf7fca5b705511c4cb43e name: haskoin-store-version: 0.31.0+version: 0.32.0 synopsis: Storage and index for Bitcoin and Bitcoin Cash-description: Store and index Bitcoin or Bitcoin Cash blocks, transactions, balances and unspent outputs.- All data is available via REST API in JSON or binary format.- Uses in-process RocksDB database to store data.- Supports Redis cache that can be shared by multiple nodes for faster access to extended public key balances, transactions and unspent outputs.+description: Please see the README on GitHub at <https://github.com/haskoin/haskoin-store#readme> category: Bitcoin, Finance, Network homepage: http://github.com/haskoin/haskoin-store#readme bug-reports: http://github.com/haskoin/haskoin-store/issues author: Jean-Pierre Rupp maintainer: jprupp@protonmail.ch-license: PublicDomain-license-file: UNLICENSE+license: MIT+license-file: LICENSE build-type: Simple source-repository head@@ -29,7 +26,6 @@ library exposed-modules: Haskoin.Store- other-modules: Haskoin.Store.BlockStore Haskoin.Store.Cache Haskoin.Store.Common@@ -40,6 +36,8 @@ Haskoin.Store.Manager Haskoin.Store.Web Paths_haskoin_store+ autogen-modules:+ Paths_haskoin_store hs-source-dirs: src build-depends:@@ -54,8 +52,8 @@ , deepseq >=1.4.4.0 , hashable >=1.3.0.0 , haskoin-core >=0.13.6- , haskoin-node >=0.13.0- , haskoin-store-data ==0.31.0+ , haskoin-node >=0.14.1+ , haskoin-store-data ==0.32.0 , hedis >=0.12.13 , http-types >=0.12.3 , monad-logger >=0.3.32@@ -96,9 +94,9 @@ , filepath , hashable >=1.3.0.0 , haskoin-core >=0.13.6- , haskoin-node >=0.13.0+ , haskoin-node >=0.14.1 , haskoin-store- , haskoin-store-data ==0.31.0+ , haskoin-store-data ==0.32.0 , monad-logger >=0.3.32 , mtl >=2.2.2 , nqe >=0.6.1@@ -118,20 +116,9 @@ other-modules: Haskoin.Store.CacheSpec Haskoin.StoreSpec- Haskoin.Store- Haskoin.Store.BlockStore- Haskoin.Store.Cache- Haskoin.Store.Common- Haskoin.Store.Database.Reader- Haskoin.Store.Database.Types- Haskoin.Store.Database.Writer- Haskoin.Store.Logic- Haskoin.Store.Manager- Haskoin.Store.Web Paths_haskoin_store hs-source-dirs: test- src ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: QuickCheck >=2.13.2@@ -147,8 +134,9 @@ , deepseq >=1.4.4.0 , hashable >=1.3.0.0 , haskoin-core >=0.13.6- , haskoin-node >=0.13.0- , haskoin-store-data ==0.31.0+ , haskoin-node >=0.14.1+ , haskoin-store ==0.32.0+ , haskoin-store-data ==0.32.0 , hedis >=0.12.13 , hspec >=2.7.1 , http-types >=0.12.3
src/Haskoin/Store/BlockStore.hs view
@@ -9,832 +9,980 @@ ( -- * Block Store BlockStore , BlockStoreMessage- , BlockStoreConfig(..)- , blockStore- , blockStorePeerConnect- , blockStorePeerConnectSTM- , blockStorePeerDisconnect- , blockStorePeerDisconnectSTM- , blockStoreHead- , blockStoreHeadSTM- , blockStoreBlock- , blockStoreBlockSTM- , blockStoreNotFound- , blockStoreNotFoundSTM- , blockStoreTx- , blockStoreTxSTM- , blockStoreTxHash- , blockStoreTxHashSTM- ) where--import Control.Applicative ((<|>))-import Control.Monad (forM, forM_, forever, mzero,- unless, void, when)-import Control.Monad.Except (ExceptT, MonadError (..),- runExceptT)-import Control.Monad.Logger (MonadLoggerIO, logDebugS,- logErrorS, logInfoS, logWarnS)-import Control.Monad.Reader (MonadReader, ReaderT (..), asks)-import Control.Monad.Trans (lift)-import Control.Monad.Trans.Maybe (MaybeT (MaybeT), runMaybeT)-import qualified Data.ByteString as B-import Data.HashMap.Strict (HashMap)-import qualified Data.HashMap.Strict as HashMap-import Data.HashSet (HashSet)-import qualified Data.HashSet as HashSet-import Data.Maybe (catMaybes, fromJust, isJust,- listToMaybe, mapMaybe)-import Data.Serialize (encode)-import Data.String (fromString)-import Data.String.Conversions (cs)-import Data.Text (Text)-import Data.Time.Clock.POSIX (posixSecondsToUTCTime)-import Data.Time.Clock.System (getSystemTime, systemSeconds)-import Data.Time.Format (defaultTimeLocale, formatTime,- iso8601DateFormat)-import Haskoin (Block (..), BlockHash (..),- BlockHeader (..), BlockHeight,- BlockNode (..), GetData (..),- InvType (..), InvVector (..),- Message (..), Network (..),- OutPoint (..), Tx (..),- TxHash (..), TxIn (..),- blockHashToHex, headerHash,- txHash, txHashToHex)-import Haskoin.Node (Chain, OnlinePeer (..), Peer,- PeerException (..), PeerManager,- chainBlockMain,- chainGetAncestor, chainGetBest,- chainGetBlock, chainGetParents,- killPeer, managerGetPeer,- managerGetPeers,- managerPeerText, sendMessage)-import Haskoin.Store.Common (StoreEvent (..), StoreRead (..),- sortTxs)-import Haskoin.Store.Data (TxData (..), TxRef (..),- UnixTime, Unspent (..))-import Haskoin.Store.Database.Reader (DatabaseReader)-import Haskoin.Store.Database.Writer (Writer, runWriter)-import Haskoin.Store.Logic (ImportException (Orphan),- deleteTx, getOldMempool,- importBlock, initBest,- newMempoolTx, revertBlock)-import Network.Socket (SockAddr)-import NQE (Inbox, Listen, Mailbox,- inboxToMailbox, query, receive,- send, sendSTM)-import System.Random (randomRIO)-import UnliftIO (Exception, MonadIO,- MonadUnliftIO, STM, TVar,- atomically, liftIO, modifyTVar,- newTVarIO, readTVar, readTVarIO,- throwIO, withAsync, writeTVar)-import UnliftIO.Concurrent (threadDelay)---- | Messages for block store actor.-data BlockStoreMessage- = BlockNewBest !BlockNode- | BlockPeerConnect !Peer !SockAddr- | BlockPeerDisconnect !Peer !SockAddr- | BlockReceived !Peer !Block- | BlockNotFound !Peer ![BlockHash]- | TxRefReceived !Peer !Tx- | TxRefAvailable !Peer ![TxHash]- | BlockPing !(Listen ())- -- ^ internal housekeeping ping---- | Mailbox for block store.-type BlockStore = Mailbox BlockStoreMessage--data BlockException- = BlockNotInChain !BlockHash- | Uninitialized- | CorruptDatabase- | AncestorNotInChain !BlockHeight !BlockHash- | MempoolImportFailed- deriving (Show, Eq, Ord, Exception)--data Syncing = Syncing- { syncingPeer :: !Peer- , syncingTime :: !UnixTime- , syncingHead :: !BlockNode- }--data PendingTx = PendingTx- { pendingTxTime :: !UnixTime- , pendingTx :: !Tx- , pendingDeps :: !(HashSet TxHash)- }- deriving (Show, Eq, Ord)---- | Block store process state.-data BlockRead = BlockRead- { mySelf :: !BlockStore- , myConfig :: !BlockStoreConfig- , myPeer :: !(TVar (Maybe Syncing))- , myTxs :: !(TVar (HashMap TxHash PendingTx))- }---- | Configuration for a block store.-data BlockStoreConfig = BlockStoreConfig- { blockConfManager :: !PeerManager- -- ^ peer manager from running node- , blockConfChain :: !Chain- -- ^ chain from a running node- , blockConfListener :: !(Listen StoreEvent)- -- ^ listener for store events- , blockConfDB :: !DatabaseReader- -- ^ RocksDB database handle- , blockConfNet :: !Network- -- ^ network constants- , blockConfWipeMempool :: !Bool- -- ^ wipe mempool at start- , blockConfPeerTimeout :: !Int- -- ^ disconnect syncing peer if inactive for this long- }--type BlockT m = ReaderT BlockRead m-type WriterT m = ReaderT Writer (ExceptT ImportException (BlockT m))--runImport ::- MonadLoggerIO m- => WriterT m a- -> BlockT m (Either ImportException a)-runImport f = do- db <- asks (blockConfDB . myConfig)- runExceptT (runWriter db f)--runRocksDB :: ReaderT DatabaseReader m a -> BlockT m a-runRocksDB f =- ReaderT $ \BlockRead {myConfig = BlockStoreConfig {blockConfDB = db}} ->- runReaderT f db--instance MonadIO m => StoreRead (BlockT m) where- getMaxGap =- runRocksDB getMaxGap- getInitialGap =- runRocksDB getInitialGap- getNetwork =- runRocksDB getNetwork- getBestBlock =- runRocksDB getBestBlock- getBlocksAtHeight =- runRocksDB . getBlocksAtHeight- getBlock =- runRocksDB . getBlock- getTxData =- runRocksDB . getTxData- getSpender =- runRocksDB . getSpender- getSpenders =- runRocksDB . getSpenders- getUnspent =- runRocksDB . getUnspent- getBalance =- runRocksDB . getBalance- getMempool =- runRocksDB getMempool- getAddressesTxs as =- runRocksDB . getAddressesTxs as- getAddressesUnspents as =- runRocksDB . getAddressesUnspents as- getAddressUnspents a =- runRocksDB . getAddressUnspents a- getAddressTxs a =- runRocksDB . getAddressTxs a---- | Run block store process.-blockStore ::- (MonadUnliftIO m, MonadLoggerIO m)- => BlockStoreConfig- -> Inbox BlockStoreMessage- -> m ()-blockStore cfg inbox = do- pb <- newTVarIO Nothing- ts <- newTVarIO HashMap.empty- runReaderT- (ini >> wipe >> run)- BlockRead { mySelf = inboxToMailbox inbox- , myConfig = cfg- , myPeer = pb- , myTxs = ts- }- where- del x n txs =- forM (zip [x ..] txs) $ \(i, tx) -> do- $(logInfoS) "BlockStore" $- "Deleting "- <> cs (show i) <> "/" <> cs (show n) <> ": "- <> txHashToHex (txRefHash tx) <> "…"- deleteTx True False (txRefHash tx)- wipeit x n txs = do- let (txs1, txs2) = splitAt 1000 txs- unless (null txs1) $ runImport (del x n txs1) >>= \case- Left e -> do- $(logErrorS) "BlockStore" $- "Could not delete mempool, database corrupt: "- <> cs (show e)- throwIO CorruptDatabase- Right _ -> wipeit (x + length txs1) n txs2- wipe- | blockConfWipeMempool cfg =- getMempool >>= \mem -> wipeit 1 (length mem) mem- | otherwise = return ()- ini =- runImport initBest >>= \case- Left e -> do- $(logErrorS) "BlockStore" $- "Could not initialize block store: "- <> fromString (show e)- throwIO e- Right () -> return ()- run = withAsync (pingMe (inboxToMailbox inbox)) . const . forever $- receive inbox >>= ReaderT . runReaderT . processBlockStoreMessage--isInSync :: (MonadLoggerIO m, StoreRead m, MonadReader BlockRead m) => m Bool-isInSync = getBestBlock >>= \case- Nothing -> do- $(logErrorS) "BlockStore" "Block database uninitialized"- throwIO Uninitialized- Just bb -> do- cb <- asks (blockConfChain . myConfig) >>= chainGetBest- return (headerHash (nodeHeader cb) == bb)--mempool :: MonadLoggerIO m => Peer -> m ()-mempool p = do- $(logDebugS) "BlockStore" "Requesting mempool from network peer"- MMempool `sendMessage` p--processBlock ::- (MonadUnliftIO m, MonadLoggerIO m)- => Peer- -> Block- -> BlockT m ()-processBlock peer block = void . runMaybeT $ do- checks <- checkPeer peer- unless checks mzero- blocknode <- MaybeT $ getBlockNode peer blockhash- pt <- managerPeerText peer =<< asks (blockConfManager . myConfig)- $(logDebugS) "BlockStore" $- "Processing block : " <> blockText blocknode Nothing- <> " (peer" <> pt <> ")"- imported <- lift . runImport $ importBlock block blocknode- either (failure pt) (success blocknode) imported- where- header = blockHeader block- blockhash = headerHash header- hexhash = blockHashToHex blockhash- success blocknode deletedtxids = do- $(logInfoS) "BlockStore" $- "Best block: " <> blockText blocknode (Just block)- notify deletedtxids- failure pt e = do- $(logErrorS) "BlockStore" $- "Error importing block: "- <> hexhash <> ": " <> cs (show e)- <> " (peer " <> pt <> ")"- killPeer (PeerMisbehaving (show e)) peer- notify deletedtxids = do- listener <- asks (blockConfListener . myConfig)- atomically $ do- mapM_ (listener . StoreTxDeleted) deletedtxids- listener (StoreBestBlock blockhash)- lift (touchPeer peer >> syncMe peer)--checkPeer :: (MonadLoggerIO m, MonadReader BlockRead m) => Peer -> m Bool-checkPeer peer =- asks (blockConfManager . myConfig)- >>= managerGetPeer peer- >>= maybe disconnected (const connected)- where- disconnected = do- $(logDebugS) "BlockStore" "Ignoring data from disconnected peer"- return False- connected = touchPeer peer >>= \case- True -> return True- False -> remove- remove = do- pt <- managerPeerText peer =<< asks (blockConfManager . myConfig)- $(logDebugS) "BlockStore" $- "Ignoring data from non-syncing peer: " <> pt- killPeer (PeerMisbehaving "Sent unpexpected data") peer- return False--getBlockNode :: (MonadUnliftIO m, MonadLoggerIO m, MonadReader BlockRead m)- => Peer -> BlockHash -> m (Maybe BlockNode)-getBlockNode peer blockhash = runMaybeT $- asks (blockConfChain . myConfig)- >>= chainGetBlock blockhash- >>= \case- Nothing -> do- p' <- managerPeerText peer =<< asks (blockConfManager . myConfig)- $(logErrorS) "BlockStore" $- "Header not found for block: "- <> blockHashToHex blockhash- <> " (peer " <> p' <> ")"- killPeer (PeerMisbehaving "Sent unknown block") peer- mzero- Just n -> return n--processNoBlocks ::- (MonadUnliftIO m, MonadLoggerIO m)- => Peer- -> [BlockHash]- -> BlockT m ()-processNoBlocks p hs = do- p' <- managerPeerText p =<< asks (blockConfManager . myConfig)- forM_ (zip [(1 :: Int) ..] hs) $ \(i, h) ->- $(logErrorS) "BlockStore" $- "Block " <> cs (show i) <> "/" <> cs (show (length hs)) <> " "- <> blockHashToHex h <> " not found (peer " <> p' <> ")"- killPeer (PeerMisbehaving "Did not find requested block(s)") p--processTx :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> Tx -> BlockT m ()-processTx p tx = do- t <- fromIntegral . systemSeconds <$> liftIO getSystemTime- p' <- managerPeerText p =<< asks (blockConfManager . myConfig)- $(logDebugS) "BlockManager" $- "Received tx " <> txHashToHex (txHash tx) <> " (peer " <> p' <> ")"- addPendingTx $ PendingTx t tx HashSet.empty--pruneOrphans :: MonadIO m => BlockT m ()-pruneOrphans = do- ts <- asks myTxs- now <- fromIntegral . systemSeconds <$> liftIO getSystemTime- atomically . modifyTVar ts . HashMap.filter $ \p ->- pendingTxTime p > now - 600--addPendingTx :: MonadIO m => PendingTx -> BlockT m ()-addPendingTx p = do- ts <- asks myTxs- atomically $ modifyTVar ts $ HashMap.insert th p- where- th = txHash (pendingTx p)--isPending :: MonadIO m => TxHash -> BlockT m Bool-isPending th = do- ts <- asks myTxs- atomically $ HashMap.member th <$> readTVar ts--pendingTxs :: MonadIO m => Int -> BlockT m [PendingTx]-pendingTxs i = asks myTxs >>= \box -> atomically $ do- pending <- readTVar box- let (selected, rest) = select pending- writeTVar box rest- return selected- where- select pend =- let eligible = HashMap.filter (null . pendingDeps) pend- orphans = HashMap.difference pend eligible- selected = take i $ sortit eligible- remaining = HashMap.filter (`notElem` selected) eligible- in (selected, remaining <> orphans)- sortit m =- let sorted = sortTxs $ map pendingTx $ HashMap.elems m- txids = map (txHash . snd) sorted- in mapMaybe (`HashMap.lookup` m) txids--fulfillOrphans :: ( MonadIO m- , MonadReader BlockRead m- )- => TxHash- -> m ()-fulfillOrphans th = asks myTxs >>= \box ->- atomically $ modifyTVar box (HashMap.map fulfill)- where- fulfill p = p {pendingDeps = HashSet.delete th (pendingDeps p)}--updateOrphans- :: ( StoreRead m- , MonadLoggerIO m- , MonadReader BlockRead m- )- => m ()-updateOrphans = do- box <- asks myTxs- pending <- readTVarIO box- let orphans = HashMap.filter (not . null . pendingDeps) pending- updated <- forM orphans $ \p -> do- let tx = pendingTx p- exists (txHash tx) >>= \case- True -> return Nothing- False -> Just <$> fill_deps p- let pruned = HashMap.map fromJust $ HashMap.filter isJust updated- atomically $ writeTVar box pruned- where- exists th = getTxData th >>= \case- Nothing -> return False- Just TxData {txDataDeleted = True} -> return False- Just TxData {txDataDeleted = False} -> return True- prev_utxos tx = catMaybes <$> mapM (getUnspent . prevOutput) (txIn tx)- fulfill p unspent =- let unspent_hash = outPointHash (unspentPoint unspent)- new_deps = HashSet.delete unspent_hash (pendingDeps p)- in p {pendingDeps = new_deps}- fill_deps p = do- let tx = pendingTx p- unspents <- prev_utxos tx- return $ foldl fulfill p unspents--newOrphanTx :: (MonadUnliftIO m, MonadLoggerIO m)- => UnixTime -> Tx -> WriterT m ()-newOrphanTx time tx = do- $(logDebugS) "BlockStore" $- "Mempool "- <> txHashToHex (txHash tx)- <> ": Orphan"- box <- lift $ asks myTxs- unspents <- catMaybes <$> mapM getUnspent prevs- let unspent_set = HashSet.fromList (map unspentPoint unspents)- missing_set = HashSet.difference prev_set unspent_set- missing_txs = HashSet.map outPointHash missing_set- atomically . modifyTVar box $- HashMap.insert- (txHash tx)- PendingTx { pendingTxTime = time- , pendingTx = tx- , pendingDeps = missing_txs- }- where- prev_set = HashSet.fromList prevs- prevs = map prevOutput (txIn tx)--importMempoolTx :: (MonadUnliftIO m, MonadLoggerIO m)- => UnixTime -> Tx -> WriterT m (Maybe [TxHash])-importMempoolTx time tx =- catchError new_mempool_tx handle_error- where- tx_hash = txHash tx- handle_error Orphan = do- newOrphanTx time tx- return Nothing- handle_error _ = do- $(logDebugS) "BlockStore" $- "Mempool " <> txHashToHex tx_hash <> ": Failed"- return Nothing- new_mempool_tx =- newMempoolTx tx time >>= \case- Just deleted -> do- $(logDebugS) "BlockStore" $- "Mempool " <> txHashToHex (txHash tx) <> ": OK"- lift $ fulfillOrphans tx_hash- return (Just deleted)- Nothing -> do- $(logDebugS) "BlockStore" $- "Mempool " <> txHashToHex (txHash tx) <> ": No action"- return Nothing--processMempool :: (MonadUnliftIO m, MonadLoggerIO m) => BlockT m ()-processMempool = do- txs <- pendingTxs 2000- unless (null txs) $ import_txs txs >>= either failed success- where- run_import p = importMempoolTx (pendingTxTime p) (pendingTx p)- add_txid p deleted = (txHash (pendingTx p), deleted)- import_pending p = fmap (add_txid p) <$> run_import p- import_txs = runImport . mapM import_pending- failed e = do- $(logErrorS) "BlockStore" $- "Mempool import failed: " <> cs (show e)- throwIO MempoolImportFailed- success = mapM_ (uncurry notify) . catMaybes- notify txid deleted = do- l <- asks (blockConfListener . myConfig)- atomically $ do- mapM_ (l . StoreTxDeleted) deleted- l (StoreMempoolNew txid)--processTxs ::- (MonadUnliftIO m, MonadLoggerIO m)- => Peer- -> [TxHash]- -> BlockT m ()-processTxs p hs = isInSync >>= \s -> when s $ do- pt <- managerPeerText p =<< asks (blockConfManager . myConfig)- $(logDebugS) "BlockStore" $- "Received inventory with "- <> cs (show (length hs))- <> " transactions from peer " <> pt- xs <- catMaybes <$> zip_counter (process_tx pt)- unless (null xs) $ go xs- where- len = length hs- zip_counter = forM (zip [(1 :: Int) ..] hs) . uncurry- have_it h = isPending h >>= \case- True -> return True- False -> getTxData h >>= \case- Nothing -> return False- Just txd -> return (not (txDataDeleted txd))- process_tx pt i h = have_it h >>= \case- True -> do_have pt i h- False -> dont_have pt i h- do_have pt i h = do- $(logDebugS) "BlockStore" $- "Tx inv " <> cs (show i) <> "/" <> cs (show len)- <> ": " <> txHashToHex h- <> ": Already have (peer " <> pt <> ")"- return Nothing- dont_have pt i h = do- $(logDebugS) "BlockStore" $- "Tx inv " <> cs (show i) <> "/" <> cs (show len)- <> ": " <> txHashToHex h- <> ": Requesting… (peer " <> pt <> ")"- return (Just h)- go xs = do- net <- asks (blockConfNet . myConfig)- let inv = if getSegWit net then InvWitnessTx else InvTx- vec = map (InvVector inv . getTxHash) xs- msg = MGetData (GetData vec)- msg `sendMessage` p--touchPeer :: (MonadIO m, MonadReader BlockRead m)- => Peer -> m Bool-touchPeer p =- getSyncingState >>= \case- Just Syncing {syncingPeer = s}- | p == s -> do- box <- asks myPeer- now <- fromIntegral . systemSeconds <$>- liftIO getSystemTime- atomically . modifyTVar box . fmap $ \x ->- x {syncingTime = now}- return True- _ -> return False--checkTime :: (MonadUnliftIO m, MonadLoggerIO m) => BlockT m ()-checkTime =- asks myPeer >>= readTVarIO >>= \case- Nothing -> return ()- Just Syncing {syncingTime = t, syncingPeer = p} -> do- n <- fromIntegral . systemSeconds <$> liftIO getSystemTime- peertout <- asks (blockConfPeerTimeout . myConfig)- when (n > t + fromIntegral peertout) $ do- p' <- managerPeerText p =<< asks (blockConfManager . myConfig)- $(logErrorS) "BlockStore" $ "Timeout syncing peer " <> p'- resetPeer- killPeer PeerTimeout p--processDisconnect- :: (MonadUnliftIO m, MonadLoggerIO m)- => Peer- -> BlockT m ()-processDisconnect p =- asks myPeer >>= readTVarIO >>= \case- Just Syncing {syncingPeer = p'} | p == p' -> dc- _ -> return ()- where- dc = do- $(logDebugS) "BlockStore" "Syncing peer disconnected"- resetPeer- getPeer >>= \case- Nothing -> $(logWarnS) "BlockStore" "No peers available"- Just peer -> do- ns <- managerPeerText peer =<<- asks (blockConfManager . myConfig)- $(logDebugS) "BlockStore" $ "New syncing peer " <> ns- syncMe peer--pruneMempool :: (MonadUnliftIO m, MonadLoggerIO m) => BlockT m ()-pruneMempool = isInSync >>= \sync -> when sync $ do- now <- fromIntegral . systemSeconds <$> liftIO getSystemTime- getOldMempool now >>= \old -> unless (null old) $ delete_txs old- where- delete_txs = mapM_ delete_it- failed txid e =- $(logErrorS) "BlockStore" $- "Could not delete old mempool tx: "- <> txHashToHex txid <> ": "- <> cs (show e)- success txids = do- $(logDebugS) "BlockStore" $- "Deleted " <> cs (show (length txids)) <> " txs"- l <- asks (blockConfListener . myConfig)- atomically $ mapM_ (l . StoreTxDeleted) txids- delete_it txid = do- $(logDebugS) "BlockStore" $- "Deleting "- <> ": " <> txHashToHex txid- <> " (old mempool tx)…"- runImport (deleteTx True False txid)- >>= either (failed txid) success--syncMe :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> BlockT m ()-syncMe peer = void . runMaybeT $ do- checksyncingpeer- reverttomainchain- syncbest <- syncbestnode- bestblock <- bestblocknode- chainbest <- chainbestnode- end syncbest bestblock chainbest- blocknodes <- selectblocks chainbest syncbest- setPeer peer (last blocknodes)- net <- asks (blockConfNet . myConfig)- let inv = if getSegWit net then InvWitnessBlock else InvBlock- vecf = InvVector inv . getBlockHash . headerHash . nodeHeader- vectors = map vecf blocknodes- pt <- managerPeerText peer =<< asks (blockConfManager . myConfig)- $(logDebugS) "BlockStore" $- "Requesting "- <> fromString (show (length vectors))- <> " blocks from peer: " <> pt- MGetData (GetData vectors) `sendMessage` peer- where- checksyncingpeer = getSyncingState >>= \case- Nothing -> return ()- Just Syncing {syncingPeer = p}- | p == peer -> return ()- | otherwise -> mzero- chainbestnode = chainGetBest =<< asks (blockConfChain . myConfig)- bestblocknode = do- bb <- lift getBestBlock >>= \case- Nothing -> do- $(logErrorS) "BlockStore" "No best block set"- throwIO Uninitialized- Just b -> return b- ch <- asks (blockConfChain . myConfig)- chainGetBlock bb ch >>= \case- Nothing -> do- $(logErrorS) "BlockStore" $- "Header not found for best block: " <> blockHashToHex bb- throwIO (BlockNotInChain bb)- Just x -> return x- syncbestnode =- asks myPeer >>= readTVarIO >>= \case- Just Syncing {syncingHead = b} -> return b- Nothing -> bestblocknode- end syncbest bestblock chainbest- | nodeHeader bestblock == nodeHeader chainbest = do- lift updateOrphans- resetPeer- mempool peer- mzero- | nodeHeader syncbest == nodeHeader chainbest =- mzero- | nodeHeight syncbest > nodeHeight bestblock + 500 =- mzero- | otherwise =- return ()- selectblocks chainbest syncbest = do- let sync_height = maxsyncheight- (nodeHeight chainbest)- (nodeHeight syncbest)- synctop <- top chainbest sync_height- ch <- asks (blockConfChain . myConfig)- parents <- chainGetParents (nodeHeight syncbest + 1) synctop ch- return $ if length parents < 500- then parents <> [chainbest]- else parents- maxsyncheight chainheight syncbestheight- | chainheight <= syncbestheight + 501 = chainheight- | otherwise = syncbestheight + 501- top chainbest syncheight =- if syncheight == nodeHeight chainbest- then return chainbest- else findancestor chainbest syncheight- findancestor chainbest syncheight = do- ch <- asks (blockConfChain . myConfig)- m <- chainGetAncestor syncheight chainbest ch- case m of- Just x -> return x- Nothing -> do- $(logErrorS) "BlockStore" $- "Could not find header for ancestor of block: "- <> blockHashToHex (headerHash (nodeHeader chainbest))- throwIO $ AncestorNotInChain- syncheight- (headerHash (nodeHeader chainbest))- reverttomainchain = do- bestblockhash <- headerHash . nodeHeader <$> bestblocknode- ch <- asks (blockConfChain . myConfig)- chainBlockMain bestblockhash ch >>= \y ->- unless y $ do_revert bestblockhash- fail_revert e = do- $(logErrorS) "BlockStore" $- "Could not revert best block: " <> cs (show e)- throwIO e- success_revert bestblockhash txids = do- listener <- asks (blockConfListener . myConfig)- atomically $ do- mapM_ (listener . StoreTxDeleted) txids- listener (StoreBlockReverted bestblockhash)- reverttomainchain- do_revert bestblockhash = do- $(logErrorS) "BlockStore" $- "Reverting best block: "- <> blockHashToHex bestblockhash- resetPeer- lift (runImport (revertBlock bestblockhash)) >>=- either fail_revert (success_revert bestblockhash)--resetPeer :: (MonadLoggerIO m, MonadReader BlockRead m) => m ()-resetPeer = do- box <- asks myPeer- atomically $ writeTVar box Nothing--setPeer :: (MonadIO m, MonadReader BlockRead m) => Peer -> BlockNode -> m ()-setPeer p b = do- box <- asks myPeer- now <- fromIntegral . systemSeconds <$> liftIO getSystemTime- atomically . writeTVar box $- Just Syncing {syncingPeer = p, syncingHead = b, syncingTime = now}--getPeer :: (MonadIO m, MonadReader BlockRead m) => m (Maybe Peer)-getPeer = runMaybeT $ MaybeT syncingpeer <|> MaybeT onlinepeer- where- syncingpeer = fmap syncingPeer <$> getSyncingState- onlinepeer =- listToMaybe . map onlinePeerMailbox <$>- (managerGetPeers =<< asks (blockConfManager . myConfig))--getSyncingState :: (MonadIO m, MonadReader BlockRead m) => m (Maybe Syncing)-getSyncingState = readTVarIO =<< asks myPeer--processBlockStoreMessage ::- (MonadUnliftIO m, MonadLoggerIO m)- => BlockStoreMessage- -> BlockT m ()-processBlockStoreMessage (BlockNewBest _) =- getPeer >>= \case- Nothing -> return ()- Just p -> syncMe p-processBlockStoreMessage (BlockPeerConnect p _) = syncMe p-processBlockStoreMessage (BlockPeerDisconnect p _sa) = processDisconnect p-processBlockStoreMessage (BlockReceived p b) = processBlock p b-processBlockStoreMessage (BlockNotFound p bs) = processNoBlocks p bs-processBlockStoreMessage (TxRefReceived p tx) = processTx p tx-processBlockStoreMessage (TxRefAvailable p ts) = processTxs p ts-processBlockStoreMessage (BlockPing r) = do- processMempool- pruneOrphans- checkTime- pruneMempool- atomically (r ())--pingMe :: MonadLoggerIO m => BlockStore -> m ()-pingMe mbox =- forever $ do- threadDelay =<< liftIO (randomRIO (100 * 1000, 1000 * 1000))- BlockPing `query` mbox--blockStorePeerConnect :: MonadIO m => Peer -> SockAddr -> BlockStore -> m ()-blockStorePeerConnect peer addr store = BlockPeerConnect peer addr `send` store--blockStorePeerDisconnect :: MonadIO m => Peer -> SockAddr -> BlockStore -> m ()-blockStorePeerDisconnect peer addr store =- BlockPeerDisconnect peer addr `send` store--blockStoreHead :: MonadIO m => BlockNode -> BlockStore -> m ()-blockStoreHead node store = BlockNewBest node `send` store--blockStoreBlock :: MonadIO m => Peer -> Block -> BlockStore -> m ()-blockStoreBlock peer block store = BlockReceived peer block `send` store--blockStoreNotFound :: MonadIO m => Peer -> [BlockHash] -> BlockStore -> m ()-blockStoreNotFound peer blocks store = BlockNotFound peer blocks `send` store--blockStoreTx :: MonadIO m => Peer -> Tx -> BlockStore -> m ()-blockStoreTx peer tx store = TxRefReceived peer tx `send` store--blockStoreTxHash :: MonadIO m => Peer -> [TxHash] -> BlockStore -> m ()-blockStoreTxHash peer txhashes store =- TxRefAvailable peer txhashes `send` store--blockStorePeerConnectSTM :: Peer -> SockAddr -> BlockStore -> STM ()-blockStorePeerConnectSTM peer addr store = BlockPeerConnect peer addr `sendSTM` store--blockStorePeerDisconnectSTM :: Peer -> SockAddr -> BlockStore -> STM ()-blockStorePeerDisconnectSTM peer addr store =- BlockPeerDisconnect peer addr `sendSTM` store--blockStoreHeadSTM :: BlockNode -> BlockStore -> STM ()-blockStoreHeadSTM node store = BlockNewBest node `sendSTM` store--blockStoreBlockSTM :: Peer -> Block -> BlockStore -> STM ()-blockStoreBlockSTM peer block store = BlockReceived peer block `sendSTM` store--blockStoreNotFoundSTM :: Peer -> [BlockHash] -> BlockStore -> STM ()-blockStoreNotFoundSTM peer blocks store = BlockNotFound peer blocks `sendSTM` store--blockStoreTxSTM :: Peer -> Tx -> BlockStore -> STM ()-blockStoreTxSTM peer tx store = TxRefReceived peer tx `sendSTM` store--blockStoreTxHashSTM :: Peer -> [TxHash] -> BlockStore -> STM ()-blockStoreTxHashSTM peer txhashes store =- TxRefAvailable peer txhashes `sendSTM` store--blockText :: BlockNode -> Maybe Block -> Text-blockText bn mblock = case mblock of- Nothing ->- height <> sep <> time <> sep <> hash- Just block ->- height <> sep <> time <> sep <> hash <> sep <> size block- where- height = cs $ show (nodeHeight bn)- systime =- posixSecondsToUTCTime (fromIntegral (blockTimestamp (nodeHeader bn)))- time =- cs $- formatTime defaultTimeLocale (iso8601DateFormat (Just "%H:%M")) systime+ , BlockStoreInbox+ , BlockStoreConfig(..)+ , blockStore+ , blockStorePeerConnect+ , blockStorePeerConnectSTM+ , blockStorePeerDisconnect+ , blockStorePeerDisconnectSTM+ , blockStoreHead+ , blockStoreHeadSTM+ , blockStoreBlock+ , blockStoreBlockSTM+ , blockStoreNotFound+ , blockStoreNotFoundSTM+ , blockStoreTx+ , blockStoreTxSTM+ , blockStoreTxHash+ , blockStoreTxHashSTM+ ) where++import Control.Monad (forM, forM_, forever, mzero,+ unless, void, when)+import Control.Monad.Except (ExceptT (..), MonadError,+ catchError, runExceptT)+import Control.Monad.Logger (MonadLoggerIO, logDebugS,+ logErrorS, logInfoS, logWarnS)+import Control.Monad.Reader (MonadReader, ReaderT (..), ask,+ asks)+import Control.Monad.Trans (lift)+import Control.Monad.Trans.Maybe (runMaybeT)+import qualified Data.ByteString as B+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as HashMap+import Data.HashSet (HashSet)+import qualified Data.HashSet as HashSet+import Data.List (delete)+import Data.Maybe (catMaybes, fromJust, isJust,+ mapMaybe)+import Data.Serialize (encode)+import Data.String (fromString)+import Data.String.Conversions (cs)+import Data.Text (Text)+import Data.Time.Clock (NominalDiffTime, UTCTime,+ diffUTCTime, getCurrentTime)+import Data.Time.Clock.POSIX (posixSecondsToUTCTime,+ utcTimeToPOSIXSeconds)+import Data.Time.Format (defaultTimeLocale, formatTime,+ iso8601DateFormat)+import Haskoin (Block (..), BlockHash (..),+ BlockHeader (..), BlockHeight,+ BlockNode (..), GetData (..),+ InvType (..), InvVector (..),+ Message (..), Network (..),+ OutPoint (..), Tx (..),+ TxHash (..), TxIn (..),+ blockHashToHex, headerHash,+ txHash, txHashToHex)+import Haskoin.Node (Chain, OnlinePeer (..), Peer,+ PeerException (..), PeerManager,+ chainBlockMain,+ chainGetAncestor, chainGetBest,+ chainGetBlock, chainGetParents,+ getPeers, killPeer, peerText,+ sendMessage, setBusy, setFree)+import Haskoin.Store.Common+import Haskoin.Store.Data (TxData (..), TxRef (..),+ Unspent (..))+import Haskoin.Store.Database.Reader (DatabaseReader)+import Haskoin.Store.Database.Writer+import Haskoin.Store.Logic (ImportException (Orphan),+ deleteUnconfirmedTx,+ getOldMempool, importBlock,+ initBest, newMempoolTx,+ revertBlock)+import NQE (Inbox, Listen, Mailbox,+ Publisher, inboxToMailbox,+ publish, query, receive, send,+ sendSTM)+import System.Random (randomRIO)+import UnliftIO (Exception, MonadIO,+ MonadUnliftIO, STM, TVar, async,+ atomically, liftIO, modifyTVar,+ newTVarIO, readTVar, readTVarIO,+ throwIO, withAsync, writeTVar)+import UnliftIO.Concurrent (threadDelay)++data BlockStoreMessage+ = BlockNewBest !BlockNode+ | BlockPeerConnect !Peer+ | BlockPeerDisconnect !Peer+ | BlockReceived !Peer !Block+ | BlockNotFound !Peer ![BlockHash]+ | TxRefReceived !Peer !Tx+ | TxRefAvailable !Peer ![TxHash]+ | BlockPing !(Listen ())++type BlockStoreInbox = Inbox BlockStoreMessage+type BlockStore = Mailbox BlockStoreMessage++data BlockException+ = BlockNotInChain !BlockHash+ | Uninitialized+ | CorruptDatabase+ | AncestorNotInChain !BlockHeight !BlockHash+ | MempoolImportFailed+ deriving (Show, Eq, Ord, Exception)++data Syncing =+ Syncing+ { syncingPeer :: !Peer+ , syncingTime :: !UTCTime+ , syncingBlocks :: ![BlockHash]+ }++data PendingTx =+ PendingTx+ { pendingTxTime :: !UTCTime+ , pendingTx :: !Tx+ , pendingDeps :: !(HashSet TxHash)+ }+ deriving (Show, Eq, Ord)++-- | Block store process state.+data BlockRead =+ BlockRead+ { mySelf :: !BlockStore+ , myConfig :: !BlockStoreConfig+ , myPeer :: !(TVar (Maybe Syncing))+ , myTxs :: !(TVar (HashMap TxHash PendingTx))+ , requested :: !(TVar (HashSet TxHash))+ }++-- | Configuration for a block store.+data BlockStoreConfig =+ BlockStoreConfig+ { blockConfManager :: !PeerManager+ -- ^ peer manager from running node+ , blockConfChain :: !Chain+ -- ^ chain from a running node+ , blockConfListener :: !(Publisher StoreEvent)+ -- ^ listener for store events+ , blockConfDB :: !DatabaseReader+ -- ^ RocksDB database handle+ , blockConfNet :: !Network+ -- ^ network constants+ , blockConfWipeMempool :: !Bool+ -- ^ wipe mempool at start+ , blockConfPeerTimeout :: !NominalDiffTime+ -- ^ disconnect syncing peer if inactive for this long+ }++type BlockT m = ReaderT BlockRead m++runImport :: MonadLoggerIO m+ => WriterT (ExceptT ImportException m) a+ -> BlockT m (Either ImportException a)+runImport f =+ ReaderT $ \r -> runExceptT $ runWriter (blockConfDB (myConfig r)) f++runRocksDB :: ReaderT DatabaseReader m a -> BlockT m a+runRocksDB f =+ ReaderT $ runReaderT f . blockConfDB . myConfig++instance MonadIO m => StoreRead (BlockT m) where+ getMaxGap =+ runRocksDB getMaxGap+ getInitialGap =+ runRocksDB getInitialGap+ getNetwork =+ runRocksDB getNetwork+ getBestBlock =+ runRocksDB getBestBlock+ getBlocksAtHeight =+ runRocksDB . getBlocksAtHeight+ getBlock =+ runRocksDB . getBlock+ getTxData =+ runRocksDB . getTxData+ getSpender =+ runRocksDB . getSpender+ getSpenders =+ runRocksDB . getSpenders+ getUnspent =+ runRocksDB . getUnspent+ getBalance =+ runRocksDB . getBalance+ getMempool =+ runRocksDB getMempool+ getAddressesTxs as =+ runRocksDB . getAddressesTxs as+ getAddressesUnspents as =+ runRocksDB . getAddressesUnspents as+ getAddressUnspents a =+ runRocksDB . getAddressUnspents a+ getAddressTxs a =+ runRocksDB . getAddressTxs a++-- | Run block store process.+blockStore ::+ (MonadUnliftIO m, MonadLoggerIO m)+ => BlockStoreConfig+ -> BlockStoreInbox+ -> m ()+blockStore cfg inbox = do+ pb <- newTVarIO Nothing+ ts <- newTVarIO HashMap.empty+ rq <- newTVarIO HashSet.empty+ runReaderT+ (ini >> wipe >> run)+ BlockRead { mySelf = inboxToMailbox inbox+ , myConfig = cfg+ , myPeer = pb+ , myTxs = ts+ , requested = rq+ }+ where+ del txs = do+ $(logInfoS) "BlockStore" $+ "Deleting " <> cs (show (length txs)) <> " transactions"+ forM_ txs $ \tx ->+ deleteUnconfirmedTx False (txRefHash tx)+ wipeit txs = do+ let (txs1, txs2) = splitAt 1000 txs+ unless (null txs1) $+ runImport (del txs1) >>= \case+ Left e -> do+ $(logErrorS) "BlockStore" $+ "Could not wipe mempool: " <> cs (show e)+ throwIO e+ Right () -> wipeit txs2+ wipe+ | blockConfWipeMempool cfg =+ getMempool >>= wipeit+ | otherwise =+ return ()+ ini = runImport initBest >>= \case+ Left e -> do+ $(logErrorS) "BlockStore" $+ "Could not initialize: " <> cs (show e)+ throwIO e+ Right () -> return ()+ run = withAsync (pingMe (inboxToMailbox inbox))+ $ const+ $ forever+ $ receive inbox >>=+ ReaderT . runReaderT . processBlockStoreMessage++isInSync :: MonadLoggerIO m => BlockT m Bool+isInSync =+ getBestBlock >>= \case+ Nothing -> do+ $(logErrorS) "BlockStore" "Block database uninitialized"+ throwIO Uninitialized+ Just bb -> do+ cb <- asks (blockConfChain . myConfig) >>= chainGetBest+ if headerHash (nodeHeader cb) == bb+ then clearSyncingState >> return True+ else return False++mempool :: MonadLoggerIO m => Peer -> m ()+mempool p = do+ $(logDebugS) "BlockStore" $+ "Requesting mempool from peer: " <> peerText p+ MMempool `sendMessage` p++processBlock :: MonadLoggerIO m => Peer -> Block -> BlockT m ()+processBlock peer block = void . runMaybeT $ do+ checkPeer peer >>= \case+ True -> return ()+ False -> do+ $(logErrorS) "BlockStore" $+ "Non-syncing peer " <> peerText peer+ <> " sent me a block: "+ <> blockHashToHex blockhash+ PeerMisbehaving "Sent unexpected block" `killPeer` peer+ mzero+ node <- getBlockNode blockhash >>= \case+ Just b -> return b+ Nothing -> do+ $(logErrorS) "BlockStore" $+ "Peer " <> peerText peer+ <> " sent unknown block: "+ <> blockHashToHex blockhash+ PeerMisbehaving "Sent unknown block" `killPeer` peer+ mzero+ $(logDebugS) "BlockStore" $+ "Processing block: " <> blockText node Nothing+ <> " from peer: " <> peerText peer+ lift $ runImport (importBlock block node) >>= \case+ Left e -> failure e+ Right () -> success node+ where+ header = blockHeader block+ blockhash = headerHash header+ hexhash = blockHashToHex blockhash+ success node = do+ $(logInfoS) "BlockStore" $+ "Best block: " <> blockText node (Just block)+ notify+ removeSyncingBlock $ headerHash $ nodeHeader node+ touchPeer+ isInSync >>= \case+ False -> syncMe+ True -> do+ updateOrphans+ mempool peer+ failure e = do+ $(logErrorS) "BlockStore" $+ "Error importing block " <> hexhash+ <> " from peer: " <> peerText peer <> ": "+ <> cs (show e)+ killPeer (PeerMisbehaving (show e)) peer+ notify = do+ listener <- asks (blockConfListener . myConfig)+ publish (StoreBestBlock blockhash) listener++setSyncingBlocks :: (MonadReader BlockRead m, MonadIO m)+ => [BlockHash] -> m ()+setSyncingBlocks hs =+ asks myPeer >>= \box ->+ atomically $ modifyTVar box $ \case+ Nothing -> Nothing+ Just x -> Just x { syncingBlocks = hs }++getSyncingBlocks :: (MonadReader BlockRead m, MonadIO m) => m [BlockHash]+getSyncingBlocks =+ asks myPeer >>= readTVarIO >>= \case+ Nothing -> return []+ Just x -> return $ syncingBlocks x++addSyncingBlocks :: (MonadReader BlockRead m, MonadIO m)+ => [BlockHash] -> m ()+addSyncingBlocks hs =+ asks myPeer >>= \box ->+ atomically $ modifyTVar box $ \case+ Nothing -> Nothing+ Just x -> Just x { syncingBlocks = syncingBlocks x <> hs }++removeSyncingBlock :: (MonadReader BlockRead m, MonadIO m)+ => BlockHash -> m ()+removeSyncingBlock h = do+ box <- asks myPeer+ atomically $ modifyTVar box $ \case+ Nothing -> Nothing+ Just x -> Just x { syncingBlocks = delete h (syncingBlocks x) }++checkPeer :: (MonadLoggerIO m, MonadReader BlockRead m) => Peer -> m Bool+checkPeer p =+ fmap syncingPeer <$> getSyncingState >>= \case+ Nothing -> return False+ Just p' -> return $ p == p'++getBlockNode :: (MonadLoggerIO m, MonadReader BlockRead m)+ => BlockHash -> m (Maybe BlockNode)+getBlockNode blockhash =+ chainGetBlock blockhash =<< asks (blockConfChain . myConfig)++processNoBlocks ::+ MonadLoggerIO m+ => Peer+ -> [BlockHash]+ -> BlockT m ()+processNoBlocks p hs = do+ forM_ (zip [(1 :: Int) ..] hs) $ \(i, h) ->+ $(logErrorS) "BlockStore" $+ "Block "+ <> cs (show i) <> "/"+ <> cs (show (length hs)) <> " "+ <> blockHashToHex h+ <> " not found by peer: "+ <> peerText p+ killPeer (PeerMisbehaving "Did not find requested block(s)") p++processTx :: MonadLoggerIO m => Peer -> Tx -> BlockT m ()+processTx p tx = do+ t <- liftIO getCurrentTime+ $(logDebugS) "BlockManager" $+ "Received tx " <> txHashToHex (txHash tx)+ <> " by peer: " <> peerText p+ addPendingTx $ PendingTx t tx HashSet.empty++pruneOrphans :: MonadIO m => BlockT m ()+pruneOrphans = do+ ts <- asks myTxs+ now <- liftIO getCurrentTime+ atomically . modifyTVar ts . HashMap.filter $ \p ->+ now `diffUTCTime` pendingTxTime p > 600++addPendingTx :: MonadIO m => PendingTx -> BlockT m ()+addPendingTx p = do+ ts <- asks myTxs+ atomically $ modifyTVar ts $ HashMap.insert th p+ where+ th = txHash (pendingTx p)++addRequestedTx :: MonadIO m => TxHash -> BlockT m ()+addRequestedTx th = do+ qbox <- asks requested+ atomically $ modifyTVar qbox $ HashSet.insert th+ liftIO $ void $ async $ do+ threadDelay 20000000+ atomically $ modifyTVar qbox $ HashSet.delete th++isPending :: MonadIO m => TxHash -> BlockT m Bool+isPending th = do+ tbox <- asks myTxs+ qbox <- asks requested+ atomically $ do+ ts <- readTVar tbox+ rs <- readTVar qbox+ return $ th `HashMap.member` ts+ || th `HashSet.member` rs++pendingTxs :: MonadIO m => Int -> BlockT m [PendingTx]+pendingTxs i = asks myTxs >>= \box -> atomically $ do+ pending <- readTVar box+ let (selected, rest) = select pending+ writeTVar box rest+ return selected+ where+ select pend =+ let eligible = HashMap.filter (null . pendingDeps) pend+ orphans = HashMap.difference pend eligible+ selected = take i $ sortit eligible+ remaining = HashMap.filter (`notElem` selected) eligible+ in (selected, remaining <> orphans)+ sortit m =+ let sorted = sortTxs $ map pendingTx $ HashMap.elems m+ txids = map (txHash . snd) sorted+ in mapMaybe (`HashMap.lookup` m) txids++fulfillOrphans :: MonadIO m => BlockRead -> TxHash -> m ()+fulfillOrphans block_read th =+ atomically $ modifyTVar box (HashMap.map fulfill)+ where+ box = myTxs block_read+ fulfill p = p {pendingDeps = HashSet.delete th (pendingDeps p)}++updateOrphans+ :: ( StoreRead m+ , MonadLoggerIO m+ , MonadReader BlockRead m+ )+ => m ()+updateOrphans = do+ box <- asks myTxs+ pending <- readTVarIO box+ let orphans = HashMap.filter (not . null . pendingDeps) pending+ updated <- forM orphans $ \p -> do+ let tx = pendingTx p+ exists (txHash tx) >>= \case+ True -> return Nothing+ False -> Just <$> fill_deps p+ let pruned = HashMap.map fromJust $ HashMap.filter isJust updated+ atomically $ writeTVar box pruned+ where+ exists th = getTxData th >>= \case+ Nothing -> return False+ Just TxData {txDataDeleted = True} -> return False+ Just TxData {txDataDeleted = False} -> return True+ prev_utxos tx = catMaybes <$> mapM (getUnspent . prevOutput) (txIn tx)+ fulfill p unspent =+ let unspent_hash = outPointHash (unspentPoint unspent)+ new_deps = HashSet.delete unspent_hash (pendingDeps p)+ in p {pendingDeps = new_deps}+ fill_deps p = do+ let tx = pendingTx p+ unspents <- prev_utxos tx+ return $ foldl fulfill p unspents++newOrphanTx :: MonadLoggerIO m+ => BlockRead+ -> UTCTime+ -> Tx+ -> WriterT m ()+newOrphanTx block_read time tx = do+ $(logDebugS) "BlockStore" $+ "Import tx "+ <> txHashToHex (txHash tx)+ <> ": Orphan"+ let box = myTxs block_read+ unspents <- catMaybes <$> mapM getUnspent prevs+ let unspent_set = HashSet.fromList (map unspentPoint unspents)+ missing_set = HashSet.difference prev_set unspent_set+ missing_txs = HashSet.map outPointHash missing_set+ atomically . modifyTVar box $+ HashMap.insert+ (txHash tx)+ PendingTx { pendingTxTime = time+ , pendingTx = tx+ , pendingDeps = missing_txs+ }+ where+ prev_set = HashSet.fromList prevs+ prevs = map prevOutput (txIn tx)++importMempoolTx+ :: (MonadLoggerIO m, MonadError ImportException m)+ => BlockRead+ -> UTCTime+ -> Tx+ -> WriterT m Bool+importMempoolTx block_read time tx =+ catchError new_mempool_tx handle_error+ where+ tx_hash = txHash tx+ handle_error Orphan = do+ newOrphanTx block_read time tx+ $(logWarnS) "BlockStore" $+ "Import tx " <> txHashToHex tx_hash+ <> ": Orphan"+ return False+ handle_error _ = do+ $(logWarnS) "BlockStore" $+ "Import tx " <> txHashToHex tx_hash+ <> ": Failed"+ return False+ seconds = floor (utcTimeToPOSIXSeconds time)+ new_mempool_tx =+ newMempoolTx tx seconds >>= \case+ True -> do+ $(logInfoS) "BlockStore" $+ "Import tx " <> txHashToHex (txHash tx)+ <> ": OK"+ fulfillOrphans block_read tx_hash+ return True+ False -> do+ $(logInfoS) "BlockStore" $+ "Import tx " <> txHashToHex (txHash tx)+ <> ": Already imported"+ return False++processMempool :: MonadLoggerIO m => BlockT m ()+processMempool = do+ txs <- pendingTxs 2000+ block_read <- ask+ unless (null txs) (import_txs block_read txs >>= success)+ where+ run_import block_read p =+ importMempoolTx+ block_read+ (pendingTxTime p)+ (pendingTx p) >>= \case+ True -> return $ Just (txHash (pendingTx p))+ False -> return Nothing+ import_txs block_read txs =+ let r = mapM (run_import block_read) txs+ in runImport r >>= \case+ Left e -> report_error e >> return []+ Right ms -> return (catMaybes ms)+ report_error e = do+ $(logErrorS) "BlockImport" $+ "Error processing mempool: " <> cs (show e)+ throwIO e+ success = mapM_ notify+ notify txid = do+ listener <- asks (blockConfListener . myConfig)+ publish (StoreMempoolNew txid) listener++processTxs ::+ MonadLoggerIO m+ => Peer+ -> [TxHash]+ -> BlockT m ()+processTxs p hs = do+ s <- isInSync+ when s $ do+ $(logDebugS) "BlockStore" $+ "Received inventory with "+ <> cs (show (length hs))+ <> " transactions from peer: "+ <> peerText p+ xs <- catMaybes <$> zip_counter process_tx+ unless (null xs) $ go xs+ where+ len = length hs+ zip_counter = forM (zip [(1 :: Int) ..] hs) . uncurry+ process_tx i h =+ isPending h >>= \case+ True -> do+ $(logDebugS) "BlockStore" $+ "Tx " <> cs (show i) <> "/" <> cs (show len)+ <> " " <> txHashToHex h <> ": "+ <> "Pending"+ return Nothing+ False -> getActiveTxData h >>= \case+ Just _ -> do+ $(logDebugS) "BlockStore" $+ "Tx " <> cs (show i) <> "/" <> cs (show len)+ <> " " <> txHashToHex h <> ": "+ <> "Already Imported"+ return Nothing+ Nothing -> do+ $(logDebugS) "BlockStore" $+ "Tx " <> cs (show i) <> "/" <> cs (show len)+ <> " " <> txHashToHex h <> ": "+ <> "Requesting"+ return (Just h)+ go xs = do+ mapM_ addRequestedTx xs+ net <- asks (blockConfNet . myConfig)+ let inv = if getSegWit net then InvWitnessTx else InvTx+ vec = map (InvVector inv . getTxHash) xs+ msg = MGetData (GetData vec)+ msg `sendMessage` p++touchPeer :: ( MonadIO m+ , MonadReader BlockRead m+ )+ => m ()+touchPeer =+ getSyncingState >>= \case+ Nothing -> return ()+ Just _ -> do+ box <- asks myPeer+ now <- liftIO getCurrentTime+ atomically $+ modifyTVar box $+ fmap $ \x -> x { syncingTime = now }++checkTime :: MonadLoggerIO m => BlockT m ()+checkTime =+ asks myPeer >>= readTVarIO >>= \case+ Nothing -> return ()+ Just Syncing { syncingTime = t+ , syncingPeer = p+ } -> do+ now <- liftIO getCurrentTime+ peer_time_out <- asks (blockConfPeerTimeout . myConfig)+ when (now `diffUTCTime` t > peer_time_out) $ do+ $(logErrorS) "BlockStore" $+ "Syncing peer timeout: " <> peerText p+ killPeer PeerTimeout p++pruneMempool :: MonadLoggerIO m => BlockT m ()+pruneMempool = do+ s <- isInSync+ when s $ do+ now <- liftIO getCurrentTime+ let seconds = floor (utcTimeToPOSIXSeconds now)+ getOldMempool seconds >>= \old ->+ unless (null old) $+ runImport (mapM_ delete_it old) >>= \case+ Left e -> do+ $(logErrorS) "BlockStore" $+ "Could not prune mempool: "+ <> cs (show e)+ throwIO e+ Right x -> return x+ where+ delete_it txid = do+ $(logDebugS) "BlockStore" $+ "Deleting "+ <> ": " <> txHashToHex txid+ <> " (old mempool tx)"+ deleteUnconfirmedTx False txid++revertToMainChain :: MonadLoggerIO m => BlockT m ()+revertToMainChain = do+ h <- headerHash . nodeHeader <$> getBest+ ch <- asks (blockConfChain . myConfig)+ chainBlockMain h ch >>= \x -> unless x $ do+ $(logWarnS) "BlockStore" $+ "Reverting best block: "+ <> blockHashToHex h+ runImport (revertBlock h) >>= \case+ Left e -> do+ $(logErrorS) "BlockStore" $+ "Could not revert block "+ <> blockHashToHex h+ <> ": " <> cs (show e)+ throwIO e+ Right () -> setSyncingBlocks []+ revertToMainChain++getBest :: MonadLoggerIO m => BlockT m BlockNode+getBest = do+ bb <- getBestBlock >>= \case+ Just b -> return b+ Nothing -> do+ $(logErrorS) "BlockStore" "No best block set"+ throwIO Uninitialized+ ch <- asks (blockConfChain . myConfig)+ chainGetBlock bb ch >>= \case+ Just x -> return x+ Nothing -> do+ $(logErrorS) "BlockStore" $+ "Header not found for best block: "+ <> blockHashToHex bb+ throwIO (BlockNotInChain bb)++getSyncBest :: MonadLoggerIO m => BlockT m BlockNode+getSyncBest = do+ bb <- getSyncingBlocks >>= \case+ [] -> getBestBlock >>= \case+ Just b -> return b+ Nothing -> do+ $(logErrorS) "BlockStore" "No best block set"+ throwIO Uninitialized+ hs -> return $ last hs+ ch <- asks (blockConfChain . myConfig)+ chainGetBlock bb ch >>= \case+ Just x -> return x+ Nothing -> do+ $(logErrorS) "BlockStore" $+ "Header not found for block: "+ <> blockHashToHex bb+ throwIO (BlockNotInChain bb)++shouldSync :: MonadLoggerIO m => BlockT m (Maybe Peer)+shouldSync =+ isInSync >>= \case+ True -> return Nothing+ False -> cont+ where+ cont = getSyncingState >>= \case+ Nothing -> return Nothing+ Just Syncing { syncingPeer = p, syncingBlocks = bs }+ | 100 > length bs -> return (Just p)+ | otherwise -> return Nothing++syncMe :: MonadLoggerIO m => BlockT m ()+syncMe = do+ revertToMainChain+ shouldSync >>= \case+ Nothing -> return ()+ Just p -> do+ bb <- getSyncBest+ bh <- getbh+ when (bb /= bh) $ do+ bns <- sel bb bh+ iv <- getiv bns+ $(logDebugS) "BlockStore" $+ "Requesting "+ <> fromString (show (length iv))+ <> " blocks from peer: " <> peerText p+ addSyncingBlocks $ map (headerHash . nodeHeader) bns+ MGetData (GetData iv) `sendMessage` p+ where+ getiv bns = do+ w <- getSegWit <$> asks (blockConfNet . myConfig)+ let i = if w then InvWitnessBlock else InvBlock+ f = InvVector i . getBlockHash . headerHash . nodeHeader+ return $ map f bns+ getbh =+ chainGetBest =<< asks (blockConfChain . myConfig)+ sel bb bh = do+ let sh = geth bb bh+ t <- top sh bh+ ch <- asks (blockConfChain . myConfig)+ ps <- chainGetParents (nodeHeight bb + 1) t ch+ return $ if 500 > length ps+ then ps <> [bh]+ else ps+ geth bb bh =+ min (nodeHeight bb + 501)+ (nodeHeight bh)+ top sh bh =+ if sh == nodeHeight bh+ then return bh+ else findAncestor sh bh++findAncestor :: (MonadLoggerIO m, MonadReader BlockRead m)+ => BlockHeight -> BlockNode -> m BlockNode+findAncestor height target = do+ ch <- asks (blockConfChain . myConfig)+ chainGetAncestor height target ch >>= \case+ Just ancestor -> return ancestor+ Nothing -> do+ let h = headerHash $ nodeHeader target+ $(logErrorS) "BlockStore" $+ "Could not find header for ancestor of block "+ <> blockHashToHex h <> " at height "+ <> cs (show (nodeHeight target))+ throwIO $ AncestorNotInChain height h++finishPeer :: (MonadLoggerIO m, MonadReader BlockRead m)+ => Peer -> m ()+finishPeer p = do+ box <- asks myPeer+ readTVarIO box >>= \case+ Just Syncing { syncingPeer = p' } | p == p' -> reset_it box+ _ -> return ()+ where+ reset_it box = do+ atomically $ writeTVar box Nothing+ $(logDebugS) "BlockStore" $ "Releasing peer: " <> peerText p+ setFree p++trySetPeer :: MonadLoggerIO m => Peer -> BlockT m Bool+trySetPeer p =+ getSyncingState >>= \case+ Just _ -> return False+ Nothing -> set_it+ where+ set_it =+ setBusy p >>= \case+ False -> return False+ True -> do+ $(logDebugS) "BlockStore" $+ "Locked peer: " <> peerText p+ box <- asks myPeer+ now <- liftIO getCurrentTime+ atomically . writeTVar box $+ Just Syncing { syncingPeer = p+ , syncingTime = now+ , syncingBlocks = []+ }+ return True++trySyncing :: MonadLoggerIO m => BlockT m ()+trySyncing =+ isInSync >>= \case+ True -> return ()+ False -> getSyncingState >>= \case+ Just _ -> return ()+ Nothing -> online_peer+ where+ recurse [] = return ()+ recurse (p : ps) =+ trySetPeer p >>= \case+ False -> recurse ps+ True -> syncMe+ online_peer = do+ ops <- getPeers =<< asks (blockConfManager . myConfig)+ let ps = map onlinePeerMailbox ops+ recurse ps++trySyncingPeer :: MonadLoggerIO m => Peer -> BlockT m ()+trySyncingPeer p =+ isInSync >>= \case+ True -> return ()+ False -> trySetPeer p >>= \case+ False -> return ()+ True -> syncMe++getSyncingState+ :: (MonadIO m, MonadReader BlockRead m) => m (Maybe Syncing)+getSyncingState =+ readTVarIO =<< asks myPeer++clearSyncingState+ :: (MonadLoggerIO m, MonadReader BlockRead m) => m ()+clearSyncingState =+ asks myPeer >>= readTVarIO >>= \case+ Nothing -> return ()+ Just Syncing { syncingPeer = p } -> finishPeer p++processBlockStoreMessage+ :: MonadLoggerIO m => BlockStoreMessage -> BlockT m ()++processBlockStoreMessage (BlockNewBest _) =+ trySyncing++processBlockStoreMessage (BlockPeerConnect p) =+ trySyncingPeer p++processBlockStoreMessage (BlockPeerDisconnect p) =+ finishPeer p++processBlockStoreMessage (BlockReceived p b) =+ processBlock p b++processBlockStoreMessage (BlockNotFound p bs) =+ processNoBlocks p bs++processBlockStoreMessage (TxRefReceived p tx) =+ processTx p tx++processBlockStoreMessage (TxRefAvailable p ts) =+ processTxs p ts++processBlockStoreMessage (BlockPing r) = do+ trySyncing+ processMempool+ pruneOrphans+ checkTime+ pruneMempool+ atomically (r ())++pingMe :: MonadLoggerIO m => BlockStore -> m ()+pingMe mbox =+ forever $ do+ delay <- liftIO $+ randomRIO ( 100 * 1000+ , 1000 * 1000 )+ threadDelay delay+ BlockPing `query` mbox++blockStorePeerConnect :: MonadIO m => Peer -> BlockStore -> m ()+blockStorePeerConnect peer store =+ BlockPeerConnect peer `send` store++blockStorePeerDisconnect+ :: MonadIO m => Peer -> BlockStore -> m ()+blockStorePeerDisconnect peer store =+ BlockPeerDisconnect peer `send` store++blockStoreHead+ :: MonadIO m => BlockNode -> BlockStore -> m ()+blockStoreHead node store =+ BlockNewBest node `send` store++blockStoreBlock+ :: MonadIO m => Peer -> Block -> BlockStore -> m ()+blockStoreBlock peer block store =+ BlockReceived peer block `send` store++blockStoreNotFound+ :: MonadIO m => Peer -> [BlockHash] -> BlockStore -> m ()+blockStoreNotFound peer blocks store =+ BlockNotFound peer blocks `send` store++blockStoreTx+ :: MonadIO m => Peer -> Tx -> BlockStore -> m ()+blockStoreTx peer tx store =+ TxRefReceived peer tx `send` store++blockStoreTxHash+ :: MonadIO m => Peer -> [TxHash] -> BlockStore -> m ()+blockStoreTxHash peer txhashes store =+ TxRefAvailable peer txhashes `send` store++blockStorePeerConnectSTM+ :: Peer -> BlockStore -> STM ()+blockStorePeerConnectSTM peer store =+ BlockPeerConnect peer `sendSTM` store++blockStorePeerDisconnectSTM+ :: Peer -> BlockStore -> STM ()+blockStorePeerDisconnectSTM peer store =+ BlockPeerDisconnect peer `sendSTM` store++blockStoreHeadSTM+ :: BlockNode -> BlockStore -> STM ()+blockStoreHeadSTM node store =+ BlockNewBest node `sendSTM` store++blockStoreBlockSTM+ :: Peer -> Block -> BlockStore -> STM ()+blockStoreBlockSTM peer block store =+ BlockReceived peer block `sendSTM` store++blockStoreNotFoundSTM+ :: Peer -> [BlockHash] -> BlockStore -> STM ()+blockStoreNotFoundSTM peer blocks store =+ BlockNotFound peer blocks `sendSTM` store++blockStoreTxSTM+ :: Peer -> Tx -> BlockStore -> STM ()+blockStoreTxSTM peer tx store =+ TxRefReceived peer tx `sendSTM` store++blockStoreTxHashSTM+ :: Peer -> [TxHash] -> BlockStore -> STM ()+blockStoreTxHashSTM peer txhashes store =+ TxRefAvailable peer txhashes `sendSTM` store++blockText :: BlockNode -> Maybe Block -> Text+blockText bn mblock = case mblock of+ Nothing ->+ height <> sep <> time <> sep <> hash+ Just block ->+ height <> sep <> time <> sep <> hash <> sep <> size block+ where+ height = cs $ show (nodeHeight bn)+ systime = posixSecondsToUTCTime+ $ fromIntegral+ $ blockTimestamp+ $ nodeHeader bn+ time =+ cs $ formatTime+ defaultTimeLocale+ (iso8601DateFormat (Just "%H:%M"))+ systime hash = blockHashToHex (headerHash (nodeHeader bn)) sep = " | " size = (<> " bytes") . cs . show . B.length . encode
src/Haskoin/Store/Cache.hs view
@@ -19,6 +19,7 @@ , CacheWriterInbox , cacheNewBlock , cacheNewTx+ , cachePing , cacheWriter , isInCache , evictFromCache@@ -413,8 +414,10 @@ , let Right v = decode v' ] -data CacheWriterMessage = CacheNewBlock+data CacheWriterMessage+ = CacheNewBlock | CacheNewTx !TxHash+ | CachePing type CacheWriterInbox = Inbox CacheWriterMessage type CacheWriter = Mailbox CacheWriterMessage@@ -585,6 +588,7 @@ -> CacheX m () cacheWriterReact CacheNewBlock = newBlockC cacheWriterReact (CacheNewTx th) = newTxC th+cacheWriterReact CachePing = syncMempoolC lenNotNull :: [XPubBal] -> Int lenNotNull bals = length $ filter (not . nullBalance . xPubBal) bals@@ -746,7 +750,7 @@ " transactions from block " <> blockHashToHex bh importMultiTxC tds- $(logInfoS) "Cache" $+ $(logDebugS) "Cache" $ "Done importing " <> cs (show (length tds)) <> " transactions from block " <> blockHashToHex bh@@ -992,14 +996,14 @@ (HashSet.difference cachepool nodepool) unless (null txs) $ do $(logDebugS) "Cache" $- "Importing " <> cs (show (length txs)) <> " mempool transactions"+ "Importing " <> cs (show (length txs))+ <> " mempool transactions" importMultiTxC (rights txs) forM_ (zip [(1 :: Int) ..] (lefts txs)) $ \(i, h) -> $(logDebugS) "Cache" $- "Ignoring cache mempool tx " <> cs (show i) <> "/" <>- cs (show (length txs)) <>- ": " <>- txHashToHex h+ "Ignoring cache mempool tx " <> cs (show i) <> "/"+ <> cs (show (length txs)) <> ": "+ <> txHashToHex h where getit th = lift (getTxData th) >>= \case@@ -1018,10 +1022,10 @@ cachePrime = cacheGetHead >>= \case Nothing -> do- $(logInfoS) "Cache" "Cache has no best block set"+ $(logDebugS) "Cache" "Cache has no best block set" lift getBestBlock >>= \case Nothing -> do- $(logInfoS) "Cache" "Best block not set yet"+ $(logDebugS) "Cache" "Best block not set yet" return Nothing Just newhead -> do ch <- asks cacheChain@@ -1030,7 +1034,7 @@ then do mem <- lift getMempool withLockWait lockKey $ do- $(logDebugS) "Cache" "Priming cache…"+ $(logInfoS) "Cache" "Priming cache…" runRedis $ do a <- redisAddToMempool mem b <- redisSetHead newhead@@ -1239,3 +1243,6 @@ cacheNewTx :: MonadIO m => TxHash -> CacheWriter -> m () cacheNewTx th = send (CacheNewTx th)++cachePing :: MonadIO m => CacheWriter -> m ()+cachePing = send CachePing
src/Haskoin/Store/Common.hs view
@@ -59,7 +59,6 @@ XPubBal (..), XPubSpec (..), XPubSummary (..), XPubUnspent (..), nullBalance, toTransaction)-import Network.Socket (SockAddr) type DeriveAddr = XPubKey -> KeyIndex -> Address @@ -268,15 +267,14 @@ -- | Events that the store can generate.-data StoreEvent = StoreBestBlock !BlockHash+data StoreEvent+ = StoreBestBlock !BlockHash | StoreMempoolNew !TxHash- | StorePeerConnected !Peer !SockAddr- | StorePeerDisconnected !Peer !SockAddr+ | StorePeerConnected !Peer+ | StorePeerDisconnected !Peer | StorePeerPong !Peer !Word64 | StoreTxAvailable !Peer ![TxHash] | StoreTxReject !Peer !TxHash !RejectCode !ByteString- | StoreTxDeleted !TxHash- | StoreBlockReverted !BlockHash -- ^ block no longer head of main chain data PubExcept = PubNoPeers
src/Haskoin/Store/Database/Writer.hs view
@@ -5,14 +5,15 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} module Haskoin.Store.Database.Writer- ( Writer+ ( WriterT+ , MemoryTx , runWriter+ , runTx ) where import Control.Applicative ((<|>)) import Control.DeepSeq (NFData)-import Control.Monad.Except (MonadError)-import Control.Monad.Reader (ReaderT)+import Control.Monad.Reader (ReaderT (..)) import qualified Control.Monad.Reader as R import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) import qualified Data.ByteString.Short as B.Short@@ -44,8 +45,8 @@ UnspentVal (..), balanceToVal, unspentToVal, valToBalance, valToUnspent)-import UnliftIO (MonadIO, TVar, atomically,- modifyTVar, newTVarIO,+import UnliftIO (MonadIO, STM, TVar, atomically,+ modifyTVar, newTVarIO, readTVar, readTVarIO) data Dirty a = Modified a | Deleted@@ -58,11 +59,14 @@ data Writer = Writer { getReader :: !DatabaseReader , getState :: !(TVar Memory) } -instance MonadIO m => StoreRead (ReaderT Writer m) where+type MemoryTx = ReaderT (TVar Memory) STM+type WriterT = ReaderT Writer++instance MonadIO m => StoreRead (WriterT m) where getInitialGap =- R.asks (databaseInitialGap . getReader)+ R.asks $ databaseInitialGap . getReader getNetwork =- R.asks (databaseNetwork . getReader)+ R.asks $ databaseNetwork . getReader getBestBlock = R.ask >>= getBestBlockI getBlocksAtHeight h =@@ -86,37 +90,7 @@ getAddressesUnspents = undefined getMaxGap =- R.asks (databaseMaxGap . getReader)--instance MonadIO m => StoreWrite (ReaderT Writer m) where- setBest h =- R.ask >>= setBestI h- insertBlock b =- R.ask >>= insertBlockI b- setBlocksAtHeight hs g =- R.ask >>= setBlocksAtHeightI hs g- insertTx t =- R.ask >>= insertTxI t- insertSpender p s =- R.ask >>= insertSpenderI p s- deleteSpender p =- R.ask >>= deleteSpenderI p- insertAddrTx a t =- R.ask >>= insertAddrTxI a t- deleteAddrTx a t =- R.ask >>= deleteAddrTxI a t- insertAddrUnspent a u =- R.ask >>= insertAddrUnspentI a u- deleteAddrUnspent a u =- R.ask >>= deleteAddrUnspentI a u- setMempool xs =- R.ask >>= setMempoolI xs- insertUnspent u =- R.ask >>= insertUnspentI u- deleteUnspent p =- R.ask >>= deleteUnspentI p- setBalance b =- R.ask >>= setBalanceI b+ R.asks $ databaseMaxGap . getReader data Memory = Memory { hBest@@ -146,54 +120,99 @@ :: !(Maybe [TxRef]) } deriving (Eq, Show) -instance MonadIO m => StoreWrite (ReaderT (TVar Memory) m) where- setBest h = do- v <- R.ask- atomically $ modifyTVar v (setBestH h)- insertBlock b = do- v <- R.ask- atomically $ modifyTVar v (insertBlockH b)- setBlocksAtHeight h g = do- v <- R.ask- atomically $ modifyTVar v (setBlocksAtHeightH h g)- insertTx t = do- v <- R.ask- atomically $ modifyTVar v (insertTxH t)- insertSpender p s = do- v <- R.ask- atomically $ modifyTVar v (insertSpenderH p s)- deleteSpender p = do- v <- R.ask- atomically $ modifyTVar v (deleteSpenderH p)- insertAddrTx a t = do- v <- R.ask- atomically $ modifyTVar v (insertAddrTxH a t)- deleteAddrTx a t = do- v <- R.ask- atomically $ modifyTVar v (deleteAddrTxH a t)- insertAddrUnspent a u = do- v <- R.ask- atomically $ modifyTVar v (insertAddrUnspentH a u)- deleteAddrUnspent a u = do- v <- R.ask- atomically $ modifyTVar v (deleteAddrUnspentH a u)- setMempool xs = do- v <- R.ask- atomically $ modifyTVar v (setMempoolH xs)- setBalance b = do- v <- R.ask- atomically $ modifyTVar v (setBalanceH b)- insertUnspent h = do- v <- R.ask- atomically $ modifyTVar v (insertUnspentH h)- deleteUnspent p = do- v <- R.ask- atomically $ modifyTVar v (deleteUnspentH p)+instance StoreWrite MemoryTx where+ setBest h =+ ReaderT $ \v -> modifyTVar v $+ setBestH h+ insertBlock b =+ ReaderT $ \v -> modifyTVar v $+ insertBlockH b+ setBlocksAtHeight h g =+ ReaderT $ \v -> modifyTVar v $+ setBlocksAtHeightH h g+ insertTx t =+ ReaderT $ \v -> modifyTVar v $+ insertTxH t+ insertSpender p s =+ ReaderT $ \v -> modifyTVar v $+ insertSpenderH p s+ deleteSpender p =+ ReaderT $ \v -> modifyTVar v $+ deleteSpenderH p+ insertAddrTx a t =+ ReaderT $ \v -> modifyTVar v $+ insertAddrTxH a t+ deleteAddrTx a t =+ ReaderT $ \v -> modifyTVar v $+ deleteAddrTxH a t+ insertAddrUnspent a u =+ ReaderT $ \v -> modifyTVar v $+ insertAddrUnspentH a u+ deleteAddrUnspent a u =+ ReaderT $ \v -> modifyTVar v $+ deleteAddrUnspentH a u+ setMempool xs =+ ReaderT $ \v -> modifyTVar v $+ setMempoolH xs+ setBalance b =+ ReaderT $ \v -> modifyTVar v $+ setBalanceH b+ insertUnspent h =+ ReaderT $ \v -> modifyTVar v $+ insertUnspentH h+ deleteUnspent p =+ ReaderT $ \v -> modifyTVar v $+ deleteUnspentH p +instance StoreRead MemoryTx where+ getInitialGap =+ error "Cannot query initial gap in STM"+ getNetwork =+ error "Cannot query network in STM"+ getBestBlock =+ error "Cannot query best block in STM"+ getBlocksAtHeight _ =+ error "Cannot query block heights in STM"+ getBlock _ =+ error "Cannot query block in STM"+ getTxData t =+ ReaderT $ fmap (getTxDataH t) . readTVar+ getSpender op =+ ReaderT $ \v -> do+ m <- getSpenderH op <$> readTVar v+ case m of+ Just (Modified s) -> return (Just s)+ Just Deleted -> return Nothing+ Nothing -> return Nothing+ getSpenders _ =+ error "Cannot query spenders in STM"+ getUnspent op =+ ReaderT $ \v -> do+ m <- getUnspentH op <$> readTVar v+ case m of+ Just (Modified u) -> return (Just (valToUnspent op u))+ Just Deleted -> return Nothing+ Nothing -> return Nothing+ getBalance a =+ ReaderT $ \v -> do+ m <- getBalanceH a <$> readTVar v+ case m of+ Just b -> return $ valToBalance a b+ Nothing -> error "Balance not pre-loaded"+ getMempool =+ ReaderT $ \v -> do+ m <- getMempoolH <$> readTVar v+ case m of+ Just mp -> return mp+ Nothing -> error "Mempool not pre-loaded"+ getAddressesTxs = undefined+ getAddressesUnspents = undefined+ getMaxGap = error "Cannot query max gap in STM"+ runWriter- :: (MonadIO m, MonadError e m)+ :: MonadIO m => DatabaseReader- -> ReaderT Writer m a+ -> WriterT m a -> m a runWriter bdb@DatabaseReader{databaseHandle = db} f = do hm <- newTVarIO emptyMemory@@ -293,77 +312,6 @@ g h i Deleted = deleteOp (UnspentKey (OutPoint h (fromIntegral i))) -setBestI :: MonadIO m => BlockHash -> Writer -> m ()-setBestI bh Writer {getState = hm} =- withMemory hm $ setBest bh--insertBlockI :: MonadIO m => BlockData -> Writer -> m ()-insertBlockI b Writer {getState = hm} =- withMemory hm $ insertBlock b--setBlocksAtHeightI :: MonadIO m- => [BlockHash]- -> BlockHeight- -> Writer- -> m ()-setBlocksAtHeightI hs g Writer {getState = hm} =- withMemory hm $ setBlocksAtHeight hs g--insertTxI :: MonadIO m => TxData -> Writer -> m ()-insertTxI t Writer {getState = hm} =- withMemory hm $ insertTx t--insertSpenderI :: MonadIO m- => OutPoint- -> Spender- -> Writer- -> m ()-insertSpenderI p s Writer {getState = hm} =- withMemory hm $ insertSpender p s--deleteSpenderI :: MonadIO m- => OutPoint- -> Writer- -> m ()-deleteSpenderI p Writer {getState = hm} =- withMemory hm $ deleteSpender p--insertAddrTxI :: MonadIO m- => Address- -> TxRef- -> Writer- -> m ()-insertAddrTxI a t Writer {getState = hm} =- withMemory hm $ insertAddrTx a t--deleteAddrTxI :: MonadIO m- => Address- -> TxRef- -> Writer- -> m ()-deleteAddrTxI a t Writer {getState = hm} =- withMemory hm $ deleteAddrTx a t--insertAddrUnspentI :: MonadIO m- => Address- -> Unspent- -> Writer- -> m ()-insertAddrUnspentI a u Writer {getState = hm} =- withMemory hm $ insertAddrUnspent a u--deleteAddrUnspentI :: MonadIO m- => Address- -> Unspent- -> Writer- -> m ()-deleteAddrUnspentI a u Writer {getState = hm} =- withMemory hm $ deleteAddrUnspent a u--setMempoolI :: MonadIO m => [TxRef] -> Writer -> m ()-setMempoolI xs Writer {getState = hm} =- withMemory hm $ setMempool xs- getBestBlockI :: MonadIO m => Writer -> m (Maybe BlockHash) getBestBlockI Writer {getState = hm, getReader = db} = runMaybeT $ MaybeT f <|> MaybeT g@@ -422,10 +370,6 @@ Just b -> return $ valToBalance a b Nothing -> withDatabaseReader db $ getBalance a -setBalanceI :: MonadIO m => Balance -> Writer -> m ()-setBalanceI b Writer {getState = hm} =- withMemory hm $ setBalance b- getUnspentI :: MonadIO m => OutPoint -> Writer@@ -436,25 +380,14 @@ Just (Modified u) -> return (Just (valToUnspent op u)) Nothing -> withDatabaseReader db (getUnspent op) -insertUnspentI :: MonadIO m => Unspent -> Writer -> m ()-insertUnspentI u Writer {getState = hm} =- withMemory hm $ insertUnspent u--deleteUnspentI :: MonadIO m => OutPoint -> Writer -> m ()-deleteUnspentI p Writer {getState = hm} =- withMemory hm $ deleteUnspent p- getMempoolI :: MonadIO m => Writer -> m [TxRef] getMempoolI Writer {getState = hm, getReader = db} = getMempoolH <$> readTVarIO hm >>= \case Just xs -> return xs Nothing -> withDatabaseReader db getMempool -withMemory :: MonadIO m- => TVar Memory- -> ReaderT (TVar Memory) m a- -> m a-withMemory = flip R.runReaderT+runTx :: MonadIO m => MemoryTx a -> WriterT m a+runTx f = ReaderT $ atomically . runReaderT f . getState emptyMemory :: Memory emptyMemory = Memory { hBest = Nothing
src/Haskoin/Store/Logic.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE LambdaCase #-}@@ -6,47 +7,63 @@ {-# LANGUAGE TupleSections #-} module Haskoin.Store.Logic ( ImportException (..)+ , MonadImport , initBest , getOldMempool , revertBlock , importBlock , newMempoolTx- , deleteTx+ , deleteUnconfirmedTx ) where -import Control.Monad (forM, forM_, guard, unless, void,- when, zipWithM_)-import Control.Monad.Except (MonadError (..))-import Control.Monad.Logger (MonadLogger, logDebugS, logErrorS,- logWarnS)-import qualified Data.ByteString as B-import qualified Data.ByteString.Short as B.Short-import Data.Either (rights)-import qualified Data.IntMap.Strict as I-import Data.List (nub, sortOn)-import Data.Maybe (fromMaybe, isNothing)-import Data.Ord (Down (Down))-import Data.Serialize (encode)-import Data.String.Conversions (cs)-import Data.Text (Text)-import Data.Word (Word32, Word64)-import Haskoin (Address, Block (..), BlockHash,- BlockHeader (..), BlockNode (..),- Network (..), OutPoint (..), Tx (..),- TxHash, TxIn (..), TxOut (..),- blockHashToHex, computeSubsidy,- eitherToMaybe, genesisBlock,- genesisNode, headerHash, isGenesis,- nullOutPoint, scriptToAddressBS,- txHash, txHashToHex)-import Haskoin.Store.Common (StoreRead (..), StoreWrite (..),- getActiveTxData, nub', sortTxs)-import Haskoin.Store.Data (Balance (..), BlockData (..),- BlockRef (..), Prev (..),- Spender (..), TxData (..), TxRef (..),- UnixTime, Unspent (..), confirmed)-import UnliftIO (Exception)+import Control.Monad (forM, forM_, guard, void, when,+ zipWithM_, (<=<), (>=>))+import Control.Monad.Except (ExceptT (..), MonadError,+ runExceptT, throwError)+import Control.Monad.Logger (LoggingT (..),+ MonadLoggerIO (..), logDebugS,+ logErrorS)+import Control.Monad.Reader (ReaderT (ReaderT), runReaderT)+import Control.Monad.Trans (lift)+import qualified Data.ByteString as B+import qualified Data.ByteString.Short as B.Short+import Data.Either (rights)+import qualified Data.IntMap.Strict as I+import Data.List (nub, sortOn)+import Data.Maybe (catMaybes, fromMaybe, isJust,+ isNothing, mapMaybe)+import Data.Ord (Down (Down))+import Data.Serialize (encode)+import Data.Word (Word32, Word64)+import Haskoin (Address, Block (..), BlockHash,+ BlockHeader (..),+ BlockNode (..), Network (..),+ OutPoint (..), Tx (..), TxHash,+ TxIn (..), TxOut (..),+ blockHashToHex, computeSubsidy,+ eitherToMaybe, genesisBlock,+ genesisNode, headerHash,+ isGenesis, nullOutPoint,+ scriptToAddressBS, txHash,+ txHashToHex)+import Haskoin.Store.Common (StoreRead (..), StoreWrite (..),+ getActiveTxData, nub', sortTxs)+import Haskoin.Store.Data (Balance (..), BlockData (..),+ BlockRef (..), Prev (..),+ Spender (..), TxData (..),+ TxRef (..), UnixTime,+ Unspent (..), confirmed)+import Haskoin.Store.Database.Writer (MemoryTx, WriterT, runTx)+import UnliftIO (Exception, MonadIO, async,+ liftIO, wait) +type MonadImport m =+ ( MonadError ImportException m+ , MonadLoggerIO m+ )++type MonadMemory = ExceptT ImportException MemoryTx+ data ImportException = PrevBlockNotBest | Orphan@@ -58,6 +75,8 @@ | TxConfirmed | InsufficientFunds | DuplicatePrevOutput+ | TxSpent+ | OrphanLoop deriving (Eq, Ord, Exception) instance Show ImportException where@@ -71,52 +90,54 @@ show TxConfirmed = "Transaction confirmed" show InsufficientFunds = "Insufficient funds" show DuplicatePrevOutput = "Duplicate previous output"+ show TxSpent = "Transaction is spent"+ show OrphanLoop = "Orphan loop" -initBest ::- ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )- => m ()+runMonadMemory :: MonadImport m => MonadMemory a -> WriterT m a+runMonadMemory f =+ runTx (runExceptT f) >>= \case+ Left e -> throwError e+ Right x -> return x++initBest :: MonadImport m => WriterT m () initBest = do+ $(logDebugS) "BlockStore" "Initializing best block" net <- getNetwork m <- getBestBlock- when (isNothing m) . void $+ when (isNothing m) . void $ do+ $(logDebugS) "BlockStore" "Importing Genesis block" importBlock (genesisBlock net) (genesisNode net) getOldMempool :: StoreRead m => UnixTime -> m [TxHash] getOldMempool now =- map txRefHash- . filter ((< now - 3600 * 72) . memRefTime . txRefBlock)- <$> getMempool+ map txRefHash . filter f <$> getMempool+ where+ f = (< now - 3600 * 72) . memRefTime . txRefBlock -newMempoolTx ::- ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )- => Tx- -> UnixTime- -> m (Maybe [TxHash])- -- ^ deleted transactions or nothing if already imported+newMempoolTx :: MonadImport m => Tx -> UnixTime -> WriterT m Bool newMempoolTx tx w = getActiveTxData (txHash tx) >>= \case- Just _ -> do- $(logDebugS) "BlockStore" $- "Transaction already in store: "- <> txHashToHex (txHash tx)- return Nothing- Nothing -> Just <$> importTx (MemRef w) w tx+ Just _ -> return False+ Nothing -> do+ freeOutputs True True [tx]+ preLoadMemory [tx]+ rbf <- isRBF (MemRef w) tx+ checkNewTx tx+ runMonadMemory $ importTx (MemRef w) w rbf tx+ return True -bestBlockData- :: ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )- => m BlockData+preLoadMemory :: MonadLoggerIO m => [Tx] -> WriterT m ()+preLoadMemory txs = do+ $(logDebugS) "BlockStore" "Pre-loading memory"+ ReaderT $ liftIO . runReaderT go+ where+ go = do+ prev_asyncs <- mapM (async . loadPrevOutputs) txs+ out_asyncs <- mapM (async . loadOutputBalances) txs+ mapM_ wait $ prev_asyncs <> out_asyncs+ runTx . setMempool =<< getMempool++bestBlockData :: MonadImport m => WriterT m BlockData bestBlockData = do h <- getBestBlock >>= \case Nothing -> do@@ -129,14 +150,7 @@ throwError BestBlockNotFound Just b -> return b -revertBlock- :: ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )- => BlockHash- -> m [TxHash]+revertBlock :: MonadImport m => BlockHash -> WriterT m () revertBlock bh = do bd <- bestBlockData >>= \b -> if headerHash (blockDataHeader b) == bh@@ -146,20 +160,14 @@ "Cannot revert non-head block: " <> blockHashToHex bh throwError BlockNotBest tds <- mapM getImportTxData (blockDataTxs bd)- setBest (prevBlock (blockDataHeader bd))- insertBlock bd {blockDataMainChain = False}- forM_ (tail tds) unConfirmTx- deleteTx False False (txHash (txData (head tds)))+ preLoadMemory $ map txData tds+ runTx $ do+ setBest (prevBlock (blockDataHeader bd))+ insertBlock bd {blockDataMainChain = False}+ forM_ (tail tds) unConfirmTx+ deleteConfirmedTx (txHash (txData (head tds))) -checkNewBlock- :: ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )- => Block- -> BlockNode- -> m ()+checkNewBlock :: MonadImport m => Block -> BlockNode -> WriterT m () checkNewBlock b n = getBestBlock >>= \case Nothing@@ -177,67 +185,72 @@ <> blockHashToHex (headerHash (blockHeader b)) throwError PrevBlockNotBest -importOrConfirm- :: ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )- => BlockNode- -> [Tx]- -> m [TxHash] -- ^ deleted transactions-importOrConfirm bn txs =- fmap concat . forM (sortTxs txs) $ \(i, tx) ->- getActiveTxData (txHash tx) >>= \case- Just td- | confirmed (txDataBlock td) -> do- $(logWarnS) "BlockStore" $- "Transaction already confirmed: "- <> txHashToHex (txHash tx)- return []- | otherwise -> do- confirmTx td (br i)- return []- Nothing ->- importTx- (br i)- (fromIntegral (blockTimestamp (nodeHeader bn)))- tx+importOrConfirm :: MonadImport m => BlockNode -> [Tx] -> WriterT m ()+importOrConfirm bn txs = do+ freeOutputs True False txs+ preLoadMemory txs+ mapM_ (uncurry action) (sortTxs txs) where br i = BlockRef {blockRefHeight = nodeHeight bn, blockRefPos = i}+ bn_time = fromIntegral . blockTimestamp $ nodeHeader bn+ action i tx =+ testPresent tx >>= \case+ False -> import_it i tx+ True -> confirm_it i tx+ confirm_it i tx =+ getActiveTxData (txHash tx) >>= \case+ Just t -> do+ $(logDebugS) "BlockStore" $+ "Confirming tx: "+ <> txHashToHex (txHash tx)+ runTx $ confTx t (Just (br i))+ return Nothing+ Nothing -> do+ $(logErrorS) "BlockStore" $+ "Cannot find tx to confirm: "+ <> txHashToHex (txHash tx)+ throwError TxNotFound+ import_it i tx = do+ $(logDebugS) "BlockStore" $+ "Importing tx: " <> txHashToHex (txHash tx)+ runMonadMemory $ importTx (br i) bn_time False tx+ return Nothing -importBlock- :: ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )- => Block- -> BlockNode- -> m [TxHash] -- ^ deleted transactions+importBlock :: MonadImport m => Block -> BlockNode -> WriterT m () importBlock b n = do+ $(logDebugS) "BlockStore" $+ "Checking new block: "+ <> blockHashToHex (headerHash (nodeHeader n)) checkNewBlock b n+ $(logDebugS) "BlockStore" "Passed check" net <- getNetwork let subsidy = computeSubsidy net (nodeHeight n)- insertBlock- BlockData- { blockDataHeight = nodeHeight n- , blockDataMainChain = True- , blockDataWork = nodeWork n- , blockDataHeader = nodeHeader n- , blockDataSize = fromIntegral (B.length (encode b))- , blockDataTxs = map txHash (blockTxns b)- , blockDataWeight = if getSegWit net then w else 0- , blockDataSubsidy = subsidy- , blockDataFees = cb_out_val - subsidy- , blockDataOutputs = ts_out_val- } bs <- getBlocksAtHeight (nodeHeight n)- setBlocksAtHeight- (nub (headerHash (nodeHeader n) : bs))- (nodeHeight n)- setBest (headerHash (nodeHeader n))+ $(logDebugS) "BlockStore" $+ "Inserting block entries for: "+ <> blockHashToHex (headerHash (nodeHeader n))+ runTx $ do+ insertBlock+ BlockData+ { blockDataHeight = nodeHeight n+ , blockDataMainChain = True+ , blockDataWork = nodeWork n+ , blockDataHeader = nodeHeader n+ , blockDataSize = fromIntegral (B.length (encode b))+ , blockDataTxs = map txHash (blockTxns b)+ , blockDataWeight = if getSegWit net then w else 0+ , blockDataSubsidy = subsidy+ , blockDataFees = cb_out_val - subsidy+ , blockDataOutputs = ts_out_val+ }+ setBlocksAtHeight+ (nub (headerHash (nodeHeader n) : bs))+ (nodeHeight n)+ setBest (headerHash (nodeHeader n)) importOrConfirm n (blockTxns b)+ $(logDebugS) "BlockStore" $+ "Finished importing transactions for: "+ <> blockHashToHex (headerHash (nodeHeader n)) where cb_out_val = sum $ map outValue $ txOut $ head $ blockTxns b@@ -250,80 +263,76 @@ s = B.length (encode b') in fromIntegral $ s * 3 + x -checkNewTx- :: ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )- => Tx- -> m ()+checkNewTx :: MonadImport m => Tx -> WriterT m () checkNewTx tx = do+ us <- runTx $ getUnspentOutputs tx when (unique_inputs < length (txIn tx)) $ do- $(logDebugS) "BlockStore" $+ $(logErrorS) "BlockStore" $ "Transaction spends same output twice: " <> txHashToHex (txHash tx) throwError DuplicatePrevOutput when (isCoinbase tx) $ do- $(logDebugS) "BlockStore" $+ $(logErrorS) "BlockStore" $ "Coinbase cannot be imported into mempool: " <> txHashToHex (txHash tx) throwError UnexpectedCoinbase- where- unique_inputs = length (nub' (map prevOutput (txIn tx)))+ when (orphanTest us tx) $ do+ $(logErrorS) "BlockStore" $+ "Orphan: " <> txHashToHex (txHash tx)+ throwError Orphan+ when (outputs > unspents us) $ do+ $(logErrorS) "BlockStore" $+ "Insufficient funds for tx: " <> txHashToHex (txHash tx)+ throwError InsufficientFunds+ where+ unspents = sum . map unspentAmount+ outputs = sum (map outValue (txOut tx))+ unique_inputs = length (nub' (map prevOutput (txIn tx))) -getUnspentOutputs- :: ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )- => Bool -- ^ only delete from mempool- -> [OutPoint]- -> m ([Unspent], [TxHash]) -- ^ unspents and transactions deleted-getUnspentOutputs mem ops = do- uns_ths <- forM ops go- let uns = map fst uns_ths- ths = concatMap snd uns_ths- return (uns, ths)+orphanTest :: [Unspent] -> Tx -> Bool+orphanTest us tx = length (prevOuts tx) > length us++getUnspentOutputs :: StoreRead m => Tx -> m [Unspent]+getUnspentOutputs tx = catMaybes <$> mapM getUnspent (prevOuts tx)++loadPrevOutputs :: MonadIO m => Tx -> WriterT m ()+loadPrevOutputs tx =+ mapM_ go ops where- go op = getUnspent op >>= \case- Nothing -> force_unspent op- Just u -> return (u, [])- force_unspent op = do- s <- getSpender op >>= \case- Nothing -> do- $(logDebugS) "BlockStore" $- "Output not found: " <> showOutput op- throwError Orphan- Just Spender {spenderHash = s} -> return s- $(logDebugS) "BlockStore" $- "Deleting to free output: " <> txHashToHex s- ths <- deleteTx True mem s+ ops = prevOuts tx+ go op = getUnspent op >>= \case+ Just u -> insert_unspent u Nothing -> do- $(logErrorS) "BlockStore" $- "Unexpected absent output: " <> showOutput op- error $ "Unexpected absent output: " <> show op- Just u -> return (u, ths)+ insert_tx (outPointHash op)+ insert_spender op+ insert_tx h = do+ mt <- getActiveTxData h+ forM_ mt $ \t -> do+ let addrs = get_addrs (txData t)+ bals <- mapM getBalance addrs+ runTx $ do+ insertTx t+ mapM_ setBalance bals+ get_addrs tx' =+ let f i o =+ if OutPoint (txHash tx') i `elem` ops+ then eitherToMaybe . scriptToAddressBS $ scriptOutput o+ else Nothing+ in catMaybes $ zipWith f [0..] (txOut tx')+ insert_spender op =+ getSpender op >>= runTx . mapM_ (insertSpender op)+ insert_unspent u = do+ mbal <- mapM getBalance (unspentAddress u)+ runTx $ do+ insertUnspent u+ mapM_ setBalance mbal -checkFunds- :: ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )- => [Unspent]- -> Tx- -> m ()-checkFunds us tx =- when (outputs > unspents) $ do- $(logDebugS) "BlockStore" $- "Insufficient funds for tx: " <> txHashToHex (txHash tx)- throwError InsufficientFunds- where- unspents = sum (map unspentAmount us)- outputs = sum (map outValue (txOut tx))+loadOutputBalances :: MonadIO m => Tx -> WriterT m ()+loadOutputBalances tx = do+ let f = eitherToMaybe . scriptToAddressBS . scriptOutput+ let addrs = mapMaybe f (txOut tx)+ forM_ addrs $ runTx . setBalance <=< getBalance prepareTxData :: Bool -> BlockRef -> Word64 -> [Unspent] -> Tx -> TxData prepareTxData rbf br tt us tx =@@ -339,72 +348,35 @@ ps = I.fromList $ zip [0 ..] $ if isCoinbase tx then [] else map mkprv us importTx- :: ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )- => BlockRef+ :: BlockRef -> Word64 -- ^ unix time+ -> Bool -- ^ RBF -> Tx- -> m [TxHash] -- ^ deleted transactions-importTx br tt tx = do- $(logDebugS) "BlockStore" $- "Importing transaction " <> txHashToHex (txHash tx)- unless (confirmed br) $ checkNewTx tx- (us, ths) <-- if isCoinbase tx- then return ([], [])- else getUnspentOutputs (not (confirmed br)) (map prevOutput (txIn tx))- unless (confirmed br) $ checkFunds us tx- rbf <- isRBF br tx+ -> MonadMemory ()+importTx br tt rbf tx = do+ us <- lift $ getUnspentOutputs tx let td = prepareTxData rbf br tt us tx- commitAddTx us td- return ths+ lift $ commitAddTx us td -unConfirmTx- :: (StoreRead m, StoreWrite m, MonadLogger m)- => TxData- -> m ()+unConfirmTx :: TxData -> MemoryTx () unConfirmTx t = confTx t Nothing -confirmTx- :: (StoreRead m, StoreWrite m, MonadLogger m)- => TxData- -> BlockRef- -> m ()-confirmTx t br = confTx t (Just br)--replaceAddressTx- :: ( StoreRead m- , StoreWrite m- , MonadLogger m- )- => TxData- -> BlockRef- -> m ()-replaceAddressTx t new =- forM_ (txDataAddresses t) $ \a -> do- deleteAddrTx- a- TxRef- { txRefBlock = txDataBlock t- , txRefHash = txHash (txData t)- }- insertAddrTx- a- TxRef- { txRefBlock = new- , txRefHash = txHash (txData t)- }+replaceAddressTx :: TxData -> BlockRef -> MemoryTx ()+replaceAddressTx t new = forM_ (txDataAddresses t) $ \a -> do+ deleteAddrTx+ a+ TxRef+ { txRefBlock = txDataBlock t+ , txRefHash = txHash (txData t)+ }+ insertAddrTx+ a+ TxRef+ { txRefBlock = new+ , txRefHash = txHash (txData t)+ } -adjustAddressOutput- :: (StoreRead m, StoreWrite m, MonadLogger m)- => OutPoint- -> TxOut- -> BlockRef- -> BlockRef- -> m ()+adjustAddressOutput :: OutPoint -> TxOut -> BlockRef -> BlockRef -> MemoryTx () adjustAddressOutput op o old new = do let pk = scriptOutput o s <- getSpender op@@ -444,11 +416,7 @@ decreaseBalance (confirmed old) a (outValue o) increaseBalance (confirmed new) a (outValue o) -confTx- :: (StoreRead m, StoreWrite m, MonadLogger m)- => TxData- -> Maybe BlockRef- -> m ()+confTx :: TxData -> Maybe BlockRef -> MemoryTx () confTx t mbr = do replaceAddressTx t new forM_ (zip [0 ..] (txOut (txData t))) $ \(n, o) -> do@@ -459,82 +427,131 @@ where new = fromMaybe (MemRef (txDataTime t)) mbr old = txDataBlock t- td = t { txDataBlock = new }+ td = t {txDataBlock = new} +freeOutputs+ :: MonadImport m+ => Bool -- ^ only delete transaction if unconfirmed+ -> Bool -- ^ only delete RBF+ -> [Tx]+ -> WriterT m ()+freeOutputs memonly rbfcheck txs = do+ as <- mapM async_get txs+ txs' <- map_asyncs as+ mapM_ (deleteSingleTx . txHash) txs'+ where+ map_asyncs as =+ sequence <$> mapM wait as >>= \case+ Right txs' ->+ return $ reverse+ $ map snd+ $ sortTxs+ $ nub'+ $ concat txs'+ Left e -> throwError e+ async_get tx = do+ g <- askLoggerIO+ ReaderT $ \r ->+ liftIO $ async $+ runLoggingT (runReaderT (get_chain tx) r) g+ get_chain tx =+ fmap (fmap concat . sequence) $+ forM (nub' (prevOuts tx)) $+ getSpender >=> \case+ Nothing ->+ return (Right [])+ Just Spender {spenderHash = s} ->+ map_chain tx <$> getChain memonly rbfcheck s+ map_chain tx = \case+ Right (tx' : _)+ | tx == tx' -> Right []+ Right ts -> Right ts+ Left e -> Left e++deleteConfirmedTx :: MonadImport m => TxHash -> WriterT m ()+deleteConfirmedTx = deleteTx False False++deleteUnconfirmedTx :: MonadImport m => Bool -> TxHash -> WriterT m ()+deleteUnconfirmedTx = deleteTx True+ deleteTx- :: ( StoreRead m- , StoreWrite m- , MonadLogger m- , MonadError ImportException m- )+ :: MonadImport m => Bool -- ^ only delete transaction if unconfirmed -> Bool -- ^ only delete RBF -> TxHash- -> m [TxHash] -- ^ deleted transactions+ -> WriterT m () deleteTx memonly rbfcheck txhash =- getActiveTxData txhash >>= \case- Nothing -> do- $(logDebugS) "BlockStore" $- "Already deleted or not found: "- <> txHashToHex txhash- return []- Just td- | memonly && confirmed (txDataBlock td) -> do- $(logDebugS) "BlockStore" $- "Will not delete confirmed tx: "- <> txHashToHex txhash- throwError TxConfirmed- | rbfcheck ->- isRBF (txDataBlock td) (txData td) >>= \case- True -> go td- False -> do- $(logDebugS) "BlockStore" $- "Will not delete non-RBF tx: "- <> txHashToHex txhash- throwError DoubleSpend- | otherwise -> go td+ getChain memonly rbfcheck txhash >>= \case+ Left e -> throwError e+ Right txs -> mapM_ (deleteSingleTx . txHash) txs++getChain+ :: (MonadLoggerIO m, StoreRead m)+ => Bool -- ^ only delete transaction if unconfirmed+ -> Bool -- ^ only delete RBF+ -> TxHash+ -> m (Either ImportException [Tx])+getChain memonly rbfcheck txhash =+ runExceptT (go txhash) where- go td = do- $(logDebugS) "BlockStore" $- "Deleting tx: " <> txHashToHex txhash- ss <- nub' . map spenderHash . I.elems <$>- getSpenders txhash- ths <- fmap concat $ forM ss $ \s -> do+ go th =+ fmap (reverse . map snd . sortTxs . nub') $+ lift (getActiveTxData th) >>= \case+ Nothing ->+ return []+ Just td+ | memonly && confirmed (txDataBlock td) -> do+ $(logErrorS) "BlockStore" $+ "Transaction already confirmed: "+ <> txHashToHex th+ throwError TxConfirmed+ | rbfcheck ->+ lift (isRBF (txDataBlock td) (txData td)) >>= \case+ True -> get_it td+ False -> do+ $(logErrorS) "BlockStore" $+ "Double-spending transaction: "+ <> txHashToHex th+ throwError DoubleSpend+ | otherwise -> get_it td+ get_it td = do+ let th = txHash (txData td)+ spenders <-+ nub' . map spenderHash . I.elems+ <$> lift (getSpenders th)+ chains <- concat <$> mapM go spenders+ return $ txData td : chains++deleteSingleTx :: MonadImport m => TxHash -> WriterT m ()+deleteSingleTx th =+ getActiveTxData th >>= \case+ Nothing -> $(logDebugS) "BlockStore" $- "Need to delete child tx: " <> txHashToHex s- deleteTx True rbfcheck s- case ths of- [] -> do- commitDelTx td- return [txhash]- _ -> getActiveTxData txhash >>= \case+ "Already deleted: " <> txHashToHex th+ Just td -> do+ $(logDebugS) "BlockStore" $+ "Deleting tx: " <> txHashToHex th+ preload td+ runTx $ commitDelTx td+ where+ preload td = do+ let ths = nub' $ map outPointHash (prevOuts (txData td))+ tds <- forM ths $ \h ->+ getActiveTxData h >>= \case Nothing -> do- $(logErrorS) "BlockStore" $- "Mysteriously gone: " <> txHashToHex txhash- return ths- Just td' -> do- commitDelTx td'- return (txhash : ths)+ $(logDebugS) "BlockStore" $+ "Could not get prev tx: " <> txHashToHex h+ throwError TxNotFound+ Just d -> return d+ preLoadMemory $ txData td : map txData tds -commitDelTx- :: (StoreRead m, StoreWrite m, MonadLogger m)- => TxData- -> m ()+commitDelTx :: TxData -> MemoryTx () commitDelTx = commitModTx False [] -commitAddTx- :: (StoreRead m, StoreWrite m, MonadLogger m)- => [Unspent]- -> TxData- -> m ()+commitAddTx :: [Unspent] -> TxData -> MemoryTx () commitAddTx = commitModTx True -commitModTx- :: (StoreRead m, StoreWrite m, MonadLogger m)- => Bool- -> [Unspent]- -> TxData- -> m ()+commitModTx :: Bool -> [Unspent] -> TxData -> MemoryTx () commitModTx add us td = do let as = txDataAddresses td forM_ as $ \a -> do@@ -554,7 +571,7 @@ mod_outputs | add = addOutputs td | otherwise = delOutputs td -updateMempool :: (StoreRead m, StoreWrite m) => TxData -> m ()+updateMempool :: TxData -> MemoryTx () updateMempool td = do mp <- getMempool setMempool (f mp)@@ -564,69 +581,53 @@ | otherwise = sortOn Down $ TxRef (txDataBlock td) (txHash (txData td)) : mp -spendOutputs :: (StoreRead m, StoreWrite m) => [Unspent] -> TxData -> m ()+spendOutputs :: [Unspent] -> TxData -> MemoryTx () spendOutputs us td = zipWithM_ (spendOutput (txHash (txData td))) [0 ..] us -addOutputs :: (StoreRead m, StoreWrite m, MonadLogger m) => TxData -> m ()+addOutputs :: TxData -> MemoryTx () addOutputs td = zipWithM_ (addOutput (txDataBlock td) . OutPoint (txHash (txData td))) [0 ..] (txOut (txData td)) -isRBF- :: (StoreRead m, MonadLogger m)- => BlockRef- -> Tx- -> m Bool+isRBF :: StoreRead m+ => BlockRef+ -> Tx+ -> m Bool isRBF br tx- | confirmed br = return False+ | confirmed br =+ return False | otherwise =- getNetwork >>= \net ->- if getReplaceByFee net- then go- else return False+ getNetwork >>= \net ->+ if getReplaceByFee net+ then go+ else return False where- go | any ((< 0xffffffff - 1) . txInSequence) (txIn tx) = return True+ go | any ((< 0xffffffff - 1) . txInSequence) (txIn tx) =+ return True | otherwise =- let hs = nub' $ map (outPointHash . prevOutput) (txIn tx)- ck [] = return False- ck (h:hs') =- getActiveTxData h >>= \case- Nothing -> do- $(logErrorS) "BlockStore" $- "Parent transaction not found: " <> txHashToHex h- error $ "Parent transaction not found: " <> show h- Just t- | confirmed (txDataBlock t) -> ck hs'- | txDataRBF t -> return True- | otherwise -> ck hs'+ carry_on+ carry_on =+ let hs = nub' $ map (outPointHash . prevOutput) (txIn tx)+ ck [] = return False+ ck (h : hs') =+ getActiveTxData h >>= \case+ Nothing -> return False+ Just t+ | confirmed (txDataBlock t) -> ck hs'+ | txDataRBF t -> return True+ | otherwise -> ck hs' in ck hs -addOutput- :: (StoreRead m, StoreWrite m, MonadLogger m)- => BlockRef- -> OutPoint- -> TxOut- -> m ()+addOutput :: BlockRef -> OutPoint -> TxOut -> MemoryTx () addOutput = modOutput True -delOutput- :: (StoreRead m, StoreWrite m)- => BlockRef- -> OutPoint- -> TxOut- -> m ()+delOutput :: BlockRef -> OutPoint -> TxOut -> MemoryTx () delOutput = modOutput False -modOutput- :: (StoreRead m, StoreWrite m)- => Bool- -> BlockRef- -> OutPoint- -> TxOut- -> m ()+modOutput :: Bool -> BlockRef -> OutPoint -> TxOut -> MemoryTx () modOutput add br op o = do mod_unspent forM_ ma $ \a -> do@@ -648,7 +649,7 @@ mod_addr_unspent | add = insertAddrUnspent | otherwise = deleteAddrUnspent -delOutputs :: (StoreRead m, StoreWrite m) => TxData -> m ()+delOutputs :: TxData -> MemoryTx () delOutputs td = forM_ (zip [0..] outs) $ \(i, o) -> do let op = OutPoint (txHash (txData td)) i@@ -656,13 +657,7 @@ where outs = txOut (txData td) -getImportTxData- :: ( StoreRead m- , MonadLogger m- , MonadError ImportException m- )- => TxHash- -> m TxData+getImportTxData :: MonadImport m => TxHash -> WriterT m TxData getImportTxData th = getActiveTxData th >>= \case Nothing -> do@@ -670,51 +665,38 @@ throwError TxNotFound Just d -> return d -getTxOut- :: Word32- -> Tx- -> Maybe TxOut+getTxOut :: Word32 -> Tx -> Maybe TxOut getTxOut i tx = do guard (fromIntegral i < length (txOut tx)) return $ txOut tx !! fromIntegral i -spendOutput- :: (StoreRead m, StoreWrite m)- => TxHash- -> Word32- -> Unspent- -> m ()+spendOutput :: TxHash -> Word32 -> Unspent -> MemoryTx () spendOutput th ix u = do insertSpender (unspentPoint u) (Spender th ix) let pk = B.Short.fromShort (unspentScript u) case scriptToAddressBS pk of Left _ -> return () Right a -> do- decreaseBalance (confirmed (unspentBlock u)) a (unspentAmount u)+ decreaseBalance+ (confirmed (unspentBlock u))+ a+ (unspentAmount u) deleteAddrUnspent a u deleteUnspent (unspentPoint u) -unspendOutputs :: (StoreRead m, StoreWrite m, MonadLogger m) => TxData -> m ()+unspendOutputs :: TxData -> MemoryTx () unspendOutputs td = mapM_ unspendOutput (prevOuts (txData td)) -unspendOutput :: (StoreRead m, StoreWrite m, MonadLogger m) => OutPoint -> m ()+unspendOutput :: OutPoint -> MemoryTx () unspendOutput op = do t <- getActiveTxData (outPointHash op) >>= \case- Nothing -> do- $(logErrorS) "BlockStore" $- "Could not find tx data: "- <> txHashToHex (outPointHash op)- error $- "Could not find tx data: "- <> show (outPointHash op)+ Nothing ->+ error $ "Could not find tx data: " <> show (outPointHash op) Just t -> return t o <- case getTxOut (outPointIndex op) (txData t) of- Nothing -> do- $(logErrorS) "BlockStore" $- "Could not find output: " <> showOutput op+ Nothing -> error $ "Could not find output: " <> show op Just o -> return o- deleteSpender op let m = eitherToMaybe (scriptToAddressBS (scriptOutput o)) u = Unspent { unspentAmount = outValue o , unspentBlock = txDataBlock t@@ -722,43 +704,29 @@ , unspentPoint = op , unspentAddress = m }+ deleteSpender op insertUnspent u forM_ m $ \a -> do insertAddrUnspent a u increaseBalance (confirmed (unspentBlock u)) a (outValue o) -modifyReceived- :: (StoreRead m, StoreWrite m)- => Address- -> (Word64 -> Word64)- -> m ()+modifyReceived :: Address -> (Word64 -> Word64) -> MemoryTx () modifyReceived a f = getBalance a >>= \b -> setBalance b {balanceTotalReceived = f (balanceTotalReceived b)} -decreaseBalance- :: (StoreRead m, StoreWrite m)- => Bool- -> Address- -> Word64- -> m ()+decreaseBalance :: Bool -> Address -> Word64 -> MemoryTx () decreaseBalance conf = modBalance conf False -increaseBalance- :: (StoreRead m, StoreWrite m)- => Bool- -> Address- -> Word64- -> m ()+increaseBalance :: Bool -> Address -> Word64 -> MemoryTx () increaseBalance conf = modBalance conf True modBalance- :: (StoreRead m, StoreWrite m)- => Bool -- ^ confirmed+ :: Bool -- ^ confirmed -> Bool -- ^ add -> Address -> Word64- -> m ()+ -> MemoryTx () modBalance conf add a val = getBalance a >>= \b -> setBalance ((g . f) b) where@@ -768,7 +736,7 @@ m | add = (+) | otherwise = subtract -modAddressCount :: (StoreRead m, StoreWrite m) => Bool -> Address -> m ()+modAddressCount :: Bool -> Address -> MemoryTx () modAddressCount add a = getBalance a >>= \b -> setBalance b {balanceTxCount = f (balanceTxCount b)}@@ -792,9 +760,15 @@ isCoinbase :: Tx -> Bool isCoinbase = all ((== nullOutPoint) . prevOutput) . txIn -showOutput :: OutPoint -> Text-showOutput OutPoint {outPointHash = h, outPointIndex = i} =- txHashToHex h <> "/" <> cs (show i)- prevOuts :: Tx -> [OutPoint] prevOuts tx = filter (/= nullOutPoint) (map prevOutput (txIn tx))++testPresent :: StoreRead m => Tx -> m Bool+testPresent tx =+ case prevOuts tx of+ [] -> isJust <$> getActiveTxData (txHash tx)+ op : _ -> getSpender op >>= \case+ Nothing ->+ return False+ Just Spender { spenderHash = s } ->+ return $ s == txHash tx
src/Haskoin/Store/Manager.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE FlexibleContexts #-} module Haskoin.Store.Manager ( StoreConfig(..) , Store(..)@@ -7,6 +8,7 @@ import Control.Monad (forever, unless, when) import Control.Monad.Logger (MonadLoggerIO) import Data.Serialize (decode)+import Data.Time.Clock (NominalDiffTime) import Data.Word (Word32) import Haskoin (BlockHash (..), Inv (..), InvType (..), InvVector (..),@@ -18,10 +20,10 @@ VarString (..), sockToHostAddress) import Haskoin.Node (Chain, ChainEvent (..),- HostPort, NodeConfig (..),- NodeEvent (..), PeerEvent (..),- PeerManager, WithConnection,- node)+ HostPort, Node (..),+ NodeConfig (..), NodeEvent (..),+ PeerEvent (..), PeerManager,+ WithConnection, withNode) import Haskoin.Store.BlockStore (BlockStore, BlockStoreConfig (..), blockStore, blockStoreBlockSTM,@@ -33,20 +35,22 @@ blockStoreTxSTM) import Haskoin.Store.Cache (CacheConfig (..), CacheWriter, cacheNewBlock, cacheNewTx,- cacheWriter, connectRedis)+ cachePing, cacheWriter,+ connectRedis) import Haskoin.Store.Common (StoreEvent (..)) import Haskoin.Store.Database.Reader (DatabaseReader (..), connectRocksDB, withDatabaseReader) import Network.Socket (SockAddr (..))-import NQE (Inbox, Listen, Process (..),- Publisher,- PublisherMessage (Event),- inboxToMailbox, newInbox,- receive, sendSTM, withProcess,- withPublisher, withSubscription)+import NQE (Inbox, Process (..), Publisher,+ newMailbox, publishSTM, receive,+ withProcess, withPublisher,+ withSubscription)+import System.Random (randomRIO) import UnliftIO (MonadIO, MonadUnliftIO, STM,- link, withAsync)+ atomically, liftIO, link,+ withAsync)+import UnliftIO.Concurrent (threadDelay) -- | Store mailboxes. data Store =@@ -85,142 +89,175 @@ -- ^ maximum number of keys in Redis cache , storeConfWipeMempool :: !Bool -- ^ wipe mempool when starting- , storeConfPeerTimeout :: !Int+ , storeConfPeerTimeout :: !NominalDiffTime -- ^ disconnect peer if message not received for this many seconds- , storeConfPeerTooOld :: !Int+ , storeConfPeerMaxLife :: !NominalDiffTime -- ^ disconnect peer if it has been connected this long- , storeConfConnect :: !WithConnection+ , storeConfConnect :: !(SockAddr -> WithConnection) -- ^ connect to peers using the function 'withConnection' } -withStore ::- (MonadLoggerIO m, MonadUnliftIO m)- => StoreConfig- -> (Store -> m a)- -> m a-withStore cfg action = do- chaininbox <- newInbox- let chain = inboxToMailbox chaininbox- maybecacheconn <-- case storeConfCache cfg of- Nothing -> return Nothing- Just redisurl -> Just <$> connectRedis redisurl- db <-- connectRocksDB- (storeConfNetwork cfg)- (storeConfInitialGap cfg)- (storeConfGap cfg)- (storeConfDB cfg)- case maybecacheconn of- Nothing -> launch db Nothing chaininbox- Just cacheconn -> do- let cachecfg =- CacheConfig- { cacheConn = cacheconn- , cacheMin = storeConfCacheMin cfg- , cacheChain = chain- , cacheMax = storeConfMaxKeys cfg- }- withProcess (withDatabaseReader db . cacheWriter cachecfg) $ \p ->- launch db (Just (cachecfg, getProcessMailbox p)) chaininbox+withStore :: (MonadLoggerIO m, MonadUnliftIO m)+ => StoreConfig -> (Store -> m a) -> m a+withStore cfg action =+ connectDB cfg >>= \db ->+ newMailbox >>= \(bi, b) ->+ withPublisher $ \pub ->+ withPublisher $ \node_pub ->+ withSubscription node_pub $ \node_sub ->+ withNode (nodeCfg cfg db node_pub) $ \node ->+ withCache cfg (nodeChain node) db pub $ \mcache ->+ withAsync (nodeForwarder b pub node_sub) $ \a1 ->+ withAsync (blockStore (blockStoreCfg cfg node pub db) bi) $ \a2 ->+ link a1 >> link a2 >>+ action Store { storeManager = nodeManager node+ , storeChain = nodeChain node+ , storeBlock = b+ , storeDB = db+ , storeCache = mcache+ , storePublisher = pub+ , storeNetwork = storeConfNetwork cfg+ }++connectDB :: MonadIO m => StoreConfig -> m DatabaseReader+connectDB cfg =+ connectRocksDB+ (storeConfNetwork cfg)+ (storeConfInitialGap cfg)+ (storeConfGap cfg)+ (storeConfDB cfg)++blockStoreCfg :: StoreConfig+ -> Node+ -> Publisher StoreEvent+ -> DatabaseReader+ -> BlockStoreConfig+blockStoreCfg cfg node pub db =+ BlockStoreConfig+ { blockConfChain = nodeChain node+ , blockConfManager = nodeManager node+ , blockConfListener = pub+ , blockConfDB = db+ , blockConfNet = storeConfNetwork cfg+ , blockConfWipeMempool = storeConfWipeMempool cfg+ , blockConfPeerTimeout = storeConfPeerTimeout cfg+ }++nodeCfg :: StoreConfig+ -> DatabaseReader+ -> Publisher NodeEvent+ -> NodeConfig+nodeCfg cfg db pub =+ NodeConfig+ { nodeConfMaxPeers = storeConfMaxPeers cfg+ , nodeConfDB = databaseHandle db+ , nodeConfPeers = storeConfInitPeers cfg+ , nodeConfDiscover = storeConfDiscover cfg+ , nodeConfEvents = pub+ , nodeConfNetAddr =+ NetworkAddress+ 0+ (sockToHostAddress (SockAddrInet 0 0))+ , nodeConfNet = storeConfNetwork cfg+ , nodeConfTimeout = storeConfPeerTimeout cfg+ , nodeConfPeerMaxLife = storeConfPeerMaxLife cfg+ , nodeConfConnect = storeConfConnect cfg+ }++withCache :: (MonadUnliftIO m, MonadLoggerIO m)+ => StoreConfig+ -> Chain+ -> DatabaseReader+ -> Publisher StoreEvent+ -> (Maybe CacheConfig -> m a)+ -> m a+withCache cfg chain db pub action =+ case storeConfCache cfg of+ Nothing ->+ action Nothing+ Just redisurl ->+ connectRedis redisurl >>= \conn ->+ withSubscription pub $ \evts ->+ withProcess (f conn) $ \p ->+ cacheWriterProcesses evts (getProcessMailbox p) $+ action (Just (c conn)) where- launch db maybecache chaininbox =- withPublisher $ \pub -> do- managerinbox <- newInbox- blockstoreinbox <- newInbox- let blockstore = inboxToMailbox blockstoreinbox- manager = inboxToMailbox managerinbox- chain = inboxToMailbox chaininbox- let nodeconfig =- NodeConfig- { nodeConfMaxPeers = storeConfMaxPeers cfg- , nodeConfDB = databaseHandle db- , nodeConfPeers = storeConfInitPeers cfg- , nodeConfDiscover = storeConfDiscover cfg- , nodeConfEvents =- storeDispatch blockstore ((`sendSTM` pub) . Event)- , nodeConfNetAddr =- NetworkAddress- 0- (sockToHostAddress (SockAddrInet 0 0))- , nodeConfNet = storeConfNetwork cfg- , nodeConfTimeout = storeConfPeerTimeout cfg- , nodeConfPeerOld = storeConfPeerTooOld cfg- , nodeConfConnect = storeConfConnect cfg- }- withAsync (node nodeconfig managerinbox chaininbox) $ \nodeasync -> do- link nodeasync- let blockstoreconfig =- BlockStoreConfig- { blockConfChain = chain- , blockConfManager = manager- , blockConfListener = (`sendSTM` pub) . Event- , blockConfDB = db- , blockConfNet = storeConfNetwork cfg- , blockConfWipeMempool = storeConfWipeMempool cfg- , blockConfPeerTimeout = storeConfPeerTimeout cfg- }- runaction =- action- Store- { storeManager = manager- , storeChain = chain- , storeBlock = blockstore- , storeDB = db- , storeCache = fst <$> maybecache- , storePublisher = pub- , storeNetwork = storeConfNetwork cfg- }- case maybecache of- Nothing ->- launch2 blockstoreconfig blockstoreinbox runaction- Just (_, cache) ->- withSubscription pub $ \evts ->- withAsync (cacheWriterEvents evts cache) $ \evtsasync ->- link evtsasync >>- launch2- blockstoreconfig- blockstoreinbox- runaction- launch2 blockstoreconfig blockstoreinbox runaction =- withAsync (blockStore blockstoreconfig blockstoreinbox) $ \blockstoreasync ->- link blockstoreasync >> runaction+ f conn cwinbox =+ withDatabaseReader db $+ cacheWriter (c conn) cwinbox+ c conn = CacheConfig+ { cacheConn = conn+ , cacheMin = storeConfCacheMin cfg+ , cacheChain = chain+ , cacheMax = storeConfMaxKeys cfg+ } +cacheWriterProcesses :: MonadUnliftIO m+ => Inbox StoreEvent+ -> CacheWriter+ -> m a+ -> m a+cacheWriterProcesses evts cwm action =+ withAsync events $ \a1 ->+ withAsync ping $ \a2 ->+ link a1 >> link a2 >> action+ where+ events = cacheWriterEvents evts cwm+ ping = forever $ do+ time <- liftIO $ randomRIO (5 * second, 15 * second)+ threadDelay time+ cachePing cwm+ second = 1000000+ cacheWriterEvents :: MonadIO m => Inbox StoreEvent -> CacheWriter -> m ()-cacheWriterEvents evts cwm = forever $ receive evts >>= (`cacheWriterDispatch` cwm)+cacheWriterEvents evts cwm =+ forever $+ receive evts >>= \e ->+ e `cacheWriterDispatch` cwm cacheWriterDispatch :: MonadIO m => StoreEvent -> CacheWriter -> m () cacheWriterDispatch (StoreBestBlock _) = cacheNewBlock cacheWriterDispatch (StoreMempoolNew th) = cacheNewTx th-cacheWriterDispatch (StoreTxDeleted th) = cacheNewTx th cacheWriterDispatch _ = const (return ()) +nodeForwarder :: MonadIO m+ => BlockStore+ -> Publisher StoreEvent+ -> Inbox NodeEvent+ -> m ()+nodeForwarder b pub sub =+ forever $ receive sub >>= atomically . storeDispatch b pub+ -- | Dispatcher of node events.-storeDispatch :: BlockStore -> Listen StoreEvent -> NodeEvent -> STM ()+storeDispatch :: BlockStore+ -> Publisher StoreEvent+ -> NodeEvent+ -> STM () -storeDispatch b l (PeerEvent (PeerConnected p a)) = do- l (StorePeerConnected p a)- blockStorePeerConnectSTM p a b+storeDispatch b pub (PeerEvent (PeerConnected p)) = do+ publishSTM (StorePeerConnected p) pub+ blockStorePeerConnectSTM p b -storeDispatch b l (PeerEvent (PeerDisconnected p a)) = do- l (StorePeerDisconnected p a)- blockStorePeerDisconnectSTM p a b+storeDispatch b pub (PeerEvent (PeerDisconnected p)) = do+ publishSTM (StorePeerDisconnected p) pub+ blockStorePeerDisconnectSTM p b storeDispatch b _ (ChainEvent (ChainBestBlock bn)) = blockStoreHeadSTM bn b -storeDispatch _ _ (ChainEvent _) = return ()+storeDispatch _ _ (ChainEvent _) =+ return () -storeDispatch _ l (PeerEvent (PeerMessage p (MPong (Pong n)))) =- l (StorePeerPong p n)+storeDispatch _ pub (PeerMessage p (MPong (Pong n))) =+ publishSTM (StorePeerPong p n) pub -storeDispatch b _ (PeerEvent (PeerMessage p (MBlock block))) =+storeDispatch b _ (PeerMessage p (MBlock block)) = blockStoreBlockSTM p block b -storeDispatch b _ (PeerEvent (PeerMessage p (MTx tx))) =+storeDispatch b _ (PeerMessage p (MTx tx)) = blockStoreTxSTM p tx b -storeDispatch b _ (PeerEvent (PeerMessage p (MNotFound (NotFound is)))) = do+storeDispatch b _ (PeerMessage p (MNotFound (NotFound is))) = do let blocks = [ BlockHash h | InvVector t h <- is@@ -228,17 +265,23 @@ ] unless (null blocks) $ blockStoreNotFoundSTM p blocks b -storeDispatch b l (PeerEvent (PeerMessage p (MInv (Inv is)))) = do+storeDispatch b pub (PeerMessage p (MInv (Inv is))) = do let txs = [TxHash h | InvVector t h <- is, t == InvTx || t == InvWitnessTx]- l (StoreTxAvailable p txs)+ publishSTM (StoreTxAvailable p txs) pub unless (null txs) $ blockStoreTxHashSTM p txs b -storeDispatch _ l (PeerEvent (PeerMessage p (MReject r))) =+storeDispatch _ pub (PeerMessage p (MReject r)) = when (rejectMessage r == MCTx) $ case decode (rejectData r) of Left _ -> return () Right th ->- l $- StoreTxReject p th (rejectCode r) (getVarString (rejectReason r))+ let reject =+ StoreTxReject+ p+ th+ (rejectCode r)+ (getVarString (rejectReason r))+ in publishSTM reject pub -storeDispatch _ _ (PeerEvent _) = return ()+storeDispatch _ _ _ =+ return ()
src/Haskoin/Store/Web.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE LambdaCase #-}@@ -16,7 +17,7 @@ import Conduit () import Control.Applicative ((<|>)) import Control.Monad (forever, when, (<=<))-import Control.Monad.Logger+import Control.Monad.Logger (MonadLoggerIO, logInfoS) import Control.Monad.Reader (ReaderT, asks, local, runReaderT) import Control.Monad.Trans (lift)@@ -43,7 +44,6 @@ import Data.Time.Clock (NominalDiffTime, diffUTCTime, getCurrentTime) import Data.Time.Clock.System (getSystemTime, systemSeconds)-import Data.Version (showVersion) import Data.Word (Word32, Word64) import Database.RocksDB (Property (..), getProperty) import qualified Haskoin.Block as H@@ -51,7 +51,7 @@ import Haskoin.Network import Haskoin.Node (Chain, OnlinePeer (..), PeerManager, chainGetBest,- managerGetPeers, sendMessage)+ getPeers, sendMessage) import Haskoin.Store.Cache (CacheT, evictFromCache, withCache) import Haskoin.Store.Common (Limits (..), PubExcept (..),@@ -75,7 +75,6 @@ setPort) import NQE (Inbox, Publisher, receive, withSubscription)-import qualified Paths_haskoin_store as P (version) import Text.Printf (printf) import UnliftIO (MonadIO, MonadUnliftIO, askRunInIO, liftIO, timeout)@@ -112,6 +111,7 @@ , webMaxLimits :: !WebLimits , webReqLog :: !Bool , webWebTimeouts :: !WebTimeouts+ , webVersion :: !String } data WebTimeouts = WebTimeouts@@ -671,7 +671,7 @@ -- | Publish a new transaction to the network. publishTx ::- (MonadUnliftIO m, StoreRead m)+ (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => Network -> Publisher StoreEvent -> PeerManager@@ -684,7 +684,7 @@ Nothing -> go s where go s =- managerGetPeers mgr >>= \case+ getPeers mgr >>= \case [] -> return $ Left PubNoPeers OnlinePeer {onlinePeerMailbox = p}:_ -> do MTx tx `sendMessage` p@@ -706,7 +706,7 @@ receive s >>= \case StoreTxReject p' h' c _ | p == p' && h' == txHash tx -> return . Left $ PubReject c- StorePeerDisconnected p' _+ StorePeerDisconnected p' | p == p' -> return $ Left PubPeerDisconnected StoreMempoolNew h' | h' == txHash tx -> return $ Right ()@@ -742,11 +742,9 @@ se <- receive sub return $ case se of- StoreBestBlock b -> Just (EventBlock b)- StoreMempoolNew t -> Just (EventTx t)- StoreTxDeleted t -> Just (EventTx t)- StoreBlockReverted b -> Just (EventBlock b)- _ -> Nothing+ StoreBestBlock b -> Just (EventBlock b)+ StoreMempoolNew t -> Just (EventTx t)+ _ -> Nothing -- GET Address Transactions -- @@ -842,11 +840,14 @@ -- GET Network Information -- scottyPeers :: MonadLoggerIO m => GetPeers -> WebT m [PeerInformation]-scottyPeers _ = getPeersInformation =<< lift (asks (storeManager . webStore))+scottyPeers _ = lift $+ getPeersInformation =<< asks (storeManager . webStore) -- | Obtain information about connected peers from peer manager process.-getPeersInformation :: MonadIO m => PeerManager -> m [PeerInformation]-getPeersInformation mgr = mapMaybe toInfo <$> managerGetPeers mgr+getPeersInformation+ :: MonadLoggerIO m => PeerManager -> m [PeerInformation]+getPeersInformation mgr =+ mapMaybe toInfo <$> getPeers mgr where toInfo op = do ver <- onlinePeerVersion op@@ -871,78 +872,84 @@ mgr <- lift $ asks (storeManager . webStore) chn <- lift $ asks (storeChain . webStore) tos <- lift $ asks webWebTimeouts- h <- lift $ healthCheck net mgr chn tos+ ver <- lift $ asks webVersion+ h <- lift $ healthCheck net mgr chn tos ver when (not (healthOK h) || not (healthSynced h)) $ S.status status503 return h healthCheck ::- (MonadUnliftIO m, StoreRead m)+ (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => Network -> PeerManager -> Chain -> WebTimeouts+ -> String -> m HealthCheck-healthCheck net mgr ch tos = do- cb <- chain_best- bb <- block_best- pc <- peer_count- tm <- get_current_time- ml <- get_mempool_last- let ck = block_ok cb- bk = block_ok bb- pk = peer_count_ok pc- bd = block_time_delta tm cb- td = tx_time_delta tm bd ml- lk = timeout_ok (blockTimeout tos) bd- tk = timeout_ok (txTimeout tos) td- sy = in_sync bb cb- ok = ck && bk && pk && lk && (tk || not sy)+healthCheck net mgr ch tos ver = do+ chain_best <- get_chain_best+ block_best <- get_block_store_best+ peer_count <- get_peer_count+ time <- get_current_time+ mempool_last <- get_mempool_last+ let chain_ok = is_block_ok chain_best+ block_ok = is_block_ok block_best+ peer_count_ok = is_peer_count_ok peer_count+ block_time_delta = get_block_time_delta time chain_best+ tx_time_delta = get_tx_time_delta time block_time_delta mempool_last+ block_timeout_ok = is_timeout_ok (blockTimeout tos) block_time_delta+ tx_timeout_ok = is_timeout_ok (txTimeout tos) tx_time_delta+ in_sync = is_in_sync block_best chain_best+ ok = chain_ok &&+ block_ok &&+ peer_count_ok &&+ block_timeout_ok &&+ (tx_timeout_ok || not in_sync) return HealthCheck- { healthBlockBest = block_hash <$> bb- , healthBlockHeight = block_height <$> bb- , healthHeaderBest = node_hash <$> cb- , healthHeaderHeight = node_height <$> cb- , healthPeers = pc+ { healthBlockBest = block_data_hash <$> block_best+ , healthBlockHeight = block_data_height <$> block_best+ , healthHeaderBest = node_hash <$> chain_best+ , healthHeaderHeight = node_height <$> chain_best+ , healthPeers = peer_count , healthNetwork = getNetworkName net , healthOK = ok- , healthSynced = sy- , healthLastBlock = bd- , healthLastTx = td- , healthVersion = showVersion P.version+ , healthSynced = in_sync+ , healthLastBlock = block_time_delta+ , healthLastTx = tx_time_delta+ , healthVersion = ver } where- block_hash = H.headerHash . blockDataHeader- block_height = blockDataHeight+ block_data_hash = H.headerHash . blockDataHeader+ block_data_height = blockDataHeight node_hash = H.headerHash . H.nodeHeader node_height = H.nodeHeight get_mempool_last = listToMaybe <$> getMempool get_current_time = fromIntegral . systemSeconds <$> liftIO getSystemTime- peer_count_ok pc = fromMaybe 0 pc > 0- block_ok = isJust+ is_peer_count_ok pc = fromMaybe 0 pc > 0+ is_block_ok = isJust node_timestamp = fromIntegral . H.blockTimestamp . H.nodeHeader- in_sync bb cb = fromMaybe False $ do+ is_in_sync bb cb = fromMaybe False $ do bh <- blockDataHeight <$> bb nh <- H.nodeHeight <$> cb return $ compute_delta bh nh <= 1- block_time_delta tm cb = do+ get_block_time_delta tm cb = do bt <- node_timestamp <$> cb return $ compute_delta bt tm- tx_time_delta tm bd ml = do+ get_tx_time_delta tm bd ml = do bd' <- bd tt <- memRefTime . txRefBlock <$> ml <|> bd return $ min (compute_delta tt tm) bd'- timeout_ok to td = fromMaybe False $ do+ is_timeout_ok to td = fromMaybe False $ do td' <- td return $ getAllowMinDifficultyBlocks net || to == 0 || td' <= to- peer_count = fmap length <$> timeout 10000000 (managerGetPeers mgr)- block_best = runMaybeT $ do+ get_peer_count = fmap length <$> timeout 10000000 (getPeers mgr)+ get_block_store_best = runMaybeT $ do h <- MaybeT getBestBlock MaybeT $ getBlock h- chain_best = timeout 10000000 $ chainGetBest ch+ get_chain_best = timeout 10000000 $ chainGetBest ch compute_delta a b = if b > a then b - a else 0 scottyDbStats :: MonadLoggerIO m => WebT m ()
test/Haskoin/StoreSpec.hs view
@@ -11,6 +11,7 @@ import qualified Data.ByteString as B import Data.ByteString.Base64 import Data.Either+import Data.List import Data.Maybe import Data.Serialize import Data.Time.Clock.POSIX@@ -32,80 +33,71 @@ } spec :: Spec-spec = do- let net = bchRegTest- describe "Download" $ do- it "gets 8 blocks" $- withTestStore net "eight-blocks" $ \TestStore {..} -> do- bs <-- replicateM 8 . receiveMatch testStoreEvents $ \case- StoreBestBlock b -> Just b- _ -> Nothing- let bestHash = last bs- bestNodeM <- chainGetBlock bestHash testStoreChain- bestNodeM `shouldSatisfy` isJust- let bestNode = fromJust bestNodeM- bestHeight = nodeHeight bestNode- bestHeight `shouldBe` 8- it "get a block and its transactions" $- withTestStore net "get-block-txs" $ \TestStore {..} ->- withDatabaseReader testStoreDB $ do- let h1 =- "5369ef2386c72acdf513ffd80aeba2a1774e2f004d120761e54a8bf614173f3e"- get_the_block h =- receive testStoreEvents >>= \case- StoreBestBlock b- | h <= 1 -> return b- | otherwise ->- get_the_block ((h :: Int) - 1)- _ -> get_the_block h- bh <- get_the_block 15- m <- getBlock bh- let bd = fromMaybe (error "Could not get block") m- t1 <- getTransaction h1- lift $ do- blockDataHeight bd `shouldBe` 15- length (blockDataTxs bd) `shouldBe` 1- head (blockDataTxs bd) `shouldBe` h1- t1 `shouldSatisfy` isJust- txHash (transactionData (fromJust t1)) `shouldBe` h1+spec = describe "Download" $ do+ it "gets 8 blocks" $+ withTestStore bchRegTest "eight-blocks" $ \TestStore {..} -> do+ bs <- replicateM 8 . receiveMatch testStoreEvents $ \case+ StoreBestBlock b -> Just b+ _ -> Nothing+ let bestHash = last bs+ bestNodeM <- chainGetBlock bestHash testStoreChain+ bestNodeM `shouldSatisfy` isJust+ let bestNode = fromJust bestNodeM+ bestHeight = nodeHeight bestNode+ bestHeight `shouldBe` 8+ it "get a block and its transactions" $+ withTestStore bchRegTest "get-block-txs" $ \TestStore {..} ->+ withDatabaseReader testStoreDB $ do+ let h1 = "5369ef2386c72acdf513ffd80aeba2a1774e2f004d120761e54a8bf614173f3e"+ get_the_block h =+ receive testStoreEvents >>= \case+ StoreBestBlock b+ | h <= 1 -> return b+ | otherwise ->+ get_the_block ((h :: Int) - 1)+ _ -> get_the_block h+ bh <- get_the_block 15+ m <- getBlock bh+ let bd = fromMaybe (error "Could not get block") m+ t1 <- getTransaction h1+ lift $ do+ blockDataHeight bd `shouldBe` 15+ length (blockDataTxs bd) `shouldBe` 1+ head (blockDataTxs bd) `shouldBe` h1+ t1 `shouldSatisfy` isJust+ txHash (transactionData (fromJust t1)) `shouldBe` h1 withTestStore :: MonadUnliftIO m => Network -> String -> (TestStore -> m a) -> m a withTestStore net t f = withSystemTempDirectory ("haskoin-store-test-" <> t <> "-") $ \w ->- runNoLoggingT $ do- let ad =- NetworkAddress- nodeNetwork- (sockToHostAddress (SockAddrInet 0 0))- cfg =- StoreConfig- { storeConfMaxPeers = 20- , storeConfInitPeers = []- , storeConfDiscover = True- , storeConfDB = w- , storeConfNetwork = net- , storeConfCache = Nothing- , storeConfGap = gap- , storeConfInitialGap = 20- , storeConfCacheMin = 100- , storeConfMaxKeys = 100 * 1000 * 1000- , storeConfWipeMempool = False- , storeConfPeerTimeout = 60- , storeConfPeerTooOld = 48 * 3600- , storeConfConnect = dummyPeerConnect net ad- }- withStore cfg $ \Store {..} ->- withSubscription storePublisher $ \sub ->- lift $- f- TestStore- { testStoreDB = storeDB- , testStoreBlockStore = storeBlock- , testStoreChain = storeChain- , testStoreEvents = sub- }+ runNoLoggingT $ do+ let ad = NetworkAddress+ nodeNetwork+ (sockToHostAddress (SockAddrInet 0 0))+ cfg = StoreConfig+ { storeConfMaxPeers = 20+ , storeConfInitPeers = []+ , storeConfDiscover = True+ , storeConfDB = w+ , storeConfNetwork = net+ , storeConfCache = Nothing+ , storeConfGap = gap+ , storeConfInitialGap = 20+ , storeConfCacheMin = 100+ , storeConfMaxKeys = 100 * 1000 * 1000+ , storeConfWipeMempool = False+ , storeConfPeerTimeout = 60+ , storeConfPeerMaxLife = 48 * 3600+ , storeConfConnect = dummyPeerConnect net ad+ }+ withStore cfg $ \Store {..} ->+ withSubscription storePublisher $ \sub ->+ lift $ f TestStore { testStoreDB = storeDB+ , testStoreBlockStore = storeBlock+ , testStoreChain = storeChain+ , testStoreEvents = sub+ } gap :: Word32 gap = 32@@ -170,7 +162,7 @@ \AAAAAAAAAAAAAAAAAAD/////DF8BAQgvRUIzMi4wL/////8BAPIFKgEAAAAjIQME7KZAozHsyrOO\ \wT6Wn6LtY47B39Th44JKsZ8BGJCvc6wAAAAA" -dummyPeerConnect :: Network -> NetworkAddress -> WithConnection+dummyPeerConnect :: Network -> NetworkAddress -> SockAddr -> WithConnection dummyPeerConnect net ad sa f = do r <- newInbox s <- newInbox@@ -192,18 +184,17 @@ outc .| awaitForever (`send` s) outc = mapMC $ \msg' -> return $ runPut (putMessage net msg')- inc =- forever $ do- x <- takeCE 24 .| foldC- case decode x of- Left _ -> error "Dummy peer not decode message header"- Right (MessageHeader _ _ len _) -> do- y <- takeCE (fromIntegral len) .| foldC- case runGet (getMessage net) $ x `B.append` y of- Left e ->- error $- "Dummy peer could not decode payload: " <> show e- Right msg' -> yield msg'+ inc = forever $ do+ x <- takeCE 24 .| foldC+ case decode x of+ Left _ ->+ error "Dummy peer not decode message header"+ Right (MessageHeader _ _ len _) -> do+ y <- takeCE (fromIntegral len) .| foldC+ case runGet (getMessage net) $ x `B.append` y of+ Right msg' -> yield msg'+ Left e -> error $+ "Dummy peer could not decode payload: " <> show e mockPeerReact :: Message -> [Message] mockPeerReact (MPing (Ping n)) = [MPong (Pong n)]@@ -214,7 +205,7 @@ hs' = map f allBlocks mockPeerReact (MGetData (GetData ivs)) = mapMaybe f ivs where- f (InvVector InvBlock h) = MBlock <$> listToMaybe (filter (l h) allBlocks)+ f (InvVector InvBlock h) = MBlock <$> find (l h) allBlocks f _ = Nothing l h b = headerHash (blockHeader b) == BlockHash h mockPeerReact _ = []