feat(cubestore): Queue - allow path lookup, if there was miss on external_id (#10543)

Author: ovr

rust/cubestore/cubestore-sql-tests/src/tests.rs

@@ -294,6 +294,10 @@ pub fn sql_tests(prefix: &str) -> Vec<(&'static str, TestFn)> {
         ),
         t("queue_custom_orphaned", queue_custom_orphaned),
         t("queue_result_by_external_id", queue_result_by_external_id),
+        t(
+            "queue_result_by_id_external_id_mismatch",
+            queue_result_by_id_external_id_mismatch,
+        ),
         t("limit_pushdown_group", limit_pushdown_group),
         t("limit_pushdown_group_order", limit_pushdown_group_order),
         t(
@@ -356,6 +360,7 @@ lazy_static::lazy_static! {
         "queue_ack_then_result_v2_with_external_id",
         "queue_custom_orphaned",
         "queue_result_by_external_id",
+        "queue_result_by_id_external_id_mismatch",
         "queue_full_workflow_v1",
         "queue_full_workflow_v2",
         "queue_full_workflow_v2_with_external_id",
@@ -11254,6 +11259,57 @@ async fn queue_full_workflow_v2_with_external_id(
     Ok(())
 }
 
+async fn queue_result_by_id_external_id_mismatch(
+    service: Box<dyn SqlClient>,
+) -> Result<(), CubeError> {
+    let add_response = service
+        .exec_query(
+            r#"QUEUE ADD PRIORITY 1 EXTERNAL_ID 'ext-match' "STANDALONE#queue:mismatch_test" "payload_mismatch";"#,
+        )
+        .await?;
+    let id = assert_queue_add_and_get_id(&add_response)?;
+
+    let ack_result = service
+        .exec_query(&format!(r#"QUEUE ACK {} "result:mismatch""#, id))
+        .await?;
+    assert_eq!(
+        ack_result.get_rows(),
+        &vec![Row::new(vec![TableValue::Boolean(true)])]
+    );
+
+    let result = service
+        .exec_query(&format!(r#"QUEUE RESULT EXTERNAL_ID "ext-match" {}"#, id))
+        .await?;
+    assert_queue_result_columns(&result);
+    assert_eq!(
+        result.get_rows(),
+        &vec![queue_result_row("result:mismatch", &id, Some("ext-match"))]
+    );
+
+    // External_id allows many reads
+    let result = service
+        .exec_query(&format!(r#"QUEUE RESULT EXTERNAL_ID "ext-match" {}"#, id))
+        .await?;
+    assert_queue_result_columns(&result);
+    assert_eq!(
+        result.get_rows(),
+        &vec![queue_result_row("result:mismatch", &id, Some("ext-match"))]
+    );
+
+    let err = service
+        .exec_query(&format!(r#"QUEUE RESULT EXTERNAL_ID "wrong-ext" {}"#, id))
+        .await;
+    assert!(err.is_err(), "Expected error for external_id mismatch");
+    let err_msg = err.unwrap_err().message;
+    assert!(
+        err_msg.contains("external_id mismatch"),
+        "Error should mention external_id mismatch, got: {}",
+        err_msg
+    );
+
+    Ok(())
+}
+
 async fn sys_cachestore_info(service: Box<dyn SqlClient>) -> Result<(), CubeError> {
     service.migration_run_next_query();
     service.exec_query("SYS CACHESTORE INFO").await?;

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

@@ -595,44 +595,45 @@ impl RocksCacheStore {
                 };
 
                 let id = queue_result.get_id();
-                let external_id = queue_result.get_row().get_external_id().clone();
+                let row_external_id = queue_result.get_row().get_external_id().clone();
+
+                if let Some(ref external_id) = external_id {
+                    if row_external_id.as_ref() != Some(external_id) {
+                        return Err(CubeError::user(format!(
+                            "Queue result (id = {}) external_id mismatch: expected {}, got {:?}",
+                            id, external_id, row_external_id
+                        )));
+                    }
+                }
 
                 return Ok(Some(QueueResultResponse::Success {
                     value: Some(queue_result.into_row().value),
                     id,
-                    external_id,
+                    external_id: row_external_id,
                 }));
             };
 
             // try external_id first (if provided), then fall back to path lookup
+            // external_id can be different for path, because path is re-used across different requests across time
             if let Some(ref external_id) = external_id {
-                let Some(queue_result) =
+                if let Some(queue_result) =
                     result_schema.get_row_by_external_id(external_id.clone())?
-                else {
-                    return Ok(None);
-                };
-
-                let id = queue_result.get_id();
-                let external_id = queue_result.get_row().get_external_id().clone();
+                {
+                    let id = queue_result.get_id();
+                    let external_id = queue_result.get_row().get_external_id().clone();
 
-                return Ok(Some(QueueResultResponse::Success {
-                    value: Some(queue_result.into_row().value),
-                    id,
-                    external_id,
-                }));
-            };
+                    return Ok(Some(QueueResultResponse::Success {
+                        value: Some(queue_result.into_row().value),
+                        id,
+                        external_id,
+                    }));
+                }
+            }
 
             let Some(queue_result) = result_schema.get_row_by_key(key)? else {
                 return Ok(None);
             };
 
-            // When external_id filter is active, only return if it matches
-            if let Some(ref external_id) = external_id {
-                if queue_result.get_row().get_external_id().as_ref() != Some(external_id) {
-                    return Ok(None);
-                }
-            }
-
             if queue_result.get_row().is_deleted() {
                 Ok(None)
             } else {