polysource/bulk-async

Polysource — async bulk actions via Symfony Messenger + live progress via Mercure SSE.

v0.1.1 2026-05-10 21:00 UTC

Requires

• php: >=8.1
• polysource/core: ^0.1
• symfony/config: ^5.4 || ^6.0 || ^7.0 || ^8.0
• symfony/dependency-injection: ^5.4 || ^6.0 || ^7.0 || ^8.0
• symfony/http-kernel: ^5.4 || ^6.0 || ^7.0 || ^8.0
• symfony/messenger: ^5.4 || ^6.0 || ^7.0 || ^8.0
• symfony/translation: ^5.4 || ^6.0 || ^7.0 || ^8.0
• symfony/uid: ^5.4 || ^6.0 || ^7.0 || ^8.0
• symfony/yaml: ^5.4 || ^6.0 || ^7.0 || ^8.0
• twig/twig: ^3.0

Requires (Dev)

• doctrine/dbal: ^3.6 || ^4.0
• doctrine/orm: ^2.20 || ^3.0
• phpunit/phpunit: ^10.5 || ^11.5
• polysource/symfony-bundle: 0.1.x-dev
• symfony/mercure: ^0.6 || ^1.0

MIT

• symfony
• async
• bulk
• Messenger
• mercure
• polysource

README

Asynchronous bulk actions for Polysource — execute over Symfony Messenger with live progress (Mercure) and cancel mid-flight.

Part of the Polysource monorepo. MIT-licensed.

When to use

The synchronous bulk action ("Retry 5 000 failed messages") times out after 30 s in production. This package fans the work out to Messenger workers, persists per-job progress every 5 records or 500 ms, and exposes a JSON endpoint + Mercure topic so the UI can show a live progress bar and a "Cancel" button.

See ADR-024.

What it ships

• BulkJob immutable VO (12 fields, 8 KiB error cap) + BulkJobStatus enum (5 states, isTerminal()).
• BulkJobStorageInterface + DoctrineBulkJobStorage + Doctrine entity.
• BulkJobMessage + BulkJobHandler — re-fetches each iteration to honour Cancelled, throttled persist (5 records OR 500 ms), per-record exception isolation.
• AsyncBulkActionDispatcher — UUID v7 + Pending persist + Messenger dispatch.
• AsyncAwareBulkActionInterface — opt-in marker (parallel interface, no BC break to BulkActionInterface).
• BulkJobResource — browsable admin resource (#[AsResource], slug bulk-jobs).
• CancelBulkJobAction — idempotent on terminal, gated POLYSOURCE_BULK_JOB_CANCEL.
• ProgressController — JSON GET /admin/bulk-jobs/{id}/progress. Two-stage gate: coarse POLYSOURCE_BULK_JOB_VIEW permission + ownership check (requester must own the job, or hold POLYSOURCE_BULK_JOB_VIEW_ANY).
• MercureBulkJobBroadcaster — gated on class_exists(HubInterface), hub failures swallowed, topic polysource/bulk-jobs/{actorId}/{id} (actor segment URL-encoded). Pair with the polysource_bulk_progress_topic(job) Twig helper so client and broadcaster always agree on the topic shape; configure your Mercure JWT subscriber claims to restrict per-actor for defence-in-depth.
• Stimulus progress_controller.js — EventSource Mercure → polling fallback auto on error.

Install

composer require polysource/bulk-async symfony/messenger
# Optional but recommended for live progress:
composer require symfony/mercure-bundle

Register the bundle:

return [
    Polysource\BulkAsync\PolysourceBulkAsyncBundle::class => ['all' => true],
];

Run the migration to create polysource_bulk_jobs.

Extend it

BulkJobStorageInterface is 3 methods. To persist jobs in Redis / Mongo / your service instead of Doctrine, implement it and alias the interface to your service in DI. The handler, the ProgressController, the Mercure broadcaster all keep working.

See extensibility map.

Documentation

• Bulk-async walkthrough