From 839ac262af5ca8e9f291cf1ae84d1b9d5edab769 Mon Sep 17 00:00:00 2001 From: Dave Lawrence Date: Wed, 26 Aug 2026 16:10:29 +0930 Subject: [PATCH 1/4] Classification - defer condition text automatch to an async task so the external Monarch search never runs in the upload request #1780 --- .../0176_conditiontext_pending_automatch.py | 18 ++++++ .../models/condition_text_matching.py | 42 ++++++++++++- .../tasks/condition_text_automatch_task.py | 22 +++++++ .../models/test_condition_text_automatch.py | 60 +++++++++++++++++++ .../settings/components/celery_settings.py | 1 + 5 files changed, 140 insertions(+), 3 deletions(-) create mode 100644 classification/migrations/0176_conditiontext_pending_automatch.py create mode 100644 classification/tasks/condition_text_automatch_task.py create mode 100644 classification/tests/models/test_condition_text_automatch.py 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..797314c2a 100644 --- a/classification/models/condition_text_matching.py +++ b/classification/models/condition_text_matching.py @@ -8,6 +8,7 @@ from typing import Optional import django +from celery.canvas import Signature from django.contrib.auth.models import User from django.contrib.postgres.fields import ArrayField from django.db import models, transaction @@ -30,6 +31,10 @@ classification_post_publish_signal, flag_types, ) +from classification.models.classification_import_run import ( + ClassificationImportRun, + classification_imports_complete_signal, +) from classification.models.condition_text_search import condition_text_search from flags.models import Flag, FlagComment, FlagResolution, flag_comment_action from genes.models import GeneSymbol, GeneSymbolAlias @@ -69,6 +74,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, processed by condition_text_automatch_task - 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 +449,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") @@ -665,6 +675,21 @@ def __bool__(self): return bool(self.terms) or bool(self.messages) +CONDITION_TEXT_AUTOMATCH_TASK_NAME = 'classification.tasks.condition_text_automatch_task.condition_text_automatch_task' + + +def queue_pending_automatch(): + """ + Launch condition_text_automatch_task for any ConditionTexts flagged pending_automatch. + During a bulk import, leave them for classification_imports_complete_signal so the whole + batch is handled by one task. + """ + if ClassificationImportRun.ongoing_imports(): + return + if ConditionText.objects.filter(pending_automatch=True).exists(): + transaction.on_commit(lambda: Signature(CONDITION_TEXT_AUTOMATCH_TASK_NAME, immutable=True).apply_async()) + + @receiver(classification_post_publish_signal, sender=Classification) def published(sender, classification: Classification, @@ -678,6 +703,7 @@ def published(sender, """ get_timer().tick("Condition Text Matching - post publish") ConditionTextMatch.sync_condition_text_classification(newly_published, attempt_automatch=True, update_counts=True) + queue_pending_automatch() @receiver(flag_comment_action, sender=Flag) @@ -690,6 +716,16 @@ def check_for_withdrawn(sender, flag_comment: FlagComment, old_resolution: FlagR cl: Classification if cl := Classification.objects.filter(flag_collection=flag.collection.id).first(): ConditionTextMatch.sync_condition_text_classification(cl.last_published_version, attempt_automatch=True, update_counts=True) + queue_pending_automatch() + + +@receiver(classification_imports_complete_signal, sender=ClassificationImportRun) +def automatch_pending_after_import(sender, **kwargs): + """ + Catch up on the automatching every publish deferred during the bulk import - the + pending_automatch flags dedupe it down to one automatch per ConditionText + """ + queue_pending_automatch() # @timed_cache(size_limit=2) diff --git a/classification/tasks/condition_text_automatch_task.py b/classification/tasks/condition_text_automatch_task.py new file mode 100644 index 000000000..6eaba7900 --- /dev/null +++ b/classification/tasks/condition_text_automatch_task.py @@ -0,0 +1,22 @@ +import celery +from django.db import transaction + +from classification.models.condition_text_matching import ConditionText, ConditionTextMatch + + +@celery.shared_task +def condition_text_automatch_task(): + """ + Automatches every ConditionText flagged pending_automatch by sync_condition_text_classification. + Runs here rather than in the publishing request because matching can call the external Monarch + search API, which can be slow or down entirely (#1780). + """ + while True: + with transaction.atomic(): + ct = ConditionText.objects.filter(pending_automatch=True).select_for_update(skip_locked=True).first() + if ct is None: + return + # claim in a short transaction so the row isn't locked during the Monarch call + ct.pending_automatch = False + ct.save() + 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..37324f050 --- /dev/null +++ b/classification/tests/models/test_condition_text_automatch.py @@ -0,0 +1,60 @@ +from unittest.mock import patch + +from django.test import TestCase + +from classification.models.classification_import_run import ClassificationImportRun +from classification.models.condition_text_matching import ( + CONDITION_TEXT_AUTOMATCH_TASK_NAME, + ConditionText, + ConditionTextMatch, + queue_pending_automatch, +) +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): + 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_task_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('classification.models.condition_text_matching.Signature') + def test_queue_defers_to_end_of_import(self, mock_signature): + self._condition_text("condition 1", pending=True) + ClassificationImportRun.record_classification_import(identifier="test-import") + + with self.captureOnCommitCallbacks(execute=True): + queue_pending_automatch() + mock_signature.assert_not_called() + + # completing the run fires classification_imports_complete_signal, which queues the task + with self.captureOnCommitCallbacks(execute=True): + ClassificationImportRun.record_classification_import(identifier="test-import", is_complete=True) + mock_signature.assert_called_once_with(CONDITION_TEXT_AUTOMATCH_TASK_NAME, immutable=True) + mock_signature.return_value.apply_async.assert_called_once() + + @patch('classification.models.condition_text_matching.Signature') + def test_queue_launches_task_when_no_import_running(self, mock_signature): + self._condition_text("condition 1", pending=True) + + with self.captureOnCommitCallbacks(execute=True): + queue_pending_automatch() + mock_signature.return_value.apply_async.assert_called_once() 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', From b809f93ac4626a44274b5bff6b4e369c74953b71 Mon Sep 17 00:00:00 2001 From: Dave Lawrence Date: Wed, 26 Aug 2026 16:17:53 +0930 Subject: [PATCH 2/4] Classification - condition text automatch as a beat sweep over pending flags instead of per-publish task dispatch #1780 --- .../models/condition_text_matching.py | 35 ++----------------- .../tasks/condition_text_automatch_task.py | 26 +++++++------- .../models/test_condition_text_automatch.py | 35 ++++++------------- variantgrid/celery.py | 8 +++++ 4 files changed, 34 insertions(+), 70 deletions(-) diff --git a/classification/models/condition_text_matching.py b/classification/models/condition_text_matching.py index 797314c2a..38b50600b 100644 --- a/classification/models/condition_text_matching.py +++ b/classification/models/condition_text_matching.py @@ -8,7 +8,6 @@ from typing import Optional import django -from celery.canvas import Signature from django.contrib.auth.models import User from django.contrib.postgres.fields import ArrayField from django.db import models, transaction @@ -31,10 +30,6 @@ classification_post_publish_signal, flag_types, ) -from classification.models.classification_import_run import ( - ClassificationImportRun, - classification_imports_complete_signal, -) from classification.models.condition_text_search import condition_text_search from flags.models import Flag, FlagComment, FlagResolution, flag_comment_action from genes.models import GeneSymbol, GeneSymbolAlias @@ -74,8 +69,8 @@ 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, processed by condition_text_automatch_task - automatching - # can call the external Monarch search API so it can't run in the publishing request (#1780) + # 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: @@ -675,21 +670,6 @@ def __bool__(self): return bool(self.terms) or bool(self.messages) -CONDITION_TEXT_AUTOMATCH_TASK_NAME = 'classification.tasks.condition_text_automatch_task.condition_text_automatch_task' - - -def queue_pending_automatch(): - """ - Launch condition_text_automatch_task for any ConditionTexts flagged pending_automatch. - During a bulk import, leave them for classification_imports_complete_signal so the whole - batch is handled by one task. - """ - if ClassificationImportRun.ongoing_imports(): - return - if ConditionText.objects.filter(pending_automatch=True).exists(): - transaction.on_commit(lambda: Signature(CONDITION_TEXT_AUTOMATCH_TASK_NAME, immutable=True).apply_async()) - - @receiver(classification_post_publish_signal, sender=Classification) def published(sender, classification: Classification, @@ -703,7 +683,6 @@ def published(sender, """ get_timer().tick("Condition Text Matching - post publish") ConditionTextMatch.sync_condition_text_classification(newly_published, attempt_automatch=True, update_counts=True) - queue_pending_automatch() @receiver(flag_comment_action, sender=Flag) @@ -716,16 +695,6 @@ def check_for_withdrawn(sender, flag_comment: FlagComment, old_resolution: FlagR cl: Classification if cl := Classification.objects.filter(flag_collection=flag.collection.id).first(): ConditionTextMatch.sync_condition_text_classification(cl.last_published_version, attempt_automatch=True, update_counts=True) - queue_pending_automatch() - - -@receiver(classification_imports_complete_signal, sender=ClassificationImportRun) -def automatch_pending_after_import(sender, **kwargs): - """ - Catch up on the automatching every publish deferred during the bulk import - the - pending_automatch flags dedupe it down to one automatch per ConditionText - """ - queue_pending_automatch() # @timed_cache(size_limit=2) diff --git a/classification/tasks/condition_text_automatch_task.py b/classification/tasks/condition_text_automatch_task.py index 6eaba7900..21bf3f7eb 100644 --- a/classification/tasks/condition_text_automatch_task.py +++ b/classification/tasks/condition_text_automatch_task.py @@ -1,22 +1,22 @@ import celery -from django.db import transaction +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(): """ - Automatches every ConditionText flagged pending_automatch by sync_condition_text_classification. - Runs here rather than in the publishing request because matching can call the external Monarch - search API, which can be slow or down entirely (#1780). + 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. """ - while True: - with transaction.atomic(): - ct = ConditionText.objects.filter(pending_automatch=True).select_for_update(skip_locked=True).first() - if ct is None: - return - # claim in a short transaction so the row isn't locked during the Monarch call - ct.pending_automatch = False - ct.save() - ConditionTextMatch.attempt_automatch(condition_text=ct) + 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 index 37324f050..6e92c6d88 100644 --- a/classification/tests/models/test_condition_text_automatch.py +++ b/classification/tests/models/test_condition_text_automatch.py @@ -3,12 +3,7 @@ from django.test import TestCase from classification.models.classification_import_run import ClassificationImportRun -from classification.models.condition_text_matching import ( - CONDITION_TEXT_AUTOMATCH_TASK_NAME, - ConditionText, - ConditionTextMatch, - queue_pending_automatch, -) +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 @@ -25,7 +20,7 @@ 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_task_automatches_each_pending_text_once(self, mock_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) @@ -36,25 +31,17 @@ def test_task_automatches_each_pending_text_once(self, mock_automatch): self.assertEqual(automatched, {pending_1.pk, pending_2.pk}) self.assertFalse(ConditionText.objects.filter(pending_automatch=True).exists()) - @patch('classification.models.condition_text_matching.Signature') - def test_queue_defers_to_end_of_import(self, mock_signature): + @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") - with self.captureOnCommitCallbacks(execute=True): - queue_pending_automatch() - mock_signature.assert_not_called() - - # completing the run fires classification_imports_complete_signal, which queues the task - with self.captureOnCommitCallbacks(execute=True): - ClassificationImportRun.record_classification_import(identifier="test-import", is_complete=True) - mock_signature.assert_called_once_with(CONDITION_TEXT_AUTOMATCH_TASK_NAME, immutable=True) - mock_signature.return_value.apply_async.assert_called_once() + condition_text_automatch_task() + mock_automatch.assert_not_called() + self.assertTrue(ConditionText.objects.filter(pending_automatch=True).exists()) - @patch('classification.models.condition_text_matching.Signature') - def test_queue_launches_task_when_no_import_running(self, mock_signature): - self._condition_text("condition 1", pending=True) + ClassificationImportRun.record_classification_import(identifier="test-import", is_complete=True) - with self.captureOnCommitCallbacks(execute=True): - queue_pending_automatch() - mock_signature.return_value.apply_async.assert_called_once() + 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'] = { From b737c09741a1217a0c4418a299967ed30b931b6e Mon Sep 17 00:00:00 2001 From: Dave Lawrence Date: Thu, 27 Aug 2026 09:56:00 +0930 Subject: [PATCH 3/4] Fix failing test --- .../tests/test_annotation_disk_space.py | 25 +++++++++++++------ 1 file changed, 18 insertions(+), 7 deletions(-) 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) From d5ce648aed9f1f1402a571a548a05328b5d1eccd Mon Sep 17 00:00:00 2001 From: Dave Lawrence Date: Mon, 31 Aug 2026 09:21:45 +0930 Subject: [PATCH 4/4] Automatch test: clear the thread-local request so the Event isn't logged against a rolled-back user #1780 --- classification/tests/models/test_condition_text_automatch.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/classification/tests/models/test_condition_text_automatch.py b/classification/tests/models/test_condition_text_automatch.py index 6e92c6d88..4407cfa53 100644 --- a/classification/tests/models/test_condition_text_automatch.py +++ b/classification/tests/models/test_condition_text_automatch.py @@ -1,6 +1,7 @@ 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 @@ -11,6 +12,9 @@ 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',