diff --git a/server/src/cli.rs b/server/src/cli.rs index 73dc9f2..c6998eb 100644 --- a/server/src/cli.rs +++ b/server/src/cli.rs @@ -1,4 +1,4 @@ -use crate::{distro::MetaDistroMgr, Config, FsConfig, Global}; +use crate::{Config, FsConfig, Global}; use std::{io, path::PathBuf, sync::Arc}; @@ -6,7 +6,6 @@ use axum::Router; use clap::Parser; use sea_orm_migration::MigratorTrait; use tower_http::trace::TraceLayer; -use tracing::debug; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; #[derive(Parser)] diff --git a/server/src/distro.rs b/server/src/distro.rs deleted file mode 100644 index 22563ff..0000000 --- a/server/src/distro.rs +++ /dev/null @@ -1,70 +0,0 @@ -use crate::{db, DistroMgr}; - -use std::{ - collections::HashMap, - path::{Path, PathBuf}, - sync::Arc, -}; - -use sea_orm::{DbConn, EntityTrait}; -use tokio::sync::Mutex; - -#[derive(Clone)] -pub struct MetaDistroMgr { - distro_dir: PathBuf, - conn: DbConn, - distros: Arc>>>, -} - -impl MetaDistroMgr { - pub async fn new>(distro_dir: P, conn: DbConn) -> crate::Result { - if !tokio::fs::try_exists(&distro_dir).await? { - tokio::fs::create_dir(&distro_dir).await?; - } - - let distro_dir = distro_dir.as_ref().to_path_buf(); - let mut map: HashMap> = HashMap::new(); - - let distros = db::Distro::find().all(&conn).await?; - - for distro in distros { - let mgr = - DistroMgr::new(distro_dir.join(&distro.name), distro.id, conn.clone()).await?; - map.insert(distro.name, Arc::new(mgr)); - } - - Ok(Self { - distro_dir, - conn, - distros: Arc::new(Mutex::new(map)), - }) - } - - pub async fn get_mgr(&self, distro: &str) -> Option> { - let map = self.distros.lock().await; - - map.get(distro).map(|mgr| Arc::clone(mgr)) - } - - pub async fn get_or_create_mgr(&self, distro: &str) -> crate::Result> { - let mut map = self.distros.lock().await; - - if let Some(mgr) = map.get(distro) { - Ok(Arc::clone(mgr)) - } else { - let distro = db::query::distro::insert(&self.conn, distro, None).await?; - - let mgr = Arc::new( - DistroMgr::new( - self.distro_dir.join(&distro.name), - distro.id, - self.conn.clone(), - ) - .await?, - ); - map.insert(distro.name, Arc::clone(&mgr)); - - Ok(mgr) - } - } -} diff --git a/server/src/main.rs b/server/src/main.rs index f7e1a95..eb1c3d0 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -2,16 +2,15 @@ mod api; mod cli; mod config; pub mod db; -mod distro; mod error; mod repo; pub use config::{Config, DbConfig, FsConfig}; pub use error::{Result, ServerError}; -use repo::DistroMgr; + +use std::sync::Arc; use clap::Parser; -use std::{path::PathBuf, sync::Arc}; pub const ANY_ARCH: &'static str = "any"; diff --git a/server/src/repo/manager.rs b/server/src/repo/manager.rs index 4bf2378..e4f0581 100644 --- a/server/src/repo/manager.rs +++ b/server/src/repo/manager.rs @@ -1,133 +1,107 @@ use super::{archive, package}; -use crate::{db, error::Result}; +use crate::db::{self, query::package::delete_stale_pkgs}; use std::{ collections::HashMap, path::{Path, PathBuf}, sync::{ - atomic::{AtomicBool, AtomicU32, Ordering}, + atomic::{AtomicU32, Ordering}, Arc, }, }; use futures::StreamExt; use sea_orm::{ - ActiveModelTrait, ColumnTrait, DbConn, EntityTrait, ModelTrait, QueryFilter, QuerySelect, - Related, RelationTrait, Set, TransactionTrait, + ActiveModelTrait, ColumnTrait, Condition, ConnectionTrait, DbConn, EntityTrait, JoinType, + ModelTrait, NotSet, QueryFilter, QuerySelect, Related, RelationTrait, Set, TransactionTrait, }; -use sea_query::{Expr, Query}; -use tokio::{ - io::AsyncRead, - sync::{Mutex, Notify, RwLock, Semaphore}, +use sea_query::{Alias, Expr, Query}; +use tokio::sync::{ + mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}, + Mutex, RwLock, }; use uuid::Uuid; -pub const ANY_ARCH: &'static str = "any"; -pub const REPOS_DIR: &'static str = "repos"; - -#[derive(Default)] -pub struct RepoState { - queued_pkgs: AtomicU32, - sync_queued: AtomicBool, - sync_notify: Notify, +struct PkgQueueMsg { + repo: i32, + path: PathBuf, } -pub struct DistroMgr { - distro_dir: PathBuf, - distro_id: i32, +/// A single instance of this struct orchestrates everything related to managing packages files on +/// disk for all repositories in the server +pub struct RepoMgr { + repos_dir: PathBuf, conn: DbConn, - repos: RwLock>>, - sync_lock: Mutex<()>, - pkg_sema: Semaphore, + pkg_queue: ( + UnboundedSender, + Mutex>, + ), + repos: RwLock>)>>, } -impl DistroMgr { - pub async fn new>(distro_dir: P, distro_id: i32, conn: DbConn) -> Result { - if !tokio::fs::try_exists(&distro_dir).await? { - tokio::fs::create_dir(&distro_dir).await?; +impl RepoMgr { + pub async fn new>(repos_dir: P, conn: DbConn) -> crate::Result { + if !tokio::fs::try_exists(&repos_dir).await? { + tokio::fs::create_dir(&repos_dir).await?; } - let repos_dir = distro_dir.as_ref().join(REPOS_DIR); + let (tx, rx) = unbounded_channel(); - if !tokio::fs::try_exists(&repos_dir).await? { - tokio::fs::create_dir(repos_dir).await?; + let mut repos = HashMap::new(); + let repo_ids: Vec = db::Repo::find() + .select_only() + .column(db::repo::Column::Id) + .into_tuple() + .all(&conn) + .await?; + + for id in repo_ids { + repos.insert(id, Default::default()); } Ok(Self { - distro_dir: distro_dir.as_ref().to_path_buf(), - distro_id, + repos_dir: repos_dir.as_ref().to_path_buf(), conn, - sync_lock: Mutex::new(()), - pkg_sema: Semaphore::new(1), - repos: RwLock::new(HashMap::new()), + pkg_queue: (tx, Mutex::new(rx)), + repos: RwLock::new(repos), }) } - pub async fn schedule_sync(&self, repo_id: i32) -> Result<()> { - let state = { - let repos = self.repos.read().await; - repos.get(&repo_id).map(Arc::clone) - }; - - if state.is_none() { - tracing::debug!("is none"); - return Ok(()); - } - - let state = state.unwrap(); - - let res = - state - .sync_queued - .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst); - - // Already a sync job scheduled, so this one can simply quit - if res.is_err() { - tracing::debug!("shit"); - return Ok(()); - } - - // If the queue is not empty, we wait for a notification that it is before syncing - if state.queued_pkgs.load(Ordering::SeqCst) > 0 { - tracing::debug!("sync waiter waiting"); - state.sync_notify.notified().await; - tracing::debug!("sync waiter notified"); - } - - self.sync_repo(repo_id).await - } - /// Generate archive databases for all known architectures in the repository, including the /// "any" architecture. - pub async fn sync_repo(&self, repo_id: i32) -> Result<()> { - let _guard = self.sync_lock.lock().await; + pub async fn sync_repo(&self, repo: i32) -> crate::Result<()> { + let lock = self + .repos + .read() + .await + .get(&repo) + .map(|(_, lock)| Arc::clone(lock)); - let repo = crate::db::query::repo::by_id(&self.conn, repo_id).await?; - - if repo.is_none() { + if lock.is_none() { return Ok(()); } - let repo = repo.unwrap(); + let lock = lock.unwrap(); + let _guard = lock.lock().await; - let mut archs = repo - .find_related(db::Package) + let archs: Vec = db::Package::find() + .filter(db::package::Column::RepoId.eq(repo)) .select_only() .column(db::package::Column::Arch) .distinct() - .into_tuple::() - .stream(&self.conn) + .into_tuple() + .all(&self.conn) .await?; - while let Some(arch) = archs.next().await.transpose()? { - self.generate_archives(&repo, &arch).await?; + for arch in archs { + self.generate_archives(repo, &arch).await?; } Ok(()) } /// Generate the archive databases for the given repository and architecture. - async fn generate_archives(&self, repo: &db::repo::Model, arch: &str) -> Result<()> { + async fn generate_archives(&self, repo: i32, arch: &str) -> crate::Result<()> { let [tmp_ar_db_path, tmp_ar_files_path, files_tmp_file_path, desc_tmp_file_path] = self.random_file_paths(); let ar_db = archive::RepoArchiveWriter::open(&tmp_ar_db_path).await?; @@ -135,18 +109,7 @@ impl DistroMgr { // Query all packages in the repo that have the given architecture or the "any" // architecture - let mut pkgs = repo - .find_related(crate::db::Package) - .filter(db::package::Column::Arch.is_in([arch, ANY_ARCH])) - .filter( - db::package::Column::Id.in_subquery( - Query::select() - .expr(db::package::Column::Id.max()) - .from(db::package::Entity) - .group_by_columns([db::package::Column::Arch, db::package::Column::Name]) - .to_owned(), - ), - ) + let mut pkgs = db::query::package::pkgs_to_sync(&self.conn, repo, arch) .stream(&self.conn) .await?; @@ -178,7 +141,7 @@ impl DistroMgr { ar_db.close().await?; ar_files.close().await?; - let repo_dir = self.distro_dir.join(&repo.name); + let repo_dir = self.repos_dir.join(repo.to_string()); // Move the db archives to their respective places tokio::fs::rename(tmp_ar_db_path, repo_dir.join(format!("{}.db.tar.gz", arch))).await?; @@ -205,225 +168,218 @@ impl DistroMgr { let _ = tokio::fs::remove_file(desc_tmp_file_path).await; let _ = tokio::fs::remove_file(files_tmp_file_path).await; - tracing::info!( - "Package archives generated for '{}' ('{}')", - &repo.name, - arch - ); + tracing::info!("Package archives generated for repo {} ('{}')", repo, arch); Ok(()) } - async fn get_or_create_repo(&self, repo: &str) -> Result { + /// Clean any remaining old package files from the database and file system + pub async fn remove_stale_pkgs(&self) -> crate::Result<()> { + let mut pkgs = db::query::package::stale_pkgs(&self.conn) + .stream(&self.conn) + .await?; + + // Ids are monotonically increasing, so the max id suffices to know which packages to + // remove later + let mut max_id = -1; + let mut removed_pkgs = 0; + + while let Some(pkg) = pkgs.next().await.transpose()? { + // Failing to remove the package file isn't the biggest problem + let _ = tokio::fs::remove_file( + self.repos_dir + .join(pkg.repo_id.to_string()) + .join(pkg.id.to_string()), + ) + .await; + + if pkg.id > max_id { + max_id = pkg.id; + } + + removed_pkgs += 1; + } + + if removed_pkgs > 0 { + db::query::package::delete_stale_pkgs(&self.conn, max_id).await?; + } + + tracing::info!("Removed {removed_pkgs} stale package(s)"); + + Ok(()) + } + + pub async fn pkg_parse_task(&self) { + loop { + // Receive the next message and immediately drop the mutex afterwards. As long as the + // quue is empty, this will lock the mutex. This is okay, as the mutex will be unlocked + // as soon as a message is received, so another worker can pick up the mutex. + let msg = { + let mut recv = self.pkg_queue.1.lock().await; + recv.recv().await + }; + + if let Some(msg) = msg { + // TODO better handle this error (retry if failure wasn't because the package is + // faulty) + let _ = self + .add_pkg_from_path(msg.path, msg.repo) + .await + .inspect_err(|e| tracing::error!("{:?}", e)); + + let old = self + .repos + .read() + .await + .get(&msg.repo) + .map(|n| n.0.fetch_sub(1, Ordering::SeqCst)); + + // Every time the queue for a repo becomes empty, we run a sync job + if old == Some(1) { + // TODO error handling + let _ = self.sync_repo(msg.repo).await; + + // TODO move this so that we only clean if entire queue is empty, not just + // queue for specific repo + let _ = self.remove_stale_pkgs().await; + } + } + } + } + + pub async fn queue_pkg(&self, repo: i32, path: PathBuf) { + self.pkg_queue.0.send(PkgQueueMsg { path, repo }).unwrap(); + self.repos.read().await.get(&repo).inspect(|n| { + n.0.fetch_add(1, Ordering::SeqCst); + }); + } + + pub async fn get_repo(&self, distro: &str, repo: &str) -> crate::Result> { + Ok(db::Repo::find() + .find_also_related(db::Distro) + .filter( + Condition::all() + .add(db::repo::Column::Name.eq(repo)) + .add(db::distro::Column::Name.eq(distro)), + ) + .one(&self.conn) + .await + .map(|res| res.map(|(repo, _)| repo.id))?) + } + + pub async fn get_or_create_repo(&self, distro: &str, repo: &str) -> crate::Result { let mut repos = self.repos.write().await; - if let Some(repo) = db::query::repo::by_name(&self.conn, repo).await? { - Ok(repo) - } else { - tokio::fs::create_dir(self.distro_dir.join(repo)).await?; - let repo = db::query::repo::insert(&self.conn, self.distro_id, repo, None).await?; - - repos.insert(repo.id, Arc::new(RepoState::default())); - - Ok(repo) - } - } - - /// Remove the repo with the given name, if it existed - pub async fn remove_repo(&self, repo: &str) -> Result { - let res = db::query::repo::by_name(&self.conn, repo).await?; - - if let Some(repo_entry) = res { - // Remove repository from database - repo_entry.delete(&self.conn).await?; - - // Remove files from file system - tokio::fs::remove_dir_all(self.distro_dir.join(repo)).await?; - - Ok(true) - } else { - Ok(false) - } - } - - /// Remove all packages from the repository with the given arch. - pub async fn remove_repo_arch(&self, repo: &str, arch: &str) -> Result { - let repo = db::query::repo::by_name(&self.conn, repo).await?; - - if let Some(repo) = repo { - let mut pkgs = repo - .find_related(db::Package) - .filter(db::package::Column::Arch.eq(arch)) - .stream(&self.conn) - .await?; - - while let Some(pkg) = pkgs.next().await.transpose()? { - let path = self - .distro_dir - .join(&repo.name) - .join(super::package::filename(&pkg)); - tokio::fs::remove_file(path).await?; - - pkg.delete(&self.conn).await?; - } - - tokio::fs::remove_file( - self.distro_dir - .join(&repo.name) - .join(format!("{}.db.tar.gz", arch)), - ) - .await?; - tokio::fs::remove_file( - self.distro_dir - .join(&repo.name) - .join(format!("{}.files.tar.gz", arch)), - ) + let distro_id: Option = db::Distro::find() + .filter(db::distro::Column::Name.eq(distro)) + .select_only() + .column(db::distro::Column::Id) + .into_tuple() + .one(&self.conn) .await?; - // If we removed all "any" packages, we need to resync all databases - if arch == ANY_ARCH { - //self.sync_repo(&repo.name).await?; - } - - Ok(true) + let distro_id = if let Some(id) = distro_id { + id } else { - Ok(false) - } + let new_distro = db::distro::ActiveModel { + id: NotSet, + name: Set(distro.to_string()), + description: NotSet, + }; + + new_distro.insert(&self.conn).await?.id + }; + + let repo_id: Option = db::Repo::find() + .filter(db::repo::Column::DistroId.eq(distro_id)) + .filter(db::repo::Column::Name.eq(repo)) + .select_only() + .column(db::repo::Column::Id) + .into_tuple() + .one(&self.conn) + .await?; + + let repo_id = if let Some(id) = repo_id { + id + } else { + let new_repo = db::repo::ActiveModel { + id: NotSet, + distro_id: Set(distro_id), + name: Set(repo.to_string()), + description: NotSet, + }; + let id = new_repo.insert(&self.conn).await?.id; + + tokio::fs::create_dir(self.repos_dir.join(id.to_string())).await?; + repos.insert(id, Default::default()); + + id + }; + + Ok(repo_id) } - pub async fn remove_pkg(&self, repo: &str, arch: &str, name: &str) -> Result { - let repo = db::query::repo::by_name(&self.conn, repo).await?; - - if let Some(repo) = repo { - let pkg = - db::query::package::by_fields(&self.conn, repo.id, arch, name, None, None).await?; - - if let Some(pkg) = pkg { - // Remove package from database & file system - tokio::fs::remove_file( - self.distro_dir - .join(&repo.name) - .join(super::package::filename(&pkg)), - ) - .await?; - pkg.delete(&self.conn).await?; - - //if arch == ANY_ARCH { - // self.sync_repo(&repo.name).await?; - //} else { - // self.generate_archives(&repo.name, arch).await?; - //} - - Ok(true) - } else { - Ok(false) - } - } else { - Ok(false) - } - } - - async fn _add_pkg_from_path>( - &self, - path: P, - repo: &db::repo::Model, - ) -> crate::Result { + async fn add_pkg_from_path>(&self, path: P, repo: i32) -> crate::Result<()> { let path_clone = path.as_ref().to_path_buf(); let pkg = tokio::task::spawn_blocking(move || package::Package::open(path_clone)) .await .unwrap()?; // TODO prevent database from being updated but file failing to move to repo dir? - let pkg = db::query::package::insert(&self.conn, repo.id, pkg).await?; + let pkg = db::query::package::insert(&self.conn, repo, pkg).await?; - let queue_path = self.distro_dir.join(&repo.name).join(pkg.id.to_string()); - tokio::fs::rename(path.as_ref(), queue_path).await?; + let dest_path = self + .repos_dir + .join(repo.to_string()) + .join(pkg.id.to_string()); + tokio::fs::rename(path.as_ref(), dest_path).await?; tracing::info!( - "Added '{}-{}' to repository '{}' ({})", + "Added '{}-{}-{}' to repository {}", pkg.name, pkg.version, - repo.name, - pkg.arch + pkg.arch, + repo, ); - Ok(pkg) + Ok(()) } - pub async fn add_pkg_from_path>( - &self, - path: P, - repo: &str, - ) -> crate::Result<(i32, String, String, String)> { - let repo = self.get_or_create_repo(repo).await?; + pub async fn remove_repo(&self, repo: i32) -> crate::Result<()> { + self.repos.write().await.remove(&repo); + db::Repo::delete_by_id(repo).exec(&self.conn).await?; + let _ = tokio::fs::remove_dir_all(self.repos_dir.join(repo.to_string())).await; - { - let repos = self.repos.read().await; - - if let Some(state) = repos.get(&repo.id) { - state.queued_pkgs.fetch_add(1, Ordering::SeqCst); - } - } - - let _guard = self.pkg_sema.acquire().await.unwrap(); - let res = self._add_pkg_from_path(path, &repo).await; - - match res { - Ok(pkg) => { - let repos = self.repos.read().await; - - if let Some(state) = repos.get(&repo.id) { - let old = state.queued_pkgs.fetch_sub(1, Ordering::SeqCst); - - if old - 1 == 0 { - state.sync_notify.notify_one(); - } - } - - Ok((repo.id, pkg.name, pkg.version, pkg.arch)) - } - Err(e) => Err(e), - } - - // If the package already exists in the database, we remove it first - //let res = db::query::package::by_fields( - // &self.conn, - // repo.id, - // &pkg.info.arch, - // &pkg.info.name, - // None, - // None, - //) - //.await?; - // - //if let Some(entry) = res { - // entry.delete(&self.conn).await?; - //} - - //let dest_pkg_path = repo_dir.join(pkg.file_name()); - // - //// Insert new package into database - //let name = pkg.info.name.clone(); - //let version = pkg.info.version.clone(); - //let arch = pkg.info.arch.clone(); - //db::query::package::insert(&self.conn, repo.id, pkg).await?; - // - //// Move the package to its final resting place - //tokio::fs::rename(tmp_file_path, dest_pkg_path).await?; - // - // Synchronize archive databases - //if arch == ANY_ARCH { - // self.generate_archives_all(&repo.name).await?; - //} else { - // self.generate_archives(&repo.name, &arch).await?; - //} + Ok(()) + } + + /// Remove all packages in the repository that have a given arch. This method marks all + /// packages with the given architecture as "pending deletion", before performing a manual sync + /// & removal of stale packages. + pub async fn remove_repo_arch(&self, repo: i32, arch: &str) -> crate::Result<()> { + db::Package::update_many() + .col_expr( + db::package::Column::State, + Expr::value(db::PackageState::PendingDeletion), + ) + .filter( + Condition::all() + .add(db::package::Column::RepoId.eq(repo)) + .add(db::package::Column::Arch.eq(arch)), + ) + .exec(&self.conn) + .await?; + + self.sync_repo(repo).await?; + self.remove_stale_pkgs().await?; + + Ok(()) } - /// Generate a path to a unique file that can be used as a temporary file pub fn random_file_paths(&self) -> [PathBuf; C] { std::array::from_fn(|_| { let uuid: uuid::fmt::Simple = Uuid::new_v4().into(); - self.distro_dir.join(uuid.to_string()) + self.repos_dir.join(uuid.to_string()) }) } } diff --git a/server/src/repo/manager2.rs b/server/src/repo/manager2.rs deleted file mode 100644 index e4f0581..0000000 --- a/server/src/repo/manager2.rs +++ /dev/null @@ -1,385 +0,0 @@ -use super::{archive, package}; -use crate::db::{self, query::package::delete_stale_pkgs}; - -use std::{ - collections::HashMap, - path::{Path, PathBuf}, - sync::{ - atomic::{AtomicU32, Ordering}, - Arc, - }, -}; - -use futures::StreamExt; -use sea_orm::{ - ActiveModelTrait, ColumnTrait, Condition, ConnectionTrait, DbConn, EntityTrait, JoinType, - ModelTrait, NotSet, QueryFilter, QuerySelect, Related, RelationTrait, Set, TransactionTrait, -}; -use sea_query::{Alias, Expr, Query}; -use tokio::sync::{ - mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}, - Mutex, RwLock, -}; -use uuid::Uuid; - -struct PkgQueueMsg { - repo: i32, - path: PathBuf, -} - -/// A single instance of this struct orchestrates everything related to managing packages files on -/// disk for all repositories in the server -pub struct RepoMgr { - repos_dir: PathBuf, - conn: DbConn, - pkg_queue: ( - UnboundedSender, - Mutex>, - ), - repos: RwLock>)>>, -} - -impl RepoMgr { - pub async fn new>(repos_dir: P, conn: DbConn) -> crate::Result { - if !tokio::fs::try_exists(&repos_dir).await? { - tokio::fs::create_dir(&repos_dir).await?; - } - - let (tx, rx) = unbounded_channel(); - - let mut repos = HashMap::new(); - let repo_ids: Vec = db::Repo::find() - .select_only() - .column(db::repo::Column::Id) - .into_tuple() - .all(&conn) - .await?; - - for id in repo_ids { - repos.insert(id, Default::default()); - } - - Ok(Self { - repos_dir: repos_dir.as_ref().to_path_buf(), - conn, - pkg_queue: (tx, Mutex::new(rx)), - repos: RwLock::new(repos), - }) - } - - /// Generate archive databases for all known architectures in the repository, including the - /// "any" architecture. - pub async fn sync_repo(&self, repo: i32) -> crate::Result<()> { - let lock = self - .repos - .read() - .await - .get(&repo) - .map(|(_, lock)| Arc::clone(lock)); - - if lock.is_none() { - return Ok(()); - } - - let lock = lock.unwrap(); - let _guard = lock.lock().await; - - let archs: Vec = db::Package::find() - .filter(db::package::Column::RepoId.eq(repo)) - .select_only() - .column(db::package::Column::Arch) - .distinct() - .into_tuple() - .all(&self.conn) - .await?; - - for arch in archs { - self.generate_archives(repo, &arch).await?; - } - - Ok(()) - } - - /// Generate the archive databases for the given repository and architecture. - async fn generate_archives(&self, repo: i32, arch: &str) -> crate::Result<()> { - let [tmp_ar_db_path, tmp_ar_files_path, files_tmp_file_path, desc_tmp_file_path] = - self.random_file_paths(); - let ar_db = archive::RepoArchiveWriter::open(&tmp_ar_db_path).await?; - let ar_files = archive::RepoArchiveWriter::open(&tmp_ar_files_path).await?; - - // Query all packages in the repo that have the given architecture or the "any" - // architecture - let mut pkgs = db::query::package::pkgs_to_sync(&self.conn, repo, arch) - .stream(&self.conn) - .await?; - - let mut commited_ids: Vec = Vec::new(); - - while let Some(pkg) = pkgs.next().await.transpose()? { - commited_ids.push(pkg.id); - - let mut files_tmp_file = tokio::fs::File::create(&files_tmp_file_path).await?; - let mut desc_tmp_file = tokio::fs::File::create(&desc_tmp_file_path).await?; - - package::write_files(&self.conn, &mut files_tmp_file, &pkg).await?; - package::write_desc(&self.conn, &mut desc_tmp_file, &pkg).await?; - - let full_name = format!("{}-{}", pkg.name, pkg.version); - - ar_db - .add_entry(&full_name, &desc_tmp_file_path, true) - .await?; - ar_files - .add_entry(&full_name, &desc_tmp_file_path, true) - .await?; - ar_files - .add_entry(&full_name, &files_tmp_file_path, false) - .await?; - } - - // Cleanup - ar_db.close().await?; - ar_files.close().await?; - - let repo_dir = self.repos_dir.join(repo.to_string()); - - // Move the db archives to their respective places - tokio::fs::rename(tmp_ar_db_path, repo_dir.join(format!("{}.db.tar.gz", arch))).await?; - tokio::fs::rename( - tmp_ar_files_path, - repo_dir.join(format!("{}.files.tar.gz", arch)), - ) - .await?; - - // Only after we have successfully written everything to disk do we update the database. - // This order ensures any failure can be recovered, as the database is our single source of - // truth. - db::Package::update_many() - .col_expr( - db::package::Column::State, - Expr::value(db::PackageState::Committed), - ) - .filter(db::package::Column::Id.is_in(commited_ids)) - .exec(&self.conn) - .await?; - - // If this fails there's no point in failing the function + if there were no packages in - // the repo, this fails anyway because the temp file doesn't exist - let _ = tokio::fs::remove_file(desc_tmp_file_path).await; - let _ = tokio::fs::remove_file(files_tmp_file_path).await; - - tracing::info!("Package archives generated for repo {} ('{}')", repo, arch); - - Ok(()) - } - - /// Clean any remaining old package files from the database and file system - pub async fn remove_stale_pkgs(&self) -> crate::Result<()> { - let mut pkgs = db::query::package::stale_pkgs(&self.conn) - .stream(&self.conn) - .await?; - - // Ids are monotonically increasing, so the max id suffices to know which packages to - // remove later - let mut max_id = -1; - let mut removed_pkgs = 0; - - while let Some(pkg) = pkgs.next().await.transpose()? { - // Failing to remove the package file isn't the biggest problem - let _ = tokio::fs::remove_file( - self.repos_dir - .join(pkg.repo_id.to_string()) - .join(pkg.id.to_string()), - ) - .await; - - if pkg.id > max_id { - max_id = pkg.id; - } - - removed_pkgs += 1; - } - - if removed_pkgs > 0 { - db::query::package::delete_stale_pkgs(&self.conn, max_id).await?; - } - - tracing::info!("Removed {removed_pkgs} stale package(s)"); - - Ok(()) - } - - pub async fn pkg_parse_task(&self) { - loop { - // Receive the next message and immediately drop the mutex afterwards. As long as the - // quue is empty, this will lock the mutex. This is okay, as the mutex will be unlocked - // as soon as a message is received, so another worker can pick up the mutex. - let msg = { - let mut recv = self.pkg_queue.1.lock().await; - recv.recv().await - }; - - if let Some(msg) = msg { - // TODO better handle this error (retry if failure wasn't because the package is - // faulty) - let _ = self - .add_pkg_from_path(msg.path, msg.repo) - .await - .inspect_err(|e| tracing::error!("{:?}", e)); - - let old = self - .repos - .read() - .await - .get(&msg.repo) - .map(|n| n.0.fetch_sub(1, Ordering::SeqCst)); - - // Every time the queue for a repo becomes empty, we run a sync job - if old == Some(1) { - // TODO error handling - let _ = self.sync_repo(msg.repo).await; - - // TODO move this so that we only clean if entire queue is empty, not just - // queue for specific repo - let _ = self.remove_stale_pkgs().await; - } - } - } - } - - pub async fn queue_pkg(&self, repo: i32, path: PathBuf) { - self.pkg_queue.0.send(PkgQueueMsg { path, repo }).unwrap(); - self.repos.read().await.get(&repo).inspect(|n| { - n.0.fetch_add(1, Ordering::SeqCst); - }); - } - - pub async fn get_repo(&self, distro: &str, repo: &str) -> crate::Result> { - Ok(db::Repo::find() - .find_also_related(db::Distro) - .filter( - Condition::all() - .add(db::repo::Column::Name.eq(repo)) - .add(db::distro::Column::Name.eq(distro)), - ) - .one(&self.conn) - .await - .map(|res| res.map(|(repo, _)| repo.id))?) - } - - pub async fn get_or_create_repo(&self, distro: &str, repo: &str) -> crate::Result { - let mut repos = self.repos.write().await; - - let distro_id: Option = db::Distro::find() - .filter(db::distro::Column::Name.eq(distro)) - .select_only() - .column(db::distro::Column::Id) - .into_tuple() - .one(&self.conn) - .await?; - - let distro_id = if let Some(id) = distro_id { - id - } else { - let new_distro = db::distro::ActiveModel { - id: NotSet, - name: Set(distro.to_string()), - description: NotSet, - }; - - new_distro.insert(&self.conn).await?.id - }; - - let repo_id: Option = db::Repo::find() - .filter(db::repo::Column::DistroId.eq(distro_id)) - .filter(db::repo::Column::Name.eq(repo)) - .select_only() - .column(db::repo::Column::Id) - .into_tuple() - .one(&self.conn) - .await?; - - let repo_id = if let Some(id) = repo_id { - id - } else { - let new_repo = db::repo::ActiveModel { - id: NotSet, - distro_id: Set(distro_id), - name: Set(repo.to_string()), - description: NotSet, - }; - let id = new_repo.insert(&self.conn).await?.id; - - tokio::fs::create_dir(self.repos_dir.join(id.to_string())).await?; - repos.insert(id, Default::default()); - - id - }; - - Ok(repo_id) - } - - async fn add_pkg_from_path>(&self, path: P, repo: i32) -> crate::Result<()> { - let path_clone = path.as_ref().to_path_buf(); - let pkg = tokio::task::spawn_blocking(move || package::Package::open(path_clone)) - .await - .unwrap()?; - - // TODO prevent database from being updated but file failing to move to repo dir? - let pkg = db::query::package::insert(&self.conn, repo, pkg).await?; - - let dest_path = self - .repos_dir - .join(repo.to_string()) - .join(pkg.id.to_string()); - tokio::fs::rename(path.as_ref(), dest_path).await?; - - tracing::info!( - "Added '{}-{}-{}' to repository {}", - pkg.name, - pkg.version, - pkg.arch, - repo, - ); - - Ok(()) - } - - pub async fn remove_repo(&self, repo: i32) -> crate::Result<()> { - self.repos.write().await.remove(&repo); - db::Repo::delete_by_id(repo).exec(&self.conn).await?; - let _ = tokio::fs::remove_dir_all(self.repos_dir.join(repo.to_string())).await; - - Ok(()) - } - - /// Remove all packages in the repository that have a given arch. This method marks all - /// packages with the given architecture as "pending deletion", before performing a manual sync - /// & removal of stale packages. - pub async fn remove_repo_arch(&self, repo: i32, arch: &str) -> crate::Result<()> { - db::Package::update_many() - .col_expr( - db::package::Column::State, - Expr::value(db::PackageState::PendingDeletion), - ) - .filter( - Condition::all() - .add(db::package::Column::RepoId.eq(repo)) - .add(db::package::Column::Arch.eq(arch)), - ) - .exec(&self.conn) - .await?; - - self.sync_repo(repo).await?; - self.remove_stale_pkgs().await?; - - Ok(()) - } - - pub fn random_file_paths(&self) -> [PathBuf; C] { - std::array::from_fn(|_| { - let uuid: uuid::fmt::Simple = Uuid::new_v4().into(); - self.repos_dir.join(uuid.to_string()) - }) - } -} diff --git a/server/src/repo/mod.rs b/server/src/repo/mod.rs index 16c62a5..953b631 100644 --- a/server/src/repo/mod.rs +++ b/server/src/repo/mod.rs @@ -1,10 +1,8 @@ mod archive; mod manager; -mod manager2; pub mod package; -pub use manager::DistroMgr; -pub use manager2::RepoMgr; +pub use manager::RepoMgr; use crate::FsConfig;