diff --git a/pkg/agent/endpoint/front.go b/pkg/agent/endpoint/front.go index 2df8a536f..0db6697ca 100644 --- a/pkg/agent/endpoint/front.go +++ b/pkg/agent/endpoint/front.go @@ -101,8 +101,8 @@ func (f *Front) WithFallback(fb Fallback) *Front { return f } -// 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. +// writeUnavailable writes a 503 with a Retry-After hint: no local worker can +// serve the request and no fallback placed it elsewhere. func (f *Front) writeUnavailable(w http.ResponseWriter, msg string) { w.Header().Set("Retry-After", "1") http.Error(w, msg, http.StatusServiceUnavailable) @@ -195,7 +195,10 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } if route == nil && f.fallback != nil { - // nothing local can serve: hand off before the local status mapping + // nothing local matched: hand off to the multi-node fallback (relay to a + // node holding the deployment) before the local status mapping. The + // serving node's relay listener installs no fallback of its own, so a + // relayed request is served or errored there and never re-relays. if f.fallback(w, r, &FallbackRequest{ APIKey: apiKey, Authenticated: authenticated, AgentName: agentName, Deployment: deployment, diff --git a/pkg/agent/endpoint/front_test.go b/pkg/agent/endpoint/front_test.go new file mode 100644 index 000000000..c029b9394 --- /dev/null +++ b/pkg/agent/endpoint/front_test.go @@ -0,0 +1,71 @@ +// Copyright 2026 LiveKit, Inc. + +package endpoint + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +func fallbackFront(t *testing.T, fb Fallback, withWorker bool) *Front { + reg := NewRegistry() + if withWorker { + m, err := ParseManifest([]*livekit.AgentHttp_AgentEndpoint{ + {Path: "/known", Methods: []string{"GET"}, Public: true}, + }) + require.NoError(t, err) + r := &Registration{WorkerID: "w1", APIKey: "proj", AgentName: "a", Deployment: "d", Manifest: m} + r.SetSession(&fakeSession{}) + require.NoError(t, reg.Register(r)) + } + f := NewFront(reg, func(*http.Request) (string, bool) { return "proj", true }, logger.GetLogger()) + if fb != nil { + f = f.WithFallback(fb) + } + return f +} + +func serveFront(f *Front, path string) *httptest.ResponseRecorder { + w := httptest.NewRecorder() + f.ServeHTTP(w, httptest.NewRequest(http.MethodGet, PathPrefix+"a/d"+path, nil)) + return w +} + +// a path no local worker matches hands off to the fallback, which is given the +// resolved identity; when the fallback serves, the front writes nothing itself. +func TestFrontFallbackFires(t *testing.T) { + var got *FallbackRequest + f := fallbackFront(t, func(w http.ResponseWriter, _ *http.Request, fr *FallbackRequest) bool { + got = fr + w.WriteHeader(http.StatusTeapot) // stands in for a relayed response + return true + }, true) + + w := serveFront(f, "/unknown") + require.Equal(t, http.StatusTeapot, w.Code) + require.NotNil(t, got) + require.Equal(t, "proj", got.APIKey) + require.True(t, got.Authenticated) + require.Equal(t, "a", got.AgentName) + require.Equal(t, "d", got.Deployment) +} + +// a declined fallback with a local worker present falls through to the front's +// own 404 for the unmatched path. +func TestFrontFallbackDeclinedMapsStatus(t *testing.T) { + f := fallbackFront(t, func(http.ResponseWriter, *http.Request, *FallbackRequest) bool { return false }, true) + require.Equal(t, http.StatusNotFound, serveFront(f, "/unknown").Code) +} + +// a declined fallback with no local worker for the deployment falls through to +// 503. +func TestFrontFallbackDeclinedNoCandidates(t *testing.T) { + f := fallbackFront(t, func(http.ResponseWriter, *http.Request, *FallbackRequest) bool { return false }, false) + require.Equal(t, http.StatusServiceUnavailable, serveFront(f, "/unknown").Code) +} diff --git a/pkg/agent/endpoint/registry.go b/pkg/agent/endpoint/registry.go index ad0e27d6b..41c562364 100644 --- a/pkg/agent/endpoint/registry.go +++ b/pkg/agent/endpoint/registry.go @@ -19,8 +19,6 @@ import ( "errors" "slices" "sync" - - "github.com/livekit/protocol/livekit" ) // DefaultDeployment is the URL segment that addresses workers registered with an @@ -53,9 +51,6 @@ type Registration struct { AgentName string Deployment string Manifest *Manifest - // Endpoints is the raw manifest as declared, kept so a multi-node layer can - // replicate the route set for remote path matching without re-deriving it. - Endpoints []*livekit.AgentHttp_AgentEndpoint // Draining is provided by the control-plane layer that owns the worker; a // draining worker takes no new streams. Worker selection uses live in-flight diff --git a/pkg/agent/endpoint/webtransport_test.go b/pkg/agent/endpoint/webtransport_test.go index 7a30590c8..d1f7470b1 100644 --- a/pkg/agent/endpoint/webtransport_test.go +++ b/pkg/agent/endpoint/webtransport_test.go @@ -91,7 +91,6 @@ func handleSession(reg *endpoint.Registry, sess *webtransport.Session) { AgentName: rw.GetAgentName(), Deployment: rw.GetDeployment(), Manifest: manifest, - Endpoints: rw.GetEndpoints(), } registration.SetSession(endpoint.NewWebTransportSession(sess, endpoint.DefaultMaxStreams)) _ = reg.Register(registration) diff --git a/pkg/service/agentservice.go b/pkg/service/agentservice.go index 64fdb1a57..8d7beb3bd 100644 --- a/pkg/service/agentservice.go +++ b/pkg/service/agentservice.go @@ -354,7 +354,6 @@ func (h *AgentHandler) registerEndpoints(w *agent.Worker, sess endpoint.Session) AgentName: w.AgentName, Deployment: w.Deployment, Manifest: manifest, - Endpoints: w.Endpoints, Draining: w.Draining, } reg.SetSession(sess)