Existence cache should invalidate if get_part from underlying store fails.

dkostyrev · dkostyrev · commit 4b9d0cd827a6 · 2026-03-17T20:29:49.000Z
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/nativelink-scheduler/src/api_worker_scheduler.rs b/nativelink-scheduler/src/api_worker_scheduler.rs
@@ -15,7 +15,7 @@
 use core::ops::{Deref, DerefMut};
 use core::sync::atomic::{AtomicU64, Ordering};
 use core::time::Duration;
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::{Instant, UNIX_EPOCH};
 
@@ -397,7 +397,7 @@ impl ApiWorkerSchedulerImpl {
         for (idx, platform_properties) in actions.iter().enumerate() {
             let candidates = self
                 .capability_index
-                .find_matching_workers(platform_properties);
+                .find_matching_workers(platform_properties, full_worker_logging);
             if candidates.is_empty() {
                 continue;
             }
diff --git a/nativelink-scheduler/src/awaited_action_db/mod.rs b/nativelink-scheduler/src/awaited_action_db/mod.rs
@@ -16,7 +16,6 @@ use core::cmp;
 use core::ops::Bound;
 use core::time::Duration;
 use std::collections::HashMap;
-use std::iter::Map;
 use std::sync::Arc;
 
 pub use awaited_action::{AwaitedAction, AwaitedActionSortKey};
diff --git a/nativelink-scheduler/src/memory_awaited_action_db.rs b/nativelink-scheduler/src/memory_awaited_action_db.rs
@@ -16,10 +16,9 @@ use core::ops::{Bound, RangeBounds};
 use core::time::Duration;
 use std::collections::hash_map::Entry;
 use std::collections::{BTreeMap, BTreeSet, HashMap};
-use std::iter::Map;
 use std::sync::Arc;
 
-use async_lock::{Mutex, RwLock};
+use async_lock::RwLock;
 use futures::{FutureExt, Stream};
 use nativelink_config::stores::EvictionPolicy;
 use nativelink_error::{Code, Error, ResultExt, error_if, make_err};
diff --git a/nativelink-scheduler/src/simple_scheduler.rs b/nativelink-scheduler/src/simple_scheduler.rs
@@ -43,13 +43,12 @@ use tokio::time::Duration;
 use tracing::{debug, error, info, info_span, warn};
 
 use crate::api_worker_scheduler::ApiWorkerScheduler;
-use crate::awaited_action_db::{AwaitedAction, AwaitedActionDb, CLIENT_KEEPALIVE_DURATION};
+use crate::awaited_action_db::{AwaitedActionDb, CLIENT_KEEPALIVE_DURATION};
 use crate::platform_property_manager::PlatformPropertyManager;
 use crate::simple_scheduler_state_manager::{SchedulerStateManager, SimpleSchedulerStateManager};
 use crate::worker::{ActionInfoWithProps, ActionsState, Worker, WorkerState, WorkerTimestamp};
 use crate::worker_registry::WorkerRegistry;
 use crate::worker_scheduler::WorkerScheduler;
-use nativelink_util::metrics::StoreType::Metrics;
 use serde::Serialize;
 use nativelink_util::metrics::EXECUTION_METRICS;
 
diff --git a/nativelink-scheduler/src/simple_scheduler_state_manager.rs b/nativelink-scheduler/src/simple_scheduler_state_manager.rs
@@ -1092,7 +1092,7 @@ where
                         warn!(state = ?awaited_action.state(), "Action already assigned");
                         return Err(make_err!(Code::Aborted, "Action already assigned"));
                     }
-                    stage.clone()
+                    (stage.clone(), false)
                 }
                 UpdateOperationType::UpdateWithError(err) => {
                     // Don't count a backpressure failure as an attempt for an action.
diff --git a/nativelink-scheduler/src/store_awaited_action_db.rs b/nativelink-scheduler/src/store_awaited_action_db.rs
@@ -37,7 +37,7 @@ use nativelink_util::store_trait::{
 };
 use nativelink_util::task::JoinHandleDropGuard;
 use tokio::sync::Notify;
-use tracing::{error, info, warn};
+use tracing::{error, warn};
 
 use crate::awaited_action_db::{
     AwaitedAction, AwaitedActionDb, AwaitedActionSubscriber, CLIENT_KEEPALIVE_DURATION,
diff --git a/nativelink-scheduler/src/worker.rs b/nativelink-scheduler/src/worker.rs
@@ -27,7 +27,6 @@ use nativelink_util::metrics_utils::{AsyncCounterWrapper, CounterWithTime, FuncC
 use nativelink_util::platform_properties::{PlatformProperties, PlatformPropertyValue};
 use serde::Serialize;
 use tokio::sync::mpsc::UnboundedSender;
-use crate::awaited_action_db::AwaitedAction;
 
 pub type WorkerTimestamp = u64;
 
@@ -61,7 +60,7 @@ pub struct PendingActionInfoData {
     pub action_info: ActionInfoWithProps,
 }
 
-#[derive(Serialize)]
+#[derive(Serialize, Debug)]
 pub struct WorkerState {
     pub id: WorkerId,
     pub platform_properties: PlatformProperties,
@@ -71,7 +70,7 @@ pub struct WorkerState {
     pub is_draining: bool,
 }
 
-#[derive(Serialize)]
+#[derive(Serialize, Debug)]
 pub struct ActionsState {
     pub executing: usize,
     pub queued: usize,
diff --git a/nativelink-store/src/existence_cache_store.rs b/nativelink-store/src/existence_cache_store.rs
@@ -293,6 +293,8 @@ impl<I: InstantWrapper> StoreDriver for ExistenceCacheStore<I> {
                 .existence_cache
                 .insert(digest, ExistenceItem(digest.size_bytes()))
                 .await;
+        } else {
+            let _ = self.existence_cache.remove(&digest).await;
         }
         result
     }
diff --git a/nativelink-store/src/fast_slow_store.rs b/nativelink-store/src/fast_slow_store.rs
@@ -18,7 +18,7 @@ use core::ops::Range;
 use core::pin::Pin;
 use std::collections::HashMap;
 use std::ffi::OsString;
-use std::sync::{Arc, LazyLock, Weak};
+use std::sync::{Arc, Weak};
 
 use async_trait::async_trait;
 use futures::{FutureExt, join};
@@ -34,7 +34,6 @@ use nativelink_util::store_trait::{
     RemoveItemCallback, Store, StoreDriver, StoreKey, StoreLike, StoreOptimizations,
     UploadSizeInfo, slow_update_store_with_file,
 };
-use opentelemetry::{InstrumentationScope, global, metrics};
 use parking_lot::Mutex;
 use tokio::sync::OnceCell;
 use tracing::{debug, trace, warn};
diff --git a/nativelink-store/src/metrics_store.rs b/nativelink-store/src/metrics_store.rs
@@ -1,10 +1,7 @@
-use crate::callback_utils::RemoveItemCallbackHolder;
 use crate::filesystem_store::FilesystemStore;
-use crate::memory_store::MemoryStore;
 use async_trait::async_trait;
 use nativelink_error::Error;
 use nativelink_metric::MetricsComponent;
-use nativelink_proto::build_event_stream::File;
 use nativelink_util::buf_channel::{DropCloserReadHalf, DropCloserWriteHalf};
 use nativelink_util::health_utils::{HealthStatus, HealthStatusIndicator};
 use nativelink_util::metrics::{StoreMetricAttrs, StoreType, STORE_METRICS};
diff --git a/nativelink-store/src/redis_store.rs b/nativelink-store/src/redis_store.rs
@@ -21,35 +21,15 @@ use core::str::FromStr;
 use core::time::Duration;
 use std::borrow::Cow;
 use std::collections::HashSet;
-use std::collections::HashMap;
-use std::ops::Index;
 use std::sync::{Arc, Weak};
 use std::time::Instant;
 
-use crate::cas_utils::is_zero_digest;
-use crate::redis_utils::ft_aggregate;
 use async_trait::async_trait;
 use bytes::Bytes;
 use const_format::formatcp;
 use futures::stream::FuturesUnordered;
 use futures::{Stream, StreamExt, TryFutureExt, TryStreamExt, future};
 use itertools::izip;
-use fred::clients::SubscriberClient;
-use fred::interfaces::{ClientLike, KeysInterface, PubsubInterface};
-use fred::prelude::{Client, EventInterface, HashesInterface, RediSearchInterface};
-use fred::types::config::{
-    Config as RedisConfig, ConnectionConfig, PerformanceConfig, ReconnectPolicy, UnresponsiveConfig,
-};
-use fred::types::redisearch::{
-    AggregateOperation, FtAggregateOptions, FtCreateOptions, IndexKind, Load, ReducerFunc,
-    SearchField, SearchReducer, SearchSchema, SearchSchemaKind, WithCursor,
-};
-use fred::types::scan::Scanner;
-use fred::types::scripts::Script;
-use fred::types::{Builder, Key as RedisKey, Map as RedisMap, SortOrder, Value as RedisValue};
-use futures::stream::FuturesUnordered;
-use futures::{FutureExt, Stream, StreamExt, TryStreamExt, future};
-use itertools::{Itertools, izip};
 use nativelink_config::stores::{RedisMode, RedisSpec};
 use nativelink_error::{Code, Error, ResultExt, make_err, make_input_err};
 use nativelink_metric::MetricsComponent;
@@ -1652,85 +1632,7 @@ where
     where
         K: SchedulerIndexProvider + Send,
     {
-        let index_values: Vec<_> = index.iter().map(|k| k.index_value()).collect();
-        let sanitized_fields: Vec<String> = index_values
-            .iter()
-            .map(|v| try_sanitize(v.as_ref()))
-            .map(|s| s.clone().unwrap_or_default().to_string())
-            .collect();
-        let index_name = format!(
-            "{}",
-            get_index_name!(K::KEY_PREFIX, K::INDEX_NAME, K::MAYBE_SORT_KEY)
-        );
-
-        let client = self.get_client().await?;
-
-        let query = if sanitized_fields.is_empty() {
-            "*".to_string()
-        } else {
-            format!("@{}:{{ {} }}", K::INDEX_NAME, sanitized_fields.join("|"))
-        };
-
-        let result: RedisValue = client
-            .client
-            .clone()
-            .ft_aggregate(
-                index_name,
-                query,
-                FtAggregateOptions {
-                    pipeline: vec![AggregateOperation::GroupBy {
-                        fields: vec![format!("@{}", K::INDEX_NAME).into()],
-                        reducers: vec![SearchReducer {
-                            func: ReducerFunc::Count,
-                            args: vec![],
-                            name: Some("cnt".into()),
-                        }],
-                    }],
-                    ..Default::default()
-                },
-            )
-            .await?;
-
-        if !result.is_array() {
-            return Err(Error::new(Code::Internal, "Expected array".to_string()));
-        }
-
-        let lookup: HashMap<&Cow<str>, usize> = index_values
-            .iter()
-            .enumerate()
-            .map(|(i, k)| (k, i))
-            .collect();
-        let mut counts = vec![0; index.len()];
-
-        let result = result.into_array();
-        if result.len() < 2 {
-            return Ok(counts);
-        }
-
-        result
-            .into_iter()
-            .skip(1)
-            .map(|map| map.into_map().unwrap())
-            .for_each(|map| {
-                let key = map
-                    .get(&RedisKey::from_static_str(K::INDEX_NAME))
-                    .err_tip(|| "Missing index field in RedisStore::count_by_index")
-                    .unwrap();
-
-                let cnt_value = map
-                    .get(&RedisKey::from_static_str("cnt"))
-                    .err_tip(|| "Missing 'cnt' field in RedisStore::count_by_index")
-                    .unwrap();
-
-                let count = cnt_value
-                    .as_usize()
-                    .err_tip(|| "Count value is not an integer in RedisStore::count_by_index")
-                    .unwrap();
-
-                let val = lookup.get(&key.as_str().unwrap()).unwrap_or(&0);
-                counts[val.clone()] = count;
-            });
-        Ok(counts)
+        Err(make_err!(Code::Unimplemented, "Not implemented"))
     }
 
     async fn search_by_index_prefix<K>(
diff --git a/nativelink-util/BUILD.bazel b/nativelink-util/BUILD.bazel
@@ -126,6 +126,7 @@ rust_test_suite(
         "//nativelink-error",
         "//nativelink-metric",
         "//nativelink-proto",
+        "//nativelink-worker",
         "@crates//:axum",
         "@crates//:bytes",
         "@crates//:futures",
diff --git a/nativelink-util/src/store_trait.rs b/nativelink-util/src/store_trait.rs
@@ -41,7 +41,6 @@ use crate::common::DigestInfo;
 use crate::digest_hasher::{DigestHasher, DigestHasherFunc, default_digest_hasher_func};
 use crate::fs;
 use crate::health_utils::{HealthRegistryBuilder, HealthStatus, HealthStatusIndicator};
-use crate::metrics::{StoreMetricAttrs, StoreType};
 
 static DEFAULT_DIGEST_SIZE_HEALTH_CHECK: OnceLock<usize> = OnceLock::new();
 /// Default digest size for health check data. Any change in this value
diff --git a/nativelink-util/src/telemetry.rs b/nativelink-util/src/telemetry.rs
@@ -35,8 +35,8 @@ use opentelemetry_sdk::trace::SdkTracerProvider;
 use opentelemetry_semantic_conventions::attribute::ENDUSER_ID;
 use prost::Message;
 use std::env;
-use std::sync::{OnceLock};
-use tracing::{debug, info};
+use std::sync::OnceLock;
+use tracing::debug;
 use tracing::metadata::LevelFilter;
 use tracing_opentelemetry::{MetricsLayer, layer};
 use tracing_subscriber::filter::Directive;
diff --git a/nativelink-util/tests/metrics_test.rs b/nativelink-util/tests/metrics_test.rs
@@ -15,7 +15,7 @@
 use nativelink_util::action_messages::{ActionResult, ActionStage};
 use nativelink_util::metrics::{
     CACHE_METRICS, CacheMetricAttrs, EXECUTION_METRICS, ExecutionMetricAttrs, ExecutionStage,
-    WORKER_METRICS, make_execution_attributes,
+    LOCAL_WORKER_METRICS, make_execution_attributes,
 };
 use opentelemetry::KeyValue;
 
@@ -85,18 +85,12 @@ fn test_execution_metric_attrs() {
 
 #[test]
 fn test_make_execution_attributes() {
-    let attrs = make_execution_attributes("test_instance", Some("worker_456"), Some(100));
+    let attrs = make_execution_attributes("test_instance", Some(100));
 
-    assert_eq!(attrs.len(), 3);
+    assert_eq!(attrs.len(), 2);
     assert!(attrs.iter().any(
         |kv| kv.key.as_str() == "execution_instance" && kv.value.to_string() == "test_instance"
     ));
-    assert!(
-        attrs
-            .iter()
-            .any(|kv| kv.key.as_str() == "execution_worker_id"
-                && kv.value.to_string() == "worker_456")
-    );
     assert!(
         attrs
             .iter()
@@ -110,7 +104,7 @@ fn test_metrics_lazy_initialization() {
     // Verify that the lazy static initialization works
     let _cache_metrics = &*CACHE_METRICS;
     let _execution_metrics = &*EXECUTION_METRICS;
-    let _worker_metrics = &*WORKER_METRICS;
+    let _worker_metrics = &*LOCAL_WORKER_METRICS;
 
     // If we got here without panicking, the metrics were initialized successfully
 }
diff --git a/nativelink-worker/src/local_worker.rs b/nativelink-worker/src/local_worker.rs
diff --git a/nativelink-worker/src/running_actions_manager.rs b/nativelink-worker/src/running_actions_manager.rs
diff --git a/nativelink-worker/tests/utils/local_worker_test_utils.rs b/nativelink-worker/tests/utils/local_worker_test_utils.rs
diff --git a/src/bin/cas_speed_check.rs b/src/bin/cas_speed_check.rs

Original file line number	Diff line number	Diff line change
`@@ -1092,7 +1092,7 @@ where`
`1092`	`1092`	`warn!(state = ?awaited_action.state(), "Action already assigned");`
`1093`	`1093`	`return Err(make_err!(Code::Aborted, "Action already assigned"));`
`1094`	`1094`	`}`
`1095`		`- stage.clone()`
	`1095`	`+ (stage.clone(), false)`
`1096`	`1096`	`}`
`1097`	`1097`	`UpdateOperationType::UpdateWithError(err) => {`
`1098`	`1098`	`// Don't count a backpressure failure as an attempt for an action.`
Original file line number	Diff line number	Diff line change
`@@ -293,6 +293,8 @@ impl<I: InstantWrapper> StoreDriver for ExistenceCacheStore<I> {`
`293`	`293`	`.existence_cache`
`294`	`294`	`.insert(digest, ExistenceItem(digest.size_bytes()))`
`295`	`295`	`.await;`
	`296`	`+ } else {`
	`297`	`+ let _ = self.existence_cache.remove(&digest).await;`
`296`	`298`	`}`
`297`	`299`	`result`
`298`	`300`	`}`