Refactor separating queue backend from batch management (ESB v3) by mmenozzi · Pull Request #46 · webgriffe/esb

mmenozzi · 2026-01-07T17:17:13Z

This is a more clean way to fix the same problem of the PR #45 but it will require a major version bump.

…ondition-fix-v3

…sion

azambon · 2026-02-02T10:57:13Z

Tests are red because of Roave BC break checks, but the tests themselves are green.

Copilot

Pull request overview

This PR refactors ESB v3 queueing by splitting the queue backend (Beanstalk + Elasticsearch operations) from producer-side batch buffering to address the concurrency issue described in PR #45, while also completing the release_delay → error_retry_delay rename and updating the supported PHP/tooling matrix.

Changes:

Introduce QueueBackendInterface + BeanstalkElasticsearchQueueBackend and move producer batching into a new BatchManager created per produceAndQueueJobs() call.
Rename worker retry delay config/log context from release_delay to error_retry_delay and remove deprecated config support/docs.
Update supported/runtime matrix (PHP 8.5, Ubuntu 24.04 CI) and remove deprecated pager exception shims / deprecated pager argument handling.

Reviewed changes

Copilot reviewed 25 out of 25 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
tests/Integration/FailingJobHandlingTest.php	Updates config and log assertions to use `error_retry_delay`.
src/WorkerInstance.php	Switches from queue manager to queue backend; updates retry delay log context key.
src/Service/QueueBackendInterface.php	Renames/defines the unified backend interface and adds producer-oriented methods.
src/Service/BeanstalkElasticsearchQueueBackend.php	Replaces `QueueManager` with a backend implementation; removes internal batching and adds `enqueueJobs()`.
src/Service/BatchManagerInterface.php	Renames the producer batching interface (producer-side).
src/Service/BatchManagerFactory.php	Adds a factory to build per-call batch managers.
src/Service/BatchManager.php	New batching implementation that buffers jobs and flushes via `QueueBackendInterface`.
src/ProducerInstance.php	Uses per-call `BatchManager` instead of shared batching state to avoid concurrency issues.
src/Model/FlowConfig.php	Removes fallback to deprecated `release_delay`, returning `error_retry_delay` only.
src/FlowExtension.php	Updates Symfony DI wiring to use the new backend + batch manager factory per flow.
src/FlowConfiguration.php	Removes deprecated `release_delay` config node.
src/Console/Pager/AsyncPager.php	Removes deprecated boolean argument handling in `setCurrentPage()`.
src/Console/Pager/*.php (removed)	Drops deprecated pager exception shim classes for v3.
composer.json	Expands PHP constraint to include 8.5.
README.md	Updates requirements and removes deprecated `release_delay` docs.
Dockerfile	Defaults container build ARG to PHP 8.5.
.github/workflows/ci.yaml	Moves CI to Ubuntu 24.04 and adds PHP 8.5 to the matrix.

Comments suppressed due to low confidence (3)

src/Service/BeanstalkElasticsearchQueueBackend.php:22

DEFAULT_BATCH_ID is no longer used after moving batching out of the queue backend. Please remove this constant to avoid dead code.
src/Service/QueueBackendInterface.php:72
enqueueJobs() is documented as JobInterface[], but BeanstalkElasticsearchQueueBackend::enqueueJobs() relies on unsetting by UUID key. Either tighten this contract to array<string, JobInterface> (UUID as key) or change implementations to not depend on array keys (e.g., filter by UUID set).
src/Service/BeanstalkElasticsearchQueueBackend.php:247
This code unsets $jobs[$uuid] based on the bulk response _id, which only works if $jobs is keyed by UUID. If callers pass a numerically-indexed list (JobInterface[]), failed jobs won’t be removed and will still be put into Beanstalk. Consider filtering by UUID without relying on array keys, or enforce/validate array<string, JobInterface>.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Copilot

Pull request overview

Copilot reviewed 25 out of 25 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (4)

src/Service/QueueBackendInterface.php:72

enqueueJobs() is documented as taking JobInterface[], but BeanstalkElasticsearchQueueBackend::enqueueJobs() relies on UUID keys to drop failed items. Update this PHPDoc to the intended shape (e.g. array<string, JobInterface>) or change the backend implementation to not depend on array keys.
src/Service/BeanstalkElasticsearchQueueBackend.php:22
DEFAULT_BATCH_ID is no longer used after moving batching logic into BatchManager, but the constant remains declared. Removing it avoids dead code and confusion about where batching is handled.
src/Service/BeanstalkElasticsearchQueueBackend.php:247
unset($jobs[$uuid]) assumes the $jobs array is keyed by job UUID, but the interface/docs describe it as a plain JobInterface[] list. If $jobs is numerically indexed, failed ES items won't be removed and will still be enqueued to Beanstalk. Either change the contract to array<string, JobInterface> or filter the list without relying on array keys.
src/Service/BeanstalkElasticsearchQueueBackend.php:228
PHPDoc @param array<JobInterface> $jobs is not valid PHPDoc generic syntax for PHPStan/Psalm in this codebase. Use JobInterface[] or (preferably, given the UUID-keyed expectation) array<string, JobInterface> to match the actual shape.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot

Pull request overview

Copilot reviewed 26 out of 26 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (2)

src/Service/BeanstalkElasticsearchQueueBackend.php:247

In the bulk-index error handling, jobs are removed from the $jobs array via unset($jobs[$uuid]), which only works if $jobs is keyed by UUID. If a caller passes a numerically indexed array (as allowed by the JobInterface[] PHPDoc), failed-to-index jobs will still be enqueued to Beanstalk, leading to buried jobs and worker errors when the job is missing in Elasticsearch. Consider normalizing $jobs to a uuid=>Job map at the start of the method, or filter out failed UUIDs without relying on array keys.
src/Service/BeanstalkElasticsearchQueueBackend.php:21
DEFAULT_BATCH_ID is no longer used after moving batching out of this class. Keeping it is misleading and can be removed to avoid confusion.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot

Pull request overview

Copilot reviewed 26 out of 26 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (2)

src/Service/BeanstalkElasticsearchQueueBackend.php:246

enqueueJobs() assumes the $jobs array is keyed by job UUID (unset($jobs[$uuid])). However ElasticSearch::bulkIndexJobs() accepts a plain JobInterface[] (numeric keys), and QueueBackendInterface::enqueueJobs() is documented the same way; in that case the failed job won't be removed and it will still be put on Beanstalk even though indexing failed.
src/Service/BeanstalkElasticsearchQueueBackend.php:226
PHPDoc uses array<JobInterface> here, but the rest of the codebase uses JobInterface[] for lists (and this method works with a list). Using a consistent phpdoc format improves static analysis and IDE support.

Copilot

Pull request overview

Copilot reviewed 26 out of 26 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (2)

src/Service/BeanstalkElasticsearchQueueBackend.php:248

The per-item error handling in enqueueJobs() removes failed UUIDs by scanning the full $jobs array for each error item (nested foreach), which is O(n²) for large batches. Since jobs have stable UUIDs, re-key the batch by UUID once and then unset failed jobs in O(1).
src/Service/QueueBackendInterface.php:72
QueueBackendInterface::enqueueJobs() returns Promise, but BatchManagerInterface/ProducerInstanceInterface now expose integer counts of enqueued jobs. With the current signature, the backend cannot communicate how many jobs were actually enqueued (especially when some are skipped due to indexing errors), which makes the public counts potentially inaccurate.

+            $jobsCount = count($this->batch);
+            if ($jobsCount > 0) {
+                yield $this->queueBackend->enqueueJobs($this->batch);
+            }


Copilot

Pull request overview

Copilot reviewed 26 out of 26 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (4)

src/Service/QueueBackendInterface.php:72

enqueueJobs() is used as part of the batching path that ultimately feeds ProducerInstance::produceAndQueueJobs() (which now returns an int count). Returning Promise<null> here makes it impossible to propagate an accurate “jobs actually enqueued” count (e.g. when some jobs are dropped after bulk-index errors). Consider returning the number of jobs successfully enqueued.
src/Service/BeanstalkElasticsearchQueueBackend.php:240
array_combine() throws a ValueError when given empty arrays. Since enqueueJobs() is a public API, it should handle an empty $jobs input gracefully instead of crashing.
src/Service/BeanstalkElasticsearchQueueBackend.php:282
enqueueJobs() currently resolves to null, so callers can’t tell how many jobs were actually enqueued (especially after filtering out bulk-index failures). Returning the number of jobs that were successfully put into Beanstalk makes produceAndQueueJobs() counters and HTTP responses accurate.
src/Service/BeanstalkElasticsearchQueueBackend.php:21
DEFAULT_BATCH_ID is now unused after moving batching responsibilities into BatchManager, so it can be removed to avoid dead code.

+    public function flush(): Promise
+    {
+        return call(function () {
+            $jobsCount = count($this->batch);
+            if ($jobsCount > 0) {
+                yield $this->queueBackend->enqueueJobs($this->batch);
+            }
+            $this->batch = [];
+
+            return $jobsCount;
+        });
+    }


mmenozzi added the 3.x label Jan 7, 2026

WIP refactor separating queue backend from batch management

1d797a9

mmenozzi force-pushed the queue-manger-race-condition-fix-v3 branch from 1dfac07 to 1d797a9 Compare January 7, 2026 17:24

azambon added 9 commits January 8, 2026 17:44

Fixes the test errors

743d1d0

Removes the deprecated release_delay configuration parameter

1efdbdf

Merge remote-tracking branch 'origin/master' into queue-manger-race-c…

315fd01

…ondition-fix-v3

Removes a bunch of unused deprecated exception classes

6a918e9

Removes the deprecated pagination functions

cd6d99b

Fix CS

560dfca

Updates the build to include PHP 8.5 and run on the latest Ubuntu ver…

e3cf9f6

…sion

Adds PHP 8.5 to the allowed PHP versions in the composer.json file

b041d2f

Switches to PHP 8.5 in a few more places

0f6bb39

azambon changed the title ~~WIP refactor separating queue backend from batch management~~ Refactor separating queue backend from batch management (ESB v3) Feb 2, 2026

azambon requested a review from Copilot February 9, 2026 10:48

Copilot started reviewing on behalf of azambon February 9, 2026 10:48 View session

Copilot AI reviewed Feb 9, 2026

View reviewed changes

Comment thread src/FlowExtension.php Outdated

Comment thread src/ProducerInstance.php

Comment thread src/Service/BatchManager.php Outdated

azambon and others added 3 commits February 9, 2026 12:01

Apply suggestion from @Copilot

b0758a7

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Apply suggestion from @Copilot

9fe1797

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Updates the return type defined in a docblock

3482a70

azambon requested a review from Copilot February 9, 2026 11:11

Copilot started reviewing on behalf of azambon February 9, 2026 11:11 View session

Copilot AI reviewed Feb 9, 2026

View reviewed changes

Comment thread src/Model/FlowConfig.php Outdated

azambon added 2 commits February 9, 2026 12:20

Updates the interface with the new return type

315fa2d

Renames a method to reflect the new config field it reads

5c9fcee

azambon requested a review from Copilot February 9, 2026 11:30

Copilot started reviewing on behalf of azambon February 9, 2026 11:30 View session

Copilot AI reviewed Feb 9, 2026

View reviewed changes

Comment thread src/WorkerInstance.php Outdated

Comment thread src/ProducerInstanceInterface.php Outdated

azambon added 2 commits February 9, 2026 12:45

Fixes a docblock parameter type

0ed421c

Implements a suggestion by Copilot

1eefdcd

azambon requested a review from Copilot July 13, 2026 10:03

Copilot started reviewing on behalf of azambon July 13, 2026 10:03 View session

Copilot AI reviewed Jul 13, 2026

View reviewed changes

Fixes a possible issue when unsetting failed jobs

87486c0

azambon requested a review from Copilot July 13, 2026 10:20

Copilot started reviewing on behalf of azambon July 13, 2026 10:20 View session

Copilot AI reviewed Jul 13, 2026

View reviewed changes

Comment thread src/Service/BatchManager.php

Comment on lines +57 to +60

$jobsCount = count($this->batch);

if ($jobsCount > 0) {

yield $this->queueBackend->enqueueJobs($this->batch);

}

Switches back to using an array of jobs indexed by job uuid

c83bb3d

azambon requested a review from Copilot July 13, 2026 10:33

Copilot started reviewing on behalf of azambon July 13, 2026 10:33 View session

Copilot AI reviewed Jul 13, 2026

View reviewed changes

Comment thread src/Service/BatchManager.php

Comment on lines +54 to +65

public function flush(): Promise

{

return call(function () {

$jobsCount = count($this->batch);

if ($jobsCount > 0) {

yield $this->queueBackend->enqueueJobs($this->batch);

}

$this->batch = [];

return $jobsCount;

});

}

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Refactor separating queue backend from batch management (ESB v3)#46

Refactor separating queue backend from batch management (ESB v3)#46
mmenozzi wants to merge 19 commits into
masterfrom
queue-manger-race-condition-fix-v3

mmenozzi commented Jan 7, 2026

Uh oh!

azambon commented Feb 2, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

mmenozzi commented Jan 7, 2026

Uh oh!

azambon commented Feb 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

azambon commented Feb 2, 2026 •

edited

Loading