[2.x] feat: pausable queues

framework/core/js/src/admin/AdminApplication.tsx

@@ -94,6 +94,8 @@ export interface AdminApplicationData extends ApplicationData {
   dbOptions: Record<string, string>;
   phpVersion: string;
   queueDriver: string;
+  pausedQueues: string[];
+  knownQueues: string[];
   schedulerStatus: string;
   sessionDriver: string;
 }

framework/core/js/src/admin/components/AdvancedPage.tsx

@@ -10,6 +10,7 @@ import ItemList from '../../common/utils/ItemList';
 import InfoTile from '../../common/components/InfoTile';
 import { MaintenanceMode } from '../../common/Application';
 import Button from '../../common/components/Button';
+import Switch from '../../common/components/Switch';
 import classList from '../../common/utils/classList';
 import ExtensionBisect from './ExtensionBisect';
 import { DatabaseDriver } from '../AdminApplication';
@@ -215,9 +216,58 @@ export default class AdvancedPage<CustomAttrs extends IPageAttrs = IPageAttrs> e
       items.add('content', this.queueCustomContent(), 100);
     }
 
+    if (driver !== 'sync') {
+      items.add('pause', this.queuePauseControl(), 110);
+    }
+
     return items;
   }
 
+  queuePauseControl() {
+    const paused = app.data.pausedQueues || [];
+    const knownQueues = app.data.knownQueues || ['default'];
+    const allPaused = paused.includes('*');
+
+    const toggle = (queue: string, value: boolean) =>
+      app
+        .request({
+          method: 'POST',
+          url: app.forum.attribute('apiUrl') + '/queue/pause',
+          body: { paused: value, queue },
+        })
+        .then(() => {
+          app.data.pausedQueues = value ? [...paused, queue] : paused.filter((name) => name !== queue);
+          m.redraw();
+        });
+
+    // With a single known queue, separate "all" and per-queue switches are
+    // redundant — offer one switch that pauses everything.
+    if (knownQueues.length <= 1) {
+      return (
+        <div className="Form-group AdvancedPage-queuePause">
+          <Switch state={allPaused || paused.includes(knownQueues[0])} onchange={(value: boolean) => toggle('*', value)}>
+            {app.translator.trans('core.admin.advanced.queue.pause_label')}
+          </Switch>
+          <p className="helpText">{app.translator.trans('core.admin.advanced.queue.pause_help')}</p>
+        </div>
+      );
+    }
+
+    return (
+      <div className="Form-group AdvancedPage-queuePause">
+        <Switch state={allPaused} onchange={(value: boolean) => toggle('*', value)}>
+          {app.translator.trans('core.admin.advanced.queue.pause_all_label')}
+        </Switch>
+        {knownQueues.map((queue) => (
+          <Switch state={allPaused || paused.includes(queue)} disabled={allPaused} onchange={(value: boolean) => toggle(queue, value)}>
+            {app.translator.trans('core.admin.advanced.queue.pause_queue_label', { queue })}
+          </Switch>
+        ))}
+        <p className="helpText">{app.translator.trans('core.admin.advanced.queue.pause_help')}</p>
+      </div>
+    );
+  }
+
   queueSyncContent() {
     return (
       <InfoTile icon="fas fa-bolt" className="InfoTile--muted">

framework/core/js/src/admin/components/DashboardPage.tsx

@@ -46,6 +46,28 @@ export default class DashboardPage extends AdminPage {
       );
     }
 
+    if (app.data.pausedQueues?.length) {
+      items.add(
+        'queuePaused',
+        <AlertWidget
+          alert={{
+            type: 'warning',
+            dismissible: false,
+            controls: [
+              <Link className="Button Button--link" href={app.route('advanced')}>
+                {app.translator.trans('core.admin.dashboard.queue_paused_manage')}
+              </Link>,
+            ],
+          }}
+        >
+          {app.data.pausedQueues.includes('*')
+            ? app.translator.trans('core.admin.dashboard.queue_paused_all_warning')
+            : app.translator.trans('core.admin.dashboard.queue_paused_warning', { queues: app.data.pausedQueues.join(', ') })}
+        </AlertWidget>,
+        105
+      );
+    }
+
     if (app.data.maintenanceMode) {
       items.add(
         'maintenanceMode',

framework/core/locale/core.yml

@@ -69,6 +69,13 @@ core:
       pgsql:
         search_configuration: Search configuration to use
       queue:
+        pause_label: Pause queue processing
+        pause_all_label: Pause all queues
+        pause_queue_label: "Pause the {queue} queue"
+        pause_help: >-
+          While paused, workers stop picking up new jobs from the affected
+          queues — jobs accumulate until processing is resumed. Jobs that are
+          already running are not affected.
         section_label: Queue
         sync_info: Your forum is using the synchronous queue driver. Jobs are processed immediately in the main process thread.
         sync_help: For better performance and user experience, consider using the database queue driver. Learn more in the documentation.
@@ -194,6 +201,9 @@ core:
 
     # These translations are used in the Dashboard page.
     dashboard:
+      queue_paused_warning: "Queue processing is paused (queue: {queues}). Jobs will accumulate until it is resumed."
+      queue_paused_all_warning: All queue processing is paused. Jobs will accumulate until it is resumed.
+      queue_paused_manage: Manage
       clear_cache_button: Clear Cache
       description: Your forum at a glance.
       info_button: System Info

framework/core/src/Admin/Content/AdminPayload.php

@@ -30,6 +30,24 @@
 
 class AdminPayload
 {
+    /**
+     * The names of the queues currently paused on the active connection —
+     * powers the dashboard warning and the Advanced page toggle, regardless
+     * of the queue backend in use.
+     *
+     * @return string[]
+     */
+    protected function pausedQueues(): array
+    {
+        /** @var \Flarum\Queue\QueueFactory $factory */
+        $factory = $this->container->make(\Illuminate\Contracts\Queue\Factory::class);
+
+        /** @var \Illuminate\Contracts\Queue\Queue $queue */
+        $queue = $this->container->make(\Illuminate\Contracts\Queue\Queue::class);
+
+        return $factory->pausedQueues((string) $queue->getConnectionName());
+    }
+
     public function __construct(
         protected Container $container,
         protected SettingsRepositoryInterface $settings,
@@ -84,6 +102,8 @@ public function __invoke(Document $document, Request $request): void
 
         $document->payload['schedulerStatus'] = $this->appInfo->getSchedulerStatus();
         $document->payload['queueDriver'] = $this->appInfo->identifyQueueDriver();
+        $document->payload['pausedQueues'] = $this->pausedQueues();
+        $document->payload['knownQueues'] = $this->container->make('flarum.queue.queues');
         $document->payload['sessionDriver'] = $this->appInfo->identifySessionDriver(true);
 
         /**

framework/core/src/Api/Controller/PauseQueueController.php

@@ -0,0 +1,54 @@
+<?php
+
+/*
+ * This file is part of Flarum.
+ *
+ * For detailed copyright and license information, please view the
+ * LICENSE file that was distributed with this source code.
+ */
+
+namespace Flarum\Api\Controller;
+
+use Flarum\Http\RequestUtil;
+use Flarum\Queue\QueueFactory;
+use Illuminate\Contracts\Queue\Queue;
+use Illuminate\Queue\SyncQueue;
+use Illuminate\Support\Arr;
+use Laminas\Diactoros\Response\EmptyResponse;
+use Psr\Http\Message\ResponseInterface;
+use Psr\Http\Message\ServerRequestInterface;
+use Psr\Http\Server\RequestHandlerInterface;
+
+class PauseQueueController implements RequestHandlerInterface
+{
+    public function __construct(
+        protected QueueFactory $factory,
+        protected Queue $queue
+    ) {
+    }
+
+    public function handle(ServerRequestInterface $request): ResponseInterface
+    {
+        RequestUtil::getActor($request)->assertAdmin();
+
+        // Sync jobs execute inline and never pass through the worker, so
+        // there is nothing to pause.
+        if ($this->queue instanceof SyncQueue) {
+            return new EmptyResponse(409);
+        }
+
+        $body = $request->getParsedBody();
+        $paused = (bool) Arr::get($body, 'paused', true);
+        $queue = (string) Arr::get($body, 'queue', 'default');
+
+        $connection = $this->queue->getConnectionName();
+
+        if ($paused) {
+            $this->factory->pause($connection, $queue);
+        } else {
+            $this->factory->resume($connection, $queue);
+        }
+
+        return new EmptyResponse(204);
+    }
+}

framework/core/src/Api/routes.php

@@ -190,6 +190,13 @@
         $route->toController(Controller\ClearCacheController::class)
     );
 
+    // Pause or resume queue processing
+    $map->post(
+        '/queue/pause',
+        'queue.pause',
+        $route->toController(Controller\PauseQueueController::class)
+    );
+
     // List available mail drivers, available fields and validation status
     $map->get(
         '/mail/settings',

framework/core/src/Queue/Console/PauseCommand.php

@@ -0,0 +1,61 @@
+<?php
+
+/*
+ * This file is part of Flarum.
+ *
+ * For detailed copyright and license information, please view the
+ * LICENSE file that was distributed with this source code.
+ */
+
+namespace Flarum\Queue\Console;
+
+use Illuminate\Console\Command;
+use Illuminate\Contracts\Queue\Factory;
+use Illuminate\Contracts\Queue\Queue;
+use Illuminate\Queue\Worker;
+use Symfony\Component\Console\Attribute\AsCommand;
+
+/**
+ * Flarum's counterpart to Illuminate's queue:pause: the upstream command
+ * type-hints the concrete QueueManager, which Flarum replaces with its own
+ * factory. The queue argument accepts `queue` or `connection:queue`; the
+ * connection defaults to the active queue connection's name.
+ */
+#[AsCommand(name: 'queue:pause')]
+class PauseCommand extends Command
+{
+    protected $signature = 'queue:pause {queue=default : The name of the queue to pause, optionally prefixed with a connection (connection:queue)} {--all : Pause all queues on the connection}';
+
+    protected $description = 'Pause processing of jobs on a queue';
+
+    public function handle(Factory $factory, Queue $connection): int
+    {
+        if (! Worker::$pausable) {
+            $this->components->error('Queue pausing is currently disabled.');
+
+            return 1;
+        }
+
+        [$connectionName, $queue] = $this->option('all')
+            ? [$connection->getConnectionName(), '*']
+            : $this->parseQueue($this->argument('queue'), $connection);
+
+        $factory->pause($connectionName, $queue);
+
+        $this->components->info("Job processing on queue [{$connectionName}:{$queue}] has been paused.");
+
+        return 0;
+    }
+
+    /**
+     * @return array{string, string}
+     */
+    protected function parseQueue(string $argument, Queue $connection): array
+    {
+        if (str_contains($argument, ':')) {
+            return explode(':', $argument, 2);
+        }
+
+        return [$connection->getConnectionName(), $argument];
+    }
+}

framework/core/src/Queue/Console/ResumeCommand.php

@@ -0,0 +1,52 @@
+<?php
+
+/*
+ * This file is part of Flarum.
+ *
+ * For detailed copyright and license information, please view the
+ * LICENSE file that was distributed with this source code.
+ */
+
+namespace Flarum\Queue\Console;
+
+use Illuminate\Console\Command;
+use Illuminate\Contracts\Queue\Factory;
+use Illuminate\Contracts\Queue\Queue;
+use Symfony\Component\Console\Attribute\AsCommand;
+
+/**
+ * Flarum's counterpart to Illuminate's queue:resume — see PauseCommand for
+ * why the upstream commands cannot be reused directly.
+ */
+#[AsCommand(name: 'queue:resume')]
+class ResumeCommand extends Command
+{
+    protected $signature = 'queue:resume {queue=default : The name of the queue to resume, optionally prefixed with a connection (connection:queue)} {--all : Resume all queues on the connection}';
+
+    protected $description = 'Resume processing of jobs on a paused queue';
+
+    public function handle(Factory $factory, Queue $connection): int
+    {
+        [$connectionName, $queue] = $this->option('all')
+            ? [$connection->getConnectionName(), '*']
+            : $this->parseQueue($this->argument('queue'), $connection);
+
+        $factory->resume($connectionName, $queue);
+
+        $this->components->info("Job processing on queue [{$connectionName}:{$queue}] has been resumed.");
+
+        return 0;
+    }
+
+    /**
+     * @return array{string, string}
+     */
+    protected function parseQueue(string $argument, Queue $connection): array
+    {
+        if (str_contains($argument, ':')) {
+            return explode(':', $argument, 2);
+        }
+
+        return [$connection->getConnectionName(), $argument];
+    }
+}