Merge pull request #7541 from ellemouton/statusClientBehind

watchtower: let client recover after data loss

docs/release-notes/release-notes-0.16.1.md

@@ -25,6 +25,8 @@
   needed](https://github.com/lightningnetwork/lnd/pull/7405). Also add a new
   `exclude_exhausted_sessions` boolean flag to the relevant lncli wtclient
   commands.
+* [Recover from StateUpdateCodeClientBehind 
+  error](https://github.com/lightningnetwork/lnd/pull/7541) after data loss. 
 
 ## Misc
 

watchtower/wtclient/client_test.go

@@ -1466,7 +1466,7 @@ var clientTests = []clientTest{
 			h.waitServerUpdates(hints[:numUpdates/2], time.Second)
 
 			// Stop the client, which should have no more backups.
-			h.client.Stop()
+			require.NoError(h.t, h.client.Stop())
 
 			// Record the policy that the first half was stored
 			// under. We'll expect the second half to also be stored
@@ -1527,7 +1527,7 @@ var clientTests = []clientTest{
 
 			// Restart the client, so we can ensure the deduping is
 			// maintained across restarts.
-			h.client.Stop()
+			require.NoError(h.t, h.client.Stop())
 			h.startClient()
 
 			// Try to back up the full range of retributions. Only
@@ -2013,6 +2013,58 @@ var clientTests = []clientTest{
 			require.NoError(h.t, err)
 		},
 	},
+	{
+		// Demonstrate that the client is unable to recover after
+		// deleting its database by skipping through key indices until
+		// it gets to one that does not result in the
+		// CreateSessionCodeAlreadyExists error code being returned from
+		// the server.
+		name: "continue after client database deletion",
+		cfg: harnessCfg{
+			localBalance:  localBalance,
+			remoteBalance: remoteBalance,
+			policy: wtpolicy.Policy{
+				TxPolicy:   defaultTxPolicy,
+				MaxUpdates: 5,
+			},
+		},
+		fn: func(h *testHarness) {
+			const (
+				numUpdates = 5
+				chanID     = 0
+			)
+
+			// Generate numUpdates retributions.
+			hints := h.advanceChannelN(chanID, numUpdates)
+
+			// Back half of the states up.
+			h.backupStates(chanID, 0, numUpdates/2, nil)
+
+			// Wait for the updates to be populated in the server's
+			// database.
+			h.waitServerUpdates(hints[:numUpdates/2], waitTime)
+
+			// Now stop the client and reset its database.
+			require.NoError(h.t, h.client.Stop())
+
+			db := wtmock.NewClientDB()
+			h.clientDB = db
+			h.clientCfg.DB = db
+
+			// Restart the client.
+			h.startClient()
+
+			// We need to re-register the channel due to the client
+			// db being reset.
+			h.registerChannel(0)
+
+			// Attempt to back up the remaining tasks.
+			h.backupStates(chanID, numUpdates/2, numUpdates, nil)
+
+			// Show that the server does get the remaining updates.
+			h.waitServerUpdates(hints[numUpdates/2:], waitTime)
+		},
+	},
 }
 
 // TestClient executes the client test suite, asserting the ability to backup

watchtower/wtclient/errors.go

@@ -34,4 +34,9 @@ var (
 	// revoked state because the channel had not been previously registered
 	// with the client.
 	ErrUnregisteredChannel = errors.New("channel is not registered")
+
+	// ErrSessionKeyAlreadyUsed indicates that the client attempted to
+	// create a new session with a tower with a session key that has already
+	// been used in the past.
+	ErrSessionKeyAlreadyUsed = errors.New("session key already used")
 )

watchtower/wtclient/interface.go

@@ -49,8 +49,8 @@ type DB interface {
 	// (tower, blob type) pair until CreateClientSession is invoked for that
 	// tower and index, at which point a new index for that tower can be
 	// reserved. Multiple calls to this method before CreateClientSession is
-	// invoked should return the same index.
-	NextSessionKeyIndex(wtdb.TowerID, blob.Type) (uint32, error)
+	// invoked should return the same index unless forceNext is true.
+	NextSessionKeyIndex(wtdb.TowerID, blob.Type, bool) (uint32, error)
 
 	// CreateClientSession saves a newly negotiated client session to the
 	// client's database. This enables the session to be used across

watchtower/wtclient/session_negotiator.go

@@ -1,6 +1,7 @@
 package wtclient
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -272,6 +273,7 @@ retryWithBackoff:
 		}
 	}
 
+tryNextCandidate:
 	for {
 		select {
 		case <-n.quit:
@@ -302,28 +304,39 @@ retryWithBackoff:
 		n.log.Debugf("Attempting session negotiation with tower=%x",
 			towerPub)
 
-		// Before proceeding, we will reserve a session key index to use
-		// with this specific tower. If one is already reserved, the
-		// existing index will be returned.
-		keyIndex, err := n.cfg.DB.NextSessionKeyIndex(
-			tower.ID, n.cfg.Policy.BlobType,
-		)
-		if err != nil {
-			n.log.Debugf("Unable to reserve session key index "+
-				"for tower=%x: %v", towerPub, err)
-			continue
-		}
+		var forceNextKey bool
+		for {
+			// Before proceeding, we will reserve a session key
+			// index to use with this specific tower. If one is
+			// already reserved, the existing index will be
+			// returned.
+			keyIndex, err := n.cfg.DB.NextSessionKeyIndex(
+				tower.ID, n.cfg.Policy.BlobType, forceNextKey,
+			)
+			if err != nil {
+				n.log.Debugf("Unable to reserve session key "+
+					"index for tower=%x: %v", towerPub, err)
 
-		// We'll now attempt the CreateSession dance with the tower to
-		// get a new session, trying all addresses if necessary.
-		err = n.createSession(tower, keyIndex)
-		if err != nil {
-			// An unexpected error occurred, updpate our backoff.
+				goto tryNextCandidate
+			}
+
+			// We'll now attempt the CreateSession dance with the
+			// tower to get a new session, trying all addresses if
+			// necessary.
+			err = n.createSession(tower, keyIndex)
+			if err == nil {
+				return
+			} else if errors.Is(err, ErrSessionKeyAlreadyUsed) {
+				forceNextKey = true
+				continue
+			}
+
+			// An unexpected error occurred, update our backoff.
 			updateBackoff()
 
 			n.log.Debugf("Session negotiation with tower=%x "+
-				"failed, trying again -- reason: %v",
-				tower.IdentityKey.SerializeCompressed(), err)
+				"failed, trying again -- reason: %v", towerPub,
+				err)
 
 			goto retryWithBackoff
 		}
@@ -360,7 +373,10 @@ func (n *sessionNegotiator) createSession(tower *Tower, keyIndex uint32) error {
 		err = n.tryAddress(sessionKey, keyIndex, tower, lnAddr)
 		tower.Addresses.ReleaseLock(addr)
 		switch {
-		case err == ErrPermanentTowerFailure:
+		case errors.Is(err, ErrSessionKeyAlreadyUsed):
+			return err
+
+		case errors.Is(err, ErrPermanentTowerFailure):
 			// TODO(conner): report to iterator? can then be reset
 			// with restart
 			fallthrough
@@ -454,12 +470,7 @@ func (n *sessionNegotiator) tryAddress(sessionKey keychain.SingleKeyECDH,
 	}
 
 	switch createSessionReply.Code {
-	case wtwire.CodeOK, wtwire.CreateSessionCodeAlreadyExists:
-
-		// TODO(conner): add last-applied to create session reply to
-		// handle case where we lose state, session already exists, and
-		// we want to possibly resume using the session
-
+	case wtwire.CodeOK:
 		// TODO(conner): validate reward address
 		rewardPkScript := createSessionReply.Data
 
@@ -500,6 +511,16 @@ func (n *sessionNegotiator) tryAddress(sessionKey keychain.SingleKeyECDH,
 			return ErrNegotiatorExiting
 		}
 
+	case wtwire.CreateSessionCodeAlreadyExists:
+		// TODO(conner): use the last-applied in the create session
+		//  reply to handle case where we lose state, session already
+		//  exists, and we want to possibly resume using the session.
+		//  NOTE that this should not be done until the server code
+		//  has been adapted to first check that the CreateSession
+		//  request is for the same blob-type as the initial session.
+
+		return ErrSessionKeyAlreadyUsed
+
 	// TODO(conner): handle error codes properly
 	case wtwire.CreateSessionCodeRejectBlobType:
 		return fmt.Errorf("tower rejected blob type: %v",

watchtower/wtdb/client_db.go

@@ -643,9 +643,10 @@ func (c *ClientDB) ListTowers() ([]*Tower, error) {
 // particular tower id. The index is reserved for that tower until
 // CreateClientSession is invoked for that tower and index, at which point a new
 // index for that tower can be reserved. Multiple calls to this method before
-// CreateClientSession is invoked should return the same index.
+// CreateClientSession is invoked should return the same index unless forceNext
+// is true.
 func (c *ClientDB) NextSessionKeyIndex(towerID TowerID,
-	blobType blob.Type) (uint32, error) {
+	blobType blob.Type, forceNext bool) (uint32, error) {
 
 	var index uint32
 	err := kvdb.Update(c.db, func(tx kvdb.RwTx) error {
@@ -654,33 +655,48 @@ func (c *ClientDB) NextSessionKeyIndex(towerID TowerID,
 			return ErrUninitializedDB
 		}
 
-		// Check the session key index to see if a key has already been
-		// reserved for this tower. If so, we'll deserialize and return
-		// the index directly.
 		var err error
-		index, err = getSessionKeyIndex(keyIndex, towerID, blobType)
-		if err == nil {
-			return nil
+		if !forceNext {
+			// Check the session key index to see if a key has
+			// already been reserved for this tower. If so, we'll
+			// deserialize and return the index directly.
+			index, err = getSessionKeyIndex(
+				keyIndex, towerID, blobType,
+			)
+			if err == nil {
+				return nil
+			}
 		}
 
-		// Otherwise, generate a new session key index since the node
-		// doesn't already have reserved index. The error is ignored
-		// since NextSequence can't fail inside Update.
-		index64, _ := keyIndex.NextSequence()
+		// By default, we use the next available bucket sequence as the
+		// key index. But if forceNext is true, then it is assumed that
+		// some data loss occurred and so the sequence is incremented a
+		// by a jump of 1000 so that we can arrive at a brand new key
+		// index quicker.
+		currentSequence := keyIndex.Sequence()
+		nextIndex := currentSequence + 1
+		if forceNext {
+			nextIndex = currentSequence + 1000
+		}
+
+		if err = keyIndex.SetSequence(nextIndex); err != nil {
+			return fmt.Errorf("could not set next bucket "+
+				"sequence: %w", err)
+		}
 
 		// As a sanity check, assert that the index is still in the
 		// valid range of unhardened pubkeys. In the future, we should
 		// move to only using hardened keys, and this will prevent any
 		// overlap from occurring until then. This also prevents us from
 		// overflowing uint32s.
-		if index64 > math.MaxInt32 {
+		if nextIndex > math.MaxInt32 {
 			return fmt.Errorf("exhausted session key indexes")
 		}
 
 		// Create the key that will used to be store the reserved index.
 		keyBytes := createSessionKeyIndexKey(towerID, blobType)
 
-		index = uint32(index64)
+		index = uint32(nextIndex)
 
 		var indexBuf [4]byte
 		byteOrder.PutUint32(indexBuf[:], index)

watchtower/wtdb/client_db_test.go

@@ -62,12 +62,12 @@ func (h *clientDBHarness) listSessions(id *wtdb.TowerID,
 	return sessions
 }
 
-func (h *clientDBHarness) nextKeyIndex(id wtdb.TowerID,
-	blobType blob.Type) uint32 {
+func (h *clientDBHarness) nextKeyIndex(id wtdb.TowerID, blobType blob.Type,
+	forceNext bool) uint32 {
 
 	h.t.Helper()
 
-	index, err := h.db.NextSessionKeyIndex(id, blobType)
+	index, err := h.db.NextSessionKeyIndex(id, blobType, forceNext)
 	require.NoError(h.t, err, "unable to create next session key index")
 	require.NotZero(h.t, index, "next key index should never be 0")
 
@@ -309,7 +309,7 @@ func testCreateClientSession(h *clientDBHarness) {
 	h.insertSession(session, wtdb.ErrNoReservedKeyIndex)
 
 	// Now, reserve a session key for this tower.
-	keyIndex := h.nextKeyIndex(session.TowerID, blobType)
+	keyIndex := h.nextKeyIndex(session.TowerID, blobType, false)
 
 	// The client session hasn't been updated with the reserved key index
 	// (since it's still zero). Inserting should fail due to the mismatch.
@@ -318,7 +318,7 @@ func testCreateClientSession(h *clientDBHarness) {
 	// Reserve another key for the same index. Since no session has been
 	// successfully created, it should return the same index to maintain
 	// idempotency across restarts.
-	keyIndex2 := h.nextKeyIndex(session.TowerID, blobType)
+	keyIndex2 := h.nextKeyIndex(session.TowerID, blobType, false)
 	require.Equalf(h.t, keyIndex, keyIndex2, "next key index should "+
 		"be idempotent: want: %v, got %v", keyIndex, keyIndex2)
 
@@ -335,11 +335,23 @@ func testCreateClientSession(h *clientDBHarness) {
 	// session already existing.
 	h.insertSession(session, wtdb.ErrClientSessionAlreadyExists)
 
-	// Finally, assert that reserving another key index succeeds with a
-	// different key index, now that the first one has been finalized.
-	keyIndex3 := h.nextKeyIndex(session.TowerID, blobType)
+	// Assert that reserving another key index succeeds with a different key
+	// index, now that the first one has been finalized.
+	keyIndex3 := h.nextKeyIndex(session.TowerID, blobType, false)
 	require.NotEqualf(h.t, keyIndex, keyIndex3, "key index still "+
 		"reserved after creating session")
+
+	// Show that calling NextSessionKeyIndex again now will result in the
+	// same key being returned as long as forceNext remains false.
+	keyIndex4 := h.nextKeyIndex(session.TowerID, blobType, false)
+	require.Equal(h.t, keyIndex3, keyIndex4)
+
+	// Finally, assert that if the forceNext param of the
+	// NextSessionKeyIndex method is true, then the key index returned will
+	// be different.
+	keyIndex5 := h.nextKeyIndex(session.TowerID, blobType, true)
+	require.NotEqual(h.t, keyIndex5, keyIndex4)
+	require.Equal(h.t, keyIndex3+1000, keyIndex5)
 }
 
 // testFilterClientSessions asserts that we can correctly filter client sessions
@@ -352,7 +364,7 @@ func testFilterClientSessions(h *clientDBHarness) {
 	towerSessions := make(map[wtdb.TowerID][]wtdb.SessionID)
 	for i := 0; i < numSessions; i++ {
 		tower := h.newTower()
-		keyIndex := h.nextKeyIndex(tower.ID, blobType)
+		keyIndex := h.nextKeyIndex(tower.ID, blobType, false)
 		sessionID := wtdb.SessionID([33]byte{byte(i)})
 		h.insertSession(&wtdb.ClientSession{
 			ClientSessionBody: wtdb.ClientSessionBody{
@@ -499,7 +511,9 @@ func testRemoveTower(h *clientDBHarness) {
 				MaxUpdates: 100,
 			},
 			RewardPkScript: []byte{0x01, 0x02, 0x03},
-			KeyIndex:       h.nextKeyIndex(tower.ID, blobType),
+			KeyIndex: h.nextKeyIndex(
+				tower.ID, blobType, false,
+			),
 		},
 		ID: wtdb.SessionID([33]byte{0x01}),
 	}
@@ -581,7 +595,7 @@ func testCommitUpdate(h *clientDBHarness) {
 	)
 
 	// Reserve a session key index and insert the session.
-	session.KeyIndex = h.nextKeyIndex(session.TowerID, blobType)
+	session.KeyIndex = h.nextKeyIndex(session.TowerID, blobType, false)
 	h.insertSession(session, nil)
 
 	// Now, try to commit the update that failed initially which should
@@ -774,7 +788,7 @@ func testAckUpdate(h *clientDBHarness) {
 	h.ackUpdate(&session.ID, 1, 0, wtdb.ErrClientSessionNotFound)
 
 	// Reserve a session key and insert the client session.
-	session.KeyIndex = h.nextKeyIndex(session.TowerID, blobType)
+	session.KeyIndex = h.nextKeyIndex(session.TowerID, blobType, false)
 	h.insertSession(session, nil)
 
 	// Now, try to ack update 1. This should fail since update 1 was never
@@ -1051,7 +1065,9 @@ func (h *clientDBHarness) randSession(t *testing.T,
 				MaxUpdates: maxUpdates,
 			},
 			RewardPkScript: []byte{0x01, 0x02, 0x03},
-			KeyIndex:       h.nextKeyIndex(towerID, blobType),
+			KeyIndex: h.nextKeyIndex(
+				towerID, blobType, false,
+			),
 		},
 		ID: id,
 	}

watchtower/wtmock/client_db.go

@@ -401,9 +401,10 @@ func (m *ClientDB) CreateClientSession(session *wtdb.ClientSession) error {
 // particular tower id. The index is reserved for that tower until
 // CreateClientSession is invoked for that tower and index, at which point a new
 // index for that tower can be reserved. Multiple calls to this method before
-// CreateClientSession is invoked should return the same index.
-func (m *ClientDB) NextSessionKeyIndex(towerID wtdb.TowerID,
-	blobType blob.Type) (uint32, error) {
+// CreateClientSession is invoked should return the same index unless forceNext
+// is set to true.
+func (m *ClientDB) NextSessionKeyIndex(towerID wtdb.TowerID, blobType blob.Type,
+	forceNext bool) (uint32, error) {
 
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -413,15 +414,24 @@ func (m *ClientDB) NextSessionKeyIndex(towerID wtdb.TowerID,
 		blobType: blobType,
 	}
 
-	if index, err := m.getSessionKeyIndex(key); err == nil {
-		return index, nil
+	if !forceNext {
+		if index, err := m.getSessionKeyIndex(key); err == nil {
+			return index, nil
+		}
 	}
 
-	m.nextIndex++
-	index := m.nextIndex
-	m.indexes[key] = index
+	// By default, we use the next available bucket sequence as the key
+	// index. But if forceNext is true, then it is assumed that some data
+	// loss occurred and so the sequence is incremented a by a jump of 1000
+	// so that we can arrive at a brand new key index quicker.
+	nextIndex := m.nextIndex + 1
+	if forceNext {
+		nextIndex = m.nextIndex + 1000
+	}
+	m.nextIndex = nextIndex
+	m.indexes[key] = nextIndex
 
-	return index, nil
+	return nextIndex, nil
 }
 
 func (m *ClientDB) getSessionKeyIndex(key keyIndexKey) (uint32, error) {