Keep the latest queued offer or answer and simplify the handshaker test

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
+// OnRemoteOffer hands an offer to Listen without blocking, keeping only the most
-// doesn't block, discards the message if connection wasn't ready
+// recent one if several arrive before Listen reads them.
 func (h *Handshaker) OnRemoteOffer(offer OfferAnswer) {
-	select {
+	enqueueLatest(h.remoteOffersCh, offer)
-	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
-	}
 }
 
-// OnRemoteAnswer handles an offer from the remote peer and returns true if the message was accepted, false otherwise
+// OnRemoteAnswer hands an answer to Listen without blocking, keeping only the most
-// doesn't block, discards the message if connection wasn't ready
+// 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:
 	}
 }
 

client/internal/peer/handshaker_test.go

@@ -5,35 +5,14 @@ import (
 	"time"
 
 	log "github.com/sirupsen/logrus"
-	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+	"github.com/stretchr/testify/assert"
-
-	signal "github.com/netbirdio/netbird/shared/signal/client"
-	sProto "github.com/netbirdio/netbird/shared/signal/proto"
 )
 
 func newTestHandshaker(t *testing.T) *Handshaker {
 	t.Helper()
-
+	// The tests exercise the answer path, whose Listen branch dispatches to the
-	localKey, err := wgtypes.GeneratePrivateKey()
+	// relay listener without sending an answer, so no signaler/ICE/relay is needed.
-	if err != nil {
+	return NewHandshaker(log.WithField("test", t.Name()), ConnConfig{}, nil, nil, nil, 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)
 }
 
 // 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)
+	processed := make(chan *OfferAnswer, 4)
-	h.AddRelayListener(func(*OfferAnswer) { processed <- struct{}{} })
+	h.AddRelayListener(func(o *OfferAnswer) { processed <- o })
 
-	// Delivered before Listen is reading, exactly as when the peer is woken by the
+	// Delivered before Listen is reading, as when the peer is woken by the remote's
-	// remote's signal and the message is delivered right after Open.
+	// signal and the message is delivered right after Open.
-	h.OnRemoteAnswer(OfferAnswer{
+	h.OnRemoteAnswer(OfferAnswer{WgListenPort: 51820})
-		WgListenPort:   51820,
-		IceCredentials: IceCredentials{UFrag: "ufrag", Pwd: "pwd"},
-	})
 
 	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")
 	}
 }