Class: Batchwatch::BatchJob

Inherits:
Object
  • Object
show all
Defined in:
lib/batchwatch/job.rb

Overview

One high-level batch job: deadline guard (#178), poll loop (#179), partial completion (#180).

Created by Batchwatch::Client#batch. result blocks, polling with backoff and jitter until the batch is terminal or the deadline fires. For an async / worker context that does not want a blocked thread, poll_once drives the same state machine one step at a time.

The clock / sleep / rng seams exist for deterministic tests: the backoff schedule and its jitter are pure functions of these, so a fake clock and a fake rng pin the exact poll count with no real sleeps. Production leaves them as the real time / random.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(bw, model, deadline: nil, on_deadline: nil, provider: "openai", poll: nil, cancel: nil, result_of: nil, succeeded: nil, expired: nil, poll_base: nil, poll_floor: nil, poll_ceiling: POLL_CEILING_S, poll_backoff: POLL_BACKOFF, poll_jitter: POLL_JITTER, use_p50_cadence: true, poll_interval: nil, clock: nil, sleep: nil, rng: nil) ⇒ BatchJob

Returns a new instance of BatchJob.



234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
 
# File 'lib/batchwatch/job.rb', line 234

def initialize(bw, model, deadline: nil, on_deadline: nil,
               provider: "openai", poll: nil, cancel: nil, result_of: nil,
               succeeded: nil, expired: nil,
               poll_base: nil, poll_floor: nil,
               poll_ceiling: POLL_CEILING_S, poll_backoff: POLL_BACKOFF,
               poll_jitter: POLL_JITTER, use_p50_cadence: true,
               poll_interval: nil, clock: nil, sleep: nil, rng: nil)
  @bw = bw
  @model = model
  @provider = provider
  @on_deadline = on_deadline
  @deadline_s = Batchwatch.seconds(deadline)
  @poll = poll || Batchwatch.method(:default_poll)
  @cancel = cancel || Batchwatch.method(:default_cancel)
  @result_of = result_of || Batchwatch.method(:default_result)
  @succeeded = succeeded || Batchwatch.method(:default_succeeded)
  @expired = expired || Batchwatch.method(:default_expired)

  # poll_interval is #178's name for the (then-fixed) poll cadence. #179
  # renamed it to poll_base (the FIRST interval before backoff) and split out
  # an explicit poll_floor (the rate-limit guard). We honour the old name as
  # an alias so #178's two-callable usage - and its tests, which set a tiny
  # interval to make the deadline fire fast - keep working untouched (epic
  # rule #4). When only poll_interval is given, a small value is also taken
  # as the floor so the deadline-guard tests still poll as fast as they
  # asked, not clamped up to the 5s default.
  if poll_base.nil?
    poll_base = poll_interval.nil? ? POLL_BASE_S : poll_interval
  end
  if poll_floor.nil?
    poll_floor = poll_interval.nil? ? POLL_FLOOR_S : [poll_interval, POLL_FLOOR_S].min
  end

  # Poll cadence (#179). The floor and ceiling are the two guard rails; the
  # base is stored raw because EVERY interval that reaches a sleep goes
  # through clamp, which is the ONE place the floor and ceiling are applied
  # (P7: one canonical site, so the rate-limit floor cannot be
  # half-removed). A base below the floor therefore still polls at the floor.
  @poll_floor = [0.001, poll_floor.to_f].max
  @poll_ceiling = [@poll_floor, poll_ceiling.to_f].max
  @poll_base = [0.0, poll_base.to_f].max
  @poll_backoff = [1.0, poll_backoff.to_f].max
  @poll_jitter = [[poll_jitter.to_f, 1.0].min, 0.0].max
  @use_p50_cadence = use_p50_cadence

  # Injectable seams for deterministic tests; real time/random otherwise.
  @clock = clock || -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) }
  @sleep = sleep || ->(s) { Kernel.sleep(s) }
  @rand = rng || -> { Kernel.rand }

  @handle = nil          # the provider's batch object
  @submitted_at = nil    # monotonic clock at submit
  @result = nil
  @fell_back = false     # did the deadline guard fire?
  @expired_flag = false  # did the batch hit the 24h expiry?
  @resolved = false      # has result run to completion?
  @interval = nil        # current backoff interval, lazily seeded
  @poll_count = 0        # how many times we asked "are you done?"
  @retries = nil         # lazy per-job retry registry (#180)
end

Instance Attribute Details

#handle ⇒ Object (readonly)

Returns the value of attribute handle.



295
296
297
 
# File 'lib/batchwatch/job.rb', line 295

def handle
  @handle
end

Instance Method Details

#expired ⇒ Object

true once the batch was seen to reach the 24h expiry. A distinct terminal state, never folded into a timeout.



390
391
392
 
# File 'lib/batchwatch/job.rb', line 390

def expired
  @expired_flag
end

#fell_back ⇒ Object

true once the deadline guard has fired and the fallback was used.



395
396
397
 
# File 'lib/batchwatch/job.rb', line 395

def fell_back
  @fell_back
end

#next_interval ⇒ Object

Seconds to sleep before the next poll_once, jitter applied. Already includes the rate-limit floor and the jitter, so an async driver can sleep exactly this and match the blocking loop's cadence.



376
377
378
379
 
# File 'lib/batchwatch/job.rb', line 376

def next_interval
  @interval = seed_interval if @interval.nil?
  sleep_for
end

#poll_count ⇒ Object

How many times we have asked the batch whether it is done. This is the number that gets someone rate-limited, so it is exposed for a caller - and a test - to assert a bound on.



384
385
386
 
# File 'lib/batchwatch/job.rb', line 384

def poll_count
  @poll_count
end

#poll_once ⇒ Object

Take one non-blocking step of the poll loop and report the state.

The non-blocking variant (#179): an async or worker context can drive the wait itself instead of handing us a thread. Call it, read the returned PollState, and - if it is RUNNING - sleep next_interval (which we have already advanced) before the next call. result is still the way to get the value once a step returns a terminal state; this only reports where the batch is.

Unlike result this never sleeps and never falls back - it is a single question. The deadline is still honoured by result's own loop.



361
362
363
364
365
366
367
368
369
370
371
 
# File 'lib/batchwatch/job.rb', line 361

def poll_once
  if @handle.nil?
    raise BatchJobError,
          "batchwatch: poll_once() before submit() - call " \
          "job.submit(-> { client.batches.create(...) }) first"
  end
  @interval = seed_interval if @interval.nil?
  state = poll_state
  @interval = advance_interval(@interval) if state == PollState::RUNNING
  state
end

#result ⇒ Object

Block until the batch finishes, or the deadline fires.

Returns the batch result if it completed in time, otherwise the result of on_deadline. Idempotent: a second call returns the same result without re-waiting or re-reporting.

Raises BatchJobError if called before submit.



330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
 
# File 'lib/batchwatch/job.rb', line 330

def result
  return @result if @resolved

  if @handle.nil?
    raise BatchJobError,
          "batchwatch: result() before submit() - call " \
          "job.submit(-> { client.batches.create(...) }) first"
  end

  completed_in_time = wait_loop
  if completed_in_time && @succeeded.call(@handle)
    @result = @result_of.call(@handle)
    report_outcome(fell_back: false)
  else
    @result = fallback(timed_out: !completed_in_time)
  end
  @resolved = true
  @result
end

#retry_failed(resubmit, result: nil) ⇒ Object

Resubmit ONLY the failed subset, once, with an idempotency key.

resubmit is a callable the caller hands us - resubmit.call(failed_ids) - that recreates a batch over just those ids. Same epic rule: we take the callable, never the payload. We call it with the list of failed custom_ids and remember the handle it returns as a fresh child job.

Idempotent by construction: calling it twice submits ONE retry. Duplicating a 20,000-request job is an expensive failure, so the key is the sorted set of failed ids - a pure function of WHAT is being retried, not of when. A second call with the same failure set returns the same child without resubmitting; a genuinely different failure set is a different retry and does submit.

Returns the child BatchJob, or nil when there is nothing to retry (an empty failure set is a no-op, not an empty batch).



440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
 
# File 'lib/batchwatch/job.rb', line 440

def retry_failed(resubmit, result: nil)
  res = result.nil? ? split : result
  failed_ids = res.failed_ids.sort
  return nil if failed_ids.empty?

  @retries ||= {}
  key = failed_ids
  return @retries[key] if @retries.key?(key)

  child = @bw.batch(@model, deadline: nil, on_deadline: @on_deadline,
                            provider: @provider)
  child.submit(-> { resubmit.call(failed_ids.dup) })
  @retries[key] = child
  child
end

#split(results = nil, custom_ids = nil) ⇒ Object

Split the batch into landed / failed / expired, mapped by custom_id.

A real batch is not binary: a job of 20,000 requests can come back with some landed, some failed per-request, and some never returned before the 24h expiry. This is where silent data loss lives - a naive "the job completed" reads all three as success. BatchResult separates them, with counts and the per-request ids, so the caller can act.

results is the per-request outcomes the caller downloaded from THEIR provider (we never fetch content). Each item is a mapping carrying a custom_id and either a success or an error - the shape the OpenAI / Anthropic batch output lines already have. custom_ids is the full set the caller SUBMITTED; any id in it that has no result line is expired (outstanding at the 24h cutoff). Mapping is by custom_id, never by index: provider result ordering is not guaranteed.

Pass neither and we read the whole-batch terminal status off the handle instead - a whole-batch expiry becomes an all-expired result, a whole-batch failure an all-failed one.



420
421
422
 
# File 'lib/batchwatch/job.rb', line 420

def split(results = nil, custom_ids = nil)
  BatchResult.from_results(self, results, custom_ids)
end

#submit(create = nil, &block) ⇒ Object

Run the caller's batch-create callable and remember the handle.

create is a zero-argument callable (a block, proc, or lambda) that returns the provider's batch object - typically -> { client.batches.create(...) }. We call it, store what it returns, and start the deadline clock. We never inspect the payload it built; the handle is opaque to us except for the questions the poll/result callables ask.

Raising from create is the caller's own error and is left untouched - the batch never started, so there is nothing for us to guard.

Raises:



309
310
311
312
313
314
315
316
317
318
319
320
321
 
# File 'lib/batchwatch/job.rb', line 309

def submit(create = nil, &block)
  create ||= block
  raise BatchJobError, "batchwatch: submit() needs a callable" if create.nil?

  unless @handle.nil?
    raise BatchJobError,
          "batchwatch: this job was already submitted - create a new " \
          "bw.batch(...) for a second batch"
  end
  @handle = create.call
  @submitted_at = @clock.call
  @handle
end

#with {|_self| ... } ⇒ Object

A block form so the partial-completion cleanup has an obvious home; the base job has nothing to do on the way out. Yields self and returns the block's value. We do NOT force a result the caller never asked for.

Yields:

  • (_self)

Yield Parameters:



461
462
463
 
# File 'lib/batchwatch/job.rb', line 461

def with
  yield self
end