executor: move block registration retries to utils

This function will be reused in many other places.

Author: starius

executor.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -14,6 +13,7 @@ import (
 	"github.com/lightninglabs/loop/loopdb"
 	"github.com/lightninglabs/loop/sweep"
 	"github.com/lightninglabs/loop/sweepbatcher"
+	"github.com/lightninglabs/loop/utils"
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/queue"
 )
@@ -67,40 +67,15 @@ func (s *executor) run(mainCtx context.Context,
 	statusChan chan<- SwapInfo,
 	abandonChans map[lntypes.Hash]chan struct{}) error {
 
-	var (
-		err            error
-		blockEpochChan <-chan int32
-		blockErrorChan <-chan error
-		batcherErrChan chan error
+	blockEpochChan, blockErrorChan, err := utils.RegisterBlockEpochNtfnWithRetry(
+		mainCtx, s.lnd.ChainNotifier,
 	)
-
-	for {
-		blockEpochChan, blockErrorChan, err =
-			s.lnd.ChainNotifier.RegisterBlockEpochNtfn(mainCtx)
-		if err == nil {
-			break
-		}
-
-		if strings.Contains(err.Error(),
-			"in the process of starting") {
-
-			log.Warnf("LND chain notifier server not ready yet, " +
-				"retrying with delay")
-
-			// Give chain notifier some time to start and try to
-			// re-attempt block epoch subscription.
-			select {
-			case <-time.After(500 * time.Millisecond):
-				continue
-
-			case <-mainCtx.Done():
-				return err
-			}
-		}
-
+	if err != nil {
 		return err
 	}
 
+	var batcherErrChan chan error
+
 	// Before starting, make sure we have an up-to-date block height.
 	// Otherwise, we might reveal a preimage for a swap that is already
 	// expired.

utils/chainnotifier.go

@@ -0,0 +1,70 @@
+package utils
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"google.golang.org/grpc/status"
+)
+
+const (
+	// chainNotifierStartupMessage is returned by lnd while the chain
+	// notifier RPC sub-server is initialising.
+	chainNotifierStartupMessage = "chain notifier RPC is still in the " +
+		"process of starting"
+
+	// chainNotifierRetryBackoff is the delay used between subscription
+	// attempts while the chain notifier is still starting.
+	chainNotifierRetryBackoff = 500 * time.Millisecond
+)
+
+// BlockEpochRegistrar represents the ability to subscribe to block epoch
+// notifications.
+type BlockEpochRegistrar interface {
+	RegisterBlockEpochNtfn(ctx context.Context) (chan int32, chan error,
+		error)
+}
+
+// RegisterBlockEpochNtfnWithRetry keeps retrying block epoch subscriptions as
+// long as lnd reports that the chain notifier sub-server is still starting.
+func RegisterBlockEpochNtfnWithRetry(ctx context.Context,
+	registrar BlockEpochRegistrar) (chan int32, chan error, error) {
+
+	for {
+		blockChan, errChan, err := registrar.RegisterBlockEpochNtfn(ctx)
+		if err == nil {
+			return blockChan, errChan, nil
+		}
+
+		if !isChainNotifierStartingErr(err) {
+			return nil, nil, err
+		}
+
+		log.Warnf("Chain notifier RPC not ready yet, retrying: %v",
+			err)
+
+		select {
+		case <-time.After(chainNotifierRetryBackoff):
+			continue
+
+		case <-ctx.Done():
+			return nil, nil, ctx.Err()
+		}
+	}
+}
+
+// isChainNotifierStartingErr checks whether an error indicates that lnd's chain
+// notifier has not started yet.
+func isChainNotifierStartingErr(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	st, ok := status.FromError(err)
+	if ok && strings.Contains(st.Message(), chainNotifierStartupMessage) {
+		return true
+	}
+
+	return strings.Contains(err.Error(), chainNotifierStartupMessage)
+}

utils/chainnotifier_test.go

@@ -0,0 +1,82 @@
+package utils
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// stubBlockEpochRegistrar implements BlockEpochRegistrar and records the
+// number of attempts before succeeding.
+type stubBlockEpochRegistrar struct {
+	mu           sync.Mutex
+	attempts     int
+	succeedAfter int
+}
+
+// RegisterBlockEpochNtfn simulates a chain notifier that returns a startup
+// error until the configured number of attempts has been exhausted.
+func (s *stubBlockEpochRegistrar) RegisterBlockEpochNtfn(
+	context.Context) (chan int32, chan error, error) {
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	s.attempts++
+	if s.attempts <= s.succeedAfter {
+		return nil, nil, status.Error(
+			codes.Unknown, chainNotifierStartupMessage,
+		)
+	}
+
+	return make(chan int32), make(chan error), nil
+}
+
+// Attempts returns the total number of registration attempts made.
+func (s *stubBlockEpochRegistrar) Attempts() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s.attempts
+}
+
+// TestRegisterBlockEpochNtfnWithRetry ensures we retry until the notifier
+// becomes available.
+func TestRegisterBlockEpochNtfnWithRetry(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+	t.Cleanup(cancel)
+
+	stub := &stubBlockEpochRegistrar{
+		succeedAfter: 1,
+	}
+
+	blockChan, errChan, err := RegisterBlockEpochNtfnWithRetry(ctx, stub)
+	require.NoError(t, err)
+	require.NotNil(t, blockChan)
+	require.NotNil(t, errChan)
+	require.Equal(t, 2, stub.Attempts())
+}
+
+// TestRegisterBlockEpochNtfnWithRetryContextCancel ensures we propagate the
+// caller's context error if the notifier never becomes ready.
+func TestRegisterBlockEpochNtfnWithRetryContextCancel(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
+	t.Cleanup(cancel)
+
+	stub := &stubBlockEpochRegistrar{
+		succeedAfter: 100,
+	}
+
+	_, _, err := RegisterBlockEpochNtfnWithRetry(ctx, stub)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+	require.GreaterOrEqual(t, stub.Attempts(), 1)
+}

utils/log.go

@@ -0,0 +1,24 @@
+package utils
+
+import (
+	"github.com/btcsuite/btclog/v2"
+	"github.com/lightningnetwork/lnd/build"
+)
+
+// Subsystem defines the sub system name of this package.
+const Subsystem = "UTILS"
+
+// log is a logger that is initialized with no output filters. This means the
+// package will not perform any logging by default until the caller requests it.
+var log btclog.Logger
+
+// The default amount of logging is none.
+func init() {
+	UseLogger(build.NewSubLogger(Subsystem, nil))
+}
+
+// UseLogger uses a specified Logger to output package logging info. This should
+// be used in preference to SetLogWriter if the caller is also using btclog.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}