use std::sync::Arc;
use knot_git::{Haves, RefUpdate, Repo, Staging, Wants};
use knot_index::{Index, Resolved};
use knot_pack::{FetchError, HaveOids, MaxWireBytes, PackLimits, UpstreamRefs, WantOids};
use knot_runtime::ReqwestHttp;
use knot_types::{AccountDid, BranchName, Oid, RepoDid};
use url::Url;
use crate::cob_write::WriteError;
use crate::config::ValidatedConfig;
use crate::repolock::{self, LockMode};
const MAX_PACK_BYTES: usize = 256 * 1024 * 1024;
enum Upstream {
Local(Repo),
Remote(ReqwestHttp, Url),
}
fn internal(error: impl std::fmt::Display) -> WriteError {
WriteError::internal(error.to_string())
}
/// a 400 on the want means the fork doesn't have the head.
fn pack_failed(error: FetchError, source: &RepoDid, head: Oid) -> WriteError {
match error {
FetchError::Status(status) if status.get() == 400 => WriteError::bad_gateway(format!(
"the fork {source} doesn't have the head commit {head}"
)),
other => map_fetch(other),
}
}
fn map_fetch(error: FetchError) -> WriteError {
match error {
FetchError::PackTooLarge { limit } => WriteError::request_too_large(format!(
"upstream exceeded this knot's limit of {} bytes",
limit.get()
)),
FetchError::Pack(inner) => {
WriteError::bad_gateway(format!("upstream sent an unusable pack: {inner}"))
}
other => WriteError::bad_gateway(other.to_string()),
}
}
async fn upstream(config: &ValidatedConfig, source: &RepoDid) -> Result<Upstream, WriteError> {
let layout = config.layout().map_err(internal)?;
let index = Index::new(
layout.meta_path(&config.knot_did).map_err(internal)?,
config.layout().map_err(internal)?,
);
index.rebuild()?;
if matches!(index.ownership_of(source), Resolved::Ready(Some(_))) {
return Ok(Upstream::Local(layout.open(source)?));
}
let transport = ReqwestHttp::new(config.http_limits()).map_err(internal)?;
let atproto = config.atproto(transport.clone()).map_err(internal)?;
let identity = atproto
.resolve_identity(&AccountDid::new(source.as_str()).map_err(internal)?)
.await
.map_err(|error| {
WriteError::invalid_request(format!("fork source can't be resolved: {error}"))
})?;
Ok(Upstream::Remote(
transport,
fork_url(identity.pds.url(), source)?,
))
}
/// only the endpoint's origin and path carry over; a did doc can't add a query or credentials.
pub(super) fn fork_url(endpoint: &Url, source: &RepoDid) -> Result<Url, WriteError> {
if !matches!(endpoint.scheme(), "https" | "http") || endpoint.host().is_none() {
return Err(WriteError::invalid_request(
"source repo's knot url isn't an http url",
));
}
let mut url = endpoint.clone();
url.set_query(None);
url.set_fragment(None);
let _ = url.set_username("");
let _ = url.set_password(None);
url.path_segments_mut()
.map_err(|()| WriteError::invalid_request("source repo's knot url can't be a base"))?
.pop_if_empty()
.push(source.as_str());
Ok(url)
}
async fn advertised(upstream: &Upstream) -> Result<UpstreamRefs, WriteError> {
advertised_under(upstream, &["HEAD", "refs/heads/"]).await
}
async fn advertised_under(
upstream: &Upstream,
prefixes: &[&str],
) -> Result<UpstreamRefs, WriteError> {
match upstream {
Upstream::Local(repo) => knot_pack::local_refs(repo, prefixes).map_err(map_fetch),
Upstream::Remote(http, url) => knot_pack::remote_refs(http, url, prefixes)
.await
.map_err(map_fetch),
}
}
pub async fn branch_tip(
config: &ValidatedConfig,
source: &RepoDid,
branch: &BranchName,
) -> Result<Oid, WriteError> {
let refs = advertised(&upstream(config, source).await?).await?;
refs.refs
.iter()
.find(|record| record.name == branch.head_ref())
.map(|record| record.target)
.ok_or_else(|| WriteError::invalid_request(format!("the fork has no branch {branch}")))
}
pub async fn fetch_head(
config: &Arc<ValidatedConfig>,
repo: &RepoDid,
source: &RepoDid,
head: Oid,
) -> Result<(), WriteError> {
let layout = config.layout().map_err(internal)?;
let target = layout
.open(repo)
.map_err(|_| WriteError::not_found("No such repository on this knot"))?;
if target.contains(head) {
return Ok(());
}
let upstream = upstream(config, source).await?;
let refs = advertised(&upstream).await?;
let haves: Vec<Oid> = target
.branches()?
.into_iter()
.map(|branch| branch.target)
.collect();
let scratch = layout.scratch_dir().to_path_buf();
std::fs::create_dir_all(&scratch).map_err(internal)?;
let pack = tempfile::Builder::new()
.prefix("pull-pack.")
.tempfile_in(&scratch)
.map_err(internal)?;
let (wants, have_oids, limit) = (
WantOids::new(vec![head]),
HaveOids::new(haves.clone()),
MaxWireBytes::new(MAX_PACK_BYTES),
);
let mut out = std::fs::File::create(pack.path()).map_err(internal)?;
pack_into(
&upstream,
&wants,
&have_oids,
refs.object_format,
limit,
&mut out,
)
.await
.map_err(|error| pack_failed(error, source, head))?;
drop(out);
// the lifetime lock keeps maintenance from pruning objects this transfer just wrote.
let _lock = repolock::acquire(
&config.output_root,
&crate::plan::repo_key(repo.as_str()),
LockMode::Lifetime,
)
.map_err(internal)?;
ingest(&target, pack.path(), &[head], &haves)
}
async fn pack_into(
upstream: &Upstream,
wants: &WantOids,
haves: &HaveOids,
object_format: knot_types::ObjectFormat,
limit: MaxWireBytes,
out: &mut std::fs::File,
) -> Result<(), FetchError> {
match upstream {
Upstream::Local(source) => knot_pack::local_pack_to(source, wants, haves, limit, out),
Upstream::Remote(http, url) => {
knot_pack::remote_pack_streamed(http, url, wants, haves, object_format, limit, out)
.await
}
}
}
/// fetched before the repo exists, a dead source can't leave one behind
pub struct Seed {
refs: UpstreamRefs,
pack: Option<tempfile::NamedTempFile>,
}
pub async fn seed(config: &ValidatedConfig, source: &RepoDid) -> Result<Seed, WriteError> {
let upstream = upstream(config, source).await?;
let mut refs = advertised_under(&upstream, &["HEAD", "refs/heads/", "refs/tags/"]).await?;
if refs.object_format != config.object_format {
return Err(WriteError::invalid_request(format!(
"the fork source {source} is a {} repo, and this knot hosts {} repos",
refs.object_format.kind(),
config.object_format.kind()
)));
}
refs.refs.retain(|record| {
let name = record.name.as_str();
name.starts_with("refs/heads/") || name.starts_with("refs/tags/")
});
let tips = refs.tips();
if tips.is_empty() {
return Ok(Seed { refs, pack: None });
}
let scratch = config
.layout()
.map_err(internal)?
.scratch_dir()
.to_path_buf();
std::fs::create_dir_all(&scratch).map_err(internal)?;
let pack = tempfile::Builder::new()
.prefix("fork-pack.")
.tempfile_in(&scratch)
.map_err(internal)?;
let mut out = std::fs::File::create(pack.path()).map_err(internal)?;
pack_into(
&upstream,
&WantOids::new(tips),
&HaveOids::default(),
refs.object_format,
MaxWireBytes::new(MAX_PACK_BYTES),
&mut out,
)
.await
.map_err(map_fetch)?;
Ok(Seed {
refs,
pack: Some(pack),
})
}
pub fn plant(config: &ValidatedConfig, repo: &RepoDid, seed: Seed) -> Result<(), WriteError> {
let target = config.layout().map_err(internal)?.open(repo)?;
let _lock = repolock::acquire(
&config.output_root,
&crate::plan::repo_key(repo.as_str()),
LockMode::Exclusive,
)
.map_err(internal)?;
let Some(pack) = &seed.pack else {
return Ok(());
};
ingest(&target, pack.path(), &seed.refs.tips(), &[])?;
let updates: Vec<RefUpdate> = seed
.refs
.refs
.iter()
.map(|record| RefUpdate::Create {
name: record.name.clone(),
new: record.target,
})
.collect();
target.update_refs(&updates)?;
let head = seed
.refs
.head_symref
.as_ref()
.filter(|head| seed.refs.find(head).is_some())
.or_else(|| {
let tip = seed.refs.head_target?;
seed.refs
.refs
.iter()
.find(|record| {
record.name.as_str().starts_with("refs/heads/") && record.target == tip
})
.map(|record| &record.name)
});
if let Some(branch) = head.and_then(|head| head.branch_name()) {
target.set_head(&branch)?;
}
crate::commit_graph::refresh(target.path());
Ok(())
}
fn ingest(
repo: &Repo,
pack: &std::path::Path,
wants: &[Oid],
haves: &[Oid],
) -> Result<(), WriteError> {
let staging = Staging::new(repo).map_err(internal)?;
knot_pack::ingest_pack_file(
&staging.repo().objects_dir(),
pack,
&PackLimits::default(),
repo.object_format().kind(),
)
.map_err(|error| WriteError::bad_gateway(format!("fork source pack is unusable: {error}")))?;
let connected = staging
.repo()
.select_pack_objects_filtered(
Wants::new(wants),
Haves::new(haves),
knot_git::Filter::None,
knot_pack::selection_budget(),
)?
.send
.iter()
.all(|oid| staging.repo().contains(*oid));
if !connected {
return Err(WriteError::bad_gateway(
"fork source sent an incomplete pack",
));
}
staging.migrate_into(repo).map_err(internal)
}
#[cfg(test)]
mod tests {
use knot_types::HttpStatus;
use super::*;
#[test]
fn a_400_to_the_want_says_the_fork_lacks_the_head() {
let fork = RepoDid::new("did:plc:abcdefghijklmnopqrstuvwx").unwrap();
let head = Oid::from_hex(&"ab".repeat(20)).unwrap();
let lacking = pack_failed(FetchError::Status(HttpStatus::new(400)), &fork, head);
assert_eq!(lacking.status(), http::StatusCode::BAD_GATEWAY);
let body = lacking.body();
assert_eq!(body["error"], "UpstreamFailure");
let message = body["message"].as_str().unwrap();
assert!(
message.contains(fork.as_str()) && message.contains(&head.to_hex()),
"{message}"
);
let other = pack_failed(FetchError::Status(HttpStatus::new(500)), &fork, head);
assert!(other.body()["message"].as_str().unwrap().contains("500"));
}
}