haskoin-store 0.23.6 → 0.23.7
raw patch · 7 files changed
+176/−116 lines, 7 files
Files
- CHANGELOG.md +5/−0
- app/Main.hs +5/−0
- haskoin-store.cabal +2/−2
- src/Haskoin/Store/BlockStore.hs +105/−49
- src/Haskoin/Store/Logic.hs +44/−54
- src/Haskoin/Store/Manager.hs +14/−11
- test/Haskoin/StoreSpec.hs +1/−0
CHANGELOG.md view
@@ -4,6 +4,11 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## 0.23.6+### Added+- Ability to wipe mempool at start.+- Improvements to mempool processing code.+ ## 0.23.5 ### Changed - Tighten the locking loop to avoid slow cache building.
app/Main.hs view
@@ -49,6 +49,7 @@ , configRedisURL :: !String , configRedisMin :: !Int , configRedisMax :: !Integer+ , configWipeMempool :: !Bool } defPort :: Int@@ -155,6 +156,8 @@ help "Maximum number of keys in Redis xpub cache" <> showDefault <> value defRedisMax+ configWipeMempool <-+ switch $ long "wipemempool" <> help "Wipe mempool when starting" pure Config { configWebLimits = WebLimits {..}@@ -220,6 +223,7 @@ , configRedisURL = redisurl , configRedisMin = cachemin , configRedisMax = redismax+ , configWipeMempool = wipemempool } = runStderrLoggingT . filterLogger l $ do $(logInfoS) "Main" $@@ -241,6 +245,7 @@ , storeConfInitialGap = maxLimitInitialGap limits , storeConfCacheMin = cachemin , storeConfMaxKeys = redismax+ , storeConfWipeMempool = wipemempool } in withStore scfg $ \st -> let wcfg =
haskoin-store.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: ed9717f6673f6dc980e5bf539fbd715a4d5c18968c4aec11058a10f84aecd382+-- hash: 5647240d12c600ec320ddcf3af88d203d5be73e9a92d370ce030d34be475dd67 name: haskoin-store-version: 0.23.6+version: 0.23.7 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.
src/Haskoin/Store/BlockStore.hs view
@@ -13,7 +13,8 @@ import Control.Applicative ((<|>)) import Control.Monad (forM, forM_, forever, guard, mzero, unless, void, when)-import Control.Monad.Except (ExceptT, runExceptT)+import Control.Monad.Except (ExceptT, MonadError (..),+ runExceptT) import Control.Monad.Logger (MonadLoggerIO, logDebugS, logErrorS, logInfoS, logWarnS) import Control.Monad.Reader (MonadReader, ReaderT (..), asks)@@ -41,8 +42,9 @@ import Haskoin.Node (Chain, Manager) import Haskoin.Store.Common (BlockStore, BlockStoreMessage (..),- StoreEvent (..), StoreRead (..),- StoreWrite (..), UnixTime)+ BlockTx (..), StoreEvent (..),+ StoreRead (..), StoreWrite (..),+ UnixTime) import Haskoin.Store.Database.Reader (DatabaseReader) import Haskoin.Store.Database.Writer (DatabaseWriter, runDatabaseWriter)@@ -56,13 +58,15 @@ import System.Random (randomRIO) import UnliftIO (Exception, MonadIO, MonadUnliftIO, TVar, atomically,- liftIO, newTVarIO, readTVarIO,- throwIO, withAsync, writeTVar)+ liftIO, modifyTVar, newTVarIO,+ readTVar, readTVarIO, throwIO,+ withAsync, writeTVar) import UnliftIO.Concurrent (threadDelay) data BlockException = BlockNotInChain !BlockHash | Uninitialized+ | CorruptDatabase | AncestorNotInChain !BlockHeight !BlockHash deriving (Show, Eq, Ord, Exception)@@ -74,25 +78,28 @@ } -- | Block store process state.-data BlockRead = BlockRead- { mySelf :: !BlockStore- , myConfig :: !BlockStoreConfig- , myPeer :: !(TVar (Maybe Syncing))- }+data BlockRead =+ BlockRead+ { mySelf :: !BlockStore+ , myConfig :: !BlockStoreConfig+ , myPeer :: !(TVar (Maybe Syncing))+ , myTxs :: !(TVar [Tx])+ } -- | Configuration for a block store. data BlockStoreConfig = BlockStoreConfig- { blockConfManager :: !Manager+ { blockConfManager :: !Manager -- ^ peer manager from running node- , blockConfChain :: !Chain+ , blockConfChain :: !Chain -- ^ chain from a running node- , blockConfListener :: !(Listen StoreEvent)+ , blockConfListener :: !(Listen StoreEvent) -- ^ listener for store events- , blockConfDB :: !DatabaseReader+ , blockConfDB :: !DatabaseReader -- ^ RocksDB database handle- , blockConfNet :: !Network+ , blockConfNet :: !Network -- ^ network constants+ , blockConfWipeMempool :: !Bool } type BlockT m = ReaderT BlockRead m@@ -139,10 +146,35 @@ -> m () blockStore cfg inbox = do pb <- newTVarIO Nothing+ ts <- newTVarIO [] runReaderT- (ini >> run)- BlockRead {mySelf = inboxToMailbox inbox, myConfig = cfg, myPeer = pb}+ (ini >> wipe >> run)+ BlockRead+ { mySelf = inboxToMailbox inbox+ , myConfig = cfg+ , myPeer = pb+ , myTxs = ts+ } where+ del txs =+ forM (zip [(1 :: Integer) ..] txs) $ \(i, tx) -> do+ $(logDebugS) "BlockStore" $+ "Wiping mempool tx " <> cs (show i) <> "/" <>+ cs (show (length txs)) <>+ ": " <>+ txHashToHex (blockTxHash tx)+ deleteTx True (blockTxHash tx)+ wipe+ | blockConfWipeMempool cfg = do+ txs <- getMempool+ runImport (del txs) >>= \case+ Left e -> do+ $(logErrorS) "BlockStore" $+ "Could not delete mempool, database corrupt: " <>+ cs (show e)+ throwIO CorruptDatabase+ Right _ -> return ()+ | otherwise = return () ini = do runImport initBest >>= \case Left e -> do@@ -226,19 +258,39 @@ m = "I do not like peers that cannot find them blocks" processTx :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> Tx -> BlockT m ()-processTx _p tx =- isInSync >>= \sync ->- when sync $ do- now <- fromIntegral . systemSeconds <$> liftIO getSystemTime- runImport (newMempoolTx tx now) >>= \case- Right (Just deleted) -> do- l <- blockConfListener <$> asks myConfig+processTx _p tx = asks myTxs >>= \b -> atomically (modifyTVar b (tx :))++processMempool :: (MonadUnliftIO m, MonadLoggerIO m) => BlockT m ()+processMempool = do+ g >>= \case+ [] -> return ()+ txs -> do+ output <-+ runImport . forM (zip [(1 :: Int) ..] txs) $ \(i, tx) -> do+ now <- fromIntegral . systemSeconds <$> liftIO getSystemTime $(logInfoS) "BlockStore" $- "New mempool tx: " <> txHashToHex (txHash tx)- atomically $ do- mapM_ (l . StoreTxDeleted) deleted- l (StoreMempoolNew (txHash tx))- _ -> return ()+ "New mempool tx " <> cs (show i) <> "/" <>+ cs (show (length txs)) <>+ ": " <>+ txHashToHex (txHash tx)+ fmap (txHash tx, ) <$> newMempoolTx tx now `catchError` h+ case output of+ Left e -> do+ $(logErrorS) "BlockStore" $+ "Importing mempool failed: " <> cs (show e)+ Right xs -> do+ l <- asks (blockConfListener . myConfig)+ atomically $ forM_ (catMaybes xs) $ \(th, deleted) -> do+ mapM_ (l . StoreTxDeleted) deleted+ l (StoreMempoolNew th)+ where+ h _ = return Nothing+ g =+ asks myTxs >>= \b ->+ atomically $ do+ txs <- readTVar b+ writeTVar b []+ return txs processOrphans :: (MonadUnliftIO m, MonadLoggerIO m) => BlockT m ()@@ -257,25 +309,28 @@ orphans <- getOrphans case orphans of [] -> return ()- _ ->- $(logInfoS) "BlockStore" $- "Attempting to import " <> cs (show (length orphans)) <>- " orphan transactions"- ops <-- zip (map snd orphans) <$>- mapM (runImport . uncurry importOrphan) orphans- let tths =- [ (txHash tx, hs)- | (tx, emths) <- ops- , let Right (Just hs) = emths- ]- ihs = map fst tths- dhs = concatMap snd tths- l <- blockConfListener <$> asks myConfig- atomically $ do- mapM_ (l . StoreTxDeleted) dhs- mapM_ (l . StoreMempoolNew) ihs-+ _ -> go orphans+ where+ h _ = return Nothing+ go os = do+ output <-+ runImport . forM (zip [(1 :: Int) ..] os) $ \(i, (t, tx)) -> do+ $(logInfoS) "BlockStore" $+ "Attempting to import orphan tx " <> cs (show i) <> "/" <>+ cs (show (length os)) <>+ ": " <>+ txHashToHex (txHash tx)+ fmap (txHash tx, ) <$> importOrphan t tx `catchError` h+ case output of+ Left e -> do+ $(logErrorS) "BlockStore" $+ "Importing orphans failed: " <> cs (show e)+ Right xs -> do+ l <- asks (blockConfListener . myConfig)+ atomically $+ forM_ (catMaybes xs) $ \(th, deleted) -> do+ mapM_ (l . StoreTxDeleted) deleted+ l (StoreMempoolNew th) processTxs :: (MonadUnliftIO m, MonadLoggerIO m)@@ -498,6 +553,7 @@ processBlockStoreMessage (BlockTxReceived p tx) = processTx p tx processBlockStoreMessage (BlockTxAvailable p ts) = processTxs p ts processBlockStoreMessage (BlockPing r) = do+ processMempool processOrphans checkTime pruneMempool@@ -506,5 +562,5 @@ pingMe :: MonadLoggerIO m => BlockStore -> m () pingMe mbox = forever $ do- threadDelay =<< liftIO (randomRIO (5 * 1000 * 1000, 10 * 1000 * 1000))+ threadDelay =<< liftIO (randomRIO (100 * 1000, 1000 * 1000)) BlockPing `query` mbox
src/Haskoin/Store/Logic.hs view
@@ -17,8 +17,7 @@ import Control.Monad (forM, forM_, unless, void, when, zipWithM_)-import Control.Monad.Except (MonadError, catchError,- throwError)+import Control.Monad.Except (MonadError (..)) import Control.Monad.Logger (MonadLogger, logErrorS, logWarnS) import qualified Data.ByteString as B@@ -38,11 +37,10 @@ OutPoint (..), Tx (..), TxHash, TxIn (..), TxOut (..), addrToString, blockHashToHex,- genesisBlock,- genesisNode, headerHash,- isGenesis, nullOutPoint,- scriptToAddressBS, txHash,- txHashToHex)+ genesisBlock, genesisNode,+ headerHash, isGenesis,+ nullOutPoint, scriptToAddressBS,+ txHash, txHashToHex) import Haskoin.Store.Common (Balance (..), BlockData (..), BlockRef (..), BlockTx (..), Prev (..), Spender (..),@@ -64,7 +62,6 @@ | BestBlockUnknown | BestBlockNotFound !Text | BlockNotBest !Text- | OrphanTx !Text | TxNotFound !Text | NoUnspent !Text | TxInvalidOp !Text@@ -113,23 +110,7 @@ => UnixTime -> Tx -> m (Maybe [TxHash]) -- ^ deleted transactions or nothing if import failed-importOrphan t tx = do- go `catchError` ex- where- go = do- maybetxids <-- newMempoolTx tx t >>= \case- Just ths -> return (Just ths)- Nothing -> return Nothing- deleteOrphanTx (txHash tx)- return maybetxids- ex (OrphanTx _) = do- return Nothing- ex _ = do- $(logWarnS) "BlockStore" $- "Deleted bad orphan tx: " <> txHashToHex (txHash tx)- deleteOrphanTx (txHash tx)- return Nothing+importOrphan t tx = deleteOrphanTx (txHash tx) >> newMempoolTx tx t newMempoolTx :: ( StoreRead m@@ -155,7 +136,7 @@ $(logWarnS) "BlockStore" $ "Orphan tx: " <> txHashToHex (txHash tx) insertOrphanTx tx w- throwError $ OrphanTx (txHashToHex (txHash tx))+ return Nothing else f f = do us <-@@ -163,10 +144,14 @@ t <- getImportTx (outPointHash op) getTxOutput (outPointIndex op) t let ds = map spenderHash (mapMaybe outputSpender us)+ h e = do+ $(logErrorS) "BlockStore" $+ "Could not import mempool tx " <> txHashToHex (txHash tx) <>+ ": " <>+ cs (show e)+ return Nothing if null ds- then do- ths <- importTx (MemRef w) w tx- return (Just ths)+ then fmap Just (importTx (MemRef w) w tx) `catchError` h else g ds g ds = do net <- getNetwork@@ -333,32 +318,34 @@ $(logErrorS) "BlockStore" $ "Insufficient funds for tx: " <> txHashToHex (txHash tx) throwError (InsufficientFunds (txHashToHex th))- zipWithM_ (spendOutput br (txHash tx)) [0 ..] us- zipWithM_- (\i o -> newOutput br (OutPoint (txHash tx) i) o)- [0 ..]- (txOut tx)- rbf <- getrbf- let t =- Transaction- { transactionBlock = br- , transactionVersion = txVersion tx- , transactionLockTime = txLockTime tx- , transactionInputs =- if iscb- then zipWith mkcb (txIn tx) ws- else zipWith3 mkin us (txIn tx) ws- , transactionOutputs = map mkout (txOut tx)- , transactionDeleted = False- , transactionRBF = rbf- , transactionTime = tt- }- let (d, _) = fromTransaction t- insertTx d- updateAddressCounts (txAddresses t) (+ 1)- unless (confirmed br) $ insertIntoMempool (txHash tx) (memRefTime br)+ commit us return ths where+ commit us = do+ zipWithM_ (spendOutput br (txHash tx)) [0 ..] us+ zipWithM_+ (\i o -> newOutput br (OutPoint (txHash tx) i) o)+ [0 ..]+ (txOut tx)+ rbf <- getrbf+ let t =+ Transaction+ { transactionBlock = br+ , transactionVersion = txVersion tx+ , transactionLockTime = txLockTime tx+ , transactionInputs =+ if iscb+ then zipWith mkcb (txIn tx) ws+ else zipWith3 mkin us (txIn tx) ws+ , transactionOutputs = map mkout (txOut tx)+ , transactionDeleted = False+ , transactionRBF = rbf+ , transactionTime = tt+ }+ let (d, _) = fromTransaction t+ insertTx d+ updateAddressCounts (txAddresses t) (+ 1)+ unless (confirmed br) $ insertIntoMempool (txHash tx) (memRefTime br) uns op = getUnspent op >>= \case Just u -> return (u, [])@@ -398,7 +385,10 @@ let hs = nub $ map (outPointHash . prevOutput) (txIn tx) in fmap or . forM hs $ \h -> getTxData h >>= \case- Nothing -> throwError (TxNotFound (txHashToHex h))+ Nothing -> do+ $(logErrorS) "BlockStore" $+ "Transaction not found: " <> txHashToHex h+ throwError (TxNotFound (txHashToHex h)) Just t | confirmed (txDataBlock t) -> return False | txDataRBF t -> return True
src/Haskoin/Store/Manager.hs view
@@ -58,26 +58,28 @@ -- | Configuration for a 'Store'. data StoreConfig = StoreConfig- { storeConfMaxPeers :: !Int+ { storeConfMaxPeers :: !Int -- ^ max peers to connect to- , storeConfInitPeers :: ![HostPort]+ , storeConfInitPeers :: ![HostPort] -- ^ static set of peers to connect to- , storeConfDiscover :: !Bool+ , storeConfDiscover :: !Bool -- ^ discover new peers- , storeConfDB :: !FilePath+ , storeConfDB :: !FilePath -- ^ RocksDB database path- , storeConfNetwork :: !Network+ , storeConfNetwork :: !Network -- ^ network constants- , storeConfCache :: !(Maybe String)+ , storeConfCache :: !(Maybe String) -- ^ Redis cache configuration- , storeConfInitialGap :: !Word32+ , storeConfInitialGap :: !Word32 -- ^ gap on extended public key with no transactions- , storeConfGap :: !Word32+ , storeConfGap :: !Word32 -- ^ gap for extended public keys- , storeConfCacheMin :: !Int+ , storeConfCacheMin :: !Int -- ^ cache xpubs with more than this many used addresses- , storeConfMaxKeys :: !Integer+ , storeConfMaxKeys :: !Integer -- ^ maximum number of keys in Redis cache+ , storeConfWipeMempool :: !Bool+ -- ^ wipe mempool when starting } withStore ::@@ -90,7 +92,7 @@ let chain = inboxToMailbox chaininbox maybecacheconn <- case storeConfCache cfg of- Nothing -> return Nothing+ Nothing -> return Nothing Just redisurl -> Just <$> connectRedis redisurl db <- connectRocksDB@@ -142,6 +144,7 @@ , blockConfListener = (`sendSTM` pub) . Event , blockConfDB = db , blockConfNet = storeConfNetwork cfg+ , blockConfWipeMempool = storeConfWipeMempool cfg } runaction = action
test/Haskoin/StoreSpec.hs view
@@ -86,6 +86,7 @@ , storeConfInitialGap = 20 , storeConfCacheMin = 100 , storeConfMaxKeys = 100 * 1000 * 1000+ , storeConfWipeMempool = False } withStore cfg $ \Store {..} -> withSubscription storePublisher $ \sub -> lift $