Merge pull request #340 from d2army/transfers-master

Making more deterministic workflow resiliency

Author: d2army

CHANGELOG.md

@@ -1,4 +1,7 @@
 # Changelog
 
+## 0.0.4
+- Patching
+
 ## 0.0.1
 - First release

lib/temporal/concerns/input_deserializer.rb

@@ -0,0 +1,26 @@
+module Temporal
+  module Concerns
+    module InputDeserializer
+      def deserialize(input)
+        JSON.deserialize(input)
+      rescue Oj::ParseError
+        # Copied over from the Cadence side, similar situation happening with Temporal
+        #
+        # cadence official go-client serializes / deserializes input in a different format than this ruby client
+        # adding additional deserialization logic here to help read input that is passed from go-client
+        # https://github.com/uber-go/cadence-client/blob/0.18.x/internal/encoding.go#L45-L58
+        #
+        # this ruby client serializes / deserializes everything as one big string like below:
+        # [1012474654, "second input"]
+        #
+        # while go client serializes input as separate input followed by line break
+        # 1012474654
+        # second input
+        args = input.split(/\n/)
+        res = args.map do |arg|
+          JSON.deserialize(arg)
+        end
+      end
+    end
+  end
+end

lib/temporal/version.rb

@@ -1,3 +1,3 @@
 module Temporal
-  VERSION = '0.0.3'.freeze
+  VERSION = '0.0.4'.freeze
 end

lib/temporal/workflow/history/event.rb

@@ -1,7 +1,11 @@
+require 'temporal/concerns/input_deserializer'
+
 module Temporal
   class Workflow
     class History
       class Event
+        include ::Temporal::Concerns::InputDeserializer
+        include Concerns::Payloads
         EVENT_TYPES = %w[
           ACTIVITY_TASK_STARTED
           ACTIVITY_TASK_COMPLETED
@@ -52,6 +56,19 @@ def originating_event_id
           end
         end
 
+        def target_attributes
+          case type
+          when 'ACTIVITY_TASK_SCHEDULED'
+            {
+              activity_id: attributes.activity_id&.to_i, # activity_id is a string from thrift
+              activity_type: attributes.activity_type&.name,
+              input: from_payloads(attributes.input)
+            }
+          else
+            {}
+          end
+        end
+
         private
 
         def extract_attributes(raw_event)

lib/temporal/workflow/history/event_target.rb

@@ -5,6 +5,7 @@ class Workflow
     class History
       class EventTarget
         class UnexpectedEventType < InternalError; end
+        class UnexpectedCommandType < InternalError; end
 
         ACTIVITY_TYPE                         = :activity
         CANCEL_ACTIVITY_REQUEST_TYPE          = :cancel_activity_request
@@ -19,7 +20,7 @@ class UnexpectedEventType < InternalError; end
         UPSERT_SEARCH_ATTRIBUTES_REQUEST_TYPE = :upsert_search_attributes_request
 
         # NOTE: The order is important, first prefix match wins (will be a longer match)
-        TARGET_TYPES = {
+        EVENT_TARGET_TYPES = {
           'ACTIVITY_TASK_CANCEL_REQUESTED'             => CANCEL_ACTIVITY_REQUEST_TYPE,
           'ACTIVITY_TASK'                              => ACTIVITY_TYPE,
           'REQUEST_CANCEL_ACTIVITY_TASK'               => CANCEL_ACTIVITY_REQUEST_TYPE,
@@ -38,25 +39,58 @@ class UnexpectedEventType < InternalError; end
           'WORKFLOW_EXECUTION'                         => WORKFLOW_TYPE,
         }.freeze
 
-        attr_reader :id, :type
+        WORKFLOW_TARGET_TYPES = {
+          'Temporal::Workflow::Command::ScheduleActivity'            => ACTIVITY_TYPE,
+          'Temporal::Workflow::Command::RequestActivityCancellation' => CANCEL_ACTIVITY_REQUEST_TYPE,
+          'Temporal::Workflow::Command::RecordMarker'                => MARKER_TYPE,
+          'Temporal::Workflow::Command::StartTimer'                  => TIMER_TYPE,
+          'Temporal::Workflow::Command::CancelTimer'                 => CANCEL_TIMER_REQUEST_TYPE,
+          'Temporal::Workflow::Command::CompleteWorkflow'            => WORKFLOW_TYPE,
+          'Temporal::Workflow::Command::FailWorkflow'                => WORKFLOW_TYPE,
+          'Temporal::Workflow::Command::StartChildWorkflow'          => CHILD_WORKFLOW_TYPE,
+          'Temporal::Workflow::Command::SignalExternalWorkflow'      => EXTERNAL_WORKFLOW_TYPE,
+          'Temporal::Workflow::Command::CancelExternalWorkflow'      => CANCEL_EXTERNAL_WORKFLOW_REQUEST_TYPE,
+          'Temporal::Workflow::Command::UpsertSearchAttributes'      => UPSERT_SEARCH_ATTRIBUTES_REQUEST_TYPE,
+          'Temporal::Workflow::Command::ContinueAsNew'               => WORKFLOW_TYPE,
+        }.freeze
+
+        COMMAND_ATTRIBUTE_LISTS = {
+          'Temporal::Workflow::Command::ScheduleActivity'            => [:activity_id, :activity_type, :input],
+        }
+        attr_reader :id, :type, :attributes
 
         def self.workflow
           @workflow ||= new(1, WORKFLOW_TYPE)
         end
 
         def self.from_event(event)
-          _, target_type = TARGET_TYPES.find { |type, _| event.type.start_with?(type) }
+          _, target_type = EVENT_TARGET_TYPES.find { |type, _| event.type.start_with?(type) }
 
           unless target_type
             raise UnexpectedEventType, "Unexpected event #{event.type}"
           end
 
-          new(event.originating_event_id, target_type)
+          new(event.originating_event_id, target_type, attributes: event.target_attributes)
+        end
+
+        def self.from_command(command_id, command)
+
+          command_type = command.class.name
+          target_type = WORKFLOW_TARGET_TYPES[command_type]
+
+          unless target_type
+            raise UnexpectedCommandType, "Unexpected command type #{command_type}"
+          end
+
+          attribute_list = COMMAND_ATTRIBUTE_LISTS.fetch(command_type, [])
+
+          new(command_id, target_type, attributes: command.to_h.slice(*attribute_list))
         end
 
-        def initialize(id, type)
+        def initialize(id, type, attributes: {})
           @id = id
           @type = type
+          @attributes = attributes
 
           freeze
         end
@@ -74,7 +108,7 @@ def hash
         end
 
         def to_s
-          "#{type} (#{id})"
+          "#{type}: #{id} (#{attributes})"
         end
       end
     end

lib/temporal/workflow/state_manager.rb

@@ -66,7 +66,7 @@ def schedule(command)
         validate_append_command(command)
         commands << [command_id, command]
 
-        return [event_target_from(command_id, command), cancelation_id]
+        [History::EventTarget.from_command(command_id, command), cancelation_id]
       end
 
       def release?(release_name)
@@ -309,32 +309,6 @@ def apply_event(event)
         end
       end
 
-      def event_target_from(command_id, command)
-        target_type =
-          case command
-          when Command::ScheduleActivity
-            History::EventTarget::ACTIVITY_TYPE
-          when Command::RequestActivityCancellation
-            History::EventTarget::CANCEL_ACTIVITY_REQUEST_TYPE
-          when Command::RecordMarker
-            History::EventTarget::MARKER_TYPE
-          when Command::StartTimer
-            History::EventTarget::TIMER_TYPE
-          when Command::CancelTimer
-            History::EventTarget::CANCEL_TIMER_REQUEST_TYPE
-          when Command::CompleteWorkflow, Command::FailWorkflow
-            History::EventTarget::WORKFLOW_TYPE
-          when Command::StartChildWorkflow
-            History::EventTarget::CHILD_WORKFLOW_TYPE
-          when Command::UpsertSearchAttributes
-            History::EventTarget::UPSERT_SEARCH_ATTRIBUTES_REQUEST_TYPE
-          when Command::SignalExternalWorkflow
-            History::EventTarget::EXTERNAL_WORKFLOW_TYPE
-          end
-
-        History::EventTarget.new(command_id, target_type)
-      end
-
       def dispatch(history_target, name, *attributes)
         dispatcher.dispatch(history_target, name, attributes)
       end
@@ -352,8 +326,9 @@ def discard_command(history_target)
             "A command in the history of previous executions, #{history_target}, was not scheduled upon replay. " + NONDETERMINISM_ERROR_SUGGESTION
         end
 
-        replay_target = event_target_from(replay_command_id, replay_command)
-        if history_target != replay_target
+        replay_target = History::EventTarget.from_command(replay_command_id, replay_command)
+
+        if history_target != replay_target || history_target.attributes != replay_target.attributes
           raise NonDeterministicWorkflowError,
             "Unexpected command.  The replaying code is issuing: #{replay_target}, "\
             "but the history of previous executions recorded: #{history_target}. " + NONDETERMINISM_ERROR_SUGGESTION

lib/temporal/workflow/task_processor.rb

@@ -94,6 +94,7 @@ def queue_time_ms
 
       def fetch_full_history
         events = task.history.events.to_a
+    
         next_page_token = task.next_page_token
         while !next_page_token.empty? do
           response = connection.get_workflow_execution_history(

spec/fabricators/grpc/history_event_fabricator.rb

@@ -1,5 +1,6 @@
 require 'securerandom'
 require 'temporal/concerns/payloads'
+require 'temporal/json'
 
 class TestSerializer
   extend Temporal::Concerns::Payloads
@@ -10,9 +11,10 @@ class TestSerializer
 Fabricator(:api_history_event, from: Temporalio::Api::History::V1::HistoryEvent) do
   event_id { 1 }
   event_time { Time.now }
+  transient input: nil
 end
 
-Fabricator(:api_workflow_execution_started_event, from: :api_history_event) do
+Fabricator(:api_workflow_execution_started_eevent, from: :api_history_event) do
   transient :headers, :search_attributes
   event_type { Temporalio::Api::Enums::V1::EventType::EVENT_TYPE_WORKFLOW_EXECUTION_STARTED }
   event_time { Time.now }
@@ -24,7 +26,7 @@ class TestSerializer
     Temporalio::Api::History::V1::WorkflowExecutionStartedEventAttributes.new(
       workflow_type: Fabricate(:api_workflow_type),
       task_queue: Fabricate(:api_task_queue),
-      input: nil,
+      input: Temporal::JSON.serialize(attrs[:input]),
       workflow_execution_timeout: 60,
       workflow_task_timeout: 15,
       original_execution_run_id: SecureRandom.uuid,

spec/unit/lib/temporal/workflow/history/event_spec.rb

@@ -41,4 +41,31 @@
       it { is_expected.to eq(raw_event.event_id) }
     end
   end
+
+
+  describe '#target_attributes' do
+    subject { described_class.new(raw_event).target_attributes }
+
+    context 'when event is ActivityTaskScheduled' do
+      let(:input) { ['foo', 'bar', { 'foo' => 'bar' }] }
+      let(:raw_event) do
+        Fabricate(:activity_task_scheduled_event_thrift, eventId: 42, input: input)
+      end
+
+      it {
+        is_expected.to eq({ activity_id: 42, activity_type: 'TestActivity', input: input })
+      }
+    end
+
+    context 'when event is WorkflowTaskScheduled' do
+      let(:input) { ['foo', 'bar', { 'foo' => 'bar' }] }
+      let(:raw_event) do
+        Fabricate(:workflow_task_scheduled_event_thrift, eventId: 42)
+      end
+
+      it {
+        is_expected.to eq({})
+      }
+    end
+  end
 end

spec/unit/lib/temporal/workflow/history/event_target_spec.rb

@@ -1,24 +1,105 @@
 require 'temporal/workflow/history/event_target'
 require 'temporal/workflow/history/event'
+require 'temporal/workflow/command'
 
 describe Temporal::Workflow::History::EventTarget do
   describe '.from_event' do
     subject { described_class.from_event(event) }
     let(:event) { Temporal::Workflow::History::Event.new(raw_event) }
 
     context 'when event is TIMER_STARTED' do
-      let(:raw_event) { Fabricate(:api_timer_started_event) }
+      let(:raw_event) { Fabricate(:api_timer_started_event, eventId: 42) }
 
-      it 'sets type to timer' do
+      it 'sets id and type' do
+        expect(subject.id).to eq(42)
         expect(subject.type).to eq(described_class::TIMER_TYPE)
+        expect(subject.attributes).to eq({})
       end
     end
 
     context 'when event is TIMER_CANCELED' do
-      let(:raw_event) { Fabricate(:api_timer_canceled_event) }
+      let(:raw_event) { Fabricate(:api_timer_canceled_event,  eventId: 42) }
 
-      it 'sets type to cancel_timer_request' do
+      it 'sets id and type' do
+        expect(subject.id).to eq(42)
         expect(subject.type).to eq(described_class::CANCEL_TIMER_REQUEST_TYPE)
+        expect(subject.attributes).to eq({})
+      end
+    end
+
+    context 'when event is ScheduleActivity' do
+      let(:input) { ['foo', 'bar', { 'foo' => 'bar' }] }
+      let(:raw_event) { Fabricate(:activity_task_scheduled_event_thrift, eventId: 42, input: input) }
+
+      it 'sets id, type and attributes' do
+        expect(subject.id).to eq(42)
+        expect(subject.type).to eq(described_class::ACTIVITY_TYPE)
+        expect(subject.attributes).to eq({ activity_id: 42, activity_type: 'TestActivity', input: input })
+      end
+    end
+  end
+
+  describe '.from_decision' do
+    subject { described_class.from_command(42, decision) }
+
+    context 'when decision is ScheduleActivity' do
+      let(:raw_decision) { { activity_type: 'foo', activity_id: 123, input: ['bar'], domain: 'domain' } }
+      let(:decision) { Temporal::Workflow::Command::ScheduleActivity.new(**raw_decision) }
+
+      it 'sets id, type' do
+        expect(subject.id).to eq(42)
+        expect(subject.type).to eq(described_class::ACTIVITY_TYPE)
+      end
+
+      it 'sets and slice the attributes' do
+        expect(raw_decision).to include(subject.attributes)
+        expect(subject.attributes.keys).to eq(%i[activity_id activity_type input])
+      end
+    end
+
+    context 'when decision is StartTimer' do
+      let(:raw_decision) { { timeout: 10, timer_id: 123 } }
+      let(:decision) { Temporal::Workflow::Command::StartTimer.new(**raw_decision) }
+
+      it 'sets id, type' do
+        expect(subject.id).to eq(42)
+        expect(subject.type).to eq(described_class::TIMER_TYPE)
+      end
+
+      it 'sets empty attributes' do
+        expect(subject.attributes.keys).to eq([])
+      end
+    end
+  end
+
+  describe '#==' do
+    subject do
+      described_class.new(id, type, attributes: attributes) ==
+        described_class.new(42, 'type', attributes: { foo: 'bar' })
+    end
+    let(:id) { 42 }
+    let(:type) { 'type' }
+    let(:attributes) { { foo: 'bar' } }
+
+    context 'when all value are the same' do
+      it 'returns true' do
+        expect(subject).to eq(true)
+      end
+    end
+
+    context 'when id are different' do
+      let(:id) { 1 }
+
+      it 'returns false' do
+        expect(subject).to eq(false)
+      end
+    end
+
+    context 'when type are different' do
+      let(:type) { 'other_type' }
+
+      it 'returns false' do
+        expect(subject).to eq(false)
       end
     end