lnd.xprv/watchtower/wtclient/session_queue.go

689 lines
21 KiB
Go
Raw Normal View History

package wtclient
import (
"container/list"
"fmt"
"sort"
"sync"
"time"
"github.com/btcsuite/btcd/btcec"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/lightningnetwork/lnd/input"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/watchtower/wtdb"
"github.com/lightningnetwork/lnd/watchtower/wtserver"
"github.com/lightningnetwork/lnd/watchtower/wtwire"
)
// retryInterval is the default duration we will wait between attempting to
// connect back out to a tower if the prior state update failed.
const retryInterval = 2 * time.Second
// reserveStatus is an enum that signals how full a particular session is.
type reserveStatus uint8
const (
// reserveAvailable indicates that the session has space for at least
// one more backup.
reserveAvailable reserveStatus = iota
// reserveExhausted indicates that all slots in the session have been
// allocated.
reserveExhausted
)
// sessionQueueConfig bundles the resources required by the sessionQueue to
// perform its duties. All entries MUST be non-nil.
type sessionQueueConfig struct {
// ClientSession provides access to the negotiated session parameters
// and updating its persistent storage.
ClientSession *wtdb.ClientSession
// ChainHash identifies the chain for which the session's justice
// transactions are targeted.
ChainHash chainhash.Hash
// Dial allows the client to dial the tower using it's public key and
// net address.
Dial func(*btcec.PrivateKey,
*lnwire.NetAddress) (wtserver.Peer, error)
// SendMessage encodes, encrypts, and writes a message to the given peer.
SendMessage func(wtserver.Peer, wtwire.Message) error
// ReadMessage receives, decypts, and decodes a message from the given
// peer.
ReadMessage func(wtserver.Peer) (wtwire.Message, error)
// Signer facilitates signing of inputs, used to construct the witnesses
// for justice transaction inputs.
Signer input.Signer
// DB provides access to the client's stable storage.
DB DB
// MinBackoff defines the initial backoff applied by the session
// queue before reconnecting to the tower after a failed or partially
// successful batch is sent. Subsequent backoff durations will grow
// exponentially up until MaxBackoff.
MinBackoff time.Duration
// MaxBackoff defines the maximum backoff applied by the session
// queue before reconnecting to the tower after a failed or partially
// successful batch is sent. If the exponential backoff produces a
// timeout greater than this value, the backoff duration will be clamped
// to MaxBackoff.
MaxBackoff time.Duration
}
// sessionQueue implements a reliable queue that will encrypt and send accepted
// backups to the watchtower specified in the config's ClientSession. Calling
// Quit will attempt to perform a clean shutdown by receiving an ACK from the
// tower for all pending backups before exiting. The clean shutdown can be
// aborted by using ForceQuit, which will attempt to shutdown the queue
// immediately.
type sessionQueue struct {
started sync.Once
stopped sync.Once
forced sync.Once
cfg *sessionQueueConfig
commitQueue *list.List
pendingQueue *list.List
queueMtx sync.Mutex
queueCond *sync.Cond
localInit *wtwire.Init
towerAddr *lnwire.NetAddress
seqNum uint16
retryBackoff time.Duration
quit chan struct{}
forceQuit chan struct{}
shutdown chan struct{}
}
// newSessionQueue intiializes a fresh sessionQueue.
func newSessionQueue(cfg *sessionQueueConfig) *sessionQueue {
localInit := wtwire.NewInitMessage(
lnwire.NewRawFeatureVector(wtwire.WtSessionsRequired),
cfg.ChainHash,
)
towerAddr := &lnwire.NetAddress{
IdentityKey: cfg.ClientSession.Tower.IdentityKey,
Address: cfg.ClientSession.Tower.Addresses[0],
}
sq := &sessionQueue{
cfg: cfg,
commitQueue: list.New(),
pendingQueue: list.New(),
localInit: localInit,
towerAddr: towerAddr,
seqNum: cfg.ClientSession.SeqNum,
retryBackoff: cfg.MinBackoff,
quit: make(chan struct{}),
forceQuit: make(chan struct{}),
shutdown: make(chan struct{}),
}
sq.queueCond = sync.NewCond(&sq.queueMtx)
sq.restoreCommittedUpdates()
return sq
}
// Start idempotently starts the sessionQueue so that it can begin accepting
// 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()
})
}
// Stop idempotently stops the sessionQueue by initiating a clean shutdown that
// will clear all pending tasks in the queue before returning to the caller.
func (q *sessionQueue) Stop() {
q.stopped.Do(func() {
log.Debugf("Stopping session queue %s", q.ID())
close(q.quit)
q.signalUntilShutdown()
// Skip log if we also force quit.
select {
case <-q.forceQuit:
return
default:
}
log.Debugf("Session queue %s successfully stopped", q.ID())
})
}
// ForceQuit idempotently aborts any clean shutdown in progress and returns to
// he caller after all lingering goroutines have spun down.
func (q *sessionQueue) ForceQuit() {
q.forced.Do(func() {
log.Infof("Force quitting session queue %s", q.ID())
close(q.forceQuit)
q.signalUntilShutdown()
log.Infof("Session queue %s unclean shutdown complete", q.ID())
})
}
// ID returns the wtdb.SessionID for the queue, which can be used to uniquely
// identify this a particular queue.
func (q *sessionQueue) ID() *wtdb.SessionID {
return &q.cfg.ClientSession.ID
}
// AcceptTask attempts to queue a backupTask for delivery to the sessionQueue's
// tower. The session will only be accepted if the queue is not already
// exhausted and the task is successfully bound to the ClientSession.
func (q *sessionQueue) AcceptTask(task *backupTask) (reserveStatus, bool) {
q.queueCond.L.Lock()
// Examine the current reserve status of the session queue.
curStatus := q.reserveStatus()
switch curStatus {
// The session queue is exhausted, and cannot accept the task because it
// is full. Reject the task such that it can be tried against a
// different session.
case reserveExhausted:
q.queueCond.L.Unlock()
return curStatus, false
// The session queue is not exhausted. Compute the sweep and reward
// outputs as a function of the session parameters. If the outputs are
// dusty or uneconomical to backup, the task is rejected and will not be
// tried again.
//
// TODO(conner): queue backups and retry with different session params.
case reserveAvailable:
err := task.bindSession(q.cfg.ClientSession)
if err != nil {
q.queueCond.L.Unlock()
log.Debugf("SessionQueue %s rejected backup chanid=%s "+
"commit-height=%d: %v", q.ID(), task.id.ChanID,
task.id.CommitHeight, err)
return curStatus, false
}
}
// The sweep and reward outputs satisfy the session's policy, queue the
// task for final signing and delivery.
q.pendingQueue.PushBack(task)
// Finally, compute the session's *new* reserve status. This will be
// used by the client to determine if it can continue using this session
// queue, or if it should negotiate a new one.
newStatus := q.reserveStatus()
q.queueCond.L.Unlock()
q.queueCond.Signal()
return newStatus, true
}
// updateWithSeqNum stores a CommittedUpdate with its assigned sequence number.
// This allows committed updates to be sorted after a restart, and added to the
// commitQueue in the proper order for delivery.
type updateWithSeqNum struct {
seqNum uint16
update *wtdb.CommittedUpdate
}
// restoreCommittedUpdates processes any CommittedUpdates loaded on startup by
// sorting them in ascending order of sequence numbers and adding them to the
// commitQueue. These will be sent before any pending updates are processed.
func (q *sessionQueue) restoreCommittedUpdates() {
committedUpdates := q.cfg.ClientSession.CommittedUpdates
// Construct and unordered slice of all committed updates with their
// assigned sequence numbers.
sortedUpdates := make([]updateWithSeqNum, 0, len(committedUpdates))
for seqNum, update := range committedUpdates {
sortedUpdates = append(sortedUpdates, updateWithSeqNum{
seqNum: seqNum,
update: update,
})
}
// Sort the resulting slice by increasing sequence number.
sort.Slice(sortedUpdates, func(i, j int) bool {
return sortedUpdates[i].seqNum < sortedUpdates[j].seqNum
})
// Finally, add the sorted, committed updates to he commitQueue. These
// updates will be prioritized before any new tasks are assigned to the
// sessionQueue. The queue will begin uploading any tasks in the
// commitQueue as soon as it is started, e.g. during client
// initialization when detecting that this session has unacked updates.
for _, update := range sortedUpdates {
q.commitQueue.PushBack(update)
}
}
// sessionManager is the primary event loop for the sessionQueue, and is
// responsible for encrypting and sending accepted tasks to the tower.
func (q *sessionQueue) sessionManager() {
defer close(q.shutdown)
for {
q.queueCond.L.Lock()
for q.commitQueue.Len() == 0 &&
q.pendingQueue.Len() == 0 {
q.queueCond.Wait()
select {
case <-q.quit:
if q.commitQueue.Len() == 0 &&
q.pendingQueue.Len() == 0 {
q.queueCond.L.Unlock()
return
}
case <-q.forceQuit:
q.queueCond.L.Unlock()
return
default:
}
}
q.queueCond.L.Unlock()
// Exit immediately if a force quit has been requested. If the
// either of the queues still has state updates to send to the
// tower, we may never exit in the above case if we are unable
// to reach the tower for some reason.
select {
case <-q.forceQuit:
return
default:
}
// Initiate a new connection to the watchtower and attempt to
// drain all pending tasks.
q.drainBackups()
}
}
// drainBackups attempts to send all pending updates in the queue to the tower.
func (q *sessionQueue) drainBackups() {
// First, check that we are able to dial this session's tower.
conn, err := q.cfg.Dial(q.cfg.ClientSession.SessionPrivKey, q.towerAddr)
if err != nil {
log.Errorf("Unable to dial watchtower at %v: %v",
q.towerAddr, err)
q.increaseBackoff()
select {
case <-time.After(q.retryBackoff):
case <-q.forceQuit:
}
return
}
defer conn.Close()
// Begin draining the queue of pending state updates. Before the first
// update is sent, we will precede it with an Init message. If the first
// is successful, subsequent updates can be streamed without sending an
// Init.
for sendInit := true; ; sendInit = false {
// Generate the next state update to upload to the tower. This
// method will first proceed in dequeueing committed updates
// before attempting to dequeue any pending updates.
stateUpdate, isPending, err := q.nextStateUpdate()
if err != nil {
log.Errorf("Unable to get next state update: %v", err)
return
}
// Now, send the state update to the tower and wait for a reply.
err = q.sendStateUpdate(
conn, stateUpdate, q.localInit, sendInit, isPending,
)
if err != nil {
log.Errorf("Unable to send state update: %v", err)
q.increaseBackoff()
select {
case <-time.After(q.retryBackoff):
case <-q.forceQuit:
}
return
}
// If the last task was backed up successfully, we'll exit and
// continue once more tasks are added to the queue. We'll also
// clear any accumulated backoff as this batch was able to be
// sent reliably.
if stateUpdate.IsComplete == 1 {
q.resetBackoff()
return
}
// Always apply a small delay between sends, which makes the
// unit tests more reliable. If we were requested to back off,
// when we will do so.
select {
case <-time.After(time.Millisecond):
case <-q.forceQuit:
return
}
}
}
// nextStateUpdate returns the next wtwire.StateUpdate to upload to the tower.
// If any committed updates are present, this method will reconstruct the state
// update from the committed update using the current last applied value found
// in the database. Otherwise, it will select the next pending update, craft the
// payload, and commit an update before returning the state update to send. The
// boolean value in the response is true if the state update is taken from the
// pending queue, allowing the caller to remove the update from either the
// commit or pending queue if the update is successfully acked.
func (q *sessionQueue) nextStateUpdate() (*wtwire.StateUpdate, bool, error) {
var (
seqNum uint16
update *wtdb.CommittedUpdate
isLast bool
isPending bool
)
q.queueCond.L.Lock()
switch {
// If the commit queue is non-empty, parse the next committed update.
case q.commitQueue.Len() > 0:
next := q.commitQueue.Front()
updateWithSeq := next.Value.(updateWithSeqNum)
seqNum = updateWithSeq.seqNum
update = updateWithSeq.update
// If this is the last item in the commit queue and no items
// exist in the pending queue, we will use the IsComplete flag
// in the StateUpdate to signal that the tower can release the
// connection after replying to free up resources.
isLast = q.commitQueue.Len() == 1 && q.pendingQueue.Len() == 0
q.queueCond.L.Unlock()
log.Debugf("Reprocessing committed state update for "+
"session=%s seqnum=%d", q.ID(), seqNum)
// Otherwise, craft and commit the next update from the pending queue.
default:
isPending = true
// Determine the current sequence number to apply for this
// pending update.
seqNum = q.seqNum + 1
// Obtain the next task from the queue.
next := q.pendingQueue.Front()
task := next.Value.(*backupTask)
// If this is the last item in the pending queue, we will use
// the IsComplete flag in the StateUpdate to signal that the
// tower can release the connection after replying to free up
// resources.
isLast = q.pendingQueue.Len() == 1
q.queueCond.L.Unlock()
hint, encBlob, err := task.craftSessionPayload(q.cfg.Signer)
if err != nil {
// TODO(conner): mark will not send
return nil, false, fmt.Errorf("unable to craft "+
"session payload: %v", err)
}
// TODO(conner): special case other obscure errors
update = &wtdb.CommittedUpdate{
BackupID: task.id,
Hint: hint,
EncryptedBlob: encBlob,
}
log.Debugf("Committing state update for session=%s seqnum=%d",
q.ID(), seqNum)
}
// Before sending the task to the tower, commit the state update
// to disk using the assigned sequence number. If this task has already
// been committed, the call will succeed and only be used for the
// purpose of obtaining the last applied value to send to the tower.
//
// This step ensures that if we crash before receiving an ack that we
// will retransmit the same update. If the tower successfully received
// the update from before, it will reply with an ACK regardless of what
// we send the next time. This step ensures that if we reliably send the
// same update for a given sequence number, to prevent us from thinking
// we backed up a state when we instead backed up another.
lastApplied, err := q.cfg.DB.CommitUpdate(q.ID(), seqNum, update)
if err != nil {
// TODO(conner): mark failed/reschedule
return nil, false, fmt.Errorf("unable to commit state update "+
"for session=%s seqnum=%d: %v", q.ID(), seqNum, err)
}
stateUpdate := &wtwire.StateUpdate{
SeqNum: seqNum,
LastApplied: lastApplied,
Hint: update.Hint,
EncryptedBlob: update.EncryptedBlob,
}
// Set the IsComplete flag if this is the last queued item.
if isLast {
stateUpdate.IsComplete = 1
}
return stateUpdate, isPending, nil
}
// sendStateUpdate sends a wtwire.StateUpdate to the watchtower and processes
// the ACK before returning. If sendInit is true, this method will first send
// the localInit message and verify that the tower supports our required feature
// bits. And error is returned if any part of the send fails. The boolean return
// variable indicates whether or not we should back off before attempting to
// send the next state update.
func (q *sessionQueue) sendStateUpdate(conn wtserver.Peer,
stateUpdate *wtwire.StateUpdate, localInit *wtwire.Init,
sendInit, isPending bool) error {
// If this is the first message being sent to the tower, we must send an
// Init message to establish that server supports the features we
// require.
if sendInit {
// Send Init to tower.
err := q.cfg.SendMessage(conn, q.localInit)
if err != nil {
return err
}
// Receive Init from tower.
remoteMsg, err := q.cfg.ReadMessage(conn)
if err != nil {
return err
}
remoteInit, ok := remoteMsg.(*wtwire.Init)
if !ok {
return fmt.Errorf("watchtower responded with %T to "+
"Init", remoteMsg)
}
// Validate Init.
err = q.localInit.CheckRemoteInit(
remoteInit, wtwire.FeatureNames,
)
if err != nil {
return err
}
}
// Send StateUpdate to tower.
err := q.cfg.SendMessage(conn, stateUpdate)
if err != nil {
return err
}
// Receive StateUpdate from tower.
remoteMsg, err := q.cfg.ReadMessage(conn)
if err != nil {
return err
}
stateUpdateReply, ok := remoteMsg.(*wtwire.StateUpdateReply)
if !ok {
return fmt.Errorf("watchtower responded with %T to StateUpdate",
remoteMsg)
}
// Process the reply from the tower.
switch stateUpdateReply.Code {
// The tower reported a successful update, validate the response and
// record the last applied returned.
case wtwire.CodeOK:
// TODO(conner): handle other error cases properly, ban towers, etc.
default:
err := fmt.Errorf("received error code %s in "+
"StateUpdateReply from tower=%x session=%s",
stateUpdateReply.Code,
conn.RemotePub().SerializeCompressed(), q.ID())
log.Warnf("Unable to upload state update: %v", err)
return err
}
lastApplied := stateUpdateReply.LastApplied
err = q.cfg.DB.AckUpdate(q.ID(), stateUpdate.SeqNum, lastApplied)
switch {
case err == wtdb.ErrUnallocatedLastApplied:
// TODO(conner): borked watchtower
err = fmt.Errorf("unable to ack update=%d session=%s: %v",
stateUpdate.SeqNum, q.ID(), err)
log.Errorf("Failed to ack update: %v", err)
return err
case err == wtdb.ErrLastAppliedReversion:
// TODO(conner): borked watchtower
err = fmt.Errorf("unable to ack update=%d session=%s: %v",
stateUpdate.SeqNum, q.ID(), err)
log.Errorf("Failed to ack update: %v", err)
return err
case err != nil:
err = fmt.Errorf("unable to ack update=%d session=%s: %v",
stateUpdate.SeqNum, q.ID(), err)
log.Errorf("Failed to ack update: %v", err)
return err
}
log.Infof("Removing update session=%s seqnum=%d is_pending=%v "+
"from memory", q.ID(), stateUpdate.SeqNum, isPending)
q.queueCond.L.Lock()
if isPending {
// If a pending update was successfully sent, increment the
// sequence number and remove the item from the queue. This
// ensures the total number of backups in the session remains
// unchanged, which maintains the external view of the session's
// reserve status.
q.seqNum++
q.pendingQueue.Remove(q.pendingQueue.Front())
} else {
// Otherwise, simply remove the update from the committed queue.
// This has no effect on the queues reserve status since the
// update had already been committed.
q.commitQueue.Remove(q.commitQueue.Front())
}
q.queueCond.L.Unlock()
return nil
}
// reserveStatus returns a reserveStatus indicating whether or not the
// sessionQueue can accept another task. reserveAvailable is returned when a
// task can be accepted, and reserveExhausted is returned if the all slots in
// the session have been allocated.
//
// NOTE: This method MUST be called with queueCond's exclusive lock held.
func (q *sessionQueue) reserveStatus() reserveStatus {
numPending := uint32(q.pendingQueue.Len())
maxUpdates := uint32(q.cfg.ClientSession.Policy.MaxUpdates)
log.Debugf("SessionQueue %s reserveStatus seqnum=%d pending=%d "+
"max-updates=%d", q.ID(), q.seqNum, numPending, maxUpdates)
if uint32(q.seqNum)+numPending < maxUpdates {
return reserveAvailable
}
return reserveExhausted
}
// resetBackoff returns the connection backoff the minimum configured backoff.
func (q *sessionQueue) resetBackoff() {
q.retryBackoff = q.cfg.MinBackoff
}
// increaseBackoff doubles the current connection backoff, clamping to the
// configured maximum backoff if it would exceed the limit.
func (q *sessionQueue) increaseBackoff() {
q.retryBackoff *= 2
if q.retryBackoff > q.cfg.MaxBackoff {
q.retryBackoff = q.cfg.MaxBackoff
}
}
// signalUntilShutdown strobes the sessionQueue's condition variable until the
// main event loop exits.
func (q *sessionQueue) signalUntilShutdown() {
for {
select {
case <-time.After(time.Millisecond):
q.queueCond.Signal()
case <-q.shutdown:
return
}
}
}
// sessionQueueSet maintains a mapping of SessionIDs to their corresponding
// sessionQueue.
type sessionQueueSet map[wtdb.SessionID]*sessionQueue
// Add inserts a sessionQueue into the sessionQueueSet.
func (s *sessionQueueSet) Add(sessionQueue *sessionQueue) {
(*s)[*sessionQueue.ID()] = sessionQueue
}
// ApplyAndWait executes the nil-adic function returned from getApply for each
// sessionQueue in the set in parallel, then waits for all of them to finish
// before returning to the caller.
func (s *sessionQueueSet) ApplyAndWait(getApply func(*sessionQueue) func()) {
var wg sync.WaitGroup
for _, sessionq := range *s {
wg.Add(1)
go func(sq *sessionQueue) {
defer wg.Done()
getApply(sq)()
}(sessionq)
}
wg.Wait()
}