Converted WorkflowExecutionAlreadyStartedFailure handling to use RPC status

lib/temporal/connection/grpc.rb

@@ -8,6 +8,7 @@
 require 'temporal/connection/errors'
 require 'temporal/connection/serializer'
 require 'temporal/connection/serializer/failure'
+require 'gen/temporal/api/errordetails/v1/message_pb'
 require 'temporal/concerns/payloads'
 
 module Temporal
@@ -125,9 +126,14 @@ def start_workflow_execution(
 
         client.start_workflow_execution(request)
       rescue ::GRPC::AlreadyExists => e
-        # Feel like there should be cleaner way to do this...
-        run_id = e.details[/RunId: ([a-f0-9]+-[a-f0-9]+-[a-f0-9]+-[a-f0-9]+-[a-f0-9]+)/, 1]
-        raise Temporal::WorkflowExecutionAlreadyStartedFailure.new(e.details, run_id)
+        error_details_from(e).each do |error|
+          case error
+          when Temporal::Api::ErrorDetails::V1::WorkflowExecutionAlreadyStartedFailure
+            raise Temporal::WorkflowExecutionAlreadyStartedFailure.new(e.details, error.run_id)
+          end
+        end
+
+        raise Temporal::ApiError, e.details # unhandled error type
       end
 
       SERVER_MAX_GET_WORKFLOW_EXECUTION_HISTORY_POLL = 30
@@ -141,7 +147,7 @@ def get_workflow_execution_history(
         event_type: :all,
         timeout: nil
       )
-        if wait_for_new_event 
+        if wait_for_new_event
           if timeout.nil?
             # This is an internal error.  Wrappers should enforce this.
             raise "You must specify a timeout when wait_for_new_event = true."
@@ -369,7 +375,17 @@ def signal_with_start_workflow_execution(
           request.workflow_id_reuse_policy = policy
         end
 
+
         client.signal_with_start_workflow_execution(request)
+      rescue ::GRPC::AlreadyExists => e
+        error_details_from(e).each do |error|
+          case error
+          when Temporal::Api::ErrorDetails::V1::WorkflowExecutionAlreadyStartedFailure
+            raise Temporal::WorkflowExecutionAlreadyStartedFailure.new(e.details, error.run_id)
+          end
+        end
+
+        raise Temporal::ApiError, e.details # unhandled error type
       end
 
       def reset_workflow_execution(namespace:, workflow_id:, run_id:, reason:, workflow_task_event_id:)
@@ -533,9 +549,18 @@ def serialize_status_filter(value)
 
         sym = Temporal::Workflow::Status::API_STATUS_MAP.invert[value]
         status = Temporal::Api::Enums::V1::WorkflowExecutionStatus.resolve(sym)
-        
+
         Temporal::Api::Filter::V1::StatusFilter.new(status: status)
       end
+
+      def error_details_from(error)
+        error.to_rpc_status&.details&.map do |any_error|
+          type = Google::Protobuf::DescriptorPool.generated_pool.lookup any_error.type_url.split('/').last
+          next if type.nil?
+
+          any_error.unpack type.msgclass
+        end&.compact
+      end
     end
   end
 end

spec/spec_helper.rb

@@ -16,6 +16,7 @@
 
 require 'temporal'
 require 'google/protobuf/well_known_types'
+require 'google/rpc/status_pb'
 require 'pry'
 
 Dir[File.expand_path('config/*.rb', __dir__)].sort.each { |f| require f }

spec/unit/lib/temporal/grpc_client_spec.rb

@@ -4,20 +4,40 @@
   let(:namespace) { 'test-namespace' }
   let(:workflow_id) { SecureRandom.uuid }
   let(:run_id) { SecureRandom.uuid }
-  let(:now) { Time.now}
+  let(:now) { Time.now.utc }
+  let(:already_started_error) do
+    detail_error = Google::Protobuf::Any.new.tap do |any|
+      any.pack(Temporal::Api::ErrorDetails::V1::WorkflowExecutionAlreadyStartedFailure.new(start_request_id: SecureRandom.uuid, run_id: run_id))
+    end
+    rpc_status = Google::Rpc::Status.new(
+      code: 6,
+      message: 'Workflow execution already finished successfully. WorkflowId: TestWorkflow-1, RunId: baaf1d86-4459-4ecd-a288-47aeae55245d. Workflow Id reuse policy: allow duplicate workflow Id if last run failed.',
+      details: [detail_error],
+    )
+    GRPC::AlreadyExists.new('details', { 'grpc-status-details-bin' => Google::Rpc::Status.encode(rpc_status) })
+  end
+
+  let(:invalid_already_exists_error) do
+    detail_error = Google::Protobuf::Any.new.tap do |any|
+      any.pack(Temporal::Api::ErrorDetails::V1::NotFoundFailure.new())
+    end
+    rpc_status = Google::Rpc::Status.new(
+      code: 6,
+      message: 'some error message',
+      details: [detail_error],
+    )
+    GRPC::AlreadyExists.new('details', { 'grpc-status-details-bin' => Google::Rpc::Status.encode(rpc_status) })
+  end
 
   before do
     allow(subject).to receive(:client).and_return(grpc_stub)
-    
+
     allow(Time).to receive(:now).and_return(now)
   end
 
   describe '#start_workflow_execution' do
     it 'provides the existing run_id when the workflow is already started' do
-      allow(grpc_stub).to receive(:start_workflow_execution).and_raise(
-        GRPC::AlreadyExists,
-        'Workflow execution already finished successfully. WorkflowId: TestWorkflow-1, RunId: baaf1d86-4459-4ecd-a288-47aeae55245d. Workflow Id reuse policy: allow duplicate workflow Id if last run failed.'
-      )
+      allow(grpc_stub).to receive(:start_workflow_execution).and_raise(already_started_error)
 
       expect do
         subject.start_workflow_execution(
@@ -31,19 +51,35 @@
           memo: {}
         )
       end.to raise_error(Temporal::WorkflowExecutionAlreadyStartedFailure) do |e|
-        expect(e.run_id).to eql('baaf1d86-4459-4ecd-a288-47aeae55245d')
+        expect(e.run_id).to eql(run_id)
       end
     end
+
+    it 'handles unexpected already exists error' do
+      allow(grpc_stub).to receive(:start_workflow_execution).and_raise(invalid_already_exists_error)
+
+      expect do
+        subject.start_workflow_execution(
+          namespace: namespace,
+          workflow_id: workflow_id,
+          workflow_name: 'Test',
+          task_queue: 'test',
+          execution_timeout: 0,
+          run_timeout: 0,
+          task_timeout: 0,
+          memo: {}
+        )
+      end.to raise_error(Temporal::ApiError)
+    end
   end
-  
+
   describe '#signal_with_start_workflow' do
     let(:temporal_response) do
       Temporal::Api::WorkflowService::V1::SignalWithStartWorkflowExecutionResponse.new(run_id: 'xxx')
     end
 
-    before { allow(grpc_stub).to receive(:signal_with_start_workflow_execution).and_return(temporal_response) }
-
     it 'starts a workflow with a signal with scalar arguments' do
+      allow(grpc_stub).to receive(:signal_with_start_workflow_execution).and_return(temporal_response)
       subject.signal_with_start_workflow_execution(
         namespace: namespace,
         workflow_id: workflow_id,
@@ -71,6 +107,46 @@
         expect(request.signal_input.payloads[0].data).to eq('"what do you get if you multiply six by nine?"')
       end
     end
+
+    it 'provides the existing run_id when the workflow is already started' do
+      allow(grpc_stub).to receive(:signal_with_start_workflow_execution).and_raise(already_started_error)
+
+      expect do
+        subject.signal_with_start_workflow_execution(
+          namespace: namespace,
+          workflow_id: workflow_id,
+          workflow_name: 'workflow_name',
+          task_queue: 'task_queue',
+          input: ['foo'],
+          execution_timeout: 1,
+          run_timeout: 2,
+          task_timeout: 3,
+          signal_name: 'the question',
+          signal_input: 'what do you get if you multiply six by nine?',
+        )
+      end.to raise_error(Temporal::WorkflowExecutionAlreadyStartedFailure) do |e|
+        expect(e.run_id).to eql(run_id)
+      end
+    end
+
+    it 'handles invalid already exists errors' do
+      allow(grpc_stub).to receive(:signal_with_start_workflow_execution).and_raise(invalid_already_exists_error)
+
+      expect do
+        subject.signal_with_start_workflow_execution(
+          namespace: namespace,
+          workflow_id: workflow_id,
+          workflow_name: 'workflow_name',
+          task_queue: 'task_queue',
+          input: ['foo'],
+          execution_timeout: 1,
+          run_timeout: 2,
+          task_timeout: 3,
+          signal_name: 'the question',
+          signal_input: 'what do you get if you multiply six by nine?',
+        )
+      end.to raise_error(Temporal::ApiError)
+    end
   end
 
   describe "#list_namespaces" do
@@ -148,7 +224,7 @@
         end
       end
 
-      it 'demands a timeout to be specified' do 
+      it 'demands a timeout to be specified' do
         expect do
           subject.get_workflow_execution_history(
             namespace: namespace,
@@ -161,7 +237,7 @@
         end
       end
 
-      it 'disallows a timeout larger than the server timeout' do 
+      it 'disallows a timeout larger than the server timeout' do
         expect do
           subject.get_workflow_execution_history(
             namespace: namespace,
@@ -261,7 +337,7 @@
         end
       end
     end
-    
+
     describe '#list_closed_workflow_executions' do
       let(:namespace) { 'test-namespace' }
       let(:from) { Time.now - 600 }