Class: HastCI::Session

Inherits:
Object
  • Object
show all
Defined in:
lib/hastci/session.rb

Constant Summary collapse

STOP_REASONS = 
i[user_interrupt server_cancelled].freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config:, api_client: nil, ack_worker: nil, task_buffer: nil, poll_interval: DEFAULT_POLL_INTERVAL, buffer_min_size: DEFAULT_BUFFER_MIN_SIZE, buffer_max_size: DEFAULT_BUFFER_MAX_SIZE, sleeper: Kernel.method(:sleep), seeding_poll_interval: DEFAULT_SEEDING_POLL_INTERVAL, error_collector: ErrorCollector.new) ⇒ Session

Returns a new instance of Session.



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/hastci/session.rb', line 31

def initialize(
  config:,
  api_client: nil,
  ack_worker: nil,
  task_buffer: nil,
  poll_interval: DEFAULT_POLL_INTERVAL,
  buffer_min_size: DEFAULT_BUFFER_MIN_SIZE,
  buffer_max_size: DEFAULT_BUFFER_MAX_SIZE,
  sleeper: Kernel.method(:sleep),
  seeding_poll_interval: DEFAULT_SEEDING_POLL_INTERVAL,
  error_collector: ErrorCollector.new
)
  @config = config
  @api_client = api_client || ApiClient.new(config: config)
  @ack_worker = ack_worker
  @task_buffer = task_buffer
  @poll_interval = poll_interval
  @buffer_min_size = buffer_min_size
  @buffer_max_size = buffer_max_size
  @sleeper = sleeper
  @seeding_poll_interval = seeding_poll_interval
  @error_collector = error_collector

  @run_id = nil
  @status = nil
  @role = nil
  @heartbeat = nil

  @stop_reason = nil
  @stop_mutex = Mutex.new
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



63
64
65
 
# File 'lib/hastci/session.rb', line 63

def config
  @config
end

#error_collector ⇒ Object (readonly)

Returns the value of attribute error_collector.



63
64
65
 
# File 'lib/hastci/session.rb', line 63

def error_collector
  @error_collector
end

#role ⇒ Object (readonly)

Returns the value of attribute role.



63
64
65
 
# File 'lib/hastci/session.rb', line 63

def role
  @role
end

#run_id ⇒ Object (readonly)

Returns the value of attribute run_id.



63
64
65
 
# File 'lib/hastci/session.rb', line 63

def run_id
  @run_id
end

#status ⇒ Object (readonly)

Returns the value of attribute status.



63
64
65
 
# File 'lib/hastci/session.rb', line 63

def status
  @status
end

Class Method Details

.run(config:, **options) ⇒ Object 



15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/hastci/session.rb', line 15

def self.run(config:, **options)
  session = new(config: config, **options)

  begin
    yield session
  ensure
    begin
      cleanup_result = session.cleanup
    ensure
      session.shutdown!
    end
  end

  cleanup_result
end

Instance Method Details

#ack(result) ⇒ Object 



183
184
185
186
187
 
# File 'lib/hastci/session.rb', line 183

def ack(result)
  ensure_connected!

  @ack_worker.enqueue(result)
end

#cancelled? ⇒ Boolean

Returns:

  • (Boolean) 


91
92
93
 
# File 'lib/hastci/session.rb', line 91

def cancelled?
  stop_reason == :server_cancelled
end

#cleanup ⇒ Object 



213
214
215
216
217
218
219
220
221
222
 
# File 'lib/hastci/session.rb', line 213

def cleanup
  flush_ok = flush_acks!(timeout: 10)
  HastCI.logger.info("[HastCI] Cleanup: flush_ok=#{flush_ok}")

  {
    flush_ok: flush_ok,
    first_error: first_error,
    stop_reason: stop_reason
  }
end

#connect! ⇒ Object 



107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/hastci/session.rb', line 107

def connect!
  run_state = @api_client.init_run

  @run_id = run_state.fetch(:run_id)
  @status = run_state.fetch(:status)
  @role = run_state.fetch(:role)

  HastCI.logger.info("[HastCI] Connected: run_id=#{@run_id} role=#{@role}")

  @ack_worker ||= AckWorker.new(api_client: @api_client, error_collector: @error_collector)
  @ack_worker.start

  self
end

#disconnect! ⇒ Object 



195
196
197
198
199
 
# File 'lib/hastci/session.rb', line 195

def disconnect!
  @task_buffer&.stop
  @ack_worker&.stop
  @api_client.disconnect!
end

#each_task ⇒ Object 



166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/hastci/session.rb', line 166

def each_task
  return enum_for(:each_task) unless block_given?

  ensure_connected!

  loop do
    return if stopping?

    check_for_fatal_error!

    task = @task_buffer.next_task
    return if task.nil?

    yield task
  end
end

#first_error ⇒ Object 



65
66
67
 
# File 'lib/hastci/session.rb', line 65

def first_error
  @error_collector.first_error
end

#flush_acks!(timeout: 10) ⇒ Object 



189
190
191
192
193
 
# File 'lib/hastci/session.rb', line 189

def flush_acks!(timeout: 10)
  return true unless @ack_worker

  @ack_worker.flush(timeout: timeout)
end

#request_stop!(reason) ⇒ Object

Raises:

  • (ArgumentError) 


69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/hastci/session.rb', line 69

def request_stop!(reason)
  raise ArgumentError, "Invalid stop reason: #{reason}" unless STOP_REASONS.include?(reason)

  should_stop_buffer = false
  @stop_mutex.synchronize do
    return if @stop_reason

    @stop_reason = reason
    should_stop_buffer = user_initiated_stop?(reason)
  end

  @task_buffer&.stop if should_stop_buffer
end

#seed(tasks:) ⇒ Object 



158
159
160
161
162
163
164
 
# File 'lib/hastci/session.rb', line 158

def seed(tasks:)
  ensure_connected!

  HastCI.logger.info("[HastCI] Seeding #{tasks.size} tasks...")
  @api_client.seed(run_id: run_id, tasks: tasks)
  HastCI.logger.info("[HastCI] Seeding complete")
end

#seeder? ⇒ Boolean

Returns:

  • (Boolean) 


154
155
156
 
# File 'lib/hastci/session.rb', line 154

def seeder?
  role == :seeder
end

#shutdown! ⇒ Object 



224
225
226
227
 
# File 'lib/hastci/session.rb', line 224

def shutdown!
  @heartbeat&.stop
  disconnect!
end

#start! ⇒ Object 



95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/hastci/session.rb', line 95

def start!
  connect! unless @run_id
  start_heartbeat!

  unless seeder?
    wait_for_ready!
    return if stopping?
  end

  start_task_buffer!
end

#start_heartbeat! ⇒ Object 



201
202
203
204
205
206
207
208
209
210
211
 
# File 'lib/hastci/session.rb', line 201

def start_heartbeat!
  return if @heartbeat

  @heartbeat = Heartbeat.new(
    api_client: @api_client,
    run_id: @run_id,
    interval: @config.heartbeat_interval,
    error_collector: @error_collector
  )
  @heartbeat.start
end

#start_task_buffer! ⇒ Object 



122
123
124
125
126
127
 
# File 'lib/hastci/session.rb', line 122

def start_task_buffer!
  return if stopping?

  @task_buffer ||= build_task_buffer
  @task_buffer.start
end

#stop_reason ⇒ Object 



87
88
89
 
# File 'lib/hastci/session.rb', line 87

def stop_reason
  @stop_mutex.synchronize { @stop_reason }
end

#stopping? ⇒ Boolean

Returns:

  • (Boolean) 


83
84
85
 
# File 'lib/hastci/session.rb', line 83

def stopping?
  @stop_mutex.synchronize { !@stop_reason.nil? }
end

#wait_for_ready! ⇒ Object 



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/hastci/session.rb', line 129

def wait_for_ready!
  return unless @status == :seeding

  HastCI.logger.info("[HastCI] Waiting for seeding (timeout: #{@config.seeding_timeout}s)...")
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + @config.seeding_timeout

  while @status == :seeding
    return if stopping?

    if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline
      raise FatalApiError, "Timed out waiting for seeding to complete (#{@config.seeding_timeout}s). " \
        "The seeder may have failed."
    end

    run_state = @api_client.run_status(run_id: @run_id)
    @status = run_state.fetch(:status)

    break unless @status == :seeding

    @sleeper.call(@seeding_poll_interval)
  end

  HastCI.logger.info("[HastCI] Seeding complete, status=#{@status}")
end