[client, android, ios] Coalesce peer list change notifications to the mobile listener (#7546)

* [client] Coalesce peer list change notifications to the mobile listener

Every peer state change spawned a goroutine to call the platform
listener. During a reconnect storm this pinned hundreds of OS threads
in JNI and let the UI call back into the engine from each of them.
Deliver peer list changes from a single goroutine per listener and
collapse pending changes into the latest count.

* [client] Cover a pending wake-up when the peer list deliverer is replaced

The replacement test waited for the old callback to finish before
swapping listeners, so it never exercised the stop check that runs
after a wake-up. Block the old callback, queue a peer list change and
swap while it is blocked, then assert the old listener never sees the
new count.

* [client] Signal peer list deliverer exit and wait for it in the test

The replacement test sampled the old listener after a fixed sleep, so a
late stale delivery could slip past it. Close a done channel when the
deliverer goroutine returns and let the test wait on it instead.

* [client] Drop the test-only peer list deliverer exit channel

The done channel was only read by the replacement test. Production code
cannot wait on it, since joining the deliverer would block on a mobile
callback. The tests now drive the deliverer loop directly and check that
setListener and removeListener close its stop channel.

client/internal/peer/notifier.go

@@ -12,6 +12,8 @@ type notifier struct {
 	serverStateLock    sync.Mutex
 	listenersLock      sync.Mutex
 	listener           Listener
+	peerListWake       chan struct{}
+	peerListStop       chan struct{}
 	currentClientState bool
 	lastNotification   ClientState
 	lastNumberOfPeers  int
@@ -62,7 +64,6 @@ func (n *notifier) setNetworkAvailable(available bool) {
 func (n *notifier) setListener(listener Listener) {
 	n.serverStateLock.Lock()
 	lastNotification := n.effectiveState(n.lastNotification)
-	numOfPeers := n.lastNumberOfPeers
 	fqdnAddress := n.lastFqdnAddress
 	address := n.lastIPAddress
 	n.serverStateLock.Unlock()
@@ -70,17 +71,19 @@ func (n *notifier) setListener(listener Listener) {
 	n.listenersLock.Lock()
 	defer n.listenersLock.Unlock()
 
+	n.stopPeerListDelivererLocked()
 	n.listener = listener
 
 	listener.OnAddressChanged(fqdnAddress, address)
 	notifyListener(listener, lastNotification)
-	// run on go routine to avoid on Java layer to call go functions on same thread
+	n.startPeerListDelivererLocked(listener)
-	go listener.OnPeersListChanged(numOfPeers)
+	n.wakePeerListDelivererLocked()
 }
 
 func (n *notifier) removeListener() {
 	n.listenersLock.Lock()
 	defer n.listenersLock.Unlock()
+	n.stopPeerListDelivererLocked()
 	n.listener = nil
 }
 
@@ -178,15 +181,56 @@ func (n *notifier) peerListChanged(numOfPeers int) {
 	n.serverStateLock.Unlock()
 
 	n.listenersLock.Lock()
-	listener := n.listener
+	defer n.listenersLock.Unlock()
-	n.listenersLock.Unlock()
+	n.wakePeerListDelivererLocked()
+}
 
-	if listener == nil {
+func (n *notifier) startPeerListDelivererLocked(listener Listener) {
+	wake := make(chan struct{}, 1)
+	stop := make(chan struct{})
+	n.peerListWake = wake
+	n.peerListStop = stop
+	go n.deliverPeerListChanges(listener, wake, stop)
+}
+
+func (n *notifier) stopPeerListDelivererLocked() {
+	if n.peerListStop == nil {
 		return
 	}
+	close(n.peerListStop)
+	n.peerListStop = nil
+	n.peerListWake = nil
+}
 
-	// run on go routine to avoid on Java layer to call go functions on same thread
+func (n *notifier) wakePeerListDelivererLocked() {
-	go listener.OnPeersListChanged(numOfPeers)
+	if n.peerListWake == nil {
+		return
+	}
+	select {
+	case n.peerListWake <- struct{}{}:
+	default:
+	}
+}
+
+func (n *notifier) deliverPeerListChanges(listener Listener, wake <-chan struct{}, stop <-chan struct{}) {
+	for {
+		select {
+		case <-stop:
+			return
+		case <-wake:
+		}
+		select {
+		case <-stop:
+			return
+		default:
+		}
+
+		n.serverStateLock.Lock()
+		numOfPeers := n.lastNumberOfPeers
+		n.serverStateLock.Unlock()
+
+		listener.OnPeersListChanged(numOfPeers)
+	}
 }
 
 func (n *notifier) localAddressChanged(fqdn, address string) {

client/internal/peer/notifier_test.go

@@ -2,7 +2,9 @@ package peer
 
 import (
 	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 )
 
 type mocListener struct {
@@ -115,3 +117,156 @@ func Test_notifier_RemoveListener(t *testing.T) {
 		t.Errorf("invalid state: %d", listener.peers)
 	}
 }
+
+type coalescingListener struct {
+	final       int
+	calls       atomic.Int32
+	inFlight    atomic.Int32
+	maxInFlight atomic.Int32
+	last        atomic.Int32
+	done        chan struct{}
+	entered     chan struct{}
+	release     chan struct{}
+	once        sync.Once
+}
+
+func (l *coalescingListener) OnStateChanged(ClientState)      {}
+func (l *coalescingListener) OnConnected()                    {}
+func (l *coalescingListener) OnDisconnected()                 {}
+func (l *coalescingListener) OnConnecting()                   {}
+func (l *coalescingListener) OnDisconnecting()                {}
+func (l *coalescingListener) OnAddressChanged(string, string) {}
+
+func (l *coalescingListener) OnPeersListChanged(size int) {
+	current := l.inFlight.Add(1)
+	for {
+		seen := l.maxInFlight.Load()
+		if current <= seen || l.maxInFlight.CompareAndSwap(seen, current) {
+			break
+		}
+	}
+	if l.calls.Add(1) == 1 && l.entered != nil {
+		close(l.entered)
+	}
+	if l.release != nil {
+		<-l.release
+	}
+	time.Sleep(time.Millisecond)
+	l.last.Store(int32(size))
+	l.inFlight.Add(-1)
+	if size == l.final {
+		l.once.Do(func() { close(l.done) })
+	}
+}
+
+func Test_notifier_PeerListChangedCoalesces(t *testing.T) {
+	const events = 1000
+	listener := &coalescingListener{final: events, done: make(chan struct{})}
+	n := newNotifier()
+	n.setListener(listener)
+
+	for i := 1; i <= events; i++ {
+		n.peerListChanged(i)
+	}
+
+	select {
+	case <-listener.done:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("last peer count not delivered, last seen: %d", listener.last.Load())
+	}
+
+	if got := listener.maxInFlight.Load(); got != 1 {
+		t.Errorf("concurrent deliveries: %d, expected 1", got)
+	}
+	if got := listener.calls.Load(); got >= events {
+		t.Errorf("deliveries not coalesced: %d calls for %d events", got, events)
+	}
+}
+
+func Test_notifier_SetListenerStopsPreviousDeliverer(t *testing.T) {
+	old := &coalescingListener{final: -1}
+	replacement := &coalescingListener{final: 7, done: make(chan struct{})}
+	n := newNotifier()
+	n.setListener(old)
+	oldStop := n.peerListStop
+
+	n.peerListChanged(7)
+	n.setListener(replacement)
+
+	select {
+	case <-oldStop:
+	default:
+		t.Fatal("old deliverer not stopped on listener replacement")
+	}
+	waitFor(t, replacement.done, "replacement listener not notified")
+}
+
+func Test_notifier_RemoveListenerStopsDeliverer(t *testing.T) {
+	n := newNotifier()
+	n.setListener(&coalescingListener{final: -1})
+	stop := n.peerListStop
+
+	n.removeListener()
+
+	select {
+	case <-stop:
+	default:
+		t.Fatal("deliverer not stopped on listener removal")
+	}
+}
+
+func Test_notifier_DelivererExitsAfterInFlightCallback(t *testing.T) {
+	listener := &coalescingListener{
+		final:   -1,
+		entered: make(chan struct{}),
+		release: make(chan struct{}),
+	}
+	n := newNotifier()
+	wake := make(chan struct{}, 1)
+	stop := make(chan struct{})
+	exited := make(chan struct{})
+	go func() {
+		n.deliverPeerListChanges(listener, wake, stop)
+		close(exited)
+	}()
+
+	wake <- struct{}{}
+	waitFor(t, listener.entered, "listener not called")
+
+	n.peerListChanged(7)
+	wake <- struct{}{}
+	close(stop)
+	close(listener.release)
+
+	waitFor(t, exited, "deliverer did not exit after stop")
+	if got := listener.calls.Load(); got != 1 {
+		t.Errorf("deliverer ran %d callbacks after stop, expected only the in-flight one", got)
+	}
+	if got := listener.last.Load(); got == 7 {
+		t.Errorf("deliverer delivered the peer count queued after stop")
+	}
+}
+
+func Test_notifier_DelivererPrefersStopOverPendingWake(t *testing.T) {
+	listener := &coalescingListener{final: -1}
+	n := newNotifier()
+	wake := make(chan struct{}, 1)
+	stop := make(chan struct{})
+
+	wake <- struct{}{}
+	close(stop)
+	n.deliverPeerListChanges(listener, wake, stop)
+
+	if got := listener.calls.Load(); got != 0 {
+		t.Errorf("deliverer ran %d callbacks with stop closed, expected 0", got)
+	}
+}
+
+func waitFor(t *testing.T, ch <-chan struct{}, msg string) {
+	t.Helper()
+	select {
+	case <-ch:
+	case <-time.After(5 * time.Second):
+		t.Fatal(msg)
+	}
+}