wtclient+lnrpc: move RegisteredTowers to Manager

Move the `RegisteredTowers` method from the `Client` to the `Manager`
interface.

lnrpc/wtclientrpc/wtclient.go

@@ -263,23 +263,7 @@ func (c *WatchtowerClient) ListTowers(ctx context.Context,
 		req.IncludeSessions, req.ExcludeExhaustedSessions,
 	)
 
-	anchorTowers, err := c.cfg.AnchorClient.RegisteredTowers(opts...)
-	if err != nil {
-		return nil, err
-	}
-
-	// Collect all the anchor client towers.
-	rpcTowers := make(map[wtdb.TowerID]*Tower)
-	for _, tower := range anchorTowers {
-		rpcTower := marshallTower(
-			tower, PolicyType_ANCHOR, req.IncludeSessions,
-			ackCounts, committedUpdateCounts,
-		)
-
-		rpcTowers[tower.ID] = rpcTower
-	}
-
-	legacyTowers, err := c.cfg.Client.RegisteredTowers(opts...)
+	towersPerBlobType, err := c.cfg.ClientMgr.RegisteredTowers(opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -287,20 +271,32 @@ func (c *WatchtowerClient) ListTowers(ctx context.Context,
 	// Collect all the legacy client towers. If it has any of the same
 	// towers that the anchors client has, then just add the session info
 	// for the legacy client to the existing tower.
-	for _, tower := range legacyTowers {
-		rpcTower := marshallTower(
-			tower, PolicyType_LEGACY, req.IncludeSessions,
-			ackCounts, committedUpdateCounts,
-		)
-
-		t, ok := rpcTowers[tower.ID]
-		if !ok {
-			rpcTowers[tower.ID] = rpcTower
-			continue
+	rpcTowers := make(map[wtdb.TowerID]*Tower)
+	for blobType, towers := range towersPerBlobType {
+		policyType := PolicyType_LEGACY
+		if blobType.IsAnchorChannel() {
+			policyType = PolicyType_ANCHOR
 		}
 
-		t.SessionInfo = append(t.SessionInfo, rpcTower.SessionInfo...)
-		t.Sessions = append(t.Sessions, rpcTower.Sessions...)
+		for _, tower := range towers {
+			rpcTower := marshallTower(
+				tower, policyType, req.IncludeSessions,
+				ackCounts, committedUpdateCounts,
+			)
+
+			t, ok := rpcTowers[tower.ID]
+			if !ok {
+				rpcTowers[tower.ID] = rpcTower
+				continue
+			}
+
+			t.SessionInfo = append(
+				t.SessionInfo, rpcTower.SessionInfo...,
+			)
+			t.Sessions = append(
+				t.Sessions, rpcTower.Sessions...,
+			)
+		}
 	}
 
 	towers := make([]*Tower, 0, len(rpcTowers))

watchtower/wtclient/client.go

@@ -95,11 +95,6 @@ type RegisteredTower struct {
 // Client is the primary interface used by the daemon to control a client's
 // lifecycle and backup revoked states.
 type Client interface {
-	// RegisteredTowers retrieves the list of watchtowers registered with
-	// the client.
-	RegisteredTowers(...wtdb.ClientSessionListOption) ([]*RegisteredTower,
-		error)
-
 	// LookupTower retrieves a registered watchtower through its public key.
 	LookupTower(*btcec.PublicKey,
 		...wtdb.ClientSessionListOption) (*RegisteredTower, error)
@@ -1564,17 +1559,13 @@ func (c *TowerClient) handleStaleTower(msg *staleTowerMsg) error {
 	return nil
 }
 
-// RegisteredTowers retrieves the list of watchtowers registered with the
+// registeredTowers retrieves the list of watchtowers registered with the
 // client.
-func (c *TowerClient) RegisteredTowers(opts ...wtdb.ClientSessionListOption) (
-	[]*RegisteredTower, error) {
-
-	// Retrieve all of our towers along with all of our sessions.
-	towers, err := c.cfg.DB.ListTowers()
-	if err != nil {
-		return nil, err
-	}
+func (c *TowerClient) registeredTowers(towers []*wtdb.Tower,
+	opts ...wtdb.ClientSessionListOption) ([]*RegisteredTower, error) {
 
+	// Generate a filter that will fetch all the client's sessions
+	// regardless of if they are active or not.
 	opts = append(opts, wtdb.WithPreEvalFilterFn(c.genSessionFilter(false)))
 
 	clientSessions, err := c.cfg.DB.ListClientSessions(nil, opts...)
@@ -1582,8 +1573,8 @@ func (c *TowerClient) RegisteredTowers(opts ...wtdb.ClientSessionListOption) (
 		return nil, err
 	}
 
-	// Construct a lookup map that coalesces all of the sessions for a
-	// specific watchtower.
+	// Construct a lookup map that coalesces all the sessions for a specific
+	// watchtower.
 	towerSessions := make(
 		map[wtdb.TowerID]map[wtdb.SessionID]*wtdb.ClientSession,
 	)

watchtower/wtclient/manager.go

@@ -15,6 +15,7 @@ import (
 	"github.com/lightningnetwork/lnd/subscribe"
 	"github.com/lightningnetwork/lnd/tor"
 	"github.com/lightningnetwork/lnd/watchtower/blob"
+	"github.com/lightningnetwork/lnd/watchtower/wtdb"
 	"github.com/lightningnetwork/lnd/watchtower/wtpolicy"
 )
 
@@ -36,6 +37,12 @@ type TowerClientManager interface {
 
 	// Stats returns the in-memory statistics of the client since startup.
 	Stats() ClientStats
+
+	// RegisteredTowers retrieves the list of watchtowers registered with
+	// the client. It returns a set of registered towers per client policy
+	// type.
+	RegisteredTowers(opts ...wtdb.ClientSessionListOption) (
+		map[blob.Type][]*RegisteredTower, error)
 }
 
 // Config provides the TowerClient with access to the resources it requires to
@@ -321,3 +328,29 @@ func (m *Manager) Stats() ClientStats {
 
 	return resp
 }
+
+// RegisteredTowers retrieves the list of watchtowers being used by the various
+// clients.
+func (m *Manager) RegisteredTowers(opts ...wtdb.ClientSessionListOption) (
+	map[blob.Type][]*RegisteredTower, error) {
+
+	towers, err := m.cfg.DB.ListTowers()
+	if err != nil {
+		return nil, err
+	}
+
+	m.clientsMu.Lock()
+	defer m.clientsMu.Unlock()
+
+	resp := make(map[blob.Type][]*RegisteredTower)
+	for _, client := range m.clients {
+		towers, err := client.registeredTowers(towers, opts...)
+		if err != nil {
+			return nil, err
+		}
+
+		resp[client.Policy().BlobType] = towers
+	}
+
+	return resp, nil
+}