use std::collections::HashMap;
use std::fmt::Write as _;
use std::fs::File;
use std::io::{BufWriter, Read as _, Seek as _, Write};
use std::sync::Arc;
use std::time::Duration;
use axum::Json;
use axum::body::Body;
use axum::extract::{Path, Query, State};
use axum::http::{HeaderMap, StatusCode, header};
use axum::response::{IntoResponse, Response};
use bytes::Bytes;
use kellnr_common::background_tasks::BackgroundTasks;
use kellnr_common::binary_body::BinaryBody;
use kellnr_db::error::DbError;
use kellnr_db::{ChannelInfo, ToolchainTargetInfo, ToolchainTargetStatus, ToolchainWithTargets};
use kellnr_http::error::{ApiError, ApiResult};
use kellnr_http::identity::AdminUser;
use kellnr_http::state::{BackgroundTasksState, DbState, SettingsState, ToolchainStorageState};
use kellnr_storage::storage_error::StorageError;
use kellnr_storage::storage_key::StorageKey;
use kellnr_storage::toolchain_storage::{ToolchainArchiveId, ToolchainStorage};
use serde::{Deserialize, Serialize};
use tokio::sync::Semaphore;
use tracing::trace;
use utoipa::ToSchema;
#[derive(Debug, Serialize, Deserialize, ToSchema)]
pub struct ToolchainResponse {
pub success: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
}
#[derive(Debug, Serialize, Deserialize, ToSchema)]
pub struct SetChannelRequest {
pub name: String,
pub version: String,
}
#[derive(Debug, Deserialize, ToSchema, utoipa::IntoParams)]
pub struct UploadQuery {
pub name: String,
pub version: String,
pub target: String,
pub date: String,
#[serde(default)]
pub channel: Option<String>,
}
fn internal_error(message: &str, error: impl std::fmt::Display) -> ApiError {
tracing::error!("{message}: {error}");
ApiError::new(message, StatusCode::INTERNAL_SERVER_ERROR)
}
fn storage_not_configured() -> ApiError {
ApiError::new(
"Toolchain storage not configured",
StatusCode::SERVICE_UNAVAILABLE,
)
}
#[utoipa::path(
get,
path = "/",
tag = "toolchains",
responses(
(status = 200, description = "List of toolchains", body = Vec<ToolchainWithTargets>),
(status = 401, description = "Not authenticated"),
(status = 403, description = "Admin access required")
),
security(("session_cookie" = []))
)]
pub async fn list_toolchains(
_user: AdminUser,
State(db): DbState,
) -> ApiResult<Json<Vec<ToolchainWithTargets>>> {
trace!("Listing toolchains");
db.list_toolchains()
.await
.map(Json)
.map_err(|e| internal_error("Failed to load toolchains", e))
}
#[utoipa::path(
put,
path = "/",
tag = "toolchains",
params(UploadQuery),
request_body(content = inline(BinaryBody), content_type = "application/octet-stream", description = "Combined toolchain archive (.tar.xz)"),
responses(
(status = 200, description = "Toolchain uploaded", body = ToolchainResponse),
(status = 401, description = "Not authenticated"),
(status = 403, description = "Admin access required"),
(status = 409, description = "Target or archive already exists"),
(status = 413, description = "Archive is too large"),
(status = 503, description = "Storage not configured")
),
security(("session_cookie" = []))
)]
pub async fn upload_toolchain(
_user: AdminUser,
State(db): DbState,
State(storage): ToolchainStorageState,
State(tasks): BackgroundTasksState,
Query(params): Query<UploadQuery>,
body: Bytes,
) -> ApiResult<Json<ToolchainResponse>> {
trace!(
name = %params.name,
version = %params.version,
target = %params.target,
"Uploading toolchain"
);
let storage = storage.as_ref().ok_or_else(storage_not_configured)?;
let toolchain = db
.get_toolchain_by_version(¶ms.name, ¶ms.version)
.await
.map_err(|e| internal_error("Failed to load toolchain", e))?;
if let Some(tc) = &toolchain
&& tc.targets.iter().any(|t| t.target == params.target)
{
return Err(ApiError::conflict(format!(
"Target {} already exists for {}-{}",
params.target, params.name, params.version
)));
}
let size = body.len() as i64;
let archive_id = ToolchainArchiveId {
name: ¶ms.name,
version: ¶ms.version,
date: ¶ms.date,
target: ¶ms.target,
};
let (path, hash) = storage.put(&archive_id, body).await.map_err(|e| match e {
StorageError::ToolchainArchiveExists { .. } => ApiError::conflict(e.to_string()),
e => internal_error("Failed to store archive", e),
})?;
let target_id =
match add_target_to_db(&db, toolchain.as_ref(), ¶ms, &path, &hash, size).await {
Ok(id) => id,
Err(e) => {
if let Err(e) = storage.delete(&path).await {
tracing::warn!("Failed to delete archive {path} after failed upload: {e}");
}
return Err(e);
}
};
let bg_storage = storage.clone();
let bg_db = db.clone();
let bg_name = params.name.clone();
let bg_version = params.version.clone();
let bg_target = params.target.clone();
let bg_date = params.date.clone();
let bg_path = path.clone();
spawn_target_processing(&tasks, db.clone(), target_id, async move {
let archive_bytes = {
let mut archive = None;
for attempt in 1..=3 {
match bg_storage.get(&bg_path).await {
Ok(bytes) => {
archive = Some(bytes);
break;
}
Err(e) => {
if attempt == 3 {
tracing::warn!(
"Failed to read archive for component extraction after {attempt} attempts: {e}"
);
if let Err(status_err) = bg_db
.set_target_status(target_id, ToolchainTargetStatus::Failed)
.await
{
tracing::warn!(
"Failed to set target status to failed after archive read error: {status_err}"
);
}
return;
}
tokio::time::sleep(Duration::from_millis(200 * attempt as u64)).await;
}
}
}
let Some(bytes) = archive else {
tracing::warn!("Archive read retry loop finished without archive bytes");
return;
};
bytes
};
let archive_id = ToolchainArchiveId {
name: &bg_name,
version: &bg_version,
date: &bg_date,
target: &bg_target,
};
match extract_and_store_components(
&bg_storage,
&bg_db,
target_id,
&archive_id,
archive_bytes,
)
.await
{
Ok(()) => {
if let Err(e) = bg_db
.set_target_status(target_id, ToolchainTargetStatus::Ready)
.await
{
tracing::warn!("Failed to set target status to ready: {e}");
}
tracing::info!(
"Successfully extracted components for {}-{}-{}",
bg_name,
bg_version,
bg_target
);
}
Err(e) => {
if let Err(status_err) = bg_db
.set_target_status(target_id, ToolchainTargetStatus::Failed)
.await
{
tracing::warn!("Failed to set target status to failed: {status_err}");
}
tracing::warn!("Failed to extract components from archive: {e}");
}
}
});
Ok(Json(ToolchainResponse {
success: true,
message: Some(format!(
"Uploaded {}-{}-{} ({} bytes); components are being processed",
params.name, params.version, params.target, size
)),
}))
}
fn spawn_target_processing(
tasks: &BackgroundTasks,
db: Arc<dyn kellnr_db::DbProvider>,
target_id: i64,
processing: impl Future<Output = ()> + Send + 'static,
) -> tokio::task::JoinHandle<()> {
tasks.spawn_tracked(async move {
if let Err(e) = tokio::spawn(processing).await {
tracing::error!("Processing of toolchain target {target_id} stopped: {e}");
if let Err(e) = db
.set_target_status(target_id, ToolchainTargetStatus::Failed)
.await
{
tracing::warn!("Failed to set target status to failed: {e}");
}
}
})
}
async fn add_target_to_db(
db: &Arc<dyn kellnr_db::DbProvider>,
toolchain: Option<&ToolchainWithTargets>,
params: &UploadQuery,
path: &StorageKey,
hash: &str,
size: i64,
) -> ApiResult<i64> {
let toolchain_id = match toolchain {
Some(tc) => tc.id,
None => db
.add_toolchain(
¶ms.name,
¶ms.version,
¶ms.date,
params.channel.clone(),
)
.await
.map_err(|e| internal_error("Failed to create toolchain", e))?,
};
match db
.add_toolchain_target(toolchain_id, ¶ms.target, path.as_str(), hash, size)
.await
{
Ok(target_id) => Ok(target_id),
Err(e) => {
if toolchain.is_none()
&& let Err(e) = db.delete_toolchain(¶ms.name, ¶ms.version).await
{
tracing::warn!(
"Failed to delete toolchain {}-{} after failed upload: {e}",
params.name,
params.version
);
}
Err(internal_error("Failed to add target", e))
}
}
}
#[utoipa::path(
delete,
path = "/{name}/{version}",
tag = "toolchains",
params(
("name" = String, Path, description = "Toolchain name"),
("version" = String, Path, description = "Toolchain version")
),
responses(
(status = 200, description = "Toolchain deleted", body = ToolchainResponse),
(status = 401, description = "Not authenticated"),
(status = 403, description = "Admin access required"),
(status = 404, description = "Toolchain not found"),
(status = 503, description = "Storage not configured")
),
security(("session_cookie" = []))
)]
pub async fn delete_toolchain(
_user: AdminUser,
State(db): DbState,
State(storage): ToolchainStorageState,
Path((name, version)): Path<(String, String)>,
) -> ApiResult<Json<ToolchainResponse>> {
trace!(name = %name, version = %version, "Deleting toolchain");
let storage = storage.as_ref().ok_or_else(storage_not_configured)?;
let toolchain = db
.get_toolchain_by_version(&name, &version)
.await
.map_err(|e| internal_error("Failed to load toolchain", e))?;
let Some(tc) = toolchain else {
return Err(ApiError::not_found(format!(
"Toolchain {name}-{version} not found"
)));
};
db.delete_toolchain(&name, &version)
.await
.map_err(|e| internal_error("Failed to delete toolchain", e))?;
for target in &tc.targets {
delete_target_archives(storage, target).await;
}
Ok(Json(ToolchainResponse {
success: true,
message: Some(format!(
"Deleted {name}-{version} with {} target(s)",
tc.targets.len()
)),
}))
}
#[utoipa::path(
delete,
path = "/{name}/{version}/targets/{target}",
tag = "toolchains",
params(
("name" = String, Path, description = "Toolchain name"),
("version" = String, Path, description = "Toolchain version"),
("target" = String, Path, description = "Target triple")
),
responses(
(status = 200, description = "Target deleted", body = ToolchainResponse),
(status = 401, description = "Not authenticated"),
(status = 403, description = "Admin access required"),
(status = 404, description = "Toolchain or target not found"),
(status = 503, description = "Storage not configured")
),
security(("session_cookie" = []))
)]
pub async fn delete_toolchain_target(
_user: AdminUser,
State(db): DbState,
State(storage): ToolchainStorageState,
Path((name, version, target)): Path<(String, String, String)>,
) -> ApiResult<Json<ToolchainResponse>> {
trace!(
name = %name,
version = %version,
target = %target,
"Deleting toolchain target"
);
let storage = storage.as_ref().ok_or_else(storage_not_configured)?;
let toolchain = db
.get_toolchain_by_version(&name, &version)
.await
.map_err(|e| internal_error("Failed to load toolchain", e))?;
let Some(tc) = toolchain else {
return Err(ApiError::not_found(format!(
"Toolchain {name}-{version} not found"
)));
};
let Some(t) = tc.targets.iter().find(|t| t.target == target) else {
return Err(ApiError::not_found(format!(
"Target {target} of {name}-{version} not found"
)));
};
let last_target = tc.targets.len() == 1;
if last_target {
db.delete_toolchain(&name, &version)
.await
.map_err(|e| internal_error("Failed to delete toolchain", e))?;
} else {
db.delete_toolchain_target(&name, &version, &target)
.await
.map_err(|e| internal_error("Failed to delete target", e))?;
}
delete_target_archives(storage, t).await;
if last_target {
return Ok(Json(ToolchainResponse {
success: true,
message: Some(format!("Deleted {name}-{version} (last target removed)")),
}));
}
Ok(Json(ToolchainResponse {
success: true,
message: Some(format!("Deleted {name}-{version}-{target}")),
}))
}
async fn delete_target_archives(storage: &ToolchainStorage, target: &ToolchainTargetInfo) {
if let Err(e) = storage
.delete(&StorageKey::from_stored(target.storage_path.clone()))
.await
{
tracing::warn!("Failed to delete archive from storage: {e}");
}
for component in &target.components {
if let Err(e) = storage
.delete(&StorageKey::from_stored(component.storage_path.clone()))
.await
{
tracing::warn!("Failed to delete component archive from storage: {e}");
}
}
}
#[utoipa::path(
get,
path = "/channels",
tag = "toolchains",
responses(
(status = 200, description = "List of channels", body = Vec<ChannelInfo>),
(status = 401, description = "Not authenticated"),
(status = 403, description = "Admin access required")
),
security(("session_cookie" = []))
)]
pub async fn list_channels(
_user: AdminUser,
State(db): DbState,
) -> ApiResult<Json<Vec<ChannelInfo>>> {
trace!("Listing channels");
db.get_channels()
.await
.map(Json)
.map_err(|e| internal_error("Failed to load channels", e))
}
#[utoipa::path(
put,
path = "/channels/{channel}",
tag = "toolchains",
params(
("channel" = String, Path, description = "Channel name (e.g., stable, beta, nightly)")
),
request_body = SetChannelRequest,
responses(
(status = 200, description = "Channel updated", body = ToolchainResponse),
(status = 401, description = "Not authenticated"),
(status = 403, description = "Admin access required"),
(status = 404, description = "Toolchain not found")
),
security(("session_cookie" = []))
)]
pub async fn set_channel(
_user: AdminUser,
State(db): DbState,
Path(channel): Path<String>,
Json(req): Json<SetChannelRequest>,
) -> ApiResult<Json<ToolchainResponse>> {
trace!(
channel = %channel,
name = %req.name,
version = %req.version,
"Setting channel"
);
db.set_channel(&channel, &req.name, &req.version)
.await
.map_err(|e| match e {
DbError::ToolchainNotFound(..) => ApiError::not_found(e.to_string()),
e => internal_error("Failed to set channel", e),
})?;
Ok(Json(ToolchainResponse {
success: true,
message: Some(format!(
"Channel {} now points to {}-{}",
channel, req.name, req.version
)),
}))
}
#[utoipa::path(
get,
path = "/{manifest_file}",
tag = "toolchains",
params(
("manifest_file" = String, Path, description = "Manifest filename (e.g., channel-rust-stable.toml)")
),
responses(
(status = 200, description = "Channel manifest (TOML)", content_type = "text/toml"),
(status = 404, description = "Channel not found or no targets available"),
(status = 503, description = "Targets are still being processed")
),
security(("session_cookie" = []))
)]
pub async fn get_channel_manifest(
State(db): DbState,
State(settings): SettingsState,
Path(manifest_file): Path<String>,
) -> ApiResult<Response> {
trace!(manifest_file = %manifest_file, "Getting channel manifest");
let (channel, want_sha256) = manifest_file
.strip_prefix("channel-rust-")
.and_then(|s| {
s.strip_suffix(".toml.sha256")
.map(|c| (c, true))
.or_else(|| s.strip_suffix(".toml").map(|c| (c, false)))
})
.ok_or(StatusCode::NOT_FOUND)?;
let toolchain = match db
.get_toolchain_by_channel(channel)
.await
.map_err(|e| internal_error("Failed to load toolchains", e))?
{
Some(tc) => tc,
None => db
.get_toolchain_by_version("rust", channel)
.await
.map_err(|e| internal_error("Failed to load toolchains", e))?
.ok_or(StatusCode::NOT_FOUND)?,
};
let has_ready_targets = toolchain
.targets
.iter()
.any(|t| t.status == ToolchainTargetStatus::Ready);
if !has_ready_targets {
let has_processing = toolchain
.targets
.iter()
.any(|t| t.status == ToolchainTargetStatus::Processing);
if has_processing {
return Err(StatusCode::SERVICE_UNAVAILABLE.into());
}
return Err(StatusCode::NOT_FOUND.into());
}
let manifest = generate_manifest(&toolchain, &settings);
if want_sha256 {
let hash = sha256::digest(manifest.as_bytes());
Ok(hash.into_response())
} else {
let mut headers = HeaderMap::new();
headers.insert(
header::CONTENT_TYPE,
"text/toml; charset=utf-8".parse().unwrap(),
);
Ok((headers, manifest).into_response())
}
}
#[utoipa::path(
get,
path = "/{date}/{filename}",
tag = "toolchains",
params(
("date" = String, Path, description = "Release date (e.g., 2024-01-15)"),
("filename" = String, Path, description = "Archive filename")
),
responses(
(status = 200, description = "Toolchain archive (xz, gzip, or raw)", content_type = "application/octet-stream"),
(status = 404, description = "Archive not found"),
(status = 503, description = "Storage not configured")
),
security(("session_cookie" = []))
)]
pub async fn download_archive(
State(storage): ToolchainStorageState,
Path((date, filename)): Path<(String, String)>,
) -> ApiResult<Response> {
trace!(date = %date, filename = %filename, "Downloading toolchain archive");
let storage = storage.as_ref().ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
let path = StorageKey::toolchain_file(&date, &filename);
let archive = storage.get(&path).await.map_err(|e| {
tracing::warn!("Failed to get toolchain archive {}: {}", path, e);
StatusCode::NOT_FOUND
})?;
let path_ref = std::path::Path::new(&filename);
let ext = path_ref.extension().and_then(|e| e.to_str()).unwrap_or("");
let content_type = if ext.eq_ignore_ascii_case("xz") {
"application/x-xz"
} else if ext.eq_ignore_ascii_case("gz") {
"application/gzip"
} else {
"application/octet-stream"
};
let mut headers = HeaderMap::new();
headers.insert(header::CONTENT_TYPE, content_type.parse().unwrap());
headers.insert(
header::CONTENT_DISPOSITION,
format!("attachment; filename=\"{filename}\"")
.parse()
.unwrap(),
);
Ok((headers, Body::from(archive)).into_response())
}
static EXTRACTION_PERMITS: Semaphore = Semaphore::const_new(1);
async fn extract_and_store_components(
storage: &Arc<ToolchainStorage>,
db: &Arc<dyn kellnr_db::DbProvider>,
target_id: i64,
archive_id: &ToolchainArchiveId<'_>,
archive_bytes: Bytes,
) -> Result<(), String> {
let _permit = EXTRACTION_PERMITS
.acquire()
.await
.map_err(|e| format!("Failed to wait for component extraction: {e}"))?;
let components = {
let (name, version, target) = (
archive_id.name.to_string(),
archive_id.version.to_string(),
archive_id.target.to_string(),
);
tokio::task::spawn_blocking(move || {
split_components(&archive_bytes, &name, &version, &target)
})
.await
.map_err(|e| format!("Component extraction stopped: {e}"))??
};
tracing::debug!(
"Extracted {} components in two streamed passes, compressing and storing",
components.len()
);
let mut stored_paths: Vec<StorageKey> = Vec::new();
for (component_name, component_tar) in components {
let component_xz = tokio::task::spawn_blocking(move || compress_component(component_tar))
.await
.map_err(|e| format!("Compression of {component_name} stopped: {e}"))?
.map_err(|e| format!("Failed to compress {component_name}: {e}"))?;
let component_hash = sha256::digest(&component_xz[..]);
let component_size = component_xz.len() as i64;
let component_storage_path = StorageKey::toolchain_component(archive_id, &component_name);
let in_use = db
.is_toolchain_component_stored_at(component_storage_path.as_str())
.await
.map_err(|e| format!("Failed to check component {component_name}: {e}"))?;
let component_xz = Bytes::from(component_xz);
let stored = if in_use {
storage.put_raw(&component_storage_path, component_xz).await
} else {
storage
.put_raw_overwrite(&component_storage_path, component_xz)
.await
};
stored.map_err(|e| format!("Failed to store component {component_name}: {e}"))?;
stored_paths.push(component_storage_path.clone());
if let Err(e) = db
.add_toolchain_component(
target_id,
&component_name,
component_storage_path.as_str(),
&component_hash,
component_size,
)
.await
{
for p in &stored_paths {
let _ = storage.delete(p).await;
}
return Err(format!(
"Failed to add component {component_name} to DB: {e}"
));
}
tracing::debug!(
component = component_name,
path = component_storage_path.as_str(),
size = component_size,
"Stored component archive"
);
}
Ok(())
}
fn split_components(
archive_bytes: &[u8],
name: &str,
version: &str,
target: &str,
) -> Result<Vec<(String, File)>, String> {
let prefix = format!("{name}-{version}-{target}/");
let meta_components_path = format!("{prefix}components");
let meta_installer_path = format!("{prefix}rust-installer-version");
let mut component_names: Vec<String> = Vec::new();
let mut installer_version = String::new();
let xz_decoder = xz2::read::XzDecoder::new(archive_bytes);
let mut archive = tar::Archive::new(xz_decoder);
for entry in archive
.entries()
.map_err(|e| format!("Failed to read tar entries: {e}"))?
{
let mut entry = entry.map_err(|e| format!("Failed to read tar entry: {e}"))?;
let entry_path = entry
.path()
.map_err(|e| format!("Failed to read entry path: {e}"))?
.to_string_lossy()
.to_string();
if entry_path == meta_components_path {
let mut contents = String::new();
entry
.read_to_string(&mut contents)
.map_err(|e| format!("Failed to read components file: {e}"))?;
component_names = contents
.lines()
.filter(|l| !l.is_empty())
.map(ToString::to_string)
.collect();
} else if entry_path == meta_installer_path {
entry
.read_to_string(&mut installer_version)
.map_err(|e| format!("Failed to read installer version: {e}"))?;
}
if !component_names.is_empty() && !installer_version.is_empty() {
break;
}
}
if component_names.is_empty() {
return Err("No components file found in archive".to_string());
}
let mut builders: HashMap<String, tar::Builder<BufWriter<File>>> = HashMap::new();
for cn in &component_names {
let file =
tempfile::tempfile().map_err(|e| format!("Failed to create temporary file: {e}"))?;
let mut builder = tar::Builder::new(BufWriter::new(file));
builder.mode(tar::HeaderMode::Deterministic);
if !installer_version.is_empty() {
append_metadata_file(
&mut builder,
&format!("{cn}-{version}-{target}/rust-installer-version"),
installer_version.as_bytes(),
)?;
}
append_metadata_file(
&mut builder,
&format!("{cn}-{version}-{target}/components"),
format!("{cn}\n").as_bytes(),
)?;
builders.insert(cn.clone(), builder);
}
let xz_decoder = xz2::read::XzDecoder::new(archive_bytes);
let mut archive = tar::Archive::new(xz_decoder);
for entry in archive
.entries()
.map_err(|e| format!("Failed to read tar entries: {e}"))?
{
let mut entry = entry.map_err(|e| format!("Failed to read tar entry: {e}"))?;
let entry_path = entry
.path()
.map_err(|e| format!("Failed to read entry path: {e}"))?
.to_string_lossy()
.to_string();
if entry_path == meta_components_path || entry_path == meta_installer_path {
continue;
}
if !entry_path.starts_with(&prefix) {
continue;
}
let rel_path = &entry_path[prefix.len()..];
let component_name = rel_path.split('/').next().unwrap_or("");
let Some(builder) = builders.get_mut(component_name) else {
continue;
};
let new_path = format!("{component_name}-{version}-{target}/{rel_path}");
let mut header = tar::Header::new_gnu();
header.set_size(entry.size());
header.set_mode(entry.header().mode().unwrap_or(0o644));
header.set_entry_type(entry.header().entry_type());
if let Ok(mtime) = entry.header().mtime() {
header.set_mtime(mtime);
}
if let Ok(Some(link)) = entry.header().link_name() {
header
.set_link_name(link)
.map_err(|e| format!("Failed to set link name: {e}"))?;
}
header.set_cksum();
builder
.append_data(&mut header, &new_path, &mut entry)
.map_err(|e| format!("Failed to append to tar: {e}"))?;
}
component_names
.into_iter()
.map(|component_name| {
let builder = builders
.remove(&component_name)
.ok_or_else(|| format!("Missing builder for {component_name}"))?;
let file = builder
.into_inner()
.and_then(|w| w.into_inner().map_err(std::io::IntoInnerError::into_error))
.and_then(|mut f| f.rewind().map(|()| f))
.map_err(|e| format!("Failed to finish tar for {component_name}: {e}"))?;
Ok((component_name, file))
})
.collect()
}
fn compress_component(mut component_tar: File) -> std::io::Result<Vec<u8>> {
let mut xz_encoder = xz2::write::XzEncoder::new(Vec::new(), 6);
std::io::copy(&mut component_tar, &mut xz_encoder)?;
xz_encoder.finish()
}
fn append_metadata_file<W: Write>(
builder: &mut tar::Builder<W>,
path: &str,
content: &[u8],
) -> Result<(), String> {
let mut header = tar::Header::new_gnu();
header.set_size(content.len() as u64);
header.set_mode(0o644);
header.set_entry_type(tar::EntryType::Regular);
header.set_cksum();
builder
.append_data(&mut header, path, content)
.map_err(|e| format!("Failed to add {path}: {e}"))
}
fn generate_manifest(
toolchain: &ToolchainWithTargets,
settings: &Arc<kellnr_settings::Settings>,
) -> String {
let base_url = format!("{}/api/v1/toolchains/dist", settings.origin.base_url());
let mut manifest = String::new();
let _ = write!(
manifest,
r#"manifest-version = "2"
date = "{}"
"#,
toolchain.date
);
let _ = write!(
manifest,
r#"
[pkg.rust]
version = "{}"
"#,
toolchain.version
);
let ready_targets: Vec<_> = toolchain
.targets
.iter()
.filter(|t| t.status == ToolchainTargetStatus::Ready)
.collect();
for target_info in &ready_targets {
let archive_url = format!("{}/{}", base_url, target_info.storage_path);
let _ = write!(
manifest,
r#"
[pkg.rust.target.{}]
available = true
xz_url = "{}"
xz_hash = "{}"
"#,
target_info.target, archive_url, target_info.hash
);
for component in &target_info.components {
let _ = write!(
manifest,
r#"
[[pkg.rust.target.{}.components]]
pkg = "{}"
target = "{}"
"#,
target_info.target, component.name, target_info.target
);
}
}
let mut component_pkgs: std::collections::BTreeMap<
String,
Vec<(&str, &kellnr_db::ToolchainComponentInfo)>,
> = std::collections::BTreeMap::new();
for target_info in &ready_targets {
for component in &target_info.components {
component_pkgs
.entry(component.name.clone())
.or_default()
.push((&target_info.target, component));
}
}
for (pkg_name, targets) in &component_pkgs {
let _ = write!(
manifest,
r#"
[pkg.{}]
version = "{}"
"#,
pkg_name, toolchain.version
);
for (target, comp) in targets {
let comp_url = format!("{}/{}", base_url, comp.storage_path);
let _ = write!(
manifest,
r#"
[pkg.{}.target.{}]
available = true
xz_url = "{}"
xz_hash = "{}"
"#,
pkg_name, target, comp_url, comp.hash
);
}
}
manifest
}
#[cfg(test)]
mod tests {
use std::io::Cursor;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use axum::Router;
use axum::body::{Body, to_bytes};
use axum::http::{Request, StatusCode, header};
use axum::response::Response;
use axum::routing::{delete, get, put};
use bytes::Bytes;
use kellnr_common::account_name::UserName;
use kellnr_common::plain_token::PlainToken;
use kellnr_db::mock::MockDb;
use kellnr_db::test_utils::test_drop_table;
use kellnr_db::{DbProvider, ToolchainTargetInfo, ToolchainWithTargets, UserFlags};
use kellnr_http::error::ErrorDetails;
use kellnr_http::state::AppStateData;
use kellnr_http::test_support::{
encode_cookies, session_cookie, sqlite_test_state, test_state,
};
use kellnr_storage::cached_crate_storage::DynStorage;
use kellnr_storage::fs_storage::FSStorage;
use kellnr_storage::toolchain_storage::ToolchainStorage;
use kellnr_test_db::TestDb;
use mockall::predicate::*;
use serde::de::DeserializeOwned;
use tempfile::TempDir;
use tower::ServiceExt;
use super::*;
const TARGET: &str = "x86_64-unknown-linux-gnu";
const ARCHIVE_PATH: &str = "2024-01-15/rust-1.0.0-x86_64-unknown-linux-gnu.tar.xz";
const UPLOAD_URL: &str = "/api/v1/toolchains?name=rust&version=1.0.0&target=x86_64-unknown-linux-gnu&date=2024-01-15";
async fn admin_state(
toolchain_storage: Option<Arc<ToolchainStorage>>,
) -> (AppStateData, TestDb, String) {
let (state, db) = sqlite_test_state().await;
let cookie = session_cookie(&db, "admin", UserFlags::admin()).await;
let state = AppStateData {
toolchain_storage,
..state
};
(state, db, cookie)
}
fn create_test_router(state: AppStateData) -> Router {
let api_routes = Router::new()
.route("/", get(list_toolchains))
.route("/", put(upload_toolchain))
.route("/{name}/{version}", delete(delete_toolchain))
.route(
"/{name}/{version}/targets/{target}",
delete(delete_toolchain_target),
)
.route("/channels", get(list_channels))
.route("/channels/{channel}", put(set_channel));
let dist_routes = Router::new()
.route("/{manifest_file}", get(get_channel_manifest))
.route("/{date}/{filename}", get(download_archive));
Router::new()
.nest("/api/v1/toolchains", api_routes)
.nest("/api/v1/toolchains/dist", dist_routes)
.with_state(state)
}
async fn send(state: AppStateData, request: Request<Body>) -> Response<Body> {
create_test_router(state).oneshot(request).await.unwrap()
}
async fn get_with(state: AppStateData, uri: &str, cookie: &str) -> Response<Body> {
let mut request = Request::get(uri);
if !cookie.is_empty() {
request = request.header(header::COOKIE, cookie);
}
send(state, request.body(Body::empty()).unwrap()).await
}
async fn parse_response<T: DeserializeOwned>(response: Response<Body>) -> T {
let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
serde_json::from_slice(&bytes).unwrap()
}
async fn error_detail(response: Response<Body>) -> String {
let details: ErrorDetails = parse_response(response).await;
details.errors[0].detail.clone()
}
async fn text(response: Response<Body>) -> String {
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
String::from_utf8(body.to_vec()).unwrap()
}
fn sample_archive() -> Bytes {
Bytes::from(vec![0xFD, 0x37, 0x7A, 0x58, 0x5A, 0x00])
}
fn upload_request(uri: &str, cookie: &str) -> Request<Body> {
Request::put(uri)
.header(header::COOKIE, cookie)
.header(header::CONTENT_TYPE, "application/octet-stream")
.body(Body::from(sample_archive()))
.unwrap()
}
fn delete_request(uri: &str, cookie: &str) -> Request<Body> {
Request::delete(uri)
.header(header::COOKIE, cookie)
.body(Body::empty())
.unwrap()
}
fn set_channel_request(channel: &str, version: &str, cookie: &str) -> Request<Body> {
Request::put(format!("/api/v1/toolchains/channels/{channel}"))
.header(header::COOKIE, cookie)
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(format!(
r#"{{"name": "rust", "version": "{version}"}}"#
)))
.unwrap()
}
async fn add_toolchain(db: &TestDb, targets: &[&str]) -> Vec<i64> {
let id = db
.add_toolchain("rust", "1.0.0", "2024-01-15", Some("stable".to_string()))
.await
.unwrap();
let mut ids = Vec::new();
for target in targets {
let target_id = db
.add_toolchain_target(
id,
target,
&format!("2024-01-15/rust-1.0.0-{target}.tar.xz"),
"abc123",
1024,
)
.await
.unwrap();
db.set_target_status(target_id, ToolchainTargetStatus::Ready)
.await
.unwrap();
ids.push(target_id);
}
ids
}
async fn toolchain(db: &TestDb) -> Option<ToolchainWithTargets> {
db.get_toolchain_by_version("rust", "1.0.0").await.unwrap()
}
fn target_names(toolchain: &ToolchainWithTargets) -> Vec<&str> {
toolchain
.targets
.iter()
.map(|t| t.target.as_str())
.collect()
}
fn stored(path: &str) -> StorageKey {
StorageKey::from_stored(path.to_string())
}
fn fs_toolchain_storage(dir: &TempDir) -> Arc<ToolchainStorage> {
let storage: DynStorage = Box::new(FSStorage::new(dir.path().to_str().unwrap()).unwrap());
Arc::new(ToolchainStorage::new(storage))
}
#[tokio::test]
async fn admin_endpoints_forbid_non_admin() {
let (state, db) = sqlite_test_state().await;
let cookie = session_cookie(&db, "user", UserFlags::normal()).await;
let requests = [
Request::get("/api/v1/toolchains")
.header(header::COOKIE, &cookie)
.body(Body::empty())
.unwrap(),
upload_request(UPLOAD_URL, &cookie),
delete_request("/api/v1/toolchains/rust/1.0.0", &cookie),
delete_request(
"/api/v1/toolchains/rust/1.0.0/targets/x86_64-unknown-linux-gnu",
&cookie,
),
Request::get("/api/v1/toolchains/channels")
.header(header::COOKIE, &cookie)
.body(Body::empty())
.unwrap(),
set_channel_request("stable", "1.0.0", &cookie),
];
for request in requests {
let uri = request.uri().to_string();
let response = send(state.clone(), request).await;
assert_eq!(StatusCode::FORBIDDEN, response.status(), "{uri}");
}
}
#[tokio::test]
async fn test_list_toolchains_returns_data() {
let (state, db, cookie) = admin_state(None).await;
add_toolchain(&db, &[TARGET]).await;
let response = get_with(state, "/api/v1/toolchains", &cookie).await;
assert_eq!(StatusCode::OK, response.status());
let toolchains: Vec<ToolchainWithTargets> = parse_response(response).await;
assert_eq!(toolchains.len(), 1);
assert_eq!(toolchains[0].name, "rust");
assert_eq!(toolchains[0].version, "1.0.0");
assert_eq!(toolchains[0].targets.len(), 1);
}
#[tokio::test]
async fn test_list_channels_returns_data() {
let (state, db, cookie) = admin_state(None).await;
add_toolchain(&db, &[]).await;
let response = get_with(state, "/api/v1/toolchains/channels", &cookie).await;
assert_eq!(StatusCode::OK, response.status());
let channels: Vec<ChannelInfo> = parse_response(response).await;
assert_eq!(channels.len(), 1);
assert_eq!(channels[0].name, "stable");
assert_eq!(channels[0].version, "1.0.0");
}
#[tokio::test]
async fn test_upload_toolchain_success() {
let temp_dir = TempDir::new().unwrap();
let (state, db, cookie) = admin_state(Some(fs_toolchain_storage(&temp_dir))).await;
let response = send(
state,
upload_request(&format!("{UPLOAD_URL}&channel=stable"), &cookie),
)
.await;
assert_eq!(StatusCode::OK, response.status());
let result: ToolchainResponse = parse_response(response).await;
assert!(result.success);
assert!(result.message.unwrap().contains("Uploaded"));
let toolchain = toolchain(&db).await.unwrap();
assert_eq!(toolchain.date, "2024-01-15");
assert_eq!(toolchain.channel.as_deref(), Some("stable"));
assert_eq!(target_names(&toolchain), vec![TARGET]);
assert_eq!(toolchain.targets[0].storage_path, ARCHIVE_PATH);
}
#[tokio::test]
async fn test_set_channel_success() {
let (state, db, cookie) = admin_state(None).await;
add_toolchain(&db, &[]).await;
let response = send(state, set_channel_request("beta", "1.0.0", &cookie)).await;
assert_eq!(StatusCode::OK, response.status());
let result: ToolchainResponse = parse_response(response).await;
assert!(result.success);
assert_eq!(
toolchain(&db).await.unwrap().channel.as_deref(),
Some("beta")
);
}
#[tokio::test]
async fn test_delete_toolchain_target_success() {
let temp_dir = TempDir::new().unwrap();
let (state, db, cookie) = admin_state(Some(fs_toolchain_storage(&temp_dir))).await;
add_toolchain(&db, &[TARGET, "aarch64-unknown-linux-gnu"]).await;
let response = send(
state,
delete_request(
"/api/v1/toolchains/rust/1.0.0/targets/x86_64-unknown-linux-gnu",
&cookie,
),
)
.await;
assert_eq!(StatusCode::OK, response.status());
let result: ToolchainResponse = parse_response(response).await;
assert!(result.success);
let toolchain = toolchain(&db).await.unwrap();
assert_eq!(target_names(&toolchain), vec!["aarch64-unknown-linux-gnu"]);
}
#[tokio::test]
async fn test_delete_last_target_auto_deletes_toolchain() {
let temp_dir = TempDir::new().unwrap();
let (state, db, cookie) = admin_state(Some(fs_toolchain_storage(&temp_dir))).await;
add_toolchain(&db, &[TARGET]).await;
let response = send(
state,
delete_request(
"/api/v1/toolchains/rust/1.0.0/targets/x86_64-unknown-linux-gnu",
&cookie,
),
)
.await;
assert_eq!(StatusCode::OK, response.status());
let result: ToolchainResponse = parse_response(response).await;
assert!(result.success);
let msg = result.message.unwrap();
assert!(
msg.contains("last target removed"),
"Expected 'last target removed' in message: {msg}"
);
assert!(toolchain(&db).await.is_none());
}
#[tokio::test]
async fn test_delete_toolchain_success() {
let temp_dir = TempDir::new().unwrap();
let (state, db, cookie) = admin_state(Some(fs_toolchain_storage(&temp_dir))).await;
add_toolchain(&db, &[TARGET, "aarch64-unknown-linux-gnu"]).await;
let response = send(
state,
delete_request("/api/v1/toolchains/rust/1.0.0", &cookie),
)
.await;
assert_eq!(StatusCode::OK, response.status());
let result: ToolchainResponse = parse_response(response).await;
assert!(result.success);
assert!(result.message.unwrap().contains("Deleted"));
assert!(toolchain(&db).await.is_none());
}
#[tokio::test]
async fn test_delete_toolchain_not_found() {
let temp_dir = TempDir::new().unwrap();
let (state, _db, cookie) = admin_state(Some(fs_toolchain_storage(&temp_dir))).await;
let response = send(
state,
delete_request("/api/v1/toolchains/rust/99.99.99", &cookie),
)
.await;
assert_eq!(StatusCode::NOT_FOUND, response.status());
assert!(error_detail(response).await.contains("not found"));
}
#[tokio::test]
async fn test_upload_toolchain_storage_not_configured() {
let (state, _db, cookie) = admin_state(None).await;
let response = send(state, upload_request(UPLOAD_URL, &cookie)).await;
assert_eq!(StatusCode::SERVICE_UNAVAILABLE, response.status());
assert!(error_detail(response).await.contains("not configured"));
}
#[tokio::test]
async fn test_upload_duplicate_target_conflict() {
let temp_dir = TempDir::new().unwrap();
let (state, db, cookie) = admin_state(Some(fs_toolchain_storage(&temp_dir))).await;
add_toolchain(&db, &[TARGET]).await;
let response = send(state, upload_request(UPLOAD_URL, &cookie)).await;
assert_eq!(StatusCode::CONFLICT, response.status());
assert!(error_detail(response).await.contains("already exists"));
}
#[tokio::test]
async fn test_set_channel_toolchain_not_found() {
let (state, _db, cookie) = admin_state(None).await;
let response = send(state, set_channel_request("stable", "99.99.99", &cookie)).await;
assert_eq!(StatusCode::NOT_FOUND, response.status());
assert!(error_detail(response).await.contains("not found"));
}
#[tokio::test]
async fn test_get_manifest_invalid_format_not_found() {
let response = get_with(test_state(), "/api/v1/toolchains/dist/stable.toml", "").await;
assert_eq!(StatusCode::NOT_FOUND, response.status());
}
#[tokio::test]
async fn test_get_manifest_channel_not_found() {
let (state, _db) = sqlite_test_state().await;
let response = get_with(
state,
"/api/v1/toolchains/dist/channel-rust-nonexistent.toml",
"",
)
.await;
assert_eq!(StatusCode::NOT_FOUND, response.status());
}
#[tokio::test]
async fn test_download_archive_not_found() {
let temp_dir = TempDir::new().unwrap();
let state = AppStateData {
toolchain_storage: Some(fs_toolchain_storage(&temp_dir)),
..test_state()
};
let response = get_with(
state,
"/api/v1/toolchains/dist/2024-01-15/nonexistent.tar.xz",
"",
)
.await;
assert_eq!(StatusCode::NOT_FOUND, response.status());
}
#[tokio::test]
async fn test_download_archive_storage_not_configured() {
let response = get_with(
test_state(),
"/api/v1/toolchains/dist/2024-01-15/rust-1.0.0-x86_64-unknown-linux-gnu.tar.xz",
"",
)
.await;
assert_eq!(StatusCode::SERVICE_UNAVAILABLE, response.status());
}
#[tokio::test]
async fn test_get_manifest_returns_valid_toml() {
let (state, db) = sqlite_test_state().await;
add_toolchain(&db, &[TARGET]).await;
let response = get_with(
state,
"/api/v1/toolchains/dist/channel-rust-stable.toml",
"",
)
.await;
assert_eq!(StatusCode::OK, response.status());
assert_eq!(
response.headers().get(header::CONTENT_TYPE).unwrap(),
"text/toml; charset=utf-8"
);
let manifest = text(response).await;
assert!(manifest.contains("manifest-version = \"2\""));
assert!(manifest.contains("date = \"2024-01-15\""));
assert!(manifest.contains("[pkg.rust]"));
assert!(manifest.contains("version = \"1.0.0\""));
assert!(manifest.contains("[pkg.rust.target.x86_64-unknown-linux-gnu]"));
assert!(manifest.contains("available = true"));
assert!(manifest.contains("xz_url = \""));
assert!(manifest.contains("xz_hash = \"abc123\""));
}
#[tokio::test]
async fn test_get_manifest_by_version_falls_back_when_no_channel() {
let (state, db) = sqlite_test_state().await;
add_toolchain(&db, &[TARGET]).await;
let response = get_with(state, "/api/v1/toolchains/dist/channel-rust-1.0.0.toml", "").await;
assert_eq!(StatusCode::OK, response.status());
let manifest = text(response).await;
assert!(manifest.contains("manifest-version = \"2\""));
assert!(manifest.contains("version = \"1.0.0\""));
assert!(manifest.contains("[pkg.rust.target.x86_64-unknown-linux-gnu]"));
}
#[tokio::test]
async fn test_manifest_includes_component_packages() {
let (state, db) = sqlite_test_state().await;
let target_id = add_toolchain(&db, &[TARGET]).await[0];
for (name, hash, size) in [("rustc", "rustc_hash", 512), ("cargo", "cargo_hash", 256)] {
db.add_toolchain_component(
target_id,
name,
&format!("2024-01-15/{name}-1.0.0-x86_64-unknown-linux-gnu.tar.xz"),
hash,
size,
)
.await
.unwrap();
}
let response = get_with(
state,
"/api/v1/toolchains/dist/channel-rust-stable.toml",
"",
)
.await;
assert_eq!(StatusCode::OK, response.status());
let manifest = text(response).await;
assert!(manifest.contains("[[pkg.rust.target.x86_64-unknown-linux-gnu.components]]"));
assert!(manifest.contains("pkg = \"rustc\""));
assert!(manifest.contains("pkg = \"cargo\""));
assert!(manifest.contains("[pkg.rustc]"));
assert!(manifest.contains("[pkg.rustc.target.x86_64-unknown-linux-gnu]"));
assert!(manifest.contains("xz_hash = \"rustc_hash\""));
assert!(manifest.contains("[pkg.cargo]"));
assert!(manifest.contains("[pkg.cargo.target.x86_64-unknown-linux-gnu]"));
assert!(manifest.contains("xz_hash = \"cargo_hash\""));
}
fn two_component_archive() -> Vec<u8> {
use std::io::Write;
let mut tar_builder = tar::Builder::new(Vec::new());
let version_data = b"3\n";
let mut header = tar::Header::new_gnu();
header.set_size(version_data.len() as u64);
header.set_mode(0o644);
header.set_entry_type(tar::EntryType::Regular);
header.set_cksum();
tar_builder
.append_data(
&mut header,
"rust-1.0.0-x86_64-unknown-linux-gnu/rust-installer-version",
&version_data[..],
)
.unwrap();
let components_data = b"rustc\ncargo\n";
let mut header = tar::Header::new_gnu();
header.set_size(components_data.len() as u64);
header.set_mode(0o644);
header.set_entry_type(tar::EntryType::Regular);
header.set_cksum();
tar_builder
.append_data(
&mut header,
"rust-1.0.0-x86_64-unknown-linux-gnu/components",
&components_data[..],
)
.unwrap();
let manifest_data = b"file:bin/rustc\n";
let mut header = tar::Header::new_gnu();
header.set_size(manifest_data.len() as u64);
header.set_mode(0o644);
header.set_entry_type(tar::EntryType::Regular);
header.set_cksum();
tar_builder
.append_data(
&mut header,
"rust-1.0.0-x86_64-unknown-linux-gnu/rustc/manifest.in",
&manifest_data[..],
)
.unwrap();
let manifest_data = b"file:bin/cargo\n";
let mut header = tar::Header::new_gnu();
header.set_size(manifest_data.len() as u64);
header.set_mode(0o644);
header.set_entry_type(tar::EntryType::Regular);
header.set_cksum();
tar_builder
.append_data(
&mut header,
"rust-1.0.0-x86_64-unknown-linux-gnu/cargo/manifest.in",
&manifest_data[..],
)
.unwrap();
let tar_data = tar_builder.into_inner().unwrap();
let mut xz_encoder = xz2::write::XzEncoder::new(Vec::new(), 1);
xz_encoder.write_all(&tar_data).unwrap();
xz_encoder.finish().unwrap()
}
const ARCHIVE_ID: ToolchainArchiveId<'static> = ToolchainArchiveId {
name: "rust",
version: "1.0.0",
date: "2024-01-15",
target: TARGET,
};
#[tokio::test]
async fn test_extract_components_from_archive() {
let xz_data = two_component_archive();
let tmp_dir = TempDir::new().unwrap();
let toolchain_dir = tmp_dir.path().join("toolchains");
std::fs::create_dir_all(&toolchain_dir).unwrap();
let fs = Box::new(FSStorage::new(toolchain_dir.to_str().unwrap()).unwrap()) as DynStorage;
let storage = Arc::new(ToolchainStorage::new(fs));
let db = TestDb::sqlite().await;
let target_id = add_toolchain(&db, &[TARGET]).await[0];
let result = extract_and_store_components(
&storage,
&db.provider(),
target_id,
&ARCHIVE_ID,
Bytes::from(xz_data),
)
.await;
assert!(result.is_ok(), "extraction failed: {result:?}");
let mut components: Vec<String> = toolchain(&db).await.unwrap().targets[0]
.components
.iter()
.map(|c| c.name.clone())
.collect();
components.sort_unstable();
assert_eq!(components, vec!["cargo", "rustc"]);
let rustc_exists = storage
.exists(&stored(
"2024-01-15/rustc-1.0.0-x86_64-unknown-linux-gnu.tar.xz",
))
.await
.unwrap();
assert!(rustc_exists, "rustc component archive not found in storage");
let cargo_exists = storage
.exists(&stored(
"2024-01-15/cargo-1.0.0-x86_64-unknown-linux-gnu.tar.xz",
))
.await
.unwrap();
assert!(cargo_exists, "cargo component archive not found in storage");
let rustc_archive = storage
.get(&stored(
"2024-01-15/rustc-1.0.0-x86_64-unknown-linux-gnu.tar.xz",
))
.await
.unwrap();
let mut xz_dec = xz2::read::XzDecoder::new(&rustc_archive[..]);
let mut tar_bytes = Vec::new();
xz_dec.read_to_end(&mut tar_bytes).unwrap();
let mut archive = tar::Archive::new(Cursor::new(&tar_bytes));
let mut entry_names: Vec<String> = Vec::new();
for entry in archive.entries().unwrap() {
let mut entry = entry.unwrap();
let path = entry.path().unwrap().to_string_lossy().to_string();
if path.ends_with("manifest.in") {
let mut contents = String::new();
entry.read_to_string(&mut contents).unwrap();
assert_eq!("file:bin/rustc\n", contents);
}
entry_names.push(path);
}
assert!(
entry_names
.iter()
.any(|n| n.contains("rust-installer-version")),
"missing rust-installer-version in component archive: {entry_names:?}"
);
assert!(
entry_names.iter().any(|n| n.contains("components")),
"missing components file in component archive: {entry_names:?}"
);
assert!(
entry_names.iter().any(|n| n.contains("manifest.in")),
"missing manifest.in in component archive: {entry_names:?}"
);
}
const RUSTC_COMPONENT: &str = "2024-01-15/rustc-1.0.0-x86_64-unknown-linux-gnu.tar.xz";
async fn extract_over_leftover(
storage: &Arc<ToolchainStorage>,
in_use: bool,
) -> Result<(), String> {
storage
.put_raw(&stored(RUSTC_COMPONENT), Bytes::from_static(b"leftover"))
.await
.unwrap();
let db = TestDb::sqlite().await;
let ids = add_toolchain(&db, &[TARGET, "aarch64-unknown-linux-gnu"]).await;
if in_use {
db.add_toolchain_component(ids[1], "rustc", RUSTC_COMPONENT, "hash", 8)
.await
.unwrap();
}
extract_and_store_components(
storage,
&db.provider(),
ids[0],
&ARCHIVE_ID,
Bytes::from(two_component_archive()),
)
.await
}
#[tokio::test]
async fn extract_components_replaces_leftover_component_archive() {
let dir = TempDir::new().unwrap();
let storage = fs_toolchain_storage(&dir);
let result = extract_over_leftover(&storage, false).await;
assert!(result.is_ok(), "extraction failed: {result:?}");
let stored = storage.get(&stored(RUSTC_COMPONENT)).await.unwrap();
assert_ne!(Bytes::from_static(b"leftover"), stored);
}
#[tokio::test]
async fn extract_components_keeps_component_archive_of_other_target() {
let dir = TempDir::new().unwrap();
let storage = fs_toolchain_storage(&dir);
let result = extract_over_leftover(&storage, true).await;
assert!(result.is_err());
let stored = storage.get(&stored(RUSTC_COMPONENT)).await.unwrap();
assert_eq!(Bytes::from_static(b"leftover"), stored);
}
#[tokio::test]
async fn test_get_manifest_sha256_returns_ok() {
let (state, db) = sqlite_test_state().await;
add_toolchain(&db, &[TARGET]).await;
let response = get_with(
state,
"/api/v1/toolchains/dist/channel-rust-stable.toml.sha256",
"",
)
.await;
assert_eq!(StatusCode::OK, response.status());
let hash_content = text(response).await;
let hash = hash_content.trim();
assert_eq!(
hash.len(),
64,
"SHA256 hash should be 64 hex characters, got: {hash}"
);
assert!(
hash.chars().all(|c| c.is_ascii_hexdigit()),
"SHA256 hash should only contain hex characters"
);
}
fn ready_toolchain(targets: &[&str]) -> ToolchainWithTargets {
ToolchainWithTargets {
id: 1,
name: "rust".to_string(),
version: "1.0.0".to_string(),
date: "2024-01-15".to_string(),
channel: Some("stable".to_string()),
created: "2024-01-15".to_string(),
targets: targets
.iter()
.zip(1..)
.map(|(target, id)| ToolchainTargetInfo {
id,
target: (*target).to_string(),
storage_path: format!("2024-01-15/rust-1.0.0-{target}.tar.xz"),
hash: "abc123".to_string(),
size: 1024,
components: vec![],
status: ToolchainTargetStatus::Ready,
})
.collect(),
}
}
#[test]
fn manifest_urls_use_the_origin_path_and_omit_the_default_port() {
let mut settings = kellnr_settings::test_settings();
settings.origin.hostname = "kellnr.example.com".to_string();
settings.origin.protocol = kellnr_settings::Protocol::Https;
settings.origin.port = 443;
settings.origin.path = "/registry".to_string();
let manifest = generate_manifest(
&ready_toolchain(&["x86_64-unknown-linux-gnu"]),
&Arc::new(settings),
);
assert!(
manifest.contains(
"xz_url = \"https://kellnr.example.com/registry/api/v1/toolchains/dist/2024-01-15/rust-1.0.0-x86_64-unknown-linux-gnu.tar.xz\""
),
"unexpected manifest: {manifest}"
);
}
#[tokio::test]
async fn delete_unknown_target_of_single_target_toolchain_is_not_found() {
let temp_dir = TempDir::new().unwrap();
let (state, db, cookie) = admin_state(Some(fs_toolchain_storage(&temp_dir))).await;
add_toolchain(&db, &[TARGET]).await;
let response = send(
state,
delete_request(
"/api/v1/toolchains/rust/1.0.0/targets/x86_64-unknown-linux-gnux",
&cookie,
),
)
.await;
assert_eq!(StatusCode::NOT_FOUND, response.status());
assert_eq!(target_names(&toolchain(&db).await.unwrap()), vec![TARGET]);
}
#[tokio::test]
async fn upload_does_not_create_toolchain_when_archive_cannot_be_stored() {
let temp_dir = TempDir::new().unwrap();
let storage = fs_toolchain_storage(&temp_dir);
storage
.put_raw(&stored(ARCHIVE_PATH), Bytes::from_static(b"orphan"))
.await
.unwrap();
let (state, db, cookie) = admin_state(Some(storage)).await;
let response = send(state, upload_request(UPLOAD_URL, &cookie)).await;
assert_eq!(StatusCode::CONFLICT, response.status());
assert!(toolchain(&db).await.is_none());
}
#[tokio::test]
async fn upload_removes_archive_and_new_toolchain_when_target_cannot_be_added() {
let temp_dir = TempDir::new().unwrap();
let storage = fs_toolchain_storage(&temp_dir);
let (state, db, cookie) = admin_state(Some(storage.clone())).await;
test_drop_table(&db, "toolchain_target").await.unwrap();
let response = send(state, upload_request(UPLOAD_URL, &cookie)).await;
assert_eq!(StatusCode::INTERNAL_SERVER_ERROR, response.status());
assert!(!error_detail(response).await.contains("no such table"));
assert!(!storage.exists(&stored(ARCHIVE_PATH)).await.unwrap());
assert!(toolchain(&db).await.is_none());
}
fn admin_cookie() -> String {
encode_cookies([("kellnr_session_id", "admin_session")])
}
fn admin_db(targets: &'static [&'static str]) -> MockDb {
let mut mock_db = MockDb::new();
mock_db
.expect_validate_session()
.with(eq(PlainToken::from("admin_session")))
.returning(|_| {
Ok(kellnr_db::SessionInfo {
name: UserName::new("admin").unwrap(),
is_admin: true,
is_read_only: false,
})
});
mock_db
.expect_get_toolchain_by_version()
.returning(|_, _| Ok(Some(ready_toolchain(targets))));
mock_db
}
fn db_error() -> DbError {
DbError::InitializationError("secret detail".to_string())
}
async fn state_with_archive(
mock_db: MockDb,
dir: &TempDir,
) -> (AppStateData, Arc<ToolchainStorage>) {
let storage = fs_toolchain_storage(dir);
storage
.put_raw(&stored(ARCHIVE_PATH), Bytes::from_static(b"archive"))
.await
.unwrap();
let state = AppStateData {
db: Arc::new(mock_db),
toolchain_storage: Some(storage.clone()),
..test_state()
};
(state, storage)
}
#[tokio::test]
async fn delete_toolchain_keeps_archives_when_database_delete_fails() {
let temp_dir = TempDir::new().unwrap();
let mut mock_db = admin_db(&[TARGET]);
mock_db
.expect_delete_toolchain()
.returning(|_, _| Err(db_error()));
let (state, storage) = state_with_archive(mock_db, &temp_dir).await;
let response = send(
state,
delete_request("/api/v1/toolchains/rust/1.0.0", &admin_cookie()),
)
.await;
assert_eq!(StatusCode::INTERNAL_SERVER_ERROR, response.status());
assert!(!error_detail(response).await.contains("secret detail"));
assert!(storage.exists(&stored(ARCHIVE_PATH)).await.unwrap());
}
#[tokio::test]
async fn delete_target_keeps_archives_when_database_delete_fails() {
let temp_dir = TempDir::new().unwrap();
let mut mock_db = admin_db(&[TARGET, "aarch64-unknown-linux-gnu"]);
mock_db
.expect_delete_toolchain_target()
.returning(|_, _, _| Err(db_error()));
let (state, storage) = state_with_archive(mock_db, &temp_dir).await;
let response = send(
state,
delete_request(
"/api/v1/toolchains/rust/1.0.0/targets/x86_64-unknown-linux-gnu",
&admin_cookie(),
),
)
.await;
assert_eq!(StatusCode::INTERNAL_SERVER_ERROR, response.status());
assert!(storage.exists(&stored(ARCHIVE_PATH)).await.unwrap());
}
#[tokio::test]
async fn database_errors_are_not_returned_to_the_client() {
let temp_dir = TempDir::new().unwrap();
let (state, db, cookie) = admin_state(Some(fs_toolchain_storage(&temp_dir))).await;
test_drop_table(&db, "toolchain").await.unwrap();
let response = send(
state,
delete_request("/api/v1/toolchains/rust/1.0.0", &cookie),
)
.await;
assert_eq!(StatusCode::INTERNAL_SERVER_ERROR, response.status());
assert!(!error_detail(response).await.contains("no such table"));
}
#[tokio::test]
async fn target_is_marked_failed_when_processing_panics() {
let db = TestDb::sqlite().await;
let id = db
.add_toolchain("rust", "1.0.0", "2024-01-15", None)
.await
.unwrap();
let target_id = db
.add_toolchain_target(id, TARGET, ARCHIVE_PATH, "abc123", 1024)
.await
.unwrap();
spawn_target_processing(&BackgroundTasks::new(), db.provider(), target_id, async {
panic!("extraction bug")
})
.await
.unwrap();
let status = &toolchain(&db).await.unwrap().targets[0].status;
assert_eq!(*status, ToolchainTargetStatus::Failed);
}
#[tokio::test(start_paused = true)]
async fn shutdown_waits_for_target_processing() {
let tasks = BackgroundTasks::new();
let finished = Arc::new(AtomicBool::new(false));
let finished_in_task = finished.clone();
spawn_target_processing(&tasks, test_state().db, 7, async move {
tokio::time::sleep(Duration::from_secs(5)).await;
finished_in_task.store(true, Ordering::SeqCst);
});
tasks.request_shutdown();
tasks.wait_for_shutdown(Duration::from_secs(30)).await;
assert!(finished.load(Ordering::SeqCst));
}
}