diff --git a/client/internal/peer/handshaker.go b/client/internal/peer/handshaker.go index e236a63fb..6ecb2a947 100644 --- a/client/internal/peer/handshaker.go +++ b/client/internal/peer/handshaker.go @@ -167,29 +167,38 @@ func (h *Handshaker) SendOffer() error { return h.sendOffer() } -// OnRemoteOffer handles an offer from the remote peer and returns true if the message was accepted, false otherwise -// doesn't block, discards the message if connection wasn't ready +// OnRemoteOffer hands an offer to Listen without blocking, keeping only the most +// recent one if several arrive before Listen reads them. func (h *Handshaker) OnRemoteOffer(offer OfferAnswer) { - select { - case h.remoteOffersCh <- offer: - return - default: - h.log.Warnf("skipping remote offer message because receiver not ready") - // connection might not be ready yet to receive so we ignore the message - return - } + enqueueLatest(h.remoteOffersCh, offer) } -// OnRemoteAnswer handles an offer from the remote peer and returns true if the message was accepted, false otherwise -// doesn't block, discards the message if connection wasn't ready +// OnRemoteAnswer hands an answer to Listen without blocking, keeping only the most +// recent one if several arrive before Listen reads them. func (h *Handshaker) OnRemoteAnswer(answer OfferAnswer) { + enqueueLatest(h.remoteAnswerCh, answer) +} + +// enqueueLatest delivers msg on a one-slot channel without blocking. When the slot +// already holds an unread message the older one is discarded in favor of msg, so a +// message arriving before Listen starts reading is held rather than dropped, and +// the newest wins if several arrive first. Safe because there is a single producer +// (the engine loop): after draining the stale value the send always has room. +func enqueueLatest(ch chan OfferAnswer, msg OfferAnswer) { select { - case h.remoteAnswerCh <- answer: + case ch <- msg: return default: - // connection might not be ready yet to receive so we ignore the message - h.log.Warnf("skipping remote answer message because receiver not ready") - return + } + + select { + case <-ch: + default: + } + + select { + case ch <- msg: + default: } } diff --git a/client/internal/peer/handshaker_test.go b/client/internal/peer/handshaker_test.go index 361b30f52..5e203d78b 100644 --- a/client/internal/peer/handshaker_test.go +++ b/client/internal/peer/handshaker_test.go @@ -5,35 +5,14 @@ import ( "time" log "github.com/sirupsen/logrus" - "golang.zx2c4.com/wireguard/wgctrl/wgtypes" - - signal "github.com/netbirdio/netbird/shared/signal/client" - sProto "github.com/netbirdio/netbird/shared/signal/proto" + "github.com/stretchr/testify/assert" ) func newTestHandshaker(t *testing.T) *Handshaker { t.Helper() - - localKey, err := wgtypes.GeneratePrivateKey() - if err != nil { - t.Fatalf("generate local key: %v", err) - } - remoteKey, err := wgtypes.GeneratePrivateKey() - if err != nil { - t.Fatalf("generate remote key: %v", err) - } - - signaler := NewSignaler(&signal.MockClient{ - ReadyFunc: func() bool { return true }, - SendFunc: func(*sProto.Message) error { return nil }, - }, localKey) - - cfg := ConnConfig{ - Key: remoteKey.PublicKey().String(), - LocalKey: localKey.PublicKey().String(), - } - - return NewHandshaker(log.WithField("test", t.Name()), cfg, signaler, nil, nil, nil) + // The tests exercise the answer path, whose Listen branch dispatches to the + // relay listener without sending an answer, so no signaler/ICE/relay is needed. + return NewHandshaker(log.WithField("test", t.Name()), ConnConfig{}, nil, nil, nil, nil) } // TestHandshakerHoldsSignalArrivingBeforeListen covers the case where a peer is @@ -42,28 +21,43 @@ func newTestHandshaker(t *testing.T) *Handshaker { // message must be held rather than dropped, or the connection cannot proceed until // the remote re-sends. This is the path taken when an eager peer connects to a // lazily-managed one. -// -// The answer path is used because its Listen branch dispatches to the same -// listeners as the offer path without also sending an answer, so it exercises the -// buffered-channel behavior (which covers both channels) without needing a relay. func TestHandshakerHoldsSignalArrivingBeforeListen(t *testing.T) { h := newTestHandshaker(t) - processed := make(chan struct{}, 4) - h.AddRelayListener(func(*OfferAnswer) { processed <- struct{}{} }) + processed := make(chan *OfferAnswer, 4) + h.AddRelayListener(func(o *OfferAnswer) { processed <- o }) - // Delivered before Listen is reading, exactly as when the peer is woken by the - // remote's signal and the message is delivered right after Open. - h.OnRemoteAnswer(OfferAnswer{ - WgListenPort: 51820, - IceCredentials: IceCredentials{UFrag: "ufrag", Pwd: "pwd"}, - }) + // Delivered before Listen is reading, as when the peer is woken by the remote's + // signal and the message is delivered right after Open. + h.OnRemoteAnswer(OfferAnswer{WgListenPort: 51820}) go h.Listen(t.Context()) select { case <-processed: case <-time.After(2 * time.Second): - t.Fatal("signal that arrived before Listen was ready was dropped") + assert.Fail(t, "remote-answer dispatch: signal delivered before Listen was ready was dropped") + } +} + +// TestHandshakerKeepsLatestSignalBeforeListen covers several signals arriving +// before Listen reads: the newest must win (matching the latest-offer contract), +// rather than the first being kept and later ones discarded. +func TestHandshakerKeepsLatestSignalBeforeListen(t *testing.T) { + h := newTestHandshaker(t) + + processed := make(chan *OfferAnswer, 4) + h.AddRelayListener(func(o *OfferAnswer) { processed <- o }) + + h.OnRemoteAnswer(OfferAnswer{WgListenPort: 1111}) + h.OnRemoteAnswer(OfferAnswer{WgListenPort: 2222}) + + go h.Listen(t.Context()) + + select { + case got := <-processed: + assert.Equal(t, 2222, got.WgListenPort, "remote-answer dispatch: the latest queued signal should be processed") + case <-time.After(2 * time.Second): + assert.Fail(t, "remote-answer dispatch: queued signal was dropped") } }