async fn wait_for_task(
        &self,
        job_id: Self::JobId,
        started_at: Instant,
        task: JoinHandle<anyhow::Result<Self::JobArtifacts>>,
        stop_receiver: &mut watch::Receiver<bool>,
    ) -> anyhow::Result<()> {
        let attempts = self.get_job_attempts(&job_id).await?;
        let max_attempts = self.max_attempts();
        if attempts == max_attempts {
            METRICS.max_attempts_reached[&(Self::SERVICE_NAME, format!("{job_id:?}"))].inc();
            tracing::error!(
                "Max attempts ({max_attempts}) reached for {} job {:?}",
                Self::SERVICE_NAME,
                job_id,
            );
        }

        let result = loop {
            tracing::trace!(
                "Polling {} task with id {:?}. Is finished: {}",
                Self::SERVICE_NAME,
                job_id,
                task.is_finished()
            );
            if task.is_finished() {
                break task.await;
            }
            if tokio::time::timeout(
                Duration::from_millis(Self::POLLING_INTERVAL_MS),
                stop_receiver.changed(),
            )
            .await
            .is_ok()
            {
                // Stop request received, return early.
                // Exit will be processed/reported by the main loop.
                return Ok(());
            }
        };
        let error_message = match result {
            Ok(Ok(data)) => {
                tracing::debug!(
                    "{} Job {:?} finished successfully",
                    Self::SERVICE_NAME,
                    job_id
                );
                METRICS.attempts[&Self::SERVICE_NAME].observe(attempts as usize);
                return self
                    .save_result(job_id, started_at, data)
                    .await
                    .context("save_result()");
            }
            Ok(Err(error)) => error.to_string(),
            Err(error) => try_extract_panic_message(error),
        };
        tracing::error!(
            "Error occurred while processing {} job {:?}: {:?}",
            Self::SERVICE_NAME,
            job_id,
            error_message
        );

        self.save_failure(job_id, started_at, error_message).await;
        Ok(())
    }
Polls task handle, saving its outcome.
core/lib/queued_job_processor/src/lib.rs