Move Checking -> Failed handling out of Selection

Selection before was written in a way that wouldn't work for restarts.
This moves the code out and makes it stateless. In the future we will be
able to clear the candidates and update the ConnectionState and
everything will restart properly.

Relates to #190

agent.go

@@ -380,32 +380,63 @@ func (a *Agent) startConnectivityChecks(isControlling bool, remoteUfrag, remoteP
 		// TODO this should be dynamic, and grow when the connection is stable
 		a.requestConnectivityCheck()
 		agent.connectivityTicker = time.NewTicker(a.taskLoopInterval)
+		go a.connectivityChecks()
+	})
+}
 
-		go func() {
-			contact := func() {
-				if err := a.run(func(a *Agent) {
-					a.selector.ContactCandidates()
-				}); err != nil {
-					a.log.Warnf("taskLoop failed: %v", err)
+func (a *Agent) connectivityChecks() {
+	lastConnectionState := ConnectionState(0)
+	checkingDuration := time.Time{}
+
+	contact := func() {
+		if err := a.run(func(a *Agent) {
+			defer func() {
+				lastConnectionState = a.connectionState
+			}()
+
+			switch a.connectionState {
+			case ConnectionStateFailed:
+				// The connection is currently failed so don't send any checks
+				// In the future it may be restarted though
+				return
+			case ConnectionStateChecking:
+				// We have just entered checking for the first time so update our checking timer
+				if lastConnectionState != a.connectionState {
+					checkingDuration = time.Now()
 				}
-			}
 
-			for {
-				select {
-				case <-a.forceCandidateContact:
-					contact()
-				case <-a.connectivityTicker.C:
-					contact()
-				case <-a.done:
+				// We have been in checking longer then Disconnect+Failed timeout, set the connection to Failed
+				if time.Since(checkingDuration) > a.disconnectTimeout+a.failedTimeout {
+					a.updateConnectionState(ConnectionStateFailed)
 					return
 				}
 			}
-		}()
-	})
+
+			a.selector.ContactCandidates()
+		}); err != nil {
+			a.log.Warnf("taskLoop failed: %v", err)
+		}
+	}
+
+	for {
+		select {
+		case <-a.forceCandidateContact:
+			contact()
+		case <-a.connectivityTicker.C:
+			contact()
+		case <-a.done:
+			return
+		}
+	}
 }
 
 func (a *Agent) updateConnectionState(newState ConnectionState) {
 	if a.connectionState != newState {
+		// Connection has gone to failed, release all gathered candidates
+		if newState == ConnectionStateFailed {
+			a.deleteAllCandidates()
+		}
+
 		a.log.Infof("Setting new connection state: %s", newState)
 		a.connectionState = newState
 
@@ -527,7 +558,6 @@ func (a *Agent) validateSelectedPair() bool {
 
 	switch {
 	case totalTimeToFailure != 0 && disconnectedTime > totalTimeToFailure:
-		a.deleteAllCandidates()
 		a.updateConnectionState(ConnectionStateFailed)
 	case a.disconnectTimeout != 0 && disconnectedTime > a.disconnectTimeout:
 		a.updateConnectionState(ConnectionStateDisconnected)

agent_test.go

@@ -357,20 +357,24 @@ func TestConnectivityOnStartup(t *testing.T) {
 	report := test.CheckRoutines(t)
 	defer report()
 
-	stunServerURL := &URL{
-		Scheme: SchemeTypeSTUN,
-		Host:   "1.2.3.4",
-		Port:   3478,
-		Proto:  ProtoTypeUDP,
-	}
+	// Create a network with two interfaces
+	wan, err := vnet.NewRouter(&vnet.RouterConfig{
+		CIDR:          "0.0.0.0/0",
+		LoggerFactory: logging.NewDefaultLoggerFactory(),
+	})
+	assert.NoError(t, err)
 
-	natType := &vnet.NATType{
-		MappingBehavior:   vnet.EndpointIndependent,
-		FilteringBehavior: vnet.EndpointIndependent,
-	}
-	v, err := buildVNet(natType, natType)
-	require.NoError(t, err, "should succeed")
-	defer v.close()
+	net0 := vnet.NewNet(&vnet.NetConfig{
+		StaticIPs: []string{"192.168.0.1"},
+	})
+	assert.NoError(t, wan.AddNet(net0))
+
+	net1 := vnet.NewNet(&vnet.NetConfig{
+		StaticIPs: []string{"192.168.0.2"},
+	})
+	assert.NoError(t, wan.AddNet(net1))
+
+	assert.NoError(t, wan.Start())
 
 	aNotifier, aConnected := onConnected()
 	bNotifier, bConnected := onConnected()
@@ -379,11 +383,10 @@ func TestConnectivityOnStartup(t *testing.T) {
 	wg.Add(2)
 
 	cfg0 := &AgentConfig{
-		Urls:             []*URL{stunServerURL},
 		Trickle:          true,
 		NetworkTypes:     supportedNetworkTypes,
 		MulticastDNSMode: MulticastDNSModeDisabled,
-		Net:              v.net0,
+		Net:              net0,
 
 		taskLoopInterval: time.Hour,
 	}
@@ -399,11 +402,10 @@ func TestConnectivityOnStartup(t *testing.T) {
 	require.NoError(t, aAgent.GatherCandidates())
 
 	cfg1 := &AgentConfig{
-		Urls:             []*URL{stunServerURL},
 		Trickle:          true,
 		NetworkTypes:     supportedNetworkTypes,
 		MulticastDNSMode: MulticastDNSModeDisabled,
-		Net:              v.net1,
+		Net:              net1,
 		taskLoopInterval: time.Hour,
 	}
 
@@ -425,6 +427,7 @@ func TestConnectivityOnStartup(t *testing.T) {
 	<-aConnected
 	<-bConnected
 
+	assert.NoError(t, wan.Stop())
 	if !closePipe(t, aConn, bConn) {
 		return
 	}
@@ -1269,9 +1272,9 @@ func TestAgentCredentials(t *testing.T) {
 	assert.EqualError(t, err, ErrLocalPwdInsufficientBits.Error())
 }
 
-// Assert that Agent on Failure flushes all existing candidates
+// Assert that Agent on Failure deletes all existing candidates
 // User can then do an ICE Restart to bring agent back
-func TestConnectionStateFailedFlushCandidates(t *testing.T) {
+func TestConnectionStateFailedDeleteAllCandidates(t *testing.T) {
 	lim := test.TimeOut(time.Second * 5)
 	defer lim.Stop()
 
@@ -1282,7 +1285,6 @@ func TestConnectionStateFailedFlushCandidates(t *testing.T) {
 	KeepaliveInterval := time.Duration(0)
 
 	cfg := &AgentConfig{
-		Urls:              []*URL{},
 		NetworkTypes:      supportedNetworkTypes,
 		DisconnectTimeout: &oneSecond,
 		FailedTimeout:     &oneSecond,
@@ -1297,18 +1299,84 @@ func TestConnectionStateFailedFlushCandidates(t *testing.T) {
 	assert.NoError(t, err)
 
 	isFailed := make(chan interface{})
-	err = aAgent.OnConnectionStateChange(func(c ConnectionState) {
+	assert.NoError(t, aAgent.OnConnectionStateChange(func(c ConnectionState) {
 		if c == ConnectionStateFailed {
 			close(isFailed)
 		}
-	})
-	assert.NoError(t, err)
+	}))
 
 	connect(aAgent, bAgent)
 	<-isFailed
 
-	assert.Equal(t, len(aAgent.remoteCandidates), 0)
-	assert.Equal(t, len(bAgent.localCandidates), 0)
+	done := make(chan struct{})
+	assert.NoError(t, aAgent.run(func(agent *Agent) {
+		assert.Equal(t, len(aAgent.remoteCandidates), 0)
+		assert.Equal(t, len(aAgent.localCandidates), 0)
+		close(done)
+	}))
+	<-done
+
+	assert.NoError(t, aAgent.Close())
+	assert.NoError(t, bAgent.Close())
+}
+
+// Assert that the ICE Agent can go directly from Connecting -> Failed on both sides
+func TestConnectionStateConnectingToFailed(t *testing.T) {
+	lim := test.TimeOut(time.Second * 5)
+	defer lim.Stop()
+
+	report := test.CheckRoutines(t)
+	defer report()
+
+	oneSecond := time.Second
+	KeepaliveInterval := time.Duration(0)
+
+	cfg := &AgentConfig{
+		DisconnectTimeout: &oneSecond,
+		FailedTimeout:     &oneSecond,
+		KeepaliveInterval: &KeepaliveInterval,
+		taskLoopInterval:  250 * time.Millisecond,
+	}
+
+	aAgent, err := NewAgent(cfg)
+	assert.NoError(t, err)
+
+	bAgent, err := NewAgent(cfg)
+	assert.NoError(t, err)
+
+	var isFailed sync.WaitGroup
+	var isChecking sync.WaitGroup
+
+	isFailed.Add(2)
+	isChecking.Add(2)
+
+	connectionStateCheck := func(c ConnectionState) {
+		switch c {
+		case ConnectionStateFailed:
+			isFailed.Done()
+		case ConnectionStateChecking:
+			isChecking.Done()
+		case ConnectionStateConnected:
+		case ConnectionStateCompleted:
+			t.Errorf("Unexpected ConnectionState: %v", c)
+		}
+	}
+
+	assert.NoError(t, aAgent.OnConnectionStateChange(connectionStateCheck))
+	assert.NoError(t, bAgent.OnConnectionStateChange(connectionStateCheck))
+
+	go func() {
+		_, err := aAgent.Accept(context.TODO(), "InvalidFrag", "InvalidPwd")
+		assert.Error(t, err)
+	}()
+
+	go func() {
+		_, err := bAgent.Dial(context.TODO(), "InvalidFrag", "InvalidPwd")
+		assert.Error(t, err)
+	}()
+
+	isChecking.Wait()
+	isFailed.Wait()
 
 	assert.NoError(t, aAgent.Close())
 	assert.NoError(t, bAgent.Close())

selection.go

@@ -26,31 +26,6 @@ type controllingSelector struct {
 
 func (s *controllingSelector) Start() {
 	s.startTime = time.Now()
-	go func() {
-		select {
-		case <-s.agent.done:
-			return
-		case <-time.After(s.agent.candidateSelectionTimeout):
-		}
-
-		err := s.agent.run(func(a *Agent) {
-			if s.nominatedPair == nil {
-				p := s.agent.getBestValidCandidatePair()
-				if p == nil {
-					s.log.Trace("check timeout reached and no valid candidate pair found, marking connection as failed")
-					s.agent.updateConnectionState(ConnectionStateFailed)
-				} else {
-					s.log.Tracef("check timeout reached, nominating (%s, %s)", p.local.String(), p.remote.String())
-					s.nominatedPair = p
-					s.nominatePair(p)
-				}
-			}
-		})
-
-		if err != nil {
-			s.log.Errorf("error processing checkCandidatesTimeout handler %v", err.Error())
-		}
-	}()
 }
 
 func (s *controllingSelector) isNominatable(c Candidate) bool {
@@ -193,13 +168,11 @@ func (s *controllingSelector) PingCandidate(local, remote Candidate) {
 }
 
 type controlledSelector struct {
-	startTime time.Time
-	agent     *Agent
-	log       logging.LeveledLogger
+	agent *Agent
+	log   logging.LeveledLogger
 }
 
 func (s *controlledSelector) Start() {
-	s.startTime = time.Now()
 }
 
 func (s *controlledSelector) ContactCandidates() {
@@ -209,12 +182,7 @@ func (s *controlledSelector) ContactCandidates() {
 			s.agent.checkKeepalive()
 		}
 	} else {
-		if time.Since(s.startTime) > s.agent.candidateSelectionTimeout {
-			s.log.Trace("check timeout reached and no valid candidate pair found, marking connection as failed")
-			s.agent.updateConnectionState(ConnectionStateFailed)
-		} else {
-			s.agent.pingAllCandidates()
-		}
+		s.agent.pingAllCandidates()
 	}
 }