ambatukam

package module
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Jun 26, 2026 License: MIT Imports: 17 Imported by: 0

README

Ambatukam Go

Ambatukam Go

Composable, idiomatic Go HTTP resilience.
Retry · Circuit Breaker · Bulkhead · Rate Limiter · Timeout · Hooks

CI Go Reference Go Report Card License Stars

One library. One API. Zero dependencies. Production-grade resilience in 10 lines.

Why Ambatukam Go?

Every Go backend that calls external services needs the same five things: retry on transient failures, circuit breaker to fail fast when downstream is down, bulkhead to limit concurrency, rate limiting to respect API quotas, and per-attempt timeout to bound latency.

Most teams stitch together 3–5 different libraries and write glue code nobody owns.

Ambatukam Go is one library with one API.

Feature Ambatukam Go Stitched Stack
Retry with backoff + jitter ✅ cenkalti/backoff
Circuit breaker (closed/open/half-open) ✅ sony/gobreaker
Bulkhead (concurrency limit) ✅ DIY or slok/goresilience
Rate limiter (token bucket) ✅ golang.org/x/time/rate
Per-attempt timeout ✅ manual
Body buffering for safe POST retry ✅ DIY (often broken)
Retry-After header support ✅ most libs skip
Generic JSON helpers ✅ DIY
Request ID propagation ✅ DIY
Hooks (auth, logging, metrics) ✅ varies
Composable policies (Chain) ✅ manual
Zero dependencies ✅ n deps

Install

go get github.com/farhanturu/ambatukam-go

Requires Go 1.21+ (uses generics, slog, atomic.Int64).

Quick Start

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/farhanturu/ambatukam-go"
)

func main() {
    client := ambatukam.New(
        ambatukam.WithTimeout(ambatukam.TimeoutConfig{Timeout: 2 * time.Second}),
        ambatukam.WithRetry(ambatukam.RetryConfig{MaxRetries: 3}),
        ambatukam.WithCircuitBreaker(ambatukam.CircuitConfig{FailureThreshold: 5}),
        ambatukam.WithBulkhead(ambatukam.BulkheadConfig{MaxConcurrent: 10}),
        ambatukam.WithRateLimit(ambatukam.RateLimitConfig{Rate: 10, Burst: 5}),
    )
    defer client.Close()

    resp, err := client.Get(context.Background(), "https://api.example.com/users")
    if err != nil {
        log.Fatal(err)
    }
    defer resp.Body.Close()
    fmt.Println("status:", resp.StatusCode)
}

Architecture

┌──────────────────────────────────────────────────────────┐
│                      Client.Do()                         │
│                                                          │
│   ┌────────┐  ┌──────────┐  ┌────────┐  ┌──────────┐   │
│   │ Retry  │→│ Circuit   │→│Bulkhead│→│Rate Limit│   │
│   │        │  │ Breaker   │  │        │  │          │   │
│   └────────┘  └──────────┘  └────────┘  └──────────┘   │
│        ↓           ↓             ↓            ↓         │
│   ┌──────────────────────────────────────────────────┐  │
│   │     Timeout · Request ID · Hooks                 │  │
│   └──────────────────────────────────────────────────┘  │
│                         ↓                                │
│                   http.Client.Do()                       │
└──────────────────────────────────────────────────────────┘

Policies are composable middleware — outer-to-inner order: retry → circuit → timeout → HTTP.

Features

🔄 Retry with Backoff

ambatukam.WithRetry(ambatukam.RetryConfig{
    MaxRetries:     3,
    InitialBackoff: 100 * time.Millisecond,
    MaxBackoff:     5 * time.Second,
    Multiplier:     2.0,
    Jitter:         0.2,
})

Three strategies: ExponentialBackoff, ConstantBackoff, LinearBackoff.

Body buffering is automatic — POST bodies are read once and replayed on each retry. Only idempotent methods (GET, HEAD, PUT, DELETE, OPTIONS, TRACE) retry by default; opt in for POST with a custom ShouldRetry.

⚡ Circuit Breaker

ambatukam.WithCircuitBreaker(ambatukam.CircuitConfig{
    FailureThreshold: 5,
    OpenDuration:     30 * time.Second,
    HalfOpenMaxReqs:  1,
})

Three-state machine: closed → open → half-open. Race-safe under concurrent load with atomic generation counters.

🚧 Bulkhead (Concurrency Limit)

ambatukam.WithBulkhead(ambatukam.BulkheadConfig{
    MaxConcurrent: 10,
    MaxQueue:      100,
    QueueTimeout:  50 * time.Millisecond,
})

Limits in-flight requests to downstream. Optional bounded queue with timeout.

🚦 Rate Limiter

ambatukam.WithRateLimit(ambatukam.RateLimitConfig{
    Rate:        10,                     // tokens per second
    Burst:       5,                      // bucket capacity
    WaitTimeout: 100 * time.Millisecond, // 0 = fail fast
})

Token bucket. Rate <= 0 denies all requests (fail-closed).

⏱️ Timeout

ambatukam.WithTimeout(ambatukam.TimeoutConfig{Timeout: 2 * time.Second})

Per-attempt deadline. Parent ctx cancellation takes precedence.

🏷️ Request ID Propagation

ambatukam.WithRequestID("X-Request-ID") // empty = default header

Auto-generates a 12-byte hex ID per request, or propagates an existing one.

🪝 Hooks

ambatukam.WithHooks(ambatukam.Hooks{
    BeforeRequest: func(req *http.Request) error {
        req.Header.Set("Authorization", "Bearer "+token)
        return nil
    },
    OnRetry: func(req *http.Request, attempt int, nextDelay time.Duration) {
        log.Printf("retrying %s (attempt %d, delay %v)", req.URL, attempt, nextDelay)
    },
    OnStateChange: func(name string, from, to ambatukam.State) {
        metrics.Gauge("circuit_state").Set(string(to))
    },
})

Four callbacks: BeforeRequest, AfterResponse, OnRetry, OnStateChange. All optional.

📦 Generic JSON Helpers

type User struct {
    Name string `json:"name"`
    Age  int    `json:"age"`
}

u, err := ambatukam.GetJSON[User](client, ctx, "https://api.example.com/users/1")

created, err := ambatukam.PostJSON[User](client, ctx, "https://api.example.com/users", User{Name: "bob"})

Auto-handles JSON encode/decode, content-type, and 4xx/5xx errors as RequestError.

Preset Configs

Ready-to-use configurations for common scenarios:

// Balanced production defaults
client := ambatukam.New(ambatukam.ProductionConfig()...)

// Strict, fast-fail for fragile downstreams
client := ambatukam.New(ambatukam.AggressiveConfig()...)

// Generous config for critical services
client := ambatukam.New(ambatukam.ConservativeConfig()...)
Preset Retries Timeout Circuit Threshold Bulkhead
Production 3 30s 5 failures NumCPU×4
Aggressive 1 5s 3 failures NumCPU×2
Conservative 5 60s 20 failures NumCPU×8

Patterns

Stripe / Payment Gateway

client := ambatukam.New(
    ambatukam.WithTimeout(ambatukam.TimeoutConfig{Timeout: 10 * time.Second}),
    ambatukam.WithRetry(ambatukam.RetryConfig{
        MaxRetries: 3,
        Backoff:    ambatukam.ConstantBackoff(500 * time.Millisecond),
    }),
    ambatukam.WithCircuitBreaker(ambatukam.CircuitConfig{FailureThreshold: 5}),
)

Microservice with Auth + Tracing

client := ambatukam.New(
    ambatukam.WithRequestID("X-Request-ID"),
    ambatukam.WithHooks(ambatukam.Hooks{
        BeforeRequest: func(req *http.Request) error {
            req.Header.Set("Authorization", "Bearer "+getToken())
            return nil
        },
    }),
    ambatukam.WithRetry(ambatukam.DefaultRetryConfig()),
)

Third-Party API with Rate Limit

client := ambatukam.New(
    ambatukam.WithRateLimit(ambatukam.RateLimitConfig{
        Rate:        5,
        Burst:       10,
        WaitTimeout: 2 * time.Second,
    }),
    ambatukam.WithTimeout(ambatukam.TimeoutConfig{Timeout: 30 * time.Second}),
)

Custom Composition Order

client := ambatukam.New(ambatukam.WithPolicy(ambatukam.Chain(
    ambatukam.NewRetry(ambatukam.DefaultRetryConfig()),
    ambatukam.NewCircuitBreaker(ambatukam.DefaultCircuitConfig()),
    ambatukam.NewTimeout(ambatukam.TimeoutConfig{Timeout: 5 * time.Second}),
)))

Order matters: outer-to-inner is [retry [circuit [timeout [HTTP]]]].

Error Handling

Use errors.Is to distinguish error types:

resp, err := client.Get(ctx, url)
switch {
case errors.Is(err, ambatukam.ErrCircuitOpen):    // downstream is down
case errors.Is(err, ambatukam.ErrMaxRetries):     // gave up after N attempts
case errors.Is(err, ambatukam.ErrTimeout):        // attempt hit its deadline
case errors.Is(err, ambatukam.ErrBulkheadFull):   // at concurrency cap
case errors.Is(err, ambatukam.ErrRateLimited):    // rate-limited
case errors.Is(err, ambatukam.ErrNilRequest):     // programming error
case errors.Is(err, context.Canceled):            // ctx was canceled
}

For full context (method, URL, status, attempts):

var reqErr *ambatukam.RequestError
if errors.As(err, &reqErr) {
    log.Printf("%s %s returned %d after %d attempts",
        reqErr.Method, reqErr.URL, reqErr.Status, reqErr.Attempts)
}

Mark errors as non-retryable:

resp, err := client.Get(ctx, url)
if err != nil {
    return ambatukam.Permanent(err) // skip retry
}

Benchmarks

Measured on Intel Core i5-8250U @ 1.60GHz, Linux, Go 1.21 (go test -bench=. -benchmem -benchtime=2s).

Setup ns/op B/op allocs/op
http.Client (raw stdlib) 98,785 5,106 63
Ambatukam Go (no policies) 98,325 4,466 57
Ambatukam Go (retry=3) 96,879 4,505 58
Ambatukam Go (full stack) 235,341 16,757 120
Ambatukam Go (parallel) 26,064 8,214 77

Run locally: go test -bench=. -benchmem -benchtime=2s ./...

Migration

From cenkalti/backoff + sony/gobreaker 
// Before: two libraries, manual wiring
import (
    "github.com/cenkalti/backoff/v4"
    "github.com/sony/gobreaker"
)

// After: one library, one config
import "github.com/farhanturu/ambatukam-go"

client := ambatukam.New(
    ambatukam.WithRetry(ambatukam.RetryConfig{MaxRetries: 3}),
    ambatukam.WithCircuitBreaker(ambatukam.CircuitConfig{FailureThreshold: 5}),
)
From hashicorp/go-retryablehttp

Ambatukam Go's *Client is a drop-in *http.Client. Wrap your existing transport via WithHTTPClient, or use Do/Get/Post directly. Adds circuit breaker, bulkhead, rate limit, hooks, and request ID.

From slok/goresilience

Both use a runner/middleware pattern. See MIGRATION.md for a detailed guide.

Documentation

Document Description
README.md You are here
COOKBOOK.md Recipes for common patterns
FAQ.md Frequently asked questions
MIGRATION.md Migrating from other libraries
CONTRIBUTING.md How to contribute
SECURITY.md Security policy

Troubleshooting

Problem Solution
POST isn't being retried Only idempotent methods retry by default. Use custom ShouldRetry or idempotency-key header.
Circuit opens too often Lower FailureThreshold or increase OpenDuration. Use OnStateChange hook to monitor.
Bulkhead denies immediately Increase MaxQueue or MaxConcurrent. See COOKBOOK.
Rate limit denies unexpectedly Rate == 0 = disabled, Rate < 0 = deny all. Verify your config value.
Need debug logging Use ambatukam.WithDebug() for verbose logging.

Roadmap

v1.0 (current)

Retry, circuit breaker, bulkhead, rate limiter, timeout, request ID, hooks, generic JSON helpers, permanent errors, preset configs.

v1.1 (next)

Hedged requests (parallel speculative retries), fallback strategy (return stale data on failure), request deduplication (singleflight).

v2.0

OpenTelemetry tracing + Prometheus metrics, adaptive timeout (based on p99 latency), distributed (Redis-backed) circuit breaker, gRPC support.

Contributing

PRs welcome. Run go test ./... and go vet ./... before submitting; add a test for any new behavior. See CONTRIBUTING.md.

License

MIT — see LICENSE.

Ambatukam Go
Built with ❤️ by farhanturu

Documentation

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	ErrCircuitOpen  = errors.New("ambatukam: circuit breaker is open")
	ErrMaxRetries   = errors.New("ambatukam: max retries exceeded")
	ErrNilRequest   = errors.New("ambatukam: nil request")
	ErrTimeout      = errors.New("ambatukam: per-attempt timeout exceeded")
	ErrBulkheadFull = errors.New("ambatukam: bulkhead full")
	ErrRateLimited  = errors.New("ambatukam: rate limited")
)

Functions

func GetJSON 

func GetJSON[T any](c *Client, ctx context.Context, url string) (T, error)

GetJSON performs a GET request and decodes the response body as JSON into T.

If the response status is 4xx or 5xx, returns an error wrapping the status code. If decoding fails, returns an error wrapping the underlying decode error.

All configured policies (retry, circuit, bulkhead, rate limit, timeout) apply normally to this request.

The returned *RequestError (on non-2xx) always has Attempts=1 since this is the final response after any retries configured on the Client.

Example

ExampleGetJSON demonstrates the typed JSON helper. 

srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	fmt.Fprint(w, `{"name":"alice","age":30}`)
}))
defer srv.Close()

client := ambatukam.New()
defer client.Close()

u, err := ambatukam.GetJSON[testUser](client, context.Background(), srv.URL)
if err != nil {
	fmt.Println("error:", err)
	return
}
fmt.Println(u.Name, u.Age)

Output:
alice 30

func Permanent 

func Permanent(err error) error

Permanent wraps err to mark it as non-retryable.

Example

ExamplePermanent demonstrates marking an error as non-retryable. 

package main

import (
	"errors"
	"fmt"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	err := ambatukam.Permanent(errors.New("do not retry me"))
	fmt.Println(err)
}

Output:
permanent: do not retry me

func PostJSON 

func PostJSON[T any](c *Client, ctx context.Context, url string, body any) (T, error)

PostJSON marshals body as JSON, performs a POST with content-type "application/json", and decodes the response body as JSON into T.

If the response status is 4xx or 5xx, returns an error wrapping the status code. If decoding fails, returns an error wrapping the underlying decode error.

All configured policies apply normally.

The returned *RequestError (on non-2xx) always has Attempts=1 since this is the final response after any retries configured on the Client.

Example

ExamplePostJSON demonstrates the typed JSON POST helper. 

srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(201)
	fmt.Fprint(w, `{"name":"bob","age":25}`)
}))
defer srv.Close()

client := ambatukam.New()
defer client.Close()

out, err := ambatukam.PostJSON[testUser](client, context.Background(), srv.URL, testUser{Name: "alice", Age: 30})
if err != nil {
	fmt.Println("error:", err)
	return
}
fmt.Println(out.Name, out.Age)

Output:
bob 25

Types

type Backoff 

type Backoff interface {
	NextDelay(attempt int) time.Duration
}

Backoff produces the delay to wait before the next retry attempt. attempt is 0-indexed: attempt=0 is the delay before the first retry (i.e., after the initial request failed).

func ConstantBackoff 

func ConstantBackoff(d time.Duration) Backoff

ConstantBackoff returns a Backoff that always returns d.

func ExponentialBackoff 

func ExponentialBackoff(initial, max time.Duration, multiplier float64) Backoff

ExponentialBackoff returns a Backoff that grows exponentially from initial up to max, with symmetric jitter of ±20% applied per delay.

func LinearBackoff 

func LinearBackoff(initial, max, step time.Duration) Backoff

LinearBackoff returns a Backoff that grows linearly: initial + step*attempt, capped at max.

type BulkheadConfig 

type BulkheadConfig struct {
	MaxConcurrent uint32

	// MaxQueue is the number of requests that can wait when MaxConcurrent is reached.
	// Set to 0 to disable queueing (fail-fast on capacity).
	MaxQueue uint32

	// QueueTimeout is how long a queued request waits for a slot before
	// returning ErrBulkheadFull.
	//
	// Special behavior: if MaxQueue > 0 and QueueTimeout == 0, requests wait
	// up to 1 second as a safety net (avoiding unbounded waits). To get
	// fail-fast behavior, set MaxQueue = 0 instead.
	QueueTimeout time.Duration
}

type BulkheadPolicy 

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

BulkheadPolicy caps the number of in-flight requests reaching the next policy (typically the HTTP transport / downstream). Additional requests either wait on a bounded queue (up to cfg.MaxQueue) or fail fast with ErrBulkheadFull when the queue is full or queueing is disabled.

Implementation notes:

  • The semaphore `sem` is a buffered channel of struct{} with capacity cfg.MaxConcurrent. Sending into it is non-blocking when a slot is free; it would block once the bulkhead is full.
  • The `queue` channel is used as a non-blocking admission gate: the Execute path briefly inserts and removes a marker to test whether there is queue capacity left, then waits on the semaphore for up to cfg.QueueTimeout (or 1s as a safety net when timeout is 0).
  • inFlight and denied counters are atomic for observability via InFlight() / Denied().

func NewBulkhead 

func NewBulkhead(cfg BulkheadConfig) *BulkheadPolicy

NewBulkhead constructs a BulkheadPolicy. Zero-valued cfg.MaxConcurrent defaults to runtime.NumCPU()*2 (floored at 1) so the policy is always usable out of the box.

Example

ExampleNewBulkhead demonstrates configuring a bulkhead. 

package main

import (
	"fmt"
	"time"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	cfg := ambatukam.BulkheadConfig{
		MaxConcurrent: 5,
		MaxQueue:      10,
		QueueTimeout:  100 * time.Millisecond,
	}

	client := ambatukam.New(ambatukam.WithBulkhead(cfg))
	defer client.Close()

	fmt.Println("max concurrent:", cfg.MaxConcurrent)
}

Output:
max concurrent: 5

func (*BulkheadPolicy) Denied 

func (b *BulkheadPolicy) Denied() uint64

Denied returns the total count of denied requests since startup (observability).

func (*BulkheadPolicy) Execute 

func (b *BulkheadPolicy) Execute(ctx context.Context, req *http.Request, next PolicyFunc) (*http.Response, error)

Execute runs a request through the bulkhead. Admitted requests call next. Denied requests return (nil, ErrBulkheadFull) (wrapped with "queue timeout" when the wait timed out).

func (*BulkheadPolicy) InFlight 

func (b *BulkheadPolicy) InFlight() uint32

InFlight returns the current count of in-flight requests (observability).

func (*BulkheadPolicy) WithLogger 

func (b *BulkheadPolicy) WithLogger(l *slog.Logger) *BulkheadPolicy

WithLogger sets a non-nil logger on the policy.

type CircuitBreakerPolicy 

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

CircuitBreakerPolicy is a circuit breaker resilience policy.

State machine:

  • Closed: consecutive failures are counted via cb.failures. When the counter reaches cfg.FailureThreshold, the breaker transitions to Open and records the time the open window began.
  • Open: every request fails fast with ErrCircuitOpen. When the open window has elapsed (cfg.OpenDuration), the next request triggers the Open->HalfOpen transition and is admitted as a trial.
  • HalfOpen: up to cfg.HalfOpenMaxReqs trial requests are admitted. A successful trial closes the breaker. A failed trial re-opens the breaker.

Concurrency:

  • cb.mu protects state, openedAt, halfOpenPermits, halfOpenInFlight. It is never held across next(ctx, req).
  • cb.failures is read atomically (fast Closed hot path) and reset to 0 atomically on every success in Closed state and on the HalfOpen->Closed transition.
  • cb.generation is a monotonic counter incremented on every state transition. A request entering HalfOpen records the generation; on response the result is only applied if the breaker is still on the same generation. Otherwise the breaker has moved on and the trial is discarded (no stale writes).

func NewCircuitBreaker 

func NewCircuitBreaker(cfg CircuitConfig) *CircuitBreakerPolicy

NewCircuitBreaker constructs a CircuitBreakerPolicy. Zero-valued config fields are filled with the package defaults via DefaultCircuitConfig: FailureThreshold=5, OpenDuration=30s, HalfOpenMaxReqs=1.

Example

ExampleNewCircuitBreaker demonstrates configuring the circuit breaker. 

package main

import (
	"fmt"
	"time"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	cfg := ambatukam.DefaultCircuitConfig()
	cfg.FailureThreshold = 3
	cfg.OpenDuration = 10 * time.Second

	client := ambatukam.New(ambatukam.WithCircuitBreaker(cfg))
	defer client.Close()

	fmt.Println("failure threshold:", cfg.FailureThreshold)
}

Output:
failure threshold: 3

func (*CircuitBreakerPolicy) Execute 

func (cb *CircuitBreakerPolicy) Execute(ctx context.Context, req *http.Request, next PolicyFunc) (*http.Response, error)

Execute runs a single request through the breaker. A request denied by the breaker returns (nil, ErrCircuitOpen) without invoking next. Admitted requests call next and the result is routed through shouldTrip/onSuccess/ onFailure.

func (*CircuitBreakerPolicy) State 

func (cb *CircuitBreakerPolicy) State() State

State returns the current breaker state. Safe to call concurrently.

func (*CircuitBreakerPolicy) WithHooks 

func (cb *CircuitBreakerPolicy) WithHooks(h Hooks) *CircuitBreakerPolicy

WithHooks installs user callbacks fired on circuit state changes. Only OnStateChange is used by this policy.

func (*CircuitBreakerPolicy) WithLogger 

func (cb *CircuitBreakerPolicy) WithLogger(l *slog.Logger) *CircuitBreakerPolicy

WithLogger sets a non-nil logger on the policy.

func (*CircuitBreakerPolicy) WithName 

func (cb *CircuitBreakerPolicy) WithName(name string) *CircuitBreakerPolicy

WithName sets the breaker identifier reported in OnStateChange callbacks. An empty name leaves the existing name unchanged (default: "default").

type CircuitConfig 

type CircuitConfig struct {
	FailureThreshold uint32
	OpenDuration     time.Duration
	HalfOpenMaxReqs  uint32
	ShouldTrip       func(resp *http.Response, err error) bool
}

func DefaultCircuitConfig 

func DefaultCircuitConfig() CircuitConfig

DefaultCircuitConfig returns the circuit breaker configuration used when none is provided. Callers can extend these defaults programmatically.

type Client 

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

func New 

func New(opts ...Option) *Client

func NewDefaultClient 

func NewDefaultClient() *Client

NewDefaultClient is a convenience constructor: New(DefaultConfig()...).

Example

ExampleNewDefaultClient demonstrates the one-call production-default client. 

package main

import (
	"fmt"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	client := ambatukam.NewDefaultClient()
	defer client.Close()
	fmt.Println("ok")
}

Output:
ok

func (*Client) Close 

func (c *Client) Close() error

func (*Client) Do 

func (c *Client) Do(req *http.Request) (*http.Response, error)

func (*Client) DoWithContext 

func (c *Client) DoWithContext(ctx context.Context, req *http.Request) (*http.Response, error)

DoWithContext is like Do but lets the caller supply the request context. It is equivalent to Do(req.WithContext(ctx)).

func (*Client) Get 

func (c *Client) Get(ctx context.Context, url string) (*http.Response, error)
Example

ExampleClient_Get demonstrates a basic GET request with all three policies. 

package main

import (
	"context"
	"fmt"
	"net/http"
	"net/http/httptest"
	"time"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		fmt.Fprint(w, `{"hello":"world"}`)
	}))
	defer srv.Close()

	client := ambatukam.New(
		ambatukam.WithTimeout(ambatukam.TimeoutConfig{Timeout: 2 * time.Second}),
		ambatukam.WithRetry(ambatukam.DefaultRetryConfig()),
		ambatukam.WithCircuitBreaker(ambatukam.DefaultCircuitConfig()),
	)
	defer client.Close()

	resp, err := client.Get(context.Background(), srv.URL)
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	defer resp.Body.Close()
	fmt.Println("status:", resp.StatusCode)
}

Output:
status: 200

func (*Client) Post 

func (c *Client) Post(ctx context.Context, url, contentType string, body io.Reader) (*http.Response, error)
Example

ExampleClient_Post demonstrates a POST with a JSON body. 

package main

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, _ := io.ReadAll(r.Body)
		fmt.Println("server received:", string(body))
		w.WriteHeader(http.StatusCreated)
	}))
	defer srv.Close()

	client := ambatukam.New()
	defer client.Close()

	resp, err := client.Post(
		context.Background(),
		srv.URL+"/users",
		"application/json",
		io.NopCloser(strings.NewReader(`{"name":"alice"}`)),
	)
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	defer resp.Body.Close()
	fmt.Println("status:", resp.StatusCode)
}

Output:
server received: {"name":"alice"}
status: 201

func (*Client) RoundTrip 

func (c *Client) RoundTrip(req *http.Request) (*http.Response, error)

RoundTrip implements http.RoundTripper. It enables using an amba *Client as the Transport of any *http.Client, including third-party libraries that accept only http.RoundTripper.

Equivalent to c.Do(req).

Example

ExampleClient_RoundTrip shows how to use an amba *Client as the Transport of a standard *http.Client. 

package main

import (
	"fmt"
	"net/http"
	"net/http/httptest"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(200)
	}))
	defer srv.Close()

	amba := ambatukam.New(ambatukam.WithRetry(ambatukam.RetryConfig{MaxRetries: 2}))
	defer amba.Close()

	hc := &http.Client{Transport: amba.Transport()}
	resp, err := hc.Get(srv.URL)
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	defer resp.Body.Close()
	fmt.Println("status:", resp.StatusCode)
}

Output:
status: 200

func (*Client) Transport 

func (c *Client) Transport() http.RoundTripper

Transport returns an http.RoundTripper that delegates to this Client. Convenient for `&http.Client{Transport: ambaClient.Transport()}`.

type Hooks 

type Hooks struct {
	// BeforeRequest is called immediately before each HTTP attempt.
	// It can mutate the request (e.g. add Authorization header) or return
	// an error to abort the attempt. Returning an error short-circuits
	// the retry loop with that error.
	BeforeRequest func(req *http.Request) error

	// AfterResponse is called after each HTTP attempt completes (success or failure).
	// It is purely observational; its return value is ignored. Use it for logging,
	// metrics emission, or response inspection.
	AfterResponse func(req *http.Request, resp *http.Response, err error)

	// OnRetry is called between retry attempts, before the backoff sleep.
	// `attempt` is the 0-indexed attempt number that just completed (the next
	// attempt will be `attempt+1`). `nextDelay` is the planned sleep duration.
	OnRetry func(req *http.Request, attempt int, nextDelay time.Duration)

	// OnStateChange is called when a circuit breaker changes state.
	// `name` is the breaker's identifier (defaults to "default" if not set).
	// `from` and `to` are the old and new states (e.g. StateClosed → StateOpen).
	OnStateChange func(name string, from, to State)
}

Hooks are user callbacks invoked at key lifecycle points. All fields are optional. Any nil callback is skipped.

type Option 

type Option func(*Client)

func AggressiveConfig 

func AggressiveConfig() []Option

AggressiveConfig returns a strict, fast-fail configuration for protecting fragile downstream services. Lower retry count, lower circuit threshold, short timeouts.

func ConservativeConfig 

func ConservativeConfig() []Option

ConservativeConfig returns a generous configuration for critical services that must not fail. Many retries, slow trip, larger timeouts.

func DefaultConfig 

func DefaultConfig() []Option

DefaultConfig returns a sensible production-default *Client configuration. Use with New() to get a fully-configured Client: 

client := ambatukam.New(ambatukam.DefaultConfig()...)

Defaults:

  • Retry: 3 attempts, exponential 100ms..5s, jitter 0.2
  • Circuit: 5 failures to trip, 30s open duration, 1 half-open probe
  • Timeout: 30s per attempt
  • Bulkhead: runtime.NumCPU()*4 concurrent, no queue (fail fast)
  • RateLimit: disabled (use WithRateLimit to enable)
  • No request ID (use WithRequestID to enable)
  • No hooks (use WithHooks to enable)

func ProductionConfig 

func ProductionConfig() []Option

ProductionConfig returns a balanced default configuration suitable for most production services. Use with New(): 

client := ambatukam.New(ambatukam.ProductionConfig()...)

Tuning: 3 retries, 30s timeout, 5-failure circuit, NumCPU*4 concurrent.

func WithBulkhead 

func WithBulkhead(cfg BulkheadConfig) Option

func WithCircuitBreaker 

func WithCircuitBreaker(cfg CircuitConfig) Option

func WithDebug 

func WithDebug() Option

WithDebug enables verbose DEBUG-level logging via slog to stderr. All policy decisions (retries, circuit transitions, rate-limit waits) are logged.

Equivalent to WithLogger(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug}))).

func WithHTTPClient 

func WithHTTPClient(hc *http.Client) Option
Example

ExampleWithHTTPClient demonstrates swapping in a custom *http.Client. 

package main

import (
	"fmt"
	"net/http"
	"time"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	client := ambatukam.New(
		ambatukam.WithHTTPClient(&http.Client{Timeout: 5 * time.Second}),
	)
	defer client.Close()
	fmt.Println("custom http.Client")
}

Output:
custom http.Client

func WithHooks 

func WithHooks(h Hooks) Option

WithHooks installs user callbacks invoked at key request lifecycle points (before each attempt, after each response, between retries, and on circuit breaker state changes). Any nil field in Hooks is skipped.

Hooks are propagated to registered retry and circuit breaker policies inside New(), so declaration order does not matter — WithHooks can be passed before or after WithRetry/WithCircuitBreaker with the same effect.

func WithLogger 

func WithLogger(l *slog.Logger) Option

func WithPolicy 

func WithPolicy(p Policy) Option

func WithRateLimit 

func WithRateLimit(cfg RateLimitConfig) Option

func WithRequestID 

func WithRequestID(header string) Option

WithRequestID enables automatic X-Request-ID generation/propagation. Pass an empty string to use the default header name.

func WithRequestIDPolicy 

func WithRequestIDPolicy(p *RequestIDPolicy) Option

WithRequestIDPolicy registers a custom-configured RequestIDPolicy. Use this when you need to set a custom header name or generator (see RequestIDPolicy.WithHeader, RequestIDPolicy.WithGenerator).

func WithRetry 

func WithRetry(cfg RetryConfig) Option

func WithTimeout 

func WithTimeout(cfg TimeoutConfig) Option

type PermanentError 

type PermanentError struct{ Err error }

PermanentError signals that a retry must NOT be attempted for this error. Wrap any error with Permanent() to mark it as non-retryable; the retry policy will propagate it immediately without further attempts.

func (*PermanentError) Error 

func (e *PermanentError) Error() string

func (*PermanentError) Unwrap 

func (e *PermanentError) Unwrap() error

type Policy 

type Policy interface {
	Execute(ctx context.Context, req *http.Request, next PolicyFunc) (*http.Response, error)
}

func Chain 

func Chain(policies ...Policy) Policy

Chain composes policies so the FIRST argument is OUTERMOST. Execution order: P0.Execute -> P1.Execute -> ... -> Pn.Execute -> next. Each policy sees the request before passing to next, and sees the result/error after.

The returned Policy delegates Execute by threading the supplied `next` into the innermost policy, so when used via WithPolicy(Chain(...)) the Client.Do chain builder's terminal (c.hc.Do) reaches the last policy in the chain.

Example

ExampleChain demonstrates manual composition of policies. 

package main

import (
	"fmt"
	"time"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	client := ambatukam.New(ambatukam.WithPolicy(ambatukam.Chain(
		ambatukam.NewTimeout(ambatukam.TimeoutConfig{Timeout: 2 * time.Second}),
		ambatukam.NewRetry(ambatukam.DefaultRetryConfig()),
		ambatukam.NewCircuitBreaker(ambatukam.DefaultCircuitConfig()),
	)))
	defer client.Close()

	fmt.Println("chained: ok")
}

Output:
chained: ok

type PolicyFunc 

type PolicyFunc func(ctx context.Context, req *http.Request) (*http.Response, error)

func (PolicyFunc) Execute 

func (f PolicyFunc) Execute(ctx context.Context, req *http.Request, next PolicyFunc) (*http.Response, error)

Execute makes PolicyFunc satisfy the Policy interface so a bare function can be used wherever a Policy is expected (e.g. WithPolicy or as a return value from Chain). The function is invoked directly; `next` is intentionally ignored — a PolicyFunc is treated as a terminal behavior.

type RateLimitConfig 

type RateLimitConfig struct {
	// Rate is the tokens per second. Special values:
	//   Rate == 0 (default): rate limiting is disabled — all requests pass through.
	//   Rate < 0: deny all requests (fail-closed).
	//   Rate > 0: normal token-bucket behavior.
	Rate float64
	// Burst is the bucket capacity (default 1 if 0).
	Burst uint32
	// WaitTimeout is how long to wait for a token when none is available.
	// 0 = fail fast; >0 = wait up to this long.
	WaitTimeout time.Duration
}

RateLimitConfig configures a token-bucket rate limiter policy.

Rate is the steady-state token replenishment rate in tokens per second. A non-positive Rate denies every request immediately with ErrRateLimited.

Burst is the bucket capacity — the maximum number of tokens that can be consumed back-to-back without replenishment. Zero is replaced with 1 inside NewRateLimit.

WaitTimeout controls behaviour when no token is available. Zero means fail fast (the request returns ErrRateLimited immediately). A positive value means the policy waits up to that duration for a token to become available; if the wait would exceed WaitTimeout the request is denied.

Note: because the bucket is checked, then released, then re-checked between waits, this policy is best-effort under high concurrency. For strict admission control, consider a dedicated library. The race window is bounded by WaitTimeout: a goroutine that releases the mutex with a small remaining deficit and then sleeps WaitTimeout may find another goroutine has already consumed the freshly-refilled token, in which case it loops and tries again.

type RateLimitPolicy 

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

RateLimitPolicy is a token-bucket rate limiter.

State machine (per policy instance):

  • The bucket holds `tokens` (a float) and is refilled continuously at `cfg.Rate` tokens per second up to a cap of `cfg.Burst`.
  • On every Execute call the bucket is refilled based on elapsed wall time since `lastRefill`, then the policy either consumes one token and forwards the request, or — depending on `cfg.WaitTimeout` — either fails fast with ErrRateLimited or sleeps up to the time required for one token to be available (capped by WaitTimeout).

Concurrency:

  • `mu` serialises refill + token consumption. It is never held across `next(ctx, req)` — the mutex is released before the HTTP call runs, so the rate limiter does not throttle concurrency, only throughput.
  • There is an inherent best-effort race between releasing the mutex and the next refill: a goroutine that wakes after a short deficit can find another goroutine has already consumed the freshly available token. The waiter then loops and tries again. The race window is bounded by WaitTimeout. This is acceptable for typical HTTP rate limiting; for strict admission control use a dedicated library such as golang.org/x/time/rate.

func NewRateLimit 

func NewRateLimit(cfg RateLimitConfig) *RateLimitPolicy

NewRateLimit constructs a RateLimitPolicy.

Zero-valued fields are normalised:

  • cfg.Burst == 0 → cfg.Burst = 1
  • cfg.WaitTimeout < 0 → cfg.WaitTimeout = 0

cfg.Rate semantics:

  • cfg.Rate == 0 → disabled: every request passes through unchanged.
  • cfg.Rate < 0 → closedAll: every request returns ErrRateLimited.
  • cfg.Rate > 0 → normal token-bucket behaviour.
Example

ExampleNewRateLimit demonstrates building a rate-limit policy directly. 

package main

import (
	"fmt"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	client := ambatukam.New(ambatukam.WithRateLimit(ambatukam.RateLimitConfig{
		Rate:  10,
		Burst: 5,
	}))
	defer client.Close()
	fmt.Println("rate limited client ready")
}

Output:
rate limited client ready

func (*RateLimitPolicy) Execute 

func (r *RateLimitPolicy) Execute(ctx context.Context, req *http.Request, next PolicyFunc) (*http.Response, error)

Execute runs a request through the rate limiter.

Behaviour:

  • disabled (Rate == 0): the request is forwarded unchanged.
  • closedAll (Rate < 0): every request returns ErrRateLimited immediately.
  • Bucket has a token: the token is consumed and the request is forwarded.
  • Bucket is empty, WaitTimeout == 0: returns ErrRateLimited immediately.
  • Bucket is empty, WaitTimeout > 0: sleeps up to WaitTimeout (capped by the actual time-to-token), then retries. Honours ctx cancellation.

func (*RateLimitPolicy) WithLogger 

func (r *RateLimitPolicy) WithLogger(l *slog.Logger) *RateLimitPolicy

WithLogger sets a non-nil logger on the policy.

type RequestError 

type RequestError struct {
	Method   string
	URL      string
	Status   int
	Attempts int
	Err      error
}

RequestError captures full context about a failed request: method, URL, status code, attempt count, and the underlying error.

func (*RequestError) Error 

func (e *RequestError) Error() string

func (*RequestError) Unwrap 

func (e *RequestError) Unwrap() error

type RequestIDPolicy 

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

RequestIDPolicy adds or propagates a request ID header on every request. Useful for distributed tracing and log correlation across services.

func NewRequestIDPolicy 

func NewRequestIDPolicy() *RequestIDPolicy

NewRequestIDPolicy returns a RequestIDPolicy that uses the default "X-Request-ID" header and an internal hex-encoded random generator. Use WithHeader and WithGenerator to customise.

Example

ExampleNewRequestIDPolicy demonstrates registering a request-ID policy. 

package main

import (
	"fmt"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	client := ambatukam.New(ambatukam.WithRequestIDPolicy(ambatukam.NewRequestIDPolicy()))
	defer client.Close()
	fmt.Println("request ID policy enabled")
}

Output:
request ID policy enabled

func (*RequestIDPolicy) Execute 

func (r *RequestIDPolicy) Execute(ctx context.Context, req *http.Request, next PolicyFunc) (*http.Response, error)

Execute sets the configured request ID header on req when not already present, then invokes next.

func (*RequestIDPolicy) WithGenerator 

func (r *RequestIDPolicy) WithGenerator(gen func() string) *RequestIDPolicy

WithGenerator overrides the ID generator (e.g., for UUID v7). Nil generators are ignored.

func (*RequestIDPolicy) WithHeader 

func (r *RequestIDPolicy) WithHeader(name string) *RequestIDPolicy

WithHeader overrides the header name (default "X-Request-ID"). Empty names are ignored to keep a sane default.

type RetryConfig 

type RetryConfig struct {
	MaxRetries     int
	InitialBackoff time.Duration
	MaxBackoff     time.Duration
	Multiplier     float64
	Jitter         float64
	Backoff        Backoff
	ShouldRetry    func(resp *http.Response, err error) bool
}

func DefaultRetryConfig 

func DefaultRetryConfig() RetryConfig

DefaultRetryConfig returns the retry configuration used when none is provided. Callers can extend these defaults programmatically.

type RetryPolicy 

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

RetryPolicy retries failed requests with configurable backoff. The request body is buffered once before any attempt so each retry can re-read it safely (safe POST retry).

func NewRetry 

func NewRetry(cfg RetryConfig) *RetryPolicy

NewRetry constructs a RetryPolicy. Zero-valued cfg fields are filled with the package defaults via applyRetryDefaults. Defaults: MaxRetries floored at 0, backoff defaults to exponential 100ms..5s with multiplier 2, jitter clamped to [0,1].

Example

ExampleNewRetry demonstrates configuring the retry policy. 

package main

import (
	"fmt"
	"time"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	cfg := ambatukam.DefaultRetryConfig()
	cfg.MaxRetries = 5
	cfg.Backoff = ambatukam.ExponentialBackoff(50*time.Millisecond, time.Second, 2.0)

	client := ambatukam.New(ambatukam.WithRetry(cfg))
	defer client.Close()

	fmt.Println("max retries:", cfg.MaxRetries)
}

Output:
max retries: 5

func (*RetryPolicy) Execute 

func (r *RetryPolicy) Execute(ctx context.Context, req *http.Request, next PolicyFunc) (*http.Response, error)

Execute runs the request through next, retrying transient failures per config.

func (*RetryPolicy) WithHooks 

func (r *RetryPolicy) WithHooks(h Hooks) *RetryPolicy

WithHooks installs user callbacks fired during the retry lifecycle. Each non-nil callback is invoked at the appropriate point: BeforeRequest before each attempt (errors abort the loop), AfterResponse after each attempt, OnRetry between attempts before the backoff sleep.

func (*RetryPolicy) WithLogger 

func (r *RetryPolicy) WithLogger(l *slog.Logger) *RetryPolicy

WithLogger sets a non-nil logger on the policy.

type State 

type State string

State is the publicly-visible state of a circuit breaker. 

const (
	StateClosed   State = "closed"
	StateOpen     State = "open"
	StateHalfOpen State = "half-open"
)

type TimeoutConfig 

type TimeoutConfig struct {
	Timeout time.Duration
}

type TimeoutPolicy 

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

TimeoutPolicy applies a per-attempt deadline to the request passed to its `next` PolicyFunc. If the deadline fires before the downstream call returns, the error returned to the caller is wrapped with ErrTimeout so callers can distinguish a per-attempt timeout from other failures (e.g. parent context cancellation).

A non-positive timeout disables the policy entirely.

func NewTimeout 

func NewTimeout(cfg TimeoutConfig) *TimeoutPolicy

NewTimeout constructs a TimeoutPolicy from a TimeoutConfig.

Example

ExampleNewTimeout demonstrates configuring the per-attempt timeout. 

package main

import (
	"fmt"
	"time"

	"github.com/farhanturu/ambatukam-go"
)

func main() {
	client := ambatukam.New(
		ambatukam.WithTimeout(ambatukam.TimeoutConfig{Timeout: 500 * time.Millisecond}),
	)
	defer client.Close()

	fmt.Println("timeout: ok")
}

Output:
timeout: ok

func (*TimeoutPolicy) Execute 

func (t *TimeoutPolicy) Execute(ctx context.Context, req *http.Request, next PolicyFunc) (*http.Response, error)

Execute runs the downstream call under a child context with the configured timeout. If the timeout fires (child context's Err() is DeadlineExceeded) while the parent context is still healthy, the returned error is wrapped with ErrTimeout. If the parent context is already canceled (or its own deadline already fired), the underlying error is returned unchanged so the caller can distinguish between a per-attempt timeout and an outer cancel.

The child context is attached to the request via req.WithContext so the downstream HTTP transport observes the per-attempt deadline (the transport reads from req.Context, not from any ctx argument the caller passes).

Source Files

View all Source files

Directories

Path Synopsis
examples
basic command
Package main is a runnable example for ambatukam.
 Package main is a runnable example for ambatukam.

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.