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 <noreply@anthropic.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

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
 }

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())
+	}
+}

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)