haskoin-store 1.8.0 → 1.8.1

raw patch · 12 files changed

+1340/−86 lines, 12 filesdep ~haskoin-store-dataPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: haskoin-store-data

API changes (from Hackage documentation)

Files

• app/Main.hs +14/−0
• haskoin-store.cabal +5/−5
• src/Haskoin/Store.hs +74/−2
• src/Haskoin/Store/BlockStore.hs +190/−18
• src/Haskoin/Store/Cache.hs +197/−9
• src/Haskoin/Store/Common.hs +166/−11
• src/Haskoin/Store/Database/Reader.hs +89/−2
• src/Haskoin/Store/Database/Types.hs +101/−18
• src/Haskoin/Store/Database/Writer.hs +81/−3
• src/Haskoin/Store/Logic.hs +137/−11
• src/Haskoin/Store/Manager.hs +48/−3
• src/Haskoin/Store/Web.hs +238/−4

@@ -86,6 +86,7 @@   ) import UnliftIO.Environment (lookupEnv) +-- | Version of the package, set at build time. haskoinStoreVersion :: String  #ifdef CURRENT_PACKAGE_VERSION@@ -94,6 +95,9 @@ haskoinStoreVersion = "Unavailable" #endif +-- | Command line configuration. Each field has a command line option and an+-- environment variable that sets its default; run @haskoin-store --help@ for+-- the list. data Config = Config   { dir :: !FilePath,     host :: !String,@@ -140,11 +144,13 @@     blockCacheMB :: !Int   } +-- | Read an environment variable with a parser, or return the default if it is unset or invalid. env :: (MonadIO m) => String -> a -> (String -> Maybe a) -> m a env e d p = do   ms <- lookupEnv e   return $ fromMaybe d $ p =<< ms +-- | Default configuration, taken from environment variables where set. defConfig :: (MonadIO m) => m Config defConfig = do   dir <-@@ -260,9 +266,11 @@         env "TX_TIMEOUT" d.txTimeout readMaybe       return WebLimits {..} +-- | Names of the supported networks, separated by pipes. netNames :: String netNames = intercalate "|" $ map (.name) allNets +-- | Parse a boolean from @yes@, @true@, @on@, @1@ or their opposites, ignoring case. parseBool :: String -> Maybe Bool parseBool str = case map toLower str of   "yes" -> Just True@@ -275,6 +283,7 @@   "0" -> Just False   _ -> Nothing +-- | Command line parser, with defaults from the given configuration. config :: Config -> Parser Config config c = do   dir <-@@ -592,12 +601,15 @@         <> value c.blockCacheMB   pure Config {..} +-- | Network by name. networkReader :: String -> Either String Network networkReader s =   case netByName s of     Just net -> Right net     Nothing -> Left "Network name invalid" +-- | Parse the command line and run the store. If no peers are given and discovery+-- is off, discovery is turned on. main :: IO () main = do   c <- execParser . opts =<< defConfig@@ -615,6 +627,8 @@           <> Options.Applicative.header             ("haskoin-store version " <> haskoinStoreVersion) +-- | Run the store with the web server, or without it if disabled, until killed.+-- Data goes in a directory named after the network inside the data directory. run :: Config -> IO () run cfg =   withContext $ \ctx ->

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           haskoin-store-version:        1.8.0+version:        1.8.1 synopsis:       Storage and index for Bitcoin and Bitcoin Cash description:    Please see the README on GitHub at <https://github.com/jprupp/haskoin-store#readme> category:       Bitcoin, Finance, Network@@ -54,7 +54,7 @@     , hashtables >=1.2.4.2     , haskoin-core >=1.6.1     , haskoin-node >=2.3.0-    , haskoin-store-data ==1.8.0+    , haskoin-store-data ==1.8.1     , hedis >=0.16.1     , http-types >=0.12.3     , lens >=4.18.1@@ -107,7 +107,7 @@     , haskoin-core >=1.6.1     , haskoin-node >=2.3.0     , haskoin-store-    , haskoin-store-data ==1.8.0+    , haskoin-store-data ==1.8.1     , hedis >=0.16.1     , http-types >=0.12.3     , lens >=4.18.1@@ -171,7 +171,7 @@     , haskoin-core >=1.6.1     , haskoin-node >=2.3.0     , haskoin-store-    , haskoin-store-data ==1.8.0+    , haskoin-store-data ==1.8.1     , hedis >=0.16.1     , hspec >=2.7.1     , http-types >=0.12.3@@ -226,7 +226,7 @@     , haskoin-core >=1.6.1     , haskoin-node >=2.3.0     , haskoin-store-    , haskoin-store-data ==1.8.0+    , haskoin-store-data ==1.8.1     , hedis >=0.16.1     , http-types >=0.12.3     , lens >=4.18.1

@@ -1,7 +1,79 @@ {-# LANGUAGE DuplicateRecordFields #-} +-- |+-- Module      : Haskoin.Store+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- A block chain indexer for Bitcoin (BTC) and Bitcoin Cash (BCH).+--+-- Haskoin Store connects to the peer-to-peer network with+-- <https://hackage.haskell.org/package/haskoin-node haskoin-node>, downloads+-- blocks and mempool transactions, and indexes them in RocksDB. The indices+-- answer questions about blocks, transactions, address balances, unspent+-- outputs and extended public keys. An optional Redis cache speeds up queries+-- about extended public keys with many addresses. A REST API built on these+-- indices is in "Haskoin.Store.Web".+--+-- = Running a store+--+-- 'withStore' starts everything from a 'StoreConfig': the database, the node,+-- the block store that imports data, and the cache if configured. The 'Store'+-- it returns has a 'DatabaseReader' and, if the cache is on, a 'CacheConfig'.+-- Run queries with 'withDB', or inside 'withCache' to use the cache.+--+-- @+-- 'withStore' cfg $ \\store ->+--   'withSubscription' store.pub $ \\events -> do+--     best <- 'withDB' store.db 'getBestBlock'+--     ...+-- @+--+-- = Modules+--+-- * "Haskoin.Store.Data": types of the data returned by the store and the API.+-- * "Haskoin.Store.Common": the 'StoreReadBase' and 'StoreReadExtra' query+--   interfaces.+-- * "Haskoin.Store.Database.Reader": queries against RocksDB.+-- * "Haskoin.Store.BlockStore": the process that imports blocks and+--   transactions.+-- * "Haskoin.Store.Cache": the Redis cache for extended public keys.+-- * "Haskoin.Store.Web": the REST API server.+-- = Purpose+--+-- A block chain is easy to append to and hard to query. Bitcoin nodes store+-- blocks, and can answer "give me block N", but not "what is the balance of this+-- address" or "which transactions belong to this wallet". Wallets, block+-- explorers and payment processors all need those answers, quickly, from many+-- clients at once. Haskoin Store maintains the indices that make them cheap, and+-- serves them over HTTP.+--+-- = Design+--+-- * __No full node required.__ The store talks to the peer-to-peer network itself+--   through haskoin-node, so a deployment is one program and a database.+--+-- * __Every import is atomic.__ A block or a batch of mempool transactions is+--   applied as one RocksDB write. A crash leaves the database at the previous+--   block, never half-way through one, so there is no recovery step.+--+-- * __Indices are designed around reads.__ Keys are laid out so that every API+--   query is a seek and a short scan, newest first; see+--   "Haskoin.Store.Database.Types".+--+-- * __One query interface, several back ends.__ The database, the import buffer+--   and the Redis cache all implement 'StoreReadBase' and 'StoreReadExtra', so+--   the same query code runs against any of them.+--+-- * __The mempool is a first-class citizen.__ Unconfirmed transactions are indexed+--   like confirmed ones, ordered before them, and kept across restarts, because+--   wallets care about them most. module Haskoin.Store-  ( Store (..),+  ( -- * Store+    Store (..),     StoreConfig (..),     StoreEvent (..),     withStore,@@ -25,7 +97,7 @@     Limits (..),     Start (..), -    -- * Useful Fuctions+    -- * Useful Functions     getTransaction,     getTransactions,     getDefaultBalance,

@@ -12,6 +12,63 @@ {-# LANGUAGE NoFieldSelectors #-} {-# OPTIONS_GHC -Wno-ambiguous-fields #-} +-- |+-- Module      : Haskoin.Store.BlockStore+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- The block store process downloads blocks and transactions from peers and+-- imports them into the database.+--+-- = Blocks+--+-- When the best header of the node is ahead of the best block in the database,+-- the block store picks one online peer as its syncing peer and asks it for+-- blocks in batches of 'blockWindow'. Each block is imported as soon as it+-- arrives. If the best block of the database is no longer in the main chain of+-- headers, blocks are reverted until it is. A syncing peer that sends a block+-- that was not requested, cannot find a block, or stays silent for longer than+-- 'peerTimeout' is disconnected.+--+-- = Mempool+--+-- Once the database is in sync, the block store asks each peer for its mempool+-- (if 'syncMempool' is set) and requests every transaction that peers+-- announce. Received transactions are queued and imported in batches on every+-- clock tick, in dependency order. Transactions that spend unknown outputs are+-- kept for up to an hour, waiting for their parents. Mempool transactions older+-- than 'mempoolTimeout' days are deleted.+--+-- = Feeding the block store+--+-- The block store does not listen to the node by itself.+-- "Haskoin.Store.Manager" forwards node events to it with the+-- @blockStore*@ functions of this module.+-- = Why it works this way+--+-- * __Blocks come from one peer at a time.__ Blocks must be imported in order,+--   so downloading from several peers at once would only buffer blocks that cannot+--   be imported yet. One syncing peer, kept busy with a sliding window of+--   requests, uses the network well without unbounded buffering.+--+-- * __The window refills early.__ The next batch is requested when fewer than a+--   quarter of the window are still outstanding, so the peer never sits idle+--   while the store imports, and the backlog never grows much beyond one window.+--+-- * __Mempool sync waits for the tip.__ Unconfirmed transactions only make sense+--   against the current chain, so they are requested once the database is in+--   sync.+--+-- * __Orphans get a grace period.__ A transaction can arrive before its parent.+--   It is held for up to an hour, waiting for the parent, instead of being+--   dropped and requested again.+--+-- * __Imports are batched.__ Mempool transactions are imported up to 2000 at a+--   time on each clock tick, which amortises the cost of a database write over+--   many transactions during bursts. module Haskoin.Store.BlockStore   ( -- * Block Store     BlockStore,@@ -178,57 +235,95 @@   ) import UnliftIO.Concurrent (threadDelay) +-- | Messages handled by the block store process. data BlockStoreMessage-  = BlockNewBest !BlockNode-  | BlockPeerConnect !Peer-  | BlockPeerDisconnect !Peer-  | BlockReceived !Peer !Block-  | BlockNotFound !Peer ![BlockHash]-  | TxRefReceived !Peer !Tx-  | TxRefAvailable !Peer ![TxHash]-  | BlockPing !(Listen ())+  = -- | the node has a new best header+    BlockNewBest !BlockNode+  | -- | a peer connected+    BlockPeerConnect !Peer+  | -- | a peer disconnected+    BlockPeerDisconnect !Peer+  | -- | a peer sent a block+    BlockReceived !Peer !Block+  | -- | a peer could not find blocks+    BlockNotFound !Peer ![BlockHash]+  | -- | a peer sent a transaction+    TxRefReceived !Peer !Tx+  | -- | a peer announced transactions+    TxRefAvailable !Peer ![TxHash]+  | -- | clock tick for housekeeping; reply when done+    BlockPing !(Listen ()) +-- | Fatal errors of the block store. data BlockException-  = BlockNotInChain !BlockHash-  | Uninitialized-  | CorruptDatabase-  | AncestorNotInChain !BlockHeight !BlockHash-  | MempoolImportFailed+  = -- | the header of a block in the database is not known to the node+    BlockNotInChain !BlockHash+  | -- | the database has no best block+    Uninitialized+  | -- | the database is inconsistent+    CorruptDatabase+  | -- | the node has no header at this height below this block+    AncestorNotInChain !BlockHeight !BlockHash+  | -- | importing mempool transactions failed+    MempoolImportFailed   deriving (Show, Eq, Ord, Exception) +-- | State of the sync against the syncing peer. data Syncing = Syncing-  { peer :: !Peer,+  { -- | peer that blocks are downloaded from+    peer :: !Peer,+    -- | when it last delivered a block, or when syncing started     time :: !UTCTime,+    -- | blocks requested and not yet imported, in chain order     blocks :: ![BlockHash]   } +-- | Transaction received from a peer and waiting to be imported. data PendingTx = PendingTx-  { time :: !UTCTime,+  { -- | when it was received+    time :: !UTCTime,+    -- | the transaction     tx :: !Tx,+    -- | parent transactions that are missing; empty if it can be imported     deps :: !(HashSet TxHash)   }   deriving (Show, Eq, Ord)  -- | Block store process state. data BlockStore = BlockStore-  { mailbox :: !(Mailbox BlockStoreMessage),+  { -- | mailbox of the process+    mailbox :: !(Mailbox BlockStoreMessage),+    -- | configuration     config :: !BlockStoreConfig,+    -- | syncing peer, if any     peer :: !(TVar (Maybe Syncing)),+    -- | transactions received and waiting to be imported     txs :: !(TVar (HashMap TxHash PendingTx)),+    -- | transactions requested from peers in the last twenty seconds     requested :: !(TVar (HashSet TxHash)),+    -- | StatsD metrics, if enabled     metrics :: !(Maybe StoreMetrics),+    -- | labels of the peers whose mempool was already requested     syncedMempool :: !(TVar (HashSet Text))   } +-- | StatsD metrics of the block store. data StoreMetrics = StoreMetrics-  { blocks :: !StatGauge,+  { -- | height of the best block (@store.blocks@)+    blocks :: !StatGauge,+    -- | height of the best header (@store.headers@)     headers :: !StatGauge,+    -- | transactions waiting to be imported (@store.queued_txs@)     queuedTxs :: !StatGauge,+    -- | online peers (@store.peers@)     peers :: !StatGauge,+    -- | transactions in the mempool (@store.mempool@)     mempool :: !StatGauge,+    -- | milliseconds to import a block (@store.block@)     block :: !StatTiming   } +-- | Register the block store metrics, if StatsD is configured, with their current values. newStoreMetrics :: (MonadIO m) => BlockStoreConfig -> m (Maybe StoreMetrics) newStoreMetrics cfg =   forM cfg.stats $ \s -> liftIO $ do@@ -251,6 +346,7 @@     g s x = newStatGauge s ("store." <> x)     t s x = newStatTiming s ("store." <> x) +-- | Update the gauge of the best block height. setStoreHeight :: (MonadIO m) => BlockT m () setStoreHeight = void $ runMaybeT $ do   m <- MaybeT (asks (.metrics))@@ -258,24 +354,28 @@   b <- MaybeT (getBlock h)   setGauge m.blocks (fromIntegral b.height) +-- | Update the gauge of the best header height. setHeadersHeight :: (MonadIO m) => BlockT m () setHeadersHeight = void $ runMaybeT $ do   m <- MaybeT (asks (.metrics))   n <- liftIO . chainGetBest =<< asks (.config.chain)   setGauge m.headers (fromIntegral n.height) +-- | Update the gauge of transactions waiting to be imported. setPendingTxs :: (MonadIO m) => BlockT m () setPendingTxs = void $ runMaybeT $ do   m <- MaybeT (asks (.metrics))   p <- readTVarIO =<< asks (.txs)   setGauge m.queuedTxs (HashMap.size p) +-- | Update the gauge of online peers. setPeersConnected :: (MonadIO m) => BlockT m () setPeersConnected = void $ runMaybeT $ do   m <- MaybeT (asks (.metrics))   p <- liftIO . getPeers =<< asks (.config.peerMgr)   setGauge m.peers (length p) +-- | Record how long a successful block import took. withBlockTiming :: (MonadIO m) => BlockT m (Either l r) -> BlockT m (Either l r) withBlockTiming go = do   asks (.metrics) >>= \case@@ -291,7 +391,8 @@  -- | Configuration for a block store. data BlockStoreConfig = BlockStoreConfig-  { ctx :: !Ctx,+  { -- | secp256k1 context+    ctx :: !Ctx,     -- | peer manager from running node     peerMgr :: !PeerMgr,     -- | chain from a running node@@ -314,13 +415,16 @@     mempoolTimeout :: !Int,     -- | disconnect syncing peer if inactive for this long     peerTimeout :: !NominalDiffTime,+    -- | StatsD client, if metrics are enabled     stats :: !(Maybe Stats),     -- | log messages here     logger :: !Logger   } +-- | Monad transformer of the block store process. type BlockT m = ReaderT BlockStore m +-- | Log a message with the logger of the block store. logDebug, logInfo, logWarn, logError ::   (MonadIO m, MonadReader BlockStore m) => Text -> m () logDebug = blockLog LevelDebug@@ -328,9 +432,11 @@ logWarn = blockLog LevelWarn logError = blockLog LevelError +-- | Log a message at a level with the logger of the block store. blockLog :: (MonadIO m, MonadReader BlockStore m) => LogLevel -> Text -> m () blockLog l msg = asks (.config.logger) >>= \lg -> logAt l lg msg +-- | Run an import against the database, returning the import error if it fails. runImport ::   (MonadIO m) =>   Network ->@@ -382,6 +488,7 @@     inBlock = maybe HashSet.empty (HashSet.fromList . map txHash . (.txs)) block     notInBlock = not . (`HashSet.member` inBlock) +-- | Run a database query from the block store. runRocksDB :: ReaderT DatabaseReader m a -> BlockT m a runRocksDB f =   ReaderT $ runReaderT f . (.config.db)@@ -502,6 +609,7 @@       withAsync (pingMe (inboxToMailbox inbox)) $ \a ->         link a >> runBlockStoreLoop inbox +-- | Handle messages forever. runBlockStoreLoop ::   (InChan mbox, MonadUnliftIO m) =>   mbox BlockStoreMessage ->@@ -512,6 +620,8 @@     msg <- receive inbox     ReaderT $ runReaderT $ processBlockStoreMessage msg +-- | Whether the best block in the database is the best header of the node. If+-- so, release the syncing peer. isInSync :: (MonadIO m) => BlockT m Bool isInSync =   getBestBlock >>= \case@@ -524,12 +634,15 @@         then clearSyncingState >> return True         else return False +-- | Run an action only if the database is in sync. guardSync :: (MonadIO m) => BlockT m () -> BlockT m () guardSync f = isInSync >>= \s -> when s f +-- | Run an action only if the mempool is indexed. guardMempool :: (Monad m) => BlockT m () -> BlockT m () guardMempool f = asks (.config.noMempool) >>= \n -> unless n f +-- | Ask a peer for its mempool, once per peer, if in sync and mempool sync is enabled. syncMempool :: (MonadUnliftIO m) => Peer -> BlockT m () syncMempool p =   guardMempool . guardSync $ do@@ -542,6 +655,9 @@       atomically . modifyTVar m $ HashSet.insert p.label       liftIO $ MMempool `sendMessage` p +-- | Import a block from the syncing peer, then ask for more blocks or, if now in+-- sync, sync the mempool. Disconnects the peer if it is not the syncing peer, if+-- the block is unknown to the node, or if the block cannot be imported. processBlock ::   (MonadUnliftIO m) =>   Peer ->@@ -603,6 +719,7 @@           <> cs (show e)       killPeer peer +-- | Set the blocks requested from the syncing peer. setSyncingBlocks ::   (MonadReader BlockStore m, MonadIO m) =>   [BlockHash] ->@@ -614,12 +731,14 @@         Nothing -> Nothing         Just x -> Just (x :: Syncing) {blocks = hs} +-- | Blocks requested from the syncing peer and not yet imported. getSyncingBlocks :: (MonadReader BlockStore m, MonadIO m) => m [BlockHash] getSyncingBlocks =   asks (.peer) >>= readTVarIO >>= \case     Nothing -> return []     Just x -> return x.blocks +-- | Add blocks to those requested from the syncing peer. addSyncingBlocks ::   (MonadReader BlockStore m, MonadIO m) =>   [BlockHash] ->@@ -631,6 +750,7 @@         Nothing -> Nothing         Just x -> Just (x :: Syncing) {blocks = x.blocks <> hs} +-- | Remove a block from those requested from the syncing peer. removeSyncingBlock ::   (MonadReader BlockStore m, MonadIO m) =>   BlockHash ->@@ -642,11 +762,13 @@       Nothing -> Nothing       Just x -> Just (x :: Syncing) {blocks = delete h x.blocks} +-- | Whether a peer is the syncing peer. checkPeer :: (MonadIO m, MonadReader BlockStore m) => Peer -> m Bool checkPeer p = fmap (fromMaybe False) <$> runMaybeT $ do   Syncing {peer} <- MaybeT getSyncingState   return $ peer == p +-- | Header of a block from the node. getBlockNode ::   (MonadIO m, MonadReader BlockStore m) =>   BlockHash ->@@ -655,6 +777,7 @@   ch <- asks (.config.chain)   liftIO $ chainGetBlock ch blockhash +-- | Disconnect a peer that could not find requested blocks. processNoBlocks ::   (MonadIO m) =>   Peer ->@@ -673,6 +796,7 @@         <> p.label   killPeer p +-- | Queue a transaction received from a peer for import. processTx :: (MonadIO m) => Peer -> Tx -> BlockT m () processTx p tx = guardMempool $ do   t <- liftIO getCurrentTime@@ -683,6 +807,7 @@       <> p.label   addPendingTx $ PendingTx t tx HashSet.empty +-- | Drop queued transactions received more than an hour ago. pruneOrphans :: (MonadIO m) => BlockT m () pruneOrphans = guardMempool $ do   ts <- asks (.txs)@@ -691,6 +816,7 @@   atomically . modifyTVar' ts . HashMap.filter $ \p ->     now `diffUTCTime` p.time <= 3600 +-- | Queue a transaction for import. addPendingTx :: (MonadIO m) => PendingTx -> BlockT m () addPendingTx p = do   ts <- asks (.txs)@@ -702,6 +828,7 @@   where     th = txHash p.tx +-- | Remember for twenty seconds that a transaction was requested, so that it is not requested again. addRequestedTx :: (MonadIO m) => TxHash -> BlockT m () addRequestedTx th = do   qbox <- asks (.requested)@@ -710,6 +837,7 @@     threadDelay 20000000     atomically $ modifyTVar qbox $ HashSet.delete th +-- | Whether a transaction is queued or was requested recently. isPending :: (MonadIO m) => TxHash -> BlockT m Bool isPending th = do   tbox <- asks (.txs)@@ -721,6 +849,8 @@       th `HashMap.member` ts         || th `HashSet.member` rs +-- | Take up to this many queued transactions that have no missing parents, in+-- dependency order, out of the queue. pendingTxs :: (MonadIO m) => Int -> BlockT m [PendingTx] pendingTxs i = do   selected <-@@ -743,6 +873,7 @@           txids = map (txHash . snd) sorted        in mapMaybe (`HashMap.lookup` m) txids +-- | Mark a transaction as no longer missing for the queued transactions that spend it. fulfillOrphans :: (MonadIO m) => BlockStore -> TxHash -> m () fulfillOrphans block_read th =   atomically $ modifyTVar box (HashMap.map fulfill)@@ -750,6 +881,8 @@     box = block_read.txs     fulfill p = p {deps = HashSet.delete th p.deps} +-- | Drop queued transactions that are already in the database, and recompute the+-- missing parents of the others. updateOrphans ::   ( StoreReadBase m,     MonadIO m,@@ -783,6 +916,7 @@       unspents <- prev_utxos tx       return $ foldl fulfill p unspents +-- | Queue a transaction that spends unknown outputs, recording which parents are missing. newOrphanTx ::   (MonadIO m) =>   BlockStore ->@@ -811,6 +945,9 @@     prev_set = HashSet.fromList prevs     prevs = map (.outpoint) tx.inputs +-- | Import a transaction into the mempool. Returns 'True' if imported. A+-- transaction that spends unknown outputs is queued as an orphan; other errors+-- are ignored. importMempoolTx :: BlockStore -> UTCTime -> Tx -> ImportM Bool importMempoolTx block_read time tx =   catch new_mempool_tx handle_error@@ -831,6 +968,7 @@       when t $ fulfillOrphans block_read tx_hash       return t +-- | Import up to 2000 queued transactions. processMempool :: (MonadIO m) => BlockT m () processMempool = guardMempool $ do   txs <- pendingTxs 2000@@ -848,6 +986,7 @@         "Error processing mempool: " <> cs (show e)       throwIO e +-- | Delete up to 1000 mempool transactions older than the configured number of days. pruneMempool :: (MonadIO m) => BlockT m () pruneMempool =   guardMempool $ do@@ -867,6 +1006,8 @@               "Error pruning mempool: " <> cs (show e)           Right () -> return () +-- | Request from a peer the announced transactions that are neither queued,+-- recently requested nor in the database. Only while in sync. processTxs ::   (MonadIO m) =>   Peer ->@@ -928,6 +1069,7 @@           msg = MGetData (GetData vec)       liftIO $ msg `sendMessage` p +-- | Record that the syncing peer just delivered something. touchPeer ::   ( MonadIO m,     MonadReader BlockStore m@@ -944,6 +1086,7 @@           fmap $             \x -> (x :: Syncing) {time = now} +-- | Disconnect the syncing peer if it has been silent for too long. checkTime :: (MonadIO m) => BlockT m () checkTime =   asks (.peer) >>= readTVarIO >>= \case@@ -960,6 +1103,7 @@             "Syncing peer timeout: " <> p.label           killPeer p +-- | Revert blocks until the best block in the database is in the main chain of headers. revertToMainChain :: (MonadIO m) => BlockT m () revertToMainChain = do   h <- headerHash . (.header) <$> getBest@@ -981,6 +1125,7 @@       Right () -> setSyncingBlocks []     revertToMainChain +-- | Header of the best block in the database. getBest :: (MonadIO m) => BlockT m BlockNode getBest = do   bb <-@@ -998,6 +1143,7 @@           <> blockHashToHex bb       throwIO (BlockNotInChain bb) +-- | Header of the last block requested from the syncing peer, or of the best block if none. getSyncBest :: (MonadIO m) => BlockT m BlockNode getSyncBest = do   bb <-@@ -1083,6 +1229,7 @@       ps <- liftIO $ chainGetParents ch (bb.height + 1) top       return $ ps <> [top] +-- | Ancestor of a header at a height. Throws 'AncestorNotInChain' if missing. findAncestor ::   (MonadIO m, MonadReader BlockStore m) =>   BlockHeight ->@@ -1101,6 +1248,7 @@           <> cs (show target.height)       throwIO $ AncestorNotInChain height h +-- | Forget a peer: release it if it was the syncing peer, and forget that its mempool was requested. finishPeer ::   (MonadIO m, MonadReader BlockStore m) =>   Peer ->@@ -1118,6 +1266,7 @@       logDebug $ "Releasing peer: " <> p.label       liftIO $ setFree p +-- | Make a peer the syncing peer if there is none and the peer is not busy. trySetPeer :: (MonadIO m) => Peer -> BlockT m Bool trySetPeer p =   getSyncingState >>= \case@@ -1141,6 +1290,7 @@                 }           return True +-- | If not in sync and there is no syncing peer, pick the fastest free peer and start syncing. trySyncing :: (MonadIO m) => BlockT m () trySyncing =   isInSync >>= \case@@ -1160,6 +1310,7 @@       let ps = map (.mailbox) ops       recurse ps +-- | Start syncing against a newly connected peer if needed, or request its mempool if in sync. trySyncingPeer :: (MonadUnliftIO m) => Peer -> BlockT m () trySyncingPeer p =   isInSync >>= \case@@ -1169,11 +1320,13 @@         False -> return ()         True -> syncMe +-- | State of the sync against the syncing peer, if any. getSyncingState ::   (MonadIO m, MonadReader BlockStore m) => m (Maybe Syncing) getSyncingState =   readTVarIO =<< asks (.peer) +-- | Release the syncing peer, if any. clearSyncingState ::   (MonadIO m, MonadReader BlockStore m) => m () clearSyncingState =@@ -1181,6 +1334,7 @@     Nothing -> return ()     Just Syncing {peer = p} -> finishPeer p +-- | Handle a message sent to the block store. processBlockStoreMessage ::   (MonadUnliftIO m) =>   BlockStoreMessage ->@@ -1239,6 +1393,7 @@   checkTime   atomically (r ()) +-- | Send a clock tick to the block store every 0.1 to 1 second, waiting for each to be handled. pingMe :: (MonadIO m) => Mailbox BlockStoreMessage -> m () pingMe mbox =   loopForever $ do@@ -1246,80 +1401,96 @@     delay <- liftIO $ randomRIO (100 * 1000, 1000 * 1000)     threadDelay delay +-- | Tell the block store that a peer connected. blockStorePeerConnect :: (MonadIO m) => Peer -> BlockStore -> m () blockStorePeerConnect peer store =   BlockPeerConnect peer `send` store.mailbox +-- | Tell the block store that a peer disconnected. blockStorePeerDisconnect ::   (MonadIO m) => Peer -> BlockStore -> m () blockStorePeerDisconnect peer store =   BlockPeerDisconnect peer `send` store.mailbox +-- | Tell the block store that the node has a new best header. blockStoreHead ::   (MonadIO m) => BlockNode -> BlockStore -> m () blockStoreHead node store =   BlockNewBest node `send` store.mailbox +-- | Give the block store a block received from a peer. blockStoreBlock ::   (MonadIO m) => Peer -> Block -> BlockStore -> m () blockStoreBlock peer block store =   BlockReceived peer block `send` store.mailbox +-- | Tell the block store that a peer could not find blocks. blockStoreNotFound ::   (MonadIO m) => Peer -> [BlockHash] -> BlockStore -> m () blockStoreNotFound peer blocks store =   BlockNotFound peer blocks `send` store.mailbox +-- | Give the block store a transaction received from a peer. blockStoreTx ::   (MonadIO m) => Peer -> Tx -> BlockStore -> m () blockStoreTx peer tx store =   TxRefReceived peer tx `send` store.mailbox +-- | Tell the block store that a peer announced transactions. blockStoreTxHash ::   (MonadIO m) => Peer -> [TxHash] -> BlockStore -> m () blockStoreTxHash peer txhashes store =   TxRefAvailable peer txhashes `send` store.mailbox +-- | 'STM' version of 'blockStorePeerConnect'. blockStorePeerConnectSTM ::   Peer -> BlockStore -> STM () blockStorePeerConnectSTM peer store =   BlockPeerConnect peer `sendSTM` store.mailbox +-- | 'STM' version of 'blockStorePeerDisconnect'. blockStorePeerDisconnectSTM ::   Peer -> BlockStore -> STM () blockStorePeerDisconnectSTM peer store =   BlockPeerDisconnect peer `sendSTM` store.mailbox +-- | 'STM' version of 'blockStoreHead'. blockStoreHeadSTM ::   BlockNode -> BlockStore -> STM () blockStoreHeadSTM node store =   BlockNewBest node `sendSTM` store.mailbox +-- | 'STM' version of 'blockStoreBlock'. blockStoreBlockSTM ::   Peer -> Block -> BlockStore -> STM () blockStoreBlockSTM peer block store =   BlockReceived peer block `sendSTM` store.mailbox +-- | 'STM' version of 'blockStoreNotFound'. blockStoreNotFoundSTM ::   Peer -> [BlockHash] -> BlockStore -> STM () blockStoreNotFoundSTM peer blocks store =   BlockNotFound peer blocks `sendSTM` store.mailbox +-- | 'STM' version of 'blockStoreTx'. blockStoreTxSTM ::   Peer -> Tx -> BlockStore -> STM () blockStoreTxSTM peer tx store =   TxRefReceived peer tx `sendSTM` store.mailbox +-- | 'STM' version of 'blockStoreTxHash'. blockStoreTxHashSTM ::   Peer -> [TxHash] -> BlockStore -> STM () blockStoreTxHashSTM peer txhashes store =   TxRefAvailable peer txhashes `sendSTM` store.mailbox +-- | Number of transactions queued or requested by the block store. blockStorePendingTxs ::   (MonadIO m) => BlockStore -> m Int blockStorePendingTxs =   atomically . blockStorePendingTxsSTM +-- | 'STM' version of 'blockStorePendingTxs'. blockStorePendingTxsSTM ::   BlockStore -> STM Int blockStorePendingTxsSTM BlockStore {..} = do@@ -1327,6 +1498,7 @@   y <- readTVar requested   return $ HashSet.size $ x `HashSet.union` y +-- | Describe a block for logs: height, time, hash and, if the block is given, size. blockText :: BlockNode -> Maybe Block -> Text blockText bn mblock = case mblock of   Nothing ->

@@ -14,6 +14,57 @@ {-# LANGUAGE TupleSections #-} {-# LANGUAGE NoFieldSelectors #-} +-- |+-- Module      : Haskoin.Store.Cache+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- Redis cache for extended public keys (xpubs).+--+-- Answering a query about an xpub from the database means deriving its+-- addresses and reading the index of every one of them, which is slow for+-- wallets with many used addresses. When such an xpub is queried, its balances,+-- transactions and unspent outputs are copied to Redis, and later queries are+-- answered from there. Only xpubs with at least 'minAddrs' used addresses are+-- cached, and only while the store is in sync.+--+-- A cache writer process ('cacheWriter') keeps cached xpubs up to date as new+-- blocks and mempool transactions arrive. When Redis holds more than 'maxKeys'+-- keys, the xpubs used least recently are evicted.+--+-- = Redis keys+--+-- * @b\<xpub\>@: hash of derivation path to balance of each address.+-- * @t\<xpub\>@: sorted set of transactions, scored by 'blockRefScore'.+-- * @u\<xpub\>@: sorted set of unspent outputs, scored by 'blockRefScore'.+-- * @a\<address\>@: xpub and derivation path of a cached address.+-- * @max@: sorted set of cached xpubs, scored by time of last use.+-- * @mempool@: sorted set of mempool transactions.+-- * @head@: last block that the cache has processed.+-- * @xlock-\<xpub\>@, @lock@, @i\<xpub\>@: locks.+--+-- Here @\<xpub\>@ and @\<address\>@ stand for their binary serialization.+-- = Why a cache, and why Redis+--+-- Most xpubs are small and fast to answer from RocksDB. A few, such as the+-- wallets of exchanges and payment processors, have thousands of used addresses+-- and are queried constantly. Answering them from the database means deriving and+-- reading every address each time. Keeping their balances, transactions and+-- unspent outputs ready in sorted sets turns those queries into a single range+-- read.+--+-- Redis fits because sorted sets are exactly the shape of these queries, and+-- because several store instances can share one cache. Only large xpubs are+-- cached ('minAddrs'), because small ones gain nothing, and xpubs not used for a+-- while are evicted, so the cache holds the working set rather than everything+-- ever queried.+--+-- Locks keep two store instances, or two requests, from building the cache of+-- the same xpub at the same time. A request that cannot get a lock in time is+-- answered from the database rather than made to wait. module Haskoin.Store.Cache   ( CacheConfig (..),     CacheMetrics,@@ -140,12 +191,14 @@   ) import UnliftIO.Concurrent (threadDelay) +-- | Run a Redis command. Throws 'RedisError' if Redis returns an error. runRedis :: (MonadIO m) => Redis (Either Reply a) -> CacheX m a runRedis action = do   conn <- asks (.redis)   logger <- asks (.logger)   withRedis logger conn action +-- | Run a Redis command on a connection. Logs and throws 'RedisError' if Redis returns an error. withRedis ::   (MonadIO m) => Logger -> Connection -> Redis (Either Reply a) -> m a withRedis logger conn action =@@ -155,29 +208,49 @@       logAt LevelError logger $ "Got error from Redis: " <> cs (show e)       throwIO (RedisError e) +-- | Cache configuration. data CacheConfig = CacheConfig-  { redis :: !Connection,+  { -- | connection to Redis+    redis :: !Connection,+    -- | cache xpubs with at least this many used addresses     minAddrs :: !Int,+    -- | evict xpubs when Redis holds more keys than this     maxKeys :: !Integer,+    -- | header chain, used to know whether the store is in sync     chain :: !Chain,+    -- | StatsD metrics, if enabled     metrics :: !(Maybe CacheMetrics),+    -- | log messages here     logger :: !Logger   } +-- | StatsD metrics of the cache. data CacheMetrics = CacheMetrics-  { hits :: !StatCounter,+  { -- | xpub balance queries answered from the cache (@cache.hits@)+    hits :: !StatCounter,+    -- | xpub balance queries not in the cache (@cache.misses@)     misses :: !StatCounter,+    -- | locks acquired (@cache.lock_acquired@)     lockAcquired :: !StatCounter,+    -- | locks released (@cache.lock_released@)     lockReleased :: !StatCounter,+    -- | locks that could not be acquired (@cache.lock_failed@)     lockFailed :: !StatCounter,+    -- | xpub balances served (@cache.xpub_balances@)     xBalances :: !StatCounter,+    -- | xpub unspent outputs served (@cache.xpub_unspents@)     xUnspents :: !StatCounter,+    -- | xpub transactions served (@cache.xpub_txs@)     xTxs :: !StatCounter,+    -- | xpub transaction counts served (@cache.xpub_tx_count@)     xTxCount :: !StatCounter,+    -- | milliseconds to cache an xpub (@cache.index@)     indexTime :: !StatTiming,+    -- | height of the last block processed by the cache (@cache.height@)     height :: !StatGauge   } +-- | Register the cache metrics with a StatsD client. newCacheMetrics ::   (MonadIO m) =>   Logger ->@@ -209,6 +282,7 @@     c x = newStatCounter s x 10     d x = newStatTiming s x 10 +-- | Record how long an action takes in a timing metric. withTiming ::   (MonadUnliftIO m) =>   (CacheMetrics -> StatTiming) ->@@ -226,29 +300,41 @@     time = round . (* 1000) <$> liftIO getPOSIXTime     end m t1 = time >>= addTiming (df m) . subtract t1 +-- | Run an action with the metrics, if they are enabled. withMetrics ::   (MonadIO m) => (CacheMetrics -> CacheX m a) -> CacheX m () withMetrics go = asks (.metrics) >>= mapM_ go +-- | Monad transformer that may use the cache. With 'Nothing', every query goes+-- to the underlying store. Run it with 'withCache'. type CacheT = ReaderT (Maybe CacheConfig) +-- | Monad transformer that uses the cache. type CacheX = ReaderT CacheConfig +-- | Log a message with the logger of the cache. logDebug, logInfo, logWarn, logError :: (MonadIO m) => Text -> CacheX m () logDebug = cacheLog LevelDebug logInfo = cacheLog LevelInfo logWarn = cacheLog LevelWarn logError = cacheLog LevelError +-- | Log a message at a level with the logger of the cache. cacheLog :: (MonadIO m) => LogLevel -> Text -> CacheX m () cacheLog l msg = asks (.logger) >>= \lg -> logAt l lg msg +-- | Errors of the cache. data CacheError-  = RedisError Reply-  | RedisTxError !String-  | LogicError !String+  = -- | Redis returned an error+    RedisError Reply+  | -- | a Redis transaction failed+    RedisTxError !String+  | -- | the cache is inconsistent with the database+    LogicError !String   deriving (Show, Eq, Generic, NFData, Exception) +-- | Connect to Redis at a URL such as @redis:\/\/localhost:6379@, or with the+-- default settings if the URL is empty. Calls 'error' if the URL is invalid. connectRedis :: (MonadIO m) => String -> m Connection connectRedis redisurl = do   conninfo <-@@ -333,21 +419,27 @@       Nothing -> xPubTxCount xpub xbals       Just cfg -> runReaderT (xPubTxCount xpub xbals) cfg +-- | Run queries with the cache, or without it if 'Nothing'. withCache :: Maybe CacheConfig -> CacheT m a -> m a withCache s f = runReaderT f s +-- | Prefix of the keys of xpub balances. balancesPfx :: ByteString balancesPfx = "b" +-- | Prefix of the keys of xpub transactions. txSetPfx :: ByteString txSetPfx = "t" +-- | Prefix of the keys of xpub unspent outputs. utxoPfx :: ByteString utxoPfx = "u" +-- | Prefix of the keys that mark an xpub as being cached. idxPfx :: ByteString idxPfx = "i" +-- | Key of the lock held while caching an xpub. xpubLock :: XPubSpec -> ByteString xpubLock xpub = "xlock-" <> encode xpub @@ -375,6 +467,7 @@ startScore (Just (AtBlock h)) = Just (Just (blockRefScore (BlockRef h maxBound))) startScore (Just AtTx {}) = Nothing +-- | Transactions of an xpub, from the cache if it is cached, or from the store otherwise. getXPubTxs ::   (MonadUnliftIO m, StoreReadExtra m) =>   XPubSpec ->@@ -406,6 +499,7 @@         incrementCounter s.xTxs (length txs)       return txs +-- | Number of transactions of an xpub, from the cache if it is cached, or from the store otherwise. getXPubTxCount ::   (MonadUnliftIO m, StoreReadExtra m) =>   XPubSpec ->@@ -420,6 +514,7 @@       return (fromInteger count)     Nothing -> lift $ xPubTxCount xpub xbals +-- | Unspent outputs of an xpub, from the cache if it is cached, or from the store otherwise. getXPubUnspents ::   (MonadUnliftIO m, StoreReadExtra m) =>   XPubSpec ->@@ -470,6 +565,8 @@       withMetrics $ \s -> incrementCounter s.xUnspents (length us)       return $ mapMaybe (uncurry i) us +-- | Balances of an xpub, from the cache if it is cached. Otherwise read them from+-- the store and cache the xpub if it has enough used addresses. getXPubBalances ::   (MonadUnliftIO m, StoreReadExtra m) =>   XPubSpec ->@@ -497,21 +594,26 @@       newXPubC x bals       return bals +-- | Whether an xpub is cached. Always 'False' without a cache. isInCache :: (MonadIO m) => XPubSpec -> CacheT m Bool isInCache xpub =   ReaderT $ \case     Nothing -> return False     Just cfg -> runReaderT (isXPubCached xpub) cfg +-- | Whether an xpub is cached. isXPubCached :: (MonadIO m) => XPubSpec -> CacheX m Bool isXPubCached xpub = runRedis (redisIsXPubCached xpub) +-- | Redis command to check whether an xpub is cached. redisIsXPubCached :: (RedisCtx m f) => XPubSpec -> m (f Bool) redisIsXPubCached xpub = Redis.exists (balancesPfx <> encode xpub) +-- | Redis command to count the transactions of an xpub. redisGetXPubTxCount :: (RedisCtx m f) => XPubSpec -> m (f Integer) redisGetXPubTxCount xpub = Redis.zcard (txSetPfx <> encode xpub) +-- | Page of the cached transactions of an xpub. cacheGetXPubTxs ::   (StoreReadBase m, MonadIO m) =>   XPubSpec ->@@ -563,6 +665,7 @@         else id     f t s = TxRef {txid = t, block = scoreBlockRef s} +-- | Page of the cached unspent outputs of an xpub, with their positions. cacheGetXPubUnspents ::   (StoreReadBase m, MonadIO m) =>   XPubSpec ->@@ -614,12 +717,22 @@         else id     f o s = (scoreBlockRef s, o) +-- | Redis command to read the cached balances of an xpub, sorted by path. redisGetXPubBalances :: (Functor f, RedisCtx m f) => XPubSpec -> m (f [XPubBal]) redisGetXPubBalances xpub =   fmap (sort . map (uncurry f)) <$> getAllFromMap (balancesPfx <> encode xpub)   where     f p b = XPubBal {path = p, balance = b} +-- | Score of a position in a Redis sorted set, such that sorting by score+-- ascending sorts from newest to oldest. Mempool transactions get negative+-- scores from their time, so they come first; confirmed ones get positive+-- scores that decrease with height and position.+--+-- One number has to order mempool and confirmed transactions in a single sorted+-- set, newest first, exactly as the database does. Floating point is what Redis+-- offers, so the height and position are packed into the 53 bits a double holds+-- exactly. blockRefScore :: BlockRef -> Double blockRefScore BlockRef {height = h, position = p} =   fromIntegral (0x001fffffffffffff - (h' .|. p'))@@ -630,6 +743,7 @@   where     t' = fromIntegral (t .&. 0x001fffffffffffff) +-- | Inverse of 'blockRefScore'. scoreBlockRef :: Double -> BlockRef scoreBlockRef s   | s < 0 = MemRef {timestamp = n}@@ -640,6 +754,9 @@     h = fromIntegral (m `shift` (-26))     p = fromIntegral (m .&. 0x03ffffff) +-- | Redis command to read members of a sorted set, decoded, with their scores,+-- from a minimum score if given, skipping an offset, up to a count (@0@ for+-- all). getFromSortedSet ::   (Applicative f, RedisCtx m f, Serialize a) =>   ByteString ->@@ -684,6 +801,7 @@     ys <- map (\(x, s) -> (,s) <$> decode x) <$> xs     return (rights ys) +-- | Redis command to read all fields of a hash, decoded. getAllFromMap ::   (Functor f, RedisCtx m f, Serialize k, Serialize v) =>   ByteString ->@@ -699,40 +817,57 @@         let v = fromRight undefined $ decode v'       ] +-- | Messages handled by the cache writer. data CacheWriterMessage-  = CacheNewBlock-  | CacheNewTx !TxHash+  = -- | a new block was imported+    CacheNewBlock+  | -- | a mempool transaction was added or removed+    CacheNewTx !TxHash   | -- | internal: a batch of coalesced 'CacheNewTx' events     CacheNewTxs ![TxHash]-  | CacheSyncMempool !(Listen ())+  | -- | sync the mempool with the database, then reply+    CacheSyncMempool !(Listen ()) +-- | Inbox of the cache writer process. type CacheWriterInbox = Inbox CacheWriterMessage +-- | Mailbox to send messages to the cache writer. type CacheWriter = Mailbox CacheWriterMessage +-- | Xpub that a cached address belongs to. data AddressXPub = AddressXPub-  { spec :: !XPubSpec,+  { -- | the xpub+    spec :: !XPubSpec,+    -- | derivation path of the address below the xpub     path :: ![KeyIndex]   }   deriving (Show, Eq, Generic, NFData, Serialize) +-- | Key of the sorted set of mempool transactions. mempoolSetKey :: ByteString mempoolSetKey = "mempool" +-- | Prefix of the keys of cached addresses. addrPfx :: ByteString addrPfx = "a" +-- | Key of the last block processed by the cache. bestBlockKey :: ByteString bestBlockKey = "head" +-- | Key of the sorted set of cached xpubs by time of last use. maxKey :: ByteString maxKey = "max" +-- | Address of an extended public key for a type of address. xPubAddrFunction :: Ctx -> DeriveType -> XPubKey -> Address xPubAddrFunction ctx DeriveNormal = xPubAddr ctx xPubAddrFunction ctx DeriveP2SH = xPubCompatWitnessAddr ctx xPubAddrFunction ctx DeriveP2WPKH = xPubWitnessAddr ctx +-- | Run the cache writer process: bring the cache up to date with the database,+-- then handle messages forever. Notifications of new mempool transactions that+-- pile up are handled together, up to 1000 at a time. cacheWriter ::   (MonadUnliftIO m, StoreReadExtra m) =>   CacheConfig ->@@ -768,6 +903,7 @@                   requeueSTM other inbox                   return (reverse acc) +-- | Try to take a lock that expires after five minutes. Returns 'True' if taken. lockIt :: (MonadIO m) => ByteString -> CacheX m Bool lockIt l = do   go >>= \case@@ -807,6 +943,7 @@                 }         Redis.setOpts l "locked" opts +-- | Extend a lock for another five minutes. refreshLock :: (MonadIO m) => ByteString -> CacheX m () refreshLock l = void . runRedis $ do   let opts =@@ -820,9 +957,11 @@           }   Redis.setOpts l "locked" opts +-- | Release a lock. unlockIt :: (MonadIO m) => ByteString -> CacheX m () unlockIt l = void . runRedis . Redis.del $ NonEmpty.singleton l +-- | Run an action holding a lock, refreshing it every two and a half minutes. Returns 'Nothing' if the lock is taken. withLock ::   (MonadUnliftIO m) =>   ByteString ->@@ -861,6 +1000,7 @@               threadDelay n               go (waited + n) (min max_delay (n * 2)) +-- | Evict the least recently used xpubs while Redis holds more than the maximum number of keys. pruneDB ::   (MonadUnliftIO m, StoreReadBase m) => CacheX m () pruneDB = void . withLock "lock" $ do@@ -871,16 +1011,19 @@     void . delXPubKeys . map fst       =<< runRedis (getFromSortedSet maxKey Nothing 0 (fromIntegral (s - x))) +-- | Record that xpubs were just used. touchKeys :: (MonadIO m) => [XPubSpec] -> CacheX m () touchKeys xpubs = do   now <- round <$> liftIO getPOSIXTime :: (MonadIO m) => m Int   runRedis $ redisTouchKeys now xpubs +-- | Redis command to record that xpubs were used at a time. redisTouchKeys :: (Monad f, RedisCtx m f, Real a) => a -> [XPubSpec] -> m (f ()) redisTouchKeys _ [] = return $ return () redisTouchKeys now xpubs =   void <$> Redis.zadd maxKey (map ((realToFrac now,) . encode) xpubs) +-- | Handle a message sent to the cache writer. cacheWriterReact ::   (MonadUnliftIO m, StoreReadExtra m) =>   CacheWriterMessage ->@@ -904,9 +1047,12 @@   syncMempoolC   atomically $ l () +-- | Number of used addresses in a list of xpub balances. lenNotNull :: [XPubBal] -> Int lenNotNull = length . filter (not . nullBalance . (.balance)) +-- | Cache an xpub, if it has enough used addresses, the store is in sync, and no+-- other thread is caching it already. newXPubC ::   (MonadUnliftIO m, StoreReadExtra m) =>   XPubSpec ->@@ -971,6 +1117,7 @@       conn <- asks (.redis)       liftIO (Redis.runRedis conn red) <&> isRight +-- | Whether the best block of the store is the best header of the node. inSync ::   (MonadUnliftIO m, StoreReadExtra m) =>   CacheX m Bool@@ -982,6 +1129,8 @@       cb <- liftIO $ chainGetBest ch       return $ cb.height > 0 && headerHash cb.header == bb +-- | Bring the cache up to date with the best block of the store, processing every+-- block since the last one it processed, across reorganisations. newBlockC ::   (MonadUnliftIO m, StoreReadExtra m) =>   CacheX m ()@@ -1025,6 +1174,7 @@           let blocks = reverse left_parents <> right_parents <> pure right_node           return $ map (headerHash . (.header)) blocks +-- | Update cached xpubs with the transactions of a block. importBlockC ::   (MonadUnliftIO m, StoreReadExtra m) =>   BlockHash ->@@ -1054,6 +1204,9 @@         "Could not get block: "           <> blockHashToHex bh +-- | Update cached xpubs with transactions: their transaction sets, unspent+-- outputs, balances and the mempool. Addresses past the gap are derived and+-- added when new addresses get used. importMultiTxC ::   (MonadUnliftIO m, StoreReadExtra m) =>   [TxData] ->@@ -1139,6 +1292,7 @@     allxpubs =       HashSet.fromList . map (.spec) . HashMap.elems +-- | Redis commands to add or remove transactions and unspent outputs of cached xpubs. redisImportMultiTx ::   (Monad f, RedisCtx m f) =>   Network ->@@ -1203,6 +1357,7 @@           return $ a >> b >> return ()     txaddrops td = txInputs net ctx td <> txOutputs net ctx td +-- | Redis commands to update the balances of cached addresses. redisUpdateBalances ::   (Monad f, RedisCtx m f) =>   HashMap Address AddressXPub ->@@ -1218,6 +1373,8 @@     xpubbal ainfo bal =       XPubBal {path = ainfo.path, balance = bal} +-- | Add newly derived addresses to their cached xpubs, with their balances,+-- transactions and unspent outputs, and keep deriving while they are used. cacheAddAddresses ::   (StoreReadExtra m, MonadUnliftIO m) =>   [(Address, AddressXPub)] ->@@ -1285,6 +1442,7 @@       let f ts = (i.spec, ts)        in f <$> lift (getAddressTxs a def) +-- | Addresses to derive and add to cached xpubs to keep the gap after the used ones. getNewAddrs ::   Ctx ->   KeyIndex ->@@ -1297,6 +1455,7 @@       Nothing -> []       Just bals -> addrsToAdd ctx gap bals a +-- | Update cached xpubs with new or deleted transactions, if in sync. syncNewTxC ::   (MonadUnliftIO m, StoreReadExtra m) =>   [TxHash] ->@@ -1310,6 +1469,7 @@           "Synchronizing transaction: " <> txHashToHex (txHash tx.tx)       importMultiTxC txs +-- | Update cached xpubs with every mempool difference between the cache and the store, if in sync. syncMempoolC ::   (MonadUnliftIO m, StoreReadExtra m) =>   CacheX m ()@@ -1327,12 +1487,15 @@         "Synchronizing " <> cs (show (length txs)) <> " mempool transactions"       importMultiTxC txs +-- | Mempool as seen by the cache, newest first. cacheGetMempool :: (MonadIO m) => CacheX m [(UnixTime, TxHash)] cacheGetMempool = runRedis redisGetMempool +-- | Last block processed by the cache. cacheGetHead :: (MonadIO m) => CacheX m (Maybe BlockHash) cacheGetHead = runRedis redisGetHead +-- | Set the last block processed by the cache. cacheSetHead :: (MonadIO m) => BlockData -> CacheX m () cacheSetHead bd = do   logDebug $ "Cache head set to: " <> blockHashToHex bh@@ -1341,10 +1504,12 @@   where     bh = headerHash bd.header +-- | Xpub and path of each address, or 'Nothing' if the address is not cached. cacheGetAddrsInfo ::   (MonadIO m) => [Address] -> CacheX m [Maybe AddressXPub] cacheGetAddrsInfo as = runRedis (redisGetAddrsInfo as) +-- | Redis command to add transactions to the mempool set. redisAddToMempool :: (Applicative f, RedisCtx m f) => [TxRef] -> m (f Integer) redisAddToMempool [] = return (pure 0) redisAddToMempool btxs =@@ -1353,6 +1518,7 @@       (\btx -> (blockRefScore btx.block, encode btx.txid))       btxs +-- | Redis command to remove transactions from the mempool set. redisRemFromMempool ::   (Applicative f, RedisCtx m f) => [TxHash] -> m (f Integer) redisRemFromMempool xs =@@ -1360,10 +1526,12 @@     Nothing -> return (pure 0)     Just xs' -> zrem mempoolSetKey $ NonEmpty.map encode xs' +-- | Redis command to record the xpub and path of an address. redisSetAddrInfo ::   (Functor f, RedisCtx m f) => Address -> AddressXPub -> m (f ()) redisSetAddrInfo a i = void <$> Redis.set (addrPfx <> encode a) (encode i) +-- | Remove xpubs from the cache. Returns the number of keys deleted. cacheDelXPubs ::   (MonadIO m, StoreReadBase m) =>   [XPubSpec] ->@@ -1372,6 +1540,7 @@   Just cache -> runReaderT (delXPubKeys xpubs) cache   Nothing -> return 0 +-- | Remove xpubs from the cache. Returns the number of keys deleted. delXPubKeys ::   (MonadIO m, StoreReadBase m) =>   [XPubSpec] ->@@ -1387,6 +1556,7 @@       return $ (xpub,) <$> bs   runRedis $ fmap sum . sequence <$> forM xbals (uncurry redisDelXPubKeys) +-- | Redis commands to remove an xpub and its addresses from the cache. redisDelXPubKeys ::   (Monad f, RedisCtx m f) => XPubSpec -> [XPubBal] -> m (f Integer) redisDelXPubKeys xpub bals = go $ map (.balance.address) bals@@ -1408,6 +1578,7 @@         bal' <- balcount         return $ addrs' + txset' + utxo' + bal' +-- | Redis command to add transactions to an xpub. redisAddXPubTxs ::   (Applicative f, RedisCtx m f) => XPubSpec -> [TxRef] -> m (f Integer) redisAddXPubTxs _ [] = return (pure 0)@@ -1415,6 +1586,7 @@   zadd (txSetPfx <> encode xpub) $     map (\t -> (blockRefScore t.block, encode t.txid)) btxs +-- | Redis command to remove transactions from an xpub. redisRemXPubTxs ::   (Applicative f, RedisCtx m f) => XPubSpec -> [TxHash] -> m (f Integer) redisRemXPubTxs xpub txhs =@@ -1422,6 +1594,7 @@     Nothing -> return (pure 0)     Just txhs' -> zrem (txSetPfx <> encode xpub) (NonEmpty.map encode txhs') +-- | Redis command to add unspent outputs to an xpub. redisAddXPubUnspents ::   (Applicative f, RedisCtx m f) =>   XPubSpec ->@@ -1433,6 +1606,7 @@   zadd (utxoPfx <> encode xpub) $     map (\(p, r) -> (blockRefScore r, encode p)) utxo +-- | Redis command to remove unspent outputs from an xpub. redisRemXPubUnspents ::   (Applicative f, RedisCtx m f) => XPubSpec -> [OutPoint] -> m (f Integer) redisRemXPubUnspents _ [] =@@ -1442,6 +1616,7 @@     Nothing -> return (pure 0)     Just ops' -> zrem (utxoPfx <> encode xpub) (NonEmpty.map encode ops') +-- | Redis commands to set balances of an xpub and record its addresses. redisAddXPubBalances ::   (Monad f, RedisCtx m f) => XPubSpec -> [XPubBal] -> m (f ()) redisAddXPubBalances xpub bals =@@ -1459,9 +1634,11 @@             }       return $ sequence_ ys +-- | Redis command to set the last block processed by the cache. redisSetHead :: (RedisCtx m f) => BlockHash -> m (f Redis.Status) redisSetHead bh = Redis.set bestBlockKey (encode bh) +-- | Redis commands to read the xpub and path of addresses. redisGetAddrsInfo ::   (Monad f, RedisCtx m f) => [Address] -> m (f [Maybe AddressXPub]) redisGetAddrsInfo [] = return (pure [])@@ -1471,6 +1648,8 @@     is' <- sequence is     return $ map (eitherToMaybe . decode =<<) is' +-- | Addresses to derive after a newly used address, so that the gap of unused+-- addresses after the last used one on its chain is kept. addrsToAdd ::   Ctx ->   KeyIndex ->@@ -1506,12 +1685,14 @@     changeaddrs = map xpubf (keys changepaths)     changelist = map pathToList changepaths +-- | Sort transactions so that each comes after those it spends. sortTxData :: [TxData] -> [TxData] sortTxData tds =   let txm = Map.fromList (map (\d -> (txHash d.tx, d)) tds)       ths = map (txHash . snd) (sortTxs (map (.tx) tds))    in mapMaybe (`Map.lookup` txm) ths +-- | Addresses and outpoints of the outputs spent by a transaction. txInputs :: Network -> Ctx -> TxData -> [(Address, OutPoint)] txInputs net ctx td =   let is = td.tx.inputs@@ -1521,6 +1702,7 @@       f (Left _) _ = Nothing    in catMaybes (zipWith f as is) +-- | Addresses and outpoints of the outputs of a transaction. txOutputs :: Network -> Ctx -> TxData -> [(Address, OutPoint)] txOutputs net ctx td =   let ps =@@ -1538,11 +1720,13 @@       f (Left _) _ = Nothing    in catMaybes (zipWith f as ps) +-- | Redis command to read the last block processed by the cache. redisGetHead :: (Functor f, RedisCtx m f) => m (f (Maybe BlockHash)) redisGetHead = do   x <- Redis.get bestBlockKey   return $ (eitherToMaybe . decode =<<) <$> x +-- | Redis command to read the mempool set, newest first. redisGetMempool :: (Applicative f, RedisCtx m f) => m (f [(UnixTime, TxHash)]) redisGetMempool = do   xs <- getFromSortedSet mempoolSetKey Nothing 0 0@@ -1550,6 +1734,7 @@   where     f t s = ((scoreBlockRef s).timestamp, t) +-- | Text form of an xpub for logs, with a suffix for its address type. xpubText ::   (StoreReadBase m) =>   XPubSpec ->@@ -1563,11 +1748,14 @@   ctx <- lift getCtx   return . cs $ suffix <> xPubExport net ctx xpub.key +-- | Tell the cache writer that a new block was imported. cacheNewBlock :: (MonadIO m) => CacheWriter -> m () cacheNewBlock = send CacheNewBlock +-- | Tell the cache writer that a mempool transaction was added or removed. cacheNewTx :: (MonadIO m) => TxHash -> CacheWriter -> m () cacheNewTx = send . CacheNewTx +-- | Ask the cache writer to sync the mempool, and wait until it is done. cacheSyncMempool :: (MonadIO m) => CacheWriter -> m () cacheSyncMempool = query CacheSyncMempool

@@ -8,6 +8,39 @@ {-# LANGUAGE TupleSections #-} {-# LANGUAGE NoFieldSelectors #-} +-- |+-- Module      : Haskoin.Store.Common+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- Interfaces shared by every part of the store.+--+-- 'StoreReadBase' and 'StoreReadExtra' abstract over anything that can answer+-- queries about blocks, transactions, addresses and extended public keys: the+-- RocksDB database ("Haskoin.Store.Database.Reader"), the in-memory write+-- buffer used while importing ("Haskoin.Store.Database.Writer"), and the Redis+-- cache ("Haskoin.Store.Cache"). 'StoreWrite' is implemented by the writer.+--+-- This module also has helpers for pagination ('Limits'), for deriving+-- addresses from extended public keys, and for merging sorted streams.+-- = Why the store is a set of type classes+--+-- The same questions are asked in three places: by the web server against the+-- database, by the import logic against a buffer of uncommitted changes, and by+-- the cache, which answers some questions itself and passes others on. Writing+-- the queries once against a class, and providing instances for each place, keeps+-- those three in agreement and lets the cache sit transparently in front of the+-- database.+--+-- = Why pages start at a transaction+--+-- Paging by offset alone gets slower the deeper you go, because the store must+-- read and drop every result before the page, and it shifts when new+-- transactions arrive. 'Start' lets a client continue from the last transaction+-- it saw: the store seeks straight to it, and new arrivals do not move the page. module Haskoin.Store.Common   ( Limits (..),     Start (..),@@ -93,36 +126,61 @@   ) import UnliftIO (MonadIO, liftIO) +-- | Function to derive the address at an index of an extended public key. type DeriveAddr = XPubKey -> KeyIndex -> Address +-- | Number of items to skip. type Offset = Word32 +-- | Maximum number of items to return; @0@ means no limit. type Limit = Word32 +-- | Where a page of results ordered from newest to oldest starts. data Start-  = AtTx !TxHash-  | AtBlock !BlockHeight+  = -- | at this transaction; for an unconfirmed transaction, at the time it+    -- entered the mempool, and for a confirmed one, at its position in the+    -- chain+    AtTx !TxHash+  | -- | at this block height, including every transaction in the mempool if it+    -- is above the best block+    AtBlock !BlockHeight   deriving (Eq, Show) +-- | Pagination of a query. Results are ordered from newest to oldest. The+-- 'start' is applied first, then the 'offset', then the 'limit'. data Limits = Limits-  { limit :: !Word32,+  { -- | maximum number of results; @0@ for no limit+    limit :: !Word32,+    -- | number of results to skip after the start     offset :: !Word32,+    -- | where to start, or from the newest result if 'Nothing'     start :: !(Maybe Start)   }   deriving (Eq, Show) +-- | No limit, no offset, start from the newest result. defaultLimits :: Limits defaultLimits = Limits {limit = 0, offset = 0, start = Nothing}  instance Default Limits where   def = defaultLimits +-- | Basic queries that every store answers. class (Monad m) => StoreReadBase m where+  -- | Network the store indexes.   getNetwork :: m Network+  -- | Context for secp256k1 operations, needed to parse and derive keys.   getCtx :: m Ctx+  -- | Hash of the best block imported, or 'Nothing' if none was imported yet.   getBestBlock :: m (Maybe BlockHash)+  -- | Hashes of all blocks imported at a height, including those on side+  -- chains.   getBlocksAtHeight :: BlockHeight -> m [BlockHash]+  -- | Data of a block imported in the past, whether still in the main chain+  -- or not. Check 'Haskoin.Store.Data.main' to know.   getBlock :: BlockHash -> m (Maybe BlockData)+  -- | Data of a transaction, including deleted ones. Check+  -- 'Haskoin.Store.Data.deleted' to know.   getTxData :: TxHash -> m (Maybe TxData)    -- | Data of many transactions, in the order of their hashes. Stores that@@ -130,8 +188,11 @@   getManyTxData :: [TxHash] -> m [Maybe TxData]   getManyTxData = mapM getTxData +  -- | Transaction input that spends an output, if it is spent.   getSpender :: OutPoint -> m (Maybe Spender)+  -- | Balance of an address, or 'Nothing' if it was never used.   getBalance :: Address -> m (Maybe Balance)+  -- | Unspent output, or 'Nothing' if it does not exist or is spent.   getUnspent :: OutPoint -> m (Maybe Unspent)   -- | Whole mempool, newest first.   getMempool :: m [(UnixTime, TxHash)]@@ -142,61 +203,110 @@   getMempoolPage :: Limits -> m [(UnixTime, TxHash)]   getMempoolPage l = applyLimits l <$> getMempool +-- | Queries about addresses and extended public keys, which need indices or+-- configuration beyond the basic store. class (StoreReadBase m) => StoreReadExtra m where+  -- | Transactions involving any of the addresses, newest first, without+  -- duplicates.   getAddressesTxs :: [Address] -> Limits -> m [TxRef]+  -- | Unspent outputs of any of the addresses, newest first.   getAddressesUnspents :: [Address] -> Limits -> m [Unspent]+  -- | Number of unused addresses to look ahead when deriving the addresses of+  -- an extended public key that has no transactions yet.   getInitialGap :: m Word32+  -- | Number of consecutive unused addresses after which derivation of an+  -- extended public key stops.   getMaxGap :: m Word32+  -- | Transactions whose hash, read as a number, matches the short identifier+  -- used by the Blockchain.info compatibility API. There may be more than one.   getNumTxData :: Word64 -> m [TxData]+  -- | Balances of the addresses, in the same order. Addresses never used get+  -- a zero balance.   getBalances :: [Address] -> m [Balance]+  -- | Transactions involving an address, newest first.   getAddressTxs :: Address -> Limits -> m [TxRef]+  -- | Unspent outputs of an address, newest first.   getAddressUnspents :: Address -> Limits -> m [Unspent]+  -- | Balances of the used addresses of an extended public key, with their+  -- derivation paths.   xPubBals :: XPubSpec -> m [XPubBal]+  -- | Unspent outputs of an extended public key, newest first. The second+  -- argument is the result of 'xPubBals' for the same key.   xPubUnspents :: XPubSpec -> [XPubBal] -> Limits -> m [XPubUnspent]+  -- | Transactions of an extended public key, newest first. The second+  -- argument is the result of 'xPubBals' for the same key.   xPubTxs :: XPubSpec -> [XPubBal] -> Limits -> m [TxRef]+  -- | Number of transactions of an extended public key. The second argument+  -- is the result of 'xPubBals' for the same key.   xPubTxCount :: XPubSpec -> [XPubBal] -> m Word32 +-- | Changes to the index. Implemented by the writer, which buffers them in+-- memory and commits them to the database in one batch. class StoreWrite m where+  -- | Set the best block.   setBest :: BlockHash -> m ()+  -- | Add or replace the data of a block.   insertBlock :: BlockData -> m ()+  -- | Set the hashes of the blocks at a height.   setBlocksAtHeight :: [BlockHash] -> BlockHeight -> m ()+  -- | Add or replace the data of a transaction.   insertTx :: TxData -> m ()+  -- | Add a transaction to the index of an address.   insertAddrTx :: Address -> TxRef -> m ()+  -- | Remove a transaction from the index of an address.   deleteAddrTx :: Address -> TxRef -> m ()+  -- | Add an unspent output to the index of an address.   insertAddrUnspent :: Address -> Unspent -> m ()+  -- | Remove an unspent output from the index of an address.   deleteAddrUnspent :: Address -> Unspent -> m ()+  -- | Add a transaction to the mempool, with the time it arrived.   addToMempool :: TxHash -> UnixTime -> m ()+  -- | Remove a transaction from the mempool.   deleteFromMempool :: TxHash -> m ()+  -- | Set the balance of an address.   setBalance :: Balance -> m ()+  -- | Add an unspent output.   insertUnspent :: Unspent -> m ()+  -- | Remove an unspent output.   deleteUnspent :: OutPoint -> m () +-- | Data of a block, or 'Nothing' if it is unknown or not in the main chain. getActiveBlock :: (StoreReadExtra m) => BlockHash -> m (Maybe BlockData) getActiveBlock bh =   getBlock bh >>= \case     Just b | b.main -> return (Just b)     _ -> return Nothing +-- | Data of a transaction, or 'Nothing' if it is unknown or was deleted. getActiveTxData :: (StoreReadBase m) => TxHash -> m (Maybe TxData) getActiveTxData th =   getTxData th >>= \case     Just td | not td.deleted -> return (Just td)     _ -> return Nothing +-- | Balance of an address, or a zero balance if the address was never used. getDefaultBalance :: (StoreReadBase m) => Address -> m Balance getDefaultBalance a =   getBalance a >>= \case     Nothing -> return $ zeroBalance a     Just b -> return b +-- | Infinite list of the addresses of an extended public key, with their+-- indices, from the given index on. deriveAddresses :: DeriveAddr -> XPubKey -> Word32 -> [(Word32, Address)] deriveAddresses derive xpub start = map (\i -> (i, derive xpub i)) [start ..] +-- | Address derivation function for a type of address: legacy pay-to-public-key+-- hash, pay-to-witness-public-key-hash nested in pay-to-script-hash, or native+-- pay-to-witness-public-key-hash. deriveFunction :: Ctx -> DeriveType -> DeriveAddr deriveFunction ctx DeriveNormal i = fst . deriveAddr ctx i deriveFunction ctx DeriveP2SH i = fst . deriveCompatWitnessAddr ctx i deriveFunction ctx DeriveP2WPKH i = fst . deriveWitnessAddr ctx i +-- | Summary of the balances of an extended public key, from the balances of its+-- addresses. The received total only counts external addresses, and the+-- indices are those of the last used external and change addresses. xPubSummary :: XPubSpec -> [XPubBal] -> XPubSummary xPubSummary _xspec xbals =   XPubSummary@@ -215,6 +325,7 @@     xt = [b | b@XPubBal {path = [0, _]} <- bs]     rx = sum [b.balance.received | b <- xt] +-- | Transaction with its metadata, or 'Nothing' if it is unknown. getTransaction ::   (StoreReadBase m) => TxHash -> m (Maybe Transaction) getTransaction h = do@@ -230,6 +341,7 @@   ctx <- getCtx   map (fmap (toTransaction net ctx)) <$> getManyTxData hs +-- | Transactions that match a short numeric identifier; see 'getNumTxData'. getNumTransaction ::   (StoreReadExtra m) => Word64 -> m [Transaction] getNumTransaction i = do@@ -237,6 +349,8 @@   ctx <- getCtx   map (toTransaction net ctx) <$> getNumTxData i +-- | First block in the main chain whose timestamp is at or after the given time,+-- or 'Nothing' if there is none. blockAtOrAfter ::   (MonadIO m, StoreReadExtra m) =>   Chain ->@@ -251,6 +365,8 @@       let t = fromIntegral x.header.timestamp        in return $ t `compare` q +-- | Last block in the main chain whose timestamp is at or before the given time,+-- or 'Nothing' if there is none. blockAtOrBefore ::   (MonadIO m, StoreReadExtra m) =>   Chain ->@@ -265,6 +381,8 @@       let t = fromIntegral x.header.timestamp        in return $ t `compare` q +-- | First block in the main chain whose median time past is at or after the given+-- time, or 'Nothing' if there is none. blockAtOrAfterMTP ::   (MonadIO m, StoreReadExtra m) =>   Chain ->@@ -281,40 +399,64 @@  -- | Events that the store can generate. data StoreEvent-  = StoreBestBlock !BlockHash-  | StoreMempoolNew !TxHash-  | StoreMempoolDelete !TxHash-  | StorePeerConnected !Peer-  | StorePeerDisconnected !Peer-  | StorePeerPong !Peer !Word64-  | StoreTxAnnounce !Peer ![TxHash]-  | StoreTxReject !Peer !TxHash !RejectCode !ByteString+  = -- | a block was imported and is now the best block+    StoreBestBlock !BlockHash+  | -- | a transaction entered the mempool+    StoreMempoolNew !TxHash+  | -- | a transaction left the mempool without being confirmed, for example+    -- because it was replaced or a conflicting transaction was confirmed+    StoreMempoolDelete !TxHash+  | -- | a peer connected+    StorePeerConnected !Peer+  | -- | a peer disconnected+    StorePeerDisconnected !Peer+  | -- | a peer answered a ping with this nonce+    StorePeerPong !Peer !Word64+  | -- | a peer announced transactions+    StoreTxAnnounce !Peer ![TxHash]+  | -- | a peer rejected a transaction, with a code and a reason+    StoreTxReject !Peer !TxHash !RejectCode !ByteString +-- | Apply the offset and limit of 'Limits' to a list. The start is ignored. applyLimits :: Limits -> [a] -> [a] applyLimits Limits {..} = applyLimit limit . applyOffset offset +-- | Drop the first items of a list. applyOffset :: Offset -> [a] -> [a] applyOffset = drop . fromIntegral +-- | Keep the first items of a list; @0@ keeps them all. applyLimit :: Limit -> [a] -> [a] applyLimit 0 = id applyLimit l = take (fromIntegral l) +-- | Fold the offset into the limit, so that a query returns enough results to+-- apply the offset afterwards, for example when merging results from several+-- sources.+--+-- When results from several sources are merged, the offset can only be applied+-- to the merged stream. Each source must therefore return the first+-- @offset + limit@ results. deOffset :: Limits -> Limits deOffset l = case l.limit of   0 -> l {offset = 0}   _ -> l {limit = l.limit + l.offset, offset = 0} +-- | Conduit version of 'applyLimits'. applyLimitsC :: (Monad m) => Limits -> ConduitT i i m () applyLimitsC Limits {..} = applyOffsetC offset >> applyLimitC limit +-- | Conduit version of 'applyOffset'. applyOffsetC :: (Monad m) => Offset -> ConduitT i i m () applyOffsetC = dropC . fromIntegral +-- | Conduit version of 'applyLimit'. applyLimitC :: (Monad m) => Limit -> ConduitT i i m () applyLimitC 0 = mapC id applyLimitC l = takeC (fromIntegral l) +-- | Sort transactions so that every transaction comes after those it spends.+-- Each transaction is returned with its position in the original list. sortTxs :: [Tx] -> [(Word32, Tx)] sortTxs txs = ts   where@@ -332,6 +474,7 @@       let f (tx, i, _) = (fromIntegral i, tx)        in map (f . n) vs +-- | Remove duplicates from a list. The order of the result is unspecified. nub' :: (Hashable a) => [a] -> [a] nub' = H.toList . H.fromList @@ -347,6 +490,7 @@     go = action >>= \_ -> go {-# INLINE loopForever #-} +-- | Current time in microseconds since the Unix epoch. microseconds :: (MonadIO m) => m Integer microseconds =   let f t =@@ -354,6 +498,15 @@           + toInteger (systemNanoseconds t) `div` 1000    in liftIO $ f <$> getSystemTime +-- | Stream every result of a paged query, fetching one page at a time.+--+-- If a function to get the transaction of a result is given, each page starts+-- at the last transaction of the previous one. Otherwise pages are fetched by+-- offset, doubling their size each time up to 1024 results.+--+-- This is how endpoints that combine many sources, such as the transactions of+-- every address of an xpub, avoid loading everything: each source is read page by+-- page, only as far as the merged output needs. streamThings ::   (Monad m) =>   (Limits -> m [a]) ->@@ -383,6 +536,8 @@           [] -> return ()           ls -> mapM_ yield ls >> go l' (last ls) +-- | Merge streams sorted in decreasing order into one stream sorted in decreasing+-- order, dropping duplicates. joinDescStreams ::   (Monad m, Ord a) =>   [ConduitT () a m ()] ->

@@ -9,6 +9,37 @@ {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE NoFieldSelectors #-} +-- |+-- Module      : Haskoin.Store.Database.Reader+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- Queries against the RocksDB database of the store.+--+-- 'DatabaseReaderT' implements 'StoreReadBase' and 'StoreReadExtra' by reading+-- the indices directly from RocksDB. The database is opened with+-- 'withDatabaseReader', which also creates or migrates it.+--+-- Data is split across column families by kind; see 'addrTxCF' and the other+-- column family accessors. The keys stored in each are described in+-- "Haskoin.Store.Database.Types".+-- = Why column families+--+-- Each kind of record lives in its own column family. Their keys and access+-- patterns differ: balances are small point lookups, address transactions are+-- scans, transactions are large values. Separate column families let each have+-- its own key prefix length for filters and seeks, and keep scans over one kind+-- of data from wading through another.+--+-- = Why addresses of xpubs are cached+--+-- Deriving an address from an extended public key costs an elliptic curve+-- operation and two hashes, and an xpub query may need hundreds of them before it+-- reads a single balance. Addresses depend only on the xpub, so the derivations+-- of recently queried xpubs are kept and shared. The cache never goes stale. module Haskoin.Store.Database.Reader   ( -- * RocksDB Database Access     DatabaseReader (..),@@ -59,14 +90,22 @@ import Text.Read (readMaybe) import UnliftIO +-- | Monad transformer to query the database. type DatabaseReaderT = ReaderT DatabaseReader +-- | Open database and its settings. data DatabaseReader = DatabaseReader-  { db :: !DB,+  { -- | RocksDB handle, with the column families of the store+    db :: !DB,+    -- | consecutive unused addresses after which xpub derivation stops     maxGap :: !Word32,+    -- | unused addresses to look at for an xpub with no transactions     initGap :: !Word32,+    -- | network constants     net :: !Network,+    -- | StatsD metrics, if enabled     metrics :: !(Maybe DataMetrics),+    -- | secp256k1 context     ctx :: !Ctx,     -- | addresses derived from recently queried xpubs     derived :: !(IORef DerivedCache)@@ -87,6 +126,7 @@     older :: !(HashMap XPubSpec Derived)   } +-- | Number of xpubs in each generation of the cache of derived addresses. derivedCacheSize :: Int derivedCacheSize = 512 @@ -113,21 +153,35 @@               | otherwise = c {newer = HashMap.insert xpub d c.newer}          in (c', d) +-- | StatsD metrics of database reads. Counters count the items read. data DataMetrics = DataMetrics-  { dataBestCount :: !StatCounter,+  { -- | best block reads (@db.best_block@)+    dataBestCount :: !StatCounter,+    -- | blocks read (@db.blocks@)     dataBlockCount :: !StatCounter,+    -- | transactions read (@db.txs@)     dataTxCount :: !StatCounter,+    -- | mempool entries read (@db.mempool@)     dataMempoolCount :: !StatCounter,+    -- | balances read (@db.balances@)     dataBalanceCount :: !StatCounter,+    -- | unspent outputs read (@db.unspents@)     dataUnspentCount :: !StatCounter,+    -- | address transactions read (@db.address_txs@)     dataAddrTxCount :: !StatCounter,+    -- | xpub balances read (@db.xpub_balances@)     dataXPubBals :: !StatCounter,+    -- | xpub unspent outputs read (@db.xpub_unspents@)     dataXPubUnspents :: !StatCounter,+    -- | xpub transactions read (@db.xpub_txs@)     dataXPubTxs :: !StatCounter,+    -- | xpub transaction counts (@db.xpub_tx_count@)     dataXPubTxCount :: !StatCounter,+    -- | open iterators (@db.iterators@)     dataIters :: !StatGauge   } +-- | Register the database metrics with a StatsD client. Samples are sent at a rate of one in ten. createDataMetrics :: Stats -> IO DataMetrics createDataMetrics s = do   dataBestCount <- newStatCounter s "db.best_block" n@@ -146,15 +200,31 @@   where     n = 10 +-- | Run an action with the metrics, if they are enabled. withMetrics :: (MonadReader DatabaseReader m) => (DataMetrics -> m a) -> m () withMetrics go = asks (.metrics) >>= mapM_ go +-- | Version of the database layout. Databases of versions 18 and 19 are+-- migrated when opened; other versions are refused.+--+-- Bumping the version is how a change to the database layout is made safe: an+-- old database is either migrated in place when it is opened or refused, never+-- read with the wrong layout. dataVersion :: Word32 dataVersion = 20 +-- | Run a query against a database. withDB :: DatabaseReader -> DatabaseReaderT m a -> m a withDB = flip runReaderT +-- | Open the database and run a query against it. Creates the database and its+-- column families if missing, and migrates it from older versions.+--+-- Arguments, in order: network, secp256k1 context, initial xpub gap, maximum+-- xpub gap, directory, whether to use Bloom filters, maximum open files, total+-- memtable size in bytes, write-ahead log size in bytes after which column+-- families are flushed, block cache size in bytes, and metrics. 'Nothing' leaves+-- a setting to RocksDB. withDatabaseReader ::   (MonadUnliftIO m) =>   Network ->@@ -247,30 +317,39 @@   where     c p = def {prefixLength = p, bloomFilter = bloom, cacheIndexAndFilterBlocks = cached} +-- | Column family for transactions by address ('AddrTxKey'). addrTxCF :: DB -> ColumnFamily addrTxCF db = columnFamilies db !! 0 +-- | Column family for unspent outputs by address ('AddrOutKey'). addrOutCF :: DB -> ColumnFamily addrOutCF db = columnFamilies db !! 1 +-- | Column family for transactions ('TxKey'). txCF :: DB -> ColumnFamily txCF db = columnFamilies db !! 2 +-- | Column family for unspent outputs ('UnspentKey'). unspentCF :: DB -> ColumnFamily unspentCF db = columnFamilies db !! 4 +-- | Column family for blocks ('BlockKey'). blockCF :: DB -> ColumnFamily blockCF db = columnFamilies db !! 5 +-- | Column family for block hashes by height ('HeightKey'). heightCF :: DB -> ColumnFamily heightCF db = columnFamilies db !! 6 +-- | Column family for address balances ('BalKey'). balanceCF :: DB -> ColumnFamily balanceCF db = columnFamilies db !! 7 +-- | Column family for mempool transactions ('MemTxKey' and 'MemTimeKey'). mempoolCF :: DB -> ColumnFamily mempoolCF db = columnFamilies db !! 8 +-- | Set the version of a new database, migrate an old one, or call 'error' if the version is not supported. initRocksDB :: DatabaseReader -> IO () initRocksDB DatabaseReader {db = db} =   retrieve db VersionKey >>= \case@@ -350,9 +429,12 @@     entry (MemTimeKey t h, ()) = Just (t, h)     entry (MemTimeKeyS, ()) = Nothing +-- | Set the version of a new database. setInitRocksDB :: DB -> IO () setInitRocksDB db = insert db VersionKey dataVersion +-- | Stream the transactions of an address, newest first, from a starting point.+-- If the start is a transaction that is not found, nothing is streamed. addressConduit ::   (MonadIO m) =>   Address ->@@ -395,6 +477,8 @@                       .| (dropWhileC (cond . fst) >> mapC id)           Nothing -> return () +-- | Stream the unspent outputs of an address, newest first, from a starting+-- point. If the start is a transaction that is not found, nothing is streamed. unspentConduit ::   (MonadIO m) =>   Network ->@@ -432,6 +516,7 @@                       .| (dropWhileC (cond . fst) >> mapC id)           Nothing -> return () +-- | Run an action with an iterator on a column family, counting it in the metrics. withIterCFStat ::   (MonadUnliftIO m) =>   DB ->@@ -486,6 +571,8 @@ retrieveManyCF db cf keys =   map (fmap (either error id . decode)) <$> multiGetCF db cf (map encode keys) +-- | Merge streams sorted newest first, drop duplicates, and apply the offset and+-- limit. joinConduits ::   (Monad m, Ord o) =>   [ConduitT () o m ()] ->

@@ -8,6 +8,49 @@ {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE NoFieldSelectors #-} +{- |+Module      : Haskoin.Store.Database.Types+Copyright   : (c) Haskoin Developers+License     : MIT+Maintainer  : jprupp@protonmail.ch+Stability   : experimental+Portability : POSIX++Keys and values stored in RocksDB.++Every key type serializes to a distinct first byte, so that keys of different+types never collide even when they share a column family. Many key types+have short variants (suffixed @A@, @B@ or @S@) that serialize to a prefix of+the full key, for use with the prefix queries of "Database.RocksDB.Query".+++-----------+--------------+----------------------------------------++| Prefix    | Key          | Value                                  |++===========+==============+========================================++| 32 x 0x00 | 'BestKey'    | hash of the best block                 |++-----------+--------------+----------------------------------------++| 0x01      | 'BlockKey'   | 'BlockData'                            |++-----------+--------------+----------------------------------------++| 0x02      | 'TxKey'      | 'TxData'                               |++-----------+--------------+----------------------------------------++| 0x03      | 'HeightKey'  | hashes of the blocks at a height       |++-----------+--------------+----------------------------------------++| 0x04      | 'BalKey'     | 'BalVal'                               |++-----------+--------------+----------------------------------------++| 0x05      | 'AddrTxKey'  | nothing                                |++-----------+--------------+----------------------------------------++| 0x06      | 'AddrOutKey' | 'OutVal'                               |++-----------+--------------+----------------------------------------++| 0x07      | 'MemKey'     | whole mempool (legacy)                 |++-----------+--------------+----------------------------------------++| 0x08      | 'MemTxKey'   | time a transaction entered the mempool |++-----------+--------------+----------------------------------------++| 0x09      | 'UnspentKey' | 'UnspentVal'                           |++-----------+--------------+----------------------------------------++| 0x0a      | 'VersionKey' | database version                       |++-----------+--------------+----------------------------------------++| 0x0b      | 'MemTimeKey' | nothing                                |++-----------+--------------+----------------------------------------++-} module Haskoin.Store.Database.Types   ( AddrTxKey (..),     AddrOutKey (..),@@ -85,8 +128,11 @@     AddrTxKey !Address !TxRef   | -- | short key that matches all entries     AddrTxKeyA !Address-  | AddrTxKeyB !Address !BlockRef-  | AddrTxKeyS+  | -- | short key that matches the entries of an address at a position in+    -- the chain+    AddrTxKeyB !Address !BlockRef+  | -- | short key that matches the entries of every address+    AddrTxKeyS   deriving (Show, Eq, Ord, Generic, Hashable)  instance Serialize AddrTxKey where@@ -122,8 +168,10 @@     AddrOutKey !Address !BlockRef !OutPoint   | -- | short key for all spent or unspent outputs     AddrOutKeyA !Address-  | AddrOutKeyB !Address !BlockRef-  | AddrOutKeyS+  | -- | short key for the outputs of an address at a position in the chain+    AddrOutKeyB !Address !BlockRef+  | -- | short key for the outputs of every address+    AddrOutKeyS   deriving (Show, Read, Eq, Ord, Generic, Hashable)  instance Serialize AddrOutKey where@@ -148,6 +196,8 @@  instance Key AddrOutKey +-- | Value of an unspent output of an address: amount in satoshi and output+-- script. data OutVal = OutVal !Word64 !ByteString   deriving (Show, Read, Eq, Ord, Generic, Hashable, Serialize) @@ -155,8 +205,11 @@  -- | Transaction database key. data TxKey-  = TxKey !TxHash-  | TxKeyS !(Word32, Word16)+  = -- | key of a transaction+    TxKey !TxHash+  | -- | short key that matches transactions whose hash starts with these six+    -- bytes; see 'decodeTxKey'+    TxKeyS !(Word32, Word16)   deriving (Show, Read, Eq, Ord, Generic, Hashable)  instance Serialize TxKey where@@ -171,6 +224,10 @@     guard . (== 0x02) =<< getWord8     TxKey <$> get +-- | Split a short numeric transaction identifier, as used by the Blockchain.info+-- compatibility API, into the first six bytes of the transaction hash and the+-- five high bits of the seventh byte. Only the low 53 bits of the identifier+-- are used. decodeTxKey :: Word64 -> ((Word32, Word16), Word8) decodeTxKey i =   let masked = i .&. 0x001fffffffffffff@@ -193,9 +250,12 @@  -- | Unspent output database key. data UnspentKey-  = UnspentKey !OutPoint-  | UnspentKeyS !TxHash-  | UnspentKeyB+  = -- | key of an unspent output+    UnspentKey !OutPoint+  | -- | short key that matches the unspent outputs of a transaction+    UnspentKeyS !TxHash+  | -- | short key that matches every unspent output+    UnspentKeyB   deriving (Show, Read, Eq, Ord, Generic, Hashable)  instance Serialize UnspentKey where@@ -220,6 +280,7 @@  instance KeyValue UnspentKey UnspentVal +-- | Build an 'Unspent' from an address output entry. Returns 'Nothing' for a short key. toUnspent :: Network -> Ctx -> AddrOutKey -> OutVal -> Maybe Unspent toUnspent net ctx (AddrOutKey _address block outpoint) (OutVal value script) =   Just@@ -232,7 +293,8 @@       } toUnspent _ _ _ _ = Nothing --- | Mempool transaction database key.+-- | Mempool transaction database key. Legacy: databases of version 18 kept the+-- whole mempool under this key. It is only read to migrate them. data MemKey   = MemKey   deriving (Show, Read)@@ -252,8 +314,10 @@ -- time it entered the mempool as its value. Replaces 'MemKey', which kept -- the whole mempool in a single value. data MemTxKey-  = MemTxKey !TxHash-  | MemTxKeyS+  = -- | key of a mempool transaction+    MemTxKey !TxHash+  | -- | short key that matches every mempool transaction+    MemTxKeyS   deriving (Show, Read, Eq, Ord, Generic, Hashable)  instance Serialize MemTxKey where@@ -276,8 +340,10 @@ -- hash, as the pairs in 'getMempool' do, so a page of the mempool is read -- with a short scan instead of loading and sorting every entry. data MemTimeKey-  = MemTimeKey !UnixTime !TxHash-  | MemTimeKeyS+  = -- | entry of a mempool transaction, by time and hash+    MemTimeKey !UnixTime !TxHash+  | -- | short key that matches every entry+    MemTimeKeyS   deriving (Show, Read, Eq, Ord, Generic, Hashable)  instance Serialize MemTimeKey where@@ -332,8 +398,10 @@  -- | Address balance database key. data BalKey-  = BalKey !Address-  | BalKeyS+  = -- | key of the balance of an address+    BalKey !Address+  | -- | short key that matches every balance+    BalKeyS   deriving (Show, Read, Eq, Ord, Generic, Hashable)  instance Serialize BalKey where@@ -381,19 +449,28 @@  instance KeyValue VersionKey Word32 +-- | Balance of an address as stored in the database. The address itself is in+-- the key. data BalVal = BalVal-  { confirmed :: !Word64,+  { -- | confirmed balance in satoshi+    confirmed :: !Word64,+    -- | unconfirmed balance in satoshi     unconfirmed :: !Word64,+    -- | number of unspent outputs     utxo :: !Word64,+    -- | number of transactions     txs :: !Word64,+    -- | total amount received in satoshi, including unconfirmed     received :: !Word64   }   deriving (Show, Read, Eq, Ord, Generic, Hashable, Serialize, NFData) +-- | Build a 'Balance' from an address and its stored value. valToBalance :: Address -> BalVal -> Balance valToBalance address BalVal {..} =   Balance {..} +-- | Stored value of a 'Balance'. balanceToVal :: Balance -> BalVal balanceToVal Balance {..} =   BalVal {..}@@ -409,16 +486,22 @@         received = 0       } +-- | Unspent output as stored in the database. The outpoint is in the key. data UnspentVal = UnspentVal-  { block :: !BlockRef,+  { -- | block or mempool position of the transaction+    block :: !BlockRef,+    -- | amount in satoshi     value :: !Word64,+    -- | output script     script :: !ByteString   }   deriving (Show, Read, Eq, Ord, Generic, Hashable, Serialize, NFData) +-- | Split an 'Unspent' into its key and its stored value. unspentToVal :: Unspent -> (OutPoint, UnspentVal) unspentToVal Unspent {..} = (outpoint, UnspentVal {..}) +-- | Build an 'Unspent' from its outpoint and stored value, decoding the address from the script. valToUnspent :: Network -> Ctx -> OutPoint -> UnspentVal -> Unspent valToUnspent net ctx outpoint UnspentVal {..} =   Unspent {..}

@@ -8,6 +8,28 @@ {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE NoFieldSelectors #-} +-- |+-- Module      : Haskoin.Store.Database.Writer+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- Buffered writes to the database.+--+-- 'WriterT' implements 'StoreWrite' by recording every change in in-memory hash+-- tables, and 'StoreReadBase' by reading those tables first and falling back to+-- the database. When the action given to 'runWriter' finishes, all the changes+-- are written to RocksDB in a single atomic batch. A block or a set of+-- transactions is thus either fully imported or not at all.+-- = Why buffer writes+--+-- Importing a block touches thousands of keys, and many of them several times:+-- an output is created and spent in the same block, a balance changes with every+-- transaction of its address. Buffering collapses those into one final value per+-- key, and a single atomic batch makes the whole import all-or-nothing. A crash+-- or an invalid block leaves the database exactly as it was. module Haskoin.Store.Database.Writer   ( WriterT,     MempoolChanges (..),@@ -59,12 +81,17 @@     writeIORef,   ) +-- | Environment of a batch of writes. data Writer = Writer-  { reader :: !DatabaseReader,+  { -- | database to read from and write to+    reader :: !DatabaseReader,+    -- | changes made so far     memory :: !Memory,+    -- | log messages here     logger :: !Logger   } +-- | Monad transformer for a batch of writes. Run it with 'runWriter'. type WriterT = ReaderT Writer  -- | Logger passed to 'runWriter'.@@ -83,36 +110,56 @@   getBalance = getBalanceI   getMempool = getMempoolI +-- | Change to the best block: 'Nothing' if unchanged, @Just Nothing@ if deleted. type BestRef = IORef (Maybe (Maybe BlockHash)) +-- | Changed blocks: 'Nothing' for a deletion. type BlockTable = H.BasicHashTable BlockHash (Maybe BlockData) +-- | Changed lists of blocks at a height. type HeightTable = H.BasicHashTable BlockHeight [BlockHash] +-- | Changed transactions: 'Nothing' for a deletion. type TxTable = H.BasicHashTable TxHash (Maybe TxData) +-- | Changed unspent outputs: 'Nothing' for a deletion. type UnspentTable = H.BasicHashTable OutPoint (Maybe Unspent) +-- | Changed balances: 'Nothing' for a deletion. type BalanceTable = H.BasicHashTable Address (Maybe Balance) +-- | Changed address transactions: 'Nothing' for a deletion. type AddrTxTable = H.BasicHashTable (Address, TxRef) (Maybe ()) +-- | Changed address unspent outputs: 'Nothing' for a deletion. type AddrOutTable = H.BasicHashTable (Address, BlockRef, OutPoint) (Maybe OutVal)  -- | Mempool changes in this batch: 'Nothing' for a deletion. type MempoolTable = H.BasicHashTable TxHash (Maybe UnixTime) +-- | Changes made in a batch, not yet written to the database. data Memory = Memory-  { net :: !Network,+  { -- | network constants+    net :: !Network,+    -- | secp256k1 context     ctx :: !Ctx,+    -- | best block     best :: !BestRef,+    -- | blocks     blockTable :: !BlockTable,+    -- | blocks at each height     heightTable :: !HeightTable,+    -- | transactions     txTable :: !TxTable,+    -- | unspent outputs     unspentTable :: !UnspentTable,+    -- | address balances     balanceTable :: !BalanceTable,+    -- | address transactions     addressTable :: !AddrTxTable,+    -- | address unspent outputs     outputTable :: !AddrOutTable,+    -- | mempool     mempoolTable :: !MempoolTable   } @@ -162,6 +209,7 @@     ReaderT $ \Writer {memory = Memory {..}} ->       liftIO $ H.insert unspentTable p Nothing +-- | Read a value from the changes of the batch, or from the database if the batch did not change it. getLayered ::   (MonadIO m) =>   (Memory -> IO (Maybe a)) ->@@ -173,6 +221,8 @@       Just x -> return x       Nothing -> runReaderT g db +-- | Run a batch of writes and commit all its changes to the database in one+-- atomic write. Reads within the batch see its own changes. runWriter ::   (MonadIO m) =>   Network ->@@ -187,7 +237,9 @@ -- | Transactions that a batch added to the mempool or removed from it, -- compared to what the database held before the batch was written. data MempoolChanges = MempoolChanges-  { added :: ![TxHash],+  { -- | transactions that entered the mempool+    added :: ![TxHash],+    -- | transactions that left the mempool     removed :: ![TxHash]   }   deriving (Eq, Show)@@ -210,6 +262,7 @@   liftIO $ writeBatch db (ops <> mops)   return (x, changes) +-- | Database operations for all changes of a batch except the mempool. hashMapOps :: (MonadIO m) => DB -> Memory -> m [BatchOp] hashMapOps db Memory {..} =   mconcat@@ -224,6 +277,7 @@         unspentOps db unspentTable       ] +-- | Operation for the change of the best block, if any. bestBlockOp :: (MonadIO m) => BestRef -> m [BatchOp] bestBlockOp r =   readIORef r >>= \case@@ -231,35 +285,41 @@     Just Nothing -> return [deleteOp BestKey]     Just (Just b) -> return [insertOp BestKey b] +-- | Operations for changed blocks. blockHashOps :: (MonadIO m) => DB -> BlockTable -> m [BatchOp] blockHashOps db t = map (uncurry f) <$> liftIO (H.toList t)   where     f k (Just d) = insertOpCF (blockCF db) (BlockKey k) d     f k Nothing = deleteOpCF (blockCF db) (BlockKey k) +-- | Operations for changed block heights. blockHeightOps :: (MonadIO m) => DB -> HeightTable -> m [BatchOp] blockHeightOps db t = map (uncurry f) <$> liftIO (H.toList t)   where     f = insertOpCF (heightCF db) . HeightKey +-- | Operations for changed transactions. txOps :: (MonadIO m) => DB -> TxTable -> m [BatchOp] txOps db t = map (uncurry f) <$> liftIO (H.toList t)   where     f k (Just t') = insertOpCF (txCF db) (TxKey k) t'     f k Nothing = deleteOpCF (txCF db) (TxKey k) +-- | Operations for changed balances. balOps :: (MonadIO m) => DB -> BalanceTable -> m [BatchOp] balOps db t = map (uncurry f) <$> liftIO (H.toList t)   where     f a (Just b) = insertOpCF (balanceCF db) (BalKey a) (balanceToVal b)     f a Nothing = deleteOpCF (balanceCF db) (BalKey a) +-- | Operations for changed address transactions. addrTxOps :: (MonadIO m) => DB -> AddrTxTable -> m [BatchOp] addrTxOps db t = map (uncurry f) <$> liftIO (H.toList t)   where     f (a, t') (Just ()) = insertOpCF (addrTxCF db) (AddrTxKey a t') ()     f (a, t') Nothing = deleteOpCF (addrTxCF db) (AddrTxKey a t') +-- | Operations for changed address unspent outputs. addrOutOps :: (MonadIO m) => DB -> AddrOutTable -> m [BatchOp] addrOutOps db t = map (uncurry f) <$> liftIO (H.toList t)   where@@ -300,6 +360,7 @@       | o == n = []       | otherwise = deleteOpCF cf (MemTimeKey o h) : mempoolInsertOps db h n +-- | Operations for changed unspent outputs. unspentOps :: (MonadIO m) => DB -> UnspentTable -> m [BatchOp] unspentOps db t = map (uncurry f) <$> liftIO (H.toList t)   where@@ -308,25 +369,32 @@     f p Nothing =       deleteOpCF (unspentCF db) (UnspentKey p) +-- | Read the best block, with the changes of the batch applied. getBestBlockI :: (MonadIO m) => WriterT m (Maybe BlockHash) getBestBlockI = getLayered getBestBlockH getBestBlock +-- | Read the blocks at a height, with the changes of the batch applied. getBlocksAtHeightI :: (MonadIO m) => BlockHeight -> WriterT m [BlockHash] getBlocksAtHeightI bh =   getLayered (getBlocksAtHeightH bh) (getBlocksAtHeight bh) +-- | Read a block, with the changes of the batch applied. getBlockI :: (MonadIO m) => BlockHash -> WriterT m (Maybe BlockData) getBlockI bh = getLayered (getBlockH bh) (getBlock bh) +-- | Read a transaction, with the changes of the batch applied. getTxDataI :: (MonadIO m) => TxHash -> WriterT m (Maybe TxData) getTxDataI th = getLayered (getTxDataH th) (getTxData th) +-- | Read the spender of an output, with the changes of the batch applied. getSpenderI :: (MonadIO m) => OutPoint -> WriterT m (Maybe Spender) getSpenderI op = getLayered (getSpenderH op) (getSpender op) +-- | Read the balance of an address, with the changes of the batch applied. getBalanceI :: (MonadIO m) => Address -> WriterT m (Maybe Balance) getBalanceI a = getLayered (getBalanceH a) (getBalance a) +-- | Read an unspent output, with the changes of the batch applied. getUnspentI :: (MonadIO m) => OutPoint -> WriterT m (Maybe Unspent) getUnspentI op = getLayered (getUnspentH op) (getUnspent op) @@ -342,6 +410,7 @@         added = [(t, h) | (h, Just t) <- changes]     return $ sortOn Down (kept <> added) +-- | Empty set of changes. newMemory ::   (MonadIO m) =>   Network ->@@ -359,26 +428,35 @@   mempoolTable <- liftIO H.new   return Memory {..} +-- | Read the best block from the changes of the batch: 'Nothing' if unchanged, @Just Nothing@ if deleted. getBestBlockH :: Memory -> IO (Maybe (Maybe BlockHash)) getBestBlockH = readIORef . (.best) +-- | Read the blocks at a height from the changes of the batch: 'Nothing' if unchanged, @Just Nothing@ if deleted. getBlocksAtHeightH :: BlockHeight -> Memory -> IO (Maybe [BlockHash]) getBlocksAtHeightH h s = H.lookup s.heightTable h +-- | Read a block from the changes of the batch: 'Nothing' if unchanged, @Just Nothing@ if deleted. getBlockH :: BlockHash -> Memory -> IO (Maybe (Maybe BlockData)) getBlockH h s = H.lookup s.blockTable h +-- | Read a transaction from the changes of the batch: 'Nothing' if unchanged, @Just Nothing@ if deleted. getTxDataH :: TxHash -> Memory -> IO (Maybe (Maybe TxData)) getTxDataH t s = H.lookup s.txTable t +-- | Read the spender of an output from the transaction changed in the batch:+-- 'Nothing' if the transaction did not change, @Just Nothing@ if the output is+-- not spent. getSpenderH :: OutPoint -> Memory -> IO (Maybe (Maybe Spender)) getSpenderH op s = do   fmap (f =<<) <$> getTxDataH op.hash s   where     f = IntMap.lookup (fromIntegral op.index) . (.spenders) +-- | Read the balance of an address from the changes of the batch: 'Nothing' if unchanged, @Just Nothing@ if deleted. getBalanceH :: Address -> Memory -> IO (Maybe (Maybe Balance)) getBalanceH a s = H.lookup s.balanceTable a +-- | Read an unspent output from the changes of the batch: 'Nothing' if unchanged, @Just Nothing@ if deleted. getUnspentH :: OutPoint -> Memory -> IO (Maybe (Maybe Unspent)) getUnspentH p Memory {..} = H.lookup unspentTable p

@@ -10,6 +10,47 @@ {-# LANGUAGE NoFieldSelectors #-} {-# OPTIONS_GHC -Wno-ambiguous-fields #-} +-- |+-- Module      : Haskoin.Store.Logic+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- Rules to import and revert blocks and transactions.+--+-- These functions keep every index consistent: transactions, unspent outputs,+-- spenders, address transactions, address unspent outputs, balances and the+-- mempool. They run in 'ImportM', which buffers all changes and commits them in+-- one atomic batch, so a failure leaves the database untouched.+--+-- When a block is imported, transactions already in the mempool are confirmed,+-- new ones are imported, and mempool transactions that conflict with the block+-- are deleted together with their descendants. When a block is reverted, its+-- transactions go back to the mempool and its coinbase is deleted.+--+-- On networks with replace-by-fee, a new mempool transaction may replace the+-- unconfirmed transactions it conflicts with. Elsewhere a conflicting mempool+-- transaction is rejected as a double spend.+-- = Why these rules+--+-- The indices are only useful if they always agree with each other: a balance+-- must equal the sum of the unspent outputs of its address, a spent output must+-- name its spender, and a transaction must appear in the index of every address+-- it touches. Every function here changes all of them together, inside one+-- atomic batch, so the database moves from one consistent state to the next.+--+-- The store does not validate scripts or consensus rules. It follows the chain+-- with the most work, as reported by haskoin-node, and checks only what it needs+-- to keep the indices consistent: that inputs exist and are unspent, and that a+-- transaction does not spend more than it has. A block that fails these checks+-- is not imported, and the block store disconnects the peer that sent it.+--+-- The mempool is treated as provisional. A confirmed block always wins: mempool+-- transactions that conflict with it are deleted together with everything that+-- spends them. Between mempool transactions, replace-by-fee decides on networks+-- that support it; elsewhere the first one seen stays. module Haskoin.Store.Logic   ( ImportException (..),     ImportM,@@ -82,22 +123,36 @@ -- failed batch commits nothing. type ImportM = WriterT IO +-- | Log a message with the logger of the writer. logDebug, logError :: Text -> ImportM () logDebug msg = writerLogger >>= \l -> logAt LevelDebug l msg logError msg = writerLogger >>= \l -> logAt LevelError l msg +-- | Reasons why an import can fail. data ImportException-  = PrevBlockNotBest-  | Orphan-  | UnexpectedCoinbase-  | BestBlockNotFound-  | BlockNotBest-  | TxNotFound-  | DoubleSpend-  | TxConfirmed-  | InsufficientFunds-  | DuplicatePrevOutput-  | TxSpent+  = -- | the block does not build on the best block+    PrevBlockNotBest+  | -- | the transaction spends an output that is unknown or already spent+    Orphan+  | -- | a coinbase transaction cannot go into the mempool+    UnexpectedCoinbase+  | -- | the database has no best block, or it cannot be read+    BestBlockNotFound+  | -- | only the best block can be reverted+    BlockNotBest+  | -- | a transaction that should be in the database is not+    TxNotFound+  | -- | the transaction conflicts with another and the network does not have+    -- replace-by-fee+    DoubleSpend+  | -- | a confirmed transaction cannot be deleted or replaced+    TxConfirmed+  | -- | the outputs of the transaction are worth more than its inputs+    InsufficientFunds+  | -- | the transaction spends the same output twice+    DuplicatePrevOutput+  | -- | a transaction cannot be deleted while another one spends it+    TxSpent   deriving (Eq, Ord, Exception)  instance Show ImportException where@@ -113,6 +168,7 @@   show DuplicatePrevOutput = "Duplicate previous output"   show TxSpent = "Transaction is spent" +-- | Import the genesis block if the database has no best block yet. initBest :: ImportM () initBest = do   logDebug "Initializing best block"@@ -122,6 +178,13 @@     logDebug "Importing Genesis block"     importBlock (genesisBlock net) (genesisNode net) +-- | Import a transaction into the mempool with the time it arrived. Returns+-- 'False' if it is already known.+--+-- Conflicting unconfirmed transactions are deleted together with their+-- descendants if the network has replace-by-fee; otherwise the transaction is+-- rejected with 'DoubleSpend'. Throws an 'ImportException' if the transaction is+-- invalid, for example if it spends unknown outputs. newMempoolTx :: Tx -> UnixTime -> ImportM Bool newMempoolTx tx w =   getActiveTxData (txHash tx) >>= \case@@ -137,6 +200,7 @@       _ <- importTx (MemRef w) w rbf tx       return True +-- | Data of the best block. Throws 'BestBlockNotFound' if there is none. bestBlockData :: ImportM BlockData bestBlockData = do   h <-@@ -151,6 +215,9 @@       throwIO BestBlockNotFound     Just b -> return b +-- | Revert the best block: make its parent the best block, mark it as not in the+-- main chain, move its transactions back to the mempool, and delete its+-- coinbase. Throws 'BlockNotBest' if the block is not the best block. revertBlock :: BlockHash -> ImportM () revertBlock bh = do   bd <-@@ -180,6 +247,8 @@   logDebug $     "Deleted coinbase: " <> txHashToHex (txHash (tds !! 0).tx) +-- | Check that a block builds on the best block, or that it is the genesis block+-- of an empty database. checkNewBlock :: Block -> BlockNode -> ImportM () checkNewBlock b n =   getBestBlock >>= \case@@ -198,6 +267,9 @@               <> blockHashToHex (headerHash b.header)           throwIO PrevBlockNotBest +-- | Import the transactions of a block, in dependency order. Mempool transactions+-- that conflict with the block are deleted first. Transactions already in the+-- mempool are confirmed; others are imported. importOrConfirm :: BlockNode -> [Tx] -> ImportM [TxData] importOrConfirm bn txns = do   logDebug "Freeing outputs..."@@ -229,6 +301,8 @@         "Importing tx: " <> txHashToHex (txHash tx)       importTx (br i) bn_time False tx +-- | Import a block that builds on the best block, and make it the best block.+-- Returns its data and the data of its transactions. importBlock :: Block -> BlockNode -> ImportM (BlockData, [TxData]) importBlock b n = do   logDebug $@@ -307,9 +381,12 @@     outputs = sum (map (.value) tx.outputs)     unique_inputs = length (nub' (map (.outpoint) tx.inputs)) +-- | Unspent outputs spent by a transaction, or 'Nothing' for those that are missing. getUnspentOutputs :: (StoreReadBase m) => Tx -> m [Maybe Unspent] getUnspentOutputs tx = mapM getUnspent (prevOuts tx) +-- | Data of a new transaction, from its replace-by-fee flag, position, time and+-- the outputs it spends. prepareTxData :: Bool -> BlockRef -> Word64 -> Tx -> [Unspent] -> TxData prepareTxData rbf br tt tx us =   TxData@@ -325,6 +402,7 @@     mkprv Unspent {..} = Prev script value     ps = I.fromList $ zip [0 ..] $ map mkprv us +-- | Import a transaction at a position in the chain or the mempool. Throws 'Orphan' if it spends a missing output. importTx ::   BlockRef ->   -- | unix time@@ -346,12 +424,15 @@   commitAddTx td   return td +-- | Move a confirmed transaction back to the mempool. unConfirmTx :: TxData -> ImportM TxData unConfirmTx t = confTx t Nothing +-- | Confirm a mempool transaction at a position in a block. confirmTx :: TxData -> BlockRef -> ImportM TxData confirmTx t br = confTx t (Just br) +-- | Move a transaction to a new position in the indices of all its addresses. replaceAddressTx :: TxData -> BlockRef -> ImportM () replaceAddressTx t new = do   net <- getNetwork@@ -370,6 +451,8 @@           txid = txHash t.tx         } +-- | Move an unspent output to a new position in the chain, updating the indices+-- and balance of its address. adjustAddressOutput ::   OutPoint ->   TxOut ->@@ -422,6 +505,8 @@       decreaseBalance (confirmed old) a o.value       increaseBalance (confirmed new) a o.value +-- | Confirm a transaction at a position in a block, or move it back to the mempool+-- if no position is given. confTx :: TxData -> Maybe BlockRef -> ImportM TxData confTx t mbr = do   replaceAddressTx t new@@ -452,6 +537,9 @@         insertTx s'         updateSpendersRBF s' +-- | Delete the transactions that spend the same outputs as this one, with their+-- descendants. The first flag refuses to delete confirmed transactions, and the+-- second refuses any deletion on networks without replace-by-fee. freeOutputs :: Bool -> Bool -> Tx -> ImportM () freeOutputs memonly rbfcheck tx =   getConflicts memonly rbfcheck tx >>= mapM_ (deleteSingleTx . txHash)@@ -477,9 +565,13 @@   return . reverse . map snd . sortTxs . M.elems $     M.fromList [(txHash t, t) | t <- concat chains] +-- | Delete a transaction and its descendants, confirmed or not. deleteConfirmedTx :: TxHash -> ImportM () deleteConfirmedTx = deleteTx False False +-- | Delete an unconfirmed transaction and its descendants. If the flag is set and+-- the network has no replace-by-fee, throw 'DoubleSpend' instead. Does nothing if+-- the transaction is not found. deleteUnconfirmedTx :: Bool -> TxHash -> ImportM () deleteUnconfirmedTx rbfcheck th =   getActiveTxData th >>= \case@@ -488,6 +580,7 @@       logDebug $         "Not found or already deleted: " <> txHashToHex th +-- | Delete a transaction and its descendants. deleteTx ::   -- | only delete transaction if unconfirmed   Bool ->@@ -504,6 +597,8 @@       <> txHashToHex th   mapM_ (\t -> let h = txHash t in deleteSingleTx h >> return h) chain +-- | A transaction and all its descendants, in an order in which they can be+-- deleted: spenders first. getChain ::   -- | only delete transaction if unconfirmed   Bool ->@@ -557,6 +652,7 @@           return $ M.elems txm'         else go txm' pdg' +-- | Delete a transaction that nothing spends. Throws 'TxSpent' otherwise. deleteSingleTx :: TxHash -> ImportM () deleteSingleTx th =   getActiveTxData th >>= \case@@ -576,12 +672,15 @@               <> txHashToHex th           throwIO TxSpent +-- | Remove a transaction from every index and mark it as deleted. commitDelTx :: TxData -> ImportM () commitDelTx = commitModTx False +-- | Add a transaction to every index. commitAddTx :: TxData -> ImportM () commitAddTx = commitModTx True +-- | Add a transaction to every index, or remove it if the flag is 'False'. commitModTx :: Bool -> TxData -> ImportM () commitModTx add tx_data = do   net <- getNetwork@@ -610,6 +709,7 @@       | add = addOutputs br tx       | otherwise = delOutputs br tx +-- | Add a transaction to the mempool if it is unconfirmed, or remove it otherwise. updateMempool :: TxData -> ImportM () updateMempool td@TxData {deleted = True} =   deleteFromMempool (txHash td.tx)@@ -618,14 +718,20 @@ updateMempool td@TxData {block = BlockRef {}} =   deleteFromMempool (txHash td.tx) +-- | Mark the outputs spent by a transaction as spent. spendOutputs :: Tx -> ImportM () spendOutputs tx =   zipWithM_ (spendOutput (txHash tx)) [0 ..] (prevOuts tx) +-- | Add the outputs of a transaction as unspent. addOutputs :: BlockRef -> Tx -> ImportM () addOutputs br tx =   zipWithM_ (addOutput br . OutPoint (txHash tx)) [0 ..] tx.outputs +-- | Whether an unconfirmed transaction can be replaced by fee: it signals BIP125+-- in one of its inputs, or it spends an unconfirmed transaction that can be+-- replaced. Always 'False' for confirmed transactions and on networks without+-- replace-by-fee. isRBF ::   (StoreReadBase m) =>   BlockRef ->@@ -654,12 +760,15 @@                 | otherwise -> ck hs'        in ck hs +-- | Add an unspent output. addOutput :: BlockRef -> OutPoint -> TxOut -> ImportM () addOutput = modOutput True +-- | Delete an output. delOutput :: BlockRef -> OutPoint -> TxOut -> ImportM () delOutput = modOutput False +-- | Add an unspent output, or delete it if the flag is 'False', updating the indices and balance of its address. modOutput ::   Bool ->   BlockRef ->@@ -694,12 +803,14 @@       | add = insertAddrUnspent       | otherwise = deleteAddrUnspent +-- | Delete the outputs of a transaction. delOutputs :: BlockRef -> Tx -> ImportM () delOutputs br tx =   forM_ (zip [0 ..] tx.outputs) $ \(i, o) -> do     let op = OutPoint (txHash tx) i     delOutput br op o +-- | Data of a transaction. Throws 'TxNotFound' if it is missing or deleted. getImportTxData :: TxHash -> ImportM TxData getImportTxData th =   getActiveTxData th >>= \case@@ -708,11 +819,13 @@       throwIO TxNotFound     Just d -> return d +-- | Output of a transaction at an index, or 'Nothing' if out of range. getTxOut :: Word32 -> Tx -> Maybe TxOut getTxOut i tx = do   guard (fromIntegral i < length tx.outputs)   return $ tx.outputs !! fromIntegral i +-- | Record which input spends an output. insertSpender :: OutPoint -> Spender -> ImportM () insertSpender op s = do   td <- getImportTxData op.hash@@ -721,6 +834,7 @@       td' = (td :: TxData) {spenders = p'}   insertTx td' +-- | Forget which input spends an output. deleteSpender :: OutPoint -> ImportM () deleteSpender op = do   td <- getImportTxData op.hash@@ -729,6 +843,7 @@       td' = (td :: TxData) {spenders = p'}   insertTx td' +-- | Spend an unspent output with an input of a transaction. spendOutput :: TxHash -> Word32 -> OutPoint -> ImportM () spendOutput th ix op = do   u <-@@ -744,9 +859,11 @@     decreaseBalance (confirmed u.block) a u.value     deleteAddrUnspent a u +-- | Make the outputs spent by a transaction unspent again. unspendOutputs :: Tx -> ImportM () unspendOutputs = mapM_ unspendOutput . prevOuts +-- | Make an output unspent again. unspendOutput :: OutPoint -> ImportM () unspendOutput op = do   net <- getNetwork@@ -774,17 +891,21 @@     insertAddrUnspent a u     increaseBalance (confirmed u.block) a o.value +-- | Change the total received by an address. modifyReceived :: Address -> (Word64 -> Word64) -> ImportM () modifyReceived a f = do   b <- getDefaultBalance a   setBalance b {received = f b.received} +-- | Remove an output from the confirmed or unconfirmed balance of an address. decreaseBalance :: Bool -> Address -> Word64 -> ImportM () decreaseBalance conf = modBalance conf False +-- | Add an output to the confirmed or unconfirmed balance of an address. increaseBalance :: Bool -> Address -> Word64 -> ImportM () increaseBalance conf = modBalance conf True +-- | Add or remove an output from the confirmed or unconfirmed balance of an address, and its count of unspent outputs. modBalance ::   -- | confirmed   Bool ->@@ -805,6 +926,7 @@       | add = (+)       | otherwise = subtract +-- | Increase or decrease the number of transactions of an address. modAddressCount :: Bool -> Address -> ImportM () modAddressCount add a = do   b <- getDefaultBalance a@@ -814,10 +936,12 @@       | add = (+ 1)       | otherwise = subtract 1 +-- | Addresses of a list of outputs, without duplicates. txOutAddrs :: Network -> Ctx -> [TxOut] -> [Address] txOutAddrs net ctx =   nub' . rights . map (scriptToAddressBS net ctx . (.script)) +-- | Addresses of the outputs spent by a transaction, without duplicates. txInAddrs :: Network -> Ctx -> [Prev] -> [Address] txInAddrs net ctx =   nub' . rights . map (scriptToAddressBS net ctx . (.script))@@ -831,8 +955,10 @@     prevs = I.elems t.prevs     outs = t.tx.outputs +-- | Outputs spent by a transaction, without the null outpoint of a coinbase. prevOuts :: Tx -> [OutPoint] prevOuts tx = filter (/= nullOutPoint) (map (.outpoint) tx.inputs) +-- | Whether a transaction is in the database and not deleted. testPresent :: (StoreReadBase m) => Tx -> m Bool testPresent tx = isJust <$> getActiveTxData (txHash tx)

@@ -5,6 +5,22 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE NoFieldSelectors #-} +-- |+-- Module      : Haskoin.Store.Manager+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- Wires the components of a store together: the RocksDB database, the+-- peer-to-peer node, the block store that imports data, and the optional Redis+-- cache with its writer process. Node events are forwarded to the block store+-- and turned into 'StoreEvent's.+-- Keeping the wiring in one place means each component only knows about the+-- interfaces it uses: the block store does not know about the cache, and the+-- node does not know about either. 'withStore' starts them in dependency order+-- and stops them all together. module Haskoin.Store.Manager   ( StoreConfig (..),     Store (..),@@ -106,15 +122,24 @@   ) import UnliftIO.Concurrent (threadDelay) --- | Store mailboxes.+-- | Store mailboxes. Handles to the running components of a store, given by+-- 'withStore'. data Store = Store-  { peerMgr :: !PeerMgr,+  { -- | peer manager of the node+    peerMgr :: !PeerMgr,+    -- | block header chain of the node     chain :: !Chain,+    -- | process that imports blocks and transactions     block :: !BlockStore,+    -- | database reader     db :: !DatabaseReader,+    -- | cache configuration, if Redis is configured     cache :: !(Maybe CacheConfig),+    -- | store events are published here     pub :: !(Publisher StoreEvent),+    -- | network constants     net :: !Network,+    -- | secp256k1 context     ctx :: !Ctx   } @@ -181,6 +206,12 @@     logger :: !Logger   } +-- | Run a store and pass its handles to the given action. Everything is stopped+-- when the action returns, and an exception in any component is re-thrown in+-- the calling thread.+--+-- The database is opened first, then the node, the cache (if 'redis' is set)+-- and the block store. withStore ::   (MonadUnliftIO m) =>   StoreConfig ->@@ -210,6 +241,7 @@               ctx = cfg.ctx             } +-- | Open the database described by the configuration and run an action with its reader. connectDB ::   (MonadUnliftIO m) =>   StoreConfig ->@@ -231,6 +263,7 @@     stats     f +-- | Configuration of the block store, from the store configuration. blockStoreCfg ::   StoreConfig ->   Node ->@@ -255,6 +288,8 @@       logger = withSource "BlockStore" cfg.logger     } +-- | Configuration of the node, from the store configuration. The node keeps its+-- headers in the default column family of the store database. nodeCfg ::   StoreConfig ->   DatabaseReader ->@@ -279,6 +314,8 @@       logger = cfg.logger     } +-- | If Redis is configured, connect to it, start the cache writer, and pass the+-- cache configuration to the action. Otherwise pass 'Nothing'. withCache ::   (MonadUnliftIO m) =>   StoreConfig ->@@ -315,6 +352,7 @@           logger = logger         } +-- | Run the loop that feeds store events to the cache writer while the action runs. cacheWriterProcesses ::   (MonadUnliftIO m) =>   Int ->@@ -325,6 +363,8 @@ cacheWriterProcesses interval evts cwm action =   withAsync (cacheWriterEvents interval evts cwm) $ \a1 -> link a1 >> action +-- | Forward store events to the cache writer, and ask it to sync the mempool+-- every given number of seconds. cacheWriterEvents :: (MonadUnliftIO m) => Int -> Inbox StoreEvent -> CacheWriter -> m () cacheWriterEvents interval evts cwm =   withAsync mempool . const $@@ -336,12 +376,14 @@       threadDelay (interval * 1000 * 1000)       cacheSyncMempool cwm +-- | Tell the cache writer about new blocks and mempool changes. cacheWriterDispatch :: (MonadIO m) => StoreEvent -> CacheWriter -> m () cacheWriterDispatch (StoreBestBlock _) = cacheNewBlock cacheWriterDispatch (StoreMempoolNew t) = cacheNewTx t cacheWriterDispatch (StoreMempoolDelete t) = cacheNewTx t cacheWriterDispatch _ = const (return ()) +-- | Forward every node event to the block store and the store publisher, forever. nodeForwarder ::   (MonadIO m) =>   BlockStore ->@@ -351,7 +393,10 @@ nodeForwarder b pub sub =   loopForever $ receive sub >>= atomically . storeDispatch b pub --- | Dispatcher of node events.+-- | Dispatcher of node events. Peer connections, blocks, transactions,+-- inventories and not-found messages go to the block store; peer events,+-- pongs, transaction announcements and rejections are published as store+-- events; new best headers go to the block store. storeDispatch ::   BlockStore ->   Publisher StoreEvent ->

@@ -14,6 +14,48 @@ {-# LANGUAGE NoFieldSelectors #-} {-# OPTIONS_GHC -Wno-deprecations #-} +-- |+-- Module      : Haskoin.Store.Web+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- REST API server of Haskoin Store, built with Scotty.+--+-- 'runWeb' serves the endpoints described in "Haskoin.Store.WebCommon" and a+-- set of endpoints compatible with the Blockchain.info API under+-- @\/blockchain\/@. Responses are JSON by default, pretty-printed with+-- @?pretty=true@, or binary if the request has the header+-- @Accept: application\/octet-stream@. New blocks and mempool transactions are+-- streamed at @\/events@, as lines of JSON or over a WebSocket.+--+-- Limits on the size of responses are set with 'WebLimits'. Errors are+-- returned as JSON with an HTTP status that depends on the 'Except'.+-- = Design of the API+--+-- * __Two encodings of the same data.__ JSON is easy to use from anywhere. The+--   binary encoding is several times smaller and faster to produce and parse, and+--   is what the Haskell client uses.+--+-- * __Limits protect the server.__ A single request for every transaction of a+--   busy address could take minutes and gigabytes. 'WebLimits' caps list sizes+--   and offsets, and lower caps apply to full transactions, which are much larger+--   than references.+--+-- * __Health is measured, not assumed.__ The store reports itself unhealthy when+--   it lags behind the network, stops receiving blocks or transactions, has too+--   few peers or a backlog of transactions. Load balancers can route around an+--   instance that is still syncing or has lost its peers.+--+-- * __Slow event clients are dropped.__ Each @\/events@ client gets a bounded+--   queue. A client that falls too far behind is disconnected rather than allowed+--   to make the server buffer events without limit.+--+-- * __Compatibility endpoints ease migration.__ The @\/blockchain\/@ endpoints+--   follow the Blockchain.info API so that existing wallets can switch to a+--   self-hosted server with few changes. module Haskoin.Store.Web   ( -- * Web     WebConfig (..),@@ -222,22 +264,37 @@ import UnliftIO.Concurrent (threadDelay) import Web.Scotty.Trans qualified as S +-- | Scotty application with access to the web state. type ScottyT m = S.ScottyT (ReaderT WebState m) +-- | Scotty request handler with access to the web state. type ActionT m = S.ActionT (ReaderT WebState m) +-- | Limits of the API. A value of @0@ means no limit. data WebLimits = WebLimits-  { maxItemCount :: !Word32,+  { -- | maximum number of items in a list response+    maxItemCount :: !Word32,+    -- | maximum number of full transactions in a list response     maxFullItemCount :: !Word32,+    -- | maximum offset of a paged query     maxOffset :: !Word32,+    -- | number of items in a list response when no limit is given     defItemCount :: !Word32,+    -- | consecutive unused addresses after which xpub derivation stops     xpubGap :: !Word32,+    -- | unused addresses to look at for an xpub with no transactions     xpubGapInit :: !Word32,+    -- | seconds without a new transaction before the store is reported+    -- unhealthy     txTimeout :: !Word64,+    -- | seconds without a new block before the store is reported unhealthy     blockTimeout :: !Word64   }   deriving (Eq, Show) +-- | 200000 items, 5000 full transactions, offset up to 50000, 100 items by+-- default, xpub gaps of 32 and 20, transaction timeout of 30 minutes and+-- block timeout of four hours. instance Default WebLimits where   def =     WebLimits@@ -251,34 +308,63 @@         blockTimeout = 4 * 3600       } +-- | Configuration of the web server. data WebConfig = WebConfig-  { host :: !String,+  { -- | address to listen on, such as @*@ for all+    host :: !String,+    -- | port to listen on     port :: !Int,+    -- | running store to serve     store :: !Store,+    -- | blocks that the database may lag behind the headers before the store+    -- is reported unhealthy     maxLaggingBlocks :: !Int,+    -- | transactions that may be waiting to be imported before the store is+    -- reported unhealthy     maxPendingTxs :: !Int,+    -- | fewest peers before the store is reported unhealthy     minPeers :: !Int,+    -- | limits of the API     limits :: !WebLimits,+    -- | version reported by the health check     version :: !String,+    -- | the mempool is not indexed, so do not report it in health checks     noMempool :: !Bool,+    -- | StatsD client, if metrics are enabled     stats :: !(Maybe Stats),+    -- | microseconds between updates of the price ticker     tickerRefresh :: !Int,+    -- | URL of the price ticker service     tickerURL :: !String,+    -- | URL of the historical price service     priceHistoryURL :: !String,+    -- | disable the xpub endpoints     noXPub :: !Bool,+    -- | disable the Blockchain.info compatible endpoints     noBlockchainInfo :: !Bool,+    -- | seconds between health checks     healthCheckInterval :: !Int,+    -- | log messages here     logger :: !Logger   } +-- | State shared by all request handlers. data WebState = WebState-  { config :: !WebConfig,+  { -- | configuration+    config :: !WebConfig,+    -- | latest prices by currency code     ticker :: !(TVar (HashMap Text BinfoTicker)),+    -- | StatsD metrics, if enabled     metrics :: !(Maybe WebMetrics),+    -- | HTTP session for price requests     session :: !Wreq.Session,+    -- | result of the latest health check     health :: !(TVar HealthCheck)   } +-- | StatsD metrics of the web server: a timing for each kind of request,+-- under @web.\<name\>@, error counters and a gauge of connected event+-- clients. data WebMetrics = WebMetrics   { -- Addresses     addressTx :: !StatTiming,@@ -341,6 +427,7 @@     events :: !StatGauge   } +-- | Register the web server metrics with a StatsD client. createMetrics :: (MonadIO m) => Stats -> m WebMetrics createMetrics s = liftIO $ do   -- Addresses@@ -415,9 +502,11 @@     g x = newStatGauge s ("web." <> x) 0     c x = newStatCounter s ("web." <> x) 10 +-- | Increase a gauge while an action runs. withGaugeIO :: (MonadUnliftIO m) => StatGauge -> m a -> m a withGaugeIO g = bracket_ (incrementGauge g 1) (decrementGauge g 1) +-- | Increase a gauge of the web metrics while a handler runs. withGaugeIncrease ::   (MonadUnliftIO m) =>   (WebMetrics -> StatGauge) ->@@ -430,6 +519,7 @@       s <- liftWith $ \run -> withGaugeIO (gf m) (run go)       restoreT $ return s +-- | Record how long a handler takes in a timing of the web metrics. withMetrics ::   (MonadUnliftIO m) => (WebMetrics -> StatTiming) -> ActionT m a -> ActionT m a withMetrics df go =@@ -445,7 +535,14 @@       let ms = t2 - t1       liftIO $ addTiming (df m) ms -data SerialAs = SerialAsBinary | SerialAsJSON | SerialAsPrettyJSON+-- | Encoding of a response.+data SerialAs+  = -- | binary, for @Accept: application\/octet-stream@+    SerialAsBinary+  | -- | compact JSON+    SerialAsJSON+  | -- | indented JSON, for @?pretty=true@+    SerialAsPrettyJSON   deriving (Eq, Show)  instance@@ -514,6 +611,9 @@ -- Path Handlers -- ------------------- +-- | Run the web server forever. Also starts a thread that updates prices, unless+-- the Blockchain.info endpoints are disabled, and a thread that runs health+-- checks. runWeb :: (MonadUnliftIO m) => WebConfig -> m () runWeb config = do   ticker <- newTVarIO HashMap.empty@@ -545,6 +645,7 @@     opts = S.defaultOptions {S.settings = settings defaultSettings}     settings = setPort config.port . setHost (fromString config.host) +-- | Historical prices of the coin of the network in a currency at several times. Returns an empty list on failure. getRates ::   (MonadUnliftIO m) =>   Logger ->@@ -571,6 +672,7 @@         & Wreq.param "base" .~ [T.toUpper $ T.pack net.name]     opts = base & Wreq.param "quote" .~ [currency] +-- | Update the prices of the coin of the network periodically, for BTC and BCH only. price ::   (MonadUnliftIO m) =>   Logger ->@@ -596,6 +698,7 @@           | net == bch = Just "bch"           | otherwise = Nothing +-- | Respond with an error and stop handling the request. raise :: (MonadIO m) => Except -> ActionT m a raise e = do   askl (.metrics) >>= mapM_ \m -> do@@ -610,6 +713,7 @@   S.json e   S.finish +-- | HTTP status of an error. errStatus :: Except -> Status errStatus ThingNotFound = status404 errStatus BadRequest = status400@@ -620,6 +724,7 @@ errStatus ServerTimeout = status500 errStatus RequestTooLarge = status413 +-- | Register the handlers of every endpoint, except those disabled in the configuration. handlePaths ::   (MonadUnliftIO m) =>   WebConfig ->@@ -857,6 +962,7 @@     json_list f = toJSONList . map f     net = cfg.store.net +-- | Register a handler for an API resource, with compact JSON by default. pathCompact ::   (ApiResource a b, MonadUnliftIO m) =>   ActionT m a ->@@ -867,6 +973,8 @@ pathCompact parser action encJson encValue =   pathCommon parser action encJson encValue False +-- | Register a handler for an API resource: parse the request, run the handler,+-- and encode the response as JSON or binary, as the client asked. pathCommon ::   (ApiResource a b, MonadUnliftIO m) =>   ActionT m a ->@@ -887,11 +995,13 @@     toProxy = const Proxy     proxy = toProxy parser +-- | Respond with JSON. streamEncoding :: (MonadIO m) => Encoding -> ActionT m () streamEncoding e = do   S.setHeader "Content-Type" "application/json; charset=utf-8"   S.raw (encodingToLazyByteString e) +-- | Encode a response in binary, JSON or pretty JSON. protoSerial ::   (Serial a) =>   SerialAs ->@@ -904,20 +1014,24 @@ protoSerial SerialAsPrettyJSON _ g =   encodePretty' defConfig {confTrailingNewline = True} . g +-- | Allow requests from any origin. setHeaders :: (MonadIO m) => S.ActionT m () setHeaders = S.setHeader "Access-Control-Allow-Origin" "*" +-- | Respond with JSON, pretty if the @pretty@ parameter or the default says so. setupJSON :: (MonadUnliftIO m) => Bool -> S.ActionT m SerialAs setupJSON pretty = do   S.setHeader "Content-Type" "application/json"   p <- param "pretty" `rescue` return pretty   return $ if p then SerialAsPrettyJSON else SerialAsJSON +-- | Respond with binary. setupBinary :: (MonadIO m) => S.ActionT m SerialAs setupBinary = do   S.setHeader "Content-Type" "application/octet-stream"   return SerialAsBinary +-- | Choose the encoding of the response from the @Accept@ header of the request. setupContentType :: (MonadUnliftIO m) => Bool -> S.ActionT m SerialAs setupContentType pretty = do   accept <- S.header "accept"@@ -928,6 +1042,7 @@  -- GET Block / GET Blocks -- +-- | Handler for @GET \/block\/:block@: a block by hash. scottyBlock ::   (MonadUnliftIO m) => GetBlock -> ActionT m BlockData scottyBlock (GetBlock h (NoTx noTx)) = withMetrics (.block) $ do@@ -937,6 +1052,7 @@     Just b -> do       return $ pruneTx noTx b +-- | Blocks by hash, without duplicates, leaving out unknown ones. getBlocks ::   (MonadUnliftIO m) =>   [H.BlockHash] ->@@ -945,18 +1061,21 @@ getBlocks hs notx =   (pruneTx notx <$>) . catMaybes <$> mapM getBlock (nub hs) +-- | Handler for @GET \/blocks?blocks=…@: several blocks by hash. scottyBlocks ::   (MonadUnliftIO m) => GetBlocks -> ActionT m [BlockData] scottyBlocks (GetBlocks hs (NoTx notx)) =   withMetrics (.block) $     getBlocks hs notx +-- | Keep only the coinbase in the list of transactions of a block, if the flag is 'True'. pruneTx :: Bool -> BlockData -> BlockData pruneTx False b = b pruneTx True BlockData {..} = BlockData {txs = take 1 txs, ..}  -- GET BlockRaw -- +-- | Handler for @GET \/block\/:block\/raw@: a block by hash, serialized as in the peer-to-peer protocol. scottyBlockRaw ::   (MonadUnliftIO m) =>   GetBlockRaw ->@@ -965,6 +1084,7 @@   withMetrics (.blockRaw) $     RawResult <$> getRawBlock h +-- | Block by hash with all its transactions. Responds with an error if not found. getRawBlock ::   (MonadUnliftIO m) =>   H.BlockHash ->@@ -972,6 +1092,7 @@ getRawBlock h =   getBlock h >>= maybe (raise ThingNotFound) return >>= lift . toRawBlock +-- | Rebuild a block from its data and its transactions. toRawBlock :: (StoreReadBase m) => BlockData -> m H.Block toRawBlock b = do   txs <- map transactionData . catMaybes <$> getTransactions b.txs@@ -979,6 +1100,7 @@  -- GET BlockBest / BlockBestRaw -- +-- | Handler for @GET \/block\/best@: the best block. scottyBlockBest ::   (MonadUnliftIO m) => GetBlockBest -> ActionT m BlockData scottyBlockBest (GetBlockBest (NoTx notx)) =@@ -986,6 +1108,7 @@     maybe (raise ThingNotFound) (return . pruneTx notx) <=< runMaybeT $       MaybeT getBestBlock >>= MaybeT . getBlock +-- | Handler for @GET \/block\/best\/raw@: the best block, serialized. scottyBlockBestRaw ::   (MonadUnliftIO m) =>   GetBlockBestRaw ->@@ -997,6 +1120,7 @@  -- GET BlockLatest -- +-- | Handler for @GET \/block\/latest@: the best block and its ancestors, up to 100 blocks. scottyBlockLatest ::   (MonadUnliftIO m) =>   GetBlockLatest ->@@ -1013,6 +1137,7 @@  -- GET BlockHeight / BlockHeights / BlockHeightRaw -- +-- | Handler for @GET \/block\/height\/:height@: the blocks at a height. scottyBlockHeight ::   (MonadUnliftIO m) => GetBlockHeight -> ActionT m [BlockData] scottyBlockHeight (GetBlockHeight h (NoTx notx)) =@@ -1020,6 +1145,7 @@     getBlocksAtHeight (fromIntegral h)       >>= (`getBlocks` notx) +-- | Handler for @GET \/block\/heights?heights=…@: the blocks at several heights. scottyBlockHeights ::   (MonadUnliftIO m) =>   GetBlockHeights ->@@ -1029,6 +1155,7 @@     mapM (getBlocksAtHeight . fromIntegral) heights       >>= (`getBlocks` notx) . concat +-- | Handler for @GET \/block\/height\/:height\/raw@: the blocks at a height, serialized. scottyBlockHeightRaw ::   (MonadUnliftIO m) =>   GetBlockHeightRaw ->@@ -1040,6 +1167,7 @@  -- GET BlockTime / BlockTimeRaw -- +-- | Handler for @GET \/block\/time\/:time@: the last block with a timestamp at or before a time. scottyBlockTime ::   (MonadUnliftIO m) =>   GetBlockTime ->@@ -1051,6 +1179,7 @@       Nothing -> raise ThingNotFound       Just b -> return $ pruneTx notx b +-- | Handler for @GET \/block\/mtp\/:time@: the first block with a median time past at or after a time. scottyBlockMTP ::   (MonadUnliftIO m) =>   GetBlockMTP ->@@ -1062,6 +1191,7 @@       Nothing -> raise ThingNotFound       Just b -> return $ pruneTx notx b +-- | Handler for @GET \/block\/time\/:time\/raw@: the last block with a timestamp at or before a time, serialized. scottyBlockTimeRaw ::   (MonadUnliftIO m) =>   GetBlockTimeRaw ->@@ -1073,6 +1203,7 @@       Nothing -> raise ThingNotFound       Just b -> RawResult <$> lift (toRawBlock b) +-- | Handler for @GET \/block\/mtp\/:time\/raw@: the first block with a median time past at or after a time, serialized. scottyBlockMTPRaw ::   (MonadUnliftIO m) =>   GetBlockMTPRaw ->@@ -1086,17 +1217,20 @@  -- GET Transactions -- +-- | Handler for @GET \/transaction\/:txid@: a transaction. scottyTx :: (MonadUnliftIO m) => GetTx -> ActionT m Transaction scottyTx (GetTx txid) =   withMetrics (.tx) $     getTransaction txid >>= maybe (raise ThingNotFound) return +-- | Handler for @GET \/transactions?txids=…@: several transactions. scottyTxs ::   (MonadUnliftIO m) => GetTxs -> ActionT m [Transaction] scottyTxs (GetTxs txids) =   withMetrics (.tx) $     catMaybes <$> getTransactions (nub txids) +-- | Handler for @GET \/transaction\/:txid\/raw@: a transaction, serialized. scottyTxRaw ::   (MonadUnliftIO m) => GetTxRaw -> ActionT m (RawResult Tx) scottyTxRaw (GetTxRaw txid) =@@ -1106,6 +1240,7 @@         Nothing -> raise ThingNotFound         Just tx -> return $ RawResult (transactionData tx) +-- | Handler for @GET \/transactions\/raw?txids=…@: several transactions, serialized. scottyTxsRaw ::   (MonadUnliftIO m) =>   GetTxsRaw ->@@ -1115,6 +1250,7 @@     RawResultList . map transactionData . catMaybes       <$> getTransactions (nub txids) +-- | Transactions of a block. Responds with an error if not found. getTxsBlock ::   (MonadUnliftIO m) =>   H.BlockHash ->@@ -1124,6 +1260,7 @@     Nothing -> raise ThingNotFound     Just b -> catMaybes <$> getTransactions b.txs +-- | Handler for @GET \/transactions\/block\/:block@: the transactions of a block. scottyTxsBlock ::   (MonadUnliftIO m) =>   GetTxsBlock ->@@ -1131,6 +1268,7 @@ scottyTxsBlock (GetTxsBlock h) =   withMetrics (.txBlock) $ getTxsBlock h +-- | Handler for @GET \/transactions\/block\/:block\/raw@: the transactions of a block, serialized. scottyTxsBlockRaw ::   (MonadUnliftIO m) =>   GetTxsBlockRaw ->@@ -1141,6 +1279,7 @@  -- GET TransactionAfterHeight -- +-- | Handler for @GET \/transaction\/:txid\/after\/:height@: whether any ancestor of a transaction is a coinbase confirmed after a height. scottyTxAfter ::   (MonadUnliftIO m) =>   GetTxAfter ->@@ -1189,6 +1328,7 @@  -- POST Transaction -- +-- | Handler for @POST \/transactions@: broadcast a transaction to half of the peers and return its hash. scottyPostTx :: (MonadUnliftIO m) => PostTx -> ActionT m TxId scottyPostTx (PostTx tx) =   withMetrics (.txPost) $ do@@ -1215,6 +1355,7 @@  -- GET Mempool / Events -- +-- | Handler for @GET \/mempool@: hashes of mempool transactions, newest first. scottyMempool ::   (MonadUnliftIO m) => GetMempool -> ActionT m [TxHash] scottyMempool (GetMempool limitM (OffsetParam o)) =@@ -1224,6 +1365,9 @@         l = Limits (validateLimit wl' False limitM) (fromIntegral o) Nothing     map snd <$> getMempoolPage l +-- | Middleware that serves @\/events@ over a WebSocket: each new block or mempool+-- transaction is sent as a JSON message. Clients that do not answer pings, or+-- that fall more than 10000 events behind, are disconnected. webSocketEvents :: WebState -> Middleware webSocketEvents s =   websocketsOr defaultConnectionOptions (wrap . events)@@ -1267,6 +1411,7 @@                   [("Content-Type", "application/json")]               } +-- | Handler for @GET \/events@: stream of new blocks and mempool transactions, one per line in JSON, until the client falls more than 10000 events behind. scottyEvents :: (MonadUnliftIO m) => ActionT m () scottyEvents =   withGaugeIncrease (.events) $ do@@ -1323,18 +1468,21 @@  -- GET Address Transactions -- +-- | Handler for @GET \/address\/:address\/transactions@: transactions of an address. scottyAddrTxs ::   (MonadUnliftIO m) => GetAddrTxs -> ActionT m [TxRef] scottyAddrTxs (GetAddrTxs addr pLimits) =   withMetrics (.addressTx) $     getAddressTxs addr =<< paramToLimits False pLimits +-- | Handler for @GET \/address\/transactions?addresses=…@: transactions of several addresses. scottyAddrsTxs ::   (MonadUnliftIO m) => GetAddrsTxs -> ActionT m [TxRef] scottyAddrsTxs (GetAddrsTxs addrs pLimits) =   withMetrics (.addressTx) $     getAddressesTxs addrs =<< paramToLimits False pLimits +-- | Handler for @GET \/address\/:address\/transactions\/full@: full transactions of an address. scottyAddrTxsFull ::   (MonadUnliftIO m) =>   GetAddrTxsFull ->@@ -1344,6 +1492,7 @@     txs <- getAddressTxs addr =<< paramToLimits True pLimits     catMaybes <$> getTransactions (map (.txid) txs) +-- | Handler for @GET \/address\/transactions\/full?addresses=…@: full transactions of several addresses. scottyAddrsTxsFull ::   (MonadUnliftIO m) =>   GetAddrsTxsFull ->@@ -1353,21 +1502,25 @@     txs <- getAddressesTxs addrs =<< paramToLimits True pLimits     catMaybes <$> getTransactions (map (.txid) txs) +-- | Handler for @GET \/address\/:address\/balance@: balance of an address. scottyAddrBalance :: (MonadUnliftIO m) => GetAddrBalance -> ActionT m Balance scottyAddrBalance (GetAddrBalance addr) =   withMetrics (.addressBalance) $ getDefaultBalance addr +-- | Handler for @GET \/address\/balances?addresses=…@: balances of several addresses. scottyAddrsBalance ::   (MonadUnliftIO m) => GetAddrsBalance -> ActionT m [Balance] scottyAddrsBalance (GetAddrsBalance addrs) =   withMetrics (.addressBalance) $ getBalances addrs +-- | Handler for @GET \/address\/:address\/unspent@: unspent outputs of an address. scottyAddrUnspent ::   (MonadUnliftIO m) => GetAddrUnspent -> ActionT m [Unspent] scottyAddrUnspent (GetAddrUnspent addr pLimits) =   withMetrics (.addressUnspent) $     getAddressUnspents addr =<< paramToLimits False pLimits +-- | Handler for @GET \/address\/unspent?addresses=…@: unspent outputs of several addresses. scottyAddrsUnspent ::   (MonadUnliftIO m) => GetAddrsUnspent -> ActionT m [Unspent] scottyAddrsUnspent (GetAddrsUnspent addrs pLimits) =@@ -1376,6 +1529,7 @@  -- GET XPubs -- +-- | Handler for @GET \/xpub\/:xpub@: summary of an extended public key. scottyXPub ::   (MonadUnliftIO m) => GetXPub -> ActionT m XPubSummary scottyXPub (GetXPub xpub deriv (NoCache noCache)) =@@ -1383,6 +1537,7 @@     let xs = XPubSpec xpub deriv      in xPubSummary xs <$> lift (runNoCache noCache $ xPubBals xs) +-- | Handler for @DELETE \/xpub\/:xpub@: remove an extended public key from the cache. Returns 'True' if it was cached. scottyDelXPub ::   (MonadUnliftIO m) =>   DelCachedXPub ->@@ -1393,6 +1548,7 @@     n <- lift (withCache c $ cacheDelXPubs [XPubSpec xpub deriv])     return $ GenericResult (n > 0) +-- | Transactions of an extended public key, optionally without the cache. getXPubTxs ::   (MonadUnliftIO m) =>   XPubKey ->@@ -1406,12 +1562,14 @@   xbals <- xPubBals xs   lift $ runNoCache nocache $ xPubTxs xs xbals limits +-- | Handler for @GET \/xpub\/:xpub\/transactions@: transactions of an extended public key. scottyXPubTxs ::   (MonadUnliftIO m) => GetXPubTxs -> ActionT m [TxRef] scottyXPubTxs (GetXPubTxs xpub deriv plimits (NoCache nocache)) =   withMetrics (.xpubTx) $     getXPubTxs xpub deriv plimits nocache +-- | Handler for @GET \/xpub\/:xpub\/transactions\/full@: full transactions of an extended public key. scottyXPubTxsFull ::   (MonadUnliftIO m) =>   GetXPubTxsFull ->@@ -1422,6 +1580,7 @@       lift . runNoCache nocache . getTransactions . map (.txid)         =<< getXPubTxs xpub deriv plimits nocache +-- | Handler for @GET \/xpub\/:xpub\/balances@: balances of the used addresses of an extended public key. scottyXPubBalances ::   (MonadUnliftIO m) => GetXPubBalances -> ActionT m [XPubBal] scottyXPubBalances (GetXPubBalances xpub deriv (NoCache noCache)) =@@ -1430,6 +1589,7 @@       runNoCache noCache $         xPubBals (XPubSpec xpub deriv) +-- | Handler for @GET \/xpub\/:xpub\/unspent@: unspent outputs of an extended public key. scottyXPubUnspent ::   (MonadUnliftIO m) =>   GetXPubUnspent ->@@ -1445,6 +1605,7 @@ -- Blockchain.info API Compatibility -- --------------------------------------- +-- | Currency symbol of the coin of a network, in the Blockchain.info format. netBinfoSymbol :: Network -> BinfoSymbol netBinfoSymbol net   | net == btc =@@ -1475,6 +1636,7 @@           local = False         } +-- | Currency symbol of a fiat currency from its price ticker. binfoTickerToSymbol :: Text -> BinfoTicker -> BinfoSymbol binfoTickerToSymbol code t =   BinfoSymbol@@ -1492,6 +1654,7 @@       "GBP" -> "British pound"       x -> x +-- | Addresses and xpubs in a request parameter, separated by pipes or commas. getBinfoAddrsParam ::   (MonadUnliftIO m) =>   Text ->@@ -1506,6 +1669,9 @@       Nothing -> raise (UserError "invalid address")       Just xs -> return $ HashSet.fromList xs +-- | Xpubs and addresses in the @active@, @activeP2SH@ and @activeBech32@+-- parameters. Xpubs in each parameter derive legacy, nested SegWit and native+-- SegWit addresses. getBinfoActive ::   (MonadUnliftIO m) =>   ActionT m (HashSet XPubSpec, HashSet Address)@@ -1529,13 +1695,16 @@     xpub (BinfoXpub x) = Just x     xpub (BinfoAddr _) = Nothing +-- | Whether to identify transactions by short number, unless @txidindex@ is false. getNumTxId :: (MonadUnliftIO m) => ActionT m Bool getNumTxId = fmap not $ param "txidindex" `rescue` return False +-- | Height of the best header. getChainHeight :: (MonadUnliftIO m) => ActionT m H.BlockHeight getChainHeight =   fmap (.height) $ liftIO . chainGetBest =<< askl (.config.store.chain) +-- | Handler for @GET\/POST \/blockchain\/unspent@: unspent outputs of addresses and xpubs, as Blockchain.info. scottyBinfoUnspent :: (MonadUnliftIO m) => ActionT m () scottyBinfoUnspent =   withMetrics (.binfoUnspent) $ do@@ -1558,6 +1727,8 @@     get_limit = fmap (min 1000) $ param "limit" `rescue` return 250     get_min_conf = param "confirmations" `rescue` return 0 +-- | Stream the unspent outputs of xpubs and addresses, newest first, in the+-- Blockchain.info format. getBinfoUnspents ::   (StoreReadExtra m, MonadIO m) =>   Bool ->@@ -1601,6 +1772,7 @@           g a = streamThings (h a) Nothing l .| mapC f        in map g (HashSet.toList addrs) +-- | Balances of the used addresses of each xpub. getXBals ::   (StoreReadExtra m) =>   HashSet XPubSpec ->@@ -1610,9 +1782,13 @@   where     f x = (x,) . filter (not . nullBalance . (.balance)) <$> xPubBals x +-- | Balances of an xpub, or none if missing. xBals :: XPubSpec -> HashMap XPubSpec [XPubBal] -> [XPubBal] xBals = HashMap.findWithDefault [] +-- | Stream the transactions of xpubs and addresses, newest first, in the+-- Blockchain.info format, with the running balance of the given addresses after+-- each, keeping only those that pass the filter. getBinfoTxs ::   (StoreReadExtra m, MonadIO m) =>   HashMap XPubSpec [XPubBal] -> -- xpub balances@@ -1696,9 +1872,11 @@                     | c -> return ()                     | otherwise -> yield a >> go b' +-- | Whether to show addresses in CashAddr format (@cashaddr@). getCashAddr :: (MonadUnliftIO m) => ActionT m Bool getCashAddr = param "cashaddr" `rescue` return False +-- | Parse an address. Responds with not found if invalid. getAddress :: (MonadUnliftIO m) => Text -> ActionT m Address getAddress txt = do   net <- askl (.config.store.net)@@ -1706,6 +1884,7 @@     Nothing -> raise ThingNotFound     Just a -> return a +-- | Address or xpub in a path parameter. Passes to the next route if invalid. getBinfoAddr :: (MonadUnliftIO m) => Text -> ActionT m BinfoAddr getBinfoAddr param' = do   txt <- S.captureParam param'@@ -1715,6 +1894,7 @@       xpub = BinfoXpub <$> xPubImport net ctx txt   maybe S.next return (addr <|> xpub) +-- | Handler for @GET\/POST \/blockchain\/export-history@: daily history of the balance of addresses and xpubs with prices, as Blockchain.info. scottyBinfoHistory :: (MonadUnliftIO m) => ActionT m () scottyBinfoHistory =   withMetrics (.binfoExportHistory) $ do@@ -1792,6 +1972,7 @@     xpub_c endM x bs =       streamThings (xPubTxs x bs) (Just (.txid)) (lim endM) +-- | Currency code in the @currency@ parameter, USD by default, and its latest price. getPrice :: (MonadUnliftIO m) => ActionT m (Text, BinfoTicker) getPrice = do   code <- T.toUpper <$> param "currency" `rescue` return "USD"@@ -1801,9 +1982,11 @@     Nothing -> return (code, def)     Just p -> return (code, p) +-- | Symbol of the currency in the @currency@ parameter. getSymbol :: (MonadUnliftIO m) => ActionT m BinfoSymbol getSymbol = uncurry binfoTickerToSymbol <$> getPrice +-- | Handler for @GET \/blockchain\/blocks\/:milliseconds@: the blocks of the day before a time in milliseconds, as Blockchain.info. scottyBinfoBlocksDay :: (MonadUnliftIO m) => ActionT m () scottyBinfoBlocksDay =   withMetrics (.binfoBlock) $ do@@ -1823,6 +2006,7 @@           b' <- getBlock b.header.prev           (b :) <$> go t b' +-- | Handler for @GET\/POST \/blockchain\/multiaddr@: balances and transactions of addresses and xpubs, as Blockchain.info. scottyMultiAddr :: (MonadUnliftIO m) => ActionT m () scottyMultiAddr =   withMetrics (.binfoMultiaddr) $ do@@ -1983,6 +2167,7 @@       let f = map (.balance.address)        in HashSet.fromList . concatMap f . HashMap.elems +-- | Number of items requested in a parameter, within the limits. getBinfoCount :: (MonadUnliftIO m) => Text -> ActionT m Int getBinfoCount str = do   d <- askl (.config.limits.defItemCount)@@ -1990,6 +2175,7 @@   i <- min x <$> (param str `rescue` return d)   return (fromIntegral i :: Int) +-- | Offset requested in the @offset@ parameter. Responds with an error if above the maximum. getBinfoOffset ::   (MonadUnliftIO m) =>   ActionT m Int@@ -2002,6 +2188,7 @@         "offset exceeded: " <> show o <> " > " <> show x   return (fromIntegral o :: Int) +-- | Handler for @GET \/blockchain\/rawaddr\/:addr@: balance and transactions of an address or xpub, as Blockchain.info; also at @\/blockchain\/address\/:addr@ and @\/blockchain\/xpub\/:addr@. scottyRawAddr :: (MonadUnliftIO m) => ActionT m () scottyRawAddr =   withMetrics (.binfoAddressRaw) $@@ -2097,6 +2284,7 @@       ctx <- askl (.config.store.ctx)       streamEncoding $ marshalEncoding (net, ctx) ra +-- | Handler for @GET \/blockchain\/q\/getreceivedbyaddress\/:addr@: total received by an address. scottyBinfoReceived :: (MonadUnliftIO m) => ActionT m () scottyBinfoReceived =   withMetrics (.binfoQgetreceivedbyaddress) $ do@@ -2105,6 +2293,7 @@     setHeaders     S.text $ cs $ show b.received +-- | Handler for @GET \/blockchain\/q\/getsentbyaddress\/:addr@: total sent by an address. scottyBinfoSent :: (MonadUnliftIO m) => ActionT m () scottyBinfoSent =   withMetrics (.binfoQgetsentbyaddress) $ do@@ -2113,6 +2302,7 @@     setHeaders     S.text $ cs $ show $ b.received - b.confirmed - b.unconfirmed +-- | Handler for @GET \/blockchain\/q\/addressbalance\/:addr@: balance of an address. scottyBinfoAddrBalance :: (MonadUnliftIO m) => ActionT m () scottyBinfoAddrBalance =   withMetrics (.binfoQaddressbalance) $ do@@ -2121,6 +2311,7 @@     setHeaders     S.text $ cs $ show $ b.confirmed + b.unconfirmed +-- | Handler for @GET \/blockchain\/q\/addressfirstseen\/:addr@: time of the first block with a transaction of an address. scottyFirstSeen :: (MonadUnliftIO m) => ActionT m () scottyFirstSeen =   withMetrics (.binfoQaddressfirstseen) $ do@@ -2160,6 +2351,7 @@       let l = Limits 1 0 (Just (AtBlock h))       not . null <$> getAddressTxs a l +-- | Handler for @GET\/POST \/blockchain\/balance@: short balances of addresses and xpubs, as Blockchain.info. scottyShortBal :: (MonadUnliftIO m) => ActionT m () scottyShortBal =   withMetrics (.binfoBalance) $ do@@ -2204,9 +2396,11 @@       ctx <- askl (.config.store.ctx)       return (xPubExport net ctx xpub.key, sbl) +-- | Whether to respond with hex (@format=hex@). getBinfoHex :: (MonadUnliftIO m) => ActionT m Bool getBinfoHex = (== ("hex" :: Text)) <$> param "format" `rescue` return "json" +-- | Handler for @GET \/blockchain\/block-height\/:height@: the blocks at a height, as Blockchain.info. scottyBinfoBlockHeight :: (MonadUnliftIO m) => ActionT m () scottyBinfoBlockHeight =   withMetrics (.binfoBlockHeight) $ do@@ -2227,6 +2421,7 @@           bts = map (toBinfoTxSimple numtxid) txs       return $ toBinfoBlock b bts nbs +-- | Handler for @GET \/blockchain\/latestblock@: the best block, as Blockchain.info. scottyBinfoLatest :: (MonadUnliftIO m) => ActionT m () scottyBinfoLatest =   withMetrics (.binfoBlockLatest) $ do@@ -2243,6 +2438,7 @@             txids = map (encodeBinfoTxId numtxid) b.txs           } +-- | Handler for @GET \/blockchain\/rawblock\/:block@: a block with its transactions, as Blockchain.info. scottyBinfoBlock :: (MonadUnliftIO m) => ActionT m () scottyBinfoBlock =   withMetrics (.binfoBlockRaw) $ do@@ -2274,6 +2470,7 @@           ctx <- askl (.config.store.ctx)           streamEncoding $ marshalEncoding (net, ctx) y +-- | Transaction by hash or short number. Fails with 'TxIndexConflict' if a short number matches several. getBinfoTx ::   (MonadUnliftIO m) =>   BinfoTxId ->@@ -2289,6 +2486,7 @@       let tids = map (txHash . transactionData) ts        in return $ Left (TxIndexConflict tids) +-- | Handler for @GET \/blockchain\/rawtx\/:txid@: a transaction, as Blockchain.info. scottyBinfoTx :: (MonadUnliftIO m) => ActionT m () scottyBinfoTx =   withMetrics (.binfoTxRaw) $ do@@ -2310,6 +2508,7 @@       setHeaders       S.text . encodeHexLazy . runPutL . serialize $ transactionData t +-- | Handler for @GET \/blockchain\/q\/txtotalbtcoutput\/:txid@: total value of the outputs of a transaction. scottyBinfoTotalOut :: (MonadUnliftIO m) => ActionT m () scottyBinfoTotalOut =   withMetrics (.binfoQtxtotalbtcoutput) $ do@@ -2320,6 +2519,7 @@         Left e -> raise e     S.text $ cs $ show $ sum $ map (.value) tx.outputs +-- | Handler for @GET \/blockchain\/q\/txfee\/:txid@: fee of a transaction. scottyBinfoTxFees :: (MonadUnliftIO m) => ActionT m () scottyBinfoTxFees =   withMetrics (.binfoQtxfee) $ do@@ -2335,6 +2535,7 @@     f StoreInput {} = True     f StoreCoinbase {} = False +-- | Handler for @GET \/blockchain\/q\/txresult\/:txid\/:addr@: net amount that a transaction sends to an address. scottyBinfoTxResult :: (MonadUnliftIO m) => ActionT m () scottyBinfoTxResult =   withMetrics (.binfoQtxresult) $ do@@ -2353,6 +2554,7 @@     g addr StoreOutput {address = Just a} = a == addr     g _ _ = False +-- | Handler for @GET \/blockchain\/q\/txtotalbtcinput\/:txid@: total value of the inputs of a transaction. scottyBinfoTotalInput :: (MonadUnliftIO m) => ActionT m () scottyBinfoTotalInput =   withMetrics (.binfoQtxtotalbtcinput) $ do@@ -2366,6 +2568,7 @@     f StoreInput {} = True     f StoreCoinbase {} = False +-- | Handler for @GET \/blockchain\/unconfirmed-transactions@: mempool transactions, as Blockchain.info. scottyBinfoMempool :: (MonadUnliftIO m) => ActionT m () scottyBinfoMempool =   withMetrics (.binfoMempool) $ do@@ -2386,6 +2589,7 @@     ctx <- askl (.config.store.ctx)     streamEncoding $ marshalEncoding (net, ctx) mem +-- | Handler for @GET \/blockchain\/q\/getblockcount@: height of the best header. scottyBinfoGetBlockCount :: (MonadUnliftIO m) => ActionT m () scottyBinfoGetBlockCount =   withMetrics (.binfoQgetblockcount) $ do@@ -2394,6 +2598,7 @@     setHeaders     S.text $ TL.pack $ show bn.height +-- | Handler for @GET \/blockchain\/q\/latesthash@: hash of the best header. scottyBinfoLatestHash :: (MonadUnliftIO m) => ActionT m () scottyBinfoLatestHash =   withMetrics (.binfoQlatesthash) $ do@@ -2402,6 +2607,7 @@     setHeaders     S.text $ TL.fromStrict $ H.blockHashToHex $ H.headerHash bn.header +-- | Handler for @GET \/blockchain\/q\/bcperblock@: block subsidy at the next height. scottyBinfoSubsidy :: (MonadUnliftIO m) => ActionT m () scottyBinfoSubsidy =   withMetrics (.binfoQbcperblock) $ do@@ -2416,6 +2622,7 @@             fromIntegral $               H.computeSubsidy net (bn.height + 1) +-- | Handler for @GET \/blockchain\/q\/addresstohash\/:addr@: hash160 of an address in hex. scottyBinfoAddrToHash :: (MonadUnliftIO m) => ActionT m () scottyBinfoAddrToHash =   withMetrics (.binfoQaddresstohash) $ do@@ -2429,6 +2636,7 @@       | otherwise ->           return () +-- | Handler for @GET \/blockchain\/q\/hashtoaddress\/:hash@: pay-to-public-key-hash address of a hash160. scottyBinfoHashToAddr :: (MonadUnliftIO m) => ActionT m () scottyBinfoHashToAddr =   withMetrics (.binfoQhashtoaddress) $ do@@ -2439,6 +2647,7 @@     setHeaders     S.text $ TL.fromStrict addr +-- | Handler for @GET \/blockchain\/q\/addrpubkey\/:pubkey@: address of a public key. scottyBinfoAddrPubkey :: (MonadUnliftIO m) => ActionT m () scottyBinfoAddrPubkey =   withMetrics (.binfoQaddrpubkey) $ do@@ -2453,6 +2662,7 @@       Nothing -> raise ThingNotFound       Just a -> S.text $ TL.fromStrict a +-- | Handler for @GET \/blockchain\/q\/pubkeyaddr\/:addr@: public key of an address, found in a transaction input that spends from it. scottyBinfoPubKeyAddr :: (MonadUnliftIO m) => ActionT m () scottyBinfoPubKeyAddr =   withMetrics (.binfoQpubkeyaddr) $ do@@ -2497,6 +2707,7 @@         _ -> Left "Address does not have public key"     extr _ _ = Left "Incorrect input type" +-- | Handler for @GET \/blockchain\/q\/hashpubkey\/:pubkey@: hash160 of a public key in hex. scottyBinfoHashPubkey :: (MonadUnliftIO m) => ActionT m () scottyBinfoHashPubkey =   withMetrics (.binfoQhashpubkey) $ do@@ -2519,6 +2730,7 @@  -- GET Network Information -- +-- | Handler for @GET \/peers@: connected peers. scottyPeers ::   (MonadUnliftIO m) =>   GetPeers ->@@ -2545,6 +2757,7 @@             relay = ver.relay           } +-- | Handler for @GET \/health@: health of the store. scottyHealth ::   (MonadUnliftIO m) => GetHealth -> ActionT m HealthCheck scottyHealth _ =@@ -2553,6 +2766,7 @@     unless (isOK h) $ S.status status503     return h +-- | Heights of the best header and best block, and the maximum lag allowed. blockHealthCheck ::   (MonadUnliftIO m, StoreReadBase m) =>   WebConfig ->@@ -2570,6 +2784,7 @@         max = fromIntegral cfg.maxLaggingBlocks       } +-- | Age of the best header and the maximum allowed. lastBlockHealthCheck ::   (MonadUnliftIO m, StoreReadBase m) =>   Chain ->@@ -2584,6 +2799,7 @@         max = fromIntegral blockTimeout       } +-- | Age of the newest mempool transaction or best header, and the maximum allowed. lastTxHealthCheck ::   (MonadUnliftIO m, StoreReadBase m) =>   WebConfig ->@@ -2609,6 +2825,7 @@         then limits.blockTimeout         else limits.txTimeout +-- | Transactions waiting to be imported and the maximum allowed. pendingTxsHealthCheck ::   (MonadUnliftIO m, StoreReadBase m) =>   WebConfig ->@@ -2621,6 +2838,7 @@         count = fromIntegral n       } +-- | Online peers and the minimum required. peerHealthCheck ::   (MonadUnliftIO m, StoreReadBase m) =>   WebConfig ->@@ -2629,6 +2847,7 @@   count <- fromIntegral . length <$> liftIO (getPeers cfg.store.peerMgr)   return CountHealth {min = fromIntegral cfg.minPeers, count} +-- | Run every health check. healthCheck ::   (MonadUnliftIO m, StoreReadBase m) => WebConfig -> m HealthCheck healthCheck cfg = do@@ -2645,6 +2864,7 @@         ..       } +-- | Handler for @GET \/dbstats@: RocksDB memory use and statistics, as text. scottyDbStats :: (MonadUnliftIO m) => ActionT m () scottyDbStats =   withMetrics (.db) $ do@@ -2726,6 +2946,7 @@         Right b16 -> fromBytes (decodeBase16 b16)         Left e -> Left (cs e) +-- | Offset in the @offset@ parameter. Responds with an error if above the maximum. parseOffset :: (MonadUnliftIO m) => ActionT m OffsetParam parseOffset = do   res@(OffsetParam o) <- paramDef@@ -2735,6 +2956,8 @@       "offset exceeded: " <> show o <> " > " <> show limits.maxOffset   return res +-- | Turn a start parameter into a position: a transaction, the block of a block+-- hash, a height, or the last block at or before a time. parseStart ::   (MonadUnliftIO m) =>   Maybe StartParam ->@@ -2759,9 +2982,11 @@       b <- MaybeT $ blockAtOrBefore ch q       return $ AtBlock b.height +-- | Limit, offset and start in the parameters of a request. parseLimits :: (MonadUnliftIO m) => ActionT m LimitsParam parseLimits = LimitsParam <$> paramOptional <*> parseOffset <*> paramOptional +-- | Pagination of a query from request parameters, within the limits for full or short items. paramToLimits ::   (MonadUnliftIO m) =>   Bool ->@@ -2771,6 +2996,7 @@   wl <- askl (.config.limits)   Limits (validateLimit wl full limitM) (fromIntegral o) <$> parseStart startM +-- | Limit of a query: the requested one or the default, but no more than the maximum. validateLimit :: WebLimits -> Bool -> Maybe LimitParam -> Word32 validateLimit wl full limitM =   f m $ maybe d (fromIntegral . (.get)) limitM@@ -2787,6 +3013,7 @@ -- Utilities -- --------------- +-- | Run a query against the database, through the cache if there is one. runInWebReader ::   (MonadIO m) =>   CacheT (DatabaseReaderT m) a ->@@ -2796,6 +3023,7 @@   mc <- asks (.config.store.cache)   lift $ runReaderT (withCache mc f) bdb +-- | Run queries without the cache if the flag is 'True'. runNoCache ::   (MonadIO m) =>   Bool ->@@ -2805,6 +3033,7 @@ runNoCache True = local $ \s ->   s {config = s.config {store = s.config.store {cache = Nothing}}} +-- | Middleware that logs every request with its response status: errors at error level, others at debug level. logIt :: Logger -> Middleware logIt logger app req respond =   app req $ \res -> do@@ -2815,6 +3044,7 @@       else logError logger msg     respond res +-- | Describe a request for logs. fmtReq :: Request -> Text fmtReq req =   let m = requestMethod req@@ -2823,9 +3053,11 @@       q = rawQueryString req    in T.decodeUtf8 (m <> " " <> p <> q <> " " <> cs (show v)) +-- | Describe a response status for logs. fmtStatus :: Status -> Text fmtStatus s = cs (show (statusCode s)) <> " " <> cs (statusMessage s) +-- | Run a handler, or another if the first one fails. rescue ::   (MonadUnliftIO m) =>   S.ActionT m a ->@@ -2833,11 +3065,13 @@   S.ActionT m a x `rescue` y = x `S.catch` \(_ :: S.ScottyException) -> y +-- | Read a request parameter from the path, query string or form. param ::   (MonadUnliftIO m, S.Parsable a) =>   Text ->   S.ActionT m a param t = S.queryParam t `rescue` S.formParam t +-- | Read a field of the environment from a monad transformer. askl :: (MonadTrans t, MonadReader r m) => (r -> a) -> t m a askl f = lift (asks f)