Do not require heartbeating to be done in context (#280)

Fixes #279

Author: cretz

temporalio/lib/temporalio/internal/worker/activity_worker.rb

@@ -251,7 +251,7 @@ def run_activity(defn, activity, input)
             impl = @worker._activity_interceptors.reverse_each.reduce(impl) do |acc, int|
               int.intercept_activity(acc)
             end
-            impl.init(OutboundImplementation.new(self))
+            impl.init(OutboundImplementation.new(self, activity.info.task_token))
 
             # Execute
             result = impl.execute(input)
@@ -381,15 +381,16 @@ def execute(input)
         end
 
         class OutboundImplementation < Temporalio::Worker::Interceptor::Activity::Outbound
-          def initialize(worker)
+          def initialize(worker, task_token)
             super(nil) # steep:ignore
             @worker = worker
+            @task_token = task_token
           end
 
           def heartbeat(input)
             @worker.bridge_worker.record_activity_heartbeat(
               Bridge::Api::CoreInterface::ActivityHeartbeat.new(
-                task_token: Activity::Context.current.info.task_token,
+                task_token: @task_token,
                 details: ProtoUtils.convert_to_payload_array(@worker.worker.options.client.data_converter,
                                                              input.details)
               ).to_proto

temporalio/lib/temporalio/worker/thread_pool.rb

@@ -125,7 +125,7 @@ def _worker_task_completed
 
       private
 
-      def locked_assign_worker(&block)
+      def locked_assign_worker(&block) # rubocop:disable Naming/PredicateMethod
         # keep growing if the pool is not at the minimum yet
         worker, = @ready.pop || locked_add_busy_worker
         if worker

temporalio/sig/temporalio/internal/worker/activity_worker.rbs

@@ -49,7 +49,7 @@ module Temporalio
         end
 
         class OutboundImplementation < Temporalio::Worker::Interceptor::Activity::Outbound
-          def initialize: (ActivityWorker worker) -> void
+          def initialize: (ActivityWorker worker, String task_token) -> void
         end
       end
     end

temporalio/test/worker_activity_test.rb

@@ -361,6 +361,39 @@ def test_heartbeat_details
                  execute_activity(HeartbeatDetailsActivity, retry_max_attempts: 2, heartbeat_timeout: 0.8)
   end
 
+  class BackgroundHeartbeatActivity < Temporalio::Activity::Definition
+    def execute
+      # First attempt sends a heartbeat with details and fails,
+      # next attempt just returns the first attempt's details
+      if Temporalio::Activity::Context.current.info.attempt == 1
+        # Send heartbeat in background, and wait on background call
+        # to complete before raising error
+        queue = Queue.new
+        context = Temporalio::Activity::Context.current
+        if Fiber.current_scheduler
+          Fiber.schedule do
+            context.heartbeat('some detail')
+            queue.push(nil)
+          end
+        else
+          Thread.new do
+            context.heartbeat('some detail')
+            queue.push(nil)
+          end
+        end
+        queue.pop
+        raise 'Intentional error'
+      else
+        "details: #{Temporalio::Activity::Context.current.info.heartbeat_details}"
+      end
+    end
+  end
+
+  def test_background_heartbeat
+    assert_equal 'details: ["some detail"]',
+                 execute_activity(BackgroundHeartbeatActivity, retry_max_attempts: 2, heartbeat_timeout: 0.8)
+  end
+
   class ShieldingActivity < Temporalio::Activity::Definition
     attr_reader :canceled, :levels_reached