Merge pull request #3184 from wpaulino/wtclient-subserver

multi: add watchtower client RPC subserver

watchtower/wtclient/candidate_iterator.go

@@ -2,6 +2,7 @@ package wtclient
 
 import (
 	"container/list"
+	"net"
 	"sync"
 
 	"github.com/lightningnetwork/lnd/watchtower/wtdb"
@@ -10,6 +11,20 @@ import (
 // TowerCandidateIterator provides an abstraction for iterating through possible
 // watchtower addresses when attempting to create a new session.
 type TowerCandidateIterator interface {
+	// AddCandidate adds a new candidate tower to the iterator. If the
+	// candidate already exists, then any new addresses are added to it.
+	AddCandidate(*wtdb.Tower)
+
+	// RemoveCandidate removes an existing candidate tower from the
+	// iterator. An optional address can be provided to indicate a stale
+	// tower address to remove it. If it isn't provided, then the tower is
+	// completely removed from the iterator.
+	RemoveCandidate(wtdb.TowerID, net.Addr)
+
+	// IsActive determines whether a given tower is exists within the
+	// iterator.
+	IsActive(wtdb.TowerID) bool
+
 	// Reset clears any internal iterator state, making previously taken
 	// candidates available as long as they remain in the set.
 	Reset() error
@@ -18,17 +33,14 @@ type TowerCandidateIterator interface {
 	// to return results in any particular order.  If no more candidates are
 	// available, ErrTowerCandidatesExhausted is returned.
 	Next() (*wtdb.Tower, error)
-
-	// TowerIDs returns the set of tower IDs contained in the iterator,
-	// which can be used to filter candidate sessions for the active tower.
-	TowerIDs() map[wtdb.TowerID]struct{}
 }
 
 // towerListIterator is a linked-list backed TowerCandidateIterator.
 type towerListIterator struct {
 	mu            sync.Mutex
-	candidates    *list.List
+	queue         *list.List
 	nextCandidate *list.Element
+	candidates    map[wtdb.TowerID]*wtdb.Tower
 }
 
 // Compile-time constraint to ensure *towerListIterator implements the
@@ -39,11 +51,13 @@ var _ TowerCandidateIterator = (*towerListIterator)(nil)
 // of lnwire.NetAddresses.
 func newTowerListIterator(candidates ...*wtdb.Tower) *towerListIterator {
 	iter := &towerListIterator{
-		candidates: list.New(),
+		queue:      list.New(),
+		candidates: make(map[wtdb.TowerID]*wtdb.Tower),
 	}
 
 	for _, candidate := range candidates {
-		iter.candidates.PushBack(candidate)
+		iter.queue.PushBack(candidate.ID)
+		iter.candidates[candidate.ID] = candidate
 	}
 	iter.Reset()
 
@@ -57,22 +71,11 @@ func (t *towerListIterator) Reset() error {
 	defer t.mu.Unlock()
 
 	// Reset the next candidate to the front of the linked-list.
-	t.nextCandidate = t.candidates.Front()
+	t.nextCandidate = t.queue.Front()
 
 	return nil
 }
 
-// TowerIDs returns the set of tower IDs contained in the iterator, which can be
-// used to filter candidate sessions for the active tower.
-func (t *towerListIterator) TowerIDs() map[wtdb.TowerID]struct{} {
-	ids := make(map[wtdb.TowerID]struct{})
-	for e := t.candidates.Front(); e != nil; e = e.Next() {
-		tower := e.Value.(*wtdb.Tower)
-		ids[tower.ID] = struct{}{}
-	}
-	return ids
-}
-
 // Next returns the next candidate tower. This iterator will always return
 // candidates in the order given when the iterator was instantiated.  If no more
 // candidates are available, ErrTowerCandidatesExhausted is returned.
@@ -80,18 +83,76 @@ func (t *towerListIterator) Next() (*wtdb.Tower, error) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
-	// If the next candidate is nil, we've exhausted the list.
-	if t.nextCandidate == nil {
-		return nil, ErrTowerCandidatesExhausted
+	for t.nextCandidate != nil {
+		// Propose the tower at the front of the list.
+		towerID := t.nextCandidate.Value.(wtdb.TowerID)
+
+		// Check whether this tower is still considered a candidate. If
+		// it's not, we'll proceed to the next.
+		tower, ok := t.candidates[towerID]
+		if !ok {
+			nextCandidate := t.nextCandidate.Next()
+			t.queue.Remove(t.nextCandidate)
+			t.nextCandidate = nextCandidate
+			continue
+		}
+
+		// Set the next candidate to the subsequent element.
+		t.nextCandidate = t.nextCandidate.Next()
+		return tower, nil
 	}
 
-	// Propose the tower at the front of the list.
-	tower := t.nextCandidate.Value.(*wtdb.Tower)
+	return nil, ErrTowerCandidatesExhausted
+}
 
-	// Set the next candidate to the subsequent element.
-	t.nextCandidate = t.nextCandidate.Next()
+// AddCandidate adds a new candidate tower to the iterator. If the candidate
+// already exists, then any new addresses are added to it.
+func (t *towerListIterator) AddCandidate(candidate *wtdb.Tower) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
 
-	return tower, nil
+	if tower, ok := t.candidates[candidate.ID]; !ok {
+		t.queue.PushBack(candidate.ID)
+		t.candidates[candidate.ID] = candidate
+
+		// If we've reached the end of our queue, then this candidate
+		// will become the next.
+		if t.nextCandidate == nil {
+			t.nextCandidate = t.queue.Back()
+		}
+	} else {
+		for _, addr := range candidate.Addresses {
+			tower.AddAddress(addr)
+		}
+	}
+}
+
+// RemoveCandidate removes an existing candidate tower from the iterator. An
+// optional address can be provided to indicate a stale tower address to remove
+// it. If it isn't provided, then the tower is completely removed from the
+// iterator.
+func (t *towerListIterator) RemoveCandidate(candidate wtdb.TowerID, addr net.Addr) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	tower, ok := t.candidates[candidate]
+	if !ok {
+		return
+	}
+	if addr != nil {
+		tower.RemoveAddress(addr)
+	} else {
+		delete(t.candidates, candidate)
+	}
+}
+
+// IsActive determines whether a given tower is exists within the iterator.
+func (t *towerListIterator) IsActive(tower wtdb.TowerID) bool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	_, ok := t.candidates[tower]
+	return ok
 }
 
 // TODO(conner): implement graph-backed candidate iterator for public towers.

watchtower/wtclient/candidate_iterator_test.go

@@ -0,0 +1,157 @@
+package wtclient
+
+import (
+	"encoding/binary"
+	"math/rand"
+	"net"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/btcsuite/btcd/btcec"
+	"github.com/davecgh/go-spew/spew"
+	"github.com/lightningnetwork/lnd/watchtower/wtdb"
+)
+
+func init() {
+	rand.Seed(time.Now().Unix())
+}
+
+func randAddr(t *testing.T) net.Addr {
+	var ip [4]byte
+	if _, err := rand.Read(ip[:]); err != nil {
+		t.Fatal(err)
+	}
+	var port [2]byte
+	if _, err := rand.Read(port[:]); err != nil {
+		t.Fatal(err)
+
+	}
+	return &net.TCPAddr{
+		IP:   net.IP(ip[:]),
+		Port: int(binary.BigEndian.Uint16(port[:])),
+	}
+}
+
+func randTower(t *testing.T) *wtdb.Tower {
+	priv, err := btcec.NewPrivateKey(btcec.S256())
+	if err != nil {
+		t.Fatalf("unable to create private key: %v", err)
+	}
+	pubKey := priv.PubKey()
+	pubKey.Curve = nil
+	return &wtdb.Tower{
+		ID:          wtdb.TowerID(rand.Uint64()),
+		IdentityKey: pubKey,
+		Addresses:   []net.Addr{randAddr(t)},
+	}
+}
+
+func copyTower(tower *wtdb.Tower) *wtdb.Tower {
+	t := &wtdb.Tower{
+		ID:          tower.ID,
+		IdentityKey: tower.IdentityKey,
+		Addresses:   make([]net.Addr, len(tower.Addresses)),
+	}
+	copy(t.Addresses, tower.Addresses)
+	return t
+}
+
+func assertActiveCandidate(t *testing.T, i TowerCandidateIterator,
+	c *wtdb.Tower, active bool) {
+
+	isCandidate := i.IsActive(c.ID)
+	if isCandidate && !active {
+		t.Fatalf("expected tower %v to no longer be an active candidate",
+			c.ID)
+	}
+	if !isCandidate && active {
+		t.Fatalf("expected tower %v to be an active candidate", c.ID)
+	}
+}
+
+func assertNextCandidate(t *testing.T, i TowerCandidateIterator, c *wtdb.Tower) {
+	t.Helper()
+
+	tower, err := i.Next()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(tower, c) {
+		t.Fatalf("expected tower: %v\ngot: %v", spew.Sdump(c),
+			spew.Sdump(tower))
+	}
+}
+
+// TestTowerCandidateIterator asserts the internal state of a
+// TowerCandidateIterator after a series of updates to its candidates.
+func TestTowerCandidateIterator(t *testing.T) {
+	t.Parallel()
+
+	// We'll start our test by creating an iterator of four candidate
+	// towers. We'll use copies of these towers within the iterator to
+	// ensure the iterator properly updates the state of its candidates.
+	const numTowers = 4
+	towers := make([]*wtdb.Tower, 0, numTowers)
+	for i := 0; i < numTowers; i++ {
+		towers = append(towers, randTower(t))
+	}
+	towerCopies := make([]*wtdb.Tower, 0, numTowers)
+	for _, tower := range towers {
+		towerCopies = append(towerCopies, copyTower(tower))
+	}
+	towerIterator := newTowerListIterator(towerCopies...)
+
+	// We should expect to see all of our candidates in the order that they
+	// were added.
+	for _, expTower := range towers {
+		tower, err := towerIterator.Next()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !reflect.DeepEqual(tower, expTower) {
+			t.Fatalf("expected tower: %v\ngot: %v",
+				spew.Sdump(expTower), spew.Sdump(tower))
+		}
+	}
+
+	if _, err := towerIterator.Next(); err != ErrTowerCandidatesExhausted {
+		t.Fatalf("expected ErrTowerCandidatesExhausted, got %v", err)
+	}
+	towerIterator.Reset()
+
+	// We'll then attempt to test the RemoveCandidate behavior of the
+	// iterator. We'll remove the address of the first tower, which should
+	// result in it not having any addresses left, but still being an active
+	// candidate.
+	firstTower := towers[0]
+	firstTowerAddr := firstTower.Addresses[0]
+	firstTower.RemoveAddress(firstTowerAddr)
+	towerIterator.RemoveCandidate(firstTower.ID, firstTowerAddr)
+	assertActiveCandidate(t, towerIterator, firstTower, true)
+	assertNextCandidate(t, towerIterator, firstTower)
+
+	// We'll then remove the second tower completely from the iterator by
+	// not providing the optional address. Since it's been removed, we
+	// should expect to see the third tower next.
+	secondTower, thirdTower := towers[1], towers[2]
+	towerIterator.RemoveCandidate(secondTower.ID, nil)
+	assertActiveCandidate(t, towerIterator, secondTower, false)
+	assertNextCandidate(t, towerIterator, thirdTower)
+
+	// We'll then update the fourth candidate with a new address. A
+	// duplicate shouldn't be added since it already exists within the
+	// iterator, but the new address should be.
+	fourthTower := towers[3]
+	assertActiveCandidate(t, towerIterator, fourthTower, true)
+	fourthTower.AddAddress(randAddr(t))
+	towerIterator.AddCandidate(fourthTower)
+	assertNextCandidate(t, towerIterator, fourthTower)
+
+	// Finally, we'll attempt to add a new candidate to the end of the
+	// iterator. Since it didn't already exist and we've reached the end, it
+	// should be available as the next candidate.
+	towerIterator.AddCandidate(secondTower)
+	assertActiveCandidate(t, towerIterator, secondTower, true)
+	assertNextCandidate(t, towerIterator, secondTower)
+}

watchtower/wtclient/client.go

@@ -2,7 +2,9 @@ package wtclient
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
+	"net"
 	"sync"
 	"time"
 
@@ -36,9 +38,48 @@ const (
 	DefaultForceQuitDelay = 10 * time.Second
 )
 
+// RegisteredTower encompasses information about a registered watchtower with
+// the client.
+type RegisteredTower struct {
+	*wtdb.Tower
+
+	// Sessions is the set of sessions corresponding to the watchtower.
+	Sessions map[wtdb.SessionID]*wtdb.ClientSession
+
+	// ActiveSessionCandidate determines whether the watchtower is currently
+	// being considered for new sessions.
+	ActiveSessionCandidate bool
+}
+
 // Client is the primary interface used by the daemon to control a client's
 // lifecycle and backup revoked states.
 type Client interface {
+	// AddTower adds a new watchtower reachable at the given address and
+	// considers it for new sessions. If the watchtower already exists, then
+	// any new addresses included will be considered when dialing it for
+	// session negotiations and backups.
+	AddTower(*lnwire.NetAddress) error
+
+	// RemoveTower removes a watchtower from being considered for future
+	// session negotiations and from being used for any subsequent backups
+	// until it's added again. If an address is provided, then this call
+	// only serves as a way of removing the address from the watchtower
+	// instead.
+	RemoveTower(*btcec.PublicKey, net.Addr) error
+
+	// RegisteredTowers retrieves the list of watchtowers registered with
+	// the client.
+	RegisteredTowers() ([]*RegisteredTower, error)
+
+	// LookupTower retrieves a registered watchtower through its public key.
+	LookupTower(*btcec.PublicKey) (*RegisteredTower, error)
+
+	// Stats returns the in-memory statistics of the client since startup.
+	Stats() ClientStats
+
+	// Policy returns the active client policy configuration.
+	Policy() wtpolicy.Policy
+
 	// RegisterChannel persistently initializes any channel-dependent
 	// parameters within the client. This should be called during link
 	// startup to ensure that the client is able to support the link during
@@ -100,10 +141,6 @@ type Config struct {
 	// new sessions will be requested immediately.
 	Policy wtpolicy.Policy
 
-	// PrivateTower is the net address of a private tower. The client will
-	// try to create all sessions with this tower.
-	PrivateTower *lnwire.NetAddress
-
 	// ChainHash identifies the chain that the client is on and for which
 	// the tower must be watching to monitor for breaches.
 	ChainHash chainhash.Hash
@@ -136,6 +173,39 @@ type Config struct {
 	MaxBackoff time.Duration
 }
 
+// newTowerMsg is an internal message we'll use within the TowerClient to signal
+// that a new tower can be considered.
+type newTowerMsg struct {
+	// addr is the tower's reachable address that we'll use to establish a
+	// connection with.
+	addr *lnwire.NetAddress
+
+	// errChan is the channel through which we'll send a response back to
+	// the caller when handling their request.
+	//
+	// NOTE: This channel must be buffered.
+	errChan chan error
+}
+
+// staleTowerMsg is an internal message we'll use within the TowerClient to
+// signal that a tower should no longer be considered.
+type staleTowerMsg struct {
+	// pubKey is the identifying public key of the watchtower.
+	pubKey *btcec.PublicKey
+
+	// addr is an optional field that when set signals that the address
+	// should be removed from the watchtower's set of addresses, indicating
+	// that it is stale. If it's not set, then the watchtower should be
+	// no longer be considered for new sessions.
+	addr net.Addr
+
+	// errChan is the channel through which we'll send a response back to
+	// the caller when handling their request.
+	//
+	// NOTE: This channel must be buffered.
+	errChan chan error
+}
+
 // TowerClient is a concrete implementation of the Client interface, offering a
 // non-blocking, reliable subsystem for backing up revoked states to a specified
 // private tower.
@@ -149,9 +219,9 @@ type TowerClient struct {
 	pipeline *taskPipeline
 
 	negotiator        SessionNegotiator
+	candidateTowers   TowerCandidateIterator
 	candidateSessions map[wtdb.SessionID]*wtdb.ClientSession
 	activeSessions    sessionQueueSet
-	targetTowerIDs    map[wtdb.TowerID]struct{}
 
 	sessionQueue *sessionQueue
 	prevTask     *backupTask
@@ -161,7 +231,10 @@ type TowerClient struct {
 	chanCommitHeights map[lnwire.ChannelID]uint64
 
 	statTicker *time.Ticker
-	stats      clientStats
+	stats      *ClientStats
+
+	newTowers   chan *newTowerMsg
+	staleTowers chan *staleTowerMsg
 
 	wg        sync.WaitGroup
 	forceQuit chan struct{}
@@ -189,26 +262,74 @@ func New(config *Config) (*TowerClient, error) {
 		cfg.WriteTimeout = DefaultWriteTimeout
 	}
 
-	// Record the tower in our database, also loading any addresses
-	// previously associated with its public key.
-	tower, err := cfg.DB.CreateTower(cfg.PrivateTower)
+	// Next, load all candidate sessions and towers from the database into
+	// the client. We will use any of these session if their policies match
+	// the current policy of the client, otherwise they will be ignored and
+	// new sessions will be requested.
+	sessions, err := cfg.DB.ListClientSessions(nil)
 	if err != nil {
 		return nil, err
 	}
 
-	log.Infof("Using private watchtower %s, offering policy %s",
-		cfg.PrivateTower, cfg.Policy)
+	candidateSessions := make(map[wtdb.SessionID]*wtdb.ClientSession)
+	sessionTowers := make(map[wtdb.TowerID]*wtdb.Tower)
+	for _, s := range sessions {
+		// Candidate sessions must be in an active state.
+		if s.Status != wtdb.CSessionActive {
+			continue
+		}
 
-	candidates := newTowerListIterator(tower)
-	targetTowerIDs := candidates.TowerIDs()
+		// Reload the tower from disk using the tower ID contained in
+		// each candidate session. We will also rederive any session
+		// keys needed to be able to communicate with the towers and
+		// authenticate session requests. This prevents us from having
+		// to store the private keys on disk.
+		tower, ok := sessionTowers[s.TowerID]
+		if !ok {
+			var err error
+			tower, err = cfg.DB.LoadTowerByID(s.TowerID)
+			if err != nil {
+				return nil, err
+			}
+		}
+		s.Tower = tower
+
+		sessionKey, err := DeriveSessionKey(cfg.SecretKeyRing, s.KeyIndex)
+		if err != nil {
+			return nil, err
+		}
+		s.SessionPrivKey = sessionKey
+
+		candidateSessions[s.ID] = s
+		sessionTowers[tower.ID] = tower
+	}
+
+	var candidateTowers []*wtdb.Tower
+	for _, tower := range sessionTowers {
+		log.Infof("Using private watchtower %s, offering policy %s",
+			tower, cfg.Policy)
+		candidateTowers = append(candidateTowers, tower)
+	}
+
+	// Load the sweep pkscripts that have been generated for all previously
+	// registered channels.
+	chanSummaries, err := cfg.DB.FetchChanSummaries()
+	if err != nil {
+		return nil, err
+	}
 
 	c := &TowerClient{
-		cfg:            cfg,
-		pipeline:       newTaskPipeline(),
-		activeSessions: make(sessionQueueSet),
-		targetTowerIDs: targetTowerIDs,
-		statTicker:     time.NewTicker(DefaultStatInterval),
-		forceQuit:      make(chan struct{}),
+		cfg:               cfg,
+		pipeline:          newTaskPipeline(),
+		candidateTowers:   newTowerListIterator(candidateTowers...),
+		candidateSessions: candidateSessions,
+		activeSessions:    make(sessionQueueSet),
+		summaries:         chanSummaries,
+		statTicker:        time.NewTicker(DefaultStatInterval),
+		stats:             new(ClientStats),
+		newTowers:         make(chan *newTowerMsg),
+		staleTowers:       make(chan *staleTowerMsg),
+		forceQuit:         make(chan struct{}),
 	}
 	c.negotiator = newSessionNegotiator(&NegotiatorConfig{
 		DB:            cfg.DB,
@@ -218,53 +339,15 @@ func New(config *Config) (*TowerClient, error) {
 		SendMessage:   c.sendMessage,
 		ReadMessage:   c.readMessage,
 		Dial:          c.dial,
-		Candidates:    candidates,
+		Candidates:    c.candidateTowers,
 		MinBackoff:    cfg.MinBackoff,
 		MaxBackoff:    cfg.MaxBackoff,
 	})
 
-	// Next, load all active sessions from the db into the client. We will
-	// use any of these session if their policies match the current policy
-	// of the client, otherwise they will be ignored and new sessions will
-	// be requested.
-	c.candidateSessions, err = c.cfg.DB.ListClientSessions()
-	if err != nil {
-		return nil, err
-	}
-
-	// Reload any towers from disk using the tower IDs contained in each
-	// candidate session. We will also rederive any session keys needed to
-	// be able to communicate with the towers and authenticate session
-	// requests. This prevents us from having to store the private keys on
-	// disk.
-	for _, s := range c.candidateSessions {
-		tower, err := c.cfg.DB.LoadTower(s.TowerID)
-		if err != nil {
-			return nil, err
-		}
-
-		sessionPriv, err := DeriveSessionKey(
-			c.cfg.SecretKeyRing, s.KeyIndex,
-		)
-		if err != nil {
-			return nil, err
-		}
-
-		s.Tower = tower
-		s.SessionPrivKey = sessionPriv
-	}
-
 	// Reconstruct the highest commit height processed for each channel
 	// under the client's current policy.
 	c.buildHighestCommitHeights()
 
-	// Finally, load the sweep pkscripts that have been generated for all
-	// previously registered channels.
-	c.summaries, err = c.cfg.DB.FetchChanSummaries()
-	if err != nil {
-		return nil, err
-	}
-
 	return c, nil
 }
 
@@ -531,12 +614,6 @@ func (c *TowerClient) nextSessionQueue() *sessionQueue {
 			continue
 		}
 
-		// Skip any sessions that are still active, but are not for the
-		// users currently configured tower.
-		if _, ok := c.targetTowerIDs[sessionInfo.TowerID]; !ok {
-			continue
-		}
-
 		candidateSession = sessionInfo
 		break
 	}
@@ -586,19 +663,38 @@ func (c *TowerClient) backupDispatcher() {
 				c.candidateSessions[session.ID] = session
 				c.stats.sessionAcquired()
 
+				// We'll continue to choose the newly negotiated
+				// session as our active session queue.
+				continue
+
 			case <-c.statTicker.C:
 				log.Infof("Client stats: %s", c.stats)
 
-				// Instead of looping, we'll jump back into the
-				// select case and await the delivery of the
-				// session to prevent us from re-requesting
-				// additional sessions.
-				goto awaitSession
+			// A new tower has been requested to be added. We'll
+			// update our persisted and in-memory state and consider
+			// its corresponding sessions, if any, as new
+			// candidates.
+			case msg := <-c.newTowers:
+				msg.errChan <- c.handleNewTower(msg)
+
+			// A tower has been requested to be removed. We'll
+			// immediately return an error as we want to avoid the
+			// possibility of a new session being negotiated with
+			// this request's tower.
+			case msg := <-c.staleTowers:
+				msg.errChan <- errors.New("removing towers " +
+					"is disallowed while a new session " +
+					"negotiation is in progress")
 
 			case <-c.forceQuit:
 				return
 			}
 
+			// Instead of looping, we'll jump back into the select
+			// case and await the delivery of the session to prevent
+			// us from re-requesting additional sessions.
+			goto awaitSession
+
 		// No active session queue but have additional sessions.
 		case c.sessionQueue == nil && len(c.candidateSessions) > 0:
 			// We've exhausted the prior session, we'll pop another
@@ -633,7 +729,7 @@ func (c *TowerClient) backupDispatcher() {
 			// we can request new sessions before the session is
 			// fully empty, which this case would handle.
 			case session := <-c.negotiator.NewSessions():
-				log.Warnf("Acquired new session with id=%s",
+				log.Warnf("Acquired new session with id=%s "+
 					"while processing tasks", session.ID)
 				c.candidateSessions[session.ID] = session
 				c.stats.sessionAcquired()
@@ -654,6 +750,20 @@ func (c *TowerClient) backupDispatcher() {
 
 				c.stats.taskReceived()
 				c.processTask(task)
+
+			// A new tower has been requested to be added. We'll
+			// update our persisted and in-memory state and consider
+			// its corresponding sessions, if any, as new
+			// candidates.
+			case msg := <-c.newTowers:
+				msg.errChan <- c.handleNewTower(msg)
+
+			// A tower has been removed, so we'll remove certain
+			// information that's persisted and also in our
+			// in-memory state depending on the request, and set any
+			// of its corresponding candidate sessions as inactive.
+			case msg := <-c.staleTowers:
+				msg.errChan <- c.handleStaleTower(msg)
 			}
 		}
 	}
@@ -883,6 +993,207 @@ func (c *TowerClient) initActiveQueue(s *wtdb.ClientSession) *sessionQueue {
 	return sq
 }
 
+// AddTower adds a new watchtower reachable at the given address and considers
+// it for new sessions. If the watchtower already exists, then any new addresses
+// included will be considered when dialing it for session negotiations and
+// backups.
+func (c *TowerClient) AddTower(addr *lnwire.NetAddress) error {
+	errChan := make(chan error, 1)
+
+	select {
+	case c.newTowers <- &newTowerMsg{
+		addr:    addr,
+		errChan: errChan,
+	}:
+	case <-c.pipeline.quit:
+		return ErrClientExiting
+	case <-c.pipeline.forceQuit:
+		return ErrClientExiting
+	}
+
+	select {
+	case err := <-errChan:
+		return err
+	case <-c.pipeline.quit:
+		return ErrClientExiting
+	case <-c.pipeline.forceQuit:
+		return ErrClientExiting
+	}
+}
+
+// handleNewTower handles a request for a new tower to be added. If the tower
+// already exists, then its corresponding sessions, if any, will be set
+// considered as candidates.
+func (c *TowerClient) handleNewTower(msg *newTowerMsg) error {
+	// We'll start by updating our persisted state, followed by our
+	// in-memory state, with the new tower. This might not actually be a new
+	// tower, but it might include a new address at which it can be reached.
+	tower, err := c.cfg.DB.CreateTower(msg.addr)
+	if err != nil {
+		return err
+	}
+	c.candidateTowers.AddCandidate(tower)
+
+	// Include all of its corresponding sessions to our set of candidates.
+	sessions, err := c.cfg.DB.ListClientSessions(&tower.ID)
+	if err != nil {
+		return fmt.Errorf("unable to determine sessions for tower %x: "+
+			"%v", tower.IdentityKey.SerializeCompressed(), err)
+	}
+	for id, session := range sessions {
+		c.candidateSessions[id] = session
+	}
+
+	return nil
+}
+
+// RemoveTower removes a watchtower from being considered for future session
+// negotiations and from being used for any subsequent backups until it's added
+// again. If an address is provided, then this call only serves as a way of
+// removing the address from the watchtower instead.
+func (c *TowerClient) RemoveTower(pubKey *btcec.PublicKey, addr net.Addr) error {
+	errChan := make(chan error, 1)
+
+	select {
+	case c.staleTowers <- &staleTowerMsg{
+		pubKey:  pubKey,
+		addr:    addr,
+		errChan: errChan,
+	}:
+	case <-c.pipeline.quit:
+		return ErrClientExiting
+	case <-c.pipeline.forceQuit:
+		return ErrClientExiting
+	}
+
+	select {
+	case err := <-errChan:
+		return err
+	case <-c.pipeline.quit:
+		return ErrClientExiting
+	case <-c.pipeline.forceQuit:
+		return ErrClientExiting
+	}
+}
+
+// handleNewTower handles a request for an existing tower to be removed. If none
+// of the tower's sessions have pending updates, then they will become inactive
+// and removed as candidates. If the active session queue corresponds to any of
+// these sessions, a new one will be negotiated.
+func (c *TowerClient) handleStaleTower(msg *staleTowerMsg) error {
+	// We'll load the tower before potentially removing it in order to
+	// retrieve its ID within the database.
+	tower, err := c.cfg.DB.LoadTower(msg.pubKey)
+	if err != nil {
+		return err
+	}
+
+	// We'll update our persisted state, followed by our in-memory state,
+	// with the stale tower.
+	if err := c.cfg.DB.RemoveTower(msg.pubKey, msg.addr); err != nil {
+		return err
+	}
+	c.candidateTowers.RemoveCandidate(tower.ID, msg.addr)
+
+	// If an address was provided, then we're only meant to remove the
+	// address from the tower, so there's nothing left for us to do.
+	if msg.addr != nil {
+		return nil
+	}
+
+	// Otherwise, the tower should no longer be used for future session
+	// negotiations and backups.
+	pubKey := msg.pubKey.SerializeCompressed()
+	sessions, err := c.cfg.DB.ListClientSessions(&tower.ID)
+	if err != nil {
+		return fmt.Errorf("unable to retrieve sessions for tower %x: "+
+			"%v", pubKey, err)
+	}
+	for sessionID := range sessions {
+		delete(c.candidateSessions, sessionID)
+	}
+
+	// If our active session queue corresponds to the stale tower, we'll
+	// proceed to negotiate a new one.
+	if c.sessionQueue != nil {
+		activeTower := c.sessionQueue.towerAddr.IdentityKey.SerializeCompressed()
+		if bytes.Equal(pubKey, activeTower) {
+			c.sessionQueue = nil
+		}
+	}
+
+	return nil
+}
+
+// RegisteredTowers retrieves the list of watchtowers registered with the
+// client.
+func (c *TowerClient) RegisteredTowers() ([]*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
+	}
+	clientSessions, err := c.cfg.DB.ListClientSessions(nil)
+	if err != nil {
+		return nil, err
+	}
+
+	// Construct a lookup map that coalesces all of the sessions for a
+	// specific watchtower.
+	towerSessions := make(
+		map[wtdb.TowerID]map[wtdb.SessionID]*wtdb.ClientSession,
+	)
+	for id, s := range clientSessions {
+		sessions, ok := towerSessions[s.TowerID]
+		if !ok {
+			sessions = make(map[wtdb.SessionID]*wtdb.ClientSession)
+			towerSessions[s.TowerID] = sessions
+		}
+		sessions[id] = s
+	}
+
+	registeredTowers := make([]*RegisteredTower, 0, len(towerSessions))
+	for _, tower := range towers {
+		isActive := c.candidateTowers.IsActive(tower.ID)
+		registeredTowers = append(registeredTowers, &RegisteredTower{
+			Tower:                  tower,
+			Sessions:               towerSessions[tower.ID],
+			ActiveSessionCandidate: isActive,
+		})
+	}
+
+	return registeredTowers, nil
+}
+
+// LookupTower retrieves a registered watchtower through its public key.
+func (c *TowerClient) LookupTower(pubKey *btcec.PublicKey) (*RegisteredTower, error) {
+	tower, err := c.cfg.DB.LoadTower(pubKey)
+	if err != nil {
+		return nil, err
+	}
+
+	towerSessions, err := c.cfg.DB.ListClientSessions(&tower.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	return &RegisteredTower{
+		Tower:                  tower,
+		Sessions:               towerSessions,
+		ActiveSessionCandidate: c.candidateTowers.IsActive(tower.ID),
+	}, nil
+}
+
+// Stats returns the in-memory statistics of the client since startup.
+func (c *TowerClient) Stats() ClientStats {
+	return c.stats.Copy()
+}
+
+// Policy returns the active client policy configuration.
+func (c *TowerClient) Policy() wtpolicy.Policy {
+	return c.cfg.Policy
+}
+
 // logMessage writes information about a message received from a remote peer,
 // using directional prepositions to signal whether the message was sent or
 // received.

watchtower/wtclient/client_test.go

@@ -26,7 +26,11 @@ import (
 	"github.com/lightningnetwork/lnd/watchtower/wtserver"
 )
 
-const csvDelay uint32 = 144
+const (
+	csvDelay uint32 = 144
+
+	towerAddrStr = "18.28.243.2:9911"
+)
 
 var (
 	revPrivBytes = []byte{
@@ -387,7 +391,6 @@ type harnessCfg struct {
 }
 
 func newHarness(t *testing.T, cfg harnessCfg) *testHarness {
-	towerAddrStr := "18.28.243.2:9911"
 	towerTCPAddr, err := net.ResolveTCPAddr("tcp", towerAddrStr)
 	if err != nil {
 		t.Fatalf("Unable to resolve tower TCP addr: %v", err)
@@ -412,6 +415,7 @@ func newHarness(t *testing.T, cfg harnessCfg) *testHarness {
 		DB:           serverDB,
 		ReadTimeout:  timeout,
 		WriteTimeout: timeout,
+		NodePrivKey:  privKey,
 		NewAddress: func() (btcutil.Address, error) {
 			return addr, nil
 		},
@@ -435,7 +439,6 @@ func newHarness(t *testing.T, cfg harnessCfg) *testHarness {
 		DB:            clientDB,
 		AuthDial:      mockNet.AuthDial,
 		SecretKeyRing: wtmock.NewSecretKeyRing(),
-		PrivateTower:  towerAddr,
 		Policy:        cfg.policy,
 		NewAddress: func() ([]byte, error) {
 			return addrScript, nil
@@ -458,6 +461,10 @@ func newHarness(t *testing.T, cfg harnessCfg) *testHarness {
 		server.Stop()
 		t.Fatalf("Unable to start wtclient: %v", err)
 	}
+	if err := client.AddTower(towerAddr); err != nil {
+		server.Stop()
+		t.Fatalf("Unable to add tower to wtclient: %v", err)
+	}
 
 	h := &testHarness{
 		t:         t,
@@ -505,7 +512,15 @@ func (h *testHarness) startServer() {
 func (h *testHarness) startClient() {
 	h.t.Helper()
 
-	var err error
+	towerTCPAddr, err := net.ResolveTCPAddr("tcp", towerAddrStr)
+	if err != nil {
+		h.t.Fatalf("Unable to resolve tower TCP addr: %v", err)
+	}
+	towerAddr := &lnwire.NetAddress{
+		IdentityKey: h.serverCfg.NodePrivKey.PubKey(),
+		Address:     towerTCPAddr,
+	}
+
 	h.client, err = wtclient.New(h.clientCfg)
 	if err != nil {
 		h.t.Fatalf("unable to create wtclient: %v", err)
@@ -513,6 +528,9 @@ func (h *testHarness) startClient() {
 	if err := h.client.Start(); err != nil {
 		h.t.Fatalf("unable to start wtclient: %v", err)
 	}
+	if err := h.client.AddTower(towerAddr); err != nil {
+		h.t.Fatalf("unable to add tower to wtclient: %v", err)
+	}
 }
 
 // chanIDFromInt creates a unique channel id given a unique integral id.

watchtower/wtclient/interface.go

@@ -17,11 +17,31 @@ type DB interface {
 	// CreateTower initialize an address record used to communicate with a
 	// watchtower. Each Tower is assigned a unique ID, that is used to
 	// amortize storage costs of the public key when used by multiple
-	// sessions.
+	// sessions. If the tower already exists, the address is appended to the
+	// list of all addresses used to that tower previously and its
+	// corresponding sessions are marked as active.
 	CreateTower(*lnwire.NetAddress) (*wtdb.Tower, error)
 
-	// LoadTower retrieves a tower by its tower ID.
-	LoadTower(wtdb.TowerID) (*wtdb.Tower, error)
+	// RemoveTower modifies a tower's record within the database. If an
+	// address is provided, then _only_ the address record should be removed
+	// from the tower's persisted state. Otherwise, we'll attempt to mark
+	// the tower as inactive by marking all of its sessions inactive. If any
+	// of its sessions has unacked updates, then ErrTowerUnackedUpdates is
+	// returned. If the tower doesn't have any sessions at all, it'll be
+	// completely removed from the database.
+	//
+	// NOTE: An error is not returned if the tower doesn't exist.
+	RemoveTower(*btcec.PublicKey, net.Addr) error
+
+	// LoadTower retrieves a tower by its public key.
+	LoadTower(*btcec.PublicKey) (*wtdb.Tower, error)
+
+	// LoadTowerByID retrieves a tower by its tower ID.
+	LoadTowerByID(wtdb.TowerID) (*wtdb.Tower, error)
+
+	// ListTowers retrieves the list of towers available within the
+	// database.
+	ListTowers() ([]*wtdb.Tower, error)
 
 	// NextSessionKeyIndex reserves a new session key derivation index for a
 	// particular tower id. The index is reserved for that tower until
@@ -38,8 +58,10 @@ type DB interface {
 
 	// ListClientSessions returns all sessions that have not yet been
 	// exhausted. This is used on startup to find any sessions which may
-	// still be able to accept state updates.
-	ListClientSessions() (map[wtdb.SessionID]*wtdb.ClientSession, error)
+	// still be able to accept state updates. An optional tower ID can be
+	// used to filter out any client sessions in the response that do not
+	// correspond to this tower.
+	ListClientSessions(*wtdb.TowerID) (map[wtdb.SessionID]*wtdb.ClientSession, error)
 
 	// FetchChanSummaries loads a mapping from all registered channels to
 	// their channel summaries.

watchtower/wtclient/session_negotiator.go

@@ -240,9 +240,6 @@ retryWithBackoff:
 		}
 	}
 
-	// Before attempting a bout of session negotiation, reset the candidate
-	// iterator to ensure the results are fresh.
-	n.cfg.Candidates.Reset()
 	for {
 		select {
 		case <-n.quit:
@@ -267,6 +264,13 @@ retryWithBackoff:
 			log.Debugf("Unable to get new tower candidate, "+
 				"retrying after %v -- reason: %v", backoff, err)
 
+			// Only reset the iterator once we've exhausted all
+			// candidates. Doing so allows us to load balance
+			// sessions better amongst all of the tower candidates.
+			if err == ErrTowerCandidatesExhausted {
+				n.cfg.Candidates.Reset()
+			}
+
 			goto retryWithBackoff
 		}
 

watchtower/wtclient/session_queue.go

@@ -145,9 +145,6 @@ func newSessionQueue(cfg *sessionQueueConfig) *sessionQueue {
 // backups.
 func (q *sessionQueue) Start() {
 	q.started.Do(func() {
-		// TODO(conner): load prior committed state updates from disk an
-		// populate in queue.
-
 		go q.sessionManager()
 	})
 }

watchtower/wtclient/stats.go

@@ -1,51 +1,97 @@
 package wtclient
 
-import "fmt"
+import (
+	"fmt"
+	"sync"
+)
 
-type clientStats struct {
-	numTasksReceived     int
-	numTasksAccepted     int
-	numTasksIneligible   int
-	numSessionsAcquired  int
-	numSessionsExhausted int
+// ClientStats is a collection of in-memory statistics of the actions the client
+// has performed since its creation.
+type ClientStats struct {
+	mu sync.Mutex
+
+	// NumTasksReceived is the total number of backups that are pending to
+	// be acknowledged by all active and exhausted watchtower sessions.
+	NumTasksReceived int
+
+	// NumTasksAccepted is the total number of backups made to all active
+	// and exhausted watchtower sessions.
+	NumTasksAccepted int
+
+	// NumTasksIneligible is the total number of backups that all active and
+	// exhausted watchtower sessions have failed to acknowledge.
+	NumTasksIneligible int
+
+	// NumSessionsAcquired is the total number of new sessions made to
+	// watchtowers.
+	NumSessionsAcquired int
+
+	// NumSessionsExhausted is the total number of watchtower sessions that
+	// have been exhausted.
+	NumSessionsExhausted int
 }
 
 // taskReceived increments the number to backup requests the client has received
 // from active channels.
-func (s *clientStats) taskReceived() {
-	s.numTasksReceived++
+func (s *ClientStats) taskReceived() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.NumTasksReceived++
 }
 
 // taskAccepted increments the number of tasks that have been assigned to active
 // session queues, and are awaiting upload to a tower.
-func (s *clientStats) taskAccepted() {
-	s.numTasksAccepted++
+func (s *ClientStats) taskAccepted() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.NumTasksAccepted++
 }
 
 // taskIneligible increments the number of tasks that were unable to satisfy the
 // active session queue's policy. These can potentially be retried later, but
 // typically this means that the balance created dust outputs, so it may not be
 // worth backing up at all.
-func (s *clientStats) taskIneligible() {
-	s.numTasksIneligible++
+func (s *ClientStats) taskIneligible() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.NumTasksIneligible++
 }
 
 // sessionAcquired increments the number of sessions that have been successfully
 // negotiated by the client during this execution.
-func (s *clientStats) sessionAcquired() {
-	s.numSessionsAcquired++
+func (s *ClientStats) sessionAcquired() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.NumSessionsAcquired++
 }
 
 // sessionExhausted increments the number of session that have become full as a
 // result of accepting backup tasks.
-func (s *clientStats) sessionExhausted() {
-	s.numSessionsExhausted++
+func (s *ClientStats) sessionExhausted() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.NumSessionsExhausted++
 }
 
 // String returns a human readable summary of the client's metrics.
-func (s clientStats) String() string {
+func (s *ClientStats) String() string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	return fmt.Sprintf("tasks(received=%d accepted=%d ineligible=%d) "+
-		"sessions(acquired=%d exhausted=%d)", s.numTasksReceived,
-		s.numTasksAccepted, s.numTasksIneligible, s.numSessionsAcquired,
-		s.numSessionsExhausted)
+		"sessions(acquired=%d exhausted=%d)", s.NumTasksReceived,
+		s.NumTasksAccepted, s.NumTasksIneligible, s.NumSessionsAcquired,
+		s.NumSessionsExhausted)
+}
+
+// Copy returns a copy of the current stats.
+func (s *ClientStats) Copy() ClientStats {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return ClientStats{
+		NumTasksReceived:     s.NumTasksReceived,
+		NumTasksAccepted:     s.NumTasksAccepted,
+		NumTasksIneligible:   s.NumTasksIneligible,
+		NumSessionsAcquired:  s.NumSessionsAcquired,
+		NumSessionsExhausted: s.NumSessionsExhausted,
+	}
 }

watchtower/wtdb/client_db.go

@@ -7,6 +7,7 @@ import (
 	"math"
 	"net"
 
+	"github.com/btcsuite/btcd/btcec"
 	"github.com/coreos/bbolt"
 	"github.com/lightningnetwork/lnd/lnwire"
 )
@@ -55,6 +56,11 @@ var (
 	// database.
 	ErrTowerNotFound = errors.New("tower not found")
 
+	// ErrTowerUnackedUpdates is an error returned when we attempt to mark a
+	// tower's sessions as inactive, but one of its sessions has unacked
+	// updates.
+	ErrTowerUnackedUpdates = errors.New("tower has unacked updates")
+
 	// ErrCorruptClientSession signals that the client session's on-disk
 	// structure deviates from what is expected.
 	ErrCorruptClientSession = errors.New("client session corrupted")
@@ -199,9 +205,11 @@ func (c *ClientDB) Close() error {
 	return c.db.Close()
 }
 
-// CreateTower initializes a database entry with the given lightning address. If
-// the tower exists, the address is append to the list of all addresses used to
-// that tower previously.
+// CreateTower initialize an address record used to communicate with a
+// watchtower. Each Tower is assigned a unique ID, that is used to amortize
+// storage costs of the public key when used by multiple sessions. If the tower
+// already exists, the address is appended to the list of all addresses used to
+// that tower previously and its corresponding sessions are marked as active.
 func (c *ClientDB) CreateTower(lnAddr *lnwire.NetAddress) (*Tower, error) {
 	var towerPubKey [33]byte
 	copy(towerPubKey[:], lnAddr.IdentityKey.SerializeCompressed())
@@ -233,6 +241,32 @@ func (c *ClientDB) CreateTower(lnAddr *lnwire.NetAddress) (*Tower, error) {
 			// address is a duplicate, this will result in no
 			// change.
 			tower.AddAddress(lnAddr.Address)
+
+			// If there are any client sessions that correspond to
+			// this tower, we'll mark them as active to ensure we
+			// load them upon restarts.
+			//
+			// TODO(wilmer): with an index of tower -> sessions we
+			// can avoid the linear lookup.
+			sessions := tx.Bucket(cSessionBkt)
+			if sessions == nil {
+				return ErrUninitializedDB
+			}
+			towerID := TowerIDFromBytes(towerIDBytes)
+			towerSessions, err := listClientSessions(
+				sessions, &towerID,
+			)
+			if err != nil {
+				return err
+			}
+			for _, session := range towerSessions {
+				err := markSessionStatus(
+					sessions, session, CSessionActive,
+				)
+				if err != nil {
+					return err
+				}
+			}
 		} else {
 			// No such tower exists, create a new tower id for our
 			// new tower. The error is unhandled since NextSequence
@@ -265,8 +299,89 @@ func (c *ClientDB) CreateTower(lnAddr *lnwire.NetAddress) (*Tower, error) {
 	return tower, nil
 }
 
-// LoadTower retrieves a tower by its tower ID.
-func (c *ClientDB) LoadTower(towerID TowerID) (*Tower, error) {
+// RemoveTower modifies a tower's record within the database. If an address is
+// provided, then _only_ the address record should be removed from the tower's
+// persisted state. Otherwise, we'll attempt to mark the tower as inactive by
+// marking all of its sessions inactive. If any of its sessions has unacked
+// updates, then ErrTowerUnackedUpdates is returned. If the tower doesn't have
+// any sessions at all, it'll be completely removed from the database.
+//
+// NOTE: An error is not returned if the tower doesn't exist.
+func (c *ClientDB) RemoveTower(pubKey *btcec.PublicKey, addr net.Addr) error {
+	return c.db.Update(func(tx *bbolt.Tx) error {
+		towers := tx.Bucket(cTowerBkt)
+		if towers == nil {
+			return ErrUninitializedDB
+		}
+		towerIndex := tx.Bucket(cTowerIndexBkt)
+		if towerIndex == nil {
+			return ErrUninitializedDB
+		}
+
+		// Don't return an error if the watchtower doesn't exist to act
+		// as a NOP.
+		pubKeyBytes := pubKey.SerializeCompressed()
+		towerIDBytes := towerIndex.Get(pubKeyBytes)
+		if towerIDBytes == nil {
+			return nil
+		}
+
+		// If an address is provided, then we should _only_ remove the
+		// address record from the database.
+		if addr != nil {
+			tower, err := getTower(towers, towerIDBytes)
+			if err != nil {
+				return err
+			}
+			tower.RemoveAddress(addr)
+			return putTower(towers, tower)
+		}
+
+		// Otherwise, we should attempt to mark the tower's sessions as
+		// inactive.
+		//
+		// TODO(wilmer): with an index of tower -> sessions we can avoid
+		// the linear lookup.
+		sessions := tx.Bucket(cSessionBkt)
+		if sessions == nil {
+			return ErrUninitializedDB
+		}
+		towerID := TowerIDFromBytes(towerIDBytes)
+		towerSessions, err := listClientSessions(sessions, &towerID)
+		if err != nil {
+			return err
+		}
+
+		// If it doesn't have any, we can completely remove it from the
+		// database.
+		if len(towerSessions) == 0 {
+			if err := towerIndex.Delete(pubKeyBytes); err != nil {
+				return err
+			}
+			return towers.Delete(towerIDBytes)
+		}
+
+		// We'll mark its sessions as inactive as long as they don't
+		// have any pending updates to ensure we don't load them upon
+		// restarts.
+		for _, session := range towerSessions {
+			if len(session.CommittedUpdates) > 0 {
+				return ErrTowerUnackedUpdates
+			}
+			err := markSessionStatus(
+				sessions, session, CSessionInactive,
+			)
+			if err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+}
+
+// LoadTowerByID retrieves a tower by its tower ID.
+func (c *ClientDB) LoadTowerByID(towerID TowerID) (*Tower, error) {
 	var tower *Tower
 	err := c.db.View(func(tx *bbolt.Tx) error {
 		towers := tx.Bucket(cTowerBkt)
@@ -285,6 +400,60 @@ func (c *ClientDB) LoadTower(towerID TowerID) (*Tower, error) {
 	return tower, nil
 }
 
+// LoadTower retrieves a tower by its public key.
+func (c *ClientDB) LoadTower(pubKey *btcec.PublicKey) (*Tower, error) {
+	var tower *Tower
+	err := c.db.View(func(tx *bbolt.Tx) error {
+		towers := tx.Bucket(cTowerBkt)
+		if towers == nil {
+			return ErrUninitializedDB
+		}
+		towerIndex := tx.Bucket(cTowerIndexBkt)
+		if towerIndex == nil {
+			return ErrUninitializedDB
+		}
+
+		towerIDBytes := towerIndex.Get(pubKey.SerializeCompressed())
+		if towerIDBytes == nil {
+			return ErrTowerNotFound
+		}
+
+		var err error
+		tower, err = getTower(towers, towerIDBytes)
+		return err
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return tower, nil
+}
+
+// ListTowers retrieves the list of towers available within the database.
+func (c *ClientDB) ListTowers() ([]*Tower, error) {
+	var towers []*Tower
+	err := c.db.View(func(tx *bbolt.Tx) error {
+		towerBucket := tx.Bucket(cTowerBkt)
+		if towerBucket == nil {
+			return ErrUninitializedDB
+		}
+
+		return towerBucket.ForEach(func(towerIDBytes, _ []byte) error {
+			tower, err := getTower(towerBucket, towerIDBytes)
+			if err != nil {
+				return err
+			}
+			towers = append(towers, tower)
+			return nil
+		})
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return towers, nil
+}
+
 // NextSessionKeyIndex reserves a new session key derivation index for a
 // particular tower id. The index is reserved for that tower until
 // CreateClientSession is invoked for that tower and index, at which point a new
@@ -384,29 +553,53 @@ func (c *ClientDB) CreateClientSession(session *ClientSession) error {
 	})
 }
 
-// ListClientSessions returns the set of all client sessions known to the db.
-func (c *ClientDB) ListClientSessions() (map[SessionID]*ClientSession, error) {
-	clientSessions := make(map[SessionID]*ClientSession)
+// ListClientSessions returns the set of all client sessions known to the db. An
+// optional tower ID can be used to filter out any client sessions in the
+// response that do not correspond to this tower.
+func (c *ClientDB) ListClientSessions(id *TowerID) (map[SessionID]*ClientSession, error) {
+	var clientSessions map[SessionID]*ClientSession
 	err := c.db.View(func(tx *bbolt.Tx) error {
 		sessions := tx.Bucket(cSessionBkt)
 		if sessions == nil {
 			return ErrUninitializedDB
 		}
+		var err error
+		clientSessions, err = listClientSessions(sessions, id)
+		return err
+	})
+	if err != nil {
+		return nil, err
+	}
 
-		return sessions.ForEach(func(k, _ []byte) error {
-			// We'll load the full client session since the client
-			// will need the CommittedUpdates and AckedUpdates on
-			// startup to resume committed updates and compute the
-			// highest known commit height for each channel.
-			session, err := getClientSession(sessions, k)
-			if err != nil {
-				return err
-			}
+	return clientSessions, nil
+}
 
-			clientSessions[session.ID] = session
+// listClientSessions returns the set of all client sessions known to the db. An
+// optional tower ID can be used to filter out any client sessions in the
+// response that do not correspond to this tower.
+func listClientSessions(sessions *bbolt.Bucket,
+	id *TowerID) (map[SessionID]*ClientSession, error) {
 
+	clientSessions := make(map[SessionID]*ClientSession)
+	err := sessions.ForEach(func(k, _ []byte) error {
+		// We'll load the full client session since the client will need
+		// the CommittedUpdates and AckedUpdates on startup to resume
+		// committed updates and compute the highest known commit height
+		// for each channel.
+		session, err := getClientSession(sessions, k)
+		if err != nil {
+			return err
+		}
+
+		// Filter out any sessions that don't correspond to the given
+		// tower if one was set.
+		if id != nil && session.TowerID != *id {
 			return nil
-		})
+		}
+
+		clientSessions[session.ID] = session
+
+		return nil
 	})
 	if err != nil {
 		return nil, err
@@ -847,6 +1040,15 @@ func putClientSessionBody(sessions *bbolt.Bucket,
 	return sessionBkt.Put(cSessionBody, b.Bytes())
 }
 
+// markSessionStatus updates the persisted state of the session to the new
+// status.
+func markSessionStatus(sessions *bbolt.Bucket, session *ClientSession,
+	status CSessionStatus) error {
+
+	session.Status = status
+	return putClientSessionBody(sessions, session)
+}
+
 // getChanSummary loads a ClientChanSummary for the passed chanID.
 func getChanSummary(chanSummaries *bbolt.Bucket,
 	chanID lnwire.ChannelID) (*ClientChanSummary, error) {

watchtower/wtdb/client_db_test.go

@@ -10,6 +10,7 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/btcsuite/btcd/btcec"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/watchtower/blob"
 	"github.com/lightningnetwork/lnd/watchtower/wtclient"
@@ -48,10 +49,10 @@ func (h *clientDBHarness) insertSession(session *wtdb.ClientSession, expErr erro
 	}
 }
 
-func (h *clientDBHarness) listSessions() map[wtdb.SessionID]*wtdb.ClientSession {
+func (h *clientDBHarness) listSessions(id *wtdb.TowerID) map[wtdb.SessionID]*wtdb.ClientSession {
 	h.t.Helper()
 
-	sessions, err := h.db.ListClientSessions()
+	sessions, err := h.db.ListClientSessions(id)
 	if err != nil {
 		h.t.Fatalf("unable to list client sessions: %v", err)
 	}
@@ -89,13 +90,81 @@ func (h *clientDBHarness) createTower(lnAddr *lnwire.NetAddress,
 		h.t.Fatalf("tower id should never be 0")
 	}
 
+	for _, session := range h.listSessions(&tower.ID) {
+		if session.Status != wtdb.CSessionActive {
+			h.t.Fatalf("expected status for session %v to be %v, "+
+				"got %v", session.ID, wtdb.CSessionActive,
+				session.Status)
+		}
+	}
+
 	return tower
 }
 
-func (h *clientDBHarness) loadTower(id wtdb.TowerID, expErr error) *wtdb.Tower {
+func (h *clientDBHarness) removeTower(pubKey *btcec.PublicKey, addr net.Addr,
+	hasSessions bool, expErr error) {
+
 	h.t.Helper()
 
-	tower, err := h.db.LoadTower(id)
+	if err := h.db.RemoveTower(pubKey, addr); err != expErr {
+		h.t.Fatalf("expected remove tower error: %v, got %v", expErr, err)
+	}
+	if expErr != nil {
+		return
+	}
+
+	if addr != nil {
+		tower, err := h.db.LoadTower(pubKey)
+		if err != nil {
+			h.t.Fatalf("expected tower %x to still exist",
+				pubKey.SerializeCompressed())
+		}
+
+		removedAddr := addr.String()
+		for _, towerAddr := range tower.Addresses {
+			if towerAddr.String() == removedAddr {
+				h.t.Fatalf("address %v not removed for tower %x",
+					removedAddr, pubKey.SerializeCompressed())
+			}
+		}
+	} else {
+		tower, err := h.db.LoadTower(pubKey)
+		if hasSessions && err != nil {
+			h.t.Fatalf("expected tower %x with sessions to still "+
+				"exist", pubKey.SerializeCompressed())
+		}
+		if !hasSessions && err == nil {
+			h.t.Fatalf("expected tower %x with no sessions to not "+
+				"exist", pubKey.SerializeCompressed())
+		}
+		if !hasSessions {
+			return
+		}
+		for _, session := range h.listSessions(&tower.ID) {
+			if session.Status != wtdb.CSessionInactive {
+				h.t.Fatalf("expected status for session %v to "+
+					"be %v, got %v", session.ID,
+					wtdb.CSessionInactive, session.Status)
+			}
+		}
+	}
+}
+
+func (h *clientDBHarness) loadTower(pubKey *btcec.PublicKey, expErr error) *wtdb.Tower {
+	h.t.Helper()
+
+	tower, err := h.db.LoadTower(pubKey)
+	if err != expErr {
+		h.t.Fatalf("expected load tower error: %v, got: %v", expErr, err)
+	}
+
+	return tower
+}
+
+func (h *clientDBHarness) loadTowerByID(id wtdb.TowerID, expErr error) *wtdb.Tower {
+	h.t.Helper()
+
+	tower, err := h.db.LoadTowerByID(id)
 	if err != expErr {
 		h.t.Fatalf("expected load tower error: %v, got: %v", expErr, err)
 	}
@@ -172,7 +241,7 @@ func testCreateClientSession(h *clientDBHarness) {
 
 	// First, assert that this session is not already present in the
 	// database.
-	if _, ok := h.listSessions()[session.ID]; ok {
+	if _, ok := h.listSessions(nil)[session.ID]; ok {
 		h.t.Fatalf("session for id %x should not exist yet", session.ID)
 	}
 
@@ -202,7 +271,7 @@ func testCreateClientSession(h *clientDBHarness) {
 	h.insertSession(session, nil)
 
 	// Verify that the session now exists in the database.
-	if _, ok := h.listSessions()[session.ID]; !ok {
+	if _, ok := h.listSessions(nil)[session.ID]; !ok {
 		h.t.Fatalf("session for id %x should exist now", session.ID)
 	}
 
@@ -218,12 +287,57 @@ func testCreateClientSession(h *clientDBHarness) {
 	}
 }
 
+// testFilterClientSessions asserts that we can correctly filter client sessions
+// for a specific tower.
+func testFilterClientSessions(h *clientDBHarness) {
+	// We'll create three client sessions, the first two belonging to one
+	// tower, and the last belonging to another one.
+	const numSessions = 3
+	towerSessions := make(map[wtdb.TowerID][]wtdb.SessionID)
+	for i := 0; i < numSessions; i++ {
+		towerID := wtdb.TowerID(1)
+		if i == numSessions-1 {
+			towerID = wtdb.TowerID(2)
+		}
+		keyIndex := h.nextKeyIndex(towerID, nil)
+		sessionID := wtdb.SessionID([33]byte{byte(i)})
+		h.insertSession(&wtdb.ClientSession{
+			ClientSessionBody: wtdb.ClientSessionBody{
+				TowerID: towerID,
+				Policy: wtpolicy.Policy{
+					MaxUpdates: 100,
+				},
+				RewardPkScript: []byte{0x01, 0x02, 0x03},
+				KeyIndex:       keyIndex,
+			},
+			ID: sessionID,
+		}, nil)
+		towerSessions[towerID] = append(towerSessions[towerID], sessionID)
+	}
+
+	// We should see the expected sessions for each tower when filtering
+	// them.
+	for towerID, expectedSessions := range towerSessions {
+		sessions := h.listSessions(&towerID)
+		if len(sessions) != len(expectedSessions) {
+			h.t.Fatalf("expected %v sessions for tower %v, got %v",
+				len(expectedSessions), towerID, len(sessions))
+		}
+		for _, expectedSession := range expectedSessions {
+			if _, ok := sessions[expectedSession]; !ok {
+				h.t.Fatalf("expected session %v for tower %v",
+					expectedSession, towerID)
+			}
+		}
+	}
+}
+
 // testCreateTower asserts the behavior of creating new Tower objects within the
 // database, and that the latest address is always prepended to the list of
 // known addresses for the tower.
 func testCreateTower(h *clientDBHarness) {
 	// Test that loading a tower with an arbitrary tower id fails.
-	h.loadTower(20, wtdb.ErrTowerNotFound)
+	h.loadTowerByID(20, wtdb.ErrTowerNotFound)
 
 	pk, err := randPubKey()
 	if err != nil {
@@ -241,7 +355,12 @@ func testCreateTower(h *clientDBHarness) {
 
 	// Load the tower from the database and assert that it matches the tower
 	// we created.
-	tower2 := h.loadTower(tower.ID, nil)
+	tower2 := h.loadTowerByID(tower.ID, nil)
+	if !reflect.DeepEqual(tower, tower2) {
+		h.t.Fatalf("loaded tower mismatch, want: %v, got: %v",
+			tower, tower2)
+	}
+	tower2 = h.loadTower(pk, err)
 	if !reflect.DeepEqual(tower, tower2) {
 		h.t.Fatalf("loaded tower mismatch, want: %v, got: %v",
 			tower, tower2)
@@ -272,7 +391,12 @@ func testCreateTower(h *clientDBHarness) {
 
 	// Load the tower from the database, and assert that it matches the
 	// tower returned from creation.
-	towerNewAddr2 := h.loadTower(tower.ID, nil)
+	towerNewAddr2 := h.loadTowerByID(tower.ID, nil)
+	if !reflect.DeepEqual(towerNewAddr, towerNewAddr2) {
+		h.t.Fatalf("loaded tower mismatch, want: %v, got: %v",
+			towerNewAddr, towerNewAddr2)
+	}
+	towerNewAddr2 = h.loadTower(pk, nil)
 	if !reflect.DeepEqual(towerNewAddr, towerNewAddr2) {
 		h.t.Fatalf("loaded tower mismatch, want: %v, got: %v",
 			towerNewAddr, towerNewAddr2)
@@ -290,6 +414,82 @@ func testCreateTower(h *clientDBHarness) {
 	}
 }
 
+// testRemoveTower asserts the behavior of removing Tower objects as a whole and
+// removing addresses from Tower objects within the database.
+func testRemoveTower(h *clientDBHarness) {
+	// Generate a random public key we'll use for our tower.
+	pk, err := randPubKey()
+	if err != nil {
+		h.t.Fatalf("unable to generate pubkey: %v", err)
+	}
+
+	// Removing a tower that does not exist within the database should
+	// result in a NOP.
+	h.removeTower(pk, nil, false, nil)
+
+	// We'll create a tower with two addresses.
+	addr1 := &net.TCPAddr{IP: []byte{0x01, 0x00, 0x00, 0x00}, Port: 9911}
+	addr2 := &net.TCPAddr{IP: []byte{0x02, 0x00, 0x00, 0x00}, Port: 9911}
+	h.createTower(&lnwire.NetAddress{
+		IdentityKey: pk,
+		Address:     addr1,
+	}, nil)
+	h.createTower(&lnwire.NetAddress{
+		IdentityKey: pk,
+		Address:     addr2,
+	}, nil)
+
+	// We'll then remove the second address. We should now only see the
+	// first.
+	h.removeTower(pk, addr2, false, nil)
+
+	// We'll then remove the first address. We should now see that the tower
+	// has no addresses left.
+	h.removeTower(pk, addr1, false, nil)
+
+	// Removing the tower as a whole from the database should succeed since
+	// there aren't any active sessions for it.
+	h.removeTower(pk, nil, false, nil)
+
+	// We'll then recreate the tower, but this time we'll create a session
+	// for it.
+	tower := h.createTower(&lnwire.NetAddress{
+		IdentityKey: pk,
+		Address:     addr1,
+	}, nil)
+
+	session := &wtdb.ClientSession{
+		ClientSessionBody: wtdb.ClientSessionBody{
+			TowerID: tower.ID,
+			Policy: wtpolicy.Policy{
+				MaxUpdates: 100,
+			},
+			RewardPkScript: []byte{0x01, 0x02, 0x03},
+			KeyIndex:       h.nextKeyIndex(tower.ID, nil),
+		},
+		ID: wtdb.SessionID([33]byte{0x01}),
+	}
+	h.insertSession(session, nil)
+	update := randCommittedUpdate(h.t, 1)
+	h.commitUpdate(&session.ID, update, nil)
+
+	// We should not be able to fully remove it from the database since
+	// there's a session and it has unacked updates.
+	h.removeTower(pk, nil, true, wtdb.ErrTowerUnackedUpdates)
+
+	// Removing the tower after all sessions no longer have unacked updates
+	// should result in the sessions becoming inactive.
+	h.ackUpdate(&session.ID, 1, 1, nil)
+	h.removeTower(pk, nil, true, nil)
+
+	// Creating the tower again should mark all of the sessions active once
+	// again.
+	h.createTower(&lnwire.NetAddress{
+		IdentityKey: pk,
+		Address:     addr1,
+	}, nil)
+}
+
 // testChanSummaries tests the process of a registering a channel and its
 // associated sweep pkscript.
 func testChanSummaries(h *clientDBHarness) {
@@ -357,7 +557,7 @@ func testCommitUpdate(h *clientDBHarness) {
 	// Assert that the committed update appears in the client session's
 	// CommittedUpdates map when loaded from disk and that there are no
 	// AckedUpdates.
-	dbSession := h.listSessions()[session.ID]
+	dbSession := h.listSessions(nil)[session.ID]
 	checkCommittedUpdates(h.t, dbSession, []wtdb.CommittedUpdate{
 		*update1,
 	})
@@ -374,7 +574,7 @@ func testCommitUpdate(h *clientDBHarness) {
 	}
 
 	// Assert that the loaded ClientSession is the same as before.
-	dbSession = h.listSessions()[session.ID]
+	dbSession = h.listSessions(nil)[session.ID]
 	checkCommittedUpdates(h.t, dbSession, []wtdb.CommittedUpdate{
 		*update1,
 	})
@@ -396,7 +596,7 @@ func testCommitUpdate(h *clientDBHarness) {
 
 	// Check that both updates now appear as committed on the ClientSession
 	// loaded from disk.
-	dbSession = h.listSessions()[session.ID]
+	dbSession = h.listSessions(nil)[session.ID]
 	checkCommittedUpdates(h.t, dbSession, []wtdb.CommittedUpdate{
 		*update1,
 		*update2,
@@ -410,7 +610,7 @@ func testCommitUpdate(h *clientDBHarness) {
 	h.commitUpdate(&session.ID, update4, wtdb.ErrCommitUnorderedUpdate)
 
 	// Assert that the ClientSession loaded from disk remains unchanged.
-	dbSession = h.listSessions()[session.ID]
+	dbSession = h.listSessions(nil)[session.ID]
 	checkCommittedUpdates(h.t, dbSession, []wtdb.CommittedUpdate{
 		*update1,
 		*update2,
@@ -467,7 +667,7 @@ func testAckUpdate(h *clientDBHarness) {
 
 	// Assert that the ClientSession loaded from disk has one update in it's
 	// AckedUpdates map, and that the committed update has been removed.
-	dbSession := h.listSessions()[session.ID]
+	dbSession := h.listSessions(nil)[session.ID]
 	checkCommittedUpdates(h.t, dbSession, nil)
 	checkAckedUpdates(h.t, dbSession, map[uint16]wtdb.BackupID{
 		1: update1.BackupID,
@@ -487,7 +687,7 @@ func testAckUpdate(h *clientDBHarness) {
 	h.ackUpdate(&session.ID, 2, 2, nil)
 
 	// Assert that both updates exist as AckedUpdates when loaded from disk.
-	dbSession = h.listSessions()[session.ID]
+	dbSession = h.listSessions(nil)[session.ID]
 	checkCommittedUpdates(h.t, dbSession, nil)
 	checkAckedUpdates(h.t, dbSession, map[uint16]wtdb.BackupID{
 		1: update1.BackupID,
@@ -620,10 +820,18 @@ func TestClientDB(t *testing.T) {
 			name: "create client session",
 			run:  testCreateClientSession,
 		},
+		{
+			name: "filter client sessions",
+			run:  testFilterClientSessions,
+		},
 		{
 			name: "create tower",
 			run:  testCreateTower,
 		},
+		{
+			name: "remove tower",
+			run:  testRemoveTower,
+		},
 		{
 			name: "chan summaries",
 			run:  testChanSummaries,

watchtower/wtdb/client_session.go

@@ -18,6 +18,10 @@ const (
 	// CSessionActive indicates that the ClientSession is active and can be
 	// used for backups.
 	CSessionActive CSessionStatus = 0
+
+	// CSessionInactive indicates that the ClientSession is inactive and
+	// cannot be used for backups.
+	CSessionInactive CSessionStatus = 1
 )
 
 // ClientSession encapsulates a SessionInfo returned from a successful

watchtower/wtdb/tower.go

@@ -1,6 +1,8 @@
 package wtdb
 
 import (
+	"encoding/hex"
+	"fmt"
 	"io"
 	"net"
 
@@ -62,6 +64,19 @@ func (t *Tower) AddAddress(addr net.Addr) {
 	t.Addresses = append([]net.Addr{addr}, t.Addresses...)
 }
 
+// RemoveAddress removes the given address from the tower's in-memory list of
+// addresses. If the address doesn't exist, then this will act as a NOP.
+func (t *Tower) RemoveAddress(addr net.Addr) {
+	addrStr := addr.String()
+	for i, address := range t.Addresses {
+		if address.String() != addrStr {
+			continue
+		}
+		t.Addresses = append(t.Addresses[:i], t.Addresses[i+1:]...)
+		return
+	}
+}
+
 // LNAddrs generates a list of lnwire.NetAddress from a Tower instance's
 // addresses. This can be used to have a client try multiple addresses for the
 // same Tower.
@@ -79,6 +94,15 @@ func (t *Tower) LNAddrs() []*lnwire.NetAddress {
 	return addrs
 }
 
+// String returns a user-friendly identifier of the tower.
+func (t *Tower) String() string {
+	pubKey := hex.EncodeToString(t.IdentityKey.SerializeCompressed())
+	if len(t.Addresses) == 0 {
+		return pubKey
+	}
+	return fmt.Sprintf("%v@%v", pubKey, t.Addresses[0])
+}
+
 // Encode writes the Tower to the passed io.Writer. The TowerID is not
 // serialized, since it acts as the key.
 func (t *Tower) Encode(w io.Writer) error {

watchtower/wtmock/client_db.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/btcsuite/btcd/btcec"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/watchtower/wtdb"
 )
@@ -37,9 +38,11 @@ func NewClientDB() *ClientDB {
 	}
 }
 
-// CreateTower initializes a database entry with the given lightning address. If
-// the tower exists, the address is append to the list of all addresses used to
-// that tower previously.
+// CreateTower initialize an address record used to communicate with a
+// watchtower. Each Tower is assigned a unique ID, that is used to amortize
+// storage costs of the public key when used by multiple sessions. If the tower
+// already exists, the address is appended to the list of all addresses used to
+// that tower previously and its corresponding sessions are marked as active.
 func (m *ClientDB) CreateTower(lnAddr *lnwire.NetAddress) (*wtdb.Tower, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -52,6 +55,15 @@ func (m *ClientDB) CreateTower(lnAddr *lnwire.NetAddress) (*wtdb.Tower, error) {
 	if ok {
 		tower = m.towers[towerID]
 		tower.AddAddress(lnAddr.Address)
+
+		towerSessions, err := m.listClientSessions(&towerID)
+		if err != nil {
+			return nil, err
+		}
+		for id, session := range towerSessions {
+			session.Status = wtdb.CSessionActive
+			m.activeSessions[id] = session
+		}
 	} else {
 		towerID = wtdb.TowerID(atomic.AddUint64(&m.nextTowerID, 1))
 		tower = &wtdb.Tower{
@@ -67,8 +79,83 @@ func (m *ClientDB) CreateTower(lnAddr *lnwire.NetAddress) (*wtdb.Tower, error) {
 	return copyTower(tower), nil
 }
 
-// LoadTower retrieves a tower by its tower ID.
-func (m *ClientDB) LoadTower(towerID wtdb.TowerID) (*wtdb.Tower, error) {
+// RemoveTower modifies a tower's record within the database. If an address is
+// provided, then _only_ the address record should be removed from the tower's
+// persisted state. Otherwise, we'll attempt to mark the tower as inactive by
+// marking all of its sessions inactive. If any of its sessions has unacked
+// updates, then ErrTowerUnackedUpdates is returned. If the tower doesn't have
+// any sessions at all, it'll be completely removed from the database.
+//
+// NOTE: An error is not returned if the tower doesn't exist.
+func (m *ClientDB) RemoveTower(pubKey *btcec.PublicKey, addr net.Addr) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	tower, err := m.loadTower(pubKey)
+	if err == wtdb.ErrTowerNotFound {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+
+	if addr != nil {
+		tower.RemoveAddress(addr)
+		m.towers[tower.ID] = tower
+		return nil
+	}
+
+	towerSessions, err := m.listClientSessions(&tower.ID)
+	if err != nil {
+		return err
+	}
+	if len(towerSessions) == 0 {
+		var towerPK towerPK
+		copy(towerPK[:], pubKey.SerializeCompressed())
+		delete(m.towerIndex, towerPK)
+		delete(m.towers, tower.ID)
+		return nil
+	}
+
+	for id, session := range towerSessions {
+		if len(session.CommittedUpdates) > 0 {
+			return wtdb.ErrTowerUnackedUpdates
+		}
+		session.Status = wtdb.CSessionInactive
+		m.activeSessions[id] = session
+	}
+
+	return nil
+}
+
+// LoadTower retrieves a tower by its public key.
+func (m *ClientDB) LoadTower(pubKey *btcec.PublicKey) (*wtdb.Tower, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.loadTower(pubKey)
+}
+
+// loadTower retrieves a tower by its public key.
+//
+// NOTE: This method requires the database's lock to be acquired.
+func (m *ClientDB) loadTower(pubKey *btcec.PublicKey) (*wtdb.Tower, error) {
+	var towerPK towerPK
+	copy(towerPK[:], pubKey.SerializeCompressed())
+
+	towerID, ok := m.towerIndex[towerPK]
+	if !ok {
+		return nil, wtdb.ErrTowerNotFound
+	}
+	tower, ok := m.towers[towerID]
+	if !ok {
+		return nil, wtdb.ErrTowerNotFound
+	}
+
+	return copyTower(tower), nil
+}
+
+// LoadTowerByID retrieves a tower by its tower ID.
+func (m *ClientDB) LoadTowerByID(towerID wtdb.TowerID) (*wtdb.Tower, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -79,6 +166,19 @@ func (m *ClientDB) LoadTower(towerID wtdb.TowerID) (*wtdb.Tower, error) {
 	return nil, wtdb.ErrTowerNotFound
 }
 
+// ListTowers retrieves the list of towers available within the database.
+func (m *ClientDB) ListTowers() ([]*wtdb.Tower, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	towers := make([]*wtdb.Tower, 0, len(m.towers))
+	for _, tower := range m.towers {
+		towers = append(towers, copyTower(tower))
+	}
+
+	return towers, nil
+}
+
 // MarkBackupIneligible records that particular commit height is ineligible for
 // backup. This allows the client to track which updates it should not attempt
 // to retry after startup.
@@ -86,13 +186,28 @@ func (m *ClientDB) MarkBackupIneligible(chanID lnwire.ChannelID, commitHeight ui
 	return nil
 }
 
-// ListClientSessions returns the set of all client sessions known to the db.
-func (m *ClientDB) ListClientSessions() (map[wtdb.SessionID]*wtdb.ClientSession, error) {
+// ListClientSessions returns the set of all client sessions known to the db. An
+// optional tower ID can be used to filter out any client sessions in the
+// response that do not correspond to this tower.
+func (m *ClientDB) ListClientSessions(
+	tower *wtdb.TowerID) (map[wtdb.SessionID]*wtdb.ClientSession, error) {
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
+	return m.listClientSessions(tower)
+}
+
+// listClientSessions returns the set of all client sessions known to the db. An
+// optional tower ID can be used to filter out any client sessions in the
+// response that do not correspond to this tower.
+func (m *ClientDB) listClientSessions(
+	tower *wtdb.TowerID) (map[wtdb.SessionID]*wtdb.ClientSession, error) {
 
 	sessions := make(map[wtdb.SessionID]*wtdb.ClientSession)
 	for _, session := range m.activeSessions {
+		if tower != nil && *tower != session.TowerID {
+			continue
+		}
 		sessions[session.ID] = session
 	}