Client improvements (#69)

antstorm · web-flow · commit 40055d3e787c · 2022-11-18T16:26:17.000Z
* Replace generic errors with specific ones in Client::Implementation

* Serialize RetryPolicy for #start_workflow

* Add missing types for Thermite::Config and Rutie

* Require schema-less URLs when initializing a Connection

* Raise specific error when workflow is already started

* Switch UnsupportedQuery to QueryFailed error
diff --git a/Steepfile b/Steepfile
@@ -6,7 +6,7 @@ target :lib do
   check 'lib'
 
   repo_path 'vendor/rbs/gem_rbs_collection/gems'
-  library 'protobuf', 'json', 'securerandom', 'socket'
+  library 'protobuf', 'json', 'securerandom', 'socket', 'uri'
 
   ignore 'lib/gen/*.rb'
 
diff --git a/lib/temporal/client/implementation.rb b/lib/temporal/client/implementation.rb
@@ -119,7 +119,7 @@ def handle_start_workflow(input)
           workflow_run_timeout: input.run_timeout,
           workflow_task_timeout: input.task_timeout,
           workflow_id_reuse_policy: Workflow::IDReusePolicy.to_raw(input.id_reuse_policy),
-          retry_policy: nil, # TODO: serialize retry policy
+          retry_policy: input.retry_policy&.to_proto,
           cron_schedule: input.cron_schedule,
           memo: memo,
           search_attributes: search_attributes,
@@ -154,8 +154,7 @@ def handle_start_workflow(input)
       rescue Temporal::Bridge::Error => e
         # TODO: Raise a better error from the bridge
         if e.message.include?('AlreadyExists')
-          # TODO: Replace with a more specific error
-          raise Temporal::Error, 'Workflow already exists'
+          raise Temporal::Error::WorkflowExecutionAlreadyStarted, 'Workflow execution already started'
         else
           raise # re-raise
         end
@@ -196,17 +195,13 @@ def handle_query_workflow(input)
 
         if response.query_rejected
           status = Workflow::ExecutionStatus.from_raw(response.query_rejected.status)
-          # TODO: Replace with a specific error when implemented
-          raise Temporal::Error, "Query rejected, workflow status: #{status}"
+          raise Temporal::Error::QueryRejected, status
         end
 
         converter.from_payloads(response.query_result)&.first
       rescue Temporal::Bridge::Error => e
         # TODO: Raise a better error from the bridge
-        if e.message.include?('unknown queryType')
-          # TODO: Replace with a more specific error
-          raise Temporal::Error, 'Unsupported query'
-        end
+        raise Temporal::Error::QueryFailed, e.message
       end
 
       def handle_signal_workflow(input)
diff --git a/lib/temporal/connection.rb b/lib/temporal/connection.rb
@@ -1,6 +1,7 @@
 require 'temporal/api/workflowservice/v1/request_response_pb'
 require 'temporal/bridge'
 require 'temporal/runtime'
+require 'uri'
 
 module Temporal
   # A connection to the Temporal server.
@@ -13,8 +14,9 @@ class Connection
     # @param host [String] `host:port` for the Temporal server. For local development, this is
     #   often `"localhost:7233"`.
     def initialize(host)
+      url = parse_url(host)
       runtime = Temporal::Runtime.instance
-      @core_connection = Temporal::Bridge::Connection.connect(runtime.core_runtime, host)
+      @core_connection = Temporal::Bridge::Connection.connect(runtime.core_runtime, url)
     end
 
     # @param request [Temporal::Api::WorkflowService::V1::RegisterNamespaceRequest]
@@ -718,5 +720,17 @@ def list_batch_operations(request, metadata: {}, timeout: nil)
 
       Temporal::Api::WorkflowService::V1::ListBatchOperationsResponse.decode(response)
     end
+
+    private
+
+    def parse_url(url)
+      # Turn this into a valid URI before parsing
+      uri = URI.parse(url.include?('://') ? url : "//#{url}")
+      raise Temporal::Error, 'Target host as URL with scheme are not supported' if uri.scheme
+
+      # TODO: Add support for mTLS
+      uri.scheme = 'http'
+      uri.to_s
+    end
   end
 end
diff --git a/lib/temporal/errors.rb b/lib/temporal/errors.rb
@@ -3,6 +3,20 @@ module Temporal
   class Error < StandardError
     # Superclass for RPC and proto related errors
     class RPCError < Error; end
+
     class UnexpectedResponse < RPCError; end
+
+    class WorkflowExecutionAlreadyStarted < RPCError; end
+
+    class QueryFailed < RPCError; end
+
+    class QueryRejected < RPCError
+      attr_reader :status
+
+      def initialize(status)
+        super("Query rejected, workflow status: #{status}")
+        @status = status
+      end
+    end
   end
 end
diff --git a/lib/temporal/retry_policy.rb b/lib/temporal/retry_policy.rb
@@ -1,3 +1,4 @@
+require 'temporal/api/common/v1/message_pb'
 require 'temporal/errors'
 
 module Temporal
@@ -67,5 +68,15 @@ def validate!
         raise Invalid, 'Maximum interval cannot be less than initial interval' if max_interval < initial_interval
       end
     end
+
+    def to_proto
+      Temporal::Api::Common::V1::RetryPolicy.new(
+        initial_interval: Google::Protobuf::Duration.new(seconds: initial_interval),
+        backoff_coefficient: backoff,
+        maximum_interval: max_interval ? Google::Protobuf::Duration.new(seconds: max_interval) : nil,
+        maximum_attempts: max_attempts,
+        non_retryable_error_types: non_retriable_errors.map(&:name).compact,
+      )
+    end
   end
 end
diff --git a/sig/temporal/bridge.rbs b/sig/temporal/bridge.rbs
@@ -1,7 +1,12 @@
 BRIDGE_DIR: String
 
 class Rutie
-  def initialize: (Symbol, release: String | Symbol) -> void
+  def initialize: (
+    Symbol bridge,
+    ?lib_path: String | Symbol,
+    ?lib_suffix: String | Symbol,
+    ?lib_prefix: String | Symbol
+  ) -> void
   def init: (String, String) -> void
 end
 
diff --git a/sig/temporal/connection.rbs b/sig/temporal/connection.rbs
@@ -43,5 +43,9 @@ module Temporal
     def get_cluster_info: (Temporal::Api::WorkflowService::V1::GetClusterInfoRequest request, ?metadata: Hash[String, String], ?timeout: Integer?) -> Temporal::Api::WorkflowService::V1::GetClusterInfoResponse
     def get_system_info: (Temporal::Api::WorkflowService::V1::GetSystemInfoRequest request, ?metadata: Hash[String, String], ?timeout: Integer?) -> Temporal::Api::WorkflowService::V1::GetSystemInfoResponse
     def list_task_queue_partitions: (Temporal::Api::WorkflowService::V1::ListTaskQueuePartitionsRequest request, ?metadata: Hash[String, String], ?timeout: Integer?) -> Temporal::Api::WorkflowService::V1::ListTaskQueuePartitionsResponse
+
+    private
+
+    def parse_url: (String) -> String
   end
 end
diff --git a/sig/temporal/errors.rbs b/sig/temporal/errors.rbs
@@ -5,5 +5,17 @@ module Temporal
 
     class UnexpectedResponse < RPCError
     end
+
+    class WorkflowExecutionAlreadyStarted < RPCError
+    end
+
+    class QueryFailed < RPCError
+    end
+
+    class QueryRejected < RPCError
+      attr_reader status: Temporal::Workflow::ExecutionStatus::values
+
+      def initialize: (Temporal::Workflow::ExecutionStatus::values) -> void
+    end
   end
 end
diff --git a/sig/temporal/retry_policy.rbs b/sig/temporal/retry_policy.rbs
@@ -1,24 +1,23 @@
 module Temporal
-  class Error < StandardError
-  end
-
   class RetryPolicy
     Invalid: Temporal::Error
 
     attr_reader initial_interval: Integer
     attr_reader backoff: Float
     attr_reader max_interval: Integer?
     attr_reader max_attempts: Integer
-    attr_reader non_retriable_errors: Array[StandardError]
+    attr_reader non_retriable_errors: Array[Class]
 
     def initialize: (
       ?initial_interval: Integer,
       ?backoff: Float,
       ?max_interval: nil,
       ?max_attempts: Integer,
-      ?non_retriable_errors: Array[StandardError]
+      ?non_retriable_errors: Array[Class]
     ) -> void
 
     def validate!: -> nil
+
+    def to_proto: -> Temporal::Api::Common::V1::RetryPolicy
   end
 end
diff --git a/sig/thermite_patch.rbs b/sig/thermite_patch.rbs
@@ -0,0 +1,13 @@
+module Thermite
+  class Config
+    def target_arch: -> String
+    def target_os: -> String
+    def cargo_target_path: (String, *String) -> String
+    def rust_toplevel_dir: -> String
+
+    private
+
+    @target_arch: String?
+    @target_os: String?
+  end
+end
diff --git a/spec/integration/client_spec.rb b/spec/integration/client_spec.rb
@@ -54,13 +54,13 @@ def terminate_workflow(input)
 
   subject { described_class.new(connection, namespace) }
 
-  let(:connection) { Temporal::Connection.new("http://#{url}") }
+  let(:connection) { Temporal::Connection.new(url) }
   let(:id) { SecureRandom.uuid }
   let(:workflow) { 'kitchen_sink' }
 
   before(:all) do
     @server_pid = fork { exec("#{support_path}/go_server/main #{port} #{namespace}") }
-    Helpers::TestRPC.wait("http://#{url}", 10, 0.5)
+    Helpers::TestRPC.wait(url, 10, 0.5)
 
     @worker_pid = fork { exec("#{support_path}/go_worker/main #{url} #{namespace} #{task_queue}") }
   end
@@ -96,7 +96,10 @@ def terminate_workflow(input)
           task_queue: task_queue,
           id_reuse_policy: Temporal::Workflow::IDReusePolicy::REJECT_DUPLICATE,
         )
-      end.to raise_error(Temporal::Error, 'Workflow already exists')
+      end.to raise_error(
+        Temporal::Error::WorkflowExecutionAlreadyStarted,
+        'Workflow execution already started'
+      )
 
       # Run with a default policy again expecting it to succeed
       input = { actions: [{ result: { value: 'test return value 3' } }] }
@@ -184,10 +187,10 @@ def terminate_workflow(input)
       handle = subject.start_workflow(workflow, input, id: id, task_queue: task_queue)
       handle.result
 
-      # TODO: Change this to a wrapped error later
-      expect do
-        handle.query('other test query', 'test query arg')
-      end.to raise_error(Temporal::Error, 'Unsupported query')
+      expect { handle.query('other test query', 'test query arg') }.to raise_error do |error|
+        expect(error).to be_a(Temporal::Error::QueryFailed)
+        expect(error.message).to include('unknown queryType other test query')
+      end
     end
   end
 
diff --git a/spec/integration/connection_spec.rb b/spec/integration/connection_spec.rb
@@ -6,13 +6,13 @@
 describe Temporal::Connection do
   mock_address = '0.0.0.0:4444'.freeze
 
-  subject { described_class.new("http://#{mock_address}") }
+  subject { described_class.new(mock_address) }
 
   # TODO: For some reason the Bridge doesn't play well with the server in the same
   #       process throwing SegFaults in cases. Needs further investigation
   before(:all) do
     @pid = fork { exec('bundle exec ruby spec/support/mock_server.rb') }
-    Helpers::TestRPC.wait("http://#{mock_address}", 10)
+    Helpers::TestRPC.wait(mock_address, 10)
   end
   after(:all) { Process.kill('QUIT', @pid) }
 
@@ -49,7 +49,13 @@
     end
 
     it 'raises when unable to connect' do
-      expect { described_class.new('http://0.0.0.0:3333') }.to raise_error(Temporal::Bridge::Error)
+      expect { described_class.new('0.0.0.0:3333') }.to raise_error(Temporal::Bridge::Error)
+    end
+
+    it 'raises when given a URL with schema' do
+      expect do
+        described_class.new('http://localhost:3333')
+      end.to raise_error(Temporal::Error, 'Target host as URL with scheme are not supported')
     end
 
     it 'raises when incorrect request was provided' do
diff --git a/spec/unit/temporal/client/implementation_spec.rb b/spec/unit/temporal/client/implementation_spec.rb
@@ -209,7 +209,7 @@
       it 'raises Temporal::Error' do
         expect do
           subject.start_workflow(input)
-        end.to raise_error(Temporal::Error, 'Workflow already exists')
+        end.to raise_error(Temporal::Error::WorkflowExecutionAlreadyStarted, 'Workflow execution already started')
       end
     end
 
@@ -360,6 +360,26 @@
       expect(subject.query_workflow(input)).to eq('test')
     end
 
+    context 'when query is rejected' do
+      before do
+        allow(connection)
+          .to receive(:query_workflow)
+          .and_return(
+            Temporal::Api::WorkflowService::V1::QueryWorkflowResponse.new(
+              query_rejected: { status: :WORKFLOW_EXECUTION_STATUS_COMPLETED },
+            )
+          )
+      end
+
+      it 'raises an error' do
+        expect { subject.query_workflow(input) }.to raise_error do |error|
+          expect(error).to be_a(Temporal::Error::QueryRejected)
+          expect(error.message).to eq('Query rejected, workflow status: COMPLETED')
+          expect(error.status).to eq(Temporal::Workflow::ExecutionStatus::COMPLETED)
+        end
+      end
+    end
+
     context 'with header' do
       before { input.headers = { 'header' => 'test' } }
 
diff --git a/spec/unit/temporal/retry_policy_spec.rb b/spec/unit/temporal/retry_policy_spec.rb
@@ -1,19 +1,21 @@
 require 'temporal/retry_policy'
 
 describe Temporal::RetryPolicy do
+  subject { described_class.new(**valid_attributes) }
+
+  let(:valid_attributes) do
+    {
+      initial_interval: 1,
+      backoff: 1.5,
+      max_interval: 5,
+      max_attempts: 3,
+      non_retriable_errors: [StandardError],
+    }
+  end
+
   describe '#validate!' do
     subject { described_class.new(**attributes) }
 
-    let(:valid_attributes) do
-      {
-        initial_interval: 1,
-        backoff: 1.5,
-        max_interval: 5,
-        max_attempts: 3,
-        non_retriable_errors: [StandardError],
-      }
-    end
-
     let(:unlimited_attempts) do
       {
         initial_interval: 1,
@@ -118,4 +120,17 @@
       end
     end
   end
+
+  describe '#to_proto' do
+    it 'serializes itself into proto' do
+      proto = subject.to_proto
+
+      expect(proto).to be_a(Temporal::Api::Common::V1::RetryPolicy)
+      expect(proto.initial_interval.seconds).to eq(subject.initial_interval)
+      expect(proto.backoff_coefficient).to eq(subject.backoff)
+      expect(proto.maximum_interval.seconds).to eq(subject.max_interval)
+      expect(proto.maximum_attempts).to eq(subject.max_attempts)
+      expect(proto.non_retryable_error_types.to_a).to eq(['StandardError'])
+    end
+  end
 end
