diff --git a/failover.go b/failover.go index 37fbf92a..224753da 100644 --- a/failover.go +++ b/failover.go @@ -188,32 +188,40 @@ func (t *failoverTransport) RoundTrip(req *http.Request) (*http.Response, error) req.Header.Set("User-Agent", userAgent) cfg := failoverConfigFromContext(req.Context()) - maxAttempts := cfg.attempts(req.URL.Hostname()) + hostname := req.URL.Hostname() + cloud := cfg.force || isCloud(hostname) + + // attempts() reflects general failover (5xx / transport errors): > 1 only when + // failover is enabled for a cloud host. + maxAttempts := cfg.attempts(hostname) // Each attempt gets the caller's full timeout budget, reset per attempt. A - // budget shorter than minFailoverTimeout skips failover (thundering-herd - // guard); a disabled or non-cloud host already has maxAttempts == 1. + // budget shorter than minFailoverTimeout skips general failover (thundering-herd + // guard). timeout := perAttemptBudget(req.Context()) if timeout > 0 && timeout < minFailoverTimeout { maxAttempts = 1 } - // No failover: a single attempt. Re-apply the budget in case - // withFailoverTimeout detached the deadline upstream (it can't know the host - // is non-cloud), so the lone attempt is still bounded. - if maxAttempts == 1 { + // A non-cloud host has nowhere to redirect: a single attempt. A cloud host + // always goes through the failover loop, even when general failover is off, so a + // 451 region-pin rejection can still be redirected. + if !cloud { ctx, cancel := withOptionalTimeout(req.Context(), timeout) resp, err := t.base.RoundTrip(req.WithContext(ctx)) return terminate(resp, err, cancel) } - return t.failover(req, maxAttempts, timeout, cfg.backoffBase) + return t.failover(req, maxAttempts > 1, timeout, cfg.backoffBase) } // failover replays the request against successive regions, each with a fresh -// timeout budget and exponential backoff, until success, a non-retryable error, -// caller cancellation, or the attempts/regions are exhausted. -func (t *failoverTransport) failover(req *http.Request, maxAttempts int, timeout, backoffBase time.Duration) (*http.Response, error) { +// timeout budget, until success, a non-retryable result, caller cancellation, or +// the regions/attempts are exhausted. A 451 region-pin rejection always triggers a +// redirect (the project can only be served by an allowed region); other retryable +// failures (5xx, transport errors, per-attempt timeouts) trigger one only when +// failoverEnabled. +func (t *failoverTransport) failover(req *http.Request, failoverEnabled bool, timeout, backoffBase time.Duration) (*http.Response, error) { // Buffer the body so it can be re-sent to each region. var body []byte if req.Body != nil { @@ -228,8 +236,13 @@ func (t *failoverTransport) failover(req *http.Request, maxAttempts int, timeout // Attempts run on a deadline-free context (the deadline is reset per attempt). // withFailoverTimeout normally detaches it upstream; detach a raw deadline // that reaches the transport directly (e.g. in tests) too. + // + // Only when general failover is on. A request that reaches this loop solely to + // allow a region-pin redirect (failover disabled, or a budget below + // minFailoverTimeout) keeps the caller's deadline, so the redirect happens but + // the call still returns within the time the caller allowed. base := req.Context() - if !hasPerAttemptTimeout(base) { + if failoverEnabled && !hasPerAttemptTimeout(base) { if _, ok := base.Deadline(); ok { base = detachDeadline(base) } @@ -241,7 +254,7 @@ func (t *failoverTransport) failover(req *http.Request, maxAttempts int, timeout var resp *http.Response var err error - for attempt := 0; attempt < maxAttempts; attempt++ { + for attempt := 0; attempt < failoverMaxAttempts; attempt++ { attemptCtx, cancel := withOptionalTimeout(base, timeout) r := req.Clone(attemptCtx) @@ -253,9 +266,13 @@ func (t *failoverTransport) failover(req *http.Request, maxAttempts int, timeout } resp, err = t.base.RoundTrip(r) - // Stop on success, a non-retryable 4xx, caller cancellation, or the last - // attempt. terminate defers the per-attempt cancel to the body's Close. - if !isRetryable(resp, err) || attempt == maxAttempts-1 { + // A 451 region-pin rejection always redirects; other failures redirect only + // when general failover is enabled. Stop on success, a non-retryable result, + // caller cancellation, or the last attempt. terminate defers the per-attempt + // cancel to the body's Close. + regionPin := isRegionPin(resp) + retry := regionPin || (failoverEnabled && isRetryable(resp, err)) + if !retry || attempt == failoverMaxAttempts-1 { return terminate(resp, err, cancel) } @@ -272,13 +289,19 @@ func (t *failoverTransport) failover(req *http.Request, maxAttempts int, timeout if resp != nil { status = resp.StatusCode } - logger.Warnw("livekit API request failed, retrying with fallback url", err, - "failedUrl", scheme+"://"+host, "fallbackUrl", nextScheme+"://"+nextHost, - "attempt", attempt+1, "maxAttempts", maxAttempts, "status", status) + reason := "unhealthy region" + if regionPin { + reason = "region pin" + } + logger.Warnw("livekit API request rejected, retrying with fallback url", err, + "reason", reason, "failedUrl", scheme+"://"+host, "fallbackUrl", nextScheme+"://"+nextHost, + "attempt", attempt+1, "maxAttempts", failoverMaxAttempts, "status", status) drainResponse(resp) cancel() // this attempt's body is drained; release its timer - if !sleepCtx(base, backoffBase<= 500 } +// regionPinStatus is the HTTP status LiveKit Cloud middleware returns when a +// project is pinned to a region other than the one that received the API request +// (451 Unavailable For Legal Reasons). Unlike a 5xx — a transient failure worth +// retrying anywhere — a 451 is definitive: the request can only succeed against an +// allowed region, so the SDK rediscovers regions and retries there even when +// general failover is disabled. +const regionPinStatus = http.StatusUnavailableForLegalReasons + +// isRegionPin reports whether resp is a region-pin rejection (see regionPinStatus). +func isRegionPin(resp *http.Response) bool { + return resp != nil && resp.StatusCode == regionPinStatus +} + // nextRegion returns the first region whose host has not yet been attempted. func nextRegion(settings *livekit.RegionSettings, attempted map[string]struct{}) (scheme, host string, ok bool) { if settings == nil { diff --git a/failover_apitest_test.go b/failover_apitest_test.go index 50509ddf..74fe4c6f 100644 --- a/failover_apitest_test.go +++ b/failover_apitest_test.go @@ -128,6 +128,43 @@ func TestAPI_FailoverDisabled(t *testing.T) { require.Error(t, err) } +// A project pinned to a region other than the one it reaches gets a 451; the +// client re-fetches /settings/regions (which returns only the pinned region) and +// retries there. The entry point in these tests is region 0. +func TestAPI_RegionPinRedirects(t *testing.T) { + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, mockControl{PinnedRegions: []string{"region-1"}}) + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.NoError(t, err, "a 451 should redirect to the pinned region from /settings/regions") +} + +// The pinned region need not be the next one; discovery returns it directly, so +// the client redirects straight to it. +func TestAPI_RegionPinRedirectsToDistantRegion(t *testing.T) { + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, mockControl{PinnedRegions: []string{"region-2"}}) + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.NoError(t, err, "should redirect to the pinned region even when it isn't the next one") +} + +// When the pinned region is unreachable (nothing in /settings/regions matches), +// the 451 is surfaced rather than retried forever. +func TestAPI_RegionPinNoReachableRegion(t *testing.T) { + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, mockControl{PinnedRegions: []string{"region-99"}}) + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.Error(t, err, "no reachable pinned region means the 451 is surfaced") +} + +// The region-pin redirect is always active — a correctness requirement, not +// resilience — so it engages even when the caller disables general failover. +func TestAPI_RegionPinRedirectCannotBeDisabled(t *testing.T) { + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := WithFailover(failoverCtx(t, mockControl{PinnedRegions: []string{"region-1"}}), false) + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.NoError(t, err, "region-pin redirect must work even with failover disabled") +} + // An unresponsive region (per-attempt timeout) should fail over to a healthy // region, with the deadline reset so the retry has its full budget. "delay" fail // mode stalls only the failing region, unlike DelayMs which delays every region. diff --git a/go.mod b/go.mod index a6743b5d..d0e34dec 100644 --- a/go.mod +++ b/go.mod @@ -8,11 +8,11 @@ require ( github.com/google/uuid v1.6.0 github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 - github.com/livekit/media-sdk v0.0.0-20260605212526-4c11a51d3c97 - github.com/livekit/mediatransportutil v0.0.0-20260605212259-862d4a7bcb1e - github.com/livekit/protocol v1.50.1 + github.com/livekit/media-sdk v0.0.0-20260812193843-5a5218b19550 + github.com/livekit/mediatransportutil v0.0.0-20260727210231-81a5287a7109 + github.com/livekit/protocol v1.50.4 github.com/magefile/mage v1.17.2 - github.com/moby/buildkit v0.32.0 + github.com/moby/buildkit v0.32.2 github.com/moby/patternmatcher v0.6.1 github.com/pion/dtls/v3 v3.1.5 github.com/pion/interceptor v0.1.47 @@ -23,8 +23,8 @@ require ( github.com/stretchr/testify v1.11.1 github.com/twitchtv/twirp v8.1.3+incompatible go.uber.org/atomic v1.11.0 - golang.org/x/crypto v0.54.0 - google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af + golang.org/x/crypto v0.55.0 + google.golang.org/protobuf v1.36.12 ) require ( @@ -132,12 +132,12 @@ require ( golang.org/x/net v0.57.0 // indirect golang.org/x/sync v0.22.0 golang.org/x/sys v0.47.0 // indirect - golang.org/x/text v0.40.0 // indirect + golang.org/x/text v0.41.0 // indirect golang.org/x/time v0.15.0 // indirect - golang.org/x/tools v0.47.0 // indirect + golang.org/x/tools v0.48.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect - google.golang.org/grpc v1.82.1 + google.golang.org/grpc v1.83.0 gopkg.in/hraban/opus.v2 v2.0.0-20230925203106-0188a62cb302 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index ba9d3978..cb9e390a 100644 --- a/go.sum +++ b/go.sum @@ -143,12 +143,12 @@ github.com/lithammer/shortuuid/v4 v4.2.0 h1:LMFOzVB3996a7b8aBuEXxqOBflbfPQAiVzkI github.com/lithammer/shortuuid/v4 v4.2.0/go.mod h1:D5noHZ2oFw/YaKCfGy0YxyE7M0wMbezmMjPdhyEFe6Y= github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5ATTo469PQPkqzdoU7be46ryiCDO3boc= github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= -github.com/livekit/media-sdk v0.0.0-20260605212526-4c11a51d3c97 h1:AyjUVuJuVd+5Kt+KEnIUoyZVAv8pejNDqs691I5e8jM= -github.com/livekit/media-sdk v0.0.0-20260605212526-4c11a51d3c97/go.mod h1:uWrLXY4JeLYynX39htMG49Dl4BhFYY+RCeoXaLdU+Lw= -github.com/livekit/mediatransportutil v0.0.0-20260605212259-862d4a7bcb1e h1:SkgQRcG2VYEhh80Qb/zYZo8rWKJzNfJcfUQnXe6su2M= -github.com/livekit/mediatransportutil v0.0.0-20260605212259-862d4a7bcb1e/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.50.1 h1:MOaLOFedKHQ1vteYjCFb+1Jypk3kQsV6h7gV4apgc+M= -github.com/livekit/protocol v1.50.1/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= +github.com/livekit/media-sdk v0.0.0-20260812193843-5a5218b19550 h1:aqaMkSNcx2GqCPsmGpkBN6MT5mlVyPZTECw5CxTAxOc= +github.com/livekit/media-sdk v0.0.0-20260812193843-5a5218b19550/go.mod h1:TuYRjSepaakL6ATsM9V2VMuksewW1PlhA32BG7Pxty0= +github.com/livekit/mediatransportutil v0.0.0-20260727210231-81a5287a7109 h1:jLE+M9fTj4HeTZSy4T7wD1M42uOeBnbzmoCoWDsTBQI= +github.com/livekit/mediatransportutil v0.0.0-20260727210231-81a5287a7109/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= +github.com/livekit/protocol v1.50.4 h1:Pzg9p1lpu9TcxUFJqIlUGNP6m7mX8PBT+ETxXRpnmZ8= +github.com/livekit/protocol v1.50.4/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= github.com/livekit/psrpc v0.7.2 h1:6oZ+NODJ2pLyaT6VqDq1F4Qc/3TpDUSpyphj/P9MhQc= github.com/livekit/psrpc v0.7.2/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= github.com/mackerelio/go-osstat v0.2.7 h1:TCavZi10wF49bT6iQZ9eT2keGZQpC69MTDfdJej5e94= @@ -157,8 +157,8 @@ github.com/magefile/mage v1.17.2 h1:fyXVu1eadI8Ap1HCCNgEhJ5McIWiYhLR8uol64ZZc40= github.com/magefile/mage v1.17.2/go.mod h1:Yj51kqllmsgFpvvSzgrZPK9WtluG3kUhFaBUVLo4feA= github.com/maxbrunsfeld/counterfeiter/v6 v6.12.2 h1:V23nK2R2B63g2GhygF9zVGpnigmhvoZoH8d0hrZwMGY= github.com/maxbrunsfeld/counterfeiter/v6 v6.12.2/go.mod h1:Mr897yU9FmyKaQDPtRlVKibrjz40XXyOHUfyZBPSyZU= -github.com/moby/buildkit v0.32.0 h1:slXarYQoMo4cp2d9x30M9t0L4R+c0CVMov+5P1hhiHY= -github.com/moby/buildkit v0.32.0/go.mod h1:Y10FBWvqxl/Wmhdzjee1Y2wQfjifTiwxENIUdaVNdME= +github.com/moby/buildkit v0.32.2 h1:Sfy7+u6dUv/2yuBc9KCoK70Re8atuV8aPZ5UOC068Vc= +github.com/moby/buildkit v0.32.2/go.mod h1:0GB/EJ1d+4VIVqIAgy3asaoGkVXy7IrDfVy7mPhOvg8= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= github.com/moby/locker v1.0.1 h1:fOXqR41zeveg4fFODix+1Ch4mj/gT0NE1XJbp/epuBg= @@ -340,8 +340,8 @@ go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= -golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw= -golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk= +golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= +golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= golang.org/x/exp v0.0.0-20260603202125-055de637280b h1:v1uXiEBHo8QA0LiGCo7UgHMzHT4Kdfpl2zmtH5vaP1Q= golang.org/x/exp v0.0.0-20260603202125-055de637280b/go.mod h1:d2fgXJLVs4dYDHUk5lwMIfzRzSrWCfGZb0ZqeLa/Vcw= golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk= @@ -355,22 +355,22 @@ golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w= -golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= -golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= -golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q= -golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA= +golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE= +golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY= google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE= -google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA= -google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af h1:+5/Sw3GsDNlEmu7TfklWKPdQ0Ykja5VEmq2i817+jbI= -google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +google.golang.org/grpc v1.83.0 h1:JeNZEKJFbQxArAMl+hiytHauacDNqJUllNfmIMmpqnQ= +google.golang.org/grpc v1.83.0/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= diff --git a/livekitapi_test.go b/livekitapi_test.go index 70841aff..8de46c70 100644 --- a/livekitapi_test.go +++ b/livekitapi_test.go @@ -64,6 +64,7 @@ type mockControl struct { Response json.RawMessage `json:"response,omitempty"` SkipAuth bool `json:"skipAuth,omitempty"` SIPStatus *sipStatus `json:"sipStatus,omitempty"` + PinnedRegions []string `json:"pinnedRegions,omitempty"` } // sipStatus fails a SIP dial method with a SIP response (code + optional reason).