13586005509

Committed 28 Feb 2025 10:14AM UTC coverage: 68.629% (+9.9%) from 58.77%

Build # 13586005509

Build Type

Pull #9521

github

Committed by

web-flow

Commit Message

Merge 37d3a70a5 into 8532955b3

Pull Request Pull Request #9521: unit: remove GOACC, use Go 1.20 native coverage functionality

Run Details

129950 of 189351 relevant lines covered (68.63%)

23726.46 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.75

/routing/chainview/btcd.go

package chainview

import (
        "bytes"
        "encoding/hex"
        "fmt"
        "sync"
        "sync/atomic"

        "github.com/btcsuite/btcd/btcjson"
        "github.com/btcsuite/btcd/btcutil"
        "github.com/btcsuite/btcd/chaincfg/chainhash"
        "github.com/btcsuite/btcd/rpcclient"
        "github.com/btcsuite/btcd/wire"
        "github.com/lightningnetwork/lnd/blockcache"
        graphdb "github.com/lightningnetwork/lnd/graph/db"
)

// BtcdFilteredChainView is an implementation of the FilteredChainView
// interface which is backed by an active websockets connection to btcd.
type BtcdFilteredChainView struct {
        started int32 // To be used atomically.
        stopped int32 // To be used atomically.

        // bestHeight is the height of the latest block added to the
        // blockQueue from the onFilteredConnectedMethod. It is used to
        // determine up to what height we would need to rescan in case
        // of a filter update.
        bestHeightMtx sync.Mutex
        bestHeight    uint32

        btcdConn *rpcclient.Client

        // blockEventQueue is the ordered queue used to keep the order
        // of connected and disconnected blocks sent to the reader of the
        // chainView.
        blockQueue *blockEventQueue

        // blockCache is an LRU block cache.
        blockCache *blockcache.BlockCache

        // filterUpdates is a channel in which updates to the utxo filter
        // attached to this instance are sent over.
        filterUpdates chan filterUpdate

        // chainFilter is the set of utox's that we're currently watching
        // spends for within the chain.
        filterMtx   sync.RWMutex
        chainFilter map[wire.OutPoint]struct{}

        // filterBlockReqs is a channel in which requests to filter select
        // blocks will be sent over.
        filterBlockReqs chan *filterBlockReq

        quit chan struct{}
        wg   sync.WaitGroup
}

// A compile time check to ensure BtcdFilteredChainView implements the
// chainview.FilteredChainView.
var _ FilteredChainView = (*BtcdFilteredChainView)(nil)

// NewBtcdFilteredChainView creates a new instance of a FilteredChainView from
// RPC credentials for an active btcd instance.
func NewBtcdFilteredChainView(config rpcclient.ConnConfig,
        blockCache *blockcache.BlockCache) (*BtcdFilteredChainView, error) {

        chainView := &BtcdFilteredChainView{
                chainFilter:     make(map[wire.OutPoint]struct{}),
                filterUpdates:   make(chan filterUpdate),
                filterBlockReqs: make(chan *filterBlockReq),
                blockCache:      blockCache,
                quit:            make(chan struct{}),
        }

        ntfnCallbacks := &rpcclient.NotificationHandlers{
                OnFilteredBlockConnected:    chainView.onFilteredBlockConnected,
                OnFilteredBlockDisconnected: chainView.onFilteredBlockDisconnected,
        }

        // Disable connecting to btcd within the rpcclient.New method. We
        // defer establishing the connection to our .Start() method.
        config.DisableConnectOnNew = true
        config.DisableAutoReconnect = false
        chainConn, err := rpcclient.New(&config, ntfnCallbacks)
        if err != nil {
                return nil, err
        }
        chainView.btcdConn = chainConn

        chainView.blockQueue = newBlockEventQueue()

        return chainView, nil
}

// Start starts all goroutines necessary for normal operation.
//
// NOTE: This is part of the FilteredChainView interface.
func (b *BtcdFilteredChainView) Start() error {
        // Already started?
        if atomic.AddInt32(&b.started, 1) != 1 {
                return nil
        }

        log.Infof("FilteredChainView starting")

        // Connect to btcd, and register for notifications on connected, and
        // disconnected blocks.
        if err := b.btcdConn.Connect(20); err != nil {
                return err
        }
        if err := b.btcdConn.NotifyBlocks(); err != nil {
                return err
        }

        _, bestHeight, err := b.btcdConn.GetBestBlock()
        if err != nil {
                return err
        }

        b.bestHeightMtx.Lock()
        b.bestHeight = uint32(bestHeight)
        b.bestHeightMtx.Unlock()

        b.blockQueue.Start()

        b.wg.Add(1)
        go b.chainFilterer()

        return nil
}

// Stop stops all goroutines which we launched by the prior call to the Start
// method.
//
// NOTE: This is part of the FilteredChainView interface.
func (b *BtcdFilteredChainView) Stop() error {
        log.Debug("BtcdFilteredChainView stopping")
        defer log.Debug("BtcdFilteredChainView stopped")

        // Already shutting down?
        if atomic.AddInt32(&b.stopped, 1) != 1 {
                return nil
        }

        // Shutdown the rpc client, this gracefully disconnects from btcd, and
        // cleans up all related resources.
        b.btcdConn.Shutdown()
        b.btcdConn.WaitForShutdown()

        b.blockQueue.Stop()

        close(b.quit)
        b.wg.Wait()

        return nil
}

// onFilteredBlockConnected is called for each block that's connected to the
// end of the main chain. Based on our current chain filter, the block may or
// may not include any relevant transactions.
func (b *BtcdFilteredChainView) onFilteredBlockConnected(height int32,
        header *wire.BlockHeader, txns []*btcutil.Tx) {

        mtxs := make([]*wire.MsgTx, len(txns))
        b.filterMtx.Lock()
        for i, tx := range txns {
                mtx := tx.MsgTx()
                mtxs[i] = mtx

                for _, txIn := range mtx.TxIn {
                        // We can delete this outpoint from the chainFilter, as
                        // we just received a block where it was spent. In case
                        // of a reorg, this outpoint might get "un-spent", but
                        // that's okay since it would never be wise to consider
                        // the channel open again (since a spending transaction
                        // exists on the network).
                        delete(b.chainFilter, txIn.PreviousOutPoint)
                }

        }
        b.filterMtx.Unlock()

        // We record the height of the last connected block added to the
        // blockQueue such that we can scan up to this height in case of
        // a rescan. It must be protected by a mutex since a filter update
        // might be trying to read it concurrently.
        b.bestHeightMtx.Lock()
        b.bestHeight = uint32(height)
        b.bestHeightMtx.Unlock()

        block := &FilteredBlock{
                Hash:         header.BlockHash(),
                Height:       uint32(height),
                Transactions: mtxs,
        }

        b.blockQueue.Add(&blockEvent{
                eventType: connected,
                block:     block,
        })
}

// onFilteredBlockDisconnected is a callback which is executed once a block is
// disconnected from the end of the main chain.
func (b *BtcdFilteredChainView) onFilteredBlockDisconnected(height int32,
        header *wire.BlockHeader) {

        log.Debugf("got disconnected block at height %d: %v", height,
                header.BlockHash())

        filteredBlock := &FilteredBlock{
                Hash:   header.BlockHash(),
                Height: uint32(height),
        }

        b.blockQueue.Add(&blockEvent{
                eventType: disconnected,
                block:     filteredBlock,
        })
}

// filterBlockReq houses a request to manually filter a block specified by
// block hash.
type filterBlockReq struct {
        blockHash *chainhash.Hash
        resp      chan *FilteredBlock
        err       chan error
}

// FilterBlock takes a block hash, and returns a FilteredBlocks which is the
// result of applying the current registered UTXO sub-set on the block
// corresponding to that block hash. If any watched UTOX's are spent by the
// selected lock, then the internal chainFilter will also be updated.
//
// NOTE: This is part of the FilteredChainView interface.
func (b *BtcdFilteredChainView) FilterBlock(blockHash *chainhash.Hash) (*FilteredBlock, error) {
        req := &filterBlockReq{
                blockHash: blockHash,
                resp:      make(chan *FilteredBlock, 1),
                err:       make(chan error, 1),
        }

        select {
        case b.filterBlockReqs <- req:
        case <-b.quit:
                return nil, fmt.Errorf("FilteredChainView shutting down")
        }

        return <-req.resp, <-req.err
}

// chainFilterer is the primary goroutine which: listens for new blocks coming
// and dispatches the relevant FilteredBlock notifications, updates the filter
// due to requests by callers, and finally is able to preform targeted block
// filtration.
//
// TODO(roasbeef): change to use loadfilter RPC's
func (b *BtcdFilteredChainView) chainFilterer() {
        defer b.wg.Done()

        // filterBlock is a helper function that scans the given block, and
        // notes which transactions spend outputs which are currently being
        // watched. Additionally, the chain filter will also be updated by
        // removing any spent outputs.
        filterBlock := func(blk *wire.MsgBlock) []*wire.MsgTx {
                b.filterMtx.Lock()
                defer b.filterMtx.Unlock()

                var filteredTxns []*wire.MsgTx
                for _, tx := range blk.Transactions {
                        var txAlreadyFiltered bool
                        for _, txIn := range tx.TxIn {
                                prevOp := txIn.PreviousOutPoint
                                if _, ok := b.chainFilter[prevOp]; !ok {
                                        continue
                                }

                                delete(b.chainFilter, prevOp)

                                // Only add this txn to our list of filtered
                                // txns if it is the first previous outpoint to
                                // cause a match.
                                if txAlreadyFiltered {
                                        continue
                                }

                                filteredTxns = append(filteredTxns, tx.Copy())
                                txAlreadyFiltered = true

                        }
                }

                return filteredTxns
        }

        decodeJSONBlock := func(block *btcjson.RescannedBlock,
                height uint32) (*FilteredBlock, error) {
                hash, err := chainhash.NewHashFromStr(block.Hash)
                if err != nil {
                        return nil, err

                }
                txs := make([]*wire.MsgTx, 0, len(block.Transactions))
                for _, str := range block.Transactions {
                        b, err := hex.DecodeString(str)
                        if err != nil {
                                return nil, err
                        }
                        tx := &wire.MsgTx{}
                        err = tx.Deserialize(bytes.NewReader(b))
                        if err != nil {
                                return nil, err
                        }
                        txs = append(txs, tx)
                }
                return &FilteredBlock{
                        Hash:         *hash,
                        Height:       height,
                        Transactions: txs,
                }, nil
        }

        for {
                select {
                // The caller has just sent an update to the current chain
                // filter, so we'll apply the update, possibly rewinding our
                // state partially.
                case update := <-b.filterUpdates:

                        // First, we'll add all the new UTXO's to the set of
                        // watched UTXO's, eliminating any duplicates in the
                        // process.
                        log.Tracef("Updating chain filter with new UTXO's: %v",
                                update.newUtxos)

                        b.filterMtx.Lock()
                        for _, newOp := range update.newUtxos {
                                b.chainFilter[newOp] = struct{}{}
                        }
                        b.filterMtx.Unlock()

                        // Apply the new TX filter to btcd, which will cause
                        // all following notifications from and calls to it
                        // return blocks filtered with the new filter.
                        b.btcdConn.LoadTxFilter(false, []btcutil.Address{},
                                update.newUtxos)

                        // All blocks gotten after we loaded the filter will
                        // have the filter applied, but we will need to rescan
                        // the blocks up to the height of the block we last
                        // added to the blockQueue.
                        b.bestHeightMtx.Lock()
                        bestHeight := b.bestHeight
                        b.bestHeightMtx.Unlock()

                        // If the update height matches our best known height,
                        // then we don't need to do any rewinding.
                        if update.updateHeight == bestHeight {
                                continue
                        }

                        // Otherwise, we'll rewind the state to ensure the
                        // caller doesn't miss any relevant notifications.
                        // Starting from the height _after_ the update height,
                        // we'll walk forwards, rescanning one block at a time
                        // with btcd applying the newly loaded filter to each
                        // block.
                        for i := update.updateHeight + 1; i < bestHeight+1; i++ {
                                blockHash, err := b.btcdConn.GetBlockHash(int64(i))
                                if err != nil {
                                        log.Warnf("Unable to get block hash "+
                                                "for block at height %d: %v",
                                                i, err)
                                        continue
                                }

                                // To avoid dealing with the case where a reorg
                                // is happening while we rescan, we scan one
                                // block at a time, skipping blocks that might
                                // have gone missing.
                                rescanned, err := b.btcdConn.RescanBlocks(
                                        []chainhash.Hash{*blockHash})
                                if err != nil {
                                        log.Warnf("Unable to rescan block "+
                                                "with hash %v at height %d: %v",
                                                blockHash, i, err)
                                        continue
                                }

                                // If no block was returned from the rescan, it
                                // means no matching transactions were found.
                                if len(rescanned) != 1 {
                                        log.Tracef("rescan of block %v at "+
                                                "height=%d yielded no "+
                                                "transactions", blockHash, i)
                                        continue
                                }
                                decoded, err := decodeJSONBlock(
                                        &rescanned[0], uint32(i))
                                if err != nil {
                                        log.Errorf("Unable to decode block: %v",
                                                err)
                                        continue
                                }
                                b.blockQueue.Add(&blockEvent{
                                        eventType: connected,
                                        block:     decoded,
                                })
                        }

                // We've received a new request to manually filter a block.
                case req := <-b.filterBlockReqs:
                        // First we'll fetch the block itself as well as some
                        // additional information including its height.
                        block, err := b.GetBlock(req.blockHash)
                        if err != nil {
                                req.err <- err
                                req.resp <- nil
                                continue
                        }
                        header, err := b.btcdConn.GetBlockHeaderVerbose(req.blockHash)
                        if err != nil {
                                req.err <- err
                                req.resp <- nil
                                continue
                        }

                        // Once we have this info, we can directly filter the
                        // block and dispatch the proper notification.
                        req.resp <- &FilteredBlock{
                                Hash:         *req.blockHash,
                                Height:       uint32(header.Height),
                                Transactions: filterBlock(block),
                        }
                        req.err <- err

                case <-b.quit:
                        return
                }
        }
}

// filterUpdate is a message sent to the chainFilterer to update the current
// chainFilter state.
type filterUpdate struct {
        newUtxos     []wire.OutPoint
        updateHeight uint32
}

// UpdateFilter updates the UTXO filter which is to be consulted when creating
// FilteredBlocks to be sent to subscribed clients. This method is cumulative
// meaning repeated calls to this method should _expand_ the size of the UTXO
// sub-set currently being watched.  If the set updateHeight is _lower_ than
// the best known height of the implementation, then the state should be
// rewound to ensure all relevant notifications are dispatched.
//
// NOTE: This is part of the FilteredChainView interface.
func (b *BtcdFilteredChainView) UpdateFilter(ops []graphdb.EdgePoint,
        updateHeight uint32) error {

        newUtxos := make([]wire.OutPoint, len(ops))
        for i, op := range ops {
                newUtxos[i] = op.OutPoint
        }

        select {

        case b.filterUpdates <- filterUpdate{
                newUtxos:     newUtxos,
                updateHeight: updateHeight,
        }:
                return nil

        case <-b.quit:
                return fmt.Errorf("chain filter shutting down")
        }
}

// FilteredBlocks returns the channel that filtered blocks are to be sent over.
// Each time a block is connected to the end of a main chain, and appropriate
// FilteredBlock which contains the transactions which mutate our watched UTXO
// set is to be returned.
//
// NOTE: This is part of the FilteredChainView interface.
func (b *BtcdFilteredChainView) FilteredBlocks() <-chan *FilteredBlock {
        return b.blockQueue.newBlocks
}

// DisconnectedBlocks returns a receive only channel which will be sent upon
// with the empty filtered blocks of blocks which are disconnected from the
// main chain in the case of a re-org.
//
// NOTE: This is part of the FilteredChainView interface.
func (b *BtcdFilteredChainView) DisconnectedBlocks() <-chan *FilteredBlock {
        return b.blockQueue.staleBlocks
}

// GetBlock is used to retrieve the block with the given hash. This function
// wraps the blockCache's GetBlock function.
func (b *BtcdFilteredChainView) GetBlock(hash *chainhash.Hash) (
        *wire.MsgBlock, error) {

        return b.blockCache.GetBlock(hash, b.btcdConn.GetBlock)
}

1	package chainview
2
3	import (
4	"bytes"
5	"encoding/hex"
6	"fmt"
7	"sync"
8	"sync/atomic"
9
10	"github.com/btcsuite/btcd/btcjson"
11	"github.com/btcsuite/btcd/btcutil"
12	"github.com/btcsuite/btcd/chaincfg/chainhash"
13	"github.com/btcsuite/btcd/rpcclient"
14	"github.com/btcsuite/btcd/wire"
15	"github.com/lightningnetwork/lnd/blockcache"
16	graphdb "github.com/lightningnetwork/lnd/graph/db"
17	)
18
19	// BtcdFilteredChainView is an implementation of the FilteredChainView
20	// interface which is backed by an active websockets connection to btcd.
21	type BtcdFilteredChainView struct {
22	started int32 // To be used atomically.
23	stopped int32 // To be used atomically.
24
25	// bestHeight is the height of the latest block added to the
26	// blockQueue from the onFilteredConnectedMethod. It is used to
27	// determine up to what height we would need to rescan in case
28	// of a filter update.
29	bestHeightMtx sync.Mutex
30	bestHeight uint32
31
32	btcdConn *rpcclient.Client
33
34	// blockEventQueue is the ordered queue used to keep the order
35	// of connected and disconnected blocks sent to the reader of the
36	// chainView.
37	blockQueue *blockEventQueue
38
39	// blockCache is an LRU block cache.
40	blockCache *blockcache.BlockCache
41
42	// filterUpdates is a channel in which updates to the utxo filter
43	// attached to this instance are sent over.
44	filterUpdates chan filterUpdate
45
46	// chainFilter is the set of utox's that we're currently watching
47	// spends for within the chain.
48	filterMtx sync.RWMutex
49	chainFilter map[wire.OutPoint]struct{}
50
51	// filterBlockReqs is a channel in which requests to filter select
52	// blocks will be sent over.
53	filterBlockReqs chan *filterBlockReq
54
55	quit chan struct{}
56	wg sync.WaitGroup
57	}
58
59	// A compile time check to ensure BtcdFilteredChainView implements the
60	// chainview.FilteredChainView.
61	var _ FilteredChainView = (*BtcdFilteredChainView)(nil)
62
63	// NewBtcdFilteredChainView creates a new instance of a FilteredChainView from
64	// RPC credentials for an active btcd instance.
65	func NewBtcdFilteredChainView(config rpcclient.ConnConfig,
66	blockCache blockcache.BlockCache) (BtcdFilteredChainView, error) {	6✔
67		6✔
68	chainView := &BtcdFilteredChainView{	6✔
69	chainFilter: make(map[wire.OutPoint]struct{}),	6✔
70	filterUpdates: make(chan filterUpdate),	6✔
71	filterBlockReqs: make(chan *filterBlockReq),	6✔
72	blockCache: blockCache,	6✔
73	quit: make(chan struct{}),	6✔
74	}	6✔
75		6✔
76	ntfnCallbacks := &rpcclient.NotificationHandlers{	6✔
77	OnFilteredBlockConnected: chainView.onFilteredBlockConnected,	6✔
78	OnFilteredBlockDisconnected: chainView.onFilteredBlockDisconnected,	6✔
79	}	6✔
80		6✔
81	// Disable connecting to btcd within the rpcclient.New method. We	6✔
82	// defer establishing the connection to our .Start() method.	6✔
83	config.DisableConnectOnNew = true	6✔
84	config.DisableAutoReconnect = false	6✔
85	chainConn, err := rpcclient.New(&config, ntfnCallbacks)	6✔
86	if err != nil {	6✔
87	return nil, err	×
88	}	×
89	chainView.btcdConn = chainConn	6✔
90		6✔
91	chainView.blockQueue = newBlockEventQueue()	6✔
92		6✔
93	return chainView, nil	6✔
94	}
95
96	// Start starts all goroutines necessary for normal operation.
97	//
98	// NOTE: This is part of the FilteredChainView interface.
99	func (b *BtcdFilteredChainView) Start() error {	6✔
100	// Already started?	6✔
101	if atomic.AddInt32(&b.started, 1) != 1 {	6✔
102	return nil	×
103	}	×
104
105	log.Infof("FilteredChainView starting")	6✔
106		6✔
107	// Connect to btcd, and register for notifications on connected, and	6✔
108	// disconnected blocks.	6✔
109	if err := b.btcdConn.Connect(20); err != nil {	6✔
110	return err	×
111	}	×
112	if err := b.btcdConn.NotifyBlocks(); err != nil {	6✔
113	return err	×
114	}	×
115
116	_, bestHeight, err := b.btcdConn.GetBestBlock()	6✔
117	if err != nil {	6✔
118	return err	×
119	}	×
120
121	b.bestHeightMtx.Lock()	6✔
122	b.bestHeight = uint32(bestHeight)	6✔
123	b.bestHeightMtx.Unlock()	6✔
124		6✔
125	b.blockQueue.Start()	6✔
126		6✔
127	b.wg.Add(1)	6✔
128	go b.chainFilterer()	6✔
129		6✔
130	return nil	6✔
131	}
132
133	// Stop stops all goroutines which we launched by the prior call to the Start
134	// method.
135	//
136	// NOTE: This is part of the FilteredChainView interface.
137	func (b *BtcdFilteredChainView) Stop() error {	6✔
138	log.Debug("BtcdFilteredChainView stopping")	6✔
139	defer log.Debug("BtcdFilteredChainView stopped")	6✔
140		6✔
141	// Already shutting down?	6✔
142	if atomic.AddInt32(&b.stopped, 1) != 1 {	6✔
143	return nil	×
144	}	×
145
146	// Shutdown the rpc client, this gracefully disconnects from btcd, and
147	// cleans up all related resources.
148	b.btcdConn.Shutdown()	6✔
149	b.btcdConn.WaitForShutdown()	6✔
150		6✔
151	b.blockQueue.Stop()	6✔
152		6✔
153	close(b.quit)	6✔
154	b.wg.Wait()	6✔
155		6✔
156	return nil	6✔
157	}
158
159	// onFilteredBlockConnected is called for each block that's connected to the
160	// end of the main chain. Based on our current chain filter, the block may or
161	// may not include any relevant transactions.
162	func (b *BtcdFilteredChainView) onFilteredBlockConnected(height int32,
163	header wire.BlockHeader, txns []btcutil.Tx) {	463✔
164		463✔
165	mtxs := make([]*wire.MsgTx, len(txns))	463✔
166	b.filterMtx.Lock()	463✔
167	for i, tx := range txns {	466✔
168	mtx := tx.MsgTx()	3✔
169	mtxs[i] = mtx	3✔
170		3✔
171	for _, txIn := range mtx.TxIn {	6✔
172	// We can delete this outpoint from the chainFilter, as	3✔
173	// we just received a block where it was spent. In case	3✔
174	// of a reorg, this outpoint might get "un-spent", but	3✔
175	// that's okay since it would never be wise to consider	3✔
176	// the channel open again (since a spending transaction	3✔
177	// exists on the network).	3✔
178	delete(b.chainFilter, txIn.PreviousOutPoint)	3✔
179	}	3✔
180
181	}
182	b.filterMtx.Unlock()	463✔
183		463✔
184	// We record the height of the last connected block added to the	463✔
185	// blockQueue such that we can scan up to this height in case of	463✔
186	// a rescan. It must be protected by a mutex since a filter update	463✔
187	// might be trying to read it concurrently.	463✔
188	b.bestHeightMtx.Lock()	463✔
189	b.bestHeight = uint32(height)	463✔
190	b.bestHeightMtx.Unlock()	463✔
191		463✔
192	block := &FilteredBlock{	463✔
193	Hash: header.BlockHash(),	463✔
194	Height: uint32(height),	463✔
195	Transactions: mtxs,	463✔
196	}	463✔
197		463✔
198	b.blockQueue.Add(&blockEvent{	463✔
199	eventType: connected,	463✔
200	block: block,	463✔
201	})	463✔
202	}
203
204	// onFilteredBlockDisconnected is a callback which is executed once a block is
205	// disconnected from the end of the main chain.
206	func (b *BtcdFilteredChainView) onFilteredBlockDisconnected(height int32,
207	header *wire.BlockHeader) {	416✔
208		416✔
209	log.Debugf("got disconnected block at height %d: %v", height,	416✔
210	header.BlockHash())	416✔
211		416✔
212	filteredBlock := &FilteredBlock{	416✔
213	Hash: header.BlockHash(),	416✔
214	Height: uint32(height),	416✔
215	}	416✔
216		416✔
217	b.blockQueue.Add(&blockEvent{	416✔
218	eventType: disconnected,	416✔
219	block: filteredBlock,	416✔
220	})	416✔
221	}	416✔
222
223	// filterBlockReq houses a request to manually filter a block specified by
224	// block hash.
225	type filterBlockReq struct {
226	blockHash *chainhash.Hash
227	resp chan *FilteredBlock
228	err chan error
229	}
230
231	// FilterBlock takes a block hash, and returns a FilteredBlocks which is the
232	// result of applying the current registered UTXO sub-set on the block
233	// corresponding to that block hash. If any watched UTOX's are spent by the
234	// selected lock, then the internal chainFilter will also be updated.
235	//
236	// NOTE: This is part of the FilteredChainView interface.
237	func (b BtcdFilteredChainView) FilterBlock(blockHash chainhash.Hash) (*FilteredBlock, error) {	2✔
238	req := &filterBlockReq{	2✔
239	blockHash: blockHash,	2✔
240	resp: make(chan *FilteredBlock, 1),	2✔
241	err: make(chan error, 1),	2✔
242	}	2✔
243		2✔
244	select {	2✔
245	case b.filterBlockReqs <- req:	2✔
246	case <-b.quit:	×
247	return nil, fmt.Errorf("FilteredChainView shutting down")	×
248	}
249
250	return <-req.resp, <-req.err	2✔
251	}
252
253	// chainFilterer is the primary goroutine which: listens for new blocks coming
254	// and dispatches the relevant FilteredBlock notifications, updates the filter
255	// due to requests by callers, and finally is able to preform targeted block
256	// filtration.
257	//
258	// TODO(roasbeef): change to use loadfilter RPC's
259	func (b *BtcdFilteredChainView) chainFilterer() {	6✔
260	defer b.wg.Done()	6✔
261		6✔
262	// filterBlock is a helper function that scans the given block, and	6✔
263	// notes which transactions spend outputs which are currently being	6✔
264	// watched. Additionally, the chain filter will also be updated by	6✔
265	// removing any spent outputs.	6✔
266	filterBlock := func(blk wire.MsgBlock) []wire.MsgTx {	8✔
267	b.filterMtx.Lock()	2✔
268	defer b.filterMtx.Unlock()	2✔
269		2✔
270	var filteredTxns []*wire.MsgTx	2✔
271	for _, tx := range blk.Transactions {	6✔
272	var txAlreadyFiltered bool	4✔
273	for _, txIn := range tx.TxIn {	8✔
274	prevOp := txIn.PreviousOutPoint	4✔
275	if _, ok := b.chainFilter[prevOp]; !ok {	6✔
276	continue	2✔
277	}
278
279	delete(b.chainFilter, prevOp)	3✔
280		3✔
281	// Only add this txn to our list of filtered	3✔
282	// txns if it is the first previous outpoint to	3✔
283	// cause a match.	3✔
284	if txAlreadyFiltered {	3✔
285	continue	×
286	}
287
288	filteredTxns = append(filteredTxns, tx.Copy())	3✔
289	txAlreadyFiltered = true	3✔
290
291	}
292	}
293
294	return filteredTxns	2✔
295	}
296
297	decodeJSONBlock := func(block *btcjson.RescannedBlock,	6✔
298	height uint32) (*FilteredBlock, error) {	7✔
299	hash, err := chainhash.NewHashFromStr(block.Hash)	1✔
300	if err != nil {	1✔
301	return nil, err	×
302		×
303	}	×
304	txs := make([]*wire.MsgTx, 0, len(block.Transactions))	1✔
305	for _, str := range block.Transactions {	2✔
306	b, err := hex.DecodeString(str)	1✔
307	if err != nil {	1✔
308	return nil, err	×
309	}	×
310	tx := &wire.MsgTx{}	1✔
311	err = tx.Deserialize(bytes.NewReader(b))	1✔
312	if err != nil {	1✔
313	return nil, err	×
314	}	×
315	txs = append(txs, tx)	1✔
316	}
317	return &FilteredBlock{	1✔
318	Hash: *hash,	1✔
319	Height: height,	1✔
320	Transactions: txs,	1✔
321	}, nil	1✔
322	}
323
324	for {	16✔
325	select {	10✔
326	// The caller has just sent an update to the current chain
327	// filter, so we'll apply the update, possibly rewinding our
328	// state partially.
329	case update := <-b.filterUpdates:	4✔
330		4✔
331	// First, we'll add all the new UTXO's to the set of	4✔
332	// watched UTXO's, eliminating any duplicates in the	4✔
333	// process.	4✔
334	log.Tracef("Updating chain filter with new UTXO's: %v",	4✔
335	update.newUtxos)	4✔
336		4✔
337	b.filterMtx.Lock()	4✔
338	for _, newOp := range update.newUtxos {	10✔
339	b.chainFilter[newOp] = struct{}{}	6✔
340	}	6✔
341	b.filterMtx.Unlock()	4✔
342		4✔
343	// Apply the new TX filter to btcd, which will cause	4✔
344	// all following notifications from and calls to it	4✔
345	// return blocks filtered with the new filter.	4✔
346	b.btcdConn.LoadTxFilter(false, []btcutil.Address{},	4✔
347	update.newUtxos)	4✔
348		4✔
349	// All blocks gotten after we loaded the filter will	4✔
350	// have the filter applied, but we will need to rescan	4✔
351	// the blocks up to the height of the block we last	4✔
352	// added to the blockQueue.	4✔
353	b.bestHeightMtx.Lock()	4✔
354	bestHeight := b.bestHeight	4✔
355	b.bestHeightMtx.Unlock()	4✔
356		4✔
357	// If the update height matches our best known height,	4✔
358	// then we don't need to do any rewinding.	4✔
359	if update.updateHeight == bestHeight {	7✔
360	continue	3✔
361	}
362
363	// Otherwise, we'll rewind the state to ensure the
364	// caller doesn't miss any relevant notifications.
365	// Starting from the height _after_ the update height,
366	// we'll walk forwards, rescanning one block at a time
367	// with btcd applying the newly loaded filter to each
368	// block.
369	for i := update.updateHeight + 1; i < bestHeight+1; i++ {	2✔
370	blockHash, err := b.btcdConn.GetBlockHash(int64(i))	1✔
371	if err != nil {	1✔
372	log.Warnf("Unable to get block hash "+	×
373	"for block at height %d: %v",	×
374	i, err)	×
375	continue	×
376	}
377
378	// To avoid dealing with the case where a reorg
379	// is happening while we rescan, we scan one
380	// block at a time, skipping blocks that might
381	// have gone missing.
382	rescanned, err := b.btcdConn.RescanBlocks(	1✔
383	[]chainhash.Hash{*blockHash})	1✔
384	if err != nil {	1✔
385	log.Warnf("Unable to rescan block "+	×
386	"with hash %v at height %d: %v",	×
387	blockHash, i, err)	×
388	continue	×
389	}
390
391	// If no block was returned from the rescan, it
392	// means no matching transactions were found.
393	if len(rescanned) != 1 {	1✔
394	log.Tracef("rescan of block %v at "+	×
395	"height=%d yielded no "+	×
396	"transactions", blockHash, i)	×
397	continue	×
398	}
399	decoded, err := decodeJSONBlock(	1✔
400	&rescanned[0], uint32(i))	1✔
401	if err != nil {	1✔
402	log.Errorf("Unable to decode block: %v",	×
403	err)	×
404	continue	×
405	}
406	b.blockQueue.Add(&blockEvent{	1✔
407	eventType: connected,	1✔
408	block: decoded,	1✔
409	})	1✔
410	}
411
412	// We've received a new request to manually filter a block.
413	case req := <-b.filterBlockReqs:	2✔
414	// First we'll fetch the block itself as well as some	2✔
415	// additional information including its height.	2✔
416	block, err := b.GetBlock(req.blockHash)	2✔
417	if err != nil {	2✔
418	req.err <- err	×
419	req.resp <- nil	×
420	continue	×
421	}
422	header, err := b.btcdConn.GetBlockHeaderVerbose(req.blockHash)	2✔
423	if err != nil {	2✔
424	req.err <- err	×
425	req.resp <- nil	×
426	continue	×
427	}
428
429	// Once we have this info, we can directly filter the
430	// block and dispatch the proper notification.
431	req.resp <- &FilteredBlock{	2✔
432	Hash: *req.blockHash,	2✔
433	Height: uint32(header.Height),	2✔
434	Transactions: filterBlock(block),	2✔
435	}	2✔
436	req.err <- err	2✔
437
438	case <-b.quit:	6✔
439	return	6✔
440	}
441	}
442	}
443
444	// filterUpdate is a message sent to the chainFilterer to update the current
445	// chainFilter state.
446	type filterUpdate struct {
447	newUtxos []wire.OutPoint
448	updateHeight uint32
449	}
450
451	// UpdateFilter updates the UTXO filter which is to be consulted when creating
452	// FilteredBlocks to be sent to subscribed clients. This method is cumulative
453	// meaning repeated calls to this method should _expand_ the size of the UTXO
454	// sub-set currently being watched. If the set updateHeight is _lower_ than
455	// the best known height of the implementation, then the state should be
456	// rewound to ensure all relevant notifications are dispatched.
457	//
458	// NOTE: This is part of the FilteredChainView interface.
459	func (b *BtcdFilteredChainView) UpdateFilter(ops []graphdb.EdgePoint,
460	updateHeight uint32) error {	4✔
461		4✔
462	newUtxos := make([]wire.OutPoint, len(ops))	4✔
463	for i, op := range ops {	10✔
464	newUtxos[i] = op.OutPoint	6✔
465	}	6✔
466
467	select {	4✔
468
469	case b.filterUpdates <- filterUpdate{
470	newUtxos: newUtxos,
471	updateHeight: updateHeight,
472	}:	4✔
473	return nil	4✔
474
475	case <-b.quit:	×
476	return fmt.Errorf("chain filter shutting down")	×
477	}
478	}
479
480	// FilteredBlocks returns the channel that filtered blocks are to be sent over.
481	// Each time a block is connected to the end of a main chain, and appropriate
482	// FilteredBlock which contains the transactions which mutate our watched UTXO
483	// set is to be returned.
484	//
485	// NOTE: This is part of the FilteredChainView interface.
486	func (b BtcdFilteredChainView) FilteredBlocks() <-chan FilteredBlock {	5✔
487	return b.blockQueue.newBlocks	5✔
488	}	5✔
489
490	// DisconnectedBlocks returns a receive only channel which will be sent upon
491	// with the empty filtered blocks of blocks which are disconnected from the
492	// main chain in the case of a re-org.
493	//
494	// NOTE: This is part of the FilteredChainView interface.
495	func (b BtcdFilteredChainView) DisconnectedBlocks() <-chan FilteredBlock {	2✔
496	return b.blockQueue.staleBlocks	2✔
497	}	2✔
498
499	// GetBlock is used to retrieve the block with the given hash. This function
500	// wraps the blockCache's GetBlock function.
501	func (b BtcdFilteredChainView) GetBlock(hash chainhash.Hash) (
502	*wire.MsgBlock, error) {	2✔
503		2✔
504	return b.blockCache.GetBlock(hash, b.btcdConn.GetBlock)	2✔
505	}	2✔

lightningnetwork / lnd / 13586005509

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous