13157733617

Committed 05 Feb 2025 12:49PM UTC coverage: 57.712% (-1.1%) from 58.82%

Build # 13157733617

Build Type

Pull #9447

github

Committed by

yyforyongyu

Commit Message

sweep: rename methods for clarity

We now rename "third party" to "unknown" as the inputs can be spent via
an older sweeping tx, a third party (anchor), or a remote party (pin).
In fee bumper we don't have the info to distinguish the above cases, and
leave them to be further handled by the sweeper as it has more context.

Pull Request Pull Request #9447: sweep: start tracking input spending status in the fee bumper

Run Details

83 of 87 new or added lines in 2 files covered. (95.4%)

19472 existing lines in 252 files now uncovered.

103634 of 179570 relevant lines covered (57.71%)

24840.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.96

/watchtower/wtclient/client.go

package wtclient

import (
        "bytes"
        "crypto/rand"
        "errors"
        "fmt"
        "math/big"
        "net"
        "sync"
        "time"

        "github.com/btcsuite/btcd/btcec/v2"
        "github.com/btcsuite/btclog/v2"
        "github.com/lightningnetwork/lnd/channeldb"
        "github.com/lightningnetwork/lnd/keychain"
        "github.com/lightningnetwork/lnd/lnwallet"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/watchtower/wtdb"
        "github.com/lightningnetwork/lnd/watchtower/wtpolicy"
        "github.com/lightningnetwork/lnd/watchtower/wtserver"
        "github.com/lightningnetwork/lnd/watchtower/wtwire"
)

const (
        // DefaultReadTimeout specifies the default duration we will wait during
        // a read before breaking out of a blocking read.
        DefaultReadTimeout = 15 * time.Second

        // DefaultWriteTimeout specifies the default duration we will wait
        // during a write before breaking out of a blocking write.
        DefaultWriteTimeout = 15 * time.Second

        // DefaultStatInterval specifies the default interval between logging
        // metrics about the client's operation.
        DefaultStatInterval = time.Minute

        // DefaultSessionCloseRange is the range over which we will generate a
        // random number of blocks to delay closing a session after its last
        // channel has been closed.
        DefaultSessionCloseRange = 288

        // DefaultMaxTasksInMemQueue is the maximum number of items to be held
        // in the in-memory queue.
        DefaultMaxTasksInMemQueue = 2000
)

// genSessionFilter constructs a filter that can be used to select sessions only
// if they match the policy of the client (namely anchor vs legacy). If
// activeOnly is set, then only active sessions will be returned.
func (c *client) genSessionFilter(
        activeOnly bool) wtdb.ClientSessionFilterFn {

        return func(session *wtdb.ClientSession) bool {
                if c.cfg.Policy.TxPolicy != session.Policy.TxPolicy {
                        return false
                }

                if !activeOnly {
                        return true
                }

                return session.Status == wtdb.CSessionActive
        }
}

// ExhaustedSessionFilter constructs a wtdb.ClientSessionFilterFn filter
// function that will filter out any sessions that have been exhausted. A
// session is considered exhausted only if it has no un-acked updates and the
// sequence number of the session is equal to the max updates of the session
// policy.
func ExhaustedSessionFilter() wtdb.ClientSessWithNumCommittedUpdatesFilterFn {
        return func(session *wtdb.ClientSession, numUnAcked uint16) bool {
                return session.SeqNum < session.Policy.MaxUpdates ||
                        numUnAcked > 0
        }
}

// RegisteredTower encompasses information about a registered watchtower with
// the client.
type RegisteredTower struct {
        *wtdb.Tower

        // Sessions is the set of sessions corresponding to the watchtower.
        Sessions map[wtdb.SessionID]*wtdb.ClientSession

        // ActiveSessionCandidate determines whether the watchtower is currently
        // being considered for new sessions.
        ActiveSessionCandidate bool
}

// BreachRetributionBuilder is a function that can be used to construct a
// BreachRetribution from a channel ID and a commitment height.
type BreachRetributionBuilder func(id lnwire.ChannelID,
        commitHeight uint64) (*lnwallet.BreachRetribution,
        channeldb.ChannelType, error)

// newTowerMsg is an internal message we'll use within the client to signal
// that a new tower can be considered.
type newTowerMsg struct {
        // tower holds the info about the new Tower or new tower address
        // required to connect to it.
        tower *Tower

        // errChan is the channel through which we'll send a response back to
        // the caller when handling their request.
        //
        // NOTE: This channel must be buffered.
        errChan chan error
}

// staleTowerMsg is an internal message we'll use within the client to
// signal that a tower should no longer be considered.
type staleTowerMsg struct {
        // id is the unique database identifier for the tower.
        id wtdb.TowerID

        // pubKey is the identifying public key of the watchtower.
        pubKey *btcec.PublicKey

        // addr is an optional field that when set signals that the address
        // should be removed from the watchtower's set of addresses, indicating
        // that it is stale. If it's not set, then the watchtower should be
        // no longer be considered for new sessions.
        addr net.Addr

        // errChan is the channel through which we'll send a response back to
        // the caller when handling their request.
        //
        // NOTE: This channel must be buffered.
        errChan chan error
}

// deactivateTowerMsg is an internal message we'll use within the TowerClient
// to signal that a tower should be marked as inactive.
type deactivateTowerMsg struct {
        // id is the unique database identifier for the tower.
        id wtdb.TowerID

        // pubKey is the identifying public key of the watchtower.
        pubKey *btcec.PublicKey

        // errChan is the channel through which we'll send a response back to
        // the caller when handling their request.
        //
        // NOTE: This channel must be buffered.
        errChan chan error
}

// terminateSessMsg is an internal message we'll use within the TowerClient to
// signal that a session should be terminated.
type terminateSessMsg struct {
        // id is the session identifier.
        id wtdb.SessionID

        // errChan is the channel through which we'll send a response back to
        // the caller when handling their request.
        //
        // NOTE: This channel must be buffered.
        errChan chan error
}

// clientCfg holds the configuration values required by a client.
type clientCfg struct {
        *Config

        // Policy is the session policy the client will propose when creating
        // new sessions with the tower. If the policy differs from any active
        // sessions recorded in the database, those sessions will be ignored and
        // new sessions will be requested immediately.
        Policy wtpolicy.Policy

        getSweepScript func(lnwire.ChannelID) ([]byte, bool)
}

// client manages backing up revoked states for all states that fall under a
// specific policy type.
type client struct {
        cfg *clientCfg

        log btclog.Logger

        pipeline *DiskOverflowQueue[*wtdb.BackupID]

        negotiator        SessionNegotiator
        candidateTowers   TowerCandidateIterator
        candidateSessions map[wtdb.SessionID]*ClientSession
        activeSessions    *sessionQueueSet

        sessionQueue *sessionQueue
        prevTask     *wtdb.BackupID

        statTicker *time.Ticker
        stats      *clientStats

        newTowers         chan *newTowerMsg
        staleTowers       chan *staleTowerMsg
        deactivateTowers  chan *deactivateTowerMsg
        terminateSessions chan *terminateSessMsg

        wg   sync.WaitGroup
        quit chan struct{}
}

// newClient initializes a new client from the provided clientCfg. An error is
// returned if the client could not be initialized.
func newClient(cfg *clientCfg) (*client, error) {
        identifier, err := cfg.Policy.BlobType.Identifier()
        if err != nil {
                return nil, err
        }
        plog := log.WithPrefix(fmt.Sprintf("(%s)", identifier))

        queueDB := cfg.DB.GetDBQueue([]byte(identifier))
        queue, err := NewDiskOverflowQueue[*wtdb.BackupID](
                queueDB, cfg.MaxTasksInMemQueue, plog,
        )
        if err != nil {
                return nil, err
        }

        c := &client{
                cfg:               cfg,
                log:               plog,
                pipeline:          queue,
                activeSessions:    newSessionQueueSet(),
                statTicker:        time.NewTicker(DefaultStatInterval),
                stats:             new(clientStats),
                newTowers:         make(chan *newTowerMsg),
                staleTowers:       make(chan *staleTowerMsg),
                deactivateTowers:  make(chan *deactivateTowerMsg),
                terminateSessions: make(chan *terminateSessMsg),
                quit:              make(chan struct{}),
        }

        candidateTowers := newTowerListIterator()
        perActiveTower := func(tower *Tower) {
                // If the tower has already been marked as active, then there is
                // no need to add it to the iterator again.
                if candidateTowers.IsActive(tower.ID) {
                        return
                }

                c.log.Infof("Using private watchtower %x, offering policy %s",
                        tower.IdentityKey.SerializeCompressed(), cfg.Policy)

                // Add the tower to the set of candidate towers.
                candidateTowers.AddCandidate(tower)
        }

        // Load all candidate sessions and towers from the database into the
        // client. We will use any of these sessions if their policies match the
        // current policy of the client, otherwise they will be ignored and new
        // sessions will be requested.
        candidateSessions, err := getTowerAndSessionCandidates(
                cfg.DB, cfg.SecretKeyRing, perActiveTower,
                wtdb.WithPreEvalFilterFn(c.genSessionFilter(true)),
                wtdb.WithPostEvalFilterFn(ExhaustedSessionFilter()),
        )
        if err != nil {
                return nil, err
        }

        c.candidateTowers = candidateTowers
        c.candidateSessions = candidateSessions

        c.negotiator = newSessionNegotiator(&NegotiatorConfig{
                DB:            cfg.DB,
                SecretKeyRing: cfg.SecretKeyRing,
                Policy:        cfg.Policy,
                ChainHash:     cfg.ChainHash,
                SendMessage:   c.sendMessage,
                ReadMessage:   c.readMessage,
                Dial:          c.dial,
                Candidates:    c.candidateTowers,
                MinBackoff:    cfg.MinBackoff,
                MaxBackoff:    cfg.MaxBackoff,
                Log:           plog,
        })

        return c, nil
}

// getTowerAndSessionCandidates loads all the towers from the DB and then
// fetches the sessions for each of tower. Sessions are only collected if they
// pass the sessionFilter check. If a tower has a session that does pass the
// sessionFilter check then the perActiveTower call-back will be called on that
// tower.
func getTowerAndSessionCandidates(db DB, keyRing ECDHKeyRing,
        perActiveTower func(tower *Tower),
        opts ...wtdb.ClientSessionListOption) (
        map[wtdb.SessionID]*ClientSession, error) {

        // Fetch all active towers from the DB.
        towers, err := db.ListTowers(func(tower *wtdb.Tower) bool {
                return tower.Status == wtdb.TowerStatusActive
        })
        if err != nil {
                return nil, err
        }

        candidateSessions := make(map[wtdb.SessionID]*ClientSession)
        for _, dbTower := range towers {
                tower, err := NewTowerFromDBTower(dbTower)
                if err != nil {
                        return nil, err
                }

                sessions, err := db.ListClientSessions(&tower.ID, opts...)
                if err != nil {
                        return nil, err
                }

                for _, s := range sessions {
                        cs, err := NewClientSessionFromDBSession(
                                s, tower, keyRing,
                        )
                        if err != nil {
                                return nil, err
                        }

                        // Add the session to the set of candidate sessions.
                        candidateSessions[s.ID] = cs
                }

                perActiveTower(tower)
        }

        return candidateSessions, nil
}

// getClientSessions retrieves the client sessions for a particular tower if
// specified, otherwise all client sessions for all towers are retrieved. An
// optional filter can be provided to filter out any undesired client sessions.
//
// NOTE: This method should only be used when deserialization of a
// ClientSession's SessionPrivKey field is desired, otherwise, the existing
// ListClientSessions method should be used.
func getClientSessions(db DB, keyRing ECDHKeyRing, forTower *wtdb.TowerID,
        opts ...wtdb.ClientSessionListOption) (
        map[wtdb.SessionID]*ClientSession, error) {

        dbSessions, err := db.ListClientSessions(forTower, opts...)
        if err != nil {
                return nil, err
        }

        // Reload the tower from disk using the tower ID contained in each
        // candidate session. We will also rederive any session keys needed to
        // be able to communicate with the towers and authenticate session
        // requests. This prevents us from having to store the private keys on
        // disk.
        sessions := make(map[wtdb.SessionID]*ClientSession)
        for _, s := range dbSessions {
                dbTower, err := db.LoadTowerByID(s.TowerID)
                if err != nil {
                        return nil, err
                }

                towerKeyDesc, err := keyRing.DeriveKey(keychain.KeyLocator{
                        Family: keychain.KeyFamilyTowerSession,
                        Index:  s.KeyIndex,
                })
                if err != nil {
                        return nil, err
                }

                sessionKeyECDH := keychain.NewPubKeyECDH(towerKeyDesc, keyRing)

                tower, err := NewTowerFromDBTower(dbTower)
                if err != nil {
                        return nil, err
                }

                sessions[s.ID] = &ClientSession{
                        ID:                s.ID,
                        ClientSessionBody: s.ClientSessionBody,
                        Tower:             tower,
                        SessionKeyECDH:    sessionKeyECDH,
                }
        }

        return sessions, nil
}

// start initializes the watchtower client by loading or negotiating an active
// session and then begins processing backup tasks from the request pipeline.
func (c *client) start() error {
        c.log.Infof("Watchtower client starting")

        // First, restart a session queue for any sessions that have
        // committed but unacked state updates. This ensures that these
        // sessions will be able to flush the committed updates after a
        // restart.
        fetchCommittedUpdates := c.cfg.DB.FetchSessionCommittedUpdates
        for _, session := range c.candidateSessions {
                committedUpdates, err := fetchCommittedUpdates(
                        &session.ID,
                )
                if err != nil {
                        return err
                }

                if len(committedUpdates) > 0 {
                        c.log.Infof("Starting session=%s to process "+
                                "%d committed backups", session.ID,
                                len(committedUpdates))

                        c.initActiveQueue(session, committedUpdates)
                }
        }

        // Now start the session negotiator, which will allow us to request new
        // session as soon as the backupDispatcher starts up.
        err := c.negotiator.Start()
        if err != nil {
                return err
        }

        // Start the task pipeline to which new backup tasks will be
        // submitted from active links.
        err = c.pipeline.Start()
        if err != nil {
                return err
        }

        c.wg.Add(1)
        go c.backupDispatcher()

        c.log.Infof("Watchtower client started successfully")

        return nil
}

// stop idempotently initiates a graceful shutdown of the watchtower client.
func (c *client) stop() error {
        var returnErr error
        c.log.Debugf("Stopping watchtower client")

        // 1. Stop the session negotiator.
        err := c.negotiator.Stop()
        if err != nil {
                returnErr = err
        }

        // 2. Stop the backup dispatcher and any other goroutines.
        close(c.quit)
        c.wg.Wait()

        // 3. If there was a left over 'prevTask' from the backup
        // dispatcher, replay that onto the pipeline.
        if c.prevTask != nil {
                err = c.pipeline.QueueBackupID(c.prevTask)
                if err != nil {
                        returnErr = err
                }
        }

        // 4. Shutdown all active session queues in parallel. These will
        // exit once all unhandled updates have been replayed to the
        // task pipeline.
        c.activeSessions.ApplyAndWait(func(s *sessionQueue) func() {
                return func() {
                        err := s.Stop(false)
                        if err != nil {
                                c.log.Errorf("could not stop session "+
                                        "queue: %s: %v", s.ID(), err)

                                returnErr = err
                        }
                }
        })

        // 5. Shutdown the backup queue, which will prevent any further
        // updates from being accepted.
        if err = c.pipeline.Stop(); err != nil {
                returnErr = err
        }

        c.log.Debugf("Client successfully stopped, stats: %s", c.stats)

        return returnErr
}

// backupState initiates a request to back up a particular revoked state. If the
// method returns nil, the backup is guaranteed to be successful unless the:
//   - justice transaction would create dust outputs when trying to abide by the
//     negotiated policy, or
//   - breached outputs contain too little value to sweep at the target sweep
//     fee rate.
func (c *client) backupState(chanID *lnwire.ChannelID,
        stateNum uint64) error {

        id := &wtdb.BackupID{
                ChanID:       *chanID,
                CommitHeight: stateNum,
        }

        return c.pipeline.QueueBackupID(id)
}

// nextSessionQueue attempts to fetch an active session from our set of
// candidate sessions. Candidate sessions with a differing policy from the
// active client's advertised policy will be ignored, but may be resumed if the
// client is restarted with a matching policy. If no candidates were found, nil
// is returned to signal that we need to request a new policy.
func (c *client) nextSessionQueue() (*sessionQueue, error) {
        // Select any candidate session at random, and remove it from the set of
        // candidate sessions.
        var candidateSession *ClientSession
        for id, sessionInfo := range c.candidateSessions {
                delete(c.candidateSessions, id)

                // Skip any sessions with policies that don't match the current
                // TxPolicy, as they would result in different justice
                // transactions from what is requested. These can be used again
                // if the client changes their configuration and restarting.
                if sessionInfo.Policy.TxPolicy != c.cfg.Policy.TxPolicy {
                        continue
                }

                candidateSession = sessionInfo
                break
        }

        // If none of the sessions could be used or none were found, we'll
        // return nil to signal that we need another session to be negotiated.
        if candidateSession == nil {
                return nil, nil
        }

        updates, err := c.cfg.DB.FetchSessionCommittedUpdates(
                &candidateSession.ID,
        )
        if err != nil {
                return nil, err
        }

        // Initialize the session queue and spin it up, so it can begin handling
        // updates. If the queue was already made active on startup, this will
        // simply return the existing session queue from the set.
        return c.getOrInitActiveQueue(candidateSession, updates), nil
}

// stopAndRemoveSession stops the session with the given ID and removes it from
// the in-memory active sessions set.
func (c *client) stopAndRemoveSession(id wtdb.SessionID, final bool) error {
        return c.activeSessions.StopAndRemove(id, final)
}

// deleteSessionFromTower dials the tower that we created the session with and
// attempts to send the tower the DeleteSession message.
func (c *client) deleteSessionFromTower(sess *wtdb.ClientSession) error {
        // First, we check if we have already loaded this tower in our
        // candidate towers iterator.
        tower, err := c.candidateTowers.GetTower(sess.TowerID)
        if errors.Is(err, ErrTowerNotInIterator) {
                // If not, then we attempt to load it from the DB.
                dbTower, err := c.cfg.DB.LoadTowerByID(sess.TowerID)
                if err != nil {
                        return err
                }

                tower, err = NewTowerFromDBTower(dbTower)
                if err != nil {
                        return err
                }
        } else if err != nil {
                return err
        }

        session, err := NewClientSessionFromDBSession(
                sess, tower, c.cfg.SecretKeyRing,
        )
        if err != nil {
                return err
        }

        localInit := wtwire.NewInitMessage(
                lnwire.NewRawFeatureVector(wtwire.AltruistSessionsRequired),
                c.cfg.ChainHash,
        )

        var (
                conn wtserver.Peer

                // addrIterator is a copy of the tower's address iterator.
                // We use this copy so that iterating through the addresses does
                // not affect any other threads using this iterator.
                addrIterator = tower.Addresses.Copy()
                towerAddr    = addrIterator.Peek()
        )
        // Attempt to dial the tower with its available addresses.
        for {
                conn, err = c.dial(
                        session.SessionKeyECDH, &lnwire.NetAddress{
                                IdentityKey: tower.IdentityKey,
                                Address:     towerAddr,
                        },
                )
                if err != nil {
                        // If there are more addrs available, immediately try
                        // those.
                        nextAddr, iteratorErr := addrIterator.Next()
                        if iteratorErr == nil {
                                towerAddr = nextAddr
                                continue
                        }

                        // Otherwise, if we have exhausted the address list,
                        // exit.
                        addrIterator.Reset()

                        return fmt.Errorf("failed to dial tower(%x) at any "+
                                "available addresses",
                                tower.IdentityKey.SerializeCompressed())
                }

                break
        }
        defer conn.Close()

        // Send Init to tower.
        err = c.sendMessage(conn, localInit)
        if err != nil {
                return err
        }

        // Receive Init from tower.
        remoteMsg, err := c.readMessage(conn)
        if err != nil {
                return err
        }

        remoteInit, ok := remoteMsg.(*wtwire.Init)
        if !ok {
                return fmt.Errorf("watchtower %s responded with %T to Init",
                        towerAddr, remoteMsg)
        }

        // Validate Init.
        err = localInit.CheckRemoteInit(remoteInit, wtwire.FeatureNames)
        if err != nil {
                return err
        }

        // Send DeleteSession to tower.
        err = c.sendMessage(conn, &wtwire.DeleteSession{})
        if err != nil {
                return err
        }

        // Receive DeleteSessionReply from tower.
        remoteMsg, err = c.readMessage(conn)
        if err != nil {
                return err
        }

        deleteSessionReply, ok := remoteMsg.(*wtwire.DeleteSessionReply)
        if !ok {
                return fmt.Errorf("watchtower %s responded with %T to "+
                        "DeleteSession", towerAddr, remoteMsg)
        }

        switch deleteSessionReply.Code {
        case wtwire.CodeOK, wtwire.DeleteSessionCodeNotFound:
                return nil
        default:
                return fmt.Errorf("received error code %v in "+
                        "DeleteSessionReply when attempting to delete "+
                        "session from tower", deleteSessionReply.Code)
        }
}

// backupDispatcher processes events coming from the taskPipeline and is
// responsible for detecting when the client needs to renegotiate a session to
// fulfill continuing demand. The event loop exits if the client is quit.
//
// NOTE: This method MUST be run as a goroutine.
func (c *client) backupDispatcher() {
        defer c.wg.Done()

        c.log.Tracef("Starting backup dispatcher")
        defer c.log.Tracef("Stopping backup dispatcher")

        for {
                switch {

                // No active session queue and no additional sessions.
                case c.sessionQueue == nil && len(c.candidateSessions) == 0:
                        c.log.Infof("Requesting new session.")

                        // Immediately request a new session.
                        c.negotiator.RequestSession()

                        // Wait until we receive the newly negotiated session.
                        // All backups sent in the meantime are queued in the
                        // revoke queue, as we cannot process them.
                awaitSession:
                        select {
                        case session := <-c.negotiator.NewSessions():
                                c.log.Infof("Acquired new session with id=%s",
                                        session.ID)
                                c.candidateSessions[session.ID] = session
                                c.stats.sessionAcquired()

                                // We'll continue to choose the newly negotiated
                                // session as our active session queue.
                                continue

                        case <-c.statTicker.C:
                                c.log.Infof("Client stats: %s", c.stats)

                        // A new tower has been requested to be added. We'll
                        // update our persisted and in-memory state and consider
                        // its corresponding sessions, if any, as new
                        // candidates.
                        case msg := <-c.newTowers:
                                msg.errChan <- c.handleNewTower(msg.tower)

                        // A tower has been requested to be removed. We'll
                        // only allow removal of it if the address in question
                        // is not currently being used for session negotiation.
                        case msg := <-c.staleTowers:
                                msg.errChan <- c.handleStaleTower(msg)

                        // A tower has been requested to be de-activated. We'll
                        // only allow this if the tower is not currently being
                        // used for session negotiation.
                        case msg := <-c.deactivateTowers:
                                msg.errChan <- c.handleDeactivateTower(msg)

                        // A request has come through to terminate a session.
                        case msg := <-c.terminateSessions:
                                msg.errChan <- c.handleTerminateSession(msg)

                        case <-c.quit:
                                return
                        }

                        // Instead of looping, we'll jump back into the select
                        // case and await the delivery of the session to prevent
                        // us from re-requesting additional sessions.
                        goto awaitSession

                // No active session queue but have additional sessions.
                case c.sessionQueue == nil && len(c.candidateSessions) > 0:
                        // We've exhausted the prior session, we'll pop another
                        // from the remaining sessions and continue processing
                        // backup tasks.
                        var err error
                        c.sessionQueue, err = c.nextSessionQueue()
                        if err != nil {
                                c.log.Errorf("error fetching next session "+
                                        "queue: %v", err)
                        }

                        if c.sessionQueue != nil {
                                c.log.Debugf("Loaded next candidate session "+
                                        "queue id=%s", c.sessionQueue.ID())
                        }

                // Have active session queue, process backups.
                case c.sessionQueue != nil:
                        if c.prevTask != nil {
                                c.processTask(c.prevTask)

                                // Continue to ensure the sessionQueue is
                                // properly initialized before attempting to
                                // process more tasks from the pipeline.
                                continue
                        }

                        // Normal operation where new tasks are read from the
                        // pipeline.
                        select {

                        // If any sessions are negotiated while we have an
                        // active session queue, queue them for future use.
                        // This shouldn't happen with the current design, so
                        // it doesn't hurt to select here just in case. In the
                        // future, we will likely allow more asynchrony so that
                        // we can request new sessions before the session is
                        // fully empty, which this case would handle.
                        case session := <-c.negotiator.NewSessions():
                                c.log.Warnf("Acquired new session with id=%s "+
                                        "while processing tasks", session.ID)
                                c.candidateSessions[session.ID] = session
                                c.stats.sessionAcquired()

                        case <-c.statTicker.C:
                                c.log.Infof("Client stats: %s", c.stats)

                        // Process each backup task serially from the queue of
                        // revoked states.
                        case task, ok := <-c.pipeline.NextBackupID():
                                // All backups in the pipeline have been
                                // processed, it is now safe to exit.
                                if !ok {
                                        return
                                }

                                c.log.Debugf("Processing %v", task)

                                c.stats.taskReceived()
                                c.processTask(task)

                        // A new tower has been requested to be added. We'll
                        // update our persisted and in-memory state and consider
                        // its corresponding sessions, if any, as new
                        // candidates.
                        case msg := <-c.newTowers:
                                msg.errChan <- c.handleNewTower(msg.tower)

                        // A tower has been removed, so we'll remove certain
                        // information that's persisted and also in our
                        // in-memory state depending on the request, and set any
                        // of its corresponding candidate sessions as inactive.
                        case msg := <-c.staleTowers:
                                msg.errChan <- c.handleStaleTower(msg)

                        // A tower has been requested to be de-activated.
                        case msg := <-c.deactivateTowers:
                                msg.errChan <- c.handleDeactivateTower(msg)

                        // A request has come through to terminate a session.
                        case msg := <-c.terminateSessions:
                                msg.errChan <- c.handleTerminateSession(msg)

                        case <-c.quit:
                                return
                        }
                }
        }
}

// processTask attempts to schedule the given backupTask on the active
// sessionQueue. The task will either be accepted or rejected, after which the
// appropriate modifications to the client's state machine will be made. After
// every invocation of processTask, the caller should ensure that the
// sessionQueue hasn't been exhausted before proceeding to the next task. Tasks
// that are rejected because the active sessionQueue is full will be cached as
// the prevTask, and should be reprocessed after obtaining a new sessionQueue.
func (c *client) processTask(task *wtdb.BackupID) {
        script, ok := c.cfg.getSweepScript(task.ChanID)
        if !ok {
                log.Infof("not processing task for unregistered channel: %s",
                        task.ChanID)

                return
        }

        backupTask := newBackupTask(*task, script)

        status, accepted := c.sessionQueue.AcceptTask(backupTask)
        if accepted {
                c.taskAccepted(task, status)
        } else {
                c.taskRejected(task, status)
        }
}

// taskAccepted processes the acceptance of a task by a sessionQueue depending
// on the state the sessionQueue is in *after* the task is added. The client's
// prevTask is always removed as a result of this call. The client's
// sessionQueue will be removed if accepting the task left the sessionQueue in
// an exhausted state.
func (c *client) taskAccepted(task *wtdb.BackupID,
        newStatus sessionQueueStatus) {

        c.log.Infof("Queued %v successfully for session %v", task,
                c.sessionQueue.ID())

        c.stats.taskAccepted()

        // If this task was accepted, we discard anything held in the prevTask.
        // Either it was nil before, or is the task which was just accepted.
        c.prevTask = nil

        switch newStatus {

        // The sessionQueue still has capacity after accepting this task.
        case sessionQueueAvailable:

        // The sessionQueue is full after accepting this task, so we will need
        // to request a new one before proceeding.
        case sessionQueueExhausted:
                c.stats.sessionExhausted()

                c.log.Debugf("Session %s exhausted", c.sessionQueue.ID())

                // This task left the session exhausted, set it to nil and
                // proceed to the next loop, so we can consume another
                // pre-negotiated session or request another.
                c.sessionQueue = nil
        }
}

// taskRejected process the rejection of a task by a sessionQueue depending on
// the state the was in *before* the task was rejected. The client's prevTask
// will cache the task if the sessionQueue was exhausted beforehand, and nil
// the sessionQueue to find a new session. If the sessionQueue was not
// exhausted and not shutting down, the client marks the task as ineligible, as
// this implies we couldn't construct a valid justice transaction given the
// session's policy.
func (c *client) taskRejected(task *wtdb.BackupID,
        curStatus sessionQueueStatus) {

        switch curStatus {

        // The sessionQueue has available capacity but the task was rejected,
        // this indicates that the task was ineligible for backup.
        case sessionQueueAvailable:
                c.stats.taskIneligible()

                c.log.Infof("Ignoring ineligible %v", task)

                err := c.cfg.DB.MarkBackupIneligible(
                        task.ChanID, task.CommitHeight,
                )
                if err != nil {
                        c.log.Errorf("Unable to mark %v ineligible: %v",
                                task, err)

                        // It is safe to not handle this error, even if we could
                        // not persist the result. At worst, this task may be
                        // reprocessed on a subsequent start up, and will either
                        // succeed do a change in session parameters or fail in
                        // the same manner.
                }

                // If this task was rejected *and* the session had available
                // capacity, we discard anything held in the prevTask. Either it
                // was nil before, or is the task which was just rejected.
                c.prevTask = nil

        // The sessionQueue rejected the task because it is full, we will stash
        // this task and try to add it to the next available sessionQueue.
        case sessionQueueExhausted:
                c.stats.sessionExhausted()

                c.log.Debugf("Session %v exhausted, %v queued for next session",
                        c.sessionQueue.ID(), task)

                // Cache the task that we pulled off, so that we can process it
                // once a new session queue is available.
                c.sessionQueue = nil
                c.prevTask = task

        // The sessionQueue rejected the task because it is shutting down. We
        // will stash this task and try to add it to the next available
        // sessionQueue.
        case sessionQueueShuttingDown:
                c.log.Debugf("Session %v is shutting down, %v queued for "+
                        "next session", c.sessionQueue.ID(), task)

                // Cache the task that we pulled off, so that we can process it
                // once a new session queue is available.
                c.sessionQueue = nil
                c.prevTask = task
        }
}

// dial connects the peer at addr using privKey as our secret key for the
// connection. The connection will use the configured Net's resolver to resolve
// the address for either Tor or clear net connections.
func (c *client) dial(localKey keychain.SingleKeyECDH,
        addr *lnwire.NetAddress) (wtserver.Peer, error) {

        return c.cfg.AuthDial(localKey, addr, c.cfg.Dial)
}

// readMessage receives and parses the next message from the given Peer. An
// error is returned if a message is not received before the server's read
// timeout, the read off the wire failed, or the message could not be
// deserialized.
func (c *client) readMessage(peer wtserver.Peer) (wtwire.Message, error) {
        // Set a read timeout to ensure we drop the connection if nothing is
        // received in a timely manner.
        err := peer.SetReadDeadline(time.Now().Add(c.cfg.ReadTimeout))
        if err != nil {
                err = fmt.Errorf("unable to set read deadline: %w", err)
                c.log.Errorf("Unable to read msg: %v", err)
                return nil, err
        }

        // Pull the next message off the wire,
        rawMsg, err := peer.ReadNextMessage()
        if err != nil {
                err = fmt.Errorf("unable to read message: %w", err)
                c.log.Errorf("Unable to read msg: %v", err)
                return nil, err
        }

        // Parse the received message according to the watchtower wire
        // specification.
        msgReader := bytes.NewReader(rawMsg)
        msg, err := wtwire.ReadMessage(msgReader, 0)
        if err != nil {
                err = fmt.Errorf("unable to parse message: %w", err)
                c.log.Errorf("Unable to read msg: %v", err)
                return nil, err
        }

        c.logMessage(peer, msg, true)

        return msg, nil
}

// sendMessage sends a watchtower wire message to the target peer.
func (c *client) sendMessage(peer wtserver.Peer,
        msg wtwire.Message) error {

        // Encode the next wire message into the buffer.
        // TODO(conner): use buffer pool
        var b bytes.Buffer
        _, err := wtwire.WriteMessage(&b, msg, 0)
        if err != nil {
                err = fmt.Errorf("unable to encode msg: %w", err)
                c.log.Errorf("Unable to send msg: %v", err)
                return err
        }

        // Set the write deadline for the connection, ensuring we drop the
        // connection if nothing is sent in a timely manner.
        err = peer.SetWriteDeadline(time.Now().Add(c.cfg.WriteTimeout))
        if err != nil {
                err = fmt.Errorf("unable to set write deadline: %w", err)
                c.log.Errorf("Unable to send msg: %v", err)
                return err
        }

        c.logMessage(peer, msg, false)

        // Write out the full message to the remote peer.
        _, err = peer.Write(b.Bytes())
        if err != nil {
                c.log.Errorf("Unable to send msg: %v", err)
        }
        return err
}

// newSessionQueue creates a sessionQueue from a ClientSession loaded from the
// database and supplying it with the resources needed by the client.
func (c *client) newSessionQueue(s *ClientSession,
        updates []wtdb.CommittedUpdate) *sessionQueue {

        return newSessionQueue(&sessionQueueConfig{
                ClientSession:          s,
                ChainHash:              c.cfg.ChainHash,
                Dial:                   c.dial,
                ReadMessage:            c.readMessage,
                SendMessage:            c.sendMessage,
                Signer:                 c.cfg.Signer,
                DB:                     c.cfg.DB,
                MinBackoff:             c.cfg.MinBackoff,
                MaxBackoff:             c.cfg.MaxBackoff,
                Log:                    c.log,
                BuildBreachRetribution: c.cfg.BuildBreachRetribution,
                TaskPipeline:           c.pipeline,
        }, updates)
}

// getOrInitActiveQueue checks the activeSessions set for a sessionQueue for the
// passed ClientSession. If it exists, the active sessionQueue is returned.
// Otherwise, a new sessionQueue is initialized and added to the set.
func (c *client) getOrInitActiveQueue(s *ClientSession,
        updates []wtdb.CommittedUpdate) *sessionQueue {

        if sq, ok := c.activeSessions.Get(s.ID); ok {
                return sq
        }

        return c.initActiveQueue(s, updates)
}

// initActiveQueue creates a new sessionQueue from the passed ClientSession,
// adds the sessionQueue to the activeSessions set, and starts the sessionQueue
// so that it can deliver any committed updates or begin accepting newly
// assigned tasks.
func (c *client) initActiveQueue(s *ClientSession,
        updates []wtdb.CommittedUpdate) *sessionQueue {

        // Initialize the session queue, providing it with all the resources it
        // requires from the client instance.
        sq := c.newSessionQueue(s, updates)

        // Add the session queue as an active session so that we remember to
        // stop it on shutdown. This method will also start the queue so that it
        // can be active in processing newly assigned tasks or to upload
        // previously committed updates.
        c.activeSessions.AddAndStart(sq)

        return sq
}

// terminateSession sets the given session's status to CSessionTerminal meaning
// that it will not be used again.
func (c *client) terminateSession(id wtdb.SessionID) error {
        errChan := make(chan error, 1)

        select {
        case c.terminateSessions <- &terminateSessMsg{
                id:      id,
                errChan: errChan,
        }:
        case <-c.pipeline.quit:
                return ErrClientExiting
        }

        select {
        case err := <-errChan:
                return err
        case <-c.pipeline.quit:
                return ErrClientExiting
        }
}

// handleTerminateSession handles a request to terminate a session. It will
// first shut down the session if it is part of the active session set, then
// it will ensure that the active session queue is set reset if it is using the
// session in question. Finally, the session's status in the DB will be updated.
func (c *client) handleTerminateSession(msg *terminateSessMsg) error {
        id := msg.id

        delete(c.candidateSessions, id)

        err := c.activeSessions.StopAndRemove(id, true)
        if err != nil {
                return fmt.Errorf("could not stop session %s: %w", id, err)
        }

        // If our active session queue corresponds to the session being
        // terminated, then we'll proceed to negotiate a new one.
        if c.sessionQueue != nil {
                if bytes.Equal(c.sessionQueue.ID()[:], id[:]) {
                        c.sessionQueue = nil
                }
        }

        return nil
}

// deactivateTower sends a tower deactivation request to the backupDispatcher
// where it will be handled synchronously. The request should result in all the
// sessions that we have with the given tower being shutdown and removed from
// our in-memory set of active sessions.
func (c *client) deactivateTower(id wtdb.TowerID,
        pubKey *btcec.PublicKey) error {

        errChan := make(chan error, 1)

        select {
        case c.deactivateTowers <- &deactivateTowerMsg{
                id:      id,
                pubKey:  pubKey,
                errChan: errChan,
        }:
        case <-c.pipeline.quit:
                return ErrClientExiting
        }

        select {
        case err := <-errChan:
                return err
        case <-c.pipeline.quit:
                return ErrClientExiting
        }
}

// handleDeactivateTower handles a request to deactivate a tower. We will remove
// it from the in-memory candidate set, and we will also stop any active
// sessions we have with this tower.
func (c *client) handleDeactivateTower(msg *deactivateTowerMsg) error {
        // Remove the tower from our in-memory candidate set so that it is not
        // used for any new session negotiations.
        err := c.candidateTowers.RemoveCandidate(msg.id, nil)
        if err != nil {
                return err
        }

        pubKey := msg.pubKey.SerializeCompressed()
        sessions, err := c.cfg.DB.ListClientSessions(&msg.id)
        if err != nil {
                return fmt.Errorf("unable to retrieve sessions for tower %x: "+
                        "%v", pubKey, err)
        }

        // Iterate over all the sessions we have for this tower and remove them
        // from our candidate set and also from our set of started, active
        // sessions.
        for sessionID := range sessions {
                delete(c.candidateSessions, sessionID)

                err = c.activeSessions.StopAndRemove(sessionID, false)
                if err != nil {
                        return fmt.Errorf("could not stop session %s: %w",
                                sessionID, err)
                }
        }

        // If our active session queue corresponds to the stale tower, we'll
        // proceed to negotiate a new one.
        if c.sessionQueue != nil {
                towerKey := c.sessionQueue.tower.IdentityKey

                if bytes.Equal(pubKey, towerKey.SerializeCompressed()) {
                        c.sessionQueue = nil
                }
        }

        return nil
}

// addTower adds a new watchtower reachable at the given address and considers
// it for new sessions. If the watchtower already exists, then any new addresses
// included will be considered when dialing it for session negotiations and
// backups.
func (c *client) addTower(tower *Tower) error {
        errChan := make(chan error, 1)

        select {
        case c.newTowers <- &newTowerMsg{
                tower:   tower,
                errChan: errChan,
        }:
        case <-c.pipeline.quit:
                return ErrClientExiting
        }

        select {
        case err := <-errChan:
                return err
        case <-c.pipeline.quit:
                return ErrClientExiting
        }
}

// handleNewTower handles a request for a new tower to be added. If the tower
// already exists, then its corresponding sessions, if any, will be set
// considered as candidates.
func (c *client) handleNewTower(tower *Tower) error {
        c.candidateTowers.AddCandidate(tower)

        // Include all of its corresponding sessions to our set of candidates.
        sessions, err := getClientSessions(
                c.cfg.DB, c.cfg.SecretKeyRing, &tower.ID,
                wtdb.WithPreEvalFilterFn(c.genSessionFilter(true)),
                wtdb.WithPostEvalFilterFn(ExhaustedSessionFilter()),
        )
        if err != nil {
                return fmt.Errorf("unable to determine sessions for tower %x: "+
                        "%v", tower.IdentityKey.SerializeCompressed(), err)
        }
        for id, session := range sessions {
                c.candidateSessions[id] = session
        }

        return nil
}

// removeTower removes a watchtower from being considered for future session
// negotiations and from being used for any subsequent backups until it's added
// again. If an address is provided, then this call only serves as a way of
// removing the address from the watchtower instead.
func (c *client) removeTower(id wtdb.TowerID, pubKey *btcec.PublicKey,
        addr net.Addr) error {

        errChan := make(chan error, 1)

        select {
        case c.staleTowers <- &staleTowerMsg{
                id:      id,
                pubKey:  pubKey,
                addr:    addr,
                errChan: errChan,
        }:
        case <-c.pipeline.quit:
                return ErrClientExiting
        }

        select {
        case err := <-errChan:
                return err
        case <-c.pipeline.quit:
                return ErrClientExiting
        }
}

// handleStaleTower handles a request for an existing tower to be removed. If
// none of the tower's sessions have pending updates, then they will become
// inactive and removed as candidates. If the active session queue corresponds
// to any of these sessions, a new one will be negotiated.
func (c *client) handleStaleTower(msg *staleTowerMsg) error {
        // We'll first update our in-memory state.
        err := c.candidateTowers.RemoveCandidate(msg.id, msg.addr)
        if err != nil {
                return err
        }

        // If an address was provided, then we're only meant to remove the
        // address from the tower.
        if msg.addr != nil {
                return nil
        }

        // Otherwise, the tower should no longer be used for future session
        // negotiations and backups.

        pubKey := msg.pubKey.SerializeCompressed()
        sessions, err := c.cfg.DB.ListClientSessions(&msg.id)
        if err != nil {
                return fmt.Errorf("unable to retrieve sessions for tower %x: "+
                        "%v", pubKey, err)
        }
        for sessionID := range sessions {
                delete(c.candidateSessions, sessionID)

                // Shutdown the session so that any pending updates are
                // replayed back onto the main task pipeline.
                err = c.activeSessions.StopAndRemove(sessionID, true)
                if err != nil {
                        c.log.Errorf("could not stop session %s: %w", sessionID,
                                err)
                }
        }

        // If our active session queue corresponds to the stale tower, we'll
        // proceed to negotiate a new one.
        if c.sessionQueue != nil {
                towerKey := c.sessionQueue.tower.IdentityKey

                if bytes.Equal(pubKey, towerKey.SerializeCompressed()) {
                        c.sessionQueue = nil
                }
        }

        return nil
}

// registeredTowers retrieves the list of watchtowers registered with the
// client.
func (c *client) registeredTowers(towers []*wtdb.Tower,
        opts ...wtdb.ClientSessionListOption) ([]*RegisteredTower, error) {

        // Generate a filter that will fetch all the client's sessions
        // regardless of if they are active or not.
        opts = append(opts, wtdb.WithPreEvalFilterFn(c.genSessionFilter(false)))

        clientSessions, err := c.cfg.DB.ListClientSessions(nil, opts...)
        if err != nil {
                return nil, err
        }

        // Construct a lookup map that coalesces all the sessions for a specific
        // watchtower.
        towerSessions := make(
                map[wtdb.TowerID]map[wtdb.SessionID]*wtdb.ClientSession,
        )
        for id, s := range clientSessions {
                sessions, ok := towerSessions[s.TowerID]
                if !ok {
                        sessions = make(map[wtdb.SessionID]*wtdb.ClientSession)
                        towerSessions[s.TowerID] = sessions
                }
                sessions[id] = s
        }

        registeredTowers := make([]*RegisteredTower, 0, len(towerSessions))
        for _, tower := range towers {
                isActive := c.candidateTowers.IsActive(tower.ID)
                registeredTowers = append(registeredTowers, &RegisteredTower{
                        Tower:                  tower,
                        Sessions:               towerSessions[tower.ID],
                        ActiveSessionCandidate: isActive,
                })
        }

        return registeredTowers, nil
}

// lookupTower retrieves the info of sessions held with the given tower handled
// by this client.
func (c *client) lookupTower(tower *wtdb.Tower,
        opts ...wtdb.ClientSessionListOption) (*RegisteredTower, error) {

        opts = append(opts, wtdb.WithPreEvalFilterFn(c.genSessionFilter(false)))

        towerSessions, err := c.cfg.DB.ListClientSessions(&tower.ID, opts...)
        if err != nil {
                return nil, err
        }

        return &RegisteredTower{
                Tower:                  tower,
                Sessions:               towerSessions,
                ActiveSessionCandidate: c.candidateTowers.IsActive(tower.ID),
        }, nil
}

// getStats returns the in-memory statistics of the client since startup.
func (c *client) getStats() ClientStats {
        return c.stats.getStatsCopy()
}

// policy returns the active client policy configuration.
func (c *client) policy() wtpolicy.Policy {
        return c.cfg.Policy
}

// logMessage writes information about a message received from a remote peer,
// using directional prepositions to signal whether the message was sent or
// received.
func (c *client) logMessage(
        peer wtserver.Peer, msg wtwire.Message, read bool) {

        var action = "Received"
        var preposition = "from"
        if !read {
                action = "Sending"
                preposition = "to"
        }

        summary := wtwire.MessageSummary(msg)
        if len(summary) > 0 {
                summary = "(" + summary + ")"
        }

        c.log.Debugf("%s %s%v %s %x@%s", action, msg.MsgType(), summary,
                preposition, peer.RemotePub().SerializeCompressed(),
                peer.RemoteAddr())
}

func newRandomDelay(max uint32) (uint32, error) {
        var maxDelay big.Int
        maxDelay.SetUint64(uint64(max))

        randDelay, err := rand.Int(rand.Reader, &maxDelay)
        if err != nil {
                return 0, err
        }

        return uint32(randDelay.Uint64()), nil
}

lightningnetwork / lnd / 13157733617

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous