From 3efa5a8a96e9f757c305b978dcc69ee6757bc975 Mon Sep 17 00:00:00 2001 From: Henry Date: Fri, 4 Sep 2026 09:31:36 +0000 Subject: [PATCH] fix(backend): stop retries from hammering a failed credential after a restart (#3357) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Claude Sonnet 5 Co-authored-by: Carlos Florencio --- .../internal/agent/runtime/dynamic/circuit.go | 91 +++- .../agent/runtime/dynamic/circuit_test.go | 401 ++++++++++++++++++ apps/backend/internal/backendapp/services.go | 1 + 3 files changed, 482 insertions(+), 11 deletions(-) create mode 100644 apps/backend/internal/agent/runtime/dynamic/circuit_test.go diff --git a/apps/backend/internal/agent/runtime/dynamic/circuit.go b/apps/backend/internal/agent/runtime/dynamic/circuit.go index b3b6b84748..ff009794da 100644 --- a/apps/backend/internal/agent/runtime/dynamic/circuit.go +++ b/apps/backend/internal/agent/runtime/dynamic/circuit.go @@ -5,6 +5,8 @@ import ( "sync" "time" + "go.uber.org/zap" + "github.com/kandev/kandev/internal/agent/runtime/routingerr" ) @@ -66,15 +68,32 @@ func WithCircuitPersistence(persistence CircuitPersistence) CircuitRegistryOptio return func(registry *CircuitRegistry) { registry.persist = persistence } } +// WithCircuitLogger sets the logger used to report a failed durable write. +// Defaults to a no-op logger. +func WithCircuitLogger(log *zap.Logger) CircuitRegistryOption { + return func(registry *CircuitRegistry) { + if log != nil { + registry.logger = log + } + } +} + type CircuitRegistry struct { mu sync.Mutex now func() time.Time circuits map[string]circuit persist CircuitPersistence + logger *zap.Logger + pending map[string]struct{} } func NewCircuitRegistry(options ...CircuitRegistryOption) *CircuitRegistry { - registry := &CircuitRegistry{now: time.Now, circuits: make(map[string]circuit)} + registry := &CircuitRegistry{ + now: time.Now, + circuits: make(map[string]circuit), + logger: zap.NewNop(), + pending: make(map[string]struct{}), + } for _, option := range options { option(registry) } @@ -107,8 +126,9 @@ func (r *CircuitRegistry) Open(key string, until time.Time, code routingerr.Code } r.mu.Lock() defer r.mu.Unlock() + r.flushPendingLocked(key) r.circuits[key] = circuit{state: CircuitOpen, until: until, code: code} - r.persistSnapshotLocked(key) + _ = r.persistSnapshotLocked(key) } func (r *CircuitRegistry) IsOpen(key string, now time.Time) bool { @@ -143,11 +163,12 @@ func (r *CircuitRegistry) AcquireProbe(key string, duration time.Duration) (Prob if !ok || entry.state == CircuitClosed || now.Before(entry.until) || now.Before(entry.probeUntil) { return ProbeLease{}, false } + r.flushPendingLocked(key) lease := ProbeLease{Key: key, ExpiresAt: now.Add(duration)} entry.state = CircuitHalfOpen entry.probeUntil = lease.ExpiresAt r.circuits[key] = entry - r.persistSnapshotLocked(key) + _ = r.persistSnapshotLocked(key) return lease, true } @@ -161,6 +182,7 @@ func (r *CircuitRegistry) ReleaseProbe(lease ProbeLease, success bool, backoff t if !ok || entry.state != CircuitHalfOpen || !entry.probeUntil.Equal(lease.ExpiresAt) { return } + r.flushPendingLocked(lease.Key) entry.probeUntil = time.Time{} if success { entry.state = CircuitClosed @@ -171,22 +193,69 @@ func (r *CircuitRegistry) ReleaseProbe(lease ProbeLease, success bool, backoff t entry.until = r.now().Add(backoff) } r.circuits[lease.Key] = entry - r.persistSnapshotLocked(lease.Key) + _ = r.persistSnapshotLocked(lease.Key) } -func (r *CircuitRegistry) persistSnapshotLocked(key string) { +// pendingFlushBudget bounds how many other pending keys a single mutation +// retries. r.mu also gates IsOpen on the routing hot path, and each retry is +// a blocking SaveCircuit call, so a mutation during a sustained persistence +// outage must hold the lock for at most one extra write, not one per pending +// key: unbounded fan-out both grows total SaveCircuit calls quadratically +// across mutations and holds the routing mutex for the sum of every pending +// write's duration. +const pendingFlushBudget = 1 + +// flushPendingLocked retries the durable write for up to pendingFlushBudget +// keys whose previous SaveCircuit failed, using each key's current in-memory +// snapshot. Called from every mutator so a transient persistence failure +// self-heals across subsequent circuit events instead of leaving the durable +// row permanently behind memory. +func (r *CircuitRegistry) flushPendingLocked(skipKey string) { + flushed := 0 + for key := range r.pending { + if key == skipKey { + continue + } + if flushed >= pendingFlushBudget { + return + } + _ = r.persistSnapshotLocked(key) + flushed++ + } +} + +// persistSnapshotLocked writes key's current snapshot and returns the +// persistence error, if any. A failure is logged at WARN once per failure +// streak (not once per retry) and the key is recorded in pending so a later +// mutation retries the write; a caller-visible error here would either fail +// routing decisions that must still complete (Open) or make every candidate +// unselectable while the store is down (AcquireProbe), so callers do not act +// on the return value directly. +func (r *CircuitRegistry) persistSnapshotLocked(key string) error { if r.persist == nil { - return + return nil } entry, ok := r.circuits[key] if !ok { - return + return nil } - // Selection remains fail-closed in memory if persistence is temporarily - // unavailable. Startup Restore and health reconciliation surface durable - // read failures to their caller. - _ = r.persist.SaveCircuit(context.Background(), CircuitSnapshot{ + err := r.persist.SaveCircuit(context.Background(), CircuitSnapshot{ Key: key, State: entry.state, Until: entry.until, Code: entry.code, ProbeUntil: entry.probeUntil, }) + if err != nil { + _, alreadyPending := r.pending[key] + r.pending[key] = struct{}{} + if !alreadyPending { + r.logger.Warn("failed to persist circuit snapshot", + zap.String("key", key), + zap.String("state", string(entry.state)), + zap.Time("until", entry.until), + zap.Error(err), + ) + } + return err + } + delete(r.pending, key) + return nil } diff --git a/apps/backend/internal/agent/runtime/dynamic/circuit_test.go b/apps/backend/internal/agent/runtime/dynamic/circuit_test.go new file mode 100644 index 0000000000..8d19dc40fd --- /dev/null +++ b/apps/backend/internal/agent/runtime/dynamic/circuit_test.go @@ -0,0 +1,401 @@ +package dynamic + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + + "github.com/kandev/kandev/internal/agent/runtime/routingerr" +) + +// pendingKeys reads r.pending directly; the test lives in package dynamic so +// it needs no exported accessor for a field production code never consumes. +func pendingKeys(r *CircuitRegistry) []string { + r.mu.Lock() + defer r.mu.Unlock() + keys := make([]string, 0, len(r.pending)) + for key := range r.pending { + keys = append(keys, key) + } + return keys +} + +// stubCircuitPersistence is an in-memory CircuitPersistence whose SaveCircuit +// can be made to fail on demand, to exercise the persistence-failure paths. +type stubCircuitPersistence struct { + mu sync.Mutex + rows map[string]CircuitSnapshot + failing bool + saves int +} + +func newStubCircuitPersistence() *stubCircuitPersistence { + return &stubCircuitPersistence{rows: make(map[string]CircuitSnapshot)} +} + +func (s *stubCircuitPersistence) SaveCircuit(_ context.Context, snapshot CircuitSnapshot) error { + s.mu.Lock() + defer s.mu.Unlock() + s.saves++ + if s.failing { + return errors.New("db is locked") + } + s.rows[snapshot.Key] = snapshot + return nil +} + +func (s *stubCircuitPersistence) LoadCircuits(_ context.Context) ([]CircuitSnapshot, error) { + s.mu.Lock() + defer s.mu.Unlock() + snapshots := make([]CircuitSnapshot, 0, len(s.rows)) + for _, row := range s.rows { + if row.State == CircuitClosed { + continue + } + snapshots = append(snapshots, row) + } + return snapshots, nil +} + +func (s *stubCircuitPersistence) setFailing(failing bool) { + s.mu.Lock() + defer s.mu.Unlock() + s.failing = failing +} + +func (s *stubCircuitPersistence) row(key string) (CircuitSnapshot, bool) { + s.mu.Lock() + defer s.mu.Unlock() + row, ok := s.rows[key] + return row, ok +} + +func (s *stubCircuitPersistence) saveCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.saves +} + +func newObservedLogger() (*zap.Logger, *observer.ObservedLogs) { + core, observed := observer.New(zapcore.WarnLevel) + return zap.New(core), observed +} + +func TestPersistSnapshotFailureIsSurfacedOnOpen(t *testing.T) { + persist := newStubCircuitPersistence() + persist.setFailing(true) + log, observed := newObservedLogger() + registry := NewCircuitRegistry(WithCircuitPersistence(persist), WithCircuitLogger(log)) + + registry.Open("credential:acme", time.Now().Add(time.Hour), routingerr.CodeProviderUnavailable) + + pending := pendingKeys(registry) + if len(pending) != 1 || pending[0] != "credential:acme" { + t.Fatalf("pending keys = %v, want [credential:acme]", pending) + } + entries := observed.FilterMessageSnippet("persist circuit").All() + if len(entries) != 1 { + t.Fatalf("warn log entries = %d, want 1", len(entries)) + } +} + +func TestPersistSnapshotFailureIsSurfacedOnAcquireProbe(t *testing.T) { + persist := newStubCircuitPersistence() + log, observed := newObservedLogger() + registry := NewCircuitRegistry( + WithCircuitPersistence(persist), + WithCircuitLogger(log), + WithCircuitClock(func() time.Time { return time.Unix(1000, 0) }), + ) + registry.Open("credential:acme", time.Unix(900, 0), routingerr.CodeProviderUnavailable) + observed.TakeAll() + + persist.setFailing(true) + lease, ok := registry.AcquireProbe("credential:acme", time.Minute) + if !ok { + t.Fatalf("AcquireProbe: expected success, lease=%#v", lease) + } + + pending := pendingKeys(registry) + if len(pending) != 1 || pending[0] != "credential:acme" { + t.Fatalf("pending keys = %v, want [credential:acme]", pending) + } + entries := observed.FilterMessageSnippet("persist circuit").All() + if len(entries) != 1 { + t.Fatalf("warn log entries = %d, want 1", len(entries)) + } +} + +func TestPersistSnapshotFailureIsSurfacedOnReleaseProbe(t *testing.T) { + persist := newStubCircuitPersistence() + log, observed := newObservedLogger() + registry := NewCircuitRegistry( + WithCircuitPersistence(persist), + WithCircuitLogger(log), + WithCircuitClock(func() time.Time { return time.Unix(1000, 0) }), + ) + registry.Open("credential:acme", time.Unix(900, 0), routingerr.CodeProviderUnavailable) + lease, ok := registry.AcquireProbe("credential:acme", time.Minute) + if !ok { + t.Fatalf("AcquireProbe: expected success") + } + observed.TakeAll() + + persist.setFailing(true) + registry.ReleaseProbe(lease, true, time.Minute) + + pending := pendingKeys(registry) + if len(pending) != 1 || pending[0] != "credential:acme" { + t.Fatalf("pending keys = %v, want [credential:acme]", pending) + } + entries := observed.FilterMessageSnippet("persist circuit").All() + if len(entries) != 1 { + t.Fatalf("warn log entries = %d, want 1", len(entries)) + } +} + +func TestMutationDoesNotFlushItsOwnPendingKey(t *testing.T) { + persist := newStubCircuitPersistence() + persist.setFailing(true) + registry := NewCircuitRegistry(WithCircuitPersistence(persist)) + + registry.Open("credential:acme", time.Now().Add(time.Hour), routingerr.CodeProviderUnavailable) + persist.setFailing(false) + registry.Open("credential:acme", time.Now().Add(2*time.Hour), routingerr.CodeProviderUnavailable) + + if got := persist.saveCount(); got != 2 { + t.Fatalf("SaveCircuit calls = %d, want 2 (one failed write and one current mutation write)", got) + } + if pending := pendingKeys(registry); len(pending) != 0 { + t.Fatalf("pending keys = %v, want empty", pending) + } +} + +// TestFailedPersistDoesNotSurviveRestart is the regression test: an Open +// whose SaveCircuit fails must not leave the durable row permanently absent. +// A later mutation that succeeds should flush the pending write so a fresh +// registry's Restore sees the binding as open, not closed. +func TestFailedPersistDoesNotSurviveRestart(t *testing.T) { + persist := newStubCircuitPersistence() + persist.setFailing(true) + registry := NewCircuitRegistry(WithCircuitPersistence(persist)) + + until := time.Now().Add(time.Hour) + registry.Open("credential:acme", until, routingerr.CodeProviderUnavailable) + + if _, ok := persist.row("credential:acme"); ok { + t.Fatalf("durable row should be absent while persistence is failing") + } + + // Restart now: a fresh registry restores from the (still-empty) store and + // reads the binding as closed. This is the bug before the fix. + restored := NewCircuitRegistry(WithCircuitPersistence(persist)) + if err := restored.Restore(context.Background()); err != nil { + t.Fatalf("Restore: %v", err) + } + if restored.IsOpen("credential:acme", time.Now()) { + t.Fatalf("restored registry should not see the lost write yet") + } + + // The store recovers; a further mutation on the original registry should + // flush the pending write. + persist.setFailing(false) + registry.Open("credential:other", time.Now().Add(time.Hour), routingerr.CodeProviderUnavailable) + + if len(pendingKeys(registry)) != 0 { + t.Fatalf("pending keys should be empty after a successful flush, got %v", pendingKeys(registry)) + } + if _, ok := persist.row("credential:acme"); !ok { + t.Fatalf("durable row for credential:acme should have been backfilled") + } + + restoredAfterRecovery := NewCircuitRegistry(WithCircuitPersistence(persist)) + if err := restoredAfterRecovery.Restore(context.Background()); err != nil { + t.Fatalf("Restore: %v", err) + } + if !restoredAfterRecovery.IsOpen("credential:acme", time.Now()) { + t.Fatalf("restored registry should see credential:acme as open after recovery") + } +} + +// TestFlushWritesCurrentSnapshotNotStale ensures the pending-flush re-reads +// the circuit's current in-memory state rather than replaying the stale +// snapshot captured at the time of the original failed write. It observes the +// flush write in isolation by triggering it through a mutation of a +// DIFFERENT key, so credential:acme's own write never runs in that call and +// the persisted row can only have come from the flush. +func TestFlushWritesCurrentSnapshotNotStale(t *testing.T) { + persist := newStubCircuitPersistence() + persist.setFailing(true) + registry := NewCircuitRegistry( + WithCircuitPersistence(persist), + WithCircuitClock(func() time.Time { return time.Unix(1000, 0) }), + ) + + // credential:acme fails to persist while opening, then advances to + // half-open via AcquireProbe -- still while persistence is failing, so + // the durable store never observes either state directly. + registry.Open("credential:acme", time.Unix(900, 0), routingerr.CodeProviderUnavailable) + lease, ok := registry.AcquireProbe("credential:acme", time.Minute) + if !ok { + t.Fatalf("AcquireProbe: expected success") + } + + persist.setFailing(false) + registry.Open("credential:other", time.Now().Add(time.Hour), routingerr.CodeProviderUnavailable) + + row, ok := persist.row("credential:acme") + if !ok { + t.Fatalf("durable row for credential:acme should have been backfilled by the flush") + } + if row.State != CircuitHalfOpen { + t.Fatalf("durable row state = %v, want %v (current snapshot, not the stale open snapshot from the original failure)", row.State, CircuitHalfOpen) + } + if !row.ProbeUntil.Equal(lease.ExpiresAt) { + t.Fatalf("durable row probeUntil = %v, want %v (current snapshot, not stale)", row.ProbeUntil, lease.ExpiresAt) + } + if pending := pendingKeys(registry); len(pending) != 0 { + t.Fatalf("pending keys should be empty after the flush, got %v", pending) + } + + // The original mutation path still ends up durable too, via its own write. + registry.ReleaseProbe(lease, true, time.Minute) + row, ok = persist.row("credential:acme") + if !ok || row.State != CircuitClosed { + t.Fatalf("durable row after ReleaseProbe = %+v, ok=%v, want closed", row, ok) + } +} + +// TestFlushDuringOutageIsLinearNotQuadratic pins the Finding-2 fix: opening N +// circuits during a sustained persistence outage must cost O(N) SaveCircuit +// calls (bounded fan-out per mutation), not O(N^2) from retrying every +// pending key on every mutation. +func TestFlushDuringOutageIsLinearNotQuadratic(t *testing.T) { + persist := newStubCircuitPersistence() + persist.setFailing(true) + registry := NewCircuitRegistry(WithCircuitPersistence(persist)) + + const n = 50 + for i := 0; i < n; i++ { + registry.Open(fmt.Sprintf("credential:c%d", i), time.Now().Add(time.Hour), routingerr.CodeProviderUnavailable) + } + + // Quadratic behavior (retry every pending key on every mutation) would + // cost n + n(n-1)/2 = 1,275 calls for n=50. Bounded fan-out costs at most + // n own writes plus pendingFlushBudget flush attempts per mutation. + if want := n * (pendingFlushBudget + 1); persist.saves > want { + t.Fatalf("SaveCircuit calls = %d, want <= %d (linear in N, bounded flush fan-out)", persist.saves, want) + } + // Lower bound: this must also prove flushing actually happens, not just + // that it stays bounded. n own writes alone (flush disabled) would leave + // saves == n; each mutation after the first also retries a pending key. + if persist.saves <= n { + t.Fatalf("SaveCircuit calls = %d, want > %d (flush must retry pending keys, not just perform each mutation's own write)", persist.saves, n) + } + if len(pendingKeys(registry)) != n { + t.Fatalf("pending keys = %d, want %d (persistence is still down)", len(pendingKeys(registry)), n) + } +} + +// gatedFailingCircuitPersistence blocks writes after setup and releases each +// write through a channel, so the lock-hold bound is tested without sleeps. +type gatedFailingCircuitPersistence struct { + blocking atomic.Bool + writeStarted chan string + releaseWrite chan struct{} +} + +func newGatedFailingCircuitPersistence() *gatedFailingCircuitPersistence { + return &gatedFailingCircuitPersistence{ + writeStarted: make(chan string), + releaseWrite: make(chan struct{}), + } +} + +func (s *gatedFailingCircuitPersistence) SaveCircuit(_ context.Context, snapshot CircuitSnapshot) error { + if s.blocking.Load() { + s.writeStarted <- snapshot.Key + <-s.releaseWrite + } + return errors.New("db is locked") +} + +func (s *gatedFailingCircuitPersistence) LoadCircuits(_ context.Context) ([]CircuitSnapshot, error) { + return nil, nil +} + +// TestFlushDuringOutageDoesNotStallRoutingHotPath pins Review Round 2's +// blocker: a mutation's pending-key flush must not hold r.mu long enough to +// stall a concurrent IsOpen call (the routing hot path) for the sum of every +// pending write's duration. With pendingFlushBudget bounded to 1, one +// mutation holds the lock for at most its own write plus a single flush +// attempt, regardless of how many keys are pending. +func TestFlushDuringOutageDoesNotStallRoutingHotPath(t *testing.T) { + persist := newGatedFailingCircuitPersistence() + registry := NewCircuitRegistry(WithCircuitPersistence(persist)) + + const n = 8 + for i := 0; i < n; i++ { + registry.Open(fmt.Sprintf("credential:c%d", i), time.Now().Add(time.Hour), routingerr.CodeProviderUnavailable) + } + if got := len(pendingKeys(registry)); got != n { + t.Fatalf("pending keys = %d, want %d", got, n) + } + persist.blocking.Store(true) + + mutationDone := make(chan struct{}) + go func() { + defer close(mutationDone) + registry.Open("credential:new", time.Now().Add(time.Hour), routingerr.CodeProviderUnavailable) + }() + <-persist.writeStarted + + isOpenDone := make(chan struct{}) + go func() { + registry.IsOpen("credential:c0", time.Now()) + close(isOpenDone) + }() + persist.releaseWrite <- struct{}{} + + writes := 1 + for { + select { + case <-persist.writeStarted: + writes++ + persist.releaseWrite <- struct{}{} + case <-mutationDone: + if writes != 2 { + t.Fatalf("blocked mutation used %d writes, want 2", writes) + } + select { + case <-isOpenDone: + case <-time.After(100 * time.Millisecond): + t.Fatal("IsOpen did not complete after the bounded mutation finished") + } + return + } + } +} + +func TestCleanPathLeavesNoPendingAndLogsNothing(t *testing.T) { + persist := newStubCircuitPersistence() + log, observed := newObservedLogger() + registry := NewCircuitRegistry(WithCircuitPersistence(persist), WithCircuitLogger(log)) + + registry.Open("credential:acme", time.Now().Add(time.Hour), routingerr.CodeProviderUnavailable) + + if pending := pendingKeys(registry); len(pending) != 0 { + t.Fatalf("pending keys = %v, want empty", pending) + } + if observed.Len() != 0 { + t.Fatalf("expected no log entries on the clean path, got %d", observed.Len()) + } +} diff --git a/apps/backend/internal/backendapp/services.go b/apps/backend/internal/backendapp/services.go index 579760ea77..fc9fdba485 100644 --- a/apps/backend/internal/backendapp/services.go +++ b/apps/backend/internal/backendapp/services.go @@ -96,6 +96,7 @@ func provideServices(cfg *config.Config, log *logger.Logger, repos *Repositories })) dynamicCircuits := dynamicruntime.NewCircuitRegistry( dynamicruntime.WithCircuitPersistence(repos.Task), + dynamicruntime.WithCircuitLogger(log.Zap()), ) if err := dynamicCircuits.Restore(context.Background()); err != nil { return nil, nil, fmt.Errorf("restore dynamic routing health: %w", err)