13980275562

Committed 20 Mar 2025 10:06PM UTC coverage: 58.6% (-10.2%) from 68.789%

Build # 13980275562

Build Type

Pull #9623

github

Committed by

web-flow

Commit Message

Merge b9b960345 into 09b674508

Pull Request Pull Request #9623: Size msg test msg

Run Details

0 of 1518 new or added lines in 42 files covered. (0.0%)

26603 existing lines in 443 files now uncovered.

96807 of 165200 relevant lines covered (58.6%)

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

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