Class: Attio::RateLimiter

Inherits:
Object
  • Object
show all
Defined in:
lib/attio/rate_limiter.rb

Overview

Rate limiter with intelligent retry and backoff strategies

Examples:

Using the rate limiter

limiter = Attio::RateLimiter.new(
  max_requests: 100,
  window_seconds: 60,
  max_retries: 3
)

limiter.execute { client.records.list }

Since:

  • 1.0.0

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(max_requests: 1000, window_seconds: 3600, max_retries: 3, enable_jitter: true) ⇒ RateLimiter

Initialize a new rate limiter

Since:

  • 1.0.0

Parameters:

  • (defaults to: 1000)

    Maximum requests per window

  • (defaults to: 3600)

    Time window in seconds

  • (defaults to: 3)

    Maximum retry attempts

  • (defaults to: true)

    Add jitter to backoff delays



24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/attio/rate_limiter.rb', line 24

def initialize(max_requests: 1000, window_seconds: 3600, max_retries: 3, enable_jitter: true)
  @max_requests = max_requests
  @window_seconds = window_seconds
  @max_retries = max_retries
  @enable_jitter = enable_jitter

  @current_limit = max_requests
  @remaining = max_requests
  @reset_at = Time.now + window_seconds

  @mutex = Mutex.new
  @request_queue = []
  @request_times = []
end

Instance Attribute Details

#current_limit ⇒ Object

Since:

  • 1.0.0



16
17
18
 
# File 'lib/attio/rate_limiter.rb', line 16

def current_limit
  @current_limit
end

#max_requests ⇒ Object (readonly)

Since:

  • 1.0.0



15
16
17
 
# File 'lib/attio/rate_limiter.rb', line 15

def max_requests
  @max_requests
end

#max_retries ⇒ Object (readonly)

Since:

  • 1.0.0



15
16
17
 
# File 'lib/attio/rate_limiter.rb', line 15

def max_retries
  @max_retries
end

#remaining ⇒ Object

Since:

  • 1.0.0



16
17
18
 
# File 'lib/attio/rate_limiter.rb', line 16

def remaining
  @remaining
end

#reset_at ⇒ Object

Since:

  • 1.0.0



16
17
18
 
# File 'lib/attio/rate_limiter.rb', line 16

def reset_at
  @reset_at
end

#window_seconds ⇒ Object (readonly)

Since:

  • 1.0.0



15
16
17
 
# File 'lib/attio/rate_limiter.rb', line 15

def window_seconds
  @window_seconds
end

Instance Method Details

#calculate_backoff(attempt, error = nil) ⇒ Object (private)

Since:

  • 1.0.0



181
182
183
184
185
186
187
188
189
190
191
192
193
194
 
# File 'lib/attio/rate_limiter.rb', line 181

private def calculate_backoff(attempt, error = nil)
  base_wait = 2**attempt

  # Use server-provided retry-after if available
  base_wait = error.retry_after if error && error.respond_to?(:retry_after) && error.retry_after

  # Add jitter to prevent thundering herd
  if @enable_jitter
    jitter = rand * base_wait * 0.1
    base_wait + jitter
  else
    base_wait
  end
end

#cleanup_old_requests ⇒ Object (private)

Since:

  • 1.0.0



176
177
178
179
 
# File 'lib/attio/rate_limiter.rb', line 176

private def cleanup_old_requests
  cutoff = Time.now - @window_seconds
  @request_times.reject! { |time| time < cutoff }
end

#execute { ... } ⇒ Object

Execute a block with rate limiting

Yields:

  • The block to execute

Raises:

Since:

  • 1.0.0

Returns:

  • The result of the block



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/attio/rate_limiter.rb', line 43

def execute
  raise ArgumentError, "Block required" unless block_given?

  @mutex.synchronize do
    wait_if_needed
    track_request
  end

  attempt = 0
  begin
    result = yield
    # Thread-safe header update
    @mutex.synchronize do
      update_from_headers(result) if result.is_a?(Hash) && result["_headers"]
    end
    result
  rescue Attio::RateLimitError => e
    attempt += 1
    raise e unless attempt <= @max_retries

    wait_time = calculate_backoff(attempt, e)
    sleep(wait_time)
    retry
  end
end

#process_queue(max_per_batch: 10) ⇒ Array

Process queued requests

Since:

  • 1.0.0

Parameters:

  • (defaults to: 10)

    Maximum requests to process

Returns:

  • Results from processed requests



133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/attio/rate_limiter.rb', line 133

def process_queue(max_per_batch: 10)
  results = []
  processed = 0

  while processed < max_per_batch
    request = @mutex.synchronize { @request_queue.shift }
    break unless request

    begin
      result = execute(&request[:block])
      results << { success: true, result: result }
    rescue StandardError => e
      results << { success: false, error: e }
    end

    processed += 1
  end

  results
end

#queue_request(priority: 5) { ... } ⇒ Object

Queue a request for later execution

Yields:

  • Block to execute

Since:

  • 1.0.0

Parameters:

  • (defaults to: 5)

    Priority (lower = higher priority)



122
123
124
125
126
127
 
# File 'lib/attio/rate_limiter.rb', line 122

def queue_request(priority: 5, &block)
  @mutex.synchronize do
    @request_queue << { priority: priority, block: block, queued_at: Time.now }
    @request_queue.sort_by! { |r| [r[:priority], r[:queued_at]] }
  end
end

#rate_limited? ⇒ Boolean

Check if rate limit is exceeded

Since:

  • 1.0.0

Returns:

  • True if rate limit would be exceeded



72
73
74
75
76
77
 
# File 'lib/attio/rate_limiter.rb', line 72

def rate_limited?
  @mutex.synchronize do
    cleanup_old_requests
    @request_times.size >= @max_requests
  end
end

#reset! ⇒ Object

Reset the rate limiter

Since:

  • 1.0.0



110
111
112
113
114
115
116
 
# File 'lib/attio/rate_limiter.rb', line 110

def reset!
  @mutex.synchronize do
    @request_times.clear
    @remaining = @max_requests
    @reset_at = Time.now + @window_seconds
  end
end

#status ⇒ Hash

Get current rate limit status

Since:

  • 1.0.0

Returns:

  • Current status



82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/attio/rate_limiter.rb', line 82

def status
  @mutex.synchronize do
    cleanup_old_requests
    {
      limit: @current_limit,
      remaining: [@remaining, @max_requests - @request_times.size].min,
      reset_at: @reset_at,
      reset_in: [@reset_at - Time.now, 0].max.to_i,
      current_usage: @request_times.size,
    }
  end
end

#track_request ⇒ Object (private)

Since:

  • 1.0.0



171
172
173
174
 
# File 'lib/attio/rate_limiter.rb', line 171

private def track_request
  @request_times << Time.now
  @remaining = [@remaining - 1, 0].max
end

#update_from_headers(response) ⇒ Object (private)

Update rate limit info from response headers NOTE: This method should be called within a mutex lock

Since:

  • 1.0.0

Parameters:

  • Response containing headers



99
100
101
102
103
104
105
106
107
 
# File 'lib/attio/rate_limiter.rb', line 99

private def update_from_headers(response)
  return unless response.is_a?(Hash)

  headers = response["_headers"] || {}

  @current_limit = headers["x-ratelimit-limit"].to_i if headers["x-ratelimit-limit"]
  @remaining = headers["x-ratelimit-remaining"].to_i if headers["x-ratelimit-remaining"]
  @reset_at = Time.at(headers["x-ratelimit-reset"].to_i) if headers["x-ratelimit-reset"]
end

#wait_if_needed ⇒ Object (private)

Since:

  • 1.0.0



154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/attio/rate_limiter.rb', line 154

private def wait_if_needed
  cleanup_old_requests

  if @request_times.size >= @max_requests
    wait_time = @request_times.first + @window_seconds - Time.now
    if wait_time > 0
      sleep(wait_time)
      cleanup_old_requests
    end
  end

  return unless @remaining <= 0 && @reset_at > Time.now

  wait_time = @reset_at - Time.now
  sleep(wait_time) if wait_time > 0
end