diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,19 @@ 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.8 +## 0.23.10 +### Fixed +- Fix transactions not recorded in cache mempool set. +- Fix transactions being downloaded multiple times. + +### Removed +- Do not store orphan transactions in database. + +### Changed +- Use sets for incoming transactions instead of lists. +- Do not do anything to the cache if there are no xpubs in it. + +## 0.23.9 ### Fixed - Wiping mempool fixed. 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: f2fa672182f2f4eca9e4e8aba0fd7f37f495eb8a8693fa00d5393b376f155502 +-- hash: abd6d08a8a871be74bde6a5b341cc3d84077ac63c8896aa0c4af02f8a6b371c8 name: haskoin-store -version: 0.23.9 +version: 0.23.10 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 @@ -20,8 +20,10 @@ import Control.Monad.Reader (MonadReader, ReaderT (..), asks) import Control.Monad.Trans (lift) import Control.Monad.Trans.Maybe (MaybeT (MaybeT), runMaybeT) +import Data.HashMap.Strict (HashMap) +import qualified Data.HashMap.Strict as HashMap import Data.Maybe (catMaybes, isNothing, - listToMaybe) + listToMaybe, mapMaybe) import Data.String (fromString) import Data.String.Conversions (cs) import Data.Time.Clock.System (getSystemTime, systemSeconds) @@ -43,16 +45,15 @@ import Haskoin.Store.Common (BlockStore, BlockStoreMessage (..), BlockTx (..), StoreEvent (..), - StoreRead (..), StoreWrite (..), - UnixTime) + StoreRead (..), UnixTime, + sortTxs) import Haskoin.Store.Database.Reader (DatabaseReader) import Haskoin.Store.Database.Writer (DatabaseWriter, runDatabaseWriter) -import Haskoin.Store.Logic (ImportException, deleteTx, - getOldMempool, getOldOrphans, - importBlock, importOrphan, - initBest, newMempoolTx, - revertBlock) +import Haskoin.Store.Logic (ImportException (TxOrphan), + deleteTx, getOldMempool, + importBlock, initBest, + newMempoolTx, revertBlock) import NQE (Inbox, Listen, inboxToMailbox, query, receive) import System.Random (randomRIO) @@ -77,13 +78,21 @@ , syncingHead :: !BlockNode } +data PendingTx = + PendingTx + { pendingTxTime :: !UnixTime + , pendingTx :: !Tx + } + deriving (Show, Eq, Ord) + -- | Block store process state. data BlockRead = BlockRead - { mySelf :: !BlockStore - , myConfig :: !BlockStoreConfig - , myPeer :: !(TVar (Maybe Syncing)) - , myTxs :: !(TVar [Tx]) + { mySelf :: !BlockStore + , myConfig :: !BlockStoreConfig + , myPeer :: !(TVar (Maybe Syncing)) + , myTxs :: !(TVar (HashMap TxHash PendingTx)) + , myPending :: !(TVar (HashMap TxHash UnixTime)) } -- | Configuration for a block store. @@ -126,7 +135,6 @@ getTxData = runRocksDB . getTxData getSpender = runRocksDB . getSpender getSpenders = runRocksDB . getSpenders - getOrphanTx = runRocksDB . getOrphanTx getUnspent = runRocksDB . getUnspent getBalance = runRocksDB . getBalance getMempool = runRocksDB getMempool @@ -134,7 +142,6 @@ runRocksDB (getAddressesTxs addrs start limit) getAddressesUnspents addrs start limit = runRocksDB (getAddressesUnspents addrs start limit) - getOrphans = runRocksDB getOrphans getAddressUnspents a s = runRocksDB . getAddressUnspents a s getAddressTxs a s = runRocksDB . getAddressTxs a s @@ -146,7 +153,8 @@ -> m () blockStore cfg inbox = do pb <- newTVarIO Nothing - ts <- newTVarIO [] + ts <- newTVarIO HashMap.empty + ps <- newTVarIO HashMap.empty runReaderT (ini >> wipe >> run) BlockRead @@ -154,6 +162,7 @@ , myConfig = cfg , myPeer = pb , myTxs = ts + , myPending = ps } where del txs = @@ -164,22 +173,20 @@ ": " <> txHashToHex (blockTxHash tx) deleteTx True (blockTxHash tx) + wipeit txs = do + let (txs1, txs2) = splitAt 1000 txs + case txs1 of + [] -> return () + _ -> + runImport (del txs1) >>= \case + Left e -> do + $(logErrorS) "BlockStore" $ + "Could not delete mempool, database corrupt: " <> + cs (show e) + throwIO CorruptDatabase + Right _ -> wipeit txs2 wipe - | blockConfWipeMempool cfg = do - mem <- getMempool - let go txs = do - let (txs1, txs2) = splitAt 1000 txs - case txs1 of - [] -> return () - _ -> - runImport (del txs1) >>= \case - Left e -> do - $(logErrorS) "BlockStore" $ - "Could not delete mempool, database corrupt: " <> - cs (show e) - throwIO CorruptDatabase - Right _ -> go txs2 - go mem + | blockConfWipeMempool cfg = getMempool >>= wipeit | otherwise = return () ini = do runImport initBest >>= \case @@ -264,104 +271,134 @@ m = "I do not like peers that cannot find them blocks" processTx :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> Tx -> BlockT m () -processTx _p tx = asks myTxs >>= \b -> atomically (modifyTVar b (tx :)) +processTx _p tx = do + t <- fromIntegral . systemSeconds <$> liftIO getSystemTime + addPendingTx $ PendingTx t 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 " <> 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) +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 + +prunePendingTxs :: MonadIO m => BlockT m () +prunePendingTxs = do + ts <- asks myTxs + 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 where - h _ = return Nothing - g = - asks myTxs >>= \b -> - atomically $ do - txs <- readTVar b - writeTVar b [] - return txs + th = txHash (pendingTx p) -processOrphans :: - (MonadUnliftIO m, MonadLoggerIO m) => BlockT m () -processOrphans = - isInSync >>= \sync -> - when sync $ do - now <- fromIntegral . systemSeconds <$> liftIO getSystemTime - old <- getOldOrphans now - case old of - [] -> return () - _ -> do - $(logInfoS) "BlockStore" $ - "Removing " <> cs (show (length old)) <> - " expired orphan transactions" - void . runImport $ mapM_ deleteOrphanTx old - orphans <- getOrphans - case orphans of - [] -> return () - _ -> go orphans +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 + +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 where - h _ = return Nothing - go os = do + sortit pend = + mapMaybe (flip HashMap.lookup pend . txHash . snd) . + sortTxs . map (pendingTx) $ + HashMap.elems pend + +processMempool :: (MonadUnliftIO m, MonadLoggerIO m) => BlockT m () +processMempool = do + prunePendingHashes + allPendingTxs >>= \txs -> + if null txs + then return () + else go txs >> prunePendingTxs + where + go ps = do output <- - runImport . forM (zip [(1 :: Int) ..] os) $ \(i, (t, tx)) -> do + 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 _ _ = return (Left Nothing) $(logInfoS) "BlockStore" $ - "Attempting to import orphan tx " <> cs (show i) <> "/" <> - cs (show (length os)) <> + "New mempool tx " <> cs (show i) <> "/" <> + cs (show (length ps)) <> ": " <> - txHashToHex (txHash tx) - fmap (txHash tx, ) <$> importOrphan t tx `catchError` h + txHashToHex th + catchError + (maybe (Left Nothing) (Right . (th, )) <$> newMempoolTx tx t) + (h p) case output of Left e -> do $(logErrorS) "BlockStore" $ - "Importing orphans failed: " <> cs (show e) + "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) + forM_ xs $ \case + Left (Just p) -> addPendingTx p + Left Nothing -> return () + Right (th, deleted) -> do + l <- asks (blockConfListener . myConfig) + atomically $ do + mapM_ (l . StoreTxDeleted) deleted + l (StoreMempoolNew th) processTxs :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> [TxHash] -> ReaderT BlockRead m () -processTxs p hs = - isInSync >>= \sync -> - when sync $ do - xs <- - fmap catMaybes . forM hs $ \h -> - runMaybeT $ do - t <- lift $ getTxData h - guard (isNothing t) - return (getTxHash h) - unless (null xs) $ do - $(logInfoS) "BlockStore" $ - "Requesting " <> fromString (show (length xs)) <> - " new transactions" - net <- blockConfNet <$> asks myConfig - let inv = - if getSegWit net - then InvWitnessTx - else InvTx - MGetData (GetData (map (InvVector inv) xs)) `sendMessage` p +processTxs p hs = do + sync <- isInSync + when sync $ do + xs <- + fmap catMaybes . forM hs $ \h -> + 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) + where + go xs = do + forM_ (zip [(1 :: Int) ..] xs) $ \(i, h) -> + $(logInfoS) "BlockStore" $ + "Requesting transaction " <> cs (show i) <> "/" <> + cs (show (length xs)) <> + ": " <> + txHashToHex h + net <- blockConfNet <$> asks myConfig + let inv = + if getSegWit net + then InvWitnessTx + else InvTx + MGetData (GetData (map (InvVector inv . getTxHash) xs)) `sendMessage` p checkTime :: (MonadUnliftIO m, MonadLoggerIO m) => ReaderT BlockRead m () checkTime = @@ -560,7 +597,6 @@ processBlockStoreMessage (BlockTxAvailable p ts) = processTxs p ts processBlockStoreMessage (BlockPing r) = do processMempool - processOrphans checkTime pruneMempool atomically (r ()) diff --git a/src/Haskoin/Store/Cache.hs b/src/Haskoin/Store/Cache.hs --- a/src/Haskoin/Store/Cache.hs +++ b/src/Haskoin/Store/Cache.hs @@ -124,8 +124,6 @@ getBlocksAtHeight = lift . getBlocksAtHeight getBlock = lift . getBlock getTxData = lift . getTxData - getOrphanTx = lift . getOrphanTx - getOrphans = lift getOrphans getSpenders = lift . getSpenders getSpender = lift . getSpender getBalance = lift . getBalance @@ -441,6 +439,9 @@ maxKey :: ByteString maxKey = "max" +isAnythingCached :: MonadLoggerIO m => CacheT m Bool +isAnythingCached = runRedis $ Redis.exists maxKey + xPubAddrFunction :: DeriveType -> XPubKey -> Address xPubAddrFunction DeriveNormal = xPubAddr xPubAddrFunction DeriveP2SH = xPubCompatWitnessAddr @@ -508,12 +509,15 @@ pruneDB :: (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => CacheT m Integer pruneDB = - withLockWait $ do - x <- asks cacheMax - s <- runRedis Redis.dbsize - if s > x - then flush (s - x) - else return 0 + withLockWait $ + isAnythingCached >>= \case + True -> do + x <- asks cacheMax + s <- runRedis Redis.dbsize + if s > x + then flush (s - x) + else return 0 + False -> return 0 where flush n = do case min 1000 (n `div` 64) of @@ -551,20 +555,22 @@ (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => XPubSpec -> CacheT m ([XPubBal], Bool) -newXPubC xpub = do - bals <- lift $ xPubBals xpub - x <- asks cacheMin - xpubtxt <- xpubText xpub - let n = lenNotNull bals - if x <= n - then do - go bals - return (bals, True) - else do - $(logDebugS) "Cache" $ - "Not caching xpub with " <> cs (show n) <> " used addresses: " <> - xpubtxt - return (bals, False) +newXPubC xpub = + cachePrime >> do + bals <- lift $ xPubBals xpub + x <- asks cacheMin + xpubtxt <- xpubText xpub + let n = lenNotNull bals + if x <= n + then do + go bals + return (bals, True) + else do + $(logDebugS) "Cache" $ + "Not caching xpub with " <> cs (show n) <> + " used addresses: " <> + xpubtxt + return (bals, False) where op XPubUnspent {xPubUnspent = u} = (unspentPoint u, unspentBlock u) go bals = do @@ -593,15 +599,15 @@ newBlockC :: (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => CacheT m () newBlockC = - lift getBestBlock >>= \case - Nothing -> $(logErrorS) "Cache" "Best block not set yet" - Just newhead -> do - cacheGetHead >>= \case - Nothing -> do - $(logInfoS) "Cache" "Cache has no best block set" - importBlockC newhead - newBlockC - Just cachehead -> go newhead cachehead + isAnythingCached >>= \case + False -> return () + True -> + cachePrime >>= \case + Nothing -> return () + Just cachehead -> + lift getBestBlock >>= \case + Nothing -> return () + Just newhead -> go newhead cachehead where go newhead cachehead | cachehead == newhead = syncMempoolC @@ -652,12 +658,17 @@ newTxC :: (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => TxHash -> CacheT m () newTxC th = - lift (getTxData th) >>= \case - Just txd -> do - $(logDebugS) "Cache" $ "Importing transaction: " <> txHashToHex th - importMultiTxC [txd] - Nothing -> - $(logErrorS) "Cache" $ "Transaction not found: " <> txHashToHex th + isAnythingCached >>= \case + True -> + lift (getTxData th) >>= \case + Just txd -> do + $(logDebugS) "Cache" $ + "Importing transaction: " <> txHashToHex th + importMultiTxC [txd] + Nothing -> + $(logErrorS) "Cache" $ + "Transaction not found: " <> txHashToHex th + False -> return () importBlockC :: (MonadUnliftIO m, StoreRead m, MonadLoggerIO m) => BlockHash -> CacheT m () importBlockC bh = @@ -716,17 +727,14 @@ xpubtxt addrs' <- withLockWait $ - getxbals xpubs >>= \xmap -> - if null xmap - then return [] - else do - let addrmap' = faddrmap (HashMap.keysSet xmap) addrmap - runRedis $ do - x <- redisImportMultiTx addrmap' unspentmap txs - y <- redisUpdateBalances addrmap' balmap - z <- redisTouchKeys now (HashMap.keys xmap) - return $ x >> y >> z >> return () - return $ getNewAddrs gap xmap (HashMap.elems addrmap') + getxbals xpubs >>= \xmap -> do + let addrmap' = faddrmap (HashMap.keysSet xmap) addrmap + runRedis $ do + x <- redisImportMultiTx addrmap' unspentmap txs + y <- redisUpdateBalances addrmap' balmap + z <- redisTouchKeys now (HashMap.keys xmap) + return $ x >> y >> z >> return () + return $ getNewAddrs gap xmap (HashMap.elems addrmap') cacheAddAddresses addrs' where faddrmap xmap = HashMap.filter (\a -> addressXPubSpec a `elem` xmap) @@ -795,7 +803,8 @@ b <- case txDataBlock tx of b@MemRef {} -> - redisAddToMempool + redisAddToMempool $ + (: []) BlockTx { blockTxHash = txHash (txData tx) , blockTxBlock = b @@ -905,6 +914,25 @@ cacheGetHead :: MonadLoggerIO m => CacheT m (Maybe BlockHash) cacheGetHead = runRedis redisGetHead +cachePrime :: + (StoreRead m, MonadUnliftIO m, MonadLoggerIO m) + => CacheT m (Maybe BlockHash) +cachePrime = + cacheGetHead >>= \case + Nothing -> do + $(logInfoS) "Cache" "Cache has no best block set" + lift getBestBlock >>= \case + Nothing -> do + $(logInfoS) "Cache" "Best block not set yet" + return Nothing + Just newhead -> do + mem <- lift getMempool + withLockWait . runRedis $ do + a <- redisAddToMempool mem + b <- redisSetHead newhead + return $ a >> b >> return (Just newhead) + Just cachehead -> return $ Just cachehead + cacheSetHead :: (MonadLoggerIO m, StoreRead m) => BlockHash -> CacheT m () cacheSetHead bh = do $(logDebugS) "Cache" $ "Cache head set to: " <> blockHashToHex bh @@ -914,11 +942,13 @@ MonadLoggerIO m => [Address] -> CacheT m [Maybe AddressXPub] cacheGetAddrsInfo as = runRedis (redisGetAddrsInfo as) -redisAddToMempool :: RedisCtx m f => BlockTx -> m (f Integer) -redisAddToMempool btx = +redisAddToMempool :: RedisCtx m f => [BlockTx] -> m (f Integer) +redisAddToMempool btxs = zadd mempoolSetKey - [(blockRefScore (blockTxBlock btx), encode (blockTxHash btx))] + (map (\btx -> + (blockRefScore (blockTxBlock btx), encode (blockTxHash btx))) + btxs) redisRemFromMempool :: RedisCtx m f => TxHash -> m (f Integer) redisRemFromMempool th = zrem mempoolSetKey [encode th] diff --git a/src/Haskoin/Store/Common.hs b/src/Haskoin/Store/Common.hs --- a/src/Haskoin/Store/Common.hs +++ b/src/Haskoin/Store/Common.hs @@ -194,8 +194,6 @@ getBlocksAtHeight :: BlockHeight -> m [BlockHash] getBlock :: BlockHash -> m (Maybe BlockData) getTxData :: TxHash -> m (Maybe TxData) - getOrphanTx :: TxHash -> m (Maybe (UnixTime, Tx)) - getOrphans :: m [(UnixTime, Tx)] getSpenders :: TxHash -> m (IntMap Spender) getSpender :: OutPoint -> m (Maybe Spender) getBalance :: Address -> m Balance @@ -308,8 +306,6 @@ insertAddrUnspent :: Address -> Unspent -> m () deleteAddrUnspent :: Address -> Unspent -> m () setMempool :: [BlockTx] -> m () - insertOrphanTx :: Tx -> UnixTime -> m () - deleteOrphanTx :: TxHash -> m () setBalance :: Balance -> m () insertUnspent :: Unspent -> m () deleteUnspent :: OutPoint -> m () diff --git a/src/Haskoin/Store/Database/Memory.hs b/src/Haskoin/Store/Database/Memory.hs --- a/src/Haskoin/Store/Database/Memory.hs +++ b/src/Haskoin/Store/Database/Memory.hs @@ -5,7 +5,6 @@ , withMemoryDatabase , emptyMemoryDatabase , getMempoolH - , getOrphanTxH , getSpenderH , getSpendersH , getUnspentH @@ -25,15 +24,14 @@ isJust) import Data.Word (Word32) import Haskoin (Address, BlockHash, BlockHeight, - Network, OutPoint (..), Tx, - TxHash, headerHash, txHash) + Network, OutPoint (..), TxHash, + headerHash, txHash) import Haskoin.Store.Common (Balance (..), BlockData (..), BlockRef, BlockTx (..), Limit, Spender, StoreRead (..), StoreWrite (..), TxData (..), - UnixTime, Unspent (..), - applyLimit, scriptToStringAddr, - zeroBalance) + Unspent (..), applyLimit, + scriptToStringAddr, zeroBalance) import Haskoin.Store.Database.Types (BalVal, OutVal (..), UnspentVal, balanceToVal, unspentToVal, valToBalance, valToUnspent) @@ -65,7 +63,6 @@ , hAddrTx :: !(HashMap Address (HashMap BlockRef (HashMap TxHash Bool))) , hAddrOut :: !(HashMap Address (HashMap BlockRef (HashMap OutPoint (Maybe OutVal)))) , hMempool :: !(Maybe [BlockTx]) - , hOrphans :: !(HashMap TxHash (Maybe (UnixTime, Tx))) } deriving (Eq, Show) emptyMemoryDatabase :: MemoryDatabase @@ -81,7 +78,6 @@ , hAddrTx = M.empty , hAddrOut = M.empty , hMempool = Nothing - , hOrphans = M.empty } getBestBlockH :: MemoryDatabase -> Maybe BlockHash @@ -112,12 +108,6 @@ getMempoolH :: MemoryDatabase -> Maybe [BlockTx] getMempoolH = hMempool -getOrphansH :: MemoryDatabase -> [(UnixTime, Tx)] -getOrphansH = catMaybes . M.elems . hOrphans - -getOrphanTxH :: TxHash -> MemoryDatabase -> Maybe (Maybe (UnixTime, Tx)) -getOrphanTxH h = M.lookup h . hOrphans - getAddressesTxsH :: [Address] -> Maybe BlockRef -> Maybe Limit -> MemoryDatabase -> [BlockTx] getAddressesTxsH addrs start limit db = applyLimit limit xs @@ -268,13 +258,6 @@ setMempoolH :: [BlockTx] -> MemoryDatabase -> MemoryDatabase setMempoolH xs db = db {hMempool = Just xs} -insertOrphanTxH :: Tx -> UnixTime -> MemoryDatabase -> MemoryDatabase -insertOrphanTxH tx u db = - db {hOrphans = M.insert (txHash tx) (Just (u, tx)) (hOrphans db)} - -deleteOrphanTxH :: TxHash -> MemoryDatabase -> MemoryDatabase -deleteOrphanTxH h db = db {hOrphans = M.insert h Nothing (hOrphans db)} - getUnspentH :: Network -> OutPoint -> MemoryDatabase -> Maybe (Maybe Unspent) getUnspentH net op db = do m <- M.lookup (outPointHash op) (hUnspent db) @@ -323,9 +306,6 @@ getSpenders t = do v <- R.asks memoryDatabase >>= readTVarIO return . I.map fromJust . I.filter isJust $ getSpendersH t v - getOrphanTx h = do - v <- R.asks memoryDatabase >>= readTVarIO - return . join $ getOrphanTxH h v getUnspent p = do v <- R.asks memoryDatabase >>= readTVarIO net <- R.asks memoryNetwork @@ -343,9 +323,6 @@ v <- R.asks memoryDatabase >>= readTVarIO net <- R.asks memoryNetwork return $ getAddressesUnspentsH net addr start limit v - getOrphans = do - v <- R.asks memoryDatabase >>= readTVarIO - return $ getOrphansH v getAddressTxs addr start limit = do v <- R.asks memoryDatabase >>= readTVarIO return $ getAddressTxsH addr start limit v @@ -391,12 +368,6 @@ setMempool xs = do v <- R.asks memoryDatabase atomically $ modifyTVar v (setMempoolH xs) - insertOrphanTx t u = do - v <- R.asks memoryDatabase - atomically $ modifyTVar v (insertOrphanTxH t u) - deleteOrphanTx h = do - v <- R.asks memoryDatabase - atomically $ modifyTVar v (deleteOrphanTxH h) setBalance b = do v <- R.asks memoryDatabase atomically $ modifyTVar v (setBalanceH b) diff --git a/src/Haskoin/Store/Database/Reader.hs b/src/Haskoin/Store/Database/Reader.hs --- a/src/Haskoin/Store/Database/Reader.hs +++ b/src/Haskoin/Store/Database/Reader.hs @@ -24,22 +24,20 @@ import Database.RocksDB.Query (insert, matching, matchingAsList, matchingSkip, retrieve) import Haskoin (Address, BlockHash, BlockHeight, - Network, OutPoint (..), Tx, - TxHash) + Network, OutPoint (..), TxHash) import Haskoin.Store.Common (Balance, BlockData, BlockRef (..), BlockTx (..), Limit, Spender, StoreRead (..), - TxData, UnixTime, Unspent (..), - applyLimit, applyLimitC, - zeroBalance) + TxData, Unspent (..), applyLimit, + applyLimitC, zeroBalance) import Haskoin.Store.Database.Types (AddrOutKey (..), AddrTxKey (..), BalKey (..), BestKey (..), BlockKey (..), HeightKey (..), MemKey (..), OldMemKey (..), - OrphanKey (..), SpenderKey (..), - TxKey (..), UnspentKey (..), - VersionKey (..), toUnspent, - valToBalance, valToUnspent) + SpenderKey (..), TxKey (..), + UnspentKey (..), VersionKey (..), + toUnspent, valToBalance, + valToUnspent) import UnliftIO (MonadIO, liftIO) type DatabaseReaderT = ReaderT DatabaseReader @@ -150,18 +148,6 @@ where f (t, h) = BlockTx {blockTxBlock = MemRef t, blockTxHash = h} -getOrphansDB :: - MonadIO m - => DatabaseReader - -> m [(UnixTime, Tx)] -getOrphansDB DatabaseReader {databaseReadOptions = opts, databaseHandle = db} = - liftIO . runResourceT . runConduit $ - matching db opts OrphanKeyS .| mapC snd .| sinkList - -getOrphanTxDB :: MonadIO m => TxHash -> DatabaseReader -> m (Maybe (UnixTime, Tx)) -getOrphanTxDB h DatabaseReader {databaseReadOptions = opts, databaseHandle = db} = - retrieve db opts (OrphanKey h) - getAddressesTxsDB :: MonadIO m => [Address] @@ -238,7 +224,6 @@ getTxData t = ask >>= getTxDataDB t getSpender p = ask >>= getSpenderDB p getSpenders t = ask >>= getSpendersDB t - getOrphanTx h = ask >>= getOrphanTxDB h getUnspent a = ask >>= getUnspentDB a getBalance a = ask >>= getBalanceDB a getMempool = ask >>= getMempoolDB @@ -246,8 +231,8 @@ ask >>= getAddressesTxsDB addrs start limit getAddressesUnspents addrs start limit = ask >>= getAddressesUnspentsDB addrs start limit - getOrphans = ask >>= getOrphansDB getAddressUnspents a b c = ask >>= getAddressUnspentsDB a b c getAddressTxs a b c = ask >>= getAddressTxsDB a b c getMaxGap = asks databaseMaxGap getInitialGap = asks databaseInitialGap + diff --git a/src/Haskoin/Store/Database/Types.hs b/src/Haskoin/Store/Database/Types.hs --- a/src/Haskoin/Store/Database/Types.hs +++ b/src/Haskoin/Store/Database/Types.hs @@ -11,7 +11,6 @@ , HeightKey(..) , MemKey(..) , OldMemKey(..) - , OrphanKey(..) , SpenderKey(..) , TxKey(..) , UnspentKey(..) @@ -40,7 +39,7 @@ import Database.RocksDB.Query (Key, KeyValue) import GHC.Generics (Generic) import Haskoin (Address, BlockHash, BlockHeight, - Network, OutPoint (..), Tx, TxHash) + Network, OutPoint (..), TxHash) import Haskoin.Store.Common (Balance (..), BlockData, BlockRef, BlockTx (..), Spender, TxData, UnixTime, Unspent (..), getUnixTime, @@ -240,29 +239,6 @@ instance Key MemKey instance KeyValue MemKey [(UnixTime, TxHash)] - --- | Orphan pool transaction database key. -data OrphanKey - = OrphanKey - { orphanKey :: !TxHash - } - | OrphanKeyS - deriving (Show, Read, Eq, Ord, Generic, Hashable) - -instance Serialize OrphanKey - -- 0x08 ยท TxHash - where - put (OrphanKey h) = do - putWord8 0x08 - put h - -- 0x08 - put OrphanKeyS = putWord8 0x08 - get = do - guard . (== 0x08) =<< getWord8 - OrphanKey <$> get - -instance Key OrphanKey -instance KeyValue OrphanKey (UnixTime, Tx) -- | Block entry database key. newtype BlockKey = BlockKey diff --git a/src/Haskoin/Store/Database/Writer.hs b/src/Haskoin/Store/Database/Writer.hs --- a/src/Haskoin/Store/Database/Writer.hs +++ b/src/Haskoin/Store/Database/Writer.hs @@ -23,28 +23,27 @@ import Database.RocksDB (BatchOp) import Database.RocksDB.Query (deleteOp, insertOp, writeBatch) import Haskoin (Address, BlockHash, BlockHeight, - OutPoint (..), Tx, TxHash) + OutPoint (..), TxHash) import Haskoin.Store.Common (Balance, BlockData, BlockRef (..), BlockTx (..), Spender, StoreRead (..), StoreWrite (..), TxData, - UnixTime, Unspent, nullBalance, + Unspent, nullBalance, zeroBalance) import Haskoin.Store.Database.Memory (MemoryDatabase (..), MemoryState (..), emptyMemoryDatabase, - getMempoolH, getOrphanTxH, - getSpenderH, getSpendersH, - getUnspentH, withMemoryDatabase) + getMempoolH, getSpenderH, + getSpendersH, getUnspentH, + withMemoryDatabase) import Haskoin.Store.Database.Reader (DatabaseReader (..), withDatabaseReader) import Haskoin.Store.Database.Types (AddrOutKey (..), AddrTxKey (..), BalKey (..), BalVal (..), BestKey (..), BlockKey (..), HeightKey (..), MemKey (..), - OrphanKey (..), OutVal, - SpenderKey (..), TxKey (..), - UnspentKey (..), + OutVal, SpenderKey (..), + TxKey (..), UnspentKey (..), UnspentVal (..)) import UnliftIO (MonadIO, newTVarIO, readTVarIO) @@ -91,7 +90,6 @@ addrTxOps (hAddrTx db) <> addrOutOps (hAddrOut db) <> maybeToList (mempoolOp <$> hMempool db) <> - orphanOps (hOrphans db) <> unspentOps (hUnspent db) bestBlockOp :: Maybe BlockHash -> [BatchOp] @@ -172,12 +170,6 @@ insertOp MemKey . map (\BlockTx {blockTxBlock = MemRef t, blockTxHash = h} -> (t, h)) -orphanOps :: HashMap TxHash (Maybe (UnixTime, Tx)) -> [BatchOp] -orphanOps = map (uncurry f) . M.toList - where - f h (Just x) = insertOp (OrphanKey h) x - f h Nothing = deleteOp (OrphanKey h) - unspentOps :: HashMap TxHash (IntMap (Maybe UnspentVal)) -> [BatchOp] unspentOps = concatMap (uncurry f) . M.toList where @@ -228,14 +220,6 @@ setMempoolI :: MonadIO m => [BlockTx] -> DatabaseWriter -> m () setMempoolI xs DatabaseWriter {databaseWriterState = hm} = withMemoryDatabase hm $ setMempool xs -insertOrphanTxI :: MonadIO m => Tx -> UnixTime -> DatabaseWriter -> m () -insertOrphanTxI t p DatabaseWriter {databaseWriterState = hm} = - withMemoryDatabase hm $ insertOrphanTx t p - -deleteOrphanTxI :: MonadIO m => TxHash -> DatabaseWriter -> m () -deleteOrphanTxI t DatabaseWriter {databaseWriterState = hm} = - withMemoryDatabase hm $ deleteOrphanTx t - getBestBlockI :: MonadIO m => DatabaseWriter -> m (Maybe BlockHash) getBestBlockI DatabaseWriter {databaseWriterState = hm, databaseWriterReader = db} = runMaybeT $ MaybeT f <|> MaybeT g @@ -264,13 +248,6 @@ f = withMemoryDatabase hm $ getTxData th g = withDatabaseReader db $ getTxData th -getOrphanTxI :: MonadIO m => TxHash -> DatabaseWriter -> m (Maybe (UnixTime, Tx)) -getOrphanTxI h DatabaseWriter {databaseWriterReader = db, databaseWriterState = hm} = - fmap join . runMaybeT $ MaybeT f <|> MaybeT g - where - f = getOrphanTxH h <$> readTVarIO (memoryDatabase hm) - g = Just <$> withDatabaseReader db (getOrphanTx h) - getSpenderI :: MonadIO m => OutPoint -> DatabaseWriter -> m (Maybe Spender) getSpenderI op DatabaseWriter {databaseWriterReader = db, databaseWriterState = hm} = fmap join . runMaybeT $ MaybeT f <|> MaybeT g @@ -344,13 +321,11 @@ getTxData t = R.ask >>= getTxDataI t getSpender p = R.ask >>= getSpenderI p getSpenders t = R.ask >>= getSpendersI t - getOrphanTx h = R.ask >>= getOrphanTxI h getUnspent a = R.ask >>= getUnspentI a getBalance a = R.ask >>= getBalanceI a getMempool = R.ask >>= getMempoolI getAddressesTxs = undefined getAddressesUnspents = undefined - getOrphans = undefined getMaxGap = R.asks (databaseMaxGap . databaseWriterReader) instance MonadIO m => StoreWrite (ReaderT DatabaseWriter m) where @@ -365,8 +340,6 @@ insertAddrUnspent a u = R.ask >>= insertAddrUnspentI a u deleteAddrUnspent a u = R.ask >>= deleteAddrUnspentI a u setMempool xs = R.ask >>= setMempoolI xs - insertOrphanTx t p = R.ask >>= insertOrphanTxI t p - deleteOrphanTx t = R.ask >>= deleteOrphanTxI t insertUnspent u = R.ask >>= insertUnspentI u deleteUnspent p = R.ask >>= deleteUnspentI p setBalance b = R.ask >>= setBalanceI b 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 @@ -4,11 +4,9 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TemplateHaskell #-} module Haskoin.Store.Logic - ( ImportException + ( ImportException (..) , initBest - , getOldOrphans , getOldMempool - , importOrphan , revertBlock , importBlock , newMempoolTx @@ -18,8 +16,8 @@ import Control.Monad (forM, forM_, unless, void, when, zipWithM_) import Control.Monad.Except (MonadError (..)) -import Control.Monad.Logger (MonadLogger, logErrorS, - logWarnS) +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) @@ -58,6 +56,7 @@ data ImportException = PrevBlockNotBest !Text + | TxOrphan | UnconfirmedCoinbase !Text | BestBlockUnknown | BestBlockNotFound !Text @@ -92,39 +91,24 @@ (isNothing m) (void (importBlock (genesisBlock net) (genesisNode net))) -getOldOrphans :: StoreRead m => UnixTime -> m [TxHash] -getOldOrphans now = - map (txHash . snd) . filter ((< now - 600) . fst) <$> getOrphans - getOldMempool :: StoreRead m => UnixTime -> m [TxHash] getOldMempool now = map blockTxHash . filter ((< now - 3600 * 72) . memRefTime . blockTxBlock) <$> getMempool -importOrphan :: - ( StoreRead m - , StoreWrite m - , MonadLogger m - , MonadError ImportException m - ) - => UnixTime - -> Tx - -> m (Maybe [TxHash]) -- ^ deleted transactions or nothing if import failed -importOrphan t tx = deleteOrphanTx (txHash tx) >> newMempoolTx tx t - newMempoolTx :: - ( StoreRead m - , StoreWrite m - , MonadLogger m - , MonadError ImportException m - ) + (StoreRead m, StoreWrite m, MonadLogger m, MonadError ImportException m) => Tx -> UnixTime - -> m (Maybe [TxHash]) -- ^ deleted transactions or nothing if import failed + -> m (Maybe [TxHash]) + -- ^ deleted transactions or nothing if already imported newMempoolTx tx w = getTxData (txHash tx) >>= \case Just x - | not (txDataDeleted x) -> return Nothing + | not (txDataDeleted x) -> do + $(logDebugS) "BlockStore" $ + "Transaction already in store: " <> txHashToHex (txHash tx) + return Nothing _ -> go where go = do @@ -135,8 +119,7 @@ then do $(logWarnS) "BlockStore" $ "Orphan tx: " <> txHashToHex (txHash tx) - insertOrphanTx tx w - return Nothing + throwError TxOrphan else f f = do us <- @@ -144,14 +127,8 @@ 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 fmap Just (importTx (MemRef w) w tx) `catchError` h + then fmap Just (importTx (MemRef w) w tx) else g ds g ds = do net <- getNetwork @@ -297,6 +274,9 @@ -> Tx -> m [TxHash] -- ^ deleted transactions importTx br tt tx = do + unless (confirmed br) $ + $(logDebugS) "BlockStore" $ + "Importing transaction " <> txHashToHex (txHash tx) when (length (nub (map prevOutput (txIn tx))) < length (txIn tx)) $ do $(logErrorS) "BlockStore" $ "Transaction spends same output twice: " <> txHashToHex (txHash tx) @@ -364,6 +344,10 @@ fromString (show (outPointIndex op)) throwError (NoUnspent (cs (show op))) Just Spender {spenderHash = s} -> do + $(logWarnS) "BlockStore" $ + "Deleting transaction " <> txHashToHex s <> + " because it conflicts with " <> + txHashToHex (txHash tx) ths <- deleteTx True s getUnspent op >>= \case Nothing -> do @@ -513,11 +497,13 @@ deleteTx mo h = do getTxData h >>= \case Nothing -> do - $(logErrorS) "BlockStore" $ "Cannot find tx to delete: " <> txHashToHex h + $(logErrorS) "BlockStore" $ + "Cannot find tx to delete: " <> txHashToHex h throwError (TxNotFound (txHashToHex h)) Just t | txDataDeleted t -> do - $(logWarnS) "BlockStore" $ "Already deleted tx: " <> txHashToHex h + $(logWarnS) "BlockStore" $ + "Already deleted tx: " <> txHashToHex h return [] | mo && confirmed (txDataBlock t) -> do $(logErrorS) "BlockStore" $ @@ -526,8 +512,16 @@ | otherwise -> go t where go t = do + $(logWarnS) "BlockStore" $ "Deleting transaction: " <> txHashToHex h ss <- nub . map spenderHash . I.elems <$> getSpenders h - ths <- concat <$> mapM (deleteTx True) ss + ths <- + fmap concat $ + forM ss $ \s -> do + $(logWarnS) "BlockStore" $ + "Deleting descendant " <> txHashToHex s <> + " to delete parent " <> + txHashToHex h + deleteTx True s forM_ (take (length (txOut (txData t))) [0 ..]) $ \n -> delOutput (OutPoint h n) let ps = filter (/= nullOutPoint) (map prevOutput (txIn (txData t))) diff --git a/src/Haskoin/Store/Web.hs b/src/Haskoin/Store/Web.hs --- a/src/Haskoin/Store/Web.hs +++ b/src/Haskoin/Store/Web.hs @@ -236,7 +236,6 @@ getTxData th = runInWebReader (getTxData th) (getTxData th) getSpender op = runInWebReader (getSpender op) (getSpender op) getSpenders th = runInWebReader (getSpenders th) (getSpenders th) - getOrphanTx th = runInWebReader (getOrphanTx th) (getOrphanTx th) getUnspent op = runInWebReader (getUnspent op) (getUnspent op) getBalance a = runInWebReader (getBalance a) (getBalance a) getBalances as = runInWebReader (getBalances as) (getBalances as) @@ -249,7 +248,6 @@ runInWebReader (getAddressesUnspents addrs start limit) (getAddressesUnspents addrs start limit) - getOrphans = runInWebReader getOrphans getOrphans xPubBals xpub = runInWebReader (xPubBals xpub) (xPubBals xpub) xPubSummary xpub = runInWebReader (xPubSummary xpub) (xPubSummary xpub) xPubUnspents xpub start offset limit = @@ -269,7 +267,6 @@ getTxData = lift . getTxData getSpender = lift . getSpender getSpenders = lift . getSpenders - getOrphanTx = lift . getOrphanTx getUnspent = lift . getUnspent getBalance = lift . getBalance getBalances = lift . getBalances @@ -277,7 +274,6 @@ getAddressesTxs addrs start limit = lift (getAddressesTxs addrs start limit) getAddressesUnspents addrs start limit = lift (getAddressesUnspents addrs start limit) - getOrphans = lift getOrphans xPubBals = lift . xPubBals xPubSummary = lift . xPubSummary xPubUnspents xpub start offset limit =