From 204585c620ea940ddb99ca49f0f1a088d851ec1f Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 27 Mar 2023 14:14:26 +0200 Subject: [PATCH 1/4] watchtower: demo StateUpdateCodeClientBehind error In this commit, a test is added to demonstrate how clients can end up getting the StateUpdateCodeClientBehind error from a tower server. This can happen if a client ever deletes their db. If they do this then the sessions they create with the tower will have the same IDs as the sessions created in the now deleted db. This is because the session keys (and thus session IDs) are calculated deterministically from a counter (which is reset if the db is deleted). The tower server then throws this error because the client would say that the sequence ID is 1 for the next update. --- watchtower/wtclient/client_test.go | 59 +++++++++++++++++++++++++++++- 1 file changed, 57 insertions(+), 2 deletions(-) diff --git a/watchtower/wtclient/client_test.go b/watchtower/wtclient/client_test.go index f870af527..dc1d87842 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,61 @@ var clientTests = []clientTest{ require.NoError(h.t, err) }, }, + { + // Demonstrate that the client is unable to upload state updates + // to a tower if the client deletes its database after already + // having created and started to use a session with a tower. + // This happens because the session key is generated + // deterministically and will only be unique for new sessions + // if the same DB is used. The server therefore rejects these + // updates with the StateUpdateCodeClientBehind error. + name: "demonstrate the StateUpdateCodeClientBehind error", + 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 not get the remaining + // updates. + h.waitServerUpdates(nil, waitTime) + }, + }, } // TestClient executes the client test suite, asserting the ability to backup From 24016c35c7f011ec8aa31376175b202a8ebf7343 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 29 Mar 2023 12:15:17 +0200 Subject: [PATCH 2/4] watchtower: add forceNext to NextSessionKeyIndex This commit adds a forceNext boolean parameter to NextSessionKeyIndex. Setting this param to true will force the index to cycle over 1000 key indices before returning the new key. --- watchtower/wtclient/interface.go | 4 +-- watchtower/wtclient/session_negotiator.go | 2 +- watchtower/wtdb/client_db.go | 44 +++++++++++++++-------- watchtower/wtdb/client_db_test.go | 42 +++++++++++++++------- watchtower/wtmock/client_db.go | 28 ++++++++++----- 5 files changed, 81 insertions(+), 39 deletions(-) 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..aafc44b2a 100644 --- a/watchtower/wtclient/session_negotiator.go +++ b/watchtower/wtclient/session_negotiator.go @@ -306,7 +306,7 @@ retryWithBackoff: // 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, + tower.ID, n.cfg.Policy.BlobType, false, ) if err != nil { n.log.Debugf("Unable to reserve session key index "+ 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) { From ba33f6a697697f397a48ed87de0b6e1d595c7013 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 29 Mar 2023 12:40:25 +0200 Subject: [PATCH 3/4] wtclient: spin session key indices if required If the tower returns CreateSessionCodeAlreadyExists in response to the CreateSession message from the client, then skip forward a few key indices until we find one that the server does not return the error for. This will allow a client to recover after a data loss incident. --- watchtower/wtclient/client_test.go | 19 +++--- watchtower/wtclient/errors.go | 5 ++ watchtower/wtclient/session_negotiator.go | 71 +++++++++++++++-------- 3 files changed, 59 insertions(+), 36 deletions(-) diff --git a/watchtower/wtclient/client_test.go b/watchtower/wtclient/client_test.go index dc1d87842..7be616d77 100644 --- a/watchtower/wtclient/client_test.go +++ b/watchtower/wtclient/client_test.go @@ -2014,14 +2014,12 @@ var clientTests = []clientTest{ }, }, { - // Demonstrate that the client is unable to upload state updates - // to a tower if the client deletes its database after already - // having created and started to use a session with a tower. - // This happens because the session key is generated - // deterministically and will only be unique for new sessions - // if the same DB is used. The server therefore rejects these - // updates with the StateUpdateCodeClientBehind error. - name: "demonstrate the StateUpdateCodeClientBehind error", + // 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, @@ -2063,9 +2061,8 @@ var clientTests = []clientTest{ // Attempt to back up the remaining tasks. h.backupStates(chanID, numUpdates/2, numUpdates, nil) - // Show that the server does not get the remaining - // updates. - h.waitServerUpdates(nil, waitTime) + // Show that the server does get the remaining updates. + h.waitServerUpdates(hints[numUpdates/2:], waitTime) }, }, } 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/session_negotiator.go b/watchtower/wtclient/session_negotiator.go index aafc44b2a..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, false, - ) - 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", From 381dd8d46055a1231e2775366cd95e8a9e315732 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 27 Mar 2023 14:21:52 +0200 Subject: [PATCH 4/4] docs: add release note for 7541 --- docs/release-notes/release-notes-0.16.1.md | 2 ++ 1 file changed, 2 insertions(+) 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