From efddbe4d7e152ac51a3d8689607a03d7d4480ab2 Mon Sep 17 00:00:00 2001 From: Shreyas Kalyan Date: Thu, 3 Sep 2026 14:34:01 -0400 Subject: [PATCH] l1: default-route unresolvable S3 backend selectors instead of rejecting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The backend-selector trust interceptor enforced the forwarded (x-blacksmith-s3-endpoint, x-blacksmith-s3-bucket) pair fail-closed: anything not in the node's backends map got an InvalidArgument trust rejection. That made every config race build-visible, and let one stale host-level env value (the FA boot probe presented Doppler's MINIO_ENDPOINT as its identity) knock whole hosts off the L1 ring into direct-S3 fallback — dialing that same stale value. 2026-09-03: a legacy us-west MinIO teardown left 134 hosts in s3_fallback with no working cache path and 450k+ selector rejections in 24h. Resolution now never rejects, mirroring the actions cache's shard router: a missing/duplicate/unknown selector routes to the map's designated default backend, and a bucket unusable for the resolved entry uses that entry's default bucket. The L1 owns which backends exist; upstream needs no endpoint knowledge of its own to be served. Tenant isolation is unchanged — the storage-prefix interceptor stays fail-closed; the selector only ever picked the shard. Every defaulted resolution is metered (bazel_remote_s3_backend_selector_defaulted_total{reason}) and rate-limit logged: nonzero during a rollout = FA/L1 map drift; sustained nonzero = a pin serving from a shard it wasn't allocated on. RejectionReasonS3BackendSelector stays defined: L1s predating this change still mint it and clients must keep degrading it to a miss. Co-authored-by: Cursor --- cache/s3proxy/multi.go | 11 +- cache/trust_rejection.go | 8 +- config/s3.go | 44 ++++- config/s3_test.go | 38 ++-- main.go | 36 ++-- server/grpc_s3_backend.go | 184 +++++++++++------- server/grpc_s3_backend_test.go | 290 +++++++++++++---------------- server/grpc_storage_prefix.go | 5 +- server/grpc_storage_prefix_test.go | 30 +++ server/grpc_tenant_metadata.go | 34 ++-- 10 files changed, 387 insertions(+), 293 deletions(-) diff --git a/cache/s3proxy/multi.go b/cache/s3proxy/multi.go index 9fba1cf..4618da3 100644 --- a/cache/s3proxy/multi.go +++ b/cache/s3proxy/multi.go @@ -27,11 +27,12 @@ import ( var ( // backendUnknown is belt-and-braces coverage for interceptor/config - // drift and stays flat in practice: the gRPC interceptor rejects - // unknown selectors at the boundary, and the HTTP listener carries no - // selector at all. Do NOT alert on it — the interceptor rejection - // counters (bazel_remote_s3_backend_selector_rejected_total{reason}) - // are the signal. + // drift and stays flat in practice: the gRPC interceptor resolves every + // selector to a configured key (unresolvable ones route to the default + // backend), and the HTTP listener carries no selector at all. Do NOT + // alert on it — the interceptor's defaulting counter + // (bazel_remote_s3_backend_selector_defaulted_total{reason}) is the + // drift signal. backendUnknown = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "bazel_remote_s3_backend_unknown_total", Help: "Requests carrying a backend selector not present in the configured backends map (refused; indicates an interceptor/config mismatch).", diff --git a/cache/trust_rejection.go b/cache/trust_rejection.go index 02faa2b..048ecc6 100644 --- a/cache/trust_rejection.go +++ b/cache/trust_rejection.go @@ -22,9 +22,13 @@ const TrustRejectionErrorDomain = "cache.blacksmith.sh" // cause (missing/duplicate/unknown/invalid) travels in ErrorInfo.Metadata // under "cause". const ( - // RejectionReasonS3BackendSelector marks rejections from the S3 + // RejectionReasonS3BackendSelector marked rejections from the S3 // backend-selector trust interceptor (missing, duplicate, or - // non-allowlisted x-blacksmith-s3-endpoint metadata). + // non-allowlisted x-blacksmith-s3-endpoint metadata). Current L1s no + // longer mint it — unresolvable selectors route to the default backend + // instead of rejecting — but the constant stays: L1s predating the + // change still send it, and upstream clients must keep degrading it to + // a metered miss. RejectionReasonS3BackendSelector = "S3_BACKEND_SELECTOR_REJECTED" // RejectionReasonStoragePrefix marks rejections from the storage-prefix diff --git a/config/s3.go b/config/s3.go index 73cabb8..3ef98c0 100644 --- a/config/s3.go +++ b/config/s3.go @@ -140,20 +140,46 @@ type S3BackendConfig struct { Default bool `yaml:"default"` } -// AllowedBackends returns, for each valid backend selector, the bucket set -// the fail-closed gRPC interceptor accepts for it: the entry's default -// bucket plus its extra_buckets. Only valid to call after validateConfig has -// passed (bucket resolution cannot fail then). -func (s3c *S3CloudStorageConfig) AllowedBackends() (map[string]map[string]bool, error) { - allowed := make(map[string]map[string]bool, len(s3c.Backends)) +// S3BackendRoutingEntry is one backends-map entry as the gRPC routing +// interceptor needs it: the entry's resolved default bucket (used when a +// request carries no usable bucket) and the full allowed set (default plus +// extra_buckets). +type S3BackendRoutingEntry struct { + DefaultBucket string + Buckets map[string]bool +} + +// RoutingBackends resolves the backends map for the gRPC routing +// interceptor: for each selector key, its default bucket and allowed bucket +// set; plus the map's designated default key, which serves every request +// whose selector is missing or not in the map. Exactly one entry must be +// marked default (the same invariant s3proxy.NewMulti enforces). Only valid +// to call after validateConfig has passed (bucket resolution cannot fail +// then). +func (s3c *S3CloudStorageConfig) RoutingBackends() (defaultKey string, entries map[string]S3BackendRoutingEntry, err error) { + entries = make(map[string]S3BackendRoutingEntry, len(s3c.Backends)) for key := range s3c.Backends { buckets, err := s3c.allowedBucketsForBackend(key) if err != nil { - return nil, err + return "", nil, err } - allowed[key] = buckets + backend := s3c.Backends[key] + bucket := backend.Bucket + if bucket == "" { + bucket = s3c.Bucket + } + entries[key] = S3BackendRoutingEntry{DefaultBucket: bucket, Buckets: buckets} + if backend.Default { + if defaultKey != "" { + return "", nil, fmt.Errorf("multiple s3.backends entries marked as default") + } + defaultKey = key + } + } + if defaultKey == "" { + return "", nil, fmt.Errorf("no s3.backends entry marked as default") } - return allowed, nil + return defaultKey, entries, nil } // allowedBucketsForBackend resolves one backends-map entry's allowed bucket diff --git a/config/s3_test.go b/config/s3_test.go index 7840e46..6091caa 100644 --- a/config/s3_test.go +++ b/config/s3_test.go @@ -48,27 +48,37 @@ func TestS3BackendsMapConfig(t *testing.T) { t.Fatalf("ConnRecycleInterval = %v, want 1m", s3.ConnRecycleInterval) } - // The allowlist pairs each selector with its bucket set: the entry's - // (possibly inherited) default bucket plus any extra_buckets. - allowed, err := s3.AllowedBackends() + // The routing table pairs each selector with its default bucket and + // bucket set (the entry's — possibly inherited — default bucket plus any + // extra_buckets), and names the map's default key. + defaultKey, entries, err := s3.RoutingBackends() if err != nil { t.Fatal(err) } - wantBuckets := map[string][]string{ - "http://minio-a.example.com:9000": {"shared-bucket"}, - "https://minio-b.example.com:9000": {"bucket-b", "bucket-b-pre-rename"}, + if defaultKey != "http://minio-a.example.com:9000" { + t.Fatalf("default key = %q, want backend a", defaultKey) } - for key, buckets := range wantBuckets { - got, ok := allowed[key] + want := map[string]struct { + defaultBucket string + buckets []string + }{ + "http://minio-a.example.com:9000": {"shared-bucket", []string{"shared-bucket"}}, + "https://minio-b.example.com:9000": {"bucket-b", []string{"bucket-b", "bucket-b-pre-rename"}}, + } + for key, w := range want { + got, ok := entries[key] if !ok { - t.Fatalf("expected %q in allowed backends %v", key, allowed) + t.Fatalf("expected %q in routing backends %v", key, entries) + } + if got.DefaultBucket != w.defaultBucket { + t.Fatalf("backend %q default bucket = %q, want %q", key, got.DefaultBucket, w.defaultBucket) } - if len(got) != len(buckets) { - t.Fatalf("backend %q allowed buckets = %v, want %v", key, got, buckets) + if len(got.Buckets) != len(w.buckets) { + t.Fatalf("backend %q allowed buckets = %v, want %v", key, got.Buckets, w.buckets) } - for _, bucket := range buckets { - if !got[bucket] { - t.Fatalf("backend %q allowed buckets = %v, missing %q", key, got, bucket) + for _, bucket := range w.buckets { + if !got.Buckets[bucket] { + t.Fatalf("backend %q allowed buckets = %v, missing %q", key, got.Buckets, bucket) } } } diff --git a/main.go b/main.go index b7f2c30..dc31b7f 100644 --- a/main.go +++ b/main.go @@ -497,21 +497,35 @@ func startGrpcServer(c *config.Config, grpcServer **grpc.Server, unaryInterceptors = append(unaryInterceptors, server.GRPCStoragePrefixUnaryServerInterceptor(authSecret)) } - // Multi-backend S3 mode: every cache RPC must carry exactly one - // allowlisted (endpoint, bucket) pair (the tenant's pinned backing-store - // endpoint and bucket, forwarded by the trusted upstream) or it is - // rejected fail-closed — same trust model as the storage prefix above. - // Health and capabilities RPCs are exempt. Only installed when a - // backends map is configured; single-backend deployments ignore both - // metadata keys. + // Multi-backend S3 mode: cache RPCs carry the tenant's pinned + // backing-store (endpoint, bucket) pair as gRPC metadata, forwarded by + // the trusted upstream. Requests resolve against the backends map; + // anything unresolvable (missing/unknown selector or bucket) routes to + // the map's default backend / the entry's default bucket, metered per + // cause — the L1 owns which backends exist, the upstream needs no + // endpoint knowledge of its own to be served. Tenant isolation is the + // storage-prefix interceptor's job (fail-closed above); the selector + // only picks the shard. Health and capabilities RPCs are exempt. Only + // installed when a backends map is configured; single-backend + // deployments ignore both metadata keys. if c.S3CloudStorage != nil && len(c.S3CloudStorage.Backends) > 0 { - allowed, err := c.S3CloudStorage.AllowedBackends() + defaultKey, entries, err := c.S3CloudStorage.RoutingBackends() if err != nil { return err } - log.Printf("Routing S3 operations by forwarded (endpoint, bucket) gRPC metadata, fail-closed (%d allowlisted backends)", len(allowed)) - streamInterceptors = append(streamInterceptors, server.GRPCS3BackendStreamServerInterceptor(allowed)) - unaryInterceptors = append(unaryInterceptors, server.GRPCS3BackendUnaryServerInterceptor(allowed)) + routing := server.S3BackendRouting{ + DefaultKey: defaultKey, + Backends: make(map[string]server.S3BackendRoutingEntry, len(entries)), + } + for key, entry := range entries { + routing.Backends[key] = server.S3BackendRoutingEntry{ + DefaultBucket: entry.DefaultBucket, + Buckets: entry.Buckets, + } + } + log.Printf("Routing S3 operations by forwarded (endpoint, bucket) gRPC metadata (%d backends, default %q; unresolvable selectors route to the default)", len(entries), defaultKey) + streamInterceptors = append(streamInterceptors, server.GRPCS3BackendStreamServerInterceptor(routing)) + unaryInterceptors = append(unaryInterceptors, server.GRPCS3BackendUnaryServerInterceptor(routing)) } if c.TLSConfig != nil { diff --git a/server/grpc_s3_backend.go b/server/grpc_s3_backend.go index 9c3a6b4..f7a8b61 100644 --- a/server/grpc_s3_backend.go +++ b/server/grpc_s3_backend.go @@ -2,6 +2,7 @@ package server import ( "context" + "fmt" "github.com/buchgr/bazel-remote/v2/cache" @@ -11,109 +12,148 @@ import ( "google.golang.org/grpc/metadata" ) -// S3-backend trust interceptors: when an L1 bazel-remote node is configured +// S3-backend routing interceptors: when an L1 bazel-remote node is configured // with a map of allowlisted S3 backends (multi-shard MinIO), the trusted // upstream (FA host) forwards each tenant's pinned backing-store endpoint and // bucket as gRPC metadata (cache.S3BackendGRPCMetadataKey and -// cache.S3BucketGRPCMetadataKey). These interceptors validate the forwarded -// (endpoint, bucket) pair against the allowlist and lift it onto the request +// cache.S3BucketGRPCMetadataKey). These interceptors resolve the forwarded +// pair against the map and lift the resolved selection onto the request // context, from which the s3proxy routes reads and write-through to the // matching MinIO cluster and bucket. // -// The trust model mirrors the storage-prefix interceptors (fail-closed): -// when a backends map is configured, every cache RPC must carry exactly one -// allowlisted endpoint and exactly one bucket from that endpoint's allowed -// set, or it is rejected at the boundary — a request routed to a guessed -// backend or bucket would read or write another shard's keyspace. Health -// and capabilities RPCs are exempt, exactly like the storage-prefix contract. -// These interceptors are only installed in multi-backend mode; single-backend -// deployments ignore both metadata keys entirely (backward compatible). +// Resolution never rejects: a request whose selector is missing or not in +// the map routes to the map's designated DEFAULT backend, and a request +// whose bucket is missing or not in the resolved entry's allowed set uses +// that entry's default bucket — mirroring the actions cache's shard router, +// where placement values are backend-authored hints and anything unresolvable +// falls back to the default client. The L1 owns which MinIO backends exist +// (its config is the map); the upstream does not need any endpoint knowledge +// of its own to be served. Every defaulted resolution is metered by cause +// (s3BackendSelectorDefaulted) and rate-limit logged: a nonzero rate during +// a backends-map rollout is the config-drift signal (FA forwarding values an +// L1's map does not — or does not yet — contain), and sustained defaulting +// means a namespace's pin and the map have diverged, which quietly serves +// that tenant from the default shard (cold reads) until reconciled. // -// Rejections carry the cache.TrustRejectionErrorDomain ErrorInfo marker so -// the upstream degrades them to metered misses instead of failed builds (see -// trustRejection). +// History: these interceptors originally enforced the pair fail-closed +// (InvalidArgument trust rejections). That turned every config race into a +// build-visible event, and — because the FA boot probe presented the host's +// Doppler MINIO_ENDPOINT as its identity — let one stale host-level env +// value knock whole hosts off the L1 ring into direct-S3 fallback (dialing +// that same stale value). See the 2026-09-03 us-west legacy-cluster +// teardown: 134 hosts in s3_fallback with no working cache path. +// Tenant isolation does not depend on this check — the storage-prefix +// interceptor (still fail-closed) scopes every request's keyspace; the +// selector only picks which shard serves it. +// +// Health and capabilities RPCs are exempt, exactly like the storage-prefix +// contract. These interceptors are only installed in multi-backend mode; +// single-backend deployments ignore both metadata keys entirely (backward +// compatible). + +// S3BackendRouting is the routing table the interceptors resolve against: +// every configured selector key with its default bucket and allowed bucket +// set, plus the designated default key for requests that carry no usable +// selector (built from config.S3CloudStorageConfig.RoutingBackends). +type S3BackendRouting struct { + DefaultKey string + Backends map[string]S3BackendRoutingEntry +} + +// S3BackendRoutingEntry mirrors config.S3BackendRoutingEntry (the server +// package deliberately does not import config). +type S3BackendRoutingEntry struct { + DefaultBucket string + Buckets map[string]bool +} -// s3BackendSelectorRejected meters the fail-closed rejections above, by -// cause. Nonzero missing/unknown (or their bucket_* counterparts) while -// rolling out a backends-map change is the config-race signal: FA forwarding -// pairs an L1's allowlist does not (yet) contain, or not forwarding them at -// all (version skew). -var s3BackendSelectorRejected = promauto.NewCounterVec(prometheus.CounterOpts{ - Name: "bazel_remote_s3_backend_selector_rejected_total", - Help: "Cache RPCs rejected by the S3 backend-selector trust interceptor, by cause (missing/duplicate/unknown and bucket_missing/bucket_duplicate/bucket_unknown). Nonzero during a backends-map rollout indicates FA/L1 config-version skew.", +// s3BackendSelectorDefaulted meters resolutions that fell back to a default, +// by cause. missing/duplicate/unknown mean the selector itself was unusable +// (routed to the default backend); bucket_missing/bucket_duplicate/ +// bucket_unknown mean the bucket was unusable for the resolved entry (its +// default bucket was used). Nonzero during a backends-map rollout indicates +// FA/L1 config drift; sustained nonzero means a pinned namespace is being +// served from a default it was not allocated on — reconcile the map or the +// pin. +var s3BackendSelectorDefaulted = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "bazel_remote_s3_backend_selector_defaulted_total", + Help: "Cache RPCs whose forwarded S3 backend selector or bucket could not be resolved and fell back to a default, by cause (missing/duplicate/unknown route to the default backend; bucket_* use the resolved entry's default bucket). Nonzero indicates FA/L1 backends-map drift.", }, []string{"reason"}) -func s3BackendFromIncomingContext(ctx context.Context, allowed map[string]map[string]bool) (context.Context, error) { +// s3BackendFromIncomingContext resolves the forwarded (endpoint, bucket) +// pair against the routing table and lifts the resolved selection onto the +// context. Resolution never fails — see the package comment for the +// defaulting contract. +func s3BackendFromIncomingContext(ctx context.Context, routing S3BackendRouting) context.Context { md, _ := metadata.FromIncomingContext(ctx) - selector, cause := singleMetadataValue(md, cache.S3BackendGRPCMetadataKey) + + key, cause := singleMetadataValue(md, cache.S3BackendGRPCMetadataKey) + if cause == "" { + // Exact opaque string match against the configured backends map + // keys — no URL normalization. Web pins namespaces by these exact + // strings; anything else is drift and routes to the default. + if _, ok := routing.Backends[key]; !ok { + cause = "unknown" + } + } if cause != "" { - s3BackendSelectorRejected.WithLabelValues(cause).Inc() - return ctx, trustRejection(cache.RejectionReasonS3BackendSelector, cause, - "%s %s metadata", cause, cache.S3BackendGRPCMetadataKey) + s3BackendSelectorDefaulted.WithLabelValues(cause).Inc() + logRateLimited("s3_backend_defaulted/"+cause, + "S3 backend selector unresolvable (%s %s metadata%s); routing to the default backend %q", + cause, cache.S3BackendGRPCMetadataKey, selectorDetail(cause, key), routing.DefaultKey) + key = routing.DefaultKey } - // Exact opaque string match against the configured backends map keys — - // no URL normalization. The upstream must forward the adoption payload's - // bazelre_cache_endpoint verbatim, and the map must be keyed by the same - // strings. - buckets, ok := allowed[selector] - if !ok { - s3BackendSelectorRejected.WithLabelValues("unknown").Inc() - return ctx, trustRejection(cache.RejectionReasonS3BackendSelector, "unknown", - "unknown %s metadata value %q", cache.S3BackendGRPCMetadataKey, selector) + entry := routing.Backends[key] + + bucket, bucketCause := singleMetadataValue(md, cache.S3BucketGRPCMetadataKey) + if bucketCause == "" && !entry.Buckets[bucket] { + // Byte-verbatim match against the resolved entry's allowed set + // (default bucket plus extra_buckets) — the pair is resolved + // together: a bucket allowed on another endpoint defaults here. + bucketCause = "unknown" } - // The bucket half of the pair: same exactly-one extraction rule, causes - // namespaced bucket_* under the same rejection marker. Web snapshots the - // bucket per namespace at allocation, so two tenants on one endpoint can - // live in different buckets; validating only the endpoint would let a - // stale or mistaken bucket read another tenant's objects. - bucket, cause := singleMetadataValue(md, cache.S3BucketGRPCMetadataKey) - if cause != "" { - cause = "bucket_" + cause - s3BackendSelectorRejected.WithLabelValues(cause).Inc() - return ctx, trustRejection(cache.RejectionReasonS3BackendSelector, cause, - "%s %s metadata", cause, cache.S3BucketGRPCMetadataKey) + if bucketCause != "" { + cause := "bucket_" + bucketCause + s3BackendSelectorDefaulted.WithLabelValues(cause).Inc() + logRateLimited("s3_backend_defaulted/"+cause, + "S3 bucket unresolvable for backend %q (%s %s metadata%s); using the entry's default bucket %q", + key, bucketCause, cache.S3BucketGRPCMetadataKey, selectorDetail(bucketCause, bucket), entry.DefaultBucket) + bucket = entry.DefaultBucket } - // Byte-verbatim match against the matched entry's allowed bucket set - // (default bucket plus extra_buckets) — the pair is validated, not the - // halves independently: a bucket allowed on another endpoint is rejected. - if !buckets[bucket] { - s3BackendSelectorRejected.WithLabelValues("bucket_unknown").Inc() - return ctx, trustRejection(cache.RejectionReasonS3BackendSelector, "bucket_unknown", - "unknown %s metadata value %q for backend %q", cache.S3BucketGRPCMetadataKey, bucket, selector) + + return cache.WithS3Backend(ctx, cache.S3BackendSelection{Endpoint: key, Bucket: bucket}) +} + +// selectorDetail renders the offending value for the rate-limited log line — +// only for the "unknown" cause, where a concrete value exists to show. +func selectorDetail(cause, value string) string { + if cause != "unknown" { + return "" } - return cache.WithS3Backend(ctx, cache.S3BackendSelection{Endpoint: selector, Bucket: bucket}), nil + return fmt.Sprintf(" value %q", value) } // GRPCS3BackendUnaryServerInterceptor returns a unary interceptor that -// enforces the fail-closed backend-selection contract and lifts the forwarded -// (endpoint, bucket) pair onto the context. allowed maps each configured -// backends-map key (tenant-facing endpoint URL) to its allowed bucket set -// (config.S3CloudStorageConfig.AllowedBackends). -func GRPCS3BackendUnaryServerInterceptor(allowed map[string]map[string]bool) grpc.UnaryServerInterceptor { +// resolves the forwarded (endpoint, bucket) pair per the defaulting contract +// and lifts the resolved selection onto the context. +func GRPCS3BackendUnaryServerInterceptor(routing S3BackendRouting) grpc.UnaryServerInterceptor { return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { if exemptFromTenantMetadata(info.FullMethod) { return handler(ctx, req) } - ctx, err := s3BackendFromIncomingContext(ctx, allowed) - if err != nil { - return nil, err - } - return handler(ctx, req) + return handler(s3BackendFromIncomingContext(ctx, routing), req) } } // GRPCS3BackendStreamServerInterceptor returns a stream interceptor that -// enforces the fail-closed backend-selection contract and lifts the forwarded -// (endpoint, bucket) pair onto the context. -func GRPCS3BackendStreamServerInterceptor(allowed map[string]map[string]bool) grpc.StreamServerInterceptor { +// resolves the forwarded (endpoint, bucket) pair per the defaulting contract +// and lifts the resolved selection onto the context. +func GRPCS3BackendStreamServerInterceptor(routing S3BackendRouting) grpc.StreamServerInterceptor { return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { if exemptFromTenantMetadata(info.FullMethod) { return handler(srv, ss) } - ctx, err := s3BackendFromIncomingContext(ss.Context(), allowed) - if err != nil { - return err - } + ctx := s3BackendFromIncomingContext(ss.Context(), routing) return handler(srv, &tenantMetadataServerStream{ServerStream: ss, ctx: ctx}) } } diff --git a/server/grpc_s3_backend_test.go b/server/grpc_s3_backend_test.go index 8ba0f64..c90018e 100644 --- a/server/grpc_s3_backend_test.go +++ b/server/grpc_s3_backend_test.go @@ -6,11 +6,8 @@ import ( "github.com/buchgr/bazel-remote/v2/cache" - "google.golang.org/genproto/googleapis/rpc/errdetails" "google.golang.org/grpc" - "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" - "google.golang.org/grpc/status" ) const ( @@ -22,14 +19,17 @@ const ( bucketB = "bazel-cache-b" ) -// allowedBackends mirrors a two-entry backends map: backend A allows its -// default bucket plus a pre-rename legacy bucket, backend B allows only its -// default. The asymmetry lets tests prove the PAIR is validated, not the -// halves independently. -func allowedBackends() map[string]map[string]bool { - return map[string]map[string]bool{ - backendA: {bucketA: true, bucketALegacy: true}, - backendB: {bucketB: true}, +// testRouting mirrors a two-entry backends map with backend A as the default: +// A allows its default bucket plus a pre-rename legacy bucket, B allows only +// its default. The asymmetry lets tests prove the PAIR is resolved together, +// not the halves independently. +func testRouting() S3BackendRouting { + return S3BackendRouting{ + DefaultKey: backendA, + Backends: map[string]S3BackendRoutingEntry{ + backendA: {DefaultBucket: bucketA, Buckets: map[string]bool{bucketA: true, bucketALegacy: true}}, + backendB: {DefaultBucket: bucketB, Buckets: map[string]bool{bucketB: true}}, + }, } } @@ -42,184 +42,173 @@ func pairMD(endpoint, bucket string) metadata.MD { ) } -// requireTrustRejection asserts that err is an InvalidArgument rejection -// carrying the typed ErrorInfo marker our trust interceptors mint — the -// wire contract the upstream grpcproxy uses to degrade config-race -// rejections to metered misses instead of failing builds. -func requireTrustRejection(t *testing.T, err error, reason, cause string) { +// requireSelection asserts the resolved (endpoint, bucket) selection lifted +// onto the context. Resolution never fails — the contract under test is +// WHICH backend and bucket an input routes to, not whether it is served. +func requireSelection(t *testing.T, ctx context.Context, endpoint, bucket string) { t.Helper() - s, ok := status.FromError(err) - if !ok || s.Code() != codes.InvalidArgument { - t.Fatalf("expected InvalidArgument status, got %v", err) + selection, ok := cache.S3BackendFromContext(ctx) + if !ok { + t.Fatal("no S3 backend selection on context") } - for _, detail := range s.Details() { - info, ok := detail.(*errdetails.ErrorInfo) - if !ok { - continue - } - if info.GetDomain() != cache.TrustRejectionErrorDomain { - t.Fatalf("ErrorInfo domain = %q, want %q", info.GetDomain(), cache.TrustRejectionErrorDomain) - } - if info.GetReason() != reason { - t.Fatalf("ErrorInfo reason = %q, want %q", info.GetReason(), reason) - } - if got := info.GetMetadata()["cause"]; got != cause { - t.Fatalf("ErrorInfo cause = %q, want %q", got, cause) - } - return + if selection.Endpoint != endpoint || selection.Bucket != bucket { + t.Fatalf("selection = %+v, want (%s, %s)", selection, endpoint, bucket) } - t.Fatalf("rejection %v carries no ErrorInfo trust marker", err) } func TestS3BackendFromIncomingContext(t *testing.T) { - t.Run("no metadata is rejected (fail-closed)", func(t *testing.T) { - _, err := s3BackendFromIncomingContext(context.Background(), allowedBackends()) - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "missing") + t.Run("no metadata routes to the default backend and bucket", func(t *testing.T) { + ctx := s3BackendFromIncomingContext(context.Background(), testRouting()) + requireSelection(t, ctx, backendA, bucketA) }) - t.Run("missing selector key is rejected", func(t *testing.T) { + t.Run("missing selector key routes to the default backend", func(t *testing.T) { md := metadata.Pairs("some-other-key", "value") - _, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "missing") + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + requireSelection(t, ctx, backendA, bucketA) }) - t.Run("duplicate selector values are rejected", func(t *testing.T) { + t.Run("duplicate selector values route to the default backend", func(t *testing.T) { + // Two different forwarded values is drift, not a choosable input: + // picking either would guess. The default is the deterministic + // resolution. md := metadata.Pairs( cache.S3BackendGRPCMetadataKey, backendA, cache.S3BackendGRPCMetadataKey, backendB, cache.S3BucketGRPCMetadataKey, bucketA, ) - _, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "duplicate") + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + requireSelection(t, ctx, backendA, bucketA) }) - t.Run("unknown selector is rejected", func(t *testing.T) { - md := pairMD("http://rogue.example.com:9000", bucketA) - _, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "unknown") + t.Run("unknown selector routes to the default backend", func(t *testing.T) { + // The 2026-09-03 shape: a host or pin still naming a torn-down + // cluster. The L1 owns which backends exist; the request is served + // from the default instead of being rejected. + md := pairMD("http://minio.torn-down.example:9000", bucketA) + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + requireSelection(t, ctx, backendA, bucketA) }) - t.Run("no URL normalization: near-miss selectors are rejected", func(t *testing.T) { - // The contract is exact opaque string match against the configured - // map keys; anything that would need normalizing to match must fail. + t.Run("no URL normalization: near-miss selectors route to the default", func(t *testing.T) { + // Resolution is exact opaque string match against the configured + // map keys; anything that would need normalizing to match is + // treated as unknown and defaults. Near-misses of backend B prove + // they do NOT resolve to B. for _, nearMiss := range []string{ - "http://minio-a.example.com:9000/", // trailing slash - "minio-a.example.com:9000", // missing scheme - "HTTP://minio-a.example.com:9000", // case difference + "https://minio-b.example.com:9000/", // trailing slash + "minio-b.example.com:9000", // missing scheme + "HTTPS://minio-b.example.com:9000", // case difference } { - md := pairMD(nearMiss, bucketA) - _, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "unknown") + md := pairMD(nearMiss, bucketB) + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + // bucketB is not allowed on default backend A, so the bucket + // defaults too. + requireSelection(t, ctx, backendA, bucketA) } }) - t.Run("missing bucket is rejected (fail-closed)", func(t *testing.T) { - md := metadata.Pairs(cache.S3BackendGRPCMetadataKey, backendA) - _, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "bucket_missing") + t.Run("missing bucket uses the entry's default bucket", func(t *testing.T) { + md := metadata.Pairs(cache.S3BackendGRPCMetadataKey, backendB) + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + requireSelection(t, ctx, backendB, bucketB) }) - t.Run("duplicate bucket values are rejected", func(t *testing.T) { + t.Run("duplicate bucket values use the entry's default bucket", func(t *testing.T) { md := metadata.Pairs( cache.S3BackendGRPCMetadataKey, backendA, cache.S3BucketGRPCMetadataKey, bucketA, cache.S3BucketGRPCMetadataKey, bucketALegacy, ) - _, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "bucket_duplicate") + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + requireSelection(t, ctx, backendA, bucketA) }) - t.Run("unknown bucket is rejected", func(t *testing.T) { - md := pairMD(backendA, "rogue-bucket") - _, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "bucket_unknown") + t.Run("unknown bucket uses the entry's default bucket", func(t *testing.T) { + md := pairMD(backendB, "rogue-bucket") + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + requireSelection(t, ctx, backendB, bucketB) }) - t.Run("the pair is validated: right endpoint, other endpoint's bucket is rejected", func(t *testing.T) { - // bucketB is allowlisted — but only for backend B. Accepting it on - // backend A would let a stale pair read another shard's bucket. + t.Run("the pair resolves together: another entry's bucket defaults on this entry", func(t *testing.T) { + // bucketB exists in the map — but only for backend B. On backend A + // it is unknown and A's default bucket is used; a stale pair must + // not read another shard's bucket. md := pairMD(backendA, bucketB) - _, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "bucket_unknown") + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + requireSelection(t, ctx, backendA, bucketA) }) - t.Run("byte-verbatim bucket match: near-miss buckets are rejected", func(t *testing.T) { + t.Run("byte-verbatim bucket match: near-miss buckets use the default", func(t *testing.T) { for _, nearMiss := range []string{ "BAZEL-CACHE-A", // case difference " bazel-cache-a", // stray whitespace "bazel-cache-a/", // trailing slash } { md := pairMD(backendA, nearMiss) - _, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "bucket_unknown") + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + requireSelection(t, ctx, backendA, bucketA) } }) - t.Run("endpoint violations are reported before bucket violations", func(t *testing.T) { - // An unknown endpoint with a missing bucket must surface the - // endpoint cause: the endpoint is the routing key, and during a - // rollout the endpoint-level causes are the primary skew signal. - md := metadata.Pairs(cache.S3BackendGRPCMetadataKey, "http://rogue.example.com:9000") - _, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "unknown") + t.Run("unknown selector with a bucket valid on the default entry keeps the bucket", func(t *testing.T) { + // The halves default independently: the selector defaults to entry + // A, and the forwarded bucket is then resolved against A's allowed + // set — the legacy bucket is in it, so it is honored. + md := pairMD("http://minio.torn-down.example:9000", bucketALegacy) + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + requireSelection(t, ctx, backendA, bucketALegacy) }) t.Run("allowlisted pair is lifted onto context", func(t *testing.T) { md := pairMD(backendB, bucketB) - ctx, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - selection, ok := cache.S3BackendFromContext(ctx) - if !ok || selection.Endpoint != backendB || selection.Bucket != bucketB { - t.Fatalf("unexpected selection %+v ok=%v", selection, ok) - } + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + requireSelection(t, ctx, backendB, bucketB) }) t.Run("extra (pre-rename) bucket of the matched entry is accepted", func(t *testing.T) { md := pairMD(backendA, bucketALegacy) - ctx, err := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), allowedBackends()) - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - selection, ok := cache.S3BackendFromContext(ctx) - if !ok || selection.Endpoint != backendA || selection.Bucket != bucketALegacy { - t.Fatalf("unexpected selection %+v ok=%v", selection, ok) - } + ctx := s3BackendFromIncomingContext(metadata.NewIncomingContext(context.Background(), md), testRouting()) + requireSelection(t, ctx, backendA, bucketALegacy) }) } func TestS3BackendInterceptorExemptsHealthAndCapabilities(t *testing.T) { - interceptor := GRPCS3BackendUnaryServerInterceptor(allowedBackends()) + interceptor := GRPCS3BackendUnaryServerInterceptor(testRouting()) for _, method := range []string{ "/grpc.health.v1.Health/Check", "/build.bazel.remote.execution.v2.Capabilities/GetCapabilities", } { - handled := false + var handlerCtx context.Context _, err := interceptor(context.Background(), nil, &grpc.UnaryServerInfo{FullMethod: method}, func(ctx context.Context, req interface{}) (interface{}, error) { - handled = true + handlerCtx = ctx return nil, nil }) - if err != nil || !handled { - t.Fatalf("expected %s to bypass selector enforcement, err=%v handled=%v", method, err, handled) + if err != nil || handlerCtx == nil { + t.Fatalf("expected %s to reach the handler, err=%v", method, err) + } + // Exempt methods carry no tenant data; no selection is resolved. + if _, ok := cache.S3BackendFromContext(handlerCtx); ok { + t.Fatalf("exempt method %s carried an S3 backend selection", method) } } - // A cache RPC without the selector is rejected before the handler runs. - handled := false + // A cache RPC without any selector reaches the handler with the default + // backend resolved. + var handlerCtx context.Context _, err := interceptor(context.Background(), nil, &grpc.UnaryServerInfo{FullMethod: "/build.bazel.remote.execution.v2.ActionCache/GetActionResult"}, func(ctx context.Context, req interface{}) (interface{}, error) { - handled = true + handlerCtx = ctx return nil, nil }) - if handled { - t.Fatal("handler ran despite missing selector") + if err != nil || handlerCtx == nil { + t.Fatalf("expected selector-less cache RPC to be served, err=%v", err) } - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "missing") + requireSelection(t, handlerCtx, backendA, bucketA) } // fakeServerStream is the minimal grpc.ServerStream for interceptor tests: @@ -234,76 +223,51 @@ func (f *fakeServerStream) Context() context.Context { } func TestS3BackendStreamInterceptor(t *testing.T) { - interceptor := GRPCS3BackendStreamServerInterceptor(allowedBackends()) + interceptor := GRPCS3BackendStreamServerInterceptor(testRouting()) byteStreamRead := &grpc.StreamServerInfo{FullMethod: "/google.bytestream.ByteStream/Read"} - t.Run("missing selector is rejected fail-closed before the handler", func(t *testing.T) { - handled := false + // streamSelection runs the interceptor with the given incoming metadata + // and returns the handler stream's context. + streamSelection := func(t *testing.T, ctx context.Context) context.Context { + t.Helper() + var handlerStream grpc.ServerStream err := interceptor(nil, - &fakeServerStream{ctx: context.Background()}, + &fakeServerStream{ctx: ctx}, byteStreamRead, func(srv interface{}, ss grpc.ServerStream) error { - handled = true + handlerStream = ss return nil }) - if handled { - t.Fatal("handler ran despite missing selector") + if err != nil || handlerStream == nil { + t.Fatalf("expected stream to reach the handler, err=%v", err) } - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "missing") + return handlerStream.Context() + } + + t.Run("missing selector routes to the default backend", func(t *testing.T) { + ctx := streamSelection(t, context.Background()) + requireSelection(t, ctx, backendA, bucketA) }) - t.Run("unknown selector is rejected fail-closed before the handler", func(t *testing.T) { - md := pairMD("http://rogue.example.com:9000", bucketA) - handled := false - err := interceptor(nil, - &fakeServerStream{ctx: metadata.NewIncomingContext(context.Background(), md)}, - byteStreamRead, - func(srv interface{}, ss grpc.ServerStream) error { - handled = true - return nil - }) - if handled { - t.Fatal("handler ran despite unknown selector") - } - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "unknown") + t.Run("unknown selector routes to the default backend", func(t *testing.T) { + md := pairMD("http://minio.torn-down.example:9000", bucketA) + ctx := streamSelection(t, metadata.NewIncomingContext(context.Background(), md)) + requireSelection(t, ctx, backendA, bucketA) }) - t.Run("wrong bucket is rejected fail-closed before the handler", func(t *testing.T) { + t.Run("another entry's bucket uses the resolved entry's default", func(t *testing.T) { md := pairMD(backendA, bucketB) - handled := false - err := interceptor(nil, - &fakeServerStream{ctx: metadata.NewIncomingContext(context.Background(), md)}, - byteStreamRead, - func(srv interface{}, ss grpc.ServerStream) error { - handled = true - return nil - }) - if handled { - t.Fatal("handler ran despite wrong bucket") - } - requireTrustRejection(t, err, cache.RejectionReasonS3BackendSelector, "bucket_unknown") + ctx := streamSelection(t, metadata.NewIncomingContext(context.Background(), md)) + requireSelection(t, ctx, backendA, bucketA) }) t.Run("accepted stream's Context carries the pair", func(t *testing.T) { - md := pairMD(backendA, bucketA) - var handlerStream grpc.ServerStream - err := interceptor(nil, - &fakeServerStream{ctx: metadata.NewIncomingContext(context.Background(), md)}, - byteStreamRead, - func(srv interface{}, ss grpc.ServerStream) error { - handlerStream = ss - return nil - }) - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - selection, ok := cache.S3BackendFromContext(handlerStream.Context()) - if !ok || selection.Endpoint != backendA || selection.Bucket != bucketA { - t.Fatalf("wrapped stream Context selection = %+v ok=%v, want (%s, %s)", selection, ok, backendA, bucketA) - } + md := pairMD(backendB, bucketB) + ctx := streamSelection(t, metadata.NewIncomingContext(context.Background(), md)) + requireSelection(t, ctx, backendB, bucketB) }) - t.Run("exempt methods bypass enforcement", func(t *testing.T) { + t.Run("exempt methods bypass resolution", func(t *testing.T) { handled := false err := interceptor(nil, &fakeServerStream{ctx: context.Background()}, @@ -313,7 +277,7 @@ func TestS3BackendStreamInterceptor(t *testing.T) { return nil }) if err != nil || !handled { - t.Fatalf("expected health stream to bypass enforcement, err=%v handled=%v", err, handled) + t.Fatalf("expected health stream to bypass resolution, err=%v handled=%v", err, handled) } }) } diff --git a/server/grpc_storage_prefix.go b/server/grpc_storage_prefix.go index 1538321..e08c01e 100644 --- a/server/grpc_storage_prefix.go +++ b/server/grpc_storage_prefix.go @@ -35,9 +35,8 @@ import ( // code-based auth check. // storagePrefixRejected and authSecretRejected meter the fail-closed -// rejections below, by cause — symmetric with the selector interceptor's -// bazel_remote_s3_backend_selector_rejected_total, and the counters the -// rate-limited trust-rejection log line points operators at. The prefix and +// rejections below, by cause — the counters the rate-limited +// trust-rejection log line points operators at. The prefix and // secret are minted by the trusted upstream, never by customers, so any // nonzero series means upstream drift: FA/L1 version skew, a Doppler secret // mismatch, or a forwarding bug. diff --git a/server/grpc_storage_prefix_test.go b/server/grpc_storage_prefix_test.go index 30162c5..335be63 100644 --- a/server/grpc_storage_prefix_test.go +++ b/server/grpc_storage_prefix_test.go @@ -7,11 +7,41 @@ import ( "github.com/buchgr/bazel-remote/v2/cache" "github.com/prometheus/client_golang/prometheus/testutil" + "google.golang.org/genproto/googleapis/rpc/errdetails" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" ) +// requireTrustRejection asserts that err is an InvalidArgument rejection +// carrying the typed ErrorInfo marker our trust interceptors mint — the +// wire contract the upstream grpcproxy uses to degrade config-race +// rejections to metered misses instead of failing builds. +func requireTrustRejection(t *testing.T, err error, reason, cause string) { + t.Helper() + s, ok := status.FromError(err) + if !ok || s.Code() != codes.InvalidArgument { + t.Fatalf("expected InvalidArgument status, got %v", err) + } + for _, detail := range s.Details() { + info, ok := detail.(*errdetails.ErrorInfo) + if !ok { + continue + } + if info.GetDomain() != cache.TrustRejectionErrorDomain { + t.Fatalf("ErrorInfo domain = %q, want %q", info.GetDomain(), cache.TrustRejectionErrorDomain) + } + if info.GetReason() != reason { + t.Fatalf("ErrorInfo reason = %q, want %q", info.GetReason(), reason) + } + if got := info.GetMetadata()["cause"]; got != cause { + t.Fatalf("ErrorInfo cause = %q, want %q", got, cause) + } + return + } + t.Fatalf("rejection %v carries no ErrorInfo trust marker", err) +} + func TestStoragePrefixFromIncomingContext(t *testing.T) { t.Run("no metadata is rejected (fail-closed)", func(t *testing.T) { _, err := storagePrefixFromIncomingContext(context.Background(), "") diff --git a/server/grpc_tenant_metadata.go b/server/grpc_tenant_metadata.go index e32e6b0..4f8daa8 100644 --- a/server/grpc_tenant_metadata.go +++ b/server/grpc_tenant_metadata.go @@ -49,27 +49,33 @@ func singleMetadataValue(md metadata.MD, key string) (value string, cause string } } -// trustRejectionLogEvery rate-limits rejection logging to one line per -// (reason, cause) pair per interval. A misconfigured fleet can reject at -// full request rate, and rejections are already metered per cause -// (bazel_remote_..._rejected_total) — the log line exists so a live -// debugging session on the node sees the incident in journald at all -// (validated 2026-07-30: counters incremented, journal stayed empty), -// not to reproduce the counter's volume. -const trustRejectionLogEvery = 30 * time.Second +// rateLimitedLogEvery rate-limits interceptor logging to one line per key +// per interval. A misconfigured fleet can trip these paths at full request +// rate, and every event is already metered per cause (the *_rejected_total / +// *_defaulted_total counters) — the log line exists so a live debugging +// session on the node sees the incident in journald at all (validated +// 2026-07-30: counters incremented, journal stayed empty), not to reproduce +// the counter's volume. +const rateLimitedLogEvery = 30 * time.Second -var trustRejectionLastLog sync.Map // "reason/cause" -> *atomic.Int64 (unix nanos) +var rateLimitedLastLog sync.Map // key -> *atomic.Int64 (unix nanos) -func logTrustRejection(reason, cause, message string) { - gateAny, _ := trustRejectionLastLog.LoadOrStore(reason+"/"+cause, new(atomic.Int64)) +// logRateLimited logs at most one line per key per rateLimitedLogEvery. +func logRateLimited(key, format string, args ...interface{}) { + gateAny, _ := rateLimitedLastLog.LoadOrStore(key, new(atomic.Int64)) gate := gateAny.(*atomic.Int64) last := gate.Load() now := time.Now().UnixNano() - if now-last < int64(trustRejectionLogEvery) || !gate.CompareAndSwap(last, now) { + if now-last < int64(rateLimitedLogEvery) || !gate.CompareAndSwap(last, now) { return } - log.Printf("trust interceptor rejected request (reason=%s cause=%s, logged at most once per %v per cause; see the *_rejected_total counters for volume): %s", - reason, cause, trustRejectionLogEvery, message) + log.Printf(format, args...) +} + +func logTrustRejection(reason, cause, message string) { + logRateLimited(reason+"/"+cause, + "trust interceptor rejected request (reason=%s cause=%s, logged at most once per %v per cause; see the *_rejected_total counters for volume): %s", + reason, cause, rateLimitedLogEvery, message) } // trustRejection mints an InvalidArgument rejection carrying the typed