diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -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 = diff --git a/haskoin-store.cabal b/haskoin-store.cabal --- a/haskoin-store.cabal +++ b/haskoin-store.cabal @@ -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. diff --git a/src/Haskoin/Store/BlockStore.hs b/src/Haskoin/Store/BlockStore.hs --- a/src/Haskoin/Store/BlockStore.hs +++ b/src/Haskoin/Store/BlockStore.hs @@ -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 diff --git a/src/Haskoin/Store/Logic.hs b/src/Haskoin/Store/Logic.hs --- a/src/Haskoin/Store/Logic.hs +++ b/src/Haskoin/Store/Logic.hs @@ -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 diff --git a/src/Haskoin/Store/Manager.hs b/src/Haskoin/Store/Manager.hs --- a/src/Haskoin/Store/Manager.hs +++ b/src/Haskoin/Store/Manager.hs @@ -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 diff --git a/test/Haskoin/StoreSpec.hs b/test/Haskoin/StoreSpec.hs --- a/test/Haskoin/StoreSpec.hs +++ b/test/Haskoin/StoreSpec.hs @@ -86,6 +86,7 @@ , storeConfInitialGap = 20 , storeConfCacheMin = 100 , storeConfMaxKeys = 100 * 1000 * 1000 + , storeConfWipeMempool = False } withStore cfg $ \Store {..} -> withSubscription storePublisher $ \sub -> lift $