perf: one round trip per enqueue, and back the result poll off to 1s - #5

Producers run on Cloud Run in us-east4; the ModelQ Redis is a Vultr host in Mumbai. Measured RTT is ~271ms, so every reply this library waits for before sending the next command is a quarter of a second on a caller's request path.

1. enqueue() was five sequential round trips

$this->enqueueTask($taskDict, $payload);   // rPush ml_tasks + zAdd queued_requests
$this->redis->setex("task:{$id}", ...);    // wait
$this->addToTaskHistory($id, $taskDict);   // zAdd task_history + setex task_history:

Five commands, five waits.

Measured: across 5,434 production requests joined between ModelQ Redis and MongoDB, the gap between queued_at and the request record landing is a flat 1.355–1.41s — identical regardless of payload size, and identical across queues. Five round trips at 271ms, plus a Mongo insert.

Now one pipeline. Redis still executes them in order; only the waiting is removed.

Ordering also fixed: rPush ml_tasks used to go first, so a worker could BLPOP a task whose task:{id} key did not exist yet. State is now written before the id is advertised — free, since it's the same pipeline.

2. The result poll woke 10× a second and asked twice each time

if ($this->isCancelled($redis)) { ... }             // GET  -> wait
$taskJson = $redis->get("task_result:{$this->taskId}"); // GET  -> wait
usleep(100000);

On a 30s wait that's ~100 round trips to learn nothing, with the caller's Octane worker pinned for all of it. On flux_klein only 5.7% of tasks finish inside 30s, so 94% of those requests paid the full cost and then had to poll /fetch anyway.

Two changes:

• the pair of GETs is now one pipelined read;
• the interval starts at 100ms and doubles to a 1s ceiling.

Why backoff rather than a flat 1s

A flat 1s would regress the fast endpoints. Measured GPU time:

A flat 1s poll adds up to a second to a one-second job. Starting tight and backing off gives sub-second tasks the loop they need and long ones a quiet one:

Both constants are public (Task::POLL_MIN_INTERVAL_US, Task::POLL_MAX_INTERVAL_US) if you'd rather pin it flat.

Compatibility

• Wire format unchanged — same five keys, same values, same scores.
• isCancelled() semantics preserved exactly, including the edge case where a stored "0" still counts as cancelled (a naive (bool) cast on the pipelined reply would have got that wrong).
• A pipeline that doesn't come back cleanly is treated as "no verdict yet" and the loop keeps waiting, rather than reading a broken reply as a result.
• addToTaskHistory() removed — it was private and its only caller is now the pipeline.

Tests

20 new tests. tests/Unit/RoundTripTest.php pins the round-trip count (not the command list — the commands were never the problem) with a recording Redis double. tests/Integration/PipelinedEnqueueTest.php runs the real phpredis against a live Redis on a dedicated DB index, because a mock cannot catch a misuse of the actual pipeline API.

• Unit suite: 60 passed
• PHPStan: no errors
• Integration: 13 pre-existing failures, identical set before and after (verified by diffing the failure list against a clean main — they're payload-shape assertions unrelated to this change)

Mutation-tested:

Deploying

The app that consumes this pins modelslab/modelq: @dev; production is currently running a build that still writes the payload three times (before #4). Worth landing both.

^{Need help on this PR? Tag @codesmith-bot with what you need. Autofix is disabled.}