diff --git a/internal/meshmapper/scopes.go b/internal/meshmapper/scopes.go index 23363e6..5eb0d7b 100644 --- a/internal/meshmapper/scopes.go +++ b/internal/meshmapper/scopes.go @@ -26,6 +26,8 @@ const ( MaxScopes = 64 // per source; at most 16 configured sources (1024 imported identities) MaxBody = 64 << 10 PollInterval = 15 * time.Second // one source per tick, at most four requests/minute + + failureRetry = 5 * time.Minute // transient failures shouldn't cost a whole refresh interval ) // Cache stores source provenance and freshness separately from packet evidence. @@ -142,6 +144,7 @@ func (i *Importer) refresh(ctx context.Context, s *source, now time.Time) error response, err := i.client.Do(request) var entries []scopestore.Entry var generated time.Time + var retryAfter time.Time if err != nil { if ctx.Err() != nil { return ctx.Err() @@ -169,10 +172,7 @@ func (i *Importer) refresh(ctx context.Context, s *source, now time.Time) error if response.StatusCode == http.StatusTooManyRequests || response.StatusCode == http.StatusServiceUnavailable { // The published contract uses seconds. Ignore invalid/overflowing headers. if seconds, parseErr := strconv.ParseInt(response.Header.Get("Retry-After"), 10, 64); parseErr == nil && seconds > 0 && seconds <= int64((1<<63-1)/time.Second) { - until := now.Add(time.Duration(seconds) * time.Second) - if until.After(update.NextAttempt) { - update.NextAttempt = until - } + retryAfter = now.Add(time.Duration(seconds) * time.Second) } } } @@ -191,6 +191,13 @@ func (i *Importer) refresh(ctx context.Context, s *source, now time.Time) error } } } + if update.LastError != "" { + // Transient failures shouldn't cost a whole refresh interval, but never beat a Retry-After. + update.NextAttempt = now.Add(min(i.interval, failureRetry)) + if retryAfter.After(update.NextAttempt) { + update.NextAttempt = retryAfter + } + } if update.LastError == "HTTP 429" { i.retryAfter = update.NextAttempt } diff --git a/internal/meshmapper/scopes_test.go b/internal/meshmapper/scopes_test.go index 858ea58..b8e1b24 100644 --- a/internal/meshmapper/scopes_test.go +++ b/internal/meshmapper/scopes_test.go @@ -278,3 +278,42 @@ func TestInvalidSavedCatalogueDoesNotBlockStartupOrRefresh(t *testing.T) { t.Fatalf("normal shutdown is a task failure: %v", err) } } + +func TestRetryAfterShorterThanIntervalIsHonoured(t *testing.T) { + ctx := context.Background() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Retry-After", "1800") + w.WriteHeader(http.StatusTooManyRequests) + })) + defer server.Close() + cfg := config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": server.URL}} + imp, err := New(ctx, cfg, &memoryStore{rows: map[string]Cache{}}, scopestore.New(), nil) + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + if err := imp.refresh(ctx, &imp.sources[0], now); err != nil { + t.Fatal(err) + } + if got := imp.sources[0].cache.NextAttempt.Sub(now); got != 30*time.Minute { + t.Fatalf("next attempt after 429 in %v, want %v (server Retry-After must not be overridden by the 1h interval)", got, 30*time.Minute) + } +} + +func TestFailedRefreshRetriesSoon(t *testing.T) { + ctx := context.Background() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(500) })) + defer server.Close() + cfg := config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": server.URL}} + imp, err := New(ctx, cfg, &memoryStore{rows: map[string]Cache{}}, scopestore.New(), nil) + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + if err := imp.refresh(ctx, &imp.sources[0], now); err != nil { + t.Fatal(err) + } + if got := imp.sources[0].cache.NextAttempt.Sub(now); got != failureRetry { + t.Fatalf("next attempt after failure in %v, want %v", got, failureRetry) + } +}