Retry an action's input fetch on the worker after a transient store failure

nativelink-config/src/cas_server.rs

@@ -1299,6 +1299,15 @@ pub struct LocalWorkerConfig {
     /// `DeadlineExceeded` and may be retried by the scheduler, and a warning
     /// names the action every minute before that. Value in seconds.
     ///
+    /// A fetch that fails with a transient store error is fetched again
+    /// from this worker, with a pause doubling from 1 s to 30 s, for as
+    /// long as this leaves room for another attempt, or the action's own
+    /// timeout when that is shorter. The scheduler is not told of the
+    /// retries: it ends an executing action on its own at the action's
+    /// timeout plus `worker_timeout_s`, and at
+    /// `max_action_executing_timeout_s`, so keep this below those or the
+    /// action is requeued while the worker still retries.
+    ///
     /// Default: 10 minutes
     #[serde(
         default,

nativelink-util/src/metrics.rs

@@ -905,6 +905,11 @@ pub static WORKER_METRICS: LazyLock<WorkerMetrics> = LazyLock::new(|| {
             .with_description("Input hardlinks made again because the fast store evicted their source while the tree was staged, by kind: plain, executable, or exhausted when the retries ran out")
             .build(),
 
+        prepare_retries: meter
+            .u64_counter("worker.prepare.retries")
+            .with_description("Input fetches tried again on the same worker after a transient store failure: retried per attempt made again, then one of recovered, exhausted when the download deadline ran out, failed when a later attempt failed for good, or abandoned when a kill or the download timeout ended the fetch mid-retry")
+            .build(),
+
         dispatch_requeues: meter
             .u64_counter("scheduler.dispatch.requeues")
             .with_description("Dispatches sent back to the queue untried, by reason: declined, channel_full, unacknowledged")
@@ -954,6 +959,8 @@ pub struct WorkerMetrics {
     pub worker_disconnections: metrics::Counter<u64>,
     /// Input hardlinks made again after their source was evicted mid-staging, by kind.
     pub input_relinks: metrics::Counter<u64>,
+    /// Input fetches tried again on the worker after a transient store failure: `retried` per attempt made again, then one of `recovered`, `exhausted`, `failed` or `abandoned`.
+    pub prepare_retries: metrics::Counter<u64>,
     /// Dispatches requeued untried, by reason.
     pub dispatch_requeues: metrics::Counter<u64>,
     /// Keepalives received, cumulative.
@@ -1017,6 +1024,17 @@ pub fn record_input_relink(kind: &'static str) {
         .add(1, &[KeyValue::new("kind", kind)]);
 }
 
+/// Records an input fetch the worker tried again after a transient store
+/// failure: `retried` per attempt made again, then one terminal outcome,
+/// `recovered`, `exhausted` (the download deadline ran out), `failed` (a
+/// later attempt failed for good) or `abandoned` (a kill or the download
+/// timeout ended it mid-retry).
+pub fn record_prepare_retry(outcome: &'static str) {
+    WORKER_METRICS
+        .prepare_retries
+        .add(1, &[KeyValue::new("outcome", outcome)]);
+}
+
 /// Records a keepalive from a worker.
 pub fn record_worker_keepalive() {
     WORKER_METRICS.worker_keepalives.add(1, &[]);

nativelink-worker/Cargo.toml

@@ -78,6 +78,9 @@ serial_test = { version = "4.0.0", features = [
   "async",
 ], default-features = false }
 tempfile = { version = "3.15.0", default-features = false }
+tokio = { version = "1.52.2", features = [
+  "test-util",
+], default-features = false }
 tonic-prost = { version = "0.14.6", default-features = false }
 tracing-test = { version = "0.2.5", default-features = false, features = [
   "no-env-filter",