fix(graph): re-probe the cache after taking an action lock

crates/once-cli/src/commands/graph/analysis/actions.rs

@@ -1212,10 +1212,11 @@ async fn resolve_cacheable_declared_action(
         context.producers.executors(context.resources)?,
         once_core::Executor::resolve(action)?,
         action.resource_request().clone(),
-        |contended, lease| async move {
-            // Another process held the action, so it may have produced the
-            // result while this one waited.
-            if contended && reusable {
+        |lease| async move {
+            // Another task or process may have produced the result since the
+            // probe above, and may have released the lock before this one
+            // tried it while this one waited for capacity.
+            if reusable {
                 if let Some(outcome) = probe_cached_action(context, action_digest).await? {
                     context.claim_worker().await;
                     return Ok(outcome);

crates/once-cli/src/commands/graph/analysis/actions/recovery.rs

@@ -632,7 +632,7 @@ impl Repair<'_> {
             self.producers.executors(self.resources)?,
             once_core::Executor::resolve(&producer.action)?,
             producer.action.resource_request().clone(),
-            |_contended, lease| async move {
+            |lease| async move {
                 let attempt = match reporting {
                     Some(reporting) => Some(AttemptReport::start(reporting, producer).await),
                     None => None,
@@ -803,9 +803,7 @@ fn single_output(path: &str, digest: Digest) -> ActionResult {
 /// Waiting for admission while holding the locks would stall every other
 /// process waiting for the same action behind this one's queue position;
 /// waiting for a lock while holding a permit or another lock would pin
-/// capacity behind work that may already be done. `work` learns whether
-/// another holder was seen, in which case the action may have finished in the
-/// meantime.
+/// capacity behind work that may already be done.
 pub(super) async fn with_action_lock_and_permit<'a, T, F, Fut>(
     workspace: &Path,
     digest: Digest,
@@ -816,22 +814,20 @@ pub(super) async fn with_action_lock_and_permit<'a, T, F, Fut>(
     work: F,
 ) -> Result<T>
 where
-    F: FnOnce(bool, once_core::ExecutionPermit) -> Fut,
+    F: FnOnce(once_core::ExecutionPermit) -> Fut,
     Fut: Future<Output = Result<T>>,
 {
     let locks = Arc::new(cacheable_action_locks(workspace, digest, outputs)?);
     let mut work = Some(work);
-    let mut contended = false;
     loop {
         let permit = resources.acquire(executor, &request).await;
         let busy = try_lock_all(&locks).with_context(|| format!("locking action {digest}"))?;
         let Some(busy) = busy else {
             let permit = permit.with_lifetime_owner(Arc::clone(&locks));
             let work = work.take().context("action work already ran")?;
-            return work(contended, permit).await;
+            return work(permit).await;
         };
         drop(permit);
-        contended = true;
         wait_for_action_lock(&locks[busy], digest).await?;
     }
 }

crates/once-cli/src/commands/graph/analysis/actions/recovery/tests.rs

@@ -404,6 +404,47 @@ async fn a_producer_that_writes_different_bytes_fails_closed() {
     );
 }
 
+#[cfg(unix)]
+#[tokio::test]
+async fn an_action_queued_behind_its_own_run_reuses_that_result() {
+    let mut workspace = Workspace::new();
+    workspace.resources = Arc::new(ResourcePool::new(once_core::ResourceLimits::new(
+        1,
+        1 << 30,
+    )));
+    let command = format!("printf run >> runs.log; sleep 0.5; printf first > {PRODUCED}");
+    let produce = || AnalysisResult {
+        actions: vec![action(serde_json::json!({
+            "identifier": "produce",
+            "argv": ["/bin/sh", "-c", command],
+            "outputs": [PRODUCED],
+            "create_dirs": [".once/out"],
+            "cacheable": true,
+        }))],
+        provider: serde_json::json!({}),
+        declared_outputs: Vec::new(),
+        observations: AnalysisObservations::default(),
+    };
+    // The second build probes the cache while the first holds the only
+    // execution slot, then waits for that slot, which the first releases
+    // together with the action lock once it has stored its result.
+    let queued = async {
+        while workspace.runs() == 0 {
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+        workspace.build(produce()).await
+    };
+
+    let (first, second) = tokio::join!(workspace.build(produce()), queued);
+
+    first.unwrap();
+    assert_eq!(workspace.runs(), 1, "the queued build ran the action again");
+    assert_eq!(
+        second.unwrap().per_action_outcomes[0].cache_state,
+        EvidenceCacheState::Hit
+    );
+}
+
 #[cfg(unix)]
 #[tokio::test]
 async fn a_cancelled_waiter_leaves_the_repair_running_and_it_is_not_run_twice() {