scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-14 23:22:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Let the driver retry git operations and decide from the credential's age

Browse source 
Fabro carried its own retry loop, its own reading of what a git failure
class means for the credentials in hand, and a credential context derived
from the token snapshot. The driver now owns the loop and the decision:
a rejected credential retries only while its mint time is within the
replication horizon, a remote that could not be reached retries on its
own, a static credential fails fast, and an operation whose outcome is
unknown is never replayed. Fabro keeps its budgets as retry policies
(clone, repository probe, checkpoint push, publish push), hands the mint
time along with the token, and records the driver's attempt history as
the push attempts the events carry. Host-side git (the repository probe
and the metadata push classification) goes through the same decision
from its rendered message.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-09-11 12:44:58 -06:00
parent dfe378c8ae
commit 22cd5d3382
No known key found for this signature in database
11 changed files with 451 additions and 914 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
lib/apps/fabro-server/src/run_manifest.rs
View file

@ -1501,23 +1501,21 @@ async fn probe_github_repository(
/// Retry auth-shaped failures with the SAME token: replication of a given
/// token only makes progress, while re-minting would restart the replication
/// clock. The sandbox git retry executor owns attempt limits,
/// classification, and pacing.
/// clock. The driver's git retry owns the decision and the pacing; fabro's
/// probe policy owns the attempt count.
async fn probe_with_replication_retry<F, Fut>(
snapshot: TokenSnapshot,
mut run: F,
run: F,
) -> std::result::Result<(), String>
where
F: FnMut() -> Fut,
Fut: Future<Output = std::result::Result<(), String>>,
{
let credential_context = fabro_sandbox::CredentialContext::from_snapshot(Some(&snapshot));
fabro_sandbox::retry_git_operation(
SandboxProviderKind::LOCAL,
fabro_sandbox::retry_git_messages(
&fabro_sandbox::repository_probe_policy(),
Some(&snapshot),
"repository probe",
&fabro_sandbox::RetryPlan::repository_probe(),
|_attempt| run(),
|message| fabro_sandbox::classify_failure(message, credential_context),
run,
)
.await
}

54
lib/components/fabro-sandbox/src/clone.rs
View file

@ -23,7 +23,7 @@ use tokio::time;
use crate::clone_source::{self, GitHubRepoLayout};
use crate::credentials::{self, RepoCredentials};
use crate::exec::{ExecResultExt, SandboxExec};
use crate::git_retry::{self, CredentialContext, GitRetryReason, RetryPlan};
use crate::git_policy;
/// Whole-clone budget, shared by every network and local step.
pub(crate) const GIT_CLONE_TIMEOUT: Duration = Duration::from_mins(5);
@ -57,11 +57,6 @@ enum CloneStep {
Local,
}
struct CloneFailure {
error: crate::Error,
retry_reason: Option<GitRetryReason>,
}
/// Clone `plan` into `handle`, laid out under `workspace_root` and
/// `repos_root`, with a GitHub App token from `credentials` when one is
/// available: the clone carries it per call, and the checkout keeps it as
@ -77,8 +72,6 @@ pub(crate) async fn clone_github_repo(
) -> crate::Result<CloneOutcome> {
let layout = clone_source::github_repo_layout(&plan.origin_url, workspace_root, repos_root)?;
let token = credentials.mint_for_clone().await?;
let credential_context =
CredentialContext::from_snapshot(token.as_ref().map(|token| &token.snapshot));
let fs = handle.fs();
for dir in [workspace_root, layout.repos_owner_path.as_str()] {
@ -106,37 +99,30 @@ pub(crate) async fn clone_github_repo(
options.tag = plan.tag.clone().filter(|_| plan.commit_sha.is_none());
options.depth = plan.depth;
options.credentials = token.as_ref().map(credentials::git_credentials);
let retry_plan = RetryPlan::clone_default(Some(deadline));
// The driver retries a clone the remote refused while the token may
// still be replicating, inside what is left of the clone budget.
let policy = git_policy::clone_policy(deadline.saturating_duration_since(time::Instant::now()));
let target = layout.primary_repo_path.clone();
git_retry::retry_git_operation(
kind.clone(),
"clone",
&retry_plan,
|_attempt| {
let options = options.clone();
let target = target.clone();
let origin_url = plan.origin_url.clone();
sandbox_driver::retry_git(
&policy,
options.credentials.as_ref(),
"git clone",
|_attempt, _timeout| {
let git = &git;
async move {
git.clone_repo(&origin_url, &target, &options)
.await
.map_err(|error| CloneFailure {
retry_reason: git_retry::classify_driver_failure(
&error,
credential_context,
),
error: clone_failure_error(
crate::Error::driver_error(error),
CloneStep::Network,
has_app,
),
})
}
let options = &options;
let target = &target;
let origin_url = &plan.origin_url;
async move { git.clone_repo(origin_url, target, options).await }
},
|failure: &CloneFailure| failure.retry_reason,
)
.await
.map_err(|failure| failure.error)?;
.map_err(|failure| {
clone_failure_error(
crate::Error::driver_error(failure.error),
CloneStep::Network,
has_app,
)
})?;
run_local_step(
exec,

15
lib/components/fabro-sandbox/src/credentials.rs
View file

@ -11,6 +11,7 @@
//! [`InstallationTokenSource`].
use std::sync::Arc;
use std::time::SystemTime;
use fabro_github::GitHubCredentials;
use fabro_github::token_source::{InstallationTokenSource, ResolvedToken};
@ -119,9 +120,15 @@ impl RepoCredentials {
}
}
/// The per-call form of `token` for the driver's network operations.
/// The per-call form of `token` for the driver's network operations. The
/// mint time travels with a minted token so the driver's retry knows a
/// rejection may be replication lag; a static credential carries none.
pub(crate) fn git_credentials(token: &ResolvedToken) -> GitCredentials {
GitCredentials::new(GITHUB_TOKEN_USERNAME, token.token.expose())
let credentials = GitCredentials::new(GITHUB_TOKEN_USERNAME, token.token.expose());
match token.snapshot.minted_at() {
Some(minted_at) => credentials.minted_at(SystemTime::from(minted_at)),
None => credentials,
}
}
#[cfg(test)]
@ -143,5 +150,9 @@ mod tests {
let credentials = git_credentials(&token);
assert_eq!(credentials.username, GITHUB_TOKEN_USERNAME);
assert_eq!(credentials.password, "ghp_static");
assert!(
credentials.minted_at.is_none(),
"a static credential has no mint time"
);
}
}

10
lib/components/fabro-sandbox/src/driver_sandbox.rs
View file

@ -24,7 +24,7 @@ use fabro_types::SandboxProviderKind;
use fabro_util::workspace_glob::WorkspaceGlob;
use sandbox_driver::{
DirEntry, EventContext, ExecControls, ExecResult, ExecSpec, ExecStreamingResult, FileKind,
GrepMatch, GrepOptions, LifecycleTimers, PtyOptions, PtySession, PtySize,
GitRetryPolicy, GrepMatch, GrepOptions, LifecycleTimers, PtyOptions, PtySession, PtySize,
Sandbox as DriverHandle, SandboxProvider as DriverProvider, SandboxSource,
SandboxSpec as DriverSpec, SandboxState, Search as _, StdioProcess, WaitOptions, WalkOptions,
};
@ -37,7 +37,7 @@ use crate::clone::{self, GitHubClone};
use crate::clone_source::{self, CloneDecision, EmptyWorkspaceReason};
use crate::credentials::{self, RepoCredentials};
use crate::environment::CloneRequest;
use crate::{GitRunInfo, GitSetupIntent, RetryPlan};
use crate::{GitRunInfo, GitSetupIntent};
/// A sandbox on the worker host at `working_directory`, the fabro `local`
/// kind, served by the driver's in-process Host provider.
@ -1051,7 +1051,7 @@ impl RunSandbox {
pub async fn git_push_ref(
&self,
refspec: &str,
plan: &RetryPlan,
policy: &GitRetryPolicy,
) -> Result<PushReport, PushError> {
let Some(workspace) = &self.workspace else {
// A designated directory: push only when the checkout has an
@ -1072,12 +1072,12 @@ impl RunSandbox {
if !has_origin {
return Ok(PushReport::default());
}
return sandbox::git_push(self, None, refspec, plan).await;
return sandbox::git_push(self, None, refspec, policy).await;
};
if !workspace.repo_cloned() {
return Ok(PushReport::default());
}
sandbox::git_push(self, Some(&workspace.credentials), refspec, plan).await
sandbox::git_push(self, Some(&workspace.credentials), refspec, policy).await
}
pub fn origin_url(&self) -> Option<&str> {

278
lib/components/fabro-sandbox/src/git_policy.rs Normal file
View file

@ -0,0 +1,278 @@
//! Fabro's retry budgets for git operations against GitHub.
//!
//! The driver owns the retry loop and the decision
//! ([`sandbox_driver::retry_git`]): a remote that cannot be reached is retried,
//! a rejected credential is retried only while the token is fresh enough to
//! still be replicating to GitHub's git endpoints, a static credential fails
//! fast, and a command whose outcome is unknown is never replayed. Fabro keeps
//! what is policy: how many attempts each operation gets, how long the
//! operation may take, and when the credential it pushes with was minted.
//!
//! Retries reuse the same token on purpose. Replication of a given token
//! only makes progress, so each attempt strictly improves the odds, while
//! re-minting would restart the replication clock.
use std::future::Future;
use std::sync::{Mutex, PoisonError};
use std::time::{Duration, SystemTime};
use fabro_github::token_source::TokenSnapshot;
pub use fabro_types::run_event::GitPushRetryReason as GitRetryReason;
use sandbox_driver::{GitBackoff, GitCredentials, GitFailure, GitFailureKind, GitRetryPolicy};
use crate::credentials::GITHUB_TOKEN_USERNAME;
/// Backoff between attempts: 3s, then 9s.
///
/// GitHub's guidance for token replication is to wait a few seconds and
/// retry with the same token. Sub-second delays land inside the same
/// replication window and spend an attempt for nothing.
fn replication_backoff() -> GitBackoff {
GitBackoff::new(Duration::from_secs(3), 3.0, Duration::from_secs(10))
}
/// The clone policy: 3 attempts at replication pacing, inside whatever is
/// left of the whole-clone budget.
pub(crate) fn clone_policy(remaining: Duration) -> GitRetryPolicy {
GitRetryPolicy::new(3, replication_backoff()).max_elapsed(remaining)
}
/// Host-side repository probes use the clone's attempt count and pacing,
/// with no deadline of their own.
#[must_use]
pub fn repository_probe_policy() -> GitRetryPolicy {
GitRetryPolicy::new(3, replication_backoff())
}
/// Checkpoint pushes stay cheap: the next checkpoint re-pushes the same
/// branch anyway. Worst case about 90 seconds of wall clock.
#[must_use]
pub fn checkpoint_push_policy() -> GitRetryPolicy {
GitRetryPolicy::new(3, replication_backoff())
.max_elapsed(Duration::from_secs(90))
.per_attempt_timeout(Duration::from_mins(1))
}
/// The terminal publish push guards the whole run's value, so it gets a
/// real budget: 5 attempts with growing backoff (about 3s, 10s, 33s, 60s),
/// bounded at 4 minutes of wall clock. The bound must stay under the token
/// source's `REFRESH_MARGIN` (see the margin-invariant test) so a pinned
/// token always outlives the operation.
#[must_use]
pub fn publish_push_policy() -> GitRetryPolicy {
GitRetryPolicy::new(
5,
GitBackoff::new(Duration::from_secs(3), 10.0 / 3.0, Duration::from_mins(1)),
)
.max_elapsed(Duration::from_mins(4))
.per_attempt_timeout(Duration::from_mins(1))
}
/// The reason fabro records for a driver retry reason. A reason this build
/// does not know still retried the attempt, so it is recorded under the
/// broader class.
pub(crate) fn recorded_reason(reason: sandbox_driver::GitRetryReason) -> GitRetryReason {
match reason {
sandbox_driver::GitRetryReason::TokenReplication => GitRetryReason::TokenReplication,
_ => GitRetryReason::TransientInfra,
}
}
/// Credentials carrying only the token's mint time, which is all the
/// driver's decision reads for git that ran outside a sandbox. The token
/// itself never leaves its snapshot.
fn credential_age(snapshot: Option<&TokenSnapshot>) -> Option<GitCredentials> {
let snapshot = snapshot?;
let credentials = GitCredentials::new(GITHUB_TOKEN_USERNAME, "");
Some(match snapshot.minted_at() {
Some(minted_at) => credentials.minted_at(SystemTime::from(minted_at)),
None => credentials,
})
}
/// The driver's failure for a rendered git message, so git that ran
/// outside a sandbox (the host-side repository probe, the metadata push)
/// is classified the same way as git the driver ran.
fn classified_failure(operation: &str, message: &str) -> sandbox_driver::Error {
sandbox_driver::Error::Git(GitFailure::classified(
operation,
GitFailureKind::from_message(message),
None,
))
}
/// Whether a rendered git failure `message` is worth retrying with the
/// token behind `snapshot`: `None` means the failure is permanent for
/// these credentials or unrecognized.
#[must_use]
pub fn transient_git_failure(
message: &str,
snapshot: Option<&TokenSnapshot>,
) -> Option<GitRetryReason> {
let credentials = credential_age(snapshot);
sandbox_driver::retry_reason(&classified_failure("git", message), credentials.as_ref())
.map(recorded_reason)
}
/// Runs a host-side git operation that reports failures as rendered
/// messages under `policy`, retrying while the driver's decision says the
/// message is transient for the token behind `snapshot`. The final failure
/// comes back as the operation's own message.
pub async fn retry_git_messages<F, Fut>(
policy: &GitRetryPolicy,
snapshot: Option<&TokenSnapshot>,
operation: &str,
mut run: F,
) -> Result<(), String>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<(), String>>,
{
let credentials = credential_age(snapshot);
// The operation's own message is kept beside the classified failure the
// driver decides on, so the caller reads the message it knows.
let last_message = Mutex::new(None);
let result = sandbox_driver::retry_git(
policy,
credentials.as_ref(),
operation,
|_attempt, _timeout| {
let attempt = run();
let last_message = &last_message;
async move {
attempt.await.map_err(|message| {
let error = classified_failure(operation, &message);
*last_message.lock().unwrap_or_else(PoisonError::into_inner) = Some(message);
error
})
}
},
)
.await;
match result {
Ok(_) => Ok(()),
Err(failure) => Err(last_message
.into_inner()
.unwrap_or_else(PoisonError::into_inner)
.unwrap_or_else(|| failure.error.to_string())),
}
}
#[cfg(test)]
mod tests {
use chrono::Utc;
use fabro_github::token_source::{REFRESH_MARGIN, TokenProvenance};
use super::*;
fn snapshot(age: Duration) -> TokenSnapshot {
let now = Utc::now();
TokenSnapshot {
generation: 1,
provenance: TokenProvenance::Minted {
minted_at: now - chrono::Duration::from_std(age).unwrap(),
expires_at: now + chrono::Duration::hours(1),
},
}
}
fn static_snapshot() -> TokenSnapshot {
TokenSnapshot {
generation: 0,
provenance: TokenProvenance::Static,
}
}
#[test]
fn not_found_follows_the_credential_age() {
let message = "repository not found: Repository not found.";
assert_eq!(
transient_git_failure(message, Some(&snapshot(Duration::from_secs(5)))),
Some(GitRetryReason::TokenReplication)
);
assert_eq!(
transient_git_failure(message, Some(&snapshot(Duration::from_mins(2)))),
Some(GitRetryReason::TransientInfra)
);
assert_eq!(
transient_git_failure(message, Some(&static_snapshot())),
None
);
assert_eq!(transient_git_failure(message, None), None);
}
#[test]
fn infrastructure_failures_retry_without_credentials() {
assert_eq!(
transient_git_failure("fatal: unable to access: Could not resolve host", None),
Some(GitRetryReason::TransientInfra)
);
assert_eq!(
transient_git_failure("fatal: something else entirely", None),
None
);
}
/// `REFRESH_MARGIN` must exceed every push policy's `max_elapsed`: a
/// push resolves its token once, and the token the source returns has
/// at least the margin of validity left, so the pinned token outlives
/// the operation.
#[test]
fn refresh_margin_exceeds_every_push_policy_elapsed_bound() {
for policy in [checkpoint_push_policy(), publish_push_policy()] {
let max_elapsed = policy.max_elapsed.expect("push policies are bounded");
assert!(
REFRESH_MARGIN > max_elapsed,
"margin invariant violated: {max_elapsed:?}"
);
}
}
#[test]
fn publish_backoff_grows_toward_a_one_minute_cap() {
let backoff = publish_push_policy().backoff;
assert_eq!(backoff.delay_after(1), Duration::from_secs(3));
assert_eq!(backoff.delay_after(2), Duration::from_secs(10));
assert_eq!(backoff.delay_after(4), Duration::from_mins(1));
assert_eq!(
repository_probe_policy().backoff.delay_after(2),
Duration::from_secs(9)
);
}
#[tokio::test(start_paused = true)]
async fn host_side_retries_keep_the_operations_own_message() {
let calls = Mutex::new(0_u32);
let result = retry_git_messages(
&repository_probe_policy(),
Some(&snapshot(Duration::from_secs(1))),
"repository probe",
|| {
let attempt = {
let mut calls = calls.lock().unwrap();
*calls += 1;
*calls
};
async move {
if attempt < 3 {
Err(format!("remote: Repository not found. (attempt {attempt})"))
} else {
Ok(())
}
}
},
)
.await;
assert_eq!(result, Ok(()));
assert_eq!(*calls.lock().unwrap(), 3);
let permanent = retry_git_messages(
&repository_probe_policy(),
Some(&static_snapshot()),
"repository probe",
|| async { Err("remote: Repository not found.".to_owned()) },
)
.await;
assert_eq!(permanent, Err("remote: Repository not found.".to_owned()));
}
}

720
lib/components/fabro-sandbox/src/git_retry.rs
View file

@ -1,720 +0,0 @@
//! Retry for git operations against GitHub from clone-based sandboxes.
//!
//! Clone-based providers can mint a GitHub App installation token and use it
//! immediately. GitHub can reject that first operation before the token is
//! available to the git endpoint. On a private repository, the rejection can
//! arrive as `Repository not found.` or an authentication failure.
//!
//! Only a token minted recently makes those messages safe to retry. Static
//! PATs and pre-minted installation tokens fail fast; a mature App token can
//! still hit a service-side blip that presents the same surface, so it
//! retries as transient infrastructure.
//!
//! Retries reuse the same token on purpose. Replication of a given token only
//! makes progress, so each attempt strictly improves the odds, while
//! re-minting would restart the replication clock.
//!
//! The driver classifies what a failure was ([`GitFailureKind`]); this module
//! decides what the class means for the credentials in hand.
use std::future::Future;
use std::time::Duration;
use chrono::Utc;
use fabro_github::token_source::TokenSnapshot;
#[cfg(test)]
use fabro_github::token_source::{REFRESH_MARGIN, TokenProvenance};
use fabro_types::SandboxProviderKind;
pub use fabro_types::run_event::GitPushRetryReason as GitRetryReason;
use fabro_util::backoff::BackoffPolicy;
use sandbox_driver::GitFailureKind;
use tokio::time;
/// How long after its mint a token is presumed to still be replicating to
/// GitHub's git endpoints. Matches the observed scale of the lag (seconds,
/// occasionally tens of seconds).
pub(crate) const REPLICATION_HORIZON: Duration = Duration::from_mins(1);
/// What a git failure message tells us about retry safety.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum GitMessageClass {
Retry(GitRetryReason),
Permanent,
Unknown,
}
impl GitMessageClass {
pub(crate) fn retry_reason(self) -> Option<GitRetryReason> {
match self {
Self::Retry(reason) => Some(reason),
Self::Permanent | Self::Unknown => None,
}
}
}
/// What the operation's credentials say about retrying auth-shaped failures.
///
/// Derived from the [`TokenSnapshot`] of the token embedded for the attempt,
/// so classification reads provenance as data instead of threading booleans
/// through call stacks.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CredentialContext {
/// An installation token younger than [`REPLICATION_HORIZON`] — a 404 or
/// auth failure is likely replication lag; retry with the same token.
FreshApp,
/// An installation token older than the horizon. A 404 with it is
/// indistinguishable from a service-side blip at this layer, so it stays
/// transient rather than proving access loss.
MatureApp,
/// A PAT or pre-minted token — it cannot become valid by waiting.
Static,
/// No credentials at all.
None,
}
impl CredentialContext {
#[must_use]
pub fn from_snapshot(snapshot: Option<&TokenSnapshot>) -> Self {
match snapshot {
None => Self::None,
Some(snapshot) => match snapshot.age_at(Utc::now()) {
None => Self::Static,
Some(age) if age < REPLICATION_HORIZON => Self::FreshApp,
Some(_) => Self::MatureApp,
},
}
}
}
/// What a classified git failure means for retrying with these credentials.
///
/// The driver reads the failure; fabro decides. A remote that could not
/// be reached is retried whatever the credential. A rejected credential
/// is retried only while a just-minted App token may still be replicating
/// (`FreshApp`), retried as a service blip for a mature App token, and
/// fails fast for a static credential or none, because waiting cannot make
/// those valid. Every other class is permanent.
pub(crate) fn decide(kind: GitFailureKind, cred: CredentialContext) -> GitMessageClass {
match kind {
GitFailureKind::RemoteUnavailable => GitMessageClass::Retry(GitRetryReason::TransientInfra),
GitFailureKind::AuthRejected => match cred {
CredentialContext::FreshApp => GitMessageClass::Retry(GitRetryReason::TokenReplication),
CredentialContext::MatureApp => GitMessageClass::Retry(GitRetryReason::TransientInfra),
CredentialContext::Static | CredentialContext::None => GitMessageClass::Permanent,
},
GitFailureKind::AccessDenied
| GitFailureKind::RefNotFound
| GitFailureKind::TargetExists
| GitFailureKind::GitUnavailable => GitMessageClass::Permanent,
_ => GitMessageClass::Unknown,
}
}
/// Classify a failed git operation by its rendered message. For git that
/// ran outside a sandbox — the host-side repository probe and metadata
/// push — where the driver never saw the failure.
pub(crate) fn classify_message(message: &str, cred: CredentialContext) -> GitMessageClass {
decide(GitFailureKind::from_message(message), cred)
}
/// Classify a rendered git failure message, returning the retry reason when
/// the failure is transient for these credentials. `None` means the failure
/// is permanent or unrecognized.
#[must_use]
pub fn classify_failure(message: &str, cred: CredentialContext) -> Option<GitRetryReason> {
classify_message(message, cred).retry_reason()
}
/// Classify a sandbox-driver git failure.
///
/// The driver classifies every git failure it produces; fabro only decides
/// what the class means for these credentials. An operation whose outcome
/// is unknown (a transport break, a timeout, an incomplete operation) is
/// never retried: replaying it could overlap a clone that is still running.
#[must_use]
pub(crate) fn classify_driver_failure(
error: &sandbox_driver::Error,
cred: CredentialContext,
) -> Option<GitRetryReason> {
match error {
sandbox_driver::Error::Git(failure) => decide(failure.kind(), cred).retry_reason(),
sandbox_driver::Error::RateLimited { .. } | sandbox_driver::Error::Overloaded { .. } => {
Some(GitRetryReason::TransientInfra)
}
_ => None,
}
}
/// Backoff between attempts: 3s, then 9s.
///
/// GitHub's guidance for token replication is to wait a few seconds and retry
/// with the same token. Sub-second delays land inside the same replication
/// window and spend an attempt for nothing.
fn replication_backoff() -> BackoffPolicy {
BackoffPolicy {
initial_delay: Duration::from_secs(3),
factor: 3.0,
max_delay: Duration::from_secs(10),
jitter: false,
}
}
/// Attempt and time bounds for one retried git operation.
///
/// All bounds are optional so existing behaviors are expressible unchanged.
/// The effective deadline is the minimum of the bounds that are present
/// (`start + max_elapsed`, `outer_deadline`); each attempt runs with
/// `min(per_attempt_timeout, remaining)` over the caps that are present, and
/// no attempt or backoff starts past the effective deadline.
#[derive(Debug, Clone)]
pub struct RetryPlan {
/// Total attempts, including the first.
pub max_attempts: u32,
pub backoff: BackoffPolicy,
/// Wall clock for this whole operation.
pub max_elapsed: Option<Duration>,
/// Cap for any single attempt.
pub per_attempt_timeout: Option<Duration>,
/// Caller-supplied absolute bound.
pub outer_deadline: Option<time::Instant>,
}
impl RetryPlan {
/// Host-side repository probes use the same attempt count and pacing as
/// clone operations against a freshly minted token.
#[must_use]
pub fn repository_probe() -> Self {
Self::clone_default(None)
}
/// The clone policy both providers already trust: 3 attempts, 3s/9s
/// backoff, no plan-level bounds. Docker supplies its existing absolute
/// five-minute deadline through `outer_deadline`; Daytona supplies none.
#[must_use]
pub fn clone_default(outer_deadline: Option<time::Instant>) -> Self {
Self {
max_attempts: 3,
backoff: replication_backoff(),
max_elapsed: None,
per_attempt_timeout: None,
outer_deadline,
}
}
/// Checkpoint pushes stay cheap: the next checkpoint re-pushes the same
/// branch anyway. Worst case ~90 seconds of wall clock.
#[must_use]
pub fn checkpoint_push() -> Self {
Self {
max_attempts: 3,
backoff: replication_backoff(),
max_elapsed: Some(Duration::from_secs(90)),
per_attempt_timeout: Some(Duration::from_mins(1)),
outer_deadline: None,
}
}
/// The terminal publish push guards the whole run's value, so it gets a
/// real budget: 5 attempts with growing backoff (~3s/10s/33s/60s),
/// bounded at 4 minutes of wall clock. The 4-minute bound must stay
/// under the token source's `REFRESH_MARGIN` (see the margin-invariant
/// test) so a pinned token always outlives the operation.
#[must_use]
pub fn publish_push() -> Self {
Self {
max_attempts: 5,
backoff: BackoffPolicy {
initial_delay: Duration::from_secs(3),
factor: 10.0 / 3.0,
max_delay: Duration::from_mins(1),
jitter: false,
},
max_elapsed: Some(Duration::from_mins(4)),
per_attempt_timeout: Some(Duration::from_mins(1)),
outer_deadline: None,
}
}
/// The absolute deadline this operation must finish by, if any bound is
/// present.
pub(crate) fn effective_deadline(&self, start: time::Instant) -> Option<time::Instant> {
let elapsed_deadline = self.max_elapsed.map(|max| start + max);
match (elapsed_deadline, self.outer_deadline) {
(Some(a), Some(b)) => Some(a.min(b)),
(Some(a), None) => Some(a),
(None, Some(b)) => Some(b),
(None, None) => None,
}
}
/// Time cap for an attempt starting now: the per-attempt cap bounded by
/// the time remaining before the effective deadline.
pub(crate) fn attempt_timeout(&self, deadline: Option<time::Instant>) -> Option<Duration> {
let remaining = deadline.map(|d| d.saturating_duration_since(time::Instant::now()));
match (self.per_attempt_timeout, remaining) {
(Some(cap), Some(remaining)) => Some(cap.min(remaining)),
(Some(cap), None) => Some(cap),
(None, remaining) => remaining,
}
}
pub(crate) fn retry_delay(
&self,
attempt_number: u32,
deadline: Option<time::Instant>,
) -> Option<Duration> {
let delay = self.backoff.delay_for_attempt(attempt_number);
if deadline.is_some_and(|deadline| {
delay >= deadline.saturating_duration_since(time::Instant::now())
}) {
None
} else {
Some(delay)
}
}
}
/// Run a git operation, repeating it while the failure looks transient.
///
/// `attempt` receives the 1-based attempt number. `classify` decides whether
/// an error is worth repeating; `None` returns it to the caller untouched.
/// A retry starts only when its backoff fits before the plan's effective
/// deadline. The final error is returned as-is.
pub async fn retry_git_operation<T, E, Attempt, Fut, Classify>(
provider: SandboxProviderKind,
op: &str,
plan: &RetryPlan,
mut attempt: Attempt,
classify: Classify,
) -> Result<T, E>
where
Attempt: FnMut(u32) -> Fut,
Fut: Future<Output = Result<T, E>>,
Classify: Fn(&E) -> Option<GitRetryReason>,
{
let deadline = plan.effective_deadline(time::Instant::now());
for attempt_number in 1..plan.max_attempts.max(1) {
match attempt(attempt_number).await {
Ok(value) => return Ok(value),
Err(err) => {
let Some(reason) = classify(&err) else {
return Err(err);
};
let Some(delay) = plan.retry_delay(attempt_number, deadline) else {
return Err(err);
};
// The failure text can carry git stderr, so log the category
// rather than the message. The caller still reports the full
// error if the attempts run out.
tracing::warn!(
provider = %provider,
op,
attempt = attempt_number,
max_attempts = plan.max_attempts,
reason = %reason,
delay_ms = u64::try_from(delay.as_millis()).unwrap_or(u64::MAX),
"Git operation failed, retrying"
);
time::sleep(delay).await;
}
}
}
attempt(plan.max_attempts.max(1)).await
}
#[cfg(test)]
mod tests {
use std::sync::Mutex;
use super::*;
/// Records the attempt numbers a closure was called with.
#[derive(Default)]
struct Attempts(Mutex<Vec<u32>>);
impl Attempts {
fn record(&self, attempt: u32) {
self.0.lock().expect("attempt log mutex").push(attempt);
}
fn recorded(&self) -> Vec<u32> {
self.0.lock().expect("attempt log mutex").clone()
}
}
/// A classifier that treats every failure as worth repeating.
const ALWAYS_RETRY: fn(&String) -> Option<GitRetryReason> =
|_| Some(GitRetryReason::TokenReplication);
fn fresh_snapshot(age: Duration, ttl: Duration) -> TokenSnapshot {
let now = Utc::now();
TokenSnapshot {
generation: 1,
provenance: TokenProvenance::Minted {
minted_at: now - chrono::Duration::from_std(age).unwrap(),
expires_at: now + chrono::Duration::from_std(ttl).unwrap(),
},
}
}
#[test]
fn credential_context_reads_token_age_from_provenance() {
assert_eq!(
CredentialContext::from_snapshot(None),
CredentialContext::None
);
assert_eq!(
CredentialContext::from_snapshot(Some(&TokenSnapshot {
generation: 0,
provenance: TokenProvenance::Static,
})),
CredentialContext::Static
);
assert_eq!(
CredentialContext::from_snapshot(Some(&fresh_snapshot(
Duration::from_secs(5),
Duration::from_hours(1)
))),
CredentialContext::FreshApp
);
assert_eq!(
CredentialContext::from_snapshot(Some(&fresh_snapshot(
Duration::from_mins(2),
Duration::from_hours(1)
))),
CredentialContext::MatureApp
);
}
#[test]
fn private_repo_not_found_with_a_fresh_token_is_a_replication_lag() {
assert_eq!(
classify_message(
"repository not found: Repository not found.",
CredentialContext::FreshApp
),
GitMessageClass::Retry(GitRetryReason::TokenReplication)
);
}
#[test]
fn not_found_with_a_mature_token_is_transient_not_permanent() {
// A service-side blip is indistinguishable from access loss at this
// layer, so a mature-App 404 stays retryable.
assert_eq!(
classify_message(
"repository not found: Repository not found.",
CredentialContext::MatureApp
),
GitMessageClass::Retry(GitRetryReason::TransientInfra)
);
}
#[test]
fn not_found_with_static_or_no_credentials_is_permanent() {
for cred in [CredentialContext::Static, CredentialContext::None] {
assert_eq!(
classify_message("repository not found: Repository not found.", cred),
GitMessageClass::Permanent,
"{cred:?} cannot become valid by waiting"
);
}
}
#[test]
fn auth_failure_classification_follows_the_credential_context() {
let message = "fatal: Authentication failed for 'https://github.com/owner/repo'";
assert_eq!(
classify_message(message, CredentialContext::FreshApp),
GitMessageClass::Retry(GitRetryReason::TokenReplication)
);
assert_eq!(
classify_message(message, CredentialContext::MatureApp),
GitMessageClass::Retry(GitRetryReason::TransientInfra)
);
assert_eq!(
classify_message(message, CredentialContext::Static),
GitMessageClass::Permanent
);
}
#[test]
fn infra_failures_retry_without_credentials() {
for message in [
"fatal: unable to access: Could not resolve host: github.com",
"error: RPC failed; curl 56 recv failure",
"fatal: early EOF",
"Operation timed out",
] {
assert_eq!(
classify_message(message, CredentialContext::None),
GitMessageClass::Retry(GitRetryReason::TransientInfra),
"expected {message:?} to be transient"
);
}
}
#[test]
fn genuine_failures_are_not_retried() {
for message in [
"fatal: could not read Username for 'https://github.com'",
"remote: Permission to owner/repo.git denied",
"fatal: destination path 'repo' already exists",
] {
assert_eq!(
classify_message(message, CredentialContext::FreshApp),
GitMessageClass::Permanent,
"expected {message:?} to fail fast"
);
}
}
#[test]
fn unrecognized_failures_remain_unknown() {
assert_eq!(
classify_message(
"git operation stopped for an unexpected reason",
CredentialContext::FreshApp
),
GitMessageClass::Unknown
);
}
#[test]
fn backoff_waits_seconds_not_milliseconds() {
let plan = RetryPlan::clone_default(None);
assert_eq!(plan.backoff.delay_for_attempt(1), Duration::from_secs(3));
assert_eq!(plan.backoff.delay_for_attempt(2), Duration::from_secs(9));
}
#[test]
fn publish_backoff_grows_toward_a_one_minute_cap() {
let plan = RetryPlan::publish_push();
assert_eq!(plan.backoff.delay_for_attempt(1), Duration::from_secs(3));
assert_eq!(plan.backoff.delay_for_attempt(2), Duration::from_secs(10));
assert!(plan.backoff.delay_for_attempt(3) < Duration::from_secs(35));
assert_eq!(plan.backoff.delay_for_attempt(4), Duration::from_mins(1));
}
/// `REFRESH_MARGIN` must exceed every push plan's `max_elapsed`: a push
/// pins the token of its single successful resolve, and any token the
/// source returns has at least the margin of validity left, so the pinned
/// token must outlive the whole operation.
#[test]
fn refresh_margin_exceeds_every_push_plan_elapsed_bound() {
for plan in [RetryPlan::checkpoint_push(), RetryPlan::publish_push()] {
let max_elapsed = plan.max_elapsed.expect("push plans bound elapsed time");
assert!(
REFRESH_MARGIN > max_elapsed,
"margin invariant violated: {max_elapsed:?}"
);
}
}
#[test]
fn effective_deadline_takes_the_minimum_of_present_bounds() {
let start = time::Instant::now();
let outer = start + Duration::from_secs(30);
let unbounded = RetryPlan::clone_default(None);
assert_eq!(unbounded.effective_deadline(start), None);
let outer_only = RetryPlan::clone_default(Some(outer));
assert_eq!(outer_only.effective_deadline(start), Some(outer));
let mut both = RetryPlan::checkpoint_push();
both.outer_deadline = Some(outer);
assert_eq!(both.effective_deadline(start), Some(outer));
both.outer_deadline = Some(start + Duration::from_mins(10));
assert_eq!(
both.effective_deadline(start),
Some(start + Duration::from_secs(90))
);
}
#[tokio::test(start_paused = true)]
async fn attempt_timeout_is_capped_by_the_remaining_deadline() {
let plan = RetryPlan::checkpoint_push();
let deadline = Some(time::Instant::now() + Duration::from_secs(20));
assert_eq!(
plan.attempt_timeout(deadline),
Some(Duration::from_secs(20))
);
assert_eq!(plan.attempt_timeout(None), Some(Duration::from_mins(1)));
let unbounded = RetryPlan::clone_default(None);
assert_eq!(unbounded.attempt_timeout(None), None);
}
#[tokio::test(start_paused = true)]
async fn first_success_runs_one_attempt() {
let attempts = Attempts::default();
let result = retry_git_operation(
SandboxProviderKind::DOCKER,
"clone",
&RetryPlan::clone_default(None),
|attempt| {
attempts.record(attempt);
async move { Ok::<_, String>(attempt) }
},
ALWAYS_RETRY,
)
.await;
assert_eq!(result, Ok(1));
assert_eq!(attempts.recorded(), vec![1]);
}
#[tokio::test(start_paused = true)]
async fn retries_until_a_later_attempt_succeeds() {
let attempts = Attempts::default();
let result = retry_git_operation(
SandboxProviderKind::DOCKER,
"clone",
&RetryPlan::clone_default(None),
|attempt| {
attempts.record(attempt);
async move {
if attempt < 3 {
Err("Repository not found.".to_string())
} else {
Ok(attempt)
}
}
},
ALWAYS_RETRY,
)
.await;
assert_eq!(result, Ok(3));
assert_eq!(attempts.recorded(), vec![1, 2, 3]);
}
#[tokio::test(start_paused = true)]
async fn exhausted_attempts_return_the_final_error() {
let attempts = Attempts::default();
let result = retry_git_operation(
SandboxProviderKind::DOCKER,
"clone",
&RetryPlan::clone_default(None),
|attempt| {
attempts.record(attempt);
async move { Err::<(), _>(format!("Repository not found. (attempt {attempt})")) }
},
ALWAYS_RETRY,
)
.await;
assert_eq!(
result,
Err("Repository not found. (attempt 3)".to_string()),
"the caller should see the last failure, not the first"
);
assert_eq!(attempts.recorded(), vec![1, 2, 3]);
}
#[tokio::test(start_paused = true)]
async fn unretryable_failure_stops_immediately() {
let attempts = Attempts::default();
let result = retry_git_operation(
SandboxProviderKind::DOCKER,
"clone",
&RetryPlan::clone_default(None),
|attempt| {
attempts.record(attempt);
async move { Err::<(), _>("permission denied".to_string()) }
},
|_: &String| None,
)
.await;
assert_eq!(result, Err("permission denied".to_string()));
assert_eq!(
attempts.recorded(),
vec![1],
"a deterministic failure should not wait out the backoff"
);
}
/// Docker clone parity: the caller's absolute deadline stops retries when
/// the backoff no longer fits before it.
#[tokio::test(start_paused = true)]
async fn outer_deadline_stops_retry_when_backoff_does_not_fit() {
let attempts = Attempts::default();
let deadline = time::Instant::now() + Duration::from_secs(2);
let result = retry_git_operation(
SandboxProviderKind::DOCKER,
"clone",
&RetryPlan::clone_default(Some(deadline)),
|attempt| {
attempts.record(attempt);
async move { Err::<(), _>("temporary failure".to_string()) }
},
ALWAYS_RETRY,
)
.await;
assert_eq!(result, Err("temporary failure".to_string()));
assert_eq!(attempts.recorded(), vec![1]);
assert_eq!(time::Instant::now() + Duration::from_secs(2), deadline);
}
/// Daytona clone parity: with no bounds at all, attempts are limited only
/// by `max_attempts` and backoff.
#[tokio::test(start_paused = true)]
async fn unbounded_plan_runs_all_attempts() {
let attempts = Attempts::default();
let result = retry_git_operation(
SandboxProviderKind::DAYTONA,
"clone",
&RetryPlan::clone_default(None),
|attempt| {
attempts.record(attempt);
async move { Err::<(), _>("temporary failure".to_string()) }
},
|_: &String| Some(GitRetryReason::TransientInfra),
)
.await;
assert!(result.is_err());
assert_eq!(attempts.recorded(), vec![1, 2, 3]);
}
#[tokio::test(start_paused = true)]
async fn max_elapsed_stops_retry_when_backoff_does_not_fit() {
let attempts = Attempts::default();
let plan = RetryPlan {
max_attempts: 5,
backoff: replication_backoff(),
max_elapsed: Some(Duration::from_secs(4)),
per_attempt_timeout: None,
outer_deadline: None,
};
let result = retry_git_operation(
SandboxProviderKind::DOCKER,
"push",
&plan,
|attempt| {
attempts.record(attempt);
async move { Err::<(), _>("temporary failure".to_string()) }
},
ALWAYS_RETRY,
)
.await;
assert!(result.is_err());
// Attempt 1 fails instantly, 3s backoff fits inside 4s, attempt 2
// fails, and the 9s backoff no longer fits.
assert_eq!(attempts.recorded(), vec![1, 2]);
}
}

13
lib/components/fabro-sandbox/src/lib.rs
View file

@ -6,7 +6,7 @@ pub mod sandbox_spec;
mod clone_source;
mod git_retry;
mod git_policy;
mod managed_labels;
@ -44,8 +44,9 @@ pub use fabro_github::token_source::{
InstallationTokenSource, ResolvedToken, TokenProvenance, TokenSnapshot,
};
pub use fabro_types::{RunSandboxInstance, SandboxProviderKind};
pub use git_retry::{
CredentialContext, GitRetryReason, RetryPlan, classify_failure, retry_git_operation,
pub use git_policy::{
GitRetryReason, checkpoint_push_policy, publish_push_policy, repository_probe_policy,
retry_git_messages, transient_git_failure,
};
pub use provider::{SandboxInventory, SandboxLookupError};
pub use provider_sandbox::{attach_provider_sandbox, provider_sandbox};
@ -64,8 +65,8 @@ pub use sandbox::{
/// dependency.
pub use sandbox_driver::{
CaptureStats, DirEntry, ExecControls, ExecFailure, ExecResult, ExecSpec, ExecStreamingResult,
FileKind, GrepMatch, GrepOptions, LifecycleTimers, NetworkPolicy, OutputSink, OutputStream,
PtySession, PtySize, Resources, SandboxSource, SandboxSpec as DriverSpec, StderrTail,
StdioProcess, StdioProcessHandle, Termination, TransportError, WalkOptions,
FileKind, GitRetryPolicy, GrepMatch, GrepOptions, LifecycleTimers, NetworkPolicy, OutputSink,
OutputStream, PtySession, PtySize, Resources, SandboxSource, SandboxSpec as DriverSpec,
StderrTail, StdioProcess, StdioProcessHandle, Termination, TransportError, WalkOptions,
};
pub use sandbox_spec::{ProviderSandboxSpec, SandboxSpec};

244
lib/components/fabro-sandbox/src/sandbox.rs
View file

@ -1,16 +1,20 @@
use std::fmt::Write;
use std::time::Duration;
use chrono::{DateTime, Utc};
use fabro_github::token_source::TokenSnapshot;
use fabro_util::shell;
use sandbox_driver::{Git as _, GitCheckoutOptions, GitPushOptions, Termination};
use sandbox_driver::{
Git as _, GitAttempt, GitCheckoutOptions, GitPushOptions, GitRetryError, GitRetryPolicy,
retry_git,
};
use serde::{Deserialize, Serialize};
use tokio::time;
use crate::credentials::{self, RepoCredentials};
use crate::driver_sandbox::RunSandbox;
use crate::exec::ExecResultExt;
use crate::git_retry::{self, CredentialContext, GitRetryReason, RetryPlan};
use crate::git_policy::{self, GitRetryReason};
/// Git command prefix that disables background maintenance.
pub(crate) const GIT: &str = "git -c maintenance.auto=0 -c gc.auto=0";
@ -316,36 +320,19 @@ pub struct PushError {
pub error: crate::Error,
}
/// What a failed push attempt means for retrying. The driver classified
/// the failure; a push that did not run to completion (timed out or
/// cancelled) is never retried, because the remote may still be applying
/// it.
fn classify_push_error(error: &crate::Error, cred: CredentialContext) -> Option<GitRetryReason> {
let driver = error.driver()?;
if let sandbox_driver::Error::Git(failure) = driver {
if failure
.output()
.is_some_and(|output| output.termination() != Termination::Exited)
{
return None;
}
}
git_retry::classify_driver_failure(driver, cred)
}
/// Pushes a refspec to origin through the driver's git facet, retrying per
/// `plan` with one token for the whole operation. `credentials` is the
/// checkout's managed credentials; `None` pushes with whatever the checkout
/// already has (the local sandbox, or a workspace without a GitHub App).
/// Pushes a refspec to origin through the driver's git facet, retried by
/// the driver under `policy` with one token for the whole operation.
/// `credentials` is the checkout's managed credentials; `None` pushes with
/// whatever the checkout already has (the local sandbox, or a workspace
/// without a GitHub App).
#[tracing::instrument(name = "git_op", skip_all, fields(op = "push"))]
pub(crate) async fn git_push(
sandbox: &RunSandbox,
credentials: Option<&RepoCredentials>,
refspec: &str,
plan: &RetryPlan,
policy: &GitRetryPolicy,
) -> Result<PushReport, PushError> {
let start = time::Instant::now();
let deadline = plan.effective_deadline(start);
let git = match sandbox.git() {
Ok(git) => git,
Err(error) => {
@ -362,16 +349,18 @@ pub(crate) async fn git_push(
// makes progress, and a fresh mint would restart that clock.
let token = match credentials {
Some(credentials) => {
let resolved = match deadline {
Some(deadline) => match time::timeout_at(deadline, credentials.resolve()).await {
Ok(resolved) => resolved,
Err(_) => {
return Err(push_deadline_error(
Vec::new(),
"while acquiring credentials",
));
let resolved = match policy.max_elapsed {
Some(max_elapsed) => {
match time::timeout(max_elapsed, credentials.resolve()).await {
Ok(resolved) => resolved,
Err(_) => {
return Err(push_deadline_error(
Vec::new(),
"while acquiring credentials",
));
}
}
},
}
None => credentials.resolve().await,
};
match resolved {
@ -387,91 +376,88 @@ pub(crate) async fn git_push(
None => None,
};
let snapshot = token.as_ref().map(|token| token.snapshot);
let git_credentials = token.as_ref().map(credentials::git_credentials);
// Resolving the token spent part of the operation's budget.
let policy = match policy.max_elapsed {
Some(max_elapsed) => policy.max_elapsed(max_elapsed.saturating_sub(start.elapsed())),
None => *policy,
};
let mut attempts: Vec<PushAttempt> = Vec::new();
let label = format!("git push origin {refspec}");
loop {
let attempt_number = u32::try_from(attempts.len()).unwrap_or(u32::MAX) + 1;
let started_at = chrono::Utc::now();
let attempt_timeout = plan
.attempt_timeout(deadline)
.unwrap_or(Duration::from_mins(1));
if attempt_timeout.is_zero() {
return Err(push_deadline_error(attempts, "before the next attempt"));
let result = retry_git(
&policy,
git_credentials.as_ref(),
&label,
|_attempt, timeout| {
let mut options = GitPushOptions::default();
options.remote = Some("origin".to_owned());
options.refspec = Some(refspec.to_owned());
options.timeout = Some(timeout.unwrap_or(Duration::from_mins(1)));
options.credentials.clone_from(&git_credentials);
let git = &git;
let repo = &repo;
async move { git.push(repo, &options).await }
},
)
.await;
match result {
Ok(report) => {
tracing::info!(
refspec = %refspec,
attempts = report.attempts.len(),
token_generation = snapshot.map(|token| token.generation),
token_age_ms = snapshot.and_then(|token| token.age_ms()),
"Pushed git ref to origin"
);
Ok(PushReport {
attempts: push_attempts(report.attempts, Ok(()), snapshot),
})
}
let mut options = GitPushOptions::default();
options.remote = Some("origin".to_owned());
options.refspec = Some(refspec.to_owned());
options.timeout = Some(attempt_timeout);
options.credentials = token.as_ref().map(credentials::git_credentials);
let push_result = git
.push(&repo, &options)
.await
.map_err(|error| crate::Error::context(label.clone(), error));
match push_result {
Ok(()) => {
attempts.push(PushAttempt {
attempt: attempt_number,
started_at,
success: true,
retry_reason: None,
exec_output_tail: None,
token: snapshot,
});
tracing::info!(
refspec = %refspec,
attempt = attempt_number,
token_generation = snapshot.map(|token| token.generation),
token_age_ms = snapshot.and_then(|token| token.age_ms()),
"Pushed git ref to origin"
);
return Ok(PushReport { attempts });
}
Err(error) => {
let cred = CredentialContext::from_snapshot(snapshot.as_ref());
let retry_reason = classify_push_error(&error, cred);
attempts.push(PushAttempt {
attempt: attempt_number,
started_at,
success: false,
retry_reason,
exec_output_tail: error.default_redacted_output_tail(),
token: snapshot,
});
let exhausted = attempt_number >= plan.max_attempts.max(1);
let Some(reason) = retry_reason.filter(|_| !exhausted) else {
return Err(PushError {
report: PushReport { attempts },
error,
});
};
let Some(delay) = plan.retry_delay(attempt_number, deadline) else {
return Err(PushError {
report: PushReport { attempts },
error,
});
};
// The failure text can carry git stderr, so log the category
// rather than the message.
tracing::warn!(
refspec = %refspec,
attempt = attempt_number,
max_attempts = plan.max_attempts,
reason = %reason,
token_generation = snapshot.map(|token| token.generation),
token_age_ms = snapshot.and_then(|token| token.age_ms()),
delay_ms = u64::try_from(delay.as_millis()).unwrap_or(u64::MAX),
"Git push failed, retrying with the same token"
);
time::sleep(delay).await;
}
Err(GitRetryError { attempts, error }) => {
let error = crate::Error::context(label, error);
Err(PushError {
report: PushReport {
attempts: push_attempts(attempts, Err(&error), snapshot),
},
error,
})
}
}
}
/// The driver's attempt history as fabro records it. In a completed
/// operation every attempt but the last failed; in a failed one every
/// attempt failed, and the last attempt's failure is `outcome`'s error.
fn push_attempts(
attempts: Vec<GitAttempt>,
outcome: Result<(), &crate::Error>,
token: Option<TokenSnapshot>,
) -> Vec<PushAttempt> {
let last = attempts.len();
attempts
.into_iter()
.enumerate()
.map(|(index, attempt)| {
let is_last = index + 1 == last;
let exec_output_tail = match (attempt.failure, &outcome) {
(Some(failure), _) => {
crate::Error::driver_error(failure).default_redacted_output_tail()
}
(None, Err(error)) if is_last => error.default_redacted_output_tail(),
(None, _) => None,
};
PushAttempt {
attempt: attempt.attempt,
started_at: DateTime::<Utc>::from(attempt.started_at),
success: is_last && outcome.is_ok(),
retry_reason: attempt.retry_reason.map(git_policy::recorded_reason),
exec_output_tail,
token,
}
})
.collect()
}
fn push_deadline_error(attempts: Vec<PushAttempt>, stage: &str) -> PushError {
PushError {
report: PushReport { attempts },
@ -491,13 +477,13 @@ mod push_tests {
use fabro_github::test_support::{InstallationTokenMinter, installation_token_source};
use fabro_github::token_source::{InstallationTokenSource, REFRESH_MARGIN};
use fabro_types::SandboxProviderKind;
use sandbox_driver::ExecResult;
use sandbox_driver::{ExecResult, Termination};
use sandbox_driver_testing::ScriptedSandbox;
use tokio::sync::Mutex as AsyncMutex;
use super::*;
use crate::credentials::RepoCredentials;
use crate::git_retry::{GitRetryReason, RetryPlan};
use crate::git_policy::{GitRetryReason, checkpoint_push_policy, publish_push_policy};
const ORIGIN: &str = "https://github.com/fabro-testing/repo";
const REFSPEC: &str = "refs/heads/fabro/run/01M0DH033P2XSTHAGVBHG6922F";
@ -676,7 +662,7 @@ mod push_tests {
&sandbox.run,
Some(&credentials),
REFSPEC,
&RetryPlan::checkpoint_push(),
&checkpoint_push_policy(),
)
.await
.expect("push should recover within the checkpoint plan");
@ -720,7 +706,7 @@ mod push_tests {
&sandbox.run,
Some(&credentials),
REFSPEC,
&RetryPlan::publish_push(),
&publish_push_policy(),
)
.await
.expect("push should recover within the publish plan");
@ -751,7 +737,7 @@ mod push_tests {
&sandbox.run,
Some(&credentials),
REFSPEC,
&RetryPlan::checkpoint_push(),
&checkpoint_push_policy(),
)
.await
.expect("push recovers");
@ -776,7 +762,7 @@ mod push_tests {
&sandbox.run,
Some(&credentials),
REFSPEC,
&RetryPlan::publish_push(),
&publish_push_policy(),
)
.await
.expect_err("static credentials cannot become valid by waiting");
@ -817,7 +803,7 @@ mod push_tests {
&sandbox.run,
Some(&credentials),
REFSPEC,
&RetryPlan::checkpoint_push(),
&checkpoint_push_policy(),
)
.await
.expect("the cached token still pushes");
@ -840,7 +826,7 @@ mod push_tests {
&sandbox.run,
Some(&credentials),
REFSPEC,
&RetryPlan::checkpoint_push(),
&checkpoint_push_policy(),
)
.await
.expect_err("no token to push with");
@ -872,7 +858,7 @@ mod push_tests {
&sandbox.run,
Some(&credentials),
REFSPEC,
&RetryPlan::checkpoint_push(),
&checkpoint_push_policy(),
)
.await
.expect("push succeeds");
@ -897,7 +883,7 @@ mod push_tests {
async fn push_without_managed_credentials_reports_no_token() {
let sandbox = ScriptedGitSandbox::new(vec![ok_exec()]);
let report = git_push(&sandbox.run, None, REFSPEC, &RetryPlan::checkpoint_push())
let report = git_push(&sandbox.run, None, REFSPEC, &checkpoint_push_policy())
.await
.expect("push succeeds");
@ -912,7 +898,7 @@ mod push_tests {
"fatal: Authentication failed for 'https://github.com/fabro-testing/repo'",
)]);
let push_error = git_push(&sandbox.run, None, REFSPEC, &RetryPlan::publish_push())
let push_error = git_push(&sandbox.run, None, REFSPEC, &publish_push_policy())
.await
.expect_err("no credentials to wait on");
@ -924,7 +910,7 @@ mod push_tests {
async fn timed_out_push_is_not_retried_while_the_remote_process_may_still_run() {
let sandbox = ScriptedGitSandbox::new(vec![timed_out_exec()]);
let push_error = git_push(&sandbox.run, None, REFSPEC, &RetryPlan::publish_push())
let push_error = git_push(&sandbox.run, None, REFSPEC, &publish_push_policy())
.await
.expect_err("an unconfirmed timeout must fail without another push");
@ -938,10 +924,9 @@ mod push_tests {
let source = installation_token_source("fabro-testing/repo", Arc::new(SlowMinter));
let credentials = RepoCredentials::new(Some(source));
let sandbox = ScriptedGitSandbox::new(vec![]);
let mut plan = RetryPlan::checkpoint_push();
plan.max_elapsed = Some(Duration::from_secs(1));
let policy = checkpoint_push_policy().max_elapsed(Duration::from_secs(1));
let push_error = git_push(&sandbox.run, Some(&credentials), REFSPEC, &plan)
let push_error = git_push(&sandbox.run, Some(&credentials), REFSPEC, &policy)
.await
.expect_err("credential resolution must stop at the operation deadline");
@ -953,10 +938,9 @@ mod push_tests {
#[tokio::test(start_paused = true)]
async fn expired_retry_deadline_does_not_launch_a_zero_timeout_push() {
let sandbox = ScriptedGitSandbox::new(vec![]);
let mut plan = RetryPlan::checkpoint_push();
plan.max_elapsed = Some(Duration::ZERO);
let policy = checkpoint_push_policy().max_elapsed(Duration::ZERO);
let push_error = git_push(&sandbox.run, None, REFSPEC, &plan)
let push_error = git_push(&sandbox.run, None, REFSPEC, &policy)
.await
.expect_err("an expired operation must stop before exec");

8
lib/components/fabro-workflow/src/lifecycle/git.rs
View file

@ -87,10 +87,10 @@ pub(crate) struct PushResult {
pub(crate) async fn push_run_branch(
sandbox: &fabro_sandbox::RunSandbox,
branch: &str,
plan: &fabro_sandbox::RetryPlan,
policy: &fabro_sandbox::GitRetryPolicy,
) -> Result<fabro_sandbox::PushReport, fabro_sandbox::PushError> {
sandbox
.git_push_ref(&format!("refs/heads/{branch}:refs/heads/{branch}"), plan)
.git_push_ref(&format!("refs/heads/{branch}:refs/heads/{branch}"), policy)
.await
}
@ -333,9 +333,9 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
.as_ref()
.and_then(|g| g.run_branch.as_ref())
{
let plan = fabro_sandbox::RetryPlan::checkpoint_push();
let policy = fabro_sandbox::checkpoint_push_policy();
let (push_ok, exec_output_tail, attempts) =
match push_run_branch(self.sandbox.as_ref(), branch, &plan).await {
match push_run_branch(self.sandbox.as_ref(), branch, &policy).await {
Ok(report) => {
self.sandbox_git.record_successful_push();
(true, None, report.attempts)

4
lib/components/fabro-workflow/src/pipeline/publish.rs
View file

@ -229,8 +229,8 @@ impl Concluded {
async fn push_final_commit(&self, run_branch: &str) -> Result<(), Error> {
// The terminal push guards the whole run's value, so it gets a real
// retry budget; attempts are nearly free at this point.
let plan = fabro_sandbox::RetryPlan::publish_push();
match push_run_branch(self.services.sandbox.as_ref(), run_branch, &plan).await {
let policy = fabro_sandbox::publish_push_policy();
match push_run_branch(self.services.sandbox.as_ref(), run_branch, &policy).await {
Ok(report) => {
self.services.sandbox_git.record_successful_push();
self.services.emitter.emit(&Event::GitPush {

3
lib/components/fabro-workflow/src/run_metadata.rs
View file

@ -24,8 +24,7 @@ pub(crate) fn metadata_push_failure_is_transient(
detail: &str,
token: Option<&TokenSnapshot>,
) -> bool {
let credentials = fabro_sandbox::CredentialContext::from_snapshot(token);
fabro_sandbox::classify_failure(detail, credentials).is_some()
fabro_sandbox::transient_git_failure(detail, token).is_some()
}
#[derive(Debug, thiserror::Error)]