require "sidekiq/api"
class QueueDepthGuard
class QueueSaturated < StandardError
attr_reader :queue, :depth
def initialize(queue, depth)
@queue = queue
@depth = depth
super("queue #{queue} saturated at depth #{depth}")
end
end
LIMITS = {
"default" => 5_000,
"ingest" => 20_000,
"critical" => 500
}.freeze
def self.limit_for(queue)
LIMITS.fetch(queue.to_s, 5_000)
end
def self.depth(queue)
name = queue.to_s
ready = Sidekiq::Queue.new(name).size
scheduled = Sidekiq::ScheduledSet.new.count { |j| j.queue == name }
retries = Sidekiq::RetrySet.new.count { |j| j.queue == name }
ready + scheduled + retries
end
def self.saturated?(queue)
depth(queue) >= limit_for(queue)
end
def self.admit!(queue)
current = depth(queue)
raise QueueSaturated.new(queue.to_s, current) if current >= limit_for(queue)
current
end
def self.overflow_backoff(attempt)
base = 5
[base * (2**attempt), 300].min + rand(0..3)
end
end
class IngestJob
include Sidekiq::Job
sidekiq_options queue: "ingest", retry: 5
def perform(batch_id, record_ids, attempt = 0)
remaining = record_ids.dup
until remaining.empty?
record_id = remaining.first
begin
QueueDepthGuard.admit!("default")
rescue QueueDepthGuard::QueueSaturated => e
defer(batch_id, remaining, attempt, e)
return
end
ProcessRecordJob.perform_async(record_id)
remaining.shift
end
end
private
def defer(batch_id, remaining, attempt, error)
delay = QueueDepthGuard.overflow_backoff(attempt)
Sidekiq.logger.info(
"ingest deferred batch=#{batch_id} left=#{remaining.size} " \
"depth=#{error.depth} in=#{delay}s"
)
StatsD.increment("ingest.records_deferred", by: remaining.size)
self.class.perform_in(delay, batch_id, remaining, attempt + 1)
end
end
class EnqueueController < ApplicationController
def create
QueueDepthGuard.admit!("ingest")
batch = IngestBatch.create!(record_ids: params[:record_ids])
IngestJob.perform_async(batch.id, batch.record_ids)
render json: { batch_id: batch.id, status: "queued" }, status: :accepted
rescue QueueDepthGuard::QueueSaturated => e
retry_after = QueueDepthGuard.overflow_backoff(0)
response.set_header("Retry-After", retry_after.to_s)
render json: {
error: "queue_saturated",
queue: e.queue,
depth: e.depth,
retry_after: retry_after
}, status: :too_many_requests
end
end
When a background system can enqueue work far faster than it drains, unbounded queues grow until Redis memory is exhausted or latency for critical jobs becomes unacceptable. Backpressure is the pattern of refusing or deferring new work once the system is saturated, pushing the pressure back onto the producer instead of silently piling it up. This snippet shows a small, focused implementation of a queue-depth guard for Sidekiq that measures backlog before enqueueing and reacts when a threshold is exceeded.
The QueueDepthGuard tab wraps Sidekiq's introspection API. depth sums the size of a Sidekiq::Queue plus scheduled and retry jobs targeting that queue, giving a realistic view of pending work rather than just the ready count. saturated? compares against a per-queue limit, and admit! raises QueueSaturated when the caller must be told to back off. The guard is deliberately cheap: these are LLEN/ZCARD-style reads against Redis, so calling it on the enqueue path adds negligible latency.
The IngestJob tab is a normal worker that also acts as a producer — it fans out per-record child jobs. Before enqueuing each child it calls QueueDepthGuard.admit!. On QueueSaturated it does not drop the work; instead it reschedules the remaining batch with perform_in and an exponential-ish delay derived from overflow_backoff. This converts a hard failure into cooperative deferral, so the producer naturally slows to match consumer throughput. Recording records_deferred keeps the deferral observable.
The EnqueueController tab shows the synchronous entry point. Here saturation is surfaced to the client as HTTP 429 with a Retry-After header rather than being swallowed, because a web request cannot politely wait. This distinction matters: producers that can retry cheaply (jobs) should defer, while user-facing callers should be told to retry later.
The main trade-off is that depth is a sampled snapshot and slightly stale under high concurrency, so the threshold should sit below the true danger point. It also does not coordinate across many producers beyond what Redis observes, so thresholds act as soft limits, not exact caps. Still, a guard like this is often enough to keep queues bounded, protect latency-sensitive queues from noisy neighbors, and avoid the classic failure mode where a retry storm buries everything.
Related snips
class CommentsController < ApplicationController
before_action :set_post
def create
@comment = @post.comments.build(comment_params)System test: asserting Turbo Stream responses
class Post < ApplicationRecord
belongs_to :author, class_name: 'User'
has_many :comments, dependent: :destroy
scope :published, -> { where.not(published_at: nil).where('published_at <= ?', Time.current) }
scope :draft, -> { where(published_at: nil) }ActiveRecord scopes for reusable query logic
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
module Api
module V1
class UsersController < BaseController
def show
user = User.includes(:profile).find(params[:id])ETags for conditional requests and caching
class PostsController < ApplicationController
def index
@posts = Post.includes(:author)
.order(created_at: :desc)
.page(params[:page])
.per(10)Turbo Frames: infinite scroll with lazy-loading frame
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
Share this code
Here's the card — post it anywhere.
