use std::path::{Path, PathBuf};
use serde_json::{json, Map, Value};
use std::collections::BTreeSet;
use onevcs::rules::{Approvals, Drafts};
use onevcs::testing::EmissionResult;
use onevcs::{
ArmedMerge, ChangeRequest, ChangeSpec, Check, CheckState, DraftReason, Error, EventKind,
FailureKind, HeldBy, Holding, Hosting, Identity, Landed, Lifecycle, MergeOutcome, MergePolicy,
PreservedBranch, Provenance, Publication, PublishOutcome, PublishRequest, Recoverable,
RemoteHost, Result, Scope, Session, SessionRecord, SessionRequest, SessionToken, Sha, Vcs,
};
use crate::events::{self, Emission};
use crate::remote::DEFAULT_HOST;
use crate::state::{self, VcsState};
use crate::store::{FileStore, MemoryStore, Store};
pub const DEFAULT_BASE: &str = "main";
pub const DEFAULT_PUBLICATION: MergePolicy = MergePolicy::ChangeOpen;
pub const DEFAULT_APPROVALS: Approvals = Approvals::Required;
#[derive(Debug)]
pub struct Repository<T> {
store: T,
root: PathBuf,
trees: Trees,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Trees {
Named,
Created,
}
pub type MemoryVcs = Repository<MemoryStore<VcsState>>;
pub type FileVcs = Repository<FileStore<VcsState>>;
impl MemoryVcs {
pub fn new() -> Self {
Self::seeded(VcsState::default())
}
pub fn seeded(state: VcsState) -> Self {
Self {
store: MemoryStore::new(state),
root: std::env::temp_dir().join("onevcs-testing-memory"),
trees: Trees::Named,
}
}
pub fn state(&self) -> VcsState {
self.store
.snapshot()
.expect("an in-memory store always answers")
}
}
impl Default for MemoryVcs {
fn default() -> Self {
Self::new()
}
}
impl FileVcs {
pub fn create(path: impl Into<PathBuf>) -> Result<Self> {
Self::over(FileStore::attach(path, &VcsState::default())?)
}
pub fn seeded(path: impl Into<PathBuf>, state: VcsState) -> Result<Self> {
Self::over(FileStore::replace(path, &state)?)
}
fn over(store: FileStore<VcsState>) -> Result<Self> {
let root = store
.path()
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or_else(|| Path::new("."))
.join("worktrees");
Ok(Self {
store,
root,
trees: Trees::Created,
})
}
pub fn state(&self) -> Result<VcsState> {
self.store.snapshot()
}
}
impl<T: Store<VcsState>> Vcs for Repository<T> {
fn resolve_identity(&self, origin_or_path: &str) -> Result<Identity> {
self.store.with(|state| {
state::identity_of(state, origin_or_path)
.cloned()
.ok_or_else(|| Error::Invalid {
reason: format!(
"{origin_or_path:?} does not name a repository this provider knows; {}",
state::known(state)
),
})
})
}
fn open_session(&self, req: SessionRequest) -> Result<Session> {
let root = self.root.clone();
let (session, emission) = self.store.with(|state| {
let identity = state::identity_of(state, &req.repo)
.cloned()
.ok_or_else(|| Error::Invalid {
reason: format!(
"{:?} does not name a repository this provider knows; {}",
req.repo,
state::known(state)
),
})?;
let token = SessionToken(format!("s-testing-{}", state.sessions.len() + 1));
let run_root = root.join(&token.0);
let base = req.base.clone().unwrap_or_else(|| DEFAULT_BASE.to_owned());
state::named_branch(&base, "the base")?;
for (key, value) in &req.labels {
state::label_pair(key, value)?;
}
let session = Session {
worktree: run_root.join("worktree"),
branch: state::requested_branch(&req, &token)?,
base,
token: token.clone(),
conflict: None,
};
state.sessions.push(session.clone());
state
.session_identities
.insert(token.clone(), identity.origin.clone());
if !req.labels.is_empty() {
state
.session_labels
.insert(token.clone(), req.labels.clone());
}
let emission = Emission {
stream: token.0.clone(),
identity: Some(identity.origin.clone()),
kind: EventKind::SessionOpened,
result: EmissionResult::Success,
payload: object(json!({
"token": token.0,
"identity": identity.origin,
"branch": session.branch,
"base": session.base,
"worktree": session.worktree.display().to_string(),
"clone": run_root.join("clone").display().to_string(),
"execution_checkout": run_root.join("checkout").display().to_string(),
"publication_checkout": run_root.join("checkout").display().to_string(),
"placement": {"kind": "run-root"},
})),
};
Ok((session, emission))
})?;
if self.trees == Trees::Created {
std::fs::create_dir_all(&session.worktree).map_err(|e| Error::Invalid {
reason: format!("cannot create {}: {e}", session.worktree.display()),
})?;
}
events::emit(&emission);
Ok(session)
}
fn adopt_session(&self, token: SessionToken) -> Result<Session> {
self.store.with(|state| {
state::session_of(state, &token)
.cloned()
.ok_or_else(|| Error::Invalid {
reason: format!(
"no session {:?} is open; `onevcs session open` prints a token",
token.0
),
})
})
}
fn preserve(&self, s: &Session, provenance: Provenance) -> Result<PreservedBranch> {
let (branch, emission) = self.store.with(|state| {
let identity = state::identity_for(state, &s.token)?;
let branch = PreservedBranch {
branch: s.branch.clone(),
base: s.base.clone(),
provenance,
change_url: None,
change_base: None,
};
let row = Recoverable {
identity: identity.clone(),
branch: branch.clone(),
checkout: s.worktree.clone(),
tip: None,
stopped_because: format!("session {} was left open", s.token.0),
recover_command: recover_command(&s.branch, &s.worktree, provenance),
landed: Landed::No,
held_by: None,
net_negative: None,
session: Some(s.token.clone()),
labels: state
.session_labels
.get(&s.token)
.cloned()
.unwrap_or_default(),
on_origin: None,
retirement: None,
};
state.preserved.retain(|kept| {
kept.identity != row.identity || kept.branch.branch != row.branch.branch
});
state.preserved.push(row);
let emission = Emission {
stream: s.token.0.clone(),
identity: None,
kind: EventKind::CommitPreserved,
result: EmissionResult::Success,
payload: object(json!({
"branch": s.branch,
"sha": events::stable_sha(&[&s.token.0, &s.branch, spell(provenance)]),
"provenance": spell(provenance),
})),
};
Ok((branch, emission))
})?;
events::emit(&emission);
Ok(branch)
}
fn session(&self, token: &SessionToken) -> Result<SessionRecord> {
self.store.with(|state| {
let session = state::session_of(state, token)
.cloned()
.ok_or_else(|| unknown_session(token))?;
let identity = state::identity_for(state, token)?;
let provenance = state
.preserved
.iter()
.find(|row| row.identity == identity && row.branch.branch == session.branch)
.map_or(Provenance::Complete, |row| row.branch.provenance);
Ok(SessionRecord {
lifecycle: if state.closed_sessions.contains(token) {
Lifecycle::Closed
} else {
Lifecycle::Open
},
session,
identity,
provenance,
retried_by: None,
})
})
}
fn close_session(&self, token: &SessionToken) -> Result<Session> {
self.store.with(|state| {
let session = state::session_of(state, token)
.cloned()
.ok_or_else(|| unknown_session(token))?;
let emission = Emission {
stream: token.0.clone(),
identity: None,
kind: EventKind::SessionClosed,
result: EmissionResult::Success,
payload: object(json!({"token": token.0, "branch": session.branch})),
};
events::emit(&emission);
state.closed_sessions.insert(token.clone());
Ok(session)
})
}
fn publish(
&self,
token: &SessionToken,
request: &PublishRequest,
hosting: &dyn Hosting,
) -> Result<Publication> {
let (publication, emissions) = self.store.with(|state| {
let session = state::session_of(state, token)
.cloned()
.ok_or_else(|| unknown_session(token))?;
let identity = state::identity_for(state, token)?;
if let Some(reason) = &request.draft {
reason.checked()?;
}
let resolved = state.policy.unwrap_or(DEFAULT_PUBLICATION);
let policy = match request.policy {
Some(requested) => resolved.narrow(requested)?,
None => resolved,
};
let published = |outcome, emissions| {
(
Publication {
session: token.clone(),
branch: session.branch.clone(),
policy,
outcome,
},
emissions,
)
};
if request.defer_checks || request.adopt.is_some() {
let approvals = state.approvals.unwrap_or(DEFAULT_APPROVALS);
let adopt = deferred(
state, &session, &identity, request, policy, approvals, token,
)?;
let number = adopt
.path_segments()
.and_then(|mut segments| segments.nth(3))
.unwrap_or_default()
.to_owned();
let emission = Emission {
stream: token.0.clone(),
identity: Some(identity.clone()),
kind: EventKind::ChangeOpened,
result: EmissionResult::Success,
payload: object(json!({
"url": adopt.to_string(),
"host": "github",
"id": number,
"base": session.base,
})),
};
let (publication, emissions) = published(
PublishOutcome::ChangeReviewDraftUnchecked(adopt),
vec![emission],
);
state.publications.push(publication.clone());
return Ok((publication, emissions));
}
if state.publications.iter().any(|earlier| {
earlier.session == *token && matches!(earlier.outcome, PublishOutcome::Merged(_))
}) {
let (publication, emissions) =
published(PublishOutcome::NothingToPublish, Vec::new());
state.publications.push(publication.clone());
return Ok((publication, emissions));
}
let (outcome, emissions) = if policy == MergePolicy::LocalDirect {
match &request.draft {
Some(reason) => (
failed(&Error::Invalid {
reason: format!(
"{branch:?} was asked to publish as {asked}, and this identity \
publishes with local-direct, which opens no change request at \
all",
branch = session.branch,
asked = asked_for(reason),
),
}),
Vec::new(),
),
None => record_local_landing(&identity, &session, token),
}
} else {
match slug(&identity) {
Some(slug) => {
let mut emissions = Vec::new();
let outcome = match publish_as_change(
hosting,
&Publishing {
slug: &slug,
identity: &identity,
session: &session,
policy,
approvals: state.approvals.unwrap_or(DEFAULT_APPROVALS),
drafts: state.drafts.unwrap_or_default(),
request,
token,
},
&mut emissions,
) {
Ok(outcome) => outcome,
Err(error) => failed(&error),
};
(outcome, emissions)
}
None => (refusal(&identity), Vec::new()),
}
};
let (publication, emissions) = published(outcome, emissions);
state.publications.push(publication.clone());
Ok((publication, emissions))
})?;
for emission in &emissions {
events::emit(emission);
}
Ok(publication)
}
fn preserved(&self, scope: Scope) -> Result<Vec<Recoverable>> {
self.recoverable(scope)
}
fn recoverable(&self, scope: Scope) -> Result<Vec<Recoverable>> {
self.store.with(|state| {
let wanted = match &scope {
Scope::All => None,
Scope::Repo(repo) => Some(
state::identity_of(state, repo)
.map(|identity| identity.origin.clone())
.ok_or_else(|| Error::Invalid {
reason: format!(
"{repo:?} does not name a repository this provider knows; {}",
state::known(state)
),
})?,
),
};
Ok(state
.preserved
.iter()
.rev()
.filter(|row| wanted.as_ref().is_none_or(|key| *key == row.identity))
.cloned()
.map(|row| held(state, row))
.collect())
})
}
}
fn held(state: &VcsState, row: Recoverable) -> Recoverable {
if row.held_by.is_some() {
return row;
}
let holder = state.sessions.iter().find(|session| {
session.branch == row.branch.branch
&& state.session_identities.get(&session.token) == Some(&row.identity)
&& !state.closed_sessions.contains(&session.token)
});
Recoverable {
held_by: holder.map(|session| HeldBy {
token: Some(session.token.clone()),
worktree: session.worktree.clone(),
holding: Holding::OwnerRunning,
}),
..row
}
}
fn unknown_session(token: &SessionToken) -> Error {
Error::Invalid {
reason: format!(
"no session {:?} is open; `onevcs session open` prints a token",
token.0
),
}
}
fn slug(identity: &str) -> Option<String> {
let mut parts = identity.split('/');
let (host, owner, name) = (parts.next()?, parts.next()?, parts.next()?);
if parts.next().is_some() || host != DEFAULT_HOST || owner.is_empty() || name.is_empty() {
return None;
}
Some(format!("{owner}/{name}"))
}
fn deferred(
state: &VcsState,
session: &Session,
identity: &str,
request: &PublishRequest,
policy: MergePolicy,
approvals: Approvals,
token: &SessionToken,
) -> Result<url::Url> {
let refused = |reason: String| Error::Invalid { reason };
let adopt = match (request.defer_checks, &request.adopt) {
(true, Some(adopt)) => adopt,
(true, None) => {
return Err(refused(
"--defer-checks publishes into a draft that is already open: name it with \
--adopt <URL>. Nothing was pushed"
.to_owned(),
))
}
(false, _) => {
return Err(refused(
"--adopt names the draft a --defer-checks publication pushes into, and this \
publication does not defer its checks. Nothing was pushed"
.to_owned(),
))
}
};
if request.draft.is_some() {
return Err(refused(
"--defer-checks pushes into a draft held for review, and a draft reason asks for a \
different draft. Nothing was pushed"
.to_owned(),
));
}
if policy != MergePolicy::ChangeOpen || approvals != Approvals::Required {
return Err(refused(format!(
"--defer-checks is for a change request held for review — a change-open identity \
whose approvals are required — and {identity} does not hold this one for review. \
Nothing was pushed"
)));
}
let segments: Vec<&str> = adopt
.path_segments()
.map(|segments| segments.collect())
.unwrap_or_default();
let named = match segments.as_slice() {
[owner, name, "pull", number, ..]
if adopt.scheme() == "https"
&& !number.is_empty()
&& number.chars().all(|c| c.is_ascii_digit()) =>
{
adopt
.host_str()
.map(|host| format!("{}/{owner}/{name}", host.to_ascii_lowercase()))
}
_ => None,
};
if named.as_deref() != Some(identity) || slug(identity).is_none() {
return Err(refused(format!(
"{adopt} is not a change request of {identity}: a change request is named as \
https://{DEFAULT_HOST}/<owner>/<name>/pull/<number>. Nothing was pushed"
)));
}
let recorded = state.publications.iter().rev().find_map(|earlier| {
let carried = match &earlier.outcome {
PublishOutcome::ChangeOpen(url)
| PublishOutcome::ChangeDraft(url)
| PublishOutcome::ChangeReviewDraft(url)
| PublishOutcome::ChangeReviewDraftUnchecked(url)
| PublishOutcome::Queued(url) => Some(url),
_ => None,
};
(carried == Some(adopt)).then(|| earlier.branch.clone())
});
match recorded {
Some(branch) if branch == session.branch => Ok(adopt.clone()),
Some(branch) => Err(refused(format!(
"{adopt} carries {branch:?} by this provider's newest record of it, and session {} \
is on {:?}. Nothing was pushed",
token.0, session.branch
))),
None => Err(Error::UnrecordedChangeHead {
change: adopt.to_string(),
reason: format!(
"this provider holds no record of which branch {adopt} carries, and a \
--defer-checks publication asks the host nothing. Nothing was pushed"
),
}),
}
}
fn record_local_landing(
identity: &str,
session: &Session,
token: &SessionToken,
) -> (PublishOutcome, Vec<Emission>) {
let sha = events::stable_sha(&["publish", &token.0, &session.branch]);
let emission = Emission {
stream: token.0.clone(),
identity: Some(identity.to_owned()),
kind: EventKind::MergeCompleted,
result: EmissionResult::Success,
payload: object(json!({
"identity": identity,
"sha": sha,
"base": session.base,
"landed_at": events::timestamp(),
"landing": sha,
})),
};
(PublishOutcome::Merged(Sha(sha)), vec![emission])
}
struct Publishing<'a> {
slug: &'a str,
identity: &'a str,
session: &'a Session,
policy: MergePolicy,
approvals: Approvals,
drafts: Drafts,
request: &'a PublishRequest,
token: &'a SessionToken,
}
impl Publishing<'_> {
fn emission(&self, kind: EventKind, payload: Value) -> Emission {
Emission {
stream: self.token.0.clone(),
identity: Some(self.identity.to_owned()),
kind,
result: EmissionResult::Success,
payload: object(payload),
}
}
}
fn publish_as_change(
hosting: &dyn Hosting,
publishing: &Publishing<'_>,
emissions: &mut Vec<Emission>,
) -> Result<PublishOutcome> {
let Publishing {
slug,
session,
policy,
request,
..
} = *publishing;
let host = hosting.for_repo(slug)?;
let author = host.authenticated_user()?;
let lifecycle = !publishing.drafts.disabled.unwrap_or(false);
if lifecycle {
grace_seconds()?;
}
let existing = host.find_changes(&session.branch, &session.base)?;
let adopted = !existing.is_empty();
let change = match existing.into_iter().next() {
Some(change) => change,
None => host.open_change(ChangeSpec {
head: session.branch.clone(),
base: session.base.clone(),
title: request
.title
.as_deref()
.map_or_else(|| format!("Publish {}", session.branch), str::to_owned),
body: request.body.clone(),
draft: request.draft.clone(),
draft_awaiting_checks: lifecycle && request.draft.is_none(),
})?,
};
emissions.push(publishing.emission(
EventKind::ChangeOpened,
json!({
"url": change.url.to_string(),
"host": "github",
"id": change.id.0,
"base": change.base,
"author": author,
}),
));
if let Some(reason) = &request.draft {
if !host.is_draft(&change)? {
return Err(Error::Invalid {
reason: format!(
"{url} is open for review on the host, and this publication asked for \
{asked}. A change that is open can land, so reporting it as a draft would \
say the work is held back when nothing is holding it",
url = change.url,
asked = asked_for(reason),
),
});
}
let mut drafted = object(json!({
"url": change.url.to_string(),
"id": change.id.0,
"base": change.base,
}));
if let serde_json::Value::Object(fields) =
serde_json::to_value(reason).expect("a draft reason serializes")
{
drafted.extend(fields);
}
emissions.push(publishing.emission(EventKind::ChangeDrafted, Value::Object(drafted)));
return Ok(PublishOutcome::ChangeDraft(change.url.clone()));
}
let is_draft = match if lifecycle || adopted {
host.is_draft(&change)
} else {
Ok(false)
} {
Ok(draft) => draft,
Err(Error::NotImplemented { .. }) => false,
Err(unreadable) => return Err(unreadable),
};
let drafted = if lifecycle {
if is_draft {
emissions.push(publishing.emission(
EventKind::ChangeDrafted,
json!({
"url": change.url.to_string(),
"id": change.id.0,
"base": change.base,
"kind": "awaiting-checks",
}),
));
}
is_draft
} else {
if adopted && is_draft {
lift(host.as_ref(), publishing, &change, emissions)?;
}
false
};
let still_draft = if drafted || policy != MergePolicy::ChangeAuto {
match watch(host.as_ref(), publishing, &change, drafted, emissions)? {
Watched::Draft => true,
Watched::Ready => false,
Watched::Elapsed { error, skipped } => {
return match policy {
MergePolicy::ChangeAuto | MergePolicy::ChangeDirect => arm_on_elapse(
host.as_ref(),
publishing,
&change,
emissions,
error,
&skipped,
),
MergePolicy::ChangeOpen | MergePolicy::LocalDirect => Err(error),
};
}
}
} else {
let answered = host.change_checks(&change)?;
let checks: Vec<&Check> = answered.checks.iter().collect();
let declared = given(
declared_required(host.as_ref(), &change),
answered.complete(),
&checks,
);
let standing = standings(&checks, &declared);
if let Some(failed) = red(&checks, &standing) {
return Err(failed);
}
if through(&standing, &declared) {
record_settled(
publishing,
&change,
emissions,
skipped(&standing),
declared.unread(),
);
}
false
};
if still_draft {
if policy == MergePolicy::ChangeOpen && publishing.approvals == Approvals::Required {
emissions.push(publishing.emission(
EventKind::DraftKeptForReview,
json!({
"url": change.url.to_string(),
"id": change.id.0,
"base": change.base,
}),
));
return Ok(PublishOutcome::ChangeReviewDraft(change.url.clone()));
}
lift(host.as_ref(), publishing, &change, emissions)?;
}
if policy == MergePolicy::ChangeOpen {
return Ok(PublishOutcome::ChangeOpen(change.url.clone()));
}
Ok(match host.merge(&change, policy)? {
MergeOutcome::Merged(sha) => merged(host.as_ref(), publishing, &change, emissions, &sha),
MergeOutcome::Queued => PublishOutcome::Queued(change.url.clone()),
MergeOutcome::Open => PublishOutcome::ChangeOpen(change.url.clone()),
})
}
fn merged(
host: &dyn RemoteHost,
publishing: &Publishing<'_>,
change: &ChangeRequest,
emissions: &mut Vec<Emission>,
sha: &Sha,
) -> PublishOutcome {
let landed_at = host
.merge_time(change)
.ok()
.flatten()
.and_then(|spelled| events::moment(&spelled));
emissions.push(publishing.emission(
EventKind::ChangeMerged,
json!({
"url": change.url.to_string(),
"sha": sha.0,
"landed_at": landed_at,
"landing": sha.0,
}),
));
emissions.push(publishing.emission(
EventKind::MergeCompleted,
json!({
"identity": publishing.identity,
"sha": sha.0,
"landed_at": landed_at,
"landing": sha.0,
}),
));
PublishOutcome::Merged(sha.clone())
}
fn arm_on_elapse(
host: &dyn RemoteHost,
publishing: &Publishing<'_>,
change: &ChangeRequest,
emissions: &mut Vec<Emission>,
elapsed: Error,
skipped: &[String],
) -> Result<PublishOutcome> {
let Error::ChecksUnsettled { reason } = elapsed else {
return Err(elapsed);
};
let unarmed = |why: String| Error::ChecksUnsettled {
reason: format!(
"{reason}. {why}, so {url} is left as it was — nothing lifted it or armed its merge",
url = change.url
),
};
if !skipped.is_empty() {
return Err(unarmed(format!(
"Required check {} concluded skipped, which the host's auto-merge would read as \
passed",
skipped.join(", ")
)));
}
match host.allows_auto_merge() {
Ok(true) => {}
Ok(false) => {
return Err(unarmed(
"The repository does not allow auto-merge".to_owned(),
))
}
Err(unread) => {
return Err(unarmed(format!(
"Whether the repository allows auto-merge could not be read ({unread})"
)))
}
}
let is_draft = match host.is_draft(change) {
Ok(draft) => draft,
Err(Error::NotImplemented { .. }) => false,
Err(unreadable) => return Err(unreadable),
};
if is_draft {
lift(host, publishing, change, emissions)?;
}
if let MergeOutcome::Merged(sha) = host.merge(change, MergePolicy::ChangeAuto)? {
return Ok(merged(host, publishing, change, emissions, &sha));
}
emissions.push(publishing.emission(
EventKind::MergeArmed,
json!({"url": change.url.to_string(), "head": change.head_sha.0}),
));
Ok(PublishOutcome::Failed {
kind: FailureKind::ChecksUnsettled,
reason: format!(
"{}. Its merge is armed: the host lands {url} once its required checks pass, and \
publishing the session again watches it",
Error::ChecksUnsettled { reason },
url = change.url,
),
retained: None,
armed: Some(ArmedMerge {
change: change.url.clone(),
head: change.head_sha.0.clone(),
}),
})
}
fn lift(
host: &dyn RemoteHost,
publishing: &Publishing<'_>,
change: &ChangeRequest,
emissions: &mut Vec<Emission>,
) -> Result<()> {
host.ready_for_review(change)?;
emissions.push(publishing.emission(
EventKind::DraftLifted,
json!({"url": change.url.to_string(), "id": change.id.0}),
));
Ok(())
}
#[derive(Clone, PartialEq, Eq)]
enum Declared {
Nothing,
Names(BTreeSet<String>),
Unknown(String),
}
impl Declared {
fn unread(&self) -> Option<&str> {
match self {
Declared::Unknown(because) => Some(because),
Declared::Nothing | Declared::Names(_) => None,
}
}
}
fn declared_required(host: &dyn RemoteHost, change: &ChangeRequest) -> Declared {
match host.required_checks_on(&change.base) {
Ok(answer) if answer.checks.is_empty() && answer.complete() => Declared::Nothing,
Ok(answer) if !answer.checks.is_empty() => Declared::Names(answer.checks),
Ok(_) => Declared::Unknown(format!(
"the host answered only in part about which checks a merge into {} requires",
change.base
)),
Err(refused) => Declared::Unknown(format!(
"the host would not say which checks a merge into {} requires: {refused}",
change.base
)),
}
}
fn given(declared: Declared, complete: bool, checks: &[&Check]) -> Declared {
if matches!(declared, Declared::Unknown(_))
&& complete
&& !checks.is_empty()
&& checks.iter().all(|check| !check.required)
{
return Declared::Nothing;
}
declared
}
fn standings(checks: &[&Check], declared: &Declared) -> Vec<(String, Option<CheckState>)> {
let names: BTreeSet<String> = match declared {
Declared::Names(names) => names.clone(),
Declared::Nothing => BTreeSet::new(),
Declared::Unknown(_) => checks
.iter()
.filter(|check| check.required)
.map(|check| check.name.clone())
.collect(),
};
let rank = |state: CheckState| match state {
CheckState::Failed => 0,
CheckState::Pending => 1,
CheckState::Passed => 2,
CheckState::NoVerdict => 3,
CheckState::Skipped => 4,
};
names
.into_iter()
.map(|name| {
let state = checks
.iter()
.filter(|check| check.name == name)
.map(|check| check.state())
.min_by_key(|state| rank(*state));
(name, state)
})
.collect()
}
fn red(checks: &[&Check], standing: &[(String, Option<CheckState>)]) -> Option<Error> {
let (name, _) = standing
.iter()
.find(|(_, state)| *state == Some(CheckState::Failed))?;
let check = checks
.iter()
.find(|check| &check.name == name && check.state() == CheckState::Failed)?;
Some(Error::ChecksFailed {
reason: format!(
"required check {:?} concluded {}.",
check.name,
check
.conclusion
.as_deref()
.unwrap_or("without a conclusion")
),
})
}
fn through(standing: &[(String, Option<CheckState>)], declared: &Declared) -> bool {
(*declared == Declared::Nothing || !standing.is_empty())
&& standing
.iter()
.all(|(_, state)| matches!(state, Some(CheckState::Passed | CheckState::Skipped)))
}
fn unsettled_named(
checks: &[&Check],
standing: &[(String, Option<CheckState>)],
declared: &Declared,
) -> String {
let pending: Vec<String> = standing
.iter()
.filter_map(|(name, state)| match state {
Some(CheckState::Pending) => Some(format!("{name:?}")),
None => Some(format!("{name:?} (no run yet)")),
_ => None,
})
.collect();
let no_verdict: Vec<String> = standing
.iter()
.filter(|(_, state)| *state == Some(CheckState::NoVerdict))
.map(|(name, _)| {
let conclusion = checks
.iter()
.find(|check| &check.name == name && check.state() == CheckState::NoVerdict)
.and_then(|check| check.conclusion.as_deref())
.unwrap_or("unknown");
format!("{name:?}: {conclusion}")
})
.collect();
let skipped: Vec<String> = skipped(standing)
.iter()
.map(|name| format!("{name:?}"))
.collect();
let named = if !pending.is_empty() {
let verdicts = if no_verdict.is_empty() {
String::new()
} else {
format!("; completed with no verdict: {}", no_verdict.join(", "))
};
format!("still unsettled: {}{verdicts}", pending.join(", "))
} else if !no_verdict.is_empty() {
format!("completed with no verdict: {}", no_verdict.join(", "))
} else if !skipped.is_empty() {
format!(
"skipped on the draft, so not run on it: {}",
skipped.join(", ")
)
} else if standing.is_empty() {
match declared {
Declared::Nothing => "the host declared no required check on it at all".to_owned(),
_ => "it has marked no check required on it".to_owned(),
}
} else {
"every required check it declared had settled".to_owned()
};
let read_from = declared
.unread()
.map(|because| {
format!(
". Which checks it requires could not be read, so they were read from the \
checks it marked required: {because}"
)
})
.unwrap_or_default();
format!("{named}{read_from}")
}
fn skipped(standing: &[(String, Option<CheckState>)]) -> Vec<String> {
standing
.iter()
.filter(|(_, state)| *state == Some(CheckState::Skipped))
.map(|(name, _)| name.clone())
.collect()
}
fn record_settled(
publishing: &Publishing<'_>,
change: &ChangeRequest,
emissions: &mut Vec<Emission>,
skipped: Vec<String>,
unread: Option<&str>,
) {
let verdict = if skipped.is_empty() {
"passed"
} else {
"passed-with-skipped"
};
let mut payload = json!({
"url": change.url.to_string(),
"id": change.id.0,
"head": change.head_sha.0,
"verdict": verdict,
"skipped": skipped,
});
if let Some(because) = unread {
eprintln!(
"onevcs: warning: the required checks on {} were read from the host's own \
per-check marking, because the declaration could not be read: {because}",
change.url
);
payload["requirement"] = json!({"read_from": "host-marking", "because": because});
}
let mut settled = publishing.emission(EventKind::ChecksSettled, payload);
settled.result = if verdict == "passed" {
EmissionResult::Passed
} else {
EmissionResult::PassedWithSkipped
};
emissions.push(settled);
}
fn watch(
host: &dyn RemoteHost,
publishing: &Publishing<'_>,
change: &ChangeRequest,
drafted: bool,
emissions: &mut Vec<Emission>,
) -> Result<Watched> {
let settled = |emissions: &mut Vec<Emission>, skipped: Vec<String>, unread: Option<&str>| {
record_settled(publishing, change, emissions, skipped, unread);
};
let elapsed_naming = |checks: &[&Check],
standing: &[(String, Option<CheckState>)],
declared: &Declared,
skipped: Vec<String>| Watched::Elapsed {
error: Error::ChecksUnsettled {
reason: format!(
"the host had not settled its required checks on {}; {}",
change.url,
unsettled_named(checks, standing, declared)
),
},
skipped,
};
let unsettled =
|checks: &[&Check], standing: &[(String, Option<CheckState>)], declared: &Declared| {
elapsed_naming(checks, standing, declared, skipped(standing))
};
let answered = host.change_checks(change)?;
let complete = answered.complete();
let checks: Vec<&Check> = answered.checks.iter().collect();
let declared = given(declared_required(host, change), complete, &checks);
let unread = declared.unread();
if !drafted {
let standing = standings(&checks, &declared);
if let Some(failed) = red(&checks, &standing) {
return Err(failed);
}
if !through(&standing, &declared) {
return Ok(unsettled(&checks, &standing, &declared));
}
settled(emissions, skipped(&standing), unread);
return Ok(Watched::Ready);
}
if matches!(declared, Declared::Nothing) {
settled(emissions, Vec::new(), None);
return Ok(Watched::Draft);
}
let standing = standings(&checks, &declared);
if let Some(failed) = red(&checks, &standing) {
return Err(failed);
}
if !standing.is_empty()
&& standing
.iter()
.all(|(_, state)| *state == Some(CheckState::Passed))
{
settled(emissions, Vec::new(), unread);
return Ok(Watched::Draft);
}
let not_run: Vec<String> = standing
.iter()
.filter(|(_, state)| matches!(state, None | Some(CheckState::Skipped)))
.map(|(name, _)| name.clone())
.collect();
let running = standing
.iter()
.any(|(_, state)| matches!(state, Some(CheckState::Pending | CheckState::NoVerdict)));
let unseen = matches!(declared, Declared::Unknown(_)) && standing.is_empty();
if unseen
&& !checks.is_empty()
&& checks
.iter()
.all(|check| check.state() != CheckState::Skipped)
{
settled(emissions, Vec::new(), unread);
return Ok(Watched::Draft);
}
if running || (not_run.is_empty() && !unseen) || still_running(host, change, &checks) {
return Ok(unsettled(&checks, &standing, &declared));
}
let snapshot: Vec<Check> = answered.checks.clone();
lift(host, publishing, change, emissions)?;
let warned = publishing.drafts.warn_on_early_lift.unwrap_or(true);
let grace = grace_seconds()?;
if warned {
eprintln!(
"onevcs: warning: {} was lifted out of its draft before its required checks ran on \
it ({}): none had run within {grace}s and nothing on its head was still running, \
which is what a workflow that skips drafts looks like. {DRAFT_CI}",
change.url,
not_run.join(", ")
);
}
emissions.push(publishing.emission(
EventKind::DraftLiftedEarly,
json!({
"url": change.url.to_string(),
"id": change.id.0,
"base": change.base,
"awaited": not_run,
"grace_seconds": grace,
"warned": warned,
}),
));
let after = host.change_checks(change)?.checks;
let from_the_draft = |check: &Check| !ran_after_the_lift(check, &snapshot);
let counted: Vec<&Check> = after
.iter()
.filter(|check| !(from_the_draft(check) && check.state() == CheckState::Skipped))
.collect();
let standing = standings(&counted, &declared);
if let Some(failed) = red(&counted, &standing) {
return Err(failed);
}
let rerun = after.iter().any(|check| {
!from_the_draft(check)
&& (standing.is_empty() || standing.iter().any(|(name, _)| *name == check.name))
});
let reported: Vec<&Check> = after.iter().collect();
let elapsed = || {
let mut named = skipped(&standing);
for name in skipped(&standings(&reported, &declared)) {
if !named.contains(&name) {
named.push(name);
}
}
elapsed_naming(&counted, &standing, &declared, named)
};
if !rerun {
if still_running(host, change, &reported) {
return Ok(elapsed());
}
return Err(Error::ChecksUnsettled {
reason: format!(
"the required checks on {} did not re-run after its draft was lifted: no run \
the host attached after the lift appeared and nothing on its head is still \
running, while the runs from while it was a draft were skipped or never \
started, which is not a verdict. {DRAFT_CI}",
change.url
),
});
}
if !standing
.iter()
.all(|(_, state)| matches!(state, Some(CheckState::Passed | CheckState::Skipped)))
{
return Ok(elapsed());
}
settled(emissions, skipped(&standing), unread);
Ok(Watched::Ready)
}
const DRAFT_CI: &str = "Run the required checks on drafts: their results attach to the head \
commit, so the change is then kept a draft until they are green, and a workflow that runs \
on drafts needs no `ready_for_review` trigger. Only a workflow that skips its required \
jobs on drafts needs `ready_for_review` to run them after a lift, at the cost of one more \
run per lift";
fn still_running(host: &dyn RemoteHost, change: &ChangeRequest, checks: &[&Check]) -> bool {
checks.iter().any(|check| !check.settled())
|| host
.head_activity(change)
.is_ok_and(|answer| answer.head == change.head_sha && !answer.in_progress.is_empty())
}
enum Watched {
Draft,
Ready,
Elapsed {
error: Error,
skipped: Vec<String>,
},
}
fn ran_after_the_lift(check: &Check, snapshot: &[Check]) -> bool {
let earlier: Vec<&Check> = snapshot
.iter()
.filter(|seen| seen.name == check.name)
.collect();
if earlier.is_empty() {
return true;
}
check.started_at.as_ref().is_some_and(|started| {
earlier
.iter()
.all(|seen| seen.started_at.as_ref() != Some(started))
})
}
const DEFAULT_DRAFT_GRACE_SECONDS: f64 = 120.0;
const DRAFT_GRACE_ENV: &str = "ONEVCS_DRAFT_CHECKS_GRACE_SECONDS";
fn grace_seconds() -> Result<f64> {
let Some(raw) = std::env::var_os(DRAFT_GRACE_ENV) else {
return Ok(DEFAULT_DRAFT_GRACE_SECONDS);
};
let raw = raw.to_string_lossy().into_owned();
let seconds: f64 = raw.trim().parse().map_err(|_| Error::Invalid {
reason: format!("{DRAFT_GRACE_ENV} must be a number of seconds, not {raw:?}"),
})?;
if !seconds.is_finite() || seconds <= 0.0 {
return Err(Error::Invalid {
reason: format!(
"{DRAFT_GRACE_ENV} must be a finite number of seconds above zero, not {raw:?}"
),
});
}
Ok(seconds)
}
fn asked_for(reason: &DraftReason) -> String {
match reason {
DraftReason::AwaitingRelease {
awaiting, target, ..
} => format!("a draft awaiting {awaiting} {target}"),
DraftReason::Held { because } => {
format!("a draft held by the session that opened it ({because})")
}
}
}
fn refusal(identity: &str) -> PublishOutcome {
failed(&if identity.split('/').count() == 3 {
Error::NotImplemented {
operation: "RemoteHost for a host other than github.com",
}
} else {
Error::Invalid {
reason: format!(
"identity {identity:?} is not a hosted repository, so it cannot publish a \
change request; a local identity publishes with local-direct"
),
}
})
}
fn failed(error: &Error) -> PublishOutcome {
PublishOutcome::Failed {
kind: FailureKind::of(error),
reason: error.to_string(),
retained: None,
armed: None,
}
}
fn recover_command(branch: &str, checkout: &Path, provenance: Provenance) -> Vec<String> {
let verb = match provenance {
Provenance::IncompleteStep => "recover",
Provenance::Complete => "publish-branch",
};
vec![
"onevcs".to_owned(),
verb.to_owned(),
branch.to_owned(),
"--repo".to_owned(),
checkout.display().to_string(),
]
}
fn spell(provenance: Provenance) -> &'static str {
match provenance {
Provenance::Complete => "complete",
Provenance::IncompleteStep => "incomplete-step",
}
}
fn object(value: Value) -> Map<String, Value> {
value.as_object().cloned().unwrap_or_default()
}