Add poller queue

ESSArch_Core/WorkflowEngine/migrations/0089_alter_processstep_queue_alter_processtask_queue.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.2.14 on 2026-06-01 17:19
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('WorkflowEngine', '0088_remove_processstep_type_processstep_label_and_more'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='processstep',
+            name='queue',
+            field=models.CharField(blank=True, default='celery', max_length=255, null=True),
+        ),
+        migrations.AlterField(
+            model_name='processtask',
+            name='queue',
+            field=models.CharField(blank=True, default='celery', max_length=255, null=True),
+        ),
+    ]

ESSArch_Core/WorkflowEngine/models.py

@@ -113,7 +113,7 @@ class Meta:
     celery_id = models.UUIDField(default=uuid.uuid4, unique=True)
     name = models.CharField(max_length=255, db_index=True)
     label = models.CharField(max_length=255, blank=True)
-    queue = models.CharField(max_length=255, blank=True, null=True, default=None)
+    queue = models.CharField(max_length=255, blank=True, null=True, default='celery')
     hidden = models.BooleanField(null=True, default=None, db_index=True)
     eager = models.BooleanField(default=True)
     time_created = models.DateTimeField(auto_now_add=True)

ESSArch_Core/WorkflowEngine/serializers.py

@@ -124,7 +124,7 @@ class Meta:
             'processstep', 'processstep_pos', 'time_created', 'time_started',
             'time_done', 'retried',
             'responsible', 'hidden', 'args', 'params', 'information_package',
-            'information_package_str', 'eager',
+            'information_package_str', 'eager', 'queue',
         )
         read_only_fields = (
             'time_created', 'time_started', 'time_done', 'retried',
@@ -234,7 +234,7 @@ class Meta:
             'url', 'id', 'name', 'label', 'result', 'part_root', 'user', 'parallel',
             'run_state', 'status', 'progress', 'time_created', 'parent',
             'parent_pos', 'information_package', 'information_package_str',
-            'flow_type', 'step_position', 'responsible',
+            'flow_type', 'step_position', 'responsible', 'queue',
         )
         read_only_fields = (
             'status', 'progress', 'time_created', 'time_done',

ESSArch_Core/WorkflowEngine/util.py

@@ -60,6 +60,7 @@ def _create_on_error_tasks(parent, errors, ip=None, responsible=None, eager=Fals
             processstep_pos=on_error_idx,
             status=status,
             progress=progress,
+            queue=on_error.get('queue') or parent.queue
         )
 
 
@@ -89,6 +90,7 @@ def _create_step(parent, flow, ip, responsible, context=None):
                 information_package=ip,
                 context=context,
                 responsible=responsible,
+                queue=flow_entry.get('queue') or parent.queue,
             )
 
             on_error_tasks = list(_create_on_error_tasks(
@@ -111,7 +113,7 @@ def _create_step(parent, flow, ip, responsible, context=None):
             result_params = flow_entry.get('result_params', {})
             task = ProcessTask.objects.create(
                 name=name,
-                queue=flow_entry.get('queue', None),
+                queue=flow_entry.get('queue') or parent.queue,
                 reference=flow_entry.get('reference', None),
                 label=flow_entry.get('label'),
                 args=args,
@@ -157,6 +159,14 @@ def create_workflow(workflow_spec=None, ip=None, workflow_steps=None, name='', l
 
     logger = logging.getLogger('essarch.workflow')
 
+    queue = None
+
+    if workflow_spec:
+        queue = workflow_spec[0].get('queue')
+
+    if queue is None:
+        queue = context.get('WORKFLOW_QUEUE', 'celery')
+
     with cache.lock('create_workflow_lock', timeout=300):
         try:
             for attempt in Retrying(stop=stop_after_delay(30),
@@ -174,15 +184,15 @@ def create_workflow(workflow_spec=None, ip=None, workflow_steps=None, name='', l
                                 root_step = ProcessStep(
                                     name=name, eager=eager, information_package=ip, context=context,
                                     responsible=responsible, label=label, part_root=part_root,
-                                    run_state=run_state)
+                                    run_state=run_state, queue=queue)
                                 root_step.parent = top_root_step
                                 root_step.parent_pos = top_root_step.child_steps.count() + 1
                                 root_step.save()
                             else:
                                 root_step = ProcessStep.objects.create(
                                     name=name, eager=eager, information_package=ip, context=context,
                                     responsible=responsible, label=label, part_root=part_root,
-                                    run_state=run_state)
+                                    run_state=run_state, queue=queue)
 
                             # Create on_error tasks
                             on_error_tasks = list(_create_on_error_tasks(

ESSArch_Core/config/settings.py

@@ -596,8 +596,15 @@
 CELERY_TASK_TRACK_STARTED = True
 CELERY_WORKER_PREFETCH_MULTIPLIER = 1
 CELERY_WORKER_CANCEL_LONG_RUNNING_TASKS_ON_CONNECTION_LOSS = True
-ESSARCH_MAX_RUNNING_STEPS = 10
 ESSARCH_RUN_SINGLE_ROOT_STEP_IF_RUNNING_PART_ROOT_STEP = True
+ESSARCH_STEP_QUEUES = {
+    'celery': {
+        'max_running_steps': 10,
+    },
+    'poller': {
+        'max_running_steps': 10,
+    }
+}
 
 CELERY_BEAT_SCHEDULE = {
     'RunWorkflowPollers-every-60-seconds': {

ESSArch_Core/frontend/static/frontend/views/modals/step_info_modal.html

@@ -24,7 +24,7 @@
     <div class="row">
       <div class="col-12 col-md-6" ng-if="currentStepTask.name">
         <label>{{ "NAME" | translate }}</label>
-        <p class="white-space-nowrap m-0 py-base">{{ currentStepTask.name }}</p>
+        <p class="white-space-nowrap m-0 py-base">{{ currentStepTask.name }} ({{currentStepTask.queue}})</p>
       </div>
       <div class="col-12 col-md-6">
         <label>{{ "IDENTIFICATION" | translate }}</label>

ESSArch_Core/frontend/static/frontend/views/modals/task_info_modal.html

@@ -24,7 +24,7 @@
     <div class="row">
       <div class="col-12 col-md-6" ng-if="currentStepTask.name">
         <label>{{ "NAME" | translate }}</label>
-        <p class="m-0 py-base word-break-word">{{ currentStepTask.name }}</p>
+        <p class="m-0 py-base word-break-word">{{ currentStepTask.name }} ({{currentStepTask.queue}})</p>
       </div>
       <div class="col-12 col-md-6">
         <label>{{ "IDENTIFICATION" | translate }}</label>

ESSArch_Core/storage/serializers.py

@@ -494,7 +494,7 @@ def create(self, validated_data):
                     media_migrate_workflow_step = ProcessStep.objects.create(
                         name='Migrate Storage Medium', eager=False, information_package=None, context={},
                         responsible=user, label='Migrate Storage Medium {}'.format(medium_id),
-                        part_root=None, run_state='')
+                        part_root=None, run_state='', queue='celery')
 
                     t = ProcessTask.objects.create(
                         name='ESSArch_Core.storage.tasks.CreateMediumMigrationWorkflow',
@@ -533,7 +533,7 @@ def create(self, validated_data):
                     media_migrate_workflow_step = ProcessStep.objects.create(
                         name='Migrate Storage Medium', eager=False, information_package=None, context={},
                         responsible=user, label='Migrate Storage Medium {}'.format(StorageMedium_obj.medium_id),
-                        part_root=None, run_state='')
+                        part_root=None, run_state='', queue='celery')
 
                     t = ProcessTask.objects.create(
                         name='ESSArch_Core.storage.tasks.CreateMediumMigrationWorkflow',

ESSArch_Core/storage/tests/test_views.py

@@ -1051,26 +1051,27 @@ def test_migration_task_order_for_information_packages(self, mock_task_run, mock
             label=mock.ANY,
             part_root=mock.ANY,
             run_state=mock.ANY,
+            queue=mock.ANY,
         ))
         for ip in [ips[0], ips[5], ips[3], ips[1], ips[2], ips[4]]:
             # calls.append(mock.call(name='Migrate Information Package', eager=False, information_package=ip,
             #                        context=mock.ANY, responsible=mock.ANY, label=mock.ANY, part_root=mock.ANY,
             #                        run_state=mock.ANY))
             calls.append(mock.call(name='Write to storage methods', parallel=mock.ANY, parent=mock.ANY,
                                    parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY,
-                                   responsible=mock.ANY))
+                                   responsible=mock.ANY, queue=mock.ANY))
             calls.append(mock.call(name='Write non-containers', parallel=mock.ANY, parent=mock.ANY,
                                    parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY,
-                                   responsible=mock.ANY))
+                                   responsible=mock.ANY, queue=mock.ANY))
             calls.append(mock.call(name='Write non-containers to storage methods', parallel=mock.ANY, parent=mock.ANY,
                                    parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY,
-                                   responsible=mock.ANY))
+                                   responsible=mock.ANY, queue=mock.ANY))
             calls.append(mock.call(name='Write containers to storage methods', parallel=mock.ANY, parent=mock.ANY,
                                    parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY,
-                                   responsible=mock.ANY))
+                                   responsible=mock.ANY, queue=mock.ANY))
             calls.append(mock.call(name='Delete temporary files', parallel=mock.ANY, parent=mock.ANY,
                                    parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY,
-                                   responsible=mock.ANY))
+                                   responsible=mock.ANY, queue=mock.ANY))
 
         mock_task.assert_has_calls(calls)
 
@@ -1120,26 +1121,27 @@ def test_migration_task_order_for_storage_mediums(self, mock_task_run, mock_task
             label=mock.ANY,
             part_root=mock.ANY,
             run_state=mock.ANY,
+            queue=mock.ANY,
         ))
         for ip in [ips[0], ips[5], ips[3], ips[1], ips[2], ips[4]]:
             # calls.append(mock.call(name='Migrate Information Package', eager=False, information_package=ip,
             #                        context=mock.ANY, responsible=mock.ANY, label=mock.ANY, part_root=mock.ANY,
             #                        run_state=mock.ANY))
             calls.append(mock.call(name='Write to storage methods', parallel=mock.ANY, parent=mock.ANY,
                                    parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY,
-                                   responsible=mock.ANY))
+                                   responsible=mock.ANY, queue=mock.ANY))
             calls.append(mock.call(name='Write non-containers', parallel=mock.ANY, parent=mock.ANY,
                                    parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY,
-                                   responsible=mock.ANY))
+                                   responsible=mock.ANY, queue=mock.ANY))
             calls.append(mock.call(name='Write non-containers to storage methods', parallel=mock.ANY, parent=mock.ANY,
                                    parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY,
-                                   responsible=mock.ANY))
+                                   responsible=mock.ANY, queue=mock.ANY))
             calls.append(mock.call(name='Write containers to storage methods', parallel=mock.ANY, parent=mock.ANY,
                                    parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY,
-                                   responsible=mock.ANY))
+                                   responsible=mock.ANY, queue=mock.ANY))
             calls.append(mock.call(name='Delete temporary files', parallel=mock.ANY, parent=mock.ANY,
                                    parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY,
-                                   responsible=mock.ANY))
+                                   responsible=mock.ANY, queue=mock.ANY))
 
         mock_task.assert_has_calls(calls)
 

ESSArch_Core/tasks.py

@@ -823,9 +823,11 @@ def get_workflows():
             backend = get_backend(name)
             poll_path = poller['path']
             poll_sa = poller.get('sa')
+            poll_queue = poller.get('queue', 'poller')
             context = {
                 'WORKFLOW_POLLER': name,
-                'WORKFLOW_POLL_PATH': poll_path
+                'WORKFLOW_POLL_PATH': poll_path,
+                'WORKFLOW_QUEUE': poll_queue
             }
             for ip in backend.poll(poll_path, poll_sa):
                 profile = ip.submission_agreement.profile_workflow