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

client/internal/peer/handshaker.go

@@ -81,14 +81,19 @@ type Handshaker struct {
 
 func NewHandshaker(log *log.Entry, config ConnConfig, signaler *Signaler, ice *WorkerICE, relay *WorkerRelay, metricsStages *MetricsStages) *Handshaker {
 	h := &Handshaker{
-		log:            log,
-		config:         config,
-		signaler:       signaler,
-		ice:            ice,
-		relay:          relay,
-		metricsStages:  metricsStages,
-		remoteOffersCh: make(chan OfferAnswer),
-		remoteAnswerCh: make(chan OfferAnswer),
+		log:           log,
+		config:        config,
+		signaler:      signaler,
+		ice:           ice,
+		relay:         relay,
+		metricsStages: metricsStages,
+		// Buffered by one so an offer or answer that arrives between Open launching
+		// the Listen goroutine and it reaching its receive is held rather than
+		// dropped. A peer activated by an incoming signal receives the remote's
+		// message in that window; an unbuffered channel skips it as "receiver not
+		// ready", and the connection cannot proceed until the remote re-sends.
+		remoteOffersCh: make(chan OfferAnswer, 1),
+		remoteAnswerCh: make(chan OfferAnswer, 1),
 	}
 	// assume remote supports ICE until we learn otherwise from received offers
 	h.remoteICESupported.Store(ice != nil)
@@ -162,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:
 	}
 }
 

client/internal/peer/handshaker_test.go

@@ -0,0 +1,63 @@
+package peer
+
+import (
+	"testing"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+)
+
+func newTestHandshaker(t *testing.T) *Handshaker {
+	t.Helper()
+	// 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
+// activated by an incoming signal: the remote's offer/answer arrives in the same
+// step that opens the connection, before the Listen loop starts reading. The
+// 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.
+func TestHandshakerHoldsSignalArrivingBeforeListen(t *testing.T) {
+	h := newTestHandshaker(t)
+
+	processed := make(chan *OfferAnswer, 4)
+	h.AddRelayListener(func(o *OfferAnswer) { processed <- o })
+
+	// 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):
+		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")
+	}
+}