Repository: pion/ice
Pull request #656 (2024-03-13 15:57:11)

Ensure sequential execution of connection state callbacks
Ice update connection state callbacks need to happen sequentially otherwise a later callback could overwrite a newer callback. 

Fixes: #624 

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pion/ice/pull/656?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
Attention: Patch coverage is `0%` with `8 lines` in your changes are missing coverage. Please review.
> Project coverage is 23.20%. Comparing base [(`cb8a47a`)](https://app.codecov.io/gh/pion/ice/commit/cb8a47ac376bb3754a1f611076de323014d73084?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) to head [(`a7d084f`)](https://app.codecov.io/gh/pion/ice/pull/656?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).

| [Files](https://app.codecov.io/gh/pion/ice/pull/656?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Patch % | Lines |
|---|---|---|
| [agent\_handlers.go](https://app.codecov.io/gh/pion/ice/pull/656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-YWdlbnRfaGFuZGxlcnMuZ28=) | 0.00% | [7 Missing :warning: ](https://app.codecov.io/gh/pion/ice/pull/656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) |
| [agent.go](https://app.codecov.io/gh/pion/ice/pull/656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-YWdlbnQuZ28=) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/pion/ice/pull/656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) |

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##           master     #656       +/-   ##
===========================================
- Coverage   78.78%   23.20%   -55.58%     
===========================================
  Files          42       37        -5     
  Lines        4431     4326      -105     
===========================================
- Hits         3491     1004     -2487     
- Misses        723     3282     +2559     
+ Partials      217       40      -177     
```

| [Flag](https://app.codecov.io/gh/pion/ice/pull/656/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Coverage Δ | |
|---|---|---|
| [go](https://app.codecov.io/gh/pion/ice/pull/656/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | `?` | |
| [wasm](https://app.codecov.io/gh/pion/ice/pull/656/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | `23.20% <0.00%> (-0.03%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#carryforward-flags-in-the-pull-request-comment) to find out more.


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/pion/ice/pull/656?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).


Patch:
diff --git a/agent.go b/agent.go
index c31d7ad..afaec76 100644
--- a/agent.go
+++ b/agent.go
@@ -130,7 +130,7 @@ type Agent struct {
 
 	chanCandidate     chan Candidate
 	chanCandidatePair chan *CandidatePair
-	chanState         chan ConnectionState
+	stateNotifier     *connectionStateNotifier
 
 	loggerFactory logging.LoggerFactory
 	log           logging.LeveledLogger
@@ -227,7 +227,6 @@ func (a *Agent) taskLoop() {
 
 		after()
 
-		close(a.chanState)
 		close(a.chanCandidate)
 		close(a.chanCandidatePair)
 		close(a.taskLoopDone)
@@ -278,7 +277,6 @@ func NewAgent(config *AgentConfig) (*Agent, error) { //nolint:gocognit
 
 	a := &Agent{
 		chanTask:          make(chan task),
-		chanState:         make(chan ConnectionState),
 		chanCandidate:     make(chan Candidate),
 		chanCandidatePair: make(chan *CandidatePair),
 		tieBreaker:        globalMathRandomGenerator.Uint64(),
@@ -322,6 +320,7 @@ func NewAgent(config *AgentConfig) (*Agent, error) { //nolint:gocognit
 
 		disableActiveTCP: config.DisableActiveTCP,
 	}
+	a.stateNotifier = &connectionStateNotifier{NotificationFunc: a.onConnectionStateChange}
 
 	if a.net == nil {
 		a.net, err = stdnet.NewNet()
@@ -369,7 +368,6 @@ func NewAgent(config *AgentConfig) (*Agent, error) { //nolint:gocognit
 	// Blocking one by the other one causes deadlock.
 	// Hence, we call handlers from independent Goroutines.
 	go a.candidatePairRoutine()
-	go a.connectionStateRoutine()
 	go a.candidateRoutine()
 
 	// Restart is also used to initialize the agent for the first time
@@ -503,12 +501,7 @@ func (a *Agent) updateConnectionState(newState ConnectionState) {
 
 		a.log.Infof("Setting new connection state: %s", newState)
 		a.connectionState = newState
-
-		// Call handler after finishing current task since we may be holding the agent lock
-		// and the handler may also require it
-		a.afterRun(func(_ context.Context) {
-			a.chanState <- newState
-		})
+		a.stateNotifier.Enqueue(newState)
 	}
 }
 
diff --git a/agent_handlers.go b/agent_handlers.go
index c5a5ec0..4ec2484 100644
--- a/agent_handlers.go
+++ b/agent_handlers.go
@@ -3,6 +3,8 @@
 
 package ice
 
+import "sync"
+
 // OnConnectionStateChange sets a handler that is fired when the connection state changes
 func (a *Agent) OnConnectionStateChange(f func(ConnectionState)) error {
 	a.onConnectionStateChangeHdlr.Store(f)
@@ -47,9 +49,35 @@ func (a *Agent) candidatePairRoutine() {
 	}
 }
 
-func (a *Agent) connectionStateRoutine() {
-	for s := range a.chanState {
-		go a.onConnectionStateChange(s)
+type connectionStateNotifier struct {
+	sync.Mutex
+	states           []ConnectionState
+	running          bool
+	NotificationFunc func(ConnectionState)
+}
+
+func (c *connectionStateNotifier) Enqueue(s ConnectionState) {
+	c.Lock()
+	defer c.Unlock()
+	c.states = append(c.states, s)
+	if !c.running {
+		c.running = true
+		go c.notify()
+	}
+}
+
+func (c *connectionStateNotifier) notify() {
+	for {
+		c.Lock()
+		if len(c.states) == 0 {
+			c.running = false
+			c.Unlock()
+			return
+		}
+		s := c.states[0]
+		c.states = c.states[1:]
+		c.Unlock()
+		c.NotificationFunc(s)
 	}
 }
 


Test patch:
diff --git a/agent_handlers_test.go b/agent_handlers_test.go
new file mode 100644
index 0000000..05ed5b8
--- /dev/null
+++ b/agent_handlers_test.go
@@ -0,0 +1,71 @@
+// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+package ice
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pion/transport/v3/test"
+)
+
+func TestConnectionStateNotifier(t *testing.T) {
+	t.Run("TestManyUpdates", func(t *testing.T) {
+		report := test.CheckRoutines(t)
+		defer report()
+		updates := make(chan struct{}, 1)
+		c := &connectionStateNotifier{
+			NotificationFunc: func(_ ConnectionState) {
+				updates <- struct{}{}
+			},
+		}
+		// Enqueue all updates upfront to ensure that it
+		// doesn't block
+		for i := 0; i < 10000; i++ {
+			c.Enqueue(ConnectionStateNew)
+		}
+		done := make(chan struct{})
+		go func() {
+			for i := 0; i < 10000; i++ {
+				<-updates
+			}
+			select {
+			case <-updates:
+				t.Errorf("received more updates than expected")
+			case <-time.After(1 * time.Second):
+			}
+			close(done)
+		}()
+		<-done
+	})
+	t.Run("TestUpdateOrdering", func(t *testing.T) {
+		report := test.CheckRoutines(t)
+		defer report()
+		updates := make(chan ConnectionState)
+		c := &connectionStateNotifier{
+			NotificationFunc: func(cs ConnectionState) {
+				updates <- cs
+			},
+		}
+		done := make(chan struct{})
+		go func() {
+			for i := 0; i < 10000; i++ {
+				x := <-updates
+				if x != ConnectionState(i) {
+					t.Errorf("expected %d got %d", x, i)
+				}
+			}
+			select {
+			case <-updates:
+				t.Errorf("received more updates than expected")
+			case <-time.After(1 * time.Second):
+			}
+			close(done)
+		}()
+		for i := 0; i < 10000; i++ {
+			c.Enqueue(ConnectionState(i))
+		}
+		<-done
+	})
+}