| 1599 | 1599 | | |
| 1600 | 1600 | | /// Starts queued jobs, oldest first, while their workspace has room. |
| 1601 | 1601 | | pub async fn start_queued(&self) -> Result<()> { |
| 1602 | + | self.start_queued_except(&std::collections::HashSet::new()).await |
| 1603 | + | } |
| 1604 | + | |
| 1605 | + | /// [`Self::start_queued`], leaving `held` for a later sweep. |
| 1606 | + | async fn start_queued_except(&self, held: &std::collections::HashSet<String>) -> Result<()> { |
| 1602 | 1607 | | let queued = self |
| 1603 | 1608 | | .db |
| 1604 | 1609 | | // Self-hosted jobs are taken by their runners (runners.rs). |
| ⋯ |
| 1619 | 1624 | | .map_or(0, |count| count.n) |
| 1620 | 1625 | | }; |
| 1621 | 1626 | | for job in queued { |
| 1627 | + | if held.contains(&job.id) { |
| 1628 | + | continue; |
| 1629 | + | } |
| 1622 | 1630 | | let in_workspace = match running.get(&job.namespace) { |
| 1623 | 1631 | | Some(n) => *n, |
| 1624 | 1632 | | None => { |
| ⋯ |
| 2867 | 2875 | | } |
| 2868 | 2876 | | // Jobs whose sandbox went quiet or ran past their time. |
| 2869 | 2877 | | let running = self.db.prepare("SELECT * FROM jobs WHERE status = 'in_progress'").all().await?.results::<JobRow>()?; |
| 2878 | + | let mut lost = std::collections::HashSet::new(); |
| 2870 | 2879 | | for job in running { |
| 2871 | 2880 | | // Times in g1t's format compare as text. |
| 2872 | 2881 | | let before = |ms: u64| rfc3339(now_ms.saturating_sub(ms)); |
| ⋯ |
| 2885 | 2894 | | let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await; |
| 2886 | 2895 | | } |
| 2887 | 2896 | | self.finish_job(&job.id, "failure", Some(&reason), None).await?; |
| 2897 | + | } else if job.runner_id.is_none() && crate::never_reported(job.started_at.as_deref(), job.seen_at.as_deref(), now_ms) { |
| 2898 | + | // Its sandbox never ran it: nothing it did needs undoing. |
| 2899 | + | // It starts again at the next sweep, not this one: the same |
| 2900 | + | // sandbox takes it, and the stop of the lost one finishes |
| 2901 | + | // in that sandbox's alarm, after this. |
| 2902 | + | if self.retry_lost_start(&job, now_ms).await? { |
| 2903 | + | lost.insert(job.id.clone()); |
| 2904 | + | } |
| 2888 | 2905 | | } else if silent { |
| 2889 | 2906 | | let reason = match &job.runner_name { |
| 2890 | 2907 | | Some(name) => format!("The self-hosted runner {name} stopped answering."), |
| ⋯ |
| 2942 | 2959 | | { |
| 2943 | 2960 | | worker::console_error!("actions: the logs' sweep failed: {error}"); |
| 2944 | 2961 | | } |
| 2945 | | − | self.start_queued().await |
| 2962 | + | self.start_queued_except(&lost).await |
| 2963 | + | } |
| 2964 | + | |
| 2965 | + | /// A job whose sandbox never reported (`never_reported`): back in the |
| 2966 | + | /// queue, its token void so that sandbox can no longer report for it, |
| 2967 | + | /// and the sandbox stopped, which gives back what billing held for it. |
| 2968 | + | /// It waits as a job with no machine does, and fails once it has waited |
| 2969 | + | /// WAIT_FOR_MACHINE_MS. Returns whether it went back in the queue. |
| 2970 | + | async fn retry_lost_start(&self, job: &JobRow, now_ms: u64) -> Result<bool> { |
| 2971 | + | let since = job.waiting_since.clone().or_else(|| job.started_at.clone()).unwrap_or_else(|| rfc3339(now_ms)); |
| 2972 | + | if crate::waited_for_a_machine(Some(&since), now_ms) { |
| 2973 | + | let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await; |
| 2974 | + | let minutes = crate::WAIT_FOR_MACHINE_MS / 60_000; |
| 2975 | + | let reason = format!("No machine ran the job in {minutes} minutes: each one it was given stopped answering before it started."); |
| 2976 | + | self.finish_job(&job.id, "failure", Some(&reason), None).await?; |
| 2977 | + | return Ok(false); |
| 2978 | + | } |
| 2979 | + | // Not if it reported after all, since it was read. |
| 2980 | + | let requeued = self |
| 2981 | + | .db |
| 2982 | + | .prepare( |
| 2983 | + | "UPDATE jobs SET status = 'queued', token_hash = NULL, first_token_hash = NULL, started_at = NULL, seen_at = NULL, waiting_since = ?, reason = ? |
| 2984 | + | WHERE id = ? AND status = 'in_progress' AND seen_at = started_at RETURNING id", |
| 2985 | + | ) |
| 2986 | + | .bind(&[ |
| 2987 | + | since.as_str().into(), |
| 2988 | + | "Its machine stopped answering before it started; it waits for another.".into(), |
| 2989 | + | job.id.as_str().into(), |
| 2990 | + | ])? |
| 2991 | + | .first::<Value>(None) |
| 2992 | + | .await?; |
| 2993 | + | if requeued.is_none() { |
| 2994 | + | return Ok(false); |
| 2995 | + | } |
| 2996 | + | worker::console_warn!("actions: job {} never reported; it goes back in the queue", job.id); |
| 2997 | + | self.revoke_job_tokens(&job.id).await; |
| 2998 | + | let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await; |
| 2999 | + | Ok(true) |
| 2946 | 3000 | | } |
| 2947 | 3001 | | } |
| 2948 | 3002 | | |