fix(ingestion): bump mcp-chassis to v0.7.1; label refused REST requests by route (#155) #181
@@ -2,24 +2,25 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
chassisauth "git.d-ma.be/mathias/mcp-chassis/auth"
|
||||
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/startupretry"
|
||||
)
|
||||
|
||||
// buildJWTValidator runs OIDC discovery under policy, so a 502 from an auth
|
||||
// upstream that is not serving yet on a cold boot is retried instead of
|
||||
// exiting (#83). It honours ctx between attempts.
|
||||
func buildJWTValidator(
|
||||
ctx context.Context, logger *slog.Logger, issuer, audience string, policy startupretry.Policy,
|
||||
) (*chassisauth.JWTValidator, error) {
|
||||
return startupretry.Do(ctx, policy, func(ctx context.Context) (*chassisauth.JWTValidator, error) {
|
||||
// buildJWTValidator builds the JWT validator for issuer (nil when issuer is
|
||||
// empty: JWT auth disabled). Since mcp-chassis v0.7 construction no longer
|
||||
// fails when OIDC discovery does (mcp-chassis#4): the issuer resolves lazily,
|
||||
// and until it does a JWT is answered 503-and-retry, not 401. That replaces
|
||||
// #83's boot-time retry loop (hyperguild#155). An issuer still unresolved at
|
||||
// boot is logged, so a misconfigured issuer is visible rather than silent.
|
||||
func buildJWTValidator(ctx context.Context, logger *slog.Logger, issuer, audience string) (*chassisauth.JWTValidator, error) {
|
||||
v, err := chassisauth.NewJWTValidator(ctx, issuer, audience)
|
||||
if err != nil {
|
||||
logger.Warn("oidc discovery not ready, retrying", "err", err)
|
||||
return nil, fmt.Errorf("build jwt validator: %w", err)
|
||||
}
|
||||
return v, err
|
||||
})
|
||||
if ok, rerr := v.Ready(ctx); !ok {
|
||||
logger.Warn("jwt issuer not resolved at boot; JWTs answer 503 until it is", "issuer", issuer, "err", rerr)
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
@@ -8,12 +9,9 @@ import (
|
||||
"net/http/httptest"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/startupretry"
|
||||
)
|
||||
|
||||
// fakeIssuer answers OIDC discovery with 502 for the first fail requests, as
|
||||
@@ -41,46 +39,42 @@ func fakeIssuer(t *testing.T, fail int32) (*httptest.Server, *atomic.Int32) {
|
||||
return srv, &discoveries
|
||||
}
|
||||
|
||||
var fastPolicy = startupretry.Policy{Attempts: 8, Base: time.Millisecond, Max: 5 * time.Millisecond}
|
||||
|
||||
func TestBuildJWTValidatorRetriesDiscovery502sThenSucceeds(t *testing.T) {
|
||||
srv, discoveries := fakeIssuer(t, 3)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
v, err := buildJWTValidator(ctx, slog.New(slog.DiscardHandler), srv.URL, "brain", fastPolicy)
|
||||
// #83 retried discovery at boot because chassis v0.2.0 failed construction on
|
||||
// a 502. Since chassis v0.7 (mcp-chassis#4) the validator is built anyway and
|
||||
// resolves the issuer lazily, so boot no longer waits or exits on it
|
||||
// (hyperguild#155). An unresolved issuer is logged at boot.
|
||||
func TestBuildJWTValidatorReturnsAValidatorWhileDiscovery502s(t *testing.T) {
|
||||
srv, _ := fakeIssuer(t, 1000)
|
||||
var buf bytes.Buffer
|
||||
v, err := buildJWTValidator(context.Background(), slog.New(slog.NewTextHandler(&buf, nil)), srv.URL, "brain")
|
||||
require.NoError(t, err)
|
||||
assert.NotNil(t, v)
|
||||
assert.Equal(t, int32(4), discoveries.Load(), "three 502s, then the successful discovery")
|
||||
assert.Contains(t, buf.String(), "jwt issuer not resolved at boot")
|
||||
}
|
||||
|
||||
func TestBuildJWTValidatorGivesUpAfterTheBudgetWithTheLastError(t *testing.T) {
|
||||
srv, discoveries := fakeIssuer(t, 1000)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
_, err := buildJWTValidator(ctx, slog.New(slog.DiscardHandler), srv.URL, "brain", fastPolicy)
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "status 502")
|
||||
assert.Equal(t, int32(8), discoveries.Load(), "bounded: exactly the attempt budget")
|
||||
func TestBuildJWTValidatorLogsNothingAgainstAHealthyIssuer(t *testing.T) {
|
||||
srv, _ := fakeIssuer(t, 0)
|
||||
var buf bytes.Buffer
|
||||
v, err := buildJWTValidator(context.Background(), slog.New(slog.NewTextHandler(&buf, nil)), srv.URL, "brain")
|
||||
require.NoError(t, err)
|
||||
assert.NotNil(t, v)
|
||||
assert.NotContains(t, buf.String(), "not resolved")
|
||||
}
|
||||
|
||||
func TestBuildJWTValidatorStopsRetryingWhenTheContextIsCancelled(t *testing.T) {
|
||||
func TestBuildJWTValidatorIsNilWithoutAnIssuer(t *testing.T) {
|
||||
v, err := buildJWTValidator(context.Background(), slog.New(slog.DiscardHandler), "", "brain")
|
||||
require.NoError(t, err)
|
||||
assert.Nil(t, v)
|
||||
}
|
||||
|
||||
// With the issuer unresolved, a JWT on a REST route answers 503 (retry), not
|
||||
// 401: the token may be fine, the server just cannot check it yet.
|
||||
func TestBuildMuxAnswers503ToAJWTWhileTheIssuerIsUnresolved(t *testing.T) {
|
||||
srv, _ := fakeIssuer(t, 1000)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
slow := startupretry.Policy{Attempts: 8, Base: time.Hour, Max: time.Hour}
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := buildJWTValidator(ctx, slog.New(slog.DiscardHandler), srv.URL, "brain", slow)
|
||||
done <- err
|
||||
}()
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
cancel()
|
||||
select {
|
||||
case err := <-done:
|
||||
assert.ErrorIs(t, err, context.Canceled)
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("did not return after the context was cancelled")
|
||||
}
|
||||
v, err := buildJWTValidator(context.Background(), slog.New(slog.DiscardHandler), srv.URL, "brain")
|
||||
require.NoError(t, err)
|
||||
mux, err := buildMux(restAuth{token: routesTok, validator: v, realm: "brain"}, testHandler(t), nil)
|
||||
require.NoError(t, err)
|
||||
code, _ := do(t, mux, "GET", "/pending", "Bearer aaa.bbb.ccc")
|
||||
assert.Equal(t, http.StatusServiceUnavailable, code)
|
||||
}
|
||||
|
||||
@@ -32,7 +32,6 @@ import (
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/reranker"
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/search"
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/startupretry"
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/symlinkguard"
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/vectorstore"
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/watcher"
|
||||
@@ -369,21 +368,15 @@ func main() {
|
||||
// hyperguild#157: report a symlink that appears in the brain while running.
|
||||
go symlinkguard.Watch(ctx, brainDir, symlinkScanInterval(os.Getenv), buildNotifier())
|
||||
|
||||
// Built before the watcher, the graph backfill and session ingestion, and
|
||||
// retried rather than fatal (#83). Two separate reasons:
|
||||
//
|
||||
// - On a cold cluster boot this pod can beat dex to serving and gets a
|
||||
// 502 from ingress. Nothing is misconfigured; it simply started first.
|
||||
// Exiting there meant CrashLoopBackOff redoing the backfill on every
|
||||
// attempt, with exponential backoff stretching the brain outage.
|
||||
// - Constructing it first means a genuine config error (wrong issuer,
|
||||
// bad audience) fails in seconds instead of after all that work.
|
||||
//
|
||||
// The budget stays bounded: retrying forever would turn a misconfiguration
|
||||
// into a process that looks alive and never serves.
|
||||
// Never fatal on an unreachable issuer (#83): on a cold cluster boot this
|
||||
// pod can beat dex to serving and get a 502 from ingress, and exiting meant
|
||||
// CrashLoopBackOff redoing the backfill. Since mcp-chassis v0.7 the
|
||||
// validator resolves the issuer lazily and answers JWTs 503-and-retry until
|
||||
// it does (hyperguild#155), so no boot retry is needed. A misconfigured
|
||||
// issuer no longer exits the process: it is logged at boot and every JWT
|
||||
// gets 503; the static bearer keeps working.
|
||||
jwtValidator, err := buildJWTValidator(ctx, logger,
|
||||
os.Getenv("DEX_ISSUER_URL"), os.Getenv("MCP_AUDIENCE"),
|
||||
startupretry.Policy{Attempts: 8, Base: time.Second, Max: 30 * time.Second})
|
||||
os.Getenv("DEX_ISSUER_URL"), os.Getenv("MCP_AUDIENCE"))
|
||||
if err != nil {
|
||||
logger.Error("build jwt validator", "err", err)
|
||||
os.Exit(1)
|
||||
|
||||
@@ -73,5 +73,23 @@ func mountREST(mux *http.ServeMux, h *api.Handler, auth func(http.Handler) http.
|
||||
rest.HandleFunc("POST /promote", h.Promote)
|
||||
rest.HandleFunc("POST /backfill-embeddings", h.BackfillEmbeddings)
|
||||
rest.HandleFunc("GET /pass-rate", h.PassRate)
|
||||
mux.Handle("/", auth(rest))
|
||||
mux.Handle("/", labelRESTPattern(rest, auth(rest)))
|
||||
}
|
||||
|
||||
// labelRESTPattern sets the request's Pattern to the REST route it addresses
|
||||
// before auth runs, so the metrics middleware (which reads Pattern from the
|
||||
// same *http.Request) labels a refused request by its route rather than by
|
||||
// the "/" mount (hyperguild#155). A refused request to a path no route
|
||||
// matches keeps "/", so for refused requests the label set stays bounded by
|
||||
// the registered routes. That bound does not hold for authenticated
|
||||
// requests: an unregistered path reaches the REST mux, which 404s with an
|
||||
// empty Pattern, and the metrics middleware falls back to the raw URL path
|
||||
// (hyperguild#185).
|
||||
func labelRESTPattern(rest *http.ServeMux, next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if _, pattern := rest.Handler(r); pattern != "" {
|
||||
r.Pattern = pattern
|
||||
}
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/metrics"
|
||||
)
|
||||
|
||||
// wrongTokSentinel is a presented credential distinctive enough that any
|
||||
// occurrence in a log line is the raw token, not a coincidence.
|
||||
const wrongTokSentinel = "wrong-tok-sentinel-q7zx"
|
||||
|
||||
// hyperguild#155: since #149 every REST route sits behind BearerMiddleware,
|
||||
// and chassis v0.2.0 refused without logging, so an unknown caller got
|
||||
// silent 401s. The chassis now writes an `mcp auth rejected` line, and the
|
||||
// presented token appears nowhere in the log as a bare value (PR #181
|
||||
// review: asserting on the "Bearer " header form could never fail, since
|
||||
// the chassis logs fields, not headers).
|
||||
func TestBuildMuxRejectionLogLineOmitsTheBarePresentedToken(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
prev := slog.Default()
|
||||
slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil)))
|
||||
t.Cleanup(func() { slog.SetDefault(prev) })
|
||||
|
||||
code, _ := do(t, testMux(t), "POST", "/query", "Bearer "+wrongTokSentinel)
|
||||
require.Equal(t, http.StatusUnauthorized, code)
|
||||
assert.Contains(t, buf.String(), "mcp auth rejected")
|
||||
assert.Contains(t, buf.String(), "path=/query")
|
||||
assert.NotContains(t, buf.String(), wrongTokSentinel, "the rejected token is logged as a bare value")
|
||||
assert.NotContains(t, buf.String(), routesTok, "the valid static token is logged")
|
||||
}
|
||||
|
||||
// The `mcp auth accepted` line leaves out the valid static token.
|
||||
func TestBuildMuxAcceptedLogLineOmitsTheBareStaticToken(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
prev := slog.Default()
|
||||
slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil)))
|
||||
t.Cleanup(func() { slog.SetDefault(prev) })
|
||||
|
||||
code, _ := do(t, testMux(t), "GET", "/pending", "Bearer "+routesTok)
|
||||
require.Equal(t, http.StatusOK, code)
|
||||
assert.Contains(t, buf.String(), "mcp auth accepted")
|
||||
assert.NotContains(t, buf.String(), routesTok, "the accepted static token is logged as a bare value")
|
||||
}
|
||||
|
||||
// An authenticated REST request is labelled with the REST mux's pattern:
|
||||
// the inner mux sets it on the same *http.Request the outer middleware reads.
|
||||
func TestMetricsLabelAnAuthenticatedRESTRequestWithItsRoutePattern(t *testing.T) {
|
||||
reg := metrics.New()
|
||||
code, _ := do(t, reg.Middleware(testMux(t)), "GET", "/pending", "Bearer "+routesTok)
|
||||
require.Equal(t, http.StatusOK, code)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
reg.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
assert.Contains(t, rec.Body.String(), `path="GET /pending"`)
|
||||
assert.NotContains(t, rec.Body.String(), `path="/"`)
|
||||
}
|
||||
|
||||
// hyperguild#155: a request refused by REST auth never reaches the REST mux,
|
||||
// so it was labelled path="/" and a 4xx on /query could not be told from one
|
||||
// on /write. It is now labelled with the REST pattern it was addressed to.
|
||||
func TestMetricsLabelARejectedRESTRequestWithItsRoutePattern(t *testing.T) {
|
||||
reg := metrics.New()
|
||||
code, _ := do(t, reg.Middleware(testMux(t)), "POST", "/query", "")
|
||||
require.Equal(t, http.StatusUnauthorized, code)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
reg.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
assert.Contains(t, rec.Body.String(), `path="POST /query",status="4xx"`)
|
||||
}
|
||||
|
||||
// A path no REST route matches keeps the catch-all label, so an attacker
|
||||
// probing random URLs cannot grow the series set.
|
||||
func TestMetricsLabelARejectedUnknownPathWithTheCatchAll(t *testing.T) {
|
||||
reg := metrics.New()
|
||||
code, _ := do(t, reg.Middleware(testMux(t)), "POST", "/no-such-route-zq", "")
|
||||
require.Equal(t, http.StatusUnauthorized, code)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
reg.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
assert.Contains(t, rec.Body.String(), `path="/",status="4xx"`)
|
||||
assert.NotContains(t, rec.Body.String(), "no-such-route-zq")
|
||||
}
|
||||
@@ -27,7 +27,7 @@ func okHandler(body string) http.Handler {
|
||||
// review showed a test of a COPY of the router let main.go regress unseen.
|
||||
func testMux(t *testing.T) http.Handler {
|
||||
t.Helper()
|
||||
h := api.NewHandler(t.TempDir(), slog.New(slog.NewTextHandler(os.Stderr, nil)), pipeline.Config{})
|
||||
h := testHandler(t)
|
||||
public := map[string]http.Handler{}
|
||||
for p := range publicRoutes {
|
||||
public[p] = okHandler("public " + p)
|
||||
@@ -37,6 +37,11 @@ func testMux(t *testing.T) http.Handler {
|
||||
return mux
|
||||
}
|
||||
|
||||
func testHandler(t *testing.T) *api.Handler {
|
||||
t.Helper()
|
||||
return api.NewHandler(t.TempDir(), slog.New(slog.NewTextHandler(os.Stderr, nil)), pipeline.Config{})
|
||||
}
|
||||
|
||||
func do(t *testing.T, h http.Handler, method, path, authz string) (int, string) {
|
||||
t.Helper()
|
||||
req := httptest.NewRequest(method, path, strings.NewReader(`{"query":"x"}`))
|
||||
|
||||
+1
-1
@@ -51,7 +51,7 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
git.d-ma.be/mathias/mcp-chassis v0.2.0
|
||||
git.d-ma.be/mathias/mcp-chassis v0.7.1
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect
|
||||
github.com/goccy/go-json v0.10.3 // indirect
|
||||
|
||||
+2
-2
@@ -1,5 +1,5 @@
|
||||
git.d-ma.be/mathias/mcp-chassis v0.2.0 h1:6fLmb7xqRa2nNVWsHaUbbfbArgDXJw/gDhb09clBIjo=
|
||||
git.d-ma.be/mathias/mcp-chassis v0.2.0/go.mod h1:Ks7EK2UnGAN0H3rJjKUxUagX8/ZBdtLrOlcUbv0RwH8=
|
||||
git.d-ma.be/mathias/mcp-chassis v0.7.1 h1:4h8TOBHTA0L/no4IPcr4tLcQtmKUlhE2/cXyPdNFGmM=
|
||||
git.d-ma.be/mathias/mcp-chassis v0.7.1/go.mod h1:Ks7EK2UnGAN0H3rJjKUxUagX8/ZBdtLrOlcUbv0RwH8=
|
||||
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
|
||||
@@ -1,75 +0,0 @@
|
||||
// Package startupretry retries a startup dependency that may not be serving
|
||||
// yet, instead of exiting on its first failure.
|
||||
//
|
||||
// It exists for the cold-boot ordering race (#83): on a cluster restart the
|
||||
// ingestion pod can reach OIDC discovery before dex is answering, gets a 502
|
||||
// from ingress, and dies — after having already connected postgres, backfilled
|
||||
// the graph and ingested session batches, all of which the restart redoes.
|
||||
// Nothing is misconfigured in that scenario; the pod simply started first.
|
||||
//
|
||||
// The budget is bounded on purpose. Retrying forever would turn a genuine
|
||||
// misconfiguration (wrong issuer URL, bad audience) into a process that looks
|
||||
// healthy and never serves, which is harder to diagnose than a crash loop.
|
||||
package startupretry
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Policy bounds the retry: at most Attempts calls, with the delay doubling
|
||||
// from Base and capped at Max.
|
||||
type Policy struct {
|
||||
Attempts int // total calls, including the first; < 1 means 1
|
||||
Base time.Duration // delay before the second attempt
|
||||
Max time.Duration // ceiling for the doubling delay
|
||||
}
|
||||
|
||||
// Do calls fn until it succeeds, the attempt budget is spent, or ctx ends.
|
||||
//
|
||||
// The returned error wraps the last failure, so callers keep the diagnosis
|
||||
// they would have had without the retry.
|
||||
func Do[T any](ctx context.Context, p Policy, fn func(context.Context) (T, error)) (T, error) {
|
||||
var zero T
|
||||
|
||||
attempts := p.Attempts
|
||||
if attempts < 1 {
|
||||
attempts = 1
|
||||
}
|
||||
|
||||
delay := p.Base
|
||||
var lastErr error
|
||||
|
||||
for i := range attempts {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return zero, fmt.Errorf("startupretry: %w", err)
|
||||
}
|
||||
|
||||
v, err := fn(ctx)
|
||||
if err == nil {
|
||||
return v, nil
|
||||
}
|
||||
lastErr = err
|
||||
|
||||
if i == attempts-1 {
|
||||
break
|
||||
}
|
||||
|
||||
if delay <= 0 {
|
||||
delay = time.Second
|
||||
}
|
||||
if p.Max > 0 && delay > p.Max {
|
||||
delay = p.Max
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return zero, fmt.Errorf("startupretry: %w", ctx.Err())
|
||||
case <-time.After(delay):
|
||||
}
|
||||
delay *= 2
|
||||
}
|
||||
|
||||
return zero, fmt.Errorf("startupretry: gave up after %d attempts: %w", attempts, lastErr)
|
||||
}
|
||||
@@ -1,142 +0,0 @@
|
||||
// ingestion/internal/startupretry/startupretry_test.go
|
||||
package startupretry
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func fastPolicy(attempts int) Policy {
|
||||
return Policy{Attempts: attempts, Base: time.Millisecond, Max: 2 * time.Millisecond}
|
||||
}
|
||||
|
||||
func TestDo_ReturnsImmediatelyOnFirstSuccess(t *testing.T) {
|
||||
calls := 0
|
||||
got, err := Do(context.Background(), fastPolicy(5), func(context.Context) (string, error) {
|
||||
calls++
|
||||
return "ready", nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Do: %v", err)
|
||||
}
|
||||
if got != "ready" {
|
||||
t.Errorf("got %q, want %q", got, "ready")
|
||||
}
|
||||
if calls != 1 {
|
||||
t.Errorf("calls = %d, want 1", calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDo_SurvivesTransientFailuresThenSucceeds(t *testing.T) {
|
||||
// The observed cold-boot race: dex is not serving yet, ingress answers
|
||||
// 502 for the first few attempts, then discovery works.
|
||||
calls := 0
|
||||
got, err := Do(context.Background(), fastPolicy(5), func(context.Context) (string, error) {
|
||||
calls++
|
||||
if calls < 4 {
|
||||
return "", errors.New("oidc discovery: status 502")
|
||||
}
|
||||
return "ready", nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Do returned %v — a transient 502 must not be fatal", err)
|
||||
}
|
||||
if got != "ready" {
|
||||
t.Errorf("got %q, want %q", got, "ready")
|
||||
}
|
||||
if calls != 4 {
|
||||
t.Errorf("calls = %d, want 4", calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDo_GivesUpAfterTheAttemptBudgetAndReportsTheLastError(t *testing.T) {
|
||||
// A genuinely broken issuer must still fail — retrying forever would turn
|
||||
// a config error into a silent hang, which is worse than crash-looping.
|
||||
last := errors.New("oidc discovery: status 404")
|
||||
calls := 0
|
||||
|
||||
_, err := Do(context.Background(), fastPolicy(3), func(context.Context) (string, error) {
|
||||
calls++
|
||||
return "", last
|
||||
})
|
||||
|
||||
if !errors.Is(err, last) {
|
||||
t.Fatalf("err = %v, want it to wrap the last failure", err)
|
||||
}
|
||||
if calls != 3 {
|
||||
t.Errorf("calls = %d, want 3 (the attempt budget)", calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDo_StopsPromptlyWhenTheContextIsCancelled(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
calls := 0
|
||||
|
||||
done := make(chan struct{})
|
||||
var err error
|
||||
go func() {
|
||||
defer close(done)
|
||||
_, err = Do(ctx, Policy{Attempts: 100, Base: 50 * time.Millisecond, Max: time.Second},
|
||||
func(context.Context) (string, error) {
|
||||
calls++
|
||||
if calls == 1 {
|
||||
cancel()
|
||||
}
|
||||
return "", errors.New("still down")
|
||||
})
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Do did not return after context cancellation")
|
||||
}
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("err = %v, want context.Canceled", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDo_BacksOffAndIsCappedByMax(t *testing.T) {
|
||||
var delays []time.Duration
|
||||
p := Policy{Attempts: 5, Base: 10 * time.Millisecond, Max: 25 * time.Millisecond}
|
||||
|
||||
last := time.Now()
|
||||
_, _ = Do(context.Background(), p, func(context.Context) (string, error) {
|
||||
now := time.Now()
|
||||
delays = append(delays, now.Sub(last))
|
||||
last = now
|
||||
return "", errors.New("down")
|
||||
})
|
||||
|
||||
// delays[0] is the first call (no wait); the rest are the sleeps.
|
||||
if len(delays) != 5 {
|
||||
t.Fatalf("attempts = %d, want 5", len(delays))
|
||||
}
|
||||
if delays[1] >= delays[2] {
|
||||
t.Errorf("backoff is not growing: %v then %v", delays[1], delays[2])
|
||||
}
|
||||
for i, d := range delays[1:] {
|
||||
if d > 2*p.Max {
|
||||
t.Errorf("delay %d = %v, exceeds the Max cap %v with slack", i+1, d, p.Max)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestPolicy_ZeroValueStillMakesOneAttempt(t *testing.T) {
|
||||
// A caller that forgets to configure the policy must not silently get a
|
||||
// no-op that returns a nil validator and a nil error.
|
||||
calls := 0
|
||||
_, err := Do(context.Background(), Policy{}, func(context.Context) (string, error) {
|
||||
calls++
|
||||
return "", errors.New("down")
|
||||
})
|
||||
|
||||
if calls != 1 {
|
||||
t.Errorf("calls = %d, want 1", calls)
|
||||
}
|
||||
if err == nil {
|
||||
t.Error("err = nil, want the failure surfaced")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user