diff --git a/docs/reference.md b/docs/reference.md index 04eba95..29ff11d 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -607,7 +607,7 @@ knit pull --rebase frontend Single-bundle pulls still default to the original repo path on the recorded base branch with `git pull --ff-only`, updating the recorded `baseSha`, and refuse on uncommitted changes unless `--force` (use `--rebase` for `git pull --rebase`). Use `knit pull --feature` to pull the tracked Knit feature checkout instead; feature pulls are recorded as `git.observed` nodes when the feature branch head moves. -`knit push` pushes tracked feature branches to `origin`. It does not create PRs, update GitHub metadata, or change bundle state. Selected repo pushes run in parallel; bundle artifact and history sync wait until every Git push succeeds. By default it pushes the current feature branch to `origin/` without setting upstream; use `--set-upstream` when you want git's upstream tracking configured: +`knit push` is the branch-only path: it pushes tracked feature branches to `origin` and nothing else — no PRs, no GitHub metadata, no bundle state change. For the review path, go straight from `knit commit` to `knit publish create`, which pushes the branches itself. Selected repo pushes run in parallel, at most `KNIT_GIT_JOBS` (default 8) at a time; bundle artifact and history sync wait until every Git push succeeds. By default it pushes the current feature branch to `origin/` without setting upstream; use `--set-upstream` when you want git's upstream tracking configured: ```sh knit push @@ -616,6 +616,10 @@ knit push --all knit push --set-upstream frontend ``` +Each `git push` is bounded: it is killed after `KNIT_GIT_PUSH_TIMEOUT` seconds (default 300), so a stalled connection cannot hold the command open, and a push that lost its connection is retried up to three times. A push the remote answered — rejected, non-fast-forward, stale lease, refused credentials — fails immediately, because repeating it only earns the same answer. Repos that fail are listed at the end and the command exits non-zero; the run keeps going for the other repos, and re-running pushes only what is still missing. + +Fan-out limits and retries are tunable through the environment: `KNIT_GIT_JOBS` (concurrent git pushes, default 8), `KNIT_FORGE_JOBS` (concurrent forge writes, default 4), `KNIT_GIT_PUSH_TIMEOUT` (seconds per push, default 300), and `KNIT_RETRY_BASE_MS` (backoff step, default 1000). Each must be a positive whole number; a value that is not is an error rather than a silently ignored setting. + `knit publish` publishes tracked feature branches to a code host. Knit is host-independent: it detects each repo's host from its git remote. GitHub uses `gh`, GitLab uses `glab`, Codeberg/Forgejo uses `tea` plus REST for richer metadata, and Bitbucket Cloud uses its REST API. Unrecognized remotes retain the historical GitHub fallback. ```sh @@ -632,7 +636,7 @@ knit publish status `knit publish create` auto-detects each repo's host (GitHub, GitLab, Forgejo/Codeberg, or Bitbucket) and publishes to all of them. Pass `--provider ` (or the `--github` shorthand) to restrict a run to repos on a single host. `knit request` is an alias for `knit publish`. -`knit publish create` is a best-effort two-phase operation. It pushes every selected tracked feature branch, creates missing review objects (PRs/MRs) or reuses an existing one for the same feature/base branch, stores publishing metadata in the bundle's `publications`, then rewrites the managed Knit block in every selected review body with the complete cross-repo list. The base defaults to each repo's bundle `baseBranch`; pass `--base release` to use the same base for every selected repo, or repeat `--base repo=branch` for per-repo bases. That target is recorded with the publication. A later native `knit land --target ` can deliberately replace those recorded review bases as part of its landing contract. Body sync is on by default; `--sync` is accepted for explicitness, and `--no-sync` skips that second phase. If body sync fails after review objects were created, run `knit publish sync` after fixing auth or network issues. +`knit publish create` is a best-effort two-phase operation and the whole review path after `knit commit`; it does its own branch push, so no separate `knit push` is needed. Repos are published at most `KNIT_FORGE_JOBS` (default 4) at a time, forge calls that fail because the host was momentarily unavailable (5xx, a rate limit honoring `Retry-After`, a dropped connection) are retried up to four times with 1s/2s/4s backoff, and calls the host answered (bad credentials, 404, 422) fail at once. A repo whose publish fails does not stop the others: every repo is reported, the command exits non-zero listing the failures, and re-running creates only what is missing — a repo whose review object already exists is adopted rather than duplicated. It pushes every selected tracked feature branch, creates missing review objects (PRs/MRs) or reuses an existing one for the same feature/base branch, stores publishing metadata in the bundle's `publications`, then rewrites the managed Knit block in every selected review body with the complete cross-repo list. The base defaults to each repo's bundle `baseBranch`; pass `--base release` to use the same base for every selected repo, or repeat `--base repo=branch` for per-repo bases. That target is recorded with the publication. A later native `knit land --target ` can deliberately replace those recorded review bases as part of its landing contract. Body sync is on by default; `--sync` is accepted for explicitness, and `--no-sync` skips that second phase. If body sync fails after review objects were created, run `knit publish sync` after fixing auth or network issues. For a named lane, select it through Knit itself. The generated plan records `lane`, immutable `targetBranches`, and whether the lane is `terminal`; apply retargets each open review object to its mapped branch, refreshes readiness, merges, and runs that lane's deployments. An intermediate lane like `staging` leaves the bundle open afterwards: @@ -866,7 +870,7 @@ Sparse advice is enabled by default for new workspaces. It prints a `Next:` line - Worktree creation relies on `git worktree add` and inherits its constraints, including branch checkout conflicts. - `knit fetch` fetches the `origin` remote for each selected repo. Repos without `origin` are reported as failures. - `knit pull` coordinates ordinary git pulls but does not resolve merge/rebase conflicts across repos. If git stops for a conflict, resolve that repo's git state before retrying. -- `knit push` pushes feature branches to `origin` and, when sync remotes are configured and `push-sync` is enabled, the bundle artifact to those remotes; use `knit publish create` to publish review objects. +- `knit push` pushes feature branches to `origin` and, when sync remotes are configured and `push-sync` is enabled, the bundle artifact to those remotes; it opens no review objects, so use `knit publish create` (which pushes the branches itself) for the PR path. - `knit publish` detects GitHub, GitLab, Codeberg/Forgejo, and Bitbucket Cloud from each repo remote; unrecognized remotes default to GitHub for compatibility. GitLab and Forgejo keep their CLI paths for the basic workspace loop and use REST for granular CI, review state, mergeability, SHA guards, and retargeting. Without a Forgejo REST token, those richer fields degrade to empty/unknown and the basic `tea` loop remains available. Bitbucket does not expose pre-merge conflict state, so conflicts surface as merge API errors. Bitbucket and Forgejo have no provider-native revert-PR API; GitHub and GitLab do. - `knit publish create` is not perfectly transactional. Branch pushes, review creation, and body updates happen sequentially. If phase two fails after review objects are created, run `knit publish sync`. - `knit land` resolves the host adapter per repo from its remote. A merge lands into the recorded base branch. Remote merges cannot be automatically unmerged by Knit, so failed land runs are recorded in `.knit/land-runs/`; fix the failed step and use `knit land resume`, or use `knit land rollback` to open revert PRs for the steps that already merged. diff --git a/src/commands/agents.rs b/src/commands/agents.rs index be59e1f..c2dad17 100644 --- a/src/commands/agents.rs +++ b/src/commands/agents.rs @@ -104,9 +104,14 @@ Bundle-scoped Knit commands resolve this bundle automatically from this cwd: knit status knit add knit commit --all -m "Describe the feature change" -knit push --set-upstream +knit publish create ``` +`knit publish create` pushes every tracked feature branch itself and then opens +one review object (PR/MR) per repo, so committing and publishing is the whole +path to review. Use `knit push --set-upstream` only when you want the branches +on `origin` without review objects. + {teamwork_section} Before editing a path that may have cross-repo coupling, ask Knit which prior bundle work touched it: diff --git a/src/commands/init.rs b/src/commands/init.rs index 40ff607..f64fc46 100644 --- a/src/commands/init.rs +++ b/src/commands/init.rs @@ -633,7 +633,8 @@ For a one-step stage and commit: knit --bundle feature-a commit --all -m "Describe the feature change" ``` -Push the bundle's feature branches after committing: +Push the bundle's feature branches without opening review objects (the PR path +below pushes them itself, so this is only for branch-only work): ```sh knit --bundle feature-a push --set-upstream @@ -649,8 +650,10 @@ knit --bundle feature-a sync pull --history ``` Publish review objects (PRs/MRs) against their intended base branch. `create` -auto-detects each repo's host; pass `--github` (or `--provider `) to limit -to one host. `knit request` is an alias for `knit publish`: +pushes each feature branch itself, so the review path is commit then publish — +no separate `knit push` step. It auto-detects each repo's host; pass `--github` +(or `--provider `) to limit to one host. `knit request` is an alias for +`knit publish`: ```sh knit publish create @@ -770,7 +773,7 @@ knit cherrypick --from feature-a --repo backend abc123 - `knit sync push [--bundles|--history|--views|--architecture|--kg|--all] [--remote ]...` is the one verb family for moving artifacts to the sync remotes; with no target flag it pushes bundle, history, views, and architecture. The often-large knowledge-graph slice moves only with explicit `--kg`. Bundle push is project-wide: every local bundle artifact — open, landed, archived — is swept so remote lifecycle state converges on the local ledger. Pushing an open bundle always means branches + artifact: missing or stale feature branches are pushed to git `origin` first, and a bundle whose branches cannot be pushed or verified is skipped with a warning. - `knit sync pull [--bundles|--history|--views|--architecture|--kg|--all] [--remote ]...` pulls those same artifacts from the sync remotes. Bundle pull is project-wide: open bundles created on other machines (and their recorded PRs) are localized into `.knit/bundles/`, and stale local artifacts fast-forward whatever their state; materialize checkouts for a discovered bundle with `knit --bundle bundle worktree`. - `knit pull --merge` union-merges the bundle ledger when the local and remote artifacts have diverged (two users recorded work concurrently); diverged feature branches still need a git merge in the worktree afterwards. -- `knit push --set-upstream` pushes every tracked feature branch in the resolved bundle to `origin` and sets upstream tracking. +- `knit push --set-upstream` pushes every tracked feature branch in the resolved bundle to `origin` and sets upstream tracking; it opens no review objects, so use `knit publish create` (which pushes too) for the PR path. - `knit push --remote hosted` pushes the resolved bundle's branches and artifact to the configured sync remote so it is visible in hosted dashboards. - `knit git --all status --short` runs Git across tracked checkouts. - `knit clean --archived --worktrees` removes generated worktrees left behind by archived or landed bundles whose normal cleanup did not remove them. diff --git a/src/commands/publish/mod.rs b/src/commands/publish/mod.rs index 44bca0c..549ba00 100644 --- a/src/commands/publish/mod.rs +++ b/src/commands/publish/mod.rs @@ -72,51 +72,55 @@ pub fn create_publications( .collect(); let total = jobs.len(); + let limit = crate::parallel::forge_jobs()?; if total > 1 { - println!("{}", out::muted(format!("publishing {total} repo(s)…"))); + println!("{}", out::muted(publishing_header(total, limit))); } // Workers stream their steps over a channel so every repo's push and // review object are printed the moment they exist, not after the slowest - // worker joined. The bundle is updated afterwards, once the workers have - // released their borrow of it. + // worker joined. The pool is bounded by the forge limit: a hundred-repo + // bundle must not open a hundred simultaneous writes against a host that + // rate-limits them. The bundle is updated afterwards, once the workers + // have released their borrow of it. let (tx, rx) = std::sync::mpsc::channel(); let outcomes: Vec = std::thread::scope(|scope| { let active = &active; let bundle = &bundle_snapshot; - for job in &jobs { - let job = job.clone(); - let tx = tx.clone(); - scope.spawn(move || { - let repo_id = job.repo.id.clone(); - let on_pushed = |pushed| { - let _ = tx.send(PublishEvent::Pushed { - repo_id: repo_id.clone(), - pushed, - }); - }; - let result = publish_repo_remote( - active, - bundle, - &job, - draft, - renew, - set_upstream, - &on_pushed, - ); - // The receiver outlives every worker; a send cannot fail. - let _ = tx.send(PublishEvent::Done { - repo_id, - result: Box::new(result), + let sender = tx.clone(); + crate::parallel::spawn_bounded(scope, &jobs, limit, move |job| { + let repo_id = job.repo.id.clone(); + let notes = sender.clone(); + let note_repo = repo_id.clone(); + let _notes = crate::retry::stream_notes_to(move |line| { + let _ = notes.send(PublishEvent::Note(format!( + "{}: {line}", + out::repo(¬e_repo) + ))); + }); + let pushed_id = repo_id.clone(); + let pushed_tx = sender.clone(); + let on_pushed = move |pushed| { + let _ = pushed_tx.send(PublishEvent::Pushed { + repo_id: pushed_id.clone(), + pushed, }); + }; + let result = + publish_repo_remote(active, bundle, job, draft, renew, set_upstream, &on_pushed); + // The receiver outlives every worker; a send cannot fail. + let _ = sender.send(PublishEvent::Done { + repo_id, + result: Box::new(result), }); - } + }); drop(tx); let mut outcomes = Vec::new(); let mut done = 0; for event in rx { match event { + PublishEvent::Note(line) => println!("{line}"), PublishEvent::Pushed { repo_id, pushed } => report_pushed(&repo_id, &pushed), PublishEvent::Done { repo_id, result } => { done += 1; @@ -179,7 +183,7 @@ pub fn create_publications( if !failures.is_empty() { bail!( - "PR publishing completed with failures:\n{}", + "PR publishing completed with failures:\n{}\n\nre-run `knit publish create` to retry only these repos; repos that already have a review object are left alone.", failures.join("\n") ); } @@ -233,39 +237,58 @@ pub fn create_publications_from_artifact( .collect(); let total = jobs.len(); + let limit = crate::parallel::forge_jobs()?; if total > 1 { - println!("{}", out::muted(format!("publishing {total} repo(s)…"))); + println!("{}", out::muted(publishing_header(total, limit))); } - // Same streaming shape as the worktree path: workers publish against the - // snapshot while the live artifact is updated as each result arrives. + // Same streaming shape and same bounded pool as the worktree path: + // workers publish against the snapshot while the live artifact is updated + // as each result arrives. let (tx, rx) = std::sync::mpsc::channel(); std::thread::scope(|scope| { let cwd = cwd.as_ref(); let snapshot = &bundle_snapshot; - for job in &jobs { - let job = job.clone(); - let tx = tx.clone(); - scope.spawn(move || { - let repo_id = job.repo.id.clone(); - let result = publish_repo_remote_from_artifact(cwd, snapshot, &job, draft, renew); - // The receiver outlives every worker; a send cannot fail. - let _ = tx.send((repo_id, result)); + let sender = tx.clone(); + crate::parallel::spawn_bounded(scope, &jobs, limit, move |job| { + let repo_id = job.repo.id.clone(); + let notes = sender.clone(); + let note_repo = repo_id.clone(); + let _notes = crate::retry::stream_notes_to(move |line| { + let _ = notes.send(ArtifactPublishEvent::Note(format!( + "{}: {line}", + out::repo(¬e_repo) + ))); }); - } + let result = publish_repo_remote_from_artifact(cwd, snapshot, job, draft, renew); + // The receiver outlives every worker; a send cannot fail. + let _ = sender.send(ArtifactPublishEvent::Done { + repo_id, + result: Box::new(result), + }); + }); drop(tx); - for (done, (repo_id, result)) in rx.into_iter().enumerate() { - let progress = out::progress(done + 1, total); - match result { - Ok(outcome) => apply_artifact_publish_result(&mut bundle, &outcome, &progress), - Err(error) => { - println!( - "{}: {}{progress}", - out::repo(&repo_id), - out::danger("PR create failed") - ); - failures.push(format!("{repo_id}: {error:#}")); + let mut done = 0; + for event in rx { + match event { + ArtifactPublishEvent::Note(line) => println!("{line}"), + ArtifactPublishEvent::Done { repo_id, result } => { + done += 1; + let progress = out::progress(done, total); + match *result { + Ok(outcome) => { + apply_artifact_publish_result(&mut bundle, &outcome, &progress) + } + Err(error) => { + println!( + "{}: {}{progress}", + out::repo(&repo_id), + out::danger("PR create failed") + ); + failures.push(format!("{repo_id}: {error:#}")); + } + } } } } @@ -286,7 +309,7 @@ pub fn create_publications_from_artifact( if !failures.is_empty() { bail!( - "PR publishing completed with failures:\n{}", + "PR publishing completed with failures:\n{}\n\nre-run `knit publish create` to retry only these repos; repos that already have a review object are left alone.", failures.join("\n") ); } @@ -334,6 +357,26 @@ pub fn sync_publications_from_artifact( Ok(()) } +/// Header for a multi-repo publish. The concurrency limit is named only when +/// it actually bounds the run, so the everyday three-repo bundle stays quiet. +fn publishing_header(total: usize, limit: usize) -> String { + if total > limit { + format!("publishing {total} repo(s), {limit} at a time…") + } else { + format!("publishing {total} repo(s)…") + } +} + +/// What an artifact-mode publish worker reports. The worktree path has richer +/// steps (see `remote::PublishEvent`); this one only pushes review objects. +enum ArtifactPublishEvent { + Note(String), + Done { + repo_id: String, + result: Box>, + }, +} + fn write_bundle_artifact_output(bundle: &ChangeGroup, out_path: Option<&Path>) -> Result<()> { match out_path { Some(path) => crate::store::write_json(path, bundle), diff --git a/src/commands/publish/remote.rs b/src/commands/publish/remote.rs index 67bed3f..eaaa867 100644 --- a/src/commands/publish/remote.rs +++ b/src/commands/publish/remote.rs @@ -4,14 +4,16 @@ use super::pr_body::initial_pr_body; use crate::checkout::checkout_dir; -use crate::git::{current_branch, git_output, git_output_optional, rev_parse}; +use crate::commands::push::{run_push, PushForce}; +use crate::git::{current_branch, git_output_optional, rev_parse}; use crate::ids::short_sha; use crate::model::{ChangeGroup, RepoEntry}; use crate::output as out; -use crate::providers::{self, pr_number_from_url, publication_for_repo, PrTarget, PullRequest}; +use crate::providers::{ + self, pr_number_from_url, publication_for_repo, Forge, PrTarget, PullRequest, +}; use crate::store::ActiveBundle; use anyhow::{bail, Context, Result}; -use std::ffi::OsString; use std::path::Path; #[derive(Clone)] @@ -31,6 +33,8 @@ pub(super) struct PushedInfo { /// branch is on origin, before the review object is looked up or created, so /// the caller can print progress per step instead of per joined worker. pub(super) enum PublishEvent { + /// A mid-publish line (a retry, say) to print in the main thread's stream. + Note(String), Pushed { repo_id: String, pushed: PushedInfo, @@ -86,7 +90,7 @@ pub(super) fn publish_repo_remote( let sha = rev_parse(&cwd, "HEAD") .with_context(|| format!("{}: failed to read feature branch HEAD", repo.id))?; - run_push(&cwd, branch, set_upstream) + run_push(&cwd, branch, set_upstream, PushForce::No) .with_context(|| format!("{}: failed to push {branch}", repo.id))?; let pushed = PushedInfo { sha, @@ -130,27 +134,19 @@ pub(super) fn publish_repo_remote( } } - let title = format!("{} ({})", bundle.title, repo.id); - let initial_body = initial_pr_body(bundle, &repo.id); - let url = forge.create(&target, base_branch, branch, &title, &initial_body, draft)?; - let summary = forge.view(&target, &url).unwrap_or_else(|_| PullRequest { - number: pr_number_from_url(&url).unwrap_or(0), - url: url.clone(), - state: Some("OPEN".to_string()), - title: Some(title), - base_ref_name: Some(base_branch.to_string()), - head_ref_name: Some(branch.to_string()), - body: None, - is_draft: None, - head_ref_oid: None, - mergeable: None, - merge_state_status: None, - review_decision: None, - }); + let status = create_or_adopt( + forge.as_ref(), + &target, + bundle, + repo, + base_branch, + branch, + draft, + )?; Ok(PublishRemoteResult { repo_index: job.repo_index, repo_id: repo.id.clone(), - status: PublishStatus::Created(summary), + status, }) } @@ -217,10 +213,52 @@ pub(super) fn publish_repo_remote_from_artifact( } } + let status = create_or_adopt( + forge.as_ref(), + &target, + bundle, + repo, + base_branch, + branch, + draft, + )?; + Ok(ArtifactPublishResult { + repo_index: job.repo_index, + repo_id: repo.id.clone(), + status, + }) +} + +/// Create the review object, or adopt one a previous attempt already created. +/// +/// A create can fail *after* the host stored the review: a POST whose reply +/// was lost is retried, and the retry is told the pull request already exists. +/// The review is real, so Knit looks it up and records it instead of failing +/// a run that in fact succeeded. Any other failure is the host's answer and +/// is returned unchanged. +fn create_or_adopt( + forge: &dyn Forge, + target: &PrTarget, + bundle: &ChangeGroup, + repo: &RepoEntry, + base_branch: &str, + branch: &str, + draft: bool, +) -> Result { let title = format!("{} ({})", bundle.title, repo.id); let initial_body = initial_pr_body(bundle, &repo.id); - let url = forge.create(&target, base_branch, branch, &title, &initial_body, draft)?; - let summary = forge.view(&target, &url).unwrap_or_else(|_| PullRequest { + let url = match forge.create(target, base_branch, branch, &title, &initial_body, draft) { + Ok(url) => url, + Err(error) => { + if providers::is_existing_review_error(&error) { + if let Ok(Some(existing)) = forge.find_existing(target, branch, base_branch) { + return Ok(PublishStatus::FoundExisting(existing)); + } + } + return Err(error); + } + }; + let summary = forge.view(target, &url).unwrap_or_else(|_| PullRequest { number: pr_number_from_url(&url).unwrap_or(0), url: url.clone(), state: Some("OPEN".to_string()), @@ -234,11 +272,7 @@ pub(super) fn publish_repo_remote_from_artifact( merge_state_status: None, review_decision: None, }); - Ok(ArtifactPublishResult { - repo_index: job.repo_index, - repo_id: repo.id.clone(), - status: PublishStatus::Created(summary), - }) + Ok(PublishStatus::Created(summary)) } /// Print the `pushed` line for a repo whose branch just reached origin. @@ -389,15 +423,3 @@ fn ensure_origin(repo: &RepoEntry, cwd: &Path) -> Result<()> { })?; Ok(()) } - -fn run_push(cwd: &Path, branch: &str, set_upstream: bool) -> Result<()> { - let mut args = vec![OsString::from("push")]; - if set_upstream { - args.push(OsString::from("--set-upstream")); - } - args.push(OsString::from("origin")); - args.push(OsString::from(branch)); - - git_output(cwd, args)?; - Ok(()) -} diff --git a/src/commands/push.rs b/src/commands/push.rs index bc3f12e..dc7eeeb 100644 --- a/src/commands/push.rs +++ b/src/commands/push.rs @@ -1,6 +1,7 @@ use crate::checkout::checkout_dir; use crate::git::{ - current_branch, git_output, git_output_optional, ref_commit_sha, remote_ref_sha, rev_parse, + current_branch, git_output, git_output_optional, git_output_with_timeout, ref_commit_sha, + remote_ref_sha, rev_parse, }; use crate::ids::short_sha; use crate::model::{BundleState, ChangeGroup, RepoEntry}; @@ -17,6 +18,17 @@ struct PushSuccess { sha: String, } +/// What a push worker reports. `Note` carries mid-push lines (a retry, say) +/// so they are printed in the main thread's stream rather than raced onto +/// stdout from a worker. +enum PushEvent { + Note(String), + Done { + repo_id: String, + result: Result, + }, +} + /// How `git push` may move the remote branch. Mirrors git's own flags: /// `WithLease` refuses when the remote moved since the last fetch. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -73,48 +85,70 @@ pub fn push_repos( let indexes = resolve_repo_indexes(&active, selectors, all)?; let total = indexes.len(); + let limit = crate::parallel::git_jobs()?; if total > 1 { - println!("{}", out::muted(format!("pushing {total} repo(s)…"))); + let bound = if total > limit { + format!(", {limit} at a time") + } else { + String::new() + }; + println!("{}", out::muted(format!("pushing {total} repo(s){bound}…"))); } // Report each repo the moment its push finishes rather than after the // slowest one: with many repos and a slow origin, a report batched after - // the last join reads as a hang. + // the last join reads as a hang. The pool is bounded so a hundred-repo + // bundle does not open a hundred connections at once. let (tx, rx) = std::sync::mpsc::channel(); let failures: Vec = std::thread::scope(|scope| { - for &index in &indexes { - let active = &active; - let repo = &active.bundle.repos[index]; + let active_ref = &active; + let sender = tx.clone(); + crate::parallel::spawn_bounded(scope, &indexes, limit, move |&index| { + let repo = &active_ref.bundle.repos[index]; let repo_id = repo.id.clone(); - let tx = tx.clone(); - scope.spawn(move || { - let result = push_repo(active, repo, set_upstream, force); - // The receiver outlives every worker; a send cannot fail. - let _ = tx.send((repo_id, result)); + // Retry notes travel the same channel as the result, so the main + // thread stays the only writer and lines never interleave. + let notes = sender.clone(); + let note_repo = repo_id.clone(); + let _notes = crate::retry::stream_notes_to(move |line| { + let _ = notes.send(PushEvent::Note(format!( + "{}: {line}", + out::repo(¬e_repo) + ))); }); - } + let result = push_repo(active_ref, repo, set_upstream, force); + // The receiver outlives every worker; a send cannot fail. + let _ = sender.send(PushEvent::Done { repo_id, result }); + }); drop(tx); let mut failures = Vec::new(); - for (done, (repo_id, result)) in rx.into_iter().enumerate() { - let progress = out::progress(done + 1, total); - match result { - Ok(success) => { - println!( - "{}: {} {} {}{progress}", - out::repo(&repo_id), - out::movement("pushed"), - out::branch(success.upstream), - out::sha(short_sha(&success.sha)) - ); - } - Err(error) => { - println!( - "{}: {}{progress}", - out::repo(&repo_id), - out::danger("push failed") - ); - failures.push(format!("{repo_id}: {error:#}")); + let mut done = 0; + for event in rx { + match event { + PushEvent::Note(line) => println!("{line}"), + PushEvent::Done { repo_id, result } => { + done += 1; + let progress = out::progress(done, total); + match result { + Ok(success) => { + println!( + "{}: {} {} {}{progress}", + out::repo(&repo_id), + out::movement("pushed"), + out::branch(success.upstream), + out::sha(short_sha(&success.sha)) + ); + } + Err(error) => { + println!( + "{}: {}{progress}", + out::repo(&repo_id), + out::danger("push failed") + ); + failures.push(format!("{repo_id}: {error:#}")); + } + } } } } @@ -122,7 +156,10 @@ pub fn push_repos( }); if !failures.is_empty() { - bail!("push failed:\n{}", failures.join("\n")); + bail!( + "push failed:\n{}\n\nre-run the same `knit push` to retry only these repos; branches already on origin are up to date.", + failures.join("\n") + ); } // After git branches are pushed, also sync the bundle artifact to the @@ -189,19 +226,41 @@ fn ensure_origin(repo: &RepoEntry, cwd: &Path) -> Result<()> { Ok(()) } -fn run_push(cwd: &Path, branch: &str, set_upstream: bool, force: PushForce) -> Result<()> { - let mut args = vec![OsString::from("push")]; - if set_upstream { - args.push(OsString::from("--set-upstream")); - } - if let Some(force_arg) = force.git_arg() { - args.push(OsString::from(force_arg)); - } - args.push(OsString::from("origin")); - args.push(OsString::from(branch)); +/// The one `git push` door for Knit's fan-out commands (`knit push`, +/// `knit publish create`, the branch/artifact coupling below). +/// +/// Every push is bounded by `KNIT_GIT_PUSH_TIMEOUT` (default 300s) so a +/// stalled connection cannot hold the command open, and a push that failed on +/// the way to the remote — a reset connection, a hung-up remote, a timeout — +/// is retried up to [`crate::retry::GIT_PUSH_ATTEMPTS`] times. A push the +/// remote *answered* (rejected, stale lease, refused credentials) is returned +/// immediately: that is an answer, and repeating it only delays it. +pub(crate) fn run_push( + cwd: &Path, + branch: &str, + set_upstream: bool, + force: PushForce, +) -> Result<()> { + let timeout = crate::retry::git_push_timeout()?; + crate::retry::retry_transient( + "push", + crate::retry::GIT_PUSH_ATTEMPTS, + crate::retry::classify_git_push, + || { + let mut args = vec![OsString::from("push")]; + if set_upstream { + args.push(OsString::from("--set-upstream")); + } + if let Some(force_arg) = force.git_arg() { + args.push(OsString::from(force_arg)); + } + args.push(OsString::from("origin")); + args.push(OsString::from(branch)); - git_output(cwd, args)?; - Ok(()) + git_output_with_timeout(cwd, args, timeout)?; + Ok(()) + }, + ) } fn read_upstream(cwd: &Path) -> Option { diff --git a/src/git.rs b/src/git.rs index e7695ca..7741bc3 100644 --- a/src/git.rs +++ b/src/git.rs @@ -387,6 +387,101 @@ where ); } +/// Like `git_output`, but bounded in time: the child is killed when `timeout` +/// expires and the failure names the command, the directory and the limit. +/// +/// `Command::output()` waits forever. That is fine for local plumbing and +/// wrong for anything that talks to a remote: one stalled TCP connection to a +/// code host would hold a whole fan-out command open with nothing to show for +/// it. Output is drained by reader threads so a chatty child cannot deadlock +/// on a full pipe, and stdin is closed so git can never sit waiting for a +/// credential prompt. +pub fn git_output_with_timeout( + cwd: &Path, + args: I, + timeout: std::time::Duration, +) -> Result +where + I: IntoIterator, + S: AsRef, +{ + use std::io::Read; + + let args = collect_args(args); + let mut child = Command::new("git") + .args(&args) + .current_dir(cwd) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .with_context(|| format!("failed to run git in {}", cwd.display()))?; + + let mut stdout_pipe = child.stdout.take(); + let mut stderr_pipe = child.stderr.take(); + let stdout_reader = std::thread::spawn(move || { + let mut buffer = Vec::new(); + if let Some(pipe) = stdout_pipe.as_mut() { + let _ = pipe.read_to_end(&mut buffer); + } + buffer + }); + let stderr_reader = std::thread::spawn(move || { + let mut buffer = Vec::new(); + if let Some(pipe) = stderr_pipe.as_mut() { + let _ = pipe.read_to_end(&mut buffer); + } + buffer + }); + + let deadline = std::time::Instant::now() + timeout; + let status = loop { + match child + .try_wait() + .with_context(|| format!("failed to wait for git in {}", cwd.display()))? + { + Some(status) => break Some(status), + None => { + if std::time::Instant::now() >= deadline { + let _ = child.kill(); + let _ = child.wait(); + break None; + } + std::thread::sleep(std::time::Duration::from_millis(50)); + } + } + }; + + let stdout = stdout_reader.join().unwrap_or_default(); + let stderr = stderr_reader.join().unwrap_or_default(); + let Some(status) = status else { + bail!( + "git {} in {} timed out after {}s", + display_args(&args), + cwd.display(), + timeout.as_secs() + ); + }; + + if status.success() { + return Ok(String::from_utf8_lossy(&stdout).trim_end().to_string()); + } + + let stderr = String::from_utf8_lossy(&stderr); + let stdout = String::from_utf8_lossy(&stdout); + let detail = if stderr.trim().is_empty() { + stdout.trim() + } else { + stderr.trim() + }; + bail!( + "git {} failed in {}: {}", + display_args(&args), + cwd.display(), + detail + ); +} + /// Like `git_output`, but with extra environment variables set for the child /// git process only (e.g. `GIT_TERMINAL_PROMPT=0` for non-interactive remote /// queries). Never used to pass credentials. diff --git a/src/lib.rs b/src/lib.rs index 9d958e5..68d7159 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,9 +7,11 @@ pub mod history; pub mod ids; pub mod model; pub mod output; +pub mod parallel; pub mod paths; pub mod providers; pub mod repo_selectors; +pub mod retry; pub mod selectors; pub mod status; pub mod store; diff --git a/src/parallel.rs b/src/parallel.rs new file mode 100644 index 0000000..699477b --- /dev/null +++ b/src/parallel.rs @@ -0,0 +1,136 @@ +//! Bounded parallelism for Knit's per-repo fan-out commands. +//! +//! `knit push` and `knit publish create` used to spawn one thread per repo. +//! That is fine for a five-repo bundle and hostile for a hundred-repo one: +//! a hundred simultaneous `git push` processes and a hundred simultaneous +//! forge API writes are how a workspace turns into a rate-limited, swap-bound +//! stall. Work is handed to a small pool instead, and the pool size is a +//! deliberate, overridable number rather than "however many repos exist". +//! +//! Git pushes and forge writes get separate limits because they fail +//! differently: git pushes are bound by network and CPU, forge writes by the +//! host's rate limiter. + +use anyhow::{bail, Result}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; + +/// Concurrent `git push` processes, unless `KNIT_GIT_JOBS` says otherwise. +pub const DEFAULT_GIT_JOBS: usize = 8; +/// Concurrent forge writes (PR creation and friends), unless +/// `KNIT_FORGE_JOBS` says otherwise. Lower than the git limit because code +/// hosts rate-limit writes far more aggressively than a git remote does. +pub const DEFAULT_FORGE_JOBS: usize = 4; + +pub fn git_jobs() -> Result { + Ok(env_number("KNIT_GIT_JOBS", DEFAULT_GIT_JOBS as u64, false)? as usize) +} + +pub fn forge_jobs() -> Result { + Ok(env_number("KNIT_FORGE_JOBS", DEFAULT_FORGE_JOBS as u64, false)? as usize) +} + +/// Read a numeric tuning knob from the environment. An unset or empty value +/// takes the default; anything that is not a whole number — or a zero where +/// zero is meaningless — is an error rather than a silently ignored setting. +pub(crate) fn env_number(name: &str, default: u64, allow_zero: bool) -> Result { + let Some(raw) = std::env::var_os(name) else { + return Ok(default); + }; + let raw = raw.to_string_lossy().trim().to_string(); + if raw.is_empty() { + return Ok(default); + } + match raw.parse::() { + Ok(0) if !allow_zero => { + bail!("{name} must be a positive whole number, got `{raw}`.") + } + Ok(value) => Ok(value), + Err(_) => bail!( + "{name} must be a{} whole number, got `{raw}`.", + if allow_zero { "" } else { " positive" } + ), + } +} + +/// Run `run` over every job on at most `limit` threads, inside the caller's +/// [`std::thread::scope`]. +/// +/// The workers are spawned and left running: the caller keeps the scope open +/// and drains its result channel while they work, so per-repo output still +/// streams as each repo finishes instead of arriving after the last join. +pub fn spawn_bounded<'scope, 'env, T, F>( + scope: &'scope std::thread::Scope<'scope, 'env>, + jobs: &'env [T], + limit: usize, + run: F, +) where + T: Sync + 'env, + F: Fn(&'env T) + Send + Sync + 'env, +{ + if jobs.is_empty() { + return; + } + let workers = limit.clamp(1, jobs.len()); + let next = Arc::new(AtomicUsize::new(0)); + let run = Arc::new(run); + for _ in 0..workers { + let next = Arc::clone(&next); + let run = Arc::clone(&run); + scope.spawn(move || loop { + let index = next.fetch_add(1, Ordering::Relaxed); + let Some(job) = jobs.get(index) else { + return; + }; + run(job); + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::AtomicI64; + + #[test] + fn env_number_reads_defaults_and_rejects_nonsense() { + assert_eq!(env_number("KNIT_TEST_UNSET_LIMIT", 8, false).unwrap(), 8); + // Values are validated, not silently ignored: a typo in a tuning knob + // must not quietly run with the default. + std::env::set_var("KNIT_TEST_LIMIT", "zero"); + assert!(env_number("KNIT_TEST_LIMIT", 8, false).is_err()); + std::env::set_var("KNIT_TEST_LIMIT", "0"); + assert!(env_number("KNIT_TEST_LIMIT", 8, false).is_err()); + assert_eq!(env_number("KNIT_TEST_LIMIT", 8, true).unwrap(), 0); + std::env::set_var("KNIT_TEST_LIMIT", "3"); + assert_eq!(env_number("KNIT_TEST_LIMIT", 8, false).unwrap(), 3); + std::env::remove_var("KNIT_TEST_LIMIT"); + } + + #[test] + fn bounded_pool_runs_every_job_without_exceeding_the_limit() { + let jobs: Vec = (0..50).collect(); + let done = AtomicUsize::new(0); + let live = AtomicI64::new(0); + let peak = AtomicI64::new(0); + std::thread::scope(|scope| { + spawn_bounded(scope, &jobs, 4, |_job| { + let now = live.fetch_add(1, Ordering::SeqCst) + 1; + peak.fetch_max(now, Ordering::SeqCst); + std::thread::sleep(std::time::Duration::from_millis(2)); + done.fetch_add(1, Ordering::SeqCst); + live.fetch_sub(1, Ordering::SeqCst); + }); + }); + assert_eq!(done.load(Ordering::SeqCst), 50); + assert!(peak.load(Ordering::SeqCst) <= 4, "{peak:?}"); + } + + #[test] + fn bounded_pool_tolerates_no_jobs() { + let jobs: Vec = Vec::new(); + std::thread::scope(|scope| { + spawn_bounded(scope, &jobs, 4, |_job| unreachable!()); + }); + } +} diff --git a/src/providers/github/transport.rs b/src/providers/github/transport.rs index 9ab84f1..21c0dde 100644 --- a/src/providers/github/transport.rs +++ b/src/providers/github/transport.rs @@ -15,7 +15,16 @@ pub(super) fn github_api_output( body: Option<&str>, ) -> Result { if use_native_github_api(target) { - return native_github_api_output(method, endpoint, body); + // Same retry contract as the `gh` path below (which retries inside + // `cli_output`): a host that was briefly unavailable gets another + // chance, a host that answered does not. + let action = format!("GitHub {method} /{}", endpoint.trim_start_matches('/')); + return crate::retry::retry_transient( + &action, + crate::retry::FORGE_ATTEMPTS, + crate::retry::classify_forge, + || native_github_api_output(method, endpoint, body), + ); } let mut args = vec![OsString::from("api")]; @@ -105,6 +114,13 @@ pub(super) fn native_github_api_output( Ok(text.trim_end().to_string()) } Err(ureq::Error::Status(status, response)) => { + // Read `Retry-After` before the body consumes the response: when + // GitHub says how long to wait, Knit waits exactly that long + // instead of guessing with its own backoff ladder. + let retry_after = response + .header("retry-after") + .and_then(|value| value.trim().parse::().ok()) + .map(std::time::Duration::from_secs); let detail = response.into_string().unwrap_or_default(); let detail = detail.trim(); if status == 401 || looks_like_github_auth_failure(detail) { @@ -112,7 +128,14 @@ pub(super) fn native_github_api_output( "GitHub API request failed during {operation}: HTTP {status}: {detail}\nHint: GitHub rejected GH_TOKEN/GITHUB_TOKEN. Replace the saved GitHub credential with an active token that can access this repository, then retry." ); } - bail!("GitHub API request failed during {operation}: HTTP {status}: {detail}"); + Err(anyhow::Error::new(crate::retry::HttpFailure { + status, + retry_after, + detail: detail.to_string(), + message: format!( + "GitHub API request failed during {operation}: HTTP {status}: {detail}" + ), + })) } Err(ureq::Error::Transport(transport)) => { bail!("GitHub API request failed during {operation}: {transport}") diff --git a/src/providers/mod.rs b/src/providers/mod.rs index 954bd7a..936ec26 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -397,6 +397,9 @@ fn checks_state(runs: &[CheckRun]) -> ChecksState { /// For `gh`, an invalid `GITHUB_TOKEN` or `GH_TOKEN` in the environment overrides /// `gh auth login`. When a host-token call fails with an auth error, Knit retries /// once without those variables so interactive credentials can succeed. +/// +/// Calls that fail because the host was momentarily unavailable are retried +/// with backoff; see [`crate::retry`] for what counts as transient. pub(crate) fn cli_output( bin: &str, cwd: &Path, @@ -412,10 +415,45 @@ where .map(|arg| arg.as_ref().to_os_string()) .collect::>(); - match run_cli_output(bin, cwd, &args, stdin, false) { + // Forge CLIs are the single door to every code host Knit talks to, so + // this is where a host that is briefly unavailable (5xx, a rate limit, a + // dropped connection) is given another chance. Anything the host actually + // decided — bad credentials, a missing repo, a rejected payload — is + // returned on the first attempt. + crate::retry::retry_transient( + &cli_action_label(bin, &args), + crate::retry::FORGE_ATTEMPTS, + crate::retry::classify_forge, + || cli_output_once(bin, cwd, &args, stdin), + ) +} + +/// How a retried forge call names itself in the streamed retry line: +/// `gh pr create`, `glab mr list`, `gh api`. +fn cli_action_label(bin: &str, args: &[OsString]) -> String { + let words = args + .iter() + .map(|arg| arg.to_string_lossy().to_string()) + .take_while(|arg| !arg.starts_with('-')) + .take(2) + .collect::>(); + if words.is_empty() { + bin.to_string() + } else { + format!("{bin} {}", words.join(" ")) + } +} + +fn cli_output_once( + bin: &str, + cwd: &Path, + args: &[OsString], + stdin: Option<&str>, +) -> Result { + match run_cli_output(bin, cwd, args, stdin, false) { Ok(output) => Ok(output), Err(first) if should_retry_gh_without_env_token(bin, &first) => { - match run_cli_output(bin, cwd, &args, stdin, true) { + match run_cli_output(bin, cwd, args, stdin, true) { Ok(output) => { warn_gh_env_token_override(); Ok(output) @@ -552,6 +590,17 @@ fn should_retry_gh_without_env_token(bin: &str, err: &anyhow::Error) -> bool { bin == "gh" && !gh_env_token_vars().is_empty() && looks_like_gh_auth_failure(&err.to_string()) } +/// Whether a failed review creation means "this review already exists". +/// +/// A create that is retried after a lost reply hits this: the first attempt +/// did reach the host. Publishing then adopts the existing review instead of +/// reporting a failure for work that succeeded. +pub(crate) fn is_existing_review_error(error: &anyhow::Error) -> bool { + let message = format!("{error:#}").to_ascii_lowercase(); + message.contains("already exists") + || message.contains("a pull request for these commits already exists") +} + pub(crate) fn is_gh_checks_access_error(err: &anyhow::Error) -> bool { err.chain().any(|cause| { let message = cause.to_string().to_ascii_lowercase(); diff --git a/src/retry.rs b/src/retry.rs new file mode 100644 index 0000000..80f05fd --- /dev/null +++ b/src/retry.rs @@ -0,0 +1,493 @@ +//! Bounded retries for the network calls Knit fans out across repos. +//! +//! Exactly two failure families are retried: a `git push` whose connection +//! broke, and a forge API call the host could not serve right now (5xx, rate +//! limits, dropped connections). Everything that reports a decision the host +//! already made — bad credentials, 404, 422, a rejected non-fast-forward push +//! — is handed straight back to the caller, because asking again only earns +//! the same answer more slowly and hides the real problem. +//! +//! Retries are announced as they happen ([`note`]). A per-repo worker installs +//! a sink with [`stream_notes_to`] so its notes travel through the same +//! channel as its result and the main thread stays the only writer. + +use anyhow::{Error, Result}; +use std::cell::RefCell; +use std::fmt; +use std::time::Duration; + +/// A `git push` gets three tries: the second covers a dropped connection, the +/// third covers a remote that was restarting. +pub const GIT_PUSH_ATTEMPTS: u32 = 3; +/// Forge writes get four tries, matching the 1s/2s/4s backoff ladder. +pub const FORGE_ATTEMPTS: u32 = 4; +/// A hostile or mistaken `Retry-After` must not turn "slow" into "hung". +const MAX_RETRY_AFTER: Duration = Duration::from_secs(60); + +/// Whether a failed call may be repeated, and how long to wait first. +pub enum Retryable { + No, + Yes { + reason: String, + retry_after: Option, + }, +} + +impl Retryable { + fn yes(reason: impl Into, retry_after: Option) -> Self { + Self::Yes { + reason: reason.into(), + retry_after, + } + } +} + +/// An HTTP failure from a forge, carried as a typed error so the retry +/// classifier reads the status and `Retry-After` the host actually sent +/// instead of guessing from prose. +#[derive(Debug)] +pub struct HttpFailure { + pub status: u16, + pub retry_after: Option, + pub detail: String, + pub message: String, +} + +impl fmt::Display for HttpFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(&self.message) + } +} + +impl std::error::Error for HttpFailure {} + +/// Where a thread's retry notes go while a fan-out command is streaming. +type NoteSink = Box; + +thread_local! { + static NOTES: RefCell> = const { RefCell::new(None) }; +} + +/// Restores the previous (empty) note routing when it drops. +pub struct NoteScope(()); + +impl Drop for NoteScope { + fn drop(&mut self) { + NOTES.with(|notes| { + *notes.borrow_mut() = None; + }); + } +} + +/// Route this thread's retry notes to `sink` until the returned guard drops. +#[must_use = "notes are routed only while the guard is alive"] +pub fn stream_notes_to(sink: impl Fn(String) + 'static) -> NoteScope { + NOTES.with(|notes| { + *notes.borrow_mut() = Some(Box::new(sink)); + }); + NoteScope(()) +} + +/// Report something that happened mid-call, such as a retry. Goes to this +/// thread's sink when a fan-out command installed one, else straight out. +pub fn note(message: impl Into) { + let message = message.into(); + let routed = NOTES.with(|notes| match notes.borrow().as_ref() { + Some(sink) => { + sink(message.clone()); + true + } + None => false, + }); + if !routed { + crate::human!("{message}"); + } +} + +/// Base backoff step. Tests set `KNIT_RETRY_BASE_MS=0` so a retry path can be +/// exercised without waiting for it. +fn base_delay() -> Result { + Ok(Duration::from_millis(crate::parallel::env_number( + "KNIT_RETRY_BASE_MS", + 1000, + true, + )?)) +} + +/// How long one `git push` may run before Knit kills it. +pub fn git_push_timeout() -> Result { + Ok(Duration::from_secs(crate::parallel::env_number( + "KNIT_GIT_PUSH_TIMEOUT", + 300, + false, + )?)) +} + +/// Run `attempt`, repeating it while `classify` calls the failure transient. +/// `action` names the call in the streamed retry line ("push", "gh pr create"). +pub fn retry_transient( + action: &str, + max_attempts: u32, + classify: impl Fn(&Error) -> Retryable, + attempt: impl FnMut() -> Result, +) -> Result { + retry_with(action, max_attempts, base_delay()?, classify, attempt) +} + +/// [`retry_transient`] with the backoff step supplied instead of read from the +/// environment, so tests can exercise the ladder without waiting on it. +fn retry_with( + action: &str, + max_attempts: u32, + base: Duration, + classify: impl Fn(&Error) -> Retryable, + mut attempt: impl FnMut() -> Result, +) -> Result { + let mut number = 1; + loop { + let error = match attempt() { + Ok(value) => return Ok(value), + Err(error) => error, + }; + if number >= max_attempts { + return Err(error); + } + let Retryable::Yes { + reason, + retry_after, + } = classify(&error) + else { + return Err(error); + }; + let delay = retry_after + .map(|after| after.min(MAX_RETRY_AFTER)) + .unwrap_or_else(|| base * 2u32.pow(number - 1)); + note(format!( + "retrying {action} ({}/{max_attempts}) after {reason}{}…", + number + 1, + wait_clause(delay) + )); + if !delay.is_zero() { + std::thread::sleep(delay); + } + number += 1; + } +} + +fn wait_clause(delay: Duration) -> String { + if delay.is_zero() { + String::new() + } else if delay.as_millis() % 1000 == 0 { + format!(", waiting {}s", delay.as_secs()) + } else { + format!(", waiting {}ms", delay.as_millis()) + } +} + +/// Push failures that describe a decision, not a bad moment. Checked first: +/// a rejected push, a lease that no longer holds, or a refused credential +/// must reach the user on the first attempt. +const GIT_PUSH_FINAL: &[&str] = &[ + "non-fast-forward", + "rejected", + "stale info", + "authentication failed", + "permission denied", + "could not read username", + "could not read password", + "access denied", + "repository not found", + "does not appear to be a git repository", + "hook declined", +]; + +/// Push failures worth repeating, with the wording used in the retry line. +const GIT_PUSH_TRANSIENT: &[(&str, &str)] = &[ + ("connection reset", "connection reset"), + ("early eof", "early EOF"), + ("rpc failed", "an RPC failure"), + ("the remote end hung up", "the remote end hanging up"), + ("could not resolve host", "an unresolved host"), + ("temporary failure in name resolution", "an unresolved host"), + ("timed out", "a timeout"), + ("timeout", "a timeout"), + ("connection refused", "a refused connection"), + ("unexpectedly closed connection", "a closed connection"), + ("broken pipe", "a broken pipe"), + ("ssh_exchange_identification", "a dropped SSH handshake"), +]; + +pub fn classify_git_push(error: &Error) -> Retryable { + classify_git_push_message(&format!("{error:#}")) +} + +fn classify_git_push_message(message: &str) -> Retryable { + let lower = message.to_ascii_lowercase(); + if GIT_PUSH_FINAL.iter().any(|marker| lower.contains(marker)) { + return Retryable::No; + } + match GIT_PUSH_TRANSIENT + .iter() + .find(|(marker, _)| lower.contains(marker)) + { + Some((_, reason)) => Retryable::yes(*reason, None), + None => Retryable::No, + } +} + +/// Forge failures that are answers, not weather. `403` is deliberately absent: +/// a plain 403 is final, but a secondary-rate-limit 403 is not, so the status +/// is judged with its body in [`classify_http`]. +const FORGE_FINAL: &[&str] = &[ + "http 401", + "http 404", + "http 422", + "bad credentials", + "unauthorized", + "authentication failed", + "requires authentication", + "must be authenticated", + "gh auth login", + "insufficient_scope", + "resource not accessible", + "not found", + "already exists", + "validation failed", +]; + +const FORGE_TRANSIENT: &[(&str, &str)] = &[ + ("connection reset", "connection reset"), + ("connection refused", "a refused connection"), + ("connection closed", "a closed connection"), + ("connection failed", "a failed connection"), + ("timed out", "a timeout"), + ("timeout", "a timeout"), + ("could not resolve host", "an unresolved host"), + ("temporary failure in name resolution", "an unresolved host"), + ("dns failure", "an unresolved host"), + ("network error", "a network error"), + ("network is unreachable", "an unreachable network"), + ("broken pipe", "a broken pipe"), + ("unexpected eof", "an unexpected EOF"), +]; + +pub fn classify_forge(error: &Error) -> Retryable { + if let Some(failure) = error + .chain() + .find_map(|cause| cause.downcast_ref::()) + { + return classify_http(failure.status, &failure.detail, failure.retry_after); + } + classify_forge_message(&format!("{error:#}")) +} + +fn classify_forge_message(message: &str) -> Retryable { + let lower = message.to_ascii_lowercase(); + if FORGE_FINAL.iter().any(|marker| lower.contains(marker)) { + return Retryable::No; + } + if let Some(status) = http_status_in(&lower) { + return classify_http(status, &lower, retry_after_in(&lower)); + } + match FORGE_TRANSIENT + .iter() + .find(|(marker, _)| lower.contains(marker)) + { + Some((_, reason)) => Retryable::yes(*reason, None), + None => Retryable::No, + } +} + +fn classify_http(status: u16, detail: &str, retry_after: Option) -> Retryable { + let lower = detail.to_ascii_lowercase(); + if status >= 500 { + return Retryable::yes(format!("HTTP {status}"), retry_after); + } + if status == 429 { + return Retryable::yes("HTTP 429 (rate limited)", retry_after); + } + if status == 403 && is_rate_limited_body(&lower) { + return Retryable::yes("HTTP 403 (rate limited)", retry_after); + } + Retryable::No +} + +fn is_rate_limited_body(lower: &str) -> bool { + lower.contains("secondary rate limit") + || lower.contains("abuse detection") + || lower.contains("rate limit exceeded") + || lower.contains("exceeded a secondary") +} + +/// First `HTTP ` in a message, however the CLI phrased it: `gh` writes +/// `(HTTP 502)`, Knit's own transport writes `HTTP 502: ...`. +fn http_status_in(lower: &str) -> Option { + let mut rest = lower; + while let Some(at) = rest.find("http ") { + let after = &rest[at + "http ".len()..]; + let digits: String = after.chars().take_while(|c| c.is_ascii_digit()).collect(); + if digits.len() == 3 { + if let Ok(status) = digits.parse::() { + return Some(status); + } + } + rest = after; + } + None +} + +fn retry_after_in(lower: &str) -> Option { + let at = lower.find("retry-after: ")?; + let after = &lower[at + "retry-after: ".len()..]; + let digits: String = after.chars().take_while(|c| c.is_ascii_digit()).collect(); + digits.parse::().ok().map(Duration::from_secs) +} + +#[cfg(test)] +mod tests { + use super::*; + use anyhow::anyhow; + + fn is_retryable(decision: &Retryable) -> bool { + matches!(decision, Retryable::Yes { .. }) + } + + #[test] + fn transient_push_failures_are_retried_and_decisions_are_not() { + for message in [ + "git push failed: Connection reset by peer", + "fatal: the remote end hung up unexpectedly", + "error: RPC failed; curl 56 recv failure", + "fatal: unable to access: Could not resolve host: github.com", + "git push origin knit/x in /tmp/x timed out after 300s", + ] { + assert!( + is_retryable(&classify_git_push_message(message)), + "{message}" + ); + } + for message in [ + "! [rejected] knit/x -> knit/x (non-fast-forward)", + "! [rejected] knit/x -> knit/x (stale info)", + "fatal: Authentication failed for 'https://github.com/acme/backend'", + "remote: Permission denied to acme.", + ] { + assert!( + !is_retryable(&classify_git_push_message(message)), + "{message}" + ); + } + } + + #[test] + fn forge_failures_are_retried_only_when_the_host_was_unavailable() { + for message in [ + "gh pr create failed: Bad gateway (HTTP 502)", + "GitHub API request failed during POST /repos/a/b/pulls: HTTP 503: unavailable", + "HTTP 429: too many requests", + "HTTP 403: You have exceeded a secondary rate limit", + "GitHub API request failed during GET /x: io: connection reset by peer", + ] { + assert!(is_retryable(&classify_forge_message(message)), "{message}"); + } + for message in [ + "HTTP 401: Bad credentials", + "HTTP 404: Not Found", + "HTTP 422: Validation Failed: A pull request already exists for acme:knit/x.", + "HTTP 403: Resource not accessible by integration", + "HTTP 403: Forbidden", + "gh: command not found", + ] { + assert!(!is_retryable(&classify_forge_message(message)), "{message}"); + } + } + + #[test] + fn typed_http_failures_carry_status_and_retry_after() { + let error = Error::new(HttpFailure { + status: 429, + retry_after: Some(Duration::from_secs(7)), + detail: "too many requests".to_string(), + message: "HTTP 429".to_string(), + }) + .context("backend: failed to create the PR"); + match classify_forge(&error) { + Retryable::Yes { retry_after, .. } => { + assert_eq!(retry_after, Some(Duration::from_secs(7))); + } + Retryable::No => panic!("a 429 with Retry-After must be retried"), + } + } + + #[test] + fn status_and_retry_after_parsing() { + assert_eq!(http_status_in("bad gateway (http 502)"), Some(502)); + assert_eq!(http_status_in("http 429: slow down"), Some(429)); + assert_eq!(http_status_in("no status here"), None); + assert_eq!( + retry_after_in("http 429 (retry-after: 12s)"), + Some(Duration::from_secs(12)) + ); + } + + #[test] + fn retries_stop_at_the_attempt_limit_and_report_the_last_error() { + let mut attempts = 0; + let error = retry_with( + "push", + 3, + Duration::ZERO, + classify_git_push, + || -> Result<()> { + attempts += 1; + Err(anyhow!("fatal: Connection reset by peer")) + }, + ) + .unwrap_err(); + assert_eq!(attempts, 3); + assert!(error.to_string().contains("Connection reset")); + } + + #[test] + fn a_final_failure_is_not_repeated() { + let mut attempts = 0; + let error = retry_with( + "push", + 3, + Duration::ZERO, + classify_git_push, + || -> Result<()> { + attempts += 1; + Err(anyhow!("! [rejected] knit/x -> knit/x (non-fast-forward)")) + }, + ) + .unwrap_err(); + assert_eq!(attempts, 1); + assert!(error.to_string().contains("rejected")); + } + + #[test] + fn a_retry_that_succeeds_returns_the_value_and_announces_itself() { + let seen = std::rc::Rc::new(RefCell::new(Vec::new())); + let recorder = std::rc::Rc::clone(&seen); + let scope = stream_notes_to(move |line| recorder.borrow_mut().push(line)); + let mut attempts = 0; + let value = retry_with("push", 3, Duration::ZERO, classify_git_push, || { + attempts += 1; + if attempts == 1 { + Err(anyhow!("fatal: Connection reset by peer")) + } else { + Ok(7) + } + }) + .unwrap(); + drop(scope); + assert_eq!(value, 7); + assert_eq!(seen.borrow().len(), 1); + assert!( + seen.borrow()[0].contains("retrying push (2/3) after connection reset"), + "{:?}", + seen.borrow() + ); + } +} diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 4eac882..57a1f92 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -199,13 +199,19 @@ pub fn install_parallel_push_hook(repo: &Path, gate: &Path, id: &str, peer: &str install_parallel_gate_hook(repo, "pre-push", gate, id, peer); } -pub fn install_parallel_gate_hook(repo: &Path, hook: &str, gate: &Path, id: &str, peer: &str) { - fs::create_dir_all(gate).unwrap(); - // Resolve the repository's own hooks directory explicitly. Agent and IDE - // sessions may inject a process-wide core.hooksPath; honoring it here - // would install this temporary synchronization hook outside the fixture - // repository and contaminate unrelated tests or real Git operations. - let git_dir = PathBuf::from(git(repo, ["rev-parse", "--git-dir"]).trim()); +/// Install a git hook in this fixture repository only. +/// +/// Resolve the repository's own hooks directory explicitly. Agent and IDE +/// sessions may inject a process-wide core.hooksPath; honoring it here would +/// install a temporary test hook outside the fixture repository and +/// contaminate unrelated tests or real Git operations. +/// +/// The *common* git dir, not the per-worktree one: git runs hooks from the +/// shared directory, so a hook written into `.git/worktrees//hooks` +/// silently never runs — which is exactly the kind of quietly dead test +/// scaffolding these hooks exist to avoid. +pub fn write_hook(repo: &Path, hook: &str, script: &str) { + let git_dir = PathBuf::from(git(repo, ["rev-parse", "--git-common-dir"]).trim()); let git_dir = if git_dir.is_absolute() { git_dir } else { @@ -213,9 +219,122 @@ pub fn install_parallel_gate_hook(repo: &Path, hook: &str, gate: &Path, id: &str }; let hook_path = git_dir.join("hooks").join(hook); fs::create_dir_all(hook_path.parent().unwrap()).unwrap(); + fs::write(&hook_path, script).unwrap(); + make_executable(&hook_path); +} + +/// A pre-push hook that fails its first `failures` invocations with `message` +/// and counts every attempt in `state/count`. +pub fn install_flaky_push_hook(repo: &Path, state: &Path, message: &str, failures: u32) { + fs::create_dir_all(state).unwrap(); + write_hook( + repo, + "pre-push", + &format!( + r#"#!/bin/sh +set -eu +state={state} +count=$(cat "$state/count" 2>/dev/null || echo 0) +count=$((count + 1)) +printf '%s\n' "$count" > "$state/count" +if [ "$count" -le {failures} ]; then + printf '%s\n' {message} >&2 + exit 1 +fi +"#, + state = shell_quote(&state.to_string_lossy()), + message = shell_quote(message), + failures = failures + ), + ); +} + +/// A pre-push hook that hangs, so a small `KNIT_GIT_PUSH_TIMEOUT` can prove +/// the push is bounded rather than left to stall forever. +pub fn install_slow_push_hook(repo: &Path, seconds: u32) { + write_hook( + repo, + "pre-push", + &format!( + r#"#!/bin/sh +set -eu +sleep {seconds} +"# + ), + ); +} + +/// A pre-push hook that records how many pushes were in flight at once, so a +/// test can assert the pool never exceeded its limit. +pub fn install_concurrency_probe_hook(repo: &Path, state: &Path) { + fs::create_dir_all(state).unwrap(); + write_hook( + repo, + "pre-push", + &format!( + r#"#!/bin/sh +set -eu +state={state} +lock="$state/lock" +enter() {{ + while ! mkdir "$lock" 2>/dev/null; do sleep 0.01; done + live=$(cat "$state/live" 2>/dev/null || echo 0) + live=$((live + $1)) + printf '%s\n' "$live" > "$state/live" + peak=$(cat "$state/peak" 2>/dev/null || echo 0) + if [ "$live" -gt "$peak" ]; then printf '%s\n' "$live" > "$state/peak"; fi + rmdir "$lock" +}} +enter 1 +sleep 0.3 +enter -1 +"#, + state = shell_quote(&state.to_string_lossy()) + ), + ); +} + +pub fn read_counter(state: &Path, name: &str) -> u32 { + fs::read_to_string(state.join(name)) + .map(|value| value.trim().parse().unwrap_or(0)) + .unwrap_or(0) +} + +/// Make the fake `gh` fail `pr create` for `repo` with `stderr`. When `once` +/// is set the failure is spent after one attempt, so the retry can succeed. +pub fn fake_gh_fail_create(fake_gh_dir: &Path, repo: &str, stderr: &str, once: bool) { + fs::write( + fake_gh_dir.join(format!("create-fail-{repo}")), + format!("{stderr}\n"), + ) + .unwrap(); + if once { + fs::write(fake_gh_dir.join(format!("create-fail-once-{repo}")), "").unwrap(); + } +} + +/// Make the fake `gh` report an existing PR once a create attempt has reached +/// it: the shape of a create whose reply was lost after the host stored it. +pub fn fake_gh_existing_after_create(fake_gh_dir: &Path, repo: &str) { fs::write( - &hook_path, - format!( + fake_gh_dir.join(format!("existing-after-create-{repo}")), + "", + ) + .unwrap(); +} + +pub fn fake_gh_create_attempts(fake_gh_dir: &Path, repo: &str) -> usize { + fs::read_to_string(fake_gh_dir.join(format!("create-attempts-{repo}"))) + .map(|value| value.lines().count()) + .unwrap_or(0) +} + +pub fn install_parallel_gate_hook(repo: &Path, hook: &str, gate: &Path, id: &str, peer: &str) { + fs::create_dir_all(gate).unwrap(); + write_hook( + repo, + hook, + &format!( r#"#!/bin/sh set -eu gate={gate} @@ -236,9 +355,7 @@ done id = shell_quote(id), peer = shell_quote(peer) ), - ) - .unwrap(); - make_executable(&hook_path); + ); } pub fn shell_quote(value: &str) -> String { @@ -777,7 +894,16 @@ repo="$(basename "$PWD")" case "$sub" in list) - printf '[]\n' + if [ -f "$GH_FAKE_DIR/existing-after-create-$repo" ] && [ -f "$GH_FAKE_DIR/create-attempted-$repo" ]; then + case "$repo" in + backend) number=101 ;; + frontend) number=202 ;; + *) number=303 ;; + esac + printf '[{"number":%s,"url":"https://github.com/acme/%s/pull/%s","state":"OPEN","title":"%s PR","baseRefName":"main","headRefName":"knit/venue-capacity","body":"Existing body","isDraft":false,"headRefOid":"%s-head","mergeable":"MERGEABLE","mergeStateStatus":"CLEAN","reviewDecision":""}]\n' "$number" "$repo" "$number" "$repo" "$repo" + else + printf '[]\n' + fi ;; create) base="main" @@ -796,6 +922,18 @@ case "$sub" in printf '%s\n' "$base" > "$GH_FAKE_DIR/create-$repo.base" printf '%s\n' "$args" > "$GH_FAKE_DIR/create-$repo.args" cat > "$GH_FAKE_DIR/create-$repo.md" + printf 'x\n' >> "$GH_FAKE_DIR/create-attempts-$repo" + touch "$GH_FAKE_DIR/create-attempted-$repo" + if [ -f "$GH_FAKE_DIR/create-gate-$repo" ]; then + sh "$GH_FAKE_DIR/create-gate-$repo" + fi + if [ -f "$GH_FAKE_DIR/create-fail-$repo" ]; then + cat "$GH_FAKE_DIR/create-fail-$repo" >&2 + if [ -f "$GH_FAKE_DIR/create-fail-once-$repo" ]; then + rm -f "$GH_FAKE_DIR/create-fail-once-$repo" "$GH_FAKE_DIR/create-fail-$repo" + fi + exit 1 + fi case "$repo" in backend) number=101 ;; frontend) number=202 ;; diff --git a/tests/project.rs b/tests/project.rs index c6fb8b8..aca141e 100644 --- a/tests/project.rs +++ b/tests/project.rs @@ -1547,7 +1547,10 @@ fn bundle_agents_are_written_at_bundle_root_not_in_repo_checkouts() { assert!(bundle_agents.contains("## Agent Teamwork")); assert!(bundle_agents.contains("minimum capable subagent/model")); assert!(bundle_agents.contains("knit commit --all")); - assert!(bundle_agents.contains("knit push --set-upstream")); + // The review path is commit then publish: `knit publish create` pushes the + // branches itself, so the worktree guide must not teach a separate push. + assert!(bundle_agents.contains("knit publish create")); + assert!(bundle_agents.contains("`knit push --set-upstream` only when")); assert!(!bundle_agents.contains("knit --bundle")); assert_eq!( fs::read_to_string(&backend_agents_path).unwrap(), diff --git a/tests/publish.rs b/tests/publish.rs index 883509e..ef46be6 100644 --- a/tests/publish.rs +++ b/tests/publish.rs @@ -490,3 +490,170 @@ fn pr_create_can_override_base_branch() { fs::remove_dir_all(root).unwrap(); } + +/// One bundle with one committed repo and a fake `gh` on PATH: the shared +/// setup for the publish-resilience tests below. +fn one_repo_ready_to_publish(root: &std::path::Path) -> (std::path::PathBuf, std::path::PathBuf) { + let (_remote, backend, _collaborator) = init_remote_repo(root, "backend"); + let workspace = root.join("workspace"); + fs::create_dir_all(&workspace).unwrap(); + knit(&workspace, ["bundle", "venue capacity"]); + knit(&workspace, ["bundle", "add", backend.to_str().unwrap()]); + let feature = workspace.join(".knit/worktrees/venue-capacity/backend"); + append_line(&feature.join("app.txt"), "publish resilience"); + knit(&workspace, ["commit", "--all", "-m", "Publish resilience"]); + (workspace, feature) +} + +#[test] +fn pr_create_retries_a_host_that_was_briefly_unavailable() { + let root = unique_temp_dir(); + let (workspace, _feature) = one_repo_ready_to_publish(&root); + let fake_gh_dir = root.join("fake-gh"); + let fake_bin = root.join("fake-bin"); + write_fake_gh(&fake_bin, &fake_gh_dir); + fake_gh_fail_create(&fake_gh_dir, "backend", "gh: Bad gateway (HTTP 502)", true); + + let create = knit_with_fake_gh_env( + &workspace, + ["publish", "create", "--github"], + &fake_bin, + &fake_gh_dir, + &[("KNIT_RETRY_BASE_MS", "0")], + ); + assert!( + create.contains("backend: retrying gh pr create (2/4) after HTTP 502"), + "{create}" + ); + assert!(create.contains("created"), "{create}"); + assert_eq!(fake_gh_create_attempts(&fake_gh_dir, "backend"), 2); + + let bundle = read_bundle(&workspace); + assert_eq!(bundle["publications"].as_array().unwrap().len(), 1); + + fs::remove_dir_all(root).unwrap(); +} + +#[test] +fn pr_create_does_not_retry_a_refused_credential() { + let root = unique_temp_dir(); + let (workspace, _feature) = one_repo_ready_to_publish(&root); + let fake_gh_dir = root.join("fake-gh"); + let fake_bin = root.join("fake-bin"); + write_fake_gh(&fake_bin, &fake_gh_dir); + fake_gh_fail_create( + &fake_gh_dir, + "backend", + "gh: Bad credentials (HTTP 401)", + false, + ); + + let create = knit_fails_with_fake_gh_env( + &workspace, + ["publish", "create", "--github"], + &fake_bin, + &fake_gh_dir, + &[("KNIT_RETRY_BASE_MS", "0")], + ); + // GitHub answered. Repeating the call would only repeat the answer. + assert_eq!( + fake_gh_create_attempts(&fake_gh_dir, "backend"), + 1, + "{create}" + ); + assert!(!create.contains("retrying"), "{create}"); + assert!(create.contains("PR create failed"), "{create}"); + assert!(create.contains("re-run `knit publish create`"), "{create}"); + + fs::remove_dir_all(root).unwrap(); +} + +#[test] +fn pr_create_adopts_the_review_a_lost_create_already_made() { + let root = unique_temp_dir(); + let (workspace, _feature) = one_repo_ready_to_publish(&root); + let fake_gh_dir = root.join("fake-gh"); + let fake_bin = root.join("fake-bin"); + write_fake_gh(&fake_bin, &fake_gh_dir); + // The shape of a create whose reply was lost: the PR exists on the host, + // and the next create is told so. + fake_gh_existing_after_create(&fake_gh_dir, "backend"); + fake_gh_fail_create( + &fake_gh_dir, + "backend", + "gh: A pull request already exists for acme:knit/venue-capacity. (HTTP 422)", + false, + ); + + let create = knit_with_fake_gh_env( + &workspace, + ["publish", "create", "--github"], + &fake_bin, + &fake_gh_dir, + &[("KNIT_RETRY_BASE_MS", "0")], + ); + assert!(create.contains("exists"), "{create}"); + assert!(!create.contains("PR create failed"), "{create}"); + // A 422 is an answer, so the create itself is never repeated: the + // existing review is looked up instead. + assert_eq!(fake_gh_create_attempts(&fake_gh_dir, "backend"), 1); + + let bundle = read_bundle(&workspace); + let publications = bundle["publications"].as_array().unwrap(); + assert_eq!(publications.len(), 1); + assert_eq!( + publications[0]["url"].as_str(), + Some("https://github.com/acme/backend/pull/101") + ); + + fs::remove_dir_all(root).unwrap(); +} + +#[test] +fn pr_create_bounds_forge_writes_and_still_publishes_every_repo() { + let root = unique_temp_dir(); + let workspace = root.join("workspace"); + fs::create_dir_all(&workspace).unwrap(); + let names = ["one", "two", "three", "four", "five"]; + let mut paths = Vec::new(); + for name in names { + let (_remote, repo, _collaborator) = init_remote_repo(&root, name); + paths.push(repo); + } + knit(&workspace, ["bundle", "venue capacity"]); + let mut add: Vec = vec!["bundle".to_string(), "add".to_string()]; + add.extend(paths.iter().map(|path| path.to_str().unwrap().to_string())); + knit(&workspace, &add); + for name in names { + let feature = workspace.join(format!(".knit/worktrees/venue-capacity/{name}")); + append_line(&feature.join("app.txt"), "bounded publish"); + } + knit(&workspace, ["commit", "--all", "-m", "Bounded publish"]); + + let fake_gh_dir = root.join("fake-gh"); + let fake_bin = root.join("fake-bin"); + write_fake_gh(&fake_bin, &fake_gh_dir); + + let create = knit_with_fake_gh_env( + &workspace, + ["publish", "create", "--github"], + &fake_bin, + &fake_gh_dir, + &[("KNIT_FORGE_JOBS", "2")], + ); + assert!( + create.contains("publishing 5 repo(s), 2 at a time"), + "{create}" + ); + assert_eq!(create.matches(": pushed ").count(), 5, "{create}"); + assert_eq!(create.matches(": created ").count(), 5, "{create}"); + assert!(create.contains("(5/5)"), "{create}"); + for name in names { + assert_eq!(fake_gh_create_attempts(&fake_gh_dir, name), 1, "{name}"); + } + + let bundle = read_bundle(&workspace); + assert_eq!(bundle["publications"].as_array().unwrap().len(), 5); + + fs::remove_dir_all(root).unwrap(); +} diff --git a/tests/sync.rs b/tests/sync.rs index 9085606..6e41dc0 100644 --- a/tests/sync.rs +++ b/tests/sync.rs @@ -2450,3 +2450,249 @@ fn sync_push_history_sends_only_what_the_remote_does_not_have_yet() { fs::remove_dir_all(root).unwrap(); } + +/// One bundle, one repo, committed and ready to push. The shared setup for +/// the push-resilience tests below. +#[cfg(unix)] +fn one_repo_ready_to_push(root: &Path) -> (PathBuf, PathBuf, PathBuf, String) { + let (remote, backend, _collaborator) = init_remote_repo(root, "backend"); + let workspace = root.join("workspace"); + fs::create_dir_all(&workspace).unwrap(); + knit(&workspace, ["bundle", "venue capacity"]); + knit(&workspace, ["bundle", "add", backend.to_str().unwrap()]); + let feature = workspace.join(".knit/worktrees/venue-capacity/backend"); + append_line(&feature.join("app.txt"), "push resilience"); + knit(&workspace, ["commit", "--all", "-m", "Push resilience"]); + let sha = git(&feature, ["rev-parse", "HEAD"]); + (workspace, feature, remote, sha) +} + +#[test] +#[cfg(unix)] +fn push_bounds_how_many_repos_are_pushed_at_once() { + let root = unique_temp_dir(); + let workspace = root.join("workspace"); + fs::create_dir_all(&workspace).unwrap(); + let names = ["one", "two", "three", "four", "five"]; + let mut remotes = Vec::new(); + let mut paths = Vec::new(); + for name in names { + let (remote, repo, _collaborator) = init_remote_repo(&root, name); + remotes.push(remote); + paths.push(repo); + } + + knit(&workspace, ["bundle", "venue capacity"]); + let mut add: Vec = vec!["bundle".to_string(), "add".to_string()]; + add.extend(paths.iter().map(|path| path.to_str().unwrap().to_string())); + knit(&workspace, &add); + for name in names { + let feature = workspace.join(format!(".knit/worktrees/venue-capacity/{name}")); + append_line(&feature.join("app.txt"), "bounded push"); + } + knit(&workspace, ["commit", "--all", "-m", "Bounded push"]); + + // Every repo's pre-push hook reports into one shared counter, so the + // probe measures how many pushes the pool really ran side by side. + let probe = root.join("push-probe"); + for name in names { + install_concurrency_probe_hook( + &workspace.join(format!(".knit/worktrees/venue-capacity/{name}")), + &probe, + ); + } + + let push = knit_with_env(&workspace, ["push", "--all"], &[("KNIT_GIT_JOBS", "2")]); + assert!(push.contains("pushing 5 repo(s), 2 at a time"), "{push}"); + assert_eq!(push.matches(": pushed ").count(), 5, "{push}"); + assert!(push.contains("(5/5)"), "{push}"); + let peak = read_counter(&probe, "peak"); + assert!((1..=2).contains(&peak), "peak concurrency was {peak}"); + for (name, remote) in names.iter().zip(remotes) { + let feature = workspace.join(format!(".knit/worktrees/venue-capacity/{name}")); + assert_eq!( + git(&remote, ["rev-parse", "refs/heads/knit/venue-capacity"]), + git(&feature, ["rev-parse", "HEAD"]), + "{name}" + ); + } + + fs::remove_dir_all(root).unwrap(); +} + +#[test] +#[cfg(unix)] +fn push_rejects_a_job_limit_that_is_not_a_positive_number() { + let root = unique_temp_dir(); + let (workspace, _feature, _remote, _sha) = one_repo_ready_to_push(&root); + + let zero = knit_fails_with_env(&workspace, ["push", "backend"], &[("KNIT_GIT_JOBS", "0")]); + assert!( + zero.contains("KNIT_GIT_JOBS must be a positive whole number"), + "{zero}" + ); + let words = knit_fails_with_env( + &workspace, + ["push", "backend"], + &[("KNIT_GIT_JOBS", "lots")], + ); + assert!( + words.contains("KNIT_GIT_JOBS must be a positive whole number"), + "{words}" + ); + + fs::remove_dir_all(root).unwrap(); +} + +#[test] +#[cfg(unix)] +fn a_stalled_push_is_killed_and_names_the_repo_branch_and_limit() { + let root = unique_temp_dir(); + let (workspace, feature, _remote, _sha) = one_repo_ready_to_push(&root); + install_slow_push_hook(&feature, 10); + + let push = knit_fails_with_env( + &workspace, + ["push", "backend"], + &[("KNIT_GIT_PUSH_TIMEOUT", "1"), ("KNIT_RETRY_BASE_MS", "0")], + ); + assert!(push.contains("timed out after 1s"), "{push}"); + assert!(push.contains("backend"), "{push}"); + assert!(push.contains("knit/venue-capacity"), "{push}"); + // A timeout is a bad moment, not an answer: it is retried, and the + // failure still names what to do next. + assert!( + push.contains("retrying push (2/3) after a timeout"), + "{push}" + ); + assert!(push.contains("re-run the same `knit push`"), "{push}"); + + fs::remove_dir_all(root).unwrap(); +} + +#[test] +#[cfg(unix)] +fn a_dropped_push_is_retried_until_it_lands() { + let root = unique_temp_dir(); + let (workspace, feature, remote, sha) = one_repo_ready_to_push(&root); + let state = root.join("push-state"); + install_flaky_push_hook( + &feature, + &state, + "fatal: the remote end hung up unexpectedly", + 1, + ); + + let push = knit_with_env( + &workspace, + ["push", "backend"], + &[("KNIT_RETRY_BASE_MS", "0")], + ); + assert!( + push.contains("backend: retrying push (2/3) after the remote end hanging up"), + "{push}" + ); + assert!(push.contains("pushed"), "{push}"); + assert_eq!(read_counter(&state, "count"), 2); + assert_eq!( + git(&remote, ["rev-parse", "refs/heads/knit/venue-capacity"]), + sha + ); + + fs::remove_dir_all(root).unwrap(); +} + +#[test] +#[cfg(unix)] +fn a_rejected_push_is_reported_without_retrying() { + let root = unique_temp_dir(); + let (workspace, feature, _remote, _sha) = one_repo_ready_to_push(&root); + let state = root.join("push-state"); + install_flaky_push_hook( + &feature, + &state, + "! [rejected] knit/venue-capacity -> knit/venue-capacity (non-fast-forward)", + 99, + ); + + let push = knit_fails_with_env( + &workspace, + ["push", "backend"], + &[("KNIT_RETRY_BASE_MS", "0")], + ); + // The remote answered. Asking again would only be told the same thing. + assert_eq!(read_counter(&state, "count"), 1, "{push}"); + assert!(push.contains("push failed"), "{push}"); + assert!(!push.contains("retrying"), "{push}"); + + fs::remove_dir_all(root).unwrap(); +} + +#[test] +#[cfg(unix)] +fn one_failing_repo_does_not_stop_the_others_and_a_rerun_finishes_the_job() { + let root = unique_temp_dir(); + let (backend_remote, backend, _backend_collaborator) = init_remote_repo(&root, "backend"); + let (frontend_remote, frontend, _frontend_collaborator) = init_remote_repo(&root, "frontend"); + let workspace = root.join("workspace"); + fs::create_dir_all(&workspace).unwrap(); + knit(&workspace, ["bundle", "venue capacity"]); + knit( + &workspace, + [ + "bundle", + "add", + backend.to_str().unwrap(), + frontend.to_str().unwrap(), + ], + ); + let backend_feature = workspace.join(".knit/worktrees/venue-capacity/backend"); + let frontend_feature = workspace.join(".knit/worktrees/venue-capacity/frontend"); + append_line(&backend_feature.join("app.txt"), "partial backend"); + append_line(&frontend_feature.join("app.txt"), "partial frontend"); + knit(&workspace, ["commit", "--all", "-m", "Partial push"]); + + let state = root.join("frontend-state"); + install_flaky_push_hook( + &frontend_feature, + &state, + "! [rejected] knit/venue-capacity -> knit/venue-capacity (non-fast-forward)", + 99, + ); + + let push = knit_fails_with_env( + &workspace, + ["push", "--all"], + &[("KNIT_RETRY_BASE_MS", "0")], + ); + assert!(push.contains("backend"), "{push}"); + assert!(push.contains("push failed"), "{push}"); + assert!(push.contains("re-run the same `knit push`"), "{push}"); + // The healthy repo still reached origin. + assert_eq!( + git( + &backend_remote, + ["rev-parse", "refs/heads/knit/venue-capacity"] + ), + git(&backend_feature, ["rev-parse", "HEAD"]) + ); + + // Re-running after the cause is fixed only does the work that is missing: + // the already-pushed repo is a no-op, the failed one lands. + install_flaky_push_hook(&frontend_feature, &state, "unused", 0); + let again = knit_with_env( + &workspace, + ["push", "--all"], + &[("KNIT_RETRY_BASE_MS", "0")], + ); + assert_eq!(again.matches(": pushed ").count(), 2, "{again}"); + assert_eq!( + git( + &frontend_remote, + ["rev-parse", "refs/heads/knit/venue-capacity"] + ), + git(&frontend_feature, ["rev-parse", "HEAD"]) + ); + + fs::remove_dir_all(root).unwrap(); +}