haskoin-store 0.37.0 → 0.37.1
raw patch · 5 files changed
+257/−315 lines, 5 filesdep ~haskoin-storedep ~haskoin-store-data
Dependency ranges changed: haskoin-store, haskoin-store-data
Files
- haskoin-store.cabal +6/−6
- src/Haskoin/Store/Cache.hs +228/−294
- src/Haskoin/Store/Common.hs +10/−0
- src/Haskoin/Store/Database/Reader.hs +1/−1
- src/Haskoin/Store/Manager.hs +12/−14
haskoin-store.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 233badef6c274ea380714d9bc0492c96784d8cceb6012c6f476caaf9f45a8352+-- hash: a3e1d274d97d15841ac32d9f355a41cae196b058822e0d107e4c5ddd928e27b3 name: haskoin-store-version: 0.37.0+version: 0.37.1 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at <https://github.com/haskoin/haskoin-store#readme> category: Bitcoin, Finance, Network@@ -53,7 +53,7 @@ , hashable >=1.3.0.0 , haskoin-core >=0.13.6 , haskoin-node >=0.14.1- , haskoin-store-data ==0.37.0+ , haskoin-store-data ==0.37.1 , hedis >=0.12.13 , http-types >=0.12.3 , monad-logger >=0.3.32@@ -96,7 +96,7 @@ , haskoin-core >=0.13.6 , haskoin-node >=0.14.1 , haskoin-store- , haskoin-store-data ==0.37.0+ , haskoin-store-data ==0.37.1 , monad-logger >=0.3.32 , mtl >=2.2.2 , nqe >=0.6.1@@ -135,8 +135,8 @@ , hashable >=1.3.0.0 , haskoin-core >=0.13.6 , haskoin-node >=0.14.1- , haskoin-store ==0.37.0- , haskoin-store-data ==0.37.0+ , haskoin-store ==0.37.1+ , haskoin-store-data ==0.37.1 , hedis >=0.12.13 , hspec >=2.7.1 , http-types >=0.12.3
src/Haskoin/Store/Cache.hs view
@@ -18,7 +18,6 @@ , CacheWriter , CacheWriterInbox , cacheNewBlock- , cacheNewTx , cachePing , cacheWriter , isInCache@@ -29,7 +28,7 @@ import Control.Monad (forM, forM_, forever, unless, void, when) import Control.Monad.Logger (MonadLoggerIO, logDebugS, logErrorS,- logInfoS, logWarnS)+ logWarnS) import Control.Monad.Reader (ReaderT (..), ask, asks) import Control.Monad.Trans (lift) import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)@@ -37,14 +36,14 @@ import Data.ByteString (ByteString) import qualified Data.ByteString.Short as BSS import Data.Default (def)-import Data.Either (lefts, rights)+import Data.Either (rights) import Data.HashMap.Strict (HashMap) import qualified Data.HashMap.Strict as HashMap import qualified Data.HashSet as HashSet import qualified Data.IntMap.Strict as I import Data.List (sort) import qualified Data.Map.Strict as Map-import Data.Maybe (catMaybes, isNothing, mapMaybe)+import Data.Maybe (catMaybes, mapMaybe) import Data.Serialize (Serialize, decode, encode) import Data.String.Conversions (cs) import Data.Text (Text)@@ -65,7 +64,7 @@ blockHashToHex, derivePubPath, eitherToMaybe, headerHash, pathToList, scriptToAddressBS,- txHash, txHashToHex, xPubAddr,+ txHash, xPubAddr, xPubCompatWitnessAddr, xPubExport, xPubWitnessAddr) import Haskoin.Node (Chain, chainBlockMain,@@ -86,11 +85,11 @@ runRedis :: MonadLoggerIO m => Redis (Either Reply a) -> CacheX m a runRedis action = asks cacheConn >>= \conn ->- liftIO (Redis.runRedis conn action) >>= \case- Right x -> return x- Left e -> do- $(logErrorS) "Cache" $ "Got error from Redis: " <> cs (show e)- throwIO (RedisError e)+ liftIO (Redis.runRedis conn action) >>= \case+ Right x -> return x+ Left e -> do+ $(logErrorS) "Cache" $ "Got error from Redis: " <> cs (show e)+ throwIO (RedisError e) data CacheConfig = CacheConfig { cacheConn :: !Connection@@ -410,7 +409,6 @@ data CacheWriterMessage = CacheNewBlock- | CacheNewTx !TxHash | CachePing type CacheWriterInbox = Inbox CacheWriterMessage@@ -434,9 +432,6 @@ maxKey :: ByteString maxKey = "max" -isAnythingCached :: MonadLoggerIO m => CacheX m Bool-isAnythingCached = runRedis $ Redis.exists maxKey- xPubAddrFunction :: DeriveType -> XPubKey -> Address xPubAddrFunction DeriveNormal = xPubAddr xPubAddrFunction DeriveP2SH = xPubCompatWitnessAddr@@ -480,7 +475,7 @@ liftIO . Redis.runRedis conn $ do let opts = Redis.SetOpts- { Redis.setSeconds = Just 30+ { Redis.setSeconds = Just 300 , Redis.setMilliseconds = Nothing , Redis.setCondition = Just Redis.Nx }@@ -497,15 +492,15 @@ ": not locked" Just bs -> if read (cs bs) == i- then do- void $ runRedis (Redis.del ["lock"])- $(logDebugS) "Cache" $- "Released lock with value " <>- cs (show i)- else- $(logErrorS) "Cache" $- "Could not release lock: value is not " <>- cs (show i)+ then do+ void $ runRedis (Redis.del ["lock"])+ $(logDebugS) "Cache" $+ "Released lock with value " <>+ cs (show i)+ else+ $(logErrorS) "Cache" $+ "Could not release lock: value is not " <>+ cs (show i) withLock :: (MonadLoggerIO m, MonadUnliftIO m)@@ -518,30 +513,24 @@ pruneDB :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadBase m) => CacheX m Integer-pruneDB =- isAnythingCached >>= \case- True -> do- x <- asks cacheMax- s <- runRedis Redis.dbsize- if s > x- then flush (s - x)- else return 0- False -> return 0+pruneDB = do+ x <- asks cacheMax+ s <- runRedis Redis.dbsize+ if s > x then flush (s - x) else return 0 where flush n = case min 1000 (n `div` 64) of- 0 -> return 0- x -> do- m <- withLock $ do- ks <-- fmap (map fst) . runRedis $- getFromSortedSet maxKey Nothing 0 (fromIntegral x)- $(logDebugS) "Cache" $- "Pruning " <> cs (show (length ks)) <> " old xpubs"- delXPubKeys ks- case m of- Nothing -> return 0- Just y -> return y+ 0 -> return 0+ x -> do+ m <- withLock $ do+ ks <- fmap (map fst) . runRedis $+ getFromSortedSet maxKey Nothing 0 (fromIntegral x)+ $(logDebugS) "Cache" $+ "Pruning " <> cs (show (length ks)) <> " old xpubs"+ delXPubKeys ks+ case m of+ Nothing -> return 0+ Just y -> return y touchKeys :: MonadLoggerIO m => [XPubSpec] -> CacheX m () touchKeys xpubs = do@@ -556,9 +545,13 @@ cacheWriterReact :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheWriterMessage -> CacheX m ()-cacheWriterReact CacheNewBlock = newBlockC-cacheWriterReact (CacheNewTx th) = newTxC th-cacheWriterReact CachePing = pruneDB >> newBlockC >> syncMempoolC+cacheWriterReact CacheNewBlock = do+ newBlockC+ syncMempoolC+cacheWriterReact CachePing = do+ pruneDB+ newBlockC+ syncMempoolC lenNotNull :: [XPubBal] -> Int lenNotNull bals = length $ filter (not . nullBalance . xPubBal) bals@@ -566,21 +559,16 @@ newXPubC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => XPubSpec -> CacheX m (Bool, [XPubBal])-newXPubC xpub =- cachePrime >>= \case- Nothing -> (False,) <$> lift (xPubBals xpub)- Just _ -> do- bals <- lift $ xPubBals xpub- x <- asks cacheMin- t <- xpubText xpub- let n = lenNotNull bals- if x <= n- then go bals- else do- $(logDebugS) "Cache" $- "Not caching xpub with " <> cs (show n) <>- " used addresses: " <> t- return (False, bals)+newXPubC xpub = do+ bals <- lift $ xPubBals xpub+ x <- asks cacheMin+ t <- xpubText xpub+ let n = lenNotNull bals+ if x <= n+ then go bals+ else do+ $(logDebugS) "Cache" $ "Not caching xpub: " <> t+ return (False, bals) where op XPubUnspent {xPubUnspent = u} = (unspentPoint u, unspentBlock u) go bals = do@@ -611,102 +599,73 @@ $(logDebugS) "Cache" $ "Done caching xpub: " <> xpubtxt return (True, bals) -newTxC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m)- => TxHash -> CacheX m ()-newTxC th = do- m <- withLock $ isAnythingCached >>= \case- False ->- $(logDebugS) "Cache" $- "Not importing tx " <> txHashToHex th <> " because cache is empty"- True ->- runRedis (Redis.zscore mempoolSetKey (encode th)) >>= \case- Nothing ->- lift (getTxData th) >>= \case- Just td -> do- importMultiTxC [td]- $(logDebugS) "Cache" $- "Updated cache with tx: " <> txHashToHex th- Nothing ->- $(logErrorS) "Cache" $- "Tx not found in db: " <> txHashToHex th- Just _ ->- $(logDebugS) "Cache" $- "Tx already in cache: " <> txHashToHex th- when (isNothing m) $- $(logErrorS) "Cache" $- "Could not get lock to cache tx: " <> txHashToHex th- newBlockC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m () newBlockC =- withLock f >>= \m ->- when (isNothing m) $- $(logErrorS) "Cache" "Could not get lock to add add block to cache"+ lift getBestBlock >>= \m ->+ forM_ m $ \bb ->+ in_sync bb >>= \s ->+ when s $ void $ withLock $ f bb where- f = isAnythingCached >>= \case+ f bb = cacheGetHead >>= go bb+ in_sync bb =+ asks cacheChain >>= \ch ->+ chainGetBest ch >>= \cb ->+ return $ headerHash (nodeHeader cb) == bb+ go bb Nothing =+ importBlockC bb+ go bb (Just cb)+ | cb == bb =+ $(logDebugS) "Cache" "Cache in sync"+ | otherwise =+ sync bb cb+ sync bb cb =+ asks cacheChain >>= \ch ->+ chainBlockMain bb ch >>= \case False ->- $(logDebugS) "Cache" "Not syncing blocks because cache is empty"- True -> cachePrime >>= \case- Nothing -> return ()- Just cachehead -> lift getBestBlock >>= \case- Nothing -> return ()- Just newhead -> go newhead cachehead- go newhead cachehead- | cachehead == newhead =- $(logDebugS) "Cache" $- "Blocks in sync: " <> blockHashToHex cachehead- | otherwise = do- ch <- asks cacheChain- chainBlockMain newhead ch >>= \case- False ->- $(logErrorS) "Cache" $- "New head not in main chain: " <> blockHashToHex newhead- True -> chainGetBlock cachehead ch >>= \case- Nothing ->- $(logErrorS) "Cache" $- "Cache head block node not found: " <>- blockHashToHex cachehead- Just cacheheadnode -> chainBlockMain cachehead ch >>= \case- False -> do- $(logDebugS) "Cache" $- "Reverting cache head not in main chain: " <>- blockHashToHex cachehead- removeHeadC >> f- True -> chainGetBlock newhead ch >>= \case- Nothing -> do- $(logErrorS) "Cache" $- "Cache head node not found: " <>- blockHashToHex newhead- throwIO $- LogicError $- "Cache head node not found: " <>- cs (blockHashToHex newhead)- Just newheadnode ->- next newheadnode cacheheadnode- next newheadnode cacheheadnode = do- ch <- asks cacheChain- chainGetAncestor (nodeHeight cacheheadnode + 1) newheadnode ch >>= \case+ $(logErrorS) "Cache" $+ "New head not in main chain: " <> blockHashToHex bb+ True ->+ chainGetBlock cb ch >>= \case Nothing ->- $(logWarnS) "Cache" $- "Ancestor not found at height " <>- cs (show (nodeHeight cacheheadnode + 1)) <>- " for block: " <>- blockHashToHex (headerHash (nodeHeader newheadnode))- Just newcachenode ->- importBlockC (headerHash (nodeHeader newcachenode)) >> f+ $(logErrorS) "Cache" $+ "Cache head block node not found: " <>+ blockHashToHex cb+ Just cn ->+ chainBlockMain cb ch >>= \case+ False -> do+ $(logDebugS) "Cache" $+ "Reverting cache head not in main chain: " <>+ blockHashToHex cb+ removeHeadC >> f bb+ True ->+ chainGetBlock bb ch >>= \case+ Just nn -> next bb nn cn+ Nothing -> do+ $(logErrorS) "Cache" $+ "Cache head node not found: "+ <> blockHashToHex bb+ throwIO $+ LogicError $+ "Cache head node not found: "+ <> cs (blockHashToHex bb)+ next bb nn cn =+ asks cacheChain >>= \ch ->+ chainGetAncestor (nodeHeight cn + 1) nn ch >>= \case+ Nothing ->+ $(logWarnS) "Cache" $+ "Ancestor not found at height "+ <> cs (show (nodeHeight cn + 1))+ <> " for block: "+ <> blockHashToHex (headerHash (nodeHeader nn))+ Just cn' ->+ importBlockC (headerHash (nodeHeader cn')) >> f bb importBlockC :: (MonadUnliftIO m, StoreReadExtra m, MonadLoggerIO m) => BlockHash -> CacheX m () importBlockC bh = lift (getBlock bh) >>= \case- Nothing -> do- $(logErrorS) "Cache" $ "Could not get block: " <> blockHashToHex bh- throwIO . LogicError . cs $- "Could not get block: " <> blockHashToHex bh- Just bd ->- go bd- where- go bd = do+ Just bd -> do let ths = blockDataTxs bd tds <- sortTxData . catMaybes <$> mapM (lift . getTxData) ths $(logDebugS) "Cache" $@@ -719,23 +678,31 @@ " transactions from block " <> blockHashToHex bh cacheSetHead bh+ Nothing -> do+ $(logErrorS) "Cache" $+ "Could not get block: "+ <> blockHashToHex bh+ throwIO . LogicError . cs $+ "Could not get block: "+ <> blockHashToHex bh removeHeadC :: (StoreReadExtra m, MonadUnliftIO m, MonadLoggerIO m) => CacheX m () removeHeadC = void . runMaybeT $ do- bh <- MaybeT cacheGetHead- bd <- MaybeT (lift (getBlock bh))- lift $ do- tds <-- sortTxData . catMaybes <$>- mapM (lift . getTxData) (blockDataTxs bd)- $(logDebugS) "Cache" $ "Reverting head: " <> blockHashToHex bh- importMultiTxC tds- $(logWarnS) "Cache" $- "Reverted block head " <> blockHashToHex bh <> " to parent " <>- blockHashToHex (prevBlock (blockDataHeader bd))- cacheSetHead (prevBlock (blockDataHeader bd))+ bh <- MaybeT cacheGetHead+ bd <- MaybeT (lift (getBlock bh))+ lift $ do+ tds <- sortTxData . catMaybes <$>+ mapM (lift . getTxData) (blockDataTxs bd)+ $(logDebugS) "Cache" $ "Reverting head: " <> blockHashToHex bh+ importMultiTxC tds+ $(logWarnS) "Cache" $+ "Reverted block head "+ <> blockHashToHex bh+ <> " to parent "+ <> blockHashToHex (prevBlock (blockDataHeader bd))+ cacheSetHead (prevBlock (blockDataHeader bd)) importMultiTxC :: (MonadUnliftIO m, StoreReadExtra m, MonadLoggerIO m)@@ -743,16 +710,20 @@ importMultiTxC txs = do $(logDebugS) "Cache" $ "Processing " <> cs (show (length txs)) <> " txs" $(logDebugS) "Cache" $- "Getting address information for " <> cs (show (length alladdrs)) <>- " addresses"+ "Getting address information for "+ <> cs (show (length alladdrs))+ <> " addresses" addrmap <- getaddrmap let addrs = HashMap.keys addrmap $(logDebugS) "Cache" $- "Getting balances for " <> cs (show (HashMap.size addrmap)) <>- " addresses"+ "Getting balances for "+ <> cs (show (HashMap.size addrmap))+ <> " addresses" balmap <- getbalances addrs $(logDebugS) "Cache" $- "Getting unspent data for " <> cs (show (length allops)) <> " outputs"+ "Getting unspent data for "+ <> cs (show (length allops))+ <> " outputs" unspentmap <- getunspents gap <- lift getMaxGap now <- systemSeconds <$> liftIO getSystemTime@@ -760,13 +731,13 @@ forM_ (zip [(1 :: Int) ..] xpubs) $ \(i, xpub) -> do xpubtxt <- xpubText xpub $(logDebugS) "Cache" $- "Affected xpub " <> cs (show i) <> "/" <> cs (show (length xpubs)) <>- ": " <>- xpubtxt+ "Affected xpub "+ <> cs (show i) <> "/" <> cs (show (length xpubs))+ <> ": " <> xpubtxt addrs' <- do $(logDebugS) "Cache" $- "Getting xpub balances for " <> cs (show (length xpubs)) <>- " xpubs"+ "Getting xpub balances for "+ <> cs (show (length xpubs)) <> " xpubs" xmap <- getxbals xpubs let addrmap' = faddrmap (HashMap.keysSet xmap) addrmap $(logDebugS) "Cache" "Starting Redis import pipeline"@@ -790,10 +761,9 @@ getbalances addrs = HashMap.fromList . zip addrs <$> mapM (lift . getDefaultBalance) addrs getxbals xpubs = do- bals <-- runRedis . fmap sequence . forM xpubs $ \xpub -> do- bs <- redisGetXPubBalances xpub- return $ (xpub, ) <$> bs+ bals <- runRedis . fmap sequence . forM xpubs $ \xpub -> do+ bs <- redisGetXPubBalances xpub+ return $ (xpub, ) <$> bs return $ HashMap.filter (not . null) (HashMap.fromList bals) allops = map snd $ concatMap txInputs txs <> concatMap txOutputs txs alladdrs =@@ -815,49 +785,43 @@ where uns p i = case HashMap.lookup p unspentmap of- Just u ->- redisAddXPubUnspents (addressXPubSpec i) [(p, unspentBlock u)]- Nothing -> redisRemXPubUnspents (addressXPubSpec i) [p]+ Just u ->+ redisAddXPubUnspents (addressXPubSpec i) [(p, unspentBlock u)]+ Nothing -> redisRemXPubUnspents (addressXPubSpec i) [p] addtx tx a p = case HashMap.lookup a addrmap of- Just i -> do- x <-- redisAddXPubTxs- (addressXPubSpec i)- [ TxRef- { txRefHash = txHash (txData tx)- , txRefBlock = txDataBlock tx- }- ]- y <- uns p i- return $ x >> y >> return ()- Nothing -> return (pure ())+ Just i -> do+ let tr = TxRef { txRefHash = txHash (txData tx)+ , txRefBlock = txDataBlock tx+ }+ x <- redisAddXPubTxs (addressXPubSpec i) [tr]+ y <- uns p i+ return $ x >> y >> return ()+ Nothing -> return (pure ()) remtx tx a p = case HashMap.lookup a addrmap of- Just i -> do- x <- redisRemXPubTxs (addressXPubSpec i) [txHash (txData tx)]- y <- uns p i- return $ x >> y >> return ()- Nothing -> return (pure ())+ Just i -> do+ x <- redisRemXPubTxs (addressXPubSpec i) [txHash (txData tx)]+ y <- uns p i+ return $ x >> y >> return ()+ Nothing -> return (pure ()) importtxentries tx = if txDataDeleted tx- then do- x <- mapM (uncurry (remtx tx)) (txaddrops tx)- y <- redisRemFromMempool [txHash (txData tx)]- return $ sequence_ x >> void y- else do- a <- sequence <$> mapM (uncurry (addtx tx)) (txaddrops tx)- b <-- case txDataBlock tx of- b@MemRef {} ->- redisAddToMempool $- (: [])- TxRef- { txRefHash = txHash (txData tx)- , txRefBlock = b- }- _ -> redisRemFromMempool [txHash (txData tx)]- return $ a >> b >> return ()+ then do+ x <- mapM (uncurry (remtx tx)) (txaddrops tx)+ y <- redisRemFromMempool [txHash (txData tx)]+ return $ sequence_ x >> void y+ else do+ a <- sequence <$> mapM (uncurry (addtx tx)) (txaddrops tx)+ b <-+ case txDataBlock tx of+ b@MemRef {} ->+ let tr = TxRef { txRefHash = txHash (txData tx)+ , txRefBlock = b+ }+ in redisAddToMempool [tr]+ _ -> redisRemFromMempool [txHash (txData tx)]+ return $ a >> b >> return () txaddrops td = txInputs td <> txOutputs td redisUpdateBalances ::@@ -867,10 +831,10 @@ -> m (f ()) redisUpdateBalances addrmap balmap = fmap (fmap mconcat . sequence) . forM (HashMap.keys addrmap) $ \a ->- case (HashMap.lookup a addrmap, HashMap.lookup a balmap) of- (Just ainfo, Just bal) ->- redisAddXPubBalances (addressXPubSpec ainfo) [xpubbal ainfo bal]- _ -> return (pure ())+ case (HashMap.lookup a addrmap, HashMap.lookup a balmap) of+ (Just ainfo, Just bal) ->+ redisAddXPubBalances (addressXPubSpec ainfo) [xpubbal ainfo bal]+ _ -> return (pure ()) where xpubbal ainfo bal = XPubBal {xPubBalPath = addressXPubPath ainfo, xPubBal = bal}@@ -896,7 +860,10 @@ c <- forM (HashMap.toList txmap) (uncurry redisAddXPubTxs) return $ sequence_ a >> sequence_ b >> sequence_ c $(logDebugS) "Cache" "Completed Redis pipeline"- let xpubs = HashSet.toList . HashSet.fromList . map addressXPubSpec $ Map.elems amap+ let xpubs = HashSet.toList+ . HashSet.fromList+ . map addressXPubSpec+ $ Map.elems amap $(logDebugS) "Cache" "Getting xpub balances" xmap <- getbals xpubs gap <- lift getMaxGap@@ -904,12 +871,10 @@ addrs' = getNewAddrs gap xmap notnulls cacheAddAddresses addrs' where- getbals xpubs =- runRedis $ do- bs <- sequence <$> forM xpubs redisGetXPubBalances- return $- HashMap.filter (not . null) . HashMap.fromList . zip xpubs <$>- bs+ getbals xpubs = runRedis $ do+ bs <- sequence <$> forM xpubs redisGetXPubBalances+ return $+ HashMap.filter (not . null) . HashMap.fromList . zip xpubs <$> bs amap = Map.fromList addrs getnotnull = let f xpub =@@ -921,57 +886,66 @@ g = filter (not . nullBalance . xPubBal) in concatMap (uncurry f) . HashMap.toList . HashMap.map g getbal a i =- let f b =- ( addressXPubSpec i- , [XPubBal {xPubBal = b, xPubBalPath = addressXPubPath i}])+ let f b = ( addressXPubSpec i+ , [XPubBal {xPubBal = b, xPubBalPath = addressXPubPath i}] ) in f <$> lift (getDefaultBalance a) getutxo a i =- let f us =- ( addressXPubSpec i- , map (\u -> (unspentPoint u, unspentBlock u)) us)+ let f us = ( addressXPubSpec i+ , map (\u -> (unspentPoint u, unspentBlock u)) us ) in f <$> lift (getAddressUnspents a def) gettxmap a i = let f ts = (addressXPubSpec i, ts) in f <$> lift (getAddressTxs a def) -getNewAddrs ::- KeyIndex- -> HashMap XPubSpec [XPubBal]- -> [AddressXPub]- -> [(Address, AddressXPub)]-getNewAddrs gap xpubs = concatMap f+getNewAddrs :: KeyIndex+ -> HashMap XPubSpec [XPubBal]+ -> [AddressXPub]+ -> [(Address, AddressXPub)]+getNewAddrs gap xpubs =+ concatMap $ \a ->+ case HashMap.lookup (addressXPubSpec a) xpubs of+ Nothing -> []+ Just bals -> addrsToAdd gap bals a++cacheCoolKey :: ByteString+cacheCoolKey = "cooldown"++isCool :: (MonadUnliftIO m, MonadLoggerIO m) => CacheX m Bool+isCool =+ runRedis (Redis.get cacheCoolKey) >>= \case+ Nothing -> return True+ Just bs -> do+ let t = read (cs bs)+ now <- microseconds+ return (cooldown <= now - t) where- f a =- case HashMap.lookup (addressXPubSpec a) xpubs of- Nothing -> []- Just bals -> addrsToAdd gap bals a+ cooldown = 250000 +startCooldown :: (MonadUnliftIO m, MonadLoggerIO m) => CacheX m ()+startCooldown = do+ now <- microseconds+ void $ runRedis (Redis.set cacheCoolKey (cs (show now)))+ syncMempoolC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m ()-syncMempoolC = do+syncMempoolC =+ void . withLock $ isCool >>= \cool -> when cool $ do nodepool <- HashSet.fromList . map txRefHash <$> lift getMempool cachepool <- HashSet.fromList . map txRefHash <$> cacheGetMempool- txs <-- mapM getit . HashSet.toList $- mappend- (HashSet.difference nodepool cachepool)- (HashSet.difference cachepool nodepool)+ txs <- mapM getit . HashSet.toList $+ mappend+ (HashSet.difference nodepool cachepool)+ (HashSet.difference cachepool nodepool) unless (null txs) $ do- $(logDebugS) "Cache" $- "Importing " <> cs (show (length txs))- <> " mempool transactions"+ $(logDebugS) "Cache" "Importing mempool transactions" importMultiTxC (rights txs)- forM_ (zip [(1 :: Int) ..] (lefts txs)) $ \(i, h) ->- $(logDebugS) "Cache" $- "Ignoring cache mempool tx " <> cs (show i) <> "/"- <> cs (show (length txs)) <> ": "- <> txHashToHex h+ startCooldown where getit th = lift (getTxData th) >>= \case- Nothing -> return (Left th)- Just tx -> return (Right tx)+ Nothing -> return (Left th)+ Just tx -> return (Right tx) cacheGetMempool :: MonadLoggerIO m => CacheX m [TxRef] cacheGetMempool = runRedis redisGetMempool@@ -979,43 +953,6 @@ cacheGetHead :: MonadLoggerIO m => CacheX m (Maybe BlockHash) cacheGetHead = runRedis redisGetHead -cachePrime ::- (StoreReadBase m, MonadUnliftIO m, MonadLoggerIO m)- => CacheX m (Maybe BlockHash)-cachePrime = cacheGetHead >>= \case- Nothing -> do- $(logDebugS) "Cache" "Cache has no best block set"- lift getBestBlock >>= \case- Nothing -> do- $(logDebugS) "Cache" "Best block not set yet"- return Nothing- Just newhead -> do- ch <- asks cacheChain- chbest <- chainGetBest ch- if headerHash (nodeHeader chbest) == newhead- then do- m <- withLock $ do- $(logInfoS) "Cache" "Priming cache"- mem <- lift getMempool- runRedis $ do- a <- redisAddToMempool mem- b <- redisSetHead newhead- return $ a >> b- $(logDebugS) "Cache" "Primed"- return newhead- case m of- Nothing -> do- $(logErrorS) "Cache"- "Could not get lock to prime cache"- return Nothing- Just x -> return (Just x)- else do- $(logDebugS)- "Cache"- "Not priming cache because not in sync"- return Nothing- Just cachehead -> return (Just cachehead)- cacheSetHead :: (MonadLoggerIO m, StoreReadBase m) => BlockHash -> CacheX m () cacheSetHead bh = do $(logDebugS) "Cache" $ "Cache head set to: " <> blockHashToHex bh@@ -1209,9 +1146,6 @@ cacheNewBlock :: MonadIO m => CacheWriter -> m () cacheNewBlock = send CacheNewBlock--cacheNewTx :: MonadIO m => TxHash -> CacheWriter -> m ()-cacheNewTx th = send (CacheNewTx th) cachePing :: MonadIO m => CacheWriter -> m () cachePing = send CachePing
src/Haskoin/Store/Common.hs view
@@ -27,6 +27,7 @@ , applyLimitsC , sortTxs , nub'+ , microseconds ) where import Conduit (ConduitT, dropC, mapC, takeC)@@ -45,6 +46,8 @@ import Data.List (sortBy) import Data.Maybe (catMaybes, listToMaybe) import Data.Serialize (Serialize (..))+import Data.Time.Clock.System (getSystemTime, systemNanoseconds,+ systemSeconds) import Data.Word (Word32, Word64) import GHC.Generics (Generic) import Haskoin (Address, BlockHash,@@ -65,6 +68,7 @@ XPubSummary (..), XPubUnspent (..), nullBalance, toTransaction, zeroBalance)+import UnliftIO (MonadIO, liftIO) type DeriveAddr = XPubKey -> KeyIndex -> Address @@ -370,3 +374,9 @@ nub' :: (Eq a, Hashable a) => [a] -> [a] nub' = H.toList . H.fromList++microseconds :: MonadIO m => m Integer+microseconds =+ let f t = toInteger (systemSeconds t) * 1000000+ + toInteger (systemNanoseconds t) `div` 1000+ in liftIO $ f <$> getSystemTime
src/Haskoin/Store/Database/Reader.hs view
@@ -224,7 +224,7 @@ -> DatabaseReader -> m [Unspent] getAddressUnspentsDB a limits bdb@DatabaseReader{databaseHandle = db} =- liftIO $ withIterCF db (unspentCF db) $ \it -> runConduit $+ liftIO $ withIterCF db (addrOutCF db) $ \it -> runConduit $ x it .| applyLimitsC limits .| mapC (uncurry toUnspent) .| sinkList where x it = case start limits of
src/Haskoin/Store/Manager.hs view
@@ -35,9 +35,8 @@ blockStoreTxHashSTM, blockStoreTxSTM, withBlockStore) import Haskoin.Store.Cache (CacheConfig (..), CacheWriter,- cacheNewBlock, cacheNewTx,- cachePing, cacheWriter,- connectRedis)+ cacheNewBlock, cachePing,+ cacheWriter, connectRedis) import Haskoin.Store.Common (StoreEvent (..)) import Haskoin.Store.Database.Reader (DatabaseReader (..), DatabaseReaderT,@@ -184,12 +183,13 @@ where f conn cwinbox = runReaderT (cacheWriter (c conn) cwinbox) db- c conn = CacheConfig- { cacheConn = conn- , cacheMin = storeConfCacheMin cfg- , cacheChain = chain- , cacheMax = storeConfMaxKeys cfg- }+ c conn =+ CacheConfig+ { cacheConn = conn+ , cacheMin = storeConfCacheMin cfg+ , cacheChain = chain+ , cacheMax = storeConfMaxKeys cfg+ } cacheWriterProcesses :: MonadUnliftIO m => Inbox StoreEvent@@ -203,10 +203,9 @@ where events = cacheWriterEvents evts cwm ping = forever $ do- time <- liftIO $ randomRIO (300 * second, 600 * second)+ time <- liftIO $ randomRIO (600000, 1200000) threadDelay time cachePing cwm- second = 1000000 cacheWriterEvents :: MonadIO m => Inbox StoreEvent -> CacheWriter -> m () cacheWriterEvents evts cwm =@@ -215,9 +214,8 @@ e `cacheWriterDispatch` cwm cacheWriterDispatch :: MonadIO m => StoreEvent -> CacheWriter -> m ()-cacheWriterDispatch (StoreBestBlock _) = cacheNewBlock-cacheWriterDispatch (StoreMempoolNew th) = cacheNewTx th-cacheWriterDispatch _ = const (return ())+cacheWriterDispatch (StoreBestBlock _) = cacheNewBlock+cacheWriterDispatch _ = const (return ()) nodeForwarder :: MonadIO m => BlockStore