Class: Shikibu::ReplayEngine

Inherits:
Object
  • Object
show all
Defined in:
lib/shikibu/replay.rb

Overview

Orchestrates workflow execution with deterministic replay

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(storage:, worker_id:, hooks: nil) ⇒ ReplayEngine

Returns a new instance of ReplayEngine.



8
9
10
11
12
 
# File 'lib/shikibu/replay.rb', line 8

def initialize(storage:, worker_id:, hooks: nil)
  @storage = storage
  @worker_id = worker_id
  @hooks = hooks
end

Instance Attribute Details

#hooks ⇒ Object (readonly)

Returns the value of attribute hooks.



6
7
8
 
# File 'lib/shikibu/replay.rb', line 6

def hooks
  @hooks
end

#storage ⇒ Object (readonly)

Returns the value of attribute storage.



6
7
8
 
# File 'lib/shikibu/replay.rb', line 6

def storage
  @storage
end

#worker_id ⇒ Object (readonly)

Returns the value of attribute worker_id.



6
7
8
 
# File 'lib/shikibu/replay.rb', line 6

def worker_id
  @worker_id
end

Instance Method Details

#execute_compensation_from_registry(instance_id, comp) ⇒ Object

Execute a single compensation from registry

Parameters:

  • instance_id (String) —

    Instance ID

  • comp (Hash) —

    Compensation record from DB



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/shikibu/replay.rb', line 116

def execute_compensation_from_registry(instance_id, comp)
  compensation_fn = Shikibu.get_compensation(comp[:activity_name])

  if compensation_fn.nil?
    # Inline block or unregistered compensation - cannot recover, skip with warning
    record_compensation_skipped(instance_id, comp)
    return
  end

  # Execute the compensation
  args = comp[:args] || {}
  symbolized_args = args.transform_keys(&:to_sym)
  compensation_fn.call(nil, **symbolized_args)

  # Record success
  storage.append_history(
    instance_id: instance_id,
    activity_id: "compensation:#{comp[:id]}",
    event_type: EventType::COMPENSATION_EXECUTED,
    event_data: {
      compensation_id: comp[:id],
      activity_id: comp[:activity_id],
      activity_name: comp[:activity_name]
    }
  )
rescue StandardError => e
  # Record failure but continue
  storage.append_history(
    instance_id: instance_id,
    activity_id: "compensation:#{comp[:id]}",
    event_type: EventType::COMPENSATION_FAILED,
    event_data: {
      compensation_id: comp[:id],
      activity_id: comp[:activity_id],
      activity_name: comp[:activity_name],
      error_type: e.class.name,
      error_message: e.message
    }
  )
end

#execute_compensations_from_db(instance_id) ⇒ Object

Execute compensations from DB (for crash recovery) Uses global registry to find compensation functions

Parameters:

  • instance_id (String) —

    Instance ID



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/shikibu/replay.rb', line 94

def execute_compensations_from_db(instance_id)
  compensations = storage.get_compensations(instance_id)

  # Get already executed compensation IDs from history (idempotency)
  history = storage.get_history(instance_id)
  executed_ids = history
                 .select { |e| e[:event_type] == EventType::COMPENSATION_EXECUTED }
                 .map { |e| e[:data]&.dig(:compensation_id) }
                 .compact
                 .to_set

  # Execute each compensation in order (already LIFO from DB)
  compensations.each do |comp|
    next if executed_ids.include?(comp[:id])

    execute_compensation_from_registry(instance_id, comp)
  end
end

#record_compensation_skipped(instance_id, comp) ⇒ Object

Record that a compensation was skipped (inline or unregistered)



158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'lib/shikibu/replay.rb', line 158

def record_compensation_skipped(instance_id, comp)
  storage.append_history(
    instance_id: instance_id,
    activity_id: "compensation:#{comp[:id]}",
    event_type: EventType::COMPENSATION_FAILED,
    event_data: {
      compensation_id: comp[:id],
      activity_id: comp[:activity_id],
      activity_name: comp[:activity_name],
      error_type: 'CompensationNotFound',
      error_message: "Compensation '#{comp[:activity_name]}' not found in registry " \
                     '(inline blocks cannot be recovered after crash)'
    }
  )
end

#resume_compensating_workflow(instance_id) ⇒ nil

Resume a workflow that was in compensating state when it crashed This executes remaining compensations from DB using the global registry

Parameters:

  • instance_id (String) —

    Instance ID

Returns:

  • (nil)

Raises:



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/shikibu/replay.rb', line 74

def resume_compensating_workflow(instance_id)
  # Acquire lock
  raise LockNotAcquiredError, instance_id unless storage.try_acquire_lock(instance_id, worker_id, timeout: 300)

  begin
    execute_compensations_from_db(instance_id)

    # Clear compensations and update status
    storage.clear_compensations(instance_id)
    storage.update_instance_status(instance_id, Status::FAILED)

    nil
  ensure
    storage.release_lock(instance_id, worker_id)
  end
end

#resume_workflow(instance_id) ⇒ Object?

Resume a workflow from its current state

Parameters:

  • instance_id (String) —

    Instance ID

Returns:

  • (Object, nil) —

    Workflow result or nil if suspended

Raises:



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/shikibu/replay.rb', line 47

def resume_workflow(instance_id)
  instance = storage.get_instance(instance_id)
  raise WorkflowNotFoundError, instance_id unless instance

  # Handle crash recovery for compensating workflows (Romancy/Edda compatible)
  return resume_compensating_workflow(instance_id) if instance[:status] == Status::COMPENSATING

  workflow_class = Shikibu.get_workflow(instance[:workflow_name])
  raise WorkflowNotRegisteredError, instance[:workflow_name] unless workflow_class

  # Load history and build cache
  history = storage.get_history(instance_id)
  history_cache = build_history_cache(history)

  execute_workflow(
    instance_id,
    workflow_class,
    instance[:input_data],
    replaying: true,
    history_cache: history_cache
  )
end

#start_workflow(workflow_class, instance_id:, **input) ⇒ Object?

Start a new workflow instance

Parameters:

  • workflow_class (Class) —

    Workflow class

  • instance_id (String) —

    Instance ID

  • input (Hash, Object) —

    Input parameters (Hash for untyped, typed object for typed workflows)

Returns:

  • (Object, nil) —

    Workflow result or nil if suspended



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/shikibu/replay.rb', line 19

def start_workflow(workflow_class, instance_id:, **input)
  # Serialize input for storage (handles typed inputs)
  serialized_input = workflow_class.serialize_input(input)

  # Save workflow definition
  storage.save_workflow_definition(
    workflow_name: workflow_class.workflow_name,
    source_hash: workflow_class.source_hash,
    source_code: workflow_class.source_code
  )

  # Create instance record
  storage.create_instance(
    instance_id: instance_id,
    workflow_name: workflow_class.workflow_name,
    source_hash: workflow_class.source_hash,
    owner_service: 'default',
    input_data: serialized_input,
    status: Status::RUNNING
  )

  # Execute the workflow
  execute_workflow(instance_id, workflow_class, serialized_input, replaying: false)
end