From 7e4efd6d116c7439b1f939788dfc936ddc0e581f Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sat, 27 Jun 2026 23:14:27 +0200 Subject: [PATCH 1/6] feat: auto failover for API with LK Cloud automatically fail over 5xx and transport errors --- .github/workflows/test-api.yml | 57 ++++++ agent_client.go | 2 +- agent_dispatch_client.go | 3 +- agent_simulation_client.go | 3 +- apitest/failover_test.go | 139 +++++++++++++ connectorclient.go | 3 +- egressclient.go | 3 +- failover.go | 352 +++++++++++++++++++++++++++++++++ failover_internal_test.go | 41 ++++ ingressclient.go | 3 +- phonenumberclient.go | 3 +- regionurlprovider.go | 5 +- roomclient.go | 3 +- sipclient.go | 3 +- 14 files changed, 601 insertions(+), 19 deletions(-) create mode 100644 .github/workflows/test-api.yml create mode 100644 apitest/failover_test.go create mode 100644 failover.go create mode 100644 failover_internal_test.go diff --git a/.github/workflows/test-api.yml b/.github/workflows/test-api.yml new file mode 100644 index 00000000..a38e5afc --- /dev/null +++ b/.github/workflows/test-api.yml @@ -0,0 +1,57 @@ +# 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. + +name: Test API + +on: + workflow_dispatch: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + failover: + runs-on: ubuntu-latest + services: + mock-server: + image: livekit/test-server:latest + ports: + - 9999:9999 + - 10000:10000 + - 10001:10001 + - 10002:10002 + steps: + - uses: actions/checkout@1af3b93b6815bc44a9784bd300feb67ff0d1eeb3 # v6 + + - name: Set up Opus + run: | + sudo apt-get update + sudo apt-get install -y libsoxr-dev libopus-dev libopusfile-dev + + - name: Set up Go + uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c # v6 + with: + go-version: 1.26.0 + + - name: Wait for mock server + run: | + for i in $(seq 1 30); do + curl -sf http://127.0.0.1:9999/settings/regions >/dev/null && exit 0 + sleep 1 + done + echo "mock server did not become ready" && exit 1 + + - name: Run API tests + run: go test -v -count=1 ./apitest/... diff --git a/agent_client.go b/agent_client.go index 10fc9de1..076d663a 100644 --- a/agent_client.go +++ b/agent_client.go @@ -29,7 +29,7 @@ func NewAgentClient(url string, apiKey string, apiSecret string, opts ...AgentCl } c := &AgentClient{ authBase: authBase{apiKey, apiSecret}, - httpClient: &http.Client{}, + httpClient: newAPIHTTPClient(), } for _, opt := range opts { opt(c) diff --git a/agent_dispatch_client.go b/agent_dispatch_client.go index 71a724bd..f84b44d6 100644 --- a/agent_dispatch_client.go +++ b/agent_dispatch_client.go @@ -16,7 +16,6 @@ package lksdk import ( "context" - "net/http" "github.com/twitchtv/twirp" @@ -31,7 +30,7 @@ type AgentDispatchClient struct { func NewAgentDispatchServiceClient(url string, apiKey string, secretKey string, opts ...twirp.ClientOption) *AgentDispatchClient { url = signalling.ToHttpURL(url) - client := livekit.NewAgentDispatchServiceProtobufClient(url, &http.Client{}, opts...) + client := livekit.NewAgentDispatchServiceProtobufClient(url, newAPIHTTPClient(), opts...) return &AgentDispatchClient{ agentDispatchService: client, diff --git a/agent_simulation_client.go b/agent_simulation_client.go index 52552858..b2851468 100644 --- a/agent_simulation_client.go +++ b/agent_simulation_client.go @@ -16,7 +16,6 @@ package lksdk import ( "context" - "net/http" "github.com/twitchtv/twirp" @@ -29,7 +28,7 @@ type AgentSimulationClient struct { } func NewAgentSimulationClient(url string, apiKey string, apiSecret string, opts ...twirp.ClientOption) *AgentSimulationClient { - client := livekit.NewAgentSimulationProtobufClient(url, &http.Client{}, opts...) + client := livekit.NewAgentSimulationProtobufClient(url, newAPIHTTPClient(), opts...) return &AgentSimulationClient{ simulationClient: client, authBase: authBase{apiKey, apiSecret}, diff --git a/apitest/failover_test.go b/apitest/failover_test.go new file mode 100644 index 00000000..459beaab --- /dev/null +++ b/apitest/failover_test.go @@ -0,0 +1,139 @@ +// 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 apitest holds black-box API tests that exercise the public server SDK +// against the shared mock LiveKit API server (livekit/livekit cmd/test-server). +// Point them at a running instance with LK_TEST_SERVER_URL (default +// http://127.0.0.1:9999); they skip when no server is reachable. In CI the +// server is booted as a Docker container. +// +// See cmd/test-server/README.md for the X-Lk-Mock-* control protocol. Mock +// directives are passed to the SDK via twirp request headers, which the +// failover transport forwards to the discovery fetch and every retry. +package apitest + +import ( + "context" + "net/http" + "os" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/twitchtv/twirp" + + "github.com/livekit/protocol/livekit" + lksdk "github.com/livekit/server-sdk-go/v2" +) + +const ( + hdrFailRegions = "X-Lk-Mock-Fail-Regions" + hdrFailMode = "X-Lk-Mock-Fail-Mode" + hdrFailStatus = "X-Lk-Mock-Fail-Status" + hdrRegionsStat = "X-Lk-Mock-Regions-Status" +) + +func testServerURL(t *testing.T) string { + url := os.Getenv("LK_TEST_SERVER_URL") + if url == "" { + url = "http://127.0.0.1:9999" + } + resp, err := http.Get(url + "/settings/regions") + if err != nil { + t.Skipf("mock test server not reachable at %s (set LK_TEST_SERVER_URL): %v", url, err) + } + _ = resp.Body.Close() + return url +} + +// failoverCtx returns a context that forces failover on (the mock is not a +// cloud host) with a tiny backoff, carrying the given X-Lk-Mock-* directives as +// twirp request headers. +func failoverCtx(t *testing.T, mode lksdk.FailoverMode, directives map[string]string) context.Context { + ctx := lksdk.WithFailoverConfig(context.Background(), lksdk.FailoverConfig{ + Mode: mode, + BackoffBase: time.Millisecond, + }) + if len(directives) > 0 { + h := make(http.Header) + for k, v := range directives { + h.Set(k, v) + } + var err error + ctx, err = twirp.WithHTTPRequestHeaders(ctx, h) + require.NoError(t, err) + } + return ctx +} + +func TestAPI_Healthy(t *testing.T) { + client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") + room, err := client.CreateRoom(failoverCtx(t, lksdk.FailoverOn, nil), &livekit.CreateRoomRequest{Name: "api-test"}) + require.NoError(t, err) + require.Equal(t, "api-test", room.Name, "the mock echoes the request name") + require.NotEmpty(t, room.Sid) +} + +func TestAPI_PrimaryUnavailable(t *testing.T) { + client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0"}) + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.NoError(t, err, "should fail over to a healthy region") +} + +func TestAPI_TwoRegionsUnavailable(t *testing.T) { + client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0,1"}) + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.NoError(t, err, "should fail over to the third region") +} + +func TestAPI_AllUnavailable(t *testing.T) { + client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0,1,2,3"}) + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.Error(t, err) +} + +func TestAPI_ClientErrorNotRetried(t *testing.T) { + client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0", hdrFailStatus: "400"}) + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.Error(t, err) + var terr twirp.Error + require.ErrorAs(t, err, &terr) + require.Equal(t, twirp.InvalidArgument, terr.Code(), "a 4xx must surface as a typed error, not fail over") +} + +func TestAPI_TransportError(t *testing.T) { + client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0", hdrFailMode: "drop"}) + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.NoError(t, err, "a dropped connection should fail over to a healthy region") +} + +func TestAPI_RegionDiscoveryUnreachable(t *testing.T) { + client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0", hdrRegionsStat: "500"}) + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.Error(t, err, "no fallback hosts means the original error is surfaced") +} + +func TestAPI_FailoverDisabledForNonCloud(t *testing.T) { + client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") + // Auto mode against 127.0.0.1 (non-cloud) must not fail over. + ctx := failoverCtx(t, lksdk.FailoverAuto, map[string]string{hdrFailRegions: "0"}) + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.Error(t, err) +} diff --git a/connectorclient.go b/connectorclient.go index a67f3c68..c025627c 100644 --- a/connectorclient.go +++ b/connectorclient.go @@ -16,7 +16,6 @@ package lksdk import ( "context" - "net/http" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/utils/xtwirp" @@ -32,7 +31,7 @@ type ConnectorClient struct { func NewConnectorClient(url string, apiKey string, secretKey string, opts ...twirp.ClientOption) *ConnectorClient { opts = append(opts, xtwirp.DefaultClientOptions()...) url = signalling.ToHttpURL(url) - client := livekit.NewConnectorProtobufClient(url, &http.Client{}, opts...) + client := livekit.NewConnectorProtobufClient(url, newAPIHTTPClient(), opts...) return &ConnectorClient{ connector: client, authBase: authBase{ diff --git a/egressclient.go b/egressclient.go index 60d37f94..42173e61 100644 --- a/egressclient.go +++ b/egressclient.go @@ -16,7 +16,6 @@ package lksdk import ( "context" - "net/http" "github.com/twitchtv/twirp" @@ -33,7 +32,7 @@ type EgressClient struct { func NewEgressClient(url string, apiKey string, secretKey string, opts ...twirp.ClientOption) *EgressClient { opts = append(opts, xtwirp.DefaultClientOptions()...) url = signalling.ToHttpURL(url) - client := livekit.NewEgressProtobufClient(url, &http.Client{}, opts...) + client := livekit.NewEgressProtobufClient(url, newAPIHTTPClient(), opts...) return &EgressClient{ egressClient: client, authBase: authBase{ diff --git a/failover.go b/failover.go new file mode 100644 index 00000000..334a4741 --- /dev/null +++ b/failover.go @@ -0,0 +1,352 @@ +// 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 lksdk + +import ( + "bytes" + "context" + "io" + "net/http" + "net/url" + "strings" + "sync" + "time" + + "google.golang.org/protobuf/encoding/protojson" + + "github.com/livekit/protocol/livekit" + + "github.com/livekit/server-sdk-go/v2/signalling" +) + +// FailoverMode controls when API requests fail over to alternative LiveKit +// Cloud regions on a retryable error. +type FailoverMode int + +const ( + // FailoverAuto enables region failover only for LiveKit Cloud hosts. This + // is the default. + FailoverAuto FailoverMode = iota + // FailoverOn forces region failover on regardless of host. Primarily for + // tests pointing at a non-cloud host (e.g. a local mock server). + FailoverOn + // FailoverOff disables region failover entirely. + FailoverOff +) + +const ( + defaultMaxAttempts = 3 + defaultBackoffBase = 200 * time.Millisecond +) + +// FailoverConfig tunes the region-failover retry loop. A zero value is valid +// and uses the defaults (auto mode, 3 attempts, 200ms base backoff). +type FailoverConfig struct { + // Mode selects when failover is active. Defaults to FailoverAuto. + Mode FailoverMode + // MaxAttempts is the total number of attempts including the first, so the + // original host plus (MaxAttempts-1) fallback regions. Defaults to 3. + MaxAttempts int + // BackoffBase is the initial delay before the first retry; each subsequent + // retry doubles it. Defaults to 200ms. Set to 0 to retry without delay. + BackoffBase time.Duration +} + +func (c FailoverConfig) withDefaults() FailoverConfig { + if c.MaxAttempts <= 0 { + c.MaxAttempts = defaultMaxAttempts + } + if c.BackoffBase < 0 { + c.BackoffBase = 0 + } else if c.BackoffBase == 0 { + c.BackoffBase = defaultBackoffBase + } + return c +} + +func (c FailoverConfig) enabledFor(hostname string) bool { + switch c.Mode { + case FailoverOff: + return false + case FailoverOn: + return true + default: + return isCloud(hostname) + } +} + +type failoverConfigKey struct{} + +// WithFailoverConfig returns a context that overrides the region-failover +// behavior for API requests made with it. Pass the returned context to any +// service client method. +func WithFailoverConfig(ctx context.Context, cfg FailoverConfig) context.Context { + return context.WithValue(ctx, failoverConfigKey{}, cfg) +} + +func failoverConfigFromContext(ctx context.Context) FailoverConfig { + if cfg, ok := ctx.Value(failoverConfigKey{}).(FailoverConfig); ok { + return cfg.withDefaults() + } + return FailoverConfig{}.withDefaults() +} + +// newAPIHTTPClient returns the *http.Client used by every API service client. +// It wraps the default transport with region-failover retries. +func newAPIHTTPClient() *http.Client { + return &http.Client{ + Transport: &failoverTransport{base: http.DefaultTransport, regions: sharedAPIRegions}, + } +} + +// failoverTransport orchestrates region failover for a single API request. On +// a retryable failure (any transport error or HTTP 5xx) it discovers +// alternative regions via /settings/regions and replays the request — body and +// headers intact — against the next region, with exponential backoff. 4xx +// responses are returned immediately. +type failoverTransport struct { + base http.RoundTripper + regions *apiRegionCache +} + +func (t *failoverTransport) RoundTrip(req *http.Request) (*http.Response, error) { + cfg := failoverConfigFromContext(req.Context()) + enabled := cfg.enabledFor(req.URL.Hostname()) + + // Buffer the body once so it can be replayed against each region. + var body []byte + if req.Body != nil { + b, err := io.ReadAll(req.Body) + _ = req.Body.Close() + if err != nil { + return nil, err + } + body = b + } + + originalScheme, originalHost := req.URL.Scheme, req.URL.Host + attempted := map[string]struct{}{strings.ToLower(originalHost): {}} + + maxAttempts := cfg.MaxAttempts + if !enabled { + maxAttempts = 1 + } + + var ( + settings *livekit.RegionSettings + fetchedRegions bool + resp *http.Response + err error + ) + scheme, host := originalScheme, originalHost + + for attempt := 0; attempt < maxAttempts; attempt++ { + if ctxErr := req.Context().Err(); ctxErr != nil { + return nil, ctxErr + } + + r := req.Clone(req.Context()) + r.URL.Scheme = scheme + r.URL.Host = host + r.Host = host + if body != nil { + buf := body + r.Body = io.NopCloser(bytes.NewReader(buf)) + r.ContentLength = int64(len(buf)) + r.GetBody = func() (io.ReadCloser, error) { return io.NopCloser(bytes.NewReader(buf)), nil } + } + + resp, err = t.base.RoundTrip(r) + if !isRetryable(resp, err) { + return resp, err + } + if attempt == maxAttempts-1 { + break // out of attempts; surface the last result + } + + // discover regions lazily + if !fetchedRegions { + settings = t.regions.get(originalScheme, originalHost, req.Header) + fetchedRegions = true + } + nextScheme, nextHost, ok := nextRegion(settings, attempted) + if !ok { + break // no untried region left + } + + drainResponse(resp) + if !sleepCtx(req.Context(), cfg.BackoffBase<= 500 +} + +// 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 { + return "", "", false + } + for _, region := range settings.Regions { + u, err := url.Parse(signalling.ToHttpURL(region.Url)) + if err != nil || u.Host == "" { + continue + } + if _, seen := attempted[strings.ToLower(u.Host)]; seen { + continue + } + return u.Scheme, u.Host, true + } + return "", "", false +} + +func drainResponse(resp *http.Response) { + if resp != nil && resp.Body != nil { + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + } +} + +func sleepCtx(ctx context.Context, d time.Duration) bool { + if d <= 0 { + return ctx.Err() == nil + } + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} + +var sharedAPIRegions = newAPIRegionCache() + +// apiRegionCache fetches and caches the LiveKit Cloud region list per host for +// the API failover path. Unlike the signaling regionURLProvider it honors the +// request scheme (so it works against an http mock server) and forwards the +// caller's headers (auth plus any custom headers) to the discovery endpoint. +type apiRegionCache struct { + mu sync.Mutex + client *http.Client + cache map[string]*apiRegionEntry +} + +type apiRegionEntry struct { + settings *livekit.RegionSettings + fetchedAt time.Time + ttl time.Duration +} + +func newAPIRegionCache() *apiRegionCache { + return &apiRegionCache{ + client: &http.Client{Timeout: 5 * time.Second}, + cache: make(map[string]*apiRegionEntry), + } +} + +// get returns the region list for host, fetching it if the cache is stale. +// It is best-effort: on a fetch failure it falls back to a stale cached list, +// and returns nil if nothing is available (the caller then stops failing over). +func (c *apiRegionCache) get(scheme, host string, headers http.Header) *livekit.RegionSettings { + key := strings.ToLower(host) + + c.mu.Lock() + if entry := c.cache[key]; entry != nil && time.Since(entry.fetchedAt) < entry.ttl { + defer c.mu.Unlock() + return entry.settings + } + c.mu.Unlock() + + settings, ttl, err := c.fetch(scheme, host, headers) + if err != nil { + c.mu.Lock() + defer c.mu.Unlock() + if entry := c.cache[key]; entry != nil { + return entry.settings // serve stale on failure + } + return nil + } + + // A zero TTL (e.g. Cache-Control: max-age=0) means "do not cache". + if ttl > 0 { + c.mu.Lock() + c.cache[key] = &apiRegionEntry{settings: settings, fetchedAt: time.Now(), ttl: ttl} + c.mu.Unlock() + } + return settings +} + +func (c *apiRegionCache) fetch(scheme, host string, headers http.Header) (*livekit.RegionSettings, time.Duration, error) { + u := url.URL{Scheme: scheme, Host: host, Path: "/settings/regions"} + req, err := http.NewRequest(http.MethodGet, u.String(), nil) + if err != nil { + return nil, 0, err + } + // Forward the caller's headers (Authorization and any custom headers), + // minus body-specific ones, so a validly-signed token reaches the + // discovery endpoint and test directives propagate. + for k, vv := range headers { + switch http.CanonicalHeaderKey(k) { + case "Content-Type", "Content-Length": + continue + } + for _, v := range vv { + req.Header.Add(k, v) + } + } + + resp, err := c.client.Do(req) + if err != nil { + return nil, 0, err + } + defer drainResponse(resp) + + if resp.StatusCode != http.StatusOK { + return nil, 0, &TwirpRegionError{StatusCode: resp.StatusCode} + } + b, err := io.ReadAll(resp.Body) + if err != nil { + return nil, 0, err + } + settings := &livekit.RegionSettings{} + if err := protojson.Unmarshal(b, settings); err != nil { + return nil, 0, err + } + ttl := parseRegionSettingsMaxAge(resp.Header.Get("Cache-Control")) + return settings, ttl, nil +} + +// TwirpRegionError indicates the /settings/regions endpoint returned a non-200 +// status while attempting region failover. +type TwirpRegionError struct { + StatusCode int +} + +func (e *TwirpRegionError) Error() string { + return "failed to fetch region settings: status " + http.StatusText(e.StatusCode) +} diff --git a/failover_internal_test.go b/failover_internal_test.go new file mode 100644 index 00000000..6ed10a13 --- /dev/null +++ b/failover_internal_test.go @@ -0,0 +1,41 @@ +// 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 lksdk + +import "testing" + +func TestFailoverEnabledFor(t *testing.T) { + cases := []struct { + mode FailoverMode + host string + want bool + }{ + // Auto: only *.livekit.cloud project domains. + {FailoverAuto, "myproject.livekit.cloud", true}, + {FailoverAuto, "myproject.region.livekit.cloud", true}, + {FailoverAuto, "myproject.livekit.io", false}, + {FailoverAuto, "example.com", false}, + {FailoverAuto, "127.0.0.1", false}, + {FailoverAuto, "notlivekit.cloud", false}, + // On/Off override the host check. + {FailoverOn, "127.0.0.1", true}, + {FailoverOff, "myproject.livekit.cloud", false}, + } + for _, c := range cases { + if got := (FailoverConfig{Mode: c.mode}).enabledFor(c.host); got != c.want { + t.Errorf("enabledFor(mode=%v, host=%q) = %v, want %v", c.mode, c.host, got, c.want) + } + } +} diff --git a/ingressclient.go b/ingressclient.go index da3e4a0e..3e000846 100644 --- a/ingressclient.go +++ b/ingressclient.go @@ -16,7 +16,6 @@ package lksdk import ( "context" - "net/http" "github.com/twitchtv/twirp" @@ -33,7 +32,7 @@ type IngressClient struct { func NewIngressClient(url string, apiKey string, secretKey string, opts ...twirp.ClientOption) *IngressClient { opts = append(opts, xtwirp.DefaultClientOptions()...) url = signalling.ToHttpURL(url) - client := livekit.NewIngressProtobufClient(url, &http.Client{}, opts...) + client := livekit.NewIngressProtobufClient(url, newAPIHTTPClient(), opts...) return &IngressClient{ ingressClient: client, authBase: authBase{ diff --git a/phonenumberclient.go b/phonenumberclient.go index ba9c0824..4f2b332b 100644 --- a/phonenumberclient.go +++ b/phonenumberclient.go @@ -16,7 +16,6 @@ package lksdk import ( "context" - "net/http" "time" "github.com/twitchtv/twirp" @@ -35,7 +34,7 @@ type PhoneNumberClient struct { func NewPhoneNumberClient(url string, apiKey string, secretKey string, opts ...twirp.ClientOption) *PhoneNumberClient { opts = append(opts, xtwirp.DefaultClientOptions()...) return &PhoneNumberClient{ - phoneNumberClient: livekit.NewPhoneNumberServiceProtobufClient(signalling.ToHttpURL(url), &http.Client{}, opts...), + phoneNumberClient: livekit.NewPhoneNumberServiceProtobufClient(signalling.ToHttpURL(url), newAPIHTTPClient(), opts...), authBase: authBase{ apiKey: apiKey, apiSecret: secretKey, diff --git a/regionurlprovider.go b/regionurlprovider.go index dbdeeae9..33f67207 100644 --- a/regionurlprovider.go +++ b/regionurlprovider.go @@ -182,7 +182,8 @@ func parseCloudURL(serverURL string) (string, error) { return parsedURL.Hostname(), nil } -// isCloud reports whether the hostname belongs to LiveKit Cloud. +// isCloud reports whether the hostname belongs to a LiveKit Cloud project +// (a *.livekit.cloud subdomain). var isCloud = func(hostname string) bool { - return strings.HasSuffix(hostname, "livekit.cloud") || strings.HasSuffix(hostname, "livekit.io") + return strings.HasSuffix(hostname, ".livekit.cloud") } diff --git a/roomclient.go b/roomclient.go index 86e89844..adfad779 100644 --- a/roomclient.go +++ b/roomclient.go @@ -16,7 +16,6 @@ package lksdk import ( "context" - "net/http" "github.com/google/uuid" "github.com/twitchtv/twirp" @@ -35,7 +34,7 @@ type RoomServiceClient struct { func NewRoomServiceClient(url string, apiKey string, secretKey string, opts ...twirp.ClientOption) *RoomServiceClient { opts = append(opts, xtwirp.DefaultClientOptions()...) url = signalling.ToHttpURL(url) - client := livekit.NewRoomServiceProtobufClient(url, &http.Client{}, opts...) + client := livekit.NewRoomServiceProtobufClient(url, newAPIHTTPClient(), opts...) return &RoomServiceClient{ roomService: client, authBase: authBase{ diff --git a/sipclient.go b/sipclient.go index 1fa0faed..acded63f 100644 --- a/sipclient.go +++ b/sipclient.go @@ -16,7 +16,6 @@ package lksdk import ( "context" - "net/http" "time" "github.com/twitchtv/twirp" @@ -38,7 +37,7 @@ type SIPClient struct { func NewSIPClient(url string, apiKey string, secretKey string, opts ...twirp.ClientOption) *SIPClient { opts = append(opts, xtwirp.DefaultClientOptions()...) return &SIPClient{ - sipClient: livekit.NewSIPProtobufClient(signalling.ToHttpURL(url), &http.Client{}, opts...), + sipClient: livekit.NewSIPProtobufClient(signalling.ToHttpURL(url), newAPIHTTPClient(), opts...), authBase: authBase{ apiKey: apiKey, apiSecret: secretKey, From b952fb63e25c0c8c0b08ff5e8f5a514e3bc70518 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sat, 27 Jun 2026 23:17:50 +0200 Subject: [PATCH 2/6] fix codeql perms --- .github/workflows/test-api.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/test-api.yml b/.github/workflows/test-api.yml index a38e5afc..f439c318 100644 --- a/.github/workflows/test-api.yml +++ b/.github/workflows/test-api.yml @@ -14,6 +14,9 @@ name: Test API +permissions: + contents: read + on: workflow_dispatch: push: From 2e82b8c86ccd279e1340232326c284cbd7950bc4 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sun, 28 Jun 2026 09:12:08 +0200 Subject: [PATCH 3/6] tweaks --- apitest/failover_test.go | 2 +- failover.go | 28 ++++++++++++++++++---------- failover_internal_test.go | 2 +- 3 files changed, 20 insertions(+), 12 deletions(-) diff --git a/apitest/failover_test.go b/apitest/failover_test.go index 459beaab..6a019ee3 100644 --- a/apitest/failover_test.go +++ b/apitest/failover_test.go @@ -61,7 +61,7 @@ func testServerURL(t *testing.T) string { // cloud host) with a tiny backoff, carrying the given X-Lk-Mock-* directives as // twirp request headers. func failoverCtx(t *testing.T, mode lksdk.FailoverMode, directives map[string]string) context.Context { - ctx := lksdk.WithFailoverConfig(context.Background(), lksdk.FailoverConfig{ + ctx := lksdk.WithFailoverOptions(context.Background(), lksdk.FailoverOptions{ Mode: mode, BackoffBase: time.Millisecond, }) diff --git a/failover.go b/failover.go index 334a4741..6ccedf48 100644 --- a/failover.go +++ b/failover.go @@ -51,9 +51,9 @@ const ( defaultBackoffBase = 200 * time.Millisecond ) -// FailoverConfig tunes the region-failover retry loop. A zero value is valid +// FailoverOptions tunes the region-failover retry loop. A zero value is valid // and uses the defaults (auto mode, 3 attempts, 200ms base backoff). -type FailoverConfig struct { +type FailoverOptions struct { // Mode selects when failover is active. Defaults to FailoverAuto. Mode FailoverMode // MaxAttempts is the total number of attempts including the first, so the @@ -64,7 +64,7 @@ type FailoverConfig struct { BackoffBase time.Duration } -func (c FailoverConfig) withDefaults() FailoverConfig { +func (c FailoverOptions) withDefaults() FailoverOptions { if c.MaxAttempts <= 0 { c.MaxAttempts = defaultMaxAttempts } @@ -76,7 +76,7 @@ func (c FailoverConfig) withDefaults() FailoverConfig { return c } -func (c FailoverConfig) enabledFor(hostname string) bool { +func (c FailoverOptions) enabledFor(hostname string) bool { switch c.Mode { case FailoverOff: return false @@ -89,18 +89,18 @@ func (c FailoverConfig) enabledFor(hostname string) bool { type failoverConfigKey struct{} -// WithFailoverConfig returns a context that overrides the region-failover +// WithFailoverOptions returns a context that overrides the region-failover // behavior for API requests made with it. Pass the returned context to any // service client method. -func WithFailoverConfig(ctx context.Context, cfg FailoverConfig) context.Context { +func WithFailoverOptions(ctx context.Context, cfg FailoverOptions) context.Context { return context.WithValue(ctx, failoverConfigKey{}, cfg) } -func failoverConfigFromContext(ctx context.Context) FailoverConfig { - if cfg, ok := ctx.Value(failoverConfigKey{}).(FailoverConfig); ok { +func failoverConfigFromContext(ctx context.Context) FailoverOptions { + if cfg, ok := ctx.Value(failoverConfigKey{}).(FailoverOptions); ok { return cfg.withDefaults() } - return FailoverConfig{}.withDefaults() + return FailoverOptions{}.withDefaults() } // newAPIHTTPClient returns the *http.Client used by every API service client. @@ -186,6 +186,14 @@ func (t *failoverTransport) RoundTrip(req *http.Request) (*http.Response, error) break // no untried region left } + status := 0 + 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) + drainResponse(resp) if !sleepCtx(req.Context(), cfg.BackoffBase< Date: Sun, 28 Jun 2026 16:35:35 +0200 Subject: [PATCH 4/6] address feedback --- .github/workflows/test-api.yml | 2 +- failover.go | 219 +++++------------- ...ilover_test.go => failover_apitest_test.go | 74 +++--- failover_internal_test.go | 31 +-- regionurlprovider.go | 217 +++++++++-------- regionurlprovider_test.go | 8 +- 6 files changed, 239 insertions(+), 312 deletions(-) rename apitest/failover_test.go => failover_apitest_test.go (57%) diff --git a/.github/workflows/test-api.yml b/.github/workflows/test-api.yml index f439c318..c2f7f5c5 100644 --- a/.github/workflows/test-api.yml +++ b/.github/workflows/test-api.yml @@ -57,4 +57,4 @@ jobs: echo "mock server did not become ready" && exit 1 - name: Run API tests - run: go test -v -count=1 ./apitest/... + run: go test -v -count=1 -run '^TestAPI_' . diff --git a/failover.go b/failover.go index 6ccedf48..bf08ee8f 100644 --- a/failover.go +++ b/failover.go @@ -21,86 +21,68 @@ import ( "net/http" "net/url" "strings" - "sync" "time" - "google.golang.org/protobuf/encoding/protojson" - "github.com/livekit/protocol/livekit" "github.com/livekit/server-sdk-go/v2/signalling" ) -// FailoverMode controls when API requests fail over to alternative LiveKit -// Cloud regions on a retryable error. -type FailoverMode int - -const ( - // FailoverAuto enables region failover only for LiveKit Cloud hosts. This - // is the default. - FailoverAuto FailoverMode = iota - // FailoverOn forces region failover on regardless of host. Primarily for - // tests pointing at a non-cloud host (e.g. a local mock server). - FailoverOn - // FailoverOff disables region failover entirely. - FailoverOff -) - +// Total attempts (the original request plus fallback regions) and the base +// retry backoff are fixed, not user-configurable, so retries can't be tuned to +// values that could overwhelm the server. const ( - defaultMaxAttempts = 3 - defaultBackoffBase = 200 * time.Millisecond + failoverMaxAttempts = 3 + failoverBackoffBase = 200 * time.Millisecond ) -// FailoverOptions tunes the region-failover retry loop. A zero value is valid -// and uses the defaults (auto mode, 3 attempts, 200ms base backoff). -type FailoverOptions struct { - // Mode selects when failover is active. Defaults to FailoverAuto. - Mode FailoverMode - // MaxAttempts is the total number of attempts including the first, so the - // original host plus (MaxAttempts-1) fallback regions. Defaults to 3. - MaxAttempts int - // BackoffBase is the initial delay before the first retry; each subsequent - // retry doubles it. Defaults to 200ms. Set to 0 to retry without delay. - BackoffBase time.Duration +// failoverConfig is the resolved per-request region-failover configuration. The +// public API exposes only the enabled toggle (default true); force and +// backoffBase are internal test-only knobs. +type failoverConfig struct { + enabled bool + // force bypasses the cloud-host check. Internal testing only. + force bool + // backoffBase overrides the retry backoff base. Internal testing only. + backoffBase time.Duration } -func (c FailoverOptions) withDefaults() FailoverOptions { - if c.MaxAttempts <= 0 { - c.MaxAttempts = defaultMaxAttempts - } - if c.BackoffBase < 0 { - c.BackoffBase = 0 - } else if c.BackoffBase == 0 { - c.BackoffBase = defaultBackoffBase +// attempts returns the total request attempts for a host; 1 means no failover. +// Failover only engages when enabled and the host is a LiveKit Cloud domain. +// force bypasses the cloud-host check and is for internal testing only. +func (c failoverConfig) attempts(hostname string) int { + if c.enabled && (c.force || isCloud(hostname)) { + return failoverMaxAttempts } - return c + return 1 } -func (c FailoverOptions) enabledFor(hostname string) bool { - switch c.Mode { - case FailoverOff: - return false - case FailoverOn: - return true - default: - return isCloud(hostname) - } -} +type failoverEnabledKey struct{} +type failoverForceKey struct{} -type failoverConfigKey struct{} +// WithFailover returns a context that enables or disables region failover for +// API requests made with it (enabled by default). Failover only engages for +// LiveKit Cloud hosts. Pass the returned context to any service client method. +func WithFailover(ctx context.Context, enabled bool) context.Context { + return context.WithValue(ctx, failoverEnabledKey{}, enabled) +} -// WithFailoverOptions returns a context that overrides the region-failover -// behavior for API requests made with it. Pass the returned context to any -// service client method. -func WithFailoverOptions(ctx context.Context, cfg FailoverOptions) context.Context { - return context.WithValue(ctx, failoverConfigKey{}, cfg) +// withFailoverForce returns a context that forces failover on regardless of +// host and overrides the retry backoff. Internal, test-only. +func withFailoverForce(ctx context.Context, backoff time.Duration) context.Context { + return context.WithValue(ctx, failoverForceKey{}, backoff) } -func failoverConfigFromContext(ctx context.Context) FailoverOptions { - if cfg, ok := ctx.Value(failoverConfigKey{}).(FailoverOptions); ok { - return cfg.withDefaults() +func failoverConfigFromContext(ctx context.Context) failoverConfig { + cfg := failoverConfig{enabled: true, backoffBase: failoverBackoffBase} + if enabled, ok := ctx.Value(failoverEnabledKey{}).(bool); ok { + cfg.enabled = enabled + } + if backoff, ok := ctx.Value(failoverForceKey{}).(time.Duration); ok { + cfg.force = true + cfg.backoffBase = backoff } - return FailoverOptions{}.withDefaults() + return cfg } // newAPIHTTPClient returns the *http.Client used by every API service client. @@ -118,12 +100,12 @@ func newAPIHTTPClient() *http.Client { // responses are returned immediately. type failoverTransport struct { base http.RoundTripper - regions *apiRegionCache + regions *regionCache } func (t *failoverTransport) RoundTrip(req *http.Request) (*http.Response, error) { cfg := failoverConfigFromContext(req.Context()) - enabled := cfg.enabledFor(req.URL.Hostname()) + maxAttempts := cfg.attempts(req.URL.Hostname()) // Buffer the body once so it can be replayed against each region. var body []byte @@ -139,11 +121,6 @@ func (t *failoverTransport) RoundTrip(req *http.Request) (*http.Response, error) originalScheme, originalHost := req.URL.Scheme, req.URL.Host attempted := map[string]struct{}{strings.ToLower(originalHost): {}} - maxAttempts := cfg.MaxAttempts - if !enabled { - maxAttempts = 1 - } - var ( settings *livekit.RegionSettings fetchedRegions bool @@ -176,9 +153,11 @@ func (t *failoverTransport) RoundTrip(req *http.Request) (*http.Response, error) break // out of attempts; surface the last result } - // discover regions lazily + // discover regions lazily; honor the request scheme (so it works against + // an http mock) and forward the caller's headers to the discovery fetch. if !fetchedRegions { - settings = t.regions.get(originalScheme, originalHost, req.Header) + discoveryURL := url.URL{Scheme: originalScheme, Host: originalHost, Path: "/settings/regions"} + settings, _ = t.regions.get(originalHost, discoveryURL.String(), req.Header, 0) fetchedRegions = true } nextScheme, nextHost, ok := nextRegion(settings, attempted) @@ -195,7 +174,7 @@ func (t *failoverTransport) RoundTrip(req *http.Request) (*http.Response, error) "attempt", attempt+1, "maxAttempts", maxAttempts, "status", status) drainResponse(resp) - if !sleepCtx(req.Context(), cfg.BackoffBase< 0 { - c.mu.Lock() - c.cache[key] = &apiRegionEntry{settings: settings, fetchedAt: time.Now(), ttl: ttl} - c.mu.Unlock() - } - return settings -} - -func (c *apiRegionCache) fetch(scheme, host string, headers http.Header) (*livekit.RegionSettings, time.Duration, error) { - u := url.URL{Scheme: scheme, Host: host, Path: "/settings/regions"} - req, err := http.NewRequest(http.MethodGet, u.String(), nil) - if err != nil { - return nil, 0, err - } - // Forward the caller's headers (Authorization and any custom headers), - // minus body-specific ones, so a validly-signed token reaches the - // discovery endpoint and test directives propagate. - for k, vv := range headers { - switch http.CanonicalHeaderKey(k) { - case "Content-Type", "Content-Length": - continue - } - for _, v := range vv { - req.Header.Add(k, v) - } - } - - resp, err := c.client.Do(req) - if err != nil { - return nil, 0, err - } - defer drainResponse(resp) - - if resp.StatusCode != http.StatusOK { - return nil, 0, &TwirpRegionError{StatusCode: resp.StatusCode} - } - b, err := io.ReadAll(resp.Body) - if err != nil { - return nil, 0, err - } - settings := &livekit.RegionSettings{} - if err := protojson.Unmarshal(b, settings); err != nil { - return nil, 0, err - } - ttl := parseRegionSettingsMaxAge(resp.Header.Get("Cache-Control")) - return settings, ttl, nil -} +// sharedAPIRegions is the process-wide region cache used by the API failover +// path. The RTC signaling path has its own regionURLProvider over a separate +// cache, but both share the same regionCache discovery/fetch/TTL logic. +var sharedAPIRegions = newRegionCache() // TwirpRegionError indicates the /settings/regions endpoint returned a non-200 // status while attempting region failover. diff --git a/apitest/failover_test.go b/failover_apitest_test.go similarity index 57% rename from apitest/failover_test.go rename to failover_apitest_test.go index 6a019ee3..4752e5f3 100644 --- a/apitest/failover_test.go +++ b/failover_apitest_test.go @@ -12,16 +12,18 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package apitest holds black-box API tests that exercise the public server SDK -// against the shared mock LiveKit API server (livekit/livekit cmd/test-server). -// Point them at a running instance with LK_TEST_SERVER_URL (default -// http://127.0.0.1:9999); they skip when no server is reachable. In CI the -// server is booted as a Docker container. +// API failover tests that exercise the public server SDK against the shared +// mock LiveKit API server (livekit/livekit cmd/test-server). Point them at a +// running instance with LK_TEST_SERVER_URL (default http://127.0.0.1:9999); +// they skip when no server is reachable. In CI the server is booted as a Docker +// container. // // See cmd/test-server/README.md for the X-Lk-Mock-* control protocol. Mock // directives are passed to the SDK via twirp request headers, which the -// failover transport forwards to the discovery fetch and every retry. -package apitest +// failover transport forwards to the discovery fetch and every retry. These +// tests are in-package so they can use the internal test-force hook (the public +// API only exposes WithFailover(ctx, bool), which is cloud-gated). +package lksdk import ( "context" @@ -34,7 +36,6 @@ import ( "github.com/twitchtv/twirp" "github.com/livekit/protocol/livekit" - lksdk "github.com/livekit/server-sdk-go/v2" ) const ( @@ -59,12 +60,9 @@ func testServerURL(t *testing.T) string { // failoverCtx returns a context that forces failover on (the mock is not a // cloud host) with a tiny backoff, carrying the given X-Lk-Mock-* directives as -// twirp request headers. -func failoverCtx(t *testing.T, mode lksdk.FailoverMode, directives map[string]string) context.Context { - ctx := lksdk.WithFailoverOptions(context.Background(), lksdk.FailoverOptions{ - Mode: mode, - BackoffBase: time.Millisecond, - }) +// twirp request headers. force/backoff are internal, test-only knobs. +func failoverCtx(t *testing.T, directives map[string]string) context.Context { + ctx := withFailoverForce(context.Background(), time.Millisecond) if len(directives) > 0 { h := make(http.Header) for k, v := range directives { @@ -78,37 +76,37 @@ func failoverCtx(t *testing.T, mode lksdk.FailoverMode, directives map[string]st } func TestAPI_Healthy(t *testing.T) { - client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") - room, err := client.CreateRoom(failoverCtx(t, lksdk.FailoverOn, nil), &livekit.CreateRoomRequest{Name: "api-test"}) + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + room, err := client.CreateRoom(failoverCtx(t, nil), &livekit.CreateRoomRequest{Name: "api-test"}) require.NoError(t, err) require.Equal(t, "api-test", room.Name, "the mock echoes the request name") require.NotEmpty(t, room.Sid) } func TestAPI_PrimaryUnavailable(t *testing.T) { - client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") - ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0"}) + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, map[string]string{hdrFailRegions: "0"}) _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) require.NoError(t, err, "should fail over to a healthy region") } func TestAPI_TwoRegionsUnavailable(t *testing.T) { - client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") - ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0,1"}) + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, map[string]string{hdrFailRegions: "0,1"}) _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) require.NoError(t, err, "should fail over to the third region") } func TestAPI_AllUnavailable(t *testing.T) { - client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") - ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0,1,2,3"}) + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, map[string]string{hdrFailRegions: "0,1,2,3"}) _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) require.Error(t, err) } func TestAPI_ClientErrorNotRetried(t *testing.T) { - client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") - ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0", hdrFailStatus: "400"}) + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, map[string]string{hdrFailRegions: "0", hdrFailStatus: "400"}) _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) require.Error(t, err) var terr twirp.Error @@ -117,23 +115,35 @@ func TestAPI_ClientErrorNotRetried(t *testing.T) { } func TestAPI_TransportError(t *testing.T) { - client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") - ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0", hdrFailMode: "drop"}) + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, map[string]string{hdrFailRegions: "0", hdrFailMode: "drop"}) _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) require.NoError(t, err, "a dropped connection should fail over to a healthy region") } func TestAPI_RegionDiscoveryUnreachable(t *testing.T) { - client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") - ctx := failoverCtx(t, lksdk.FailoverOn, map[string]string{hdrFailRegions: "0", hdrRegionsStat: "500"}) + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, map[string]string{hdrFailRegions: "0", hdrRegionsStat: "500"}) _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) require.Error(t, err, "no fallback hosts means the original error is surfaced") } -func TestAPI_FailoverDisabledForNonCloud(t *testing.T) { - client := lksdk.NewRoomServiceClient(testServerURL(t), "devkey", "secret") - // Auto mode against 127.0.0.1 (non-cloud) must not fail over. - ctx := failoverCtx(t, lksdk.FailoverAuto, map[string]string{hdrFailRegions: "0"}) +func TestAPI_FailoverNotCloudHost(t *testing.T) { + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + // Enabled (the default) but not forced; 127.0.0.1 is not a cloud host, so + // failover must not engage. + h := make(http.Header) + h.Set(hdrFailRegions, "0") + ctx, err := twirp.WithHTTPRequestHeaders(context.Background(), h) + require.NoError(t, err) + _, err = client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.Error(t, err) +} + +func TestAPI_FailoverDisabled(t *testing.T) { + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + // Forced on, but explicitly disabled via WithFailover. + ctx := WithFailover(failoverCtx(t, map[string]string{hdrFailRegions: "0"}), false) _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) require.Error(t, err) } diff --git a/failover_internal_test.go b/failover_internal_test.go index bce0190f..9ebc4136 100644 --- a/failover_internal_test.go +++ b/failover_internal_test.go @@ -16,26 +16,27 @@ package lksdk import "testing" -func TestFailoverEnabledFor(t *testing.T) { +func TestFailoverAttempts(t *testing.T) { cases := []struct { - mode FailoverMode + cfg failoverConfig host string - want bool + want int }{ - // Auto: only *.livekit.cloud project domains. - {FailoverAuto, "myproject.livekit.cloud", true}, - {FailoverAuto, "myproject.region.livekit.cloud", true}, - {FailoverAuto, "myproject.livekit.io", false}, - {FailoverAuto, "example.com", false}, - {FailoverAuto, "127.0.0.1", false}, - {FailoverAuto, "notlivekit.cloud", false}, - // On/Off override the host check. - {FailoverOn, "127.0.0.1", true}, - {FailoverOff, "myproject.livekit.cloud", false}, + // Enabled (the default): only *.livekit.cloud project domains fail over. + {failoverConfig{enabled: true}, "myproject.livekit.cloud", failoverMaxAttempts}, + {failoverConfig{enabled: true}, "myproject.region.livekit.cloud", failoverMaxAttempts}, + {failoverConfig{enabled: true}, "myproject.livekit.io", 1}, + {failoverConfig{enabled: true}, "example.com", 1}, + {failoverConfig{enabled: true}, "127.0.0.1", 1}, + {failoverConfig{enabled: true}, "notlivekit.cloud", 1}, + // force bypasses the cloud-host check; disabled never fails over. + {failoverConfig{enabled: true, force: true}, "127.0.0.1", failoverMaxAttempts}, + {failoverConfig{enabled: false, force: true}, "myproject.livekit.cloud", 1}, + {failoverConfig{enabled: false}, "myproject.livekit.cloud", 1}, } for _, c := range cases { - if got := (FailoverOptions{Mode: c.mode}).enabledFor(c.host); got != c.want { - t.Errorf("enabledFor(mode=%v, host=%q) = %v, want %v", c.mode, c.host, got, c.want) + if got := c.cfg.attempts(c.host); got != c.want { + t.Errorf("attempts(cfg=%+v, host=%q) = %v, want %v", c.cfg, c.host, got, c.want) } } } diff --git a/regionurlprovider.go b/regionurlprovider.go index 33f67207..08d0ba7f 100644 --- a/regionurlprovider.go +++ b/regionurlprovider.go @@ -18,6 +18,9 @@ import ( const ( regionHostnameProviderSettingsCacheTime = 3 * time.Second + // regionDiscoveryTimeout bounds a single /settings/regions fetch so a slow + // or unreachable endpoint doesn't stall the failover path. + regionDiscoveryTimeout = 2 * time.Second ) // regionSettingsURL builds the LiveKit Cloud region-discovery URL for a hostname. @@ -25,133 +28,159 @@ var regionSettingsURL = func(cloudHostname string) string { return "https://" + cloudHostname + "/settings/regions" } +// regionURLProvider supplies LiveKit Cloud region lists to the RTC signaling +// reconnect path. It is a thin token/https adapter over the shared regionCache, +// adding a default cache TTL and support for server-pushed region lists. type regionURLProvider struct { - hostnameSettingsCache map[string]*hostnameSettingsCacheItem // hostname -> regionSettings - - mutex sync.RWMutex - httpClient *http.Client -} - -type hostnameSettingsCacheItem struct { - regionSettings *livekit.RegionSettings - updatedAt time.Time - cacheTTL time.Duration // from Cache-Control max-age; falls back to the default + cache *regionCache } func newRegionURLProvider() *regionURLProvider { - return ®ionURLProvider{ - hostnameSettingsCache: make(map[string]*hostnameSettingsCacheItem), - httpClient: &http.Client{ - Timeout: 5 * time.Second, - }, - } + return ®ionURLProvider{cache: newRegionCache()} } func (r *regionURLProvider) RefreshRegionSettings(cloudHostname, token string) error { - r.mutex.RLock() - hostnameSettings := r.hostnameSettingsCache[cloudHostname] - r.mutex.RUnlock() - - if hostnameSettings != nil { - ttl := hostnameSettings.cacheTTL - if ttl <= 0 { - ttl = regionHostnameProviderSettingsCacheTime - } - if time.Since(hostnameSettings.updatedAt) < ttl { - return nil - } - } + _, err := r.RegionSettings(cloudHostname, token) + return err +} - settingsURL := regionSettingsURL(cloudHostname) - req, err := http.NewRequest("GET", settingsURL, nil) +// RegionSettings returns the cached region list for a hostname, refreshing it if +// stale. The returned list is owned by the cache and must not be mutated; the +// caller is responsible for tracking its own per-failover attempt state. +func (r *regionURLProvider) RegionSettings(cloudHostname, token string) (*livekit.RegionSettings, error) { + headers := http.Header{"Authorization": []string{"Bearer " + token}} + settings, err := r.cache.get(cloudHostname, regionSettingsURL(cloudHostname), headers, regionHostnameProviderSettingsCacheTime) if err != nil { - return errors.New("refreshRegionSettings failed to create request: " + err.Error()) + return nil, err } - req.Header = http.Header{ - "Authorization": []string{"Bearer " + token}, + if settings == nil { + return nil, errors.New("no regions available") } + if len(settings.Regions) == 0 { + logger.Warnw("no regions returned", nil, "cloudHostname", cloudHostname) + } + return settings, nil +} - resp, err := r.httpClient.Do(req) - if err != nil { - return err +// SetServerReportedRegions stores a region list pushed by the server (e.g. on a +// reconnect LeaveRequest), overriding the cached /settings/regions list and +// resetting the cache TTL so the next failover uses it without re-fetching. +func (r *regionURLProvider) SetServerReportedRegions(cloudHostname string, regions *livekit.RegionSettings) { + r.cache.set(cloudHostname, regions, regionHostnameProviderSettingsCacheTime) +} + +// regionCache fetches and caches the LiveKit Cloud region list per host. It is +// shared by the API failover path (which honors the request scheme and forwards +// the caller's headers) and the RTC signaling path (which fetches over https +// with a bearer token). The caller supplies the discovery URL so each path +// keeps its own URL scheme and test seams. +type regionCache struct { + mu sync.Mutex + client *http.Client + cache map[string]*regionCacheEntry +} + +type regionCacheEntry struct { + settings *livekit.RegionSettings + fetchedAt time.Time + ttl time.Duration +} + +func newRegionCache() *regionCache { + return ®ionCache{ + client: &http.Client{Timeout: regionDiscoveryTimeout}, + cache: make(map[string]*regionCacheEntry), } - defer resp.Body.Close() +} - if resp.StatusCode != http.StatusOK { - return errors.New("refreshRegionSettings failed to fetch region settings. http status: " + resp.Status) +// get returns the cached region list for key, fetching discoveryURL if the +// cache is stale. The server's Cache-Control max-age sets the TTL; when absent, +// defaultTTL is used (0 means "do not cache"). Best-effort: on a fetch failure +// it serves a stale cached list when available, otherwise returns the error. +func (c *regionCache) get(key, discoveryURL string, headers http.Header, defaultTTL time.Duration) (*livekit.RegionSettings, error) { + key = strings.ToLower(key) + + c.mu.Lock() + if entry := c.cache[key]; entry != nil && time.Since(entry.fetchedAt) < entry.ttl { + defer c.mu.Unlock() + return entry.settings, nil } + c.mu.Unlock() - respBody, err := io.ReadAll(resp.Body) + settings, ttl, err := c.fetch(discoveryURL, headers) if err != nil { - return errors.New("refreshRegionSettings failed to read response body: " + err.Error()) - } - regions := &livekit.RegionSettings{} - if err := protojson.Unmarshal(respBody, regions); err != nil { - return errors.New("refreshRegionSettings failed to decode region settings: " + err.Error()) + c.mu.Lock() + defer c.mu.Unlock() + if entry := c.cache[key]; entry != nil { + return entry.settings, nil // serve stale on failure + } + return nil, err } - ttl := regionHostnameProviderSettingsCacheTime - if maxAge := parseRegionSettingsMaxAge(resp.Header.Get("Cache-Control")); maxAge > 0 { - ttl = maxAge + if ttl <= 0 { + ttl = defaultTTL } - - r.mutex.Lock() - item := &hostnameSettingsCacheItem{ - regionSettings: regions, - updatedAt: time.Now(), - cacheTTL: ttl, + if ttl > 0 { + c.mu.Lock() + c.cache[key] = ®ionCacheEntry{settings: settings, fetchedAt: time.Now(), ttl: ttl} + c.mu.Unlock() } - r.hostnameSettingsCache[cloudHostname] = item - r.mutex.Unlock() + return settings, nil +} - if len(item.regionSettings.Regions) == 0 { - logger.Warnw("no regions returned", nil, "cloudHostname", cloudHostname) +// set stores a region list pushed out-of-band (e.g. by the server on reconnect), +// overriding any cached list and keeping the existing TTL, or defaultTTL. +func (c *regionCache) set(key string, settings *livekit.RegionSettings, defaultTTL time.Duration) { + if settings == nil { + return } - - return nil + key = strings.ToLower(key) + c.mu.Lock() + defer c.mu.Unlock() + ttl := defaultTTL + if existing := c.cache[key]; existing != nil && existing.ttl > 0 { + ttl = existing.ttl + } + c.cache[key] = ®ionCacheEntry{settings: settings, fetchedAt: time.Now(), ttl: ttl} } -// RegionSettings returns the cached region list for a hostname, refreshing it if -// stale. The returned list is owned by the cache and must not be mutated; the -// caller is responsible for tracking its own per-failover attempt state. -func (r *regionURLProvider) RegionSettings(cloudHostname, token string) (*livekit.RegionSettings, error) { - if err := r.RefreshRegionSettings(cloudHostname, token); err != nil { - r.mutex.RLock() - cached := r.hostnameSettingsCache[cloudHostname] - r.mutex.RUnlock() - if cached == nil { - return nil, err +func (c *regionCache) fetch(discoveryURL string, headers http.Header) (*livekit.RegionSettings, time.Duration, error) { + req, err := http.NewRequest(http.MethodGet, discoveryURL, nil) + if err != nil { + return nil, 0, err + } + // Forward the caller's headers (Authorization and any custom headers), + // minus body-specific ones, so a validly-signed token reaches the discovery + // endpoint and test directives propagate. + for k, vv := range headers { + switch http.CanonicalHeaderKey(k) { + case "Content-Type", "Content-Length": + continue + } + for _, v := range vv { + req.Header.Add(k, v) } - // a cached list exists; fall through and use it } - r.mutex.RLock() - defer r.mutex.RUnlock() - item := r.hostnameSettingsCache[cloudHostname] - if item == nil { - return nil, errors.New("no regions available") + resp, err := c.client.Do(req) + if err != nil { + return nil, 0, err } - return item.regionSettings, nil -} + defer drainResponse(resp) -// SetServerReportedRegions stores a region list pushed by the server (e.g. on a -// reconnect LeaveRequest), overriding the cached /settings/regions list and -// resetting the cache TTL so the next failover uses it without re-fetching. -func (r *regionURLProvider) SetServerReportedRegions(cloudHostname string, regions *livekit.RegionSettings) { - if regions == nil { - return + if resp.StatusCode != http.StatusOK { + return nil, 0, &TwirpRegionError{StatusCode: resp.StatusCode} } - r.mutex.Lock() - defer r.mutex.Unlock() - ttl := regionHostnameProviderSettingsCacheTime - if existing := r.hostnameSettingsCache[cloudHostname]; existing != nil && existing.cacheTTL > 0 { - ttl = existing.cacheTTL + b, err := io.ReadAll(resp.Body) + if err != nil { + return nil, 0, err } - r.hostnameSettingsCache[cloudHostname] = &hostnameSettingsCacheItem{ - regionSettings: regions, - updatedAt: time.Now(), - cacheTTL: ttl, + settings := &livekit.RegionSettings{} + if err := protojson.Unmarshal(b, settings); err != nil { + return nil, 0, err } + ttl := parseRegionSettingsMaxAge(resp.Header.Get("Cache-Control")) + return settings, ttl, nil } // parseRegionSettingsMaxAge extracts the max-age (seconds) from a Cache-Control diff --git a/regionurlprovider_test.go b/regionurlprovider_test.go index 97a0290b..64e18e73 100644 --- a/regionurlprovider_test.go +++ b/regionurlprovider_test.go @@ -80,7 +80,7 @@ func (rs *regionSettingsServer) setRegions(regions *livekit.RegionSettings) { func newTestProvider(rs *regionSettingsServer) *regionURLProvider { p := newRegionURLProvider() // trust the httptest TLS certificate - p.httpClient = rs.server.Client() + p.cache.client = rs.server.Client() return p } @@ -139,9 +139,9 @@ func TestRegionURLProvider_RefreshRepopulates(t *testing.T) { require.Equal(t, "wss://a.example.com", settings.GetRegions()[0].Url) // expire the cache and serve a different region list - p.mutex.Lock() - p.hostnameSettingsCache[rs.hostname].updatedAt = time.Now().Add(-2 * regionHostnameProviderSettingsCacheTime) - p.mutex.Unlock() + p.cache.mu.Lock() + p.cache.cache[strings.ToLower(rs.hostname)].fetchedAt = time.Now().Add(-2 * regionHostnameProviderSettingsCacheTime) + p.cache.mu.Unlock() rs.setRegions(&livekit.RegionSettings{ Regions: []*livekit.RegionInfo{{Region: "b", Url: "wss://b.example.com"}}, }) From de3bafbfe82d8c5fe69b645a6d4534f0ac8cee68 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Tue, 30 Jun 2026 16:21:14 +0200 Subject: [PATCH 5/6] clean up failover logic, handle per-attempt timeout --- agent_client.go | 36 +++--- agent_dispatch_client.go | 6 +- agent_simulation_client.go | 12 +- auth.go | 11 +- connectorclient.go | 10 +- egressclient.go | 18 +-- failover.go | 228 ++++++++++++++++++++++++++++--------- failover_apitest_test.go | 33 ++++++ failover_internal_test.go | 227 +++++++++++++++++++++++++++++++++++- ingressclient.go | 8 +- phonenumberclient.go | 12 +- roomclient.go | 26 ++--- sipclient.go | 34 +++--- 13 files changed, 527 insertions(+), 134 deletions(-) diff --git a/agent_client.go b/agent_client.go index 076d663a..66c9326e 100644 --- a/agent_client.go +++ b/agent_client.go @@ -53,7 +53,7 @@ func WithTwirpClientOptions(opts ...twirp.ClientOption) AgentClientOption { } func (c *AgentClient) CreateAgent(ctx context.Context, req *livekit.CreateAgentRequest) (*livekit.CreateAgentResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -61,7 +61,7 @@ func (c *AgentClient) CreateAgent(ctx context.Context, req *livekit.CreateAgentR } func (c *AgentClient) CreateAgentV2(ctx context.Context, req *livekit.CreateAgentV2Request) (*livekit.CreateAgentV2Response, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -69,7 +69,7 @@ func (c *AgentClient) CreateAgentV2(ctx context.Context, req *livekit.CreateAgen } func (c *AgentClient) ListAgents(ctx context.Context, req *livekit.ListAgentsRequest) (*livekit.ListAgentsResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -77,7 +77,7 @@ func (c *AgentClient) ListAgents(ctx context.Context, req *livekit.ListAgentsReq } func (c *AgentClient) ListAgentVersions(ctx context.Context, req *livekit.ListAgentVersionsRequest) (*livekit.ListAgentVersionsResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -85,7 +85,7 @@ func (c *AgentClient) ListAgentVersions(ctx context.Context, req *livekit.ListAg } func (c *AgentClient) DeleteAgent(ctx context.Context, req *livekit.DeleteAgentRequest) (*livekit.DeleteAgentResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -93,7 +93,7 @@ func (c *AgentClient) DeleteAgent(ctx context.Context, req *livekit.DeleteAgentR } func (c *AgentClient) UpdateAgent(ctx context.Context, req *livekit.UpdateAgentRequest) (*livekit.UpdateAgentResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -101,7 +101,7 @@ func (c *AgentClient) UpdateAgent(ctx context.Context, req *livekit.UpdateAgentR } func (c *AgentClient) RestartAgent(ctx context.Context, req *livekit.RestartAgentRequest) (*livekit.RestartAgentResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -109,7 +109,7 @@ func (c *AgentClient) RestartAgent(ctx context.Context, req *livekit.RestartAgen } func (c *AgentClient) RollbackAgent(ctx context.Context, req *livekit.RollbackAgentRequest) (*livekit.RollbackAgentResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -117,7 +117,7 @@ func (c *AgentClient) RollbackAgent(ctx context.Context, req *livekit.RollbackAg } func (c *AgentClient) ListAgentSecrets(ctx context.Context, req *livekit.ListAgentSecretsRequest) (*livekit.ListAgentSecretsResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -125,7 +125,7 @@ func (c *AgentClient) ListAgentSecrets(ctx context.Context, req *livekit.ListAge } func (c *AgentClient) UpdateAgentSecrets(ctx context.Context, req *livekit.UpdateAgentSecretsRequest) (*livekit.UpdateAgentSecretsResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -133,7 +133,7 @@ func (c *AgentClient) UpdateAgentSecrets(ctx context.Context, req *livekit.Updat } func (c *AgentClient) DeployAgent(ctx context.Context, req *livekit.DeployAgentRequest) (*livekit.DeployAgentResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -141,7 +141,7 @@ func (c *AgentClient) DeployAgent(ctx context.Context, req *livekit.DeployAgentR } func (c *AgentClient) DeployAgentV2(ctx context.Context, req *livekit.DeployAgentV2Request) (*livekit.DeployAgentV2Response, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -149,7 +149,7 @@ func (c *AgentClient) DeployAgentV2(ctx context.Context, req *livekit.DeployAgen } func (c *AgentClient) PromoteAgent(ctx context.Context, req *livekit.PromoteAgentRequest) (*livekit.PromoteAgentResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -157,7 +157,7 @@ func (c *AgentClient) PromoteAgent(ctx context.Context, req *livekit.PromoteAgen } func (c *AgentClient) GetClientSettings(ctx context.Context, req *livekit.ClientSettingsRequest) (*livekit.ClientSettingsResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -165,7 +165,7 @@ func (c *AgentClient) GetClientSettings(ctx context.Context, req *livekit.Client } func (c *AgentClient) CreatePrivateLink(ctx context.Context, req *livekit.CreatePrivateLinkRequest) (*livekit.CreatePrivateLinkResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -173,7 +173,7 @@ func (c *AgentClient) CreatePrivateLink(ctx context.Context, req *livekit.Create } func (c *AgentClient) DestroyPrivateLink(ctx context.Context, req *livekit.DestroyPrivateLinkRequest) (*livekit.DestroyPrivateLinkResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -181,7 +181,7 @@ func (c *AgentClient) DestroyPrivateLink(ctx context.Context, req *livekit.Destr } func (c *AgentClient) ListPrivateLinks(ctx context.Context, req *livekit.ListPrivateLinksRequest) (*livekit.ListPrivateLinksResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } @@ -189,7 +189,7 @@ func (c *AgentClient) ListPrivateLinks(ctx context.Context, req *livekit.ListPri } func (c *AgentClient) GetPrivateLinkStatus(ctx context.Context, req *livekit.GetPrivateLinkStatusRequest) (*livekit.GetPrivateLinkStatusResponse, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{Admin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{Admin: true}) if err != nil { return nil, err } diff --git a/agent_dispatch_client.go b/agent_dispatch_client.go index f84b44d6..ccd6ca4a 100644 --- a/agent_dispatch_client.go +++ b/agent_dispatch_client.go @@ -42,7 +42,7 @@ func NewAgentDispatchServiceClient(url string, apiKey string, secretKey string, } func (c *AgentDispatchClient) CreateDispatch(ctx context.Context, req *livekit.CreateAgentDispatchRequest) (*livekit.AgentDispatch, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) if err != nil { return nil, err } @@ -51,7 +51,7 @@ func (c *AgentDispatchClient) CreateDispatch(ctx context.Context, req *livekit.C } func (c *AgentDispatchClient) DeleteDispatch(ctx context.Context, req *livekit.DeleteAgentDispatchRequest) (*livekit.AgentDispatch, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) if err != nil { return nil, err } @@ -60,7 +60,7 @@ func (c *AgentDispatchClient) DeleteDispatch(ctx context.Context, req *livekit.D } func (c *AgentDispatchClient) ListDispatch(ctx context.Context, req *livekit.ListAgentDispatchRequest) (*livekit.ListAgentDispatchResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) if err != nil { return nil, err } diff --git a/agent_simulation_client.go b/agent_simulation_client.go index b2851468..58ab4d70 100644 --- a/agent_simulation_client.go +++ b/agent_simulation_client.go @@ -36,7 +36,7 @@ func NewAgentSimulationClient(url string, apiKey string, apiSecret string, opts } func (c *AgentSimulationClient) CreateSimulationRun(ctx context.Context, req *livekit.SimulationRun_Create_Request) (*livekit.SimulationRun_Create_Response, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{SimulationAdmin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{SimulationAdmin: true}) if err != nil { return nil, err } @@ -44,7 +44,7 @@ func (c *AgentSimulationClient) CreateSimulationRun(ctx context.Context, req *li } func (c *AgentSimulationClient) ConfirmSimulationSourceUpload(ctx context.Context, req *livekit.SimulationRun_ConfirmSourceUpload_Request) (*livekit.SimulationRun_ConfirmSourceUpload_Response, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{SimulationAdmin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{SimulationAdmin: true}) if err != nil { return nil, err } @@ -52,7 +52,7 @@ func (c *AgentSimulationClient) ConfirmSimulationSourceUpload(ctx context.Contex } func (c *AgentSimulationClient) GetSimulationRun(ctx context.Context, req *livekit.SimulationRun_Get_Request) (*livekit.SimulationRun_Get_Response, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{SimulationAdmin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{SimulationAdmin: true}) if err != nil { return nil, err } @@ -60,7 +60,7 @@ func (c *AgentSimulationClient) GetSimulationRun(ctx context.Context, req *livek } func (c *AgentSimulationClient) ListSimulationRuns(ctx context.Context, req *livekit.SimulationRun_List_Request) (*livekit.SimulationRun_List_Response, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{SimulationAdmin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{SimulationAdmin: true}) if err != nil { return nil, err } @@ -68,7 +68,7 @@ func (c *AgentSimulationClient) ListSimulationRuns(ctx context.Context, req *liv } func (c *AgentSimulationClient) CancelSimulationRun(ctx context.Context, req *livekit.SimulationRun_Cancel_Request) (*livekit.SimulationRun_Cancel_Response, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{SimulationAdmin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{SimulationAdmin: true}) if err != nil { return nil, err } @@ -76,7 +76,7 @@ func (c *AgentSimulationClient) CancelSimulationRun(ctx context.Context, req *li } func (c *AgentSimulationClient) CreateScenarioFromSession(ctx context.Context, req *livekit.Scenario_CreateFromSession_Request) (*livekit.Scenario_CreateFromSession_Response, error) { - ctx, err := c.withAuth(ctx, withAgentGrant{SimulationAdmin: true}) + ctx, err := c.prepareContext(ctx, withAgentGrant{SimulationAdmin: true}) if err != nil { return nil, err } diff --git a/auth.go b/auth.go index f9ad66e2..cf6dd727 100644 --- a/auth.go +++ b/auth.go @@ -52,7 +52,11 @@ func (g withAgentGrant) Apply(t *auth.AccessToken) { t.SetAgentGrant((*auth.AgentGrant)(&g)) } -func (b authBase) withAuth(ctx context.Context, opt authOption, options ...authOption) (context.Context, error) { +// prepareContext builds the context for an outgoing API request: it signs an +// access token for the given grants and attaches it as a request header, then +// detaches a long-enough deadline so failover can reset it per attempt (see +// withFailoverTimeout). +func (b authBase) prepareContext(ctx context.Context, opt authOption, options ...authOption) (context.Context, error) { at := auth.NewAccessToken(b.apiKey, b.apiSecret) opt.Apply(at) for _, opt := range options { @@ -78,5 +82,10 @@ func (b authBase) withAuth(ctx context.Context, opt authOption, options ...authO } } + // Detach a long-enough deadline so it isn't enforced across failover retries + // (twirp re-checks the context after each request); the transport re-applies + // the budget per attempt. A no-op for short or deadline-free requests. + ctx = withFailoverTimeout(ctx) + return twirp.WithHTTPRequestHeaders(ctx, ctxH) } diff --git a/connectorclient.go b/connectorclient.go index c025627c..35452718 100644 --- a/connectorclient.go +++ b/connectorclient.go @@ -42,7 +42,7 @@ func NewConnectorClient(url string, apiKey string, secretKey string, opts ...twi } func (c *ConnectorClient) ConnectTwilioCall(ctx context.Context, req *livekit.ConnectTwilioCallRequest) (*livekit.ConnectTwilioCallResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomCreate: true, Room: req.RoomName}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomCreate: true, Room: req.RoomName}) if err != nil { return nil, err } @@ -50,7 +50,7 @@ func (c *ConnectorClient) ConnectTwilioCall(ctx context.Context, req *livekit.Co } func (c *ConnectorClient) DialWhatsAppCall(ctx context.Context, req *livekit.DialWhatsAppCallRequest) (*livekit.DialWhatsAppCallResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomCreate: true, Room: req.RoomName}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomCreate: true, Room: req.RoomName}) if err != nil { return nil, err } @@ -58,7 +58,7 @@ func (c *ConnectorClient) DialWhatsAppCall(ctx context.Context, req *livekit.Dia } func (c *ConnectorClient) AcceptWhatsAppCall(ctx context.Context, req *livekit.AcceptWhatsAppCallRequest) (*livekit.AcceptWhatsAppCallResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomCreate: true, Room: req.RoomName}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomCreate: true, Room: req.RoomName}) if err != nil { return nil, err } @@ -66,7 +66,7 @@ func (c *ConnectorClient) AcceptWhatsAppCall(ctx context.Context, req *livekit.A } func (c *ConnectorClient) ConnectWhatsAppCall(ctx context.Context, req *livekit.ConnectWhatsAppCallRequest) (*livekit.ConnectWhatsAppCallResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomCreate: true}) if err != nil { return nil, err } @@ -74,7 +74,7 @@ func (c *ConnectorClient) ConnectWhatsAppCall(ctx context.Context, req *livekit. } func (c *ConnectorClient) DisconnectWhatsAppCall(ctx context.Context, req *livekit.DisconnectWhatsAppCallRequest) (*livekit.DisconnectWhatsAppCallResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomCreate: true}) if err != nil { return nil, err } diff --git a/egressclient.go b/egressclient.go index 42173e61..c0520a3b 100644 --- a/egressclient.go +++ b/egressclient.go @@ -43,7 +43,7 @@ func NewEgressClient(url string, apiKey string, secretKey string, opts ...twirp. } func (c *EgressClient) StartRoomCompositeEgress(ctx context.Context, req *livekit.RoomCompositeEgressRequest) (*livekit.EgressInfo, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomRecord: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomRecord: true}) if err != nil { return nil, err } @@ -51,7 +51,7 @@ func (c *EgressClient) StartRoomCompositeEgress(ctx context.Context, req *liveki } func (c *EgressClient) StartParticipantEgress(ctx context.Context, req *livekit.ParticipantEgressRequest) (*livekit.EgressInfo, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomRecord: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomRecord: true}) if err != nil { return nil, err } @@ -59,7 +59,7 @@ func (c *EgressClient) StartParticipantEgress(ctx context.Context, req *livekit. } func (c *EgressClient) StartTrackCompositeEgress(ctx context.Context, req *livekit.TrackCompositeEgressRequest) (*livekit.EgressInfo, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomRecord: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomRecord: true}) if err != nil { return nil, err } @@ -67,7 +67,7 @@ func (c *EgressClient) StartTrackCompositeEgress(ctx context.Context, req *livek } func (c *EgressClient) StartTrackEgress(ctx context.Context, req *livekit.TrackEgressRequest) (*livekit.EgressInfo, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomRecord: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomRecord: true}) if err != nil { return nil, err } @@ -75,7 +75,7 @@ func (c *EgressClient) StartTrackEgress(ctx context.Context, req *livekit.TrackE } func (c *EgressClient) StartWebEgress(ctx context.Context, req *livekit.WebEgressRequest) (*livekit.EgressInfo, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomRecord: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomRecord: true}) if err != nil { return nil, err } @@ -83,7 +83,7 @@ func (c *EgressClient) StartWebEgress(ctx context.Context, req *livekit.WebEgres } func (c *EgressClient) UpdateLayout(ctx context.Context, req *livekit.UpdateLayoutRequest) (*livekit.EgressInfo, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomRecord: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomRecord: true}) if err != nil { return nil, err } @@ -91,7 +91,7 @@ func (c *EgressClient) UpdateLayout(ctx context.Context, req *livekit.UpdateLayo } func (c *EgressClient) UpdateStream(ctx context.Context, req *livekit.UpdateStreamRequest) (*livekit.EgressInfo, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomRecord: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomRecord: true}) if err != nil { return nil, err } @@ -99,7 +99,7 @@ func (c *EgressClient) UpdateStream(ctx context.Context, req *livekit.UpdateStre } func (c *EgressClient) ListEgress(ctx context.Context, req *livekit.ListEgressRequest) (*livekit.ListEgressResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomRecord: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomRecord: true}) if err != nil { return nil, err } @@ -107,7 +107,7 @@ func (c *EgressClient) ListEgress(ctx context.Context, req *livekit.ListEgressRe } func (c *EgressClient) StopEgress(ctx context.Context, req *livekit.StopEgressRequest) (*livekit.EgressInfo, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomRecord: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomRecord: true}) if err != nil { return nil, err } diff --git a/failover.go b/failover.go index bf08ee8f..d9ea8ec0 100644 --- a/failover.go +++ b/failover.go @@ -17,6 +17,7 @@ package lksdk import ( "bytes" "context" + "errors" "io" "net/http" "net/url" @@ -36,20 +37,27 @@ const ( failoverBackoffBase = 200 * time.Millisecond ) -// failoverConfig is the resolved per-request region-failover configuration. The -// public API exposes only the enabled toggle (default true); force and -// backoffBase are internal test-only knobs. +// minFailoverTimeout gates failover on the request timeout: a shorter budget +// gets a single attempt, since a retry is unlikely to help and risks +// thundering-herd retries across regions. Deadline-free requests are exempt. +var minFailoverTimeout = 5 * time.Second + +// perAttemptTimeoutKey carries the caller's original timeout budget once its +// deadline has been detached (see withFailoverTimeout), so the transport can +// re-apply the full budget to each attempt instead of sharing one shrinking +// deadline across retries. +type perAttemptTimeoutKey struct{} + +// failoverConfig is the resolved per-request region-failover configuration. type failoverConfig struct { - enabled bool - // force bypasses the cloud-host check. Internal testing only. - force bool - // backoffBase overrides the retry backoff base. Internal testing only. - backoffBase time.Duration + enabled bool + force bool // bypass the cloud-host check (test-only) + backoffBase time.Duration // retry backoff base (test-only) } // attempts returns the total request attempts for a host; 1 means no failover. -// Failover only engages when enabled and the host is a LiveKit Cloud domain. -// force bypasses the cloud-host check and is for internal testing only. +// Failover engages only when enabled and the host is a LiveKit Cloud domain +// (or force is set). func (c failoverConfig) attempts(hostname string) int { if c.enabled && (c.force || isCloud(hostname)) { return failoverMaxAttempts @@ -85,6 +93,66 @@ func failoverConfigFromContext(ctx context.Context) failoverConfig { return cfg } +// withFailoverTimeout prepares a context for region failover. When the request +// has a deadline long enough to fail over (>= minFailoverTimeout) and failover +// is enabled, it detaches the deadline — so it isn't enforced across retries by +// twirp or the transport — and stashes the original budget for the transport to +// re-apply per attempt. Shorter or deadline-free requests are returned +// unchanged. Called once where the context enters the API client (prepareContext). +func withFailoverTimeout(ctx context.Context) context.Context { + if !failoverConfigFromContext(ctx).enabled { + return ctx + } + deadline, ok := ctx.Deadline() + if !ok { + return ctx + } + if timeout := time.Until(deadline); timeout >= minFailoverTimeout { + return context.WithValue(detachDeadline(ctx), perAttemptTimeoutKey{}, timeout) + } + return ctx +} + +// detachDeadline returns a context with parent's values that propagates parent's +// explicit cancellation but not its deadline. The failover loop resets the +// timeout per attempt, so the original deadline must not fire mid-failover. +func detachDeadline(parent context.Context) context.Context { + if parent.Done() == nil { + return parent // nothing to detach or forward + } + detached, cancel := context.WithCancel(context.WithoutCancel(parent)) + go func() { + <-parent.Done() + // Forward explicit cancellation only; the original deadline is reset per + // attempt, so let it lapse. + if errors.Is(context.Cause(parent), context.DeadlineExceeded) { + return + } + cancel() + }() + return detached +} + +// perAttemptBudget is the per-attempt timeout for a request: the budget stashed +// by withFailoverTimeout, or the remaining time on a raw deadline (e.g. when the +// transport is exercised directly). Zero means no timeout. +func perAttemptBudget(ctx context.Context) time.Duration { + if d, ok := ctx.Value(perAttemptTimeoutKey{}).(time.Duration); ok { + return d + } + if deadline, ok := ctx.Deadline(); ok { + return time.Until(deadline) + } + return 0 +} + +// hasPerAttemptTimeout reports whether withFailoverTimeout already detached the +// deadline and stashed the budget (so the transport needn't detach again). +func hasPerAttemptTimeout(ctx context.Context) bool { + _, ok := ctx.Value(perAttemptTimeoutKey{}).(time.Duration) + return ok +} + // newAPIHTTPClient returns the *http.Client used by every API service client. // It wraps the default transport with region-failover retries. func newAPIHTTPClient() *http.Client { @@ -107,7 +175,31 @@ func (t *failoverTransport) RoundTrip(req *http.Request) (*http.Response, error) cfg := failoverConfigFromContext(req.Context()) maxAttempts := cfg.attempts(req.URL.Hostname()) - // Buffer the body once so it can be replayed against each region. + // 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. + 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 { + 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) +} + +// 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) { + // Buffer the body so it can be re-sent to each region. var body []byte if req.Body != nil { b, err := io.ReadAll(req.Body) @@ -118,51 +210,47 @@ func (t *failoverTransport) RoundTrip(req *http.Request) (*http.Response, error) body = b } - originalScheme, originalHost := req.URL.Scheme, req.URL.Host - attempted := map[string]struct{}{strings.ToLower(originalHost): {}} + // 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. + base := req.Context() + if !hasPerAttemptTimeout(base) { + if _, ok := base.Deadline(); ok { + base = detachDeadline(base) + } + } - var ( - settings *livekit.RegionSettings - fetchedRegions bool - resp *http.Response - err error - ) - scheme, host := originalScheme, originalHost + scheme, host := req.URL.Scheme, req.URL.Host + tried := map[string]struct{}{strings.ToLower(host): {}} + var regions *livekit.RegionSettings // discovered lazily on the first failure + var resp *http.Response + var err error for attempt := 0; attempt < maxAttempts; attempt++ { - if ctxErr := req.Context().Err(); ctxErr != nil { - return nil, ctxErr - } + attemptCtx, cancel := withOptionalTimeout(base, timeout) - r := req.Clone(req.Context()) - r.URL.Scheme = scheme - r.URL.Host = host - r.Host = host + r := req.Clone(attemptCtx) + r.URL.Scheme, r.URL.Host, r.Host = scheme, host, host if body != nil { - buf := body - r.Body = io.NopCloser(bytes.NewReader(buf)) - r.ContentLength = int64(len(buf)) - r.GetBody = func() (io.ReadCloser, error) { return io.NopCloser(bytes.NewReader(buf)), nil } + r.Body = io.NopCloser(bytes.NewReader(body)) + r.ContentLength = int64(len(body)) + r.GetBody = func() (io.ReadCloser, error) { return io.NopCloser(bytes.NewReader(body)), nil } } - resp, err = t.base.RoundTrip(r) - if !isRetryable(resp, err) { - return resp, err - } - if attempt == maxAttempts-1 { - break // out of attempts; surface the last result + + // 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 { + return terminate(resp, err, cancel) } - // discover regions lazily; honor the request scheme (so it works against - // an http mock) and forward the caller's headers to the discovery fetch. - if !fetchedRegions { - discoveryURL := url.URL{Scheme: originalScheme, Host: originalHost, Path: "/settings/regions"} - settings, _ = t.regions.get(originalHost, discoveryURL.String(), req.Header, 0) - fetchedRegions = true + if regions == nil { + u := url.URL{Scheme: req.URL.Scheme, Host: req.URL.Host, Path: "/settings/regions"} + regions, _ = t.regions.get(req.URL.Host, u.String(), req.Header, 0) } - nextScheme, nextHost, ok := nextRegion(settings, attempted) + nextScheme, nextHost, ok := nextRegion(regions, tried) if !ok { - break // no untried region left + return terminate(resp, err, cancel) // no untried region left } status := 0 @@ -174,20 +262,58 @@ func (t *failoverTransport) RoundTrip(req *http.Request) (*http.Response, error) "attempt", attempt+1, "maxAttempts", maxAttempts, "status", status) drainResponse(resp) - if !sleepCtx(req.Context(), cfg.backoffBase<= 500 } diff --git a/failover_apitest_test.go b/failover_apitest_test.go index 4752e5f3..24dfd6d6 100644 --- a/failover_apitest_test.go +++ b/failover_apitest_test.go @@ -147,3 +147,36 @@ func TestAPI_FailoverDisabled(t *testing.T) { _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) require.Error(t, err) } + +// An unresponsive region (per-attempt timeout) should fail over to a healthy +// region, with the deadline reset so the retry has its full budget. +func TestAPI_TimeoutFailsOver(t *testing.T) { + setMinFailoverTimeout(t, 50*time.Millisecond) // let a short test timeout fail over + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + // Region 0 stalls well past the deadline; regions 1+ are healthy. + ctx := failoverCtx(t, map[string]string{ + hdrFailRegions: "0", + hdrFailMode: "delay", + "X-Lk-Mock-Delay-Ms": "5000", + }) + ctx, cancel := context.WithTimeout(ctx, 300*time.Millisecond) + defer cancel() + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.NoError(t, err, "an unresponsive region should fail over to a healthy one") +} + +// A request whose timeout is below the thundering-herd threshold must not fail +// over: a stalled region surfaces the timeout instead of retrying elsewhere. +func TestAPI_ShortTimeoutNotRetried(t *testing.T) { + setMinFailoverTimeout(t, 5*time.Second) + client := NewRoomServiceClient(testServerURL(t), "devkey", "secret") + ctx := failoverCtx(t, map[string]string{ + hdrFailRegions: "0", + hdrFailMode: "delay", + "X-Lk-Mock-Delay-Ms": "5000", + }) + ctx, cancel := context.WithTimeout(ctx, 300*time.Millisecond) // < threshold + defer cancel() + _, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: "api-test"}) + require.Error(t, err, "a short-timeout request must not fail over to another region") +} diff --git a/failover_internal_test.go b/failover_internal_test.go index 9ebc4136..16c18d48 100644 --- a/failover_internal_test.go +++ b/failover_internal_test.go @@ -14,7 +14,18 @@ package lksdk -import "testing" +import ( + "context" + "errors" + "io" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/livekit/protocol/livekit" +) func TestFailoverAttempts(t *testing.T) { cases := []struct { @@ -40,3 +51,217 @@ func TestFailoverAttempts(t *testing.T) { } } } + +type stubAttempt struct { + at time.Time + deadline time.Time + hasDL bool +} + +// stubRoundTripper records the context deadline of each attempt and delegates +// the response/error to behave. +type stubRoundTripper struct { + mu sync.Mutex + attempts []stubAttempt + behave func(attempt int, ctx context.Context) (*http.Response, error) +} + +func (s *stubRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { + s.mu.Lock() + i := len(s.attempts) + dl, ok := r.Context().Deadline() + s.attempts = append(s.attempts, stubAttempt{at: time.Now(), deadline: dl, hasDL: ok}) + s.mu.Unlock() + return s.behave(i, r.Context()) +} + +func (s *stubRoundTripper) count() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.attempts) +} + +func stubResponse(status int) *http.Response { + return &http.Response{StatusCode: status, Body: io.NopCloser(strings.NewReader("{}"))} +} + +// newStubTransport wires a failoverTransport to a stub base and a region cache +// pre-populated with two fallback regions, so failover has somewhere to go. +func newStubTransport(stub http.RoundTripper, host string) *failoverTransport { + rc := newRegionCache() + rc.cache[strings.ToLower(host)] = ®ionCacheEntry{ + settings: &livekit.RegionSettings{Regions: []*livekit.RegionInfo{ + {Url: "http://" + host}, + {Url: "http://r1.example.com"}, + {Url: "http://r2.example.com"}, + }}, + fetchedAt: time.Now(), + ttl: time.Hour, + } + return &failoverTransport{base: stub, regions: rc} +} + +func stubRequest(ctx context.Context, host string) *http.Request { + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, + "http://"+host+"/twirp/livekit.RoomService/CreateRoom", strings.NewReader("{}")) + return req +} + +// setMinFailoverTimeout lowers the thundering-herd threshold so tests can use +// short timeouts and still exercise failover. +func setMinFailoverTimeout(t *testing.T, d time.Duration) { + t.Helper() + orig := minFailoverTimeout + minFailoverTimeout = d + t.Cleanup(func() { minFailoverTimeout = orig }) +} + +// Each retry must get the caller's full timeout budget, not the shrinking +// remainder of a single deadline. +func TestFailoverResetsTimeoutPerAttempt(t *testing.T) { + const host = "primary.example.com" + const budget = 200 * time.Millisecond + setMinFailoverTimeout(t, 10*time.Millisecond) // allow failover at this short budget + + stub := &stubRoundTripper{behave: func(attempt int, _ context.Context) (*http.Response, error) { + if attempt < 2 { + return stubResponse(http.StatusServiceUnavailable), nil // 5xx -> fail over + } + return stubResponse(http.StatusOK), nil + }} + tr := newStubTransport(stub, host) + + // force enables failover for the non-cloud host; 30ms backoff makes the + // shrinking-budget bug obvious if the timeout weren't reset. + ctx := withFailoverForce(context.Background(), 30*time.Millisecond) + ctx, cancel := context.WithTimeout(ctx, budget) + defer cancel() + + resp, err := tr.RoundTrip(stubRequest(ctx, host)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + _ = resp.Body.Close() + + if stub.count() != 3 { + t.Fatalf("expected 3 attempts, got %d", stub.count()) + } + for i, a := range stub.attempts { + if !a.hasDL { + t.Fatalf("attempt %d had no deadline", i) + } + if remaining := a.deadline.Sub(a.at); remaining < budget*3/4 { + t.Errorf("attempt %d started with only %v of the %v budget; timeout was not reset", i, remaining, budget) + } + } +} + +// An unresponsive region (per-attempt timeout) is retried against the next +// region, with a fresh budget. +func TestFailoverRetriesOnTimeout(t *testing.T) { + const host = "primary.example.com" + setMinFailoverTimeout(t, 10*time.Millisecond) // allow failover at this short budget + + stub := &stubRoundTripper{behave: func(attempt int, ctx context.Context) (*http.Response, error) { + if attempt == 0 { + // Region 0 is unresponsive: block until the attempt's deadline fires. + <-ctx.Done() + return nil, ctx.Err() + } + return stubResponse(http.StatusOK), nil + }} + tr := newStubTransport(stub, host) + + ctx := withFailoverForce(context.Background(), time.Millisecond) + ctx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + + resp, err := tr.RoundTrip(stubRequest(ctx, host)) + if err != nil { + t.Fatalf("a timed-out region should fail over, got error: %v", err) + } + _ = resp.Body.Close() + if n := stub.count(); n != 2 { + t.Fatalf("expected 2 attempts (timeout then success), got %d", n) + } + // The retry must get its own fresh budget, not the remainder after the first + // attempt already consumed the whole 50ms. + if a := stub.attempts[1]; a.hasDL && a.deadline.Sub(a.at) < 40*time.Millisecond { + t.Errorf("retry after timeout got only %v budget; not reset", a.deadline.Sub(a.at)) + } +} + +// A short timeout (< minFailoverTimeout) must not fail over, to avoid +// thundering-herd retries across regions. +func TestFailoverShortTimeoutDoesNotRetry(t *testing.T) { + const host = "primary.example.com" + setMinFailoverTimeout(t, 5*time.Second) + + stub := &stubRoundTripper{behave: func(_ int, _ context.Context) (*http.Response, error) { + return stubResponse(http.StatusServiceUnavailable), nil // 5xx would normally fail over + }} + tr := newStubTransport(stub, host) + + ctx := withFailoverForce(context.Background(), time.Millisecond) // failover otherwise forced on + ctx, cancel := context.WithTimeout(ctx, 500*time.Millisecond) // but budget < 5s + defer cancel() + + resp, err := tr.RoundTrip(stubRequest(ctx, host)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + _ = resp.Body.Close() + if n := stub.count(); n != 1 { + t.Fatalf("a short-timeout request must not fail over; made %d attempts", n) + } +} + +// Explicit caller cancellation is terminal — we stop rather than failing over. +func TestFailoverDoesNotRetryOnCancel(t *testing.T) { + const host = "primary.example.com" + + ctx, cancel := context.WithCancel(withFailoverForce(context.Background(), time.Millisecond)) + defer cancel() + + stub := &stubRoundTripper{behave: func(_ int, attemptCtx context.Context) (*http.Response, error) { + cancel() // caller gives up mid-flight + <-attemptCtx.Done() + return nil, attemptCtx.Err() + }} + tr := newStubTransport(stub, host) + + _, err := tr.RoundTrip(stubRequest(ctx, host)) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected Canceled, got %v", err) + } + if n := stub.count(); n != 1 { + t.Fatalf("a cancelled request must not be retried; made %d attempts", n) + } +} + +// A 5xx within the budget is still retried (sanity check that the timeout +// changes didn't disable failover). +func TestFailoverRetriesOn5xxWithinBudget(t *testing.T) { + const host = "primary.example.com" + + stub := &stubRoundTripper{behave: func(attempt int, _ context.Context) (*http.Response, error) { + if attempt == 0 { + return stubResponse(http.StatusBadGateway), nil + } + return stubResponse(http.StatusOK), nil + }} + tr := newStubTransport(stub, host) + + ctx := withFailoverForce(context.Background(), time.Millisecond) + resp, err := tr.RoundTrip(stubRequest(ctx, host)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected 200 after failover, got %d", resp.StatusCode) + } + if n := stub.count(); n != 2 { + t.Fatalf("expected 2 attempts (5xx then success), got %d", n) + } +} diff --git a/ingressclient.go b/ingressclient.go index 3e000846..be76fefc 100644 --- a/ingressclient.go +++ b/ingressclient.go @@ -47,7 +47,7 @@ func (c *IngressClient) CreateIngress(ctx context.Context, in *livekit.CreateIng return nil, ErrInvalidParameter } - ctx, err := c.withAuth(ctx, withVideoGrant{IngressAdmin: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{IngressAdmin: true}) if err != nil { return nil, err } @@ -59,7 +59,7 @@ func (c *IngressClient) UpdateIngress(ctx context.Context, in *livekit.UpdateIng return nil, ErrInvalidParameter } - ctx, err := c.withAuth(ctx, withVideoGrant{IngressAdmin: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{IngressAdmin: true}) if err != nil { return nil, err } @@ -71,7 +71,7 @@ func (c *IngressClient) ListIngress(ctx context.Context, in *livekit.ListIngress return nil, ErrInvalidParameter } - ctx, err := c.withAuth(ctx, withVideoGrant{IngressAdmin: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{IngressAdmin: true}) if err != nil { return nil, err } @@ -83,7 +83,7 @@ func (c *IngressClient) DeleteIngress(ctx context.Context, in *livekit.DeleteIng return nil, ErrInvalidParameter } - ctx, err := c.withAuth(ctx, withVideoGrant{IngressAdmin: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{IngressAdmin: true}) if err != nil { return nil, err } diff --git a/phonenumberclient.go b/phonenumberclient.go index 4f2b332b..00f248e8 100644 --- a/phonenumberclient.go +++ b/phonenumberclient.go @@ -48,7 +48,7 @@ func (p *PhoneNumberClient) SearchPhoneNumbers(ctx context.Context, in *livekit. return nil, ErrInvalidParameter } - ctx, err := p.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := p.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -66,7 +66,7 @@ func (p *PhoneNumberClient) PurchasePhoneNumber(ctx context.Context, in *livekit return nil, ErrInvalidParameter } - ctx, err := p.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := p.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -79,7 +79,7 @@ func (p *PhoneNumberClient) ListPhoneNumbers(ctx context.Context, in *livekit.Li return nil, ErrInvalidParameter } - ctx, err := p.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := p.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -92,7 +92,7 @@ func (p *PhoneNumberClient) GetPhoneNumber(ctx context.Context, in *livekit.GetP return nil, ErrInvalidParameter } - ctx, err := p.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := p.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -105,7 +105,7 @@ func (p *PhoneNumberClient) UpdatePhoneNumber(ctx context.Context, in *livekit.U return nil, ErrInvalidParameter } - ctx, err := p.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := p.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -118,7 +118,7 @@ func (p *PhoneNumberClient) ReleasePhoneNumbers(ctx context.Context, in *livekit return nil, ErrInvalidParameter } - ctx, err := p.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := p.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } diff --git a/roomclient.go b/roomclient.go index adfad779..bb7a7efe 100644 --- a/roomclient.go +++ b/roomclient.go @@ -45,7 +45,7 @@ func NewRoomServiceClient(url string, apiKey string, secretKey string, opts ...t } func (c *RoomServiceClient) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomCreate: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomCreate: true}) if err != nil { return nil, err } @@ -54,7 +54,7 @@ func (c *RoomServiceClient) CreateRoom(ctx context.Context, req *livekit.CreateR } func (c *RoomServiceClient) ListRooms(ctx context.Context, req *livekit.ListRoomsRequest) (*livekit.ListRoomsResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomList: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomList: true}) if err != nil { return nil, err } @@ -63,7 +63,7 @@ func (c *RoomServiceClient) ListRooms(ctx context.Context, req *livekit.ListRoom } func (c *RoomServiceClient) DeleteRoom(ctx context.Context, req *livekit.DeleteRoomRequest) (*livekit.DeleteRoomResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomCreate: true}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomCreate: true}) if err != nil { return nil, err } @@ -72,7 +72,7 @@ func (c *RoomServiceClient) DeleteRoom(ctx context.Context, req *livekit.DeleteR } func (c *RoomServiceClient) ListParticipants(ctx context.Context, req *livekit.ListParticipantsRequest) (*livekit.ListParticipantsResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) if err != nil { return nil, err } @@ -81,7 +81,7 @@ func (c *RoomServiceClient) ListParticipants(ctx context.Context, req *livekit.L } func (c *RoomServiceClient) GetParticipant(ctx context.Context, req *livekit.RoomParticipantIdentity) (*livekit.ParticipantInfo, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) if err != nil { return nil, err } @@ -90,7 +90,7 @@ func (c *RoomServiceClient) GetParticipant(ctx context.Context, req *livekit.Roo } func (c *RoomServiceClient) RemoveParticipant(ctx context.Context, req *livekit.RoomParticipantIdentity) (*livekit.RemoveParticipantResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) if err != nil { return nil, err } @@ -102,7 +102,7 @@ func (c *RoomServiceClient) RemoveParticipant(ctx context.Context, req *livekit. // stop when the participant leaves the room or `RemoveParticipant` has been called in the destination room. // A participant can be forwarded to multiple rooms. The destination room will be created if it does not exist. func (c *RoomServiceClient) ForwardParticipant(ctx context.Context, req *livekit.ForwardParticipantRequest) (*livekit.ForwardParticipantResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room, DestinationRoom: req.DestinationRoom}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room, DestinationRoom: req.DestinationRoom}) if err != nil { return nil, err } @@ -113,7 +113,7 @@ func (c *RoomServiceClient) ForwardParticipant(ctx context.Context, req *livekit // The participant will be removed from the current room and added to the destination room. // From other observers' perspective, the participant would've disconnected from the previous room and joined the new one. func (c *RoomServiceClient) MoveParticipant(ctx context.Context, req *livekit.MoveParticipantRequest) (*livekit.MoveParticipantResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room, DestinationRoom: req.DestinationRoom}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room, DestinationRoom: req.DestinationRoom}) if err != nil { return nil, err } @@ -121,7 +121,7 @@ func (c *RoomServiceClient) MoveParticipant(ctx context.Context, req *livekit.Mo } func (c *RoomServiceClient) MutePublishedTrack(ctx context.Context, req *livekit.MuteRoomTrackRequest) (*livekit.MuteRoomTrackResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) if err != nil { return nil, err } @@ -130,7 +130,7 @@ func (c *RoomServiceClient) MutePublishedTrack(ctx context.Context, req *livekit } func (c *RoomServiceClient) UpdateParticipant(ctx context.Context, req *livekit.UpdateParticipantRequest) (*livekit.ParticipantInfo, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) if err != nil { return nil, err } @@ -138,7 +138,7 @@ func (c *RoomServiceClient) UpdateParticipant(ctx context.Context, req *livekit. } func (c *RoomServiceClient) UpdateSubscriptions(ctx context.Context, req *livekit.UpdateSubscriptionsRequest) (*livekit.UpdateSubscriptionsResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) if err != nil { return nil, err } @@ -146,7 +146,7 @@ func (c *RoomServiceClient) UpdateSubscriptions(ctx context.Context, req *liveki } func (c *RoomServiceClient) UpdateRoomMetadata(ctx context.Context, req *livekit.UpdateRoomMetadataRequest) (*livekit.Room, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) if err != nil { return nil, err } @@ -154,7 +154,7 @@ func (c *RoomServiceClient) UpdateRoomMetadata(ctx context.Context, req *livekit } func (c *RoomServiceClient) SendData(ctx context.Context, req *livekit.SendDataRequest) (*livekit.SendDataResponse, error) { - ctx, err := c.withAuth(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) + ctx, err := c.prepareContext(ctx, withVideoGrant{RoomAdmin: true, Room: req.Room}) if err != nil { return nil, err } diff --git a/sipclient.go b/sipclient.go index acded63f..f71d7a24 100644 --- a/sipclient.go +++ b/sipclient.go @@ -51,7 +51,7 @@ func (s *SIPClient) CreateSIPInboundTrunk(ctx context.Context, in *livekit.Creat return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -64,7 +64,7 @@ func (s *SIPClient) CreateSIPOutboundTrunk(ctx context.Context, in *livekit.Crea return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -77,7 +77,7 @@ func (s *SIPClient) UpdateSIPInboundTrunk(ctx context.Context, in *livekit.Updat return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -90,7 +90,7 @@ func (s *SIPClient) UpdateSIPOutboundTrunk(ctx context.Context, in *livekit.Upda return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -104,7 +104,7 @@ func (s *SIPClient) GetSIPInboundTrunksByIDs(ctx context.Context, ids []string) return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -126,7 +126,7 @@ func (s *SIPClient) GetSIPOutboundTrunksByIDs(ctx context.Context, ids []string) return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -149,7 +149,7 @@ func (s *SIPClient) ListSIPTrunk(ctx context.Context, in *livekit.ListSIPTrunkRe return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -162,7 +162,7 @@ func (s *SIPClient) ListSIPInboundTrunk(ctx context.Context, in *livekit.ListSIP return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -175,7 +175,7 @@ func (s *SIPClient) ListSIPOutboundTrunk(ctx context.Context, in *livekit.ListSI return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -188,7 +188,7 @@ func (s *SIPClient) DeleteSIPTrunk(ctx context.Context, in *livekit.DeleteSIPTru return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -201,7 +201,7 @@ func (s *SIPClient) CreateSIPDispatchRule(ctx context.Context, in *livekit.Creat return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -214,7 +214,7 @@ func (s *SIPClient) UpdateSIPDispatchRule(ctx context.Context, in *livekit.Updat return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -228,7 +228,7 @@ func (s *SIPClient) GetSIPDispatchRulesByIDs(ctx context.Context, ids []string) return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -249,7 +249,7 @@ func (s *SIPClient) ListSIPDispatchRule(ctx context.Context, in *livekit.ListSIP return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -262,7 +262,7 @@ func (s *SIPClient) DeleteSIPDispatchRule(ctx context.Context, in *livekit.Delet return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Admin: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Admin: true}) if err != nil { return nil, err } @@ -275,7 +275,7 @@ func (s *SIPClient) CreateSIPParticipant(ctx context.Context, in *livekit.Create return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Call: true}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Call: true}) if err != nil { return nil, err } @@ -297,7 +297,7 @@ func (s *SIPClient) TransferSIPParticipant(ctx context.Context, in *livekit.Tran return nil, ErrInvalidParameter } - ctx, err := s.withAuth(ctx, withSIPGrant{Call: true}, withVideoGrant{RoomAdmin: true, Room: in.RoomName}) + ctx, err := s.prepareContext(ctx, withSIPGrant{Call: true}, withVideoGrant{RoomAdmin: true, Room: in.RoomName}) if err != nil { return nil, err } From 3639875d44e5e70b618de44c7ee006358c78c7d6 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Tue, 30 Jun 2026 16:50:08 +0200 Subject: [PATCH 6/6] sane default timeouts for CreateSIPParticipant --- failover.go | 16 ++++++++++++---- sipclient.go | 34 ++++++++++++++++------------------ 2 files changed, 28 insertions(+), 22 deletions(-) diff --git a/failover.go b/failover.go index d9ea8ec0..a6b1fc6c 100644 --- a/failover.go +++ b/failover.go @@ -39,9 +39,17 @@ const ( // minFailoverTimeout gates failover on the request timeout: a shorter budget // gets a single attempt, since a retry is unlikely to help and risks -// thundering-herd retries across regions. Deadline-free requests are exempt. +// thundering-herd retries across regions. var minFailoverTimeout = 5 * time.Second +const ( + // defaultRequestTimeout bounds a request when the caller sets no deadline. + defaultRequestTimeout = 10 * time.Second + // sipDialTimeout is the longer default for calls that dial a phone + // (CreateSIPParticipant with WaitUntilAnswered, TransferSIPParticipant). + sipDialTimeout = 30 * time.Second +) + // perAttemptTimeoutKey carries the caller's original timeout budget once its // deadline has been detached (see withFailoverTimeout), so the transport can // re-apply the full budget to each attempt instead of sharing one shrinking @@ -134,8 +142,8 @@ func detachDeadline(parent context.Context) context.Context { } // perAttemptBudget is the per-attempt timeout for a request: the budget stashed -// by withFailoverTimeout, or the remaining time on a raw deadline (e.g. when the -// transport is exercised directly). Zero means no timeout. +// by withFailoverTimeout, the remaining time on a raw deadline (e.g. when the +// transport is exercised directly), or the default when the caller set neither. func perAttemptBudget(ctx context.Context) time.Duration { if d, ok := ctx.Value(perAttemptTimeoutKey{}).(time.Duration); ok { return d @@ -143,7 +151,7 @@ func perAttemptBudget(ctx context.Context) time.Duration { if deadline, ok := ctx.Deadline(); ok { return time.Until(deadline) } - return 0 + return defaultRequestTimeout } // hasPerAttemptTimeout reports whether withFailoverTimeout already detached the diff --git a/sipclient.go b/sipclient.go index f71d7a24..2b07f92b 100644 --- a/sipclient.go +++ b/sipclient.go @@ -16,7 +16,6 @@ package lksdk import ( "context" - "time" "github.com/twitchtv/twirp" "google.golang.org/protobuf/types/known/emptypb" @@ -275,19 +274,19 @@ func (s *SIPClient) CreateSIPParticipant(ctx context.Context, in *livekit.Create return nil, ErrInvalidParameter } - ctx, err := s.prepareContext(ctx, withSIPGrant{Call: true}) - if err != nil { - return nil, err - } - - // CreateSIPParticipant will wait for LiveKit Participant to be created and that can take some time. - // Default deadline is too short, thus, we must set a higher deadline for it (if not specified by the user). - if _, ok := ctx.Deadline(); !ok { + // Dialing a phone and waiting for an answer takes longer than a normal + // request, so apply a longer default deadline (before prepareContext, which + // detaches it for failover). Set it before auth so failover sees the budget. + if _, ok := ctx.Deadline(); !ok && in.WaitUntilAnswered { var cancel func() - ctx, cancel = context.WithTimeout(ctx, 30*time.Second) + ctx, cancel = context.WithTimeout(ctx, sipDialTimeout) defer cancel() } + ctx, err := s.prepareContext(ctx, withSIPGrant{Call: true}) + if err != nil { + return nil, err + } return s.sipClient.CreateSIPParticipant(ctx, in) } @@ -297,19 +296,18 @@ func (s *SIPClient) TransferSIPParticipant(ctx context.Context, in *livekit.Tran return nil, ErrInvalidParameter } - ctx, err := s.prepareContext(ctx, withSIPGrant{Call: true}, withVideoGrant{RoomAdmin: true, Room: in.RoomName}) - if err != nil { - return nil, err - } - - // TransferSIPParticipant will wait for call to be transferred and that can take some time. - // Default deadline is too short, thus, we must set a higher deadline for it (if not specified by the user). + // Transferring a call dials a phone and can take a while, so apply a longer + // default deadline (before prepareContext, which detaches it for failover). if _, ok := ctx.Deadline(); !ok { var cancel func() - ctx, cancel = context.WithTimeout(ctx, 30*time.Second) + ctx, cancel = context.WithTimeout(ctx, sipDialTimeout) defer cancel() } + ctx, err := s.prepareContext(ctx, withSIPGrant{Call: true}, withVideoGrant{RoomAdmin: true, Room: in.RoomName}) + if err != nil { + return nil, err + } return s.sipClient.TransferSIPParticipant(ctx, in) }