class ReindexCheckpoint < ApplicationRecord
enum status: { idle: 0, running: 1, done: 2, failed: 3 }
validates :index_name, presence: true, uniqueness: true
def self.for(index_name)
find_or_create_by!(index_name: index_name)
end
def reset!(total:)
update!(last_id: 0, processed: 0, total: total, status: :running)
end
def advance!(new_last_id:, count:)
with_lock do
# never let a retried/out-of-order batch rewind the cursor
return if new_last_id <= last_id
self.last_id = new_last_id
self.processed += count
self.status = :done if processed >= total
save!
end
end
def fail!(message)
update!(status: :failed)
Rails.logger.error("reindex #{index_name} failed: #{message}")
end
end
class CreateReindexCheckpoints < ActiveRecord::Migration[7.0]
def change
create_table :reindex_checkpoints do |t|
t.string :index_name, null: false
t.bigint :last_id, null: false, default: 0
t.integer :processed, null: false, default: 0
t.integer :total, null: false, default: 0
t.integer :status, null: false, default: 0
t.timestamps
end
add_index :reindex_checkpoints, :index_name, unique: true
end
end
class BackfillReindexJob
include Sidekiq::Job
sidekiq_options queue: :reindex, retry: 5
BATCH_SIZE = 500
INDEX_NAME = "products".freeze
def perform
checkpoint = ReindexCheckpoint.for(INDEX_NAME)
return if checkpoint.done?
slice = Product.where("id > ?", checkpoint.last_id)
.order(:id)
.limit(BATCH_SIZE)
.to_a
if slice.empty?
checkpoint.update!(status: :done)
return
end
bulk_index(slice)
checkpoint.advance!(new_last_id: slice.last.id, count: slice.size)
# chain the next slice; a short job survives deploys and retries cleanly
self.class.perform_async unless checkpoint.reload.done?
rescue => e
ReindexCheckpoint.for(INDEX_NAME).fail!(e.message)
raise
end
private
def bulk_index(records)
payload = records.flat_map do |record|
[{ index: { _index: INDEX_NAME, _id: record.id } }, record.as_indexed_json]
end
Product.__elasticsearch__.client.bulk(body: payload)
end
end
class ReindexesController < ApplicationController
before_action :require_admin
def create
checkpoint = ReindexCheckpoint.for("products")
checkpoint.reset!(total: Product.count)
BackfillReindexJob.perform_async
redirect_to reindex_path, notice: "Reindex started"
end
def show
@checkpoint = ReindexCheckpoint.for("products")
@percent = @checkpoint.total.zero? ? 0 : (@checkpoint.processed * 100 / @checkpoint.total)
respond_to do |format|
format.html
format.json { render json: { status: @checkpoint.status, percent: @percent } }
end
end
end
This snippet shows a background reindex that can crash, deploy, or get retried at any point without losing progress or re-scanning the whole table. The core idea is to break the work into deterministic ID ranges (slices), process each range, and persist a checkpoint after every completed batch so a restart resumes from the last confirmed position rather than the beginning.
In ReindexCheckpoint model, the checkpoint is a durable single-row-per-index record keyed by index_name. It stores last_id, the highest primary key already indexed, plus total, processed, and a status enum. advance! moves the cursor forward atomically and is written to guard against a lower value overwriting a higher one — retried or out-of-order batches cannot rewind progress. This makes the checkpoint the single source of truth for where the job should continue.
The migration in create_reindex_checkpoints migration backs that model. A unique index on index_name enforces one checkpoint per logical index, and last_id defaults to 0 so a fresh run naturally starts from the beginning. Keeping the cursor in postgres rather than in redis or job arguments means it survives queue flushes and Sidekiq restarts.
BackfillReindexJob is where slicing happens. Instead of loading every record, it reads checkpoint.last_id and pulls one ordered slice with where('id > ?', cursor).order(:id).limit(BATCH_SIZE). Each record is pushed through __elasticsearch__ bulk indexing, then checkpoint.advance! records the new high-water mark and perform_async re-enqueues the next slice. Re-enqueueing per batch — rather than looping in one long-lived job — keeps each execution short, bounds memory, plays nicely with Sidekiq's retry and shutdown handling, and lets a deploy interrupt the chain safely. The find_in_batches-style cursor pagination is deliberate: offset pagination degrades on large tables, while an id cursor stays index-friendly.
The controller ReindexesController exposes the operation. create resets the checkpoint via reset! and kicks off the first slice, while show reports processed/total for a progress bar. Because the job is idempotent around last_id, hitting retry or double-clicking start cannot corrupt state. The main trade-off is that mid-batch failures reprocess that batch's documents, which is safe only because Elasticsearch indexing by document id is itself idempotent.
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.
