Module: Workers::EdiFlowDispatch

Extended by:
ActiveSupport::Concern
Included in:
EdiInventoryFlowWorker, EdiPriceFlowWorker
Defined in:
app/concerns/workers/edi_flow_dispatch.rb

Overview

Turns a monolithic EDI flow worker into a Sidekiq Pro batch of one job per
ORCHESTRATOR CLASS, and gives the batch a completion callback.

The wiki's serial → parallel → back-to-serial shape
(https://github.com/sidekiq/sidekiq/wiki/Complex-Job-Workflows-with-Batches),
already used by EdiProductDataFlowWorker. Two things it buys that the
single job could not:

  1. Isolation. Edi::BaseOrchestrator.execute_flow walks every partner of
    every orchestrator in ONE job, so a partner that HANGS holds the worker's
    unique lock and silently coalesces every later scheduled run — a hung
    Menard upload stopped Wayfair, Walmart, CommerceHub, Amazon and 31
    reseller feeds for 8.5 hours on 2026-08-02. Fanned out, the lock is
    per-orchestrator and Menard can only ever stop Menard.
  2. A consumer for the results. execute_flow builds a per-partner result
    array with :error entries and returns it to a Sidekiq worker, which
    discards it. EdiFlowFinalizer reports it.

PER CLASS, NOT PER PARTNER — deliberately. Each vendor's flow paces itself
internally (Amazon sleeps between marketplaces; its twelve share two SP-API
accounts whose feed quota is metered per ACCOUNT — see
Edi::Amazon::FeedSubmissionBudget). Splitting partners into parallel jobs
discards that pacing. One job per class keeps every vendor's sequencing
exactly as it is today and still isolates vendors from each other.

ponytail: the fanned-out jobs run on the caller's queue (default), so peak
memory is now N orchestrators in flight rather than one. Realistically 3-4 do
real work per hour and each holds what the single job already held
sequentially. If peak RSS becomes the constraint, the upgrade is a dedicated
queue on the sidekiq_edi role (concurrency 4) — a capacity change with its
own DB_POOL/PgBouncer budget review, not a code change here.

Instance Method Summary collapse

Instance Method Details

#dispatch_flow(flow, options) ⇒ String, Array<Hash>

Runs +flow+ across every orchestrator class that has work for it.

A targeted request (CRM's per-partner buttons pass partner) runs inline:
fanning out a single target buys nothing and would drop the partner filter.
It still promotes partner failures to an exception, so a targeted run and a
fanned-out one fail the same way — otherwise execute_flow's :error
entries would be swallowed on exactly the path a human is watching.

Parameters:

  • flow (Symbol) —

    the flow to run, e.g. :execute_inventory_flow

  • options (Hash) —

    filters from the job payload

Options Hash (options):

  • 'orchestrator_name' (String, nil) —

    restrict to one orchestrator class

  • 'partner' (String, nil) —

    restrict to one partner, and run inline

Returns:

  • (String, Array<Hash>) —

    the batch id when fanned out, execute_flow
    results when targeted, or an empty array when nothing has work

Raises:

  • (RuntimeError) —

    when a targeted run had partner failures



54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'app/concerns/workers/edi_flow_dispatch.rb', line 54

def dispatch_flow(flow, options)
  orchestrator_name = options['orchestrator_name'].presence
  partner = options['partner'].presence
  if partner
    results = Edi::BaseOrchestrator.execute_flow(flow, orchestrator_name:, partner:)
    return EdiOrchestratorFlowWorker.promote_failures!(results, "#{flow} #{partner}")
  end

  targets = orchestrator_classes_with_work(flow, orchestrator_name)
  return [] if targets.empty?

  enqueue_batch(flow, targets)
end