From c5cc7951a3df3dc53757d90cdb5019d48ac7a03c Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Fri, 1 May 2026 22:25:37 -0400 Subject: [PATCH] feat(monitor/httpx): rate-limited HTTP client with 429/5xx retry honoring Retry-After --- .../backend/internal/httpx/client.go | 199 ++++++++++++++++++ .../backend/internal/httpx/client_test.go | 175 +++++++++++++++ 2 files changed, 374 insertions(+) create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/httpx/client.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/httpx/client_test.go diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/httpx/client.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/httpx/client.go new file mode 100644 index 00000000..e214aba6 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/httpx/client.go @@ -0,0 +1,199 @@ +// ©AngelaMos | 2026 +// client.go + +package httpx + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "time" + + "github.com/cenkalti/backoff/v4" + "golang.org/x/time/rate" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/ratelimit" +) + +const ( + defaultUserAgent = "monitor-the-situation/1.0" + defaultRequestTimeout = 30 * time.Second + defaultMaxRetryElapsed = 60 * time.Second + defaultInitialRetryBackoff = 500 * time.Millisecond + defaultMaxRetryBackoff = 10 * time.Second + defaultAPIKeyHeader = "X-API-Key" + clientErrorBodyLimit = 4096 +) + +type Config struct { + Name string + BaseURL string + Headers http.Header + BearerToken string + APIKey string + APIKeyHeader string + Rate rate.Limit + Burst int + ConsecutiveFailureBudget uint32 + BreakerTimeout time.Duration + RequestTimeout time.Duration + MaxRetryElapsed time.Duration + UserAgent string +} + +type Client struct { + cfg Config + limiter *ratelimit.Limiter[*http.Response] + hc *http.Client +} + +func New(cfg Config) *Client { + if cfg.RequestTimeout <= 0 { + cfg.RequestTimeout = defaultRequestTimeout + } + if cfg.MaxRetryElapsed <= 0 { + cfg.MaxRetryElapsed = defaultMaxRetryElapsed + } + if cfg.UserAgent == "" { + cfg.UserAgent = defaultUserAgent + } + limiter := ratelimit.New[*http.Response](ratelimit.Config{ + Name: cfg.Name, + Rate: cfg.Rate, + Burst: cfg.Burst, + ConsecutiveFailureBudget: cfg.ConsecutiveFailureBudget, + BreakerTimeout: cfg.BreakerTimeout, + }) + return &Client{ + cfg: cfg, + limiter: limiter, + hc: &http.Client{Timeout: cfg.RequestTimeout}, + } +} + +func (c *Client) GetJSON(ctx context.Context, path string, query url.Values, out any) error { + resp, err := c.Get(ctx, path, query) + if err != nil { + return err + } + defer drainAndClose(resp) + return json.NewDecoder(resp.Body).Decode(out) +} + +func (c *Client) Get(ctx context.Context, path string, query url.Values) (*http.Response, error) { + target, err := c.resolveURL(path, query) + if err != nil { + return nil, err + } + + bo := backoff.NewExponentialBackOff() + bo.MaxElapsedTime = c.cfg.MaxRetryElapsed + bo.InitialInterval = defaultInitialRetryBackoff + bo.MaxInterval = defaultMaxRetryBackoff + + var resp *http.Response + op := func() error { + r, opErr := c.limiter.Do(ctx, func(rctx context.Context) (*http.Response, error) { + return c.do(rctx, target) + }) + if opErr != nil { + return opErr + } + switch { + case r.StatusCode == http.StatusTooManyRequests: + retryAfter := parseRetryAfter(r.Header.Get("Retry-After")) + drainAndClose(r) + if retryAfter > 0 { + select { + case <-ctx.Done(): + return backoff.Permanent(ctx.Err()) + case <-time.After(retryAfter): + } + } + return errors.New("upstream 429") + case r.StatusCode >= 500 && r.StatusCode <= 599: + drainAndClose(r) + return fmt.Errorf("upstream %d", r.StatusCode) + case r.StatusCode >= 400: + body, _ := io.ReadAll(io.LimitReader(r.Body, clientErrorBodyLimit)) + drainAndClose(r) + return backoff.Permanent(fmt.Errorf("client error %d: %s", r.StatusCode, body)) + } + resp = r + return nil + } + if err := backoff.Retry(op, backoff.WithContext(bo, ctx)); err != nil { + return nil, err + } + return resp, nil +} + +func (c *Client) do(ctx context.Context, target string) (*http.Response, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) + if err != nil { + return nil, err + } + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", c.cfg.UserAgent) + for k, vs := range c.cfg.Headers { + for _, v := range vs { + req.Header.Add(k, v) + } + } + if c.cfg.BearerToken != "" { + req.Header.Set("Authorization", "Bearer "+c.cfg.BearerToken) + } + if c.cfg.APIKey != "" { + header := c.cfg.APIKeyHeader + if header == "" { + header = defaultAPIKeyHeader + } + req.Header.Set(header, c.cfg.APIKey) + } + return c.hc.Do(req) +} + +func (c *Client) resolveURL(path string, query url.Values) (string, error) { + base, err := url.Parse(c.cfg.BaseURL) + if err != nil { + return "", err + } + rel, err := url.Parse(path) + if err != nil { + return "", err + } + resolved := base.ResolveReference(rel) + if query != nil { + resolved.RawQuery = query.Encode() + } + return resolved.String(), nil +} + +func parseRetryAfter(v string) time.Duration { + if v == "" { + return 0 + } + if secs, err := strconv.Atoi(v); err == nil { + return time.Duration(secs) * time.Second + } + if t, err := http.ParseTime(v); err == nil { + d := time.Until(t) + if d > 0 { + return d + } + } + return 0 +} + +func drainAndClose(r *http.Response) { + if r == nil || r.Body == nil { + return + } + _, _ = io.Copy(io.Discard, r.Body) + _ = r.Body.Close() +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/httpx/client_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/httpx/client_test.go new file mode 100644 index 00000000..837ba359 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/httpx/client_test.go @@ -0,0 +1,175 @@ +// ©AngelaMos | 2026 +// client_test.go + +package httpx_test + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + "golang.org/x/time/rate" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/httpx" +) + +func TestClient_GetJSONHonorsRateLimit(t *testing.T) { + var hits atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + hits.Add(1) + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer srv.Close() + + c := httpx.New(httpx.Config{ + Name: "test", + BaseURL: srv.URL, + Rate: rate.Every(50 * time.Millisecond), + Burst: 1, + ConsecutiveFailureBudget: 5, + BreakerTimeout: time.Second, + }) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + start := time.Now() + for i := 0; i < 3; i++ { + var out struct { + Ok bool `json:"ok"` + } + require.NoError(t, c.GetJSON(ctx, "/", nil, &out)) + require.True(t, out.Ok) + } + require.GreaterOrEqual(t, time.Since(start), 100*time.Millisecond) + require.EqualValues(t, 3, hits.Load()) +} + +func TestClient_RetriesOn429WithRetryAfter(t *testing.T) { + var hits atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if hits.Add(1) == 1 { + w.Header().Set("Retry-After", "1") + w.WriteHeader(http.StatusTooManyRequests) + return + } + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer srv.Close() + + c := httpx.New(httpx.Config{ + Name: "retry", + BaseURL: srv.URL, + Rate: rate.Inf, + Burst: 10, + ConsecutiveFailureBudget: 5, + BreakerTimeout: time.Second, + MaxRetryElapsed: 3 * time.Second, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second) + defer cancel() + + var out struct { + Ok bool `json:"ok"` + } + require.NoError(t, c.GetJSON(ctx, "/", nil, &out)) + require.True(t, out.Ok) + require.EqualValues(t, 2, hits.Load()) +} + +func TestClient_RetriesOn5xx(t *testing.T) { + var hits atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if hits.Add(1) < 3 { + w.WriteHeader(http.StatusBadGateway) + return + } + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer srv.Close() + + c := httpx.New(httpx.Config{ + Name: "5xx", + BaseURL: srv.URL, + Rate: rate.Inf, + Burst: 10, + ConsecutiveFailureBudget: 10, + BreakerTimeout: time.Second, + MaxRetryElapsed: 5 * time.Second, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 6*time.Second) + defer cancel() + + var out struct { + Ok bool `json:"ok"` + } + require.NoError(t, c.GetJSON(ctx, "/", nil, &out)) + require.EqualValues(t, 3, hits.Load()) +} + +func TestClient_PermanentErrorOn4xx(t *testing.T) { + var hits atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + hits.Add(1) + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`bad`)) + })) + defer srv.Close() + + c := httpx.New(httpx.Config{ + Name: "perm", + BaseURL: srv.URL, + Rate: rate.Inf, + Burst: 10, + ConsecutiveFailureBudget: 10, + BreakerTimeout: time.Second, + MaxRetryElapsed: 3 * time.Second, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + var out struct{} + err := c.GetJSON(ctx, "/", nil, &out) + require.Error(t, err) + require.EqualValues(t, 1, hits.Load()) +} + +func TestClient_AddsBearerAndAPIKeyHeaders(t *testing.T) { + var bearer, apiKey, ua string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + bearer = r.Header.Get("Authorization") + apiKey = r.Header.Get("X-Custom-Key") + ua = r.Header.Get("User-Agent") + _, _ = w.Write([]byte(`{}`)) + })) + defer srv.Close() + + c := httpx.New(httpx.Config{ + Name: "headers", + BaseURL: srv.URL, + BearerToken: "bearer-abc", + APIKey: "key-xyz", + APIKeyHeader: "X-Custom-Key", + UserAgent: "monitor-test/1.0", + Rate: rate.Inf, + Burst: 1, + ConsecutiveFailureBudget: 5, + BreakerTimeout: time.Second, + }) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + var out struct{} + require.NoError(t, c.GetJSON(ctx, "/", nil, &out)) + require.Equal(t, "Bearer bearer-abc", bearer) + require.Equal(t, "key-xyz", apiKey) + require.Equal(t, "monitor-test/1.0", ua) +}