13211764208

Committed 08 Feb 2025 03:08AM UTC coverage: 49.288% (-9.5%) from 58.815%

Build # 13211764208

Build Type

Pull #9489

github

Committed by

calvinrzachman

Commit Message

itest: verify switchrpc server enforces send then track

We prevent the rpc server from allowing onion dispatches for
attempt IDs which have already been tracked by rpc clients.

This helps protect the client from leaking a duplicate onion
attempt. NOTE: This is not the only method for solving this
issue! The issue could be addressed via careful client side
programming which accounts for the uncertainty and async
nature of dispatching onions to a remote process via RPC.
This would require some lnd ChannelRouter changes for how
we intend to use these RPCs though.

Pull Request Pull Request #9489: multi: add BuildOnion, SendOnion, and TrackOnion RPCs

Run Details

474 of 990 new or added lines in 11 files covered. (47.88%)

27321 existing lines in 435 files now uncovered.

101192 of 205306 relevant lines covered (49.29%)

1.54 hits per line

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

67.9

/watchtower/wtclient/session_queue.go

package wtclient

import (
        "container/list"
        "fmt"
        "sync"
        "time"

        "github.com/btcsuite/btcd/chaincfg/chainhash"
        "github.com/btcsuite/btclog/v2"
        "github.com/lightningnetwork/lnd/input"
        "github.com/lightningnetwork/lnd/keychain"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/watchtower/wtdb"
        "github.com/lightningnetwork/lnd/watchtower/wtserver"
        "github.com/lightningnetwork/lnd/watchtower/wtwire"
)

// sessionQueueStatus is an enum that signals how full a particular session is.
type sessionQueueStatus uint8

const (
        // sessionQueueAvailable indicates that the session has space for at
        // least one more backup.
        sessionQueueAvailable sessionQueueStatus = iota

        // sessionQueueExhausted indicates that all slots in the session have
        // been allocated.
        sessionQueueExhausted

        // sessionQueueShuttingDown indicates that the session queue is
        // shutting down and so is no longer accepting any more backups.
        sessionQueueShuttingDown
)

// 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 *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(keychain.SingleKeyECDH, *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

        // BuildBreachRetribution is a function closure that allows the client
        // to fetch the breach retribution info for a certain channel at a
        // certain revoked commitment height.
        BuildBreachRetribution BreachRetributionBuilder

        // TaskPipeline is a pipeline which the sessionQueue should use to send
        // any unhandled tasks on shutdown of the queue.
        TaskPipeline *DiskOverflowQueue[*wtdb.BackupID]

        // 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

        // Log specifies the desired log output, which should be prefixed by the
        // client type, e.g. anchor or legacy.
        Log btclog.Logger
}

// sessionQueue implements a reliable queue that will encrypt and send accepted
// backups to the watchtower specified in the config's ClientSession. Calling
// Stop will attempt to perform a clean shutdown replaying any un-committed
// pending updates to the client's main task pipeline.
type sessionQueue struct {
        started sync.Once
        stopped sync.Once
        forced  sync.Once

        cfg *sessionQueueConfig
        log btclog.Logger

        commitQueue  *list.List
        pendingQueue *list.List
        queueMtx     sync.Mutex
        queueCond    *sync.Cond

        localInit *wtwire.Init
        tower     *Tower

        seqNum uint16

        retryBackoff time.Duration

        quit chan struct{}
        wg   sync.WaitGroup
}

// newSessionQueue initializes a fresh sessionQueue.
func newSessionQueue(cfg *sessionQueueConfig,
        updates []wtdb.CommittedUpdate) *sessionQueue {

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

        sq := &sessionQueue{
                cfg:          cfg,
                log:          cfg.Log,
                commitQueue:  list.New(),
                pendingQueue: list.New(),
                localInit:    localInit,
                tower:        cfg.ClientSession.Tower,
                seqNum:       cfg.ClientSession.SeqNum,
                retryBackoff: cfg.MinBackoff,
                quit:         make(chan struct{}),
        }
        sq.queueCond = sync.NewCond(&sq.queueMtx)

        // The database should return them in sorted order, and session queue's
        // sequence number will be equal to that of the last committed update.
        for _, update := range updates {
                sq.commitQueue.PushBack(update)
        }

        return sq
}

// Start idempotently starts the sessionQueue so that it can begin accepting
// backups.
func (q *sessionQueue) Start() {
        q.started.Do(func() {
                q.wg.Add(1)
                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.
// The final param should only be set to true if this is the last time that
// this session will be used. Otherwise, during normal shutdown, the final param
// should be false.
func (q *sessionQueue) Stop(final bool) error {
        var returnErr error
        q.stopped.Do(func() {
                q.log.Debugf("SessionQueue(%s) stopping ...", q.ID())

                close(q.quit)

                shutdown := make(chan struct{})
                go func() {
                        for {
                                select {
                                case <-time.After(time.Millisecond):
                                        q.queueCond.Signal()
                                case <-shutdown:
                                        return
                                }
                        }
                }()

                q.wg.Wait()
                close(shutdown)

                // Now, for any task in the pending queue that we have not yet
                // created a CommittedUpdate for, re-add the task to the main
                // task pipeline.
                updates, err := q.cfg.DB.FetchSessionCommittedUpdates(q.ID())
                if err != nil {
                        returnErr = err
                        return
                }

                unAckedUpdates := make(map[wtdb.BackupID]bool)
                for _, update := range updates {
                        unAckedUpdates[update.BackupID] = true

                        if !final {
                                continue
                        }

                        err := q.cfg.TaskPipeline.QueueBackupID(
                                &update.BackupID,
                        )
                        if err != nil {
                                log.Errorf("could not re-queue %s: %v",
                                        update.BackupID, err)
                                continue
                        }
                }

                if final {
                        err = q.cfg.DB.DeleteCommittedUpdates(q.ID())
                        if err != nil {
                                log.Errorf("could not delete committed "+
                                        "updates for session %s", q.ID())
                        }
                }

                // Push any task that was on the pending queue that there is
                // not yet a committed update for back to the main task
                // pipeline.
                q.queueCond.L.Lock()
                for q.pendingQueue.Len() > 0 {
                        next := q.pendingQueue.Front()
                        q.pendingQueue.Remove(next)

                        //nolint:forcetypeassert
                        task := next.Value.(*backupTask)

                        if unAckedUpdates[task.id] {
                                continue
                        }

                        err := q.cfg.TaskPipeline.QueueBackupID(&task.id)
                        if err != nil {
                                log.Errorf("could not re-queue backup task: "+
                                        "%v", err)
                                continue
                        }
                }
                q.queueCond.L.Unlock()

                q.log.Debugf("SessionQueue(%s) stopped", q.ID())
        })

        return returnErr
}

// 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 or shutting down and the task is successfully bound to the
// ClientSession.
func (q *sessionQueue) AcceptTask(task *backupTask) (sessionQueueStatus, bool) {
        // Exit early if the queue has started shutting down.
        select {
        case <-q.quit:
                return sessionQueueShuttingDown, false
        default:
        }

        q.queueCond.L.Lock()

        // There is a chance that sessionQueue started shutting down between
        // the last quit channel check and waiting for the lock. So check one
        // more time here.
        select {
        case <-q.quit:
                q.queueCond.L.Unlock()
                return sessionQueueShuttingDown, false
        default:
        }

        numPending := uint32(q.pendingQueue.Len())
        maxUpdates := q.cfg.ClientSession.Policy.MaxUpdates
        q.log.Debugf("SessionQueue(%s) deciding to accept %v seqnum=%d "+
                "pending=%d max-updates=%d",
                q.ID(), task.id, q.seqNum, numPending, maxUpdates)

        // Examine the current reserve status of the session queue.
        curStatus := q.status()

        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 sessionQueueExhausted:
                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 sessionQueueAvailable:
                err := task.bindSession(
                        &q.cfg.ClientSession.ClientSessionBody,
                        q.cfg.BuildBreachRetribution,
                )
                if err != nil {
                        q.queueCond.L.Unlock()
                        q.log.Debugf("SessionQueue(%s) rejected %v: %v ",
                                q.ID(), task.id, 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.status()
        q.queueCond.L.Unlock()

        q.queueCond.Signal()

        return newStatus, true
}

// 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 q.wg.Done()

        for {
                q.queueCond.L.Lock()
                for q.commitQueue.Len() == 0 &&
                        q.pendingQueue.Len() == 0 {

                        q.queueCond.Wait()

                        select {
                        case <-q.quit:
                                q.queueCond.L.Unlock()
                                return
                        default:
                        }
                }
                q.queueCond.L.Unlock()

                // Exit immediately if the sessionQueue has been stopped.
                select {
                case <-q.quit:
                        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() {
        var (
                conn      wtserver.Peer
                err       error
                towerAddr = q.tower.Addresses.Peek()
        )

        for {
                q.log.Infof("SessionQueue(%s) attempting to dial tower at %v",
                        q.ID(), towerAddr)

                // First, check that we are able to dial this session's tower.
                conn, err = q.cfg.Dial(
                        q.cfg.ClientSession.SessionKeyECDH, &lnwire.NetAddress{
                                IdentityKey: q.tower.IdentityKey,
                                Address:     towerAddr,
                        },
                )
                if err != nil {
                        // If there are more addrs available, immediately try
                        // those.
                        nextAddr, iteratorErr := q.tower.Addresses.Next()
                        if iteratorErr == nil {
                                towerAddr = nextAddr
                                continue
                        }

                        // Otherwise, if we have exhausted the address list,
                        // back off and try again later.
                        q.tower.Addresses.Reset()

                        q.log.Errorf("SessionQueue(%s) unable to dial tower "+
                                "at any available Addresses: %v", q.ID(), err)

                        q.increaseBackoff()
                        select {
                        case <-time.After(q.retryBackoff):
                        case <-q.quit:
                        }
                        return
                }

                break
        }
        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 dequeuing committed updates
                // before attempting to dequeue any pending updates.
                stateUpdate, isPending, backupID, err := q.nextStateUpdate()
                if err != nil {
                        q.log.Errorf("SessionQueue(%v) unable to get next "+
                                "state update: %v", q.ID(), err)
                        return
                }

                // Now, send the state update to the tower and wait for a reply.
                err = q.sendStateUpdate(conn, stateUpdate, sendInit, isPending)
                if err != nil {
                        q.log.Errorf("SessionQueue(%s) unable to send state "+
                                "update: %v", q.ID(), err)

                        q.increaseBackoff()
                        select {
                        case <-time.After(q.retryBackoff):
                        case <-q.quit:
                        }
                        return
                }

                q.log.Infof("SessionQueue(%s) uploaded %v seqnum=%d",
                        q.ID(), backupID, stateUpdate.SeqNum)

                // 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.quit:
                        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,
        wtdb.BackupID, 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()

                update = next.Value.(wtdb.CommittedUpdate)
                seqNum = update.SeqNum

                // 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()

                q.log.Debugf("SessionQueue(%s) reprocessing committed state "+
                        "update for %v seqnum=%d",
                        q.ID(), update.BackupID, 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
                        err := fmt.Errorf("unable to craft session payload: %w",
                                err)
                        return nil, false, wtdb.BackupID{}, err
                }
                // TODO(conner): special case other obscure errors

                update = wtdb.CommittedUpdate{
                        SeqNum: seqNum,
                        CommittedUpdateBody: wtdb.CommittedUpdateBody{
                                BackupID:      task.id,
                                Hint:          hint,
                                EncryptedBlob: encBlob,
                        },
                }

                q.log.Debugf("SessionQueue(%s) committing state update "+
                        "%v seqnum=%d", q.ID(), update.BackupID, 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(), &update)
        if err != nil {
                // TODO(conner): mark failed/reschedule
                err := fmt.Errorf("unable to commit state update for "+
                        "%v seqnum=%d: %v", update.BackupID, seqNum, err)
                return nil, false, wtdb.BackupID{}, err
        }

        stateUpdate := &wtwire.StateUpdate{
                SeqNum:        update.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, update.BackupID, 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 we should back off before attempting to send the
// next state update.
func (q *sessionQueue) sendStateUpdate(conn wtserver.Peer,
        stateUpdate *wtwire.StateUpdate, sendInit, isPending bool) error {

        towerAddr := &lnwire.NetAddress{
                IdentityKey: conn.RemotePub(),
                Address:     conn.RemoteAddr(),
        }

        // 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 %s responded with %T "+
                                "to Init", towerAddr, 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 %s responded with %T to "+
                        "StateUpdate", towerAddr, 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 %v in "+
                        "StateUpdateReply for seqnum=%d",
                        stateUpdateReply.Code, stateUpdate.SeqNum)
                q.log.Warnf("SessionQueue(%s) unable to upload state update "+
                        "to tower=%s: %v", q.ID(), towerAddr, 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 seqnum=%d: %w",
                        stateUpdate.SeqNum, err)
                q.log.Errorf("SessionQueue(%v) failed to ack update: %v",
                        q.ID(), err)
                return err

        case err == wtdb.ErrLastAppliedReversion:
                // TODO(conner): borked watchtower
                err = fmt.Errorf("unable to ack seqnum=%d: %w",
                        stateUpdate.SeqNum, err)
                q.log.Errorf("SessionQueue(%s) failed to ack update: %v",
                        q.ID(), err)
                return err

        case err != nil:
                err = fmt.Errorf("unable to ack seqnum=%d: %w",
                        stateUpdate.SeqNum, err)
                q.log.Errorf("SessionQueue(%s) failed to ack update: %v",
                        q.ID(), err)
                return err
        }

        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
}

// status returns a sessionQueueStatus indicating whether the sessionQueue can
// accept another task. sessionQueueAvailable is returned when a task can be
// accepted, and sessionQueueExhausted 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) status() sessionQueueStatus {
        numPending := uint32(q.pendingQueue.Len())
        maxUpdates := uint32(q.cfg.ClientSession.Policy.MaxUpdates)

        if uint32(q.seqNum)+numPending < maxUpdates {
                return sessionQueueAvailable
        }

        return sessionQueueExhausted

}

// 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
        }
}

// sessionQueueSet maintains a mapping of SessionIDs to their corresponding
// sessionQueue.
type sessionQueueSet struct {
        queues map[wtdb.SessionID]*sessionQueue
        mu     sync.Mutex
}

// newSessionQueueSet constructs a new sessionQueueSet.
func newSessionQueueSet() *sessionQueueSet {
        return &sessionQueueSet{
                queues: make(map[wtdb.SessionID]*sessionQueue),
        }
}

// AddAndStart inserts a sessionQueue into the sessionQueueSet and starts it.
func (s *sessionQueueSet) AddAndStart(sessionQueue *sessionQueue) {
        s.mu.Lock()
        defer s.mu.Unlock()

        s.queues[*sessionQueue.ID()] = sessionQueue

        sessionQueue.Start()
}

// StopAndRemove stops the given session queue and removes it from the
// sessionQueueSet.
func (s *sessionQueueSet) StopAndRemove(id wtdb.SessionID, final bool) error {
        s.mu.Lock()
        defer s.mu.Unlock()

        queue, ok := s.queues[id]
        if !ok {
                return nil
        }

        delete(s.queues, id)

        return queue.Stop(final)
}

// Get fetches and returns the sessionQueue with the given ID.
func (s *sessionQueueSet) Get(id wtdb.SessionID) (*sessionQueue, bool) {
        s.mu.Lock()
        defer s.mu.Unlock()

        q, ok := s.queues[id]

        return q, ok
}

// 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()) {
        s.mu.Lock()
        defer s.mu.Unlock()

        var wg sync.WaitGroup
        for _, sessionq := range s.queues {
                wg.Add(1)
                go func(sq *sessionQueue) {
                        defer wg.Done()

                        getApply(sq)()
                }(sessionq)
        }
        wg.Wait()
}

lightningnetwork / lnd / 13211764208

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