haskoin-store 0.23.10 → 0.23.11
raw patch · 4 files changed
+126/−190 lines, 4 files
Files
- CHANGELOG.md +4/−0
- haskoin-store.cabal +2/−2
- src/Haskoin/Store/BlockStore.hs +61/−57
- src/Haskoin/Store/Logic.hs +59/−131
CHANGELOG.md view
@@ -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.
haskoin-store.cabal view
@@ -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.
src/Haskoin/Store/BlockStore.hs view
@@ -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 ())
src/Haskoin/Store/Logic.hs view
@@ -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