lndclient: block until chain notifier is ready

LND commit c6f458e478f9 (v0.20.0-rc3) moved ChainNotifier startup later in
the lifecycle, so RegisterBlockEpochNtfn callers now see "chain notifier RPC
is still in the process of starting" coming from Recv().

The new BlockUntilChainNotifier config option repeatedly calls
RegisterBlockEpochNtfn during startup and only proceeds once the stream yields
its first block height, retrying solely when we detect the
ErrChainNotifierServerNotActive condition introduced by the LND commit above.

The wait is guarded by a "chainrpc" build-tag check, so deployments without
the ChainNotifier service fail fast when the option is requested.

Author: starius

lnd_services.go

@@ -143,6 +143,11 @@ type LndServicesConfig struct {
 	// block download is still in progress.
 	BlockUntilChainSynced bool
 
+	// BlockUntilChainNotifier indicates that the client should wait until
+	// the ChainNotifier RPC is accepting subscriptions. This requires lnd
+	// to be built with the "chainrpc" tag.
+	BlockUntilChainNotifier bool
+
 	// BlockUntilUnlocked denotes that the NewLndServices function should
 	// block until lnd is unlocked.
 	BlockUntilUnlocked bool
@@ -453,6 +458,33 @@ func NewLndServices(cfg *LndServicesConfig) (*GrpcLndServices, error) {
 		log.Infof("lnd is now fully synced to its chain backend")
 	}
 
+	// If requested, wait until the chain notifier RPC is ready before we
+	// return. This ensures sub-servers relying on the notifier don't fail
+	// during startup.
+	if cfg.BlockUntilChainNotifier {
+		if !hasBuildTag(services.Version, "chainrpc") {
+			cleanup()
+
+			return nil, fmt.Errorf("chain notifier build tag is " +
+				"required when waiting for chain notifier " +
+				"readiness")
+		}
+
+		log.Infof("Waiting for chain notifier RPC to be ready")
+
+		err := services.waitForChainNotifier(
+			cfg.CallerCtx, timeout, cfg.ChainSyncPollInterval,
+		)
+		if err != nil {
+			cleanup()
+
+			return nil, fmt.Errorf("error waiting for chain "+
+				"notifier readiness: %w", err)
+		}
+
+		log.Infof("Chain notifier RPC is ready")
+	}
+
 	return services, nil
 }
 
@@ -533,6 +565,60 @@ func (s *GrpcLndServices) waitForChainSync(ctx context.Context,
 	return <-update
 }
 
+// waitForChainNotifier blocks until the ChainNotifier RPC accepts block epoch
+// subscriptions and delivers at least one block height.
+func (s *GrpcLndServices) waitForChainNotifier(ctx context.Context,
+	timeout, pollInterval time.Duration) error {
+
+	mainCtx := ctx
+	if mainCtx == nil {
+		mainCtx = context.Background()
+	}
+
+	register := s.ChainNotifier.RegisterBlockEpochNtfn
+
+	for {
+		// Make new RegisterBlockEpochNtfn call.
+		subCtx, cancel := context.WithTimeout(mainCtx, timeout)
+		blockChan, errChan, err := register(subCtx)
+		if err != nil {
+			cancel()
+
+			return fmt.Errorf("register block epoch ntfn: %w", err)
+		}
+
+		// Wait for block height notification, which indicates success.
+		select {
+		case <-mainCtx.Done():
+			cancel()
+
+			return mainCtx.Err()
+
+		case err := <-errChan:
+			cancel()
+
+			// If chainNotifier is not ready yet, retry.
+			if isChainNotifierStartingErr(err) {
+				select {
+				case <-time.After(pollInterval):
+					continue
+
+				case <-mainCtx.Done():
+					return mainCtx.Err()
+				}
+			}
+
+			return err
+
+		// We got a block height. Success!
+		case <-blockChan:
+			cancel()
+
+			return nil
+		}
+	}
+}
+
 // getLndInfo queries lnd for information about the node it is connected to.
 // If the waitForUnlocked boolean is set, it will examine any errors returned
 // and back off if the failure is due to lnd currently being locked. Otherwise,
@@ -676,6 +762,37 @@ func IsUnlockError(err error) bool {
 	return false
 }
 
+// chainNotifierStartupMessage matches the error string returned by lnd
+// v0.20.0-rc3+ when a ChainNotifier RPC is invoked before the sub-server
+// finishes initialization.
+const chainNotifierStartupMessage = "chain notifier RPC is still in the " +
+	"process of starting"
+
+// isChainNotifierStartingErr reports whether err is due to the lnd
+// ChainNotifier sub-server still starting up. Starting with lnd v0.20.0-rc3
+// the notifier is initialised later in the daemon lifecycle, and the RPC layer
+// surfaces this as an Unknown gRPC status that contains the message defined in
+// chainNotifierStartupMessage. There is a PR in LND to return code Unavailable
+// instead of Unknown: https://github.com/lightningnetwork/lnd/pull/10352
+func isChainNotifierStartingErr(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	// gRPC code Unavailable means "the server can't handle this request
+	// now, retry later". LND's chain notifier returns this error when
+	// the server is starting.
+	// See https://github.com/lightningnetwork/lnd/pull/10352
+	st, ok := status.FromError(err)
+	if ok && st.Code() == codes.Unavailable {
+		return true
+	}
+
+	// TODO(ln-v0.20.0) remove the string fallback once lndclient depends on
+	// a version of lnd that returns codes.Unavailable for this condition.
+	return strings.Contains(err.Error(), chainNotifierStartupMessage)
+}
+
 // checkLndCompatibility makes sure the connected lnd instance is running on the
 // correct network, has the version RPC implemented, is the correct minimal
 // version and supports all required build tags/subservers.
@@ -809,6 +926,22 @@ func assertBuildTagsEnabled(actual *verrpc.Version,
 	return nil
 }
 
+// hasBuildTag reports whether the given version advertises the specified
+// build tag.
+func hasBuildTag(version *verrpc.Version, tag string) bool {
+	if version == nil {
+		return false
+	}
+
+	for _, t := range version.BuildTags {
+		if t == tag {
+			return true
+		}
+	}
+
+	return false
+}
+
 var (
 	defaultRPCPort         = "10009"
 	defaultLndDir          = btcutil.AppDataDir("lnd", false)

lnd_services_test.go

@@ -361,3 +361,27 @@ func TestCustomMacaroonHex(t *testing.T) {
 	_, err = NewLndServices(testCfg)
 	require.Error(t, err, "must set only one")
 }
+
+// TestIsChainNotifierStartingErr ensures we correctly detect the startup lag
+// error returned by lnd v0.20.0-rc3+.
+func TestIsChainNotifierStartingErr(t *testing.T) {
+	t.Parallel()
+
+	require.True(t, isChainNotifierStartingErr(
+		status.Error(codes.Unavailable, chainNotifierStartupMessage),
+	))
+
+	require.True(t, isChainNotifierStartingErr(
+		status.Error(codes.Unknown, chainNotifierStartupMessage),
+	))
+
+	require.True(t, isChainNotifierStartingErr(
+		status.Error(codes.Unavailable, "some other error"),
+	))
+
+	require.False(t, isChainNotifierStartingErr(nil))
+
+	require.False(t, isChainNotifierStartingErr(
+		status.Error(codes.Unknown, "some other error"),
+	))
+}