Class: Raif::ArchiveModelCompletionsJob

Inherits:
ApplicationJob
  • Object
show all
Defined in:
app/jobs/raif/archive_model_completions_job.rb

Overview

Archives Raif::ModelCompletion rows older than Raif.config.model_completion_retention_period to Raif.config.archive_storage as gzip JSONL (see Raif::ArchiveSerializer), then deletes them, one Raif::Archive batch at a time. Enabling, scheduling, storage requirements and partitioning are documented for operators at https://docs.raif.ai/learn_more/archiving.

The invariants this job exists to hold:

  • At-least-once archiving + never-delete-unarchived. A batch is deleted only after this run uploaded it, and a re-run after any crash writes a new object under a new key rather than resuming or overwriting. There is no persisted in-flight state to repair, ever.
  • The accepted cost of that is duplicate objects: a crash between the upload and the Raif::Archive insert leaves an object that no row references (those rows were not deleted, so they re-archive next run). Storage policy must therefore apply to whole prefixes and never be derived from raif_archives rows, which is also what makes Raif::Archive.purge_partition! complete.
  • Inert unless the host opts in: archive_enabled, an archive_storage adapter and a retention period must all be set, or perform returns without touching anything.
  • With partitioning, every object holds records from exactly one partition (see Raif::ArchivePartition). Per-partition erasure rests on that, so a record whose partition changes mid-upload taints the object and aborts the cull.

Deliberately NOT handled: raif_model_completion_batches rows remain; they carry their own aggregated cost columns and nothing recomputes them from children after finalization.

Constant Summary collapse

BATCH_RECORD_LIMIT =

Caps are constants, not config: public config is API; add a knob only when a host demonstrably needs it. A batch closes when EITHER cap is hit; the byte cap keeps wildly varying completion payloads (agent message arrays can be enormous) from building a multi-GB tempfile.

25_000
BATCH_UNCOMPRESSED_BYTE_LIMIT = 
512.megabytes
QUIESCENCE_PERIOD =

Rows updated more recently than this are ineligible for archiving, regardless of age (legitimately active months-old completions can't exist since batch lifetime is capped, but the guard is cheap insurance).

24.hours
KEY_PREFIX =

Key prefix when partitioning is unset.

"raif-archives/model-completions"
RESOURCE_KEY_SEGMENT =

Resource segment below a partition prefix: raif-archives/partitions//model-completions/

"model-completions"
PARTITIONS_PER_PASS =

Distinct partitions fetched per round-robin pass. The listing query is a GROUP BY over the full multi-guard eligibility scope, the one potentially expensive query in the fairness design, so it is capped.

100

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.dry_run(cutoff: Raif.config.model_completion_retention_period&.ago) ⇒ Object

Counts what a run under the given cutoff would archive (split by terminal state), plus the per-guard exclusions among cutoff-aged completions. Writes nothing; re-runnable anytime. Operators run this before enabling archiving (defaults to the configured retention period so it can be previewed while archive_enabled is still false).

Raises:

  • (ArgumentError) 


102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'app/jobs/raif/archive_model_completions_job.rb', line 102

def dry_run(cutoff: Raif.config.model_completion_retention_period&.ago)
  raise ArgumentError, "Provide a cutoff: or set Raif.config.model_completion_retention_period" if cutoff.nil?

  validate_partition_column!

  eligible = eligible_scope(cutoff)

  result = {
    cutoff: cutoff,
    eligible: eligible.count,
    eligible_terminal: eligible.where("completed_at IS NOT NULL OR failed_at IS NOT NULL").count,
    eligible_nonterminal: eligible.where(completed_at: nil, failed_at: nil).count,
    excluded_by_quiescence: base_scope(cutoff).where(completions_table[:updated_at].gteq(QUIESCENCE_PERIOD.ago)).count,
    excluded_by_active_batch: base_scope(cutoff).where(id: active_batch_members).count,
    excluded_missing_cost_event: terminal_scope(cutoff).where.not(id: completions_with_cost_event).count,
    excluded_stale_cost_event: terminal_scope(cutoff)
      .where(id: completions_with_cost_event)
      .where.not(id: completions_with_fresh_cost_event).count,
    excluded_uncopied_citations: base_scope(cutoff).where(id: completions_with_uncopied_citations).count
  }

  add_partition_report!(result, eligible)
  result
end

.eligible_scope(cutoff) ⇒ Object

A completion is safe to archive and delete only when ALL hold:

  • created_at is before the (job-frozen) retention cutoff
  • it has been quiescent: not updated within QUIESCENCE_PERIOD
  • it is not a member of a model completion batch that is still non-terminal (belt-and-suspenders alongside quiescence)
  • durability guard, TERMINAL rows only: its Raif::InferenceCostEvent exists AND is at least as fresh as the completion (event.updated_at >= completion.updated_at). A post-terminal update whose event re-sync failed leaves a stale event that missing-only repair would never revisit, so the repair job also re-syncs stale events; until then the row just waits.
  • nonterminal rows skip the durability guard: they never reached a terminal state, so no cost event exists and there is no spend to protect. These are orphaned pending rows from killed processes and crashed jobs (a third of one host's table in practice) that would otherwise be immortal. They are archived through the same path as everything else - NOT deleted outright, despite the temptation (no response, near-zero historical value): "every deleted completion exists in an archive" must hold without exception, and a delete-without-archive shortcut would be a second deletion semantics that weakens the invariant this job's safety rests on, to save pennies of mostly-redundant prompt storage.
  • durable-citations guard: its citations, if any, have been copied to its Raif::ConversationEntry source (protects hosts that haven't run the conversation entry backfill)


89
90
91
92
93
94
95
 
# File 'app/jobs/raif/archive_model_completions_job.rb', line 89

def eligible_scope(cutoff)
  base_scope(cutoff)
    .where(completions_table[:updated_at].lt(QUIESCENCE_PERIOD.ago))
    .where.not(id: active_batch_members)
    .where.not(id: terminal_without_fresh_cost_event(cutoff))
    .where.not(id: completions_with_uncopied_citations)
end

.partition_column ⇒ Object 



127
128
129
 
# File 'app/jobs/raif/archive_model_completions_job.rb', line 127

def partition_column
  Raif.config.archive_partition_column
end

.ungrouped_fallback? ⇒ Boolean

Returns:

  • (Boolean) 


131
132
133
 
# File 'app/jobs/raif/archive_model_completions_job.rb', line 131

def ungrouped_fallback?
  Raif.config.archive_partition_fallback.equal?(Raif::ArchivePartition::UNGROUPED)
end

.validate_partition_column! ⇒ Object

Column existence is validated here, at job/dry_run execution time, never at boot: Raif::Configuration#validate! is deliberately DB-free so blank-database boots (db:create, db:migrate, asset precompile) keep working. Each archived resource's job validates the column on its own resource.

Raises:



140
141
142
143
144
145
146
147
 
# File 'app/jobs/raif/archive_model_completions_job.rb', line 140

def validate_partition_column!
  return if partition_column.nil?
  return if Raif::ModelCompletion.column_names.include?(partition_column.to_s)

  raise Raif::Errors::InvalidConfigError,
    "Raif.config.archive_partition_column is :#{partition_column}, but #{Raif::ModelCompletion.table_name} " \
      "has no #{partition_column} column"
end

Instance Method Details

#perform(max_records: 100_000, max_objects: 500, max_runtime: 30.minutes) ⇒ Object 



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
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
 
# File 'app/jobs/raif/archive_model_completions_job.rb', line 257

def perform(max_records: 100_000, max_objects: 500, max_runtime: 30.minutes)
  return unless Raif.config.archive_enabled
  return if Raif.config.archive_storage.nil?
  return if Raif.config.model_completion_retention_period.nil?

  # Defense in depth: Raif::Configuration#validate! enforces this floor
  # at boot, but a destructive job must not trust that validation ran
  # (initializers can be skipped or misordered on a misconfigured node).
  # Cost/budget consumers aggregate by billing period, so a tiny
  # retention value must never be able to cull inside an open window.
  if Raif.config.model_completion_retention_period < 1.month
    raise Raif::Errors::InvalidConfigError,
      "Raif.config.model_completion_retention_period must be at least 1 month (got #{Raif.config.model_completion_retention_period.inspect})"
  end

  self.class.validate_partition_column!

  # Shared with Raif::Archive.purge_partition!; a second concurrent
  # perform (or a purge in progress) makes this run return immediately.
  Raif::ArchiveAdvisoryLock.acquire do
    # Frozen at job start so every batch in this run shares one cutoff.
    cutoff = Raif.config.model_completion_retention_period.ago
    deadline = monotonic_now + max_runtime.to_f
    records_remaining = max_records
    objects_remaining = max_objects

    # Round-robin passes: at most one object per partition per pass, so
    # many small partitions all drain within a run while one
    # large-backlog partition cannot monopolize it. Each pass excludes
    # the partitions already visited this round, paging through EVERY
    # eligible partition before any is revisited (the listing cap alone
    # would relist the same oldest cohort while it still holds the
    # oldest rows, starving partitions beyond the cap). A pass that
    # lists nothing ends the round; a new round starts only if the
    # finished round culled something. With partitioning unset there is
    # exactly one pseudo-partition and this reduces to sequential
    # batching. The runtime budget is only checked between objects: a
    # started object always finishes, so a run always stops in a safe
    # state.
    visited = []
    round_culled = false

    loop do
      selections = partition_selections(cutoff, visited)

      if selections.empty?
        break unless round_culled

        visited = []
        round_culled = false
        next
      end

      selections.each do |partition, predicate, raws|
        break if records_remaining <= 0 || objects_remaining <= 0 || monotonic_now >= deadline

        # Marked visited even when nothing archives, or an emptied
        # partition would relist in every pass for the rest of the round.
        visited.concat(raws)

        ids = eligible_batch_ids(cutoff, partition, predicate, limit: [BATCH_RECORD_LIMIT, records_remaining].min)
        next if ids.empty?

        outcome = archive_batch!(ids, cutoff, partition: partition, partition_predicate: predicate)
        next if outcome.nil?

        records_remaining -= outcome[:records]
        objects_remaining -= 1
        # A tainted batch (partition mutated during upload, object
        # cleaned up) spends budget but is not progress; the round
        # moves on to other partitions.
        round_culled ||= outcome[:culled]
      end

      break if records_remaining <= 0 || objects_remaining <= 0 || monotonic_now >= deadline
    end
  end
end