haskoin-store 0.23.9 → 0.23.10
raw patch · 11 files changed
+298/−329 lines, 11 files
Files
- CHANGELOG.md +13/−1
- haskoin-store.cabal +2/−2
- src/Haskoin/Store/BlockStore.hs +148/−112
- src/Haskoin/Store/Cache.hs +82/−52
- src/Haskoin/Store/Common.hs +0/−4
- src/Haskoin/Store/Database/Memory.hs +4/−33
- src/Haskoin/Store/Database/Reader.hs +8/−23
- src/Haskoin/Store/Database/Types.hs +1/−25
- src/Haskoin/Store/Database/Writer.hs +7/−34
- src/Haskoin/Store/Logic.hs +33/−39
- src/Haskoin/Store/Web.hs +0/−4
CHANGELOG.md view
@@ -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.
haskoin-store.cabal view
@@ -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.
src/Haskoin/Store/BlockStore.hs view
@@ -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 ())
src/Haskoin/Store/Cache.hs view
@@ -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]
src/Haskoin/Store/Common.hs view
@@ -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 ()
src/Haskoin/Store/Database/Memory.hs view
@@ -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)
src/Haskoin/Store/Database/Reader.hs view
@@ -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+
src/Haskoin/Store/Database/Types.hs view
@@ -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
src/Haskoin/Store/Database/Writer.hs view
@@ -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
src/Haskoin/Store/Logic.hs view
@@ -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)))
src/Haskoin/Store/Web.hs view
@@ -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 =