spider_lib/

crawler.rs

//! The core Crawler implementation for the `spider-lib` framework.
//!
//! This module defines the `Crawler` struct, which acts as the central orchestrator
//! for the web scraping process. It ties together the scheduler, downloader,
//! middlewares, spiders, and item pipelines to execute a crawl. The crawler
//! manages the lifecycle of requests and items, handles concurrency, supports
//! checkpointing for fault tolerance, and collects statistics for monitoring.
//!
//! It utilizes a task-based asynchronous model, spawning distinct tasks for
//! handling initial requests, downloading web pages, parsing responses, and
//! processing scraped items.

use crate::downloader::Downloader;
use crate::error::SpiderError;
use crate::item::{ParseOutput, ScrapedItem};
use crate::middleware::{Middleware, MiddlewareAction};
use crate::pipeline::Pipeline;
use crate::request::Request;
use crate::response::Response;
use crate::scheduler::Scheduler;
use crate::spider::Spider;
use crate::state::CrawlerState;
use anyhow::Result;
use futures_util::future::join_all;
use kanal::{AsyncReceiver, AsyncSender, bounded_async};

#[cfg(feature = "checkpoint")]
use crate::checkpoint::save_checkpoint;
#[cfg(feature = "checkpoint")]
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;
use tokio::sync::Mutex;
use crate::stats::StatCollector;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;
use tracing::{debug, error, info, warn};

/// The central orchestrator for the web scraping process, handling requests, responses, items, concurrency, checkpointing, and statistics collection.
pub struct Crawler<S: Spider, C> {
    scheduler: Arc<Scheduler>,
    req_rx: AsyncReceiver<Request>,
    stats: Arc<StatCollector>, // Added
    downloader: Arc<dyn Downloader<Client = C> + Send + Sync>,
    middlewares: Vec<Box<dyn Middleware<C> + Send + Sync>>,
    spider: Arc<Mutex<S>>,
    item_pipelines: Vec<Box<dyn Pipeline<S::Item>>>,
    max_concurrent_downloads: usize,
    parser_workers: usize,
    max_concurrent_pipelines: usize,
    #[cfg(feature = "checkpoint")]
    checkpoint_path: Option<PathBuf>,
    #[cfg(feature = "checkpoint")]
    checkpoint_interval: Option<Duration>,
}

impl<S, C> Crawler<S, C>
where
    S: Spider + 'static,
    S::Item: ScrapedItem,
    C: Send + Sync + 'static,
{
    /// Creates a new `Crawler` instance with the given components and configuration.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        scheduler: Arc<Scheduler>,
        req_rx: AsyncReceiver<Request>,
        downloader: Arc<dyn Downloader<Client = C> + Send + Sync>,
        middlewares: Vec<Box<dyn Middleware<C> + Send + Sync>>,
        spider: S,
        item_pipelines: Vec<Box<dyn Pipeline<S::Item>>>,
        max_concurrent_downloads: usize,
        parser_workers: usize,
        max_concurrent_pipelines: usize,
        #[cfg(feature = "checkpoint")] checkpoint_path: Option<PathBuf>,
        #[cfg(feature = "checkpoint")] checkpoint_interval: Option<Duration>,
        stats: Arc<StatCollector>,
    ) -> Self {
        Crawler {
            scheduler,
            req_rx,
            stats,
            downloader,
            middlewares,
            spider: Arc::new(Mutex::new(spider)),
            item_pipelines,
            max_concurrent_downloads,
            parser_workers,
            max_concurrent_pipelines,
            #[cfg(feature = "checkpoint")]
            checkpoint_path,
            #[cfg(feature = "checkpoint")]
            checkpoint_interval,
        }
    }

    /// Starts the crawl, orchestrating the scraping process, managing tasks, handling shutdown, checkpointing, and logging statistics.
    pub async fn start_crawl(self) -> Result<(), SpiderError> {
        info!("Crawler starting crawl");

        let Crawler {
            scheduler,
            req_rx,
            stats,
            downloader,
            middlewares,
            spider,
            item_pipelines,
            max_concurrent_downloads,
            parser_workers,
            max_concurrent_pipelines,
            #[cfg(feature = "checkpoint")]
            checkpoint_path,
            #[cfg(feature = "checkpoint")]
            checkpoint_interval,
        } = self;

        let state = CrawlerState::new();
        let pipelines = Arc::new(item_pipelines);
        let channel_capacity = max_concurrent_downloads * 2;

        let (res_tx, res_rx) = bounded_async(channel_capacity);
        let (item_tx, item_rx) = bounded_async(channel_capacity);

        let initial_requests_task =
            spawn_initial_requests_task::<S>(scheduler.clone(), spider.clone(), stats.clone());

        let downloader_task = spawn_downloader_task::<S, C>(
            scheduler.clone(),
            req_rx,
            downloader,
            Arc::new(Mutex::new(middlewares)),
            state.clone(),
            res_tx.clone(),
            max_concurrent_downloads,
            stats.clone(),
        );

        let parser_task = spawn_parser_task::<S>(
            scheduler.clone(),
            spider.clone(),
            state.clone(),
            res_rx,
            item_tx.clone(),
            parser_workers,
            stats.clone(),
        );

        let item_processor_task = spawn_item_processor_task::<S>(
            state.clone(),
            item_rx,
            pipelines.clone(),
            max_concurrent_pipelines,
            stats.clone(),
        );

        #[cfg(feature = "checkpoint")]
        if let (Some(path), Some(interval)) = (&checkpoint_path, checkpoint_interval) {
            let scheduler_clone = scheduler.clone();
            let pipelines_clone = pipelines.clone();
            let path_clone = path.clone();

            tokio::spawn(async move {
                let mut interval_timer = tokio::time::interval(interval);
                interval_timer.tick().await;
                loop {
                    tokio::select! {
                        _ = interval_timer.tick() => {
                            if let Ok(scheduler_checkpoint) = scheduler_clone.snapshot().await &&
                                let Err(e) = save_checkpoint::<S>(&path_clone, scheduler_checkpoint, &pipelines_clone).await {
                                    error!("Periodic checkpoint save failed: {}", e);
                            }
                        }
                    }
                }
            });
        }

        tokio::select! {
            _ = tokio::signal::ctrl_c() => {
                info!("Ctrl-C received, initiating graceful shutdown.");
            }
            _ = async {
                loop {
                    if scheduler.is_idle() && state.is_idle() {
                        tokio::time::sleep(Duration::from_millis(50)).await;
                        if scheduler.is_idle() && state.is_idle() {
                            break;
                        }
                    }
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
            } => {
                info!("Crawl has become idle, initiating shutdown.");
            }
        }

        info!("Initiating actor shutdowns.");

        #[cfg(feature = "checkpoint")]
        let scheduler_checkpoint = scheduler.snapshot().await?;

        drop(res_tx);
        drop(item_tx);

        scheduler.shutdown().await?;

        item_processor_task
            .await
            .map_err(|e| SpiderError::GeneralError(format!("Item processor task failed: {}", e)))?;

        parser_task
            .await
            .map_err(|e| SpiderError::GeneralError(format!("Parser task failed: {}", e)))?;

        downloader_task
            .await
            .map_err(|e| SpiderError::GeneralError(format!("Downloader task failed: {}", e)))?;

        initial_requests_task.await.map_err(|e| {
            SpiderError::GeneralError(format!("Initial requests task failed: {}", e))
        })?;

        #[cfg(feature = "checkpoint")]
        if let Some(path) = &checkpoint_path
            && let Err(e) = save_checkpoint::<S>(path, scheduler_checkpoint, &pipelines).await
        {
            error!("Final checkpoint save failed: {}", e);
        }

        // Close all pipelines
        info!("Closing item pipelines...");
        let closing_futures: Vec<_> = pipelines.iter().map(|p| p.close()).collect();
        join_all(closing_futures).await;

        info!("Crawl finished successfully.");
        Ok(())
    }

    /// Returns a cloned Arc to the `StatCollector` instance used by this crawler.
    ///
    /// This allows programmatic access to the collected statistics at any time during or after the crawl.
    pub fn get_stats(&self) -> Arc<StatCollector> {
        Arc::clone(&self.stats)
    }
}

fn spawn_initial_requests_task<S>(
    scheduler: Arc<Scheduler>,
    spider: Arc<Mutex<S>>,
    stats: Arc<StatCollector>,
) -> tokio::task::JoinHandle<()>
where
    S: Spider + 'static,
    S::Item: ScrapedItem,
{
    tokio::spawn(async move {
        match spider.lock().await.start_requests() {
            Ok(requests) => {
                for mut req in requests {
                    req.url.set_fragment(None);
                    match scheduler.enqueue_request(req).await {
                        Ok(_) => {
                            stats.increment_requests_enqueued();
                        }
                        Err(e) => {
                            error!("Failed to enqueue initial request: {}", e);
                        }
                    }
                }
            }
            Err(e) => error!("Failed to create start requests: {}", e),
        }
    })
}

#[allow(clippy::too_many_arguments)]
fn spawn_downloader_task<S, C>(
    scheduler: Arc<Scheduler>,
    req_rx: AsyncReceiver<Request>,
    downloader: Arc<dyn Downloader<Client = C> + Send + Sync>,
    middlewares: Arc<Mutex<Vec<Box<dyn Middleware<C> + Send + Sync>>>>,
    state: Arc<CrawlerState>,
    res_tx: AsyncSender<Response>,
    max_concurrent_downloads: usize,
    stats: Arc<StatCollector>,
) -> tokio::task::JoinHandle<()>
where
    S: Spider + 'static,
    S::Item: ScrapedItem,
    C: Send + Sync + 'static,
{
    let semaphore = Arc::new(Semaphore::new(max_concurrent_downloads));
    let mut tasks = JoinSet::new();

    tokio::spawn(async move {
        while let Ok(request) = req_rx.recv().await {
            let permit = match semaphore.clone().acquire_owned().await {
                Ok(permit) => permit,
                Err(_) => {
                    warn!("Semaphore closed, shutting down downloader actor.");
                    break;
                }
            };

            state.in_flight_requests.fetch_add(1, Ordering::SeqCst);
            let downloader_clone = Arc::clone(&downloader);
            let middlewares_clone = Arc::clone(&middlewares);
            let res_tx_clone = res_tx.clone();
            let state_clone = Arc::clone(&state);
            let scheduler_clone = Arc::clone(&scheduler);
            let stats_clone = Arc::clone(&stats);

            tasks.spawn(async move {
                let mut early_returned_response: Option<Response> = None;

                // Process request middlewares
                let mut processed_request_opt = Some(request); 
                for mw in middlewares_clone.lock().await.iter_mut() {
                    let req_to_process = processed_request_opt.take().expect("Request should be present before middleware processing");
                    match mw.process_request(downloader_clone.client(), req_to_process).await {
                        Ok(MiddlewareAction::Continue(req)) => {
                            processed_request_opt = Some(req);
                        }
                        Ok(MiddlewareAction::Retry(req, delay)) => {
                            stats_clone.increment_requests_retried();
                            tokio::time::sleep(delay).await;
                            if scheduler_clone.enqueue_request(*req).await.is_err() {
                                error!("Failed to re-enqueue retried request.");
                            }
                            return;
                        }
                        Ok(MiddlewareAction::Drop) => {
                            stats_clone.increment_requests_dropped();
                            debug!("Request dropped by middleware.");
                            return;
                        }
                        Ok(MiddlewareAction::ReturnResponse(resp)) => {
                            early_returned_response = Some(resp);
                            break;
                        }
                        Err(e) => {
                            error!("Request middleware error: {:?}", e);
                            return;
                        }
                    }
                }

                // Download or use early response
                // If early_returned_response is Some, request was consumed by a middleware
                // If early_returned_response is None, processed_request_opt must contain the request
                let response = match early_returned_response { 
                    Some(resp) => {
                        if resp.cached {
                            stats_clone.increment_responses_from_cache();
                        }
                        stats_clone.increment_requests_succeeded();
                        stats_clone.increment_responses_received();
                        stats_clone.record_response_status(resp.status.as_u16());
                        resp
                    },
                    None => {
                        let request_for_download = processed_request_opt.expect("Request must be available for download if not handled by middleware or early returned response");
                        stats_clone.increment_requests_sent();
                        match downloader_clone.download(request_for_download).await { 
                            Ok(resp) => {
                                stats_clone.increment_requests_succeeded();
                                stats_clone.increment_responses_received();
                                stats_clone.record_response_status(resp.status.as_u16());
                                // Corrected for Response.body being Bytes, not Option<Body>
                                stats_clone.add_bytes_downloaded(resp.body.len());
                                resp
                            },
                            Err(e) => {
                                stats_clone.increment_requests_failed();
                                error!("Download error: {:?}", e);
                                return;
                            }
                        }
                    },
                };

                // Process response middlewares
                let mut processed_response_opt = Some(response); 
                for mw in middlewares_clone.lock().await.iter_mut().rev() {
                    let res_to_process = processed_response_opt.take().expect("Response should be present before middleware processing"); // Take ownership for current middleware
                    match mw.process_response(res_to_process).await {
                        Ok(MiddlewareAction::Continue(res)) => {
                            processed_response_opt = Some(res); // Reassign for next middleware
                        }
                        Ok(MiddlewareAction::Retry(req, delay)) => {
                            stats_clone.increment_requests_retried();
                            tokio::time::sleep(delay).await;
                            if scheduler_clone.enqueue_request(*req).await.is_err() {
                                error!("Failed to re-enqueue retried request.");
                            }
                            return;
                        }
                        Ok(MiddlewareAction::Drop) => {
                            stats_clone.increment_requests_dropped();
                            debug!("Response dropped by middleware.");
                            return;
                        }
                        Ok(MiddlewareAction::ReturnResponse(_)) => {
                            // This indicates the middleware has fully handled or consumed the response.
                            // Effectively, the response is dropped from further processing by this chain.
                            debug!("ReturnResponse action encountered in process_response; this is unexpected and effectively drops the response for further processing.");
                            processed_response_opt = None; 
                            break; 
                        }
                        Err(e) => {
                            error!("Response middleware error: {:?}", e);
                            return;
                        }
                    }
                }

                // Send the final processed response, if it still exists
                if let Some(final_response) = processed_response_opt
                    && res_tx_clone.send(final_response).await.is_err() {
                    error!("Response channel closed, cannot send parsed response.");
                }

                state_clone.in_flight_requests.fetch_sub(1, Ordering::SeqCst);
                drop(permit);
            });
        }
        while let Some(res) = tasks.join_next().await {
            if let Err(e) = res {
                error!("A download task failed: {:?}", e);
            }
        }
    })
}

fn spawn_parser_task<S>(
    scheduler: Arc<Scheduler>,
    spider: Arc<Mutex<S>>,
    state: Arc<CrawlerState>,
    res_rx: AsyncReceiver<Response>,
    item_tx: AsyncSender<S::Item>,
    parser_workers: usize,
    stats: Arc<StatCollector>,
) -> tokio::task::JoinHandle<()>
where
    S: Spider + 'static,
    S::Item: ScrapedItem,
{
    let mut tasks = JoinSet::new();
    let internal_parse_tx: AsyncSender<Response>;
    let internal_parse_rx: AsyncReceiver<Response>;
    (internal_parse_tx, internal_parse_rx) = bounded_async(parser_workers * 2);

            // Spawn N parsing worker tasks

        for _ in 0..parser_workers {

            let internal_parse_rx_clone = internal_parse_rx.clone();

            let spider_clone = Arc::clone(&spider);

            let scheduler_clone = Arc::clone(&scheduler);

            let item_tx_clone = item_tx.clone();

            let state_clone = Arc::clone(&state);

            let stats_clone = Arc::clone(&stats);

    

            tasks.spawn(async move {

                while let Ok(response) = internal_parse_rx_clone.recv().await {

                    debug!("Parsing response from {}", response.url);

                    match spider_clone.lock().await.parse(response).await {

                        Ok(outputs) => {

                                                        process_crawl_outputs::<S>(

                                                            outputs,

                                                            scheduler_clone.clone(),

                                                            item_tx_clone.clone(),

                                                            state_clone.clone(),

                                                            stats_clone.clone(),

                                                    )
                        .await;
                    }
                    Err(e) => error!("Spider parsing error: {:?}", e),
                }
                state_clone.parsing_responses.fetch_sub(1, Ordering::SeqCst);
            }
        });
    }

    tokio::spawn(async move {
        while let Ok(response) = res_rx.recv().await {
            state.parsing_responses.fetch_add(1, Ordering::SeqCst);
            if internal_parse_tx.send(response).await.is_err() {
                error!("Internal parse channel closed, cannot send response to parser worker.");
                state.parsing_responses.fetch_sub(1, Ordering::SeqCst);
            }
        }

        drop(internal_parse_tx);

        while let Some(res) = tasks.join_next().await {
            if let Err(e) = res {
                error!("A parsing worker task failed: {:?}", e);
            }
        }
    })
}

async fn process_crawl_outputs<S>(
    outputs: ParseOutput<S::Item>,
    scheduler: Arc<Scheduler>,
    item_tx: AsyncSender<S::Item>,
    state: Arc<CrawlerState>,
    stats: Arc<StatCollector>,
) where
    S: Spider + 'static,
    S::Item: ScrapedItem,
{
    let (items, requests) = outputs.into_parts();
    info!(
        "Processed {} requests and {} items from spider output.",
        requests.len(),
        items.len()
    );

    stats.increment_items_scraped();

    let mut request_error_total = 0;
    for request in requests {
        match scheduler.enqueue_request(request).await {
            Ok(_) => {
                // Stat: requests_enqueued
                stats.increment_requests_enqueued();
            }
            Err(_) => {
                request_error_total += 1;
            }
        }
    }
    if request_error_total > 0 {
        error!(
            "Failed to enqueue {} requests: scheduler channel closed.",
            request_error_total
        );
    }

    let mut item_error_total = 0;
    for item in items {
        state.processing_items.fetch_add(1, Ordering::SeqCst);
        if item_tx.send(item).await.is_err() {
            item_error_total += 1;
            state.processing_items.fetch_sub(1, Ordering::SeqCst);
        }
    }
    if item_error_total > 0 {
        error!(
            "Failed to send {} scraped items: channel closed.",
            item_error_total
        );
    }
}

fn spawn_item_processor_task<S>(
    state: Arc<CrawlerState>,
    item_rx: AsyncReceiver<S::Item>,
    pipelines: Arc<Vec<Box<dyn Pipeline<S::Item>>>>,
    max_concurrent_pipelines: usize,
    stats: Arc<StatCollector>,
) -> tokio::task::JoinHandle<()>
where
    S: Spider + 'static,
    S::Item: ScrapedItem,
{
    let mut tasks = JoinSet::new();
    let semaphore = Arc::new(Semaphore::new(max_concurrent_pipelines));
    tokio::spawn(async move {
        while let Ok(item) = item_rx.recv().await {
            let permit = match semaphore.clone().acquire_owned().await {
                Ok(p) => p,
                Err(_) => {
                    warn!("Semaphore closed, shutting down item processor actor.");
                    break;
                }
            };

            let state_clone = Arc::clone(&state);
            let pipelines_clone = Arc::clone(&pipelines);
            let stats_clone = Arc::clone(&stats);
            tasks.spawn(async move {
                let mut item_to_process = Some(item);
                for pipeline in pipelines_clone.iter() {
                    if let Some(item) = item_to_process.take() {
                        match pipeline.process_item(item).await {
                            Ok(Some(next_item)) => item_to_process = Some(next_item),
                            Ok(None) => {
                                stats_clone.increment_items_dropped_by_pipeline();
                                break;
                            }
                            Err(e) => {
                                error!("Pipeline error for {}: {:?}", pipeline.name(), e);
                                stats_clone.increment_items_dropped_by_pipeline();
                                break;
                            }
                        }
                    } else {
                        break;
                    }
                }
                // If item survived all pipelines, it's processed
                if item_to_process.is_some() {
                    stats_clone.increment_items_processed();
                }
                state_clone.processing_items.fetch_sub(1, Ordering::SeqCst);
                drop(permit);
            });
        }
        while let Some(res) = tasks.join_next().await {
            if let Err(e) = res {
                error!("An item processing task failed: {:?}", e);
            }
        }
    })
}