Merge pull request coinbase#73 from stripe-private-oss-forks/nagl-RUN_WOFLO-310

Expose scheduled_time and current_attempt_scheduled_time on activity metadata

Author: nagl-stripe

lib/temporal/metadata.rb

@@ -21,7 +21,10 @@ def generate_activity_metadata(task, namespace)
           workflow_id: task.workflow_execution.workflow_id,
           workflow_name: task.workflow_type.name,
           headers: from_payload_map(task.header&.fields || {}),
-          heartbeat_details: from_details_payloads(task.heartbeat_details)
+          heartbeat_details: from_details_payloads(task.heartbeat_details),
+          # temporal doesn't render sub-second times, so we ignore the nanos field
+          scheduled_time: task.scheduled_time.seconds,
+          current_attempt_scheduled_time: task.current_attempt_scheduled_time.seconds
         )
       end
 

lib/temporal/metadata/activity.rb

@@ -3,9 +3,9 @@
 module Temporal
   module Metadata
     class Activity < Base
-      attr_reader :namespace, :id, :name, :task_token, :attempt, :workflow_run_id, :workflow_id, :workflow_name, :headers, :heartbeat_details
+      attr_reader :namespace, :id, :name, :task_token, :attempt, :workflow_run_id, :workflow_id, :workflow_name, :headers, :heartbeat_details, :scheduled_time, :current_attempt_scheduled_time
 
-      def initialize(namespace:, id:, name:, task_token:, attempt:, workflow_run_id:, workflow_id:, workflow_name:, headers: {}, heartbeat_details:)
+      def initialize(namespace:, id:, name:, task_token:, attempt:, workflow_run_id:, workflow_id:, workflow_name:, headers: {}, heartbeat_details:, scheduled_time:, current_attempt_scheduled_time:)
         @namespace = namespace
         @id = id
         @name = name
@@ -16,6 +16,8 @@ def initialize(namespace:, id:, name:, task_token:, attempt:, workflow_run_id:,
         @workflow_name = workflow_name
         @headers = headers
         @heartbeat_details = heartbeat_details
+        @scheduled_time = scheduled_time
+        @current_attempt_scheduled_time = current_attempt_scheduled_time
 
         freeze
       end
@@ -32,7 +34,9 @@ def to_h
           'run_id' => workflow_run_id,
           'activity_id' => id,
           'activity_name' => name,
-          'attempt' => attempt
+          'attempt' => attempt,
+          'scheduled_time' => scheduled_time,
+          'current_attempt_scheduled_time' => current_attempt_scheduled_time,
         }
       end
     end

lib/temporal/testing/local_workflow_context.rb

@@ -60,7 +60,9 @@ def execute_activity(activity_class, *input, **args)
           workflow_id: workflow_id,
           workflow_name: self.metadata.name,
           headers: execution_options.headers,
-          heartbeat_details: nil
+          heartbeat_details: nil,
+          scheduled_time: Time.now.to_i,
+          current_attempt_scheduled_time: Time.now.to_i,
         )
         context = LocalActivityContext.new(metadata)
 
@@ -110,7 +112,9 @@ def execute_local_activity(activity_class, *input, **args)
           workflow_id: workflow_id,
           workflow_name: self.metadata.name,
           headers: execution_options.headers,
-          heartbeat_details: nil
+          heartbeat_details: nil,
+          scheduled_time: Time.now.to_i,
+          current_attempt_scheduled_time: Time.now.to_i,
         )
         context = LocalActivityContext.new(metadata)
 

spec/fabricators/activity_metadata_fabricator.rb

@@ -11,4 +11,6 @@
   workflow_name 'TestWorkflow'
   headers { {} }
   heartbeat_details nil
+  scheduled_time 0
+  current_attempt_scheduled_time 0
 end

spec/fabricators/grpc/activity_task_fabricator.rb

@@ -11,6 +11,8 @@
   workflow_execution { Fabricate(:api_workflow_execution) }
   current_attempt_scheduled_time { Google::Protobuf::Timestamp.new.tap { |t| t.from_time(Time.now) } }
   started_time { Google::Protobuf::Timestamp.new.tap { |t| t.from_time(Time.now) } }
+  scheduled_time { Google::Protobuf::Timestamp.new.tap { |t| t.from_time(Time.now) } }
+  current_attempt_scheduled_time { Google::Protobuf::Timestamp.new.tap { |t| t.from_time(Time.now) } }  
   header do |attrs|
     fields = (attrs[:headers] || {}).each_with_object({}) do |(field, value), h|
       h[field] = Temporal.configuration.converter.to_payload(value)

spec/unit/lib/temporal/metadata/activity_spec.rb

@@ -16,6 +16,8 @@
       expect(subject.workflow_name).to eq(args.workflow_name)
       expect(subject.headers).to eq(args.headers)
       expect(subject.heartbeat_details).to eq(args.heartbeat_details)
+      expect(subject.scheduled_time).to eq(args.scheduled_time)
+      expect(subject.current_attempt_scheduled_time).to eq(args.current_attempt_scheduled_time)      
     end
 
     it { is_expected.to be_frozen }
@@ -36,7 +38,9 @@
         'namespace' => subject.namespace,
         'workflow_id' => subject.workflow_id,
         'workflow_name' => subject.workflow_name,
-        'run_id' => subject.workflow_run_id
+        'run_id' => subject.workflow_run_id,
+        'scheduled_time' => subject.scheduled_time,
+        'current_attempt_scheduled_time' => subject.current_attempt_scheduled_time,
       })
     end
   end

spec/unit/lib/temporal/testing/temporal_override_spec.rb

@@ -196,6 +196,8 @@ def execute
         expect(metadata.namespace).to eq('default-namespace')
         expect(metadata.workflow_id).to eq(workflow_id)
         expect(metadata.workflow_run_id).to eq(run_id)
+        expect(metadata.scheduled_time).to be > 0
+        expect(metadata.current_attempt_scheduled_time).to be > 0
       end
 
       describe 'execution control' do