multi: start consumers with a starting blockbeat

This is needed so the consumers have an initial state about the current
block.

contractcourt/chain_arbitrator.go

@@ -570,13 +570,16 @@ func (c *ChainArbitrator) ResolveContract(chanPoint wire.OutPoint) error {
 }
 
 // Start launches all goroutines that the ChainArbitrator needs to operate.
-func (c *ChainArbitrator) Start() error {
+func (c *ChainArbitrator) Start(beat chainio.Blockbeat) error {
 	if !atomic.CompareAndSwapInt32(&c.started, 0, 1) {
 		return nil
 	}
 
-	log.Infof("ChainArbitrator starting with config: budget=[%v]",
+	// Set the current beat.
-		&c.cfg.Budget)
+	c.beat = beat
+
+	log.Infof("ChainArbitrator starting at height %d with budget=[%v]",
+		&c.cfg.Budget, c.beat.Height())
 
 	// First, we'll fetch all the channels that are still open, in order to
 	// collect them within our set of active contracts.

contractcourt/chain_arbitrator_test.go

@@ -90,7 +90,8 @@ func TestChainArbitratorRepublishCloses(t *testing.T) {
 		chainArbCfg, db,
 	)
 
-	if err := chainArb.Start(); err != nil {
+	beat := newBeatFromHeight(0)
+	if err := chainArb.Start(beat); err != nil {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
@@ -173,7 +174,8 @@ func TestResolveContract(t *testing.T) {
 	chainArb := NewChainArbitrator(
 		chainArbCfg, db,
 	)
-	if err := chainArb.Start(); err != nil {
+	beat := newBeatFromHeight(0)
+	if err := chainArb.Start(beat); err != nil {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {

server.go

@@ -28,6 +28,7 @@ import (
 	"github.com/lightningnetwork/lnd/aliasmgr"
 	"github.com/lightningnetwork/lnd/autopilot"
 	"github.com/lightningnetwork/lnd/brontide"
+	"github.com/lightningnetwork/lnd/chainio"
 	"github.com/lightningnetwork/lnd/chainreg"
 	"github.com/lightningnetwork/lnd/chanacceptor"
 	"github.com/lightningnetwork/lnd/chanbackup"
@@ -2073,6 +2074,12 @@ func (c cleaner) run() {
 //
 //nolint:funlen
 func (s *server) Start() error {
+	// Get the current blockbeat.
+	beat, err := s.getStartingBeat()
+	if err != nil {
+		return err
+	}
+
 	var startErr error
 
 	// If one sub system fails to start, the following code ensures that the
@@ -2167,13 +2174,13 @@ func (s *server) Start() error {
 		}
 
 		cleanup = cleanup.add(s.txPublisher.Stop)
-		if err := s.txPublisher.Start(); err != nil {
+		if err := s.txPublisher.Start(beat); err != nil {
 			startErr = err
 			return
 		}
 
 		cleanup = cleanup.add(s.sweeper.Stop)
-		if err := s.sweeper.Start(); err != nil {
+		if err := s.sweeper.Start(beat); err != nil {
 			startErr = err
 			return
 		}
@@ -2218,7 +2225,7 @@ func (s *server) Start() error {
 		}
 
 		cleanup = cleanup.add(s.chainArb.Stop)
-		if err := s.chainArb.Start(); err != nil {
+		if err := s.chainArb.Start(beat); err != nil {
 			startErr = err
 			return
 		}
@@ -5152,3 +5159,35 @@ func (s *server) fetchClosedChannelSCIDs() map[lnwire.ShortChannelID]struct{} {
 
 	return closedSCIDs
 }
+
+// getStartingBeat returns the current beat. This is used during the startup to
+// initialize blockbeat consumers.
+func (s *server) getStartingBeat() (*chainio.Beat, error) {
+	// beat is the current blockbeat.
+	var beat *chainio.Beat
+
+	// We should get a notification with the current best block immediately
+	// by passing a nil block.
+	blockEpochs, err := s.cc.ChainNotifier.RegisterBlockEpochNtfn(nil)
+	if err != nil {
+		return beat, fmt.Errorf("register block epoch ntfn: %w", err)
+	}
+	defer blockEpochs.Cancel()
+
+	// We registered for the block epochs with a nil request. The notifier
+	// should send us the current best block immediately. So we need to
+	// wait for it here because we need to know the current best height.
+	select {
+	case bestBlock := <-blockEpochs.Epochs:
+		srvrLog.Infof("Received initial block %v at height %d",
+			bestBlock.Hash, bestBlock.Height)
+
+		// Update the current blockbeat.
+		beat = chainio.NewBeat(*bestBlock)
+
+	case <-s.quit:
+		srvrLog.Debug("LND shutting down")
+	}
+
+	return beat, nil
+}

sweep/fee_bumper.go

@@ -795,13 +795,16 @@ type monitorRecord struct {
 
 // Start starts the publisher by subscribing to block epoch updates and kicking
 // off the monitor loop.
-func (t *TxPublisher) Start() error {
+func (t *TxPublisher) Start(beat chainio.Blockbeat) error {
 	log.Info("TxPublisher starting...")
 
 	if t.started.Swap(true) {
 		return fmt.Errorf("TxPublisher started more than once")
 	}
 
+	// Set the current height.
+	t.currentHeight.Store(beat.Height())
+
 	t.wg.Add(1)
 	go t.monitor()
 

sweep/sweeper.go

@@ -441,7 +441,7 @@ func New(cfg *UtxoSweeperConfig) *UtxoSweeper {
 }
 
 // Start starts the process of constructing and publish sweep txes.
-func (s *UtxoSweeper) Start() error {
+func (s *UtxoSweeper) Start(beat chainio.Blockbeat) error {
 	if !atomic.CompareAndSwapUint32(&s.started, 0, 1) {
 		return nil
 	}
@@ -452,6 +452,9 @@ func (s *UtxoSweeper) Start() error {
 	// not change from here on.
 	s.relayFeeRate = s.cfg.FeeEstimator.RelayFeePerKW()
 
+	// Set the current height.
+	s.currentHeight = beat.Height()
+
 	// Start sweeper main loop.
 	s.wg.Add(1)
 	go s.collector()