15858991938

Committed 24 Jun 2025 06:51PM UTC coverage: 55.808% (-2.4%) from 58.173%

Build # 15858991938

Build Type

Pull #9148

github

Committed by

web-flow

Commit Message

Merge 0e921d6a5 into 29ff13d83

Pull Request Pull Request #9148: DynComms [2/n]: lnwire: add authenticated wire messages for Dyn*

Run Details

232 of 267 new or added lines in 5 files covered. (86.89%)

24606 existing lines in 281 files now uncovered.

108380 of 194201 relevant lines covered (55.81%)

22488.12 hits per line

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

74.23

/peer/ping_manager.go

package peer

import (
        "errors"
        "fmt"
        "sync"
        "sync/atomic"
        "time"

        "github.com/lightningnetwork/lnd/fn/v2"
        "github.com/lightningnetwork/lnd/lnwire"
)

// PingManagerConfig is a structure containing various parameters that govern
// how the PingManager behaves.
type PingManagerConfig struct {
        // NewPingPayload is a closure that returns the payload to be packaged
        // in the Ping message.
        NewPingPayload func() []byte

        // NewPongSize is a closure that returns a random value between
        // [0, lnwire.MaxPongBytes]. This random value helps to more effectively
        // pair Pong messages with Ping.
        NewPongSize func() uint16

        // IntervalDuration is the Duration between attempted pings.
        IntervalDuration time.Duration

        // TimeoutDuration is the Duration we wait before declaring a ping
        // attempt failed.
        TimeoutDuration time.Duration

        // SendPing is a closure that is responsible for sending the Ping
        // message out to our peer
        SendPing func(ping *lnwire.Ping)

        // OnPongFailure is a closure that is responsible for executing the
        // logic when a Pong message is either late or does not match our
        // expectations for that Pong
        OnPongFailure func(failureReason error, timeWaitedForPong time.Duration,
                lastKnownRTT time.Duration)
}

// PingManager is a structure that is designed to manage the internal state
// of the ping pong lifecycle with the remote peer. We assume there is only one
// ping outstanding at once.
//
// NOTE: This structure MUST be initialized with NewPingManager.
type PingManager struct {
        cfg *PingManagerConfig

        // pingTime is a rough estimate of the RTT (round-trip-time) between us
        // and the connected peer.
        // To be used atomically.
        // TODO(roasbeef): also use a WMA or EMA?
        pingTime atomic.Pointer[time.Duration]

        // pingLastSend is the time when we sent our last ping message.
        // To be used atomically.
        pingLastSend *time.Time

        // outstandingPongSize is the current size of the requested pong
        // payload.  This value can only validly range from [0,65531]. Any
        // value < 0 is interpreted as if there is no outstanding ping message.
        outstandingPongSize int32

        // pingTicker is a pointer to a Ticker that fires on every ping
        // interval.
        pingTicker *time.Ticker

        // pingTimeout is a Timer that will fire when we want to time out a
        // ping
        pingTimeout *time.Timer

        // pongChan is the channel on which the pingManager will write Pong
        // messages it is evaluating
        pongChan chan *lnwire.Pong

        started sync.Once
        stopped sync.Once

        quit chan struct{}
        wg   sync.WaitGroup
}

// NewPingManager constructs a pingManager in a valid state. It must be started
// before it does anything useful, though.
func NewPingManager(cfg *PingManagerConfig) *PingManager {
        m := PingManager{
                cfg:                 cfg,
                outstandingPongSize: -1,
                pongChan:            make(chan *lnwire.Pong, 1),
                quit:                make(chan struct{}),
        }

        return &m
}

// Start launches the primary goroutine that is owned by the pingManager.
func (m *PingManager) Start() error {
        var err error
        m.started.Do(func() {
                m.pingTicker = time.NewTicker(m.cfg.IntervalDuration)
                m.pingTimeout = time.NewTimer(0)

                m.wg.Add(1)
                go m.pingHandler()
        })

        return err
}

// getLastRTT safely retrieves the last known RTT, returning 0 if none exists.
func (m *PingManager) getLastRTT() time.Duration {
        rttPtr := m.pingTime.Load()
        if rttPtr == nil {
                return 0
        }

        return *rttPtr
}

// pendingPingWait calculates the time waited since the last ping was sent. If
// no ping time is reported, None is returned. defaultDuration.
func (m *PingManager) pendingPingWait() fn.Option[time.Duration] {
        if m.pingLastSend != nil {
                return fn.Some(time.Since(*m.pingLastSend))
        }

        return fn.None[time.Duration]()
}

// pingHandler is the main goroutine responsible for enforcing the ping/pong
// protocol.
func (m *PingManager) pingHandler() {
        defer m.wg.Done()
        defer m.pingTimeout.Stop()

        // Ensure that the pingTimeout channel is empty.
        if !m.pingTimeout.Stop() {
                <-m.pingTimeout.C
        }

        // Because we don't know if the OnPingFailure callback actually
        // disconnects a peer (dependent on user config), we should never return
        // from this loop unless the ping manager is stopped explicitly (which
        // happens on disconnect).
        for {
                select {
                case <-m.pingTicker.C:
                        // If this occurs it means that the new ping cycle has
                        // begun while there is still an outstanding ping
                        // awaiting a pong response.  This should never occur,
                        // but if it does, it implies a timeout.
                        if m.outstandingPongSize >= 0 {
                                // Ping was outstanding, meaning it timed out by
                                // the arrival of the next ping interval.
                                timeWaited := m.pendingPingWait().UnwrapOr(
                                        m.cfg.IntervalDuration,
                                )
                                lastRTT := m.getLastRTT()

                                m.cfg.OnPongFailure(
                                        errors.New("ping timed "+
                                                "out by next interval"),
                                        timeWaited, lastRTT,
                                )

                                m.resetPingState()
                        }

                        pongSize := m.cfg.NewPongSize()
                        ping := &lnwire.Ping{
                                NumPongBytes: pongSize,
                                PaddingBytes: m.cfg.NewPingPayload(),
                        }

                        // Set up our bookkeeping for the new Ping.
                        if err := m.setPingState(pongSize); err != nil {
                                // This is an internal error related to timer
                                // reset. Pass it to OnPongFailure as it's
                                // critical. Current and last RTT are not
                                // directly applicable here.
                                m.cfg.OnPongFailure(err, 0, 0)

                                m.resetPingState()

                                continue
                        }

                        m.cfg.SendPing(ping)

                case <-m.pingTimeout.C:
                        timeWaited := m.pendingPingWait().UnwrapOr(
                                m.cfg.TimeoutDuration,
                        )
                        lastRTT := m.getLastRTT()

                        m.cfg.OnPongFailure(
                                errors.New("timeout while waiting for "+
                                        "pong response"),
                                timeWaited, lastRTT,
                        )

                        m.resetPingState()

                case pong := <-m.pongChan:
                        pongSize := int32(len(pong.PongBytes))

                        // Save off values we are about to override when we call
                        // resetPingState.
                        expected := m.outstandingPongSize
                        lastPingTime := m.pingLastSend

                        // This is an unexpected pong, we'll continue.
                        if lastPingTime == nil {
                                continue
                        }

                        actualRTT := time.Since(*lastPingTime)

                        // If the pong we receive doesn't match the ping we sent
                        // out, then we fail out.
                        if pongSize != expected {
                                e := fmt.Errorf("pong response does not match "+
                                        "expected size. Expected: %d, Got: %d",
                                        expected, pongSize)

                                lastRTT := m.getLastRTT()
                                m.cfg.OnPongFailure(e, actualRTT, lastRTT)

                                m.resetPingState()

                                continue
                        }

                        // Pong is good, update RTT and reset state.
                        m.pingTime.Store(&actualRTT)
                        m.resetPingState()

                case <-m.quit:
                        return
                }
        }
}

// Stop interrupts the goroutines that the PingManager owns.
func (m *PingManager) Stop() {
        if m.pingTicker == nil {
                return
        }

        m.stopped.Do(func() {
                close(m.quit)
                m.wg.Wait()

                m.pingTicker.Stop()
                m.pingTimeout.Stop()
        })
}

// setPingState is a private method to keep track of all of the fields we need
// to set when we send out a Ping.
func (m *PingManager) setPingState(pongSize uint16) error {
        t := time.Now()
        m.pingLastSend = &t
        m.outstandingPongSize = int32(pongSize)
        if m.pingTimeout.Reset(m.cfg.TimeoutDuration) {
                return fmt.Errorf(
                        "impossible: ping timeout reset when already active",
                )
        }

        return nil
}

// resetPingState is a private method that resets all of the bookkeeping that
// is tracking a currently outstanding Ping.
func (m *PingManager) resetPingState() {
        m.pingLastSend = nil
        m.outstandingPongSize = -1

        if !m.pingTimeout.Stop() {
                select {
                case <-m.pingTimeout.C:
                default:
                }
        }
}

// GetPingTimeMicroSeconds reports back the RTT calculated by the pingManager.
func (m *PingManager) GetPingTimeMicroSeconds() int64 {
        rtt := m.pingTime.Load()

        if rtt == nil {
                return -1
        }

        return rtt.Microseconds()
}

// ReceivedPong is called to evaluate a Pong message against the expectations
// we have for it. It will cause the PingManager to invoke the supplied
// OnPongFailure function if the Pong argument supplied violates expectations.
func (m *PingManager) ReceivedPong(msg *lnwire.Pong) {
        select {
        case m.pongChan <- msg:
        case <-m.quit:
        }
}

1	package peer
2
3	import (
4	"errors"
5	"fmt"
6	"sync"
7	"sync/atomic"
8	"time"
9
10	"github.com/lightningnetwork/lnd/fn/v2"
11	"github.com/lightningnetwork/lnd/lnwire"
12	)
13
14	// PingManagerConfig is a structure containing various parameters that govern
15	// how the PingManager behaves.
16	type PingManagerConfig struct {
17	// NewPingPayload is a closure that returns the payload to be packaged
18	// in the Ping message.
19	NewPingPayload func() []byte
20
21	// NewPongSize is a closure that returns a random value between
22	// [0, lnwire.MaxPongBytes]. This random value helps to more effectively
23	// pair Pong messages with Ping.
24	NewPongSize func() uint16
25
26	// IntervalDuration is the Duration between attempted pings.
27	IntervalDuration time.Duration
28
29	// TimeoutDuration is the Duration we wait before declaring a ping
30	// attempt failed.
31	TimeoutDuration time.Duration
32
33	// SendPing is a closure that is responsible for sending the Ping
34	// message out to our peer
35	SendPing func(ping *lnwire.Ping)
36
37	// OnPongFailure is a closure that is responsible for executing the
38	// logic when a Pong message is either late or does not match our
39	// expectations for that Pong
40	OnPongFailure func(failureReason error, timeWaitedForPong time.Duration,
41	lastKnownRTT time.Duration)
42	}
43
44	// PingManager is a structure that is designed to manage the internal state
45	// of the ping pong lifecycle with the remote peer. We assume there is only one
46	// ping outstanding at once.
47	//
48	// NOTE: This structure MUST be initialized with NewPingManager.
49	type PingManager struct {
50	cfg *PingManagerConfig
51
52	// pingTime is a rough estimate of the RTT (round-trip-time) between us
53	// and the connected peer.
54	// To be used atomically.
55	// TODO(roasbeef): also use a WMA or EMA?
56	pingTime atomic.Pointer[time.Duration]
57
58	// pingLastSend is the time when we sent our last ping message.
59	// To be used atomically.
60	pingLastSend *time.Time
61
62	// outstandingPongSize is the current size of the requested pong
63	// payload. This value can only validly range from [0,65531]. Any
64	// value < 0 is interpreted as if there is no outstanding ping message.
65	outstandingPongSize int32
66
67	// pingTicker is a pointer to a Ticker that fires on every ping
68	// interval.
69	pingTicker *time.Ticker
70
71	// pingTimeout is a Timer that will fire when we want to time out a
72	// ping
73	pingTimeout *time.Timer
74
75	// pongChan is the channel on which the pingManager will write Pong
76	// messages it is evaluating
77	pongChan chan *lnwire.Pong
78
79	started sync.Once
80	stopped sync.Once
81
82	quit chan struct{}
83	wg sync.WaitGroup
84	}
85
86	// NewPingManager constructs a pingManager in a valid state. It must be started
87	// before it does anything useful, though.
88	func NewPingManager(cfg PingManagerConfig) PingManager {	28✔
89	m := PingManager{	28✔
90	cfg: cfg,	28✔
91	outstandingPongSize: -1,	28✔
92	pongChan: make(chan *lnwire.Pong, 1),	28✔
93	quit: make(chan struct{}),	28✔
94	}	28✔
95		28✔
96	return &m	28✔
97	}	28✔
98
99	// Start launches the primary goroutine that is owned by the pingManager.
100	func (m *PingManager) Start() error {	6✔
101	var err error	6✔
102	m.started.Do(func() {	12✔
103	m.pingTicker = time.NewTicker(m.cfg.IntervalDuration)	6✔
104	m.pingTimeout = time.NewTimer(0)	6✔
105		6✔
106	m.wg.Add(1)	6✔
107	go m.pingHandler()	6✔
108	})	6✔
109
110	return err	6✔
111	}
112
113	// getLastRTT safely retrieves the last known RTT, returning 0 if none exists.
114	func (m *PingManager) getLastRTT() time.Duration {	2✔
115	rttPtr := m.pingTime.Load()	2✔
116	if rttPtr == nil {	4✔
117	return 0	2✔
118	}	2✔
119
120	return *rttPtr	×
121	}
122
123	// pendingPingWait calculates the time waited since the last ping was sent. If
124	// no ping time is reported, None is returned. defaultDuration.
125	func (m *PingManager) pendingPingWait() fn.Option[time.Duration] {	1✔
126	if m.pingLastSend != nil {	2✔
127	return fn.Some(time.Since(*m.pingLastSend))	1✔
128	}	1✔
129
130	return fn.None[time.Duration]()	×
131	}
132
133	// pingHandler is the main goroutine responsible for enforcing the ping/pong
134	// protocol.
135	func (m *PingManager) pingHandler() {	6✔
136	defer m.wg.Done()	6✔
137	defer m.pingTimeout.Stop()	6✔
138		6✔
139	// Ensure that the pingTimeout channel is empty.	6✔
140	if !m.pingTimeout.Stop() {	6✔
141	<-m.pingTimeout.C	×
142	}	×
143
144	// Because we don't know if the OnPingFailure callback actually
145	// disconnects a peer (dependent on user config), we should never return
146	// from this loop unless the ping manager is stopped explicitly (which
147	// happens on disconnect).
148	for {	20✔
149	select {	14✔
150	case <-m.pingTicker.C:	4✔
151	// If this occurs it means that the new ping cycle has	4✔
152	// begun while there is still an outstanding ping	4✔
153	// awaiting a pong response. This should never occur,	4✔
154	// but if it does, it implies a timeout.	4✔
155	if m.outstandingPongSize >= 0 {	4✔
156	// Ping was outstanding, meaning it timed out by	×
157	// the arrival of the next ping interval.	×
158	timeWaited := m.pendingPingWait().UnwrapOr(	×
159	m.cfg.IntervalDuration,	×
160	)	×
161	lastRTT := m.getLastRTT()	×
162		×
163	m.cfg.OnPongFailure(	×
164	errors.New("ping timed "+	×
165	"out by next interval"),	×
166	timeWaited, lastRTT,	×
167	)	×
168		×
169	m.resetPingState()	×
170	}	×
171
172	pongSize := m.cfg.NewPongSize()	4✔
173	ping := &lnwire.Ping{	4✔
174	NumPongBytes: pongSize,	4✔
175	PaddingBytes: m.cfg.NewPingPayload(),	4✔
176	}	4✔
177		4✔
178	// Set up our bookkeeping for the new Ping.	4✔
179	if err := m.setPingState(pongSize); err != nil {	4✔
180	// This is an internal error related to timer	×
181	// reset. Pass it to OnPongFailure as it's	×
182	// critical. Current and last RTT are not	×
183	// directly applicable here.	×
184	m.cfg.OnPongFailure(err, 0, 0)	×
185		×
186	m.resetPingState()	×
187		×
188	continue	×
189	}
190
191	m.cfg.SendPing(ping)	4✔
192
193	case <-m.pingTimeout.C:	1✔
194	timeWaited := m.pendingPingWait().UnwrapOr(	1✔
195	m.cfg.TimeoutDuration,	1✔
196	)	1✔
197	lastRTT := m.getLastRTT()	1✔
198		1✔
199	m.cfg.OnPongFailure(	1✔
200	errors.New("timeout while waiting for "+	1✔
201	"pong response"),	1✔
202	timeWaited, lastRTT,	1✔
203	)	1✔
204		1✔
205	m.resetPingState()	1✔
206
207	case pong := <-m.pongChan:	3✔
208	pongSize := int32(len(pong.PongBytes))	3✔
209		3✔
210	// Save off values we are about to override when we call	3✔
211	// resetPingState.	3✔
212	expected := m.outstandingPongSize	3✔
213	lastPingTime := m.pingLastSend	3✔
214		3✔
215	// This is an unexpected pong, we'll continue.	3✔
216	if lastPingTime == nil {	3✔
217	continue	×
218	}
219
220	actualRTT := time.Since(*lastPingTime)	3✔
221		3✔
222	// If the pong we receive doesn't match the ping we sent	3✔
223	// out, then we fail out.	3✔
224	if pongSize != expected {	4✔
225	e := fmt.Errorf("pong response does not match "+	1✔
226	"expected size. Expected: %d, Got: %d",	1✔
227	expected, pongSize)	1✔
228		1✔
229	lastRTT := m.getLastRTT()	1✔
230	m.cfg.OnPongFailure(e, actualRTT, lastRTT)	1✔
231		1✔
232	m.resetPingState()	1✔
233		1✔
234	continue	1✔
235	}
236
237	// Pong is good, update RTT and reset state.
238	m.pingTime.Store(&actualRTT)	2✔
239	m.resetPingState()	2✔
240
241	case <-m.quit:	3✔
242	return	3✔
243	}
244	}
245	}
246
247	// Stop interrupts the goroutines that the PingManager owns.
248	func (m *PingManager) Stop() {	4✔
249	if m.pingTicker == nil {	5✔
250	return	1✔
251	}	1✔
252
253	m.stopped.Do(func() {	6✔
254	close(m.quit)	3✔
255	m.wg.Wait()	3✔
256		3✔
257	m.pingTicker.Stop()	3✔
258	m.pingTimeout.Stop()	3✔
259	})	3✔
260	}
261
262	// setPingState is a private method to keep track of all of the fields we need
263	// to set when we send out a Ping.
264	func (m *PingManager) setPingState(pongSize uint16) error {	4✔
265	t := time.Now()	4✔
266	m.pingLastSend = &t	4✔
267	m.outstandingPongSize = int32(pongSize)	4✔
268	if m.pingTimeout.Reset(m.cfg.TimeoutDuration) {	4✔
269	return fmt.Errorf(	×
270	"impossible: ping timeout reset when already active",	×
271	)	×
272	}	×
273
274	return nil	4✔
275	}
276
277	// resetPingState is a private method that resets all of the bookkeeping that
278	// is tracking a currently outstanding Ping.
279	func (m *PingManager) resetPingState() {	4✔
280	m.pingLastSend = nil	4✔
281	m.outstandingPongSize = -1	4✔
282		4✔
283	if !m.pingTimeout.Stop() {	5✔
284	select {	1✔
285	case <-m.pingTimeout.C:	×
286	default:	1✔
287	}
288	}
289	}
290
291	// GetPingTimeMicroSeconds reports back the RTT calculated by the pingManager.
UNCOV 292	func (m *PingManager) GetPingTimeMicroSeconds() int64 {	×
UNCOV 293	rtt := m.pingTime.Load()	×
UNCOV 294		×
UNCOV 295	if rtt == nil {	×
UNCOV 296	return -1	×
UNCOV 297	}	×
298
299	return rtt.Microseconds()	×
300	}
301
302	// ReceivedPong is called to evaluate a Pong message against the expectations
303	// we have for it. It will cause the PingManager to invoke the supplied
304	// OnPongFailure function if the Pong argument supplied violates expectations.
305	func (m PingManager) ReceivedPong(msg lnwire.Pong) {	3✔
306	select {	3✔
307	case m.pongChan <- msg:	3✔
308	case <-m.quit:	×
309	}
310	}

lightningnetwork / lnd / 15858991938

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