http_client: refactoring

Author: batazor

http/client/internal/types/errors.go

@@ -0,0 +1,8 @@
+package types
+
+import "errors"
+
+var (
+	ErrInvalidLimiterConfig = errors.New("http_client: invalid limiter config")
+	ErrDeadlineTooClose     = errors.New("http_client: deadline too close")
+)

http/client/internal/types/types.go

@@ -0,0 +1,92 @@
+package types
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Middleware wraps a RoundTripper to add functionality.
+// It receives the next RoundTripper in the chain and returns a new RoundTripper
+// that may intercept, modify, or observe requests and responses.
+type Middleware func(next http.RoundTripper) http.RoundTripper
+
+// RoundTripperFunc allows using a function as a RoundTripper.
+type RoundTripperFunc func(*http.Request) (*http.Response, error)
+
+func (f RoundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) {
+	return f(r)
+}
+
+// Limiter defines the interface for rate limiting.
+// Wait blocks until a request can be made according to the rate limit,
+// or until the context is cancelled. It returns the total time spent waiting
+// and any error that occurred (typically context.Canceled or context.DeadlineExceeded).
+type Limiter interface {
+	Wait(ctx context.Context) (time.Duration, error)
+}
+
+const (
+	LabelClient = "client"
+	LabelHost   = "host"
+	LabelMethod = "method"
+	LabelSource = "source"
+)
+
+type Metrics struct {
+	RateLimitWaitSeconds   *prometheus.HistogramVec
+	RateLimit429Total      *prometheus.CounterVec
+	DeadlineCancelledTotal *prometheus.CounterVec
+}
+
+func NewMetrics(namespace, subsystem string) *Metrics {
+	return &Metrics{
+		RateLimitWaitSeconds: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{ //nolint:exhaustruct // Prometheus options have many optional fields
+				Namespace: namespace,
+				Subsystem: subsystem,
+				Name:      "rate_limit_wait_seconds",
+				Help:      "Time spent waiting due to rate limiting (server headers or local bucket).",
+				Buckets:   prometheus.DefBuckets,
+			},
+			[]string{LabelClient, LabelHost, LabelMethod, LabelSource},
+		),
+		RateLimit429Total: prometheus.NewCounterVec(
+			prometheus.CounterOpts{ //nolint:exhaustruct // Prometheus options have many optional fields
+				Namespace: namespace,
+				Subsystem: subsystem,
+				Name:      "rate_limit_429_total",
+				Help:      "Total number of HTTP 429 responses.",
+			},
+			[]string{LabelClient, LabelHost, LabelMethod},
+		),
+		DeadlineCancelledTotal: prometheus.NewCounterVec(
+			prometheus.CounterOpts{ //nolint:exhaustruct // Prometheus options have many optional fields
+				Namespace: namespace,
+				Subsystem: subsystem,
+				Name:      "deadline_canceled_total",
+				Help:      "Requests canceled early due to short deadline.",
+			},
+			[]string{LabelClient, LabelHost, LabelMethod},
+		),
+	}
+}
+
+func (m *Metrics) Register(reg prometheus.Registerer) error {
+	if err := reg.Register(m.RateLimitWaitSeconds); err != nil {
+		return fmt.Errorf("register wait_seconds: %w", err)
+	}
+
+	if err := reg.Register(m.RateLimit429Total); err != nil {
+		return fmt.Errorf("register 429: %w", err)
+	}
+
+	if err := reg.Register(m.DeadlineCancelledTotal); err != nil {
+		return fmt.Errorf("register deadline_canceled: %w", err)
+	}
+
+	return nil
+}

http/client/limiter_tokenbucket_test.go

@@ -0,0 +1,179 @@
+package http_client
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/shortlink-org/go-sdk/http/client/internal/types"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewTokenBucketLimiter_InvalidConfig(t *testing.T) {
+	_, err := NewTokenBucketLimiter(0, 10, 0)
+	require.Error(t, err)
+	require.Equal(t, types.ErrInvalidLimiterConfig, err)
+
+	_, err = NewTokenBucketLimiter(10, 0, 0)
+	require.Error(t, err)
+	require.Equal(t, types.ErrInvalidLimiterConfig, err)
+
+	_, err = NewTokenBucketLimiter(-1, 10, 0)
+	require.Error(t, err)
+	require.Equal(t, types.ErrInvalidLimiterConfig, err)
+
+	_, err = NewTokenBucketLimiter(10, -1, 0)
+	require.Error(t, err)
+	require.Equal(t, types.ErrInvalidLimiterConfig, err)
+}
+
+func TestNewTokenBucketLimiter_JitterNormalization(t *testing.T) {
+	limiter, err := NewTokenBucketLimiter(10, 10, -0.5)
+	require.NoError(t, err)
+	require.NotNil(t, limiter)
+
+	limiter2, err := NewTokenBucketLimiter(10, 10, 1.5)
+	require.NoError(t, err)
+	require.NotNil(t, limiter2)
+}
+
+func TestTokenBucketLimiter_Wait_Immediate(t *testing.T) {
+	limiter, err := NewTokenBucketLimiter(10, 10, 0)
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	wait, err := limiter.Wait(ctx)
+	require.NoError(t, err)
+	require.Equal(t, time.Duration(0), wait)
+}
+
+func TestTokenBucketLimiter_Wait_ContextCancellation(t *testing.T) {
+	limiter, err := NewTokenBucketLimiter(0.1, 1, 0) // very slow rate
+	require.NoError(t, err)
+
+	// Consume the token
+	_, err = limiter.Wait(context.Background())
+	require.NoError(t, err)
+
+	// Now we'll have to wait, but cancel the context
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // cancel immediately
+
+	wait, err := limiter.Wait(ctx)
+	require.Error(t, err)
+	require.Equal(t, context.Canceled, err)
+	require.GreaterOrEqual(t, wait, time.Duration(0))
+}
+
+func TestTokenBucketLimiter_Wait_ContextTimeout(t *testing.T) {
+	limiter, err := NewTokenBucketLimiter(0.1, 1, 0) // very slow rate
+	require.NoError(t, err)
+
+	// Consume the token
+	_, err = limiter.Wait(context.Background())
+	require.NoError(t, err)
+
+	// Now we'll have to wait, but timeout the context
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+	defer cancel()
+
+	wait, err := limiter.Wait(ctx)
+	require.Error(t, err)
+	require.Equal(t, context.DeadlineExceeded, err)
+	require.GreaterOrEqual(t, wait, time.Duration(0))
+}
+
+func TestTokenBucketLimiter_Refill(t *testing.T) {
+	limiter, err := NewTokenBucketLimiter(10, 5, 0) // 10 tokens/sec, burst 5
+	require.NoError(t, err)
+
+	// Consume all tokens
+	for i := 0; i < 5; i++ {
+		_, err = limiter.Wait(context.Background())
+		require.NoError(t, err)
+	}
+
+	// Wait for refill
+	time.Sleep(150 * time.Millisecond) // should refill ~1.5 tokens
+
+	// Should be able to get at least one token
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	wait, err := limiter.Wait(ctx)
+	require.NoError(t, err)
+	require.Less(t, wait, 100*time.Millisecond)
+}
+
+func TestTokenBucketLimiter_Jitter(t *testing.T) {
+	limiter, err := NewTokenBucketLimiter(1, 1, 0.1) // 10% jitter
+	require.NoError(t, err)
+
+	// Consume the token
+	_, err = limiter.Wait(context.Background())
+	require.NoError(t, err)
+
+	// Wait should have some jitter applied
+	// Base wait should be ~1 second, jitter can add up to 10% = 100ms
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	wait1, err1 := limiter.Wait(ctx)
+	require.NoError(t, err1)
+
+	// Wait time should be between base (1s) and base + jitter (1.1s)
+	// but could be slightly more due to timing
+	baseWait := 1 * time.Second
+	maxWait := 1200 * time.Millisecond              // 1.2 seconds
+	require.GreaterOrEqual(t, wait1, baseWait*9/10) // at least 90% of base (allowing negative jitter)
+	require.LessOrEqual(t, wait1, maxWait)          // at most 20% more than base
+}
+
+func TestTokenBucketLimiter_BurstLimit(t *testing.T) {
+	limiter, err := NewTokenBucketLimiter(10, 3, 0) // 10 tokens/sec, burst 3
+	require.NoError(t, err)
+
+	// Consume burst
+	for i := 0; i < 3; i++ {
+		_, err = limiter.Wait(context.Background())
+		require.NoError(t, err)
+	}
+
+	// Next wait should take time even though rate is high
+	// With 10 tokens/sec, we need 0.1s (100ms) to refill 1 token
+	ctx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond)
+	defer cancel()
+
+	wait, err := limiter.Wait(ctx)
+	// Should timeout because we need to wait for refill (~100ms) but timeout is 80ms
+	require.Error(t, err)
+	require.Equal(t, context.DeadlineExceeded, err)
+	// Wait time accumulates during the actual wait and may be close to timeout
+	// It should be less than the required refill time (~100ms)
+	require.Less(t, wait, 110*time.Millisecond)   // should be less than refill time
+	require.Greater(t, wait, 50*time.Millisecond) // but more than half timeout
+}
+
+func TestTokenBucketLimiter_ConcurrentAccess(t *testing.T) {
+	limiter, err := NewTokenBucketLimiter(100, 10, 0)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancel()
+
+	done := make(chan error, 10)
+
+	// Launch 10 goroutines
+	for i := 0; i < 10; i++ {
+		go func() {
+			_, err := limiter.Wait(ctx)
+			done <- err
+		}()
+	}
+
+	// All should succeed
+	for i := 0; i < 10; i++ {
+		err := <-done
+		require.NoError(t, err)
+	}
+}

http/client/middleware/deadline/deadline.go

@@ -0,0 +1,41 @@
+package deadline
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/shortlink-org/go-sdk/http/client/internal/types"
+)
+
+type Config struct {
+	Threshold time.Duration
+	Metrics   *types.Metrics
+	Client    string
+}
+
+func Middleware(cfg Config) types.Middleware {
+	if cfg.Threshold <= 0 {
+		return func(next http.RoundTripper) http.RoundTripper { return next }
+	}
+
+	return func(next http.RoundTripper) http.RoundTripper {
+		return types.RoundTripperFunc(func(req *http.Request) (*http.Response, error) {
+			ctx := req.Context()
+
+			dl, ok := ctx.Deadline()
+			// Check if deadline is too close, accounting for potential clock skew.
+			// Using time.Until ensures we check relative to current time.
+			if ok && time.Until(dl) < cfg.Threshold {
+				if cfg.Metrics != nil {
+					cfg.Metrics.DeadlineCancelledTotal.
+						WithLabelValues(cfg.Client, req.URL.Host, req.Method).
+						Inc()
+				}
+
+				return nil, types.ErrDeadlineTooClose
+			}
+
+			return next.RoundTrip(req)
+		})
+	}
+}