Client improvements (#69)

* 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

Author: antstorm

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'
 

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)

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

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

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

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
 

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

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

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

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