diff --git a/agent.go b/agent.go index 7e9aa5c9..6eb58901 100644 --- a/agent.go +++ b/agent.go @@ -12,6 +12,7 @@ import ( "net" "net/netip" "slices" + "strconv" "strings" "sync" "sync/atomic" @@ -64,8 +65,9 @@ type Agent struct { tieBreaker uint64 lite bool - connectionState ConnectionState - gatheringState GatheringState + connectionState ConnectionState + gatheringState GatheringState + gatherGeneration uint64 mDNSMode MulticastDNSMode mDNSName string @@ -1024,7 +1026,7 @@ func (a *Agent) AddVirtualCandidate(cand Candidate, candidateConn net.PacketConn return errCandidatePacketConnNil } - return a.addCandidate(a.loop, cand, candidateConn, true) + return a.addCandidate(a.loop, cand, candidateConn, nil, true) } func isMulticastDNSCandidate(cand Candidate) bool { @@ -1410,10 +1412,20 @@ func (a *Agent) shouldAcceptRemoteCandidate(cand Candidate) bool { return true } +func (a *Agent) cleanupCandidate(cand Candidate, candidateConn net.PacketConn, reason string) { + if err := cand.close(); err != nil { + a.log.Warnf("Failed to close %s candidate: %v", reason, err) + } + if err := candidateConn.Close(); err != nil { + a.log.Warnf("Failed to close %s candidate connection: %v", reason, err) + } +} + func (a *Agent) addCandidate( ctx context.Context, cand Candidate, candidateConn net.PacketConn, + generation *uint64, errorOnDuplicate bool, ) error { if err := ctx.Err(); err != nil { @@ -1422,6 +1434,21 @@ func (a *Agent) addCandidate( var addErr error err := a.loop.Run(ctx, func(context.Context) { + candidateGeneration := a.gatherGeneration + if generation != nil { + candidateGeneration = *generation + } + if a.gatherGeneration != candidateGeneration { + a.log.Debugf( + "Ignoring candidate from different gather generation (a: %d c: %d)", + a.gatherGeneration, + candidateGeneration, + ) + a.cleanupCandidate(cand, candidateConn, "old") + + return + } + set := a.localCandidates[cand.NetworkType()] for _, candidate := range set { if candidate.Equal(cand) { @@ -1432,18 +1459,13 @@ func (a *Agent) addCandidate( } a.log.Debugf("Ignore duplicate candidate: %s", cand) - if err := cand.close(); err != nil { - a.log.Warnf("Failed to close duplicate candidate: %v", err) - } - if err := candidateConn.Close(); err != nil { - a.log.Warnf("Failed to close duplicate candidate connection: %v", err) - } + a.cleanupCandidate(cand, candidateConn, "duplicate") return } } - a.setCandidateExtensions(cand) + a.setCandidateExtensions(cand, candidateGeneration) cand.start(a, candidateConn, a.startedCh) a.setUniqueLiteCandidatePriority(cand) @@ -1467,7 +1489,7 @@ func (a *Agent) addCandidate( return addErr } -func (a *Agent) setCandidateExtensions(cand Candidate) { +func (a *Agent) setCandidateExtensions(cand Candidate, candidateGeneration uint64) { err := cand.AddExtension(CandidateExtension{ Key: "ufrag", Value: a.localUfrag, @@ -1475,6 +1497,14 @@ func (a *Agent) setCandidateExtensions(cand Candidate) { if err != nil { a.log.Errorf("Failed to add ufrag extension to candidate: %v", err) } + + err = cand.AddExtension(CandidateExtension{ + Key: "generation", + Value: strconv.FormatUint(candidateGeneration, 10), + }) + if err != nil { + a.log.Errorf("Failed to add generation extension to candidate: %v", err) + } } // GetRemoteCandidates returns the remote candidates. @@ -2146,10 +2176,10 @@ func (a *Agent) Restart(ufrag, pwd string) error { //nolint:cyclop } if runErr := a.loop.Run(a.loop, func(_ context.Context) { - // Cancel unconditionally: a gather goroutine that has started but not yet - // marked Gathering would otherwise outlive the restart and later - // overwrite the fresh New state. + // Cancel the previous gather before resetting its state. a.gatherCandidateCancel() + a.gatherGeneration++ + a.gatheringState = GatheringStateNew // Clear all agent needed to take back to fresh state a.removeUfragFromMux() @@ -2158,7 +2188,6 @@ func (a *Agent) Restart(ufrag, pwd string) error { //nolint:cyclop a.remoteUfrag = "" a.remotePwd = "" a.remoteCandidateGeneration++ - a.gatheringState = GatheringStateNew a.checklist = make([]*CandidatePair, 0) a.pairsByID = make(map[uint64]*CandidatePair) a.pendingBindingRequests = make([]bindingRequest, 0) @@ -2178,32 +2207,19 @@ func (a *Agent) Restart(ufrag, pwd string) error { //nolint:cyclop return nil } -// setGatheringState applies newState and reports whether it was applied. A write -// from a cycle canceled by Restart is dropped and reported false, so it can't -// clobber the fresh New state and wedge the next gather. -func (a *Agent) setGatheringState(gatherCtx context.Context, newState GatheringState) (bool, error) { - done := make(chan struct{}) - applied := false - if err := a.loop.Run(a.loop, func(context.Context) { //nolint:contextcheck - defer close(done) - - if gatherCtx.Err() != nil { +func (a *Agent) completeGathering(generation uint64) error { + if err := a.loop.Run(a.loop, func(context.Context) { + if generation != a.gatherGeneration || a.gatheringState != GatheringStateGathering { return } - if a.gatheringState != newState && newState == GatheringStateComplete { - a.candidateNotifier.EnqueueCandidate(nil) - } - - a.gatheringState = newState - applied = true + a.gatheringState = GatheringStateComplete + a.candidateNotifier.EnqueueCandidate(nil) }); err != nil { - return false, err + return err } - <-done - - return applied, nil + return nil } func (a *Agent) needsToCheckPriorityOnNominated() bool { diff --git a/agent_test.go b/agent_test.go index ebb2b273..98562d4c 100644 --- a/agent_test.go +++ b/agent_test.go @@ -539,7 +539,7 @@ func TestAgentCloseClearsSharedUDPMuxAbortDeadlineForOtherAgent(t *testing.T) { }) require.NoError(t, err) - require.NoError(t, agent.gatherCandidatesLocalUDPMux(context.Background())) + require.NoError(t, agent.gatherCandidatesLocalUDPMux(context.Background(), agent.gatherGeneration)) return agent } @@ -2629,9 +2629,7 @@ func TestAgentRestart(t *testing.T) { t.Run("Restart Both Sides", func(t *testing.T) { // Get all addresses of candidates concatenated - generateCandidateAddressStrings := func(candidates []Candidate, err error) (out string) { - require.NoError(t, err) - + generateCandidateAddressStrings := func(candidates []Candidate) (out string) { for _, c := range candidates { out += c.Address() + ":" out += strconv.Itoa(c.Port()) @@ -2640,14 +2638,31 @@ func TestAgentRestart(t *testing.T) { return } + candidateHasGeneration := func(generation uint64, candidate Candidate) { + genString := strconv.FormatUint(generation, 10) + ext, ok := candidate.GetExtension("generation") + + require.True(t, ok) + require.Equal(t, genString, ext.Value) + } + // Store the original candidates, confirm that after we reconnect we have new pairs connA, connB := pipe(t, &AgentConfig{ DisconnectedTimeout: &oneSecond, FailedTimeout: &oneSecond, }) defer closePipe(t, connA, connB) - connAFirstCandidates := generateCandidateAddressStrings(connA.agent.GetLocalCandidates()) - connBFirstCandidates := generateCandidateAddressStrings(connB.agent.GetLocalCandidates()) + + aFirstGeneration := connA.agent.gatherGeneration + bFirstGeneration := connB.agent.gatherGeneration + + connAFirstCandidates, err := connA.agent.GetLocalCandidates() + require.NoError(t, err) + connBFirstCandidates, err := connB.agent.GetLocalCandidates() + require.NoError(t, err) + + candidateHasGeneration(aFirstGeneration, connAFirstCandidates[0]) + candidateHasGeneration(bFirstGeneration, connBFirstCandidates[0]) aNotifier, aConnected := onConnected() require.NoError(t, connA.agent.OnConnectionStateChange(aNotifier)) @@ -2659,6 +2674,10 @@ func TestAgentRestart(t *testing.T) { require.NoError(t, connA.agent.Restart("", "")) require.NoError(t, connB.agent.Restart("", "")) + // Generation should change after Restart call + require.NotEqual(t, aFirstGeneration, connA.agent.gatherGeneration) + require.NotEqual(t, bFirstGeneration, connB.agent.gatherGeneration) + // Exchange Candidates and Credentials ufrag, pwd, err := connB.agent.GetLocalUserCredentials() require.NoError(t, err) @@ -2674,9 +2693,21 @@ func TestAgentRestart(t *testing.T) { <-aConnected <-bConnected + connASecondCandidates, err := connA.agent.GetLocalCandidates() + require.NoError(t, err) + connBSecondCandidates, err := connB.agent.GetLocalCandidates() + require.NoError(t, err) + + candidateHasGeneration(connA.agent.gatherGeneration, connASecondCandidates[0]) + candidateHasGeneration(connB.agent.gatherGeneration, connBSecondCandidates[0]) + // Assert that we have new candidates each time - require.NotEqual(t, connAFirstCandidates, generateCandidateAddressStrings(connA.agent.GetLocalCandidates())) - require.NotEqual(t, connBFirstCandidates, generateCandidateAddressStrings(connB.agent.GetLocalCandidates())) + aFirstCandidatesString := generateCandidateAddressStrings(connAFirstCandidates) + aSecondCandidatesString := generateCandidateAddressStrings(connASecondCandidates) + bFirstCandidatesString := generateCandidateAddressStrings(connBFirstCandidates) + bSecondCandidatesString := generateCandidateAddressStrings(connBSecondCandidates) + require.NotEqual(t, aFirstCandidatesString, aSecondCandidatesString) + require.NotEqual(t, bFirstCandidatesString, bSecondCandidatesString) }) } @@ -2937,8 +2968,8 @@ func TestAddCandidateClosesDuplicate(t *testing.T) { addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 13), Port: 5003}, } - require.NoError(t, agent.addCandidate(context.Background(), first, firstConn, false)) - require.NoError(t, agent.addCandidate(context.Background(), duplicate, duplicateConn, false)) + require.NoError(t, agent.addCandidate(context.Background(), first, firstConn, nil, false)) + require.NoError(t, agent.addCandidate(context.Background(), duplicate, duplicateConn, nil, false)) require.Equal(t, int32(1), duplicateCloseCount.Load()) require.Equal(t, int32(1), duplicateConn.closeCount.Load()) } @@ -2968,7 +2999,7 @@ func TestGetLocalCandidates(t *testing.T) { expectedCandidates = append(expectedCandidates, cand) - err = agent.addCandidate(context.Background(), cand, dummyConn, false) + err = agent.addCandidate(context.Background(), cand, dummyConn, nil, false) require.NoError(t, err) } @@ -3641,7 +3672,7 @@ func TestSetCandidatesUfrag(t *testing.T) { cand, errCand := NewCandidateHost(&cfg) require.NoError(t, errCand) - err = agent.addCandidate(context.Background(), cand, dummyConn, false) + err = agent.addCandidate(context.Background(), cand, dummyConn, nil, false) require.NoError(t, err) } @@ -3656,6 +3687,47 @@ func TestSetCandidatesUfrag(t *testing.T) { } } +func TestAddingCandidatesFromOtherGenerations(t *testing.T) { + var config AgentConfig + + agent, err := NewAgent(&config) + require.NoError(t, err) + defer func() { + require.NoError(t, agent.Close()) + }() + + agent.gatherGeneration = 3 + + dummyConn := &net.UDPConn{} + + for i := range 5 { + cfg := CandidateHostConfig{ + Network: "udp", + Address: "192.168.0.2", + Port: 1000 + i, + Component: 1, + } + + cand, errCand := NewCandidateHost(&cfg) + require.NoError(t, errCand) + + generation := uint64(i) //nolint:gosec + err = agent.addCandidate(context.Background(), cand, dummyConn, &generation, false) + require.NoError(t, err) + } + + actualCandidates, err := agent.GetLocalCandidates() + require.NoError(t, err) + require.Equal(t, 1, len(actualCandidates), "Only the candidate with a matching generation should be added") + + ext, ok := actualCandidates[0].GetExtension("generation") + require.True(t, ok) + + generation, err := strconv.ParseUint(ext.Value, 10, 64) + require.NoError(t, err) + require.Equal(t, agent.gatherGeneration, generation) +} + func TestAlwaysSentKeepAlive(t *testing.T) { //nolint:cyclop defer test.CheckRoutines(t)() diff --git a/gather.go b/gather.go index 1b6eb29d..74f651c1 100644 --- a/gather.go +++ b/gather.go @@ -133,8 +133,10 @@ func (a *Agent) GatherCandidates() error { a.gatherCandidateCancel = cancel done := make(chan struct{}) a.gatherCandidateDone = done + generation := a.gatherGeneration + a.gatheringState = GatheringStateGathering - go a.gatherCandidates(ctx, done) + go a.gatherCandidates(ctx, done, generation) }); runErr != nil { return runErr } @@ -142,24 +144,17 @@ func (a *Agent) GatherCandidates() error { return gatherErr } -func (a *Agent) gatherCandidates(ctx context.Context, done chan struct{}) { //nolint:cyclop +func (a *Agent) gatherCandidates(ctx context.Context, done chan struct{}, generation uint64) { //nolint:cyclop defer close(done) - applied, err := a.setGatheringState(ctx, GatheringStateGathering) - if err != nil { - a.log.Warnf("Failed to set gatheringState to GatheringStateGathering: %v", err) - - return - } - // The cycle was canceled before it started, so skip its gathering. - if !applied { + if ctx.Err() != nil { return } - a.gatherCandidatesInternal(ctx) + a.gatherCandidatesInternal(ctx, generation) switch a.continualGatheringPolicy { case GatherOnce: - if _, err := a.setGatheringState(ctx, GatheringStateComplete); err != nil { + if err := a.completeGathering(generation); err != nil { //nolint:contextcheck a.log.Warnf("Failed to set gatheringState to GatheringStateComplete: %v", err) } case GatherContinually: @@ -179,7 +174,7 @@ func (a *Agent) gatherCandidates(ctx context.Context, done chan struct{}) { //no } a.log.Infof("Initialized network monitoring with %d IP addresses", len(addrs)) } - go a.startNetworkMonitoring(ctx) + go a.startNetworkMonitoring(ctx, generation) } } @@ -264,22 +259,22 @@ func (a *Agent) applyHostRewriteForUDPMux(candidateIPs []net.IP, udpAddr *net.UD } // gatherCandidatesInternal performs the actual candidate gathering for all configured types. -func (a *Agent) gatherCandidatesInternal(ctx context.Context) { +func (a *Agent) gatherCandidatesInternal(ctx context.Context, generation uint64) { var wg sync.WaitGroup for _, t := range a.candidateTypes { switch t { case CandidateTypeHost: wg.Add(1) go func() { - a.gatherCandidatesLocal(ctx, a.networkTypes) + a.gatherCandidatesLocal(ctx, a.networkTypes, generation) wg.Done() }() case CandidateTypeServerReflexive: - a.gatherServerReflexiveCandidates(ctx, &wg) + a.gatherServerReflexiveCandidates(ctx, &wg, generation) case CandidateTypeRelay: wg.Add(1) go func() { - a.gatherCandidatesRelay(ctx, a.urls) + a.gatherCandidatesRelay(ctx, a.urls, generation) wg.Done() }() case CandidateTypePeerReflexive, CandidateTypeUnspecified: @@ -290,15 +285,15 @@ func (a *Agent) gatherCandidatesInternal(ctx context.Context) { wg.Wait() } -func (a *Agent) gatherServerReflexiveCandidates(ctx context.Context, wg *sync.WaitGroup) { +func (a *Agent) gatherServerReflexiveCandidates(ctx context.Context, wg *sync.WaitGroup, generation uint64) { replaceSrflx := a.addressRewriteMapper != nil && a.addressRewriteMapper.shouldReplace(CandidateTypeServerReflexive) if !replaceSrflx { wg.Add(1) go func() { if a.udpMuxSrflx != nil { - a.gatherCandidatesSrflxUDPMux(ctx, a.urls, a.networkTypes) + a.gatherCandidatesSrflxUDPMux(ctx, a.urls, a.networkTypes, generation) } else { - a.gatherCandidatesSrflx(ctx, a.urls, a.networkTypes) + a.gatherCandidatesSrflx(ctx, a.urls, a.networkTypes, generation) } wg.Done() }() @@ -306,14 +301,14 @@ func (a *Agent) gatherServerReflexiveCandidates(ctx context.Context, wg *sync.Wa if a.addressRewriteMapper != nil && a.addressRewriteMapper.hasCandidateType(CandidateTypeServerReflexive) { wg.Add(1) go func() { - a.gatherCandidatesSrflxMapped(ctx, a.networkTypes) + a.gatherCandidatesSrflxMapped(ctx, a.networkTypes, generation) wg.Done() }() } } //nolint:gocognit,gocyclo,cyclop,maintidx -func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []NetworkType) { +func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []NetworkType, generation uint64) { networks := map[string]struct{}{} for _, networkType := range networkTypes { if networkType.IsTCP() { @@ -325,7 +320,7 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ // When UDPMux is enabled, skip other UDP candidates if a.udpMux != nil { - if err := a.gatherCandidatesLocalUDPMux(ctx); err != nil { + if err := a.gatherCandidatesLocalUDPMux(ctx, generation); err != nil { a.log.Warnf("Failed to create host candidate for UDPMux: %s", err) } delete(networks, udp) @@ -485,16 +480,9 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ continue } - if err := a.addCandidate(ctx, candidateHost, connAndPort.conn, false); err != nil { - if closeErr := candidateHost.close(); closeErr != nil { - a.log.Warnf("Failed to close candidate: %v", closeErr) - } - closeConnAndLog( - connAndPort.conn, - a.log, - "Failed to append to localCandidates and run onCandidateHdlr: %v", - err, - ) + if err := a.addCandidate(ctx, candidateHost, connAndPort.conn, &generation, false); err != nil { + a.log.Warnf("Failed to append to localCandidates and run onCandidateHdlr: %v", err) + a.cleanupCandidate(candidateHost, connAndPort.conn, "failed") } } } @@ -524,7 +512,7 @@ func shouldFilterLocationTracked(candidateIP net.IP) bool { return shouldFilterLocationTrackedIP(addr) } -func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context) error { //nolint:gocognit,cyclop +func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context, generation uint64) error { //nolint:gocognit,cyclop if a.udpMux == nil { return errUDPMuxDisabled } @@ -594,12 +582,9 @@ func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context) error { //nolin continue } - if err := a.addCandidate(ctx, c, conn, false); err != nil { - if closeErr := c.close(); closeErr != nil { - a.log.Warnf("Failed to close candidate: %v", closeErr) - } - - closeConnAndLog(conn, a.log, "failed to add candidate: %s %d: %v", candidateIP, udpAddr.Port, err) + if err := a.addCandidate(ctx, c, conn, &generation, false); err != nil { + a.log.Warnf("failed to add candidate: %s %d: %v", candidateIP, udpAddr.Port, err) + a.cleanupCandidate(c, conn, "failed") continue } @@ -611,7 +596,8 @@ func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context) error { //nolin return nil } -func (a *Agent) gatherCandidatesSrflxMapped(ctx context.Context, networkTypes []NetworkType) { //nolint:gocognit,cyclop +//nolint:gocognit,cyclop +func (a *Agent) gatherCandidatesSrflxMapped(ctx context.Context, networkTypes []NetworkType, generation uint64) { var wg sync.WaitGroup defer wg.Wait() @@ -698,7 +684,7 @@ func (a *Agent) gatherCandidatesSrflxMapped(ctx context.Context, networkTypes [] RelAddr: currentAddr.IP.String(), RelPort: currentAddr.Port, } - c, err := NewCandidateServerReflexive(&srflxConfig) + candidate, err := NewCandidateServerReflexive(&srflxConfig) if err != nil { closeConnAndLog(currentConn, a.log, "failed to create server reflexive candidate: %s %s %d: %v", network, @@ -709,17 +695,9 @@ func (a *Agent) gatherCandidatesSrflxMapped(ctx context.Context, networkTypes [] continue } - if err := a.addCandidate(ctx, c, currentConn, false); err != nil { - if closeErr := c.close(); closeErr != nil { - a.log.Warnf("Failed to close candidate: %v", closeErr) - } + if err := a.addCandidate(ctx, candidate, currentConn, &generation, false); err != nil { a.log.Warnf("Failed to append to localCandidates and run onCandidateHdlr: %v", err) - closeConnAndLog( - currentConn, - a.log, - "closing srflx conn after addCandidate failure: %v", - err, - ) + a.cleanupCandidate(candidate, currentConn, "failed") } } }() @@ -727,7 +705,12 @@ func (a *Agent) gatherCandidatesSrflxMapped(ctx context.Context, networkTypes [] } //nolint:gocognit,cyclop -func (a *Agent) gatherCandidatesSrflxUDPMux(ctx context.Context, urls []*stun.URI, networkTypes []NetworkType) { +func (a *Agent) gatherCandidatesSrflxUDPMux( + ctx context.Context, + urls []*stun.URI, + networkTypes []NetworkType, + generation uint64, +) { var wg sync.WaitGroup defer wg.Wait() @@ -798,16 +781,9 @@ func (a *Agent) gatherCandidatesSrflxUDPMux(ctx context.Context, urls []*stun.UR return } - if err := a.addCandidate(ctx, c, conn, false); err != nil { - if closeErr := c.close(); closeErr != nil { - a.log.Warnf("Failed to close candidate: %v", closeErr) - } - closeConnAndLog( - conn, - a.log, - "Failed to append srflx mux candidate to localCandidates: %v", - err, - ) + if err := a.addCandidate(ctx, c, conn, &generation, false); err != nil { + a.log.Warnf("Failed to append srflx mux candidate to localCandidates: %v", err) + a.cleanupCandidate(c, conn, "failed") } }(*urls[i], networkType.String(), udpAddr) } @@ -837,7 +813,9 @@ func getXORMappedAddr( } //nolint:cyclop,gocognit -func (a *Agent) gatherCandidatesSrflx(ctx context.Context, urls []*stun.URI, networkTypes []NetworkType) { +func (a *Agent) gatherCandidatesSrflx( + ctx context.Context, urls []*stun.URI, networkTypes []NetworkType, generation uint64, +) { var wg sync.WaitGroup defer wg.Wait() @@ -929,11 +907,9 @@ func (a *Agent) gatherCandidatesSrflx(ctx context.Context, urls []*stun.URI, net return } - if err := a.addCandidate(ctx, c, conn, false); err != nil { - if closeErr := c.close(); closeErr != nil { - a.log.Warnf("Failed to close candidate: %v", closeErr) - } + if err := a.addCandidate(ctx, c, conn, &generation, false); err != nil { a.log.Warnf("Failed to append to localCandidates and run onCandidateHdlr: %v", err) + a.cleanupCandidate(c, conn, "failed") } } @@ -974,7 +950,7 @@ func (a *Agent) gatherCandidatesSrflx(ctx context.Context, urls []*stun.URI, net } //nolint:maintidx,gocognit,gocyclo,cyclop -func (a *Agent) gatherCandidatesRelay(ctx context.Context, urls []*stun.URI) { +func (a *Agent) gatherCandidatesRelay(ctx context.Context, urls []*stun.URI, generation uint64) { var wg sync.WaitGroup defer wg.Wait() _, ifaces, _ := localInterfaces(a.net, a.interfaceFilter, a.ipFilter, a.networkTypes, a.includeLoopback) @@ -1244,7 +1220,7 @@ func (a *Agent) gatherCandidatesRelay(ctx context.Context, urls []*stun.URI) { // Relay allocations currently produce UDP relay endpoints regardless of // whether the TURN control connection uses UDP/TCP/TLS/DTLS. - a.addRelayCandidates(ctx, relayEndpoint{ + a.addRelayCandidates(ctx, generation, relayEndpoint{ network: udp, address: rAddr.IP, port: rAddr.Port, @@ -1361,7 +1337,9 @@ func findIfaceForIP(ifaces []ifaceAddr, ip net.IP) string { return "" } -func (a *Agent) createRelayCandidate(ctx context.Context, ep relayEndpoint, ip net.IP, onClose func() error) error { +func (a *Agent) createRelayCandidate( + ctx context.Context, ep relayEndpoint, ip net.IP, generation uint64, onClose func() error, +) error { relayConfig := CandidateRelayConfig{ Network: ep.network, Component: ComponentRTP, @@ -1379,7 +1357,7 @@ func (a *Agent) createRelayCandidate(ctx context.Context, ep relayEndpoint, ip n return err } - if err := a.addCandidate(ctx, candidate, ep.conn, false); err != nil { + if err := a.addCandidate(ctx, candidate, ep.conn, &generation, false); err != nil { if closeErr := candidate.close(); closeErr != nil { a.log.Warnf("Failed to close candidate: %v", closeErr) } @@ -1391,7 +1369,7 @@ func (a *Agent) createRelayCandidate(ctx context.Context, ep relayEndpoint, ip n return nil } -func (a *Agent) addRelayCandidates(ctx context.Context, ep relayEndpoint) { //nolint:cyclop +func (a *Agent) addRelayCandidates(ctx context.Context, generation uint64, ep relayEndpoint) { //nolint:cyclop if ep.conn == nil || ep.address == nil { return } @@ -1430,7 +1408,7 @@ func (a *Agent) addRelayCandidates(ctx context.Context, ep relayEndpoint) { //no onClose = nil } - if err := a.createRelayCandidate(ctx, ep, ip, onClose); err != nil { + if err := a.createRelayCandidate(ctx, ep, ip, generation, onClose); err != nil { if idx == 0 { if ep.closeConn != nil { ep.closeConn() @@ -1448,7 +1426,7 @@ func (a *Agent) addRelayCandidates(ctx context.Context, ep relayEndpoint) { //no // startNetworkMonitoring starts a goroutine that periodically checks for network changes // and re-gathers candidates when changes are detected. This is only used with GatherContinually policy. -func (a *Agent) startNetworkMonitoring(ctx context.Context) { +func (a *Agent) startNetworkMonitoring(ctx context.Context, generation uint64) { ticker := time.NewTicker(a.networkMonitorInterval) defer ticker.Stop() @@ -1458,7 +1436,7 @@ func (a *Agent) startNetworkMonitoring(ctx context.Context) { return case <-ticker.C: if a.detectNetworkChanges() { - a.gatherCandidatesInternal(ctx) + a.gatherCandidatesInternal(ctx, generation) } } } diff --git a/gather_test.go b/gather_test.go index 88062e7e..33f5bde9 100644 --- a/gather_test.go +++ b/gather_test.go @@ -236,6 +236,79 @@ func TestAgentRestartThenGatherRepeatedly(t *testing.T) { } } +func TestCompleteGatheringIgnoresOldGeneration(t *testing.T) { + agent, err := NewAgent(&AgentConfig{}) + require.NoError(t, err) + defer func() { require.NoError(t, agent.Close()) }() + + completed := make(chan struct{}, 1) + require.NoError(t, agent.OnCandidate(func(candidate Candidate) { + if candidate == nil { + completed <- struct{}{} + } + })) + + require.NoError(t, agent.loop.Run(agent.loop, func(context.Context) { + agent.gatherGeneration = 2 + agent.gatheringState = GatheringStateGathering + })) + require.NoError(t, agent.completeGathering(1)) + + state, err := agent.GetGatheringState() + require.NoError(t, err) + require.Equal(t, GatheringStateGathering, state) + require.Never(t, func() bool { + select { + case <-completed: + return true + default: + return false + } + }, 50*time.Millisecond, time.Millisecond) +} + +func TestContinualRegatherKeepsGeneration(t *testing.T) { + agent, err := NewAgentWithOptions( + WithNet(newHostGatherNet(nil)), + WithNetworkTypes([]NetworkType{NetworkTypeUDP4}), + WithCandidateTypes([]CandidateType{CandidateTypeHost}), + WithMulticastDNSMode(MulticastDNSModeDisabled), + WithNetworkMonitorInterval(time.Millisecond), + WithIncludeLoopback(), + ) + require.NoError(t, err) + defer func() { require.NoError(t, agent.Close()) }() + + candidates := make(chan Candidate, 1) + require.NoError(t, agent.OnCandidate(func(candidate Candidate) { + if candidate != nil { + candidates <- candidate + } + })) + + generation := agent.gatherGeneration + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + agent.startNetworkMonitoring(ctx, generation) + }() + + var candidate Candidate + select { + case candidate = <-candidates: + case <-time.After(time.Second): + require.FailNow(t, "timed out waiting for continual regather") + } + cancel() + <-done + + extension, ok := candidate.GetExtension("generation") + require.True(t, ok) + require.Equal(t, strconv.FormatUint(generation, 10), extension.Value) + require.Equal(t, generation, agent.gatherGeneration) +} + func TestLoopbackCandidate(t *testing.T) { defer test.CheckRoutines(t)() @@ -1459,7 +1532,7 @@ func TestGatherCandidatesRelayCallsAddRelayCandidates(t *testing.T) { } })) - agent.gatherCandidatesRelay(context.Background(), agent.urls) + agent.gatherCandidatesRelay(context.Background(), agent.urls, agent.gatherGeneration) var cand Candidate select { @@ -1521,7 +1594,7 @@ func TestGatherCandidatesRelayRespectsInterfaceFilter(t *testing.T) { require.NoError(t, agent.OnCandidate(func(Candidate) {})) - agent.gatherCandidatesRelay(context.Background(), agent.urls) + agent.gatherCandidatesRelay(context.Background(), agent.urls, agent.gatherGeneration) listenAddrs := netCapture.listenAddresses() require.NotEmpty(t, listenAddrs) @@ -1577,7 +1650,7 @@ func TestGatherCandidatesRelayRespectsNetworkTypeAndTransport(t *testing.T) { // return client, nil } require.NoError(t, agent.OnCandidate(func(Candidate) {})) - agent.gatherCandidatesRelay(context.Background(), agent.urls) + agent.gatherCandidatesRelay(context.Background(), agent.urls, agent.gatherGeneration) require.True(t, client.allocateCalled, "TURN transport must remain independent of candidate family") candidates, err := agent.GetLocalCandidates() require.NoError(t, err) @@ -1639,7 +1712,7 @@ func TestGatherCandidatesRelayRespectsNetworkTypeAndTransport(t *testing.T) { // } })) - agent.gatherCandidatesRelay(context.Background(), agent.urls) + agent.gatherCandidatesRelay(context.Background(), agent.urls, agent.gatherGeneration) select { case <-candidateCh: @@ -1684,7 +1757,7 @@ func TestGatherCandidatesRelayDefaultClientError(t *testing.T) { } })) - agent.gatherCandidatesRelay(context.Background(), agent.urls) + agent.gatherCandidatesRelay(context.Background(), agent.urls, agent.gatherGeneration) select { case <-candidateCh: @@ -1848,7 +1921,7 @@ func TestGatherCandidatesRelayTURNOverTCPProducesUDPRelayCandidate(t *testing.T) } })) - agent.gatherCandidatesRelay(context.Background(), agent.urls) + agent.gatherCandidatesRelay(context.Background(), agent.urls, agent.gatherGeneration) select { case relay := <-relayCandidateCh: @@ -1895,7 +1968,7 @@ func TestGatherCandidatesRelayProxySkipsTURNResolution(t *testing.T) { return nil, errors.New("stop after capturing config") //nolint:err113 // test } - agent.gatherCandidatesRelay(context.Background(), agent.urls) + agent.gatherCandidatesRelay(context.Background(), agent.urls, agent.gatherGeneration) var config *turn.ClientConfig select { @@ -1953,7 +2026,7 @@ func TestGatherCandidatesSrflxMappedPortRangeError(t *testing.T) { agent.portMin = 9000 agent.portMax = 8000 - agent.gatherCandidatesSrflxMapped(context.Background(), []NetworkType{NetworkTypeUDP4}) + agent.gatherCandidatesSrflxMapped(context.Background(), []NetworkType{NetworkTypeUDP4}, agent.gatherGeneration) localCandidates, err := agent.GetLocalCandidates() require.NoError(t, err) @@ -1968,7 +2041,7 @@ func TestGatherCandidatesLocalUDPMux(t *testing.T) { require.NoError(t, agent.Close()) }() - err = agent.gatherCandidatesLocalUDPMux(context.Background()) + err = agent.gatherCandidatesLocalUDPMux(context.Background(), agent.gatherGeneration) require.ErrorIs(t, err, errUDPMuxDisabled) }) @@ -1989,7 +2062,7 @@ func TestGatherCandidatesLocalUDPMux(t *testing.T) { require.NoError(t, agent.OnCandidate(func(Candidate) {})) - err = agent.gatherCandidatesLocalUDPMux(context.Background()) + err = agent.gatherCandidatesLocalUDPMux(context.Background(), agent.gatherGeneration) require.NoError(t, err) candidates, err := agent.GetLocalCandidates() @@ -2030,7 +2103,9 @@ func TestGatherCandidatesSrflxUDPMux(t *testing.T) { require.NoError(t, agent.OnCandidate(func(Candidate) {})) - agent.gatherCandidatesSrflxUDPMux(context.Background(), []*stun.URI{stunURI}, []NetworkType{NetworkTypeUDP4}) + agent.gatherCandidatesSrflxUDPMux( + context.Background(), []*stun.URI{stunURI}, []NetworkType{NetworkTypeUDP4}, agent.gatherGeneration, + ) candidates, err := agent.GetLocalCandidates() require.NoError(t, err) @@ -2073,7 +2148,7 @@ func TestGatherCandidatesSrflxRespectsInterfaceFilter(t *testing.T) { require.NoError(t, agent.OnCandidate(func(Candidate) {})) - agent.gatherCandidatesSrflx(context.Background(), agent.urls, []NetworkType{NetworkTypeUDP4}) + agent.gatherCandidatesSrflx(context.Background(), agent.urls, []NetworkType{NetworkTypeUDP4}, agent.gatherGeneration) listenIPs := netCapture.listenCallIPs() require.NotEmpty(t, listenIPs) @@ -2117,7 +2192,7 @@ func TestGatherCandidatesSrflxUDPMuxRespectsURLTransport(t *testing.T) { Host: "127.0.0.1", Port: 3478, }, - }, []NetworkType{NetworkTypeUDP4}) + }, []NetworkType{NetworkTypeUDP4}, agent.gatherGeneration) candidates, err := agent.GetLocalCandidates() require.NoError(t, err) @@ -2791,7 +2866,7 @@ func TestAddRelayCandidatesWithRewrite(t *testing.T) { require.NoError(t, agent.OnCandidate(func(Candidate) {})) closed := 0 - agent.addRelayCandidates(t.Context(), relayEndpoint{ + agent.addRelayCandidates(t.Context(), agent.gatherGeneration, relayEndpoint{ network: udp, address: net.ParseIP("2001:db8::1"), port: 3478, relAddr: "198.51.100.77", relPort: 50000, conn: newStubPacketConn(nil), onClose: func() error { @@ -2840,7 +2915,7 @@ func TestAddRelayCandidatesSkipsNilConnOrAddress(t *testing.T) { ctx := context.Background() - agent.addRelayCandidates(ctx, relayEndpoint{ + agent.addRelayCandidates(ctx, agent.gatherGeneration, relayEndpoint{ network: NetworkTypeUDP4.String(), address: net.IPv4(10, 0, 0, 1), port: 3478, @@ -2852,7 +2927,7 @@ func TestAddRelayCandidatesSkipsNilConnOrAddress(t *testing.T) { require.NoError(t, err) assert.Len(t, cands, 0) - agent.addRelayCandidates(ctx, relayEndpoint{ + agent.addRelayCandidates(ctx, agent.gatherGeneration, relayEndpoint{ network: NetworkTypeUDP4.String(), address: nil, port: 3478, @@ -2893,7 +2968,7 @@ func TestAddRelayCandidatesSkipsWhenResolveFails(t *testing.T) { agent.loop.Close() }) - agent.addRelayCandidates(context.Background(), relayEndpoint{ + agent.addRelayCandidates(context.Background(), agent.gatherGeneration, relayEndpoint{ network: NetworkTypeUDP4.String(), address: net.IPv4(10, 0, 0, 2), port: 3478, @@ -2932,7 +3007,7 @@ func TestAddRelayCandidatesSkipsWhenResolveFails(t *testing.T) { agent.loop.Close() }) - agent.addRelayCandidates(context.Background(), relayEndpoint{ + agent.addRelayCandidates(context.Background(), agent.gatherGeneration, relayEndpoint{ network: NetworkTypeUDP4.String(), address: net.IPv4(10, 0, 0, 3), port: 3478, @@ -2977,7 +3052,7 @@ func TestCreateRelayCandidateErrorPaths(t *testing.T) { }, } - agent.addRelayCandidates(context.Background(), ep) + agent.addRelayCandidates(context.Background(), agent.gatherGeneration, ep) cands, err := agent.GetLocalCandidates() require.NoError(t, err) @@ -3005,7 +3080,7 @@ func TestCreateRelayCandidateErrorPaths(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() // force addCandidate to fail - agent.addRelayCandidates(ctx, relayEndpoint{ + agent.addRelayCandidates(ctx, agent.gatherGeneration, relayEndpoint{ network: NetworkTypeUDP4.String(), address: net.IPv4(10, 0, 0, 5), port: 3478, @@ -3044,7 +3119,7 @@ func TestGatherCandidatesLocalTCPMuxSkipsUnboundInterfaces(t *testing.T) { }) require.NoError(t, agent.OnCandidate(func(Candidate) {})) - agent.gatherCandidatesLocal(context.Background(), []NetworkType{NetworkTypeTCP4}) + agent.gatherCandidatesLocal(context.Background(), []NetworkType{NetworkTypeTCP4}, agent.gatherGeneration) cands, err := agent.GetLocalCandidates() require.NoError(t, err) @@ -3069,7 +3144,7 @@ func TestGatherCandidatesLocalHostErrorPaths(t *testing.T) { }) require.NoError(t, agent.OnCandidate(func(Candidate) {})) - assert.NoError(t, agent.gatherCandidatesLocalUDPMux(context.Background())) + assert.NoError(t, agent.gatherCandidatesLocalUDPMux(context.Background(), agent.gatherGeneration)) assert.True(t, mux.conn.closed) cands, err := agent.GetLocalCandidates() @@ -3095,7 +3170,7 @@ func TestGatherCandidatesLocalHostErrorPaths(t *testing.T) { agent.includeLoopback = true agent.mDNSName = "invalid-mdns" // no .local suffix -> NewCandidateHost parse fails - agent.gatherCandidatesLocal(context.Background(), []NetworkType{NetworkTypeUDP4}) + agent.gatherCandidatesLocal(context.Background(), []NetworkType{NetworkTypeUDP4}, agent.gatherGeneration) cands, err := agent.GetLocalCandidates() require.NoError(t, err) @@ -3121,7 +3196,7 @@ func TestGatherCandidatesLocalHostErrorPaths(t *testing.T) { agent.loop.Close() - agent.gatherCandidatesLocal(context.Background(), []NetworkType{NetworkTypeUDP4}) + agent.gatherCandidatesLocal(context.Background(), []NetworkType{NetworkTypeUDP4}, agent.gatherGeneration) agent.loop.Run(agent.loop, func(context.Context) { //nolint:errcheck,gosec assert.Empty(t, agent.localCandidates[NetworkTypeUDP4]) @@ -3155,7 +3230,7 @@ func TestGatherCandidatesLocalHostErrorPaths(t *testing.T) { agent.loop.Close() }) - agent.gatherCandidatesLocal(context.Background(), []NetworkType{NetworkTypeUDP4}) + agent.gatherCandidatesLocal(context.Background(), []NetworkType{NetworkTypeUDP4}, agent.gatherGeneration) cands, err := agent.GetLocalCandidates() require.NoError(t, err) @@ -3184,7 +3259,7 @@ func TestGatherCandidatesLocalHostErrorPaths(t *testing.T) { }) require.NoError(t, agent.OnCandidate(func(Candidate) {})) - require.NoError(t, agent.gatherCandidatesLocalUDPMux(context.Background())) + require.NoError(t, agent.gatherCandidatesLocalUDPMux(context.Background(), agent.gatherGeneration)) cands, err := agent.GetLocalCandidates() require.NoError(t, err) @@ -3947,7 +4022,7 @@ func TestGatherAddressRewriteRelayModes(t *testing.T) { require.NoError(t, agent.Close()) }) - agent.addRelayCandidates(context.Background(), relayEndpoint{ + agent.addRelayCandidates(context.Background(), agent.gatherGeneration, relayEndpoint{ network: NetworkTypeUDP4.String(), address: net.ParseIP("192.0.2.10"), port: 5000, @@ -3981,7 +4056,7 @@ func TestGatherAddressRewriteRelayModes(t *testing.T) { require.NoError(t, agent.Close()) }) - agent.addRelayCandidates(context.Background(), relayEndpoint{ + agent.addRelayCandidates(context.Background(), agent.gatherGeneration, relayEndpoint{ network: NetworkTypeUDP4.String(), address: net.ParseIP("192.0.2.20"), port: 6000, @@ -4241,7 +4316,7 @@ func TestGatherCandidatesSrflxMappedMissingExternalIPs(t *testing.T) { }, } - agent.gatherCandidatesSrflxMapped(context.Background(), []NetworkType{NetworkTypeUDP4}) + agent.gatherCandidatesSrflxMapped(context.Background(), []NetworkType{NetworkTypeUDP4}, agent.gatherGeneration) localCandidates, err := agent.GetLocalCandidates() require.NoError(t, err) diff --git a/gather_vnet_test.go b/gather_vnet_test.go index 25f1f7e6..9e7f8c1b 100644 --- a/gather_vnet_test.go +++ b/gather_vnet_test.go @@ -554,5 +554,5 @@ func TestVNetGather_TURNConnectionLeak(t *testing.T) { require.NoError(t, aAgent.Close()) }() - aAgent.gatherCandidatesRelay(context.Background(), []*stun.URI{turnServerURL}) + aAgent.gatherCandidatesRelay(context.Background(), []*stun.URI{turnServerURL}, aAgent.gatherGeneration) }