Hand a whole-view bulk action to a worker, which it never was

RunBulkStatusHandler exists to take a mark-read or archive over a whole view
out of the request — its docblock quotes the thirty-second timeout that
produced it, and BulkStatusController answers with an indicator so nobody
waits. Neither was true: the message was never listed in messenger.yaml's
routing map, and Messenger handles an unrouted message inline in whichever
process dispatched it. Every whole-view action ran inside the web request,
under the limit it was written to escape, at request concurrency — which is
where the production deadlock came from.

It gets a queue of its own rather than joining maintenance, because somebody
is watching an indicator for it and queueing it behind an embedding backfill
would be slower than the inline version it replaces. That means one new
container, worker-bulk, in every compose file and in the handbook.

The deadlock itself is retried now instead of failing the job, three times per
chunk, and a collision that outlasts those hands the job back to the queue
rather than ending it. The failure path no longer writes through an
EntityManager Doctrine has closed, which is what it was doing for every
failure that came out of a flush.

And the list catches up: nothing ever refreshed it on the strength of the job,
only on the toolbar's own "the write landed" event, which is now the instant
the work was QUEUED rather than finished.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.env

@@ -90,6 +90,11 @@ DATABASE_URL="postgresql://app@database:5432/app?serverVersion=18&charset=utf8"
 # MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages
 # MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
 MESSENGER_TRANSPORT_DSN=doctrine://default?auto_setup=0
+# The `bulk` queue only, if it should live somewhere other than the line above.
+# Unset it and it follows MESSENGER_TRANSPORT_DSN, which is what every real
+# deployment wants; it exists so the browser suite can make that one queue a
+# real, consumed one while the rest stay in memory (see compose.test.yaml).
+# MESSENGER_BULK_DSN=doctrine://default?auto_setup=0
 ###< symfony/messenger ###
 
 ###> symfony/mailer ###

assets/controllers/mail/mail_pane_controller.js

@@ -459,6 +459,45 @@ export default class extends Controller {
         this._refreshList();
     }
 
+    /**
+     * A bulk job moved a chunk, or finished.
+     *
+     * THE LIST HAD NO WAY OF FINDING OUT, and for as long as the work happened
+     * inside the request nobody could tell. mail/_job_started.stream.html.twig
+     * says the answer deliberately changes nothing in the list, because "the
+     * list finds out the way it always does, from the writes the job makes" —
+     * except that the only thing that re-read the list after a bulk action was
+     * the toolbar's own `written` event, which fires when the REQUEST lands.
+     * Inline, that was the same instant the work finished. Handed to a worker,
+     * it is the instant the work was queued: the refresh ran against a mailbox
+     * nothing had happened to yet, and the rows sat there afterwards for ever.
+     *
+     * COALESCED WHILE IT RUNS, IMMEDIATE WHEN IT ENDS, and the split is not a
+     * nicety. JobNotifier publishes on every chunk, so a run over five thousand
+     * conversations is fifty of these and the floor in _refreshList() is what
+     * keeps that to one fetch. But the floor is MIN_REFRESH_MS = 15s measured
+     * from the LAST refresh, and the bulk request itself has just caused one —
+     * the toolbar's `written` fires when the response lands, which is now the
+     * moment the work is QUEUED. A job that then finishes a second later had
+     * its nudge deferred by the remaining fourteen, so the list sat showing
+     * mail that was already archived for the rest of that window. It is the
+     * same trap refreshNow() was written for, in a new place: the difference
+     * that matters is whether anybody is waiting, and after "archive all" they
+     * are watching the list to see it happen.
+     *
+     * Per-chunk nudges keep the floor, because those really are a burst and the
+     * rows they move are moving either way.
+     */
+    onJobChanged(event) {
+        if (false === this.hasListTarget) {
+            return;
+        }
+
+        const state = event?.detail?.state;
+
+        this._refreshList({ immediate: "done" === state || "failed" === state });
+    }
+
     /**
      * Somebody pressed a button and is watching the list for the result.
      *

compose.demo.yaml

@@ -73,6 +73,13 @@ services:
     environment:
       <<: *demo-env
 
+  # The demo's mailboxes are small, so a bulk action there is quick — but it
+  # still travels through this queue, and without the consumer the indicator
+  # would sit at nothing for ever.
+  worker-bulk:
+    environment:
+      <<: *demo-env
+
   # Runs app:demo:reap every ten minutes. Drop this service and expired
   # visitors are never deleted — the database then grows by one user, one
   # account, ten threads and ten contacts for every person who followed the

compose.test.yaml

@@ -72,7 +72,24 @@ x-test-app: &test-app
     # Doctrine's when@test dbname_suffix (config/packages/doctrine.yaml) turns
     # this into `app_test` automatically.
     DATABASE_URL: postgresql://app:app@database:5432/app?serverVersion=18&charset=utf8
+    # `in-memory://` for the three shared queues, which means a message routed
+    # to one is kept and never handled — every worker process is a container
+    # this stack does not run, and an in-memory transport is per-process
+    # anyway. That is deliberate and is what keeps the suite deterministic: a
+    # spec is never racing an AI classification, a mailbox resort or a rule
+    # pass that fired in the background.
     MESSENGER_TRANSPORT_DSN: in-memory://
+    # The one exception, and the reason `bulk` has a DSN of its own in
+    # config/packages/messenger.yaml. A whole-view action HAS to actually run
+    # for bulk-select-all.spec.ts to mean anything — it selects the view,
+    # archives it and waits for the rows to go — so this queue is a real
+    # Postgres one with the worker below draining it. Pointing the shared
+    # variable here instead would switch the other three on at the same time.
+    #
+    # auto_setup=0: `messenger_messages` is created by a migration
+    # (Version20260625123836), and app-init has run it before anything else
+    # starts.
+    MESSENGER_BULK_DSN: doctrine://default?auto_setup=0
     MAILER_DSN: null://null
     APP_DEV_USER_EMAIL: e2e@plmail.test
     APP_DEV_USER_PASSWORD: e2e-password-change-me
@@ -192,6 +209,31 @@ services:
       mercure:
         condition: service_healthy
 
+  # Drains the one queue this stack makes real — see MESSENGER_BULK_DSN above.
+  #
+  # `--time-limit` is short because the stack is short-lived and a worker that
+  # has exited is easier to read in `compose ps` than one that is idle; the
+  # restart policy brings it straight back either way. `-vv` so a handler that
+  # throws says why in `compose logs worker-bulk`, which is the only window
+  # onto work that no longer happens inside a request the browser can see.
+  worker-bulk:
+    <<: *test-app
+    command: php bin/console messenger:consume bulk --time-limit=600 --memory-limit=256M -vv
+    restart: unless-stopped
+    stop_grace_period: 5s
+    # The image carries a HEALTHCHECK that curls the web server, which a worker
+    # is not. Left on, `compose up --wait` reports this container unhealthy and
+    # exits non-zero with the stack up and working — which in CI is a red run
+    # for a container doing exactly what it should. Every worker in
+    # compose.yaml disables it for the same reason.
+    healthcheck:
+      disable: true
+    depends_on:
+      database:
+        condition: service_healthy
+      app-init:
+        condition: service_completed_successfully
+
 volumes:
   app_var_test:
   app_public_assets_test:

compose.yaml

@@ -198,6 +198,27 @@ services:
       database:
         condition: service_healthy
 
+  # A whole-view mark-read or archive. On its own process for the reason the
+  # send queue is: somebody pressed a button and is watching an indicator, and
+  # this must not wait behind a backfill on `maintenance`. Drop this service and
+  # the indicator fills to nothing — the job is queued and never handled.
+  worker-bulk:
+    image: *app-image
+    restart: unless-stopped
+    command: php bin/console messenger:consume bulk --time-limit=3600 --memory-limit=256M
+    stop_grace_period: 15s
+    environment:
+      <<: *app-env
+      APP_CONTAINER_NAME: worker-bulk
+    volumes: *app-storage
+    healthcheck:
+      disable: true
+    depends_on:
+      secrets-init:
+        condition: service_completed_successfully
+      database:
+        condition: service_healthy
+
   # Fires the recurring jobs in App\Infrastructure\Scheduler\MaintenanceSchedule.
   # Separate from messenger-worker so a long prune cannot stall the async queue.
   scheduler:

config/packages/messenger.yaml

@@ -1,3 +1,10 @@
+# The DSN every queue uses unless it is given one of its own. A parameter
+# rather than repeating the env() call, because `bulk` below needs to be able
+# to fall back to it — %env(default:…)% takes a PARAMETER name, not a second
+# variable.
+parameters:
+    messenger.default_dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
+
 framework:
     messenger:
         failure_transport: failed
@@ -50,6 +57,40 @@ framework:
                     max_retries: 5
                     max_delay: 300000
 
+            # A whole-view action: mark every unread read, archive the lot.
+            #
+            # ITS OWN QUEUE RATHER THAN maintenance, and the difference is who
+            # is waiting. maintenance is where work goes when nobody is — a
+            # backfill, a sweep — and it says so two entries down. A bulk action
+            # is the opposite: somebody pressed a button, is watching an
+            # indicator fill, and cannot use the list until it does. Queued
+            # behind a ten-minute embedding backfill it would be slower than the
+            # inline version it replaced, which would be a strange thing to ship
+            # as a fix.
+            #
+            # A DSN OF ITS OWN, defaulting to everyone else's. Nothing sets
+            # MESSENGER_BULK_DSN in production and nothing should; it exists so
+            # compose.test.yaml can make THIS queue a real, consumed one while
+            # the other three stay `in-memory://` and unconsumed. Pointing the
+            # shared variable at Postgres instead would switch all four on at
+            # once, and the browser suite would suddenly be running mailbox
+            # resorts, rule passes and AI classification in the middle of specs
+            # about something else.
+            bulk:
+                dsn: '%env(default:messenger.default_dsn:MESSENGER_BULK_DSN)%'
+                options:
+                    queue_name: bulk
+                # The export ladder, for the export reason: somebody is watching
+                # this one. What reaches here is a lock collision the handler's
+                # own three in-process attempts could not clear, so the useful
+                # retry is soon and there is no point in a long tail — a bulk
+                # action nobody is still looking at is one they have given up on.
+                retry_strategy:
+                    delay: 2000
+                    multiplier: 3
+                    max_retries: 5
+                    max_delay: 60000
+
             # Bulk and deliberate: backfills, rule runs over existing mail,
             # admin sweeps. Slow is fine; blocking the other two is not.
             maintenance:
@@ -89,6 +130,25 @@ framework:
             # time and memory limits. The tradeoff of routing it here is that a
             # long sweep occupies the async consumer while it runs.
             Symfony\Component\Console\Messenger\RunCommandMessage: maintenance
+
+            # THE SAME TRAP AS THE NOTE ABOVE, AND IT HAD ALREADY SPRUNG.
+            # RunBulkStatusHandler exists to take a whole-view mark-read or
+            # archive OUT of the request — its docblock quotes the `Maximum
+            # execution time of 30 seconds exceeded` that produced it, and
+            # BulkStatusController::startJob answers with an indicator precisely
+            # so nobody waits for the work. Neither was true: the message was
+            # never listed here, an unrouted message is handled INLINE by
+            # dispatch(), and so every whole-view action ran inside the web
+            # request under the limit it was written to escape.
+            #
+            # What surfaced it was not the timeout but a deadlock in production
+            # — bulk runs colliding at request concurrency with a flag sync,
+            # each taking row locks on `message` in its own order, one of them
+            # shot by PostgreSQL. On a queue with one consumer those runs are
+            # serialised instead, which removes that pairing outright; the
+            # handler's in-process retry covers the collisions that are left.
+            App\Infrastructure\Messaging\Message\RunBulkStatusMessage: bulk
+
             Symfony\Component\Mailer\Messenger\SendEmailMessage: export
             Symfony\Component\Notifier\Message\ChatMessage: export
             Symfony\Component\Notifier\Message\SmsMessage: export

docs/de/install/docker.md

@@ -1,4 +1,4 @@
-<!-- translated-from: install/docker.md sha1:76b39de6d79bca245417628dafa8d329a76f4ba7 -->
+<!-- translated-from: install/docker.md sha1:89c280cbbe9bb45d384b184c1dd64fb5d1dd053c -->
 # Installation mit Docker Compose
 
 Der unterstützte Weg, von Anfang bis Ende: was du brauchst, was `docker compose up` tatsächlich tut,
@@ -139,10 +139,11 @@ rufe die Einrichtungsseite über die Adresse auf, die du tatsächlich verwenden
 | `worker-export` | `messenger:consume export` | Alles, was plMail verlässt, und die einzige Warteschlange, auf die jemand wartet. In einem eigenen Prozess, damit ein Versand nie hinter einem Abgleich steht |
 | `worker-ingest` | `messenger:consume ingest` | Eingehende Mail und die Arbeit, die unmittelbar darauf folgt |
 | `worker-maintenance` | `messenger:consume maintenance async` | Nachträgliche Verarbeitungen, Regelläufe über vorhandene Mail, administrative Durchläufe. Leert außerdem die stillgelegte Warteschlange `async` |
+| `worker-bulk` | `messenger:consume bulk` | Aktionen über eine ganze Ansicht: alles Ungelesene als gelesen markieren, alles archivieren. Eigener Prozess, weil jemand dabei auf eine Fortschrittsanzeige schaut und das nicht hinter einer nachträglichen Verarbeitung warten darf |
 | `scheduler` | `messenger:consume scheduler_default` | Löst alles Wiederkehrende aus. **Ohne diesen Container plant sich nichts von selbst** |
 | `ntfy` | ntfy, unter dem Profil `push` | Optional. Android-Push ohne Google — starte ihn mit `docker compose --profile push up -d` |
 
-Drei Prozesse statt drei Transports in einem Worker, weil ein Worker, der bereits in einem langen
+Vier Prozesse statt vier Transports in einem Worker, weil ein Worker, der bereits in einem langen
 Handler steckt, nichts anderes mehr annehmen kann, wie auch immer die Warteschlangen priorisiert
 sind. Genau das war das ursprüngliche Problem: Ein Klick auf Senden wartete hinter einem
 Gmail-Batch.

docs/de/install/troubleshooting.md

@@ -1,4 +1,4 @@
-<!-- translated-from: install/troubleshooting.md sha1:c2602b3577029a9a880454d2a1c78a8c4d349a7b -->
+<!-- translated-from: install/troubleshooting.md sha1:7d141d4e9169db18a2b045164621c0809700a68f -->
 # Fehlersuche
 
 Was `/healthz` bedeutet, wie du eine hängende von einer leeren Warteschlange unterscheidest, wo die
@@ -61,9 +61,10 @@ Vier Transports auf dem Doctrine-Transport, jeder mit einem eigenen Worker-Proze
 | `export` | `worker-export` | Alles, was plMail verlässt — Versand, Flag-Pushes, Gmail-Label-Änderungen, Mail aus dem Notifier. Die einzige Warteschlange, auf die jemand wartet |
 | `ingest` | `worker-ingest` | Eingehende Mail, Gmail- und Graph-Nachrichten-Batches, Kalenderabgleiche, Terminerkennung |
 | `maintenance` | `worker-maintenance` | Nachträgliche Verarbeitungen, Regelläufe über vorhandene Mail, "Jetzt ausführen"-Knöpfe im Administrationsbereich, Registrierung von Kalender-Push |
+| `bulk` | `worker-bulk` | Mark-Read- und Archiv-Läufe über eine ganze Ansicht. Jemand schaut dabei auf eine Fortschrittsanzeige, deshalb reihen sie sich nicht hinter einer nachträglichen Verarbeitung ein |
 | `async` | `worker-maintenance` | Stillgelegt. Wird geleert, damit vor der Aufteilung eingestellte Envelopes noch einen Konsumenten haben; hierher wird nichts mehr geroutet |
 
-Drei Prozesse statt drei Transports in einem Worker, weil ein Worker, der bereits in einem langen
+Vier Prozesse statt vier Transports in einem Worker, weil ein Worker, der bereits in einem langen
 Handler steckt, nichts anderes mehr annehmen kann, wie auch immer die Warteschlangen priorisiert
 sind — weshalb ein Klick auf Senden früher darauf wartete, dass ein Gmail-Batch fertig wurde.
 

docs/de/internals/architecture.md

@@ -1,4 +1,4 @@
-<!-- translated-from: internals/architecture.md sha1:1a4a42064fa940f8b0838ec7cdfa4e9acf0182bc -->
+<!-- translated-from: internals/architecture.md sha1:20b8fdd77966024edc2b0f9c02a0d15f9adf0633 -->
 # Architektur
 
 Die Schichten, was wo liegt, und die Regeln, die das so halten. Diese Seite beschreibt die
@@ -177,10 +177,19 @@ Workers, damit ein verzögerter Retry nie auf einen Neustart warten muss.
 
 Getrennte Transports allein genügen nicht. Ein Worker, der bereits in einem langen Handler
 steckt, kann eine Sendung nicht aufnehmen, wie hoch sie auch priorisiert ist — also hat jeder
-Transport seinen eigenen Prozess: `worker-export`, `worker-ingest` und `worker-maintenance` in
-`compose.yaml`. Ein vierter Transport, `async`, wird ohne Routing weitergeführt, damit
-Envelopes, die vor der Aufteilung eingereiht wurden, noch irgendwo landen können; der
-Maintenance-Worker leert ihn.
+Transport seinen eigenen Prozess: `worker-export`, `worker-ingest`, `worker-maintenance` und
+`worker-bulk` in `compose.yaml`. Ein fünfter Transport, `async`, wird ohne Routing
+weitergeführt, damit Envelopes, die vor der Aufteilung eingereiht wurden, noch irgendwo landen
+können; der Maintenance-Worker leert ihn.
+
+`bulk` ist der jüngste davon und derjenige, den du am ehesten mit `maintenance` verwechselst:
+Er trägt Mark-Read- und Archiv-Läufe über eine ganze Ansicht. Die dauern so lange wie eine
+nachträgliche Verarbeitung, sind aber im entscheidenden Punkt nicht dasselbe — jemand hat einen
+Knopf gedrückt und schaut auf eine Fortschrittsanzeige. Hinter einem Embedding-Backfill
+eingereiht wäre der Auftrag langsamer als die Inline-Variante, die er ersetzt hat. Er ist
+außerdem der einzige Transport mit einer eigenen DSN (`MESSENGER_BULK_DSN`); die gibt es, damit
+die Browser-Suite genau diese eine Warteschlange echt und konsumiert laufen lassen kann,
+während die anderen drei im Speicher und unbearbeitet bleiben.
 
 Zwei Routing-Entscheidungen sind tragend und nicht bloß ordentlich:
 

docs/install/docker.md

@@ -125,10 +125,11 @@ from the environment, or reach the setup page through the address you actually i
 | `worker-export` | `messenger:consume export` | Anything leaving plMail, and the only queue somebody is watching. On its own process so a send is never behind a sync |
 | `worker-ingest` | `messenger:consume ingest` | Mail arriving and the work that immediately follows it |
 | `worker-maintenance` | `messenger:consume maintenance async` | Backfills, rule runs over existing mail, admin sweeps. Also drains the retired `async` queue |
+| `worker-bulk` | `messenger:consume bulk` | Whole-view actions: mark every unread read, archive the lot. On its own process because somebody is watching an indicator for it and it must not wait behind a backfill |
 | `scheduler` | `messenger:consume scheduler_default` | Fires everything recurring. **Nothing schedules itself without this container** |
 | `ntfy` | ntfy, under the `push` profile | Optional. Android push without Google — start it with `docker compose --profile push up -d` |
 
-Three processes rather than three transports on one worker, because a worker already inside a long
+Four processes rather than four transports on one worker, because a worker already inside a long
 handler cannot pick up anything else however the queues are prioritised. That was the original
 problem: pressing Send behind a Gmail batch waited for the batch.
 

docs/install/troubleshooting.md

@@ -56,9 +56,10 @@ Four transports on the Doctrine transport, each with its own worker process:
 | `export` | `worker-export` | Anything leaving plMail — sends, flag pushes, Gmail label changes, mail from the notifier. The only queue somebody is waiting on |
 | `ingest` | `worker-ingest` | Mail arriving, Gmail and Graph message batches, calendar syncs, event extraction |
 | `maintenance` | `worker-maintenance` | Backfills, rule runs over existing mail, admin "run now" buttons, calendar push registration |
+| `bulk` | `worker-bulk` | Whole-view mark-read and archive runs. Somebody is watching an indicator for these, so they do not queue behind a backfill |
 | `async` | `worker-maintenance` | Retired. Drained so envelopes queued before the split still have a consumer; nothing routes here any more |
 
-Three processes rather than three transports on one worker, because a worker already inside a long
+Four processes rather than four transports on one worker, because a worker already inside a long
 handler cannot pick up anything else however the queues are prioritised — which is why pressing Send
 used to wait for a Gmail batch to finish.
 

docs/internals/architecture.md

@@ -163,9 +163,17 @@ delayed retry is never left waiting on a restart.
 
 Separate transports are not enough on their own. A worker already inside a long handler
 cannot pick up a send however it is prioritised, so each transport has its own process:
-`worker-export`, `worker-ingest` and `worker-maintenance` in `compose.yaml`. A fourth
-transport, `async`, is kept routing nothing, so envelopes queued before the split still have
-somewhere to land; the maintenance worker drains it.
+`worker-export`, `worker-ingest`, `worker-maintenance` and `worker-bulk` in `compose.yaml`. A
+fifth transport, `async`, is kept routing nothing, so envelopes queued before the split still
+have somewhere to land; the maintenance worker drains it.
+
+`bulk` is the newest of them and the one whose reason is easiest to mistake for `maintenance`:
+it carries whole-view mark-read and archive runs, which are long like a backfill but are not
+like one in the way that matters — somebody pressed a button and is watching an indicator, so
+queueing them behind an embedding backfill would make the job slower than the inline version
+it replaced. It is also the only transport with a DSN of its own
+(`MESSENGER_BULK_DSN`), which exists so the browser suite can make this one queue real and
+consumed while leaving the other three in memory and unhandled.
 
 Two routing decisions are load-bearing rather than tidy:
 

src/Infrastructure/Messaging/Handler/RunBulkStatusHandler.php

@@ -13,7 +13,9 @@ use App\Repository\Mail\MessageThreadRepository;
 use App\Service\Job\JobNotifier;
 use App\Service\Mail\ListViewResolver;
 use App\Service\Mail\ThreadStatusUpdater;
+use Doctrine\DBAL\Exception\RetryableException;
 use Doctrine\ORM\EntityManagerInterface;
+use Doctrine\Persistence\ManagerRegistry;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Messenger\Attribute\AsMessageHandler;
 use Throwable;
@@ -57,6 +59,29 @@ final readonly class RunBulkStatusHandler
      */
     private const int CHUNK = 100;
 
+    /**
+     * How many times a chunk is re-applied after a lock collision.
+     *
+     * Three, because a deadlock is not a queue: PostgreSQL shoots one of the
+     * two transactions the instant it sees the cycle, so a retry is competing
+     * with whatever else is writing right now rather than waiting out a
+     * backlog. Two attempts clear nearly all of them and the third is for the
+     * unlucky case where the same pair collides twice. Beyond that the
+     * collision is structural and should be read in the log, not hidden by a
+     * longer loop.
+     */
+    private const int LOCK_ATTEMPTS = 3;
+
+    /**
+     * Microseconds before the retry, multiplied by the attempt.
+     *
+     * Small and not zero. Both transactions rolled back at the same instant,
+     * so an immediate retry is the same race run again; a few tens of
+     * milliseconds is enough for the survivor to commit and get out of the way,
+     * and is invisible next to the flush itself.
+     */
+    private const int LOCK_BACKOFF_US = 40_000;
+
     public function __construct(
         private BackgroundJobRepository  $jobs,
         private MessageThreadRepository  $threads,
@@ -64,6 +89,7 @@ final readonly class RunBulkStatusHandler
         private ThreadStatusUpdater     $status,
         private JobNotifier             $notifier,
         private EntityManagerInterface  $em,
+        private ManagerRegistry         $registry,
         private LoggerInterface         $logger,
     ) {
     }
@@ -86,6 +112,32 @@ final readonly class RunBulkStatusHandler
 
         try {
             $this->run($job);
+        } catch (RetryableException $e) {
+            // THREE IN-PROCESS ATTEMPTS WERE NOT ENOUGH — see applyWithRetry()
+            // for what those are and why a deadlock deserves them. Reaching
+            // here means the collisions kept coming, which is bad luck rather
+            // than a broken job: nothing is corrupt, the chunk that lost rolled
+            // back whole, and the chunks before it are committed.
+            //
+            // So the job is left ACTIVE and the exception rethrown, which hands
+            // it back to the transport. That is not merely "give up more
+            // politely": redelivery is the only path that gets a fresh
+            // EntityManager — Doctrine closes this one on a failed flush and
+            // nothing after that point can write — and the `bulk` queue's retry
+            // ladder is tuned for exactly this arrival, 2s out to a minute over
+            // five attempts. Re-running re-resolves the view, so what it
+            // repeats is only the work still left to do.
+            //
+            // Marking it Failed here instead, which is what the generic branch
+            // below would do, ends the story: the redelivery finds a job that
+            // is no longer active and returns without doing anything.
+            $this->logger->warning('RunBulkStatusHandler: lock collisions outlasted the retries, handing the job back to the queue', [
+                'jobId' => $message->jobId,
+                'kind'  => $job->kind->value,
+                'error' => $e->getMessage(),
+            ]);
+
+            throw $e;
         } catch (Throwable $e) {
             $this->logger->error('RunBulkStatusHandler: bulk action failed', [
                 'jobId'     => $message->jobId,
@@ -103,6 +155,32 @@ final readonly class RunBulkStatusHandler
             // finish() on a detached entity flushes nothing at all. The failure
             // was logged and the indicator span forever, which is precisely the
             // outcome this block exists to prevent.
+            //
+            // NOT THROUGH A MANAGER DOCTRINE HAS SHUT, which is every failure
+            // that came out of a flush: UnitOfWork::commit() closes it in a
+            // `finally` before rethrowing. Writing through it then throws
+            // EntityManagerClosed from inside this catch block — which replaces
+            // the real error with one about Doctrine AND still leaves the
+            // indicator spinning, so the block fails at the only job it has,
+            // for the failures most likely to reach it.
+            //
+            // Reset rather than give up: the registry re-initialises the ghost
+            // in place, so the repositories holding this manager are working
+            // again on the next line. If even that will not write, say so and
+            // leave the row to app:jobs:reap, which is what the staleness
+            // window is for.
+            if (false === $this->em->isOpen()) {
+                $this->registry->resetManager();
+            }
+
+            if (false === $this->em->isOpen()) {
+                $this->logger->error('RunBulkStatusHandler: entity manager closed, the job could not be marked failed', [
+                    'jobId' => $message->jobId,
+                ]);
+
+                throw $e;
+            }
+
             $failed = $this->jobs->find($message->jobId);
 
             if (null !== $failed) {
@@ -158,7 +236,7 @@ final readonly class RunBulkStatusHandler
         $processed = 0;
 
         foreach (array_chunk($ids, self::CHUNK) as $chunk) {
-            $this->apply($chunk, $action, $read, $userId);
+            $this->applyWithRetry($chunk, $action, $read, $userId);
 
             $processed += count($chunk);
 
@@ -197,6 +275,68 @@ final readonly class RunBulkStatusHandler
         $this->notifier->changed($done);
     }
 
+    /**
+     * One chunk, re-applied if the database shot it for taking locks out of turn.
+     *
+     * WHAT THIS IS FOR
+     *
+     *     SQLSTATE[40P01]: Deadlock detected … while updating tuple (496,11)
+     *     in relation "message"
+     *
+     * Two transactions held a lock the other wanted. This run was updating the
+     * messages of one thread while something else — another bulk action, a sync
+     * writing remote flags onto mail this one is marking read — updated the
+     * same rows in the opposite order, and PostgreSQL broke the cycle by
+     * killing one of them. That is the database working, not failing: nothing
+     * is corrupt, the killed transaction rolled back whole, and the statement
+     * that lost will very likely win a moment later.
+     *
+     * Which is why it must not travel up as a failed job. It used to, and the
+     * person who pressed "mark all read" got an error and a dead indicator for
+     * a collision that would have cleared on its own.
+     *
+     * RESETTING IS NOT OPTIONAL. Doctrine closes the EntityManager on any
+     * failed flush, so without this the retry would throw EntityManagerClosed
+     * before it reached the database. The reset re-initialises the manager in
+     * place — the repositories injected above keep working — and it is safe
+     * here precisely because the transaction rolled back: there is no pending
+     * work to lose, and apply() clears the manager at the end of every chunk
+     * anyway.
+     *
+     * RetryableException rather than DeadlockException: DBAL's own marker for
+     * "the same statement may simply work next time", which also covers a lock
+     * wait that timed out. Anything else is a real failure and goes straight
+     * up, unretried.
+     *
+     * @param list<int> $chunk
+     */
+    private function applyWithRetry(array $chunk, string $action, bool $read, int $userId): void
+    {
+        for ($attempt = 1; ; ++$attempt) {
+            try {
+                $this->apply($chunk, $action, $read, $userId);
+
+                return;
+            } catch (RetryableException $e) {
+                if ($attempt >= self::LOCK_ATTEMPTS) {
+                    throw $e;
+                }
+
+                $this->logger->warning('RunBulkStatusHandler: lock collision, retrying the chunk', [
+                    'attempt' => $attempt,
+                    'threads' => count($chunk),
+                    'error'   => $e->getMessage(),
+                ]);
+
+                if (false === $this->em->isOpen()) {
+                    $this->registry->resetManager();
+                }
+
+                usleep(self::LOCK_BACKOFF_US * $attempt);
+            }
+        }
+    }
+
     /**
      * One chunk, grouped by account.
      *

templates/_layout/app.html.twig

@@ -220,6 +220,7 @@
            row, and a refresh landing mid-run undoes the rows it has already
            finished. See mail_pane_controller#hold. #}
         data-action="core--mercure:jobs-changed->mail--jobs#refresh
+                     core--mercure:jobs-changed->mail--mail-pane#onJobChanged
                      core--mercure:mailbox-synced->mail--mail-pane#onMailboxSynced
                      core--mercure:account-synced->mail--mail-pane#onAccountSynced
                      mail--list-toolbar:writing->mail--mail-pane#hold

templates/mail/_job_started.stream.html.twig

@@ -9,8 +9,17 @@ a toast says it started — because the list does NOT change yet, and an action
 that appears to do nothing is one people press again.
 
 What is not here: any change to the list. Five thousand rows will move as the
-job runs, and the list finds out the way it always does, from the writes the
-job makes.
+job runs, and the list finds out from the job itself — JobNotifier publishes on
+every chunk and at the end, and mail--mail-pane#onJobChanged re-reads the frame
+on the strength of it.
+
+That last sentence used to read "the list finds out the way it always does,
+from the writes the job makes", and there was no such mechanism. What actually
+re-read the list after a bulk action was the toolbar's own `written` event,
+which fires when the REQUEST lands — the same instant the work finished, for as
+long as the work happened inside the request. It was never true of a job
+handed to a worker, and the first thing that ran on one archived seventy-one
+conversations while the screen went on showing them.
 
 Variables: job
 #}

tests/Controller/Mail/BulkStatusOffloadTest.php

@@ -5,10 +5,12 @@ declare(strict_types=1);
 namespace App\Tests\Controller\Mail;
 
 use App\Domain\Enum\Job\JobKind;
+use App\Infrastructure\Messaging\Message\RunBulkStatusMessage;
 use App\Repository\Job\BackgroundJobRepository;
 use App\Repository\User\UserRepository;
 use Symfony\Bundle\FrameworkBundle\KernelBrowser;
 use Symfony\Bundle\FrameworkBundle\Test\WebTestCase;
+use Symfony\Component\Messenger\Envelope;
 
 /**
  * A whole-view bulk action is handed to a worker, not done in the request.
@@ -63,13 +65,16 @@ final class BulkStatusOffloadTest extends WebTestCase
         self::assertNotNull($job);
         self::assertSame(JobKind::MarkRead, $job->kind);
 
-        // Deliberately NOT asserting that it is still running. This
-        // environment's transport handles a dispatch inline, so by the time the
-        // response comes back the job has already finished — which is also what
-        // made the E2E version of this test corrupt other specs. What is being
-        // pinned is the contract the controller owns: a job exists, describing
-        // the right work. When it runs is the transport's business, and in
-        // production that is a worker.
+        // Deliberately NOT asserting that it is still running: when it runs
+        // is the transport's business, and asserting on timing here is what
+        // made the E2E version of this test corrupt other specs. What is
+        // pinned is the contract the controller owns — a job exists, describing
+        // the right work.
+        //
+        // That the transport is a WORKER and not this very request is the next
+        // test. It used to be a sentence of prose here saying "in production
+        // that is a worker", and that sentence was wrong for as long as it had
+        // been written.
 
         // The selection travels on the job, not in the envelope: a queue row
         // should not be the size of the work it describes.
@@ -94,6 +99,46 @@ final class BulkStatusOffloadTest extends WebTestCase
         self::assertSame(JobKind::MarkUnread, $this->jobs()->findOneBy([], ['id' => 'DESC'])?->kind);
     }
 
+    /**
+     * The envelope reaches a queue instead of being run on the spot.
+     *
+     * THIS IS THE TEST THAT WAS MISSING, and the prose above carried the gap as
+     * an assumption. Nothing routed RunBulkStatusMessage in
+     * config/packages/messenger.yaml, and Messenger handles an unrouted message
+     * INLINE in whichever process dispatched it — so every whole-view action
+     * ran inside the web request, under exactly the time limit
+     * RunBulkStatusHandler was written to escape, and at request concurrency,
+     * which is where the production deadlock came from. That file warns about
+     * this precise trap one routing entry above, for RunCommandMessage.
+     *
+     * The queue is asserted BY NAME, not "some transport got something". Which
+     * one it lands on is half the decision: `maintenance` would park a bulk
+     * action behind an embedding backfill, and `export` would park somebody's
+     * send behind the bulk action.
+     *
+     * ASKED OF THE ROUTING, not of a transport's contents, and that is not
+     * squeamishness about side effects. `bulk` is the one queue whose DSN can
+     * be overridden per environment — compose.test.yaml points it at Postgres
+     * and runs a worker, so the browser suite's bulk actions really happen —
+     * and a test that reached for InMemoryTransport::getSent() therefore
+     * passed on a developer's machine and died with a TypeError in the
+     * container the suite actually runs in. What is being pinned is which
+     * queue the envelope is addressed to, and SendersLocator answers that
+     * whatever is at the other end.
+     */
+    public function testTheJobIsHandedToTheBulkQueue(): void
+    {
+        $senders = static::getContainer()
+            ->get('messenger.senders_locator')
+            ->getSenders(new Envelope(new RunBulkStatusMessage(1)));
+
+        self::assertSame(
+            ['bulk'],
+            array_keys(iterator_to_array($senders)),
+            'a whole-view action is not addressed to the bulk queue',
+        );
+    }
+
     /**
      * An explicit list of ids stays inline, and that is deliberate.
      *

tests/Infrastructure/Messaging/Handler/RunBulkStatusHandlerTest.php

@@ -17,8 +17,12 @@ use App\Entity\Mail\MessageThread;
 use App\Entity\User\User;
 use App\Infrastructure\Messaging\Handler\RunBulkStatusHandler;
 use App\Infrastructure\Messaging\Message\RunBulkStatusMessage;
+use App\Repository\Mail\MessageThreadRepository;
 use App\Service\Label\LabelResolver;
 use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Driver\Exception as DriverException;
+use Doctrine\DBAL\Exception\DeadlockException;
+use Doctrine\Persistence\ManagerRegistry;
 use Doctrine\ORM\EntityManagerInterface;
 use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
 
@@ -122,19 +126,174 @@ final class RunBulkStatusHandlerTest extends KernelTestCase
         self::assertSame(0, $stillUnread, 'every message in the view should have been marked read');
     }
 
+    /**
+     * A deadlock on the first attempt is retried, not reported as a failure.
+     *
+     * WHAT CAME OUT OF PRODUCTION
+     *
+     *     SQLSTATE[40P01]: Deadlock detected … while updating tuple (496,11)
+     *     in relation "message"
+     *
+     * Two transactions took row locks on the same messages in opposite orders
+     * and PostgreSQL killed one of them, which is the database doing its job:
+     * the loser rolled back whole and would very likely win a moment later.
+     * What the handler did with it was mark the whole bulk action Failed, so a
+     * collision that would have cleared on its own became an error and a dead
+     * indicator in front of whoever pressed "mark all read".
+     *
+     * THE COLLISION IS INJECTED AT THE REPOSITORY, which is the one seam in
+     * apply() that a test can reach: ThreadStatusUpdater is final and the flush
+     * that really deadlocks is inside it. What matters is not WHERE the
+     * RetryableException comes from — the handler catches the marker, not a
+     * call site — but that one of them does not end the job, and that the
+     * chunk it interrupted is applied in full afterwards. Thrown once, on the
+     * first chunk, so the retry has to succeed for the assertion below to hold
+     * over all 250 threads.
+     */
+    public function testALockCollisionIsRetriedAndTheWorkStillLands(): void
+    {
+        $messageIds = $this->seedThreads();
+
+        $job = $this->job();
+
+        $threads = new class (self::getContainer()->get(ManagerRegistry::class)) extends MessageThreadRepository {
+            public int $collisions = 0;
+
+            /**
+             * @param array<string, mixed> $criteria
+             * @param array<string, string>|null $orderBy
+             *
+             * @return list<MessageThread>
+             */
+            public function findBy(array $criteria, ?array $orderBy = null, ?int $limit = null, ?int $offset = null): array
+            {
+                if (0 === $this->collisions++) {
+                    // Closed first, because that is the state a real one leaves
+                    // behind: UnitOfWork::commit() shuts the manager in a
+                    // `finally` before the exception escapes. Without it this
+                    // test would exercise the retry and skip the only line in
+                    // it that can go wrong — a retry through a manager Doctrine
+                    // has already closed throws EntityManagerClosed instead of
+                    // re-applying anything.
+                    $this->getEntityManager()->close();
+
+                    throw new DeadlockException(
+                        new class ('deadlock detected') extends \RuntimeException implements DriverException {
+                            public function getSQLState(): string
+                            {
+                                return '40P01';
+                            }
+                        },
+                        null,
+                    );
+                }
+
+                return parent::findBy($criteria, $orderBy, $limit, $offset);
+            }
+        };
+
+        $this->handler($threads)(new RunBulkStatusMessage((int) $job->id));
+
+        // The collision really happened: one throw plus one lookup per chunk.
+        self::assertGreaterThan(1, $threads->collisions, 'the repository was never reached twice');
+
+        $this->em->clear();
+
+        $reloaded = $this->em->find(BackgroundJob::class, $job->id);
+
+        self::assertNotNull($reloaded);
+        self::assertSame(JobState::Done, $reloaded->state, (string) $reloaded->failureReason);
+        self::assertSame(self::THREADS, $reloaded->processed);
+
+        $stillUnread = 0;
+
+        foreach ($messageIds as $id) {
+            if (null === $this->em->find(Message::class, $id)?->seenAt) {
+                ++$stillUnread;
+            }
+        }
+
+        self::assertSame(0, $stillUnread, 'the chunk that collided was never re-applied');
+    }
+
+    /**
+     * Collisions that outlast the retries hand the job BACK, they do not end it.
+     *
+     * The difference is whether the transport gets another go. A job marked
+     * Failed is one the redelivery skips — __invoke() returns early on anything
+     * that is no longer active — so recording a lock collision as a failure
+     * would quietly turn the `bulk` queue's five-attempt ladder into one
+     * attempt. Left active, the envelope comes back with a fresh
+     * EntityManager, re-resolves the view and finishes what is left.
+     *
+     * Four collisions against three attempts, so the retry is exhausted on the
+     * first chunk rather than by arithmetic that happens to line up.
+     */
+    public function testCollisionsThatOutlastTheRetriesLeaveTheJobToTheQueue(): void
+    {
+        $this->seedThreads();
+
+        $job = $this->job();
+
+        $threads = new class (self::getContainer()->get(ManagerRegistry::class)) extends MessageThreadRepository {
+            public int $collisions = 0;
+
+            /**
+             * @param array<string, mixed> $criteria
+             * @param array<string, string>|null $orderBy
+             *
+             * @return list<MessageThread>
+             */
+            public function findBy(array $criteria, ?array $orderBy = null, ?int $limit = null, ?int $offset = null): array
+            {
+                ++$this->collisions;
+
+                $this->getEntityManager()->close();
+
+                throw new DeadlockException(
+                    new class ('deadlock detected') extends \RuntimeException implements DriverException {
+                        public function getSQLState(): string
+                        {
+                            return '40P01';
+                        }
+                    },
+                    null,
+                );
+            }
+        };
+
+        try {
+            $this->handler($threads)(new RunBulkStatusMessage((int) $job->id));
+
+            self::fail('the collision was swallowed instead of being handed back');
+        } catch (DeadlockException) {
+            // What the transport sees, and what makes it redeliver.
+        }
+
+        self::assertSame(3, $threads->collisions, 'the chunk was not retried the documented number of times');
+
+        $this->em->clear();
+
+        $reloaded = $this->em->find(BackgroundJob::class, $job->id);
+
+        self::assertNotNull($reloaded);
+        self::assertTrue($reloaded->isActive(), 'a lock collision ended the job instead of leaving it to be redelivered');
+    }
+
     // ── fixtures ─────────────────────────────────────────────────────────────
 
-    private function handler(): RunBulkStatusHandler
+    private function handler(?MessageThreadRepository $threads = null): RunBulkStatusHandler
     {
         $container = self::getContainer();
 
         return new RunBulkStatusHandler(
             $container->get(\App\Repository\Job\BackgroundJobRepository::class),
-            $container->get(\App\Repository\Mail\MessageThreadRepository::class),
+            $threads ?? $container->get(\App\Repository\Mail\MessageThreadRepository::class),
             $container->get(\App\Service\Mail\ListViewResolver::class),
             $container->get(\App\Service\Mail\ThreadStatusUpdater::class),
             $container->get(\App\Service\Job\JobNotifier::class),
             $this->em,
+            $container->get(\Doctrine\Persistence\ManagerRegistry::class),
             $container->get(\Psr\Log\LoggerInterface::class),
         );
     }

truenas.compose.yaml

@@ -343,6 +343,34 @@ services:
             database:
                 condition: service_healthy
 
+    # A whole-view mark-read or archive. On its own process for the reason the
+    # send queue is: somebody pressed a button and is watching an indicator,
+    # and this must not wait behind a backfill on `maintenance`. Drop this
+    # service and the indicator fills to nothing — the job is queued and never
+    # handled.
+    worker-bulk:
+        image: *image
+        pull_policy: always
+        restart: unless-stopped
+        command: php bin/console messenger:consume bulk --time-limit=3600 --memory-limit=256M
+        stop_grace_period: 15s
+        environment:
+            <<: *app-env
+            APP_CONTAINER_NAME: worker-bulk
+        volumes:
+            - type: bind
+              source: *path_data
+              target: /app/var/data
+              bind:
+                  create_host_path: true
+        healthcheck:
+            disable: true
+        depends_on:
+            secrets-init:
+                condition: service_completed_successfully
+            database:
+                condition: service_healthy
+
     # Runs the scheduled jobs defined in
     # App\Infrastructure\Scheduler\MaintenanceSchedule: polling sync, push
     # renewal, monitoring pruning and the blob sweep. Drop this service and