workerlib/

lib.rs

//! This is the workerlib crate
//!

mod arguments;
mod download;
pub mod execute;
mod fetch;
pub mod git_config;
mod importer;
pub mod lorry_specs;
mod push;
mod raw_files;
pub mod workspace;

use crate::execute::{execute, Error as ExecutionError};
pub use arguments::Arguments;
use git2::{Error as Libgit2Error, Repository};
use glob::Pattern;
pub use lorry_specs::extract_lorry_specs;
pub use lorry_specs::LorrySpec;
use lorry_specs::SingleLorry;
use std::collections::BTreeMap;
use thiserror::Error;
use url::Url;

#[cfg(test)]
mod test_server;

/// Extra header exposed on Git fetch uperations to upstream servers and
/// push operations on downstream.
pub const LORRY_VERSION_HEADER: &str = concat!("Lorry-Version: ", env!("CARGO_PKG_VERSION"));

pub const DEFAULT_BRANCH_NAME: &str = "main";
pub const DEFAULT_REF_NAME: &str = "refs/heads/main";

/// Non-fatal error encountered when running a Lorry.
#[derive(Clone, Debug)]
pub enum Warning {
    /// Anytime a refspec is specified but it doesn't match any refs in the
    /// upstream repository.
    NonMatchingRefSpecs {
        /// Pattern from the configuration that did not match any refs
        pattern: String,
    },
}

impl Warning {
    /// Return a nicely formatted name
    pub fn name(&self) -> String {
        match self {
            Warning::NonMatchingRefSpecs { pattern: _ } => String::from("NonMatchingRefSpec"),
        }
    }

    /// Return the body of the error message
    pub fn message(&self) -> String {
        match self {
            Warning::NonMatchingRefSpecs { pattern } => pattern.clone(),
        }
    }
}

/// Results of a parsing refs from porcelain git push output
///
/// Example command:
/// ```txt
/// git push --porcelain origin 'refs/heads/*' 2>/dev/null`
/// ```
#[derive(Clone, Debug, Default)]
pub struct PushRefs(pub Vec<(String, String)>, pub Vec<(String, String)>);

/// Contains the status of a single mirror operation
#[derive(Default)]
pub struct MirrorStatus {
    pub push_refs: PushRefs,
    pub warnings: Vec<Warning>,
}

/// Top-level error type containing all possible errors during a mirroring operation.
#[derive(Error, Debug)]
pub enum Error {
    /// The upstream was a git repo. There was an error pulling from the remote.
    #[error("Failed to fetch upstream: {0}")]
    Fetch(#[from] fetch::Error),

    /// Indicates that the job exceeded the time it was allocated in the
    /// Lorry configuration
    #[error("job exceeded the allocated timeout: {seconds}")]
    TimeoutExceeded { seconds: u64 },

    /// An error occurred that originated in the workspace module.
    #[error("Workspace Error: {0}")]
    WorkspaceError(#[from] workspace::Error),

    /// An error occured when importing raw files into a working directory.
    #[error("LFS Importer Error: {0}")]
    LFSImporter(#[from] importer::Error),

    /// Generic command execution error not specific to any particular module.
    #[error("Command Execution Error: {0}")]
    Command(#[from] ExecutionError),

    /// Generic IO error
    #[error("IO Error")]
    Io { command: String },
    /// Tried to access a repository on the local filesystem as the downstream,
    /// but the given path is not a valid one.
    #[error(
        "Downstream is specified as a path on the local filesystem, but the path is malformed."
    )]
    InvalidPath(String),

    /// Problem parsing a downstream git url
    #[error("Failed to construct path to downstream git repo")]
    ParseError(#[from] url::ParseError),

    /// Generic Libgit2 failure not specific to any particular module.
    #[error("Internal git failure")]
    Libgit2Error(#[from] Libgit2Error),

    /// Indicates that 100% of the desired refs failed to push to the
    /// downstream.
    #[error("All ({n_attempted}) refspecs failed")]
    AllRefspecsFailed { refs: PushRefs, n_attempted: i64 },

    /// Indicates some but not all of the desired refspecs were not able to
    /// be pushed downstream for some reason.
    #[error("{n_failed} refspecs failed")]
    SomeRefspecsFailed { refs: PushRefs, n_failed: i64 },

    /// Indicates a refspec was malformed
    #[error("Invalid Glob: {pattern} - {error}")]
    InvalidGlobPattern { pattern: String, error: String },

    /// An error related to raw-file management occurred.
    #[error("Raw File Related Error: {0}")]
    RawFiles(#[from] raw_files::Error),

    /// Indicates none of the refs in the mirror configuration matches those
    /// that were available in the repository.
    #[error("No Matching Refspecs")]
    NoMatchingRefspecs,

    /// Indicates that a push command failed and we could not understand the
    /// the reason why. This may be due to a downstream server error.
    #[error("Cannot parse push output: {0}")]
    CannotParsePushOutput(String),

    /// Indicates that the server requires Sha256sum values for all raw-files
    /// but the file did not have one specified.
    #[error("Sha256sums are missing: \n{0}")]
    Sha256sumsNotSpecified(String),

    /// Indicates the ignore pattern was not a valid regular expression
    #[error("Invalid ignore pattern: {0}")]
    InvalidIgnorePattern(String),
}

impl Error {
    /// Return the underlying status code from the error if the error is the
    /// result of an command execution failure otherwise return nothing.
    #[allow(clippy::collapsible_match)]
    pub fn status(&self) -> Option<i32> {
        match self {
            Error::Fetch(e) => match e {
                fetch::Error::Command(e) => e.status(),
                _ => None,
            },
            Error::Command(e) => match e {
                ExecutionError::IO {
                    command: _,
                    source: _,
                } => None,
                ExecutionError::CommandError {
                    command: _,
                    status,
                    stderr: _,
                    stdout: _,
                } => Some(status.code().unwrap_or(-1)),
            },
            _ => None,
        }
    }
}

fn get_refs(n_refs: usize, stdout: &str) -> Result<PushRefs, Error> {
    match crate::push::Push::parse_output(stdout) {
        Ok(results) => {
            if results.1.len() == n_refs {
                return Err(Error::AllRefspecsFailed {
                    refs: results.clone(),
                    n_attempted: n_refs as i64,
                });
            }
            if !results.1.is_empty() {
                Err(Error::SomeRefspecsFailed {
                    refs: results.clone(),
                    n_failed: results.1.len() as i64,
                })
            } else {
                Ok(results)
            }
        }
        Err(message) => {
            tracing::warn!("Failed to parse git push output:\n{}", message);
            Err(Error::CannotParsePushOutput(message))
        }
    }
}

/// Match refs ignoring refs/{heads,tags}/ unless they're explicitly specified
fn match_ref(pattern: &Pattern, input: &str) -> bool {
    let pattern_str = pattern.as_str();
    if pattern_str.starts_with("refs/heads/") || pattern_str.starts_with("refs/tags/") {
        pattern.matches(input)
    } else {
        pattern.matches(
            input
                .trim_start_matches("refs/heads/")
                .trim_start_matches("refs/tags/"),
        )
    }
}

/// Given all of the refs return a vec of those with some matches and a vec of
/// any patterns that didn't result in any matches at all.
fn parse_refs(
    refs: &[String],
    ref_patterns: Option<&[String]>,
    ignore_patterns: Option<&[String]>,
) -> Result<(Vec<String>, Vec<String>), Error> {
    let ignore_globs = ignore_patterns
        .as_ref()
        .map(|ignore_patterns| {
            ignore_patterns
                .iter()
                .try_fold(Vec::new(), |mut accm, pattern| {
                    let glob_pattern =
                        Pattern::new(pattern).map_err(|e| Error::InvalidGlobPattern {
                            pattern: pattern.clone(),
                            error: e.to_string(),
                        })?;
                    accm.push(glob_pattern);
                    Ok::<Vec<Pattern>, Error>(accm)
                })
        })
        .transpose()?;

    let refs: Vec<String> = refs
        .iter()
        .filter_map(|ref_name| {
            if ignore_globs.as_ref().is_some_and(|ignore_globs| {
                ignore_globs
                    .iter()
                    .any(|pattern| match_ref(pattern, ref_name))
            }) {
                None
            } else {
                Some(ref_name.clone())
            }
        })
        .collect();

    if let Some(ref_specs) = ref_patterns {
        let mut patterns_with_matches: BTreeMap<String, bool> = BTreeMap::new();
        let globs: Vec<Pattern> = ref_specs.iter().try_fold(Vec::new(), |mut accm, pattern| {
            match Pattern::new(pattern) {
                Ok(regex) => {
                    patterns_with_matches.insert(pattern.clone(), false);
                    accm.push(regex);
                    Ok(accm)
                }
                Err(e) => Err(Error::InvalidGlobPattern {
                    pattern: pattern.clone(),
                    error: e.to_string(),
                }),
            }
        })?;

        Ok((
            refs.iter().fold(Vec::new(), |mut accm, ref_name| {
                if let Some(matching_glob) = globs.iter().find_map(|glob_pattern| {
                    if match_ref(glob_pattern, ref_name) {
                        Some(glob_pattern.to_string())
                    } else {
                        None
                    }
                }) {
                    patterns_with_matches.insert(matching_glob, true);
                    accm.push(ref_name.clone());
                };
                accm
            }),
            patterns_with_matches
                .iter()
                .filter_map(|(pattern, had_match)| {
                    if !had_match {
                        Some(pattern.clone())
                    } else {
                        None
                    }
                })
                .collect(),
        ))
    } else {
        Ok((refs.to_vec(), vec![]))
    }
}

/// Push a git based mirror to the downstream server
///
/// TODO: Add option to use libgit2 for push in addition to pull on git mirrors.
#[allow(deprecated)]
async fn push_to_mirror_server(
    lorry_spec: &SingleLorry,
    lorry_name: &str,
    url_to_push_to: &Url,
    active_repo: &workspace::Workspace,
    arguments: &arguments::Arguments,
) -> Result<MirrorStatus, Error> {
    if url_to_push_to.scheme() == "file" {
        let url_to_push_to = url_to_push_to
            .to_file_path()
            .map_err(|_| Error::InvalidPath(url_to_push_to.to_string()))?;
        // This repo is bieng pushed to the local filesystem make sure it
        // exists first
        if !url_to_push_to.exists() {
            // TODO: Support non-gitlab
            tracing::info!("Creating local repository: {:?}", url_to_push_to);
            unimplemented!("cannot support local mirror pushing yet");
        }
    }

    tracing::debug!("Pushing {} to mirror at {:?}", lorry_name, &url_to_push_to);

    let repository = Repository::open(active_repo.repository_path())?;
    let ref_names = repository
        .references()?
        .try_fold(Vec::new(), |mut accm, reference| {
            let ref_name = match reference {
                Ok(ref_name) => {
                    let name = ref_name.name().unwrap();
                    name.to_string()
                }
                Err(err) => return Err(err),
            };
            accm.push(ref_name);
            Ok(accm)
        })?;

    let (refs, missing) = parse_refs(
        ref_names.as_slice(),
        lorry_spec.ref_patterns.as_deref(),
        lorry_spec.ignore_patterns.as_deref(),
    )?;
    tracing::info!("pushing {} refs", refs.len());

    if refs.is_empty() {
        return Err(Error::NoMatchingRefspecs);
    }

    let repository_path = active_repo.repository_path();
    let push = &crate::push::Push {
        url: url_to_push_to,
        ref_names: refs.iter().map(|s| s.as_str()).collect(),
        config_path: &arguments.git_config_path,
    };

    let push_refs = match execute(push, repository_path.as_path()).await {
        Ok((stdout, _)) => get_refs(refs.len(), &stdout),
        Err(err) => match err {
            ExecutionError::IO { command, source } => {
                tracing::warn!("Command failed to spawn: {:?}", source.to_string());
                Err(Error::Io {
                    command: command.clone(),
                })
            }
            ExecutionError::CommandError {
                command: _,
                status: _,
                stderr: _,
                stdout,
            } => get_refs(refs.len(), &stdout),
        },
    }?;

    Ok(MirrorStatus {
        push_refs,
        warnings: missing
            .iter()
            .map(|pattern| Warning::NonMatchingRefSpecs {
                pattern: pattern.clone(),
            })
            .collect(),
    })
}

/// Attempt to mirror a repository from an upstream host into the configured
/// downstream server.
///
/// NOTE: The only supported downstream configuration is currently Gitlab.
///
/// Rough outline of the order of operations is:
///
/// * Attempt to fetch repository data from the downstream into the working
///     directory of the local mirror.
///
/// * Fetch the upstream repository on top of the working directory of the
///     local mirror applying new updates.
///
/// * Push the local mirror back into the downstream updating it with new refs.
///
/// Fetching the downstream repository initially ensures that Lorry can work
/// in a stateless environment and will consider the downstream mirror the
/// source of truth.
///
/// This workflow applies to both LFS and normal Git mirrors. In the case of
/// LFS the raw files are also downloaded into the working directory prior to
/// running download operations.
///
/// In the event that the downstream repository becomes corrupted in someway
/// the procedure is to delete and reinitialize an empty repository in which
/// case the upstream mirror will be re-imported from scratch.
///
/// TODO: This code can be factored out better
pub async fn try_mirror(
    lorry_details: &LorrySpec,
    lorry_name: &str,
    downstream_url: &url::Url,
    workspace: &workspace::Workspace,
    arguments: &Arguments,
) -> Result<MirrorStatus, Error> {
    match lorry_details {
        LorrySpec::Git(single_lorry) => {
            tracing::info!("Ensuring local mirror is consistent with downstream");
            crate::fetch::Fetch {
                git_repo: workspace,
                target_url: downstream_url,
                use_git_binary: arguments.use_git_binary,
                git_config_path: arguments.git_config_path.as_path(),
            }
            .fetch()
            .await?;
            tracing::info!("Fetching upstream repository into local mirror");
            crate::fetch::Fetch {
                git_repo: workspace,
                target_url: &url::Url::parse(single_lorry.url.as_str()).unwrap(), // FIXME
                use_git_binary: arguments.use_git_binary,
                git_config_path: arguments.git_config_path.as_path(),
            }
            .fetch()
            .await?;

            push_to_mirror_server(
                single_lorry,
                lorry_name,
                downstream_url,
                workspace,
                arguments,
            )
            .await
        }
        LorrySpec::RawFiles(raw_files) => {
            // check for missing sha256sums if configuration disallows them
            if arguments.sha256sum_required {
                let missing_sha256sums = raw_files.missing_sha256sums();
                if !missing_sha256sums.is_empty() {
                    let mut message = String::default();
                    missing_sha256sums
                        .iter()
                        .for_each(|url| message.push_str(&format!("{}\n", url)));
                    return Err(Error::Sha256sumsNotSpecified(message));
                }
            }
            tracing::info!("Fetching raw files from downstream to ensure consistency");
            let mut lfs_url = downstream_url.clone();
            lfs_url.set_path(&format!("{}/info/lfs", lfs_url.path()));
            let fetch_err = execute(
                &raw_files::FetchDownstreamRawFiles {
                    url: downstream_url,
                    lfs_url: &lfs_url,
                    worktree: workspace.lfs_data_path().as_path(),
                    config_path: arguments.git_config_path.as_path(),
                },
                &workspace.repository_path(),
            )
            .await;
            // TODO: I would prefer to explicitly detect if the repository
            // exists or not but the code calling the Gitlab API needs work.
            // BUG: Be aware that if the downstream fails for some other reason
            // like the server being down that this will create a new
            // repository and push its contents up causing a rejection error.
            // If this happens then you need to manually delete the mirror
            // directory and on the next run it will properly import the
            // downstream sources.
            if let Err(err) = fetch_err {
                tracing::warn!("Fetch failed but might not be an error: {}", err);
            } else {
                tracing::info!("Local mirror is consistent with downstream");
            }
            let importer = importer::Importer(raw_files.clone());
            let helper = raw_files::Helper(workspace.clone());

            // Initialize the LFS repository if it hasn't been already creating
            // the first commit which enables tracking of everything within.
            helper
                .initial_commit_if_missing(arguments.git_config_path.as_path())
                .await?;

            // Download any new raw files
            let modified = importer
                .ensure(&workspace.lfs_data_path(), arguments)
                .await?;
            if modified {
                // Import new data that was just downloaded
                helper
                    .import_data(arguments.git_config_path.as_path())
                    .await?;
            } else {
                tracing::info!("No new files were added or removed, nothing to do")
            };

            tracing::info!("Synchronizing local raw-file mirror to downstream");
            let mut lfs_url = downstream_url.clone();
            lfs_url.set_path(&format!("{}/info/lfs", lfs_url.path()));
            execute(
                &raw_files::PushRawFiles {
                    url: downstream_url,
                    lfs_url: &lfs_url,
                    worktree: workspace.lfs_data_path().as_path(),
                    config_path: arguments.git_config_path.as_path(),
                },
                workspace.repository_path().as_path(),
            )
            .await?;
            // LFS breaks porcelain git output so we just fake it since we only
            // supporting syncing to a single branch anyway.
            Ok(MirrorStatus {
                push_refs: PushRefs(
                    vec![(
                        DEFAULT_BRANCH_NAME.to_string(),
                        String::from("raw files pushed successfully"),
                    )],
                    vec![],
                ),
                warnings: vec![],
            })
        }
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use std::path::Path;

    use git2::Repository;
    use tempfile::tempdir;

    use crate::git_config::Config;
    use crate::test_server::{spawn_test_server, TestRepo, EXAMPLE_COMMIT};
    use crate::workspace::Workspace;

    #[test]
    fn test_parse_refs() {
        let (refs, missing) = parse_refs(
            &[
                String::from("refs/heads/master"),
                String::from("refs/tags/v1.0.0"),
                String::from("refs/tags/v1.0.1"),
                String::from("refs/tags/v1.0.2-rc1"),
                String::from("some-random-string"),
            ],
            Some(&[
                String::from("refs/heads/ma*"),
                String::from("v1.0.1"),
                String::from("refs/tags/v*"),
                String::from("notgonnamatch"),
            ]),
            // ignore RC releases
            Some(&[
                String::from("refs/tags/*-rc*"),
                String::from("*-rc*"),
                String::from("*ef*"), // Will not match refs/...
            ]),
        )
        .unwrap();
        assert!(refs.iter().any(|key| *key == "refs/heads/master"));
        assert!(refs.iter().any(|key| *key == "refs/tags/v1.0.0"));
        assert!(refs.iter().any(|key| *key == "refs/tags/v1.0.1"));
        assert!(!refs.iter().any(|key| *key == "some-random-string"));
        assert!(!refs.iter().any(|key| *key == "refs/tags/v1.0.2-rc1"));
        assert!(missing.first().unwrap() == "notgonnamatch");
        assert!(missing.len() == 1);
    }

    #[test]
    fn test_parse_refs_conflict() {
        let (refs, missing) = parse_refs(
            &[
                String::from("refs/heads/master"),
                String::from("refs/heads/v1.0.0"),
                String::from("refs/tags/v1.0.0"),
            ],
            Some(&[String::from("v1.0.0")]),
            Some(&[String::from("master")]), // pointless
        )
        .unwrap();
        assert!(refs.len() == 2);
        assert!(refs.iter().any(|key| *key == "refs/tags/v1.0.0"));
        assert!(refs.iter().any(|key| *key == "refs/heads/v1.0.0"));
        assert!(missing.is_empty());
    }

    #[tokio::test]
    async fn test_try_mirror() {
        let upstream_repo = TestRepo((String::from("hello.git"), vec![EXAMPLE_COMMIT.to_string()]));
        let downstream_repo = TestRepo((String::from("hello.git"), vec![]));
        let test_dir = tempdir().unwrap();
        let mirror_workspace_dir = test_dir.path().join("workspace");
        let git_config_path = test_dir.path().join("gitconfig");
        let git_config = Config(git_config_path.to_path_buf());
        git_config
            .setup("hello@example.org", 1, false, Path::new("/dev/null"), None)
            .unwrap();
        let repos_upstream_dir = test_dir.path().join("repos_upstream");
        let repos_downstream_dir = test_dir.path().join("repos_downstream");
        let test_workspace =
            Workspace::new(test_dir.path().join("workspace").as_path(), "test_fetch");
        test_workspace.init_if_missing(false).unwrap();
        // Upstream server
        let upstream_address =
            spawn_test_server(repos_upstream_dir.as_path(), &[upstream_repo.clone()])
                .await
                .unwrap();
        // Downstream server
        let downstream_address =
            spawn_test_server(repos_downstream_dir.as_path(), &[downstream_repo.clone()])
                .await
                .unwrap();

        let workspace = Workspace::new(mirror_workspace_dir.as_path(), "test-repo");
        workspace.init_if_missing(false).unwrap();

        try_mirror(
            &LorrySpec::Git(SingleLorry {
                url: upstream_repo.address(&upstream_address).to_string(),
                ..Default::default()
            }),
            "test_lorry",
            &downstream_repo.address(&downstream_address),
            &workspace,
            &Arguments {
                working_area: test_dir.path().to_path_buf(),
                use_git_binary: Some(true),
                git_config_path,
                ..Default::default()
            },
        )
        .await
        .unwrap();

        // ensure that downstream contains the git from the upstream
        let repository = Repository::open_bare(repos_downstream_dir.join("hello.git")).unwrap();
        let mut walk = repository.revwalk().unwrap();
        walk.push_head().unwrap();
        let last_commit_id = walk.next().unwrap().unwrap();
        let last_commit = repository.find_commit(last_commit_id).unwrap();
        let last_commit_message = last_commit.message().unwrap();
        assert!(last_commit_message == "Test Commit: 1/1")
    }
}