From 0bf964dad785cb3775ee142769ffbf4dc84ac4a3 Mon Sep 17 00:00:00 2001 From: riccardom Date: Sun, 28 Jun 2026 11:27:58 +0200 Subject: [PATCH] Do not process intermediate one if new ones are fresher just use the freshest --- client/internal/mapsync.go | 57 ++++++++++++++++----------------- client/internal/mapsync_test.go | 56 ++++++++++++++++++++++++++------ 2 files changed, 75 insertions(+), 38 deletions(-) diff --git a/client/internal/mapsync.go b/client/internal/mapsync.go index 9254c4728..083c841bd 100644 --- a/client/internal/mapsync.go +++ b/client/internal/mapsync.go @@ -17,33 +17,29 @@ import ( // 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 -// coalesces: it keeps converging toward the latest target rather than replaying -// the intermediate ones. +// coalesces: it keeps converging toward the latest target and the intermediate one +// is SKIPPED — never applied on its own (logged, no onConverged). // // Convergence is a single comparison: appliedGen == targetGen. targetGen // increments on every SetTarget (an internal generation counter, so it also covers // config-only updates that carry no network-map serial). // -// onConverged still fires once per MAP, not once per convergence: every update's -// receive time is recorded in `pending` and the whole list is drained on each -// settle. So a map that was superseded mid-flight (its apply coalesced into a newer -// target) still gets its finish-sync signal when the client reaches a converged -// state covering it — the signal is delayed at worst, never lost. That log/metric -// is a strong problem indicator and must not silently disappear. +// onConverged fires once for each — and only each — map that is actually processed +// (i.e. converged as the target). Skipped/superseded maps and dropped-on-error maps +// do NOT fire it. So "sync finished in X" / RecordSyncDuration always corresponds +// to a real, completed alignment. 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 applied map, with the elapsed time since that + // 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) - mu sync.Mutex - target *mgmProto.SyncResponse - targetGen uint64 - appliedGen uint64 - // pending holds the receive time of every update accepted since the last settle, - // oldest first. Drained on convergence so onConverged fires once per map. - pending []time.Time + mu sync.Mutex + target *mgmProto.SyncResponse + targetGen uint64 + appliedGen uint64 + targetSetAt time.Time wake chan struct{} } @@ -61,6 +57,12 @@ func newMapStateManager(apply func(*mgmProto.SyncResponse) (bool, error), onConv // staleness of the network map is still enforced inside apply (updateNetworkMap). func (m *mapStateManager) SetTarget(update *mgmProto.SyncResponse) error { m.mu.Lock() + // A target that has not settled yet (targetGen > appliedGen) is being superseded + // before it converged: we coalesce to the latest map and never apply this one on + // its own. It is SKIPPED — logged here, and it will not fire onConverged. + if m.target != nil && m.targetGen > m.appliedGen { + log.Debugf("sync map (gen %d) superseded before convergence, skipping", m.targetGen) + } m.target = update // Bump an internal generation counter, NOT the map serial: config-only updates // (relay token rotation, STUN/TURN) arrive with NetworkMap == nil and carry no @@ -68,7 +70,7 @@ func (m *mapStateManager) SetTarget(update *mgmProto.SyncResponse) error { // target regardless of payload. Map-serial staleness is enforced separately // inside apply (updateNetworkMap). m.targetGen++ - m.pending = append(m.pending, time.Now()) + m.targetSetAt = time.Now() m.mu.Unlock() select { @@ -119,21 +121,21 @@ func (m *mapStateManager) run(ctx context.Context) { continue } - // This pass converged. Mark applied + signal every pending map. + // This pass converged. Mark applied and signal this one map. m.settle(tg, true) // if a newer target arrived mid-pass, settle is a no-op (targetGen != tg) and - // ag apply it; pending carries over until the next settle. + // ag apply it; this generation was skipped (logged in + // SetTarget) and is not signaled. } } // settle marks generation tg as processed so the loop goes idle instead of // re-applying the same target. It is a no-op when a newer target arrived during the -// pass (targetGen != tg), leaving appliedGen + pending behind so that target -// re-applies and its maps are signaled at the next settle. +// pass (targetGen != tg), leaving appliedGen behind so that target re-applies — the +// just-finished generation was already counted as skipped. // -// When signal is true (the pass converged) it drains pending and fires onConverged -// once per map. When false (the target was dropped on error) it discards pending -// without signaling — those maps did not converge; management re-delivers them. +// When signal is true (the pass converged) it fires onConverged once for this map; +// when false (the target was dropped on error) it does not — the map did not converge. func (m *mapStateManager) settle(tg uint64, signal bool) { m.mu.Lock() if m.targetGen != tg { @@ -141,13 +143,10 @@ func (m *mapStateManager) settle(tg uint64, signal bool) { return } m.appliedGen = tg - toSignal := m.pending - m.pending = nil + setAt := m.targetSetAt m.mu.Unlock() if signal && m.onConverged != nil { - for _, setAt := range toSignal { - m.onConverged(time.Since(setAt)) - } + m.onConverged(time.Since(setAt)) } } diff --git a/client/internal/mapsync_test.go b/client/internal/mapsync_test.go index 6208660b6..600499dd9 100644 --- a/client/internal/mapsync_test.go +++ b/client/internal/mapsync_test.go @@ -42,14 +42,14 @@ func TestMapStateManager_ConvergesThenStops(t *testing.T) { require.EqualValues(t, 3, atomic.LoadInt32(&passes), "apply must not run after convergence") } -// every map management sends is reported via onConverged exactly once, in order -// — mirroring the legacy per-message handleSync timing. -func TestMapStateManager_SignalsEveryMap(t *testing.T) { - var converged atomic.Int32 +// 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) { + converged := make(chan struct{}, 8) apply := func(*mgmProto.SyncResponse) (bool, error) { - return false, nil // each map converges in one pass + return false, nil // converge in one pass } - m := newMapStateManager(apply, func(time.Duration) { converged.Add(1) }) + m := newMapStateManager(apply, func(time.Duration) { converged <- struct{}{} }) ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -58,12 +58,50 @@ func TestMapStateManager_SignalsEveryMap(t *testing.T) { const maps = 3 for i := 0; i < maps; i++ { require.NoError(t, m.SetTarget(&mgmProto.SyncResponse{})) + select { // wait for this map to converge before sending the next (no coalescing) + case <-converged: + case <-time.After(2 * time.Second): + t.Fatalf("map %d not signaled", i) + } } - require.Eventually(t, func() bool { return converged.Load() == maps }, 2*time.Second, 10*time.Millisecond) // no extra signals once the stream goes quiet - time.Sleep(100 * time.Millisecond) - require.EqualValues(t, maps, converged.Load()) + select { + case <-converged: + t.Fatal("unexpected extra onConverged") + case <-time.After(100 * time.Millisecond): + } +} + +// a map superseded before it converges is skipped: only the latest (processed) map +// fires onConverged, not the skipped one. +func TestMapStateManager_SkippedMapNotSignaled(t *testing.T) { + release := make(chan struct{}) + var applies, converged atomic.Int32 + apply := func(*mgmProto.SyncResponse) (bool, error) { + applies.Add(1) + <-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) }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go m.run(ctx) + + // map1 is picked up; its apply blocks on release + require.NoError(t, m.SetTarget(&mgmProto.SyncResponse{})) + require.Eventually(t, func() bool { return applies.Load() >= 1 }, 2*time.Second, 5*time.Millisecond) + + // map2 supersedes map1 before it settled -> map1 is skipped + require.NoError(t, m.SetTarget(&mgmProto.SyncResponse{})) + close(release) // let both applies proceed + + // only the processed (latest) map signals; the skipped one does not + require.Eventually(t, func() bool { return converged.Load() == 1 }, 2*time.Second, 10*time.Millisecond) + time.Sleep(150 * time.Millisecond) + require.EqualValues(t, 1, converged.Load(), "skipped map must not fire onConverged") + require.EqualValues(t, 2, applies.Load(), "both targets entered apply (map1 once, map2 once)") } // an apply error drops the target: no retry of the same target, no onConverged,