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: 5311cf67a21f834e0890ad311ec5a86ad5678aacdf47bd4e63e0c36ed240332c +-- hash: d8f0a497d75c2e8bea6249b9809c79b72d8b13a39a166553bae1a4e57466d42d name: haskoin-store -version: 0.26.1 +version: 0.26.2 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. @@ -57,7 +57,7 @@ , hashable >=1.3.0.0 , haskoin-core >=0.13.3 , haskoin-node >=0.13.0 - , haskoin-store-data ==0.26.1 + , haskoin-store-data ==0.26.2 , hedis >=0.12.13 , http-types >=0.12.3 , monad-logger >=0.3.32 @@ -99,7 +99,7 @@ , haskoin-core >=0.13.3 , haskoin-node >=0.13.0 , haskoin-store - , haskoin-store-data ==0.26.1 + , haskoin-store-data ==0.26.2 , monad-logger >=0.3.32 , mtl >=2.2.2 , nqe >=0.6.1 @@ -149,7 +149,7 @@ , hashable >=1.3.0.0 , haskoin-core >=0.13.3 , haskoin-node >=0.13.0 - , haskoin-store-data ==0.26.1 + , haskoin-store-data ==0.26.2 , hedis >=0.12.13 , hspec >=2.7.1 , http-types >=0.12.3 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 @@ -39,7 +39,7 @@ import qualified Data.HashMap.Strict as HashMap import qualified Data.HashSet as HashSet import qualified Data.IntMap.Strict as I -import Data.List (nub, sort) +import Data.List (sort) import qualified Data.Map.Strict as Map import Data.Maybe (catMaybes, mapMaybe) import Data.Serialize (decode, encode) @@ -66,8 +66,8 @@ txHash, txHashToHex, xPubAddr, xPubCompatWitnessAddr, xPubExport, xPubWitnessAddr) -import Haskoin.Node (Chain, chainGetAncestor, - chainGetBlock, chainGetSplitBlock) +import Haskoin.Node (Chain, chainBlockMain, + chainGetAncestor, chainGetBlock) import Haskoin.Store.Common (Limit, Offset, StoreRead (..), sortTxs, xPubBals, xPubBalsTxs, xPubBalsUnspents, xPubTxs) @@ -630,50 +630,51 @@ where go newhead cachehead | cachehead == newhead = syncMempoolC - | otherwise = do - ch <- asks cacheChain - chainGetBlock newhead ch >>= \case - Nothing -> do - $(logErrorS) "Cache" $ - "No header for new head: " <> blockHashToHex newhead - throwIO . LogicError . cs $ - "No header for new head: " <> blockHashToHex newhead - Just newheadnode -> - chainGetBlock cachehead ch >>= \case - Nothing -> do - $(logErrorS) "Cache" $ - "No header for cache head: " <> + | otherwise = + asks cacheChain >>= \ch -> + 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 -> go2 newheadnode cacheheadnode - go2 newheadnode cacheheadnode - | nodeHeight cacheheadnode > nodeHeight newheadnode = do - $(logErrorS) "Cache" $ - "Cache head is above new best block: " <> - blockHashToHex (headerHash (nodeHeader newheadnode)) - | otherwise = do - ch <- asks cacheChain - split <- chainGetSplitBlock cacheheadnode newheadnode ch - if split == cacheheadnode - then if prevBlock (nodeHeader newheadnode) == - headerHash (nodeHeader cacheheadnode) - then do - importBlockC (headerHash (nodeHeader newheadnode)) - newBlockC - else go3 newheadnode cacheheadnode - else removeHeadC >> newBlockC - go3 newheadnode cacheheadnode = do - ch <- asks cacheChain - chainGetAncestor (nodeHeight cacheheadnode + 1) newheadnode ch >>= \case - Nothing -> do - $(logErrorS) "Cache" $ - "Could not get expected ancestor block at height " <> - cs (show (nodeHeight cacheheadnode + 1)) <> - " for: " <> - blockHashToHex (headerHash (nodeHeader newheadnode)) - throwIO $ LogicError "Could not get expected ancestor block" - Just a -> do - importBlockC (headerHash (nodeHeader a)) - newBlockC + Just cacheheadnode -> + chainBlockMain cachehead ch >>= \case + False -> do + $(logWarnS) "Cache" $ + "Reverting cache head not in main chain: " <> + blockHashToHex cachehead + removeHeadC >> newBlockC + 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 = + asks cacheChain >>= \ch -> + chainGetAncestor (nodeHeight cacheheadnode + 1) newheadnode ch >>= \case + Nothing -> do + let txt = + "Ancestor not found at height " <> + cs (show (nodeHeight cacheheadnode + 1)) <> + " for block: " <> + blockHashToHex (headerHash (nodeHeader newheadnode)) + $(logErrorS) "Cache" txt + throwIO $ LogicError $ cs txt + Just newcachenode -> + importBlockC (headerHash (nodeHeader newcachenode)) >> + newBlockC importBlockC :: (MonadUnliftIO m, StoreRead m, MonadLoggerIO m) => BlockHash -> CacheT m () importBlockC bh = @@ -688,11 +689,15 @@ go bd = do let ths = blockDataTxs bd tds <- sortTxData . catMaybes <$> mapM (lift . getTxData) ths - $(logInfoS) "Cache" $ + $(logDebugS) "Cache" $ "Importing " <> cs (show (length tds)) <> " transactions from block " <> blockHashToHex bh importMultiTxC tds + $(logInfoS) "Cache" $ + "Done importing " <> cs (show (length tds)) <> + " transactions from block " <> + blockHashToHex bh cacheSetHead bh removeHeadC :: (StoreRead m, MonadUnliftIO m, MonadLoggerIO m) => CacheT m () @@ -704,8 +709,11 @@ tds <- sortTxData . catMaybes <$> mapM (lift . getTxData) (blockDataTxs bd) - $(logWarnS) "Cache" $ "Reverting head: " <> blockHashToHex bh + $(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 :: @@ -713,39 +721,51 @@ => [TxData] -> CacheT m () importMultiTxC txs = do - forM_ (zip [(1 :: Int) ..] txs) $ \(i, tx) -> - $(logDebugS) "Cache" $ - "Import tx " <> cs (show i) <> "/" <> cs (show (length txs)) <> ": " <> - txHashToHex (txHash (txData tx)) + $(logDebugS) "Cache" $ "Processing " <> cs (show (length txs)) <> " txs…" + $(logDebugS) "Cache" $ + "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…" balmap <- getbalances addrs + $(logDebugS) "Cache" $ + "Getting unspent data for " <> cs (show (length allops)) <> " outputs…" unspentmap <- getunspents gap <- getMaxGap now <- systemSeconds <$> liftIO getSystemTime - let xpubs = allxpubs addrmap + let xpubs = allxpubsls addrmap forM_ (zip [(1 :: Int) ..] xpubs) $ \(i, xpub) -> do xpubtxt <- xpubText xpub $(logDebugS) "Cache" $ "Affected xpub " <> cs (show i) <> "/" <> cs (show (length xpubs)) <> ": " <> xpubtxt + $(logDebugS) "Cache" $ "Acquiring lock…" addrs' <- - withLockWait lockKey $ - getxbals xpubs >>= \xmap -> do + withLockWait lockKey $ do + $(logDebugS) "Cache" $ + "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…" runRedis $ do x <- redisImportMultiTx addrmap' unspentmap txs y <- redisUpdateBalances addrmap' balmap z <- redisTouchKeys now (HashMap.keys xmap) return $ x >> y >> z >> return () + $(logDebugS) "Cache" $ "Completed Redis pipeline" return $ getNewAddrs gap xmap (HashMap.elems addrmap') cacheAddAddresses addrs' where + alladdrsls = HashSet.toList alladdrs faddrmap xmap = HashMap.filter (\a -> addressXPubSpec a `elem` xmap) getaddrmap = - HashMap.fromList . catMaybes . zipWith (\a -> fmap (a, )) alladdrs <$> - cacheGetAddrsInfo alladdrs + HashMap.fromList . catMaybes . zipWith (\a -> fmap (a, )) alladdrsls <$> + cacheGetAddrsInfo alladdrsls getunspents = HashMap.fromList . catMaybes . zipWith (\p -> fmap (p, )) allops <$> lift (mapM getUnspent allops) @@ -758,8 +778,12 @@ return $ (xpub, ) <$> bs return $ HashMap.filter (not . null) (HashMap.fromList bals) allops = map snd $ concatMap txInputs txs <> concatMap txOutputs txs - alladdrs = nub . map fst $ concatMap txInputs txs <> concatMap txOutputs txs - allxpubs addrmap = nub . map addressXPubSpec $ HashMap.elems addrmap + alladdrs = + HashSet.fromList . map fst $ + concatMap txInputs txs <> concatMap txOutputs txs + allxpubsls addrmap = HashSet.toList (allxpubs addrmap) + allxpubs addrmap = + HashSet.fromList . map addressXPubSpec $ HashMap.elems addrmap redisImportMultiTx :: (Monad f, RedisCtx m f) @@ -839,19 +863,27 @@ (StoreRead m, MonadUnliftIO m, MonadLoggerIO m) => [(Address, AddressXPub)] -> CacheT m () -cacheAddAddresses [] = return () +cacheAddAddresses [] = $(logDebugS) "Cache" "No further addresses to add" cacheAddAddresses addrs = do $(logDebugS) "Cache" $ "Adding " <> cs (show (length addrs)) <> " new generated addresses" + $(logDebugS) "Cache" $ "Getting balances…" balmap <- HashMap.fromListWith (<>) <$> mapM (uncurry getbal) addrs + $(logDebugS) "Cache" $ "Getting unspent outputs…" utxomap <- HashMap.fromListWith (<>) <$> mapM (uncurry getutxo) addrs + $(logDebugS) "Cache" $ "Getting transactions…" txmap <- HashMap.fromListWith (<>) <$> mapM (uncurry gettxmap) addrs - runRedis $ do - a <- forM (HashMap.toList balmap) (uncurry redisAddXPubBalances) - b <- forM (HashMap.toList utxomap) (uncurry redisAddXPubUnspents) - c <- forM (HashMap.toList txmap) (uncurry redisAddXPubTxs) - return $ sequence a >> sequence b >> sequence c >> return () - let xpubs = nub (map addressXPubSpec (Map.elems amap)) + $(logDebugS) "Cache" $ "Acquiring lock…" + withLockWait lockKey $ do + $(logDebugS) "Cache" $ "Running Redis pipeline…" + runRedis $ do + a <- forM (HashMap.toList balmap) (uncurry redisAddXPubBalances) + b <- forM (HashMap.toList utxomap) (uncurry redisAddXPubUnspents) + c <- forM (HashMap.toList txmap) (uncurry redisAddXPubTxs) + return $ sequence a >> sequence b >> sequence c >> return () + $(logDebugS) "Cache" $ "Completed Redis pipeline" + let xpubs = HashSet.toList . HashSet.fromList . map addressXPubSpec $ Map.elems amap + $(logDebugS) "Cache" $ "Getting xpub balances…" xmap <- getbals xpubs gap <- getMaxGap let notnulls = getnotnull balmap @@ -1021,7 +1053,7 @@ redisRemXPubTxs :: (Applicative f, RedisCtx m f) => XPubSpec -> [TxHash] -> m (f Integer) -redisRemXPubTxs _ [] = return (pure 0) +redisRemXPubTxs _ [] = return (pure 0) redisRemXPubTxs xpub txhs = zrem (txSetPfx <> encode xpub) (map encode txhs) redisAddXPubUnspents ::