Add memos

Author: jeffschoner-stripe

examples/spec/integration/await_workflow_result_spec.rb

@@ -95,7 +95,9 @@
     run_id = Temporal.start_workflow(
       LoopWorkflow,
       2, # it continues as new if this arg is > 1
-      { options: { workflow_id: workflow_id } },
+      options: {
+        workflow_id: workflow_id,
+      },
     )
 
     expect do

examples/spec/integration/continue_as_new_spec.rb

@@ -0,0 +1,44 @@
+require 'workflows/loop_workflow'
+
+describe LoopWorkflow do
+  it 'workflow continues as new into a new run' do
+    workflow_id = SecureRandom.uuid
+    memo = {
+        'my-memo' => 'foo',
+    }
+    run_id = Temporal.start_workflow(
+      LoopWorkflow,
+      2, # it continues as new if this arg is > 1
+      options: {
+        workflow_id: workflow_id,
+        memo: memo,
+      },
+    )
+
+    # First run will throw because it continued as new
+    next_run_id = nil
+    expect do
+      Temporal.await_workflow_result(
+        LoopWorkflow,
+        workflow_id: workflow_id,
+        run_id: run_id,
+      )
+    end.to raise_error(Temporal::WorkflowRunContinuedAsNew) do |error|
+      next_run_id = error.new_run_id
+    end
+
+    expect(next_run_id).to_not eq(nil)
+
+    # Second run will not throw because it returns rather than continues as new.
+    final_result = Temporal.await_workflow_result(
+      LoopWorkflow,
+      workflow_id: workflow_id,
+      run_id: next_run_id,
+    )
+
+    expect(final_result[:count]).to eq(1)
+
+    # memo should be copied to the next run automatically
+    expect(final_result[:memo]).to eq(memo)
+  end
+end

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

@@ -8,6 +8,9 @@ def execute(count)
       return workflow.continue_as_new(count - 1)
     end
 
-    return count
+    return {
+      count: count,
+      memo: workflow.metadata.memo,
+    }
   end
 end

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