From bc22926fe06085bca82a5c45175403e3233f1fdc Mon Sep 17 00:00:00 2001 From: riccardom Date: Fri, 26 Jun 2026 17:57:21 +0200 Subject: [PATCH] Drop in case of error, will reconcile with next update --- client/internal/mapsync.go | 42 ++++++++++++++++++----------- client/internal/mapsync_test.go | 47 +++++++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 16 deletions(-) diff --git a/client/internal/mapsync.go b/client/internal/mapsync.go index db06d3495..463f4193c 100644 --- a/client/internal/mapsync.go +++ b/client/internal/mapsync.go @@ -90,14 +90,16 @@ func (m *mapStateManager) run(ctx context.Context) { 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): - } + // Log and DROP this target — do not retry it. A deterministic failure + // (e.g. a malformed peer in the map) would otherwise spin every pass + // making no progress. Management is the source of truth and re-delivers + // the full map on the next sync, so dropping is safe; peers already + // applied this convergence stay (idempotent diffs) and the remainder is + // reconciled by the next target. Mirrors the legacy handleSync path, + // where the apply error was logged by the gRPC client and the update + // dropped. No onConverged: this target did not converge. + log.Errorf("apply sync pass, dropping update: %v", err) + m.markProcessed(tg) continue } @@ -108,16 +110,24 @@ func (m *mapStateManager) run(ctx context.Context) { } // 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 { + if m.markProcessed(tg) && m.onConverged != nil { m.onConverged(time.Since(setAt)) } // if a newer target arrived mid-pass, ag apply it } } + +// markProcessed records that the loop has finished working generation tg (whether +// it converged or the pass was dropped on error), so it 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 behind so that target re-applies. +// Returns true when tg was the latest target (i.e. this was a genuine settle). +func (m *mapStateManager) markProcessed(tg uint64) bool { + m.mu.Lock() + defer m.mu.Unlock() + if m.targetGen != tg { + return false + } + m.appliedGen = tg + return true +} diff --git a/client/internal/mapsync_test.go b/client/internal/mapsync_test.go index 866b34aa2..9ba0043b2 100644 --- a/client/internal/mapsync_test.go +++ b/client/internal/mapsync_test.go @@ -2,6 +2,7 @@ package internal import ( "context" + "errors" "sync/atomic" "testing" "time" @@ -41,6 +42,52 @@ func TestMapStateManager_ConvergesThenStops(t *testing.T) { require.EqualValues(t, 3, atomic.LoadInt32(&passes), "apply must not run after convergence") } +// an apply error drops the target: no retry of the same target, no onConverged, +// the loop goes idle — and a fresh target is still applied afterwards. +func TestMapStateManager_DropsTargetOnError(t *testing.T) { + applied := make(chan struct{}, 8) + var failNext atomic.Bool + failNext.Store(true) + + apply := func(*mgmProto.SyncResponse) (bool, error) { + applied <- struct{}{} + if failNext.Load() { + return false, errors.New("boom") + } + return false, nil // converge in one pass + } + var converged atomic.Int32 + m := newMapStateManager(apply, func(time.Duration) { converged.Add(1) }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go m.run(ctx) + + // first target errors -> applied once, then dropped (no retry, no onConverged) + require.NoError(t, m.SetTarget(&mgmProto.SyncResponse{})) + select { + case <-applied: + case <-time.After(2 * time.Second): + t.Fatal("errored target not applied") + } + select { + case <-applied: + t.Fatal("errored target must not be retried") + case <-time.After(150 * time.Millisecond): + } + require.EqualValues(t, 0, converged.Load(), "onConverged must not fire on error") + + // a new target is still processed normally and converges + failNext.Store(false) + require.NoError(t, m.SetTarget(&mgmProto.SyncResponse{})) + select { + case <-applied: + case <-time.After(2 * time.Second): + t.Fatal("new target after error not applied") + } + require.Eventually(t, func() bool { return converged.Load() == 1 }, 2*time.Second, 10*time.Millisecond) +} + // 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) {