loopd: gate manager startup on chain notifier readiness

- add a readiness probe in Daemon.initialize that blocks until lnd’s
  chain notifier accepts (and maintains) a block-epoch subscription
  before starting static-address managers
- treat initial RegisterBlockEpochNtfn RPC failures and early stream
  shutdowns with exponential backoff, logging retries without crashing
  loopd
- keep a background drainer running on the readiness subscription so the
  notifier stays happy once startup proceeds

Author: ffranr

loopd/daemon.go

@@ -33,6 +33,8 @@ import (
 	"github.com/lightningnetwork/lnd/macaroons"
 	"go.etcd.io/bbolt"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/encoding/protojson"
 	"gopkg.in/macaroon-bakery.v2/bakery"
 )
@@ -45,6 +47,9 @@ var (
 	// errOnlyStartOnce is the error that is returned if the daemon is
 	// started more than once.
 	errOnlyStartOnce = fmt.Errorf("daemon can only be started once")
+
+	chainNotifierStartupMessage = "chain notifier RPC is still in the " +
+		"process of starting"
 )
 
 // ListenerCfg holds closures used to retrieve listeners for the gRPC services.
@@ -404,6 +409,15 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 
 	infof("Swap server address: %v", d.cfg.Server.Host)
 
+	infof("Waiting for chain notifier to become available 3")
+	err := waitForChainNotifierReady(
+		d.mainCtx, d.lnd.ChainNotifier,
+	)
+	if err != nil {
+		return fmt.Errorf("chain notifier unavailable: %w", err)
+	}
+	infof("Chain notifier ready")
+
 	// Check if we need to migrate the database.
 	if needSqlMigration(d.cfg) {
 		infof("Boltdb found, running migration")
@@ -1095,3 +1109,105 @@ func allowCORS(handler http.Handler, origin string) http.Handler {
 		handler.ServeHTTP(w, r)
 	})
 }
+
+// waitForChainNotifierReady blocks until lnd's chain notifier accepts a block
+// epoch subscription or the provided context is canceled.
+func waitForChainNotifierReady(ctx context.Context,
+	notifier lndclient.ChainNotifierClient) error {
+
+	const (
+		initialBackoff = 200 * time.Millisecond
+		maxBackoff     = 5 * time.Second
+		streamGrace    = 2 * time.Second
+	)
+
+	backoff := initialBackoff
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+
+		default:
+		}
+
+		attemptCtx, cancel := context.WithCancel(ctx)
+		blockChan, errChan, err := notifier.RegisterBlockEpochNtfn(attemptCtx)
+
+		if err != nil {
+			cancel()
+
+			isStartupErr := status.Code(err) == codes.Unavailable ||
+				strings.Contains(err.Error(),
+					chainNotifierStartupMessage)
+
+			if !isStartupErr {
+				return err
+			}
+
+			warnf("Chain notifier RPC not ready, retrying in %v: %v",
+				backoff, err)
+		} else {
+			// Registration succeeded; wait briefly to ensure the
+			// notifier keeps the stream open. If it pushes the startup
+			// error down errChan, retry as well.
+			timer := time.NewTimer(streamGrace)
+
+			select {
+			case <-blockChan:
+				go drainReadinessNtfn(
+					attemptCtx, cancel, blockChan, errChan,
+				)
+				return nil
+
+			case err = <-errChan:
+				cancel()
+
+				warnf("Chain notifier stream ended early, "+
+					"retrying: %v", err)
+
+			case <-timer.C:
+				go drainReadinessNtfn(
+					attemptCtx, cancel, blockChan, errChan,
+				)
+				return nil
+
+			case <-ctx.Done():
+				cancel()
+				return ctx.Err()
+			}
+		}
+
+		select {
+		case <-time.After(backoff):
+
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+
+		backoff *= 2
+		if backoff > maxBackoff {
+			backoff = maxBackoff
+		}
+	}
+}
+
+// drainReadinessNtfn discards notifications until the daemon shuts down,
+// allowing the readiness subscription to stay open without affecting lnd logs.
+func drainReadinessNtfn(ctx context.Context, cancel func(),
+	blockChan <-chan int32, errChan <-chan error) {
+
+	defer cancel()
+
+	for {
+		select {
+		case <-blockChan:
+
+		case <-errChan:
+			return
+
+		case <-ctx.Done():
+			return
+		}
+	}
+}