diff --git a/docs/release-notes/release-notes-0.16.1.md b/docs/release-notes/release-notes-0.16.1.md index ebe476295..3cd4890d5 100644 --- a/docs/release-notes/release-notes-0.16.1.md +++ b/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 diff --git a/watchtower/wtclient/client_test.go b/watchtower/wtclient/client_test.go index f870af527..7be616d77 100644 --- a/watchtower/wtclient/client_test.go +++ b/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 diff --git a/watchtower/wtclient/errors.go b/watchtower/wtclient/errors.go index c6884bb35..2cb070ac4 100644 --- a/watchtower/wtclient/errors.go +++ b/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") ) diff --git a/watchtower/wtclient/interface.go b/watchtower/wtclient/interface.go index 8a6d9b320..06126ae21 100644 --- a/watchtower/wtclient/interface.go +++ b/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 diff --git a/watchtower/wtclient/session_negotiator.go b/watchtower/wtclient/session_negotiator.go index db0f543a1..d137d45e1 100644 --- a/watchtower/wtclient/session_negotiator.go +++ b/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", diff --git a/watchtower/wtdb/client_db.go b/watchtower/wtdb/client_db.go index 5a18095b8..a949021d1 100644 --- a/watchtower/wtdb/client_db.go +++ b/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) diff --git a/watchtower/wtdb/client_db_test.go b/watchtower/wtdb/client_db_test.go index 26d02bcf3..80557f1f8 100644 --- a/watchtower/wtdb/client_db_test.go +++ b/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, } diff --git a/watchtower/wtmock/client_db.go b/watchtower/wtmock/client_db.go index 8d43a0e45..5b77da744 100644 --- a/watchtower/wtmock/client_db.go +++ b/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) {