From 5bec1e8f03947c0c973ae2aadc0e7a533373764a Mon Sep 17 00:00:00 2001 From: riccardom Date: Fri, 26 Jun 2026 17:02:43 +0200 Subject: [PATCH] Adds map state manager --- client/internal/engine.go | 42 +++++------- client/internal/mapsync.go | 118 ++++++++++++++++++++++++++++++++ client/internal/mapsync_test.go | 78 +++++++++++++++++++++ 3 files changed, 212 insertions(+), 26 deletions(-) create mode 100644 client/internal/mapsync.go create mode 100644 client/internal/mapsync_test.go diff --git a/client/internal/engine.go b/client/internal/engine.go index d64532155..692ad1e50 100644 --- a/client/internal/engine.go +++ b/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 diff --git a/client/internal/mapsync.go b/client/internal/mapsync.go new file mode 100644 index 000000000..87d2b3553 --- /dev/null +++ b/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 apply it + } +} diff --git a/client/internal/mapsync_test.go b/client/internal/mapsync_test.go new file mode 100644 index 000000000..866b34aa2 --- /dev/null +++ b/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") + } +}