multi: use utils.RegisterBlockEpochNtfnWithRetry

This is needed after LND release v0.20.0-beta.rc3, which delays the chain
notifier until after headers finish syncing (commit c6f458e478), so when Loop
boots every static-address manager calls RegisterBlockEpochNtfn immediately,
sees "chain notifier RPC is still in the process of starting", treats it as
fatal, and Loop shuts down; LND v0.20.0-beta.rc2 didn't have the delay.

Author: starius

instantout/manager.go

@@ -11,6 +11,7 @@ import (
 	"github.com/lightninglabs/loop/fsm"
 	"github.com/lightninglabs/loop/instantout/reservation"
 	"github.com/lightninglabs/loop/swapserverrpc"
+	"github.com/lightninglabs/loop/utils"
 	"github.com/lightningnetwork/lnd/lntypes"
 )
 
@@ -68,8 +69,8 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
 		return err
 	}
 
-	newBlockChan, newBlockErrChan, err := m.cfg.ChainNotifier.
-		RegisterBlockEpochNtfn(ctx)
+	newBlockChan, newBlockErrChan, err :=
+		utils.RegisterBlockEpochNtfnWithRetry(ctx, m.cfg.ChainNotifier)
 	if err != nil {
 		close(initChan)
 		return err

instantout/reservation/actions.go

@@ -7,6 +7,7 @@ import (
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/lightninglabs/loop/fsm"
 	"github.com/lightninglabs/loop/swapserverrpc"
+	"github.com/lightninglabs/loop/utils"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 )
 
@@ -103,8 +104,8 @@ func (f *FSM) SubscribeToConfirmationAction(ctx context.Context,
 		return f.HandleError(err)
 	}
 
-	blockChan, errBlockChan, err := f.cfg.ChainNotifier.RegisterBlockEpochNtfn(
-		callCtx,
+	blockChan, errBlockChan, err := utils.RegisterBlockEpochNtfnWithRetry(
+		callCtx, f.cfg.ChainNotifier,
 	)
 	if err != nil {
 		f.Errorf("unable to subscribe to block notifications: %v", err)
@@ -158,8 +159,9 @@ func (f *FSM) AsyncWaitForExpiredOrSweptAction(ctx context.Context,
 
 	notifCtx, cancel := context.WithCancel(ctx)
 
-	blockHeightChan, errEpochChan, err := f.cfg.ChainNotifier.
-		RegisterBlockEpochNtfn(notifCtx)
+	blockHeightChan, errEpochChan, err := utils.RegisterBlockEpochNtfnWithRetry(
+		notifCtx, f.cfg.ChainNotifier,
+	)
 	if err != nil {
 		cancel()
 		return f.HandleError(err)

instantout/reservation/manager.go

@@ -11,6 +11,7 @@ import (
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/lightninglabs/loop/fsm"
 	reservationrpc "github.com/lightninglabs/loop/swapserverrpc"
+	"github.com/lightninglabs/loop/utils"
 )
 
 // Manager manages the reservation state machines.
@@ -49,8 +50,8 @@ func (m *Manager) Run(ctx context.Context, height int32,
 		return err
 	}
 
-	newBlockChan, newBlockErrChan, err := m.cfg.ChainNotifier.
-		RegisterBlockEpochNtfn(runCtx)
+	newBlockChan, newBlockErrChan, err :=
+		utils.RegisterBlockEpochNtfnWithRetry(runCtx, m.cfg.ChainNotifier)
 	if err != nil {
 		return err
 	}

staticaddr/address/manager.go

@@ -16,6 +16,7 @@ import (
 	"github.com/lightninglabs/loop/staticaddr/version"
 	"github.com/lightninglabs/loop/swap"
 	staticaddressrpc "github.com/lightninglabs/loop/swapserverrpc"
+	"github.com/lightninglabs/loop/utils"
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/keychain"
 	"github.com/lightningnetwork/lnd/lnwallet"
@@ -69,7 +70,7 @@ func NewManager(cfg *ManagerConfig, currentHeight int32) *Manager {
 // Run runs the address manager.
 func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
 	newBlockChan, newBlockErrChan, err :=
-		m.cfg.ChainNotifier.RegisterBlockEpochNtfn(ctx)
+		utils.RegisterBlockEpochNtfnWithRetry(ctx, m.cfg.ChainNotifier)
 
 	if err != nil {
 		return err

staticaddr/deposit/manager.go

@@ -15,6 +15,7 @@ import (
 	"github.com/lightninglabs/loop"
 	"github.com/lightninglabs/loop/fsm"
 	staticaddressrpc "github.com/lightninglabs/loop/swapserverrpc"
+	"github.com/lightninglabs/loop/utils"
 	"github.com/lightningnetwork/lnd/lnrpc/walletrpc"
 	"github.com/lightningnetwork/lnd/lnwallet"
 )
@@ -101,7 +102,8 @@ func NewManager(cfg *ManagerConfig) *Manager {
 
 // Run runs the address manager.
 func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
-	newBlockChan, newBlockErrChan, err := m.cfg.ChainNotifier.RegisterBlockEpochNtfn(ctx) //nolint:lll
+	newBlockChan, newBlockErrChan, err :=
+		utils.RegisterBlockEpochNtfnWithRetry(ctx, m.cfg.ChainNotifier)
 	if err != nil {
 		log.Errorf("unable to register block epoch notifier: %v", err)
 

staticaddr/loopin/actions.go

@@ -20,6 +20,7 @@ import (
 	"github.com/lightninglabs/loop/staticaddr/version"
 	"github.com/lightninglabs/loop/swap"
 	"github.com/lightninglabs/loop/swapserverrpc"
+	"github.com/lightninglabs/loop/utils"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/invoices"
@@ -497,8 +498,9 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context,
 	}
 
 	// Subscribe to new blocks.
-	registerBlocks := f.cfg.ChainNotifier.RegisterBlockEpochNtfn
-	blockChan, blockChanErr, err := registerBlocks(ctx)
+	blockChan, blockChanErr, err := utils.RegisterBlockEpochNtfnWithRetry(
+		ctx, f.cfg.ChainNotifier,
+	)
 	if err != nil {
 		err = fmt.Errorf("unable to subscribe to new blocks: %w", err)
 

staticaddr/loopin/manager.go

@@ -22,6 +22,7 @@ import (
 	"github.com/lightninglabs/loop/staticaddr/address"
 	"github.com/lightninglabs/loop/staticaddr/deposit"
 	"github.com/lightninglabs/loop/swapserverrpc"
+	"github.com/lightninglabs/loop/utils"
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/lnwallet"
@@ -158,8 +159,8 @@ func NewManager(cfg *Config, currentHeight uint32) *Manager {
 
 // Run runs the static address loop-in manager.
 func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
-	registerBlockNtfn := m.cfg.ChainNotifier.RegisterBlockEpochNtfn
-	newBlockChan, newBlockErrChan, err := registerBlockNtfn(ctx)
+	newBlockChan, newBlockErrChan, err :=
+		utils.RegisterBlockEpochNtfnWithRetry(ctx, m.cfg.ChainNotifier)
 	if err != nil {
 		log.Errorf("unable to register for block notifications: %v",
 			err)

staticaddr/withdraw/manager.go

@@ -20,6 +20,7 @@ import (
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/loop/staticaddr/deposit"
 	staticaddressrpc "github.com/lightninglabs/loop/swapserverrpc"
+	"github.com/lightninglabs/loop/utils"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/lnrpc/walletrpc"
@@ -154,7 +155,7 @@ func NewManager(cfg *ManagerConfig, currentHeight uint32) *Manager {
 // Run runs the deposit withdrawal manager.
 func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
 	newBlockChan, newBlockErrChan, err :=
-		m.cfg.ChainNotifier.RegisterBlockEpochNtfn(ctx)
+		utils.RegisterBlockEpochNtfnWithRetry(ctx, m.cfg.ChainNotifier)
 
 	if err != nil {
 		log.Errorf("unable to register for block epoch "+

sweepbatcher/sweep_batch.go

@@ -842,8 +842,9 @@ func (b *batch) Run(ctx context.Context) error {
 	clock := b.cfg.clock
 	startTime := clock.Now()
 
-	blockChan, blockErrChan, err :=
-		b.chainNotifier.RegisterBlockEpochNtfn(runCtx)
+	blockChan, blockErrChan, err := utils.RegisterBlockEpochNtfnWithRetry(
+		runCtx, b.chainNotifier,
+	)
 	if err != nil {
 		return fmt.Errorf("block registration error: %w", err)
 	}