feat(cubestore): bound the priority inversion of ADD_AND_RETRIEVE

Priority ordering is enforced by the selection step, not by the claim:
QUEUE PENDING returns items highest priority first and reconcile takes
toProcessLimit off the top of that list. QUEUE RETRIEVE is priority blind and
is safe only because the path it gets comes from that sorted list.

ADD_AND_RETRIEVE selects itself, so it had nothing to compensate for being
priority blind. It now claims the item only while there is almost nothing to
jump over:

| condition                          | meaning                            |
|------------------------------------|------------------------------------|
| active < concurrency               | a slot is free, as before           |
| pending * 2 < concurrency          | the backlog is shallow              |

A claimed item goes straight to active and never becomes pending, so a burst
onto an idle queue still fast tracks every query. Items start to accumulate in
pending only when the concurrency budget is exhausted, which is exactly when
the fast track should step aside and let reconcile pick by priority.

The condition is written as a multiplication, because concurrency / 2
truncates to 0 for a queue with concurrency 1 (the pre-aggregation build
queue), which would disable the fast track there entirely. An item which is
already pending is not counted as its own backlog, otherwise claiming an
existing item would never happen for the small concurrency queues.

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

Author: ovr

packages/cubejs-query-orchestrator/DEVELOPMENT.md

@@ -230,17 +230,24 @@ active and returns the payload. Between those two calls another node can take th
 concurrency slot, so the enqueueing node often pays for the second round-trip and gets
 nothing back.
 
-`QUEUE ADD_AND_RETRIEVE` inserts **and** claims the item in one atomic operation when the
-prefix has a free concurrency slot, so the enqueueing request can go straight to executing:
+`QUEUE ADD_AND_RETRIEVE` inserts **and** claims the item in one atomic operation, so the
+enqueueing request can go straight to executing:
 
 ```
 QUEUE ADD_AND_RETRIEVE [EXCLUSIVE] [PRIORITY ?n] [ORPHANED ?ttl] [EXTERNAL_ID ?id]
     ?path ?payload ?concurrency
 ```
 
-`payload IS NULL` in the response means the item was not claimed — the budget was full, the
-item was already active, or it belongs to another process — and the caller falls back to
-the normal path with nothing lost, because the item is enqueued either way.
+The item is claimed when both hold for its prefix:
+
+- a concurrency slot is free — `active < concurrency`, the same budget
+  `QUEUE RETRIEVE CONCURRENCY` uses;
+- the backlog is shallow — `pending < concurrency / 2`, not counting the item itself.
+
+`payload IS NULL` in the response means the item was not claimed — one of the two
+conditions failed, the item was already active, or it belongs to another process — and the
+caller falls back to the normal path with nothing lost, because the item is enqueued either
+way.
 
 ```mermaid
 sequenceDiagram
@@ -287,10 +294,18 @@ Everything after the claim is unchanged: `MERGE_EXTRA`, `HEARTBEAT`, `ACK` and
 `RESULT_BLOCKING` behave exactly as in the normal path, and a fast-tracked item is a
 regular active item — `TO_CANCEL` will reclaim it if the heartbeat stops.
 
-The concurrency budget is the same one `QUEUE RETRIEVE CONCURRENCY` uses: prefix scoped,
-and blind to both exclusivity and priority. Being priority blind is the trade-off of the
-fast track — it claims the item being added even when a higher priority item is pending in
-the same prefix, so it should not be used for queues that rely on priority ordering.
+Priority ordering is enforced by the *selection* step, not by the claim: `QUEUE PENDING`
+returns items highest priority first (oldest first within a priority) and reconcile takes
+`toProcessLimit` off the top of that list. `QUEUE RETRIEVE <path>` itself is priority blind
+— it is safe only because the path it is given came from that sorted list.
+
+The fast track selects itself, so it is priority blind with nothing to compensate. That is
+what the `pending < concurrency / 2` condition bounds: with a shallow backlog there is
+almost nothing to jump over, and with a deep one the fast track steps aside and lets
+reconcile pick by priority. Note that a claimed item goes straight to active and never
+becomes pending, so a burst onto an idle queue still fast-tracks every query — items only
+start accumulating in pending once the concurrency budget is exhausted, which is exactly
+when the condition should stop firing.
 
 > **Status:** the `QUEUE ADD_AND_RETRIEVE` command exists in Cube Store. The driver still
 > emits `QUEUE ADD`; wiring the fast track into `CubeStoreQueueDriver.addToQueue` needs a

rust/cubestore/cubestore/src/cachestore/cache_rocksstore.rs

@@ -1467,12 +1467,24 @@ impl CacheStore for RocksCacheStore {
             let queue_schema = QueueItemRocksTable::new(db_ref.clone());
             let (pending, mut active) = Self::queue_prefix_counters(&queue_schema, &payload.path)?;
 
-            let claim = active.len() < (payload.concurrency as usize);
-
             let index_key = QueueItemIndexKey::ByPath(payload.path.clone());
             let id_row_opt = queue_schema
                 .get_single_opt_row_by_index(&index_key, &QueueItemRocksIndex::ByPath)?;
 
+            // An item which is already pending is not a part of its own backlog
+            let backlog = match &id_row_opt {
+                Some(id_row) if id_row.get_row().get_status() == &QueueItemStatus::Pending => {
+                    pending.saturating_sub(1)
+                }
+                _ => pending,
+            };
+
+            // A deep backlog is left to QUEUE PENDING + reconcile, which pick by priority.
+            // Claiming the item being added ignores priority, it's acceptable only while
+            // there is almost nothing to jump over.
+            let claim = active.len() < (payload.concurrency as usize)
+                && backlog * 2 < (payload.concurrency as u64);
+
             if let Some(id_row) = id_row_opt {
                 let id = id_row.get_id();
                 let claim_result = if claim {
@@ -2945,6 +2957,57 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_queue_add_and_retrieve_backlog() -> Result<(), CubeError> {
+        init_test_logger().await;
+
+        let (_, cachestore) = RocksCacheStore::prepare_test_cachestore(
+            "test_queue_add_and_retrieve_backlog",
+            Config::test("test_queue_add_and_retrieve_backlog"),
+        );
+
+        for path in ["prefix:path1", "prefix:path2"] {
+            cachestore
+                .queue_add(QueueAddPayload {
+                    path: path.to_string(),
+                    value: "v".to_string(),
+                    priority: 0,
+                    orphaned: None,
+                    process_id: None,
+                    exclusive: false,
+                    external_id: None,
+                })
+                .await?;
+        }
+
+        // 2 pending items is not less than a half of 4, the backlog is left to reconcile
+        // even though all the concurrency slots are free
+        let res = cachestore
+            .queue_add_and_retrieve(queue_add_and_retrieve_payload("prefix:path3", "v3", 4))
+            .await?;
+        assert!(res.added);
+        assert_eq!(res.payload, None);
+        assert_eq!(res.active, Vec::<String>::new());
+        assert_eq!(res.pending, 3);
+
+        assert_queue_item_status(&cachestore, "path3", QueueItemStatus::Pending, false).await?;
+
+        // 3 pending items is less than a half of 7
+        let res = cachestore
+            .queue_add_and_retrieve(queue_add_and_retrieve_payload("prefix:path4", "v4", 7))
+            .await?;
+        assert!(res.added);
+        assert_eq!(res.payload, Some("v4".to_string()));
+        assert_eq!(res.active, vec!["path4".to_string()]);
+        assert_eq!(res.pending, 3);
+
+        assert_queue_item_status(&cachestore, "path4", QueueItemStatus::Active, true).await?;
+
+        RocksCacheStore::cleanup_test_cachestore("test_queue_add_and_retrieve_backlog");
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_queue_add_and_retrieve_exclusive() -> Result<(), CubeError> {
         init_test_logger().await;