diff --git a/hoglet.go b/hoglet.go index 1efeeb0..999e993 100644 --- a/hoglet.go +++ b/hoglet.go @@ -49,6 +49,9 @@ type Breaker interface { type ObserverFactory interface { // ObserverForCall returns an [Observer] for the incoming call. // It is called with the current [State] of the circuit, before calling the wrapped function. + // + // An error rejects the call. No [Observer] is returned in that case, so anything the factory claimed for the + // call must be released before returning: nothing else will. ObserverForCall(context.Context, State) (Observer, error) } diff --git a/limiter.go b/limiter.go index 91b05d2..78379ad 100644 --- a/limiter.go +++ b/limiter.go @@ -29,6 +29,9 @@ func ConcurrencyLimiter(limit int64, block bool) BreakerMiddleware { }) } +// concurrencyLimiter holds the shared state of both limiter variants. Both acquire a permit before delegating to its +// [concurrencyLimiter.ObserverForCall], which is therefore responsible for releasing it again — including when the +// inner factory rejects the call. type concurrencyLimiter struct { sem *semaphore.Weighted next ObserverFactory @@ -37,6 +40,11 @@ type concurrencyLimiter struct { func (cl concurrencyLimiter) ObserverForCall(ctx context.Context, state State) (Observer, error) { o, err := cl.next.ObserverForCall(ctx, state) if err != nil { + // No [Observer] is returned on error, so release here or never. + // Leaking a permit here is terminal: the limiter sits outside the circuit's state + // check, so every call dropped while the circuit is open would permanently shrink the + // effective limit until no call can reach the circuit to ever close it again. + cl.sem.Release(1) return nil, err } return ObserverFunc(func(b bool) { diff --git a/limiter_test.go b/limiter_test.go index 586b465..eaa34e9 100644 --- a/limiter_test.go +++ b/limiter_test.go @@ -4,6 +4,7 @@ import ( "context" "sync" "testing" + "testing/synctest" "time" "github.com/exaring/hoglet" @@ -22,7 +23,12 @@ func (mo mockPanickingObservable) Observe(shouldPanic bool) { type mockObserverFactory struct{} -func (mof mockObserverFactory) ObserverForCall(ctx context.Context, state hoglet.State) (hoglet.Observer, error) { +func (mof mockObserverFactory) ObserverForCall(_ context.Context, state hoglet.State) (hoglet.Observer, error) { + // abuse the state argument to control the result of the call, standing in for a [hoglet.Circuit] rejecting calls + // while open + if state == hoglet.StateOpen { + return nil, hoglet.ErrCircuitOpen + } return &mockPanickingObservable{}, nil } @@ -128,6 +134,56 @@ func Test_ConcurrencyLimiter(t *testing.T) { } } +// Test_ConcurrencyLimiter_ReleasesOnInnerError ensures the limiter releases its permit when the inner factory rejects +// the call. A leak there is terminal: the limiter sits outside the circuit's state check, so calls dropped while the +// circuit is open would drain the permits until nothing can reach the circuit to ever close it again. +func Test_ConcurrencyLimiter_ReleasesOnInnerError(t *testing.T) { + tests := []struct { + name string + block bool + }{ + {name: "non-blocking", block: false}, + {name: "blocking", block: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + const limit = 2 + + of, err := hoglet.ConcurrencyLimiter(limit, tt.block).Wrap(mockObserverFactory{}) + require.NoError(t, err) + + // Every acquisition is bounded, so the blocking variant fails instead of hanging on a leaked permit. + // Costs no wall-clock time under synctest. + call := func(state hoglet.State) (hoglet.Observer, error) { + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + + return of.ObserverForCall(ctx, state) + } + + // One more dropped call than there are permits: had they leaked, the last one would be rejected by the + // limiter instead of the circuit. + for i := range limit + 1 { + o, err := call(hoglet.StateOpen) + require.ErrorIs(t, err, hoglet.ErrCircuitOpen, "call %d", i) + assert.Nil(t, o) // nothing is handed back that could release the permit for us + } + + // The circuit closes again: every permit must still be available. + for i := range limit { + _, err := call(hoglet.StateClosed) + require.NoError(t, err, "call %d after recovery: permit leaked while circuit was open", i) + } + + // The limit is still enforced, i.e. we did not release more than we held. + _, err = call(hoglet.StateClosed) + assert.Error(t, err, "limit should be reached with all %d permits held", limit) + }) + }) + } +} + func ptr[T any](in T) *T { return &in }