feat: Queue - support fast track feature - #11618
Conversation
…E ADD_AND_RETRIEVE `QUEUE ADD_AND_RETRIEVE` inserts and claims a queue item in one atomic operation, but nothing on the JS side used it. Enqueueing a query cost five Cube Store round-trips before it could start executing, and between `QUEUE ADD` and `QUEUE RETRIEVE` another node could take the concurrency slot, so the enqueueing node often paid for the retrieval and got nothing back. `QueueDriverConnectionInterface` gains `addAndRetrieve`, whose response always carries a claim slot. A claim is an ownership transfer: the caller owns an active item and must execute and acknowledge it, so it gets its own method instead of riding along on `addToQueue`. `QueryQueue` hands the claimed `QueryDef` straight to `sendProcessMessageFn` and skips reconcile, the retrieval and the queue state lookup. | Step | Normal | Fast track | |---|---|---| | `QUEUE ADD` | 1 round-trip | folded into one command | | `QUEUE TO_CANCEL` + `QUEUE LIST` (reconcile) | 2 round-trips | skipped | | `QUEUE RETRIEVE` | 1 round-trip | folded into one command | | `QUEUE LIST` (the `Waiting for query` event) | 1 round-trip | skipped, the claim carries the state | | Window for another node to steal the slot | between ADD and RETRIEVE | none | Off by default, `CUBEJS_QUEUE_FAST_TRACK=true` makes `QueryQueue` call `addAndRetrieve`. Claiming is then up to the driver: the Cube Store one negotiates the `queueAddAndRetrieve` capability and falls back to `QUEUE ADD`, the memory one never claims. An unclaimed response carries a `null` claim and the caller continues through reconcile with nothing lost, because the item is enqueued either way. The flag is opt-in because a `sendProcessMessageFn` override which drops the claim leaves an active item with nobody running it, and the query stalls until `TO_CANCEL` reclaims it. Two side effects, both documented in DEVELOPMENT.md: `queueSize` drops by one on the fast track since a claimed item is never pending, and submission-time orphan collection no longer runs — it still runs after every completed query, and a submission only skips reconcile while the concurrency budget is free, which is exactly when there is nothing to reclaim. Also subscribes the `streamStarted` listener before the query is dispatched instead of after, which retires the race the removed TODO described: a stream handler which starts fast used to emit the event before anyone listened for it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
github-actions
Bot
added
javascript
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #11618 +/- ##
=======================================
Coverage 59.33% 59.33%
=======================================
Files 228 228
Lines 18222 18242 +20
Branches 3675 3684 +9
=======================================
+ Hits 10812 10824 +12
- Misses 6863 6866 +3
- Partials 547 552 +5
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
…iver Follow-up on the `QUEUE ADD_AND_RETRIEVE` fast track. `executeInQueue` used to read `CUBEJS_QUEUE_FAST_TRACK` and pick between `addToQueue` and `addAndRetrieve`, which made the queue responsible for a decision only the driver can finish anyway — it is the one which negotiates the `queueAddAndRetrieve` capability. It now always calls `addAndRetrieve`, and the Cube Store driver combines the flag with the capability in `useFastTrack()`, next to the identical `useExternalId()`. A driver which does not claim returns a `null` claim, so the caller only ever sees "claimed or not". `keyScore` and `orphanedTime` are gone from both methods. Cube Store never read them, it derives its own from `priority` and `options.orphanedTimeout`; the memory driver now does the same instead of having them computed for it by the caller. That also drops a latent `NaN`: a query carrying an `orphanedTimeout` key with an `undefined` value used to produce `undefined * 1000` as its orphan deadline, which never compares as orphaned. It falls back to the queue default now. Also dropped comments which restate the code they sit on.
|
Preview deployment for your docs. Learn more about Mintlify Previews.
💡 Tip: Enable Workflows to automatically generate PRs for you. |
`addAndRetrieve` was a second insert method which differed from `addToQueue` only in whether the driver also claimed the item, and every caller had to pick one. There is one `addToQueue` again: `AddToQueueResponse` always carries a claim slot, and the driver decides whether to fill it. The Cube Store driver picks `QUEUE ADD_AND_RETRIEVE` over `QUEUE ADD` in one place, so the non-fast path no longer builds the command twice by re-entering itself, and the memory driver loses its override entirely.
|
Claude finished @ovr's task in 4m 50s —— View job Review complete — no blockers; 1 medium (claim hand-off leaks an active item on failure), 3 low (breaking interface arity, weak assertion, test gaps). 4 inline comments posted.Full reviewTodos
What's goodThe shape of this is right. Keeping the decision "do I claim?" entirely inside the driver — one A few specific things I checked and found correct:
Findings
Smaller observations (not posted inline)
Not verifiedI couldn't run the test suite or the linter — |
The four settings were hardcoded, so comparing two queue configurations meant editing the file between runs. Defaults are unchanged. Measured with them on a release Cube Store 1.7.25, 1000 queries, `WORKERS=0`, `BENCH_CONCURRENCY=500`, 64KB responses, 16KB payloads: | Driver call | Fast track off | Fast track on | | --- | --- | --- | | `retrieveForProcessing` | 1000 | 0 | | `getQueryStageState` | 1000 | 0 | | `getQueriesToCancel` (reconcile) | 1980 | 1000 | | `getActiveAndToProcess` (reconcile) | 1980 | 1000 | | total driver calls | 10960 | 7000 | Wall clock is 12s either way: the bench is bound by its 1500ms handler and its 10ms pusher interval, so it measures round trips, not latency. At `BENCH_CONCURRENCY=50` only 50 of 1000 queries are claimed - the budget saturates after the initial fill and the claim condition steps aside, which is what keeps the fast track from jumping the priority order.
|
Claude finished @ovr's task in 5m 11s —— View job Review complete — no blockers; 2 medium, 4 low. Feature is correctly gated (env var + Cube Store 1.7.25 capability), and the Cube Store side reserves a slot per pending item so the claim can't jump the priority queue. 6 inline comments posted.Full review — findings, verification notes, and checklistChecklist
What I verifiedI cross-checked the JS side against the Rust implementation it depends on, since most of the risk in this change lives at that seam:
I did not run the build or test suite — this is a fresh checkout without installed workspace deps, and the fast-track tests need a live Cube Store. Worth confirming Findings
Things I liked
|
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
|
|
||
| if (claim) { | ||
| // The item is active already, there is nothing for reconcile to pick up | ||
| await this.sendProcessMessageFn(this.claimedQuery(queryKeyHash, queueId, claim)); |
There was a problem hiding this comment.
The claim hand-off has no failure handling, so a failure leaks an active queue item.
By the time addToQueue returns a claim, Cube Store has already moved the item to active and this process owns the processing lock (as DEVELOPMENT.md puts it, "a receiver which drops the hand-off leaves the item active with nobody running it"). Two things on this line can throw:
claimedQuery()throws'Queue driver claimed a query without reporting its queue id'whenclaim[1]andqueueIdare both null;sendProcessMessageFnis a documented extension point (message bus / worker dispatch) and can reject.
Either way the exception propagates out of executeInQueue and nothing frees the lock — the item sits active with no runner until TO_CANCEL reclaims it on the heartbeat/orphan timeout, and the fast track has just skipped the reconcileQueue() that would have run TO_CANCEL. The equivalent call in processQuery (line ~814) is deliberately wrapped:
try {
await this.sendProcessMessageFn(claimed);
} catch (e: any) {
this.logger('Error while sending process message', { ... });
}Suggest the same shape here, plus queueConnection.freeProcessingLock(queryKeyHash, processingId, true) on failure so reconcile can re-pick the item immediately instead of waiting out the orphan timeout. (For the null-queueId case specifically, falling back to reconcileQueue() is likely better than throwing — the item is still recoverable, the caller's query is not.)
|
|
||
| // The handler can emit `streamStarted` and finish before executeInQueue awaits for | ||
| // it, which is safe only because the listener is subscribed before the dispatch | ||
| expect(stream).toBeDefined(); |
There was a problem hiding this comment.
expect(null).toBeDefined() passes, and null is exactly the failure mode this test guards against — waitForQueryStream resolves null when the streamStarted event is missed. The assertion only fails indirectly, via for await (const chunk of null) throwing a TypeError two lines down, which reads as an unrelated crash.
| expect(stream).toBeDefined(); | |
| expect(stream).not.toBeNull(); |
| }); | ||
|
|
||
| // eslint-disable-next-line no-unused-expressions | ||
| options.cacheAndQueueDriver === 'cubestore' && describe('with CUBEJS_QUEUE_FAST_TRACK enabled', () => { |
There was a problem hiding this comment.
Two gaps in this block, both on paths the PR specifically changes:
- No fast-track stream test. The listener-reordering in
waitForQueryStreamexists because the fast track dispatches sooner than reconcile did — that's the narrowest window in the whole change, and it's only exercised with the flag off (stream handler which starts immediately, line 367). AqueryHandler: 'stream'+persistent: truecase here would also coverADD_AND_RETRIEVE EXCLUSIVE, which Cube Store rejects outright without anx-process-idheader (cachestore.rs:393) — currently nothing checks that combination works at all. - The
added === 0 && claim !== nullbranch is untested.CubeStoreQueueDriver.ts:155explicitly documents it ("An item which already existed is never added twice, but it still can be claimed"), andQueryQueue.ts:322depends on it by reading the query def out ofclaim[4]instead of callinggetQueryDef. The third test here only asserts the opposite case (an active item is not re-claimed). A pending-but-unclaimed item that a secondaddToQueuethen claims would pin the branch down.
Nit on line 609: jest.setTimeout isn't scoped to the describe, so this 10s value applies to every test declared after it in the file too (there's a matching one at line 498 already).
|
|
||
| if (claim) { | ||
| // The item is active already, there is nothing for reconcile to pick up | ||
| await this.sendProcessMessageFn(this.claimedQuery(queryKeyHash, queueId, claim)); |
There was a problem hiding this comment.
The dispatch isn't guarded the way processQuery guards it. In processQuery (line ~813) a throwing sendProcessMessageFn is caught and logged as Error while sending process message, and the caller keeps waiting on getResultBlocking until the orphan reclaim picks the item up. Here the rejection escapes the outer try and is rethrown out of executeInQueue, so a custom sendProcessMessageFn (the whole point of the option — handing the query to another process) turns a queue-dispatch hiccup into a failed API request, while the item stays Active in Cube Store with nobody processing it.
The default handler ((claimed) => { this.executeQuery(claimed); }) never throws, so this only bites custom implementations — but that's exactly the configuration this feature is aimed at.
| await this.sendProcessMessageFn(this.claimedQuery(queryKeyHash, queueId, claim)); | |
| if (claim) { | |
| // The item is active already, there is nothing for reconcile to pick up | |
| const claimed = this.claimedQuery(queryKeyHash, queueId, claim); | |
| try { | |
| await this.sendProcessMessageFn(claimed); | |
| } catch (e: any) { | |
| this.logger('Error while sending process message', { | |
| queueId: claimed.queueId, | |
| queryKey: claimed.query.queryKey, | |
| requestId: claimed.query.requestId, | |
| error: (e.stack || e).toString(), | |
| queuePrefix: this.redisQueuePrefix | |
| }); | |
| } | |
| } else { |
| if (claimedQueueId === null) { | ||
| throw new Error('Queue driver claimed a query without reporting its queue id'); | ||
| } |
There was a problem hiding this comment.
Throwing here is the worst of both worlds: by the time we get a claim back, Cube Store has already flipped the item to Active, so the throw fails the request and leaves the item wedged until the orphan/heartbeat reclaim. Since a claim without an id is a driver bug rather than a user-visible condition, it'd be safer to log it and fall back to reconcileQueue() (which will re-derive the id via retrieveForProcessing / skip processing and free the lock) instead of aborting the request.
Not reachable with the current Cube Store driver (ADD_AND_RETRIEVE always returns id), so this is about the failure mode, not a live bug.
| const [active, toProcess] = claim ? [claim[2], undefined] : await queueConnection.getQueryStageState(true); | ||
|
|
||
| this.logger('Waiting for query', { | ||
| ...waitingContext, | ||
| queueSize, | ||
| activeQueryKeys: active, | ||
| toProcessQueryKeys: toProcess, | ||
| active: active.indexOf(queryKeyHash) !== -1, | ||
| queueIndex: toProcess.indexOf(queryKeyHash), | ||
| queueIndex: toProcess ? toProcess.indexOf(queryKeyHash) : -1, | ||
| fastTrack: !!claim, |
There was a problem hiding this comment.
Skipping getQueryStageState is the right call for latency, but it silently changes the shape of the Waiting for query event: on the fast track toProcessQueryKeys becomes undefined (not []) and queueIndex is -1 rather than the real position. Anything downstream that treats toProcessQueryKeys as an array (log processors, Cube Cloud dashboards, logger assertions) will now see undefined. [claim[2], []] would keep the event schema stable at no cost, and queueIndex: -1 is already correct for a claimed item since it's never pending.
| }); | ||
|
|
||
| // eslint-disable-next-line no-unused-expressions | ||
| options.cacheAndQueueDriver === 'cubestore' && describe('with CUBEJS_QUEUE_FAST_TRACK enabled', () => { |
There was a problem hiding this comment.
Good coverage of the happy path, the dedup path and the budget-exhausted path. Two gaps worth closing before this graduates from experimental:
- Persistent / exclusive queries.
buildAddCommandemitsEXCLUSIVEforqueryKey.persistent, and Cube Store hard-errors onQUEUE ADD_AND_RETRIEVE EXCLUSIVEwhen the connection has nox-process-id(sql/cachestore.rs:392, and again inqueue_add_and_retrieve).WebSocketConnectiondoes send the header, so it should work — but nothing here proves it, and this is the one combination where the fast track can fail with a user error rather than degrading to the normal path. - Streams on the fast track. The
stream handler which starts immediatelytest exercises the reorderedwaitForQueryStreamsubscription, but only with fast track off. The fast track shortens the window between subscription and dispatch further; running that same test inside this describe would pin the interaction the PR body is really about.
| const command = fastTrack ? 'ADD_AND_RETRIEVE' : 'ADD'; | ||
| const rows = await this.driver.query<CubeStoreClaimResponse & { added: string }>(`QUEUE ${command}${modifiers}${fastTrack ? ' ?' : ''}`, values); |
There was a problem hiding this comment.
Minor typing nit: the row type claims active / payload / extra unconditionally, but plain QUEUE ADD only returns id, added and pending (sql/cachestore.rs). It's safe today because decodeClaimFromRow is only reached when fastTrack is true, but the type is lying about the non-fast-track branch. CubeStoreClaimResponse & { added: string } vs. { id, added, pending } | CubeStoreClaimResponse & { added: string } would keep the compiler honest if someone later reads rows[0].active outside the fastTrack guard.
The pusher pushed a query every 10ms, so a run only ever measured a queue under a burst. `BENCH_PERIOD_MS` spreads the queries evenly over that window instead, which is what decides whether the queue ever backlogs. Unset keeps the 10ms push, so existing runs are unchanged. Concurrency 10 with the harness' 1500ms handler serves 6.67 queries/s. Sweeping the arrival rate across it, 1000 queries, single process, release Cube Store 1.7.25: | Period | Arrival | Driver calls off | on | Claimed | | --- | --- | --- | --- | --- | | 60s | 16.7/s, 2.5x over capacity | 14059 | 14234 | 10/1000 | | 180s | 5.6/s, 0.84x | 11000 | 7000 | 1000/1000 | | 300s | 3.3/s, 0.5x | 11000 | 7000 | 1000/1000 | Below capacity every query is claimed and it is 11 round trips per query against 7 - RETRIEVE and LIST go to zero, reconciles halve. Above capacity the fast track claims one budget's worth and stands down, and enabling it costs a little rather than nothing (+1.2% single process, +3.0% with two workers - small, one run each, but the sign is consistent). Pacing also fixes a measurement problem: the 10ms push at concurrency 10 timed out 150-160 queries per run against the 120s continueWaitTimeout and ran only ~860 handlers. Paced, every run completes all 1000 with no continue-wait at all.
|
Claude finished @ovr's task in 1m 49s —— View job Re-review of the new commit — bench pacing only; no new blockers. 1 low posted inline, 6 prior threads re-affirmed, 2 resolved.Full reviewTodos
Scope of this passOnly one commit landed since my last review (
No production code changed, so everything I said about The pacing changeThis is the right instrument for the question the PR is asking. Pushing every 10ms at concurrency 10 only ever measured a queue in permanent overload — where the fast track is structurally unable to help, because Two details I checked:
The measurement fix buried at the end of the commit message is the more important part: the unpaced run was timing out 150–160 of 1000 queries against Finding
Prior threadsRe-affirmed, still unresolved and still applicable (nothing on these lines changed):
Resolved the two Not verifiedSame as before: no |
| totalQueries: 1_000, | ||
| currency: parseInt(process.env.BENCH_CONCURRENCY || '50', 10), | ||
| totalQueries, | ||
| pushIntervalMs: periodMs > 0 ? Math.max(1, Math.round(periodMs / totalQueries)) : 10, |
There was a problem hiding this comment.
Two ways the realised arrival rate can differ from the one the run is labelled with, both worth knowing when the numbers end up in a commit message:
- Integer rounding is amplified by
totalQueries. The effective window istotalQueries * round(periodMs / totalQueries), so anything that doesn't divide evenly drifts by up tototalQueries / 2ms. At the ratios in the commit table (60000/1000 = 60ms) it's exact, but e.g.BENCH_PERIOD_MS=45500 BENCH_TOTAL_QUERIES=1000runs a 46s window while reporting 45.5s. TheMath.max(1, …)floor is the extreme case: anyperiodMs < totalQueriessilently becomes a 1ms push, i.e. a burst, which is exactly the regime the flag exists to avoid. setIntervaldoesn't await the callback, so pacing is preserved (good — theawait runningat line 289 doesn't back-pressure the pusher), but the timer can't sustain sub-~1msticks and drifts under event-loop load at small intervals.
Cheap fix: log the derived interval and the implied arrival rate alongside benchSettings (it's already printed each tick), so a run is self-describing rather than relying on the operator to recompute it. Not blocking — bench-only.
1 participant
QUEUE ADD_AND_RETRIEVEinserts and claims a queue item in one atomic operation, but nothing on the JS side used it. Enqueueing a query cost five Cube Store round-trips before it could start executing, and betweenQUEUE ADDandQUEUE RETRIEVEanother node could take the concurrency slot, so the enqueueing node often paid for the retrieval and got nothing back.