✨ Config option to timeout idle connections within the reqwest pool

[neon-mmd]

Work Expected From The Issue

Provide a config option under the server section of the config to allow users to timeout idle connections within the reqwest connection pool after a specified period of time to reduce network resource usage. Also, add the new request pool_idle_timeout for the same in the ClientBuilder struct in the aggregator.rs file and provide it with the newly added config option value passed via the function parameters.

The issue expects the following files to be changed:

• src/results/aggregator.rs
• websurfx/config.lua

Note

All the files that are expected to be changed are located under the codebase (websurfx directory).

Reason Behind These Changes

The reason behind having these changes is to allow the user to tweak the reqwest connection pool timeouts which can help reduce server resources automatically.

Sample Code

The sample codes for both the files as mentioned above have been provided below:

aggregator.rs

//! This module provides the functionality to scrape and gathers all the results from the upstream
//! search engines and then removes duplicate results.

use super::user_agent::random_user_agent;
use crate::handler::{file_path, FileType};
use crate::models::{
    aggregation_models::{EngineErrorInfo, SearchResult, SearchResults},
    engine_models::{EngineError, EngineHandler},
};
use error_stack::Report;
use futures::stream::FuturesUnordered;
use regex::Regex;
use reqwest::{Client, ClientBuilder};
use std::time::{SystemTime, UNIX_EPOCH};
use std::{fs::File, io::BufRead};
use std::{
    io::{BufReader, Read},
    time::Duration,
};
use tokio::task::JoinHandle;

/// A constant for holding the prebuilt Client globally in the app.
static CLIENT: std::sync::OnceLock<Client> = std::sync::OnceLock::new();

/// Aliases for long type annotations

type FutureVec =
    FuturesUnordered<JoinHandle<Result<Vec<(String, SearchResult)>, Report<EngineError>>>>;

/// The function aggregates the scraped results from the user-selected upstream search engines.
/// These engines can be chosen either from the user interface (UI) or from the configuration file.
/// The code handles this process by matching the selected search engines and adding them to a vector.
/// This vector is then used to create an asynchronous task vector using `tokio::spawn`, which returns
/// a future. This future is awaited in another loop. Once the results are collected, they are filtered
/// to remove any errors and ensure only proper results are included. If an error is encountered, it is
/// sent to the UI along with the name of the engine and the type of error. This information is finally
/// placed in the returned `SearchResults` struct.
///
/// Additionally, the function eliminates duplicate results. If two results are identified as coming from
/// multiple engines, their names are combined to indicate that the results were fetched from these upstream
/// engines. After this, all the data in the `Vec` is removed and placed into a struct that contains all
/// the aggregated results in a vector. Furthermore, the query used is also added to the struct. This step is
/// necessary to ensure that the search bar in the search remains populated even when searched from the query URL.
///
/// Overall, this function serves to aggregate scraped results from user-selected search engines, handling errors,
/// removing duplicates, and organizing the data for display in the UI.
///
/// # Example:
///
/// If you search from the url like `https://127.0.0.1/search?q=huston` then the search bar should
/// contain the word huston and not remain empty.
///
/// # Arguments
///
/// * `query` - Accepts a string to query with the above upstream search engines.
/// * `page` - Accepts an u32 page number.
/// * `random_delay` - Accepts a boolean value to add a random delay before making the request.
/// * `debug` - Accepts a boolean value to enable or disable debug mode option.
/// * `upstream_search_engines` - Accepts a vector of search engine names which was selected by the
/// * `request_timeout` - Accepts a time (secs) as a value which controls the server request timeout.
/// user through the UI or the config file.
///
/// # Error
///
/// Returns an error a reqwest and scraping selector errors if any error occurs in the results
/// function in either `searx` or `duckduckgo` or both otherwise returns a `SearchResults struct`
/// containing appropriate values.
pub async fn aggregate(
    query: &str,
    page: u32,
    random_delay: bool,
    debug: bool,
    upstream_search_engines: &[EngineHandler],
    request_timeout: u8,
    safe_search: u8,
    config: &Config
) -> Result<SearchResults, Box<dyn std::error::Error>> {
    let client = CLIENT.get_or_init(|| {
        ClientBuilder::new()
            .timeout(Duration::from_secs(request_timeout as u64)) // Add timeout to request to avoid DDOSing the server
+           .pool_idle_timeout(Duration::from_secs(config.pool_idle_connection_timeout as u64))
            .https_only(true)
            .gzip(true)
            .brotli(true)
            .build()
            .unwrap()
    });

    let user_agent: &str = random_user_agent();

    // Add a random delay before making the request.
    if random_delay || !debug {
        let nanos = SystemTime::now().duration_since(UNIX_EPOCH)?.subsec_nanos() as f32;
        let delay = ((nanos / 1_0000_0000 as f32).floor() as u64) + 1;
        tokio::time::sleep(Duration::from_secs(delay)).await;
    }

    let mut names: Vec<&str> = Vec::with_capacity(0);

    // create tasks for upstream result fetching
    let tasks: FutureVec = FutureVec::new();

    for engine_handler in upstream_search_engines {
        let (name, search_engine) = engine_handler.to_owned().into_name_engine();
        names.push(name);
        let query: String = query.to_owned();
        tasks.push(tokio::spawn(async move {
            search_engine
                .results(&query, page, user_agent, client, safe_search)
                .await
        }));
    }

    // get upstream responses
    let mut responses = Vec::with_capacity(tasks.len());

    for task in tasks {
        if let Ok(result) = task.await {
            responses.push(result)
        }
    }

    // aggregate search results, removing duplicates and handling errors the upstream engines returned
    let mut result_map: Vec<(String, SearchResult)> = Vec::new();
    let mut engine_errors_info: Vec<EngineErrorInfo> = Vec::new();

    let mut handle_error = |error: &Report<EngineError>, engine_name: &'static str| {
        log::error!("Engine Error: {:?}", error);
        engine_errors_info.push(EngineErrorInfo::new(
            error.downcast_ref::<EngineError>().unwrap(),
            engine_name,
        ));
    };

    for _ in 0..responses.len() {
        let response = responses.pop().unwrap();
        let engine = names.pop().unwrap();

        if result_map.is_empty() {
            match response {
                Ok(results) => result_map = results,
                Err(error) => handle_error(&error, engine),
            };
            continue;
        }

        match response {
            Ok(result) => {
                result.into_iter().for_each(|(key, value)| {
                    match result_map.iter().find(|(key_s, _)| key_s == &key) {
                        Some(value) => value.1.to_owned().add_engines(engine),
                        None => result_map.push((key, value)),
                    };
                });
            }
            Err(error) => handle_error(&error, engine),
        };
    }

    if safe_search >= 3 {
        let mut blacklist_map: Vec<(String, SearchResult)> = Vec::new();
        filter_with_lists(
            &mut result_map,
            &mut blacklist_map,
            file_path(FileType::BlockList)?,
        )?;

        filter_with_lists(
            &mut blacklist_map,
            &mut result_map,
            file_path(FileType::AllowList)?,
        )?;

        drop(blacklist_map);
    }

    let results: Vec<SearchResult> = result_map.iter().map(|(_, value)| value.clone()).collect();

    Ok(SearchResults::new(results, &engine_errors_info))
}

/// Filters a map of search results using a list of regex patterns.
///
/// # Arguments
///
/// * `map_to_be_filtered` - A mutable reference to a `Vec` of search results to filter, where the filtered results will be removed from.
/// * `resultant_map` - A mutable reference to a `Vec` to hold the filtered results.
/// * `file_path` - A `&str` representing the path to a file containing regex patterns to use for filtering.
///
/// # Errors
///
/// Returns an error if the file at `file_path` cannot be opened or read, or if a regex pattern is invalid.
pub fn filter_with_lists(
    map_to_be_filtered: &mut Vec<(String, SearchResult)>,
    resultant_map: &mut Vec<(String, SearchResult)>,
    file_path: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let mut reader = BufReader::new(File::open(file_path)?);

    for line in reader.by_ref().lines() {
        let re = Regex::new(line?.trim())?;

        // Iterate over each search result in the map and check if it matches the regex pattern
        for (index, (url, search_result)) in map_to_be_filtered.clone().into_iter().enumerate() {
            if re.is_match(&url.to_lowercase())
                || re.is_match(&search_result.title.to_lowercase())
                || re.is_match(&search_result.description.to_lowercase())
            {
                // If the search result matches the regex pattern, move it from the original map to the resultant map
                resultant_map.push(map_to_be_filtered.remove(index));
            }
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use smallvec::smallvec;
    use std::io::Write;
    use tempfile::NamedTempFile;

    #[test]
    fn test_filter_with_lists() -> Result<(), Box<dyn std::error::Error>> {
        // Create a map of search results to filter
        let mut map_to_be_filtered = Vec::new();
        map_to_be_filtered.push((
            "https://www.example.com".to_owned(),
            SearchResult {
                title: "Example Domain".to_owned(),
                url: "https://www.example.com".to_owned(),
                description: "This domain is for use in illustrative examples in documents."
                    .to_owned(),
                engine: smallvec!["Google".to_owned(), "Bing".to_owned()],
            },
        ));
        map_to_be_filtered.push((
            "https://www.rust-lang.org/".to_owned(),
            SearchResult {
                title: "Rust Programming Language".to_owned(),
                url: "https://www.rust-lang.org/".to_owned(),
                description: "A systems programming language that runs blazingly fast, prevents segfaults, and guarantees thread safety.".to_owned(),
                engine: smallvec!["Google".to_owned(), "DuckDuckGo".to_owned()],
            },)
        );

        // Create a temporary file with regex patterns
        let mut file = NamedTempFile::new()?;
        writeln!(file, "example")?;
        writeln!(file, "rust")?;
        file.flush()?;

        let mut resultant_map = Vec::new();
        filter_with_lists(
            &mut map_to_be_filtered,
            &mut resultant_map,
            file.path().to_str().unwrap(),
        )?;

        assert_eq!(resultant_map.len(), 2);
        assert!(resultant_map
            .iter()
            .any(|(key, _)| key == "https://www.example.com"));
        assert!(resultant_map
            .iter()
            .any(|(key, _)| key == "https://www.rust-lang.org/"));
        assert_eq!(map_to_be_filtered.len(), 0);

        Ok(())
    }

    #[test]
    fn test_filter_with_lists_wildcard() -> Result<(), Box<dyn std::error::Error>> {
        let mut map_to_be_filtered = Vec::new();
        map_to_be_filtered.push((
            "https://www.example.com".to_owned(),
            SearchResult {
                title: "Example Domain".to_owned(),
                url: "https://www.example.com".to_owned(),
                description: "This domain is for use in illustrative examples in documents."
                    .to_owned(),
                engine: smallvec!["Google".to_owned(), "Bing".to_owned()],
            },
        ));
        map_to_be_filtered.push((
            "https://www.rust-lang.org/".to_owned(),
            SearchResult {
                title: "Rust Programming Language".to_owned(),
                url: "https://www.rust-lang.org/".to_owned(),
                description: "A systems programming language that runs blazingly fast, prevents segfaults, and guarantees thread safety.".to_owned(),
                engine: smallvec!["Google".to_owned(), "DuckDuckGo".to_owned()],
            },
        ));

        // Create a temporary file with a regex pattern containing a wildcard
        let mut file = NamedTempFile::new()?;
        writeln!(file, "ex.*le")?;
        file.flush()?;

        let mut resultant_map = Vec::new();

        filter_with_lists(
            &mut map_to_be_filtered,
            &mut resultant_map,
            file.path().to_str().unwrap(),
        )?;

        assert_eq!(resultant_map.len(), 1);
        assert!(resultant_map
            .iter()
            .any(|(key, _)| key == "https://www.example.com"));
        assert_eq!(map_to_be_filtered.len(), 1);
        assert!(map_to_be_filtered
            .iter()
            .any(|(key, _)| key == "https://www.rust-lang.org/"));

        Ok(())
    }

    #[test]
    fn test_filter_with_lists_file_not_found() {
        let mut map_to_be_filtered = Vec::new();

        let mut resultant_map = Vec::new();

        // Call the `filter_with_lists` function with a non-existent file path
        let result = filter_with_lists(
            &mut map_to_be_filtered,
            &mut resultant_map,
            "non-existent-file.txt",
        );

        assert!(result.is_err());
    }

    #[test]
    fn test_filter_with_lists_invalid_regex() {
        let mut map_to_be_filtered = Vec::new();
        map_to_be_filtered.push((
            "https://www.example.com".to_owned(),
            SearchResult {
                title: "Example Domain".to_owned(),
                url: "https://www.example.com".to_owned(),
                description: "This domain is for use in illustrative examples in documents."
                    .to_owned(),
                engine: smallvec!["Google".to_owned(), "Bing".to_owned()],
            },
        ));

        let mut resultant_map = Vec::new();

        // Create a temporary file with an invalid regex pattern
        let mut file = NamedTempFile::new().unwrap();
        writeln!(file, "example(").unwrap();
        file.flush().unwrap();

        let result = filter_with_lists(
            &mut map_to_be_filtered,
            &mut resultant_map,
            file.path().to_str().unwrap(),
        );

        assert!(result.is_err());
    }
}

config.lua

 -- ### General ###
logging = true -- an option to enable or disable logs.
debug = false -- an option to enable or disable debug mode.
threads = 10 -- the amount of threads that the app will use to run (the value should be greater than 0).

 -- ### Server ###
port = "8080" -- port on which server should be launched
binding_ip = "127.0.0.1" --ip address on the which server should be launched.
production_use = false -- whether to use production mode or not (in other words this option should be used if it is to be used to host it on the server to provide a service to a large number of users (more than one))
 -- if production_use is set to true
 -- There will be a random delay before sending the request to the search engines, this is to prevent DDoSing the upstream search engines from a large number of simultaneous requests.
request_timeout = 30 -- timeout for the search requests sent to the upstream search engines to be fetched (value in seconds).
+pool_idle_connection_timeout = 30 -- timeout for the idle connections in the reqwest HTTP connection pool (value in seconds).
rate_limiter = {
	number_of_requests = 20, -- The number of request that are allowed within a provided time limit.
	time_limit = 3, -- The time limit in which the quantity of requests that should be accepted.
}

 -- ### Search ###
 -- Filter results based on different levels. The levels provided are:
 -- {{
 -- 0 - None
 -- 1 - Low
 -- 2 - Moderate
 -- 3 - High
 -- 4 - Aggressive
 -- }}
safe_search = 2

 -- ### Website ###
 -- The different colorschemes provided are:
 -- {{
 -- catppuccin-mocha
 -- dark-chocolate
 -- dracula
 -- gruvbox-dark
 -- monokai
 -- nord
 -- oceanic-next
 -- one-dark
 -- solarized-dark
 -- solarized-light
 -- tokyo-night
 -- tomorrow-night
 -- }}
colorscheme = "catppuccin-mocha" -- the colorscheme name which should be used for the website theme
 -- The different themes provided are:
 -- {{
 -- simple
 -- }}
theme = "simple" -- the theme name which should be used for the website
 -- The different animations provided are:
 -- {{
 -- simple-frosted-glow
 -- }}
animation = "simple-frosted-glow" -- the animation name which should be used with the theme or `nil` if you don't want any animations.

 -- ### Caching ###
redis_url = "redis://127.0.0.1:8082" -- redis connection url address on which the client should connect on.
cache_expiry_time = 600 -- This option takes the expiry time of the search results (value in seconds and the value should be greater than or equal to 60 seconds).
 -- ### Search Engines ###
upstream_search_engines = {
    DuckDuckGo = true,
    Searx = false,
    Brave = false,
    Startpage = false,
    LibreX = false,
    Mojeek = false,
    Bing = false,
} -- select the upstream search engines from which the results should be fetched.

[github-actions]

The issue has been unlocked and is now ready for dev. If you would like to work on this issue, you can comment to have it assigned to you. You can learn more in our contributing guide https://github.com/neon-mmd/websurfx/blob/rolling/CONTRIBUTING.md

[KekmaTime]

@neon-mmd i would like to work on this issue

[neon-mmd]

@neon-mmd i would like to work on this issue

Yes, sure, we would be glad to assign this issue to you, but our suggestion would be to do it step by step like one issue at a time. This allows others who might be willing to contribute to take up these issue as well (in other words, it will not block them from contributing). Actually, we are not suggesting that we will not assign them to you at all. We are just recommending that it would be better to do it step by step so if anyone is willing to contribute, they can also do so without getting blocked. If nobody does take this up, then we will for sure assign it to you. 🙂 .

[KekmaTime]

@neon-mmd i would like to work on this issue

Yes, sure, we would be glad to assign this issue to you, but our suggestion would be to do it step by step like one issue at a time. This allows others who might be willing to contribute to take up these issue as well (in other words, it will not block them from contributing). Actually, we are not suggesting that we will not assign them to you at all. We are just recommending that it would be better to do it step by step so if anyone is willing to contribute, they can also do so without getting blocked. If nobody does take this up, then we will for sure assign it to you. 🙂 .

@neon-mmd can i work on it now?

[neon-mmd]

@neon-mmd i would like to work on this issue

Yes, sure, we would be glad to assign this issue to you, but our suggestion would be to do it step by step like one issue at a time. This allows others who might be willing to contribute to take up these issue as well (in other words, it will not block them from contributing). Actually, we are not suggesting that we will not assign them to you at all. We are just recommending that it would be better to do it step by step so if anyone is willing to contribute, they can also do so without getting blocked. If nobody does take this up, then we will for sure assign it to you. 🙂 .

@neon-mmd can i work on it now?

Ok sure, we will assign this issue to you. You may start working on it right away 🚀 🙂 .