Class: EdiOrchestratorFlowWorker

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

Overview

Runs ONE EDI flow for ONE orchestrator class — every partner of that vendor,
sequentially, exactly as Edi::BaseOrchestrator.execute_flow always has.

The unit of the fan-out in Workers::EdiFlowDispatch: the vendor keeps its own
internal pacing (Amazon sleeps between marketplaces) while a partner that hangs
can no longer hold a lock that suppresses every OTHER vendor's feeds.

Uniqueness is per (flow, orchestrator) because the lock digest is built from
the args, so a wedged Menard run leaves Wayfair's next run free — the whole
point. lock_ttl still bounds a wedge that outlives its schedule interval; see
EdiInventoryFlowWorker for why an advisory lock is the wrong tool here.

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.promote_failures!(results, label) ⇒ Array<Hash>

Turns execute_flow's per-partner :error entries into a raised exception.

execute_flow rescues per partner so one bad vendor partner can't abort the
rest — correct, but it means the caller "succeeds" with errors buried in a
return value nobody reads. Raising puts them in status.failures for
EdiFlowFinalizer, and surfaces them on the targeted path too
(Workers::EdiFlowDispatch#dispatch_flow), so both fail the same way.

Deliberately does NOT retry: retry: 0 is the global default, and re-running
would re-submit the feeds that DID succeed. The next scheduled run is the
retry.

Parameters:

Returns:

  • (Array<Hash>) —

    the results, when none of them errored

Raises:

  • (RuntimeError) —

    when any entry has result: :error



35
36
37
38
39
40
41
 
# File 'app/workers/edi_orchestrator_flow_worker.rb', line 35

def self.promote_failures!(results, label)
  failures = Array(results).select { |r| r[:result] == :error }
  return results if failures.empty?

  raise "#{label}: #{failures.size}/#{Array(results).size} partner(s) failed — " \
        "#{failures.map { |f| "#{f[:partner]}: #{f[:error]}" }.join('; ')}"
end

Instance Method Details

#perform(flow, orchestrator_name) ⇒ Array<Hash>

Returns result entries from the dispatcher.

Parameters:

  • flow (String) —

    e.g. 'execute_inventory_flow'

  • orchestrator_name (String) —

    concrete EDI orchestrator class name

Returns:

  • (Array<Hash>) —

    result entries from the dispatcher

Raises:

  • (RuntimeError) —

    when any partner of this orchestrator errored



47
48
49
50
 
# File 'app/workers/edi_orchestrator_flow_worker.rb', line 47

def perform(flow, orchestrator_name)
  results = Edi::BaseOrchestrator.execute_flow(flow.to_sym, orchestrator_name: orchestrator_name)
  self.class.promote_failures!(results, "#{orchestrator_name} #{flow}")
end