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

48.67

/htlcswitch/quiescer.go

package htlcswitch

import (
        "fmt"
        "sync"
        "time"

        "github.com/btcsuite/btclog/v2"
        "github.com/lightningnetwork/lnd/fn/v2"
        "github.com/lightningnetwork/lnd/lntypes"
        "github.com/lightningnetwork/lnd/lnwire"
)

var (
        // ErrInvalidStfu indicates that the Stfu we have received is invalid.
        // This can happen in instances where we have not sent Stfu but we have
        // received one with the initiator field set to false.
        ErrInvalidStfu = fmt.Errorf("stfu received is invalid")

        // ErrStfuAlreadySent indicates that this channel has already sent an
        // Stfu message for this negotiation.
        ErrStfuAlreadySent = fmt.Errorf("stfu already sent")

        // ErrStfuAlreadyRcvd indicates that this channel has already received
        // an Stfu message for this negotiation.
        ErrStfuAlreadyRcvd = fmt.Errorf("stfu already received")

        // ErrNoQuiescenceInitiator indicates that the caller has requested the
        // quiescence initiator for a channel that is not yet quiescent.
        ErrNoQuiescenceInitiator = fmt.Errorf(
                "indeterminate quiescence initiator: channel is not quiescent",
        )

        // ErrPendingRemoteUpdates indicates that we have received an Stfu while
        // the remote party has issued updates that are not yet bilaterally
        // committed.
        ErrPendingRemoteUpdates = fmt.Errorf(
                "stfu received with pending remote updates",
        )

        // ErrPendingLocalUpdates indicates that we are attempting to send an
        // Stfu while we have issued updates that are not yet bilaterally
        // committed.
        ErrPendingLocalUpdates = fmt.Errorf(
                "stfu send attempted with pending local updates",
        )

        // ErrQuiescenceTimeout indicates that the quiescer has been quiesced
        // beyond the allotted time.
        ErrQuiescenceTimeout = fmt.Errorf(
                "quiescence timeout",
        )
)

const defaultQuiescenceTimeout = 30 * time.Second

type StfuReq = fn.Req[fn.Unit, fn.Result[lntypes.ChannelParty]]

// Quiescer is the public interface of the quiescence mechanism. Callers of the
// quiescence API should not need any methods besides the ones detailed here.
type Quiescer interface {
        // IsQuiescent returns true if the state machine has been driven all the
        // way to completion. If this returns true, processes that depend on
        // channel quiescence may proceed.
        IsQuiescent() bool

        // QuiescenceInitiator determines which ChannelParty is the initiator of
        // quiescence for the purposes of downstream protocols. If the channel
        // is not currently quiescent, this method will return
        // ErrNoDownstreamLeader.
        QuiescenceInitiator() fn.Result[lntypes.ChannelParty]

        // InitStfu instructs the quiescer that we intend to begin a quiescence
        // negotiation where we are the initiator. We don't yet send stfu yet
        // because we need to wait for the link to give us a valid opportunity
        // to do so.
        InitStfu(req StfuReq)

        // RecvStfu is called when we receive an Stfu message from the remote.
        RecvStfu(stfu lnwire.Stfu) error

        // CanRecvUpdates returns true if we haven't yet received an Stfu which
        // would mark the end of the remote's ability to send updates.
        CanRecvUpdates() bool

        // CanSendUpdates returns true if we haven't yet sent an Stfu which
        // would mark the end of our ability to send updates.
        CanSendUpdates() bool

        // SendOwedStfu sends Stfu if it owes one. It returns an error if the
        // state machine is in an invalid state.
        SendOwedStfu() error

        // OnResume accepts a no return closure that will run when the quiescer
        // is resumed.
        OnResume(hook func())

        // Resume runs all of the deferred actions that have accumulated while
        // the channel has been quiescent and then resets the quiescer state to
        // its initial state.
        Resume()
}

// QuiescerCfg is a config structure used to initialize a quiescer giving it the
// appropriate functionality to interact with the channel state that the
// quiescer must syncrhonize with.
type QuiescerCfg struct {
        // chanID marks what channel we are managing the state machine for. This
        // is important because the quiescer needs to know the ChannelID to
        // construct the Stfu message.
        chanID lnwire.ChannelID

        // channelInitiator indicates which ChannelParty originally opened the
        // channel. This is used to break ties when both sides of the channel
        // send Stfu claiming to be the initiator.
        channelInitiator lntypes.ChannelParty

        // sendMsg is a function that can be used to send an Stfu message over
        // the wire.
        sendMsg func(lnwire.Stfu) error

        // timeoutDuration is the Duration that we will wait from the moment the
        // channel is considered quiescent before we call the onTimeout function
        timeoutDuration time.Duration

        // onTimeout is a function that will be called in the event that the
        // Quiescer has not been resumed before the timeout is reached. If
        // Quiescer.Resume is called before the timeout has been raeached, then
        // onTimeout will not be called until the quiescer reaches a quiescent
        // state again.
        onTimeout func()
}

// QuiescerLive is a state machine that tracks progression through the
// quiescence protocol.
type QuiescerLive struct {
        cfg QuiescerCfg

        // log is a quiescer-scoped logging instance.
        log btclog.Logger

        // localInit indicates whether our path through this state machine was
        // initiated by our node. This can be true or false independently of
        // remoteInit.
        localInit bool

        // remoteInit indicates whether we received Stfu from our peer where the
        // message indicated that the remote node believes it was the initiator.
        // This can be true or false independently of localInit.
        remoteInit bool

        // sent tracks whether or not we have emitted Stfu for sending.
        sent bool

        // received tracks whether or not we have received Stfu from our peer.
        received bool

        // activeQuiescenceRequest is a possibly None Request that we should
        // resolve when we complete quiescence.
        activeQuiescenceReq fn.Option[StfuReq]

        // resumeQueue is a slice of hooks that will be called when the quiescer
        // is resumed. These are actions that needed to be deferred while the
        // channel was quiescent.
        resumeQueue []func()

        // timeoutTimer is a field that is used to hold onto the timeout job
        // when we reach quiescence.
        timeoutTimer *time.Timer

        sync.RWMutex
}

// NewQuiescer creates a new quiescer for the given channel.
func NewQuiescer(cfg QuiescerCfg) Quiescer {
        logPrefix := fmt.Sprintf("Quiescer(%v):", cfg.chanID)

        return &QuiescerLive{
                cfg: cfg,
                log: log.WithPrefix(logPrefix),
        }
}

// RecvStfu is called when we receive an Stfu message from the remote.
func (q *QuiescerLive) RecvStfu(msg lnwire.Stfu) error {
        q.Lock()
        defer q.Unlock()

        return q.recvStfu(msg)
}

// recvStfu is called when we receive an Stfu message from the remote.
func (q *QuiescerLive) recvStfu(msg lnwire.Stfu) error {
        // At the time of this writing, this check that we have already received
        // an Stfu is not strictly necessary, according to the specification.
        // However, it is fishy if we do and it is unclear how we should handle
        // such a case so we will err on the side of caution.
        if q.received {
                return fmt.Errorf("%w for channel %v", ErrStfuAlreadyRcvd,
                        q.cfg.chanID)
        }

        // We need to check that the Stfu we are receiving is valid.
        if !q.sent && !msg.Initiator {
                return fmt.Errorf("%w for channel %v", ErrInvalidStfu,
                        q.cfg.chanID)
        }

        if !q.canRecvStfu() {
                return fmt.Errorf("%w for channel %v", ErrPendingRemoteUpdates,
                        q.cfg.chanID)
        }

        q.received = true

        // If the remote party sets the initiator bit to true then we will
        // remember that they are making a claim to the initiator role. This
        // does not necessarily mean they will get it, though.
        q.remoteInit = msg.Initiator

        // Since we just received an Stfu, we may have a newly quiesced state.
        // If so, we will try to resolve any outstanding StfuReqs.
        q.tryResolveStfuReq()

        if q.isQuiescent() {
                q.startTimeout()
        }

        return nil
}

// MakeStfu is called when we are ready to send an Stfu message. It returns the
// Stfu message to be sent.
func (q *QuiescerLive) MakeStfu() fn.Result[lnwire.Stfu] {
        q.RLock()
        defer q.RUnlock()

        return q.makeStfu()
}

// makeStfu is called when we are ready to send an Stfu message. It returns the
// Stfu message to be sent.
func (q *QuiescerLive) makeStfu() fn.Result[lnwire.Stfu] {
        if q.sent {
                return fn.Errf[lnwire.Stfu]("%w for channel %v",
                        ErrStfuAlreadySent, q.cfg.chanID)
        }

        if !q.canSendStfu() {
                return fn.Errf[lnwire.Stfu]("%w for channel %v",
                        ErrPendingLocalUpdates, q.cfg.chanID)
        }

        stfu := lnwire.Stfu{
                ChanID:    q.cfg.chanID,
                Initiator: q.localInit,
        }

        return fn.Ok(stfu)
}

// OweStfu returns true if we owe the other party an Stfu. We owe the remote an
// Stfu when we have received but not yet sent an Stfu, or we are the initiator
// but have not yet sent an Stfu.
func (q *QuiescerLive) OweStfu() bool {
        q.RLock()
        defer q.RUnlock()

        return q.oweStfu()
}

// oweStfu returns true if we owe the other party an Stfu. We owe the remote an
// Stfu when we have received but not yet sent an Stfu, or we are the initiator
// but have not yet sent an Stfu.
func (q *QuiescerLive) oweStfu() bool {
        return (q.received || q.localInit) && !q.sent
}

// NeedStfu returns true if the remote owes us an Stfu. They owe us an Stfu when
// we have sent but not yet received an Stfu.
func (q *QuiescerLive) NeedStfu() bool {
        q.RLock()
        defer q.RUnlock()

        return q.needStfu()
}

// needStfu returns true if the remote owes us an Stfu. They owe us an Stfu when
// we have sent but not yet received an Stfu.
func (q *QuiescerLive) needStfu() bool {
        q.RLock()
        defer q.RUnlock()

        return q.sent && !q.received
}

// IsQuiescent returns true if the state machine has been driven all the way to
// completion. If this returns true, processes that depend on channel quiescence
// may proceed.
func (q *QuiescerLive) IsQuiescent() bool {
        q.RLock()
        defer q.RUnlock()

        return q.isQuiescent()
}

// isQuiescent returns true if the state machine has been driven all the way to
// completion. If this returns true, processes that depend on channel quiescence
// may proceed.
func (q *QuiescerLive) isQuiescent() bool {
        return q.sent && q.received
}

// QuiescenceInitiator determines which ChannelParty is the initiator of
// quiescence for the purposes of downstream protocols. If the channel is not
// currently quiescent, this method will return ErrNoQuiescenceInitiator.
func (q *QuiescerLive) QuiescenceInitiator() fn.Result[lntypes.ChannelParty] {
        q.RLock()
        defer q.RUnlock()

        return q.quiescenceInitiator()
}

// quiescenceInitiator determines which ChannelParty is the initiator of
// quiescence for the purposes of downstream protocols. If the channel is not
// currently quiescent, this method will return ErrNoQuiescenceInitiator.
func (q *QuiescerLive) quiescenceInitiator() fn.Result[lntypes.ChannelParty] {
        switch {
        case !q.isQuiescent():
                return fn.Err[lntypes.ChannelParty](ErrNoQuiescenceInitiator)

        case q.localInit && q.remoteInit:
                // In the case of a tie, the channel initiator wins.
                return fn.Ok(q.cfg.channelInitiator)

        case q.localInit:
                return fn.Ok(lntypes.Local)

        case q.remoteInit:
                return fn.Ok(lntypes.Remote)
        }

        // unreachable
        return fn.Err[lntypes.ChannelParty](ErrNoQuiescenceInitiator)
}

// CanSendUpdates returns true if we haven't yet sent an Stfu which would mark
// the end of our ability to send updates.
func (q *QuiescerLive) CanSendUpdates() bool {
        q.RLock()
        defer q.RUnlock()

        return q.canSendUpdates()
}

// canSendUpdates returns true if we haven't yet sent an Stfu which would mark
// the end of our ability to send updates.
func (q *QuiescerLive) canSendUpdates() bool {
        return !q.sent && !q.localInit
}

// CanRecvUpdates returns true if we haven't yet received an Stfu which would
// mark the end of the remote's ability to send updates.
func (q *QuiescerLive) CanRecvUpdates() bool {
        q.RLock()
        defer q.RUnlock()

        return q.canRecvUpdates()
}

// canRecvUpdates returns true if we haven't yet received an Stfu which would
// mark the end of the remote's ability to send updates.
func (q *QuiescerLive) canRecvUpdates() bool {
        return !q.received
}

// CanSendStfu returns true if we can send an Stfu.
func (q *QuiescerLive) CanSendStfu(numPendingLocalUpdates uint64) bool {
        q.RLock()
        defer q.RUnlock()

        return q.canSendStfu()
}

// canSendStfu returns true if we can send an Stfu.
func (q *QuiescerLive) canSendStfu() bool {
        return !q.sent
}

// CanRecvStfu returns true if we can receive an Stfu.
func (q *QuiescerLive) CanRecvStfu() bool {
        q.RLock()
        defer q.RUnlock()

        return q.canRecvStfu()
}

// canRecvStfu returns true if we can receive an Stfu.
func (q *QuiescerLive) canRecvStfu() bool {
        return !q.received
}

// SendOwedStfu sends Stfu if it owes one. It returns an error if the state
// machine is in an invalid state.
func (q *QuiescerLive) SendOwedStfu() error {
        q.Lock()
        defer q.Unlock()

        return q.sendOwedStfu()
}

// sendOwedStfu sends Stfu if it owes one. It returns an error if the state
// machine is in an invalid state.
func (q *QuiescerLive) sendOwedStfu() error {
        if !q.oweStfu() || !q.canSendStfu() {
                return nil
        }

        err := q.makeStfu().Sink(q.cfg.sendMsg)

        if err == nil {
                q.sent = true

                // Since we just sent an Stfu, we may have a newly quiesced
                // state. If so, we will try to resolve any outstanding
                // StfuReqs.
                q.tryResolveStfuReq()

                if q.isQuiescent() {
                        q.startTimeout()
                }
        }

        return err
}

// TryResolveStfuReq attempts to resolve the active quiescence request if the
// state machine has reached a quiescent state.
func (q *QuiescerLive) TryResolveStfuReq() {
        q.Lock()
        defer q.Unlock()

        q.tryResolveStfuReq()
}

// tryResolveStfuReq attempts to resolve the active quiescence request if the
// state machine has reached a quiescent state.
func (q *QuiescerLive) tryResolveStfuReq() {
        q.activeQuiescenceReq.WhenSome(
                func(req StfuReq) {
                        if q.isQuiescent() {
                                req.Resolve(q.quiescenceInitiator())
                                q.activeQuiescenceReq = fn.None[StfuReq]()
                        }
                },
        )
}

// InitStfu instructs the quiescer that we intend to begin a quiescence
// negotiation where we are the initiator. We don't yet send stfu yet because
// we need to wait for the link to give us a valid opportunity to do so.
func (q *QuiescerLive) InitStfu(req StfuReq) {
        q.Lock()
        defer q.Unlock()

        q.initStfu(req)
}

// initStfu instructs the quiescer that we intend to begin a quiescence
// negotiation where we are the initiator. We don't yet send stfu yet because
// we need to wait for the link to give us a valid opportunity to do so.
func (q *QuiescerLive) initStfu(req StfuReq) {
        if q.localInit {
                req.Resolve(fn.Errf[lntypes.ChannelParty](
                        "quiescence already requested",
                ))

                return
        }

        q.localInit = true
        q.activeQuiescenceReq = fn.Some(req)
}

// OnResume accepts a no return closure that will run when the quiescer is
// resumed.
func (q *QuiescerLive) OnResume(hook func()) {
        q.Lock()
        defer q.Unlock()

        q.onResume(hook)
}

// onResume accepts a no return closure that will run when the quiescer is
// resumed.
func (q *QuiescerLive) onResume(hook func()) {
        q.resumeQueue = append(q.resumeQueue, hook)
}

// Resume runs all of the deferred actions that have accumulated while the
// channel has been quiescent and then resets the quiescer state to its initial
// state.
func (q *QuiescerLive) Resume() {
        q.Lock()
        defer q.Unlock()

        q.resume()
}

// resume runs all of the deferred actions that have accumulated while the
// channel has been quiescent and then resets the quiescer state to its initial
// state.
func (q *QuiescerLive) resume() {
        q.log.Debug("quiescence terminated, resuming htlc traffic")

        // since we are resuming we want to cancel the quiescence timeout
        // action.
        q.cancelTimeout()

        for _, hook := range q.resumeQueue {
                hook()
        }
        q.localInit = false
        q.remoteInit = false
        q.sent = false
        q.received = false
        q.resumeQueue = nil
}

// startTimeout starts the timeout function that fires if the quiescer remains
// in a quiesced state for too long. If this function is called multiple times
// only the last one will have an effect.
func (q *QuiescerLive) startTimeout() {
        if q.cfg.onTimeout == nil {
                return
        }

        old := q.timeoutTimer

        q.timeoutTimer = time.AfterFunc(q.cfg.timeoutDuration, q.cfg.onTimeout)

        if old != nil {
                old.Stop()
        }
}

// cancelTimeout cancels the timeout function that would otherwise fire if the
// quiescer remains in a quiesced state too long. If this function is called
// before startTimeout or after another call to cancelTimeout, the effect will
// be a noop.
func (q *QuiescerLive) cancelTimeout() {
        if q.timeoutTimer != nil {
                q.timeoutTimer.Stop()
                q.timeoutTimer = nil
        }
}

type quiescerNoop struct{}

var _ Quiescer = (*quiescerNoop)(nil)

func (q *quiescerNoop) InitStfu(req StfuReq) {
        req.Resolve(fn.Errf[lntypes.ChannelParty]("quiescence not supported"))
}
func (q *quiescerNoop) RecvStfu(_ lnwire.Stfu) error { return nil }
func (q *quiescerNoop) CanRecvUpdates() bool         { return true }
func (q *quiescerNoop) CanSendUpdates() bool         { return true }
func (q *quiescerNoop) SendOwedStfu() error          { return nil }
func (q *quiescerNoop) IsQuiescent() bool            { return false }
func (q *quiescerNoop) OnResume(hook func())         { hook() }
func (q *quiescerNoop) Resume()                      {}
func (q *quiescerNoop) QuiescenceInitiator() fn.Result[lntypes.ChannelParty] {
        return fn.Err[lntypes.ChannelParty](ErrNoQuiescenceInitiator)
}

1	package htlcswitch
2
3	import (
4	"fmt"
5	"sync"
6	"time"
7
8	"github.com/btcsuite/btclog/v2"
9	"github.com/lightningnetwork/lnd/fn/v2"
10	"github.com/lightningnetwork/lnd/lntypes"
11	"github.com/lightningnetwork/lnd/lnwire"
12	)
13
14	var (
15	// ErrInvalidStfu indicates that the Stfu we have received is invalid.
16	// This can happen in instances where we have not sent Stfu but we have
17	// received one with the initiator field set to false.
18	ErrInvalidStfu = fmt.Errorf("stfu received is invalid")
19
20	// ErrStfuAlreadySent indicates that this channel has already sent an
21	// Stfu message for this negotiation.
22	ErrStfuAlreadySent = fmt.Errorf("stfu already sent")
23
24	// ErrStfuAlreadyRcvd indicates that this channel has already received
25	// an Stfu message for this negotiation.
26	ErrStfuAlreadyRcvd = fmt.Errorf("stfu already received")
27
28	// ErrNoQuiescenceInitiator indicates that the caller has requested the
29	// quiescence initiator for a channel that is not yet quiescent.
30	ErrNoQuiescenceInitiator = fmt.Errorf(
31	"indeterminate quiescence initiator: channel is not quiescent",
32	)
33
34	// ErrPendingRemoteUpdates indicates that we have received an Stfu while
35	// the remote party has issued updates that are not yet bilaterally
36	// committed.
37	ErrPendingRemoteUpdates = fmt.Errorf(
38	"stfu received with pending remote updates",
39	)
40
41	// ErrPendingLocalUpdates indicates that we are attempting to send an
42	// Stfu while we have issued updates that are not yet bilaterally
43	// committed.
44	ErrPendingLocalUpdates = fmt.Errorf(
45	"stfu send attempted with pending local updates",
46	)
47
48	// ErrQuiescenceTimeout indicates that the quiescer has been quiesced
49	// beyond the allotted time.
50	ErrQuiescenceTimeout = fmt.Errorf(
51	"quiescence timeout",
52	)
53	)
54
55	const defaultQuiescenceTimeout = 30 * time.Second
56
57	type StfuReq = fn.Req[fn.Unit, fn.Result[lntypes.ChannelParty]]
58
59	// Quiescer is the public interface of the quiescence mechanism. Callers of the
60	// quiescence API should not need any methods besides the ones detailed here.
61	type Quiescer interface {
62	// IsQuiescent returns true if the state machine has been driven all the
63	// way to completion. If this returns true, processes that depend on
64	// channel quiescence may proceed.
65	IsQuiescent() bool
66
67	// QuiescenceInitiator determines which ChannelParty is the initiator of
68	// quiescence for the purposes of downstream protocols. If the channel
69	// is not currently quiescent, this method will return
70	// ErrNoDownstreamLeader.
71	QuiescenceInitiator() fn.Result[lntypes.ChannelParty]
72
73	// InitStfu instructs the quiescer that we intend to begin a quiescence
74	// negotiation where we are the initiator. We don't yet send stfu yet
75	// because we need to wait for the link to give us a valid opportunity
76	// to do so.
77	InitStfu(req StfuReq)
78
79	// RecvStfu is called when we receive an Stfu message from the remote.
80	RecvStfu(stfu lnwire.Stfu) error
81
82	// CanRecvUpdates returns true if we haven't yet received an Stfu which
83	// would mark the end of the remote's ability to send updates.
84	CanRecvUpdates() bool
85
86	// CanSendUpdates returns true if we haven't yet sent an Stfu which
87	// would mark the end of our ability to send updates.
88	CanSendUpdates() bool
89
90	// SendOwedStfu sends Stfu if it owes one. It returns an error if the
91	// state machine is in an invalid state.
92	SendOwedStfu() error
93
94	// OnResume accepts a no return closure that will run when the quiescer
95	// is resumed.
96	OnResume(hook func())
97
98	// Resume runs all of the deferred actions that have accumulated while
99	// the channel has been quiescent and then resets the quiescer state to
100	// its initial state.
101	Resume()
102	}
103
104	// QuiescerCfg is a config structure used to initialize a quiescer giving it the
105	// appropriate functionality to interact with the channel state that the
106	// quiescer must syncrhonize with.
107	type QuiescerCfg struct {
108	// chanID marks what channel we are managing the state machine for. This
109	// is important because the quiescer needs to know the ChannelID to
110	// construct the Stfu message.
111	chanID lnwire.ChannelID
112
113	// channelInitiator indicates which ChannelParty originally opened the
114	// channel. This is used to break ties when both sides of the channel
115	// send Stfu claiming to be the initiator.
116	channelInitiator lntypes.ChannelParty
117
118	// sendMsg is a function that can be used to send an Stfu message over
119	// the wire.
120	sendMsg func(lnwire.Stfu) error
121
122	// timeoutDuration is the Duration that we will wait from the moment the
123	// channel is considered quiescent before we call the onTimeout function
124	timeoutDuration time.Duration
125
126	// onTimeout is a function that will be called in the event that the
127	// Quiescer has not been resumed before the timeout is reached. If
128	// Quiescer.Resume is called before the timeout has been raeached, then
129	// onTimeout will not be called until the quiescer reaches a quiescent
130	// state again.
131	onTimeout func()
132	}
133
134	// QuiescerLive is a state machine that tracks progression through the
135	// quiescence protocol.
136	type QuiescerLive struct {
137	cfg QuiescerCfg
138
139	// log is a quiescer-scoped logging instance.
140	log btclog.Logger
141
142	// localInit indicates whether our path through this state machine was
143	// initiated by our node. This can be true or false independently of
144	// remoteInit.
145	localInit bool
146
147	// remoteInit indicates whether we received Stfu from our peer where the
148	// message indicated that the remote node believes it was the initiator.
149	// This can be true or false independently of localInit.
150	remoteInit bool
151
152	// sent tracks whether or not we have emitted Stfu for sending.
153	sent bool
154
155	// received tracks whether or not we have received Stfu from our peer.
156	received bool
157
158	// activeQuiescenceRequest is a possibly None Request that we should
159	// resolve when we complete quiescence.
160	activeQuiescenceReq fn.Option[StfuReq]
161
162	// resumeQueue is a slice of hooks that will be called when the quiescer
163	// is resumed. These are actions that needed to be deferred while the
164	// channel was quiescent.
165	resumeQueue []func()
166
167	// timeoutTimer is a field that is used to hold onto the timeout job
168	// when we reach quiescence.
169	timeoutTimer *time.Timer
170
171	sync.RWMutex
172	}
173
174	// NewQuiescer creates a new quiescer for the given channel.
175	func NewQuiescer(cfg QuiescerCfg) Quiescer {	3✔
176	logPrefix := fmt.Sprintf("Quiescer(%v):", cfg.chanID)	3✔
177		3✔
178	return &QuiescerLive{	3✔
179	cfg: cfg,	3✔
180	log: log.WithPrefix(logPrefix),	3✔
181	}	3✔
182	}	3✔
183
184	// RecvStfu is called when we receive an Stfu message from the remote.
185	func (q *QuiescerLive) RecvStfu(msg lnwire.Stfu) error {	3✔
186	q.Lock()	3✔
187	defer q.Unlock()	3✔
188		3✔
189	return q.recvStfu(msg)	3✔
190	}	3✔
191
192	// recvStfu is called when we receive an Stfu message from the remote.
193	func (q *QuiescerLive) recvStfu(msg lnwire.Stfu) error {	3✔
194	// At the time of this writing, this check that we have already received	3✔
195	// an Stfu is not strictly necessary, according to the specification.	3✔
196	// However, it is fishy if we do and it is unclear how we should handle	3✔
197	// such a case so we will err on the side of caution.	3✔
198	if q.received {	3✔
UNCOV 199	return fmt.Errorf("%w for channel %v", ErrStfuAlreadyRcvd,	×
UNCOV 200	q.cfg.chanID)	×
UNCOV 201	}	×
202
203	// We need to check that the Stfu we are receiving is valid.
204	if !q.sent && !msg.Initiator {	3✔
205	return fmt.Errorf("%w for channel %v", ErrInvalidStfu,	×
206	q.cfg.chanID)	×
207	}	×
208
209	if !q.canRecvStfu() {	3✔
210	return fmt.Errorf("%w for channel %v", ErrPendingRemoteUpdates,	×
211	q.cfg.chanID)	×
212	}	×
213
214	q.received = true	3✔
215		3✔
216	// If the remote party sets the initiator bit to true then we will	3✔
217	// remember that they are making a claim to the initiator role. This	3✔
218	// does not necessarily mean they will get it, though.	3✔
219	q.remoteInit = msg.Initiator	3✔
220		3✔
221	// Since we just received an Stfu, we may have a newly quiesced state.	3✔
222	// If so, we will try to resolve any outstanding StfuReqs.	3✔
223	q.tryResolveStfuReq()	3✔
224		3✔
225	if q.isQuiescent() {	6✔
226	q.startTimeout()	3✔
227	}	3✔
228
229	return nil	3✔
230	}
231
232	// MakeStfu is called when we are ready to send an Stfu message. It returns the
233	// Stfu message to be sent.
UNCOV 234	func (q *QuiescerLive) MakeStfu() fn.Result[lnwire.Stfu] {	×
UNCOV 235	q.RLock()	×
UNCOV 236	defer q.RUnlock()	×
UNCOV 237		×
UNCOV 238	return q.makeStfu()	×
UNCOV 239	}	×
240
241	// makeStfu is called when we are ready to send an Stfu message. It returns the
242	// Stfu message to be sent.
243	func (q *QuiescerLive) makeStfu() fn.Result[lnwire.Stfu] {	3✔
244	if q.sent {	3✔
UNCOV 245	return fn.Errf[lnwire.Stfu]("%w for channel %v",	×
UNCOV 246	ErrStfuAlreadySent, q.cfg.chanID)	×
UNCOV 247	}	×
248
249	if !q.canSendStfu() {	3✔
250	return fn.Errf[lnwire.Stfu]("%w for channel %v",	×
251	ErrPendingLocalUpdates, q.cfg.chanID)	×
252	}	×
253
254	stfu := lnwire.Stfu{	3✔
255	ChanID: q.cfg.chanID,	3✔
256	Initiator: q.localInit,	3✔
257	}	3✔
258		3✔
259	return fn.Ok(stfu)	3✔
260	}
261
262	// OweStfu returns true if we owe the other party an Stfu. We owe the remote an
263	// Stfu when we have received but not yet sent an Stfu, or we are the initiator
264	// but have not yet sent an Stfu.
265	func (q *QuiescerLive) OweStfu() bool {	×
266	q.RLock()	×
267	defer q.RUnlock()	×
268		×
269	return q.oweStfu()	×
270	}	×
271
272	// oweStfu returns true if we owe the other party an Stfu. We owe the remote an
273	// Stfu when we have received but not yet sent an Stfu, or we are the initiator
274	// but have not yet sent an Stfu.
275	func (q *QuiescerLive) oweStfu() bool {	3✔
276	return (q.received \|\| q.localInit) && !q.sent	3✔
277	}	3✔
278
279	// NeedStfu returns true if the remote owes us an Stfu. They owe us an Stfu when
280	// we have sent but not yet received an Stfu.
UNCOV 281	func (q *QuiescerLive) NeedStfu() bool {	×
UNCOV 282	q.RLock()	×
UNCOV 283	defer q.RUnlock()	×
UNCOV 284		×
UNCOV 285	return q.needStfu()	×
UNCOV 286	}	×
287
288	// needStfu returns true if the remote owes us an Stfu. They owe us an Stfu when
289	// we have sent but not yet received an Stfu.
UNCOV 290	func (q *QuiescerLive) needStfu() bool {	×
UNCOV 291	q.RLock()	×
UNCOV 292	defer q.RUnlock()	×
UNCOV 293		×
UNCOV 294	return q.sent && !q.received	×
UNCOV 295	}	×
296
297	// IsQuiescent returns true if the state machine has been driven all the way to
298	// completion. If this returns true, processes that depend on channel quiescence
299	// may proceed.
UNCOV 300	func (q *QuiescerLive) IsQuiescent() bool {	×
UNCOV 301	q.RLock()	×
UNCOV 302	defer q.RUnlock()	×
UNCOV 303		×
UNCOV 304	return q.isQuiescent()	×
UNCOV 305	}	×
306
307	// isQuiescent returns true if the state machine has been driven all the way to
308	// completion. If this returns true, processes that depend on channel quiescence
309	// may proceed.
310	func (q *QuiescerLive) isQuiescent() bool {	3✔
311	return q.sent && q.received	3✔
312	}	3✔
313
314	// QuiescenceInitiator determines which ChannelParty is the initiator of
315	// quiescence for the purposes of downstream protocols. If the channel is not
316	// currently quiescent, this method will return ErrNoQuiescenceInitiator.
UNCOV 317	func (q *QuiescerLive) QuiescenceInitiator() fn.Result[lntypes.ChannelParty] {	×
UNCOV 318	q.RLock()	×
UNCOV 319	defer q.RUnlock()	×
UNCOV 320		×
UNCOV 321	return q.quiescenceInitiator()	×
UNCOV 322	}	×
323
324	// quiescenceInitiator determines which ChannelParty is the initiator of
325	// quiescence for the purposes of downstream protocols. If the channel is not
326	// currently quiescent, this method will return ErrNoQuiescenceInitiator.
327	func (q *QuiescerLive) quiescenceInitiator() fn.Result[lntypes.ChannelParty] {	3✔
328	switch {	3✔
UNCOV 329	case !q.isQuiescent():	×
UNCOV 330	return fn.Err[lntypes.ChannelParty](ErrNoQuiescenceInitiator)	×
331
UNCOV 332	case q.localInit && q.remoteInit:	×
UNCOV 333	// In the case of a tie, the channel initiator wins.	×
UNCOV 334	return fn.Ok(q.cfg.channelInitiator)	×
335
336	case q.localInit:	3✔
337	return fn.Ok(lntypes.Local)	3✔
338
UNCOV 339	case q.remoteInit:	×
UNCOV 340	return fn.Ok(lntypes.Remote)	×
341	}
342
343	// unreachable
344	return fn.Err[lntypes.ChannelParty](ErrNoQuiescenceInitiator)	×
345	}
346
347	// CanSendUpdates returns true if we haven't yet sent an Stfu which would mark
348	// the end of our ability to send updates.
349	func (q *QuiescerLive) CanSendUpdates() bool {	3✔
350	q.RLock()	3✔
351	defer q.RUnlock()	3✔
352		3✔
353	return q.canSendUpdates()	3✔
354	}	3✔
355
356	// canSendUpdates returns true if we haven't yet sent an Stfu which would mark
357	// the end of our ability to send updates.
358	func (q *QuiescerLive) canSendUpdates() bool {	3✔
359	return !q.sent && !q.localInit	3✔
360	}	3✔
361
362	// CanRecvUpdates returns true if we haven't yet received an Stfu which would
363	// mark the end of the remote's ability to send updates.
364	func (q *QuiescerLive) CanRecvUpdates() bool {	3✔
365	q.RLock()	3✔
366	defer q.RUnlock()	3✔
367		3✔
368	return q.canRecvUpdates()	3✔
369	}	3✔
370
371	// canRecvUpdates returns true if we haven't yet received an Stfu which would
372	// mark the end of the remote's ability to send updates.
373	func (q *QuiescerLive) canRecvUpdates() bool {	3✔
374	return !q.received	3✔
375	}	3✔
376
377	// CanSendStfu returns true if we can send an Stfu.
378	func (q *QuiescerLive) CanSendStfu(numPendingLocalUpdates uint64) bool {	×
379	q.RLock()	×
380	defer q.RUnlock()	×
381		×
382	return q.canSendStfu()	×
383	}	×
384
385	// canSendStfu returns true if we can send an Stfu.
386	func (q *QuiescerLive) canSendStfu() bool {	3✔
387	return !q.sent	3✔
388	}	3✔
389
390	// CanRecvStfu returns true if we can receive an Stfu.
391	func (q *QuiescerLive) CanRecvStfu() bool {	×
392	q.RLock()	×
393	defer q.RUnlock()	×
394		×
395	return q.canRecvStfu()	×
396	}	×
397
398	// canRecvStfu returns true if we can receive an Stfu.
399	func (q *QuiescerLive) canRecvStfu() bool {	3✔
400	return !q.received	3✔
401	}	3✔
402
403	// SendOwedStfu sends Stfu if it owes one. It returns an error if the state
404	// machine is in an invalid state.
405	func (q *QuiescerLive) SendOwedStfu() error {	3✔
406	q.Lock()	3✔
407	defer q.Unlock()	3✔
408		3✔
409	return q.sendOwedStfu()	3✔
410	}	3✔
411
412	// sendOwedStfu sends Stfu if it owes one. It returns an error if the state
413	// machine is in an invalid state.
414	func (q *QuiescerLive) sendOwedStfu() error {	3✔
415	if !q.oweStfu() \|\| !q.canSendStfu() {	6✔
416	return nil	3✔
417	}	3✔
418
419	err := q.makeStfu().Sink(q.cfg.sendMsg)	3✔
420		3✔
421	if err == nil {	6✔
422	q.sent = true	3✔
423		3✔
424	// Since we just sent an Stfu, we may have a newly quiesced	3✔
425	// state. If so, we will try to resolve any outstanding	3✔
426	// StfuReqs.	3✔
427	q.tryResolveStfuReq()	3✔
428		3✔
429	if q.isQuiescent() {	6✔
430	q.startTimeout()	3✔
431	}	3✔
432	}
433
434	return err	3✔
435	}
436
437	// TryResolveStfuReq attempts to resolve the active quiescence request if the
438	// state machine has reached a quiescent state.
439	func (q *QuiescerLive) TryResolveStfuReq() {	×
440	q.Lock()	×
441	defer q.Unlock()	×
442		×
443	q.tryResolveStfuReq()	×
444	}	×
445
446	// tryResolveStfuReq attempts to resolve the active quiescence request if the
447	// state machine has reached a quiescent state.
448	func (q *QuiescerLive) tryResolveStfuReq() {	3✔
449	q.activeQuiescenceReq.WhenSome(	3✔
450	func(req StfuReq) {	6✔
451	if q.isQuiescent() {	6✔
452	req.Resolve(q.quiescenceInitiator())	3✔
453	q.activeQuiescenceReq = fn.None[StfuReq]()	3✔
454	}	3✔
455	},
456	)
457	}
458
459	// InitStfu instructs the quiescer that we intend to begin a quiescence
460	// negotiation where we are the initiator. We don't yet send stfu yet because
461	// we need to wait for the link to give us a valid opportunity to do so.
462	func (q *QuiescerLive) InitStfu(req StfuReq) {	3✔
463	q.Lock()	3✔
464	defer q.Unlock()	3✔
465		3✔
466	q.initStfu(req)	3✔
467	}	3✔
468
469	// initStfu instructs the quiescer that we intend to begin a quiescence
470	// negotiation where we are the initiator. We don't yet send stfu yet because
471	// we need to wait for the link to give us a valid opportunity to do so.
472	func (q *QuiescerLive) initStfu(req StfuReq) {	3✔
473	if q.localInit {	3✔
UNCOV 474	req.Resolve(fn.Errf[lntypes.ChannelParty](	×
UNCOV 475	"quiescence already requested",	×
UNCOV 476	))	×
UNCOV 477		×
UNCOV 478	return	×
UNCOV 479	}	×
480
481	q.localInit = true	3✔
482	q.activeQuiescenceReq = fn.Some(req)	3✔
483	}
484
485	// OnResume accepts a no return closure that will run when the quiescer is
486	// resumed.
UNCOV 487	func (q *QuiescerLive) OnResume(hook func()) {	×
UNCOV 488	q.Lock()	×
UNCOV 489	defer q.Unlock()	×
UNCOV 490		×
UNCOV 491	q.onResume(hook)	×
UNCOV 492	}	×
493
494	// onResume accepts a no return closure that will run when the quiescer is
495	// resumed.
UNCOV 496	func (q *QuiescerLive) onResume(hook func()) {	×
UNCOV 497	q.resumeQueue = append(q.resumeQueue, hook)	×
UNCOV 498	}	×
499
500	// Resume runs all of the deferred actions that have accumulated while the
501	// channel has been quiescent and then resets the quiescer state to its initial
502	// state.
UNCOV 503	func (q *QuiescerLive) Resume() {	×
UNCOV 504	q.Lock()	×
UNCOV 505	defer q.Unlock()	×
UNCOV 506		×
UNCOV 507	q.resume()	×
UNCOV 508	}	×
509
510	// resume runs all of the deferred actions that have accumulated while the
511	// channel has been quiescent and then resets the quiescer state to its initial
512	// state.
UNCOV 513	func (q *QuiescerLive) resume() {	×
UNCOV 514	q.log.Debug("quiescence terminated, resuming htlc traffic")	×
UNCOV 515		×
UNCOV 516	// since we are resuming we want to cancel the quiescence timeout	×
UNCOV 517	// action.	×
UNCOV 518	q.cancelTimeout()	×
UNCOV 519		×
UNCOV 520	for _, hook := range q.resumeQueue {	×
UNCOV 521	hook()	×
UNCOV 522	}	×
UNCOV 523	q.localInit = false	×
UNCOV 524	q.remoteInit = false	×
UNCOV 525	q.sent = false	×
UNCOV 526	q.received = false	×
UNCOV 527	q.resumeQueue = nil	×
528	}
529
530	// startTimeout starts the timeout function that fires if the quiescer remains
531	// in a quiesced state for too long. If this function is called multiple times
532	// only the last one will have an effect.
533	func (q *QuiescerLive) startTimeout() {	3✔
534	if q.cfg.onTimeout == nil {	3✔
UNCOV 535	return	×
UNCOV 536	}	×
537
538	old := q.timeoutTimer	3✔
539		3✔
540	q.timeoutTimer = time.AfterFunc(q.cfg.timeoutDuration, q.cfg.onTimeout)	3✔
541		3✔
542	if old != nil {	3✔
543	old.Stop()	×
544	}	×
545	}
546
547	// cancelTimeout cancels the timeout function that would otherwise fire if the
548	// quiescer remains in a quiesced state too long. If this function is called
549	// before startTimeout or after another call to cancelTimeout, the effect will
550	// be a noop.
UNCOV 551	func (q *QuiescerLive) cancelTimeout() {	×
UNCOV 552	if q.timeoutTimer != nil {	×
UNCOV 553	q.timeoutTimer.Stop()	×
UNCOV 554	q.timeoutTimer = nil	×
UNCOV 555	}	×
556	}
557
558	type quiescerNoop struct{}
559
560	var _ Quiescer = (*quiescerNoop)(nil)
561
562	func (q *quiescerNoop) InitStfu(req StfuReq) {	×
563	req.Resolve(fn.Errf[lntypes.ChannelParty]("quiescence not supported"))	×
564	}	×
565	func (q *quiescerNoop) RecvStfu(_ lnwire.Stfu) error { return nil }	×
566	func (q *quiescerNoop) CanRecvUpdates() bool { return true }	×
567	func (q *quiescerNoop) CanSendUpdates() bool { return true }	×
568	func (q *quiescerNoop) SendOwedStfu() error { return nil }	×
569	func (q *quiescerNoop) IsQuiescent() bool { return false }	×
570	func (q *quiescerNoop) OnResume(hook func()) { hook() }	×
571	func (q *quiescerNoop) Resume() {}	×
572	func (q *quiescerNoop) QuiescenceInitiator() fn.Result[lntypes.ChannelParty] {	×
573	return fn.Err[lntypes.ChannelParty](ErrNoQuiescenceInitiator)	×
574	}	×

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