Worker specific task queues sample. (#34)

Closes #19

Author: robholland

README.md

@@ -27,6 +27,7 @@ Prerequisites:
   through workflows/activities.
 * [message_passing_simple](message_passing_simple) - Simple workflow that accepts signals, queries, and updates.
 * [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.
 
 ## Development
 

test/worker_specific_task_queues/file_processing_workflow_test.rb

@@ -0,0 +1,70 @@
+# frozen_string_literal: true
+
+require 'test'
+require 'worker_specific_task_queues/file_processing_workflow'
+require 'worker_specific_task_queues/normal_activities'
+require 'securerandom'
+require 'temporalio/testing'
+require 'temporalio/worker'
+
+module WorkerSpecificTaskQueues
+  class FileProcessingWorkflowTest < Test
+    class DownloadFileActivity < Temporalio::Activity::Definition
+      def execute(url)
+        url
+      end
+    end
+
+    class WorkOnFileActivity < Temporalio::Activity::Definition
+      @failed = false
+
+      class << self
+        attr_accessor :failed
+      end
+
+      def execute(_file_path)
+        # This should fail the first time it's called
+        unless self.class.failed
+          self.class.failed = true
+          raise 'Fake failure'
+        end
+
+        nil
+      end
+    end
+
+    class CleanupFileActivity < Temporalio::Activity::Definition
+      def execute(_file_path)
+        nil
+      end
+    end
+
+    def test_workflow
+      # Run test server until completion of the block
+      Temporalio::Testing::WorkflowEnvironment.start_local do |env|
+        unique_task_queue = "tq-#{SecureRandom.uuid}"
+        unique_worker = Temporalio::Worker.new(
+          client: env.client,
+          task_queue: unique_task_queue,
+          activities: [WorkerSpecificActivities::DownloadFileActivity, WorkerSpecificActivities::WorkOnFileActivity,
+                       WorkerSpecificActivities::CleanupFileActivity]
+        )
+
+        # Run worker until completion of the block
+        worker = Temporalio::Worker.new(
+          client: env.client,
+          task_queue: "tq-#{SecureRandom.uuid}",
+          activities: [WorkerSpecificTaskQueues::NormalActivities::GetUniqueTaskQueueActivity.new(unique_task_queue)],
+          workflows: [WorkerSpecificTaskQueues::FileProcessingWorkflow]
+        )
+        unique_worker.run do
+          worker.run do
+            # Run workflow
+            env.client.execute_workflow(WorkerSpecificTaskQueues::FileProcessingWorkflow, 2,
+                                        id: "wf-#{SecureRandom.uuid}", task_queue: worker.task_queue)
+          end
+        end
+      end
+    end
+  end
+end

worker_specific_task_queues/README.md

@@ -0,0 +1,30 @@
+# Worker-Specific Task Queues
+
+Use a unique Task Queue for each Worker in order to have certain Activities run on a specific Worker.
+
+This is useful in scenarios where multiple Activities need to run in the same process or on the same host, for example to share memory or disk. This sample has a file processing Workflow, where one Activity downloads the file to disk and other Activities process it and clean it up.
+
+This strategy is:
+
+- Each Worker process runs two workers:
+  - One worker listens on the shared `worker-specific-task-queues-sample` Task Queue.
+  - Another worker listens on a uniquely generated Task Queue.
+- Create a `GetUniqueTaskQueue` Activity that returns one of the uniquely generated Task Queues (that only one Worker is listening on—i.e. the **Worker-specific Task Queue**). It doesn't matter where this Activity is run, so it can be executed on the shared Task Queue. In this sample, the unique Task Queue is simply a UUID, but you can inject smart logic here to uniquely identify the Worker.
+- The Workflow and the first Activity are run on the shared `worker-specific-task-queues-sample` Task Queue. The rest of the Activities that do the file processing are run on the Worker-specific Task Queue.
+
+Activities have been artificially slowed with `sleep(3)` to simulate slow activities.
+
+### Running this sample
+
+1. Make sure Temporal Server is running locally (see [temporalio/docker-compose](https://github.com/temporalio/docker-compose))
+2. Run the following to start the worker:
+```
+bundle exec ruby worker.rb
+```
+
+3. In another terminal, run the workflow:
+```
+bundle exec ruby run_workflow.rb
+```
+
+You should see output in the worker terminal showing the file being downloaded, processed, and cleaned up. 

worker_specific_task_queues/file_processing_workflow.rb

@@ -0,0 +1,70 @@
+# frozen_string_literal: true
+
+require_relative 'normal_activities'
+require_relative 'worker_specific_activities'
+require 'temporalio/workflow'
+
+module WorkerSpecificTaskQueues
+  class FileProcessingWorkflow < Temporalio::Workflow::Definition
+    def execute(max_attempts)
+      attempt = 0
+
+      loop do
+        attempt += 1
+        begin
+          process_file
+          return
+        rescue StandardError => e
+          # If it's at max attempts, re-raise to fail the workflow
+          if attempt >= max_attempts
+            Temporalio::Workflow.logger.error(
+              "File processing failed and reached #{attempt} attempts, failing workflow: #{e.message}"
+            )
+            raise
+          end
+          # Otherwise, just warn and continue
+          Temporalio::Workflow.logger.warn(
+            "File processing failed on attempt #{attempt}, trying again: #{e.message}"
+          )
+        end
+      end
+    end
+
+    private
+
+    def process_file
+      # Get a unique task queue from any worker
+      unique_worker_task_queue = Temporalio::Workflow.execute_activity(
+        NormalActivities::GetUniqueTaskQueueActivity,
+        start_to_close_timeout: 60
+      )
+
+      # Download the file on the specific worker
+      download_path = Temporalio::Workflow.execute_activity(
+        WorkerSpecificActivities::DownloadFileActivity,
+        'https://temporal.io',
+        task_queue: unique_worker_task_queue,
+        schedule_to_close_timeout: 300,
+        heartbeat_timeout: 60
+      )
+
+      # Process the file on the same worker
+      Temporalio::Workflow.execute_activity(
+        WorkerSpecificActivities::WorkOnFileActivity,
+        download_path,
+        task_queue: unique_worker_task_queue,
+        schedule_to_close_timeout: 300,
+        heartbeat_timeout: 60
+      )
+
+      # Clean up the file on the same worker
+      Temporalio::Workflow.execute_activity(
+        WorkerSpecificActivities::CleanupFileActivity,
+        download_path,
+        task_queue: unique_worker_task_queue,
+        schedule_to_close_timeout: 300,
+        heartbeat_timeout: 60
+      )
+    end
+  end
+end

worker_specific_task_queues/normal_activities.rb

@@ -0,0 +1,19 @@
+# frozen_string_literal: true
+
+require 'temporalio/activity'
+
+module WorkerSpecificTaskQueues
+  module NormalActivities
+    class GetUniqueTaskQueueActivity < Temporalio::Activity::Definition
+      def initialize(unique_task_queue)
+        @unique_task_queue = unique_task_queue
+      end
+
+      def execute
+        # Return the known worker-specific task queue
+        # that was provided during initialization
+        @unique_task_queue
+      end
+    end
+  end
+end

worker_specific_task_queues/run_workflow.rb

@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+require_relative 'file_processing_workflow'
+require 'logger'
+require 'temporalio/client'
+
+# Create client with logger
+client = Temporalio::Client.connect(
+  'localhost:7233',
+  'default',
+  logger: Logger.new($stdout, level: Logger::INFO)
+)
+
+# Run workflow
+client.execute_workflow(
+  WorkerSpecificTaskQueues::FileProcessingWorkflow,
+  3, # max_attempts
+  id: 'file-processing-workflow',
+  task_queue: 'worker-specific-task-queues-sample'
+)
+
+puts 'Workflow completed successfully'

worker_specific_task_queues/worker.rb

@@ -0,0 +1,47 @@
+# frozen_string_literal: true
+
+require_relative 'file_processing_workflow'
+require_relative 'normal_activities'
+require_relative 'worker_specific_activities'
+require 'logger'
+require 'securerandom'
+require 'temporalio/client'
+require 'temporalio/worker'
+
+# Create client with logger
+client = Temporalio::Client.connect(
+  'localhost:7233',
+  'default',
+  logger: Logger.new($stdout, level: Logger::INFO)
+)
+
+# Create a unique task queue for this worker
+unique_task_queue = SecureRandom.uuid
+
+# Create worker for shared task queue
+shared_worker = Temporalio::Worker.new(
+  client: client,
+  task_queue: 'worker-specific-task-queues-sample',
+  workflows: [WorkerSpecificTaskQueues::FileProcessingWorkflow],
+  activities: [
+    # Pass the unique task queue to the activity so it can return it when called
+    WorkerSpecificTaskQueues::NormalActivities::GetUniqueTaskQueueActivity.new(unique_task_queue)
+  ]
+)
+
+# Create worker for unique task queue
+unique_worker = Temporalio::Worker.new(
+  client: client,
+  task_queue: unique_task_queue,
+  activities: [
+    WorkerSpecificTaskQueues::WorkerSpecificActivities::DownloadFileActivity.new,
+    WorkerSpecificTaskQueues::WorkerSpecificActivities::WorkOnFileActivity.new,
+    WorkerSpecificTaskQueues::WorkerSpecificActivities::CleanupFileActivity.new
+  ]
+)
+
+puts "Running worker with unique task queue: #{unique_task_queue}"
+
+# Run both workers using run_all for concurrent execution
+# The workers need to be passed as separate arguments, not as an array
+Temporalio::Worker.run_all(shared_worker, unique_worker, shutdown_signals: ['SIGINT'])

worker_specific_task_queues/worker_specific_activities.rb

@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+require 'digest'
+require 'logger'
+require 'net/http'
+require 'tempfile'
+require 'temporalio/activity'
+require 'uri'
+
+module WorkerSpecificTaskQueues
+  module WorkerSpecificActivities
+    class DownloadFileActivity < Temporalio::Activity::Definition
+      def execute(url)
+        # Simulate slow activity
+        sleep(3)
+
+        file = Tempfile.create
+
+        Temporalio::Activity::Context.current.logger.info("Downloading #{url} to #{file.path}")
+        file.write(Net::HTTP.get(URI(url)))
+        file.close
+
+        file.path
+      end
+    end
+
+    class WorkOnFileActivity < Temporalio::Activity::Definition
+      def execute(file_path)
+        # Simulate slow activity
+        sleep(3)
+
+        # Calculate checksum to simulate work
+        checksum = Digest::SHA256.file(file_path).hexdigest
+        Temporalio::Activity::Context.current.logger.info("Did some work on #{file_path}, checksum: #{checksum}")
+
+        nil
+      end
+    end
+
+    class CleanupFileActivity < Temporalio::Activity::Definition
+      def execute(file_path)
+        # Simulate slow activity
+        sleep(3)
+
+        Temporalio::Activity::Context.current.logger.info("Removing #{file_path}")
+        File.unlink(file_path)
+
+        nil
+      end
+    end
+  end
+end