lnrpc/wtclientrpc: correctly merge tower client info

Prior to this commit, the wtclient would request tower info from both
the legacy client and the anchors client and would then merge the
results returned. This is incorrect since the two clients will have
different session info and different "active" statuses for the same
tower. This commit thus ensures that we dont lose this info.
This commit is contained in:
Elle Mouton committed 2023-03-16 14:59:08 +02:00
1 parent c1be420696
commit b635629e3c
4 files changed
+421 -167

No files matched your search

+92 -27
View File
@@ -1,10 +1,13 @@
package wtclientrpc
import (
"bytes"
"context"
"encoding/binary"
"errors"
"fmt"
"net"
"sort"
"strconv"
"github.com/btcsuite/btcd/btcec/v2"
@@ -274,30 +277,46 @@ func (c *WatchtowerClient) ListTowers(ctx context.Context,
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...)
if err != nil {
return nil, err
}
// Filter duplicates.
towers := make(map[wtdb.TowerID]*wtclient.RegisteredTower)
for _, tower := range anchorTowers {
towers[tower.Tower.ID] = tower
}
// 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 {
towers[tower.Tower.ID] = tower
}
rpcTowers := make([]*Tower, 0, len(towers))
for _, tower := range towers {
rpcTower := marshallTower(
tower, req.IncludeSessions, ackCounts,
committedUpdateCounts,
tower, PolicyType_LEGACY, req.IncludeSessions,
ackCounts, committedUpdateCounts,
)
rpcTowers = append(rpcTowers, rpcTower)
t, ok := rpcTowers[tower.ID]
if !ok {
rpcTowers[tower.ID] = rpcTower
continue
}
t.SessionInfo = append(t.SessionInfo, rpcTower.SessionInfo...)
}
return &ListTowersResponse{Towers: rpcTowers}, nil
towers := make([]*Tower, 0, len(rpcTowers))
for _, tower := range rpcTowers {
towers = append(towers, tower)
}
return &ListTowersResponse{Towers: towers}, nil
}
// GetTowerInfo retrieves information for a registered watchtower.
@@ -317,18 +336,37 @@ func (c *WatchtowerClient) GetTowerInfo(ctx context.Context,
req.IncludeSessions,
)
var tower *wtclient.RegisteredTower
tower, err = c.cfg.Client.LookupTower(pubKey, opts...)
if err == wtdb.ErrTowerNotFound {
tower, err = c.cfg.AnchorClient.LookupTower(pubKey, opts...)
// Get the tower and its sessions from anchors client.
tower, err := c.cfg.AnchorClient.LookupTower(pubKey, opts...)
if err != nil {
return nil, err
}
rpcTower := marshallTower(
tower, PolicyType_ANCHOR, req.IncludeSessions, ackCounts,
committedUpdateCounts,
)
// Get the tower and its sessions from legacy client.
tower, err = c.cfg.Client.LookupTower(pubKey, opts...)
if err != nil {
return nil, err
}
return marshallTower(
tower, req.IncludeSessions, ackCounts, committedUpdateCounts,
), nil
rpcLegacyTower := marshallTower(
tower, PolicyType_LEGACY, req.IncludeSessions, ackCounts,
committedUpdateCounts,
)
if !bytes.Equal(rpcTower.Pubkey, rpcLegacyTower.Pubkey) {
return nil, fmt.Errorf("legacy and anchor clients returned " +
"inconsistent results for the given tower")
}
rpcTower.SessionInfo = append(
rpcTower.SessionInfo, rpcLegacyTower.SessionInfo...,
)
return rpcTower, nil
}
// constructFunctionalOptions is a helper function that constructs a list of
@@ -437,8 +475,8 @@ func (c *WatchtowerClient) Policy(ctx context.Context,
// marshallTower converts a client registered watchtower into its corresponding
// RPC type.
func marshallTower(tower *wtclient.RegisteredTower, includeSessions bool,
ackCounts map[wtdb.SessionID]uint16,
func marshallTower(tower *wtclient.RegisteredTower, policyType PolicyType,
includeSessions bool, ackCounts map[wtdb.SessionID]uint16,
pendingCounts map[wtdb.SessionID]uint16) *Tower {
rpcAddrs := make([]string, 0, len(tower.Addresses))
@@ -448,8 +486,24 @@ func marshallTower(tower *wtclient.RegisteredTower, includeSessions bool,
var rpcSessions []*TowerSession
if includeSessions {
rpcSessions = make([]*TowerSession, 0, len(tower.Sessions))
// To ensure that the output order is deterministic for a given
// set of sessions, we put the sessions into a slice and order
// them based on session ID.
sessions := make([]*wtdb.ClientSession, 0, len(tower.Sessions))
for _, session := range tower.Sessions {
sessions = append(sessions, session)
}
sort.Slice(sessions, func(i, j int) bool {
id1 := sessions[i].ID
id2 := sessions[j].ID
return binary.BigEndian.Uint64(id1[:]) <
binary.BigEndian.Uint64(id2[:])
})
rpcSessions = make([]*TowerSession, 0, len(tower.Sessions))
for _, session := range sessions {
satPerVByte := session.Policy.SweepFeeRate.FeePerKVByte() / 1000
rpcSessions = append(rpcSessions, &TowerSession{
NumBackups: uint32(ackCounts[session.ID]),
@@ -463,11 +517,22 @@ func marshallTower(tower *wtclient.RegisteredTower, includeSessions bool,
}
}
return &Tower{
Pubkey: tower.IdentityKey.SerializeCompressed(),
Addresses: rpcAddrs,
rpcTower := &Tower{
Pubkey: tower.IdentityKey.SerializeCompressed(),
Addresses: rpcAddrs,
SessionInfo: []*TowerSessionInfo{{
PolicyType: policyType,
ActiveSessionCandidate: tower.ActiveSessionCandidate,
NumSessions: uint32(len(tower.Sessions)),
Sessions: rpcSessions,
}},
// The below fields are populated for backwards compatibility
// but will be removed in a future commit when the proto fields
// are removed.
ActiveSessionCandidate: tower.ActiveSessionCandidate,
NumSessions: uint32(len(tower.Sessions)),
Sessions: rpcSessions,
}
return rpcTower
}