Allow users to run sync jobs immediately

src/pretix/base/services/datasync.py

@@ -52,6 +52,22 @@ def periodic_reset_in_flight(sender, **kwargs):
         sq.set_sync_error('timeout', [], 'Timeout')
 
 
+def run_sync(queue):
+        grouped = groupby(sorted(queue, key=lambda q: (q.sync_provider, q.event.pk)), lambda q: (q.sync_provider, q.event))
+        for (target, event), queued_orders in grouped:
+            target_cls, meta = datasync_providers.get(identifier=target, active_in=event)
+
+            if not target_cls:
+                # sync plugin not found (plugin deactivated or uninstalled) -> drop outstanding jobs
+                num_deleted, _ = OrderSyncQueue.objects.filter(pk__in=[sq.pk for sq in queued_orders]).delete()
+                logger.info("Deleted %d queue entries from %r because plugin %s inactive", num_deleted, event, target)
+                continue
+
+            with scope(organizer=event.organizer):
+                with target_cls(event=event) as p:
+                    p.sync_queued_orders(queued_orders)
+
+
 @app.task()
 def sync_all():
     with scopes_disabled():
@@ -70,16 +86,20 @@ def sync_all():
             .prefetch_related("event")
             [:1000]
         )
-        grouped = groupby(sorted(queue, key=lambda q: (q.sync_provider, q.event.pk)), lambda q: (q.sync_provider, q.event))
+        run_sync(queue)
-        for (target, event), queued_orders in grouped:
-            target_cls, meta = datasync_providers.get(identifier=target, active_in=event)
 
-            if not target_cls:
-                # sync plugin not found (plugin deactivated or uninstalled) -> drop outstanding jobs
-                num_deleted, _ = OrderSyncQueue.objects.filter(pk__in=[sq.pk for sq in queued_orders]).delete()
-                logger.info("Deleted %d queue entries from %r because plugin %s inactive", num_deleted, event, target)
-                continue
 
-            with scope(organizer=event.organizer):
+@app.task()
-                with target_cls(event=event) as p:
+def sync_single(queue_item_id: int):
-                    p.sync_queued_orders(queued_orders)
+    with scopes_disabled():
+        queue = (
+            OrderSyncQueue.objects
+            .filter(
+                pk=queue_item_id,
+                in_flight=False,
+                not_before__lt=now(),
+                need_manual_retry__isnull=True,
+            )
+            .prefetch_related("event")
+        )
+        run_sync(queue)

src/pretix/control/views/datasync.py

@@ -35,6 +35,7 @@ from django.views.generic import ListView
 from pretix.base.datasync.datasync import datasync_providers
 from pretix.base.models import Event, Order
 from pretix.base.models.datasync import OrderSyncQueue
+from pretix.base.services.datasync import sync_single
 from pretix.control.permissions import (
     AdministratorPermissionRequiredMixin, EventPermissionRequiredMixin,
     OrganizerPermissionRequiredMixin,
@@ -86,10 +87,14 @@ class ControlSyncJob(OrderView):
                 messages.success(self.request, _('The sync job has been canceled.'))
         elif self.request.POST.get("run_job_now"):
             job = self.order.queued_sync_jobs.get(pk=self.request.POST.get("run_job_now"))
-            job.not_before = now()
+            if job.in_flight:
-            job.need_manual_retry = None
+                messages.success(self.request, _('The sync job is already in progress.'))
-            job.save()
+            else:
-            messages.success(self.request, _('The sync job has been set to run as soon as possible.'))
+                job.not_before = now()
+                job.need_manual_retry = None
+                job.save()
+                sync_single.apply_async(args=(job.pk,))
+                messages.success(self.request, _('The sync job has been set to run as soon as possible.'))
 
         return redirect(self.get_order_url())