From 75ab5721f2fd0e6929226ea8add3748cea985e6f Mon Sep 17 00:00:00 2001 From: Landon Date: Sun, 12 Apr 2026 12:28:27 -0500 Subject: [PATCH] fix: preventing `task_rety()` from overwriting completed `Tests` --- app/api/tasks.py | 5 ++ app/api/utils/scanner.py | 126 ++++++++++++++++++++++++--------------- app/api/utils/updater.py | 5 ++ 3 files changed, 88 insertions(+), 48 deletions(-) diff --git a/app/api/tasks.py b/app/api/tasks.py index f127fa7..e535036 100644 --- a/app/api/tasks.py +++ b/app/api/tasks.py @@ -1585,6 +1585,11 @@ def run_test( # get test test = Test.objects.get(id=test_id) + # idempotency guard: once a Test is complete, do not execute it again. + if test.time_completed is not None: + logger.info(f'skipping run_test for completed test_id: {str(test_id)}') + return None + # define objects for flowrun objects = [_flow_obj( parent=str(test.page.id), diff --git a/app/api/utils/scanner.py b/app/api/utils/scanner.py index b046454..ad4c965 100644 --- a/app/api/utils/scanner.py +++ b/app/api/utils/scanner.py @@ -10,6 +10,7 @@ from .updater import update_flowrun from .manager import record_task from .tester import Tester +from django.core.cache import cache from datetime import datetime from cursion import settings import os, asyncio, uuid, boto3, random, time @@ -377,53 +378,83 @@ def check_scan_completion( # start Test if test_id present if test_id is not None: - # update flowrun - if flowrun_id and flowrun_id != 'None': - time.sleep(random.uniform(0.1, 5)) - update_flowrun(**{ - 'flowrun_id': str(flowrun_id), - 'node_index': node_index, - 'message': f'starting test comparison algorithm for {scan.page.page_url} | test_id: {str(test_id)}', - 'objects': objects - }) - - # get task_id from scan.system - task_id = None - for task in scan.system['tasks']: - if task.get('component') == sender: - task_id = task.get('task_id') - - # record task data in test - record_task( - resource_type='test', - resource_id=str(test_id), - task_id=str(task_id), - task_method='run_test', - kwargs={ - 'test_id': str(test_id), - 'alert_id': str(alert_id) if alert_id is not None else None, - 'flowrun_id': str(flowrun_id) if flowrun_id is not None else None, - 'node_index': str(node_index) if node_index is not None else None, - 'track_id': track_id - } - ) - - print('\n---------------\nScan Complete\nStarting Test...\n---------------\n') - test = Test.objects.get(id=test_id) - updated_test = Tester(test=test).run_test() - - # update flowrun - if flowrun_id and flowrun_id != 'None': - objects[-1]['status'] = updated_test.status - update_flowrun(**{ - 'flowrun_id': str(flowrun_id), - 'node_index': node_index, - 'message': ( - f'test for {scan.page.page_url} completed with status: '+ - f'{"❌ FAILED" if updated_test.status == 'failed' else "✅ PASSED"} | test_id: {str(test_id)}' - ), - 'objects': objects - }) + # derive tracking identity from Test system data + run_track_id = str(test_id) + try: + test = Test.objects.get(id=test_id) + first_task = ((test.system or {}).get('tasks') or [{}])[0] + task_kwargs = first_task.get('kwargs') or {} + run_track_id = str(task_kwargs.get('track_id') or test_id) + except Exception: + test = None + + # if Test already completed, do not re-run it. + if test is not None and test.time_completed is not None: + if flowrun_id and flowrun_id != 'None': + objects[-1]['status'] = test.status + update_flowrun(**{ + 'flowrun_id': str(flowrun_id), + 'node_index': node_index, + 'message': f'skipping run_test for completed test_id: {str(test_id)}', + 'objects': objects + }) + return scan + + # avoid duplicate test launches from concurrent component completions + run_test_lock_key = f'scanner:run_test:{str(test_id)}' + if not cache.add(run_test_lock_key, '1', timeout=600): + return scan + + try: + # update flowrun + if flowrun_id and flowrun_id != 'None': + time.sleep(random.uniform(0.1, 5)) + update_flowrun(**{ + 'flowrun_id': str(flowrun_id), + 'node_index': node_index, + 'message': f'starting test comparison algorithm for {scan.page.page_url} | test_id: {str(test_id)}', + 'objects': objects + }) + + # get task_id from scan.system + task_id = None + for task in scan.system['tasks']: + if task.get('component') == sender: + task_id = task.get('task_id') + + # record task data in test + record_task( + resource_type='test', + resource_id=str(test_id), + task_id=str(task_id), + task_method='run_test', + kwargs={ + 'test_id': str(test_id), + 'alert_id': str(alert_id) if alert_id is not None else None, + 'flowrun_id': str(flowrun_id) if flowrun_id is not None else None, + 'node_index': str(node_index) if node_index is not None else None, + 'track_id': run_track_id + } + ) + + print('\n---------------\nScan Complete\nStarting Test...\n---------------\n') + test = Test.objects.get(id=test_id) + updated_test = Tester(test=test).run_test() + + # update flowrun + if flowrun_id and flowrun_id != 'None': + objects[-1]['status'] = updated_test.status + update_flowrun(**{ + 'flowrun_id': str(flowrun_id), + 'node_index': node_index, + 'message': ( + f'test for {scan.page.page_url} completed with status: '+ + f'{"❌ FAILED" if updated_test.status == 'failed' else "✅ PASSED"} | test_id: {str(test_id)}' + ), + 'objects': objects + }) + finally: + cache.delete(run_test_lock_key) if alert_id is not None and alert_id != 'None': print('running alert from `cursion.check_scan_completion`') @@ -756,4 +787,3 @@ def _yellowlab( return scan - diff --git a/app/api/utils/updater.py b/app/api/utils/updater.py index 15f8aed..7fe2dae 100644 --- a/app/api/utils/updater.py +++ b/app/api/utils/updater.py @@ -36,6 +36,11 @@ def update_flowrun(**kwargs) -> object: # stale copies of nodes/edges/logs. flowrun = FlowRun.objects.select_for_update().get(id=flowrun_id) + # Ignore stale worker updates after completion; late async tasks should + # not mutate finished runs. + if flowrun.time_completed is not None: + return flowrun + # set timestamp timestamp = timezone.now().strftime('%Y-%m-%d %H:%M:%S.%f')