authmailbox: make Subscribe non-blocking, add fallback support

Refactor MultiSubscription.Subscribe to be non-blocking. It now also
initiates subscriptions to any fallback servers in addition to the
primary server specified in the tap address.

Author: ffranr

authmailbox/multi_subscription.go

@@ -5,12 +5,20 @@ import (
 	"fmt"
 	"net/url"
 	"sync"
+	"time"
 
 	"github.com/lightninglabs/taproot-assets/asset"
+	"github.com/lightninglabs/taproot-assets/fn"
 	lfn "github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/keychain"
 )
 
+const (
+	// DefaultTimeout is the default timeout we use for RPC and database
+	// operations.
+	DefaultTimeout = 30 * time.Second
+)
+
 // clientSubscriptions holds the subscriptions and cancel functions for a
 // specific mailbox client.
 type clientSubscriptions struct {
@@ -131,6 +139,10 @@ type MultiSubscription struct {
 	// message channel that can be used to receive messages from any
 	// subscribed account, regardless of which mailbox server it belongs to.
 	msgQueue *lfn.ConcurrentQueue[*ReceivedMessages]
+
+	// ContextGuard provides a wait group and main quit channel that can be
+	// used to create guarded contexts.
+	*fn.ContextGuard
 }
 
 // MultiSubscriptionConfig holds the configuration parameters for creating a
@@ -154,15 +166,64 @@ func NewMultiSubscription(cfg MultiSubscriptionConfig) *MultiSubscription {
 		cfg:      cfg,
 		registry: newClientRegistry(),
 		msgQueue: queue,
+		ContextGuard: &fn.ContextGuard{
+			DefaultTimeout: DefaultTimeout,
+			Quit:           make(chan struct{}),
+		},
 	}
 }
 
 // Subscribe adds a new subscription for the specified client URL and receiver
-// key. It starts a new mailbox client if one does not already exist for the
-// given URL. The subscription will receive messages that match the provided
-// filter and will send them to the shared message queue.
-func (m *MultiSubscription) Subscribe(ctx context.Context, serverURL url.URL,
-	receiverKey keychain.KeyDescriptor, filter MessageFilter) error {
+// key in a non-blocking manner. It starts a goroutine that will establish the
+// subscription asynchronously.
+func (m *MultiSubscription) Subscribe(ctx context.Context,
+	primaryServerURL url.URL, receiverKey keychain.KeyDescriptor,
+	filter MessageFilter) error {
+
+	// Formulate the list of server URLs to attempt subscription to. This
+	// includes the primary server URL followed by any fallback URLs.
+	mboxServerURLs := append(
+		[]url.URL{primaryServerURL}, m.cfg.FallbackMboxURLs...,
+	)
+
+	// Attempt to subscribe to all mailbox servers in parallel and in a
+	// non-blocking manner.
+	m.Goroutine(func() error {
+		errMap, err := fn.ParSliceErrCollect(
+			ctx, mboxServerURLs,
+			func(ctx context.Context, serverURL url.URL) error {
+				return m.subscribeBlocking(
+					ctx, serverURL, receiverKey, filter,
+				)
+			},
+		)
+		if err != nil {
+			return fmt.Errorf("parallel subscription attempt "+
+				"failed: %w", err)
+		}
+
+		for idx, subErr := range errMap {
+			serverURL := mboxServerURLs[idx]
+
+			log.ErrorS(ctx, "Subscription to server failed", subErr,
+				"server_addr", serverURL.String())
+		}
+
+		return nil
+	}, func(err error) {
+		log.ErrorS(ctx, "Subscription goroutine exited with error", err)
+	})
+
+	return nil
+}
+
+// subscribeBlocking is the blocking implementation of Subscribe. It starts a
+// new mailbox client if one does not already exist for the given URL. The
+// subscription will receive messages that match the provided filter and will
+// send them to the shared message queue.
+func (m *MultiSubscription) subscribeBlocking(ctx context.Context,
+	serverURL url.URL, receiverKey keychain.KeyDescriptor,
+	filter MessageFilter) error {
 
 	// Get or create a client for the given server URL. This call is
 	// thread-safe and will handle locking internally.