fix: use wildcard query when Redis index value is empty (#2069)

* fix: use wildcard query when Redis index value is empty

When searching Redis with an empty filter, the query was generating
`@nl_state:{  }` which causes a syntax error at offset 10. This fix
uses `*` (match all) when the sanitized field is empty.

Also updates the test mock to handle wildcard queries and skip entries
without version fields (e.g., those created via HSET).

Fixes: Syntax error at offset 10 near state

* complete merge to comply with CI

* fix: resolve merge conflict in search_by_index_prefix

Merged main into fix-redis-query-syntax branch and resolved conflict
in the search_by_index_prefix method to work with RecoverablePool.

Changes:
- Extract Client from ClientWithPermit using client.client.clone()
- Pass sanitized_field as parameter to closure for proper ownership
- Maintain wildcard query fix (*) for empty filter values
- Maintain input sanitization security
,

Author: MarcusSorealheis

nativelink-scheduler/tests/redis_store_awaited_action_db_test.rs

@@ -217,37 +217,56 @@ impl Mocks for FakeRedisBackend {
         }
 
         if actual.cmd == Str::from_static("FT.AGGREGATE") {
-            // The query is @field:value where value might be wrapped in braces.
+            // The query is either "*" (match all) or @field:{ value }.
             let query = actual.args[1]
                 .clone()
                 .into_string()
                 .expect("Aggregate query should be a string");
-            assert_eq!(&query[..1], "@");
-            let mut parts = query[1..].split(':');
-            let field = parts.next().expect("No field name");
-            let value = parts.next().expect("No value");
-            let value = value
-                .strip_prefix("{ ")
-                .and_then(|s| s.strip_suffix(" }"))
-                .unwrap_or(value);
             // Lazy implementation making assumptions.
             assert_eq!(
                 actual.args[2..6],
                 vec!["LOAD".into(), 2.into(), "data".into(), "version".into()]
             );
             let mut results = vec![RedisValue::Integer(0)];
-            for fields in self.table.lock().values() {
-                if let Some(key_value) = fields.get(field) {
-                    if *key_value == RedisValue::Bytes(Bytes::from(value.to_owned())) {
+
+            if query == "*" {
+                // Wildcard query - return all records that have both data and version fields.
+                // Some entries (e.g., from HSET) may not have version field.
+                for fields in self.table.lock().values() {
+                    if let (Some(data), Some(version)) = (fields.get("data"), fields.get("version"))
+                    {
                         results.push(RedisValue::Array(vec![
                             RedisValue::Bytes(Bytes::from("data")),
-                            fields.get("data").expect("No data field").clone(),
+                            data.clone(),
                             RedisValue::Bytes(Bytes::from("version")),
-                            fields.get("version").expect("No version field").clone(),
+                            version.clone(),
                         ]));
                     }
                 }
+            } else {
+                // Field-specific query: @field:{ value }
+                assert_eq!(&query[..1], "@");
+                let mut parts = query[1..].split(':');
+                let field = parts.next().expect("No field name");
+                let value = parts.next().expect("No value");
+                let value = value
+                    .strip_prefix("{ ")
+                    .and_then(|s| s.strip_suffix(" }"))
+                    .unwrap_or(value);
+                for fields in self.table.lock().values() {
+                    if let Some(key_value) = fields.get(field) {
+                        if *key_value == RedisValue::Bytes(Bytes::from(value.to_owned())) {
+                            results.push(RedisValue::Array(vec![
+                                RedisValue::Bytes(Bytes::from("data")),
+                                fields.get("data").expect("No data field").clone(),
+                                RedisValue::Bytes(Bytes::from("version")),
+                                fields.get("version").expect("No version field").clone(),
+                            ]));
+                        }
+                    }
+                }
             }
+
             results[0] = u32::try_from(results.len() - 1).unwrap_or(u32::MAX).into();
             return Ok(RedisValue::Array(vec![
                 RedisValue::Array(results),

nativelink-store/src/redis_store.rs

@@ -1328,39 +1328,44 @@ impl SchedulerStore for RedisStore {
             get_index_name!(K::KEY_PREFIX, K::INDEX_NAME, K::MAYBE_SORT_KEY)
         );
 
-        let run_ft_aggregate =
-            |client: Arc<ClientWithPermit>, index_name: String, field: String| async move {
-                ft_aggregate(
-                    client.client.clone(),
-                    index_name,
-                    format!("@{}:{{ {} }}", K::INDEX_NAME, field),
-                    FtAggregateOptions {
-                        load: Some(Load::Some(vec![
-                            SearchField {
-                                identifier: DATA_FIELD_NAME.into(),
-                                property: None,
-                            },
-                            SearchField {
-                                identifier: VERSION_FIELD_NAME.into(),
-                                property: None,
-                            },
-                        ])),
-                        cursor: Some(WithCursor {
-                            count: Some(MAX_COUNT_PER_CURSOR),
-                            max_idle: Some(CURSOR_IDLE_MS),
+        let run_ft_aggregate = |client: Arc<ClientWithPermit>,
+                                index_name: String,
+                                sanitized_field: String| async move {
+            ft_aggregate(
+                client.client.clone(),
+                index_name,
+                if sanitized_field.is_empty() {
+                    "*".to_string()
+                } else {
+                    format!("@{}:{{ {} }}", K::INDEX_NAME, sanitized_field)
+                },
+                FtAggregateOptions {
+                    load: Some(Load::Some(vec![
+                        SearchField {
+                            identifier: DATA_FIELD_NAME.into(),
+                            property: None,
+                        },
+                        SearchField {
+                            identifier: VERSION_FIELD_NAME.into(),
+                            property: None,
+                        },
+                    ])),
+                    cursor: Some(WithCursor {
+                        count: Some(MAX_COUNT_PER_CURSOR),
+                        max_idle: Some(CURSOR_IDLE_MS),
+                    }),
+                    pipeline: vec![AggregateOperation::SortBy {
+                        properties: K::MAYBE_SORT_KEY.map_or_else(Vec::new, |v| {
+                            vec![(format!("@{v}").into(), SortOrder::Asc)]
                         }),
-                        pipeline: vec![AggregateOperation::SortBy {
-                            properties: K::MAYBE_SORT_KEY.map_or_else(Vec::new, |v| {
-                                vec![(format!("@{v}").into(), SortOrder::Asc)]
-                            }),
-                            max: None,
-                        }],
-                        ..Default::default()
-                    },
-                )
-                .await
-                .map(|stream| (stream, client))
-            };
+                        max: None,
+                    }],
+                    ..Default::default()
+                },
+            )
+            .await
+            .map(|stream| (stream, client))
+        };
 
         let client = Arc::new(self.get_client().await?);
         let (stream, client_guard) = if let Ok(result) =