Skip to content

Feature: reconcile ontologyRank against Solr state (skip already-current ontologies) #300

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mdorf merged 1 commit into from
Jun 23, 2026
Merged

Feature: reconcile ontologyRank against Solr state (skip already-current ontologies) #300

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 46 additions & 65 deletions lib/ontologies_linked_data/services/rank_solr_propagator.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
# frozen_string_literal: true

require 'redis'

module LinkedData
module Services
# Propagates the current ontology rank values into the denormalized
Expand All @@ -15,70 +13,60 @@ module Services
# the indexer uses), which derives normalizedScore from the Redis-stored
# bioportalScore/umlsScore using the configured weights.
#
# Updates use true Solr atomic updates ({ id, ontologyRank: { set: score } })
# so no per-class triplestore read is needed. Batches are sent without a
# commit; a single commit is issued *between* ontologies (when no updates are
# in flight) to keep the transaction log bounded without pausing replicas
# mid-stream — committing during the update stream backs up SolrCloud's
# leader->replica forwarding queue and causes "distributed update stalled"
# 500s. Each Solr call is retried with backoff so a transient stall is
# survived rather than aborting the whole run.
# For each ontology the propagator first asks Solr how many of its docs are
# NOT already at the current rank (a cheap rows:0 count with a negative range
# filter). If none are stale it skips the ontology entirely — so even the
# first run skips ontologies whose Solr rank already matches (e.g. set at
# index time and never drifted), and a re-run resumes from Solr's actual
# state. Otherwise it cursor-scans just the stale docs and issues true atomic
# updates ({ id, ontologyRank: { set: score } }) with no triplestore read.
#
# Ontologies whose rank is unchanged since the last successful propagation
# are skipped (tracked per acronym in Redis). Pass force: true to ignore the
# skip cache and re-propagate everything (e.g. after a collection rebuild).
# Batches are sent without a commit; a single commit is issued between
# ontologies (no updates in flight) to keep the transaction log bounded
# without stalling replica forwarding mid-stream. Every Solr call is retried
# with backoff so a transient stall/blip is survived rather than aborting.
#
# See https://github.com/ncbo/ncbo_cron/issues/132
class RankSolrPropagator
DEFAULT_BATCH_SIZE = 2_500
PROGRESS_EVERY = 50_000 # log per-ontology progress every N docs
MAX_RETRIES = 5
RETRY_BASE_SLEEP = 5 # seconds; exponential backoff: 5, 10, 20, 40, 80
# Redis key holding the last successfully propagated rank per acronym.
LAST_PROPAGATED_REDIS_FIELD = 'ontology_rank_solr_propagated'
# Half-width of the match band. normalizedScore is rounded to 3 dp, so a
# +/-0.0005 band uniquely identifies a stored pfloat value despite float
# representation.
SCORE_TOLERANCE = 0.0005

# batch_size precedence: explicit arg > RANK_SOLR_BATCH_SIZE env > default.
# The env override lets the batch size be tuned on staging without a deploy.
def initialize(logger: nil, batch_size: nil)
@logger = logger || Logger.new($stdout)
@batch_size = (batch_size || ENV['RANK_SOLR_BATCH_SIZE'] || DEFAULT_BATCH_SIZE).to_i
@retry_count = 0
end

# Reads the rank map and writes each ontology's normalizedScore onto all of
# its term_search docs. Unchanged ontologies are skipped unless force: true.
# Returns the number of ontologies whose docs were updated.
def propagate(rank_map = nil, force: false)
# Writes each ontology's normalizedScore onto its stale term_search docs,
# skipping ontologies already at the current rank. Returns the number of
# ontologies that needed an update.
def propagate(rank_map = nil)
rank_map ||= LinkedData::Models::Ontology.rank

if rank_map.nil? || rank_map.empty?
@logger.warn('RankSolrPropagator: empty rank map; nothing to propagate')
return 0
end

last_propagated = force ? {} : load_last_propagated
total_onts = rank_map.size
updated = 0
skipped = 0
@retry_count = 0

rank_map.each_with_index do |(acronym, rank_info), i|
score = rank_info[:normalizedScore].to_f

if !force && last_propagated[acronym] == score
skipped += 1
next
end

log("[#{i + 1}/#{total_onts}] #{acronym} rank=#{score}")
count = update_ontology_rank(acronym, score)

last_propagated[acronym] = score
save_last_propagated(last_propagated)
updated += 1 if count.positive?
count.positive? ? (updated += 1) : (skipped += 1)
end

summary = "done; updated #{updated}, skipped #{skipped} unchanged (of #{total_onts}); Solr retries: #{@retry_count}"
summary = "done; updated #{updated}, skipped #{skipped} already-current (of #{total_onts}); Solr retries: #{@retry_count}"
if @retry_count.zero?
log(summary)
else
Expand All @@ -91,25 +79,30 @@ def propagate(rank_map = nil, force: false)

private

# Cursor-scans all term_search docs for the acronym, issues batched atomic
# updates to ontologyRank (no commit), then commits once. Returns the
# number of docs updated.
# Counts docs whose ontologyRank is not already the target, then cursor-scans
# just those stale docs and atomic-updates them. Returns docs updated.
# Scanning the stale set is safe under cursorMark: docs we update leave the
# set only behind the (id-ascending) cursor, so no stale doc is skipped.
def update_ontology_rank(acronym, score)
query = "submissionAcronym:\"#{acronym}\""
num_found = (solr_search(query, rows: 0).dig('response', 'numFound') || 0).to_i

if num_found.zero?
log(" #{acronym}: no term_search docs; skipping")
acronym_q = "submissionAcronym:\"#{acronym}\""
lo = format('%.4f', score - SCORE_TOLERANCE)
hi = format('%.4f', score + SCORE_TOLERANCE)
stale_fq = "-ontologyRank:[#{lo} TO #{hi}]"

stale = solr_count(acronym_q, stale_fq)
if stale.zero?
log(" #{acronym}: already current; skipping")
return 0
end
log(" #{acronym}: #{num_found} docs to update")
log(" #{acronym}: #{stale} docs to update")

cursor = '*'
total = 0
last_logged = 0

loop do
resp = solr_search(query, fl: 'id', rows: @batch_size, sort: 'id asc', cursorMark: cursor)
resp = solr_search(acronym_q, fq: stale_fq, fl: 'id', rows: @batch_size,
sort: 'id asc', cursorMark: cursor)
docs = resp.dig('response', 'docs') || []

unless docs.empty?
Expand All @@ -120,7 +113,7 @@ def update_ontology_rank(acronym, score)
total += batch.size

if total - last_logged >= PROGRESS_EVERY
log(" #{acronym}: #{total}/#{num_found} docs…")
log(" #{acronym}: #{total}/#{stale} docs…")
last_logged = total
end
end
Expand All @@ -139,12 +132,19 @@ def update_ontology_rank(acronym, score)
total
end

def solr_count(query, fq = nil)
params = { rows: 0 }
params[:fq] = fq if fq
(solr_search(query, **params).dig('response', 'numFound') || 0).to_i
end

def solr_search(query, **params)
with_retry('search') { LinkedData::Models::Class.search(query, params) }
end

# Retries transient Solr errors (e.g. distributed-update stalls under load)
# with exponential backoff. Re-raises once retries are exhausted.
# Retries transient Solr/network errors (distributed-update stalls,
# connection blips, timeouts) with exponential backoff. Re-raises once
# retries are exhausted.
def with_retry(what)
attempts = 0
begin
Expand Down Expand Up @@ -177,25 +177,6 @@ def log(msg)
@logger.info("RankSolrPropagator: #{msg}")
@logger.flush if @logger.respond_to?(:flush)
end

def redis
@redis ||= Redis.new(host: LinkedData.settings.ontology_analytics_redis_host,
port: LinkedData.settings.ontology_analytics_redis_port)
end

def load_last_propagated
raw = redis.get(LAST_PROPAGATED_REDIS_FIELD)
raw ? Marshal.load(raw) : {}
rescue StandardError => e
@logger.warn("RankSolrPropagator: could not load skip cache (#{e.class}: #{e.message}); propagating all")
{}
end

def save_last_propagated(map)
redis.set(LAST_PROPAGATED_REDIS_FIELD, Marshal.dump(map))
rescue StandardError => e
@logger.warn("RankSolrPropagator: could not save skip cache (#{e.class}: #{e.message})")
end
end
end
end
27 changes: 6 additions & 21 deletions test/services/test_rank_solr_propagator.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@ def self.after_suite

def setup
self.class.after_suite
clear_propagation_cache
# Index a small ontology's terms into the real local term_search collection.
submission_parse(ACRONYM, 'BRO Ontology',
'./test/data/ontology_files/BRO_v3.5.owl', 1,
Expand All @@ -24,13 +23,6 @@ def setup

def teardown
self.class.after_suite
clear_propagation_cache
end

def clear_propagation_cache
Redis.new(host: LinkedData.settings.ontology_analytics_redis_host,
port: LinkedData.settings.ontology_analytics_redis_port)
.del(LinkedData::Services::RankSolrPropagator::LAST_PROPAGATED_REDIS_FIELD)
end

def stub_rank(score)
Expand Down Expand Up @@ -87,24 +79,17 @@ def test_propagate_returns_zero_for_empty_rank_map
assert_equal 0, LinkedData::Services::RankSolrPropagator.new.propagate
end

def test_propagate_skips_unchanged_ontology_on_second_run
# Once Solr already holds the target rank, a fresh propagator (no cache, no
# shared state) must skip the ontology purely from Solr's state — this is the
# same mechanism that lets the first run skip already-current ontologies.
def test_skips_ontology_already_current_in_solr
stub_rank(0.642)

first = LinkedData::Services::RankSolrPropagator.new.propagate
assert_equal 1, first, 'first run should propagate the ontology'
assert_equal 1, first, 'first run should propagate the stale ontology'

# Rank unchanged -> the second run should skip it entirely.
second = LinkedData::Services::RankSolrPropagator.new.propagate
assert_equal 0, second, 'unchanged ontology should be skipped on the second run'
end

def test_force_repropagates_even_when_unchanged
stub_rank(0.642)

LinkedData::Services::RankSolrPropagator.new.propagate
# force: true ignores the skip cache and re-propagates.
forced = LinkedData::Services::RankSolrPropagator.new.propagate(nil, force: true)
assert_equal 1, forced, 'force should re-propagate even when rank is unchanged'
assert_equal 0, second, 'ontology already at the target rank in Solr must be skipped'
end

# Manifests the Solr-stall condition: the first atomic-update POST raises a
Expand Down
Loading