Add memos

Author: jeffschoner-stripe

examples/spec/integration/await_workflow_result_spec.rb

@@ -92,10 +92,17 @@
 
   it 'raises Temporal::WorkflowRunContinuedAsNew when the workflow continues as new' do
     workflow_id = SecureRandom.uuid
+    expected_memo = {
+      'my-memo' => 'bar'
+    }
     run_id = Temporal.start_workflow(
       LoopWorkflow,
       2, # it continues as new if this arg is > 1
-      { options: { workflow_id: workflow_id } },
+      expected_memo,
+      options: {
+        workflow_id: workflow_id,
+        memo: expected_memo,
+      },
     )
 
     expect do

examples/spec/integration/metadata_workflow_spec.rb

@@ -7,13 +7,15 @@
     workflow_id = 'task-queue-' + SecureRandom.uuid
     run_id = Temporal.start_workflow(
       subject,
-      { options: { workflow_id: workflow_id } },
+      options: { workflow_id: workflow_id }
     )
+
     actual_result = Temporal.await_workflow_result(
       subject,
       workflow_id: workflow_id,
       run_id: run_id,
     )
+
     expect(actual_result.task_queue).to eq(Temporal.configuration.task_queue)
   end
 
@@ -51,4 +53,39 @@
     )
     expect(Time.now - actual_result.run_started_at).to be_between(0, 30)
   end
+
+  it 'gets memo from workflow execution info' do
+    workflow_id = 'memo_execution_test_wf-' + SecureRandom.uuid
+    run_id = Temporal.start_workflow(subject, options: { workflow_id: workflow_id, memo: { 'foo' => 'bar' } })
+
+    actual_result = Temporal.await_workflow_result(
+      subject,
+      workflow_id: workflow_id,
+      run_id: run_id,
+    )
+    expect(actual_result.memo['foo']).to eq('bar')
+
+    expect(Temporal.fetch_workflow_execution_info(
+      'ruby-samples', workflow_id, nil
+    ).memo).to eq({ 'foo' => 'bar' })
+  end
+
+  it 'gets memo from workflow context with no memo' do
+    workflow_id = 'memo_context_no_memo_test_wf-' + SecureRandom.uuid
+
+    run_id = Temporal.start_workflow(
+      subject,
+      options: { workflow_id: workflow_id }
+    )
+
+    actual_result = Temporal.await_workflow_result(
+      subject,
+      workflow_id: workflow_id,
+      run_id: run_id,
+    )
+    expect(actual_result.memo).to eq({})
+    expect(Temporal.fetch_workflow_execution_info(
+      'ruby-samples', workflow_id, nil
+    ).memo).to eq({})
+  end
 end

examples/workflows/loop_workflow.rb

@@ -1,7 +1,13 @@
 require 'activities/hello_world_activity'
 
 class LoopWorkflow < Temporal::Workflow
-  def execute(count)
+  def execute(count, assert_memo)
+    assert_memo.each do |memo_key, memo_value|
+      if workflow.metadata.memo[memo_key] != memo_value
+        raise "Memo #{memo_key} was not #{memo_value}!"
+      end
+    end
+
     HelloWorldActivity.execute!('Alice')
 
     if count > 1

examples/workflows/metadata_workflow.rb

@@ -2,4 +2,4 @@ class MetadataWorkflow < Temporal::Workflow
   def execute
     workflow.metadata
   end
-end
+end

lib/temporal/client.rb

@@ -57,7 +57,8 @@ def start_workflow(workflow, *input, options: {}, **args)
           run_timeout: compute_run_timeout(execution_options),
           task_timeout: execution_options.timeouts[:task],
           workflow_id_reuse_policy: options[:workflow_id_reuse_policy],
-          headers: execution_options.headers
+          headers: execution_options.headers,
+          memo: execution_options.memo,
         )
       else
         raise ArgumentError, 'If signal_input is provided, you must also provide signal_name' if signal_name.nil?
@@ -73,6 +74,7 @@ def start_workflow(workflow, *input, options: {}, **args)
           task_timeout: execution_options.timeouts[:task],
           workflow_id_reuse_policy: options[:workflow_id_reuse_policy],
           headers: execution_options.headers,
+          memo: execution_options.memo,
           signal_name: signal_name,
           signal_input: signal_input
         )
@@ -119,7 +121,8 @@ def schedule_workflow(workflow, cron_schedule, *input, options: {}, **args)
         task_timeout: execution_options.timeouts[:task],
         workflow_id_reuse_policy: options[:workflow_id_reuse_policy],
         headers: execution_options.headers,
-        cron_schedule: cron_schedule
+        cron_schedule: cron_schedule,
+        memo: execution_options.memo
       )
 
       response.run_id

lib/temporal/concerns/payloads.rb

@@ -21,6 +21,10 @@ def from_signal_payloads(payloads)
         from_payloads(payloads)&.first
       end
 
+      def from_payload_map(payload_map)
+        payload_map.map { |key, value| [key, from_payload(value)] }.to_h
+      end
+
       def to_payloads(data)
         payload_converter.to_payloads(data)
       end
@@ -41,6 +45,10 @@ def to_signal_payloads(data)
         to_payloads([data])
       end
 
+      def to_payload_map(data)
+        data.transform_values(&method(:to_payload))
+      end
+
       private
 
       def payload_converter

lib/temporal/connection/grpc.rb

@@ -81,7 +81,8 @@ def start_workflow_execution(
         task_timeout:,
         workflow_id_reuse_policy: nil,
         headers: nil,
-        cron_schedule: nil
+        cron_schedule: nil,
+        memo: nil
       )
         request = Temporal::Api::WorkflowService::V1::StartWorkflowExecutionRequest.new(
           identity: identity,
@@ -101,7 +102,10 @@ def start_workflow_execution(
           header: Temporal::Api::Common::V1::Header.new(
             fields: headers
           ),
-          cron_schedule: cron_schedule
+          cron_schedule: cron_schedule,
+          memo: Temporal::Api::Common::V1::Memo.new(
+            fields: to_payload_map(memo || {})
+          )
         )
 
         if workflow_id_reuse_policy
@@ -305,7 +309,8 @@ def signal_with_start_workflow_execution(
         headers: nil,
         cron_schedule: nil,
         signal_name:,
-        signal_input:
+        signal_input:,
+        memo: nil
       )
         request = Temporal::Api::WorkflowService::V1::SignalWithStartWorkflowExecutionRequest.new(
           identity: identity,
@@ -327,7 +332,10 @@ def signal_with_start_workflow_execution(
           ),
           cron_schedule: cron_schedule,
           signal_name: signal_name,
-          signal_input: to_signal_payloads(signal_input)
+          signal_input: to_signal_payloads(signal_input),
+          memo: Temporal::Api::Common::V1::Memo.new(
+            fields: to_payload_map(memo || {})
+          ),
         )
 
         if workflow_id_reuse_policy

lib/temporal/connection/serializer/continue_as_new.rb

@@ -19,7 +19,8 @@ def to_proto
                 workflow_run_timeout: object.timeouts[:execution],
                 workflow_task_timeout: object.timeouts[:task],
                 retry_policy: Temporal::Connection::Serializer::RetryPolicy.new(object.retry_policy).to_proto,
-                header: serialize_headers(object.headers)
+                header: serialize_headers(object.headers),
+                memo: serialize_memo(object.memo)
               )
           )
         end
@@ -31,6 +32,12 @@ def serialize_headers(headers)
 
           Temporal::Api::Common::V1::Header.new(fields: object.headers)
         end
+
+        def serialize_memo(memo)
+          return unless memo
+
+          Temporal::Api::Common::V1::Memo.new(fields: to_payload_map(memo))
+        end
       end
     end
   end

lib/temporal/connection/serializer/start_child_workflow.rb

@@ -22,7 +22,8 @@ def to_proto
                 workflow_run_timeout: object.timeouts[:run],
                 workflow_task_timeout: object.timeouts[:task],
                 retry_policy: Temporal::Connection::Serializer::RetryPolicy.new(object.retry_policy).to_proto,
-                header: serialize_headers(object.headers)
+                header: serialize_headers(object.headers),
+                memo: serialize_memo(object.memo),
               )
           )
         end
@@ -34,6 +35,12 @@ def serialize_headers(headers)
 
           Temporal::Api::Common::V1::Header.new(fields: object.headers)
         end
+
+        def serialize_memo(memo)
+          return unless memo
+
+          Temporal::Api::Common::V1::Memo.new(fields: object.memo)
+        end
       end
     end
   end

lib/temporal/execution_options.rb

@@ -3,7 +3,7 @@
 
 module Temporal
   class ExecutionOptions
-    attr_reader :name, :namespace, :task_queue, :retry_policy, :timeouts, :headers
+    attr_reader :name, :namespace, :task_queue, :retry_policy, :timeouts, :headers, :memo
 
     def initialize(object, options, defaults = nil)
       # Options are treated as overrides and take precedence
@@ -13,6 +13,7 @@ def initialize(object, options, defaults = nil)
       @retry_policy = options[:retry_policy] || {}
       @timeouts = options[:timeouts] || {}
       @headers = options[:headers] || {}
+      @memo = options[:memo] || {}
 
       # For Temporal::Workflow and Temporal::Activity use defined values as the next option
       if has_executable_concern?(object)