Class: SourceReconciliationWorker

Inherits:
Object
  • Object
show all
Includes:
Sidekiq::Job
Defined in:
app/workers/source_reconciliation_worker.rb

Overview

Restores the operational source of records that nurturing campaigns
overwrote, back to the acquisition source stamped once at creation.

auto_assign_source let 26 campaigns rewrite source_id on the
opportunities, orders and invoices of every customer they touched. Measured
2026-07-31: 92,284 records across the three types hold an influence-only
campaign as their operational source while their write-once
original_source_id records that they were acquired elsewhere — Costco,
professional referrals, blog posts, cold calls.

ORDER OF OPERATIONS MATTERS. StopAutoAssigningInfluenceOnlyCampaigns must be
deployed first. Reconciling while those campaigns still auto-assign means
Campaign#synchronize_source simply re-overwrites everything this fixes on the
customer's next touch, and the sweep quietly undoes itself.

Resumable by construction: a corrected record no longer matches the scope, so
a re-run picks up exactly where a killed worker stopped. That matters because
a kamal deploy cycles every worker mid-run.

Constant Summary collapse

BATCH_SIZE =

Records per job. Small enough that a deploy loses little work, large enough
that 92k records don't need thousands of jobs.

500
RECORDS =

Where each record type's acquisition source lives, and which table it is
corrected on. Invoice carries no original_source_id of its own — it
inherits the acquisition source of the order it bills.

Table names are spelled out rather than derived with tableize: these three
happen to inflect correctly today, but a mapping that silently produces the
wrong table name would build valid SQL against the wrong rows.

{
  'Order' => { table: 'orders', acquisition: 'orders.original_source_id' },
  'Opportunity' => { table: 'opportunities', acquisition: 'opportunities.original_source_id' },
  'Invoice' => { table: 'invoices', acquisition: 'orders.original_source_id' }
}.freeze
LOCK_NAME =

Serializes runs so two jobs for the same record type cannot read the same
batch and write it twice. Held across select-and-correct, not just the
write, because the select is what decides the work.

'source_reconciliation'
LOCK_TIMEOUT =

Short: a batch is a bounded select plus one insert and a few updates.

10
LOCK_RETRY_DELAY =

How long to stand off when another run holds the lock. Long enough that a
busy holder finishes, short enough that the sweep doesn't crawl.

30.seconds

Instance Method Summary collapse

Instance Method Details

#perform(record_type, batch_id, dry_run = false) ⇒ Integer

Returns records corrected by this job.

Parameters:

  • record_type (String) —

    one of SourceReconciliation::RECORD_TYPES

  • batch_id (String) —

    groups this run so it can be reverted whole

  • dry_run (Boolean) (defaults to: false) —

    count and log the scope, change nothing

Returns:

  • (Integer) —

    records corrected by this job



57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'app/workers/source_reconciliation_worker.rb', line 57

def perform(record_type, batch_id, dry_run = false)
  # Selection and correction under one lock: two concurrent jobs would
  # otherwise select the same batch and both write it, producing duplicate
  # manifest entries whose `from_source_id` disagree — and a manifest that
  # disagrees with itself is not an undo log.
  outcome = Source.with_advisory_lock(LOCK_NAME, timeout_seconds: LOCK_TIMEOUT) do
    run_batch(record_type, batch_id, dry_run)
  end
  return outcome unless outcome == false

  # `with_advisory_lock` returns false WITHOUT running the block when it
  # cannot acquire in time. Returning here would end the chain silently: the
  # re-enqueue lives inside the block, so a job that loses the lock takes its
  # whole record type down with it and the sweep just stops. That is exactly
  # what stalled the 2026-07-31 run at 1,000 of 30,645 orders, with nothing
  # queued, nothing retrying, and no error recorded anywhere.
  Rails.logger.info("[SourceReconciliationWorker] #{record_type} deferred — another run holds the lock")
  self.class.perform_in(LOCK_RETRY_DELAY, record_type, batch_id, dry_run) unless dry_run
  0
end