Each now stops when iteration emits a failing signal.

Author: apotonick

lib/trailblazer/macro/each.rb

@@ -1,7 +1,9 @@
 module Trailblazer
   module Macro
     # TODO: explain termini routing, Inject usage for "block activity", Out() => []
+    #       BLOG: Each() is a perfect example of how versatile the TRB mechanics are
 
+    # @api private The internals here are considered private and might change at some point.
     def self.Each(block_activity=nil, dataset_from: nil, item_key: :item, id: Macro.id_for(block_activity, macro: :Each, hint: dataset_from), collect: false, **dsl_options_for_iterated, &block)
       # TODO: 2.5. fix
       iterated_activity, outputs_from_block_activity = Trailblazer::Macro.block_activity_for(block_activity, &block)
@@ -29,6 +31,11 @@ def self.Each(block_activity=nil, dataset_from: nil, item_key: :item, id: Macro.
 
       each_activity = Trailblazer::Activity::Railway(termini: termini_from_block_activity) do
 
+        # TODO: make publicly configurable.
+        @state.update!(:fields) do |fields|
+          fields.merge(failing_semantics: [:failure, :fail_fast])
+        end
+
         step Subprocess(iterated_activity, strict: true),
             id: "ITERATED FIXME",
             Inject(:index, filter_builder: my_filter_builder) => my_lowlevel_inject_filter,
@@ -42,6 +49,8 @@ def self.call((ctx, flow_options), runner:, **circuit_options)
           # We don't really need to override/replace {circuit} as we only want to change the way it's run.
           iterated_railway = to_h[:circuit].to_h[:map].keys[1] # DISCUSS: maybe find by id?
 
+          failing_semantics = @state.get(:fields).fetch(:failing_semantics)
+
           dataset           = ctx.fetch(:dataset)
           signal = nil
 
@@ -54,6 +63,9 @@ def self.call((ctx, flow_options), runner:, **circuit_options)
 
             # we "inject" item_key and index via Runner.(..., item_key => ..) and then the input filter grabs that.
             signal, (ctx, flow_options) = runner.(iterated_railway, [ctx, flow_options], runner: runner, **circuit_options, activity: self, **each_options_for_iterated)
+
+            # Break the loop if {iterated_activity} emits a failure signal.
+            break if failing_semantics.include?(signal.to_h[:semantic]) # TODO: use generic check from older macro
           end
 
           return signal, [ctx, flow_options]

test/docs/each_test.rb

@@ -241,6 +241,7 @@ def composers_for_each(ctx, model:, **)
   end # F
 
   it "failure in Each" do
+    # last iteration fails:
     assert_invoke F::Song::Activity::Cover, params: {id: 2},
       expected_ctx_variables: {
         model: B::Song.find_by(id: 2),
@@ -249,6 +250,15 @@ def composers_for_each(ctx, model:, **)
       seq: "[]",
       terminus: :failure
 
+    # first iteration fails:
+    assert_invoke F::Song::Activity::Cover, params: {id: 999},
+      expected_ctx_variables: {
+        model: B::Song.find_by(id: 999),
+        collected_from_each: [nil],
+      },
+      seq: "[]",
+      terminus: :failure
+
     Trailblazer::Developer.wtf?(F::Song::Activity::Cover, {params: {id: 2}, seq: []})
   end