Repository: quic-go/quic-go

Issue:
retransmission of path probe packets
Packets containing PATH_CHALLENGE frames might get lost, or they might get reordered (and therefore ignored by the server, as only a packet containing a PATH_CHALLENGE that increases the largest received packet number counts).

We should retransmit the PATH_CHALLENGE for the duration of the `Probe` call.

Hints from issue comments:

Solution patch:
diff --git a/connection.go b/connection.go
index 9e72f925..ef8476eb 100644
--- a/connection.go
+++ b/connection.go
@@ -2636,18 +2636,22 @@ func (s *connection) AddPath(t *Transport) (*Path, error) {
 	if err := t.init(false); err != nil {
 		return nil, err
 	}
-	return s.getPathManager().NewPath(t, func() {
-		runner := t.connRunner()
-		s.connIDGenerator.AddConnRunner(
-			t.id(),
-			connRunnerCallbacks{
-				AddConnectionID:    func(connID protocol.ConnectionID) { runner.Add(connID, s) },
-				RemoveConnectionID: runner.Remove,
-				RetireConnectionID: runner.Retire,
-				ReplaceWithClosed:  runner.ReplaceWithClosed,
-			},
-		)
-	}), nil
+	return s.getPathManager().NewPath(
+		t,
+		200*time.Millisecond, // initial RTT estimate
+		func() {
+			runner := t.connRunner()
+			s.connIDGenerator.AddConnRunner(
+				t.id(),
+				connRunnerCallbacks{
+					AddConnectionID:    func(connID protocol.ConnectionID) { runner.Add(connID, s) },
+					RemoveConnectionID: runner.Remove,
+					RetireConnectionID: runner.Retire,
+					ReplaceWithClosed:  runner.ReplaceWithClosed,
+				},
+			)
+		},
+	), nil
 }
 
 func (s *connection) NextConnection(ctx context.Context) (Connection, error) {
diff --git a/path_manager_outgoing.go b/path_manager_outgoing.go
index 88d30a5b..95ec140c 100644
--- a/path_manager_outgoing.go
+++ b/path_manager_outgoing.go
@@ -4,8 +4,10 @@ import (
 	"context"
 	"crypto/rand"
 	"errors"
+	"slices"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/quic-go/quic-go/internal/ackhandler"
 	"github.com/quic-go/quic-go/internal/protocol"
@@ -22,21 +24,37 @@ type Path struct {
 	id          pathID
 	pathManager *pathManagerOutgoing
 	tr          *Transport
+	initialRTT  time.Duration
 
-	enablePath     func()
-	startedProbing atomic.Bool
+	enablePath func()
+	validated  atomic.Bool
 }
 
 func (p *Path) Probe(ctx context.Context) error {
-	done := make(chan struct{})
-	p.pathManager.addPath(p, p.enablePath, done)
+	path := p.pathManager.addPath(p, p.enablePath)
 
-	p.startedProbing.Store(true)
-	select {
-	case <-ctx.Done():
-		return context.Cause(ctx)
-	case <-done:
-		return nil
+	p.pathManager.enqueueProbe(p)
+	nextProbeDur := p.initialRTT
+	var timer *time.Timer
+	var timerChan <-chan time.Time
+	for {
+		select {
+		case <-ctx.Done():
+			return context.Cause(ctx)
+		case <-path.Validated():
+			p.validated.Store(true)
+			return nil
+		case <-timerChan:
+			p.pathManager.enqueueProbe(p)
+		case <-path.ProbeSent():
+		}
+
+		if timer != nil {
+			timer.Stop()
+		}
+		timer = time.NewTimer(nextProbeDur)
+		timerChan = timer.C
+		nextProbeDur *= 2 // exponential backoff
 	}
 }
 
@@ -44,7 +62,7 @@ func (p *Path) Probe(ctx context.Context) error {
 // It immediately stops sending on the old path, and sends on this new path.
 func (p *Path) Switch() error {
 	if err := p.pathManager.switchToPath(p.id); err != nil {
-		if errors.Is(err, errPathDoesNotExist) && !p.startedProbing.Load() {
+		if errors.Is(err, errPathDoesNotExist) && !p.validated.Load() {
 			return ErrPathNotValidated
 		}
 		return err
@@ -53,13 +71,17 @@ func (p *Path) Switch() error {
 }
 
 type pathOutgoing struct {
-	pathChallenge [8]byte
-	tr            *Transport
-	isValidated   bool
-	validated     chan<- struct{} // closed when the path the corresponding PATH_RESPONSE is received
-	enablePath    func()
+	pathChallenges [][8]byte // length is implicitly limited by exponential backoff
+	tr             *Transport
+	isValidated    bool
+	probeSent      chan struct{} // receives when a PATH_CHALLENGE is sent
+	validated      chan struct{} // closed when the path the corresponding PATH_RESPONSE is received
+	enablePath     func()
 }
 
+func (p *pathOutgoing) ProbeSent() <-chan struct{} { return p.probeSent }
+func (p *pathOutgoing) Validated() <-chan struct{} { return p.validated }
+
 type pathManagerOutgoing struct {
 	getConnID       func(pathID) (_ protocol.ConnectionID, ok bool)
 	retireConnID    func(pathID)
@@ -85,16 +107,28 @@ func newPathManagerOutgoing(
 	}
 }
 
-func (pm *pathManagerOutgoing) addPath(p *Path, enablePath func(), done chan<- struct{}) {
-	var b [8]byte
-	_, _ = rand.Read(b[:])
+func (pm *pathManagerOutgoing) addPath(p *Path, enablePath func()) *pathOutgoing {
 	pm.mx.Lock()
-	pm.paths[p.id] = &pathOutgoing{
-		pathChallenge: b,
-		tr:            p.tr,
-		validated:     done,
-		enablePath:    enablePath,
+	defer pm.mx.Unlock()
+
+	// path might already exist, and just being re-probed
+	if existingPath, ok := pm.paths[p.id]; ok {
+		existingPath.validated = make(chan struct{})
+		return existingPath
+	}
+
+	path := &pathOutgoing{
+		tr:         p.tr,
+		probeSent:  make(chan struct{}, 1),
+		validated:  make(chan struct{}),
+		enablePath: enablePath,
 	}
+	pm.paths[p.id] = path
+	return path
+}
+
+func (pm *pathManagerOutgoing) enqueueProbe(p *Path) {
+	pm.mx.Lock()
 	pm.pathsToProbe = append(pm.pathsToProbe, p.id)
 	pm.mx.Unlock()
 	pm.scheduleSending()
@@ -115,7 +149,7 @@ func (pm *pathManagerOutgoing) switchToPath(id pathID) error {
 	return nil
 }
 
-func (pm *pathManagerOutgoing) NewPath(t *Transport, enablePath func()) *Path {
+func (pm *pathManagerOutgoing) NewPath(t *Transport, initialRTT time.Duration, enablePath func()) *Path {
 	pm.mx.Lock()
 	defer pm.mx.Unlock()
 
@@ -126,6 +160,7 @@ func (pm *pathManagerOutgoing) NewPath(t *Transport, enablePath func()) *Path {
 		id:          id,
 		tr:          t,
 		enablePath:  enablePath,
+		initialRTT:  initialRTT,
 	}
 }
 
@@ -156,9 +191,17 @@ func (pm *pathManagerOutgoing) NextPathToProbe() (_ protocol.ConnectionID, _ ack
 		return protocol.ConnectionID{}, ackhandler.Frame{}, nil, false
 	}
 
+	var b [8]byte
+	_, _ = rand.Read(b[:])
+	p.pathChallenges = append(p.pathChallenges, b)
+
 	p.enablePath()
+	select {
+	case p.probeSent <- struct{}{}:
+	default:
+	}
 	frame := ackhandler.Frame{
-		Frame:   &wire.PathChallengeFrame{Data: p.pathChallenge},
+		Frame:   &wire.PathChallengeFrame{Data: b},
 		Handler: (*pathManagerOutgoingAckHandler)(pm),
 	}
 	return connID, frame, p.tr, true
@@ -169,14 +212,14 @@ func (pm *pathManagerOutgoing) HandlePathResponseFrame(f *wire.PathResponseFrame
 	defer pm.mx.Unlock()
 
 	for _, p := range pm.paths {
-		if f.Data == p.pathChallenge {
+		if slices.Contains(p.pathChallenges, f.Data) {
 			// path validated
 			if !p.isValidated {
+				// make sure that duplicate PATH_RESPONSE frames are ignored
 				p.isValidated = true
+				p.pathChallenges = nil
 				close(p.validated)
 			}
-			// makes sure that duplicate PATH_RESPONSE frames are ignored
-			p.validated = nil
 			break
 		}
 	}



Test patch:
diff --git a/path_manager_outgoing_test.go b/path_manager_outgoing_test.go
index db1ab260..54b86742 100644
--- a/path_manager_outgoing_test.go
+++ b/path_manager_outgoing_test.go
@@ -11,7 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestPathManagerOutgoing(t *testing.T) {
+func TestPathManagerOutgoingPathProbing(t *testing.T) {
 	connIDs := []protocol.ConnectionID{
 		protocol.ParseConnectionID([]byte{1, 2, 3, 4, 5, 6, 7, 8}),
 		protocol.ParseConnectionID([]byte{2, 3, 4, 5, 6, 7, 8, 9}),
@@ -28,7 +28,7 @@ func TestPathManagerOutgoing(t *testing.T) {
 
 	tr1 := &Transport{}
 	var enabled bool
-	p := pm.NewPath(tr1, func() { enabled = true })
+	p := pm.NewPath(tr1, time.Second, func() { enabled = true })
 	require.ErrorIs(t, p.Switch(), ErrPathNotValidated)
 
 	errChan := make(chan error, 1)
@@ -97,3 +97,105 @@ func TestPathManagerOutgoing(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, tr1, switchToTransport)
 }
+
+func TestPathManagerOutgoingRetransmissions(t *testing.T) {
+	connIDs := []protocol.ConnectionID{
+		protocol.ParseConnectionID([]byte{1, 2, 3, 4, 5, 6, 7, 8}),
+		protocol.ParseConnectionID([]byte{2, 3, 4, 5, 6, 7, 8, 9}),
+	}
+	var retiredConnIDs []protocol.ConnectionID
+	scheduledSending := make(chan struct{}, 20)
+	pm := newPathManagerOutgoing(
+		func(id pathID) (protocol.ConnectionID, bool) { return connIDs[id], true },
+		func(id pathID) { retiredConnIDs = append(retiredConnIDs, connIDs[id]) },
+		func() { scheduledSending <- struct{}{} },
+	)
+
+	_, _, _, ok := pm.NextPathToProbe()
+	require.False(t, ok)
+
+	tr1 := &Transport{}
+	initialRTT := scaleDuration(2 * time.Millisecond)
+	p := pm.NewPath(tr1, initialRTT, func() {})
+
+	pathChallengeChan := make(chan [8]byte)
+	done := make(chan struct{})
+	defer close(done)
+	go func() {
+		for {
+			select {
+			case <-scheduledSending:
+			case <-done:
+				return
+			}
+			_, f, _, ok := pm.NextPathToProbe()
+			if !ok {
+				// should never happen
+				pathChallengeChan <- [8]byte{}
+				continue
+			}
+			pathChallengeChan <- f.Frame.(*wire.PathChallengeFrame).Data
+		}
+	}()
+
+	errChan := make(chan error, 1)
+	go func() { errChan <- p.Probe(context.Background()) }()
+
+	start := time.Now()
+	var pathChallenges [][8]byte
+	for range 4 {
+		select {
+		case err := <-errChan:
+			require.NoError(t, err)
+		case pc := <-pathChallengeChan:
+			pathChallenges = append(pathChallenges, pc)
+		case <-time.After(scaleDuration(time.Second)):
+			t.Fatal("timeout")
+		}
+	}
+	took := time.Since(start)
+
+	require.NotContains(t, pathChallenges, [8]byte{})
+	require.NotEqual(t, pathChallenges[0], pathChallenges[1])
+	require.NotEqual(t, pathChallenges[0], pathChallenges[2])
+	require.NotEqual(t, pathChallenges[0], pathChallenges[3])
+	require.NotEqual(t, pathChallenges[1], pathChallenges[2])
+	require.NotEqual(t, pathChallenges[2], pathChallenges[3])
+
+	require.Greater(t, took, initialRTT*(1+2+4+8))
+	require.Less(t, took, initialRTT*(1+2+4+8)*3/2)
+
+	// receiving a PATH_RESPONSE for any of the PATH_CHALLENGES completes path validation
+	pm.HandlePathResponseFrame(&wire.PathResponseFrame{Data: pathChallenges[2]})
+
+	select {
+	case err := <-errChan:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("timeout")
+	}
+
+	// It is valid to probe again
+	pathChallenges = pathChallenges[:0]
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() { errChan <- p.Probe(ctx) }()
+
+	for range 2 {
+		select {
+		case err := <-errChan:
+			require.NoError(t, err)
+		case pc := <-pathChallengeChan:
+			pathChallenges = append(pathChallenges, pc)
+		case <-time.After(scaleDuration(time.Second)):
+			t.Fatal("timeout")
+		}
+	}
+	// this time, don't receive a PATH_RESPONSE
+	cancel()
+	select {
+	case err := <-errChan:
+		require.ErrorIs(t, err, context.Canceled)
+	case <-time.After(time.Second):
+		t.Fatal("timeout")
+	}
+}