feat: external storage foundation types

temporalio/lib/temporalio/converters.rb

@@ -1,6 +1,15 @@
 # frozen_string_literal: true
 
 require 'temporalio/converters/data_converter'
+require 'temporalio/converters/external_storage'
+require 'temporalio/converters/storage_driver'
+require 'temporalio/converters/storage_driver_activity_info'
+require 'temporalio/converters/storage_driver_claim'
+require 'temporalio/converters/storage_driver_retrieve_context'
+require 'temporalio/converters/storage_driver_select_context'
+require 'temporalio/converters/storage_driver_store_context'
+require 'temporalio/converters/storage_driver_workflow_info'
+require 'temporalio/internal/external_storage_references'
 
 module Temporalio
   # Module for data conversion.

temporalio/lib/temporalio/converters/external_storage.rb

@@ -0,0 +1,79 @@
+# frozen_string_literal: true
+
+require 'temporalio/converters/storage_driver'
+
+module Temporalio
+  module Converters
+    # Configuration for offloading large payloads to an external storage system.
+    #
+    # This is validated when constructed rather than when used, so a misconfiguration surfaces at startup instead of on
+    # the first payload large enough to offload.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!visibility private
+    class ExternalStorage
+      # Shared with every other SDK, so the same payload offloads regardless of which one wrote it.
+      DEFAULT_PAYLOAD_SIZE_THRESHOLD = 256 * 1024
+
+      # @return [Array<StorageDriver>] Drivers available for storing and retrieving payloads.
+      attr_reader :drivers
+
+      # @return [Proc] Selector choosing which driver stores each payload. Called with a
+      #   {StorageDriverSelectContext} and a payload, returning a {StorageDriver} or nil to pass the payload through.
+      #   Only called for payloads that meet {#payload_size_threshold}, and must return one of {#drivers} or nil.
+      attr_reader :driver_selector
+
+      # @return [Integer] Minimum encoded payload size, in bytes, that is offloaded. Payloads at or above this size are
+      #   offloaded; smaller ones are left inline. Zero offloads every payload.
+      attr_reader :payload_size_threshold
+
+      # Create external storage configuration.
+      #
+      # @param drivers [Array<StorageDriver>] Drivers available for storing and retrieving payloads. At least one is
+      #   required, and every driver must have a unique, non-empty name. Retrieval routes by the name recorded in
+      #   history, so a driver must remain configured for as long as any payload it stored can still be read.
+      # @param driver_selector [Proc, nil] Selector choosing which driver stores each payload. Required when more than
+      #   one driver is given; with a single driver it may be nil, in which case that driver stores every eligible
+      #   payload.
+      # @param payload_size_threshold [Integer] Minimum encoded payload size, in bytes, to offload. Must be a
+      #   non-negative Integer.
+      def initialize(drivers:, driver_selector: nil, payload_size_threshold: DEFAULT_PAYLOAD_SIZE_THRESHOLD)
+        raise ArgumentError, 'At least one driver is required' if drivers.empty?
+
+        unless payload_size_threshold.is_a?(Integer) && payload_size_threshold >= 0
+          raise ArgumentError, 'payload_size_threshold must be a non-negative Integer'
+        end
+
+        if drivers.size > 1 && driver_selector.nil?
+          raise ArgumentError, 'driver_selector is required when more than one driver is given'
+        end
+
+        @drivers_by_name = {}
+        drivers.each do |driver|
+          name = driver.name
+          # The name is the routing key written into history and is written to a string proto field, so anything but
+          # a non-empty String cannot be resolved on read. A Symbol in particular passes a bare empty? check.
+          raise ArgumentError, 'Driver name must be a non-empty String' unless name.is_a?(String) && !name.empty?
+          raise ArgumentError, "Multiple drivers given with name '#{name}'" if @drivers_by_name.key?(name)
+
+          @drivers_by_name[name] = driver
+        end
+        @drivers_by_name.freeze
+
+        @drivers = drivers.dup.freeze
+        # Normalized so callers never have to distinguish the single-driver case.
+        @driver_selector = driver_selector || ->(_context, _payload) { @drivers.first }
+        @payload_size_threshold = payload_size_threshold
+      end
+
+      # Get the driver registered under the given name.
+      #
+      # @param name [String] Driver name recorded in the payload reference.
+      # @return [StorageDriver, nil] The driver, or nil if none is registered under that name.
+      def driver(name)
+        @drivers_by_name[name]
+      end
+    end
+  end
+end

temporalio/lib/temporalio/converters/storage_driver.rb

@@ -0,0 +1,56 @@
+# frozen_string_literal: true
+
+module Temporalio
+  module Converters
+    # Base class for storing and retrieving payloads in an external storage system.
+    #
+    # Payloads at or over the configured size threshold are handed to a driver and replaced on the wire by a small
+    # reference, so the payload data itself never reaches the Temporal server.
+    #
+    # Implementations are called concurrently and must be thread safe. Methods block, and may perform I/O directly.
+    # To work a batch concurrently, fan out with threads: threads overlap blocking I/O, since Ruby releases its global
+    # lock around it, and they work whether or not the worker is running under a fiber scheduler.
+    # {::Fiber.schedule} must not be used, since it raises when the worker is configured with a payload codec thread
+    # pool instead of a fiber scheduler. {Temporalio::Worker::ThreadPool} can be used to bound that fan out.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!visibility private
+    class StorageDriver
+      # Name of this driver instance, unique among the drivers registered on one {ExternalStorage}. This is written
+      # into history alongside every payload the driver stores and is used to route retrieval back to the same driver.
+      # Renaming a deployed driver makes every payload stored under the old name unretrievable.
+      #
+      # @return [String] Driver instance name.
+      def name
+        raise NotImplementedError
+      end
+
+      # Identifier for this driver implementation, for example +aws.s3driver+. Unlike {#name}, this is identical across
+      # every instance of the same implementation and across SDK languages.
+      #
+      # @return [String] Driver implementation identifier.
+      def type
+        raise NotImplementedError
+      end
+
+      # Store the given payloads.
+      #
+      # @param context [StorageDriverStoreContext] Context for this store operation.
+      # @param payloads [Enumerable<Api::Common::V1::Payload>] Payloads to store. This value should not be mutated.
+      # @return [Array<StorageDriverClaim>] One claim per payload, in the same order.
+      def store(context, payloads)
+        raise NotImplementedError
+      end
+
+      # Retrieve the payloads for the given claims.
+      #
+      # @param context [StorageDriverRetrieveContext] Context for this retrieve operation.
+      # @param claims [Enumerable<StorageDriverClaim>] Claims to retrieve. This value should not be mutated.
+      # @return [Array<Api::Common::V1::Payload>] One payload per claim, in the same order.
+      def retrieve(context, claims)
+        raise NotImplementedError
+      end
+    end
+  end
+end

temporalio/lib/temporalio/converters/storage_driver_activity_info.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+module Temporalio
+  module Converters
+    StorageDriverActivityInfo = Data.define(
+      :namespace,
+      :id,
+      :run_id,
+      :type
+    )
+
+    # Identity of a standalone activity a payload is being stored on behalf of.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!attribute [r] namespace
+    #   @return [String] Activity namespace.
+    # @!attribute [r] id
+    #   @return [String, nil] Activity ID, if known.
+    # @!attribute [r] run_id
+    #   @return [String, nil] Run ID, if known.
+    # @!attribute [r] type
+    #   @return [String, nil] Activity type name, if known.
+    #
+    # @!visibility private
+    class StorageDriverActivityInfo
+      # Create activity target information.
+      #
+      # @param namespace [String] Activity namespace.
+      # @param id [String, nil] Activity ID.
+      # @param run_id [String, nil] Run ID.
+      # @param type [String, nil] Activity type name.
+      def initialize(namespace:, id: nil, run_id: nil, type: nil)
+        super
+      end
+    end
+  end
+end

temporalio/lib/temporalio/converters/storage_driver_claim.rb

@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+module Temporalio
+  module Converters
+    StorageDriverClaim = Data.define(:claim_data)
+
+    # Driver-defined reference to an externally stored payload, used to retrieve it later.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!attribute [r] claim_data
+    #   @return [Hash<String, String>] Data the driver needs to retrieve the payload. This is written into history, so
+    #     it must contain everything required for retrieval and must not contain secrets.
+    #
+    # @!visibility private
+    class StorageDriverClaim
+      # Create a claim.
+      #
+      # @param claim_data [Hash<String, String>] Data identifying the stored payload.
+      def initialize(claim_data:)
+        # Copied and frozen because a claim is compared and hashed by this data, so a later mutation would silently
+        # change its identity. The values are copied individually because freezing the Hash leaves them mutable in
+        # place and still aliased to the caller's objects.
+        # steep:ignore:start
+        super(claim_data: claim_data.transform_values { |value| value.dup.freeze }.freeze)
+        # steep:ignore:end
+      end
+    end
+  end
+end

temporalio/lib/temporalio/converters/storage_driver_retrieve_context.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+require 'temporalio/cancellation'
+
+module Temporalio
+  module Converters
+    StorageDriverRetrieveContext = Data.define(:cancellation)
+
+    # Context given to {StorageDriver#retrieve}.
+    #
+    # This deliberately carries no execution target. A payload must be retrievable from its {StorageDriverClaim}
+    # alone, since the same payload can be read from a different execution than the one that stored it.
+    #
+    # @note WARNING: This API is experimental and may change in the future. Members may be added, so accept keyword
+    #   arguments defensively if constructing this yourself.
+    #
+    # @!attribute [r] cancellation
+    #   @return [Cancellation] Cancelled when the SDK abandons this retrieve operation.
+    #
+    # @!visibility private
+    class StorageDriverRetrieveContext; end # rubocop:disable Lint/EmptyClass
+  end
+end

temporalio/lib/temporalio/converters/storage_driver_select_context.rb

@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+require 'temporalio/cancellation'
+
+module Temporalio
+  module Converters
+    StorageDriverSelectContext = Data.define(
+      :target,
+      :cancellation
+    )
+
+    # Context given to the driver selector.
+    #
+    # @note WARNING: This API is experimental and may change in the future. Members may be added, so accept keyword
+    #   arguments defensively if constructing this yourself.
+    #
+    # @!attribute [r] target
+    #   @return [StorageDriverWorkflowInfo, StorageDriverActivityInfo, nil] The execution the payload is being stored
+    #     on behalf of, or nil when there is no associated execution.
+    # @!attribute [r] cancellation
+    #   @return [Cancellation] Cancelled when the SDK abandons the operation this selection is part of.
+    #
+    # @!visibility private
+    class StorageDriverSelectContext; end # rubocop:disable Lint/EmptyClass
+  end
+end

temporalio/lib/temporalio/converters/storage_driver_store_context.rb

@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+require 'temporalio/cancellation'
+
+module Temporalio
+  module Converters
+    StorageDriverStoreContext = Data.define(
+      :target,
+      :cancellation
+    )
+
+    # Context given to {StorageDriver#store}.
+    #
+    # @note WARNING: This API is experimental and may change in the future. Members may be added, so accept keyword
+    #   arguments defensively if constructing this yourself.
+    #
+    # @!attribute [r] target
+    #   @return [StorageDriverWorkflowInfo, StorageDriverActivityInfo, nil] The execution the payloads are being stored
+    #     on behalf of, or nil when there is no associated execution.
+    # @!attribute [r] cancellation
+    #   @return [Cancellation] Cancelled when the SDK abandons this store operation.
+    #
+    # @!visibility private
+    class StorageDriverStoreContext; end # rubocop:disable Lint/EmptyClass
+  end
+end

temporalio/lib/temporalio/converters/storage_driver_workflow_info.rb

@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+module Temporalio
+  module Converters
+    StorageDriverWorkflowInfo = Data.define(
+      :namespace,
+      :id,
+      :run_id,
+      :type
+    )
+
+    # Identity of the workflow a payload is being stored on behalf of. Also used for workflow activities, which
+    # store against their owning workflow rather than themselves.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!attribute [r] namespace
+    #   @return [String] Workflow namespace.
+    # @!attribute [r] id
+    #   @return [String, nil] Workflow ID, if known.
+    # @!attribute [r] run_id
+    #   @return [String, nil] Run ID, if known. Unset when the run is not yet determined, such as when starting a child
+    #     workflow or continuing as new.
+    # @!attribute [r] type
+    #   @return [String, nil] Workflow type name, if known.
+    #
+    # @!visibility private
+    class StorageDriverWorkflowInfo
+      # Create workflow target information.
+      #
+      # @param namespace [String] Workflow namespace.
+      # @param id [String, nil] Workflow ID.
+      # @param run_id [String, nil] Run ID.
+      # @param type [String, nil] Workflow type name.
+      def initialize(namespace:, id: nil, run_id: nil, type: nil)
+        super
+      end
+    end
+  end
+end

temporalio/lib/temporalio/error.rb

@@ -116,6 +116,23 @@ def initialize(details)
       end
     end
 
+    # Error that occurs when a payload stored in external storage is encountered but no external storage is
+    # configured to retrieve it. This usually means the client or worker reading the payload is configured
+    # differently from the one that wrote it.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!visibility private
+    class ExternalStorageNotConfiguredError < Error
+      # @!visibility private
+      def initialize(
+        message = '[TMPRL1105] Encountered a reference to a payload in external storage, but no external storage ' \
+                  'is configured to retrieve it. Configure external storage with a driver able to retrieve it.'
+      )
+        super
+      end
+    end
+
     # Error raised by a client for a general RPC failure.
     class RPCError < Error
       # @return [Code] Status code for the error.