ruby 52 lines · 9 steps

Streaming import progress with a Rails job

A background job builds records row by row and broadcasts throttled progress updates over Turbo Streams.

Explained by highlit
Rails Intermediate background-jobs turbo-streams progress-tracking throttling error-handling
1class DatasetImportJob < ApplicationJob
2 queue_as :default
3 
4 def perform(import)
5 rows = import.source_rows
6 total = rows.size
7 import.update!(status: :processing, total: total, processed: 0)
8 
9 rows.each_with_index do |row, index|
10 RecordBuilder.new(import, row).call
11 processed = index + 1
12 
13 if processed == total || (processed % broadcast_interval(total)).zero?
14 import.update_columns(processed: processed)
15 broadcast_progress(import, processed, total)
16 end
17 end
18 
19 import.update!(status: :completed)
20 broadcast_completion(import)
21 rescue => e
22 import.update!(status: :failed, error_message: e.message)
23 broadcast_completion(import)
24 raise
25 end
26 
27 private
28 
29 def broadcast_interval(total)
30 [(total / 100.0).ceil, 1].max
31 end
32 
33 def broadcast_progress(import, processed, total)
34 percent = ((processed.to_f / total) * 100).round
35 
36 Turbo::StreamsChannel.broadcast_replace_to(
37 import,
38 target: "import_#{import.id}_progress",
39 partial: "imports/progress",
40 locals: { import: import, percent: percent, processed: processed, total: total }
41 )
42 end
43 
44 def broadcast_completion(import)
45 Turbo::StreamsChannel.broadcast_replace_to(
46 import,
47 target: "import_#{import.id}",
48 partial: "imports/import",
49 locals: { import: import }
50 )
51 end
52end
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Broadcasting on an interval instead of every row keeps live progress updates cheap at scale.
  2. 2Persisting status and counts around the work makes an import's state recoverable and observable.
  3. 3A rescue that records the failure, notifies the client, then re-raises keeps the UI honest while preserving retry behavior.

Related explainers

rust 
use serde::Deserialize;
 
#[derive(Debug, Deserialize)]
#[serde(untagged)]

Parsing flexible JSON shapes with serde

deserialization enums json
Intermediate 6 steps
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
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
rust 
use chrono::{Duration, NaiveDate};
 
#[derive(Debug)]
pub struct DateRange {

Parsing and iterating date ranges in Rust

error-handling iterators parsing
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

Streaming import progress with a Rails job — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code