From 405d4e5f73cc9879c69266669294d1da659cd2f2 Mon Sep 17 00:00:00 2001 From: Keagan McClelland Date: Wed, 12 Jul 2023 16:26:12 -0600 Subject: [PATCH] chainntnfs: introduce system for chain state tracking and querying This change adds a new subsystem that is responsible for providing an up to date view of some global chainstate parameters. --- chainntnfs/best_block_view.go | 135 +++++++++++++++++++++++++++++ chainntnfs/best_block_view_test.go | 112 ++++++++++++++++++++++++ 2 files changed, 247 insertions(+) create mode 100644 chainntnfs/best_block_view.go create mode 100644 chainntnfs/best_block_view_test.go diff --git a/chainntnfs/best_block_view.go b/chainntnfs/best_block_view.go new file mode 100644 index 000000000..c043e68e7 --- /dev/null +++ b/chainntnfs/best_block_view.go @@ -0,0 +1,135 @@ +package chainntnfs + +import ( + "errors" + "fmt" + "sync" + "sync/atomic" + + "github.com/btcsuite/btcd/wire" +) + +// BestBlockView is an interface that allows the querying of the most +// up-to-date blockchain state with low overhead. Valid implementations of this +// interface must track the latest chain state. +type BestBlockView interface { + // BestHeight gets the most recent block height known to the view. + BestHeight() (uint32, error) + + // BestBlockHeader gets the most recent block header known to the view. + BestBlockHeader() (*wire.BlockHeader, error) +} + +// BestBlockTracker is a tiny subsystem that tracks the blockchain tip +// and saves the most recent tip information in memory for querying. It is a +// valid implementation of BestBlockView and additionally includes +// methods for starting and stopping the system. +type BestBlockTracker struct { + notifier ChainNotifier + blockNtfnStream *BlockEpochEvent + current atomic.Pointer[BlockEpoch] + mu sync.Mutex + quit chan struct{} + wg sync.WaitGroup +} + +// This is a compile time check to ensure that BestBlockTracker implements +// BestBlockView. +var _ BestBlockView = (*BestBlockTracker)(nil) + +// NewBestBlockTracker creates a new BestBlockTracker that isn't running yet. +// It will not provide up to date information unless it has been started. The +// ChainNotifier parameter must also be started prior to starting the +// BestBlockTracker. +func NewBestBlockTracker(chainNotifier ChainNotifier) *BestBlockTracker { + return &BestBlockTracker{ + notifier: chainNotifier, + blockNtfnStream: nil, + quit: make(chan struct{}), + } +} + +// BestHeight gets the most recent block height known to the +// BestBlockTracker. +func (t *BestBlockTracker) BestHeight() (uint32, error) { + epoch := t.current.Load() + if epoch == nil { + return 0, errors.New("best block height not yet known") + } + + return uint32(epoch.Height), nil +} + +// BestBlockHeader gets the most recent block header known to the +// BestBlockTracker. +func (t *BestBlockTracker) BestBlockHeader() (*wire.BlockHeader, error) { + epoch := t.current.Load() + if epoch == nil { + return nil, errors.New("best block header not yet known") + } + + return epoch.BlockHeader, nil +} + +// updateLoop is a helper that subscribes to the underlying BlockEpochEvent +// stream and updates the internal values to match the new BlockEpochs that +// are discovered. +// +// MUST be run as a goroutine. +func (t *BestBlockTracker) updateLoop() { + defer t.wg.Done() + for { + select { + case epoch, ok := <-t.blockNtfnStream.Epochs: + if !ok { + Log.Error("dead epoch stream in " + + "BestBlockTracker") + + return + } + t.current.Store(epoch) + case <-t.quit: + t.current.Store(nil) + return + } + } +} + +// Start starts the BestBlockTracker. It is an error to start it if it +// is already started. +func (t *BestBlockTracker) Start() error { + t.mu.Lock() + defer t.mu.Unlock() + + if t.blockNtfnStream != nil { + return fmt.Errorf("BestBlockTracker is already started") + } + + var err error + t.blockNtfnStream, err = t.notifier.RegisterBlockEpochNtfn(nil) + if err != nil { + return err + } + + t.wg.Add(1) + go t.updateLoop() + + return nil +} + +// Stop stops the BestBlockTracker. It is an error to stop it if it has +// not been started or if it has already been stopped. +func (t *BestBlockTracker) Stop() error { + t.mu.Lock() + defer t.mu.Unlock() + + if t.blockNtfnStream == nil { + return fmt.Errorf("BestBlockTracker is not running") + } + close(t.quit) + t.wg.Wait() + t.blockNtfnStream.Cancel() + t.blockNtfnStream = nil + + return nil +} diff --git a/chainntnfs/best_block_view_test.go b/chainntnfs/best_block_view_test.go new file mode 100644 index 000000000..2a55bc8b0 --- /dev/null +++ b/chainntnfs/best_block_view_test.go @@ -0,0 +1,112 @@ +package chainntnfs_test + +import ( + "math/rand" + "reflect" + "testing" + "testing/quick" + "time" + + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/wire" + "github.com/lightningnetwork/lnd/chainntnfs" + "github.com/lightningnetwork/lnd/lntest/mock" + "github.com/lightningnetwork/lnd/lntest/wait" + "github.com/stretchr/testify/require" +) + +type blockEpoch chainntnfs.BlockEpoch + +func (blockEpoch) Generate(r *rand.Rand, size int) reflect.Value { + var chainHash, prevBlockHash, merkleRootHash chainhash.Hash + r.Read(chainHash[:]) + r.Read(prevBlockHash[:]) + r.Read(merkleRootHash[:]) + + return reflect.ValueOf(blockEpoch(chainntnfs.BlockEpoch{ + Hash: &chainHash, + Height: r.Int31n(1000000), + BlockHeader: &wire.BlockHeader{ + Version: 2, + PrevBlock: prevBlockHash, + MerkleRoot: merkleRootHash, + Timestamp: time.Now(), + Bits: r.Uint32(), + Nonce: r.Uint32(), + }, + })) +} + +// TestBestBlockTracker ensures that the most recent event pushed on the +// underlying EpochChan is remembered by the BestBlockView functions as well +// as testing the idempotence of the BestBlockView interface. +func TestBestBlockTracker(t *testing.T) { + t.Parallel() + + notifier := &mock.ChainNotifier{ + SpendChan: nil, + EpochChan: make(chan *chainntnfs.BlockEpoch), + ConfChan: nil, + } + + chainNotifierI := chainntnfs.ChainNotifier(notifier) + + tracker := chainntnfs.NewBestBlockTracker(chainNotifierI) + require.Nil(t, tracker.Start(), + "BestBlockTacker could not be started") + + // we have to limit test cases because the poll interval of + // wait.Predicate isn't tight enough to support the usual 100 + cfg := quick.Config{MaxCount: 50} + correctness := func(epochRand blockEpoch) bool { + epoch := chainntnfs.BlockEpoch(epochRand) + notifier.EpochChan <- &epoch + + // wait for new block to propagate + err := wait.Predicate( + func() bool { + _, err := tracker.BestHeight() + return err == nil + }, + 1*time.Second, + ) + require.Nil(t, err, + "BestBlockTracker: block propagation timeout") + + height, _ := tracker.BestHeight() + header, _ := tracker.BestBlockHeader() + + return height == uint32(epoch.Height) && + header == epoch.BlockHeader + } + idempotence := func(epochRand blockEpoch) bool { + epoch := chainntnfs.BlockEpoch(epochRand) + notifier.EpochChan <- &epoch + + // wait for new block to propagate + err := wait.Predicate( + func() bool { + _, err := tracker.BestHeight() + return err == nil + }, + 1*time.Second, + ) + require.Nil(t, err, + "ChainStateTracker: block propagation timeout") + + height0, _ := tracker.BestHeight() + height1, _ := tracker.BestHeight() + header0, _ := tracker.BestBlockHeader() + header1, _ := tracker.BestBlockHeader() + + return height0 == height1 && header0 == header1 + } + err := quick.Check(correctness, &cfg) + require.Nil(t, err, + "ChainStateTracker does not give up to date info: %v", err) + + require.Nil(t, quick.Check(idempotence, &cfg), + "ChainStateTracker is not idempotent") + + require.Nil(t, tracker.Stop(), "ChainStateTracker could not be stopped") +}