diff --git a/client/internal/engine.go b/client/internal/engine.go index 970e3447a..825244360 100644 --- a/client/internal/engine.go +++ b/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) }) diff --git a/client/internal/mapsync.go b/client/internal/mapsync.go index e30e3c1da..cd4c2b01e 100644 --- a/client/internal/mapsync.go +++ b/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 } diff --git a/client/internal/mapsync_test.go b/client/internal/mapsync_test.go index ba2736d26..7ae551ef5 100644 --- a/client/internal/mapsync_test.go +++ b/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()