Documentation
¶
Overview ¶
Package engine enforces bounded concurrency with backpressure, request deduplication (singleflight), and integration with adaptive limiting and the circuit breaker. It uses only the standard library: semaphores are buffered channels, singleflight is a mutex-guarded call table. No unbounded goroutines (invariant I13); panics in submitted functions are recovered and returned as errors so they never crash the process.
Index ¶
- Variables
- type Engine
- func (e *Engine) Close()
- func (e *Engine) Drain(ctx context.Context) error
- func (e *Engine) Submit(ctx context.Context, key string, fn func(context.Context) (any, error)) (any, error)
- func (e *Engine) SubmitCPU(ctx context.Context, key string, fn func(context.Context) (any, error)) (any, error)
- func (e *Engine) SubmitClass(ctx context.Context, class WorkClass, key string, ...) (any, error)
- type Option
- func WithBreaker(b *ratelimit.Breaker) Option
- func WithCPUPool(n int) Option
- func WithFailureClassifier(classify func(error) bool) Option
- func WithIOPool(n int) Option
- func WithLimiter(l *ratelimit.Adaptive) Option
- func WithMaxInflight(n int) Option
- func WithRPSLimiter(l *ratelimit.TokenBucket) Option
- type WorkClass
Constants ¶
This section is empty.
Variables ¶
var ( // ErrOverloaded is returned when the in-flight queue is full (backpressure). ErrOverloaded = errors.New("engine: overloaded") // ErrInvalidConfig is returned by New for non-positive pool sizes. ErrInvalidConfig = errors.New("engine: invalid config") // ErrClosed is returned by Submit after Close. ErrClosed = errors.New("engine: closed") )
Sentinel errors.
Functions ¶
This section is empty.
Types ¶
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine bounds concurrency and deduplicates concurrent identical requests.
func (*Engine) Close ¶
func (e *Engine) Close()
Close rejects new work and waits for all in-flight submissions.
func (*Engine) Drain ¶
Drain rejects new submissions and waits for admitted callers and detached singleflight execution goroutines to return.
func (*Engine) Submit ¶
func (e *Engine) Submit(ctx context.Context, key string, fn func(context.Context) (any, error)) (any, error)
Submit schedules fn under backpressure and returns fn's result value and error. It returns ErrOverloaded if the in-flight queue is full, ErrCircuitOpen if the breaker is open, ErrRateLimited if concurrency exceeds the adaptive limit. Concurrent calls with the same non-empty key share a single execution (singleflight) and all receive the leader's result. An empty key opts out of de-duplication (writes/unique ops). Panics in fn are recovered as errors.
type Option ¶
type Option func(*config)
Option configures an Engine.
func WithBreaker ¶
WithBreaker attaches a circuit breaker.
func WithCPUPool ¶
WithCPUPool sets the CPU concurrency (<= GOMAXPROCS).
func WithFailureClassifier ¶
WithFailureClassifier decides whether a non-nil execution error represents provider/system health and should affect AIMD and the circuit breaker.
func WithIOPool ¶
WithIOPool sets the IO concurrency (<= DB connection pool size).
func WithLimiter ¶
WithLimiter attaches an adaptive limiter for AIMD feedback.
func WithMaxInflight ¶
WithMaxInflight sets the bounded in-flight queue (backpressure).
func WithRPSLimiter ¶
func WithRPSLimiter(l *ratelimit.TokenBucket) Option
WithRPSLimiter attaches a token-bucket request-rate limiter.
Source Files