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 and core/bus: admission is a bus lane (the Serve lane of the process's bus, or one of the engine's own), IO slots a bus semaphore, singleflight 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
- func Run[T any](ctx context.Context, e *Engine, fn func(context.Context) (T, error)) (T, error)
- func Submit[T any](ctx context.Context, e *Engine, key string, ...) (T, error)
- type Engine
- type Option
- func WithBreaker(b *ratelimit.Breaker) Option
- func WithFailureClassifier(classify func(error) bool) Option
- func WithIOPool(n int) Option
- func WithLane(l *bus.Lane) Option
- func WithLimiter(l *ratelimit.Adaptive) Option
- func WithMaxInflight(n int) Option
- func WithQuota(q *Quota) Option
- func WithRPSLimiter(l *ratelimit.TokenBucket) Option
- type Quota
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 ¶
func Run ¶ added in v0.1.13
Run runs fn on e under the rate limit, its panics recovered, but outside the lane, the breaker, the adaptive limit and the IO slots: for a call ending a transaction, which must reach its database however it is held back, and holds its connection already.
func Submit ¶ added in v0.1.13
func Submit[T any](ctx context.Context, e *Engine, key string, fn func(context.Context) (T, error)) (T, error)
Submit schedules fn on e 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.
Types ¶
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine bounds concurrency and deduplicates concurrent identical requests.
type Option ¶
type Option func(*config)
Option configures an Engine.
func WithBreaker ¶
WithBreaker attaches a circuit breaker.
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 WithLane ¶ added in v0.1.13
WithLane admits submissions on lane, such as the Serve lane of the process's bus, instead of an in-flight bound of the engine's own; it replaces the max in-flight option.
func WithLimiter ¶
WithLimiter attaches an adaptive limiter for AIMD feedback.
func WithMaxInflight ¶
WithMaxInflight sets the bounded in-flight queue (backpressure).
func WithQuota ¶ added in v0.1.12
WithQuota bounds IO by a quota shared with other engines; it replaces the IO pool, limiter, rate limiter and breaker options.
func WithRPSLimiter ¶
func WithRPSLimiter(l *ratelimit.TokenBucket) Option
WithRPSLimiter attaches a token-bucket request-rate limiter.
type Quota ¶ added in v0.1.12
type Quota struct {
// contains filtered or unexported fields
}
Quota is what bounds IO across engines: the IO slots, the adaptive limiter, the request-rate limiter and the circuit breaker. Engines of successive configurations of one process share it, so a reload never lets the old and the new configuration together exceed the IO pool, and Update applies new limits to the work already running.
func NewQuota ¶ added in v0.1.12
NewQuota returns a quota of io slots and no limiter, rate limit or breaker.
func NewQuotaOn ¶ added in v0.1.13
NewQuotaOn returns a quota whose IO slots are io, such as the IO semaphore of the process's bus, and no limiter, rate limit or breaker.
func (*Quota) Update ¶ added in v0.1.12
func (q *Quota) Update(io int, limiter *ratelimit.Adaptive, rps *ratelimit.TokenBucket, breaker *ratelimit.Breaker)
Update sets the IO slots and replaces the limiter, rate limiter and breaker (nil turns one off). Work holding a slot keeps it; when the slots shrink, new work waits until the running work fits.
Source Files