diff --git a/docs/shared-tier-admission.md b/docs/shared-tier-admission.md index c6a104a..696b9e4 100644 --- a/docs/shared-tier-admission.md +++ b/docs/shared-tier-admission.md @@ -127,15 +127,43 @@ The five legacy single-scope quota headers (`X-Saturn-Service-Tier`, `X-Saturn-Rate-Limit-Uncached-Prompt-Tokens`, `X-Saturn-Rate-Limit-Generated-Tokens`) were removed in one cutover release (ruling R8, a hard cut). In that release Saturn stops stamping them, Traefik -stops allowlisting them, the phoebe chart drops them from `trustedHeaders` -(so they leave the trust set and the strip middleware together), and Phoebe -stops reading them. There is no dual-contract window after the cutover and no +stops allowlisting them, the phoebe chart drops them from `trustedHeaders`, +and Phoebe stops reading them. The strip outlives the trust: the edge keeps +blanking client-supplied copies of the five names until every Phoebe replica +runs the R8 image (see below). There is no dual-contract window after the cutover and no fallback: a request that carries only the legacy headers and no `X-Saturn-Owner-Id` has no policy and, with admission enabled, fails closed -with 503. The four components of that release can roll in any order, because -the scoped envelope was already stamped on every route where admission runs -before the cutover, and every Phoebe that still read the legacy headers -preferred the scoped envelope whenever the owner id was present. +with 503. + +The four components of that release are safe to roll in any order only +because admission stays disabled (`admission.enabled: false`, ruling R12) for the +whole cutover, so no replica enforces a policy envelope during the rollout. +No current or pre-cutover Phoebe reads the legacy headers on dedicated +routes. The scoped envelope is stamped only by the gateway ForwardAuth, on +gateway routes; every Phoebe that still read the legacy headers preferred the +scoped envelope whenever the owner id was present. Historical per-resource +shared routes never carried the policy contract, and R8 does not change that: +they must be drained or removed before admission is enabled (see the +requirement later in this section). + +The five legacy names leave the trust set but must not leave the edge strip +lists in the same release. A Phoebe replica older than the R8 image still +parses the legacy envelope, and during a mixed rollout it would accept a +client-forged `X-Saturn-Service-Tier` / `X-Saturn-Rate-Limit-*` set on any +admission-enabled request that has no `X-Saturn-Owner-Id`. The edge strip has +always prevented that, and the R8 image cannot protect older replicas by +stripping in its own code. So the names stay stripped at the edge, without +being trusted: the traefik chart's `tf-gateway-headers` middleware blanks +them on the gateway route, and the phoebe chart's +`phoebe-inference-headers` middleware must blank them on the standard route +through a strip-only list that is not rendered into `PHOEBE_TRUSTED_HEADERS`. +On the standard route this breaks the usual equality of the strip set and the +trusted set (ruling R3) on purpose: the strip set is the trusted set plus the +five legacy names. Remove the strip-only names only in a later release, after +every Phoebe replica runs the R8 image. The recommended order is phoebe #55 +first (or in the same window), then Atlas #6715, with saturn-k8s #1073 +alongside either. Nothing reads these headers after the cutover, so dropping +the strip once no pre-R8 replica remains is harmless. While admission is disabled, Phoebe does not require a policy envelope. Once admission is enabled, a missing or structurally broken policy fails closed. diff --git a/internal/identity/trusted_headers.go b/internal/identity/trusted_headers.go index b6e1061..7a8694d 100644 --- a/internal/identity/trusted_headers.go +++ b/internal/identity/trusted_headers.go @@ -78,6 +78,8 @@ var requiredTrustedHeaders = []struct { }{ {HeaderGateway, "gateway requests will not be recognized"}, {HeaderServingMode, "every header-routed (non-gateway) inference request will be refused with 404 because the serving mode reads as absent (ruling #19)"}, + {HeaderOwnerID, "with admission enabled, every shared gateway inference request will be refused with 503 because the quota policy has no owner-id anchor (R7/R8: the legacy service-tier envelope is no longer read)"}, + {HeaderOrgID, "with admission enabled, every shared gateway inference request will be refused with 503 because the trusted organization identity reads as absent"}, } // trustedHeaderSet is the active trusted-header set. Keys are canonical diff --git a/internal/identity/trusted_headers_test.go b/internal/identity/trusted_headers_test.go index 7cc02b1..165b6f4 100644 --- a/internal/identity/trusted_headers_test.go +++ b/internal/identity/trusted_headers_test.go @@ -170,9 +170,14 @@ func TestFromRequestIgnoresHeadersOutsideActiveSet(t *testing.T) { } } -// TestFromRequestFallbackTrustsPinnedHeaders is the no-regression pin: with -// the fallback engaged (unset env), all 13 remain trusted on the real parse -// path — request handling is unchanged from before the gate existed. +// TestFromRequestFallbackTrustsPinnedHeaders is the no-regression pin for the +// gate: with the fallback engaged (unset env), all 13 pinned envelope headers +// are trusted on the real parse path, so a missing or misrendered ConfigMap +// does not change request handling relative to the chart's intended set, and +// the gate's fallback reads exactly what the ungated parser reads. Note: +// ruling R8 removed the five legacy single-scope quota headers +// (X-Saturn-Service-Tier, X-Saturn-Rate-Limit-*) from the parser entirely, +// so pre-R8 phoebe read 18 headers; see TestPinnedFallbackIsExactlyThe13. func TestFromRequestFallbackTrustsPinnedHeaders(t *testing.T) { withTrustedHeadersEnv(t, "", false) @@ -274,12 +279,45 @@ func TestLoadTrustedHeadersErrorsWhenGatewayMissing(t *testing.T) { // TestLoadTrustedHeadersNoErrorWhenRequiredHeadersPresent: a configured list // containing every required header (in any case) logs nothing at ERROR. func TestLoadTrustedHeadersNoErrorWhenRequiredHeadersPresent(t *testing.T) { - out := loadTrustedHeadersCapturingErrors(t, "x-saturn-gateway, X-SATURN-SERVING-MODE, X-Saturn-Org-Id") + out := loadTrustedHeadersCapturingErrors(t, "x-saturn-gateway, X-SATURN-SERVING-MODE, X-Saturn-Org-Id, x-saturn-owner-id") if out != "" { t.Fatalf("unexpected error log for a list with every required header: %q", out) } } +// TestLoadTrustedHeadersErrorsWhenOwnerOrOrgAnchorMissing: after R8, +// X-Saturn-Owner-Id is the only anchor for the quota policy and admission +// also requires X-Saturn-Org-Id, so a configured list that omits either +// would 503 every admitted shared request. Each omission logs an ERROR line +// naming the header and the 503 consequence, and the list is still installed +// exactly as written (neither header becomes trusted). +func TestLoadTrustedHeadersErrorsWhenOwnerOrOrgAnchorMissing(t *testing.T) { + out := loadTrustedHeadersCapturingErrors(t, "X-Saturn-Gateway,X-Saturn-Serving-Mode") + for _, missing := range []string{HeaderOwnerID, HeaderOrgID} { + if !strings.Contains(out, "omits "+missing) { + t.Fatalf("error log %q does not flag the missing %s", out, missing) + } + if _, ok := ActiveTrustedHeaders()[missing]; ok { + t.Fatalf("active set trusts %s although the configured list omits it", missing) + } + } + if !strings.Contains(out, "503") || !strings.Contains(out, "owner-id anchor") { + t.Fatalf("error log %q does not state the 503 owner-anchor consequence", out) + } + if got := len(ActiveTrustedHeaders()); got != 2 { + t.Fatalf("active set has %d headers, want exactly the 2 configured", got) + } +} + +// TestLoadTrustedHeadersNoErrorForPinnedList: the full pinned 13, configured +// explicitly, satisfies every required header and logs nothing at ERROR. +func TestLoadTrustedHeadersNoErrorForPinnedList(t *testing.T) { + out := loadTrustedHeadersCapturingErrors(t, strings.Join(pinnedTrustedHeaders, ",")) + if out != "" { + t.Fatalf("unexpected error log for the pinned list: %q", out) + } +} + // TestStripUntrustedSaturnHeaders (ruling Q-R8STRIP option b): with the pinned // set active, every X-Saturn-* header outside the trusted set and the // edge-contract identity headers is removed, in any case spelling and with all diff --git a/internal/proxy/admission.go b/internal/proxy/admission.go index 25217d7..b6ead86 100644 --- a/internal/proxy/admission.go +++ b/internal/proxy/admission.go @@ -384,16 +384,16 @@ func parseTrustedRateLimits(id identity.Identity) (admission.RateLimits, admissi } return out, nil } - scopedValues := [8]string{ - id.OrgRateLimitRequests, id.OrgRateLimitTotalPromptTokens, - id.OrgRateLimitUncachedPromptTokens, id.OrgRateLimitGeneratedTokens, - id.OwnerRateLimitRequests, id.OwnerRateLimitTotalPromptTokens, - id.OwnerRateLimitUncachedPromptTokens, id.OwnerRateLimitGeneratedTokens, - } - anyPresent := func(values []string) bool { - for _, value := range values { - if value != "" { - return true + orgNames := [4]string{identity.HeaderOrgRateLimitRequests, identity.HeaderOrgRateLimitTotalPromptTokens, identity.HeaderOrgRateLimitUncachedPromptTokens, identity.HeaderOrgRateLimitGeneratedTokens} + orgValues := [4]string{id.OrgRateLimitRequests, id.OrgRateLimitTotalPromptTokens, id.OrgRateLimitUncachedPromptTokens, id.OrgRateLimitGeneratedTokens} + ownerNames := [4]string{identity.HeaderOwnerRateLimitRequests, identity.HeaderOwnerRateLimitTotalPromptTokens, identity.HeaderOwnerRateLimitUncachedPromptTokens, identity.HeaderOwnerRateLimitGeneratedTokens} + ownerValues := [4]string{id.OwnerRateLimitRequests, id.OwnerRateLimitTotalPromptTokens, id.OwnerRateLimitUncachedPromptTokens, id.OwnerRateLimitGeneratedTokens} + anyScopedPresent := func() bool { + for _, values := range [][4]string{orgValues, ownerValues} { + for _, value := range values { + if value != "" { + return true + } } } return false @@ -403,25 +403,48 @@ func parseTrustedRateLimits(id identity.Identity) (admission.RateLimits, admissi case id.OwnerID != "": // Scoped envelope: the owner id is the structural anchor (R7). The 8 // scoped headers are per-field R4 — absent is unlimited. - organization, err := parseScope( - [4]string{identity.HeaderOrgRateLimitRequests, identity.HeaderOrgRateLimitTotalPromptTokens, identity.HeaderOrgRateLimitUncachedPromptTokens, identity.HeaderOrgRateLimitGeneratedTokens}, - [4]string{id.OrgRateLimitRequests, id.OrgRateLimitTotalPromptTokens, id.OrgRateLimitUncachedPromptTokens, id.OrgRateLimitGeneratedTokens}, - ) + organization, err := parseScope(orgNames, orgValues) if err != nil { return organization, admission.RateLimits{}, err } - owner, err := parseScope( - [4]string{identity.HeaderOwnerRateLimitRequests, identity.HeaderOwnerRateLimitTotalPromptTokens, identity.HeaderOwnerRateLimitUncachedPromptTokens, identity.HeaderOwnerRateLimitGeneratedTokens}, - [4]string{id.OwnerRateLimitRequests, id.OwnerRateLimitTotalPromptTokens, id.OwnerRateLimitUncachedPromptTokens, id.OwnerRateLimitGeneratedTokens}, - ) + owner, err := parseScope(ownerNames, ownerValues) return organization, owner, err - case anyPresent(scopedValues[:]): + case anyScopedPresent(): // Scoped limit headers without the owner-id anchor are a structural // violation (R7), not unlimited fields. return admission.RateLimits{}, admission.RateLimits{}, fmt.Errorf("incomplete trusted shared-inference rate-limit policy: scoped headers without %s", identity.HeaderOwnerID) default: // No identity anchor and no policy at all: a structural violation // (R7) that fails closed. - return admission.RateLimits{}, admission.RateLimits{}, fmt.Errorf("incomplete trusted shared-inference rate-limit policy") + return admission.RateLimits{}, admission.RateLimits{}, errNoTrustedRateLimitPolicy } } + +// errNoTrustedRateLimitPolicy is the parser's "no anchor and no scoped policy +// at all" result. The proxy call site matches it to add a log-only diagnostic +// for pre-R8 producers (see legacyQuotaHeadersPresent). +var errNoTrustedRateLimitPolicy = errors.New("incomplete trusted shared-inference rate-limit policy") + +// legacyQuotaHeaderNames are the five single-scope quota headers that R8 +// removed from the trusted envelope. Phoebe never reads them for a trust or +// limit decision; they are listed here ONLY so the proxy can log that a +// not-yet-upgraded producer (Atlas or Traefik) is still stamping them. Remove +// this diagnostic one release after the R8 cutover. +var legacyQuotaHeaderNames = []string{ + "X-Saturn-Service-Tier", + "X-Saturn-Rate-Limit-Requests", + "X-Saturn-Rate-Limit-Total-Prompt-Tokens", + "X-Saturn-Rate-Limit-Uncached-Prompt-Tokens", + "X-Saturn-Rate-Limit-Generated-Tokens", +} + +// legacyQuotaHeadersPresent reports whether the raw request carries any of the +// removed legacy quota headers. It is a logging diagnostic only. +func legacyQuotaHeadersPresent(h http.Header) bool { + for _, name := range legacyQuotaHeaderNames { + if h.Get(name) != "" { + return true + } + } + return false +} diff --git a/internal/proxy/admission_test.go b/internal/proxy/admission_test.go index 74e55e1..5bd96cb 100644 --- a/internal/proxy/admission_test.go +++ b/internal/proxy/admission_test.go @@ -990,7 +990,7 @@ func rateLimitsEqual(a, b admission.RateLimits) bool { // unlimited). R7: the completeness gate covers the IDENTITY anchor only — // X-Saturn-Owner-Id; limit headers without that anchor, a malformed present // value, and an absent policy all fail closed. R8: there is no legacy -// envelope (see TestLegacyOnlyQuotaHeadersFailClosed). +// envelope (see TestLegacyQuotaHeadersAreIgnored). func TestTrustedRateLimitPolicyParsing(t *testing.T) { // R7 structural pins — fail closed. for _, tc := range []struct { @@ -1130,29 +1130,18 @@ func TestTrustedRateLimitPolicyWithoutIdentityAnchorFailsClosed(t *testing.T) { } } -// R8 hard cut, end to end: the five legacy single-scope quota headers -// (X-Saturn-Service-Tier and X-Saturn-Rate-Limit-*) are no longer an -// envelope. A shared request under enabled admission that carries ONLY those -// five, with no X-Saturn-Owner-Id and no scoped header, has no trusted policy -// and fails closed (503) before reaching the upstream — the legacy values -// neither admit it nor bind any limit. The header names are spelled out -// literally because phoebe no longer defines constants for them. -func TestLegacyOnlyQuotaHeadersFailClosed(t *testing.T) { - var hits int - backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - hits++ - _, _ = w.Write([]byte(`{"model":"model-a","usage":{"prompt_tokens":1,"completion_tokens":1}}`)) - })) - defer backend.Close() - up, _ := url.Parse(backend.URL) - cfg := proxyAdmissionConfig(1) - mr := miniredis.RunT(t) - c := redis.NewClient(&redis.Options{Addr: mr.Addr()}) - t.Cleanup(func() { _ = c.Close() }) - s := New(&config.Settings{Admission: cfg}, logging.New(logging.ERROR), &recordingEmitter{}).WithAdmitter(admission.New(c, cfg)) +// legacyQuotaTestHeaders are the five single-scope quota headers R8 removed. +// They are spelled out literally because phoebe defines no trusted-header +// constants for them. +var legacyQuotaTestHeaders = []string{ + "X-Saturn-Service-Tier", + "X-Saturn-Rate-Limit-Requests", + "X-Saturn-Rate-Limit-Total-Prompt-Tokens", + "X-Saturn-Rate-Limit-Uncached-Prompt-Tokens", + "X-Saturn-Rate-Limit-Generated-Tokens", +} - req := sharedRequest(up) - req.Header.Del(identity.HeaderOwnerID) +func deleteScopedRateLimitHeaders(req *http.Request) { for _, header := range []string{ identity.HeaderOrgRateLimitRequests, identity.HeaderOrgRateLimitTotalPromptTokens, @@ -1165,24 +1154,169 @@ func TestLegacyOnlyQuotaHeadersFailClosed(t *testing.T) { } { req.Header.Del(header) } - req.Header.Set("X-Saturn-Service-Tier", "default") - req.Header.Set("X-Saturn-Rate-Limit-Requests", "1000000") - req.Header.Set("X-Saturn-Rate-Limit-Total-Prompt-Tokens", "1000000") - req.Header.Set("X-Saturn-Rate-Limit-Uncached-Prompt-Tokens", "1000000") - req.Header.Set("X-Saturn-Rate-Limit-Generated-Tokens", "1000000") +} - rr := httptest.NewRecorder() - s.Handler().ServeHTTP(rr, req) - if rr.Code != http.StatusServiceUnavailable { - t.Fatalf("status=%d, want 503 — legacy-only quota headers must not satisfy the policy gate", rr.Code) - } - if hits != 0 { - t.Fatalf("legacy-only request reached upstream %d times, want 0", hits) +// R8 hard cut, end to end: the five legacy single-scope quota headers +// (X-Saturn-Service-Tier and X-Saturn-Rate-Limit-*) are not an envelope. +// Legacy headers neither admit a request nor bind or break its limits: +// - legacy-only (no X-Saturn-Owner-Id, no scoped header) has no trusted +// policy and fails closed (503) before reaching the upstream; +// - with the owner-id anchor and no scoped limits, legacy zero caps are +// ignored and the request is unlimited and forwarded; +// - with the owner-id anchor and scoped limits, legacy zero or malformed +// values neither refuse the request nor change the parsed scoped limits. +// +// The last two cases are the ones that would fail if phoebe read the legacy +// headers; the first only shows that a missing anchor fails closed. +func TestLegacyQuotaHeadersAreIgnored(t *testing.T) { + newServer := func(t *testing.T) (*Server, *url.URL, *int) { + t.Helper() + hits := new(int) + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + *hits++ + _, _ = w.Write([]byte(`{"model":"model-a","usage":{"prompt_tokens":1,"completion_tokens":1}}`)) + })) + t.Cleanup(backend.Close) + up, _ := url.Parse(backend.URL) + cfg := proxyAdmissionConfig(1) + mr := miniredis.RunT(t) + c := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = c.Close() }) + s := New(&config.Settings{Admission: cfg}, logging.New(logging.ERROR), &recordingEmitter{}).WithAdmitter(admission.New(c, cfg)) + return s, up, hits } - // The parser must also see no policy at all: the legacy values are not - // part of the identity it reads. - if _, _, err := parseTrustedRateLimits(identity.FromRequest(req)); err == nil { - t.Fatal("parseTrustedRateLimits accepted a request carrying only the legacy headers") + + t.Run("legacy only without owner anchor fails closed", func(t *testing.T) { + s, up, hits := newServer(t) + req := sharedRequest(up) + req.Header.Del(identity.HeaderOwnerID) + deleteScopedRateLimitHeaders(req) + req.Header.Set("X-Saturn-Service-Tier", "default") + for _, header := range legacyQuotaTestHeaders[1:] { + req.Header.Set(header, "1000000") + } + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, req) + if rr.Code != http.StatusServiceUnavailable { + t.Fatalf("status=%d, want 503 — legacy-only quota headers must not satisfy the policy gate", rr.Code) + } + if *hits != 0 { + t.Fatalf("legacy-only request reached upstream %d times, want 0", *hits) + } + }) + + t.Run("owner anchor without scoped limits ignores legacy zero caps", func(t *testing.T) { + s, up, hits := newServer(t) + req := sharedRequest(up) + deleteScopedRateLimitHeaders(req) + req.Header.Set("X-Saturn-Service-Tier", "default") + for _, header := range legacyQuotaTestHeaders[1:] { + req.Header.Set(header, "0") + } + organization, owner, err := parseTrustedRateLimits(identity.FromRequest(req)) + if err != nil { + t.Fatalf("parseTrustedRateLimits: %v — legacy headers must not affect an anchored request", err) + } + for name, limits := range map[string]admission.RateLimits{"organization": organization, "owner": owner} { + if limits.Requests != nil || limits.TotalPromptTokens != nil || limits.UncachedPromptTokens != nil || limits.GeneratedTokens != nil { + t.Fatalf("%s limits=%+v, want all unlimited (nil) — legacy zero caps were read", name, limits) + } + } + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("status=%d, want 200 — legacy zero caps must not block the request", rr.Code) + } + if *hits != 1 { + t.Fatalf("upstream hits=%d, want 1", *hits) + } + }) + + t.Run("owner anchor with scoped limits ignores legacy zero and malformed values", func(t *testing.T) { + s, up, hits := newServer(t) + req := sharedRequest(up) + req.Header.Set("X-Saturn-Service-Tier", "gold") + req.Header.Set("X-Saturn-Rate-Limit-Requests", "0") + req.Header.Set("X-Saturn-Rate-Limit-Total-Prompt-Tokens", "0") + req.Header.Set("X-Saturn-Rate-Limit-Uncached-Prompt-Tokens", "-1") + req.Header.Set("X-Saturn-Rate-Limit-Generated-Tokens", "not-a-number") + organization, owner, err := parseTrustedRateLimits(identity.FromRequest(req)) + if err != nil { + t.Fatalf("parseTrustedRateLimits: %v — malformed legacy headers must not break an anchored request", err) + } + for name, limits := range map[string]admission.RateLimits{"organization": organization, "owner": owner} { + for field, got := range map[string]*int64{ + "Requests": limits.Requests, "TotalPromptTokens": limits.TotalPromptTokens, + "UncachedPromptTokens": limits.UncachedPromptTokens, "GeneratedTokens": limits.GeneratedTokens, + } { + if got == nil || *got != 1000000 { + t.Fatalf("%s %s=%v, want the scoped 1000000 — legacy values leaked into the parsed limits", name, field, got) + } + } + } + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("status=%d, want 200 — legacy zero/malformed values must not refuse the request", rr.Code) + } + if *hits != 1 { + t.Fatalf("upstream hits=%d, want 1", *hits) + } + }) +} + +// R8 cutover diagnostic: a legacy-only request still gets the unchanged, +// opaque 503, but the log names the likely root cause (a producer that still +// stamps only the removed legacy headers). A request with no policy and no +// legacy headers must not carry the marker. The legacy headers only ever +// reach a log line, never a trust or limit decision. +func TestLegacyOnlyQuotaHeadersLogPreR8ProducerMarker(t *testing.T) { + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"model":"model-a","usage":{"prompt_tokens":1,"completion_tokens":1}}`)) + })) + defer backend.Close() + up, _ := url.Parse(backend.URL) + cfg := proxyAdmissionConfig(1) + mr := miniredis.RunT(t) + c := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = c.Close() }) + + for _, tc := range []struct { + name string + legacy bool + wantMarker bool + }{ + {name: "legacy headers present", legacy: true, wantMarker: true}, + {name: "no policy at all", legacy: false, wantMarker: false}, + } { + t.Run(tc.name, func(t *testing.T) { + var warnBuf, errBuf bytes.Buffer + logger := &logging.Logger{Warn: log.New(&warnBuf, "", 0), Error: log.New(&errBuf, "", 0)} + s := New(&config.Settings{Admission: cfg}, logger, &recordingEmitter{}).WithAdmitter(admission.New(c, cfg)) + req := sharedRequest(up) + req.Header.Del(identity.HeaderOwnerID) + deleteScopedRateLimitHeaders(req) + if tc.legacy { + req.Header.Set("X-Saturn-Service-Tier", "default") + for _, header := range legacyQuotaTestHeaders[1:] { + req.Header.Set(header, "1000000") + } + } + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, req) + if rr.Code != http.StatusServiceUnavailable { + t.Fatalf("status=%d, want 503", rr.Code) + } + if body := strings.TrimSpace(rr.Body.String()); body != "shared inference policy unavailable" { + t.Fatalf("response body=%q, want the unchanged opaque body", body) + } + if !strings.Contains(errBuf.String(), "incomplete trusted shared-inference rate-limit policy") { + t.Fatalf("error log=%q, want the policy error", errBuf.String()) + } + if got := strings.Contains(warnBuf.String(), "legacy_envelope_present=true"); got != tc.wantMarker { + t.Fatalf("legacy marker in warn log=%v, want %v; warn log=%q", got, tc.wantMarker, warnBuf.String()) + } + }) } } @@ -2231,6 +2365,44 @@ func TestProxyForwardsOperatorAdmissionLaneDynamoHints(t *testing.T) { } } +// Only operator config selects a lane: an org with no OrganizationLanes entry +// stays on the default lane even when a higher-priority lane is configured, +// another org is mapped to it, and the client asks for that priority. +func TestUnmappedOrgStaysOnDefaultLaneDynamoHints(t *testing.T) { + seen := make(chan *http.Request, 1) + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + seen <- r.Clone(r.Context()) + _, _ = w.Write([]byte(`{"model":"model-a","usage":{"prompt_tokens":1,"completion_tokens":1}}`)) + })) + defer backend.Close() + up, _ := url.Parse(backend.URL) + mr := miniredis.RunT(t) + cfg := proxyAdmissionConfig(2) + cfg.Lanes = map[string]config.AdmissionLane{ + "default": {Weight: 1}, + "gold": {Weight: 1, DynamoPriority: 11, DynamoStrictPriority: 4}, + } + cfg.OrganizationLanes = map[string]string{"org-z": "gold"} + c := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = c.Close() }) + s := New(&config.Settings{Admission: cfg}, logging.New(logging.ERROR), &recordingEmitter{}).WithAdmitter(admission.New(c, cfg)) + req := sharedRequest(up) // org-a, full scoped envelope + req.Header.Set("X-Dynamo-Request-Priority", "11") + req.Header.Set("X-Dynamo-Request-Strict-Priority", "4") + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("status=%d", rr.Code) + } + forwarded := <-seen + if got := forwarded.Header.Get("X-Dynamo-Request-Priority"); got != "0" { + t.Fatalf("forwarded priority header=%q, want 0 — an unmapped org must not get the gold lane's 11", got) + } + if got := forwarded.Header.Get("X-Dynamo-Request-Strict-Priority"); got != "0" { + t.Fatalf("forwarded strict-priority header=%q, want 0 — an unmapped org must not get the gold lane's 4", got) + } +} + // The per-request admission error logs must aggregate, not flood: the first // occurrence logs at onset, then every 100th, each carrying the number of // suppressed occurrences since the previous line. diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index f93e4aa..7ccd220 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -623,6 +623,11 @@ func (s *Server) handleProxy(w http.ResponseWriter, r *http.Request) { organizationLimits, ownerLimits, policyErr := parseTrustedRateLimits(id) if policyErr != nil { s.log.Error.Printf("admission: invalid trusted rate-limit policy: %v", policyErr) + if errors.Is(policyErr, errNoTrustedRateLimitPolicy) && legacyQuotaHeadersPresent(r.Header) { + // Log-only diagnostic for the R8 cutover: the legacy headers + // decide nothing, but their presence names the root cause. + s.log.Warn.Printf("admission: legacy single-scope quota headers present without %s (removed by R8); legacy_envelope_present=true (pre-R8 producer; upgrade Atlas/Traefik) request_id=%s", identity.HeaderOwnerID, requestID) + } http.Error(w, "shared inference policy unavailable", http.StatusServiceUnavailable) return }