Class: TextUnifiedEmbeddingBackfillWorker

Inherits:
Object
  • Object
show all
Includes:
Sidekiq::IterableJob, Sidekiq::Job
Defined in:
app/workers/text_unified_embedding_backfill_worker.rb

Overview

Backfills Gemini Embedding 2 vectors into the unified vector space for TEXT
content, so text and images become cross-modally searchable from one query.

Dual-write, no delete: for each safe source content_type='primary' text row it
upserts a content_type='unified' sibling tagged gemini-embedding-2 (via
Embedding::TextUnifier). The candidate anti-join recognizes both valid live
shapes — a single unified row or unified_chunk_N rows — and correlates
them by embeddable type, id, and locale.

The legacy set is expected to be complete, so the nightly run should normally
be a no-op. It remains a bounded safety net for a stray legacy primary row.
Chunked records are never converted here; their exact shape is validated and
repaired by EmbeddingRefreshWorker through the normal EmbeddingWorker path.

Retiring it is a judgement call, not a cleanup: dropping the schedule entry
(config/sidekiq_production_schedule.yml) removes that safety net. NOTE this
worker does not repair stale live embeddings; the population and refresh
workers own that lifecycle.

Self-throttling: Embedding::Gemini caps at 300 req/min, and each iteration is
one batchEmbedContents request (≤100 rows), so the corpus drains without any
mass enqueue. Resumable: Sidekiq::IterableJob checkpoints the PK cursor after
every batch, so a deploy or restart resumes mid-run.

Scheduled nightly; idempotent (skips rows that already have a Gemini unified
sibling), so it's safe to re-run and converges to full coverage.

Examples:

Backfill all eligible text types (default)

TextUnifiedEmbeddingBackfillWorker.perform_async

Restrict to specific types

TextUnifiedEmbeddingBackfillWorker.perform_async('types' => %w[Post Article])

See Also:

  • doc/tasks/202606051030_TEXT_EMBEDDING_UNIFICATIONdoc/tasks/202606051030_TEXT_EMBEDDING_UNIFICATION.md

Constant Summary collapse

BATCH_SIZE =

One Gemini batchEmbedContents request per iteration.

Embedding::Gemini::MAX_BATCH_SIZE

Instance Method Summary collapse

Instance Method Details

#build_enumerator(options = nil, cursor:) ⇒ Enumerator

Returns batches of candidate records.

Parameters:

  • options (Hash, nil) (defaults to: nil) —

    serialized worker options (nil when none were enqueued)

  • cursor (Object) —

    sidekiq-iteration cursor

Options Hash (options):

  • types (Array<String>) —

    subset of text types to backfill (defaults to all)

Returns:

  • (Enumerator) —

    batches of candidate records



51
52
53
54
55
56
57
58
59
60
 
# File 'app/workers/text_unified_embedding_backfill_worker.rb', line 51

def build_enumerator(options = nil, cursor:)
  options ||= {} # sidekiq-iteration omits the arg entirely when no options were enqueued
  @types = Array(options[:types]).presence || Embedding::TextUnifier::TEXT_TYPES
  @counts = { processed: 0, skipped: 0, failed: 0 }

  scope = candidate_scope
  log_info "Starting: #{scope.count} primary rows need a Gemini unified sibling (types: #{@types.join(',')})"

  active_record_batches_enumerator(scope, cursor: cursor, batch_size: BATCH_SIZE)
end

#each_iteration(batch, *_args) ⇒ Object

Each iteration.

Parameters:

  • batch (Object) —

    the batch to each

  • _args (Array) —

    the args

Returns:

  • (Object) —

    the result



67
68
69
70
71
72
 
# File 'app/workers/text_unified_embedding_backfill_worker.rb', line 67

def each_iteration(batch, *_args)
  result = Embedding::TextUnifier.backfill(batch)
  @counts.each_key { |key| @counts[key] += result[key] }

  log_info "Progress: #{@counts.inspect}" if (@counts[:processed] % 1000).zero? && @counts[:processed].positive?
end

#on_complete ⇒ Object

On complete.

Returns:

  • (Object) —

    the result



77
78
79
 
# File 'app/workers/text_unified_embedding_backfill_worker.rb', line 77

def on_complete
  log_info "Complete: #{@counts.inspect}"
end