Merge pull request coinbase#51 from francis-li/francis/go-client-compatibility

[TFRS-3063] go client compatibility

Author: DeRauk

.circleci/config.yml

@@ -5,7 +5,7 @@ orbs:
 jobs:
   test_gem:
     docker:
-      - image: circleci/ruby:2.6.3-stretch-node
+      - image: cimg/ruby:2.7.6
     executor: ruby/default
     steps:
       - checkout
@@ -16,12 +16,49 @@ jobs:
 
   test_examples:
     docker:
-      - image: circleci/ruby:2.6.3-stretch-node
+      - image: cimg/ruby:2.7.6
+      - image: circleci/postgres:alpine
+        name: postgres
+        environment:
+          POSTGRES_PASSWORD: cadence
+      - image: ubercadence/server:master-auto-setup
+        name: cadence
+        environment:
+          DB: postgres
+          DB_PORT: 5432
+          POSTGRES_USER: postgres
+          POSTGRES_PWD: cadence
+          POSTGRES_SEEDS: postgres
+          DYNAMIC_CONFIG_FILE_PATH: config/dynamicconfig/development.yaml
+      - image: antstorm/cadence-tchannel-proxy:latest
+        name: proxy
+        environment:
+          BIND_ADDRESS: 0.0.0.0:6666
+          CADENCE_ADDRESS: cadence:7933
+
+    environment:
+      CADENCE_HOST: proxy
+      CADENCE_PORT: 6666
+
     steps:
       - checkout
       - run:
           name: Bundle Install
           command: cd examples && bundle install --path vendor/bundle
+
+      - run:
+          name: Register Domain
+          command: cd examples && bin/register_domain ruby-samples
+
+      - run:
+          name: Wait for Domain to settle
+          command: sleep 15
+
+      - run:
+          name: Boot up worker
+          command: cd examples && bin/worker
+          background: true
+
       - run:
           name: Run RSpec
           command: cd examples && bundle exec rspec
@@ -31,4 +68,4 @@ workflows:
   test:
     jobs:
       - test_gem
-      - test_examples
+      - test_examples

CHANGELOG.md

@@ -1,6 +1,9 @@
 # Changelog
 
-## 0.1.14
+## 0.1.16
+- Add activity task deserialization logic that is compatible with official go implementation
+
+## 0.1.15
 - Add rbi definition for Cadence::Crew
 
 ## 0.1.14

examples/Gemfile

@@ -3,5 +3,5 @@ source 'https://rubygems.org'
 gem 'cadence-ruby', path: '../'
 
 gem 'dry-struct', '~> 1.1.1'
-gem 'dry-types', '~> 1.2.0'
+gem 'dry-types', '~> 1.6.0'
 gem 'rspec', group: :test

examples/bin/crew

@@ -0,0 +1,56 @@
+#!/usr/bin/env ruby
+require_relative '../init'
+
+require 'cadence/crew'
+require 'cadence/worker'
+
+init = proc do |worker|
+  Dir[File.expand_path('../workflows/*.rb', __dir__)].each { |f| require f }
+  Dir[File.expand_path('../activities/*.rb', __dir__)].each { |f| require f }
+  Dir[File.expand_path('../middleware/*.rb', __dir__)].each { |f| require f }
+
+  worker.register_workflow(AsyncActivityWorkflow)
+  worker.register_workflow(AsyncHelloWorldWorkflow)
+  worker.register_workflow(AsyncTimerWorkflow)
+  worker.register_workflow(BranchingWorkflow)
+  worker.register_workflow(CancellingTimerWorkflow)
+  worker.register_workflow(CheckWorkflow)
+  worker.register_workflow(FailingWorkflow)
+  worker.register_workflow(HelloWorldWorkflow)
+  worker.register_workflow(LocalHelloWorldWorkflow)
+  worker.register_workflow(LongWorkflow)
+  worker.register_workflow(ParentWorkflow)
+  worker.register_workflow(ProcessFileWorkflow)
+  worker.register_workflow(ReleaseWorkflow)
+  worker.register_workflow(SerialHelloWorldWorkflow)
+  worker.register_workflow(SideEffectWorkflow)
+  worker.register_workflow(SimpleTimerWorkflow)
+  worker.register_workflow(TimeoutWorkflow)
+  worker.register_workflow(TripBookingWorkflow)
+
+  worker.register_activity(AsyncActivity)
+  worker.register_activity(EchoActivity)
+  worker.register_activity(GenerateFileActivity)
+  worker.register_activity(HelloWorldActivity)
+  worker.register_activity(LongRunningActivity)
+  worker.register_activity(ProcessFileActivity)
+  worker.register_activity(RandomlyFailingActivity)
+  worker.register_activity(RandomNumberActivity)
+  worker.register_activity(SleepActivity)
+  worker.register_activity(UploadFileActivity)
+  worker.register_activity(Trip::BookFlightActivity)
+  worker.register_activity(Trip::BookHotelActivity)
+  worker.register_activity(Trip::CancelCarActivity)
+  worker.register_activity(Trip::CancelFlightActivity)
+  worker.register_activity(Trip::CancelHotelActivity)
+  worker.register_activity(Trip::MakePaymentActivity)
+  worker.register_activity(Trip::RentCarActivity)
+
+  worker.add_decision_middleware(LoggingMiddleware, 'EXAMPLE')
+  worker.add_activity_middleware(LoggingMiddleware, 'EXAMPLE')
+end
+
+worker = Cadence::Worker.new(polling_ttl: 10, thread_pool_size: 20)
+crew = Cadence::Crew.new(worker, 2)
+crew.after_fork &init
+crew.dispatch

examples/bin/worker

@@ -50,10 +50,4 @@ worker.register_activity(Trip::RentCarActivity)
 worker.add_decision_middleware(LoggingMiddleware, 'EXAMPLE')
 worker.add_activity_middleware(LoggingMiddleware, 'EXAMPLE')
 
-option1 = ARGV[0]
-if option1 == '--crew'
-  crew = Cadence::Crew.new(worker, 2)
-  crew.dispatch
-else
-  worker.start
-end
+worker.start

lib/cadence/activity/task_processor.rb

@@ -32,7 +32,7 @@ def process
         context = Activity::Context.new(connection, metadata)
 
         result = middleware_chain.invoke(metadata) do
-          activity_class.execute_in_context(context, JSON.deserialize(task.input))
+          activity_class.execute_in_context(context, deserialize(task.input))
         end
 
         # Do not complete asynchronous activities, these should be completed manually
@@ -79,6 +79,25 @@ def respond_failed(reason, details)
         Cadence.logger.error("Unable to fail Activity #{activity_name}: #{error.inspect}")
         Cadence::ErrorHandler.handle(error, metadata: metadata)
       end
+
+      def deserialize(input)
+        JSON.deserialize(task.input)
+      rescue Oj::ParseError
+        # cadence official go-client serializes / deserializes input in a different format than this ruby client
+        # adding additional deserialization logic here to help read input that is passed from go-client
+        # https://github.com/uber-go/cadence-client/blob/0.18.x/internal/encoding.go#L45-L58
+        #
+        # this ruby client serializes / deserializes everything as one big string like below:
+        # [1012474654, "second input"]
+        #
+        # while go client serializes input as separate input followed by line break
+        # 1012474654
+        # second input
+        args = input.split(/\n/)
+        res = args.map do |arg|
+          JSON.deserialize(arg)
+        end
+      end
     end
   end
 end

lib/cadence/crew.rb

@@ -31,12 +31,23 @@ def stop(signal)
       crew.each { |pid| stop_worker(signal, pid) }
     end
 
+    def after_fork(&block)
+      raise 'after_fork can only be called before dispatching the worker' if crew.length.positive?
+
+      @after_fork_block = block
+    end
+
     private
 
+    attr_accessor :after_fork_block
     attr_reader :worker, :crew, :logger
 
     def dispatch_worker
-      pid = fork { worker.start }
+      pid = fork do
+        after_fork_block&.call(worker)
+        worker.start
+      end
+
       crew << pid
 
       Cadence.metrics.gauge("crew.num_workers", crew.size)
@@ -49,7 +60,7 @@ def monitor
       while crew.length.positive?
         (pid, status) = ::Process.waitpid2(-1)
         crew.delete(pid)
-        
+
         Cadence.metrics.gauge("crew.num_workers", crew.size)
         logger.info "Worker quit. pid: #{pid.to_s}, exitstatus: #{status.exitstatus}, remaining_workers: #{crew.length}"
       end

lib/cadence/version.rb

@@ -1,3 +1,3 @@
 module Cadence
-  VERSION = '0.1.15'.freeze
+  VERSION = '0.1.16'.freeze
 end

rbi/cadence-ruby.rbi

@@ -140,6 +140,8 @@ class Cadence::Worker
 end
 
 class Cadence::Crew
+    def after_fork(block); end
+
     def dispatch; end
 
     def stop(signal); end

spec/unit/lib/cadence/activity/task_processor_spec.rb

@@ -16,6 +16,7 @@
   let(:middleware_chain) { Cadence::Middleware::Chain.new }
   let(:config) { Cadence::Configuration.new }
   let(:input) { ['arg1', 'arg2'] }
+  let(:go_client_input) { "123\n\"abc\"" }
 
   describe '#process' do
     let(:context) { instance_double('Cadence::Activity::Context', async?: false) }
@@ -40,6 +41,21 @@
       allow(Cadence.metrics).to receive(:timing)
     end
 
+    context 'when input comes from go client' do
+      let(:activity_class) { double('Cadence::Activity', execute_in_context: nil) }
+      let(:task) { Fabricate(:activity_task_thrift, activity_name: activity_name, input: go_client_input) }
+
+      before do
+        allow(lookup).to receive(:find).with(activity_name).and_return(activity_class)
+      end
+
+      it 'successfully deserializes the input and ran the activity' do
+        subject.process
+
+        expect(activity_class).to have_received(:execute_in_context).with(context, [123, 'abc'])
+      end
+    end
+
     context 'when activity is not registered' do
       it 'fails the activity task' do
         subject.process