Class: CDC::Parallel::Runtime

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

Overview

Note:

Runtime is an execution facade, not a source adapter. It expects work that has already been normalized into cdc-core primitives.

High-level Ractor runtime facade for cdc-core processors.

Runtime is the primary public entry point for applications that want to execute normalized CDC work items with cdc-parallel. It wires together a ProcessorPool, a TransactionPool, and a Router so callers can submit either a single CDC::Core::ChangeEvent or a CDC::Core::TransactionEnvelope through one object.

Use this class when you want the default cdc-parallel behavior:

  • validate that the processor declared ractor_safe!
  • boot a fixed set of worker Ractors
  • route events and transaction envelopes to the right pool
  • return CDC::Core::ProcessorResult objects
  • shut down all worker resources together

Examples:

Processing a change event

runtime = CDC::Parallel::Runtime.new(
  processor: AnalyticsProcessor.new,
  size: 4,
  timeout: 5
)

result = runtime.process(change_event)
result.success? #=> true

runtime.shutdown

Processing a transaction envelope

result = runtime.process_transaction(transaction)

See Also:

Instance Method Summary collapse

Constructor Details

#initialize(processor:, size: Etc.nprocessors, timeout: nil, supervision: true, max_respawns: 3, respawn_window: 60, respawn_cooldown: 5) ⇒ void

Create a runtime with event and transaction pools.

Parameters:

  • processor (CDC::Core::Processor) —

    Ractor-safe processor used for both event and transaction processing.

  • size (Integer) (defaults to: Etc.nprocessors) —

    Number of worker Ractors per internal pool.

  • timeout (Numeric, nil) (defaults to: nil) —

    Optional timeout in seconds for result collection and shutdown waits.

  • supervision (Boolean) (defaults to: true) —

    Whether worker Ractors should be respawned after unexpected death.

  • max_respawns (Integer) (defaults to: 3) —

    Maximum crash count inside respawn_window before a worker slot enters cooldown.

  • respawn_window (Numeric) (defaults to: 60) —

    Time window, in seconds, used by the crash-loop circuit breaker.

  • respawn_cooldown (Numeric) (defaults to: 5) —

    Cooldown, in seconds, before a crash-looping slot is revived again.

Raises:

  • (UnsafeProcessorError) —

    Raised when the processor class has not declared ractor_safe!.

  • (ArgumentError) —

    Raised when size or timeout is invalid.



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/cdc/parallel/runtime.rb', line 65

def initialize(
  processor:,
  size: Etc.nprocessors,
  timeout: nil,
  supervision: true,
  max_respawns: 3,
  respawn_window: 60,
  respawn_cooldown: 5
)
  @processor = processor
  @processor.start

  pool_options = build_pool_options(
    processor:, size:, timeout:, supervision:,
    max_respawns:, respawn_window:, respawn_cooldown:
  )

  @processor_pool = ProcessorPool.new(**pool_options)
  @transaction_pool = TransactionPool.new(**pool_options)
  @router = Router.new(processor_pool: @processor_pool, transaction_pool: @transaction_pool)
  @shutdown = false
end

Instance Method Details

#process(item) ⇒ CDC::Core::ProcessorResult

Process a supported normalized CDC work item.

Supported items are CDC::Core::ChangeEvent and CDC::Core::TransactionEnvelope. Unsupported objects raise UnsupportedWorkItemError from the router.

Parameters:

  • item (CDC::Core::ChangeEvent, CDC::Core::TransactionEnvelope) —

    Normalized CDC work item.

Returns:

  • (CDC::Core::ProcessorResult)

Raises:



101
102
103
104
105
 
# File 'lib/cdc/parallel/runtime.rb', line 101

def process(item)
  raise ShutdownError, "runtime has been shut down" if @shutdown

  @router.process(item)
end

#process_transaction(transaction) ⇒ CDC::Core::ProcessorResult

Process a transaction envelope.

This method is a readability alias for transaction-oriented call sites. It delegates to #process, so it has the same validation, shutdown, and result behavior.

Parameters:

  • transaction (CDC::Core::TransactionEnvelope)

Returns:

  • (CDC::Core::ProcessorResult) 


115
116
117
 
# File 'lib/cdc/parallel/runtime.rb', line 115

def process_transaction(transaction)
  process(transaction)
end

#shutdown ⇒ void

This method returns an undefined value.

Shut down all runtime resources.

Shutdown is idempotent and cascades to the internal event and transaction pools. The processor's flush and stop lifecycle hooks are called once after all pools have been drained. After shutdown, #process raises ShutdownError.



127
128
129
130
131
132
133
134
135
 
# File 'lib/cdc/parallel/runtime.rb', line 127

def shutdown
  return if @shutdown

  @shutdown = true
  @processor_pool.shutdown
  @transaction_pool.shutdown
  @processor.flush
  @processor.stop
end