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

@@ -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
+}

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")
+}