diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ 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.11 +### Fixed +- Streamline mempool transaction importing. + ## 0.23.10 ### Fixed - Fix transactions not recorded in cache mempool set. 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: abd6d08a8a871be74bde6a5b341cc3d84077ac63c8896aa0c4af02f8a6b371c8 +-- hash: 7993f74354bd0a9ecaf5f50e8e72712a0b0dca8d92ef828cefa5bf7ce77de855 name: haskoin-store -version: 0.23.10 +version: 0.23.11 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 @@ -22,6 +22,8 @@ import Control.Monad.Trans.Maybe (MaybeT (MaybeT), runMaybeT) 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, isNothing, listToMaybe, mapMaybe) import Data.String (fromString) @@ -31,7 +33,8 @@ BlockHeight, BlockNode (..), GetData (..), InvType (..), InvVector (..), Message (..), - Network (..), Tx, TxHash (..), + Network (..), OutPoint (..), + Tx (..), TxHash (..), TxIn (..), blockHashToHex, headerHash, txHash, txHashToHex) import Haskoin.Node (OnlinePeer (..), Peer, @@ -41,7 +44,7 @@ chainGetBlock, chainGetParents, killPeer, managerGetPeers, sendMessage) -import Haskoin.Node (Chain, Manager) +import Haskoin.Node (Chain, Manager, managerGetPeer) import Haskoin.Store.Common (BlockStore, BlockStoreMessage (..), BlockTx (..), StoreEvent (..), @@ -82,17 +85,17 @@ 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)) - , myPending :: !(TVar (HashMap TxHash UnixTime)) + { mySelf :: !BlockStore + , myConfig :: !BlockStoreConfig + , myPeer :: !(TVar (Maybe Syncing)) + , myTxs :: !(TVar (HashMap TxHash PendingTx)) } -- | Configuration for a block store. @@ -154,7 +157,6 @@ blockStore cfg inbox = do pb <- newTVarIO Nothing ts <- newTVarIO HashMap.empty - ps <- newTVarIO HashMap.empty runReaderT (ini >> wipe >> run) BlockRead @@ -162,31 +164,30 @@ , myConfig = cfg , myPeer = pb , myTxs = ts - , myPending = ps } where - del txs = + del n txs = forM (zip [(1 :: Integer) ..] txs) $ \(i, tx) -> do $(logDebugS) "BlockStore" $ - "Wiping mempool tx " <> cs (show i) <> "/" <> - cs (show (length txs)) <> + "Wiping mempool tx " <> cs (show i) <> "/" <> cs (show n) <> ": " <> txHashToHex (blockTxHash tx) - deleteTx True (blockTxHash tx) - wipeit txs = do + deleteTx True False (blockTxHash tx) + wipeit n txs = do let (txs1, txs2) = splitAt 1000 txs case txs1 of [] -> return () _ -> - runImport (del txs1) >>= \case + runImport (del n txs1) >>= \case Left e -> do $(logErrorS) "BlockStore" $ "Could not delete mempool, database corrupt: " <> cs (show e) throwIO CorruptDatabase - Right _ -> wipeit txs2 + Right _ -> wipeit n txs2 wipe - | blockConfWipeMempool cfg = getMempool >>= wipeit + | blockConfWipeMempool cfg = + getMempool >>= \mem -> wipeit (length mem) mem | otherwise = return () ini = do runImport initBest >>= \case @@ -273,16 +274,7 @@ processTx :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> Tx -> BlockT m () processTx _p tx = do t <- fromIntegral . systemSeconds <$> liftIO getSystemTime - addPendingTx $ PendingTx t tx - -addPendingHash :: MonadIO m => UnixTime -> TxHash -> BlockT m () -addPendingHash t th = do - ts <- asks myTxs - ps <- asks myPending - atomically $ - HashMap.member th <$> readTVar ts >>= \case - True -> modifyTVar ps $ HashMap.delete th - False -> modifyTVar ps $ HashMap.insert th t + addPendingTx $ PendingTx t tx HashSet.empty prunePendingTxs :: MonadIO m => BlockT m () prunePendingTxs = do @@ -290,61 +282,66 @@ now <- fromIntegral . systemSeconds <$> liftIO getSystemTime atomically . modifyTVar ts $ HashMap.filter ((> now - 600) . pendingTxTime) -prunePendingHashes :: MonadIO m => BlockT m () -prunePendingHashes = do - ps <- asks myPending - now <- fromIntegral . systemSeconds <$> liftIO getSystemTime - atomically . modifyTVar ps $ HashMap.filter (> now - 10) - addPendingTx :: MonadIO m => PendingTx -> BlockT m () addPendingTx p = do ts <- asks myTxs - ps <- asks myPending - atomically $ do - modifyTVar ts $ HashMap.insert th p - modifyTVar ps $ HashMap.delete th + 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 - ps <- asks myPending - atomically $ do - a <- HashMap.member th <$> readTVar ts - b <- HashMap.member th <$> readTVar ps - return $ a && b + atomically $ HashMap.member th <$> readTVar ts allPendingTxs :: MonadIO m => BlockT m [PendingTx] allPendingTxs = do ts <- asks myTxs - ps <- asks myPending atomically $ do pend <- readTVar ts - writeTVar ts HashMap.empty - forM_ pend $ modifyTVar ps . HashMap.delete . txHash . pendingTx - return $ sortit pend + writeTVar ts $ HashMap.filter (not . null . pendingDeps) pend + return $ sortit $ HashMap.filter (null . pendingDeps) pend where sortit pend = mapMaybe (flip HashMap.lookup pend . txHash . snd) . sortTxs . map (pendingTx) $ HashMap.elems pend +fulfillOrphans :: MonadIO m => TxHash -> BlockT m () +fulfillOrphans th = do + ts <- asks myTxs + atomically $ do + pend <- readTVar ts + let pend' = HashMap.map upd pend + writeTVar ts pend' + where + upd p = p {pendingDeps = HashSet.delete th (pendingDeps p)} + + processMempool :: (MonadUnliftIO m, MonadLoggerIO m) => BlockT m () processMempool = do - prunePendingHashes allPendingTxs >>= \txs -> if null txs then return () - else go txs >> prunePendingTxs + else go txs where + pend p = do + ex <- + fmap (HashSet.fromList . catMaybes) $ + forM (txIn (pendingTx p)) $ \i -> do + let op = prevOutput i + h = outPointHash op + getUnspent op >>= \case + Nothing -> return (Just h) + Just _ -> return Nothing + addPendingTx $ p {pendingDeps = ex} go ps = do output <- runImport . forM (zip [(1 :: Int) ..] ps) $ \(i, p) -> do let tx = pendingTx p t = pendingTxTime p th = txHash tx - h x TxOrphan = return (Left (Just x)) + h x TxOrphan {} = return (Left (Just x)) h _ _ = return (Left Nothing) $(logInfoS) "BlockStore" $ "New mempool tx " <> cs (show i) <> "/" <> @@ -360,9 +357,10 @@ "Importing mempool failed: " <> cs (show e) Right xs -> do forM_ xs $ \case - Left (Just p) -> addPendingTx p + Left (Just p) -> pend p Left Nothing -> return () Right (th, deleted) -> do + fulfillOrphans th l <- asks (blockConfListener . myConfig) atomically $ do mapM_ (l . StoreTxDeleted) deleted @@ -381,19 +379,24 @@ runMaybeT $ do guard . not =<< lift (isPending h) guard . isNothing =<< lift (getTxData h) - now <- fromIntegral . systemSeconds <$> liftIO getSystemTime - lift $ addPendingHash now h return h - unless (null xs) (go xs) + unless (null xs) $ go xs where go xs = do + p' <- + do mgr <- asks (blockConfManager . myConfig) + managerGetPeer p mgr >>= \case + Nothing -> return "???" + Just op -> return . cs . show $ onlinePeerAddress op forM_ (zip [(1 :: Int) ..] xs) $ \(i, h) -> $(logInfoS) "BlockStore" $ "Requesting transaction " <> cs (show i) <> "/" <> cs (show (length xs)) <> - ": " <> - txHashToHex h - net <- blockConfNet <$> asks myConfig + " " <> + txHashToHex h <> + " from peer " <> + p' + net <- asks (blockConfNet . myConfig) let inv = if getSegWit net then InvWitnessTx @@ -444,7 +447,7 @@ $(logInfoS) "BlockStore" $ "Removing " <> cs (show (length old)) <> " old mempool transactions" forM_ old $ \txid -> - runImport (deleteTx True txid) >>= \case + runImport (deleteTx True False txid) >>= \case Left _ -> return () Right txids -> do listener <- asks (blockConfListener . myConfig) @@ -597,6 +600,7 @@ processBlockStoreMessage (BlockTxAvailable p ts) = processTxs p ts processBlockStoreMessage (BlockPing r) = do processMempool + prunePendingTxs checkTime pruneMempool atomically (r ()) 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 @@ -23,7 +23,7 @@ import Data.Either (rights) import qualified Data.IntMap.Strict as I import Data.List (nub, sort) -import Data.Maybe (fromMaybe, isNothing, mapMaybe) +import Data.Maybe (fromMaybe, isNothing) import Data.Serialize (encode) import Data.String (fromString) import Data.String.Conversions (cs) @@ -56,13 +56,14 @@ data ImportException = PrevBlockNotBest !Text - | TxOrphan + | TxOrphan !Text | UnconfirmedCoinbase !Text | BestBlockUnknown | BestBlockNotFound !Text | BlockNotBest !Text | TxNotFound !Text - | NoUnspent !Text + | OutputSpent !Text + | CannotDeleteNonRBF !Text | TxInvalidOp !Text | TxDeleted !Text | TxDoubleSpend !Text @@ -109,47 +110,7 @@ $(logDebugS) "BlockStore" $ "Transaction already in store: " <> txHashToHex (txHash tx) return Nothing - _ -> go - where - go = do - orp <- - any isNothing <$> - mapM (getTxData . outPointHash . prevOutput) (txIn tx) - if orp - then do - $(logWarnS) "BlockStore" $ - "Orphan tx: " <> txHashToHex (txHash tx) - throwError TxOrphan - else f - f = do - us <- - forM (txIn tx) $ \TxIn {prevOutput = op} -> do - t <- getImportTx (outPointHash op) - getTxOutput (outPointIndex op) t - let ds = map spenderHash (mapMaybe outputSpender us) - if null ds - then fmap Just (importTx (MemRef w) w tx) - else g ds - g ds = do - net <- getNetwork - rbf <- - if getReplaceByFee net - then and <$> mapM isrbf ds - else return False - if rbf - then r ds - else n - r ds = do - $(logWarnS) "BlockStore" $ - "Replace by fee tx: " <> txHashToHex (txHash tx) - ths <- concat <$> forM ds (deleteTx True) - dts <- importTx (MemRef w) w tx - return (Just (nub (ths <> dts))) - n = do - $(logWarnS) "BlockStore" $ "Conflicting tx: " <> txHashToHex (txHash tx) - insertDeletedMempoolTx tx w - return Nothing - isrbf th = transactionRBF <$> getImportTx th + _ -> Just <$> importTx (MemRef w) w tx revertBlock :: ( StoreRead m @@ -180,7 +141,7 @@ txs <- mapM (fmap transactionData . getImportTx) (blockDataTxs bd) ths <- nub . concat <$> - mapM (deleteTx False . txHash . snd) (reverse (sortTxs txs)) + mapM (deleteTx False False . txHash . snd) (reverse (sortTxs txs)) setBest (prevBlock (blockDataHeader bd)) insertBlock bd {blockDataMainChain = False} return ths @@ -298,16 +259,16 @@ $(logErrorS) "BlockStore" $ "Insufficient funds for tx: " <> txHashToHex (txHash tx) throwError (InsufficientFunds (txHashToHex th)) - commit us + rbf <- isRBF br tx + commit rbf us return ths where - commit us = do + commit rbf 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 @@ -342,13 +303,15 @@ txHashToHex (outPointHash op) <> " " <> fromString (show (outPointIndex op)) - throwError (NoUnspent (cs (show op))) + $(logErrorS) "BlockStore" $ + "Orphan tx: " <> txHashToHex (txHash tx) + throwError (TxOrphan (txHashToHex (txHash tx))) Just Spender {spenderHash = s} -> do $(logWarnS) "BlockStore" $ "Deleting transaction " <> txHashToHex s <> " because it conflicts with " <> txHashToHex (txHash tx) - ths <- deleteTx True s + ths <- deleteTx True True s getUnspent op >>= \case Nothing -> do $(logErrorS) "BlockStore" $ @@ -356,27 +319,11 @@ txHashToHex (outPointHash op) <> " " <> fromString (show (outPointIndex op)) - throwError (NoUnspent (cs (show op))) + throwError (OutputSpent (cs (show op))) Just u -> return (u, ths) th = txHash tx iscb = all (== nullOutPoint) (map prevOutput (txIn tx)) ws = map Just (txWitness tx) <> repeat Nothing - getrbf - | iscb = return False - | any ((< 0xffffffff - 1) . txInSequence) (txIn tx) = return True - | confirmed br = return False - | otherwise = - let hs = nub $ map (outPointHash . prevOutput) (txIn tx) - in fmap or . forM hs $ \h -> - getTxData h >>= \case - Nothing -> do - $(logErrorS) "BlockStore" $ - "Transaction not found: " <> txHashToHex h - throwError (TxNotFound (txHashToHex h)) - Just t - | confirmed (txDataBlock t) -> return False - | txDataRBF t -> return True - | otherwise -> return False mkcb ip w = StoreCoinbase { inputPoint = prevOutput ip @@ -492,104 +439,85 @@ , MonadError ImportException m ) => Bool -- ^ only delete transaction if unconfirmed + -> Bool -- ^ do RBF check before deleting transaction -> TxHash -> m [TxHash] -- ^ deleted transactions -deleteTx mo h = do - getTxData h >>= \case +deleteTx memonly rbfcheck txhash = do + getTxData txhash >>= \case Nothing -> do $(logErrorS) "BlockStore" $ - "Cannot find tx to delete: " <> txHashToHex h - throwError (TxNotFound (txHashToHex h)) + "Cannot find tx to delete: " <> txHashToHex txhash + throwError (TxNotFound (txHashToHex txhash)) Just t | txDataDeleted t -> do $(logWarnS) "BlockStore" $ - "Already deleted tx: " <> txHashToHex h + "Already deleted tx: " <> txHashToHex txhash return [] - | mo && confirmed (txDataBlock t) -> do + | memonly && confirmed (txDataBlock t) -> do $(logErrorS) "BlockStore" $ - "Will not delete confirmed tx: " <> txHashToHex h - throwError (TxConfirmed (txHashToHex h)) + "Will not delete confirmed tx: " <> txHashToHex txhash + throwError (TxConfirmed (txHashToHex txhash)) + | rbfcheck -> + isRBF (txDataBlock t) (txData t) >>= \case + True -> go t + False -> do + $(logErrorS) "BlockStore" $ + "Delete non-RBF transaction attempted: " <> + txHashToHex txhash + throwError $ CannotDeleteNonRBF (txHashToHex txhash) | otherwise -> go t where go t = do - $(logWarnS) "BlockStore" $ "Deleting transaction: " <> txHashToHex h - ss <- nub . map spenderHash . I.elems <$> getSpenders h + $(logWarnS) "BlockStore" $ + "Deleting transaction: " <> txHashToHex txhash + ss <- nub . map spenderHash . I.elems <$> getSpenders txhash ths <- fmap concat $ forM ss $ \s -> do $(logWarnS) "BlockStore" $ "Deleting descendant " <> txHashToHex s <> " to delete parent " <> - txHashToHex h - deleteTx True s + txHashToHex txhash + deleteTx True rbfcheck s forM_ (take (length (txOut (txData t))) [0 ..]) $ \n -> - delOutput (OutPoint h n) + delOutput (OutPoint txhash n) let ps = filter (/= nullOutPoint) (map prevOutput (txIn (txData t))) mapM_ unspendOutput ps - unless (confirmed (txDataBlock t)) $ deleteFromMempool h + unless (confirmed (txDataBlock t)) $ deleteFromMempool txhash insertTx t {txDataDeleted = True} updateAddressCounts (txDataAddresses t) (subtract 1) - return $ nub (h : ths) + return $ nub (txhash : ths) -insertDeletedMempoolTx :: - ( StoreRead m - , StoreWrite m - , MonadLogger m - , MonadError ImportException m - ) - => Tx - -> UnixTime - -> m () -insertDeletedMempoolTx tx w = do - us <- - forM (txIn tx) $ \TxIn {prevOutput = op} -> - getImportTx (outPointHash op) >>= getTxOutput (outPointIndex op) - rbf <- getrbf - let d = - fst $ - fromTransaction - Transaction - { transactionBlock = MemRef w - , transactionVersion = txVersion tx - , transactionLockTime = txLockTime tx - , transactionInputs = zipWith3 mkin us (txIn tx) ws - , transactionOutputs = map mkout (txOut tx) - , transactionDeleted = True - , transactionRBF = rbf - , transactionTime = w - } - insertTx d + +isRBF :: + (StoreRead m, MonadLogger m, MonadError ImportException m) + => BlockRef + -> Tx + -> m Bool +isRBF br tx = + getNetwork >>= \net -> + if getReplaceByFee net + then go + else return False where - ws = map Just (txWitness tx) <> repeat Nothing - getrbf + go + | all (== nullOutPoint) (map prevOutput (txIn tx)) = return False | any ((< 0xffffffff - 1) . txInSequence) (txIn tx) = return True + | confirmed br = return False | otherwise = let hs = nub $ map (outPointHash . prevOutput) (txIn tx) - in fmap or . forM hs $ \h -> + ck [] = return False + ck (h:hs') = getTxData h >>= \case Nothing -> do $(logErrorS) "BlockStore" $ - "Tx not found: " <> txHashToHex h + "Transaction not found: " <> txHashToHex h throwError (TxNotFound (txHashToHex h)) Just t - | confirmed (txDataBlock t) -> return False + | confirmed (txDataBlock t) -> ck hs' | txDataRBF t -> return True - | otherwise -> return False - mkin u ip wit = - StoreInput - { inputPoint = prevOutput ip - , inputSequence = txInSequence ip - , inputSigScript = scriptInput ip - , inputPkScript = outputScript u - , inputAmount = outputAmount u - , inputWitness = wit - } - mkout o = - StoreOutput - { outputAmount = outValue o - , outputScript = scriptOutput o - , outputSpender = Nothing - } + | otherwise -> ck hs' + in ck hs newOutput :: ( StoreRead m