Module: LiveKit::Failover

Defined in:
lib/livekit/failover.rb

Overview

Region failover for the Twirp API clients.

On a retryable failure (any transport error or HTTP 5xx) the client discovers alternative LiveKit Cloud regions via /settings/regions and replays the request against the next region, with exponential backoff. 4xx responses are returned immediately.

The failover argument accepted by the service clients is a boolean (default true). When enabled, failover engages for LiveKit Cloud hosts only. The total attempt count and backoff are fixed (not user-configurable) so retries can't be tuned to values that could overwhelm the server.

Constant Summary collapse

MAX_ATTEMPTS = 
3
BACKOFF_BASE = 
0.2
DEFAULT_TIMEOUT =

Default per-request timeout (seconds). Calls that dial a phone override it via TIMEOUT_HEADER (see SIPServiceClient).

10
MIN_FAILOVER_TIMEOUT =

Below this per-request timeout a retry is unlikely to help and many clients would retry in lockstep across regions, so a short request gets a single attempt (thundering-herd guard).

5
TIMEOUT_HEADER =

Internal header carrying a per-request timeout override (seconds). Consumed by the middleware and not sent to the server.

'X-Lk-Request-Timeout'

Class Method Summary collapse

Class Method Details

.attempts(enabled, host, force, timeout) ⇒ Object

Total request attempts including the initial one; 1 means no failover. Failover only engages when enabled, the host is a LiveKit Cloud domain, and the request timeout is long enough to retry. force bypasses the cloud-host check and is for internal testing only.



40
41
42
43
44
45
 
# File 'lib/livekit/failover.rb', line 40

def self.attempts(enabled, host, force, timeout)
  return 1 unless enabled && (force || cloud?(host))
  return 1 if timeout && timeout < MIN_FAILOVER_TIMEOUT

  MAX_ATTEMPTS
end

.cloud?(host) ⇒ Boolean

Failover only engages for LiveKit Cloud project domains.

Returns:

  • (Boolean) 


60
61
62
 
# File 'lib/livekit/failover.rb', line 60

def self.cloud?(host)
  !host.nil? && host.end_with?('.livekit.cloud')
end

.connection(base_url, failover) ⇒ Object

Builds a Faraday connection wired with the region-failover middleware and the LiveKit Twirp base URL. Passed to Twirp::Client in place of a URL.



49
50
51
52
53
54
55
56
57
 
# File 'lib/livekit/failover.rb', line 49

def self.connection(base_url, failover)
  url = File.join(Utils.to_http_url(base_url), '/twirp')
  Faraday.new(url: url) do |f|
    f.headers['User-Agent'] = "livekit-server-sdk-ruby/#{VERSION}"
    f.options.timeout = DEFAULT_TIMEOUT
    f.use RegionFailoverMiddleware, failover: failover
    f.adapter Faraday.default_adapter
  end
end

.fetch(origin, headers) ⇒ Object 



108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/livekit/failover.rb', line 108

def self.fetch(origin, headers)
  forward = headers.reject { |k, _| %w[content-type content-length].include?(k.to_s.downcase) }
  url = "#{origin.scheme}://#{origin.host}:#{origin.port}/settings/regions"
  resp = Faraday.get(url) do |req|
    # Short timeout so a slow/unreachable discovery endpoint doesn't stall
    # the failover path.
    req.options.timeout = 2
    forward.each { |k, v| req.headers[k] = v }
  end
  raise "region discovery failed: #{resp.status}" unless resp.status == 200

  ttl = parse_max_age(resp.headers['cache-control'])
  body = JSON.parse(resp.body || '{}')
  uris = (body['regions'] || [])
         .map { |r| r['url'] }
         .reject { |u| u.nil? || u.empty? }
         .map { |u| URI.parse(to_http(u)) }
  [uris, ttl]
end

.host_key(uri) ⇒ Object 



69
70
71
 
# File 'lib/livekit/failover.rb', line 69

def self.host_key(uri)
  "#{uri.host}:#{uri.port}".downcase
end

.parse_max_age(cache_control) ⇒ Object 



128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/livekit/failover.rb', line 128

def self.parse_max_age(cache_control)
  return 0 if cache_control.nil? || cache_control.empty?

  cache_control.split(',').each do |directive|
    directive = directive.strip.downcase
    if directive.start_with?('max-age=')
      secs = directive.sub('max-age=', '').to_i
      return secs.positive? ? secs : 0
    end
  end
  0
end

.pick_next(regions, attempted) ⇒ Object

Returns the first region whose host has not yet been attempted.



74
75
76
 
# File 'lib/livekit/failover.rb', line 74

def self.pick_next(regions, attempted)
  regions.find { |uri| !attempted.include?(host_key(uri)) }
end

.region_uris(origin, headers) ⇒ Object

Returns alternative region origins (URIs) for the given origin, fetching /settings/regions if the cache is stale. Best-effort: on a fetch failure it serves a stale cached list when available, otherwise an empty list. Forwards headers so a valid token — and any test directives — reach the discovery endpoint.



86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/livekit/failover.rb', line 86

def self.region_uris(origin, headers)
  key = host_key(origin)
  @cache_mutex.synchronize do
    entry = @cache[key]
    return entry[:uris] if entry && (Time.now - entry[:fetched_at]) < entry[:ttl]
  end

  begin
    uris, ttl = fetch(origin, headers)
  rescue StandardError
    return @cache_mutex.synchronize { @cache[key]&.fetch(:uris, nil) } || []
  end

  # A zero TTL (e.g. Cache-Control: max-age=0) means "do not cache".
  if ttl.positive?
    @cache_mutex.synchronize do
      @cache[key] = { uris: uris, fetched_at: Time.now, ttl: ttl }
    end
  end
  uris
end

.to_http(url) ⇒ Object

Normalizes a region URL to an http(s) scheme (ws -> http, wss -> https).



65
66
67
 
# File 'lib/livekit/failover.rb', line 65

def self.to_http(url)
  url.start_with?('ws') ? "http#{url[2..]}" : url
end