async fn run(
        self,
        mut stop_receiver: watch::Receiver<bool>,
        mut iterations_left: Option<usize>,
    ) -> anyhow::Result<()>
    where
        Self: Sized,
    {
        let mut backoff: u64 = Self::POLLING_INTERVAL_MS;
        while iterations_left.is_none_or(|i| i > 0) {
            if *stop_receiver.borrow() {
                tracing::warn!(
                    "Stop request received, shutting down {} component while waiting for a new job",
                    Self::SERVICE_NAME
                );
                return Ok(());
            }
            if let Some((job_id, job)) =
                Self::get_next_job(&self).await.context("get_next_job()")?
            {
                let started_at = Instant::now();
                backoff = Self::POLLING_INTERVAL_MS;
                iterations_left = iterations_left.map(|i| i - 1);

                tracing::debug!(
                    "Spawning thread processing {:?} job with id {:?}",
                    Self::SERVICE_NAME,
                    job_id
                );
                let task = self.process_job(&job_id, job, started_at).await;

                self.wait_for_task(job_id, started_at, task, &mut stop_receiver)
                    .await
                    .context("wait_for_task")?;
            } else if iterations_left.is_some() {
                tracing::info!("No more jobs to process. Server can stop now.");
                return Ok(());
            } else {
                tracing::trace!("Backing off for {} ms", backoff);
                // Error here corresponds to a timeout w/o `stop_receiver` changed; we're OK with this.
                tokio::time::timeout(Duration::from_millis(backoff), stop_receiver.changed())
                    .await
                    .ok();
                backoff = (backoff * Self::BACKOFF_MULTIPLIER).min(Self::MAX_BACKOFF_MS);
            }
        }
        tracing::info!("Requested number of jobs is processed. Server can stop now.");
        Ok(())
    }
`iterations_left`:
To run indefinitely, pass `None`,
To process one job, pass `Some(1)`,
To process a batch, pass `Some(batch_size)`.
core/lib/queued_job_processor/src/lib.rs