Adds map state manager

client/internal/engine.go

@@ -914,32 +914,10 @@ func (e *Engine) handleAutoUpdateVersion(autoUpdateSettings *mgmProto.AutoUpdate
 	e.updateManager.SetVersion(autoUpdateSettings.Version, autoUpdateSettings.AlwaysUpdate)
 }
 
-// handleSync processes one sync update to convergence. The peer apply is bounded
-// per pass (see maxPeersPerSyncPass), so syncMsgMux is acquired and released once
-// per pass via applySyncPass — the signal handler interleaves between passes.
-// Returning with no error means the map was fully applied (converged).
-func (e *Engine) handleSync(update *mgmProto.SyncResponse) error {
-	started := time.Now()
-	defer func() {
-		duration := time.Since(started)
-		log.Infof("sync finished in %s", duration)
-		e.clientMetrics.RecordSyncDuration(e.ctx, duration)
-	}()
-
-	for {
-		more, err := e.applySyncPass(update)
-		if err != nil {
-			return err
-		}
-		if !more {
-			return nil
-		}
-		// lock released between passes -> signal handler can interleave
-	}
-}
-
 // applySyncPass applies one bounded pass of the sync update under syncMsgMux and
-// returns true if more peers remained than the per-pass cap.
+// returns true if more peers remained than the per-pass cap. It is driven by the
+// mapStateManager, which re-invokes it (releasing the lock between passes) until
+// the update is fully applied.
 func (e *Engine) applySyncPass(update *mgmProto.SyncResponse) (bool, error) {
 	e.syncMsgMux.Lock()
 	defer e.syncMsgMux.Unlock()
@@ -1318,7 +1296,19 @@ func (e *Engine) receiveManagementEvents() {
 			e.config.DisableSSHAuth,
 		)
 
-		err = e.mgmClient.Sync(e.ctx, info, e.handleSync)
+		// The map-state manager converges the latest update in the background in
+		// bounded passes; the stream callback only hands it the newest target.
+		manager := newMapStateManager(e.applySyncPass, func(d time.Duration) {
+			log.Infof("sync finished in %s", d)
+			e.clientMetrics.RecordSyncDuration(e.ctx, d)
+		})
+		e.shutdownWg.Add(1)
+		go func() {
+			defer e.shutdownWg.Done()
+			manager.run(e.ctx)
+		}()
+
+		err = e.mgmClient.Sync(e.ctx, info, manager.SetTarget)
 		if err != nil {
 			// happens if management is unavailable for a long time.
 			// We want to cancel the operation of the whole client

client/internal/mapsync.go

@@ -0,0 +1,118 @@
+package internal
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	mgmProto "github.com/netbirdio/netbird/shared/management/proto"
+)
+
+// mapStateManager is the single read/write point between the management stream
+// (writes) and the convergence loop (reads/applies).
+//
+// The stream calls SetTarget with the latest full SyncResponse — the complete
+// desired state. A single background goroutine (run) applies it to the engine in
+// bounded passes via apply() until converged, releasing syncMsgMux between passes
+// so other subsystems interleave. If a newer update arrives mid-flight, the loop
+// keeps converging toward the latest target.
+//
+// State is a single comparison: appliedGen == targetGen means converged.
+// targetGen increments on every SetTarget (an internal generation counter, so it
+// also covers config-only updates that carry no network-map serial).
+type mapStateManager struct {
+	// apply performs one bounded apply pass and reports whether more passes are needed.
+	apply func(*mgmProto.SyncResponse) (bool, error)
+	// onConverged is called once per target when it is fully applied, with the
+	// elapsed time since that target was set (for the sync-duration metric).
+	onConverged func(time.Duration)
+
+	mu          sync.Mutex
+	target      *mgmProto.SyncResponse
+	targetGen   uint64
+	appliedGen  uint64
+	targetSetAt time.Time
+
+	wake chan struct{}
+}
+
+func newMapStateManager(apply func(*mgmProto.SyncResponse) (bool, error), onConverged func(time.Duration)) *mapStateManager {
+	return &mapStateManager{
+		apply:       apply,
+		onConverged: onConverged,
+		wake:        make(chan struct{}, 1),
+	}
+}
+
+// SetTarget records the latest update as the desired state and wakes the loop.
+// It returns immediately; convergence happens in the background. Serial-based
+// staleness of the network map is still enforced inside apply (updateNetworkMap).
+func (m *mapStateManager) SetTarget(update *mgmProto.SyncResponse) error {
+	m.mu.Lock()
+	m.target = update
+	m.targetGen++
+	m.targetSetAt = time.Now()
+	m.mu.Unlock()
+
+	select {
+	case m.wake <- struct{}{}:
+	default:
+	}
+	return nil
+}
+
+// run drives convergence until ctx is done. It is meant to run in its own goroutine.
+func (m *mapStateManager) run(ctx context.Context) {
+	for {
+		m.mu.Lock()
+		target, tg, ag, setAt := m.target, m.targetGen, m.appliedGen, m.targetSetAt
+		m.mu.Unlock()
+
+		// Fully converged (or nothing yet): block until a new target arrives.
+		if target == nil || ag == tg {
+			select {
+			case <-ctx.Done():
+				return
+			case <-m.wake:
+				continue
+			}
+		}
+
+		more, err := m.apply(target)
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			log.Errorf("apply sync pass: %v", err)
+			// avoid a tight error loop; retry on the next target or after a short delay
+			select {
+			case <-ctx.Done():
+				return
+			case <-m.wake:
+			case <-time.After(time.Second):
+			}
+			continue
+		}
+
+		if more {
+			// keep converging the current target; syncMsgMux was released by apply
+			// between passes so other subsystems interleave.
+			continue
+		}
+
+		// This pass converged. Mark applied only if no newer target arrived during it.
+		m.mu.Lock()
+		converged := m.targetGen == tg
+		if converged {
+			m.appliedGen = tg
+		}
+		m.mu.Unlock()
+
+		if converged && m.onConverged != nil {
+			m.onConverged(time.Since(setAt))
+		}
+		// if a newer target arrived mid-pass, ag<tg next iteration -> apply it
+	}
+}

client/internal/mapsync_test.go

@@ -0,0 +1,78 @@
+package internal
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	mgmProto "github.com/netbirdio/netbird/shared/management/proto"
+)
+
+// converges over the bounded passes (apply returns more until the 3rd pass),
+// fires onConverged exactly once, then blocks (no further apply) until a new target.
+func TestMapStateManager_ConvergesThenStops(t *testing.T) {
+	var passes int32
+	converged := make(chan struct{}, 1)
+
+	apply := func(*mgmProto.SyncResponse) (bool, error) {
+		n := atomic.AddInt32(&passes, 1)
+		return n < 3, nil // more on pass 1 and 2, converge on pass 3
+	}
+	m := newMapStateManager(apply, func(time.Duration) { converged <- struct{}{} })
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go m.run(ctx)
+
+	require.NoError(t, m.SetTarget(&mgmProto.SyncResponse{}))
+
+	select {
+	case <-converged:
+	case <-time.After(2 * time.Second):
+		t.Fatal("manager did not converge")
+	}
+	require.EqualValues(t, 3, atomic.LoadInt32(&passes))
+
+	// once converged the loop blocks: no further apply calls
+	time.Sleep(100 * time.Millisecond)
+	require.EqualValues(t, 3, atomic.LoadInt32(&passes), "apply must not run after convergence")
+}
+
+// a new target after convergence triggers a fresh apply; an idle (converged)
+// manager does not apply on its own.
+func TestMapStateManager_ReappliesOnNewTarget(t *testing.T) {
+	applied := make(chan struct{}, 8)
+	apply := func(*mgmProto.SyncResponse) (bool, error) {
+		applied <- struct{}{}
+		return false, nil // converge in one pass
+	}
+	m := newMapStateManager(apply, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go m.run(ctx)
+
+	require.NoError(t, m.SetTarget(&mgmProto.SyncResponse{}))
+	select {
+	case <-applied:
+	case <-time.After(2 * time.Second):
+		t.Fatal("first target not applied")
+	}
+
+	// converged → must stay idle (no spurious apply)
+	select {
+	case <-applied:
+		t.Fatal("unexpected apply while idle/converged")
+	case <-time.After(150 * time.Millisecond):
+	}
+
+	require.NoError(t, m.SetTarget(&mgmProto.SyncResponse{}))
+	select {
+	case <-applied:
+	case <-time.After(2 * time.Second):
+		t.Fatal("new target not applied")
+	}
+}