use std::sync::atomic::AtomicBool;
use super::live_progress::LiveProgress;
use super::{PreparedPush, PushAdvertisement, PushError, PushFailure, PushReport, protocol};
use crate::packet::Wire;
use crate::transport::TransportControl;
use crate::transport::http::{HttpRemote, RequestBody};
#[derive(Debug)]
pub enum HttpPushOutcome {
Declined,
Sent(PushReport),
}
pub async fn send_http(
remote: &HttpRemote,
prepared: PreparedPush,
control: TransportControl<'_>,
) -> Result<PushReport, PushError> {
match send_http_checked(remote, prepared, control, |_| true).await? {
HttpPushOutcome::Sent(report) => Ok(report),
HttpPushOutcome::Declined => unreachable!("unconditional HTTP push cannot decline"),
}
}
pub async fn send_http_checked(
remote: &HttpRemote,
prepared: PreparedPush,
control: TransportControl<'_>,
should_send: impl FnOnce(&PushAdvertisement) -> bool,
) -> Result<HttpPushOutcome, PushError> {
send_http_checked_with_progress(remote, prepared, control, should_send, |_| {}).await
}
pub async fn send_http_checked_with_progress(
remote: &HttpRemote,
prepared: PreparedPush,
control: TransportControl<'_>,
should_send: impl FnOnce(&PushAdvertisement) -> bool,
progress: impl FnMut(&[u8]),
) -> Result<HttpPushOutcome, PushError> {
let (format, limits) = (prepared.format, prepared.limits);
send_advertised(
remote,
format,
limits,
control,
|advertisement, offered| {
offered.validate(&prepared, &prepared.commands, control.cancel)?;
Ok(should_send(advertisement).then_some(prepared))
},
progress,
)
.await
}
pub(crate) async fn send_http_preparing(
remote: &HttpRemote,
format: crate::ObjectFormat,
limits: super::PushLimits,
control: TransportControl<'_>,
prepare: impl FnOnce(&PushAdvertisement) -> Result<PreparedPush, PushFailure>,
progress: impl FnMut(&[u8]),
) -> Result<PushReport, PushError> {
let outcome = send_advertised(
remote,
format,
limits,
control,
|advertisement, _| prepare(advertisement).map(Some),
progress,
)
.await?;
match outcome {
HttpPushOutcome::Sent(report) => Ok(report),
HttpPushOutcome::Declined => unreachable!("preparing HTTP push cannot decline"),
}
}
async fn send_advertised(
remote: &HttpRemote,
format: crate::ObjectFormat,
limits: super::PushLimits,
control: TransportControl<'_>,
prepare: impl FnOnce(
&PushAdvertisement,
&protocol::AdvertisedCapabilities,
) -> Result<Option<PreparedPush>, PushFailure>,
mut progress: impl FnMut(&[u8]),
) -> Result<HttpPushOutcome, PushError> {
#[cfg(feature = "tracing")]
let span = tracing::debug_span!(
target: "girt",
"push.http",
outcome = "incomplete",
failure_class = tracing::field::Empty,
effects = tracing::field::Empty,
accepted = tracing::field::Empty,
rejected = tracing::field::Empty,
pending = tracing::field::Empty,
unpack = tracing::field::Empty,
);
let operation = async {
let preflight = async {
control.check()?;
let (bytes, _) = remote
.discover("git-receive-pack", limits.max_advertisement_bytes, control)
.await?;
let mut reader = bytes.as_slice();
let mut wire = Wire {
reader: &mut reader,
remaining: limits.max_advertisement_bytes,
cancel: control.cancel,
};
let (offered, advertisement) =
protocol::advertise_for_selection(&mut wire, format, limits)?;
wire.end()?;
control.check()?;
Ok::<_, PushFailure>((offered, advertisement))
};
let (offered, advertisement) = preflight.await.map_err(PushError::NotSent)?;
let Some(mut prepared) = prepare(&advertisement, &offered).map_err(PushError::NotSent)?
else {
return Ok(HttpPushOutcome::Declined);
};
let caps = offered
.validate(&prepared, &prepared.commands, control.cancel)
.map_err(PushError::NotSent)?;
if caps.report_v2 || caps.sideband {
prepared.request = prepared
.request_for(caps.report_v2, caps.sideband)
.map_err(PushError::NotSent)?
.into_owned();
}
let mut report = PushReport::pending(&prepared.commands);
if prepared.commands.is_empty() {
return Ok(HttpPushOutcome::Sent(report));
}
report.mark_attempted(&prepared.request, prepared.request.len());
let body = RequestBody::new(prepared.request, prepared.pack)
.map_err(|error| PushError::NotSent(error.into()))?;
control.check().map_err(|e| PushError::NotSent(e.into()))?;
let mut sideband = LiveProgress::default();
let response = remote
.exchange_observed(
"git-receive-pack",
Some(body),
prepared.limits.max_status_bytes,
control,
|bytes| {
if caps.sideband {
sideband.observe(bytes, &mut progress);
}
},
)
.await;
let retain = AtomicBool::new(false);
let mut reader = response.body.as_slice();
let mut wire = Wire {
reader: &mut reader,
remaining: prepared.limits.max_status_bytes,
cancel: &retain,
};
let parsed = protocol::read_response(
&mut wire,
&mut report,
caps,
prepared.limits.max_status_bytes,
)
.and_then(|()| wire.end().map_err(PushFailure::from));
let result = response.result.map_err(PushFailure::from).and(parsed);
match result {
Ok(()) => Ok(HttpPushOutcome::Sent(report)),
Err(cause) => Err(PushError::Uncertain {
cause,
report: Box::new(report),
}),
}
};
#[cfg(feature = "tracing")]
let result = tracing::Instrument::instrument(operation, span.clone()).await;
#[cfg(not(feature = "tracing"))]
let result = operation.await;
#[cfg(feature = "tracing")]
crate::trace::finish(&span, &result, |error| crate::trace::push(error, &span));
#[cfg(feature = "tracing")]
if let Ok(HttpPushOutcome::Sent(report)) = &result {
crate::trace::push_report(&span, report);
}
result
}