FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Refactor Swoole concurrency to adapter-owned fan-out; remove AMQP broker by loks0n · Pull Request #85 · utopia-php/queue · GitHub

/ queue Public

Refactor Swoole concurrency to adapter-owned fan-out; remove AMQP broker - #85

Merged
loks0n merged 9 commits into
mainfrom
swoole-concurrency-2
Jun 4, 2026
ConversationCommits (9)ChecksFiles changed
Merged

Refactor Swoole concurrency to adapter-owned fan-out; remove AMQP broker#85
loks0n merged 9 commits into
mainfrom
swoole-concurrency-2

Conversation

loks0n commented Jun 4, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Summary

Reworks Swoole concurrency, removes the AMQP broker, and fixes several pre-existing bugs.

The old design ran consumer-pool-size = max-coroutines: every coroutine ran the whole receive+process loop, pinning a connection (or the Locking mutex) on a blocking BRPOP and coupling connection count to coroutine count. Now concurrency is a property of the Swoole adapter: a single receive loop fans message processing out across a bounded set of coroutines, and the broker is just primitives.

Design

  • Consumer = receive() / commit() / reject() (was consume()). Brokers hold no loop, no concurrency knowledge.
  • The base Adapter processes sequentially (Workerman); the Swoole adapter fans processing out across coroutines (semaphore Channel + WaitGroup), parameterised by workerNum and maxCoroutines.
  • Adapter::process() never throws — a failing handler/commit/reject/callback is routed to $errorCallback instead of escaping the coroutine.
  • Broker\Redis takes two connections: receive (blocking pop + claim) and commands (acks + publishing; Locking-wrapped when shared by coroutines).
  • AMQP broker removed — php-amqplib channels aren't coroutine-safe and it wasn't on the active path.

Bugs fixed

  1. Workers failed every job (Argument #1 ($job) must be of type Message, string given). Server::getArguments built the argument array keyed by declared order but iterated in insertion order, and call_user_func_array passes integer keys positionally in iteration order — so a handler with inject() before param() got swapped arguments. Fixed with ksort. Reproduced on unmodified main.
  2. PHP version mismatch — utopia-php/pools 1.0.3 needs php ≥8.4 while the images ran 8.3. Bumped the floor to 8.4 and images to phpswoole/swoole:php8.4-alpine.
  3. Broker\Redis::getJob() crashed retry() on a missing/expired job (get() → null → new Message(null)).

⚠️ Migration guide (breaking changes)

1. PHP ≥ 8.4 required

Was >=8.3. Upgrade your runtime; the Docker images now use phpswoole/swoole:php8.4-alpine.

2. Broker\Redis now takes two connections

The broker uses one connection for the blocking receive loop and another for acks/publishing, so a blocking BRPOP can't stall everything else.

// Before
$broker = new Redis(new RedisConnection('redis'));

// After — sequential (maxCoroutines = 1): pass one connection for both roles
$broker = new Redis(
    receive:  new RedisConnection('redis'),
    commands: new RedisConnection('redis'),
);

// After — concurrent (maxCoroutines > 1): dedicated receive + a locked, shared commands connection
$broker = new Redis(
    receive:  new RedisConnection('redis'),
    commands: new Locking(new RedisConnection('redis')),
);

This replaces the old "size a connection Pool to maxCoroutines" approach — you now need exactly two connections regardless of coroutine count.

3. Consumer interface: consume() → receive() / commit() / reject()

The receive loop and ack orchestration moved into the Adapter. Custom Consumer implementations must change:

// Before
public function consume(Queue $queue, callable $messageCallback, callable $successCallback, callable $errorCallback): void;

// After
public function receive(Queue $queue, int $timeout): ?Message; // block up to $timeout, return one message or null
public function commit(Queue $queue, Message $message): void;   // acknowledge
public function reject(Queue $queue, Message $message): void;   // fail
public function close(): void;

4. AMQP broker removed

Utopia\Queue\Broker\AMQP and the php-amqplib dependency are gone. There is no drop-in replacement; use Broker\Redis. (php-amqplib channels are not coroutine-safe under Swoole.)

5. Removed Result\Commit, Result\NoCommit, Error\Retryable

These were the AMQP acknowledgement protocol.

  • Remove return new Commit() / return new NoCommit() from handlers — return values were only honoured by AMQP and are ignored now.
  • Replace throw new Retryable() (AMQP requeue) with a normal exception: on Redis a thrown handler exception routes the job to the failed queue, which you re-drive with Publisher::retry().

Tests / CI

New SwooleConcurrencyTest (adapter fan-out via an in-memory Connection). Priority-ordering tests use a queue no worker consumes so they don't race the consumer. CI matrix gains Locking, SwooleConcurrency.

Ran the full matrix locally in Docker on PHP 8.4 — all green: Locking, Pool, Swoole, SwooleConcurrency, SwooleRedisCluster, Workerman. pint + phpstan clean. Verified the concurrent two-connection worker boots and processes against real Redis (success + reject paths) with no errors.

🤖 Generated with Claude Code

Refactor Swoole concurrency to receiver fan-out; remove AMQP broker
77169d2 
Move the unit of concurrency from "N consumers each running the whole
loop" to "one receiver fanning message processing out to a bounded pool
of coroutines". This decouples the concurrency level from the connection
count and makes the Locking decorator useful instead of a footgun.

- Add Concurrency\Executor with Inline (default, sequential) and
  Coroutine (bounded fan-out via a semaphore Channel + WaitGroup)
  strategies. maxCoroutines now lives on the executor.
- Broker\Redis takes two connections: a dedicated `receive` connection
  for the blocking pop loop and a `work` connection (Locking-wrapped
  when concurrent) for bookkeeping/acks. Processing is handed to the
  executor; `work` defaults to `receive` so existing single-connection
  callers are unchanged.
- Adapter\Swoole drops the N-coroutine wrapper and maxCoroutines; it is
  back to runtime setup plus coroutine-local DI context.
- Remove the AMQP broker entirely (broker, test, server, compose
  services, CI matrix entry, php-amqplib dependency). php-amqplib
  channels are not coroutine-safe and AMQP was not on the active path.
- Wire the Swoole E2E worker to the concurrent two-connection broker.
- Make the priority-ordering tests use a queue no worker consumes, so
  they no longer race the live consumer.
- Add CI matrix entries for the Concurrency, Locking and
  SwooleConcurrency suites.
- Disable composer platform-check: pools 1.0.3 declares php >=8.4 while
  the project and its Docker images target 8.3, which otherwise aborts
  the runtime guard.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

greptile-apps Bot commented Jun 4, 2026 •
edited
Loading

Copy link
Copy Markdown

Greptile Summary

This PR refactors Swoole concurrency from N parallel consumer loops to a single receive loop with bounded coroutine fan-out (Channel semaphore + WaitGroup), removes the AMQP broker, splits the Consumer interface into receive()/commit()/reject(), and fixes two pre-existing bugs: argument-ordering in Server::getArguments and a null-return crash in Broker\Redis::getJob().

  • Adapter refactoring: The base Adapter gains a process() template method (never-throws contract), and Swoole overrides consume() to fan out processing across coroutines. Coroutine-local context isolation is preserved via Coroutine::getContext().
  • Broker\Redis split: The broker now takes separate $receive and $commands connections; claim writes stay on the blocking-pop connection, while acks/publishes go through the lockable commands connection.
  • Broker\Redis constructor is now a breaking API change: callers must pass two Connection arguments instead of one, and all internal usages have been updated accordingly.

Confidence Score: 4/5

Generally safe to merge; the core concurrency refactor is architecturally sound, but Adapter::process() has a control-flow issue where a throwing success callback incorrectly invokes reject() on an already-committed message, corrupting processing counters.

The new Adapter::process() template method uses a single catch block spanning both the handler call and the post-commit successCallback. If the success callback throws, commit() has already removed the job and decremented the processing counter, yet reject() fires anyway — pushing the pid to the failed list, incrementing the failed counter, attempting a second listRemove, and decrementing processing a second time (driving it negative). This is a current defect on the changed path that affects every adapter using process().

src/Queue/Adapter.php — the process() method's single try/catch covers both handler execution and post-commit callbacks, which leads to incorrect state when successCallback throws.

Important Files Changed

Filename Overview
src/Queue/Adapter.php New process() template method introduced with a "never-throws" contract; the single catch block covers both handler failures and post-commit failures, so a throwing successCallback triggers reject() on an already-committed message.
src/Queue/Adapter/Swoole.php Consume loop refactored to single-receiver + coroutine fan-out using Channel semaphore and WaitGroup; coroutine-local context isolation and SIGTERM graceful shutdown look correct.
src/Queue/Broker/Redis.php Constructor split into receive/commands connections; receive loop claim writes remain on the blocking connection; getJob() null-crash fix is correct. Work connection is not reset when receive reconnects (pre-existing gap, already flagged).
src/Queue/Broker/Pool.php Delegating helpers consolidated into one delegate() method; new receive/commit/reject delegations each borrow an independent connection from the pool, which is correct for stateless operations.
src/Queue/Consumer.php Interface cleanly split from a monolithic consume() loop into receive()/commit()/reject() primitives.
src/Queue/Server.php Single-line ksort($arguments) fix for the two-pass parameter-injection ordering bug; no other behavioral changes.
tests/Queue/E2E/Adapter/SwooleConcurrencyTest.php New test replaces mock consumer with an in-memory Redis broker; correctly validates bounded concurrency and sequential single-coroutine operation.
tests/Queue/E2E/Adapter/InMemoryConnection.php Minimal in-memory Connection for unit tests; setArray() stores PHP arrays while getJob() expects strings, so retry tests using this connection would always treat jobs as missing — but no retry tests against InMemoryConnection currently exist.
composer.json php-amqplib removed; PHP floor bumped to >=8.4; platform-check disabled (already flagged in previous review).

Reviews (6): Last reviewed commit: "Remove dead Commit/NoCommit/Retryable; c..." | Re-trigger Greptile

greptile-apps Bot reviewed Jun 4, 2026
View reviewed changes
Comment thread src/Queue/Concurrency/Coroutine.php Outdated
Comment thread composer.json Outdated
Move concurrency to the adapter; fix arg ordering; require PHP 8.4
7d124ca 
Address review: the broker should not own the executor. With AMQP gone the
push-based constraint that forced it there is gone, so the loop moves to the
adapter and the broker becomes plain primitives.

- Consumer is now receive()/commit()/reject() instead of consume(). Brokers
  expose primitives and hold no loop, executor, or coroutine knowledge.
- The base Adapter owns the receive loop and an Executor (default Inline).
  The Swoole adapter selects the Coroutine executor and takes maxCoroutines —
  concurrency is now purely a property of the adapter. Per-message context is
  set via setContext(), overridden by Swoole to stay coroutine-local.
- Broker\Redis: receive() does the blocking pop + claim on the receive
  connection; commit()/reject() ack on the work connection. Reconnect state
  moved to instance fields so it persists across receive() calls.
- Broker\Pool delegates the new primitives.

Fix a pre-existing argument-ordering bug: Server::getArguments built the
arguments array in two passes (params then injections), so it was keyed by
declared order but not iterated in it. call_user_func_array passes
integer-keyed values positionally in iteration order, so handlers with an
inject() before a param() received arguments swapped (e.g. a string where a
Message was expected). ksort the arguments before invoking.

Require PHP 8.4 (utopia-php/pools 1.0.3 already requires it) and move the
Docker images to phpswoole/swoole:php8.4-alpine, replacing the earlier
platform-check workaround.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
loks0n changed the title Refactor Swoole concurrency to receiver fan-out; remove AMQP broker Refactor Swoole concurrency to adapter-owned fan-out; remove AMQP broker Jun 4, 2026
Inline coroutine fan-out in the Swoole adapter; drop Executor abstrac…
487b64e 
…tion

- Remove the Concurrency\Executor/Inline/Coroutine classes. The base adapter
  processes sequentially; the Swoole adapter inlines the coroutine fan-out
  (semaphore Channel + WaitGroup) directly. The Swoole adapter has a single
  model parameterised by workerNum and maxCoroutines — no inline-vs-concurrent
  split.
- Harden Adapter::process(): it now never throws. A failing commit/reject or
  error callback is routed to $errorCallback instead of escaping the coroutine
  and being swallowed by Swoole's default handler (error-visibility regression
  raised in review). The Swoole coroutine body also logs as a last resort.
- Broker\Redis::getJob() tolerates a missing job (get() returning null), which
  could otherwise crash retry() via new Message(null).
- Trim the over-explanatory doc comments added in the refactor.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
greptile-apps Bot reviewed Jun 4, 2026
View reviewed changes
Comment thread src/Queue/Adapter.php
loks0n and others added 3 commits June 4, 2026 12:18
Trim comments to non-obvious WHYs; rename Redis $work to $commands
2be4c1b 
- Cut the explanatory doc comments down to the few that carry non-obvious
  rationale (process() never throws, the two-connection split, the
  call_user_func_array argument ordering); drop the rest that just restated
  the code.
- Rename the Redis broker's second connection from $work to $commands, which
  says what it carries (acks + publishing) rather than a vague "work".

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Use constructor property promotion for Redis $receive
95064b1 
$commands keeps its body assignment because of the `?? $receive` fallback,
but $receive is a plain assignment and can be promoted (as the broker's
original single connection was).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Require both Redis connections explicitly; drop the $work fallback
3d0131c 
No implicit "$commands defaults to $receive" — callers pass both connections,
which makes the two-connection model explicit and lets both be promoted. Pass
the same connection twice when one suffices (sequential/inline). All call sites
(workers, tests) updated accordingly.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
greptile-apps Bot reviewed Jun 4, 2026
View reviewed changes
Comment thread src/Queue/Broker/Redis.php
Comment on lines +29 to 35
public function __construct(
// Blocking receive loop + claim writes (single caller).
private readonly Connection $receive,
// Acks and publishing; wrap in Locking when shared by coroutines.
private readonly Connection $commands,
) {
}

greptile-apps Bot Jun 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality

The constructor now requires two arguments, silently breaking any caller that was using new Redis($connection). The PR description states that "Existing new Redis($conn) is unchanged," but there is no default for $commands — PHP will throw ArgumentCountError: Too few arguments at runtime. Making $commands optional and defaulting to the receive connection restores backward compatibility while still allowing the recommended two-connection setup.

Suggested change
public function __construct(
// Blocking receive loop + claim writes (single caller).
private readonly Connection $receive,
// Acks and publishing; wrap in Locking when shared by coroutines.
private readonly Connection $commands,
) {
}
private readonly Connection $commands;
public function __construct(
// Blocking receive loop + claim writes (single caller).
private readonly Connection $receive,
// Acks and publishing; wrap in Locking when shared by coroutines.
// Defaults to $receive for single-connection (non-concurrent) usage.
?Connection $commands = null,
) {
$this->commands = $commands ?? $receive;
}

loks0n and others added 3 commits June 4, 2026 13:23
Drop setContext; reset context in the loop, lazy per-coroutine on Swoole
d0faa5e 
Remove the setContext seam. process() no longer manages context — the base
loop assigns a fresh container per message, and the Swoole adapter creates one
lazily per coroutine in context() (each message already runs in its own
coroutine, so it's naturally isolated). Fewer moving parts; process() is now
purely run-handler-then-commit/reject.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Flatten Adapter::process() error handling
42560b0 
Collapse the nested try/catch into one try with guarded reject/error calls.
Same guarantee (process never throws) with less nesting, and $errorCallback
now always receives the original handler error rather than a reject error
when reject() also fails.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Remove dead Commit/NoCommit/Retryable; collapse Pool delegate helpers
6db5c8f 
- Delete Result\Commit, Result\NoCommit and Error\Retryable. They were the
  AMQP acknowledgement protocol and are unreferenced now that AMQP is gone.
- Collapse Pool's delegatePublish/delegateConsumer into a single delegate()
  that takes the target pool, removing the duplicated wrapper.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
loks0n merged commit d4da9e8 into main Jun 4, 2026
9 checks passed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

greptile-apps[bot] greptile-apps[bot] left review comments

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

1 participant

Back | FazBrowse Home | New Git URL