Class: CDC::Parallel::ResultCollector

Inherits:
Object
  • Object
show all
Defined in:
lib/cdc/parallel/result_collector.rb

Overview

Converts raw worker responses into ‘CDC::Core::ProcessorResult` objects.

Ractors cannot freely share arbitrary mutable Ruby objects. Worker responses must therefore be normalized into shareable payloads before they cross back to the caller. ‘ResultCollector` owns that small translation boundary.

The worker side uses ResultCollector.worker_success and ResultCollector.worker_failure. The caller side uses ResultCollector.normalize to convert those payloads into the public result contract.

Examples:

Normalizing a processor return value

response = CDC::Parallel::ResultCollector.worker_success(value)
result = CDC::Parallel::ResultCollector.normalize(response)

Normalizing a worker exception

response = CDC::Parallel::ResultCollector.worker_failure(error)
result = CDC::Parallel::ResultCollector.normalize(response)
result.failure? #=> true

Constant Summary collapse

FAILURE_MARKER =

Internal marker used to identify serialized worker failures.

Returns:

  • (Symbol) 
:__cdc_parallel_failure__

Class Method Summary collapse

Class Method Details

.normalize(value) ⇒ CDC::Core::ProcessorResult

Normalize a worker return value into a ‘CDC::Core::ProcessorResult`.

Failure payloads become failed processor results containing a ProcessorExecutionError. Existing processor results are returned unchanged. Other values are wrapped in a successful processor result.

Parameters:

  • value (Object) —

    Raw worker response.

Returns:

  • (CDC::Core::ProcessorResult) 


77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/cdc/parallel/result_collector.rb', line 77

def self.normalize(value)
  if worker_failure?(value)
    CDC::Core::ProcessorResult.failure(
      ProcessorExecutionError.new(
        original_class: value[:class],
        original_message: value[:message],
        original_backtrace: value[:backtrace]
      )
    )
  elsif value.is_a?(CDC::Core::ProcessorResult)
    value
  else
    CDC::Core::ProcessorResult.success(value)
  end
end

.worker_failure(error) ⇒ Hash

Build a shareable failure payload that can safely cross a Ractor boundary.

Exceptions themselves are not used as the cross-Ractor payload. Instead, the class name, message, and backtrace are serialized into a simple hash that can be reconstructed as a ProcessorExecutionError by normalize.

Parameters:

  • error (Exception) —

    Exception raised inside a worker Ractor.

Returns:

  • (Hash) —

    Shareable serialized failure payload.



57
58
59
60
61
62
63
64
65
66
 
# File 'lib/cdc/parallel/result_collector.rb', line 57

def self.worker_failure(error)
  ::Ractor.make_shareable(
    {
      type: FAILURE_MARKER,
      class: error.class.name,
      message: error.message,
      backtrace: (error.backtrace || []).map { |line| String(line) }.freeze
    }
  )
end

.worker_success(value) ⇒ Object

Build a shareable success payload that can safely cross a Ractor boundary.

If the processor already returned a ‘CDC::Core::ProcessorResult`, that result is preserved. Any other shareable value will later be wrapped in a success result by normalize.

Parameters:

  • value (Object) —

    Processor return value.

Returns:

  • (Object) —

    Shareable success payload.

Raises:

  • (Ractor::Error) —

    Raised by Ruby when the value cannot be made shareable.



43
44
45
 
# File 'lib/cdc/parallel/result_collector.rb', line 43

def self.worker_success(value)
  ::Ractor.make_shareable(value)
end