From 41d45b2c3d02b46dc7c9c25bf5a3a9cdcaef07d8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Thu, 3 Sep 2026 00:22:14 -0700 Subject: [PATCH] agent endpoints: route at deployment granularity, drop the route filter Route presence like a reverse proxy keyed on (project, agent_name, deployment): any of a deployment's workers is a candidate and the worker's own router returns the real status (a 404 for a path it doesn't serve during a rolling deploy is returned, not re-relayed). This removes the whole per-node route-advertisement path: the cuckoo route filter and matcher (routematch.go), the miss-tagging / retry-past-miss machinery, and the route-depth cap that only existed to bound the filter. Text routing is now the same shape as voice job dispatch. --- pkg/agent/endpoint/front.go | 62 +----- pkg/agent/endpoint/manifest.go | 7 - pkg/agent/endpoint/routematch.go | 274 -------------------------- pkg/agent/endpoint/routematch_test.go | 166 ---------------- 4 files changed, 8 insertions(+), 501 deletions(-) delete mode 100644 pkg/agent/endpoint/routematch.go delete mode 100644 pkg/agent/endpoint/routematch_test.go diff --git a/pkg/agent/endpoint/front.go b/pkg/agent/endpoint/front.go index cc2d7966e..2df8a536f 100644 --- a/pkg/agent/endpoint/front.go +++ b/pkg/agent/endpoint/front.go @@ -30,24 +30,6 @@ import ( "github.com/livekit/protocol/logger" ) -// HeaderEndpointMiss marks a response the front produced ITSELF - a routing -// miss (no matching route, wrong method, auth required, or no local capacity) - -// as distinct from a response the worker's app returned through the bridge. A -// relay caller keys its cross-node retry on this header, so a worker's own 404 -// (e.g. GET /users/999 for a missing user) is never mistaken for "this node -// can't serve the path" and re-relayed. The value is the miss kind, so the -// caller can surface the most informative aggregate status. The header is set -// only on the private relay listener (see MarkMisses) and stripped by the relay -// caller, so it never reaches a client. -const HeaderEndpointMiss = "X-Livekit-Endpoint-Miss" - -const ( - MissNotFound = "notfound" - MissMethodNotAllowed = "methodnotallowed" - MissUnauthenticated = "unauthenticated" - MissUnavailable = "unavailable" -) - const ( // PathPrefix is the public route namespace: /agents/{agent_name}/{deployment}/{path...} PathPrefix = "/agents/" @@ -86,9 +68,6 @@ type Front struct { fallback Fallback // see WithSingleKeyFallback singleKeyFallback bool - // see MarkMisses: set on the private relay listener so a relay caller can - // tell a routing miss from a worker-app response - markMisses bool } func NewFront(registry *Registry, resolveAPIKey APIKeyResolver, log logger.Logger) *Front { @@ -108,12 +87,6 @@ type FallbackRequest struct { Authenticated bool AgentName string Deployment string - // Method is the request's HTTP method, so a multi-node layer can select - // candidates method-aware (a node serving the path under a different method - // is not a candidate). - Method string - // Path within the deployment, '/'-rooted - Path string } // Fallback serves a request elsewhere (e.g. a multi-node relay); it reports @@ -128,30 +101,11 @@ func (f *Front) WithFallback(fb Fallback) *Front { return f } -// MarkMisses tags the front's own routing-miss responses with HeaderEndpointMiss -// so a relay caller can distinguish them from worker-app responses. Set it on -// the private relay listener only; the public front must not (the header would -// leak to clients, and its misses are final anyway). -func (f *Front) MarkMisses() *Front { - f.markMisses = true - return f -} - -// writeMiss writes a front-originated miss, tagging it with the kind when this -// front marks misses (the relay listener) so the relay caller can retry past it -// and aggregate the most informative status. -func (f *Front) writeMiss(w http.ResponseWriter, status int, kind, msg string) { - if f.markMisses { - w.Header().Set(HeaderEndpointMiss, kind) - } - http.Error(w, msg, status) -} - -// writeUnavailable writes a 503 with a Retry-After hint, the miss the front +// writeUnavailable writes a 503 with a Retry-After hint, the response the front // returns when no worker can currently serve a request it did route. func (f *Front) writeUnavailable(w http.ResponseWriter, msg string) { w.Header().Set("Retry-After", "1") - f.writeMiss(w, http.StatusServiceUnavailable, MissUnavailable, msg) + http.Error(w, msg, http.StatusServiceUnavailable) } // WithSingleKeyFallback resolves unauthenticated requests to the registry's @@ -192,7 +146,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { } if apiKey == "" { w.Header().Set("WWW-Authenticate", "Bearer") - f.writeMiss(w, http.StatusUnauthorized, MissUnauthenticated, "authentication required") + http.Error(w, "authentication required", http.StatusUnauthorized) return } @@ -244,7 +198,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { // nothing local can serve: hand off before the local status mapping if f.fallback(w, r, &FallbackRequest{ APIKey: apiKey, Authenticated: authenticated, - AgentName: agentName, Deployment: deployment, Method: r.Method, Path: path, + AgentName: agentName, Deployment: deployment, }) { return } @@ -257,11 +211,11 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { switch { case restricted: w.Header().Set("WWW-Authenticate", "Bearer") - f.writeMiss(w, http.StatusUnauthorized, MissUnauthenticated, "authentication required") + http.Error(w, "authentication required", http.StatusUnauthorized) case partial: - f.writeMiss(w, http.StatusMethodNotAllowed, MissMethodNotAllowed, "method not allowed") + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) default: - f.writeMiss(w, http.StatusNotFound, MissNotFound, "not found") + http.Error(w, "not found", http.StatusNotFound) } return } @@ -291,7 +245,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { if bodyConsumed == 0 && f.fallback != nil { if f.fallback(w, r, &FallbackRequest{ APIKey: apiKey, Authenticated: authenticated, - AgentName: agentName, Deployment: deployment, Method: r.Method, Path: path, + AgentName: agentName, Deployment: deployment, }) { return } diff --git a/pkg/agent/endpoint/manifest.go b/pkg/agent/endpoint/manifest.go index 977e2c7e3..e781f2429 100644 --- a/pkg/agent/endpoint/manifest.go +++ b/pkg/agent/endpoint/manifest.go @@ -74,13 +74,6 @@ func ParseManifest(endpoints []*livekit.AgentHttp_AgentEndpoint) (*Manifest, err if err != nil { return nil, err } - // bound route depth: it caps the replicated filter's prefix count and - // the edge matcher's walk (a request past a route's depth can't match it) - if d, err := RouteDepth(ep.GetPath()); err != nil { - return nil, err - } else if d > MaxRouteDepth { - return nil, fmt.Errorf("endpoint %q exceeds max route depth %d", ep.GetPath(), MaxRouteDepth) - } if ep.GetKind() != livekit.AgentHttp_AEK_HTTP { return nil, fmt.Errorf("endpoint %q has unsupported kind %s", ep.GetPath(), ep.GetKind()) } diff --git a/pkg/agent/endpoint/routematch.go b/pkg/agent/endpoint/routematch.go deleted file mode 100644 index 9df3609a7..000000000 --- a/pkg/agent/endpoint/routematch.go +++ /dev/null @@ -1,274 +0,0 @@ -// Copyright 2026 LiveKit, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package endpoint - -import ( - "regexp" - "strings" -) - -// canonical wildcard tokens; the control-char prefix keeps them from colliding -// with any real path segment -const ( - tokStr = "\x00s" // {x} or {x:str} - any single segment - tokInt = "\x00i" // {x:int} - tokFloat = "\x00f" // {x:float} - tokUUID = "\x00u" // {x:uuid} - tokGlob = "\x00p" // {x:path} - greedy, spans the rest of the path - - segSep = "\x1f" // joins canonical segments into an index key - - // a route deeper than this is rejected at registration; it bounds both the - // filter's prefix count and the walk's depth - MaxRouteDepth = 32 - // a request longer than this skips the precise walk (falls back to - // relay-and-let-the-worker-decide); real paths are a handful of segments - maxMatchDepth = 64 -) - -var ( - intRe = regexp.MustCompile(`^[0-9]+$`) - floatRe = regexp.MustCompile(`^[0-9]+(?:\.[0-9]+)?$`) - uuidRe = regexp.MustCompile(`^[0-9a-fA-F]{8}-?[0-9a-fA-F]{4}-?[0-9a-fA-F]{4}-?[0-9a-fA-F]{4}-?[0-9a-fA-F]{12}$`) -) - -// canonicalizeTemplate turns a starlette path template into canonical segments -// (params -> typed tokens, {:path} -> glob). A segment that mixes a literal -// with a param (e.g. "{name}.json") canonicalizes to the str token: the edge -// over-approximates the shape and the worker enforces the literal part. -func canonicalizeTemplate(path string) ([]string, error) { - if _, err := ParseTemplate(path); err != nil { // validates '/', convertors, dup params - return nil, err - } - segs := splitSegments(path) - out := make([]string, 0, len(segs)) - for _, seg := range segs { - tok, glob := canonicalSegment(seg) - out = append(out, tok) - if glob { - // a path convertor spans slashes: the glob is terminal, and any - // segments after it are dropped so the walk over-approximates (a - // concrete request matching the pre-glob prefix always matches; the - // worker's router enforces whatever follows). Truncating here can - // only add matches, never drop one. - break - } - } - return out, nil -} - -// canonicalSegment maps a template segment to its canonical token; glob is true -// when the segment contains a path convertor (spans slashes). -func canonicalSegment(seg string) (tok string, glob bool) { - matches := paramRegex.FindAllStringSubmatchIndex(seg, -1) - if len(matches) == 0 { - return seg, false // pure literal - } - // any path convertor in the segment makes it a glob, whether it is the whole - // segment ("{rest:path}") or mixed with literal text ("pre{rest:path}") - - // str would be strictly narrower than the glob's `.*` and would drop matches - for _, m := range matches { - if m[4] != -1 && seg[m[4]+1:m[5]] == "path" { - return tokGlob, true - } - } - // a clean whole-segment param: "{name}" or "{name:conv}" - if m := matches[0]; len(matches) == 1 && m[0] == 0 && m[1] == len(seg) { - conv := "str" - if m[4] != -1 { - conv = seg[m[4]+1 : m[5]] - } - switch conv { - case "int": - return tokInt, false - case "float": - return tokFloat, false - case "uuid": - return tokUUID, false - default: // str and anything ParseTemplate already accepted - return tokStr, false - } - } - // a non-path param mixed with literal text: over-approximate to str (str's - // [^/]+ is wider than any typed single-segment convertor, so no false negative) - return tokStr, false -} - -// candidateTokens returns the canonical tokens a concrete request segment could -// match: the literal itself, the str wildcard (always), and any typed wildcard -// whose pattern the value satisfies. -func candidateTokens(seg string) []string { - toks := make([]string, 0, 5) - toks = append(toks, seg, tokStr) - if intRe.MatchString(seg) { - toks = append(toks, tokInt) - } - if floatRe.MatchString(seg) { - toks = append(toks, tokFloat) - } - if uuidRe.MatchString(seg) { - toks = append(toks, tokUUID) - } - return toks -} - -func splitSegments(path string) []string { - path = strings.TrimPrefix(path, "/") - path = strings.TrimSuffix(path, "/") // /x/ and /x match the same shape; worker 307s - if path == "" { - return nil - } - return strings.Split(path, "/") -} - -func extendKey(prefix, tok string) string { - if prefix == "" { - return tok - } - return prefix + segSep + tok -} - -// RouteKey qualifies a canonical path terminal with its HTTP method, so a -// terminal is a route only for the method(s) the worker declared. Methods are -// uppercase ASCII (allowlisted at registration) and canonical tokens start with -// a control char, so method and path can't collide. Prefix keys stay path-only -// (the pruned walk is method-agnostic until the terminal check). -func RouteKey(method, canonicalRoute string) string { - return method + segSep + canonicalRoute -} - -// PrefixIndex is the backing an edge matches against: whether a canonical prefix -// key is a prefix of some route, and whether it is a complete route. An exact -// map-backed index (ExactIndex) gives precise answers; a cuckoo-filter backing -// trades a bounded false-positive rate for a compact, replicable form. -type PrefixIndex interface { - HasPrefix(key string) bool - IsRoute(key string) bool -} - -// Matches reports whether the request (method + path) matches any route in the -// index, via the pruned walk. The path walk is method-agnostic (prefix keys are -// path-only); only the terminal check is method-qualified. A path deeper than -// maxMatchDepth returns true so the caller relays and lets the worker decide -// (never a spurious 404 on a deep path). -func Matches(idx PrefixIndex, method, path string) bool { - segs := splitSegments(path) - if len(segs) > maxMatchDepth { - return true - } - - live := map[string]struct{}{"": {}} - for _, seg := range segs { - // a glob at any currently-live prefix eats this segment and the rest - for p := range live { - if idx.IsRoute(RouteKey(method, extendKey(p, tokGlob))) { - return true - } - } - next := make(map[string]struct{}) - for p := range live { - for _, tok := range candidateTokens(seg) { - k := extendKey(p, tok) - if idx.HasPrefix(k) { - next[k] = struct{}{} - } - } - } - if len(next) == 0 { - return false - } - live = next - } - // consumed every segment: a live prefix that is a complete route for this - // method matches, and a glob at a live prefix matches the empty remainder too - for p := range live { - if idx.IsRoute(RouteKey(method, p)) || idx.IsRoute(RouteKey(method, extendKey(p, tokGlob))) { - return true - } - } - return false -} - -// ExactIndex is a precise, map-backed PrefixIndex built from a set of route -// templates. Used where the full route set is available locally (a single-node -// server, or a test); the cuckoo-filter backing is used where the set must be -// replicated compactly. -type ExactIndex struct { - prefixes map[string]struct{} - routes map[string]struct{} -} - -func NewExactIndex() *ExactIndex { - return &ExactIndex{ - prefixes: make(map[string]struct{}), - routes: make(map[string]struct{}), - } -} - -// Add canonicalizes a template and inserts its path prefixes plus a -// method-qualified terminal per declared method. -func (x *ExactIndex) Add(methods []string, path string) error { - segs, err := canonicalizeTemplate(path) - if err != nil { - return err - } - key := "" - for _, s := range segs { - key = extendKey(key, s) - x.prefixes[key] = struct{}{} - } - for _, m := range methods { - x.routes[RouteKey(m, key)] = struct{}{} - } - return nil -} - -func (x *ExactIndex) HasPrefix(key string) bool { - _, ok := x.prefixes[key] - return ok -} - -func (x *ExactIndex) IsRoute(key string) bool { - _, ok := x.routes[key] - return ok -} - -// CanonicalKeys returns the index keys for a route template: every prefix key -// (including the full-length one) and the terminal route key. A multi-node -// layer uses these to populate a compact backing (e.g. a cuckoo filter) that it -// replicates, then matches with Matches against a FilterIndex over it. -func CanonicalKeys(path string) (prefixes []string, route string, err error) { - segs, err := canonicalizeTemplate(path) - if err != nil { - return nil, "", err - } - key := "" - prefixes = make([]string, 0, len(segs)) - for _, s := range segs { - key = extendKey(key, s) - prefixes = append(prefixes, key) - } - return prefixes, key, nil -} - -// RouteDepth reports the canonical segment count of a template, for enforcing -// MaxRouteDepth at registration. -func RouteDepth(path string) (int, error) { - segs, err := canonicalizeTemplate(path) - if err != nil { - return 0, err - } - return len(segs), nil -} diff --git a/pkg/agent/endpoint/routematch_test.go b/pkg/agent/endpoint/routematch_test.go deleted file mode 100644 index 91158d59b..000000000 --- a/pkg/agent/endpoint/routematch_test.go +++ /dev/null @@ -1,166 +0,0 @@ -// Copyright 2026 LiveKit, Inc. - -package endpoint - -import ( - "net/http" - "testing" - - "github.com/stretchr/testify/require" -) - -// idx builds an ExactIndex whose routes all serve GET; the path-shape tests -// below match with GET via the matches helper. Method-awareness has its own test. -func idx(t *testing.T, routes ...string) *ExactIndex { - x := NewExactIndex() - for _, r := range routes { - require.NoError(t, x.Add([]string{http.MethodGet}, r), r) - } - return x -} - -// matches is a GET-method shorthand so the path-shape assertions read cleanly. -func matches(idx PrefixIndex, path string) bool { - return Matches(idx, http.MethodGet, path) -} - -func TestRouteMatchStaticAndParams(t *testing.T) { - x := idx(t, "/sms", "/orders/{id}/items", "/orders/{id}/list", "/health") - - // static - require.True(t, matches(x, "/sms")) - require.True(t, matches(x, "/health")) - require.False(t, matches(x, "/unknown")) - - // single-segment params, and the two sub-routes are distinguished - require.True(t, matches(x, "/orders/42/items")) - require.True(t, matches(x, "/orders/abc/list")) - require.False(t, matches(x, "/orders/42/refund")) // neither sub-route exists - require.False(t, matches(x, "/orders/42")) // prefix only, not a full route - require.False(t, matches(x, "/orders/42/items/x")) - - // trailing slash matches the same shape (worker 307s) - require.True(t, matches(x, "/orders/42/items/")) -} - -func TestRouteMatchTypedParams(t *testing.T) { - x := idx(t, "/orders/{id:int}", "/u/{u:uuid}") - - require.True(t, matches(x, "/orders/42")) - require.False(t, matches(x, "/orders/abc")) // not an int -> edge miss, no relay - require.False(t, matches(x, "/orders/3.14")) // float is not int - - uuid := "550e8400-e29b-41d4-a716-446655440000" - require.True(t, matches(x, "/u/"+uuid)) - require.True(t, matches(x, "/u/"+"550e8400e29b41d4a716446655440000")) // hyphenless - require.False(t, matches(x, "/u/not-a-uuid")) -} - -func TestRouteMatchUUIDvsStr(t *testing.T) { - // a uuid value is also a valid str; whichever the route declared wins - uuid := "550e8400-e29b-41d4-a716-446655440000" - - strOnly := idx(t, "/x/{v}") // str - require.True(t, matches(strOnly, "/x/"+uuid)) // uuid value matches a str route - require.True(t, matches(strOnly, "/x/anything")) - - uuidOnly := idx(t, "/x/{v:uuid}") - require.True(t, matches(uuidOnly, "/x/"+uuid)) - require.False(t, matches(uuidOnly, "/x/anything")) // non-uuid never matches a uuid route -} - -func TestRouteMatchGlob(t *testing.T) { - x := idx(t, "/files/{rest:path}", "/static") - - require.True(t, matches(x, "/files/a")) - require.True(t, matches(x, "/files/a/b/c/d.txt")) // spans segments - // /files with no trailing slash is a tolerated FALSE POSITIVE (Starlette - // requires the separating slash); the worker returns the real status - require.True(t, matches(x, "/files")) - require.False(t, matches(x, "/other/a/b")) - require.True(t, matches(x, "/static")) -} - -// a path convertor that is not the last segment must still match (the glob is -// treated as terminal, over-approximating the suffix) -func TestRouteMatchNonTerminalGlob(t *testing.T) { - x := idx(t, "/files/{rest:path}/edit") - require.True(t, matches(x, "/files/a/b/edit")) - require.True(t, matches(x, "/files/a/edit")) - // suffix is over-approximated: a path under /files matches even without - // /edit (the worker's router returns the real 404) - require.True(t, matches(x, "/files/a/b")) - require.False(t, matches(x, "/other/a/edit")) -} - -// a path convertor mixed with literal text in a segment spans slashes and must -// not be narrowed to a single-segment str -func TestRouteMatchMixedGlob(t *testing.T) { - x := idx(t, "/files/pre{rest:path}") - require.True(t, matches(x, "/files/prea")) - require.True(t, matches(x, "/files/prea/b/c")) // spans segments (would drop under str) - require.False(t, matches(x, "/other/x")) -} - -func TestRouteMatchHeterogeneousShapes(t *testing.T) { - // two different workers' route sets, same path shape, different types - ints := idx(t, "/item/{id:int}") - strs := idx(t, "/item/{slug}") - - require.True(t, matches(ints, "/item/42")) - require.False(t, matches(ints, "/item/foo")) // routes only to the str worker - require.True(t, matches(strs, "/item/foo")) - require.True(t, matches(strs, "/item/42")) // str accepts a numeric slug too -} - -func TestRouteMatchLiteralBeatsWildcardCoexist(t *testing.T) { - // a literal segment and a param at the same position coexist - x := idx(t, "/users/me", "/users/{id:int}") - require.True(t, matches(x, "/users/me")) - require.True(t, matches(x, "/users/42")) - require.False(t, matches(x, "/users/abc")) // neither: not "me", not an int -} - -// the terminal is method-qualified: the SAME path shape served under different -// methods (e.g. two workers, POST vs PUT) matches only for the declared method, -// so the cross-node presence filter is method-aware. -func TestRouteMatchMethodAware(t *testing.T) { - postOnly := NewExactIndex() - require.NoError(t, postOnly.Add([]string{http.MethodPost}, "/test")) - putOnly := NewExactIndex() - require.NoError(t, putOnly.Add([]string{http.MethodPut}, "/test")) - - require.True(t, Matches(postOnly, http.MethodPost, "/test")) - require.False(t, Matches(postOnly, http.MethodPut, "/test")) - require.True(t, Matches(putOnly, http.MethodPut, "/test")) - require.False(t, Matches(putOnly, http.MethodPost, "/test")) - - // a route declaring several methods matches each, and only those - multi := NewExactIndex() - require.NoError(t, multi.Add([]string{http.MethodGet, http.MethodHead}, "/thing/{id:int}")) - require.True(t, Matches(multi, http.MethodGet, "/thing/5")) - require.True(t, Matches(multi, http.MethodHead, "/thing/5")) - require.False(t, Matches(multi, http.MethodDelete, "/thing/5")) - require.False(t, Matches(multi, http.MethodGet, "/thing/abc")) // still shape-checked -} - -func TestRouteMatchDepthCapFallsBackToRelay(t *testing.T) { - x := idx(t, "/a/{b}") - deep := "/" // build a > maxMatchDepth path - for i := 0; i < maxMatchDepth+5; i++ { - deep += "x/" - } - require.True(t, matches(x, deep)) // over cap -> relay and let the worker decide -} - -// depth enforcement lives in the manifest layer (ParseManifest), not in the -// index: Add canonicalizes any depth without error, so the two layers stay -// decoupled. -func TestRouteDepthNotEnforcedAtAdd(t *testing.T) { - x := NewExactIndex() - long := "/" - for i := 0; i < MaxRouteDepth+2; i++ { - long += "s/" - } - require.NoError(t, x.Add([]string{http.MethodGet}, long)) -}