[client] Serialize notifier state publication

The effective state was computed under serverStateLock but published after
releasing it, so two transitions could reorder between compute and notify
and leave the listener on a state the notifier had already superseded,
e.g. NoNetwork surviving after the network came back.

Hold a publish lock across compute and notify on every publishing path.

client/internal/peer/notifier.go

@@ -5,6 +5,10 @@ import (
 )
 
 type notifier struct {
+	// publishLock orders state publication: it is held across computing the
+	// effective state and handing it to the listener, so a transition cannot
+	// overtake a newer one and leave the listener on a stale state.
+	publishLock        sync.Mutex
 	serverStateLock    sync.Mutex
 	listenersLock      sync.Mutex
 	listener           Listener
@@ -37,6 +41,9 @@ func (n *notifier) effectiveState(state ClientState) ClientState {
 // the listener when the flag flips the effective state (Connecting <->
 // NoNetwork).
 func (n *notifier) setNetworkAvailable(available bool) {
+	n.publishLock.Lock()
+	defer n.publishLock.Unlock()
+
 	n.serverStateLock.Lock()
 	if n.networkAvailable == available {
 		n.serverStateLock.Unlock()
@@ -78,6 +85,9 @@ func (n *notifier) removeListener() {
 }
 
 func (n *notifier) updateServerStates(mgmState bool, signalState bool) {
+	n.publishLock.Lock()
+	defer n.publishLock.Unlock()
+
 	n.serverStateLock.Lock()
 	calculatedState := n.calculateState(mgmState, signalState)
 
@@ -94,6 +104,9 @@ func (n *notifier) updateServerStates(mgmState bool, signalState bool) {
 }
 
 func (n *notifier) clientStart() {
+	n.publishLock.Lock()
+	defer n.publishLock.Unlock()
+
 	n.serverStateLock.Lock()
 	n.currentClientState = true
 	n.lastNotification = ClientStateConnecting
@@ -104,6 +117,9 @@ func (n *notifier) clientStart() {
 }
 
 func (n *notifier) clientStop() {
+	n.publishLock.Lock()
+	defer n.publishLock.Unlock()
+
 	n.serverStateLock.Lock()
 	n.currentClientState = false
 	n.lastNotification = ClientStateDisconnected
@@ -113,6 +129,9 @@ func (n *notifier) clientStop() {
 }
 
 func (n *notifier) clientTearDown() {
+	n.publishLock.Lock()
+	defer n.publishLock.Unlock()
+
 	n.serverStateLock.Lock()
 	n.currentClientState = false
 	n.lastNotification = ClientStateDisconnecting

client/internal/peer/notifier_concurrent_test.go

@@ -0,0 +1,108 @@
+package peer
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+type recordingListener struct {
+	mu      sync.Mutex
+	states  []ClientState
+	onState func(ClientState)
+}
+
+func (l *recordingListener) OnStateChanged(state ClientState) {
+	l.mu.Lock()
+	l.states = append(l.states, state)
+	hook := l.onState
+	l.mu.Unlock()
+
+	if hook != nil {
+		hook(state)
+	}
+}
+
+func (l *recordingListener) last() (ClientState, bool) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if len(l.states) == 0 {
+		return 0, false
+	}
+	return l.states[len(l.states)-1], true
+}
+
+func (l *recordingListener) snapshot() []ClientState {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return append([]ClientState(nil), l.states...)
+}
+
+func (l *recordingListener) OnConnected()                    {}
+func (l *recordingListener) OnDisconnected()                 {}
+func (l *recordingListener) OnConnecting()                   {}
+func (l *recordingListener) OnDisconnecting()                {}
+func (l *recordingListener) OnAddressChanged(string, string) {}
+func (l *recordingListener) OnPeersListChanged(int)          {}
+
+// TestNotifier_ConcurrentAvailabilityFlipOrdersPublication holds the first
+// transition inside the listener callback and flips availability again from
+// another goroutine while it is parked. The second flip must not publish
+// ahead of the one in flight, otherwise the listener ends up on a state the
+// notifier already superseded.
+func TestNotifier_ConcurrentAvailabilityFlipOrdersPublication(t *testing.T) {
+	n := newNotifier()
+	n.currentClientState = true
+	n.lastNotification = ClientStateConnecting
+
+	entered := make(chan struct{})
+	release := make(chan struct{})
+
+	l := &recordingListener{}
+	l.onState = func(state ClientState) {
+		if state != ClientStateNoNetwork {
+			return
+		}
+		l.mu.Lock()
+		l.onState = nil
+		l.mu.Unlock()
+		close(entered)
+		<-release
+	}
+	n.listener = l
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		n.setNetworkAvailable(false)
+	}()
+
+	<-entered
+
+	flipped := make(chan struct{})
+	go func() {
+		defer close(flipped)
+		n.setNetworkAvailable(true)
+	}()
+
+	select {
+	case <-flipped:
+		t.Fatal("the online transition published while the offline one was " +
+			"still in flight; publication is not serialized")
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	close(release)
+	<-flipped
+	wg.Wait()
+
+	got, ok := l.last()
+	if !ok {
+		t.Fatal("listener never observed a state")
+	}
+	if got != ClientStateConnecting {
+		t.Fatalf("listener holds %v after the network came back, want Connecting; sequence: %v",
+			got, l.snapshot())
+	}
+}