Saga sample (#47)

Fixes #14

Author: cretz

README.md

@@ -30,6 +30,7 @@ Prerequisites:
 * [message_passing_simple](message_passing_simple) - Simple workflow that accepts signals, queries, and updates.
 * [polling/infrequent](polling/infrequent) - Implement an infrequent polling mechanism using Temporal's automatic Activity Retry feature.
 * [rails_app](rails_app) - Basic Rails API application using Temporal workflows and activities.
+* [saga](saga) - Using undo/compensation using a very simplistic Saga pattern.
 * [sorbet_generic](sorbet_generic) - Proof of concept of how to do _advanced_ Sorbet typing with the SDK.
 * [worker_specific_task_queues](worker_specific_task_queues) - Use a unique Task Queue for each Worker to run a sequence of Activities on the same Worker.
 

saga/README.md

@@ -0,0 +1,33 @@
+# Saga
+
+This sample demonstrates undo/compensation using a very simplistic Saga pattern.
+
+To run, first see [README.md](../README.md) for prerequisites. Then, in another terminal, start the Ruby worker
+from this directory:
+
+    bundle exec ruby worker.rb
+
+Finally in another terminal, use the Ruby client to the workflow from this directory:
+
+    bundle exec ruby starter.rb
+
+By intention, this will fail with an exception and backtrace. The exception will be a
+`Temporalio::Error::WorkflowFailedError` with a cause of `Temporalio::Error::ActivityError` that will have a
+"Simulated failure" message.
+
+Looking at the worker side, the following logs will be visible (adjusted for clarity):
+
+```
+INFO -- : Withdrawing 100 from acc1000. Reference ID: 1324
+INFO -- : Depositing 100 into acc2000. Reference ID: 1324
+INFO -- : Simulate failure to trigger compensation. Reference ID: 1324
+WARN -- : Completing activity as failed
+WARN -- : Simulated failure
+<backtrace omitted>
+INFO -- : Undoing deposit of 100 into acc2000. Reference ID: 1324
+INFO -- : Undoing withdraw of 100 from acc1000. Reference ID: 1324
+```
+
+This shows a withdraw and deposit activity completing, but then an activity raised an error (by intention in this
+sample), so we undo the deposit/withdraw in reverse before re-raising that error. These steps that were performed are
+also visible in the UI.

saga/activities.rb

@@ -0,0 +1,59 @@
+# frozen_string_literal: true
+
+require 'temporalio/activity'
+
+# To get JSON additions for struct
+require 'json/add/struct'
+
+module Saga
+  module Activities
+    # Transfer details parameter we use everywhere
+    TransferDetails = Struct.new(
+      :amount,
+      :from_account,
+      :to_account,
+      :reference_id
+    )
+
+    class Withdraw < Temporalio::Activity::Definition
+      def execute(details)
+        Temporalio::Activity::Context.current.logger.info(
+          "Withdrawing #{details.amount} from #{details.from_account}. Reference ID: #{details.reference_id}"
+        )
+      end
+    end
+
+    class WithdrawCompensation < Temporalio::Activity::Definition
+      def execute(details)
+        Temporalio::Activity::Context.current.logger.info(
+          "Undoing withdraw of #{details.amount} from #{details.from_account}. Reference ID: #{details.reference_id}"
+        )
+      end
+    end
+
+    class Deposit < Temporalio::Activity::Definition
+      def execute(details)
+        Temporalio::Activity::Context.current.logger.info(
+          "Depositing #{details.amount} into #{details.to_account}. Reference ID: #{details.reference_id}"
+        )
+      end
+    end
+
+    class DepositCompensation < Temporalio::Activity::Definition
+      def execute(details)
+        Temporalio::Activity::Context.current.logger.info(
+          "Undoing deposit of #{details.amount} into #{details.to_account}. Reference ID: #{details.reference_id}"
+        )
+      end
+    end
+
+    class SomethingThatFails < Temporalio::Activity::Definition
+      def execute(details)
+        Temporalio::Activity::Context.current.logger.info(
+          "Simulate failure to trigger compensation. Reference ID: #{details.reference_id}"
+        )
+        raise Temporalio::Error::ApplicationError.new('Simulated failure', non_retryable: true)
+      end
+    end
+  end
+end

saga/saga_workflow.rb

@@ -0,0 +1,42 @@
+# frozen_string_literal: true
+
+require 'temporalio/workflow'
+require_relative 'activities'
+
+module Saga
+  class SagaWorkflow < Temporalio::Workflow::Definition
+    def execute(details)
+      # Collect compensation activities (aka activities to perform undo) in an array
+      compensations = []
+
+      # Perform some actions. Notice how we add compensations _before_ executing the associated activities. This is a
+      # user choice, but usually it is best before because the activity may have run but the end of it failed or timed
+      # out. The compensation should be smart enough to check what is about to undo before it does it.
+
+      # Withdraw money
+      compensations << Activities::WithdrawCompensation
+      Temporalio::Workflow.execute_activity(Activities::Withdraw, details, start_to_close_timeout: 30)
+
+      # Deposit money
+      compensations << Activities::DepositCompensation
+      Temporalio::Workflow.execute_activity(Activities::Deposit, details, start_to_close_timeout: 30)
+
+      # Simulate a failure. This simulates a failure after withdraw and deposit, but this could just as easily be a
+      # failure with either of those.
+      Temporalio::Workflow.execute_activity(Activities::SomethingThatFails, details, start_to_close_timeout: 30)
+
+      # Never reached
+      nil
+    rescue StandardError
+      # Perform the compensations in reverse. It is user choice on whether a compensation failure should be allowed to
+      # raise thereby swallowing the existing one, or if it should be swallowed. This sample raises the compensation
+      # error because either error fails the workflow anyways and both errors are going to be visible in history.
+      compensations.reverse_each do |compensating_activity|
+        Temporalio::Workflow.execute_activity(compensating_activity, details, start_to_close_timeout: 30)
+      end
+
+      # Re-raise
+      raise
+    end
+  end
+end

saga/starter.rb

@@ -0,0 +1,21 @@
+# frozen_string_literal: true
+
+require 'temporalio/client'
+require_relative 'activities'
+require_relative 'saga_workflow'
+
+# Create a Temporal client
+client = Temporalio::Client.connect('localhost:7233', 'default')
+
+# Run workflow that we know will fail
+client.execute_workflow(
+  Saga::SagaWorkflow,
+  Saga::Activities::TransferDetails.new(
+    amount: 100,
+    from_account: 'acc1000',
+    to_account: 'acc2000',
+    reference_id: '1324'
+  ),
+  id: 'saga-sample-workflow-id',
+  task_queue: 'saga-sample'
+)

saga/worker.rb

@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+require 'logger'
+require 'temporalio/client'
+require 'temporalio/worker'
+require_relative 'activities'
+require_relative 'saga_workflow'
+
+# Create a Temporal client
+client = Temporalio::Client.connect(
+  'localhost:7233', 'default',
+  # Enable info logging to see our activity logs
+  logger: Logger.new($stdout, level: Logger::INFO)
+)
+
+# Create worker with the activities and workflow
+worker = Temporalio::Worker.new(
+  client:,
+  task_queue: 'saga-sample',
+  activities: [Saga::Activities::Withdraw, Saga::Activities::WithdrawCompensation,
+               Saga::Activities::Deposit, Saga::Activities::DepositCompensation,
+               Saga::Activities::SomethingThatFails],
+  workflows: [Saga::SagaWorkflow]
+)
+
+# Run the worker until SIGINT
+puts 'Starting worker (ctrl+c to exit)'
+worker.run(shutdown_signals: ['SIGINT'])

test/saga/saga_workflow_test.rb

@@ -0,0 +1,53 @@
+# frozen_string_literal: true
+
+require 'test'
+require 'securerandom'
+require 'saga/activities'
+require 'saga/saga_workflow'
+require 'temporalio/testing'
+require 'temporalio/worker'
+require 'polling/infrequent/compose_greeting_activity'
+require 'polling/infrequent/test_service'
+
+module Saga
+  class SagaWorkflowTest < Test
+    def test_workflow_runs_compensations
+      # Run worker in test environment
+      Temporalio::Testing::WorkflowEnvironment.start_local do |env|
+        worker = Temporalio::Worker.new(
+          client: env.client,
+          task_queue: "tq-#{SecureRandom.uuid}",
+          activities: [Activities::Withdraw, Activities::WithdrawCompensation,
+                       Activities::Deposit, Activities::DepositCompensation,
+                       Activities::SomethingThatFails],
+          workflows: [SagaWorkflow]
+        )
+        worker.run do
+          # Start workflow
+          handle = env.client.start_workflow(
+            SagaWorkflow,
+            Saga::Activities::TransferDetails.new(
+              amount: 100,
+              from_account: 'acc1000',
+              to_account: 'acc2000',
+              reference_id: '1324'
+            ),
+            id: "wf-#{SecureRandom.uuid}",
+            task_queue: worker.task_queue
+          )
+
+          # Confirm it failed as expected
+          err = assert_raises(Temporalio::Error::WorkflowFailedError) { handle.result }
+          assert_instance_of(Temporalio::Error::ActivityError, err.cause)
+          assert_instance_of(Temporalio::Error::ApplicationError, err.cause.cause)
+          assert_equal('Simulated failure', err.cause.cause.message)
+
+          # Confirm last two events are the compensations
+          activity_events = handle.fetch_history_events.map(&:activity_task_scheduled_event_attributes).compact
+          assert_equal('DepositCompensation', activity_events[-2].activity_type.name)
+          assert_equal('WithdrawCompensation', activity_events[-1].activity_type.name)
+        end
+      end
+    end
+  end
+end