pub async fn run(mut self, mut stop_receiver: watch::Receiver<bool>) -> anyhow::Result<()> {
        self.health_updater
            .update(Health::from(HealthStatus::Ready));
        let mut last_updated_l1_batch = None;

        self.last_scanned_batch = self.determine_first_batch_to_scan().await?;
        self.drop_entries_without_inclusion_data().await?;

        while !*stop_receiver.borrow_and_update() {
            let step_outcome = self.step().await;
            let need_to_sleep = match step_outcome {
                Ok(StepOutcome::UpdatedBatch(batch_number)) => {
                    last_updated_l1_batch = Some(batch_number);
                    self.update_health(last_updated_l1_batch);
                    false
                }
                Ok(StepOutcome::NoProgress) | Ok(StepOutcome::NoInclusionDataFromMainNode) => {
                    // Update health status even if no progress was made to timely clear a previously set
                    // "affected" health.
                    self.update_health(last_updated_l1_batch);
                    true
                }
                Ok(StepOutcome::UnableToFetchInclusionData) => {
                    tracing::warn!(
                        "No inclusion data for the batch from DA layer, will retry later"
                    );
                    self.update_health(last_updated_l1_batch);
                    true
                }
                Err(err) => {
                    if err.is_retriable {
                        tracing::warn!(
                            "Error in data availability fetcher, will retry after a delay: {err:?}"
                        );
                        let health = DataAvailabilityFetcherHealth::Affected {
                            error: err.error.to_string(),
                        };
                        self.health_updater.update(health.into());
                        true
                    } else {
                        tracing::error!("Fatal error in data availability fetcher: {err:?}");
                        return Err(err.error);
                    }
                }
            };

            if need_to_sleep
                && tokio::time::timeout(self.poll_interval, stop_receiver.changed())
                    .await
                    .is_ok()
            {
                break;
            }
        }
        tracing::info!("Stop request received; data availability fetcher is shutting down");
        Ok(())
    }
Runs this component until a fatal error occurs or a stop request is received. Retriable errors
(e.g., no network connection) are handled gracefully by retrying after a delay.
core/node/node_sync/src/data_availability_fetcher/mod.rs