CI confirms the expected release gate: queue dependency installation fails because Packagist currently ends at NATS 1.3.1 and this PR requires ^1.4. Local checks used the linked NATS source from #315. Keep this draft until #315 is released, then retarget to main and rerun the registry-backed queue suite.

Benchmark results

queue — workload shapes across both concurrency axes (4 cores, 600 messages, median of 3)

Np x Mc x Ff = N consumer processes x M handler coroutines x a prefetch limit of F
unacknowledged messages. io yields and should follow coroutines; cpu does not and should follow
processes; mixed is in between. null has no handler at all, so it measures the broker
rather than the job -- the only row where prefetch can matter.

Shared CI runners — treat absolute numbers as rough, compare modes within a run. Commit de792b9.

Updated and rebased for the callback-only NATS API in #315; 185 queue unit tests and 53 NATS broker/prefetch integration tests pass locally. For the queue-specific review before this draft becomes ready: verify command-connection replacement after an ambiguous transport failure. NATS requests() intentionally disconnects without replay, while commandsConnection() currently retains the cached instance. Recovery of future operations needs explicit coverage in the broker; callback-abort and response-timeout connection usability are already covered in the NATS package.

Rebased onto the unified NATS request lifecycle in #315. The queue now passes JetStreamMessage objects to JetStream::ackBatch(); it no longer constructs acknowledgement subjects or payloads. Updated the design document. Validation passes locally: 185 queue unit tests / 442 assertions, 53 NATS broker and prefetch integration tests / 261 assertions (including batch 100 with one coroutine), Pint, PHPStan, Rector, and documentation lint. Existing package-release and command-connection recovery review gates remain open.

Rebased onto the NATS review fix 24cbdee: server-closing errors recycle the connection before invoking onError, so a throwing callback cannot leave a dead connection marked connected. Queue validation after rebase: 185 unit tests / 442 assertions and 53 NATS broker/prefetch integration tests / 261 assertions pass. Existing release and command-connection recovery gates remain unchanged.

Rebased onto a746aaf from #315. Request collection now performs keepalive checks without reconnecting/replaying an in-flight batch; ackBatch documentation explicitly requires AckPolicy::Explicit for isolation of selected messages. The queue already uses Explicit. Dependent NATS broker/prefetch integration suite passes: 53 tests / 261 assertions. Existing release and command-connection recovery gates remain unchanged.

Rebased onto NATS fix 1077972, which consumes buffered protocol data before keepalive emission or stale checks. This preserves buffered acknowledgement replies and PONGs. Dependent NATS broker/prefetch integration validation passes: 53 tests / 261 assertions. Existing release and command-connection recovery gates remain unchanged.

NATS 1.4.0 is released and available on Packagist: https://github.com/utopia-php/nats/releases/tag/1.4.0 . Rebased onto main, retargeted the PR from codex/nats-requests to main, and updated the rollout document. Existing ^1.4 constraints now resolve normally; Composer installed the published archive (mirror commit 016f48bd25f7bf7a06e5e12a9c768036945d5459), not a local path package. Validation against that archive: 185 unit tests / 442 assertions and 53 NATS broker/prefetch integration tests / 261 assertions pass. The full runner hit a port conflict when starting duplicate services, so targeted integration ran against the existing services. Board moved from Blocked to In progress. Remains draft for the separate command-connection recovery review; downstream queue/Platform/Appwrite/Cloud releases are not performed.

Closed the acknowledgement connection recovery gap in b6a535e. Reproduced both failure modes against released NATS 1.4.0: an acknowledgement write never delivered and a write delivered with its confirmation lost. Previously the next commit failed because the disconnected connection stayed cached. Now command() clears the disconnected factory-created connection, JetStream handle and consumer handles under the lock in finally; the next operation obtains a fresh connection. No retry of the failed operation and no reset of the receive connection or other in-flight messages. Tests verify successful later commits, working queue-depth reads through refreshed consumer handles, redelivery only when the original acknowledgement did not reach NATS, and no replay when commit is called again for the failed delivery. Live-connection sources retain the existing owner-managed replacement contract.

Validation: 185 unit tests / 442 assertions; 55 NATS broker/prefetch integration tests / 276 assertions; Pint, PHPStan, Rector and Vale pass. Updated the design document and removed this blocker. Remote CI is running.

Confidence Score: 4/5

The queue behavior appears mergeable, but the explicit repository requirement against implementation-coupled tests must be satisfied first.

Findings

1.  Test Mirrors Lifecycle Internals ▶

### Issue 1
packages/queue/tests/Queue/Unit/PrefetchTest.php:61-63
This test directly iterates Swoole's protected `$onWorkerStart` storage and reproduces its internal callback argument to enter `Server::start()`. A behavior-preserving lifecycle refactor could therefore break the test even when prefetch still works correctly. `RedisExpiryTest.php:48-50` has the same problem by identifying a Lua operation from its source prefix. This violates the repository directive to test observable behavior rather than mirror source code or configuration, so the requirement must be satisfied before merging.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Summary

This PR redesigns queue consumption around independently bounded prefetch and handler concurrency.

• Adds recoverable Redis batch reservations, grouped settlement and renewal, durable ownership, and Redis Cluster hash-tag validation.
• Adds incremental, independently confirmed NATS acknowledgement batching and command-connection replacement.
• Updates Swoole buffering, renewal, and bounded shutdown behavior.
• Renames public concurrency and batching configuration to coroutines and prefetch, removes priority publishing, and updates Platform to require queue 5.
• Reworks broker tests around real Redis/Dragonfly and observable worker outcomes, although one new prefetch test still mirrors protected lifecycle internals.

_{Reviews (6) · Last reviewed commit: "test(queue): assert release and prefetch..."}

Benchmark interpretation from source review: the screenshot shows Redis about 1.55x NATS throughput for the null-handler batch-16 row (10,043 vs 6,465 msg/s), while I/O/mixed handlers largely hide broker overhead. Concrete differences: Redis uses ext-redis and Lua reserve/claim/settle operations with aggregate replies; the NATS client parses/dispatches messages in PHP and ackBatch still uses an independent confirmation per message (pipelined, not serial round trips). For a normal backlog with batch > 1, NATS pull() performs an empty priority poll, a one-message blocking fetch, then a no-wait top-up; each Consumer::fetch creates and removes an inbox subscription. Benchmark defaults also differ in persistence: NATS File storage with one replica, Redis compose without AOF enabled. These are candidate contributors, not measured attribution: the table is not a CPU profile or a durability-matched comparison. A 600-message null-handler sample at these rates lasts only about 60-93 ms; longer runs and component timing are needed before assigning percentages or selecting an optimization.

Removed priority in 94c48be, net -129 lines. Publisher APIs no longer accept the flag; Redis always uses normal FIFO insertion; background batches group by queue only; NATS has one work consumer, one depth query and no priority poll. Existing normal subject/durable names remain unchanged. Added an integration regression for existing normal messages on a legacy two-consumer topology with an unused priority consumer.

Audited current Appwrite main f2f494fd and Cloud main 72a03895 for queue priority publishing; found no usage. Cloud cutover status still mentions worker_priority, but handles its absence as 404. This does not establish that every deployed priority backlog is empty. Queue 5 migration notes require stopping priority producers and draining pending/unacknowledged legacy priority deliveries with old workers before upgrade. No live resources or messages were deleted.

Validation: 184 unit tests / 440 assertions; final NATS broker suite 50 tests / 258 assertions; 14 background/prefetch integration tests passed (including Redis and NATS). The initial combined run had a test-setup error while constructing the legacy stream config; corrected the fixture and reran the entire NATS suite successfully. Pint, PHPStan, Rector and Vale pass. No throughput gain is claimed without another benchmark run.

Review of 94c48be: three correctness findings to address before merge.

[P1] Do not silently drop expired prefetched payloads on release

packages/queue/src/Queue/Broker/Redis/settle.lua:17-20

With a positive Queue::jobTtl, a prefetched message can lose its jobs payload while waiting for an earlier handler. Renewal extends only the claim, not that payload. release() then deletes the claim/processing entry, skips RPUSH because GET returned false, decrements processing and reports success. Work that never entered a handler disappears. Reproduced against Redis with jobTtl=1: publish, receive, wait 2s, release; ready=0, processing=0, claim absent, no exception. This confirms Greptile's release finding. Preserve the payload for the outstanding delivery (or make release reconstruct it safely); do not report successful release after discarding the only delivery state.

[P1] Preserve long-running Redis jobs on adapters without renewal

packages/queue/src/Queue/Broker/Redis/settle.lua:13 (claim TTL established at claim.lua:20)

Settlement now requires the 90-second claim key to remain live, but only Adapter/Swoole renews deliveries. KubernetesJob and Workerman still inherit Adapter::withAckExtension(), which merely invokes the handler. Consequently a handler lasting beyond CLAIM_TTL completes its side effects and then cannot commit or reject, even when no other consumer reclaimed it and the job payload is intact. It remains processing until recovery and can run again. This is a regression from the previous commit path, which did not require a live heartbeat. Reproduced through KubernetesJob::consume(), accelerating claim expiry with PEXPIRE during its handler: handled=1, success callbacks=0, error='Queue delivery is no longer owned by this consumer'. Provide renewal for all supported execution paths or separate durable ownership from the expiring liveness signal; simply removing the ownership check would weaken fencing.

[P2] Partition grouped Cluster settlements by compatible key placement

packages/queue/src/Queue/Broker/Redis.php:191-197

The broker has one settlement Buffer for every queue/namespace, and concatenates all waiting keys into one script. Individually valid Redis Cluster queues with different namespace hash tags therefore fail once their completions coalesce. Reproduced using one broker with {review316-a-...} and {review316-b-...} namespaces: the first completion succeeds, then the grouped A/B completions both throw 'Atomic Redis queue operations require a shared hash tag'. Both namespaces satisfy the documented placement rule individually. Partition batching by hash tag/slot (or namespace conservatively), retaining independent indexed outcomes.

The consolidated worker loop and JetStream ackBatch boundary are improvements. The remaining modeling issue is that payload retention, ownership and heartbeat lifetime have different meanings, while settlement currently assumes they expire together. Likewise, a batching boundary must represent a valid atomic operation, not just all calls sharing a broker instance.

Validation on an isolated export of the exact head:

• Full configured unit suite: 184 tests / 440 assertions pass.
• Redis reservation integration: 8 tests / 30 assertions pass.
• Redis/NATS prefetch + graceful shutdown integration: 4 tests / 12 assertions pass with isolated Packagist NATS ^1.4 installed. The initial run used an older local NATS dependency; the corrected run passed.
• Additional focused reproductions above against local Redis/Redis Cluster, with uniquely named keys removed afterward. Claim expiry was accelerated, not a 90-second wall-clock run.
• No source changes or production operations. Board item remains In review.

Checked the linked review and both Greptile comments against current head 94c48be. All three correctness findings are valid merge blockers: (1) claim.lua applies jobTtl to payload while extend.lua renews only the claim; settle.lua release removes delivery state and reports success even if payload is missing. (2) settle.lua fences on a 90-second expiring claim, while KubernetesJob and Workerman have no renewal override, so long handlers lose settlement rights without a takeover. (3) the broker-wide settlement Buffer combines keys across namespaces; RedisCluster::execute rejects distinct hash tags even when each queue is valid on its own.

Greptile Release Can Lose Work duplicates finding 1. Its test-quality finding is also valid: BufferTest and RedisReservationTest pin exact scheduler-dependent groupings; LockingTest mirrors the interface method list through reflection. Replace these with behavioral checks for delivery/outcomes, coalescing and serialization rather than exact partitions or a copied API inventory.

Recommended correction: separate payload retention, durable ownership and expiring liveness; retain fencing across actual takeover. Partition settlement batches by compatible placement (namespace is a conservative option). These need focused TTL, non-renewing-adapter and cross-namespace Cluster regressions. Source verification only in this pass; I did not rerun the review reproductions or modify code. The Appwrite/Cloud priority backlog is confirmed empty by the owner and is not a rollout blocker.

Prior-art-based consumer model proposal (design only; no changes to the correctness fixes currently being edited in the PR worktree):

Use Message, Consumer, Worker, receipt, acknowledgement, prefetch, heartbeat. No new public Delivery/Claim/Lease/Settlement objects.

• Primary reference: Dramatiq's Consumer exposes ack/nack/requeue; its Redis consumer counts cached and outstanding messages against prefetch. Its nack semantics differ from our retrying reject, so retain our explicit outcome policy rather than copying names indiscriminately.
• RabbitMQ distinguishes unacknowledged messages, prefetch and channel-scoped acknowledgement identity. Our worker should track unacknowledged messages through confirmation; compatible transport ownership determines what can batch together.
• SQS distinguishes message ID from the receipt handle for a particular receive. An opaque receipt on the existing received Message is enough; a public Delivery wrapper is unnecessary.
• BullMQ couples active-job recovery to reliable worker renewal. Copying its strict lock model without that scheduling contract would not suit our synchronous adapters.

Concrete target:

1. Keep receive/commit/reject; document commit as acknowledgement. Keep release for returning work that never started, without incrementing attempts. Replace the internal ?bool $success and 0/1/2 protocol with explicit commit/reject/release operations.
2. Worker-level prefetch means all unacknowledged messages: waiting, running and awaiting confirmation. Concurrency means running handlers. Batch size is only the count in one broker operation. Proposed queue-5 shape: job('stats-usage', maxCoroutines: 1, prefetch: 100); omitted prefetch defaults to concurrency, explicit prefetch below concurrency is rejected. This would require call-site/migration updates and is not implemented yet.
3. Redis ownership and liveness are separate: heartbeat expiry permits recovery; atomic recovery invalidates ownership. Commit/recovery must have one winner. Preserve payloads until no outstanding work needs them. Document retention changes explicitly instead of silently redefining jobTtl. This is our proposed policy, not a universal broker guarantee.
4. Keep Redis's raw reservation/finalization protocol private to the existing opaque-list compatibility constraint. Dramatiq's ID/payload layout is a separate storage migration.
5. Namespace-scope Redis command grouping; connection-scope NATS acknowledgement grouping. Make the shared buffer internal transport machinery, with one incremental result callback instead of optional callbacks plus aggregate returns. Redis can dispatch each returned script result; NATS dispatches each arriving confirmation.

Deletion targets: nullable outcome interpretation, duplicated result accumulation, generic-looking public Buffer, and redundant state/type wrappers. Keep the consolidated worker loop. A renewal capability should replace reflection only with an explicit third-party migration plan.

Addressed all three findings in #316 (comment) and both Greptile comments in 20b2327.

• Payload retention: outstanding jobs remain available through release/reclaim; jobTtl applies to retained failed/dead payloads after settlement. A missing release payload returns an error before mutating ownership or processing state.
• Ownership/liveness: durable owner tokens are separate from expiring heartbeats. KubernetesJob/Workerman can settle after heartbeat expiry if ownership is unchanged. Recovery atomically rechecks heartbeat and token, invalidates ownership and moves the message; stale completion/renewal cannot undo a takeover. Legacy processing entries remain recoverable. The existing publish-age recovery threshold still must exceed runtime for adapters without renewal.
• Cluster batching: settlement buffers are scoped by namespace, so completions from individually valid namespaces cannot create a cross-slot script.
• Tests: removed exact scheduler partition assertions and interface/private-lock reflection checks; test independent outcomes, coalescing bounds and concurrent serialization instead.

The four primary regressions fail against the previous implementation and pass after the fixes. Final validation: 183 unit tests / 443 assertions; 22 Redis/Cluster/prefetch integration tests / 82 assertions; 16 reservation tests / 54 assertions against Dragonfly. Includes an actual KubernetesJob consume path with accelerated heartbeat expiry, failed-payload TTL, missing-payload release, stale owner after reclaim, and completion/renewal winning races with reclaim. Pint, PHPStan, Rector, Vale and diff checks pass. Remote CI is running. Updated retention and rollback documentation; no production changes.

Design update requested by @loks0n: rename maxCoroutines to coroutines alongside batch → prefetch.

$server->job('stats-usage', coroutines: 1, prefetch: 100);

Apply consistently to parameter names, internal queue specifications, examples and named-argument callers in queue 5; retain positional argument order. This updates the proposed model, not a claim that the implementation has already changed.

Implemented the domain model proposal and naming update in 3addb37.

$server->job('stats-usage', coroutines: 1, prefetch: 100);

• Prefetch limits waiting, running, and unconfirmed messages together. Omission defaults to coroutines; an explicit lower value is rejected before registering the job. Positional argument order is unchanged. Queue specs, Platform configuration, examples, tests, and benchmark options use the same names.
• Redis uses explicit commit / reject / release operations throughout PHP and Lua. Release still returns work that never started without counting a failure. Existing ownership/heartbeat separation, payload retention, atomic recovery, and receipt fencing remain intact.
• The shared buffer is under Internal, with one required incremental result callback. Removed aggregate result accumulation from NATS and dispatch Redis script results individually. A transport failure only fails unresolved requests; earlier confirmations remain successful. Redis grouping remains namespace-scoped and NATS acknowledgements use the broker's command connection.
• No new public message wrappers or renewal interfaces. Existing single-message third-party renewal remains supported.

Queue 5 migration is documented: rename named arguments and Platform job keys from maxCoroutines/batch to coroutines/prefetch; replace Server::batch() with Server::prefetch(). Existing deployment environment names need not change. Map their values to prefetch: max($batch, $coroutines) to preserve the previous effective bound, or omit prefetch to use coroutines. Platform now requires queue 5. Appwrite/Cloud call-site and lock updates remain part of the package-release rollout; no deployment configuration changed here.

Validation: 187 queue unit tests / 459 assertions; 72 Redis, Redis Cluster, NATS and prefetch integration tests / 340 assertions; 16 Dragonfly reservation tests / 54 assertions; 2 Platform worker tests / 8 assertions against the local queue 5 package. New regressions cover unconfirmed messages counting against prefetch, handlers progressing before confirmations, successful confirmations surviving a later transport failure, and rejecting undersized prefetch. Pint, PHPStan, Rector, package validation, documentation lint, and benchmark shell syntax pass.

Re-reviewed head 3addb37842220b3da46691d4d04291762dc5d837. No further blocking findings in this pass.

The three earlier findings are addressed: active Redis payloads survive retention expiry, settlement checks durable ownership independently of heartbeat expiry, and settlement batches are scoped by namespace. Atomic recovery also rechecks ownership and heartbeat before changing state.

The model is substantially cleaner: coroutines controls handler concurrency, prefetch bounds waiting + running + unconfirmed messages, and commit/reject/release are explicit operations. Keeping receipts opaque and using one incremental result contract in the internal buffer avoids adding unnecessary public abstractions. I would keep this structure rather than introduce more domain types.

Independent validation against this head:

• Unit suite: 187 tests, 459 assertions, passed.
• RedisReservationTest + PrefetchTest: 20 tests, 66 assertions, passed against local services, covering retention, heartbeat expiry, recovery races, namespace isolation, and prefetch/shutdown behavior.

Platform linked tests and validation are green in CI. Queue tests, benchmark, and Greptile are still pending at the time of this review. Board status remains In review.

Checked outstanding Greptile threads against 3addb37: one unresolved finding, Test Mirrors Lua Internals. It is valid and still present.

InMemoryConnection::execute() branches on Lua source prefixes and copies argument offsets. It already differs from the real script: release with a missing payload removes processing/ownership in the fake, while settle.lua rejects before mutation. PushRecordingConnection::execute() also assumes key/argument positions and returns success without checking ownership.

Recommended fix: use a Consumer fake for adapter lifecycle/concurrency unit tests; test Redis state transitions against real Redis through observable results. Delete the PHP copy of the Lua protocol rather than extending it. This is a test-design issue, not a newly identified runtime defect. Greptile's latest summary reviewed this exact commit and lists no additional findings. No code changed during this check.

Merged #316 with the requested admin override after every check on 92eb36edaec952aa5578b2fb57ac8471ab3577f6 passed. Merge commit: 5f8b17543e520f6e33bf9325c8b40145778409b6.

Published from that merge commit and verified on Packagist:

• queue 5.0.0, mirror commit 2b7edcbf344c02c8bff72a0b7a573cf8d7af94d4.
• Platform 1.0.0-rc23, mirror commit faa60fd42bc62d98f7433c6a8f03dd6991a48d8b, requiring queue ^5.0.0.

CI tests: queue unit 145 tests / 329 assertions; queue integration 151 tests / 743 assertions; linked Platform 11 tests / 92 assertions. Validation, static checks, benchmarks, and Greptile checks all passed. Both tag-split release workflows succeeded.

One advisory Greptile test-harness coupling note remains visible; it did not fail the review check and was not resolved by this release.

Next rollout steps remain Appwrite/Cloud configuration-key and package-lock updates, staging validation on both brokers, then the stats-usage canary. No application deployments or deployment defaults changed here. Migration and rollback details are included in the release notes.

Both releases were withdrawn at @loks0n's request:

• queue 5.0.0
• Platform 1.0.0-rc23

Deleted both GitHub releases, both package-mirror version tags, and both monorepo release tags (including local tags). Verified that neither version remains in Packagist's Composer metadata. PR #316 remains merged; no source-code revert or application deployment was performed. The earlier release-completion note is superseded by this withdrawal.