ruby 60 lines · 9 steps

Rate-limited CRM sync in a Rails job

A background job pushes contacts to a CRM while a Redis sliding window and a semaphore keep it under the API's limits.

Explained by highlit
Rails Advanced background-jobs rate-limiting redis concurrency sliding-window
1class SyncContactsJob < ApplicationJob
2 queue_as :external_api
3 
4 MAX_CONCURRENCY = 5
5 RATE_LIMIT_KEY = "crm:rate_limit".freeze
6 
7 retry_on Crm::RateLimitedError, wait: :polynomially_longer, attempts: 8
8 
9 def perform(account_id)
10 account = Account.find(account_id)
11 
12 account.contacts.pending_sync.find_each do |contact|
13 throttle { Crm::Client.upsert_contact(contact) }
14 contact.update!(synced_at: Time.current, sync_status: :synced)
15 end
16 end
17 
18 private
19 
20 def throttle(&block)
21 semaphore.synchronize do
22 wait_for_rate_limit_slot
23 block.call
24 end
25 end
26 
27 def semaphore
28 @semaphore ||= Concurrent::Semaphore.new(MAX_CONCURRENCY)
29 end
30 
31 def wait_for_rate_limit_slot
32 loop do
33 allowed = redis.eval(
34 THROTTLE_SCRIPT,
35 keys: [RATE_LIMIT_KEY],
36 argv: [60, 100, Time.current.to_f]
37 )
38 break if allowed == 1
39 
40 sleep(0.25 + rand * 0.25)
41 end
42 end
43 
44 def redis
45 @redis ||= Redis.new(url: ENV.fetch("REDIS_URL"))
46 end
47 
48 THROTTLE_SCRIPT = <<~LUA.freeze
49 local window = tonumber(ARGV[1])
50 local limit = tonumber(ARGV[2])
51 local now = tonumber(ARGV[3])
52 redis.call('ZREMRANGEBYSCORE', KEYS[1], 0, now - window)
53 if redis.call('ZCARD', KEYS[1]) < limit then
54 redis.call('ZADD', KEYS[1], now, now)
55 redis.call('EXPIRE', KEYS[1], window)
56 return 1
57 end
58 return 0
59 LUA
60end
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A Redis sorted set makes an atomic sliding-window rate limiter that survives across processes.
  2. 2Combining a local semaphore with a shared rate limit bounds both in-process and cross-process concurrency.
  3. 3Declarative retries with exponential backoff let a job absorb upstream rate-limit errors gracefully.

Related explainers

ruby 
require "shellwords"
require "open3"
 
module Backup

Building safe shell commands in Ruby

shell-out subprocess command-injection
Intermediate 7 steps
ruby 
class UserAgentParser
  BROWSERS = [
    [/Edg\/([\d.]+)/, "Edge"],
    [/OPR\/([\d.]+)/, "Opera"],

Parsing user-agent strings in Ruby

regex pattern-matching lookup-tables
Intermediate 8 steps
ruby 
class LogAggregator
  BUCKET_FORMAT = "%Y-%m-%dT%H:%M"
 
  def initialize(entries)

Bucketing log entries by the minute in Ruby

aggregation hashing enumerable
Intermediate 5 steps
go 
func (w *Watcher) resetDebounce(d time.Duration) {
	if !w.timer.Stop() {
		select {
		case <-w.timer.C:

Debouncing a stream of events in Go

debounce timers channels
Advanced 7 steps
ruby 
class WeeklySignupsReport
  DEFAULT_WEEKS = 12
 
  def initialize(weeks: DEFAULT_WEEKS, source: User.all)

Building a weekly signups report in Rails

service object aggregation group by
Intermediate 7 steps
ruby 
class ApplicationController < ActionController::Base
  EXPERIMENTS = {
    checkout_button_color: %w[control blue green],
    onboarding_flow: %w[control streamlined]

How A/B test cohorts are assigned in Rails

a-b-testing cookies hashing
Intermediate 8 steps

Share this explainer

Here's the card — post it anywhere.

Rate-limited CRM sync in a Rails job — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code