Prevents skipping of intermediate map updates potentially not applied

by moving the persistence from applySync to the map state manager

client/internal/engine.go

@@ -955,8 +955,6 @@ func (e *Engine) applySyncPass(update *mgmProto.SyncResponse, firstPass bool) (b
 		return false, err
 	}
 
-	e.persistSyncResponse(update)
-
 	// only apply new changes and ignore old ones
 	more, err := e.updateNetworkMap(nm, maxPeersPerSyncPass, firstPass)
 	if err != nil {
@@ -1011,6 +1009,13 @@ func (e *Engine) updateNetbirdConfig(wCfg *mgmProto.NetbirdConfig) error {
 // (not syncMsgMux) is held for the whole Set so the store cannot be cleared (disabled /
 // engine close) mid-call and have this write resurrect a file that was just removed.
 func (e *Engine) persistSyncResponse(update *mgmProto.SyncResponse) {
+	// Only persist updates that carry a network map. Config-only updates (e.g. relay
+	// token rotation, STUN/TURN) have a nil NetworkMap; persisting them would overwrite
+	// the last full map on disk and break restore-on-restart.
+	if update.GetNetworkMap() == nil {
+		return
+	}
+
 	e.syncRespMux.RLock()
 	defer e.syncRespMux.RUnlock()
 
@@ -1304,7 +1309,7 @@ func (e *Engine) receiveManagementEvents() {
 
 		// 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) {
+		manager := newMapStateManager(e.applySyncPass, e.persistSyncResponse, func(d time.Duration) {
 			log.Infof("sync finished in %s", d)
 			e.clientMetrics.RecordSyncDuration(e.ctx, d)
 		})

client/internal/mapsync.go

@@ -38,6 +38,11 @@ type mapStateManager struct {
 	// onConverged is called once per processed map, with the elapsed time since that
 	// map was received (for the sync-duration metric / "sync finished" log).
 	onConverged func(time.Duration)
+	// persist snapshots an update to disk for restore-on-restart. Called once per
+	// update received from management (in SetTarget), including ones later coalesced
+	// or skipped from apply, so the on-disk state mirrors what management last sent.
+	// The impl skips config-only updates (nil NetworkMap). May be nil.
+	persist func(*mgmProto.SyncResponse)
 
 	mu          sync.Mutex
 	target      *mgmProto.SyncResponse
@@ -48,9 +53,10 @@ type mapStateManager struct {
 	wake chan struct{}
 }
 
-func newMapStateManager(apply func(update *mgmProto.SyncResponse, firstPass bool) (bool, error), onConverged func(time.Duration)) *mapStateManager {
+func newMapStateManager(apply func(update *mgmProto.SyncResponse, firstPass bool) (bool, error), persist func(*mgmProto.SyncResponse), onConverged func(time.Duration)) *mapStateManager {
 	return &mapStateManager{
 		apply:       apply,
+		persist:     persist,
 		onConverged: onConverged,
 		wake:        make(chan struct{}, 1),
 	}
@@ -81,6 +87,15 @@ func (m *mapStateManager) SetTarget(update *mgmProto.SyncResponse) error {
 	case m.wake <- struct{}{}:
 	default:
 	}
+
+	// Persist every update received from management — once per update (not per apply
+	// pass), and including ones that get coalesced/skipped from apply, so the on-disk
+	// state always reflects the latest map management sent. Done after waking the loop
+	// so convergence can start in parallel with the disk write. The persist impl skips
+	// config-only updates (nil NetworkMap).
+	if m.persist != nil {
+		m.persist(update)
+	}
 	return nil
 }
 

client/internal/mapsync_test.go

@@ -26,7 +26,7 @@ func TestMapStateManager_ConvergesThenStops(t *testing.T) {
 		}
 		return n < 3, nil // more on pass 1 and 2, converge on pass 3
 	}
-	m := newMapStateManager(apply, func(time.Duration) { converged <- struct{}{} })
+	m := newMapStateManager(apply, nil, func(time.Duration) { converged <- struct{}{} })
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -47,6 +47,56 @@ func TestMapStateManager_ConvergesThenStops(t *testing.T) {
 	require.EqualValues(t, 3, atomic.LoadInt32(&passes), "apply must not run after convergence")
 }
 
+// persist runs once per received update (not per apply pass), regardless of how many
+// bounded passes that target takes to converge.
+func TestMapStateManager_PersistsOncePerUpdate(t *testing.T) {
+	var passes, persists int32
+	converged := make(chan struct{}, 1)
+	apply := func(_ *mgmProto.SyncResponse, _ bool) (bool, error) {
+		n := atomic.AddInt32(&passes, 1)
+		return n < 3, nil // 3 passes for one target
+	}
+	persist := func(*mgmProto.SyncResponse) { atomic.AddInt32(&persists, 1) }
+	m := newMapStateManager(apply, persist, 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("did not converge")
+	}
+	require.EqualValues(t, 3, atomic.LoadInt32(&passes))
+	require.EqualValues(t, 1, atomic.LoadInt32(&persists), "persist once per update, not per pass")
+}
+
+// every update received from management is persisted — even one that is coalesced /
+// skipped from apply before it ever converges.
+func TestMapStateManager_PersistsEveryUpdateIncludingSkipped(t *testing.T) {
+	release := make(chan struct{})
+	var persists int32
+	apply := func(_ *mgmProto.SyncResponse, _ bool) (bool, error) {
+		<-release // hold the first apply so the second update coalesces/skips
+		return false, nil
+	}
+	persist := func(*mgmProto.SyncResponse) { atomic.AddInt32(&persists, 1) }
+	m := newMapStateManager(apply, persist, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go m.run(ctx)
+
+	require.NoError(t, m.SetTarget(&mgmProto.SyncResponse{})) // map1 -> apply blocks
+	require.NoError(t, m.SetTarget(&mgmProto.SyncResponse{})) // map2 supersedes map1 (skipped from apply)
+	close(release)
+
+	// both updates persisted even though map1 is skipped from apply
+	require.Eventually(t, func() bool { return atomic.LoadInt32(&persists) == 2 }, 2*time.Second, 10*time.Millisecond)
+}
+
 // each map that is actually processed (converged before the next arrives) fires
 // onConverged exactly once — mirroring the legacy per-message handleSync timing.
 func TestMapStateManager_SignalsEachProcessedMap(t *testing.T) {
@@ -54,7 +104,7 @@ func TestMapStateManager_SignalsEachProcessedMap(t *testing.T) {
 	apply := func(_ *mgmProto.SyncResponse, _ bool) (bool, error) {
 		return false, nil // converge in one pass
 	}
-	m := newMapStateManager(apply, func(time.Duration) { converged <- struct{}{} })
+	m := newMapStateManager(apply, nil, func(time.Duration) { converged <- struct{}{} })
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -88,7 +138,7 @@ func TestMapStateManager_SkippedMapNotSignaled(t *testing.T) {
 		<-release // hold the first apply in-flight so we can queue a newer target
 		return false, nil
 	}
-	m := newMapStateManager(apply, func(time.Duration) { converged.Add(1) })
+	m := newMapStateManager(apply, nil, func(time.Duration) { converged.Add(1) })
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -124,7 +174,7 @@ func TestMapStateManager_DropsTargetOnError(t *testing.T) {
 		return false, nil // converge in one pass
 	}
 	var converged atomic.Int32
-	m := newMapStateManager(apply, func(time.Duration) { converged.Add(1) })
+	m := newMapStateManager(apply, nil, func(time.Duration) { converged.Add(1) })
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -163,7 +213,7 @@ func TestMapStateManager_ReappliesOnNewTarget(t *testing.T) {
 		applied <- struct{}{}
 		return false, nil // converge in one pass
 	}
-	m := newMapStateManager(apply, nil)
+	m := newMapStateManager(apply, nil, nil)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()