Analysis is actually deleted when pressing delete, results are archived if needed

Author: glormph

src/backend/analysis/migrations/0063_analysis_success_completed.py

@@ -0,0 +1,29 @@
+# Generated by Django 5.2 on 2026-03-04 14:42
+
+from django.db import migrations, models
+from jobs import jobs as jj
+
+
+def fake(x, y):
+    pass
+
+def fixfield(apps, s):
+    A = apps.get_model('analysis', 'Analysis')
+    A.objects.filter(nextflowsearch__job__state=jj.Jobstates.DONE).update(success_completed=True)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('analysis', '0062_alter_analysisdeleted_date'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='analysis',
+            name='success_completed',
+            field=models.BooleanField(default=False),
+        ),
+
+        migrations.RunPython(fixfield, fake)
+    ]

src/backend/analysis/models.py

@@ -287,6 +287,7 @@ class Analysis(models.Model):
     base_rundir = models.TextField()
     editable = models.BooleanField(default=True)
     securityclass = models.IntegerField(choices=filemodels.DataSecurityClass.choices)
+    success_completed = models.BooleanField(default=False)
 
     def get_fullname(self, wftype=False):
         if wftype:

src/backend/analysis/tests.py

@@ -845,22 +845,22 @@ def test_new_analysis_and_run_and_purge(self):
         j = jm.Job.objects.last()
         self.assertEqual(j.kwargs, {'sfloc_id': anasfl.pk, 'isdir': False})
 
-        # Now purge
+        # Now delete analysis
         j.state = jj.Jobstates.DONE
         j.save()
+        rm.PDCBackedupFile.objects.filter(storedfile_id=anasfl.sfile_id).update(success=True)
         self.user.is_staff = True
         self.user.save()
         resp = self.cl.post('/analysis/delete/', content_type='application/json', data={'item_id': ana.pk})
-        resp = self.cl.post('/analysis/purge/', content_type='application/json', data={'item_id': ana.pk})
         self.assertEqual(resp.status_code, 200)
         self.assertFalse(os.path.exists(webfn))
         self.assertTrue(os.path.exists(anafn))
         self.assertTrue(os.path.exists(nfrunfn))
         # Two purge jobs and two delete dir jobs, this test is getting slow
-        self.run_job()
-        self.run_job()
-        self.run_job()
-        self.run_job()
+        self.run_job() # purge files
+        self.run_job() # rm dir
+        self.run_job() # purge
+        self.run_job() # rm dir
         self.assertFalse(os.path.exists(anafn))
         self.assertFalse(os.path.exists(nfrunfn))
 
@@ -1240,11 +1240,6 @@ def test_failing(self):
 class TestDeleteAnalysis(BaseTest):
     url = '/analysis/delete/'
 
-    def setUp(self):
-        super().setUp()
-        self.ana = am.Analysis.objects.create(user=self.user, name='test', base_rundir='testdir',
-                securityclass=rm.DataSecurityClass.NOSECURITY)
-
     def test_fail_request(self):
         resp = self.cl.get(self.url)
         self.assertEqual(resp.status_code, 405)
@@ -1275,50 +1270,45 @@ def test_fail_request(self):
         self.assertEqual(resp.status_code, 403)
         self.assertIn('Analysis is already deleted', resp.json()['error'])
 
-    def test_delete(self):
+    def test_delete_with_backup(self):
         resp = self.cl.post(self.url, content_type='application/json', data={'item_id': self.ana.pk})
         self.assertEqual(resp.status_code, 200)
         self.assertFalse(self.ana.deleted)
         self.ana.refresh_from_db()
         self.assertTrue(self.ana.deleted)
-        # FIXME test job revoked
-
-
-class TestPurgeAnalysis(BaseIntegrationTest):
-    url = '/analysis/purge/'
-
-    def setUp(self):
-        super().setUp()
-        self.ana = am.Analysis.objects.create(user=self.user, name='test', base_rundir='testdir',
-                deleted=True, securityclass=rm.DataSecurityClass.NOSECURITY)
-        self.user.is_staff = True
-        self.user.save()
-
-    def test_fail_request(self):
-        resp = self.cl.get(self.url)
-        self.assertEqual(resp.status_code, 405)
-        # wrong dict keys
-        resp = self.cl.post(self.url, content_type='application/json', data={'hello': 'test'})
-        self.assertEqual(resp.status_code, 400)
+        self.assertTrue(jm.Job.objects.filter(funcname='create_pdc_archive',
+            kwargs={'sfloc_id': self.anasfile_sfl.pk, 'isdir': False}).exists())
 
-        # No analysis
-        resp = self.cl.post(self.url, content_type='application/json',
-                data={'item_id': self.ana.pk + 1000})
-        self.assertEqual(resp.status_code, 403)
-        self.assertIn('Analysis does not exist', resp.json()['error'])
+    def test_delete_without_backup(self):
+        rm.PDCBackedupFile.objects.create(storedfile=self.anasfile, success=True)
+        resp = self.cl.post(self.url, content_type='application/json', data={'item_id': self.ana.pk})
+        self.assertEqual(resp.status_code, 200)
+        self.assertFalse(self.ana.deleted)
+        self.ana.refresh_from_db()
+        self.assertTrue(self.ana.deleted)
+        self.assertFalse(jm.Job.objects.filter(funcname='create_pdc_archive',
+            kwargs={'sfloc_id': self.anasfile_sfl.pk, 'isdir': False}).exists())
 
-        # Not allowed wrong user
-        self.user.is_staff = False
-        self.user.save()
+    def test_delete_without_backup_pending(self):
+        rm.PDCBackedupFile.objects.create(storedfile=self.anasfile, success=False)
+        job = jm.Job.objects.create(funcname='create_pdc_archive', timestamp=timezone.now(),
+            kwargs={'sfloc_id': self.anasfile_sfl.pk, 'isdir': False}, state=jj.Jobstates.PENDING)
         resp = self.cl.post(self.url, content_type='application/json', data={'item_id': self.ana.pk})
-        self.assertEqual(resp.status_code, 403)
-        self.assertIn('Only admin is authorized to purge analysis', resp.json()['error'])
-        
-        # analysis not deleted
-        self.user.is_staff = True
-        self.user.save()
-        self.ana.deleted = False
-        self.ana.save()
+        self.assertEqual(resp.status_code, 200)
+        self.assertFalse(self.ana.deleted)
+        self.ana.refresh_from_db()
+        self.assertTrue(self.ana.deleted)
+        self.assertFalse(jm.Job.objects.exclude(pk=job.pk).filter(funcname='create_pdc_archive',
+            kwargs={'sfloc_id': self.anasfile_sfl.pk, 'isdir': False}).exists())
+
+    def test_delete_fail_backup(self):
+        rm.PDCBackedupFile.objects.create(storedfile=self.anasfile, success=False)
+        job = jm.Job.objects.create(funcname='create_pdc_archive', timestamp=timezone.now(),
+            kwargs={'sfloc_id': self.anasfile_sfl.pk, 'isdir': False}, state=jj.Jobstates.WAITING)
         resp = self.cl.post(self.url, content_type='application/json', data={'item_id': self.ana.pk})
-        self.assertEqual(resp.status_code, 403)
-        self.assertIn('Analysis is not deleted', resp.json()['error'])
+        self.assertEqual(resp.status_code, 409)
+        self.assertFalse(self.ana.deleted)
+        self.ana.refresh_from_db()
+        self.assertFalse(self.ana.deleted)
+        self.assertFalse(jm.Job.objects.exclude(pk=job.pk).filter(funcname='create_pdc_archive',
+            kwargs={'sfloc_id': self.anasfile_sfl.pk, 'isdir': False}).exists())

src/backend/analysis/urls.py

@@ -13,7 +13,6 @@
     path('freeze/', views.freeze_analysis),
     path('unfreeze/', views.unfreeze_analysis),
     path('undelete/', views.undelete_analysis),
-    path('purge/', views.purge_analysis),
     path('dsets/<int:wfversion_id>/', views.get_datasets),
     path('baseanalysis/show/', views.get_base_analyses),
     path('baseanalysis/load/<int:wfversion_id>/<int:baseanid>/', views.load_base_analysis),

src/backend/analysis/views.py

@@ -20,7 +20,7 @@
 from rawstatus import models as rm
 from jobs import jobs as jj
 from jobs import views as jv
-from jobs.jobutil import create_job, jobmap
+from jobs.jobutil import create_job, check_job_error, jobmap, create_job_without_check
 from jobs import models as jm
 
 
@@ -1320,69 +1320,104 @@ def undelete_analysis(request):
 
 @login_required
 def delete_analysis(request):
+    # FIXME analysis can be properly deleted if its backed up
+    # right now this is only mark for deletion and it was for a previous
+    # setup where delete meant gone for ever
+    # There is a lot of old analyses that should be backed up or purged properly
+    # Maybe delete all those where no project is active anymore in delete_expired?
     if request.method != 'POST':
         return JsonResponse({'error': 'Must use POST'}, status=405)
     req = json.loads(request.body.decode('utf-8'))
     try:
-        analysis = am.Analysis.objects.select_related('nextflowsearch__job').get(pk=req['item_id'])
+        analysis = am.Analysis.objects.get(pk=req['item_id'])
     except am.Analysis.DoesNotExist:
         return JsonResponse({'error': 'Analysis does not exist'}, status=403)
     except KeyError:
         return JsonResponse({'error': 'Bad request'}, status=400)
     if analysis.deleted:
         return JsonResponse({'error': 'Analysis is already deleted'}, status=403)
-    if analysis.user == request.user or request.user.is_staff:
-        if not analysis.deleted:
-            analysis.deleted = True
-            analysis.save()
-            am.AnalysisDeleted.objects.update_or_create(analysis=analysis)
-            if hasattr(analysis, 'nextflowsearch'):
-                jobq = jm.Job.objects.filter(nextflowsearch__analysis=analysis)
-                jv.cancel_or_revoke_job(jobq)
-        return JsonResponse({})
-    else:
+    if analysis.user != request.user and not request.user.is_staff:
         return JsonResponse({'error': 'User is not authorized to delete this analysis'}, status=403)
+    if errmsg := do_analysis_deletion(analysis):
+        return JsonResponse({'error': f'Could not delete analysis, {errmsg}'}, status=409)
+    return JsonResponse({})
 
 
-@login_required
-def purge_analysis(request):
-    if request.method != 'POST':
-        return JsonResponse({'error': 'Must use POST'}, status=405)
-    elif not request.user.is_staff:
-        return JsonResponse({'error': 'Only admin is authorized to purge analysis'}, status=403)
-    req = json.loads(request.body.decode('utf-8'))
-    try:
-        analysis = am.Analysis.objects.get(pk=req['item_id'])
-    except am.Analysis.DoesNotExist:
-        return JsonResponse({'error': 'Analysis does not exist'}, status=403)
-    except KeyError:
-        return JsonResponse({'error': 'Bad request'}, status=400)
-    if not analysis.deleted:
-        return JsonResponse({'error': 'Analysis is not deleted, cannot purge'}, status=403)
-    analysis.purged = True
-    analysis.save()
-    webshares = rm.ServerShare.objects.filter(function=rm.ShareFunction.REPORTS)
-    # Delete files on web share here since the job tasks run on storage cannot do that
-    webfiles = rm.StoredFileLoc.objects.filter(sfile__analysisresultfile__analysis__id=analysis.pk,
-            servershare__in=webshares, active=True)
-    for webfile in webfiles.values('path', 'sfile__filename'):
-        fpath = os.path.join(settings.WEBSHARE, webfile['path'], webfile['sfile__filename'])
-        os.unlink(fpath)
-    webfiles.update(active=False, purged=True)
-    sfiles = rm.StoredFile.objects.filter(analysisresultfile__analysis__id=analysis.pk)
-    sfiles.update(deleted=True)
-    sfls = rm.StoredFileLoc.objects.filter(sfile__in=sfiles, active=True)
-    # One job per servershare as empty dir deleter needs that
-    share_pks = defaultdict(list)
-    for sfl in sfls.values('pk', 'servershare_id', 'path'):
-        share_pks[f'{sfl["servershare_id"]}__{sfl["path"]}'].append(sfl['pk'])
-    for share_path, sfloc_ids in share_pks.items():
-        shareid, path = share_path.split('__')
-        purgejob = create_job('purge_files', sfloc_ids=sfloc_ids)
-        if not purgejob['error']:
+def do_analysis_deletion(analysis):
+    '''Function to call when deleting analysis. Also used when 
+    projects are deleted. Checks if analysis is backed up'''
+    # FIXME analysisdeleted tracking no longer needed when we backup etc
+    # implement a log instead
+    am.AnalysisDeleted.objects.update_or_create(analysis=analysis)
+    if jobq := jm.Job.objects.filter(nextflowsearch__analysis=analysis).exclude(state__in=[
+                jj.Jobstates.ERROR, jj.Jobstates.DONE, jj.Jobstates.REVOKING, jj.Jobstates.CANCELED]):
+        jv.cancel_or_revoke_job(jobq)
+    if analysis.success_completed:
+        # Back up files if that for some reason (old analysis) hasnt happened earlier
+        backup_jobs = []
+        for sf in rm.StoredFile.objects.filter(analysisresultfile__analysis=analysis).exclude(
+                pdcbackedupfile__success=True).values('pk', 'filetype__is_folder'):
+            sfl_pks = [x['pk'] for x in rm.StoredFileLoc.objects.filter(sfile_id=sf['pk']).values('pk')]
+            do_backup = True
+            for exist_job in jm.Job.objects.filter(funcname='create_pdc_archive',
+                    kwargs__sfloc_id__in=sfl_pks).values('state'):
+                match exist_job['state']:
+                    case jj.Jobstates.WAITING | jj.Jobstates.HOLD:
+                        do_backup = False
+                        return 'Result file queued for backing up but job not running, please check'
+                    case x if x in jj.JOBSTATES_JOB_ACTIVE:
+                        do_backup = False
+                        continue
+                    case jj.Jobstates.ERROR:
+                        return 'Result file queued for backing up but job errored, please check'
+                    # If no backup exists and job is somehow in done/revoking/canceled,
+                    # keep looking for more jobs, if none exist -> do a backup
+            if do_backup and (sfl := rm.StoredFileLoc.objects.filter(pk__in=sfl_pks, active=True,
+                    servershare__fileservershare__server__can_backup=True).values('pk')):
+                jobkw = {'sfloc_id': sfl.first()['pk'], 'isdir': sf['filetype__is_folder']}
+                backup_jobs.append(jobkw)
+                if joberror := check_job_error('create_pdc_archive', **jobkw):
+                    return f'errors backing up result file(s): {joberror}'
+            elif do_backup:
+                return 'could not find copy of result file to back up'
+        for bupjob in backup_jobs:
+            create_job_without_check('create_pdc_archive', **bupjob)
+
+        # Delete files on web share here since the job tasks run on storage cannot do that
+        webshares = rm.ServerShare.objects.filter(function=rm.ShareFunction.REPORTS)
+        webfiles = rm.StoredFileLoc.objects.filter(sfile__analysisresultfile__analysis__id=analysis.pk,
+                servershare__in=webshares, active=True)
+        for webfile in webfiles.values('path', 'sfile__filename'):
+            fpath = os.path.join(settings.WEBSHARE, webfile['path'], webfile['sfile__filename'])
+            os.unlink(fpath)
+        webfiles.update(active=False, purged=True)
+        sfiles = rm.StoredFile.objects.filter(analysisresultfile__analysis__id=analysis.pk)
+        sfls = rm.StoredFileLoc.objects.filter(sfile__in=sfiles, active=True)
+        # One job per servershare as empty dir deleter needs that
+        share_pks = defaultdict(list)
+        for sfl in sfls.values('pk', 'servershare_id', 'path'):
+            share_pks[f'{sfl["servershare_id"]}__{sfl["path"]}'].append(sfl['pk'])
+        purgejobs, rmdirjobs = [], []
+        for share_path, sfloc_ids in share_pks.items():
+            shareid, path = share_path.split('__')
+            purgejobs.append(sfloc_ids)
+            if joberror := check_job_error('purge_files', sfloc_ids=sfloc_ids):
+                return f'error trying to queue delete files job: {joberror}'
+            rmdirkw = {'sfloc_ids': sfloc_ids, 'path': path, 'share_id': shareid}
+            rmdirjobs.append(rmdirkw)
+            if joberror := check_job_error('delete_empty_directory', sfloc_ids=sfloc_ids):
+                return f'error trying to queue delete files job: {joberror}'
+        for sfloc_ids in purgejobs:
+            create_job_without_check('purge_files', sfloc_ids=sfloc_ids)
             rm.StoredFileLoc.objects.filter(pk__in=sfloc_ids).update(active=False)
-            rmdirjob = create_job('delete_empty_directory', sfloc_ids=sfloc_ids, path=path, share_id=shareid)
-    return JsonResponse({})
+        for rmdirkw in rmdirjobs:
+            create_job_without_check('delete_empty_directory', **rmdirkw)
+        sfiles.update(deleted=True)
+
+    # No more errors, mark for deletion
+    analysis.deleted, analysis.purged = True, True
+    analysis.save()
+    return 0
 
 
 @login_required

src/backend/jobs/views.py

@@ -316,7 +316,7 @@ def analysis_run_done(request):
     if 'task' in data:
         set_task_done(data['task'])
     # Defensively, since analysis should already be not editable upon launch:
-    am.Analysis.objects.filter(pk=data['analysis_id']).update(editable=False)
+    am.Analysis.objects.filter(pk=data['analysis_id']).update(editable=False, success_completed=True)
     ana = am.Analysis.objects.values('user', 'name').get(pk=data['analysis_id'])
     hm.UserMessage.create_message(ana['user'], msgtype=hm.AnalysisMsgTypes.COMPLETED,
             analysis_id=data['analysis_id'])

src/backend/kantele/tests.py

@@ -258,8 +258,8 @@ def setUp(self):
         self.uft = rm.StoredFileType.objects.create(name='ufileft', filetype='tst', is_rawdata=False)
 
         # Analysis files
-        ana = am.Analysis.objects.create(name='anatesttest', user=self.user, base_rundir='fakerundir',
-                securityclass=rm.DataSecurityClass.NOSECURITY)
+        self.ana = am.Analysis.objects.create(name='anatesttest', user=self.user, base_rundir='fakerundir',
+                securityclass=rm.DataSecurityClass.NOSECURITY, success_completed=True)
         anaft = rm.StoredFileType.objects.create(name=settings.ANALYSIS_FT_NAME, filetype='ana',
                 is_rawdata=False)
         self.anaprod = rm.Producer.objects.create(name='analysisprod', client_id=settings.ANALYSISCLIENT_APIKEY, shortname=settings.PRODUCER_ANALYSIS_NAME)
@@ -269,7 +269,7 @@ def setUp(self):
         self.anasfile = rm.StoredFile.objects.create(rawfile=self.ana_raw, filetype=anaft,
 
                 filename=self.ana_raw.name, md5=self.ana_raw.source_md5)
-        am.AnalysisResultFile.objects.create(analysis=ana, sfile=self.anasfile)
+        am.AnalysisResultFile.objects.create(analysis=self.ana, sfile=self.anasfile)
 
         self.anasfile_sfl = rm.StoredFileLoc.objects.create(sfile=self.anasfile,
                 servershare=self.ssana, path='', active=True, purged=False)

src/frontend/home/src/Analysis.svelte

@@ -113,15 +113,6 @@ async function unDeleteAnalyses() {
   }
   updateNotif()
 }
-
-
-async function purgeAnalyses() {
-  for (let anid of selectedAnalyses) {
-    await treatItems('/analysis/purge/', 'analysis', 'purging', anid, notif);
-    refreshAnalysis(anid);
-  }
-  updateNotif()
-}
 </script>
 
 <Tabs tabshow="Analyses" notif={notif} />
@@ -130,11 +121,9 @@ async function purgeAnalyses() {
 {#if selectedAnalyses.length}
 <a class="button is-small" on:click={deleteAnalyses}>Delete analyses</a>
 <a class="button is-small" on:click={unDeleteAnalyses}>Undelete analyses</a>
-<a class="button is-small" on:click={purgeAnalyses}>Purge analyses</a>
 {:else}
 <a class="button is-small" disabled>Delete analyses</a>
 <a class="button is-small" disabled>Undelete analyses</a>
-<a class="button is-small" disabled>Purge analyses</a>
 {/if}
 
 <Table tab="Analyses"