diff --git a/ingestion/cmd/server/main.go b/ingestion/cmd/server/main.go index 53baf1d..145fef7 100644 --- a/ingestion/cmd/server/main.go +++ b/ingestion/cmd/server/main.go @@ -32,6 +32,7 @@ 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/vectorstore" "github.com/mathiasbq/hyperguild/ingestion/internal/watcher" "github.com/mathiasbq/hyperguild/ingestion/internal/webhook" @@ -314,6 +315,36 @@ func main() { } ctx := context.Background() + + // 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. + jwtValidator, err := startupretry.Do(ctx, + startupretry.Policy{Attempts: 8, Base: time.Second, Max: 30 * time.Second}, + func(ctx context.Context) (*chassisauth.JWTValidator, error) { + v, err := chassisauth.NewJWTValidator(ctx, os.Getenv("DEX_ISSUER_URL"), os.Getenv("MCP_AUDIENCE")) + if err != nil { + logger.Warn("oidc discovery not ready, retrying", "err", err) + } + return v, err + }) + if err != nil { + logger.Error("build jwt validator", "err", err) + os.Exit(1) + } + if jwtValidator != nil { + logger.Info("jwt auth enabled", "issuer", os.Getenv("DEX_ISSUER_URL")) + } + if watchInterval > 0 { watcher.Start(ctx, watcher.Config{ BrainDir: brainDir, @@ -418,15 +449,6 @@ func main() { if webhookHandler != nil { mux.Handle("POST /webhooks/brain-sync", webhookHandler) } - jwtValidator, err := chassisauth.NewJWTValidator(ctx, os.Getenv("DEX_ISSUER_URL"), os.Getenv("MCP_AUDIENCE")) - if err != nil { - logger.Error("build jwt validator", "err", err) - os.Exit(1) - } - if jwtValidator != nil { - logger.Info("jwt auth enabled", "issuer", os.Getenv("DEX_ISSUER_URL")) - } - // Resource-metadata URL is only emitted on 401 when Dex OAuth is // configured. Static-Bearer-only deployments leave this empty so // clients never see an OAuth challenge. diff --git a/ingestion/internal/startupretry/startupretry.go b/ingestion/internal/startupretry/startupretry.go new file mode 100644 index 0000000..2882f67 --- /dev/null +++ b/ingestion/internal/startupretry/startupretry.go @@ -0,0 +1,75 @@ +// 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) +} diff --git a/ingestion/internal/startupretry/startupretry_test.go b/ingestion/internal/startupretry/startupretry_test.go new file mode 100644 index 0000000..a96a017 --- /dev/null +++ b/ingestion/internal/startupretry/startupretry_test.go @@ -0,0 +1,144 @@ +// 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") + } +}