diff --git a/annotation/tests/test_annotation_disk_space.py b/annotation/tests/test_annotation_disk_space.py index 94e4e8626..ac2fe51c5 100644 --- a/annotation/tests/test_annotation_disk_space.py +++ b/annotation/tests/test_annotation_disk_space.py @@ -49,6 +49,20 @@ STANDARD = VariantAnnotationPipelineType.STANDARD +def past_vep_kwargs() -> dict: + """ Timestamps/counts a run carries once VEP has finished, ie ANNOTATION_COMPLETED - what the import + lane checks before it does anything. """ + now = timezone.now() + return { + "count": 100, + "dump_start": now - timedelta(minutes=5), + "dump_end": now - timedelta(minutes=4), + "dump_count": 100, + "annotation_start": now - timedelta(minutes=4), + "annotation_end": now, + } + + @override_settings(**get_fake_annotation_settings_dict(columns_version=2)) class AnnotationRunCleanupTestCase(TestCase): """ Each way into the cleanup module, and the one case that must not reach it. """ @@ -134,7 +148,8 @@ def test_failed_import_keeps_everything(self): with tempfile.TemporaryDirectory() as tmp_dir, \ override_settings(ANNOTATION_VCF_DUMP_DIR=tmp_dir, ANNOTATION_DELETE_TEMP_FILES_ON_SUCCESS=True): - run, paths, _ = self._run_with_output(tmp_dir) + run, paths, _ = self._run_with_output(tmp_dir, **past_vep_kwargs()) + self.assertEqual(run.status, AnnotationStatus.ANNOTATION_COMPLETED) with mock.patch.object(VEPRunner, "import_results", side_effect=RuntimeError("import blew up")), \ @@ -204,12 +219,8 @@ def _make_run(self, lo_idx, hi_idx, status=AnnotationStatus.CREATED): # count stamped as the count lane would have, so the run is ready for the run lanes run = AnnotationRun.objects.create(annotation_range_lock=lock, pipeline_type=STANDARD, count=100) if status == AnnotationStatus.ANNOTATION_COMPLETED: - now = timezone.now() # past VEP, waiting on the import lane - run.dump_start = now - timedelta(minutes=5) - run.dump_end = now - timedelta(minutes=4) - run.dump_count = 100 - run.annotation_start = now - timedelta(minutes=4) - run.annotation_end = now + for k, v in past_vep_kwargs().items(): # past VEP, waiting on the import lane + setattr(run, k, v) run.vcf_annotated_filename = "/does/not/need/to/exist.vcf.gz" run.save() self.assertEqual(run.status, AnnotationStatus.ANNOTATION_COMPLETED) diff --git a/classification/migrations/0176_conditiontext_pending_automatch.py b/classification/migrations/0176_conditiontext_pending_automatch.py new file mode 100644 index 000000000..2649cdf42 --- /dev/null +++ b/classification/migrations/0176_conditiontext_pending_automatch.py @@ -0,0 +1,18 @@ +# Generated by Django 6.1 on 2026-08-26 06:38 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('classification', '0175_ekey_gene_fusion_options'), + ] + + operations = [ + migrations.AddField( + model_name='conditiontext', + name='pending_automatch', + field=models.BooleanField(default=False), + ), + ] diff --git a/classification/models/condition_text_matching.py b/classification/models/condition_text_matching.py index 1f3c7f2ed..38b50600b 100644 --- a/classification/models/condition_text_matching.py +++ b/classification/models/condition_text_matching.py @@ -69,6 +69,9 @@ class ConditionText(TimeStampedModel, GuardianPermissionsMixin): classifications_count = models.IntegerField(default=0) classifications_count_outstanding = models.IntegerField(default=0) + # set when a new root/gene level appears, drained by the condition_text_automatch_task beat sweep - + # automatching can call the external Monarch search API so it can't run in the publishing request (#1780) + pending_automatch = models.BooleanField(default=False) class Meta: unique_together = ("normalized_text", "lab") @@ -441,13 +444,15 @@ def sync_condition_text_classification(cm: ClassificationModification, update_co debug_timer.tick("Condition Text Matching - create new entry") if attempt_automatch and (new_root or new_gene_level): - ConditionTextMatch.attempt_automatch(ct, gene_symbol=gene_symbol) - debug_timer.tick("Condition Text Matching - auto match") - elif update_counts: + ct.pending_automatch = True + + if update_counts: ct.classifications_count += 1 is_valid = root.is_valid or gene_level.is_valid or mode_of_inheritance_level.is_valid or (existing and existing.is_valid) if not is_valid: ct.classifications_count_outstanding += 1 + + if update_counts or ct.pending_automatch: ct.save() debug_timer.tick("Condition Text Matching - update count quick") diff --git a/classification/tasks/condition_text_automatch_task.py b/classification/tasks/condition_text_automatch_task.py new file mode 100644 index 000000000..21bf3f7eb --- /dev/null +++ b/classification/tasks/condition_text_automatch_task.py @@ -0,0 +1,22 @@ +import celery + +from classification.models.classification_import_run import ClassificationImportRun +from classification.models.condition_text_matching import ConditionText, ConditionTextMatch + + +@celery.shared_task +def condition_text_automatch_task(): + """ + Celery beat sweep that automatches every ConditionText flagged pending_automatch. Automatching + can call the external Monarch search API, which can be slow or down entirely, so it runs here + rather than in the publishing request (#1780). While a bulk import is ongoing the sweep stands + aside - the flags accumulate and the first sweep after completion handles the whole batch, + deduped to one automatch per distinct condition text. The flags are also the crash recovery: + anything a dead worker left behind is picked up by the next sweep. + """ + if ClassificationImportRun.ongoing_imports(): + return + for ct in ConditionText.objects.filter(pending_automatch=True).iterator(): + # single-statement claim so overlapping sweeps can't automatch the same text twice + if ConditionText.objects.filter(pk=ct.pk, pending_automatch=True).update(pending_automatch=False): + ConditionTextMatch.attempt_automatch(condition_text=ct) diff --git a/classification/tests/models/test_condition_text_automatch.py b/classification/tests/models/test_condition_text_automatch.py new file mode 100644 index 000000000..4407cfa53 --- /dev/null +++ b/classification/tests/models/test_condition_text_automatch.py @@ -0,0 +1,51 @@ +from unittest.mock import patch + +from django.test import TestCase +from threadlocals.threadlocals import set_thread_variable + +from classification.models.classification_import_run import ClassificationImportRun +from classification.models.condition_text_matching import ConditionText, ConditionTextMatch +from classification.tasks.condition_text_automatch_task import condition_text_automatch_task +from snpdb.models import Country, Lab, Organization + + +class ConditionTextAutomatchTest(TestCase): + + def setUp(self): + # NotificationBuilder logs an Event against the thread-local user - a client request in + # an earlier test leaves a rolled-back one behind + set_thread_variable('request', None) + org = Organization.objects.create(name='InstX', group_name='instx') + country = Country.objects.get_or_create(name='CountryA')[0] + self.lab = Lab.objects.create(name='Labby', organization=org, city='CityA', + country=country, group_name='instx/labby') + + def _condition_text(self, text: str, pending: bool) -> ConditionText: + return ConditionText.objects.create(normalized_text=text, lab=self.lab, pending_automatch=pending) + + @patch.object(ConditionTextMatch, 'attempt_automatch') + def test_sweep_automatches_each_pending_text_once(self, mock_automatch): + pending_1 = self._condition_text("condition 1", pending=True) + pending_2 = self._condition_text("condition 2", pending=True) + self._condition_text("condition 3", pending=False) + + condition_text_automatch_task() + + automatched = {call.kwargs["condition_text"].pk for call in mock_automatch.call_args_list} + self.assertEqual(automatched, {pending_1.pk, pending_2.pk}) + self.assertFalse(ConditionText.objects.filter(pending_automatch=True).exists()) + + @patch.object(ConditionTextMatch, 'attempt_automatch') + def test_sweep_waits_for_ongoing_import(self, mock_automatch): + self._condition_text("condition 1", pending=True) + ClassificationImportRun.record_classification_import(identifier="test-import") + + condition_text_automatch_task() + mock_automatch.assert_not_called() + self.assertTrue(ConditionText.objects.filter(pending_automatch=True).exists()) + + ClassificationImportRun.record_classification_import(identifier="test-import", is_complete=True) + + condition_text_automatch_task() + mock_automatch.assert_called_once() + self.assertFalse(ConditionText.objects.filter(pending_automatch=True).exists()) diff --git a/variantgrid/celery.py b/variantgrid/celery.py index c85fcc507..381d4110e 100644 --- a/variantgrid/celery.py +++ b/variantgrid/celery.py @@ -51,6 +51,14 @@ 'schedule': HOUR_SECS, # Check every hour, only update if hash changed } +# Condition text automatch sweep (#1780): publishing flags ConditionTexts (pending_automatch) +# instead of calling the external Monarch search API in the request; this drains the flags, and +# skips while a bulk import is ongoing so a sync's worth is handled as one deduped batch afterwards. +app.conf.beat_schedule['condition-text-automatch'] = { + 'task': 'classification.tasks.condition_text_automatch_task.condition_text_automatch_task', + 'schedule': MINUTE_SECS * 5, +} + # Reclassification timelines (issue #1523): the analytics page builds what it can in the request, # this picks up anything left over, e.g. the morning after a large sync. app.conf.beat_schedule['reclassification-events-update'] = { diff --git a/variantgrid/settings/components/celery_settings.py b/variantgrid/settings/components/celery_settings.py index 3d93b8308..5c875d18e 100644 --- a/variantgrid/settings/components/celery_settings.py +++ b/variantgrid/settings/components/celery_settings.py @@ -136,6 +136,7 @@ 'classification.tasks.classification_import_process_variants_task', 'classification.tasks.classification_import_task', 'classification.tasks.classification_candidate_search_tasks', + 'classification.tasks.condition_text_automatch_task', 'genes.tasks.gene_coverage_tasks', 'patients.tasks.extraction_matching_tasks', 'pedigree.models',