multi: pong enforcement

[ProofOfKeags]

Change Description

This change introduces pong enforcement: see #3003

• Introduce subsystem for low latency querying of global blockchain state
• Remove blockchain subscription from pingHandler
• Record most recent numPongBytes on Brontide connection
• Add verification logic to readHandler
• Launch timeout function during queuing to race the verification logic in the readHandler
• Handle API misuse of CurrentChainStateTracker
• Take some action on pong failure (disconnect?)
• Add tests to verify correctness of this code

Steps to Test

make unit

Pull Request Checklist

Testing

• Your PR passes all CI checks.
• Tests covering the positive and negative (error paths) are included.
  - [ ] Bug fixes contain tests triggering the bug to prevent regressions.

Code Style and Documentation

• The change obeys the Code Documentation and Commenting guidelines, and lines wrap at 80.
• Commits follow the Ideal Git Commit Structure.
• Any new logging statements use an appropriate subsystem and logging level.
• There is a change description in the release notes, or [skip ci] in the commit message for small changes.

📝 Please see our Contribution Guidelines for further guidance.

---

This change is 

[Roasbeef]

Nice work! I dig the attention to test coverage, and overall the testability of the new sub-systems and related gorotuines. Completed an initial pass with a few comments on: style, simplifying the deps of the new sub-system, and re-using some standard library functions throughout the diff.

[Roasbeef]

Can these variables just be created by the NewChainStateTracker constructor?

[ProofOfKeags]

This allows the tracker to be started, stopped, then started again. If we place the channel construction in the constructor itself then it would complicate the logic of the Start() function or we would introduce the limitation that it cannot be started again once it is stopped. Perhaps this is OK.

[Roasbeef]

If we place the channel construction in the constructor itself then it would complicate the logic of the Start() function or we would introduce the limitation that it cannot be started again once it is stopped. Perhaps this is OK.

Gotcha, yeah usually most of these are only ever started once, as their lifetime tracks the lifetime of the peer connection. So I think it'd be safe to move all the initialization into the main constructor.

[ellemouton]

yes agreed - should be moved to main constructor 👍

[ProofOfKeags]

Should we remove the Start function then? I don't think it makes sense to do some initialization in the constructor and some initialization in a Start method

[ellemouton]

The constructor is responsible for building the object. The start method is responsible for actually doing things and kicking off goroutines.

[Roasbeef]

So we'll maximally end up here with ~1 kB/s

[ProofOfKeags]

The goal with this was to reach maximum variance to minimize accidental collisions of pings. It's possibly overkill and I can shrink it down but I decided to cover the entire valid input space.

[Roasbeef]

Reviewed 8 of 8 files at r1, all commit messages.
Reviewable status: all files reviewed, 21 unresolved discussions (waiting on @ProofOfKeags and @saubyk)

[lightninglabs-deploy]

@morehouse: review reminder
@ellemouton: review reminder
@ProofOfKeags, remember to re-request review from reviewers when ready

[ellemouton]

@ProofOfKeags - there is a race in TestPingManager (see CI)

[ellemouton]

Getting there!

Here is a proposal for how to simplify things quite a bit:

pong.patch

[ellemouton]

construction should happen in the Brontide constructor.

Then rather call pingManger.Stop() from the Brontide Disconnect method rather than passing in the quit channel. The ping manager should have its own quit channel instead.

[ProofOfKeags]

Then rather call pingManger.Stop() from the Brontide Disconnect method rather than passing in the quit channel. The ping manager should have its own quit channel instead.

Adjusted this.

construction should happen in the Brontide constructor.

While I understand that the pattern throughout the codebase is this, I want to point out the difficulty with doing that in this case.

On the input side we allocate a couple of stack variables to help with some optimizations. We can drop the optimizations, or we will have to put those stack variables into the Brontide struct and I'm reticent to further bloat the Brontide. Dropping the optimizations I'm much more amenable to.

On the output side we would need to also store the ping, pong, and fail channels on the brontide and my goal with the "receive only" and "send only" type casts here is to enlist the go type system to help make sure that these things don't get misused. It is also why I made the pingManagerLiason and the readHandler take these channels as arguments as well, to prevent those values from being made freely available to other goroutines.

While I'm sympathetic to following patterns to make code more uniform, I worry about progressively piling more and more state into the Brontide, which in this context is serving as a global namespace and I don't think it's that controversial to say that large global namespaces are bad for reasoning about code.

How do you think about this tradeoff?

[ellemouton]

On the output side we would need to also store the ping, pong, and fail channels on the brontide and my goal with the "receive only" and "send only" t

Did you take a look at the patch I sent yesterday? it doesnt require having these channels at all and I would argue that things are way more self contained with the patch implementation where no "liason" is needed

[ellemouton]

I think we can instead pass callbacks to the ping manager. See proposed patch

[ProofOfKeags]

I took a look at the proposed patch and I think it removes some of the desired structure I had from it. This was designed to take a channels first approach to things which as far as I understand is the preferred go model of thinking, and then secondly I think it's important to not pollute the Brontide struct with additional state that only pertains to parts of it, hence my choice to thread state via arguments and ensure proper message flow by downcasting channel types into their send-only and receive-only counterparts. I'm amenable to making some adjustments, but I'd like it to be understood why some of these choices were made before we throw them out.

[ellemouton]

designed to take a channels first approach to things which as far as I understand is the preferred go model of thinking

Not necessarily. Channels are nice within a subsystem when used in a "closed" way such that how the system is used from the outside doesnt affect things. With the current impl as it currently stands, there is the whole "ping manager liason" thing which imo is leaking implementation into the brontide system. I opt for keeping things contained within the ping manager

I think it's important to not pollute the Brontide struct with additional state

where is the polluted state?

[ellemouton]

With the patch I sent, you no longer need these channels.
With these channels, this subsystem (PingManager) depends on Brontide correctly reading and writing to these channels. So implementation details leak.

The patch I sent is a pattern we use often throughout the codebase where the calling system just provides some callbacks.

[ellemouton]

reposting the suggested adjusted files instead of the patch so that it easier to view and apply.

Suggested ping manager: Note the use of PingManagerConfig with call backs instead of passing back channels and depending on Brontide using them correctly:

(Ive removed all the comments for the sake of making this comment as small as possible)

package peer

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

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

type PingManagerConfig struct {
	NewPingPayload func() []byte
	NewPongSize func() uint16
	IntervalDuration time.Duration
	TimeoutDuration time.Duration
	SendPing func(ping *lnwire.Ping)
	OnNoPongReceived func()
}

type PingManager struct {
	cfg *PingManagerConfig
	pingTime atomic.Pointer[time.Duration]
	pingLastSend *time.Time
	outstandingPongSize int32
	pingTicker *time.Ticker
	pingTimeout *time.Timer
	pongChan chan *lnwire.Pong

	started sync.Once
	stopped sync.Once

	mu   sync.Mutex
	quit chan struct{}
	wg   sync.WaitGroup
}

func NewPingManager(cfg *PingManagerConfig) *PingManager {
	m := PingManager{
		cfg:                 cfg,
		outstandingPongSize: -1,
		pongChan:            make(chan *lnwire.Pong, 1),
		quit:                make(chan struct{}),
	}

	return &m
}

func (m *PingManager) ReceivedPong(msg *lnwire.Pong) {
	select {
	case m.pongChan <- msg:
	case <-m.quit:
	}

	return
}

func (m *PingManager) Start() error {
	var err error
	m.started.Do(func() {
		err = m.start()
	})

	return err
}

func (m *PingManager) start() error {
	m.pingTicker = time.NewTicker(m.cfg.IntervalDuration)

	m.pingTimeout = time.NewTimer(0)
	defer m.pingTimeout.Stop()

	if !m.pingTimeout.Stop() {
		<-m.pingTimeout.C
	}

	m.wg.Add(1)
	go func() {
		defer m.wg.Done()
		for {
			select {
			case <-m.pingTicker.C:
				if m.outstandingPongSize >= 0 {
					m.cfg.OnNoPongReceived()
					return
				}

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

				if err := m.setPingState(pongSize); err != nil {
					m.cfg.OnNoPongReceived()

					return
				}

				m.cfg.SendPing(ping)

			case <-m.pingTimeout.C:
				m.resetPingState()

				m.cfg.OnNoPongReceived()

				return

			case pong := <-m.pongChan:
				pongSize := int32(len(pong.PongBytes))
				expected := m.outstandingPongSize
				lastPing := m.pingLastSend
				m.resetPingState()
				if pongSize != expected {
					m.cfg.OnNoPongReceived()

					return
				}

				if lastPing != nil {
					rtt := time.Since(*lastPing)
					m.pingTime.Store(&rtt)
				}
			case <-m.quit:
				return
			}
		}
	}()

	return nil
}

func (m *PingManager) Stop() error {
	close(m.quit)
	m.wg.Wait()

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

	return nil
}

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
}

func (m *PingManager) resetPingState() {
	m.pingLastSend = nil
	m.outstandingPongSize = -1
	if !m.pingTimeout.Stop() {
		select {
		case <-m.pingTimeout.C:
		default:
		}
	}
}

func (m *PingManager) GetPingTimeMicroSeconds() int64 {
	rtt := m.pingTime.Load()

	if rtt == nil {
		return -1
	}

	return rtt.Microseconds()
}

Test:

package peer

import (
	"testing"
	"time"

	"github.com/lightningnetwork/lnd/lnwire"
	"github.com/stretchr/testify/require"
)

func TestPingManager(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name     string
		delay    int
		pongSize uint16
		result   bool
	}{
		{
			name:     "Happy Path",
			delay:    0,
			pongSize: 4,
			result:   true,
		},
		{
			name:     "Bad Pong",
			delay:    0,
			pongSize: 3,
			result:   false,
		},
		{
			name:     "Timeout",
			delay:    2,
			pongSize: 4,
			result:   false,
		},
	}

	payload := make([]byte, 4)
	for _, test := range testCases {
		test := test

		t.Run(test.name, func(t *testing.T) {
			t.Parallel()

			// Set up PingManager.
			onPing := make(chan struct{})
			disconnected := make(chan struct{})
			mgr := NewPingManager(&PingManagerConfig{
				NewPingPayload: func() []byte {
					return payload
				},
				NewPongSize: func() uint16 {
					return 4
				},
				IntervalDuration: time.Second * 2,
				TimeoutDuration:  time.Second,
				SendPing: func(ping *lnwire.Ping) {
					close(onPing)
				},
				OnNoPongReceived: func() {
					close(disconnected)
				},
			})

			require.NoError(t, mgr.Start())

			// Wait for initial Ping.
			<-onPing

			// Wait for pre-determined time before sending Pong
			// response.
			time.Sleep(time.Duration(test.delay) * time.Second)

			// Send Pong back.
			res := &lnwire.Pong{
				PongBytes: make([]byte, test.pongSize),
			}
			mgr.ReceivedPong(res)

			// Evaluate result
			select {
			case <-time.NewTimer(time.Second / 2).C:
				require.True(t, test.result)
			case <-disconnected:
				require.False(t, test.result)
			}

			require.NoError(t, mgr.Stop())
		})
	}
}

Resulting brontide.go changes: (had to change the suffix to .txt to trick github into accepting the file type)

brontide.txt

[ProofOfKeags]

@ellemouton all feedback has been integrated. Let's get this shipped :)

[ellemouton]

@ProofOfKeags - can you re-push the branch to try get the CI to run? 🙏

[ProofOfKeags]

While it looks like the unit-race is still failing it appears to me that this is unrelated to the PR itself and is an artifact of the rebase.

[ellemouton]

LGTM 🔥 left a last set of comments

[ellemouton]

Since OnPongFailure is called for a variety of reasons, perhaps it makes sense to give the callback an error param so we can provide more info which Brontide can then log.

[ProofOfKeags]

LGTM 🔥 left a last set of comments

All comments addressed.

[saubyk]

Ack

[ellemouton]

@ProofOfKeags - The linter is failing and the race check failed on TestPingManager