use std::borrow::Borrow;
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::ssh::SshRemote;
#[derive(Debug)]
#[non_exhaustive]
pub enum SshPushOutcome {
Declined,
Sent(PushReport),
}
pub async fn send_ssh(
remote: &SshRemote,
prepared: &PreparedPush,
control: TransportControl<'_>,
) -> Result<PushReport, PushError> {
match send_ssh_checked_with_progress(remote, prepared, control, |_| true, |_| {}, |_| {})
.await?
{
SshPushOutcome::Sent(report) => Ok(report),
SshPushOutcome::Declined => unreachable!("unconditional SSH push cannot decline"),
}
}
pub async fn send_ssh_checked_with_progress(
remote: &SshRemote,
prepared: &PreparedPush,
control: TransportControl<'_>,
should_send: impl FnOnce(&PushAdvertisement) -> bool,
diagnostics: impl FnMut(&[u8]),
progress: impl FnMut(&[u8]),
) -> Result<SshPushOutcome, PushError> {
send_selected(
remote,
prepared.format,
prepared.limits,
control,
|advertisement, offered| {
offered.validate(prepared, &prepared.commands, control.cancel)?;
Ok(should_send(advertisement).then(|| (prepared, prepared.commands.clone())))
},
diagnostics,
progress,
)
.await
}
pub async fn send_ssh_selected_with_progress(
remote: &SshRemote,
prepared: &PreparedPush,
control: TransportControl<'_>,
select: impl FnOnce(&PushAdvertisement) -> Vec<crate::refs::RefName>,
diagnostics: impl FnMut(&[u8]),
progress: impl FnMut(&[u8]),
) -> Result<SshPushOutcome, PushError> {
send_selected(
remote,
prepared.format,
prepared.limits,
control,
|advertisement, offered| {
offered.validate_format(prepared.format)?;
let commands = prepared.selected_commands(select(advertisement))?;
Ok(Some((prepared, commands)))
},
diagnostics,
progress,
)
.await
}
pub(crate) async fn send_ssh_preparing(
remote: &SshRemote,
format: crate::ObjectFormat,
limits: super::PushLimits,
control: TransportControl<'_>,
prepare: impl FnOnce(&PushAdvertisement) -> Result<PreparedPush, PushFailure>,
diagnostics: impl FnMut(&[u8]),
progress: impl FnMut(&[u8]),
) -> Result<PushReport, PushError> {
let outcome = send_selected(
remote,
format,
limits,
control,
|advertisement, _| {
let prepared = prepare(advertisement)?;
let commands = prepared.commands.clone();
Ok(Some((prepared, commands)))
},
diagnostics,
progress,
)
.await?;
match outcome {
SshPushOutcome::Sent(report) => Ok(report),
SshPushOutcome::Declined => unreachable!("preparing SSH push cannot decline"),
}
}
async fn send_selected<P: Borrow<PreparedPush>>(
remote: &SshRemote,
format: crate::ObjectFormat,
limits: super::PushLimits,
control: TransportControl<'_>,
select: impl FnOnce(
&PushAdvertisement,
&protocol::AdvertisedCapabilities,
) -> Result<Option<(P, Vec<super::PushCommand>)>, PushFailure>,
mut diagnostics: impl FnMut(&[u8]),
mut progress: impl FnMut(&[u8]),
) -> Result<SshPushOutcome, PushError> {
#[cfg(feature = "tracing")]
let span = tracing::debug_span!(
target: "girt",
"push.ssh",
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 mut session = remote
.connect("git-receive-pack", control)
.map_err(|e| PushError::NotSent(e.into()))?;
let preflight = async {
let bytes = session
.advertise_with_diagnostics(
limits.max_advertisement_bytes,
control,
&mut diagnostics,
)
.await?;
let mut reader = bytes.as_slice();
let mut wire = Wire {
reader: &mut reader,
remaining: limits.max_advertisement_bytes,
cancel: control.cancel,
};
let (caps, advertisement) =
protocol::advertise_for_selection(&mut wire, format, limits)?;
wire.end()?;
control.check()?;
Ok::<_, PushFailure>((caps, advertisement))
};
let (offered, advertisement) = preflight.await.map_err(PushError::NotSent)?;
let selected = select(&advertisement, &offered).map_err(PushError::NotSent)?;
let Some((prepared, commands)) = selected else {
let (body, result, _) = session
.exchange_with_diagnostics(
b"0000",
&[],
limits.max_status_bytes,
control,
&mut diagnostics,
)
.await;
result.map_err(|error| PushError::NotSent(error.into()))?;
if !body.is_empty() {
return Err(PushError::NotSent(PushFailure::Protocol(
"trailing declined response bytes",
)));
}
return Ok(SshPushOutcome::Declined);
};
let prepared: &PreparedPush = prepared.borrow();
let caps = offered
.validate(prepared, &commands, control.cancel)
.map_err(PushError::NotSent)?;
let negotiated = prepared
.selected_request(&commands, caps.report_v2, caps.sideband, control.cancel)
.map_err(PushError::NotSent)?;
let mut report = PushReport::pending(&commands);
let request: &[u8] = if commands.is_empty() {
b"0000"
} else {
negotiated.as_ref()
};
let mut sideband = LiveProgress::default();
let (body, result, written) = session
.exchange_observed(
request,
if commands.iter().all(super::PushCommand::deletes) {
crate::fetch::PackInput::Slice(&[])
} else {
crate::fetch::PackInput::Stored(&prepared.pack)
},
prepared.limits.max_status_bytes,
control,
&mut diagnostics,
&mut |bytes| {
if caps.sideband && !commands.is_empty() {
sideband.observe(bytes, &mut progress);
}
},
)
.await;
if commands.is_empty() {
result.map_err(|e| PushError::NotSent(e.into()))?;
if !body.is_empty() {
return Err(PushError::NotSent(PushFailure::Protocol(
"trailing response bytes",
)));
}
return Ok(SshPushOutcome::Sent(report));
}
if written == 0
&& let Err(cause) = result
{
return Err(PushError::NotSent(cause.into()));
}
report.mark_attempted(request, written);
let retain = AtomicBool::new(false);
let mut reader = 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));
match result.map_err(PushFailure::from).and(parsed) {
Ok(()) => Ok(SshPushOutcome::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(SshPushOutcome::Sent(report)) = &result {
crate::trace::push_report(&span, report);
}
result
}