mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 09:01:31 +00:00
feat: retry a transient reconcile failure before the resync, not after it (#969)
pass and keyPass logged every failure at WARN as "retrying next cycle" and returned the callee's deadline, which for a failed callee is zero. Zero means no deadline, so the next attempt was the resync — five minutes by default. Two things were wrong with that. A fault that clears in seconds left the world unconverged for minutes, when the loop already had the mechanism to ask for sooner. And a terminal failure was announced as recoverable: "retrying next cycle" is false for a 400, which will be a 400 every cycle, forever, at WARN. Both now go through failed(), which classifies with awserrors.IsTerminal — the predicate added in #923 for exactly this decision and until now called by nothing. A terminal error is logged at ERROR and asks for no early retry; any other error asks to be revisited after DefaultErrorRetry. A deadline the callee named survives when it is sooner, and clampRevisit still caps every request at the resync, so a loop can only become more eager than it is today.
This commit is contained in:
Josh Siviour
@@ -0,0 +1,127 @@
|
||||
//test:in-package — pass and keyPass are the decision under test, and driving
|
||||
//them through Run would turn a pure classification into a timing test against
|
||||
//DefaultErrorRetry.
|
||||
|
||||
package reconciler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/awserrors"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
// A terminal failure repeats identically however often it is retried, so asking
|
||||
// to be revisited early only produces the same error sooner. The resync stays as
|
||||
// the backstop, which is why the deadline is left alone rather than suppressed.
|
||||
func TestPass_ATerminalFailureAsksForNoEarlyRetry(t *testing.T) {
|
||||
cfg := Config{
|
||||
Name: "terminal",
|
||||
Reconcile: func(context.Context) (time.Duration, error) {
|
||||
return 0, errors.New(awserrors.ErrorInvalidParameterValue)
|
||||
},
|
||||
}
|
||||
|
||||
assert.Zero(t, pass(t.Context(), cfg, "test"),
|
||||
"a 400 will be a 400 next time; retrying early only logs the same failure sooner")
|
||||
}
|
||||
|
||||
// The case the deadline exists for: a fault that clears on its own should not
|
||||
// leave the world unconverged until the resync, which is five minutes by default.
|
||||
func TestPass_ATransientFailureIsRetriedBeforeTheResync(t *testing.T) {
|
||||
cfg := Config{
|
||||
Name: "transient",
|
||||
Reconcile: func(context.Context) (time.Duration, error) {
|
||||
return 0, errors.New(awserrors.ErrorServiceUnavailable)
|
||||
},
|
||||
}
|
||||
|
||||
assert.Equal(t, DefaultErrorRetry, pass(t.Context(), cfg, "test"))
|
||||
}
|
||||
|
||||
// An error carrying no AWS code at all is the common case — a wrapped NATS or
|
||||
// S3 failure — and IsTerminal calls it transient by design, so it is retried
|
||||
// rather than silently abandoned.
|
||||
func TestPass_AnUnclassifiableFailureIsTreatedAsTransient(t *testing.T) {
|
||||
cfg := Config{
|
||||
Name: "unclassified",
|
||||
Reconcile: func(context.Context) (time.Duration, error) {
|
||||
return 0, errors.New("read zone: connection reset")
|
||||
},
|
||||
}
|
||||
|
||||
assert.Equal(t, DefaultErrorRetry, pass(t.Context(), cfg, "test"))
|
||||
}
|
||||
|
||||
// A pass partway through a transition names its own deadline, and that deadline
|
||||
// is the reason the loop revisits at all. The error retry may bring a revisit
|
||||
// forward but must never push one back.
|
||||
func TestPass_ACalleesSoonerDeadlineSurvivesATransientFailure(t *testing.T) {
|
||||
cfg := Config{
|
||||
Name: "sooner",
|
||||
Reconcile: func(context.Context) (time.Duration, error) {
|
||||
return 2 * time.Second, errors.New(awserrors.ErrorServiceUnavailable)
|
||||
},
|
||||
}
|
||||
|
||||
assert.Equal(t, 2*time.Second, pass(t.Context(), cfg, "test"),
|
||||
"the callee asked for 2s; the error retry must not defer it to DefaultErrorRetry")
|
||||
}
|
||||
|
||||
// A callee asking for later than the error retry gets the retry instead, which
|
||||
// is the same direction: sooner wins.
|
||||
func TestPass_ACalleesLaterDeadlineYieldsToTheErrorRetry(t *testing.T) {
|
||||
cfg := Config{
|
||||
Name: "later",
|
||||
Reconcile: func(context.Context) (time.Duration, error) {
|
||||
return time.Hour, errors.New(awserrors.ErrorServiceUnavailable)
|
||||
},
|
||||
}
|
||||
|
||||
assert.Equal(t, DefaultErrorRetry, pass(t.Context(), cfg, "test"))
|
||||
}
|
||||
|
||||
// The per-key path carries the same classification. A key is otherwise covered
|
||||
// only by the whole-set resync, so a transient key failure is the case with the
|
||||
// most to gain from an early revisit.
|
||||
func TestKeyPass_ClassifiesTheSameWayAsAWholeSetPass(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
err error
|
||||
want time.Duration
|
||||
}{
|
||||
{name: "terminal", err: errors.New(awserrors.ErrorInvalidParameterValue), want: 0},
|
||||
{name: "transient", err: errors.New(awserrors.ErrorServiceUnavailable), want: DefaultErrorRetry},
|
||||
{name: "unclassifiable", err: errors.New("put key: no responders"), want: DefaultErrorRetry},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
var got string
|
||||
cfg := Config{
|
||||
Name: "keys",
|
||||
ReconcileKey: func(_ context.Context, key string) (time.Duration, error) {
|
||||
got = key
|
||||
return 0, tt.err
|
||||
},
|
||||
}
|
||||
|
||||
assert.Equal(t, tt.want, keyPass(t.Context(), cfg, "i-abc"))
|
||||
assert.Equal(t, "i-abc", got, "the key that failed has to reach the callee")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A pass that succeeds keeps whatever deadline it named, so the error retry
|
||||
// cannot leak into the healthy path and make every loop revisit on a timer it
|
||||
// never asked for.
|
||||
func TestPass_ASuccessfulPassKeepsItsOwnDeadline(t *testing.T) {
|
||||
cfg := Config{
|
||||
Name: "ok",
|
||||
Reconcile: func(context.Context) (time.Duration, error) { return 90 * time.Second, nil },
|
||||
}
|
||||
|
||||
assert.Equal(t, 90*time.Second, pass(t.Context(), cfg, "test"))
|
||||
}
|
||||
@@ -23,6 +23,7 @@ import (
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/awserrors"
|
||||
"github.com/mulgadc/spinifex/spinifex/kvstore"
|
||||
"github.com/mulgadc/spinifex/spinifex/otelsetup"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
@@ -40,6 +41,10 @@ const (
|
||||
// be established, so a bucket that is briefly unreachable is retried
|
||||
// without spinning.
|
||||
retryWatch = 5 * time.Second
|
||||
// DefaultErrorRetry is how soon a pass that failed transiently asks to be
|
||||
// revisited. Waiting out the resync leaves the world unconverged for minutes
|
||||
// over a fault that usually clears in seconds.
|
||||
DefaultErrorRetry = 15 * time.Second
|
||||
)
|
||||
|
||||
// Source names the buckets to watch and the keys within them that matter.
|
||||
@@ -332,6 +337,23 @@ func settle(ctx context.Context, changes <-chan struct{}, debounce time.Duration
|
||||
}
|
||||
}
|
||||
|
||||
// failed reports a reconcile failure and returns the deadline it should be
|
||||
// retried on. A terminal error will fail identically however often it is
|
||||
// repeated, so it is reported as needing intervention and asks for nothing
|
||||
// sooner than the resync; anything else is retried promptly.
|
||||
//
|
||||
// A deadline the callee asked for survives when it is sooner: a pass partway
|
||||
// through a transition still gets revisited on its own schedule.
|
||||
func failed(ctx context.Context, loop, what string, err error, revisit time.Duration, attrs ...any) time.Duration {
|
||||
attrs = append([]any{"loop", loop}, append(attrs, "err", err)...)
|
||||
if awserrors.IsTerminal(err) {
|
||||
slog.ErrorContext(ctx, what+" failed and will not succeed unchanged", attrs...)
|
||||
return revisit
|
||||
}
|
||||
slog.WarnContext(ctx, what+" failed, retrying", attrs...)
|
||||
return Earliest(revisit, DefaultErrorRetry)
|
||||
}
|
||||
|
||||
// pass runs one reconcile, logging rather than returning a failure: the loop
|
||||
// outlives any single pass, and the next resync retries. A failed pass keeps
|
||||
// its deadline, so a caller partway through a transition still gets revisited
|
||||
@@ -340,9 +362,8 @@ func pass(ctx context.Context, cfg Config, trigger string) time.Duration {
|
||||
start := time.Now()
|
||||
revisit, err := cfg.Reconcile(ctx)
|
||||
if err != nil {
|
||||
slog.WarnContext(ctx, "reconcile pass failed, retrying next cycle",
|
||||
"loop", cfg.Name, "trigger", trigger, "duration_ms", otelsetup.Millis(time.Since(start)), "err", err)
|
||||
return revisit
|
||||
return failed(ctx, cfg.Name, "reconcile pass", err, revisit,
|
||||
"trigger", trigger, "duration_ms", otelsetup.Millis(time.Since(start)))
|
||||
}
|
||||
slog.DebugContext(ctx, "reconcile pass complete",
|
||||
"loop", cfg.Name, "trigger", trigger, "duration_ms", otelsetup.Millis(time.Since(start)),
|
||||
@@ -350,16 +371,15 @@ func pass(ctx context.Context, cfg Config, trigger string) time.Duration {
|
||||
return revisit
|
||||
}
|
||||
|
||||
// keyPass runs one per-key reconcile. Its failure is logged rather than
|
||||
// retried here: the resync re-reconciles the whole set, so a key that failed is
|
||||
// covered by the same backstop as a key the queue never saw.
|
||||
// keyPass runs one per-key reconcile. The resync re-reconciles the whole set, so
|
||||
// a key that failed is covered by the same backstop as a key the queue never
|
||||
// saw; the deadline a transient failure asks for only brings that forward.
|
||||
func keyPass(ctx context.Context, cfg Config, key string) time.Duration {
|
||||
start := time.Now()
|
||||
revisit, err := cfg.ReconcileKey(ctx, key)
|
||||
if err != nil {
|
||||
slog.WarnContext(ctx, "reconcile key failed, retrying on the next resync",
|
||||
"loop", cfg.Name, "key", key, "duration_ms", otelsetup.Millis(time.Since(start)), "err", err)
|
||||
return revisit
|
||||
return failed(ctx, cfg.Name, "reconcile key", err, revisit,
|
||||
"key", key, "duration_ms", otelsetup.Millis(time.Since(start)))
|
||||
}
|
||||
slog.DebugContext(ctx, "reconcile key complete",
|
||||
"loop", cfg.Name, "key", key, "duration_ms", otelsetup.Millis(time.Since(start)),
|
||||
|
||||
Reference in New Issue
Block a user