feat(queue): Synchronous/Asynchronous publishers + Swoole Background decorator

[loks0n]

What

Splits publishing into two explicit contracts and adds a Swoole-backed decorator that provides both, with back pressure and telemetry.

Interfaces (Utopia\Queue\Publisher\*)

interface Synchronous {                  // publish now, block, return the result
    public function publish(Queue $queue, array $payload, bool $priority = false): bool;
    public function retry(Queue $queue, ?int $limit = null): void;
    public function getQueueSize(Queue $queue, bool $failedJobs = false): int;
}

interface Asynchronous {                 // hand off; void, throws on back pressure
    public function enqueue(Queue $queue, array $payload, bool $priority = false): void;
}

enqueue() returns void — it reports acceptance, not delivery. When the buffer is full and can't accept the message, it throws BufferFullException so the caller can shed load or slow down (a bool was too easy to ignore).

Redis and Pool implement Synchronous (their former enqueue() is now publish()).

Broker\Background — wraps a Synchronous, implements both

$publisher = new Background($redisBroker, capacity: 512, coroutines: 1, timeout: -1, telemetry: $telemetry);

Coroutine\run(function () use ($publisher) {
    $publisher->start();              // spawn the reader coroutine(s)

    $publisher->enqueue($queue, $p);  // async — push onto the channel; throws BufferFullException if full past timeout
    $publisher->publish($queue, $p);  // sync — delegate straight through

    $publisher->shutdown();           // drain queued publishes, then stop the readers
});

• Bounded channel = back pressure. enqueue() pushes onto a Swoole\Coroutine\Channel sized by capacity; once full it blocks the producing coroutine until a reader drains a slot — a slow broker throttles producers instead of piling up unbounded work.
• timeout (seconds, default -1) caps that wait: when the buffer is full, enqueue() waits up to timeout then throws BufferFullException; -1 blocks indefinitely.
• Reader coroutine(s) loop pop() and delegate each dispatch to the wrapped publisher's publish().
• coroutines (default 1) sets how many readers dispatch concurrently. ⚠️ coroutines > 1 is only safe when the wrapped publisher tolerates concurrent use across coroutines — a single-connection broker (bare Redis) must not be shared, so pair it with a connection Pool. More than one reader also gives up FIFO dispatch order.
• shutdown() pushes one sentinel per reader and WaitGroup::wait()s, so everything queued is published before the readers exit.
• Safe fallback: with no reader running or outside a coroutine runtime, enqueue() publishes synchronously.
• Failed background dispatches are error_log'd (no producer to surface to). Note: a down broker makes publish() throw, so background dispatch drops+logs — the channel is a transient shock absorber, not a durable outage buffer.

Telemetry (no-op by default)

Background takes an optional Telemetry adapter and reports the one thing unique to the background buffer:

instrument	type	meaning
messaging.publisher.buffer.depth	observable gauge	channel length — publishes buffered awaiting dispatch (the back-pressure signal)

Dispatch/failure counts are intentionally not metered here — the wrapped synchronous publisher sees every publish and reports those itself, so counting them again would double-count.

⚠️ Breaking changes

• Redis::enqueue() / Pool::enqueue() → publish().
• Utopia\Queue\Publisher interface → Utopia\Queue\Publisher\Synchronous.
• The async publish verb (enqueue()) now returns void and throws BufferFullException instead of returning bool.

Appwrite's $publisher->enqueue(...) call sites and any Publisher type hints must migrate.

Tests

BackgroundTest (unit suite, in-memory fakes — no Redis/Docker): sync publish() delegation, the reader draining the channel in order, a back-pressure case (capacity 1, 20 messages, in order), enqueue throws BufferFullException when the buffer stays full past the timeout, concurrent coroutines (4 readers, 20 messages, each delivered once), the sync fallback, management-call delegation, and the buffer-depth gauge (via the Test adapter). The rest of the suite was migrated to publish().

Verified locally: BackgroundTest 8/8 ✅ · pint ✅ · phpstan No errors ✅ · rector ✅

Note: constructing Background needs ext-swoole. bin/monorepo test queue also spins up the package's compose services; the new test itself needs only ext-swoole.

🤖 Generated with Claude Code

https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ

[greptile-apps]

Greptile Summary

This PR splits the old Publisher interface into explicit Synchronous and Asynchronous contracts and ships a new Broker\Background Swoole decorator that wraps a synchronous publisher behind a bounded channel, giving callers fire-and-forget enqueue() with cooperative back-pressure via BufferFullException.

• Redis and Pool implement Synchronous (rename-only: enqueue → publish); Server.php updates its instanceof check; all test call sites are migrated.
• Broker\Background is the main new surface: a bounded Swoole\Coroutine\Channel acts as the queue between producers and one-or-more reader coroutines, with opt-in telemetry for buffer depth.
• The BackgroundTest suite (8 cases, in-memory fakes) covers sync fallback, drain order, back-pressure, concurrent coroutines, and the gauge — though retry() delegation is not exercised.

Confidence Score: 3/5

Background.php has three open defects from previous review rounds still unaddressed: a shutdown race that silently drops messages, a WaitGroup deadlock when coroutine creation fails, and fire-and-forget dispatch discarding false returns from publish.

All three previously flagged defects remain in the current code: $this->started is set to false only after waitGroup->wait() (race window), waitGroup->add() is called before Coroutine::create() whose failure is unchecked (potential indefinite hang in shutdown()), and publish() returning false inside the reader closure is silently ignored. Multiple co-located defects in the core new class drive the score down.

packages/queue/src/Queue/Broker/Background.php — shutdown(), start(), and the reader closure all need attention for the three previously flagged issues.

Important Files Changed

Filename	Overview
packages/queue/src/Queue/Broker/Background.php	New Swoole-backed decorator implementing both Synchronous/Asynchronous; has three unresolved defects flagged in previous review rounds (shutdown race, WaitGroup deadlock on coroutine creation failure, silent false-return from publish).
packages/queue/src/Queue/Broker/Redis.php	Method renamed from enqueue() to publish() to match new Synchronous interface; internal self-call in retry() also updated. Mechanical rename, no logic change.
packages/queue/src/Queue/Broker/Pool.php	Implements Synchronous instead of old Publisher; publish() delegates via FUNCTION correctly matching the renamed Redis method.
packages/queue/src/Queue/Publisher/Synchronous.php	New interface splitting the old Publisher into a synchronous contract with publish(), retry(), getQueueSize(); clean and well-documented.
packages/queue/src/Queue/Publisher/Asynchronous.php	New interface for fire-and-forget delivery with BufferFullException on back-pressure; void return is correct and well-documented.
packages/queue/src/Queue/Publisher/BufferFullException.php	Thin RuntimeException subclass for buffer-full signalling; correct and minimal.
packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php	8-case unit suite with in-memory fakes; covers sync fallback, drain order, back-pressure, concurrent coroutines, and gauge reporting. retry() delegation not tested.
packages/queue/src/Queue/Server.php	One-line change updating instanceof check from old Publisher to new Synchronous; semantically equivalent.
packages/queue/tests/Queue/E2E/Adapter/Base.php	All enqueue() call sites migrated to publish() and return type updated to Synchronous; straightforward rename.
packages/queue/phpunit.xml	BackgroundTest.php added to the unit testsuite (explicitly listed, not directory-scanned), so it won't be picked up by the e2e suite. No overlap.

_{Reviews (7): Last reviewed commit: "refactor(queue): rename BackpressureExce..." | Re-trigger Greptile}