16293817609

Committed 15 Jul 2025 12:53PM UTC coverage: 57.584% (-9.8%) from 67.338%

Build # 16293817609

Build Type

Pull #10031

github

Committed by

web-flow

Commit Message

Merge 8fd26da8b into df6c02e3a

Pull Request Pull Request #10031: Skip unnecessary disconnect and wait for disconnect to finish in shutdown

Run Details

36 of 40 new or added lines in 2 files covered. (90.0%)

28482 existing lines in 458 files now uncovered.

98681 of 171370 relevant lines covered (57.58%)

1.79 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")
)

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

lightningnetwork / lnd / 16293817609

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