Warn on unawaited workflow futures with failures

temporalio/lib/temporalio/internal/worker/workflow_instance.rb

@@ -90,6 +90,7 @@ def initialize(details)
           @buffered_signals = {} # Keyed by signal name, value is array of signal jobs
           # TODO(cretz): Should these be sets instead? Both should be fairly low counts.
           @in_progress_handlers = [] # Value is HandlerExecution
+          @futures_with_failures = [] # Value is Workflow::Future
           @patches_notified = []
           @definition = details.definition
           @interceptors = details.interceptors
@@ -226,6 +227,7 @@ def activate(activation)
             !c.cancel_workflow_execution.nil?
           end
             warn_on_any_unfinished_handlers
+            warn_on_any_unawaited_failed_futures
           end
 
           # Return success or failure
@@ -816,6 +818,23 @@ def scoped_logger_info
           @scoped_logger_info.merge({ update_id: update_info.id, update_name: update_info.name })
         end
 
+        def track_future_with_failure(future)
+          @futures_with_failures << future
+        end
+
+        def warn_on_any_unawaited_failed_futures
+          unawaited = @futures_with_failures.reject(&:awaited?)
+          return if unawaited.empty?
+
+          failures_str = JSON.generate(unawaited.map { |f| { type: f.failure.class.name, message: f.failure.message } })
+          warn(
+            "[TMPRL1103] Workflow #{info.workflow_id} finished with #{unawaited.size} unawaited failed " \
+            "future(s). These failures were silently swallowed because the future was never awaited with `wait` " \
+            "or `wait_no_raise`. This is usually a bug — if the failure is expected, await the future and handle " \
+            "the error. Unawaited failures: #{failures_str}"
+          )
+        end
+
         def warn_on_any_unfinished_handlers
           updates, signals = in_progress_handlers.select do |h|
             h.unfinished_policy == Workflow::HandlerUnfinishedPolicy::WARN_AND_ABANDON

temporalio/lib/temporalio/internal/worker/workflow_instance/context.rb

@@ -59,6 +59,10 @@ def current_deployment_version
             @instance.current_deployment_version
           end
 
+          def track_future_with_failure(future)
+            @instance.track_future_with_failure(future)
+          end
+
           def current_history_length
             @instance.current_history_length
           end

temporalio/lib/temporalio/workflow/future.rb

@@ -76,6 +76,7 @@ def initialize(&block)
         @done = false
         @result = nil
         @failure = nil
+        @awaited = false
         @block_given = block_given?
         return unless block_given?
 
@@ -89,6 +90,7 @@ def initialize(&block)
           @failure = e
         ensure
           @done = true
+          Workflow._current_or_nil&.track_future_with_failure(self) if @failure
         end
       end
 
@@ -132,12 +134,18 @@ def failure=(failure)
         @done = true
       end
 
+      # @return [Boolean] True if the future's result or failure has been observed via {wait} or {wait_no_raise}.
+      def awaited?
+        @awaited
+      end
+
       # Wait on the future to complete. This will return the success or raise the failure. To not raise, use
       # {wait_no_raise}.
       #
       # @return [Object] Result on success.
       # @raise [Exception] Failure if occurred.
       def wait
+        @awaited = true
         Workflow.wait_condition(cancellation: nil) { done? }
         Kernel.raise failure if failure? # steep:ignore
 
@@ -148,6 +156,7 @@ def wait
       #
       # @return [Object, nil] Result on success or nil on failure.
       def wait_no_raise
+        @awaited = true
         Workflow.wait_condition(cancellation: nil) { done? }
         result
       end

temporalio/test/worker_workflow_test.rb

@@ -3028,4 +3028,86 @@ def test_leftover_wait
     # This used to fail because of a workflow task failure caused by a leftover wait condition
     assert_nil handle.fetch_history_events.find(&:workflow_task_failed_event_attributes)
   end
+
+  class UnawaitedFutureError < StandardError; end
+  class UnawaitedFutureError2 < StandardError; end
+
+  class UnawaitedFailedFutureWorkflow < Temporalio::Workflow::Definition
+    def execute(scenario)
+      case scenario.to_sym
+      when :unawaited_single
+        Temporalio::Workflow::Future.new { raise UnawaitedFutureError, 'activity failed' }
+        Temporalio::Workflow.sleep(0.1)
+        'done'
+      when :unawaited_multiple
+        Temporalio::Workflow::Future.new { raise UnawaitedFutureError, 'first failure' }
+        Temporalio::Workflow::Future.new { raise UnawaitedFutureError2, 'second failure' }
+        Temporalio::Workflow.sleep(0.1)
+        'done'
+      when :awaited_with_wait
+        fut = Temporalio::Workflow::Future.new { raise UnawaitedFutureError, 'activity failed' }
+        begin
+          fut.wait
+        rescue UnawaitedFutureError
+          # handled
+        end
+        'done'
+      when :awaited_with_wait_no_raise
+        fut = Temporalio::Workflow::Future.new { raise UnawaitedFutureError, 'activity failed' }
+        fut.wait_no_raise
+        'done'
+      when :unawaited_success
+        Temporalio::Workflow::Future.new { 'ok' }
+        Temporalio::Workflow.sleep(0.1)
+        'done'
+      else
+        raise NotImplementedError
+      end
+    end
+  end
+
+  def test_unawaited_failed_future_warns
+    _, err = safe_capture_io do
+      execute_workflow(UnawaitedFailedFutureWorkflow, 'unawaited_single', logger: Logger.new($stdout))
+    end
+    lines = err.split("\n")
+    warning_lines = lines.select { |l| l.include?('TMPRL1103') }
+    assert_equal 1, warning_lines.size
+    assert_includes warning_lines.first, '1 unawaited failed future(s)'
+    assert_includes warning_lines.first, 'UnawaitedFutureError'
+    assert_includes warning_lines.first, 'activity failed'
+  end
+
+  def test_unawaited_multiple_failed_futures_warns
+    _, err = safe_capture_io do
+      execute_workflow(UnawaitedFailedFutureWorkflow, 'unawaited_multiple', logger: Logger.new($stdout))
+    end
+    lines = err.split("\n")
+    warning_lines = lines.select { |l| l.include?('TMPRL1103') }
+    assert_equal 1, warning_lines.size
+    assert_includes warning_lines.first, '2 unawaited failed future(s)'
+    assert_includes warning_lines.first, 'UnawaitedFutureError'
+    assert_includes warning_lines.first, 'UnawaitedFutureError2'
+  end
+
+  def test_awaited_failed_future_with_wait_no_warning
+    _, err = safe_capture_io do
+      execute_workflow(UnawaitedFailedFutureWorkflow, 'awaited_with_wait', logger: Logger.new($stdout))
+    end
+    refute_includes err, 'TMPRL1103'
+  end
+
+  def test_awaited_failed_future_with_wait_no_raise_no_warning
+    _, err = safe_capture_io do
+      execute_workflow(UnawaitedFailedFutureWorkflow, 'awaited_with_wait_no_raise', logger: Logger.new($stdout))
+    end
+    refute_includes err, 'TMPRL1103'
+  end
+
+  def test_unawaited_successful_future_no_warning
+    _, err = safe_capture_io do
+      execute_workflow(UnawaitedFailedFutureWorkflow, 'unawaited_success', logger: Logger.new($stdout))
+    end
+    refute_includes err, 'TMPRL1103'
+  end
 end