diff --git a/common/selection_probe.go b/common/selection_probe.go new file mode 100644 index 000000000..8ecc671e3 --- /dev/null +++ b/common/selection_probe.go @@ -0,0 +1,34 @@ +package common + +import "context" + +// selectionProbeKey marks a context as belonging to a selection-policy recovery +// probe (the shadow request the prober mirrors against a currently-excluded +// upstream). +// +// A recovery probe MUST reach the upstream even when that upstream's circuit +// breaker is open: its whole job is to gather a fresh, real health signal for +// the selection policy. If the breaker denied the probe, the prober would +// record the breaker-open error as a probe FAILURE, keeping the upstream's +// error rate high and excluding it forever — the "probe/selection wedge". So a +// probe is breaker-INELIGIBLE: it neither acquires a permit nor records an +// outcome into the breaker. The breaker still recovers on its own via its +// HalfOpen trials on real traffic; the probe's signal flows through the +// selection tracker instead. +const selectionProbeKey ContextKey = "selection_probe" + +// WithSelectionProbe marks ctx as a selection-policy recovery probe so the +// upstream executor treats it as breaker-ineligible. It does not mutate the +// request object, which the prober shares with live client traffic. +func WithSelectionProbe(ctx context.Context) context.Context { + return context.WithValue(ctx, selectionProbeKey, true) +} + +// IsSelectionProbe reports whether ctx was marked by WithSelectionProbe. +func IsSelectionProbe(ctx context.Context) bool { + if ctx == nil { + return false + } + v, _ := ctx.Value(selectionProbeKey).(bool) + return v +} diff --git a/internal/policy/prober.go b/internal/policy/prober.go index 5c8a4b0ed..464ef6f3f 100644 --- a/internal/policy/prober.go +++ b/internal/policy/prober.go @@ -390,6 +390,12 @@ func (p *Prober) mirror(req *common.NormalizedRequest, u common.Upstream, cfg *P } ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() + // Mark this as a selection-recovery probe so the upstream executor treats + // it as breaker-ineligible: the probe must reach an excluded upstream even + // while its breaker is open, otherwise the breaker-open denial is recorded + // as a probe failure and the upstream stays excluded forever. We tag the + // ctx (not req) because req is shared with live client traffic. + ctx = common.WithSelectionProbe(ctx) method, _ := req.Method() finality := req.Finality(ctx) diff --git a/upstream/breaker_eligibility_test.go b/upstream/breaker_eligibility_test.go new file mode 100644 index 000000000..0dd2f6b04 --- /dev/null +++ b/upstream/breaker_eligibility_test.go @@ -0,0 +1,38 @@ +package upstream + +import ( + "context" + "testing" + + "github.com/erpc/erpc/common" + "github.com/stretchr/testify/require" +) + +// TestUpstreamBreakerEligible_SelectionProbe is the guard for the probe/selection +// wedge follow-up: a selection-recovery probe must be breaker-INELIGIBLE so it +// reaches an excluded upstream even while the breaker is open (gathering a real +// health signal) instead of being denied a permit — which the prober would +// record as a probe failure, excluding the upstream forever. +func TestUpstreamBreakerEligible_SelectionProbe(t *testing.T) { + t.Run("a selection probe is breaker-ineligible", func(t *testing.T) { + ctx := common.WithSelectionProbe(context.Background()) + require.False(t, upstreamBreakerEligible(ctx, nil, false), + "a selection-recovery probe must not acquire/record a breaker permit") + }) + + t.Run("ordinary traffic stays breaker-eligible", func(t *testing.T) { + require.True(t, upstreamBreakerEligible(context.Background(), nil, false), + "non-probe, non-hedge requests must remain breaker-eligible") + }) + + t.Run("hedge attempts remain ineligible regardless of probe flag", func(t *testing.T) { + require.False(t, upstreamBreakerEligible(context.Background(), nil, true)) + }) +} + +// TestSelectionProbeContextRoundtrip verifies the marker helpers. +func TestSelectionProbeContextRoundtrip(t *testing.T) { + require.False(t, common.IsSelectionProbe(context.Background())) + require.True(t, common.IsSelectionProbe(common.WithSelectionProbe(context.Background()))) + require.False(t, common.IsSelectionProbe(nil)) +} diff --git a/upstream/upstream_executor.go b/upstream/upstream_executor.go index c9e283927..42b742a37 100644 --- a/upstream/upstream_executor.go +++ b/upstream/upstream_executor.go @@ -348,9 +348,11 @@ func (e *upstreamExecutor) callBreakerWithTimeout( inner func(ctx context.Context, isHedge bool) (*common.NormalizedResponse, error), isHedge bool, ) (*common.NormalizedResponse, error) { - // Breaker eligibility check — internal probes and hedge attempts do NOT - // count toward the breaker. - if e.breaker != nil && upstreamBreakerEligible(req, isHedge) { + // Breaker eligibility check — internal probes, selection-recovery probes, + // and hedge attempts do NOT count toward the breaker. Decide once so the + // permit acquisition and the outcome recording stay consistent. + breakerEligible := e.breaker != nil && upstreamBreakerEligible(ctx, req, isHedge) + if breakerEligible { if !e.breaker.TryAcquirePermit() { startTime := time.Now() return nil, common.NewErrFailsafeCircuitBreakerOpen(common.ScopeUpstream, failsafe.ErrCircuitOpen, &startTime) @@ -359,19 +361,25 @@ func (e *upstreamExecutor) callBreakerWithTimeout( resp, err := e.callWithTimeout(ctx, req, inner, isHedge) - if e.breaker != nil && upstreamBreakerEligible(req, isHedge) { + if breakerEligible { e.breaker.Record(upstreamBreakerOutcome(resp, err)) } return resp, err } -// upstreamBreakerEligible decides whether (req, isHedge) should contribute -// to the breaker counters. Hedge attempts and internal probes are excluded. -// Composite requests are also excluded. -func upstreamBreakerEligible(req *common.NormalizedRequest, isHedge bool) bool { +// upstreamBreakerEligible decides whether (ctx, req, isHedge) should contribute +// to the breaker counters. Hedge attempts and internal probes are excluded, as +// are selection-recovery probes — a probe must reach the upstream even with the +// breaker open so it can gather a real health signal for the selection policy; +// being denied a permit would otherwise record as a probe failure and wedge the +// upstream excluded. Composite requests are also excluded. +func upstreamBreakerEligible(ctx context.Context, req *common.NormalizedRequest, isHedge bool) bool { if isHedge { return false } + if common.IsSelectionProbe(ctx) { + return false + } if req == nil { return true }