Fix Future#flat when failures happen

Author: pitr-ch

lib/concurrent/edge/future.rb

@@ -1141,17 +1141,24 @@ def blocked_by
 
       def process_on_done(future)
         countdown = super(future)
-        value     = future.value!
         if countdown.nonzero?
+          internal_state = future.internal_state
+
+          unless internal_state.success?
+            complete_with internal_state
+            return countdown
+          end
+
+          value = internal_state.value
           case value
           when Future
             @BlockedBy.push value
             value.add_callback :pr_callback_notify_blocked, self
             @Countdown.value
           when Event
-            raise TypeError, 'cannot flatten to Event'
+            evaluate_to(lambda { raise TypeError, 'cannot flatten to Event' })
           else
-            raise TypeError, "returned value #{value.inspect} is not a Future"
+            evaluate_to(lambda { raise TypeError, "returned value #{value.inspect} is not a Future" })
           end
         end
         countdown
@@ -1174,6 +1181,10 @@ def clear_blocked_by!
         @BlockedBy.clear
         nil
       end
+
+      def completable?(countdown)
+        !@Future.internal_state.completed? && super(countdown)
+      end
     end
 
     # @!visibility private

spec/concurrent/edge/future_spec.rb

@@ -312,6 +312,16 @@
     it 'has flat map' do
       f = Concurrent.future { Concurrent.future { 1 } }.flat.then(&:succ)
       expect(f.value!).to eq 2
+
+      err = StandardError.new('boo')
+      f   = Concurrent.future { Concurrent.failed_future(err) }.flat
+      expect(f.reason).to eq err
+
+      f   = Concurrent.future { raise 'boo' }.flat
+      expect(f.reason.message).to eq 'boo'
+
+      f   = Concurrent.future { 'boo' }.flat
+      expect(f.reason).to be_an_instance_of TypeError
     end
   end