diff --git a/contractcourt/chain_arbitrator.go b/contractcourt/chain_arbitrator.go index 34ed7d7ff..524958f48 100644 --- a/contractcourt/chain_arbitrator.go +++ b/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]", - &c.cfg.Budget) + // Set the current beat. + 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. diff --git a/contractcourt/chain_arbitrator_test.go b/contractcourt/chain_arbitrator_test.go index de6d69900..622686f76 100644 --- a/contractcourt/chain_arbitrator_test.go +++ b/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() { diff --git a/server.go b/server.go index 799bfde6b..f83333bc6 100644 --- a/server.go +++ b/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 +} diff --git a/sweep/fee_bumper.go b/sweep/fee_bumper.go index ad02ac619..a43f87585 100644 --- a/sweep/fee_bumper.go +++ b/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() diff --git a/sweep/sweeper.go b/sweep/sweeper.go index b91a6808f..0f2675e8f 100644 --- a/sweep/sweeper.go +++ b/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()