[Feature] Failure converter (#55)

* Add a dummy implementation of a FailureConverter

* Implement FailureConverter::Basic#from_failure

* Implement FailureConverter::Basic#to_failure

* Support encoded attributes and enums

* Fix rubocop violations

* Add types for the FailureConverter::Basic

* Rubocop fixes

* Integrate FailureConverter into other parts of the SDK

* Add ExpectedResponse error

* Add missing specs

* Extend Basic failure converter from Base

* Switch to passing payload converter per-call

* Skip RBS for now due to an incompatibility

Author: antstorm

.github/workflows/ci.yml

@@ -29,7 +29,9 @@ jobs:
           go-version: "1.19"
       - run: bundle install
       - run: bundle exec rubocop
-      - run: bundle exec steep check
+      # Turned off for now because of an incompatibility in steep
+      # More info — https://github.com/soutaro/steep/issues/477
+      # - run: bundle exec steep check
       - run: bundle exec rake bridge:lint
       - run: bundle exec rake bridge:release
       - run: bundle exec rake test_server:build

.rubocop.yml

@@ -29,6 +29,10 @@ Style/IfUnlessModifier:
 Style/GuardClause:
   Enabled: false
 
+Style/MultilineBlockChain:
+  Exclude:
+    - spec/**/*_spec.rb
+
 Style/RedundantReturn:
   Enabled: false
 

lib/temporal/client/implementation.rb

@@ -1,8 +1,10 @@
 require 'socket'
 require 'temporal/api/workflowservice/v1/request_response_pb'
 require 'temporal/client/workflow_handle'
-require 'temporal/errors'
+require 'temporal/error/failure'
+require 'temporal/error/workflow_failure'
 require 'temporal/interceptor/chain'
+require 'temporal/timeout_type'
 require 'temporal/version'
 require 'temporal/workflow/execution_info'
 require 'temporal/workflow/id_reuse_policy'
@@ -275,8 +277,7 @@ def process_workflow_result_from(response, follow_runs)
         end
 
         event = events.first
-        # TODO: Use special error type for internal errors
-        raise Temporal::Error, 'Missing final history event' unless event
+        raise Temporal::Error::UnexpectedResponse, 'Missing final history event' unless event
 
         case event.event_type
         when :EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED
@@ -290,23 +291,39 @@ def process_workflow_result_from(response, follow_runs)
           attributes = event.workflow_execution_failed_event_attributes
           follow(attributes&.new_execution_run_id) if follow_runs
 
-          # TODO: Use more specific error and decode failure
-          raise Temporal::Error, 'Workflow execution failed'
+          raise Temporal::Error::WorkflowFailure.new(
+            cause: converter.from_failure(attributes&.failure),
+          )
 
         when :EVENT_TYPE_WORKFLOW_EXECUTION_CANCELED
-          # TODO: Use more specific error and decode failure
-          raise Temporal::Error, 'Workflow execution cancelled'
+          attributes = event.workflow_execution_canceled_event_attributes
+
+          raise Temporal::Error::WorkflowFailure.new(
+            cause: Temporal::Error::CancelledError.new(
+              'Workflow execution cancelled',
+              details: converter.from_payloads(attributes&.details),
+            ),
+          )
 
         when :EVENT_TYPE_WORKFLOW_EXECUTION_TERMINATED
-          # TODO: Use more specific error and decode failure
-          raise Temporal::Error, 'Workflow execution terminated'
+          attributes = event.workflow_execution_terminated_event_attributes
+
+          raise Temporal::Error::WorkflowFailure.new(
+            cause: Temporal::Error::TerminatedError.new(
+              attributes&.reason || 'Workflow execution terminated',
+            ),
+          )
 
         when :EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT
           attributes = event.workflow_execution_timed_out_event_attributes
           follow(attributes&.new_execution_run_id) if follow_runs
 
-          # TODO: Use more specific error and decode failure
-          raise Temporal::Error, 'Workflow execution timed out'
+          raise Temporal::Error::WorkflowFailure.new(
+            cause: Temporal::Error::TimeoutError.new(
+              'Workflow execution timed out',
+              type: Temporal::TimeoutType::START_TO_CLOSE,
+            ),
+          )
 
         when :EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW
           attributes = event.workflow_execution_continued_as_new_event_attributes

lib/temporal/data_converter.rb

@@ -1,14 +1,20 @@
 require 'temporal/api/common/v1/message_pb'
 require 'temporal/errors'
 require 'temporal/payload_converter'
+require 'temporal/failure_converter'
 
 module Temporal
   class DataConverter
     class MissingPayload < Temporal::Error; end
 
-    def initialize(payload_converter: Temporal::PayloadConverter::DEFAULT, payload_codecs: [])
+    def initialize(
+      payload_converter: Temporal::PayloadConverter::DEFAULT,
+      payload_codecs: [],
+      failure_converter: Temporal::FailureConverter::DEFAULT
+    )
       @payload_converter = payload_converter
       @payload_codecs = payload_codecs
+      @failure_converter = failure_converter
     end
 
     def to_payloads(data)
@@ -28,6 +34,11 @@ def to_payload_map(data)
       end
     end
 
+    def to_failure(error)
+      failure = failure_converter.to_failure(error, payload_converter)
+      encode_failure(failure)
+    end
+
     def from_payloads(payloads)
       return unless payloads
 
@@ -49,11 +60,20 @@ def from_payload_map(payload_map)
       # rubocop:enable Style/MapToHash
     end
 
+    def from_failure(failure)
+      raise ArgumentError, 'missing a failure to convert from' unless failure
+
+      failure = decode_failure(failure)
+      failure_converter.from_failure(failure, payload_converter)
+    end
+
     private
 
-    attr_reader :payload_converter, :payload_codecs
+    attr_reader :payload_converter, :payload_codecs, :failure_converter
 
     def encode(payloads)
+      return [] unless payloads
+
       payload_codecs.each do |codec|
         payloads = codec.encode(payloads)
       end
@@ -62,6 +82,8 @@ def encode(payloads)
     end
 
     def decode(payloads)
+      return [] unless payloads
+
       payload_codecs.reverse_each do |codec|
         payloads = codec.decode(payloads)
       end
@@ -76,5 +98,59 @@ def to_payload(data)
     def from_payload(payload)
       payload_converter.from_payload(payload)
     end
+
+    def encode_failure(failure)
+      failure = failure.dup
+
+      failure.encoded_attributes = failure.encoded_attributes ? encode([failure.encoded_attributes])&.first : nil
+      failure.cause = failure.cause ? encode_failure(failure.cause) : nil
+
+      if failure.application_failure_info
+        failure.application_failure_info.details = Temporal::Api::Common::V1::Payloads.new(
+          payloads: encode(failure.application_failure_info.details&.payloads).to_a,
+        )
+      elsif failure.timeout_failure_info
+        failure.timeout_failure_info.last_heartbeat_details = Temporal::Api::Common::V1::Payloads.new(
+          payloads: encode(failure.timeout_failure_info.last_heartbeat_details&.payloads).to_a,
+        )
+      elsif failure.canceled_failure_info
+        failure.canceled_failure_info.details = Temporal::Api::Common::V1::Payloads.new(
+          payloads: encode(failure.canceled_failure_info.details&.payloads).to_a,
+        )
+      elsif failure.reset_workflow_failure_info
+        failure.reset_workflow_failure_info.last_heartbeat_details = Temporal::Api::Common::V1::Payloads.new(
+          payloads: encode(failure.reset_workflow_failure_info.last_heartbeat_details&.payloads).to_a,
+        )
+      end
+
+      failure
+    end
+
+    def decode_failure(failure)
+      failure = failure.dup
+
+      failure.encoded_attributes = failure.encoded_attributes ? decode([failure.encoded_attributes])&.first : nil
+      failure.cause = failure.cause ? decode_failure(failure.cause) : nil
+
+      if failure.application_failure_info
+        failure.application_failure_info.details = Temporal::Api::Common::V1::Payloads.new(
+          payloads: decode(failure.application_failure_info.details&.payloads).to_a,
+        )
+      elsif failure.timeout_failure_info
+        failure.timeout_failure_info.last_heartbeat_details = Temporal::Api::Common::V1::Payloads.new(
+          payloads: decode(failure.timeout_failure_info.last_heartbeat_details&.payloads).to_a,
+        )
+      elsif failure.canceled_failure_info
+        failure.canceled_failure_info.details = Temporal::Api::Common::V1::Payloads.new(
+          payloads: decode(failure.canceled_failure_info.details&.payloads).to_a,
+        )
+      elsif failure.reset_workflow_failure_info
+        failure.reset_workflow_failure_info.last_heartbeat_details = Temporal::Api::Common::V1::Payloads.new(
+          payloads: decode(failure.reset_workflow_failure_info.last_heartbeat_details&.payloads).to_a,
+        )
+      end
+
+      failure
+    end
   end
 end

lib/temporal/error/failure.rb

@@ -0,0 +1,147 @@
+# TODO: Figure out the hierarchy
+require 'temporal/errors'
+
+module Temporal
+  class Error
+    class Failure < Error
+      attr_reader :raw
+
+      def initialize(message, raw: nil, cause: nil)
+        super(message)
+
+        @raw = raw
+        @cause = cause
+      end
+
+      def cause
+        @cause || super
+      end
+    end
+
+    class ApplicationError < Failure
+      attr_reader :type, :details, :non_retryable
+
+      def initialize(message, type:, details: [], non_retryable: false, raw: nil, cause: nil)
+        super(message, raw: raw, cause: cause)
+
+        @type = type
+        @details = details
+        @non_retryable = non_retryable
+      end
+
+      def retryable?
+        !non_retryable
+      end
+    end
+
+    class TimeoutError < Failure
+      attr_reader :type, :last_heartbeat_details
+
+      def initialize(message, type:, last_heartbeat_details: [], raw: nil, cause: nil)
+        super(message, raw: raw, cause: cause)
+
+        @type = type
+        @last_heartbeat_details = last_heartbeat_details
+      end
+    end
+
+    class CancelledError < Failure
+      attr_reader :details
+
+      def initialize(message, details: [], raw: nil, cause: nil)
+        super(message, raw: raw, cause: cause)
+
+        @details = details
+      end
+    end
+
+    class TerminatedError < Failure; end
+
+    class ServerError < Failure
+      attr_reader :non_retryable
+
+      def initialize(message, non_retryable:, raw: nil, cause: nil)
+        super(message, raw: raw, cause: cause)
+
+        @non_retryable = non_retryable
+      end
+
+      def retryable?
+        !non_retryable
+      end
+    end
+
+    class ResetWorkflowError < Failure
+      attr_reader :last_heartbeat_details
+
+      def initialize(message, last_heartbeat_details: [], raw: nil, cause: nil)
+        super(message, raw: raw, cause: cause)
+
+        @last_heartbeat_details = last_heartbeat_details
+      end
+    end
+
+    class ActivityError < Failure
+      attr_reader :scheduled_event_id,
+                  :started_event_id,
+                  :identity,
+                  :activity_name,
+                  :activity_id,
+                  :retry_state
+
+      def initialize(
+        message,
+        scheduled_event_id:,
+        started_event_id:,
+        identity:,
+        activity_name:,
+        activity_id:,
+        retry_state:,
+        raw: nil,
+        cause: nil
+      )
+        super(message, raw: raw, cause: cause)
+
+        @scheduled_event_id = scheduled_event_id
+        @started_event_id = started_event_id
+        @identity = identity
+        @activity_name = activity_name
+        @activity_id = activity_id
+        @retry_state = retry_state
+      end
+    end
+
+    class ChildWorkflowError < Failure
+      attr_reader :namespace,
+                  :workflow_id,
+                  :run_id,
+                  :workflow_name,
+                  :initiated_event_id,
+                  :started_event_id,
+                  :retry_state
+
+      def initialize(
+        message,
+        namespace:,
+        workflow_id:,
+        run_id:,
+        workflow_name:,
+        initiated_event_id:,
+        started_event_id:,
+        retry_state:,
+        raw: nil,
+        cause: nil
+      )
+        super(message, raw: raw, cause: cause)
+
+        @namespace = namespace
+        @workflow_id = workflow_id
+        @run_id = run_id
+        @workflow_name = workflow_name
+        @initiated_event_id = initiated_event_id
+        @started_event_id = started_event_id
+        @retry_state = retry_state
+      end
+    end
+  end
+end

lib/temporal/error/workflow_failure.rb

@@ -0,0 +1,17 @@
+require 'temporal/errors'
+
+module Temporal
+  class Error
+    # Used as a wrapper to perserve failure hierarchy in nested calls
+    # i.e. WorkflowFailure(ActivityError(WorkflowFailure(CancelledError)))
+    class WorkflowFailure < Error
+      attr_reader :cause
+
+      def initialize(cause:)
+        super
+
+        @cause = cause
+      end
+    end
+  end
+end

lib/temporal/errors.rb

@@ -1,4 +1,8 @@
 module Temporal
   # Superclass for all Temporal errors
-  class Error < StandardError; end
+  class Error < StandardError
+    # Superclass for RPC and proto related errors
+    class RPCError < Error; end
+    class UnexpectedResponse < RPCError; end
+  end
 end