From 7f03a2e86fe42f2418b1637ae0d00f3dae4351c3 Mon Sep 17 00:00:00 2001 From: Viktor Liu <17948409+lixmal@users.noreply.github.com> Date: Mon, 24 Aug 2026 17:54:11 +0900 Subject: [PATCH] [client] Hold a peer offer or answer that arrives before the handshaker starts listening (#7255) --- client/internal/peer/handshaker.go | 62 ++++++++++++++---------- client/internal/peer/handshaker_test.go | 63 +++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 24 deletions(-) create mode 100644 client/internal/peer/handshaker_test.go diff --git a/client/internal/peer/handshaker.go b/client/internal/peer/handshaker.go index 56e82e6e3..6ecb2a947 100644 --- a/client/internal/peer/handshaker.go +++ b/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: } } diff --git a/client/internal/peer/handshaker_test.go b/client/internal/peer/handshaker_test.go new file mode 100644 index 000000000..5e203d78b --- /dev/null +++ b/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") + } +}