watchtower/wtclient: reliable, asynchronous pipeline for revoked state backups

watchtower/wtclient/backup_task.go

@@ -34,9 +34,8 @@ import (
 //     necessary components are stripped out and encrypted before being sent to
 //     the tower in a StateUpdate.
 type backupTask struct {
-	chanID       lnwire.ChannelID
-	commitHeight uint64
-	breachInfo   *lnwallet.BreachRetribution
+	id         wtdb.BackupID
+	breachInfo *lnwallet.BreachRetribution
 
 	// state-dependent variables
 
@@ -96,8 +95,10 @@ func newBackupTask(chanID *lnwire.ChannelID,
 	}
 
 	return &backupTask{
-		chanID:        *chanID,
-		commitHeight:  breachInfo.RevokedStateNum,
+		id: wtdb.BackupID{
+			ChanID:       *chanID,
+			CommitHeight: breachInfo.RevokedStateNum,
+		},
 		breachInfo:    breachInfo,
 		toLocalInput:  toLocalInput,
 		toRemoteInput: toRemoteInput,
@@ -125,7 +126,7 @@ func (t *backupTask) inputs() map[wire.OutPoint]input.Input {
 // SessionInfo's policy. If no error is returned, the task has been bound to the
 // session and can be queued to upload to the tower. Otherwise, the bind failed
 // and should be rescheduled with a different session.
-func (t *backupTask) bindSession(session *wtdb.SessionInfo) error {
+func (t *backupTask) bindSession(session *wtdb.ClientSession) error {
 
 	// First we'll begin by deriving a weight estimate for the justice
 	// transaction. The final weight can be different depending on whether
@@ -154,7 +155,7 @@ func (t *backupTask) bindSession(session *wtdb.SessionInfo) error {
 	// in the current session's policy.
 	outputs, err := session.Policy.ComputeJusticeTxOuts(
 		t.totalAmt, int64(weightEstimate.Weight()),
-		t.sweepPkScript, session.RewardAddress,
+		t.sweepPkScript, session.RewardPkScript,
 	)
 	if err != nil {
 		return err

watchtower/wtclient/backup_task_internal_test.go

@@ -69,7 +69,7 @@ type backupTaskTest struct {
 	expSweepAmt      int64
 	expRewardAmt     int64
 	expRewardScript  []byte
-	session          *wtdb.SessionInfo
+	session          *wtdb.ClientSession
 	bindErr          error
 	expSweepScript   []byte
 	signer           input.Signer
@@ -205,13 +205,13 @@ func genTaskTest(
 		expSweepAmt:      expSweepAmt,
 		expRewardAmt:     expRewardAmt,
 		expRewardScript:  rewardScript,
-		session: &wtdb.SessionInfo{
+		session: &wtdb.ClientSession{
 			Policy: wtpolicy.Policy{
 				BlobType:     blobType,
 				SweepFeeRate: sweepFeeRate,
 				RewardRate:   10000,
 			},
-			RewardAddress: rewardScript,
+			RewardPkScript: rewardScript,
 		},
 		bindErr:        bindErr,
 		expSweepScript: makeAddrSlice(22),
@@ -379,7 +379,7 @@ var backupTaskTests = []backupTaskTest{
 }
 
 // TestBackupTaskBind tests the initialization and binding of a backupTask to a
-// SessionInfo. After a succesfful bind, all parameters of the justice
+// ClientSession. After a successful bind, all parameters of the justice
 // transaction should be solidified, so we assert there correctness. In an
 // unsuccessful bind, the session-dependent parameters should be unmodified so
 // that the backup task can be rescheduled if necessary. Finally, we assert that
@@ -401,14 +401,14 @@ func testBackupTask(t *testing.T, test backupTaskTest) {
 
 	// Assert that all parameters set during initialization are properly
 	// populated.
-	if task.chanID != test.chanID {
+	if task.id.ChanID != test.chanID {
 		t.Fatalf("channel id mismatch, want: %s, got: %s",
-			test.chanID, task.chanID)
+			test.chanID, task.id.ChanID)
 	}
 
-	if task.commitHeight != test.breachInfo.RevokedStateNum {
+	if task.id.CommitHeight != test.breachInfo.RevokedStateNum {
 		t.Fatalf("commit height mismatch, want: %d, got: %d",
-			test.breachInfo.RevokedStateNum, task.commitHeight)
+			test.breachInfo.RevokedStateNum, task.id.CommitHeight)
 	}
 
 	if task.totalAmt != test.expTotalAmt {

watchtower/wtclient/candidate_iterator.go

@@ -0,0 +1,82 @@
+package wtclient
+
+import (
+	"container/list"
+	"sync"
+
+	"github.com/lightningnetwork/lnd/watchtower/wtdb"
+)
+
+// TowerCandidateIterator provides an abstraction for iterating through possible
+// watchtower addresses when attempting to create a new session.
+type TowerCandidateIterator interface {
+	// Reset clears any internal iterator state, making previously taken
+	// candidates available as long as they remain in the set.
+	Reset() error
+
+	// Next returns the next candidate tower. The iterator is not required
+	// to return results in any particular order.  If no more candidates are
+	// available, ErrTowerCandidatesExhausted is returned.
+	Next() (*wtdb.Tower, error)
+}
+
+// towerListIterator is a linked-list backed TowerCandidateIterator.
+type towerListIterator struct {
+	mu            sync.Mutex
+	candidates    *list.List
+	nextCandidate *list.Element
+}
+
+// Compile-time constraint to ensure *towerListIterator implements the
+// TowerCandidateIterator interface.
+var _ TowerCandidateIterator = (*towerListIterator)(nil)
+
+// newTowerListIterator initializes a new towerListIterator from a variadic list
+// of lnwire.NetAddresses.
+func newTowerListIterator(candidates ...*wtdb.Tower) *towerListIterator {
+	iter := &towerListIterator{
+		candidates: list.New(),
+	}
+
+	for _, candidate := range candidates {
+		iter.candidates.PushBack(candidate)
+	}
+	iter.Reset()
+
+	return iter
+}
+
+// Reset clears the iterators state, and makes the address at the front of the
+// list the next item to be returned..
+func (t *towerListIterator) Reset() error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	// Reset the next candidate to the front of the linked-list.
+	t.nextCandidate = t.candidates.Front()
+
+	return nil
+}
+
+// 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.
+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
+	}
+
+	// Propose the tower at the front of the list.
+	tower := t.nextCandidate.Value.(*wtdb.Tower)
+
+	// Set the next candidate to the subsequent element.
+	t.nextCandidate = t.nextCandidate.Next()
+
+	return tower, nil
+}
+
+// TODO(conner): implement graph-backed candidate iterator for public towers.