adaptx

package
v1.5.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 17, 2026 License: MIT Imports: 11 Imported by: 0

README

adaptx — Adaptive Concurrency Limiting (AIMD · Vegas · Gradient)

CI Go Reference License: MIT Changelog

A thread-safe concurrency limiter that discovers a backend's safe operating limit instead of making you guess it. It starts at a configured limit and moves that limit once per sample window (default 1s) from latency and error feedback using one of three control laws — AIMD, Vegas, or Gradient — opening up when the backend is fast and healthy and clamping down the moment a window shows latency climb or errors. A controller for load-aware callbacks, panic recovery, and lock-light admission. Go 1.24+. Zero external dependencies (depends only on the urx panix package; testify in tests only).

go get github.com/aasyanov/urx

[!IMPORTANT] adaptx limits concurrency, not request rate, and it adapts the limit itself — windowed, not per request. Where bulkx enforces a fixed, hand-sized bound, adaptx treats the limit as a variable it servo-controls toward the backend's real capacity. Pick the algorithm that matches your overload signal: AIMD when errors signal overload, Vegas/Gradient when latency does.

The Problem

A static concurrency bound forces an impossible choice. Set it too low and you leave throughput on the table when the backend is healthy. Set it too high and you flood the backend the moment it slows down, driving it into congestion collapse — the exact failure a limiter is supposed to prevent. And the "right" number is never stable: it drifts with backend deploys, cache state, neighbouring tenants, and time of day.

  1. Capacity is unknown and moving. The safe concurrency for a database or downstream service is rarely documented and changes constantly. A hand-tuned bulkhead is stale the day after you tune it.
  2. Latency is the early warning, errors are the late one. A backend under stress slows down before it starts failing. A limiter that only reacts to errors acts too late; one that reads latency can back off while there is still time.
  3. Lockstep amplification. Many clients stepping their limits up at the same instant produce a synchronized surge — a self-inflicted thundering herd.

adaptx solves all three: it servo-controls the limit toward observed capacity, offers latency-driven laws (Vegas, Gradient) that react before failures appear, and jitters increases so a fleet of limiters does not march in lockstep.

Architectural Position

✅ Limiter              — servo-control a concurrency limit from feedback
✅ Execute[T]           — admit + run a callback, releasing the permit on return/panic
✅ Acquire / release    — manual admission when a callback does not fit
✅ Allow                — non-blocking probe without admission
✅ AdaptController      — limit + in-flight snapshot, SkipSample() to drop outliers
✅ AIMD/Vegas/Gradient  — three control laws for error- or latency-driven overload
✅ panic safety         — a panicking callback becomes a *panix.PanicError, not a crash

❌ NOT a rate limiter   — it bounds concurrency, not requests-per-second (see ratex)
❌ NOT a static bulkhead — the limit moves; for a fixed bound use bulkx
❌ NOT a circuit breaker — it throttles, it does not trip fully open (see circuitx)
❌ NOT a load shedder    — it has no per-request priority (see shedx)
❌ NOT a deadline        — it does not abort admitted work (compose with toutx)
Position in the urx Stack
┌─────────────────────────────────────────────────────────────┐
│  service code: DB pools, RPC clients, job dispatchers       │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│  adaptx   Limiter · Execute[T] · AdaptController            │
│           move the limit toward real capacity               │
└──────────────┬────────────────────────┬─────────────────────┘
               │                        │
┌──────────────▼─────────┐   ┌──────────▼─────────────────────┐
│  panix.Safe            │   │  chan + sync/atomic            │
│  (panic → PanicError)  │   │  (permit semaphore + counters) │
└────────────────────────┘   └────────────────────────────────┘

Architecture

                           adaptx
   ┌────────────────────┬────────────────────┬────────────────────┐
   │                    │                    │                    │
 Limiter (adaptx.go)   Option (options.go)  AdaptController (types.go)
   │                   config{algorithm,    execution{limit,inFlight,
 sem chan (permits)     utilization,         algorithm,skipped}
 atomic counters,       sampleWindow,...}    Limit/InFlight/
 mu: limit/debt/        │                    Algorithm/SkipSample
 window/estimators     WithAlgorithm         │
   │                   WithUtilization       Algorithm enum (types.go)
 Execute/TryExecute     WithSampleWindow     AIMD / Vegas / Gradient
 Acquire/release/Allow  WithOnLimitChange    AdaptFunc[T] (types.go)
   │                   (sync, must not block)│
 window snapshot →     ringCapacity()        ErrClosed/ErrTimeout
 aimd/vegas/gradient                         ErrCancelled/ErrNilFunc
 → permits                                   ErrDrainTimeout

How It Works

Execute(l, ctx, fn)
  │ fn == nil ? ───────────────────────────► ErrNilFunc
  │ closed ? ──────────────────────────────► ErrClosed
  │ ctx already cancelled ? ────────────────► ErrCancelled / ErrTimeout (no permit)
  │
  ├── Acquire: <-sem (block until a permit is free, ctx, or close)
  │     inFlight++ ; total++
  │
  ├── ac = {limit, inFlight, algorithm}     (admission snapshot)
  ├── (val,err) = panix.Safe(op, fn(ctx, ac))
  │
  └── release(success, latency):            (idempotent, runs once)
        record(success, latency)
          ├── success/fail counters++
          ├── peak in-flight for this window
          ├── skip (latency==0) ? ──► no ring, no window n
          └── else: ring ← sample ; window n/fails/meanRTT/minRTT
        if now−windowStart ≥ sampleWindow AND seen ≥ warmup:
          ONE adjust from snapshot {n, fails, maxInFlight, meanRTT, minRTT}
            ├── AIMD     : fails>0 → ×ratio once
            │              else if maxInFlight ≥ ceil(limit·utilization)
            │                → accumulate increaseRate credit, step=int(credit)
            │              else hold
            ├── Vegas    : queue = limit·(1 − minRTT/rtt)   rtt = window mean
            │              α = limit·tol [, ×(1 − minRTT/targetLatency)]
            │              queue<α → +1 ; queue>α·2 → ×ratio ; else hold
            └── Gradient : first window holds (avgLat unset); then avgLat = rtt
                           later avgLat = s·rtt + (1−s)·avgLat
                           fails>0 → ×ratio once
                           g=(rtt−avg)/avg ; g<−tol → +2 ; g≤tol → +1
                           else ×max(1−g·ratio, ratio)
          jitter, clamp to [min,max]
          grow → push permits (pay debt first)
          shrink → pull idle permits, else record debt
          onLimitChange(old,new) synchronously under recover (must not block)
          reset window counters
        inFlight-- ; releasePermit (pay debt or return permit)

The limit is a windowed servo, not a per-request counter. Samples accumulate for WithSampleWindow (default 1s). When the window elapses and warmup is done, the configured law runs once on the snapshot, then the window resets. Ten failures in one window are one multiplicative decrease, not ten halvings.

Admission rides a buffered-channel semaphore whose buffer is the configured maximum; the number of values currently buffered is the count of permits available to acquire. Acquiring receives a permit; releasing returns one. To grow the limit the controller pushes new permits into the channel; to shrink it, it pulls idle permits out, and for permits that are currently held it records debt — the next releases retire those permits instead of returning them. That is what makes a multiplicative decrease actually remove capacity without blocking a release.

In-flight vs live limit. Admission never takes a permit that is not in the semaphore, so in-flight never exceeds maxLimit. After a shrink, in-flight work that was already admitted may exceed the live limit until those releases pay the debt. Allow compares in-flight to the live limit without claiming a slot — it is a hint, not an admission.

Only the windowed adaptation step and the percentile snapshot take the mutex; the success/failure/reject counters are lock-free atomics. The callback runs under panix.Safe, so a panic becomes a *panix.PanicError and the permit is still released — a panicking handler can never leak capacity.

The three control laws
Algorithm Signal Grows when Backs off when Best for
AIMD window success / failure successful window at ≥ ceil(limit·utilization) any failure in the window (×ratio, once) failure-driven overload; the safe default
Vegas window mean RTT vs RTT_min estimated queue below α estimated queue above β = 2α a stable backend floor latency
Gradient window mean vs EMA average g at/below tolerance g above tolerance (proportional) drifting floor latency

AIMD is the TCP congestion-avoidance law, applied per window and gated on utilization: idle successes do not inflate the limit. Fractional WithIncreaseRate (for example 0.5) keeps a remainder so the limit grows by 1 every two eligible windows. Vegas infers queued work as limit·(1 − minRTT/rtt) — the denominator is the window mean RTT, not RTT_min — and holds the limit inside a tolerance band scaled by WithTargetLatency when the target sits above RTT_min. Gradient compares each window mean to an EMA that is initialized to the first window's RTT (not blended with zero). The first window holds the limit so a high opening RTT cannot grow concurrency before an average exists.

Keeping the feedback honest

Two mechanisms stop the controller from chasing noise. Warmup (WithWarmupSamples) ignores the first N samples so an unrepresentative cold start does not move the limit — windows still roll, but no adjust runs until seen ≥ warmup. RTT_min decay (WithMinLatencyDecay) drifts the recorded minimum toward the average on each completed window so Vegas cannot stick forever to one anomalously fast sample. The callback can also call AdaptController.SkipSample() to exclude a known outlier (a cache miss, a cold connection) from both the latency feedback and the percentile history — it still counts toward the success/failure totals.

Normative Contracts

Contract Guarantee
Bounded admission A missing permit is never taken; in-flight never exceeds maxLimit; the live limit stays in [min, max]
Shrink debt After a shrink, in-flight may exceed the live limit until released permits pay the debt
Allow is a hint Allow does not claim a slot; a concurrent admit may change the outcome
Limit floor min is floored to 1, so the limiter always makes forward progress
Context first A pre-cancelled context returns ErrCancelled/ErrTimeout without consuming a permit
Permit release The permit is released when the callback returns or panics
Idempotent release The release function runs its effect once; extra calls are no-ops
Windowed adjust The control law runs at most once per sample window, after warmup
Skip honesty SkipSample removes a call from latency feedback and history but not from success/failure totals
Warmup No adaptation occurs until warmupSamples samples have been recorded
Panic safety A panicking callback becomes a *panix.PanicError, permit still freed
Close semantics After close, admission returns ErrClosed; blocked waiters wake immediately
Idempotent Close Close() returns nil on the first and every later call
Drain timeout CloseWithTimeout returns ErrDrainTimeout if in-flight remains; the limiter stays closed
Limit-change hook Runs synchronously under recover; must not block or panic
Controller scope An AdaptController is valid only during its callback; do not retain it

Quick Start

package main

import (
	"context"
	"database/sql"
	"errors"
	"fmt"

	"github.com/aasyanov/urx/adaptx"
)

func main() {
	l := adaptx.New(
		adaptx.WithAlgorithm(adaptx.Gradient),
		adaptx.WithInitialLimit(10),
		adaptx.WithMaxLimit(200),
	)
	defer l.Close()

	rows, err := adaptx.Execute(l, context.Background(),
		func(ctx context.Context, ac adaptx.AdaptController) (*sql.Rows, error) {
			if ac.InFlight() > ac.Limit()/2 {
				return queryCheap(ctx) // shed load near saturation
			}
			return queryFull(ctx)
		})

	switch {
	case errors.Is(err, adaptx.ErrClosed):
		fmt.Println("closed")
	case err != nil:
		fmt.Println("failed:", err)
	default:
		_ = rows
		fmt.Println("ok")
	}
}

func queryCheap(context.Context) (*sql.Rows, error) { return nil, nil }
func queryFull(context.Context) (*sql.Rows, error)  { return nil, nil }

Usage Scenarios

Protect a database connection pool
l := adaptx.New(adaptx.WithAlgorithm(adaptx.Vegas), adaptx.WithMaxLimit(maxConns))
defer l.Close()

rows, err := adaptx.Execute(l, ctx,
	func(ctx context.Context, _ adaptx.AdaptController) (*sql.Rows, error) {
		return db.QueryContext(ctx, q, args...)
	})
Manual admission with a release function
release, err := l.Acquire(ctx)
if err != nil {
	return err // ErrClosed / ErrTimeout / ErrCancelled
}
start := time.Now()
err = stream(ctx) // in-flight tracked across many statements
release(err == nil, time.Since(start))
Drop outlier latencies from the feedback
resp, _ := adaptx.Execute(l, ctx,
	func(ctx context.Context, ac adaptx.AdaptController) (*Resp, error) {
		if coldStart {
			ac.SkipSample() // do not let a cold connection move the limit
		}
		return call(ctx)
	})
Non-blocking fast path
ran, val, err := adaptx.TryExecute(l, ctx,
	func(ctx context.Context, _ adaptx.AdaptController) (Result, error) {
		return compute(ctx)
	})
if !ran {
	return serveStale() // limiter saturated, no permit available
}
Probe without admission
if !l.Allow() {
	return serveCached() // at capacity, skip the expensive path
}
Observe limit changes
l := adaptx.New(adaptx.WithOnLimitChange(func(old, new int) {
	metrics.Gauge("adaptx.limit").Set(float64(new))
}))

The hook runs synchronously on the goroutine that closed the sample window. It must not block or panic; a panic is recovered and discarded.

API

Symbol Signature Description
New func New(opts ...Option) *Limiter Create a limiter with defaults + options
Execute func Execute[T any](l *Limiter, ctx context.Context, fn AdaptFunc[T]) (T, error) Admit + run a callback (recommended)
TryExecute func TryExecute[T any](l *Limiter, ctx context.Context, fn AdaptFunc[T]) (bool, T, error) Non-blocking variant; (false, …) when saturated
Limiter.Allow func (l *Limiter) Allow() bool Probe whether a permit is free (no admission)
Limiter.Acquire func (l *Limiter) Acquire(ctx context.Context) (release func(bool, time.Duration), err error) Blocking manual admission
Limiter.TryAcquire func (l *Limiter) TryAcquire() (release func(bool, time.Duration), ok bool) Non-blocking manual admission
Limiter.Limit func (l *Limiter) Limit() int Current adaptive limit
Limiter.InFlight func (l *Limiter) InFlight() int Current in-flight count
Limiter.Stats func (l *Limiter) Stats() Stats Counter + latency-percentile snapshot
Limiter.ResetStats func (l *Limiter) ResetStats() Zero counters, reset adaptive state
Limiter.Close func (l *Limiter) Close() error Idempotent shutdown; always returns nil (does not wait)
Limiter.CloseWithTimeout func (l *Limiter) CloseWithTimeout(d time.Duration) error Shutdown with custom drain; ErrDrainTimeout / ErrClosed
Limiter.IsClosed func (l *Limiter) IsClosed() bool Report closed state
Algorithm type Algorithm uint8 AIMD / Vegas / Gradient
AIMD, Vegas, Gradient constants Control-law selectors
AdaptFunc[T] func(ctx context.Context, ac AdaptController) (T, error) Unit of work for Execute / TryExecute
AdaptController
Method Signature Description
Limit Limit() int Limit at admission time
InFlight InFlight() int In-flight count at admission (excludes self)
Algorithm Algorithm() Algorithm Active adaptation algorithm
SkipSample SkipSample() Exclude this call from latency feedback (idempotent)

Configuration

Option Default Description
WithAlgorithm(a) AIMD Control law; unknown values behave as AIMD
WithInitialLimit(n) 10 Starting limit; clamped into [min, max]
WithMinLimit(n) 1 Floor; ≤ 0 ignored, final value floored to 1
WithMaxLimit(n) 1000 Ceiling and semaphore capacity; below min raised to min
WithSmoothing(f) 0.2 EMA weight per window mean RTT; outside (0, 1] ignored
WithIncreaseRate(r) 1.0 AIMD additive credit per eligible window; ≤ 0 ignored; fractional rates accumulate
WithDecreaseRatio(r) 0.5 Multiplicative backoff factor per backoff window; outside (0, 1) ignored
WithUtilization(f) 0.9 AIMD increase gate: peak in-flight ≥ ceil(limit·f); outside (0, 1] ignored
WithTargetLatency(d) 100ms Vegas operating-point RTT; scales the queue target band; ≤ 0 ignored
WithTolerance(f) 0.1 Vegas/Gradient deviation band; outside (0, 1] ignored
WithSampleWindow(d) 1s Aggregation window for one adjust and Stats percentiles; ≤ 0 ignored
WithWarmupSamples(n) 10 Samples before adaptation; 0 disables warmup
WithMinLatencyDecay(f) 0.001 RTT_min drift toward average per window; 0 disables, outside [0,1] ignored
WithJitter(f) 0.1 Fraction of an increase that may be withheld; 0 disables
WithOp(s) [opExecute] / [opTryExecute] Operation name attached to panic reports
WithOnLimitChange(fn) none Synchronous callback on every limit change; must not block

Errors

Error Condition
ErrClosed Admission attempted after close; also a second CloseWithTimeout
ErrTimeout Blocking acquire exceeded its context deadline (wraps context.DeadlineExceeded)
ErrCancelled Context cancelled before a permit was available (Execute, Acquire, TryExecute; wraps ctx.Err())
ErrNilFunc Execute/TryExecute given a nil function
ErrDrainTimeout CloseWithTimeout deadline elapsed while in-flight work remains; limiter stays closed

A panicking callback surfaces as a *panix.PanicError returned by Execute (reach it with errors.As); the permit is still released. Close() itself always returns nil; use CloseWithTimeout when a drain failure must be visible.

Pitfalls

[!WARNING] adaptx bounds concurrency, not request rate. A flood of fast successes at high utilization will raise the limit, not throttle arrivals. Serial traffic well below ceil(limit·utilization) holds AIMD still. For requests-per-second limiting use ratex; compose the two for both axes.

[!WARNING] After a shrink, in-flight may exceed the live limit. Already-admitted work is not cancelled. New admissions wait for debt to be paid. Allow is a hint: it compares in-flight to the live limit without taking a permit.

[!WARNING] Choose the algorithm to match your overload signal. Vegas and Gradient need representative latency: if your work has wildly bimodal latency (cache hit vs miss) without SkipSample, they will thrash. When failures — not latency — are the overload signal, prefer AIMD.

[!WARNING] WithOnLimitChange must not block. The hook runs on the goroutine that closed the window. A slow hook stalls every subsequent record. Panics are recovered and discarded.

[!NOTE] ResetStats snaps the limit immediately. Counters, latency estimators, window credit, and the permit pool are reset to the configured initial limit in one step. When in-flight work exceeds that initial limit the live limit is raised to the in-flight count so permits never go negative.

[!NOTE] SkipSample keeps the success/failure totals. A skipped call is removed from latency feedback, percentile history, and the AIMD window peak in-flight — it still counts as a success or failure in Stats. Use it for outlier latency, not to hide errors.

[!NOTE] Close() does not wait. It is CloseWithTimeout(0): in-flight work is not joined, drain timeout is swallowed, and the call always returns nil. Use CloseWithTimeout(DefaultCloseTimeout) when shutdown must wait up to 30s. CloseWithTimeout(0) itself returns ErrDrainTimeout if in-flight work remains; the limiter stays closed.

Safety and Concurrency

Limiter is safe for concurrent use from any number of goroutines. Admission rides a buffered-channel semaphore; the success, failure, reject, and adjustment counters are sync/atomic. A single mutex guards the adaptive state (limit, shrink debt, latency estimators, sample ring, window counters) and is taken on each completed sample and on the percentile snapshot — never on the fast admission path beyond a single Limit() read. Growing the limit pushes permits into the channel; shrinking pulls idle permits and records debt so held permits are retired on release. Already-admitted in-flight work may exceed the live limit until that debt is paid; it never exceeds maxLimit. The release function uses an atomic compare-and-swap so a double call is a no-op. The AdaptController is touched only by the goroutine running its callback. CloseWithTimeout waits with a timer/select on a drain channel, not time.Sleep. Every test runs under -race, including 50-goroutine admission stress that asserts in-flight returns to zero.

Benchmarks

Three environments, two hardware classes, two operating systems. All values are medians. B/op and allocs/op are deterministic — they depend on code, not hardware.

Environments
Laptop CI Server (Linux) CI Server (Windows)
CPU Intel Core i7-10510U, 4C/8T Intel Xeon 6973P-C AMD EPYC 7763
TDP 15W (mobile, throttles) 280W (server, stable) server, stable
OS Windows 10 Ubuntu Windows Server 2022
Go 1.26.2 1.26 1.26
GOMAXPROCS 8 4 4
Runs 3 (-count=3) 3 (-count=3) 3 (-count=3)

This gives three comparison axes: laptop vs server (hardware scaling), Linux vs Windows (OS mutex/timer impact), and serial vs parallel (channel + atomic contention under load).

Admission Path
Benchmark What it measures Laptop Linux Windows B/op allocs/op
Execute Admit + callback + release 250 ns 254.8 ns 221 ns 52 3
Execute_Parallel Execute, 8/4 goroutines 497 ns 507.2 ns 399.5 ns 52 3
TryExecute Non-blocking admit path 184 ns 237.1 ns 175.4 ns 52 3
Acquire Semaphore only (no callback) 164 ns 103.2 ns 147.8 ns 28 2
Acquire_Parallel Acquire, 8/4 goroutines 344 ns 174.6 ns 341.5 ns 28 2
TryAcquire Non-blocking acquire 123 ns 83.8 ns 109.8 ns 28 2
Allow Read-only admission check 13.4 ns 11.4 ns 5.4 ns 0 0
Limit Current limit snapshot 13.5 ns 12.3 ns 7.2 ns 0 0
Analysis

Pure CPU + channel + atomic — no I/O. Every benchmark is in-process: buffered-channel semaphore, atomic counters, and (on the adaptation step only) a mutex. The Linux vs Windows gap on the same server class is dominated by mutex and channel fast-path cost, not filesystem or timer resolution.

Windows CI is faster on the admit path. Execute is 254.8 ns (Linux) vs 221 ns (Windows) — a 1.2× spread on identical -count=3 methodology. TryExecute shows the same pattern (237.1 ns vs 175.4 ns). The three heap allocations (release closure, captured atomic.Bool, execution controller) are fixed on every admit; the remaining time is channel send/receive and atomic bumps. EPYC 7763 on the Windows runner appears to win on this hot path despite the OS overhead seen in other urx packages.

Parallel acquire is OS-sensitive. Acquire_Parallel is 174.6 ns (Linux) vs 341.5 ns (Windows) — Linux 1.8× faster under four goroutines contending on the same semaphore. Execute_Parallel inverts again (507.2 ns Linux vs 399.5 ns Windows), because the execute path adds callback setup work that amortizes channel contention differently. Pick TryAcquire/Acquire when you need raw slot reservation without the 52 B controller overhead.

Laptop sits between CI platforms on serial paths, worse on parallel. Serial Execute (250 ns) beats Linux CI but loses to Windows CI. Acquire_Parallel at 344 ns matches Windows, not Linux — the 8-thread laptop runs more goroutines than the 4-slot semaphore can serve without queueing, inflating parallel numbers versus the 4-vCPU CI matrix.

Execute / TryExecute — 3 allocs is the architectural floor. The three allocations are the release closure, the atomic.Bool it captures for double-call safety, and the execution controller handed to the callback through the panix.Safe boundary as an interface. The semaphore receive and atomic counter bumps are otherwise alloc-free.

Acquire / TryAcquire — 2 allocs / 28 B. Only the release closure and its captured atomic.Bool escape. TryAcquire is ~25% cheaper than Acquire on CI because it skips the context pre-check and blocking select.

Allow / Limit — 0 allocs, ~5–13 ns on CI. A mutex lock/unlock around limit and in-flight reads. Safe to poll from a metrics loop; Allow is cheaper when you only need a yes/no without running a callback.

Quality

Metric Value
Test functions 88
Benchmarks 8
Fuzz targets 2
Examples 4
Coverage 95.2%
Race detector All pass (go test -race -count=1 ./adaptx/)
External deps 0 (panix; testify in dev only) 
go test -race -count=1 ./adaptx/
go test -run='^$' -bench=. -benchmem -count=5 ./adaptx/
go test -fuzz=FuzzNew -fuzztime=30s ./adaptx/
go test -fuzz=FuzzExecute -fuzztime=30s ./adaptx/

File Structure

adaptx/
├── adaptx.go           # package doc + Limiter + Execute/TryExecute + Acquire + windowed adaptation
├── options.go          # config, Option, defaults, WithXxx, withClock (tests)
├── types.go            # Algorithm enum + AdaptController + private execution impl + sample
├── errors.go           # ErrClosed, ErrTimeout, ErrCancelled, ErrNilFunc, ErrDrainTimeout
├── adaptx_test.go      # unit + table-driven + race tests
├── errors_test.go      # sentinel errors.Is coverage
├── bench_test.go       # benchmarks (sequential + parallel)
├── fuzz_test.go        # FuzzNew, FuzzExecute — construction + permit-accounting invariants
├── example_test.go     # runnable GoDoc examples
├── footprint_test.go   # struct size guards
└── README.md           # this file

License

MIT — see LICENSE in the repository root.

Documentation

Overview

Package adaptx provides adaptive concurrency limiting for production Go services.

A Limiter discovers a backend's safe concurrency on its own. It starts at a configured limit and moves that limit up or down once per sample window from latency and error feedback, using one of three control laws — AIMD, Vegas, or Gradient. Where a static bulkhead (see bulkx) must be sized by hand to a fixed guess, an adaptive limiter tracks capacity as it changes: it opens up when the backend is fast and healthy, and clamps down the moment latency climbs or errors appear, so callers wait (or are turned away) instead of piling onto a struggling backend.

Quick Start 

l := adaptx.New(
    adaptx.WithAlgorithm(adaptx.Gradient),
    adaptx.WithInitialLimit(10),
)
defer l.Close()

rows, err := adaptx.Execute(l, ctx,
    func(ctx context.Context, ac adaptx.AdaptController) (*sql.Rows, error) {
        if ac.InFlight() > ac.Limit()/2 {
            return db.QueryContext(ctx, simpleSQL) // shed load near saturation
        }
        return db.QueryContext(ctx, complexSQL)
    })

The callback receives an AdaptController exposing the limit and in-flight count at admission and an AdaptController.SkipSample method to keep outlier latencies out of the feedback signal. For tracked admission without a callback, use Limiter.Acquire and call the returned release function.

Each callback is wrapped with github.com/aasyanov/urx/panix for panic recovery; a panicking function yields a *panix.PanicError instead of crashing the process, and the in-flight slot is always released.

Dependencies

adaptx depends only on the Go standard library and the urx panix package.

Index

Examples

Constants

View Source 
const (
	// DefaultInitialLimit is the concurrency limit a [Limiter] starts at before
	// any adaptation, applied when [WithInitialLimit] is not supplied.
	DefaultInitialLimit = 10

	// DefaultMinLimit is the floor the adaptive limit is never driven below,
	// applied when [WithMinLimit] is not supplied. A floor of 1 keeps the
	// limiter able to make forward progress even under sustained failure.
	DefaultMinLimit = 1

	// DefaultMaxLimit is the ceiling the adaptive limit is never driven above,
	// applied when [WithMaxLimit] is not supplied.
	DefaultMaxLimit = 1000

	// DefaultSmoothing is the EMA weight applied to each window's mean RTT when
	// updating the smoothed average, applied when [WithSmoothing] is not
	// supplied. Higher reacts faster but is noisier.
	DefaultSmoothing = 0.2

	// DefaultIncreaseRate is the additive credit [AIMD] accumulates on each
	// successful, high-utilization window, applied when [WithIncreaseRate] is
	// not supplied. Values below 1 grow the limit every few windows (0.5 → +1
	// every two windows).
	DefaultIncreaseRate = 1.0

	// DefaultDecreaseRatio is the multiplicative factor the limit is scaled by
	// on a backoff window, applied when [WithDecreaseRatio] is not supplied.
	// 0.5 halves the limit, matching TCP multiplicative decrease.
	DefaultDecreaseRatio = 0.5

	// DefaultUtilization is the in-flight fraction of the live limit that [AIMD]
	// requires before it will add credit, applied when [WithUtilization] is not
	// supplied. A window whose peak in-flight is below ceil(limit·utilization)
	// holds the limit even if every sample succeeded.
	DefaultUtilization = 0.9

	// DefaultTargetLatency is the latency [Vegas] treats as the operating point,
	// applied when [WithTargetLatency] is not supplied.
	DefaultTargetLatency = 100 * time.Millisecond

	// DefaultTolerance is the fractional latency deviation [Vegas] and
	// [Gradient] tolerate before reacting, applied when [WithTolerance] is not
	// supplied.
	DefaultTolerance = 0.1

	// DefaultSampleWindow is the interval over which samples are aggregated
	// into one control-law adjustment and over which [Stats] computes latency
	// percentiles, applied when [WithSampleWindow] is not supplied.
	DefaultSampleWindow = 1 * time.Second

	// DefaultWarmupSamples is the number of recorded samples collected before
	// adaptation begins, applied when [WithWarmupSamples] is not supplied. It
	// stops the controller from reacting to the first few unrepresentative
	// calls.
	DefaultWarmupSamples = 10

	// DefaultMinLatencyDecay is the fraction by which RTT_min drifts toward the
	// running average on each completed window, applied when
	// [WithMinLatencyDecay] is not supplied. It prevents [Vegas] from sticking
	// to an anomalously low minimum forever. 0 disables decay.
	DefaultMinLatencyDecay = 0.001

	// DefaultJitter is the fraction of each limit increase that may be randomly
	// withheld to desynchronize many limiters, applied when [WithJitter] is not
	// supplied. 0 disables jitter.
	DefaultJitter = 0.1
)
View Source 
const (
	// DefaultCloseTimeout is a suggested drain bound for [Limiter.CloseWithTimeout].
	// [Limiter.Close] itself does not wait.
	DefaultCloseTimeout = 30 * time.Second
)

Variables

View Source 
var (
	// ErrClosed is returned by [Limiter.Acquire], [Execute], and related methods
	// after [Limiter.Close] has been called. Safe to compare with == or
	// [errors.Is]. Also returned by a second [Limiter.CloseWithTimeout] after
	// the first call has already begun shutdown.
	ErrClosed = errors.New("adaptx: limiter is closed")

	// ErrTimeout is returned when a blocking acquire exceeds its deadline before
	// a slot becomes available. The joined error carries [context.DeadlineExceeded];
	// reach it with [errors.Unwrap] or test it with [errors.Is]. Safe to compare
	// with == or [errors.Is].
	ErrTimeout = errors.New("adaptx: acquire timed out")

	// ErrCancelled is returned when the caller's context is cancelled before a
	// slot becomes available. The joined error carries ctx.Err(); reach it with
	// [errors.Unwrap] or test it with [errors.Is]. Safe to compare with == or
	// [errors.Is].
	ErrCancelled = errors.New("adaptx: acquire cancelled")

	// ErrNilFunc is returned by [Execute] and [TryExecute] when the supplied
	// function is nil. Safe to compare with == or [errors.Is].
	ErrNilFunc = errors.New("adaptx: nil function")

	// ErrDrainTimeout is returned by [Limiter.CloseWithTimeout] when in-flight
	// work is still running after the drain deadline. The limiter stays closed;
	// remaining work is not cancelled. Safe to compare with == or [errors.Is].
	ErrDrainTimeout = errors.New("adaptx: drain timed out")
)

Functions

func Execute 

func Execute[T any](l *Limiter, ctx context.Context, fn AdaptFunc[T]) (T, error)

Execute admits one operation and runs fn under panic recovery. Because Go methods cannot have type parameters, Execute is a package-level generic function taking the Limiter as its first argument; it is the recommended way to use the limiter.

Execute blocks for a permit exactly as Limiter.Acquire does and reports the same admission errors: ErrClosed, ErrTimeout, or ErrCancelled. It returns ErrNilFunc if fn is nil. On admission the permit is held for the duration of fn and released even if fn panics — the callback runs under panix.Safe, so a panic becomes a *panix.PanicError. The call's latency and outcome feed the adaptive algorithm unless the callback invokes AdaptController.SkipSample.

Example

ExampleExecute shows the recommended callback form: the limiter admits the call, the callback adapts its work to the admission snapshot, and the result feeds the adaptive algorithm. 

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/aasyanov/urx/adaptx"
)

func main() {
	l := adaptx.New(
		adaptx.WithAlgorithm(adaptx.Gradient),
		adaptx.WithInitialLimit(10),
	)
	defer l.Close()

	result, err := adaptx.Execute(l, context.Background(),
		func(ctx context.Context, ac adaptx.AdaptController) (string, error) {
			if ac.InFlight() > ac.Limit()/2 {
				return "cheap", nil // shed load near saturation
			}
			return "full", nil
		})

	switch {
	case errors.Is(err, adaptx.ErrClosed):
		fmt.Println("closed")
	case err != nil:
		fmt.Println("failed:", err)
	default:
		fmt.Println("ok:", result)
	}
}

Output:
ok: full

func TryExecute 

func TryExecute[T any](l *Limiter, ctx context.Context, fn AdaptFunc[T]) (bool, T, error)

TryExecute runs fn only if a permit is immediately available, without blocking. It returns (true, val, err) when fn ran and (false, zero, nil) when no permit was free. Returns (false, zero, ErrClosed) if the limiter is closed, (false, zero, ErrNilFunc) if fn is nil, and (false, zero, ErrCancelled or ErrTimeout) when ctx is already cancelled or its deadline has expired (no permit consumed). The permit is released when fn returns or panics.

Example

ExampleTryExecute shows the non-blocking variant: when no permit is free the call is skipped rather than queued. 

package main

import (
	"context"
	"fmt"

	"github.com/aasyanov/urx/adaptx"
)

func main() {
	l := adaptx.New(adaptx.WithInitialLimit(1), adaptx.WithMaxLimit(1))
	defer l.Close()

	ran, val, err := adaptx.TryExecute(l, context.Background(),
		func(ctx context.Context, ac adaptx.AdaptController) (int, error) {
			return 7, nil
		})
	fmt.Printf("ran=%v val=%d err=%v\n", ran, val, err)
}

Output:
ran=true val=7 err=<nil>

Types

type AdaptController 

type AdaptController interface {
	// Limit returns the concurrency limit in effect at admission time.
	Limit() int

	// InFlight returns the number of operations in flight at admission time,
	// excluding this one.
	InFlight() int

	// Algorithm returns the active adaptation algorithm.
	Algorithm() Algorithm

	// SkipSample tells the limiter not to feed this call's latency and outcome
	// into the adaptive algorithm. Use it for outlier operations whose latency
	// would mislead the controller (cache misses, cold starts, admin calls).
	// The call still counts toward the success/failure totals in [Stats] but
	// does not raise the window peak in-flight used by AIMD utilization. Safe
	// to call multiple times; only the first call has an effect.
	SkipSample()
}

AdaptController exposes the admission snapshot to the Execute callback and lets it opt out of feeding its result into the adaptive algorithm. The implementation is private; callers interact only through this interface. An AdaptController is bound to a single Execute call and must not be retained after the callback returns.

The concurrency limit is decided at admission: by the time the callback runs the request is already admitted. The controller therefore exposes the limit and in-flight count captured at admission so the callback can adapt its work to the observed pressure — for example, serve a cheaper query when the limiter is near saturation. AdaptController.SkipSample removes outlier calls (cache misses, cold starts) from the feedback signal so a single anomalous latency does not mislead the controller.

type AdaptFunc 

type AdaptFunc[T any] func(ctx context.Context, ac AdaptController) (T, error)

AdaptFunc is the unit of work run by Execute and TryExecute. It receives the call context and an AdaptController, and runs under panic recovery: a panicking function becomes a *panix.PanicError.

type Algorithm 

type Algorithm uint8

Algorithm selects the strategy a Limiter uses to move its concurrency limit in response to latency and error feedback. Each law runs once per completed sample window, not once per request. 

const (
	// AIMD is Additive Increase / Multiplicative Decrease: after a successful
	// window that reached the utilization gate the limit grows by a fractional
	// credit of [WithIncreaseRate]; a window with any failure is cut once by
	// [WithDecreaseRatio]. It needs no latency target and is the safest default
	// — the same control law TCP congestion avoidance uses. Best when failures
	// (not latency) are the overload signal.
	AIMD Algorithm = iota

	// Vegas estimates queue build-up from round-trip time, in the spirit of
	// TCP Vegas. It compares the window's mean RTT against the best latency
	// seen (RTT_min) as queue = limit·(1 − minRTT/rtt), then grows the limit
	// while the estimated queue is below α and shrinks it when the queue
	// exceeds β = α·2. Best when a backend has a stable, measurable floor
	// latency.
	Vegas

	// Gradient reacts to the trend of latency: it grows the limit while the
	// window mean is at or below the EMA average and backs off in proportion
	// to how far the window mean has risen above it. Best for backends whose
	// floor latency drifts, where an absolute target would go stale.
	Gradient
)

func (Algorithm) String 

func (a Algorithm) String() string

String returns a human-readable label for the algorithm.

type Limiter 

type Limiter struct {
	// contains filtered or unexported fields
}

Limiter is a thread-safe adaptive concurrency limiter. Create one with New, run work with Execute or admit it manually with Limiter.Acquire, inspect counters with Limiter.Stats, and release resources with Limiter.Close.

It is safe for concurrent use from multiple goroutines. Admission rides a buffered-channel semaphore; only the periodic windowed adaptation step and the percentile snapshot take the mutex.

After a shrink, in-flight work may briefly exceed the live limit until released permits pay the shrink debt. New admissions never take a permit that is not in the semaphore; in-flight never exceeds WithMaxLimit.

func New 

func New(opts ...Option) *Limiter

New creates a Limiter with the given options applied on top of the package defaults (AIMD, initial limit DefaultInitialLimit, bounds DefaultMinLimit–DefaultMaxLimit). Invalid options are ignored and cross-field invariants are enforced, so New never returns an unusable limiter.

func (*Limiter) Acquire 

func (l *Limiter) Acquire(ctx context.Context) (release func(success bool, latency time.Duration), err error)

Acquire blocks until a permit is available, the context is cancelled, or the limiter is closed. It returns a release function that MUST be called exactly once with the operation outcome and measured latency; the release function is idempotent, so extra calls are no-ops.

Returns ErrClosed if the limiter has been closed, ErrTimeout if the context deadline is exceeded while waiting, or ErrCancelled if the context is cancelled. Acquire is the building block for code that cannot use the callback form of Execute; the caller owns the returned release function and must invoke it to free the permit and feed the adaptive algorithm.

Example

ExampleLimiter_Acquire shows manual admission for code that cannot use a single callback. The release function must be called exactly once with the outcome and measured latency. 

package main

import (
	"context"
	"fmt"

	"github.com/aasyanov/urx/adaptx"
)

func main() {
	l := adaptx.New(adaptx.WithInitialLimit(5))
	defer l.Close()

	release, err := l.Acquire(context.Background())
	if err != nil {
		fmt.Println("acquire failed:", err)
		return
	}
	// ... do work, measure latency ...
	release(true, 0)

	fmt.Println("in-flight:", l.InFlight())
}

Output:
in-flight: 0

func (*Limiter) Allow 

func (l *Limiter) Allow() bool

Allow reports whether a permit is currently free without acquiring it. It does not track anything or mutate any counter; use Execute, TryExecute, or Limiter.Acquire for tracked admission. Returns false once the limiter is closed.

Allow is a best-effort hint: it compares in-flight work against the live limit without claiming a slot, so a concurrent admission may change the outcome before the caller acts. After a shrink, in-flight may still exceed the live limit (debt not yet paid), in which case Allow reports false even though no new permit is available. Only the tracked entry points enforce the concurrency bound.

func (*Limiter) Close 

func (l *Limiter) Close() error

Close shuts the limiter down without waiting for in-flight work. Use Limiter.CloseWithTimeout (for example with DefaultCloseTimeout) when drain must complete before return. Close is idempotent: the first and every later call return nil. An incomplete drain is swallowed; the limiter is still closed.

func (*Limiter) CloseWithTimeout 

func (l *Limiter) CloseWithTimeout(timeout time.Duration) error

CloseWithTimeout shuts the limiter down, waiting up to timeout for in-flight operations to drain before returning. Blocked Limiter.Acquire waiters are released immediately with ErrClosed. Subsequent Limiter.Acquire, Limiter.TryAcquire, Execute, and TryExecute calls return ErrClosed. A zero or negative timeout returns immediately without waiting. If in-flight work remains after the wait, CloseWithTimeout returns ErrDrainTimeout and the limiter stays closed. The first call performs shutdown; later calls return ErrClosed.

func (*Limiter) InFlight 

func (l *Limiter) InFlight() int

InFlight returns the number of operations currently admitted and running.

func (*Limiter) IsClosed 

func (l *Limiter) IsClosed() bool

IsClosed reports whether Limiter.Close has been called.

func (*Limiter) Limit 

func (l *Limiter) Limit() int

Limit returns the current adaptive concurrency limit.

func (*Limiter) ResetStats 

func (l *Limiter) ResetStats()

ResetStats zeroes the cumulative counters and resets the adaptive state back to the initial limit, clearing the latency estimators, window counters, and sample history. It does not affect the in-flight count or the closed state. When in-flight work exceeds the configured initial limit the live limit is raised to that count so permits never go negative; the permit pool is reconciled immediately.

func (*Limiter) Stats 

func (l *Limiter) Stats() Stats

Stats returns a snapshot of limiter statistics. Latency percentiles are computed over the samples recorded within the configured sample window; with no recent samples the latency fields are zero.

func (*Limiter) TryAcquire 

func (l *Limiter) TryAcquire() (release func(success bool, latency time.Duration), ok bool)

TryAcquire attempts to take a permit without blocking. It returns the release function and true on success, or (nil, false) when no permit is immediately available or the limiter is closed. The release function MUST be called exactly once on success and is idempotent.

type Option 

type Option func(*config)

Option configures a Limiter created by New.

func WithAlgorithm 

func WithAlgorithm(a Algorithm) Option

WithAlgorithm selects the adaptation strategy. Default: AIMD. An unknown value falls back to AIMD at adaptation time.

func WithDecreaseRatio 

func WithDecreaseRatio(r float64) Option

WithDecreaseRatio sets the multiplicative backoff factor applied to the limit on a failure or overload window. Default: DefaultDecreaseRatio. Values outside (0, 1) are ignored.

func WithIncreaseRate 

func WithIncreaseRate(r float64) Option

WithIncreaseRate sets the additive credit AIMD accumulates on each successful window that meets the utilization gate. Default: DefaultIncreaseRate. Values <= 0 are ignored. Fractional rates keep a remainder so 0.5 grows the limit by 1 every two windows.

func WithInitialLimit 

func WithInitialLimit(n int) Option

WithInitialLimit sets the concurrency limit the limiter starts at before any adaptation. Default: DefaultInitialLimit. Values <= 0 are ignored; the final value is clamped into [min, max].

func WithJitter 

func WithJitter(f float64) Option

WithJitter sets the fraction of each limit increase that may be randomly withheld, desynchronizing many limiters so they do not all step up in lockstep (thundering herd). Default: DefaultJitter. 0 disables jitter. Values outside [0, 1] are ignored.

func WithMaxLimit 

func WithMaxLimit(n int) Option

WithMaxLimit sets the ceiling the adaptive limit is never driven above and the hard cap on concurrently admitted operations. Default: DefaultMaxLimit. Values <= 0 are ignored; a value below the minimum is raised to it.

func WithMinLatencyDecay 

func WithMinLatencyDecay(f float64) Option

WithMinLatencyDecay sets the fraction by which the observed minimum latency drifts toward the running average on each completed window, preventing Vegas from sticking to an anomalously low minimum. Default: DefaultMinLatencyDecay. 0 disables decay. Values outside [0, 1] are ignored.

func WithMinLimit 

func WithMinLimit(n int) Option

WithMinLimit sets the floor the adaptive limit is never driven below. Default: DefaultMinLimit. Values <= 0 are ignored; the final value is floored to 1.

func WithOnLimitChange 

func WithOnLimitChange(fn func(oldLimit, newLimit int)) Option

WithOnLimitChange registers a callback invoked synchronously whenever the adaptive limit changes, receiving the old and new values. Default: none. The callback must not block or panic; it runs on the goroutine that closed the sample window, and a panic is recovered and discarded.

func WithOp 

func WithOp(op string) Option

WithOp sets the logical operation name attached to panic reports raised by the callback (e.g. "db.query", "api.search"). Default: [opExecute] for Execute and [opTryExecute] for TryExecute. Empty values are ignored.

func WithSampleWindow 

func WithSampleWindow(d time.Duration) Option

WithSampleWindow sets the interval over which completed operations are aggregated into one control-law adjustment, and over which Stats computes latency percentiles. Default: DefaultSampleWindow. Values <= 0 are ignored.

func WithSmoothing 

func WithSmoothing(f float64) Option

WithSmoothing sets the EMA weight applied to each window's mean RTT. Default: DefaultSmoothing. Values outside (0, 1] are ignored.

func WithTargetLatency 

func WithTargetLatency(d time.Duration) Option

WithTargetLatency sets the round-trip latency Vegas treats as the operating point when scaling the queue target band. Default: DefaultTargetLatency. Values <= 0 are ignored. When target latency is at or below the observed minimum RTT the band falls back to limit·tolerance.

func WithTolerance 

func WithTolerance(f float64) Option

WithTolerance sets the fractional latency deviation Vegas and Gradient tolerate before reacting. Default: DefaultTolerance. Values outside (0, 1] are ignored.

func WithUtilization added in v1.5.2 

func WithUtilization(f float64) Option

WithUtilization sets the in-flight fraction of the live limit that AIMD requires before it will add increase credit. Default: DefaultUtilization. Values outside (0, 1] are ignored. A window whose peak in-flight is below ceil(limit·utilization) holds the limit.

func WithWarmupSamples 

func WithWarmupSamples(n int) Option

WithWarmupSamples sets the number of recorded samples collected before adaptation begins. Default: DefaultWarmupSamples. 0 disables warmup so adaptation starts on the first completed window. Negative values are ignored.

type Stats 

type Stats struct {
	Algorithm string        `json:"algorithm"`
	Limit     int           `json:"limit"`
	MinLimit  int           `json:"min_limit"`
	MaxLimit  int           `json:"max_limit"`
	InFlight  int           `json:"in_flight"`
	Total     int64         `json:"total"`
	Success   int64         `json:"success"`
	Failures  int64         `json:"failures"`
	Rejected  int64         `json:"rejected"`
	Increases int64         `json:"increases"`
	Decreases int64         `json:"decreases"`
	AvgLat    time.Duration `json:"avg_latency"`
	MinLat    time.Duration `json:"min_latency"`
	MaxLat    time.Duration `json:"max_latency"`
	P50Lat    time.Duration `json:"p50_latency"`
	P99Lat    time.Duration `json:"p99_latency"`
}

Stats holds a point-in-time snapshot of limiter counters and latency percentiles computed over the configured sample window.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.