Adding payload codec pipeline (#224)

* Revert signal_with_start (before taking a different approach)

* Add signal arguments to start_workflow (to support signal_with_start)

* Merge memo changes

* Address PR feedback

* Update method signature in temporal test fixture

* Add detail to a few error messages

* Update our FailWorkflowTask logic's call to ErrorHandler.handle

* Workflow await

* Make dispatch more generic

* Fix race condition

* Merge await into wait_for

* Update sample workflow to use wait_for, rename to WaitForWorkflow

* Reorganize and extend wait_for tests

* Check for completed futures before setting dispatcher and yielding the fiber

* Extend wait_for to take multiple futures and a condition block

* Differentiate TARGET_WILDCARD and WILDCARD, allow comparison with EventTarget objects

* Use Ruby 2.7 to be consistent with pay-server

* Turn off schedule_to_start activity timeout by default

* Refactor metadata generation

* Make task queue available on workflow metadata, add example test

* Expose workflow start time metadata

* Add memos

* Make error deserialization more resilient

* Make temporal-ruby able to deserialize larger histories

* Remove temporary test

* Expose workflow name in activity metadata in temporal-ruby's unit tester

* Add a workflow-level test

* add namespace to emitted metrics

* emit the workflow name tag during activity processing

* fix typo

* added failworkflowtaskerror

* updated register namespace to accept new params

* examples

* fixed namespace test

* empty

* updated unit tests

* removed unncessary code

* updated seconds

* Add replay flag to workflow context

* fixed nits

* Rename to replay? to history_replaying?

* updated sleep to 0.5

* updated unit tests and nits

* fixed unit tests

* added link to comment

* Merge fix

* Fix upsert_search_attributes

* added fix for nil search attributes

* added unit test

* updated unit test

* added expect to be nil

* Expose scheduled_time and current_attempt_scheduled_time on activity metadata

* Implement ParentClosePolicy for child workflows

* Add e2e test for child workflow execution

* move serialization logic farther down the stack

* Refactor serialize_parent_close_policy; add unit tests

* Expose wait_for_start for child workflow execution

* Remove future `workflow_id,run_id` annotations; simplify wait_for logic

* Add parent_run_id, parent_id to workflow metadata

* Allow opting out of child workflow futures

* Remove duplicate describe block

* Factor out workflow_id_reuse_policy serialization

* Respect workflow_id_reuse_policy for child workflows

* Add integration tests

* Use a nicer exception type

* Refactor wait_for into distinct wait_for_any and wait_for_condition methods

* Order wildcard dispatch handlers

* Remove finished handlers

* Check finished? on wait_for_any, add more unit specs

* More dispatch unit specs

* Use hash instead of list for callbacks per target

* Remove dead code, improve error messages in local workflow context

* Correct swapped arguments

* Eliminate unnecessary IDs for dispatcher handlers

* Raise on duplicate ID

* added paginated workflows

* client spec

* reset

* Downstream the rest of [activity_metadata.workflow_name](#130)

* Downstream the rest of [activity_metadata.scheduled_time](#164)

* Fix an activity_metadata related test that doesn't exist upstream

* Downstream the rest of [child workflow workflow_id_reuse_policy fixes](#172)

* Downstream [merge error fix](#180)

* added json protobuf

* Update json_protobuf.rb

* added unit test

* Remove duplicate tests in context_spec

* allow connection options to be set

* add missing comma

* add test for interceptors

* remove unused variable

* use new Temporal client in interceptor test to avoid test pollution

* add option to specify search attributes when starting workflows

* move empty check out of Temporal::Workflow::Context::Helpers.process_search_attributes

* move process_search_attributes out of ExecutionOptions.initialize

* allow default search attributes to be configured globally

* fix tests, unit test global default search attributes

* add unit test for gRPC serialization

* clean up the integration test

* Include remaining changes from upstream #188 (#98)

### Summary

This PR is a follow-up to #97 that includes the changes made to the corresponding upstream PR ([github.com/#188](#188)) after that downstream PR was eagerly merged (with a subset of the changes from upstream). The changes that were not included in the downstream PR when it was merged are: [compare](https://github.com/coinbase/temporal-ruby/pull/188/files/05b6eafeb43ac717c15ae683e6249c4de876ef3d..c6f614325695cc666e066aa218a329c9bf7504f3) (that should be identical to the changes in this PR).

With the merging of the upstream PR, I also updated the [non-upstreamed changed doc](https://paper.dropbox.com/doc/Non-upstreamed-temporal-ruby-changes--Bm39qa99fshz6nbw9RG37pgFAg-AYkwiCfkcoM66adjZMwAZ) to remove the entry on Initial workflow search attributes.

### Motivation
The intention of this PR is to synchronize upstream and downstream, with relation to the changeset applied in the upstream PR.

### Test plan
N/A

### Rollout/monitoring/revert plan
Safe to revert.

* Remove dead code from previous messy merge

* Separate wait_until handlers, execute at end

* Modify signal_with_start_workflow to cover dwillett's repro

* Disable running rubyfmt on save

* Consolidate metrics constants

* Add workflow task failure counter

* Use metric_keys.rb for filename

* Allow client identity to be configurable

* Use PID instead of thread ID in default identity

* Add poller completed metrics

* Gauge metrics for queue size and available threads per thread pool

* Add task queue and namespace to thread pool metrics

* Fix tag

* reverted proto_json changes

* fixed import

* remmoved test file

* Revert "Merge pull request #108 from stripe-private-oss-forks/calum/fixing-import"

This reverts commit 48cbe20, reversing
changes made to e805a38.

* Revert "Merge pull request #107 from stripe-private-oss-forks/calum/reverting-json-proto-changes"

This reverts commit e805a38, reversing
changes made to 3deda64.

* Exempt the necessary system workflows from normal JSON proto deserialization

* Emit task queue for workflow task failures

* Added task queue tags to workflow and activity task queue time metrics

* DynamicActivity

* Use const_get

* Cleanup error message

* Jeff feedback

* do not fail workflow task if completing it errors

It's possible for a workflow task to fail to complete even during normal
operation. For example, if a signal is added to the workflow history
concurrently with a workflow task attempting to complete the workflow
then the `RespondWorkflowTaskCompleted` call will recieve an
`InvalidArgument` error with an `UnhandledCommand` cause.

If this happens then the Ruby SDK would get this error and then try and
fail the workflow task due to how the `rescue` block encompassed the
completion function. This would then lead to the
`RespondWorkflowTaskFailed` call failing because the server doesn't
recognize the task anymore. This is because the task was actually
finalized when the original `RespondWorkflowTaskCompleted` call was made
and so it should not be interacted with anymore.

* add a comment

* use more realistic error type in tests

* Remove orphaned code post-merge

* Allow empty pages when paginating through history

* Remove unused error

* Allow ActivityException to accept args

* Improve deserialization code flow

* Use the default converter to serialize errors when configured to do so

* Get tests working

* Cleanup

* fix nit

* Fix extraneous debug spew

* added max page size param

* Show bad data on Activity error serialization failure

* Upgrade Temporal proto API to version 1.16

* Rename Temporal -> Temporalio

* Remove deprecated namespace field for activity task scheduling

* Methods for operating on custom search attributes

* Example test for custom search attribute APIs

* Unit tests

* Treat missing history submessage on GetWorkflowExecutionHistoryResponse as timeout

* Remove unnecessary &.

* Dynamic Workflows

* Add terminate-if-running workflow id reuse policy mappings

* Add integration test for terminate-if-running

* Move misplaced test

* Feedback

* Use terminate-if-running as policy for both invocations

* added payload codecs

* added search attribute payload methods

* fixed tests

* added codec tests

* added chain tests

* updated nits

* Revert "Merge branch 'master' into calum/adding-payload-codecs"

This reverts commit 145290c, reversing
changes made to d8a42d8.

* fixed tests

* fixed

* examples/lib/crypt_payload_codec.rb

---------

Co-authored-by: Nathan Glass <[email protected]>
Co-authored-by: Jeff Schoner <[email protected]>
Co-authored-by: Andrew Hoskins <[email protected]>
Co-authored-by: Christopher Brown <[email protected]>
Co-authored-by: Arya Kumar <[email protected]>
Co-authored-by: Joseph Azevedo <[email protected]>
Co-authored-by: Jeff Schoner <[email protected]>
Co-authored-by: Bryton Shoffner <[email protected]>

Author: 9 people

examples/bin/worker

@@ -1,6 +1,6 @@
 #!/usr/bin/env ruby
 require_relative '../init'
-require_relative '../lib/cryptconverter'
+require_relative '../lib/crypt_payload_codec'
 
 require 'temporal/worker'
 
@@ -11,8 +11,10 @@ Dir[File.expand_path('../middleware/*.rb', __dir__)].each { |f| require f }
 if !ENV['USE_ENCRYPTION'].nil?
   Temporal.configure do |config|
     config.task_queue = 'crypt'
-    config.converter = Temporal::CryptConverter.new(
-      payload_converter: Temporal::Configuration::DEFAULT_CONVERTER
+    config.payload_codec = Temporal::Connection::Converter::Codec::Chain.new(
+      payload_codecs: [
+        Temporal::CryptPayloadCodec.new
+      ]
     )
   end
 end

examples/lib/cryptconverter.rb renamed to examples/lib/crypt_payload_codec.rb

@@ -1,7 +1,8 @@
 require 'openssl'
+require 'temporal/connection/converter/codec/base'
 
 module Temporal
-  class CryptConverter < Temporal::Connection::Converter::Base
+  class CryptPayloadCodec < Temporal::Connection::Converter::Codec::Base
     CIPHER = 'aes-256-gcm'.freeze
     GCM_NONCE_SIZE = 12
     GCM_TAG_SIZE = 16
@@ -10,26 +11,23 @@ class CryptConverter < Temporal::Connection::Converter::Base
     METADATA_ENCODING_KEY = 'encoding'.freeze
     METADATA_ENCODING = 'binary/encrypted'.freeze
 
-    def to_payloads(data)
+    def encode(payload)
+      return nil if payload.nil?
+
       key_id = get_key_id
       key = get_key(key_id)
 
-      payloads = super(data)
-
-      Temporalio::Api::Common::V1::Payloads.new(
-        payloads: payloads.payloads.map { |payload| encrypt_payload(payload, key_id, key) }
-      )
+      encrypt_payload(payload, key_id, key)
     end
+    
+    def decode(payload)
+      return nil if payload.nil?
 
-    def from_payloads(payloads)
-      return nil if payloads.nil?
-
-      payloads.payloads.map do |payload|
-        if payload.metadata[METADATA_ENCODING_KEY] == METADATA_ENCODING
-          payload = decrypt_payload(payload)
-        end
-        from_payload(payload)
+      if payload.metadata[METADATA_ENCODING_KEY] == METADATA_ENCODING
+        payload = decrypt_payload(payload)
       end
+
+      payload
     end
 
     private

examples/spec/integration/converter_spec.rb

@@ -1,5 +1,5 @@
 require 'workflows/hello_world_workflow'
-require 'lib/cryptconverter'
+require 'lib/crypt_payload_codec'
 require 'grpc/errors'
 
 describe 'Converter', :integration do
@@ -8,8 +8,10 @@
 
     Temporal.configure do |config|
       config.task_queue = 'crypt'
-      config.converter = Temporal::CryptConverter.new(
-        payload_converter: Temporal::Configuration::DEFAULT_CONVERTER
+      config.payload_codec = Temporal::Connection::Converter::Codec::Chain.new(
+        payload_codecs: [
+          Temporal::CryptPayloadCodec.new
+        ]
       )
     end
 
@@ -65,8 +67,8 @@
     completion_event = events[:EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED].first
     result = completion_event.workflow_execution_completed_event_attributes.result
 
-    converter = Temporal.configuration.converter
+    payload_codec = Temporal.configuration.payload_codec
 
-    expect(converter.from_payloads(result)&.first).to eq('Hello World, Tom')
+    expect(payload_codec.decodes(result).payloads.first.data).to eq('"Hello World, Tom"')
   end
 end

lib/temporal/concerns/payloads.rb

@@ -2,13 +2,19 @@ module Temporal
   module Concerns
     module Payloads
       def from_payloads(payloads)
+        payloads = payload_codec.decodes(payloads)
         payload_converter.from_payloads(payloads)
       end
 
       def from_payload(payload)
+        payload = payload_codec.decode(payload)
         payload_converter.from_payload(payload)
       end
 
+      def from_payload_map_without_codec(payload_map)
+        payload_map.map { |key, value| [key, payload_converter.from_payload(value)] }.to_h
+      end
+
       def from_result_payloads(payloads)
         from_payloads(payloads)&.first
       end
@@ -30,11 +36,20 @@ def from_payload_map(payload_map)
       end
 
       def to_payloads(data)
-        payload_converter.to_payloads(data)
+        payloads = payload_converter.to_payloads(data)
+        payload_codec.encodes(payloads)
       end
 
       def to_payload(data)
-        payload_converter.to_payload(data)
+        payload = payload_converter.to_payload(data)
+        payload_codec.encode(payload)
+      end
+
+      def to_payload_map_without_codec(data)
+        # skips the payload_codec step because search attributes don't use this pipeline
+        data.transform_values do |value|
+          payload_converter.to_payload(value)
+        end
       end
 
       def to_result_payloads(data)
@@ -62,6 +77,10 @@ def to_payload_map(data)
       def payload_converter
         Temporal.configuration.converter
       end
+
+      def payload_codec
+        Temporal.configuration.payload_codec
+      end
     end
   end
 end

lib/temporal/configuration.rb

@@ -7,14 +7,15 @@
 require 'temporal/connection/converter/payload/json'
 require 'temporal/connection/converter/payload/proto_json'
 require 'temporal/connection/converter/composite'
+require 'temporal/connection/converter/codec/chain'
 
 module Temporal
   class Configuration
     Connection = Struct.new(:type, :host, :port, :credentials, :identity, keyword_init: true)
     Execution = Struct.new(:namespace, :task_queue, :timeouts, :headers, :search_attributes, keyword_init: true)
 
     attr_reader :timeouts, :error_handlers
-    attr_accessor :connection_type, :converter, :use_error_serialization_v2, :host, :port, :credentials, :identity, :logger, :metrics_adapter, :namespace, :task_queue, :headers, :search_attributes, :header_propagators
+    attr_accessor :connection_type, :converter, :use_error_serialization_v2, :host, :port, :credentials, :identity, :logger, :metrics_adapter, :namespace, :task_queue, :headers, :search_attributes, :header_propagators, :payload_codec
 
     # See https://docs.temporal.io/blog/activity-timeouts/ for general docs.
     # We want an infinite execution timeout for cron schedules and other perpetual workflows.
@@ -45,6 +46,14 @@ class Configuration
         Temporal::Connection::Converter::Payload::JSON.new
       ]
     ).freeze
+    
+    # The Payload Codec is an optional step that happens between the wire and the Payload Converter:
+    # Temporal Server <--> Wire <--> Payload Codec <--> Payload Converter <--> User code
+    # which can be useful for transformations such as compression and encryption
+    # more info at https://docs.temporal.io/security#payload-codec
+    DEFAULT_PAYLOAD_CODEC = Temporal::Connection::Converter::Codec::Chain.new(
+      payload_codecs: []
+    ).freeze
 
     def initialize
       @connection_type = :grpc
@@ -55,6 +64,7 @@ def initialize
       @task_queue = DEFAULT_TASK_QUEUE
       @headers = DEFAULT_HEADERS
       @converter = DEFAULT_CONVERTER
+      @payload_codec = DEFAULT_PAYLOAD_CODEC
       @use_error_serialization_v2 = false
       @error_handlers = []
       @credentials = :this_channel_is_insecure

lib/temporal/connection/converter/codec/base.rb

@@ -0,0 +1,35 @@
+module Temporal
+  module Connection
+    module Converter
+      module Codec
+        class Base
+          def encodes(payloads)
+            return nil if payloads.nil?
+
+            Temporalio::Api::Common::V1::Payloads.new(
+              payloads: payloads.payloads.map(&method(:encode))
+            )
+          end
+
+          def decodes(payloads)
+            return nil if payloads.nil?
+
+            Temporalio::Api::Common::V1::Payloads.new(
+              payloads: payloads.payloads.map(&method(:decode))
+            )
+          end
+
+          def encode(payload)
+            # should return Temporalio::Api::Common::V1::Payload
+            raise NotImplementedError, 'codec converter needs to implement encode'
+          end
+
+          def decode(payload)
+            # should return Temporalio::Api::Common::V1::Payload
+            raise NotImplementedError, 'codec converter needs to implement decode'
+          end
+        end
+      end
+    end
+  end
+end

lib/temporal/connection/converter/codec/chain.rb

@@ -0,0 +1,36 @@
+require 'temporal/connection/converter/codec/base'
+
+module Temporal
+  module Connection
+    module Converter
+      module Codec
+        # Performs encoding/decoding on the payloads via the given payload codecs. When encoding
+        # the codecs are applied last to first meaning the earlier encodings wrap the later ones.
+        # When decoding, the codecs are applied first to last to reverse the effect.
+        class Chain < Base
+          def initialize(payload_codecs:)
+            @payload_codecs = payload_codecs
+          end
+
+          def encode(payload)
+            payload_codecs.reverse_each do |payload_codec|
+              payload = payload_codec.encode(payload)
+            end
+            payload
+          end
+
+          def decode(payload)
+            payload_codecs.each do |payload_codec|
+              payload = payload_codec.decode(payload)
+            end
+            payload
+          end
+
+          private
+
+          attr_reader :payload_codecs
+        end
+      end
+    end
+  end
+end

lib/temporal/connection/grpc.rb

@@ -138,7 +138,7 @@ def start_workflow_execution(
             fields: to_payload_map(memo || {})
           ),
           search_attributes: Temporalio::Api::Common::V1::SearchAttributes.new(
-            indexed_fields: to_payload_map(search_attributes || {})
+            indexed_fields: to_payload_map_without_codec(search_attributes || {})
           ),
         )
 
@@ -401,7 +401,7 @@ def signal_with_start_workflow_execution(
             fields: to_payload_map(memo || {})
           ),
           search_attributes: Temporalio::Api::Common::V1::SearchAttributes.new(
-            indexed_fields: to_payload_map(search_attributes || {})
+            indexed_fields: to_payload_map_without_codec(search_attributes || {})
           ),
         )
 

lib/temporal/connection/serializer/continue_as_new.rb

@@ -43,7 +43,7 @@ def serialize_memo(memo)
         def serialize_search_attributes(search_attributes)
           return unless search_attributes
 
-          Temporalio::Api::Common::V1::SearchAttributes.new(indexed_fields: to_payload_map(search_attributes))
+          Temporalio::Api::Common::V1::SearchAttributes.new(indexed_fields: to_payload_map_without_codec(search_attributes))
         end
       end
     end

lib/temporal/connection/serializer/start_child_workflow.rb

@@ -66,7 +66,7 @@ def serialize_parent_close_policy(parent_close_policy)
         def serialize_search_attributes(search_attributes)
           return unless search_attributes
 
-          Temporalio::Api::Common::V1::SearchAttributes.new(indexed_fields: to_payload_map(search_attributes))
+          Temporalio::Api::Common::V1::SearchAttributes.new(indexed_fields: to_payload_map_without_codec(search_attributes))
         end
       end
     end