feat(job): batch await and completion result helpers

[HonestMajority]

Summary

Adds batch-level primitives to the job crate so consumers don't need to roll their own when coordinating fan-out workflows (spawn N child jobs, await all, inspect results).

Jobs::await_completions(ids, timeout) — batch counterpart of await_completion. Awaits multiple jobs concurrently with an optional timeout covering the entire batch.

JobCompletionResults trait — extension trait on Vec<JobCompletionResult> and [JobCompletionResult] providing failed_count() and all_succeeded(). Replaces identical free-function copies scattered across lana-bank consumers.

JobCompletionResult::is_completed() / is_errored() — convenience predicates used by the trait above.

Context

lana-bank's EOD pipeline has 4–5 coordinator jobs that all fan out child jobs and then await them. Each one carried its own copy of await_job_completions and failed_count. This PR moves the reusable parts into the job crate. Shutdown-aware tokio::select! remains at the consumer level since that's an application concern.

🤖 Generated with Claude Code

---

Note

Medium Risk
Introduces new public APIs and renames existing completion/result types (JobCompletionResult/JobResult -> JobOutcome/JobReturnValue), which is a breaking change for consumers and could affect completion/result handling semantics if migrated incorrectly.

Overview
Adds a batch API Jobs::await_completions(ids, timeout) that awaits multiple jobs concurrently under a single optional timeout and returns a Vec<JobOutcome>.

Refactors completion/result types into a new outcome module: JobResult/JobCompletionResult become JobReturnValue/JobOutcome, job events rename ResultUpdated -> ReturnValueUpdated, and await_completion now returns JobOutcome. Adds a JobOutcomes extension trait (for Vec<JobOutcome> and slices) with failed_count()/all_succeeded() plus convenience predicates on JobOutcome.

Updates exports and tests to use the new types and validates batch waiting, empty batches, timeouts, and the new outcome helpers.

^{Written by Cursor Bugbot for commit a4753a8. This will update automatically on new commits. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Batch timeout error misleadingly blames first job ID
  • I added a batch-specific timeout error (JobError::BatchTimedOut(Vec<JobId>)) and updated await_completions to return it so timeout diagnostics no longer misattribute the first job ID.

Or push these changes by commenting:

@cursor push 867f920fa7

Preview (867f920fa7)

diff --git a/src/error.rs b/src/error.rs
--- a/src/error.rs
+++ b/src/error.rs
@@ -51,6 +51,10 @@
         "JobError - TimedOut: job {0} did not reach terminal state within the specified timeout"
     )]
     TimedOut(JobId),
+    #[error(
+        "JobError - BatchTimedOut: jobs {0:?} did not all reach terminal state within the specified timeout"
+    )]
+    BatchTimedOut(Vec<JobId>),
 }
 
 impl From<Box<dyn std::error::Error>> for JobError {

diff --git a/src/lib.rs b/src/lib.rs
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -551,16 +551,16 @@
     /// have finished.
     ///
     /// When `timeout` is `Some(duration)`, the call returns
-    /// [`JobError::TimedOut`] if the batch has not fully resolved within the
-    /// specified duration. Pass `None` to wait indefinitely.
+    /// [`JobError::BatchTimedOut`] if the batch has not fully resolved within
+    /// the specified duration. Pass `None` to wait indefinitely.
     ///
     /// An empty `ids` slice returns an empty `Vec` immediately.
     ///
     /// # Errors
     ///
     /// Returns [`JobError::Find`] if any job in the batch does not exist.
-    /// Returns [`JobError::TimedOut`] if the timeout elapses before every job
-    /// reaches a terminal state.
+    /// Returns [`JobError::BatchTimedOut`] if the timeout elapses before every
+    /// job reaches a terminal state.
     /// Returns [`JobError::AwaitCompletionShutdown`] if the notification channel
     /// is dropped (e.g., during shutdown) before all jobs have resolved.
     #[instrument(name = "job.await_completions", skip(self))]
@@ -577,12 +577,9 @@
             .map(|id| self.await_completion(*id, None))
             .collect();
         let results = match timeout {
-            Some(duration) => {
-                let first_id = ids[0];
-                tokio::time::timeout(duration, futures::future::join_all(futs))
-                    .await
-                    .map_err(|_| JobError::TimedOut(first_id))?
-            }
+            Some(duration) => tokio::time::timeout(duration, futures::future::join_all(futs))
+                .await
+                .map_err(|_| JobError::BatchTimedOut(ids.to_vec()))?,
             None => futures::future::join_all(futs).await,
         };
         results.into_iter().collect()

diff --git a/tests/job.rs b/tests/job.rs
--- a/tests/job.rs
+++ b/tests/job.rs
@@ -1578,22 +1578,33 @@
     });
     jobs.start_poll().await?;
 
-    // Schedule a job far in the future so it never completes
-    let job_id = JobId::new();
+    // Spawn one quick job that should complete within timeout
+    let quick_id = JobId::new();
+    spawner
+        .spawn(quick_id, TestJobConfig { delay_ms: 10 })
+        .await?;
+
+    // Schedule another job far in the future so the batch never fully completes
+    let stuck_id = JobId::new();
     let schedule_at = chrono::Utc::now() + chrono::Duration::hours(24);
     spawner
-        .spawn_at(job_id, TestJobConfig { delay_ms: 50 }, schedule_at)
+        .spawn_at(stuck_id, TestJobConfig { delay_ms: 50 }, schedule_at)
         .await?;
 
+    let ids = vec![quick_id, stuck_id];
     let result = jobs
-        .await_completions(&[job_id], Some(Duration::from_millis(200)))
+        .await_completions(&ids, Some(Duration::from_secs(1)))
         .await;
 
-    assert!(
-        matches!(result, Err(JobError::TimedOut(_))),
-        "Expected TimedOut error, got: {:?}",
-        result,
-    );
+    match result {
+        Err(JobError::BatchTimedOut(timed_out_ids)) => {
+            assert_eq!(
+                timed_out_ids, ids,
+                "Batch timeout should report all batch IDs"
+            );
+        }
+        other => panic!("Expected BatchTimedOut error, got: {:?}", other),
+    }
 
     Ok(())
 }

_{This Bugbot Autofix run was free. To enable autofix for future PRs, go to the Cursor dashboard.}

[cursor]

Cursor Bugbot has reviewed your changes and found 2 potential issues.

^{Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}