From 5c6289df80d9a82dc500f5422402c9928b86b136 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 4 Aug 2026 14:51:29 -0400 Subject: [PATCH] Simplify async pull request creation Structural cleanup of the durable pull request creation feature, from a three-agent review (reuse, quality, efficiency) of the branch: - Move the supervisor out of handler/ into server/pull_request_supervisor.rs, collapse its double bookkeeping into one task-id map, and fold the five copy-pasted failure arms into attempt_pull_request_creation. - Tag pull_request.failed events with the creation id they resolve, so a publish-stage failure can never fail an unrelated explicit creation. The reducer gains PullRequestCreation::succeed/fail transition methods. - Scan pending creations through a narrow projection-cache accessor instead of materializing every run summary, raise the scan interval to 30s (notify covers the live path), and cap retries for runs whose worker cannot even record a failure. - Answer "creation already pending" POSTs before taking the per-run create lock, which a worker can hold for the whole creation. - Replace the hand-rolled per-run lock map with fabro_store::KeyedMutex. - Reuse cheap Arc'd projections (cached_run_projection) on the poll endpoint and in the worker instead of deep-cloning run summaries and diffs. - Merge ExistingPullRequest into fabro_github::CreatedPullRequest and extract one reconcile_existing_pull_request helper for both call sites. - Give the client poll loop a 15-minute deadline; document that Retry-After and the poll interval are the same constant. - Resolve a wedged pending creation (run already has a pull request) as a durable failure instead of skipping it forever. - Tests: shared wait_for_pull_request_creation helper, a pinned generation- failure assertion, and a new pipeline test proving reconciliation adopts an existing PR without an LLM call or create request. Verified: cargo build --workspace, cargo nextest run --workspace (7,767 passed), nightly clippy -D warnings, fmt --check, insta (no pending), bun typecheck in fabro-api-client. Co-Authored-By: Claude Fable 5 --- docs/internal/events.md | 6 + docs/public/api-reference/fabro-api.yaml | 5 +- .../src/commands/run/run_progress/mod.rs | 3 +- lib/apps/fabro-server/src/serve.rs | 15 +- lib/apps/fabro-server/src/server.rs | 72 +--- .../src/server/handler/pull_requests.rs | 370 +++--------------- .../src/server/pull_request_supervisor.rs | 265 +++++++++++++ lib/apps/fabro-server/src/server/tests.rs | 76 ++-- lib/components/fabro-github/src/lib.rs | 19 +- lib/components/fabro-store/src/run_state.rs | 65 +-- lib/components/fabro-store/src/slate/mod.rs | 7 + .../fabro-store/src/slate/projection_cache.rs | 21 + .../fabro-workflow/src/event/convert.rs | 5 +- .../fabro-workflow/src/event/events.rs | 5 +- .../fabro-workflow/src/pipeline/publish.rs | 6 +- .../src/pipeline/pull_request.rs | 185 +++++---- lib/foundation/fabro-client/src/client.rs | 14 + .../fabro-types/src/pull_request.rs | 16 + .../fabro-types/src/run_event/misc.rs | 6 +- .../fabro-api-client/src/api/runs-api.ts | 8 +- 20 files changed, 632 insertions(+), 537 deletions(-) create mode 100644 lib/apps/fabro-server/src/server/pull_request_supervisor.rs diff --git a/docs/internal/events.md b/docs/internal/events.md index d01c0991b..859c1687e 100644 --- a/docs/internal/events.md +++ b/docs/internal/events.md @@ -2199,6 +2199,7 @@ These legacy events may appear in older run logs. Current CLI backend runs do no "id": "...", "ts": "...", "run_id": "...", "event": "pull_request.failed", "properties": { + "creation_id": "01KYYK70WTZT2E551P3H5P0059", "error": "insufficient permissions" } } @@ -2206,8 +2207,13 @@ These legacy events may appear in older run logs. Current CLI backend runs do no | Property | Type | Description | |----------|------|-------------| +| `creation_id` | string (optional) | Explicit pull request creation this failure resolves. Absent for publish-stage failures. | | `error` | string | Error message | +When `creation_id` names the run's pending pull request creation, the run +projection marks that creation `failed`. A `pull_request.failed` event without +a `creation_id` (the workflow publish stage) does not change creation state. + ## Artifact events ### `artifact.captured` diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index b151695cc..99f8deed1 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -2580,6 +2580,10 @@ paths: server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. + + If a creation is already pending for the run, the response returns + that creation unchanged; any different `model` or `force` values in + the new request are ignored. parameters: - $ref: "#/components/parameters/RunId" requestBody: @@ -2644,7 +2648,6 @@ paths: application/json: schema: $ref: "#/components/schemas/ErrorResponse" - put: operationId: linkRunPullRequest tags: [Runs] diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs index 9e7e85030..53d336d49 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs @@ -1422,7 +1422,8 @@ mod tests { draft: true, }); emit(&mut ui, Event::PullRequestFailed { - error: "auth token expired".into(), + creation_id: None, + error: "auth token expired".into(), }); insta::assert_snapshot!(rendered(&buffer), @r" diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index 5b345d965..e22c61786 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -997,18 +997,11 @@ where } } else { cleanup_handle.abort(); - } - - if shutdown.is_cancelled() { - if let Err(join_err) = pull_request_creation_supervisor.await { - warn!(error = %join_err, "Pull request creation supervisor task panicked"); - } - } else { pull_request_creation_supervisor.abort(); - if let Err(join_err) = pull_request_creation_supervisor.await { - if !join_err.is_cancelled() { - warn!(error = %join_err, "Pull request creation supervisor task panicked"); - } + } + if let Err(join_err) = pull_request_creation_supervisor.await { + if !join_err.is_cancelled() { + warn!(error = %join_err, "Pull request creation supervisor task panicked"); } } diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 84810a68a..5445ad619 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -86,7 +86,7 @@ use fabro_slack::{blocks as slack_blocks, connection as slack_connection}; use fabro_static::EnvVars; use fabro_store::{ ArtifactKey, ArtifactStore, CachedRunProjection, Database, EventEnvelope, EventPayload, - NodeArtifact, PendingInterviewRecord, RunSummaryStore, StageArtifactEntry, StageId, + KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSummaryStore, StageArtifactEntry, StageId, }; #[cfg(test)] use fabro_types::BlockedReason; @@ -128,8 +128,7 @@ use tokio::process::Command; use tokio::runtime::Builder as TokioRuntimeBuilder; use tokio::sync::broadcast::error::RecvError; use tokio::sync::{ - Mutex as AsyncMutex, Notify, OwnedMutexGuard, RwLock as AsyncRwLock, Semaphore, broadcast, - mpsc, oneshot, + Mutex as AsyncMutex, Notify, RwLock as AsyncRwLock, Semaphore, broadcast, mpsc, oneshot, }; use tokio::task::spawn_blocking; use tokio::time::{sleep, timeout}; @@ -174,6 +173,7 @@ use crate::{ mod automation_scheduler; mod handler; +mod pull_request_supervisor; mod resource_sampler; mod session_runtime; @@ -186,9 +186,9 @@ pub(crate) use handler::graph::render_graph_bytes; pub(in crate::server) use handler::graph::{ RenderSubprocessError, render_dot_subprocess, render_graph_bytes_with_exe_override, }; -pub(crate) use handler::pull_requests::spawn_pull_request_creation_supervisor; #[cfg(test)] pub(in crate::server) use handler::system::validate_github_slug; +pub(crate) use pull_request_supervisor::spawn_pull_request_creation_supervisor; use session_runtime::SessionRuntimeManager; pub(crate) type EnvLookup = Arc Option + Send + Sync>; @@ -1127,7 +1127,7 @@ pub struct AppState { /// callers for the same run share one materialization; different runs /// proceed in parallel. See `crate::run_files` for semantics. pub(crate) files_in_flight: FilesInFlight, - pull_request_create_locks: PullRequestCreateLocks, + pull_request_create_locks: KeyedMutex, parent_link_lock: AsyncMutex<()>, pub(super) server_secrets: ServerSecrets, @@ -1160,8 +1160,6 @@ pub(crate) struct AppStores { pub(crate) variables: Arc, } -type PullRequestCreateLocks = Arc>>>>; - impl AppState { pub(crate) fn automation_store(&self) -> &AutomationStore { &self.stores.automations @@ -1255,50 +1253,6 @@ impl AskFabroReadiness { } } -struct PullRequestCreateGuard { - locks: PullRequestCreateLocks, - run_id: RunId, - mutex: Arc>, - guard: Option>, -} - -impl Drop for PullRequestCreateGuard { - fn drop(&mut self) { - self.guard.take(); - - let mut locks = self - .locks - .lock() - .expect("pull request create locks poisoned"); - if locks.get(&self.run_id).is_some_and(|mutex| { - Arc::ptr_eq(mutex, &self.mutex) && Arc::strong_count(&self.mutex) == 2 - }) { - locks.remove(&self.run_id); - } - } -} - -async fn lock_pull_request_create( - locks: &PullRequestCreateLocks, - run_id: &RunId, -) -> PullRequestCreateGuard { - let mutex = { - let mut locks = locks.lock().expect("pull request create locks poisoned"); - Arc::clone( - locks - .entry(*run_id) - .or_insert_with(|| Arc::new(AsyncMutex::new(()))), - ) - }; - let guard = mutex.clone().lock_owned().await; - PullRequestCreateGuard { - locks: Arc::clone(locks), - run_id: *run_id, - mutex, - guard: Some(guard), - } -} - pub(crate) struct AppStateConfig { pub(crate) resolved_settings: ResolvedAppStateSettings, pub(crate) registry_factory_override: Option>, @@ -1531,6 +1485,20 @@ impl AppState { .ok_or_else(|| ApiError::not_found("Run not found.")) } + /// Like [`Self::cached_run`], but returns only the shared projection — + /// no run summary clone or children count under the cache mutex. + pub(crate) async fn cached_run_projection( + &self, + run_id: &RunId, + ) -> Result, ApiError> { + self.stores + .runs + .get_cached_projection(run_id) + .await + .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))? + .ok_or_else(|| ApiError::not_found("Run not found.")) + } + pub(crate) fn session_runtimes(&self) -> &SessionRuntimeManager { &self.session_runtimes } @@ -2575,7 +2543,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result Router> { @@ -38,9 +33,10 @@ pub(super) fn routes() -> Router> { ) } -const PULL_REQUEST_CREATION_TIMEOUT: Duration = Duration::from_mins(10); -const PULL_REQUEST_CREATION_SCAN_INTERVAL: Duration = Duration::from_secs(5); -const MAX_CONCURRENT_PULL_REQUEST_CREATIONS: usize = 4; +/// Advertised via the 202 `Retry-After` header; the Rust client's poll +/// interval (`PULL_REQUEST_CREATION_POLL_INTERVAL` in `fabro-client`) matches +/// this value. +const PULL_REQUEST_CREATION_RETRY_AFTER: Duration = Duration::from_secs(1); #[expect( clippy::disallowed_types, @@ -81,7 +77,7 @@ fn pull_request_record_from_link_request( }) } -async fn load_server_github_credentials( +pub(in crate::server) async fn load_server_github_credentials( state: &AppState, ) -> Result { let settings = state.server_settings(); @@ -109,7 +105,7 @@ async fn load_server_github_credentials( } } -fn server_github_context<'a>( +pub(in crate::server) fn server_github_context<'a>( state: &'a AppState, creds: &'a fabro_github::GitHubCredentials, ) -> Result, ApiError> { @@ -135,6 +131,14 @@ fn github_pull_request_not_found_error(number: u64) -> ApiError { ) } +fn pull_request_exists_error(record: &PullRequestLink) -> ApiError { + ApiError::with_code( + StatusCode::CONFLICT, + format!("Pull request already exists at {}", record.html_url()), + "pull_request_exists", + ) +} + struct PullRequestGithubContext { record: PullRequestLink, owner: String, @@ -177,24 +181,23 @@ async fn load_pull_request_github_context( }) } -struct RunPrInputs<'a> { - goal: &'a str, - base_branch: &'a str, - run_branch: &'a str, - final_git_sha: &'a str, - diff: &'a str, - conclusion: &'a fabro_types::Conclusion, - normalized_origin: String, +pub(in crate::server) struct RunPrInputs<'a> { + pub(in crate::server) goal: &'a str, + pub(in crate::server) base_branch: &'a str, + pub(in crate::server) run_branch: &'a str, + pub(in crate::server) final_git_sha: &'a str, + pub(in crate::server) diff: &'a str, + pub(in crate::server) conclusion: &'a fabro_types::Conclusion, + pub(in crate::server) normalized_origin: String, } impl<'a> RunPrInputs<'a> { - fn extract(run_state: &'a fabro_store::RunProjection, force: bool) -> Result { + pub(in crate::server) fn extract( + run_state: &'a fabro_store::RunProjection, + force: bool, + ) -> Result { if let Some(record) = run_state.pull_request.as_ref() { - return Err(ApiError::with_code( - StatusCode::CONFLICT, - format!("Pull request already exists at {}", record.html_url()), - "pull_request_exists", - )); + return Err(pull_request_exists_error(record)); } let run_spec = &run_state.spec; let origin_url = run_spec.repo_origin_url().ok_or_else(|| { @@ -313,15 +316,16 @@ async fn create_run_pull_request( State(state): State>, Json(body): Json, ) -> Response { - let _create_guard = lock_pull_request_create(&state.pull_request_create_locks, &id).await; let Ok(run_store) = state.stores.runs.open_run(&id).await else { return ApiError::not_found("Run not found.").into_response(); }; - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let run_state = match state.cached_run_projection(&id).await { + Ok(run_state) => run_state, Err(err) => return err.into_response(), }; - let run_state = cached.projection.as_ref(); + // Answer before taking the per-run create lock: a running worker holds + // that lock for the whole creation, and an already-pending request only + // needs its current status. if let Some(creation) = run_state .pull_request_creation .as_ref() @@ -329,14 +333,10 @@ async fn create_run_pull_request( { return accepted_pull_request_creation_response(&id, creation.clone()); } - if let Err(err) = RunPrInputs::extract(run_state, body.force) { + if let Err(err) = RunPrInputs::extract(&run_state, body.force) { return err.into_response(); } - let creds = match load_server_github_credentials(state.as_ref()).await { - Ok(creds) => creds, - Err(err) => return err.into_response(), - }; - if let Err(err) = server_github_context(state.as_ref(), &creds) { + if let Err(err) = load_server_github_credentials(state.as_ref()).await { return err.into_response(); } let model = if let Some(model) = body.model { @@ -349,6 +349,7 @@ async fn create_run_pull_request( .id .to_string() }; + let _create_guard = state.pull_request_create_locks.lock(id).await; let creation_id = fabro_types::PullRequestCreationId::new(); let event = workflow_event::Event::PullRequestCreationRequested { creation_id, @@ -371,26 +372,20 @@ async fn create_run_pull_request( } }; - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let run_state = match state.cached_run_projection(&id).await { + Ok(run_state) => run_state, Err(err) => return err.into_response(), }; if !appended { - if let Some(creation) = cached - .projection + if let Some(creation) = run_state .pull_request_creation .as_ref() .filter(|creation| creation.is_pending()) { return accepted_pull_request_creation_response(&id, creation.clone()); } - if let Some(record) = cached.projection.pull_request.as_ref() { - return ApiError::with_code( - StatusCode::CONFLICT, - format!("Pull request already exists at {}", record.html_url()), - "pull_request_exists", - ) - .into_response(); + if let Some(record) = run_state.pull_request.as_ref() { + return pull_request_exists_error(record).into_response(); } return ApiError::new( StatusCode::CONFLICT, @@ -399,7 +394,7 @@ async fn create_run_pull_request( .into_response(); } - let Some(creation) = cached.projection.pull_request_creation.clone() else { + let Some(creation) = run_state.pull_request_creation.clone() else { return ApiError::new( StatusCode::INTERNAL_SERVER_ERROR, "Pull request creation was accepted but its status is unavailable.", @@ -416,12 +411,14 @@ fn accepted_pull_request_creation_response( ) -> Response { let mut response = (StatusCode::ACCEPTED, Json(creation)).into_response(); let location = format!("/api/v1/runs/{run_id}/pull_request/creation"); - if let Ok(location) = HeaderValue::from_str(&location) { - response.headers_mut().insert(header::LOCATION, location); - } - response - .headers_mut() - .insert(header::RETRY_AFTER, HeaderValue::from_static("1")); + response.headers_mut().insert( + header::LOCATION, + HeaderValue::try_from(location).expect("run ids are header-safe ASCII"), + ); + response.headers_mut().insert( + header::RETRY_AFTER, + HeaderValue::from(PULL_REQUEST_CREATION_RETRY_AFTER.as_secs()), + ); response } @@ -429,11 +426,11 @@ async fn get_run_pull_request_creation( RequireRunScoped(id): RequireRunScoped, State(state): State>, ) -> Response { - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let run_state = match state.cached_run_projection(&id).await { + Ok(run_state) => run_state, Err(err) => return err.into_response(), }; - match cached.projection.pull_request_creation.clone() { + match run_state.pull_request_creation.clone() { Some(creation) => Json(creation).into_response(), None => ApiError::with_code( StatusCode::NOT_FOUND, @@ -444,261 +441,12 @@ async fn get_run_pull_request_creation( } } -async fn append_pull_request_creation_failure( - run_store: &fabro_store::RunDatabase, - run_id: &RunId, - creation_id: fabro_types::PullRequestCreationId, - error: String, -) -> anyhow::Result<()> { - let event = workflow_event::Event::PullRequestFailed { error }; - workflow_event::append_event_if(run_store, run_id, &event, |projection| { - projection - .pull_request_creation - .as_ref() - .is_some_and(|creation| creation.id == creation_id && creation.is_pending()) - }) - .await?; - Ok(()) -} - -pub(crate) async fn process_pull_request_creation( - state: Arc, - run_id: RunId, -) -> anyhow::Result<()> { - let _create_guard = lock_pull_request_create(&state.pull_request_create_locks, &run_id).await; - let run_store = state.stores.runs.open_run(&run_id).await?; - let run_state = run_store.state().await?; - let Some(creation) = run_state - .pull_request_creation - .as_ref() - .filter(|creation| creation.is_pending()) - .cloned() - else { - return Ok(()); - }; - if run_state.pull_request.is_some() { - return Ok(()); - } - - let inputs = match RunPrInputs::extract(&run_state, creation.force) { - Ok(inputs) => inputs, - Err(err) => { - return append_pull_request_creation_failure( - &run_store, - &run_id, - creation.id, - err.detail().to_string(), - ) - .await; - } - }; - let creds = match load_server_github_credentials(state.as_ref()).await { - Ok(creds) => creds, - Err(err) => { - return append_pull_request_creation_failure( - &run_store, - &run_id, - creation.id, - err.detail().to_string(), - ) - .await; - } - }; - let github = match server_github_context(state.as_ref(), &creds) { - Ok(github) => github, - Err(err) => { - return append_pull_request_creation_failure( - &run_store, - &run_id, - creation.id, - err.detail().to_string(), - ) - .await; - } - }; - let catalog = state.catalog(); - let run_store_handle = run_store.clone().into(); - let request = pull_request::OpenPullRequestRequest { - github, - origin_url: &inputs.normalized_origin, - base_branch: inputs.base_branch, - head_branch: inputs.run_branch, - expected_head_sha: inputs.final_git_sha, - goal: inputs.goal, - diff: inputs.diff, - model: &creation.model, - draft: true, - auto_merge: None, - run_store: &run_store_handle, - llm_source: state.llm_source.as_ref(), - catalog, - conclusion: Some(inputs.conclusion), - run_state: Some(&run_state), - }; - let shutdown = state.shutdown_token(); - let result = tokio::select! { - () = shutdown.cancelled() => return Ok(()), - result = time::timeout(PULL_REQUEST_CREATION_TIMEOUT, pull_request::open_pull_request(request)) => result, - }; - let created_pull_request = match result { - Ok(Ok(created)) => created, - Ok(Err(err)) => { - return append_pull_request_creation_failure(&run_store, &run_id, creation.id, err) - .await; - } - Err(_) => { - return append_pull_request_creation_failure( - &run_store, - &run_id, - creation.id, - "Pull request creation timed out after 10 minutes.".to_string(), - ) - .await; - } - }; - - let event = workflow_event::Event::pull_request_created( - &created_pull_request.link, - &created_pull_request.base_branch, - &created_pull_request.head_branch, - inputs.final_git_sha, - &created_pull_request.title, - true, - ); - workflow_event::append_event_if(&run_store, &run_id, &event, |projection| { - projection.pull_request.is_none() - && projection - .pull_request_creation - .as_ref() - .is_some_and(|current| current.id == creation.id && current.is_pending()) - }) - .await?; - Ok(()) -} - -async fn pending_pull_request_creation_run_ids(state: &AppState) -> anyhow::Result> { - let mut pending = state - .stores - .runs - .list_cached_runs(&ListRunsQuery::default(), chrono::Utc::now()) - .await? - .into_iter() - .filter_map(|cached| { - let creation = cached.projection.pull_request_creation.as_ref()?; - (cached.projection.pull_request.is_none() && creation.is_pending()) - .then_some((cached.run_id, creation.requested_at)) - }) - .collect::>(); - pending.sort_by_key(|(run_id, requested_at)| (*requested_at, *run_id)); - Ok(pending.into_iter().map(|(run_id, _)| run_id).collect()) -} - -pub(crate) fn spawn_pull_request_creation_supervisor(state: Arc) -> JoinHandle<()> { - tokio::spawn( - run_pull_request_creation_supervisor(state) - .instrument(tracing::info_span!("pull_request_creation_supervisor")), - ) -} - -async fn run_pull_request_creation_supervisor(state: Arc) { - let shutdown = state.shutdown_token(); - let mut workers = JoinSet::new(); - let mut active = HashSet::new(); - let mut task_run_ids = std::collections::HashMap::new(); - let mut scan_requested = true; - let mut scan_interval = time::interval(PULL_REQUEST_CREATION_SCAN_INTERVAL); - scan_interval.set_missed_tick_behavior(time::MissedTickBehavior::Delay); - - loop { - if scan_requested { - match pending_pull_request_creation_run_ids(state.as_ref()).await { - Ok(pending) => { - let available = - MAX_CONCURRENT_PULL_REQUEST_CREATIONS.saturating_sub(active.len()); - let ready = pending - .into_iter() - .filter(|run_id| !active.contains(run_id)) - .take(available) - .collect::>(); - for run_id in ready { - active.insert(run_id); - let task_state = Arc::clone(&state); - let handle = workers.spawn( - async move { - let result = - process_pull_request_creation(task_state, run_id).await; - (run_id, result) - } - .instrument( - tracing::info_span!("pull_request_creation", run_id = %run_id), - ), - ); - task_run_ids.insert(handle.id(), run_id); - } - } - Err(err) => { - tracing::warn!(error = %err, "Failed to scan queued pull request creations"); - } - } - scan_requested = false; - } - - if shutdown.is_cancelled() { - break; - } - - if workers.is_empty() { - tokio::select! { - () = shutdown.cancelled() => break, - () = state.pull_request_scheduler_notified() => scan_requested = true, - _ = scan_interval.tick() => scan_requested = true, - } - continue; - } - - tokio::select! { - () = shutdown.cancelled() => break, - () = state.pull_request_scheduler_notified() => scan_requested = true, - _ = scan_interval.tick() => scan_requested = true, - joined = workers.join_next_with_id() => { - match joined { - Some(Ok((task_id, (run_id, Ok(()))))) => { - task_run_ids.remove(&task_id); - active.remove(&run_id); - scan_requested = true; - } - Some(Ok((task_id, (run_id, Err(err))))) => { - task_run_ids.remove(&task_id); - active.remove(&run_id); - tracing::warn!(run_id = %run_id, error = %err, "Pull request creation worker failed"); - } - Some(Err(err)) => { - if let Some(run_id) = task_run_ids.remove(&err.id()) { - active.remove(&run_id); - tracing::warn!(run_id = %run_id, error = %err, "Pull request creation worker stopped unexpectedly"); - } else { - tracing::warn!(error = %err, "Pull request creation worker stopped unexpectedly"); - } - } - None => {} - } - } - } - } - - while let Some(joined) = workers.join_next().await { - if let Err(err) = joined { - tracing::warn!(error = %err, "Pull request creation worker stopped during shutdown"); - } - } -} - async fn link_run_pull_request( RequireRunScoped(id): RequireRunScoped, State(state): State>, Json(body): Json, ) -> Response { - let _create_guard = lock_pull_request_create(&state.pull_request_create_locks, &id).await; + let _create_guard = state.pull_request_create_locks.lock(id).await; let pull_request = match pull_request_record_from_link_request(&body) { Ok(record) => record, Err(err) => return err.into_response(), @@ -720,15 +468,15 @@ async fn unlink_run_pull_request( RequireRunScoped(id): RequireRunScoped, State(state): State>, ) -> Response { - let _create_guard = lock_pull_request_create(&state.pull_request_create_locks, &id).await; + let _create_guard = state.pull_request_create_locks.lock(id).await; let Ok(run_store) = state.stores.runs.open_run(&id).await else { return ApiError::not_found("Run not found.").into_response(); }; - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let run_state = match state.cached_run_projection(&id).await { + Ok(run_state) => run_state, Err(err) => return err.into_response(), }; - let Some(pull_request) = cached.projection.pull_request.clone() else { + let Some(pull_request) = run_state.pull_request.clone() else { return ApiError::with_code( StatusCode::NOT_FOUND, format!("No pull request found in store. Create one first with: fabro pr create {id}"), diff --git a/lib/apps/fabro-server/src/server/pull_request_supervisor.rs b/lib/apps/fabro-server/src/server/pull_request_supervisor.rs new file mode 100644 index 000000000..2bebd4c97 --- /dev/null +++ b/lib/apps/fabro-server/src/server/pull_request_supervisor.rs @@ -0,0 +1,265 @@ +//! Background processing for durably accepted pull request creations. +//! +//! `POST /runs/{id}/pull_request` records a `pull_request.creation_requested` +//! event and returns 202; this supervisor finds pending creations (including +//! after a server restart), runs them under a bounded worker pool, and +//! records a durable success or failure result. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use fabro_types::{PullRequestCreation, PullRequestCreationId, RunId}; +use tokio::task::{self, JoinHandle, JoinSet}; +use tokio::time; +use tracing::{Instrument as _, info_span, warn}; + +use super::handler::pull_requests::{ + RunPrInputs, load_server_github_credentials, server_github_context, +}; +use super::{AppState, pull_request, workflow_event}; + +const PULL_REQUEST_CREATION_TIMEOUT: Duration = Duration::from_mins(10); +const PULL_REQUEST_CREATION_SCAN_INTERVAL: Duration = Duration::from_secs(30); +const MAX_CONCURRENT_PULL_REQUEST_CREATIONS: usize = 4; +/// Stop retrying a run after this many worker attempts that could not even +/// record a durable failure (store errors). Without a cap, such a run would +/// re-run the whole attempt — including the LLM call — on every scan. +const MAX_WORKER_FAILURES_PER_RUN: u32 = 3; + +async fn append_pull_request_creation_failure( + run_store: &fabro_store::RunDatabase, + run_id: &RunId, + creation_id: PullRequestCreationId, + error: String, +) -> anyhow::Result<()> { + let event = workflow_event::Event::PullRequestFailed { + creation_id: Some(creation_id), + error, + }; + workflow_event::append_event_if(run_store, run_id, &event, |projection| { + is_pending_creation(projection, creation_id) + }) + .await?; + Ok(()) +} + +fn is_pending_creation( + projection: &fabro_store::RunProjection, + creation_id: PullRequestCreationId, +) -> bool { + projection + .pull_request_creation + .as_ref() + .is_some_and(|creation| creation.id == creation_id && creation.is_pending()) +} + +pub(in crate::server) async fn process_pull_request_creation( + state: Arc, + run_id: RunId, +) -> anyhow::Result<()> { + let _create_guard = state.pull_request_create_locks.lock(run_id).await; + let run_store = state.stores.runs.open_run(&run_id).await?; + let Some(run_state) = state.stores.runs.get_cached_projection(&run_id).await? else { + return Ok(()); + }; + let Some(creation) = run_state + .pull_request_creation + .as_ref() + .filter(|creation| creation.is_pending()) + .cloned() + else { + return Ok(()); + }; + + match attempt_pull_request_creation(&state, &run_store, &run_id, &run_state, &creation).await? { + Ok(()) => Ok(()), + Err(error) => { + append_pull_request_creation_failure(&run_store, &run_id, creation.id, error).await + } + } +} + +/// One end-to-end creation attempt. The inner `Err` is a durable creation +/// failure for the caller to record; the inner `Ok` covers success and +/// shutdown-interrupted attempts (which stay pending). The outer `Err` is an +/// infrastructure failure — nothing was recorded, so the supervisor may retry. +async fn attempt_pull_request_creation( + state: &AppState, + run_store: &fabro_store::RunDatabase, + run_id: &RunId, + run_state: &fabro_store::RunProjection, + creation: &PullRequestCreation, +) -> anyhow::Result> { + let inputs = match RunPrInputs::extract(run_state, creation.force) { + Ok(inputs) => inputs, + Err(err) => return Ok(Err(err.detail().to_string())), + }; + let creds = match load_server_github_credentials(state).await { + Ok(creds) => creds, + Err(err) => return Ok(Err(err.detail().to_string())), + }; + let github = match server_github_context(state, &creds) { + Ok(github) => github, + Err(err) => return Ok(Err(err.detail().to_string())), + }; + let catalog = state.catalog(); + let run_store_handle = run_store.clone().into(); + let request = pull_request::OpenPullRequestRequest { + github, + origin_url: &inputs.normalized_origin, + base_branch: inputs.base_branch, + head_branch: inputs.run_branch, + expected_head_sha: inputs.final_git_sha, + goal: inputs.goal, + diff: inputs.diff, + model: &creation.model, + draft: true, + auto_merge: None, + run_store: &run_store_handle, + llm_source: state.llm_source.as_ref(), + catalog, + conclusion: Some(inputs.conclusion), + run_state: Some(run_state), + }; + let shutdown = state.shutdown_token(); + let result = tokio::select! { + () = shutdown.cancelled() => return Ok(Ok(())), + result = time::timeout(PULL_REQUEST_CREATION_TIMEOUT, pull_request::open_pull_request(request)) => result, + }; + let created_pull_request = match result { + Ok(Ok(created)) => created, + Ok(Err(err)) => return Ok(Err(err)), + Err(_) => { + return Ok(Err(format!( + "Pull request creation timed out after {} minutes.", + PULL_REQUEST_CREATION_TIMEOUT.as_secs() / 60 + ))); + } + }; + + let event = workflow_event::Event::pull_request_created( + &created_pull_request.link, + &created_pull_request.base_branch, + &created_pull_request.head_branch, + inputs.final_git_sha, + &created_pull_request.title, + true, + ); + workflow_event::append_event_if(run_store, run_id, &event, |projection| { + projection.pull_request.is_none() && is_pending_creation(projection, creation.id) + }) + .await?; + Ok(Ok(())) +} + +pub(crate) fn spawn_pull_request_creation_supervisor(state: Arc) -> JoinHandle<()> { + tokio::spawn( + run_pull_request_creation_supervisor(state) + .instrument(info_span!("pull_request_creation_supervisor")), + ) +} + +async fn run_pull_request_creation_supervisor(state: Arc) { + let shutdown = state.shutdown_token(); + let mut workers = JoinSet::new(); + let mut active: HashMap = HashMap::new(); + let mut failures: HashMap = HashMap::new(); + let mut scan_requested = true; + let mut scan_interval = time::interval(PULL_REQUEST_CREATION_SCAN_INTERVAL); + scan_interval.set_missed_tick_behavior(time::MissedTickBehavior::Delay); + + loop { + if scan_requested { + match state + .stores + .runs + .pending_pull_request_creation_run_ids() + .await + { + Ok(pending) => { + let available = + MAX_CONCURRENT_PULL_REQUEST_CREATIONS.saturating_sub(active.len()); + let ready = pending + .into_iter() + .filter(|run_id| { + !active.values().any(|active_id| active_id == run_id) + && failures.get(run_id).copied().unwrap_or(0) + < MAX_WORKER_FAILURES_PER_RUN + }) + .take(available) + .collect::>(); + for run_id in ready { + let handle = workers.spawn( + process_pull_request_creation(Arc::clone(&state), run_id) + .instrument(info_span!("pull_request_creation", run_id = %run_id)), + ); + active.insert(handle.id(), run_id); + } + } + Err(err) => { + warn!(error = %err, "Failed to scan queued pull request creations"); + } + } + scan_requested = false; + } + + if shutdown.is_cancelled() { + break; + } + + if workers.is_empty() { + tokio::select! { + () = shutdown.cancelled() => break, + () = state.pull_request_scheduler_notified() => scan_requested = true, + _ = scan_interval.tick() => scan_requested = true, + } + continue; + } + + tokio::select! { + () = shutdown.cancelled() => break, + () = state.pull_request_scheduler_notified() => scan_requested = true, + _ = scan_interval.tick() => scan_requested = true, + joined = workers.join_next_with_id() => { + match joined { + Some(Ok((task_id, result))) => { + let run_id = active.remove(&task_id); + match (run_id, result) { + (Some(run_id), Ok(())) => { + failures.remove(&run_id); + scan_requested = true; + } + (Some(run_id), Err(err)) => { + // Deliberately no immediate rescan: the run's + // creation is still pending, and re-picking it + // now would retry the whole attempt in a tight + // loop. The next interval tick retries it. + *failures.entry(run_id).or_default() += 1; + warn!(run_id = %run_id, error = %err, "Pull request creation worker failed"); + } + (None, result) => { + warn!(?result, "Pull request creation worker finished without a tracked run id"); + } + } + } + Some(Err(err)) => { + if let Some(run_id) = active.remove(&err.id()) { + *failures.entry(run_id).or_default() += 1; + warn!(run_id = %run_id, error = %err, "Pull request creation worker stopped unexpectedly"); + } else { + warn!(error = %err, "Pull request creation worker stopped unexpectedly"); + } + } + None => {} + } + } + } + } + + while let Some(joined) = workers.join_next().await { + if let Err(err) = joined { + warn!(error = %err, "Pull request creation worker stopped during shutdown"); + } + } +} diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 54ce2fbe4..573c43e2b 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -4098,6 +4098,30 @@ async fn wait_for_mock_hits(mock: &httpmock::Mock<'_>, expected: usize) { panic!("mock did not receive {expected} request(s)"); } +/// Poll `GET /runs/{id}/pull_request/creation` until the creation leaves +/// `pending`, returning the terminal creation body. +async fn wait_for_pull_request_creation(app: &Router, run_id: RunId) -> serde_json::Value { + for _ in 0..150 { + let response = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/pull_request/creation"))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let body = response_json!(response, StatusCode::OK).await; + if body["status"] != "pending" { + return body; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + panic!("pull request creation for run {run_id} did not finish"); +} + async fn title_update_event_count(state: &AppState, run_id: RunId) -> usize { let run_store = state.stores.runs.open_run(&run_id).await.unwrap(); run_store @@ -9754,28 +9778,7 @@ async fn create_run_pull_request_creates_and_persists_record() { // the durable pending event is enough to resume the operation. let supervisor = spawn_pull_request_creation_supervisor(Arc::clone(&state)); - let creation_body = tokio::time::timeout(std::time::Duration::from_secs(3), async { - loop { - let response = app - .clone() - .oneshot( - Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}/pull_request/creation"))) - .body(Body::empty()) - .unwrap(), - ) - .await - .unwrap(); - let body = response_json!(response, StatusCode::OK).await; - if body["status"] != "pending" { - break body; - } - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - } - }) - .await - .expect("pull request creation should finish"); + let creation_body = wait_for_pull_request_creation(&app, run_id).await; assert_eq!(creation_body["status"], "succeeded"); assert_eq!(creation_body["pull_request"]["number"], 42); @@ -9916,34 +9919,17 @@ async fn create_run_pull_request_persists_generation_failure() { response_json!(response, StatusCode::ACCEPTED).await; let supervisor = spawn_pull_request_creation_supervisor(Arc::clone(&state)); - let creation = tokio::time::timeout(std::time::Duration::from_secs(3), async { - loop { - let response = app - .clone() - .oneshot( - Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}/pull_request/creation"))) - .body(Body::empty()) - .unwrap(), - ) - .await - .unwrap(); - let body = response_json!(response, StatusCode::OK).await; - if body["status"] != "pending" { - break body; - } - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - } - }) - .await - .expect("pull request creation should finish"); + let creation = wait_for_pull_request_creation(&app, run_id).await; assert_eq!(creation["status"], "failed"); + // The unconfigured LLM is what fails this fixture; pin the error to the + // generation step so the test cannot pass on an earlier validation error. assert!( creation["error"] .as_str() - .is_some_and(|error| !error.is_empty()) + .is_some_and(|error| error.contains("LLM generation failed")), + "unexpected error: {:?}", + creation["error"] ); assert!(creation["pull_request"].is_null()); branch_mock.assert(); diff --git a/lib/components/fabro-github/src/lib.rs b/lib/components/fabro-github/src/lib.rs index d4d4baff8..fd4b98936 100644 --- a/lib/components/fabro-github/src/lib.rs +++ b/lib/components/fabro-github/src/lib.rs @@ -635,17 +635,9 @@ pub async fn create_installation_access_token_for_pr( .await } -/// Result of a successful pull request creation. -pub struct CreatedPullRequest { - pub html_url: String, - pub number: u64, - pub node_id: String, -} - -/// Existing open pull request found for an exact base, head branch, and head -/// commit. +/// Pull request created on, or reconciled from, GitHub. #[derive(Debug, Clone, PartialEq, Eq)] -pub struct ExistingPullRequest { +pub struct CreatedPullRequest { pub html_url: String, pub number: u64, pub node_id: String, @@ -663,7 +655,7 @@ pub async fn find_open_pull_request( base: &str, head: &str, expected_head_sha: &str, -) -> anyhow::Result> { +) -> anyhow::Result> { let client = ctx.http_client()?; find_open_pull_request_with_client(&client, ctx, owner, repo, base, head, expected_head_sha) .await @@ -681,7 +673,7 @@ pub async fn find_open_pull_request_with_client( base: &str, head: &str, expected_head_sha: &str, -) -> anyhow::Result> { +) -> anyhow::Result> { #[derive(Deserialize)] struct PullRequestHead { sha: String, @@ -736,7 +728,7 @@ pub async fn find_open_pull_request_with_client( Ok(pull_requests .into_iter() .find(|pull_request| pull_request.head.sha == expected_head_sha) - .map(|pull_request| ExistingPullRequest { + .map(|pull_request| CreatedPullRequest { html_url: pull_request.html_url, number: pull_request.number, node_id: pull_request.node_id, @@ -849,6 +841,7 @@ pub async fn create_pull_request_with_client( html_url: pr.html_url, number: pr.number, node_id: pr.node_id, + title: title.to_string(), }) } diff --git a/lib/components/fabro-store/src/run_state.rs b/lib/components/fabro-store/src/run_state.rs index 2e50d448c..d685f2bea 100644 --- a/lib/components/fabro-store/src/run_state.rs +++ b/lib/components/fabro-store/src/run_state.rs @@ -326,37 +326,39 @@ impl RunProjectionReducer for RunProjection { number: props.pr_number, }; self.pull_request = Some(pull_request.clone()); - if let Some(creation) = self.pull_request_creation.as_mut() { - if creation.is_pending() { - creation.status = PullRequestCreationStatus::Succeeded; - creation.updated_at = ts; - creation.pull_request = Some(pull_request); - creation.error = None; - } + if let Some(creation) = self + .pull_request_creation + .as_mut() + .filter(|creation| creation.is_pending()) + { + creation.succeed(pull_request, ts); } } EventBody::PullRequestLinked(props) => { self.pull_request = Some(props.pull_request.clone()); - if let Some(creation) = self.pull_request_creation.as_mut() { - if creation.is_pending() { - creation.status = PullRequestCreationStatus::Succeeded; - creation.updated_at = ts; - creation.pull_request = Some(props.pull_request.clone()); - creation.error = None; - } + if let Some(creation) = self + .pull_request_creation + .as_mut() + .filter(|creation| creation.is_pending()) + { + creation.succeed(props.pull_request.clone(), ts); } } EventBody::PullRequestUnlinked(_) => { self.pull_request = None; + // Clear the creation record too: a lingering `Succeeded` + // record would point at a pull request that is no longer + // linked, and it would block a later explicit creation. self.pull_request_creation = None; } EventBody::PullRequestFailed(props) => { - if let Some(creation) = self.pull_request_creation.as_mut() { - if creation.is_pending() { - creation.status = PullRequestCreationStatus::Failed; - creation.updated_at = ts; - creation.error = Some(props.error.clone()); - } + // Only a failure that names the pending creation resolves it; + // publish-stage failures carry no creation id and must not + // fail an unrelated explicit creation. + if let Some(creation) = self.pull_request_creation.as_mut().filter(|creation| { + Some(creation.id) == props.creation_id && creation.is_pending() + }) { + creation.fail(props.error.clone(), ts); } } EventBody::InterviewStarted(props) => { @@ -4663,7 +4665,23 @@ mod tests { .apply_event(&test_event( 2, EventBody::PullRequestFailed(PullRequestFailedProps { - error: "provider unavailable".to_string(), + creation_id: None, + error: "publish stage failure".to_string(), + }), + None, + )) + .unwrap(); + assert!( + state.pull_request_creation.as_ref().unwrap().is_pending(), + "a failure without a creation id must not resolve the creation" + ); + + state + .apply_event(&test_event( + 3, + EventBody::PullRequestFailed(PullRequestFailedProps { + creation_id: Some(first_id), + error: "provider unavailable".to_string(), }), None, )) @@ -4675,7 +4693,7 @@ mod tests { let retry_id = "01KYYK70WTZT2E551P3H5P0060".parse().unwrap(); state .apply_event(&test_event( - 3, + 4, EventBody::PullRequestCreationRequested(PullRequestCreationRequestedProps { creation_id: retry_id, model: "claude-sonnet-4-6".to_string(), @@ -4688,7 +4706,7 @@ mod tests { state .apply_event(&test_event( - 4, + 5, EventBody::PullRequestCreated(PullRequestCreatedProps { pr_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(), pr_number: 123, @@ -4706,6 +4724,7 @@ mod tests { let succeeded = state.pull_request_creation.as_ref().unwrap(); assert_eq!(succeeded.status, PullRequestCreationStatus::Succeeded); assert_eq!(succeeded.pull_request.as_ref().unwrap().number, 123); + assert_eq!(succeeded.pull_request, state.pull_request); assert!(succeeded.error.is_none()); } diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 893b0d8f0..e0eaffb96 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -364,6 +364,13 @@ impl Database { Ok(self.projection_cache.get_summary(run_id, now).await) } + /// Run ids whose latest explicit pull request creation is still pending, + /// oldest request first. + pub async fn pending_pull_request_creation_run_ids(&self) -> Result> { + self.warm_projection_cache().await?; + Ok(self.projection_cache.pending_pull_request_creations().await) + } + pub async fn put_session_run_index( &self, session_id: &SessionId, diff --git a/lib/components/fabro-store/src/slate/projection_cache.rs b/lib/components/fabro-store/src/slate/projection_cache.rs index 79ff0c03b..6d53d2a6d 100644 --- a/lib/components/fabro-store/src/slate/projection_cache.rs +++ b/lib/components/fabro-store/src/slate/projection_cache.rs @@ -189,6 +189,27 @@ impl RunProjectionCache { .map(|entry| (Arc::clone(&entry.projection), entry.last_seq)) } + /// Run ids whose latest explicit pull request creation is still pending, + /// oldest request first. Clones only ids and timestamps, so callers can + /// poll on an interval without materializing run summaries. + pub(crate) async fn pending_pull_request_creations(&self) -> Vec { + let mut pending = self + .state + .lock() + .await + .entries + .values() + .filter_map(|entry| { + let creation = entry.projection.pull_request_creation.as_ref()?; + creation + .is_pending() + .then_some((creation.requested_at, entry.run_id)) + }) + .collect::>(); + pending.sort_unstable(); + pending.into_iter().map(|(_, run_id)| run_id).collect() + } + pub(crate) async fn get_summary(&self, run_id: &RunId, now: DateTime) -> Option { let mut entry = { let state = self.state.lock().await; diff --git a/lib/components/fabro-workflow/src/event/convert.rs b/lib/components/fabro-workflow/src/event/convert.rs index 3ac58c2d8..5f81e1806 100644 --- a/lib/components/fabro-workflow/src/event/convert.rs +++ b/lib/components/fabro-workflow/src/event/convert.rs @@ -1362,9 +1362,10 @@ fn event_body_from_event(event: &Event) -> EventBody { pull_request: pull_request.clone(), }) } - Event::PullRequestFailed { error } => { + Event::PullRequestFailed { creation_id, error } => { EventBody::PullRequestFailed(fabro_types::PullRequestFailedProps { - error: error.clone(), + creation_id: *creation_id, + error: error.clone(), }) } } diff --git a/lib/components/fabro-workflow/src/event/events.rs b/lib/components/fabro-workflow/src/event/events.rs index 61a1ceeaa..f277caac3 100644 --- a/lib/components/fabro-workflow/src/event/events.rs +++ b/lib/components/fabro-workflow/src/event/events.rs @@ -751,7 +751,10 @@ pub enum Event { pull_request: PullRequestLink, }, PullRequestFailed { - error: String, + /// Set when the failure resolves an explicitly requested creation; + /// `None` for pull request failures in the workflow publish stage. + creation_id: Option, + error: String, }, } diff --git a/lib/components/fabro-workflow/src/pipeline/publish.rs b/lib/components/fabro-workflow/src/pipeline/publish.rs index 6adce32b9..30fe5b055 100644 --- a/lib/components/fabro-workflow/src/pipeline/publish.rs +++ b/lib/components/fabro-workflow/src/pipeline/publish.rs @@ -116,7 +116,8 @@ impl Concluded { .await .map_err(|error| { self.services.emitter.emit(&Event::PullRequestFailed { - error: error.clone(), + creation_id: None, + error: error.clone(), }); Error::publish_with_source("failed to create pull request", anyhow::anyhow!(error)) })?; @@ -185,7 +186,8 @@ impl Concluded { fn pull_request_error(&self, message: &str) -> Error { self.services.emitter.emit(&Event::PullRequestFailed { - error: message.to_string(), + creation_id: None, + error: message.to_string(), }); Error::publish(message) } diff --git a/lib/components/fabro-workflow/src/pipeline/pull_request.rs b/lib/components/fabro-workflow/src/pipeline/pull_request.rs index 21d9164a5..71de7edb4 100644 --- a/lib/components/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/components/fabro-workflow/src/pipeline/pull_request.rs @@ -472,23 +472,47 @@ pub struct CreatedPullRequest { pub head_branch: String, } -fn recovered_pull_request( - existing: fabro_github::ExistingPullRequest, - owner: String, - repo: String, - base_branch: &str, - head_branch: &str, -) -> CreatedPullRequest { - CreatedPullRequest { +/// Adopt an open pull request that already exists for the head branch at the +/// expected commit, e.g. when GitHub created the pull request but the caller +/// stopped before persisting the result. +async fn reconcile_existing_pull_request( + req: &OpenPullRequestRequest<'_>, + owner: &str, + repo: &str, + context: &'static str, +) -> anyhow::Result> { + let Some(existing) = github_app::find_open_pull_request( + &req.github, + owner, + repo, + req.base_branch, + req.head_branch, + req.expected_head_sha, + ) + .await? + else { + return Ok(None); + }; + info!(pr_url = %existing.html_url, pr_number = existing.number, context, "Existing pull request reconciled"); + enable_auto_merge_if_requested( + &req.github, + owner, + repo, + &existing.node_id, + existing.number, + req.auto_merge.as_ref(), + ) + .await; + Ok(Some(CreatedPullRequest { link: PullRequestLink { - owner, - repo, + owner: owner.to_string(), + repo: repo.to_string(), number: existing.number, }, title: existing.title, - base_branch: base_branch.to_string(), - head_branch: head_branch.to_string(), - } + base_branch: req.base_branch.to_string(), + head_branch: req.head_branch.to_string(), + })) } async fn enable_auto_merge_if_requested( @@ -576,34 +600,11 @@ pub async fn open_pull_request( // branch would otherwise cost a full LLM call before failing. verify_remote_head(&req, &owner, &repo).await?; - if let Some(existing) = github_app::find_open_pull_request( - &req.github, - &owner, - &repo, - req.base_branch, - req.head_branch, - req.expected_head_sha, - ) - .await - .map_err(|err| format!("failed to reconcile an existing pull request: {err:#}"))? + if let Some(existing) = reconcile_existing_pull_request(&req, &owner, &repo, "before creation") + .await + .map_err(|err| format!("failed to reconcile an existing pull request: {err:#}"))? { - info!(pr_url = %existing.html_url, pr_number = existing.number, "Existing pull request reconciled"); - enable_auto_merge_if_requested( - &req.github, - &owner, - &repo, - &existing.node_id, - existing.number, - req.auto_merge.as_ref(), - ) - .await; - return Ok(recovered_pull_request( - existing, - owner, - repo, - req.base_branch, - req.head_branch, - )); + return Ok(existing); } let content = build_pr_content( @@ -635,35 +636,10 @@ pub async fn open_pull_request( { Ok(created) => created, Err(create_err) => { - match github_app::find_open_pull_request( - &req.github, - &owner, - &repo, - req.base_branch, - req.head_branch, - req.expected_head_sha, - ) - .await + match reconcile_existing_pull_request(&req, &owner, &repo, "after a failed create") + .await { - Ok(Some(existing)) => { - info!(pr_url = %existing.html_url, pr_number = existing.number, "Pull request reconciled after create response failed"); - enable_auto_merge_if_requested( - &req.github, - &owner, - &repo, - &existing.node_id, - existing.number, - req.auto_merge.as_ref(), - ) - .await; - return Ok(recovered_pull_request( - existing, - owner, - repo, - req.base_branch, - req.head_branch, - )); - } + Ok(Some(existing)) => return Ok(existing), Ok(None) => return Err(format!("{create_err:#}")), Err(reconcile_err) => { return Err(format!( @@ -1750,6 +1726,15 @@ mod tests { async fn setup_fallback_test_harness_with_branch_sha( openai_payload_text: &str, branch_sha: &str, + ) -> FallbackHarness { + setup_fallback_test_harness_with(openai_payload_text, branch_sha, serde_json::json!([])) + .await + } + + async fn setup_fallback_test_harness_with( + openai_payload_text: &str, + branch_sha: &str, + reconcile_response: serde_json::Value, ) -> FallbackHarness { let openai_server = MockServer::start_async().await; let openai_mock = openai_server @@ -1792,7 +1777,7 @@ mod tests { }) .await; let reconcile_mock = github_server - .mock_async(|when, then| { + .mock_async(move |when, then| { when.method(GET) .path("/repos/owner/repo/pulls") .query_param("state", "open") @@ -1801,7 +1786,7 @@ mod tests { .header("authorization", "Bearer test-token"); then.status(200) .header("content-type", "application/json") - .json_body(serde_json::json!([])); + .json_body(reconcile_response); }) .await; @@ -1912,6 +1897,66 @@ mod tests { } } + /// An open pull request already exists for the head branch at the + /// expected commit — for example after a crash between GitHub creating + /// the pull request and the caller persisting it. `open_pull_request` + /// adopts it without an LLM call and without a create request. + #[tokio::test] + async fn open_pull_request_adopts_an_existing_pull_request_without_creating() { + let payload = pr_content_json("Unused", "Unused."); + let harness = setup_fallback_test_harness_with( + &payload, + "final-sha", + serde_json::json!([{ + "html_url": "https://github.com/owner/repo/pull/7", + "number": 7, + "node_id": "PR_existing", + "title": "Reconciled title", + "head": {"sha": "final-sha"} + }]), + ) + .await; + + let github_base_url = harness.github_server.url(""); + let github = github_app::GitHubContext::new(&harness.creds, &github_base_url); + + let result = open_pull_request(OpenPullRequestRequest { + github, + origin_url: "https://github.com/owner/repo.git", + base_branch: "main", + head_branch: "fabro/run/123", + expected_head_sha: "final-sha", + goal: "Fix telemetry leak", + diff: "diff --git a/src/lib.rs b/src/lib.rs\n+fn x() {}\n", + model: "gpt-5.4", + draft: false, + auto_merge: None, + run_store: &harness.run_store, + llm_source: harness.llm_source.as_ref(), + catalog: harness.catalog.clone(), + conclusion: None, + run_state: None, + }) + .await + .expect("reconciliation should adopt the existing pull request"); + + assert_eq!(result.link.number, 7); + assert_eq!(result.title, "Reconciled title"); + // Adoption must not cost an LLM call or a create request. + assert_eq!( + httpmock::Mock::new(harness.openai_mock_id, &harness.openai_server) + .calls_async() + .await, + 0 + ); + assert_eq!( + httpmock::Mock::new(harness.github_mock_id, &harness.github_server) + .calls_async() + .await, + 0 + ); + } + /// LLM returns a usable body but an empty title; the content builder /// falls back to `pr_title_from_goal` (first line, decoration stripped) /// and PR creation succeeds with that title. diff --git a/lib/foundation/fabro-client/src/client.rs b/lib/foundation/fabro-client/src/client.rs index 4b3537aa9..4204272f7 100644 --- a/lib/foundation/fabro-client/src/client.rs +++ b/lib/foundation/fabro-client/src/client.rs @@ -38,7 +38,13 @@ use crate::{AuthEntry, OAuthEntry, StoredSubject, sse}; const DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); const DEFAULT_HEALTH_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(250); +/// Matches the `Retry-After` the server sends on the 202 +/// (`PULL_REQUEST_CREATION_RETRY_AFTER` in `fabro-server`). const PULL_REQUEST_CREATION_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); +/// Overall polling deadline. The server abandons a creation attempt after 10 +/// minutes, but a creation can also sit pending behind the server's worker +/// pool (or a dead server), so the client needs its own bound. +const PULL_REQUEST_CREATION_POLL_DEADLINE: std::time::Duration = std::time::Duration::from_mins(15); type TransportFuture = BoxFuture<'static, Result<(fabro_http::HttpClient, String)>>; @@ -1440,9 +1446,17 @@ impl Client { .request_run_pull_request_creation(run_id, force, model) .await?; let creation_id = creation.id; + let deadline = std::time::Instant::now() + PULL_REQUEST_CREATION_POLL_DEADLINE; loop { match creation.status { fabro_types::PullRequestCreationStatus::Pending => { + if std::time::Instant::now() >= deadline { + bail!( + "Pull request creation {creation_id} is still pending after {} \ + minutes. Check its status with: fabro pr create {run_id}", + PULL_REQUEST_CREATION_POLL_DEADLINE.as_secs() / 60 + ); + } time::sleep(PULL_REQUEST_CREATION_POLL_INTERVAL).await; creation = self.get_run_pull_request_creation(run_id).await?; if creation.id != creation_id { diff --git a/lib/foundation/fabro-types/src/pull_request.rs b/lib/foundation/fabro-types/src/pull_request.rs index cdd922754..4938083b5 100644 --- a/lib/foundation/fabro-types/src/pull_request.rs +++ b/lib/foundation/fabro-types/src/pull_request.rs @@ -28,6 +28,9 @@ pub struct PullRequestCreation { pub force: bool, pub requested_at: DateTime, pub updated_at: DateTime, + /// Copy of the run's pull request link so that polling the creation + /// resource alone is enough to learn the outcome. Always equal to the + /// run's `pull_request` when the status is `Succeeded`. #[serde(default, skip_serializing_if = "Option::is_none")] pub pull_request: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -39,6 +42,19 @@ impl PullRequestCreation { pub fn is_pending(&self) -> bool { self.status == PullRequestCreationStatus::Pending } + + pub fn succeed(&mut self, pull_request: PullRequestLink, ts: DateTime) { + self.status = PullRequestCreationStatus::Succeeded; + self.updated_at = ts; + self.pull_request = Some(pull_request); + self.error = None; + } + + pub fn fail(&mut self, error: String, ts: DateTime) { + self.status = PullRequestCreationStatus::Failed; + self.updated_at = ts; + self.error = Some(error); + } } /// Minimal GitHub pull request reference stored on a workflow run. diff --git a/lib/foundation/fabro-types/src/run_event/misc.rs b/lib/foundation/fabro-types/src/run_event/misc.rs index 844df5dce..cde08f65f 100644 --- a/lib/foundation/fabro-types/src/run_event/misc.rs +++ b/lib/foundation/fabro-types/src/run_event/misc.rs @@ -295,5 +295,9 @@ pub struct PullRequestUnlinkedProps { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct PullRequestFailedProps { - pub error: String, + /// Set when the failure resolves an explicitly requested creation; absent + /// for pull request failures in the workflow publish stage. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub creation_id: Option, + pub error: String, } diff --git a/lib/packages/fabro-api-client/src/api/runs-api.ts b/lib/packages/fabro-api-client/src/api/runs-api.ts index 36e9eb8d9..e3d936afe 100644 --- a/lib/packages/fabro-api-client/src/api/runs-api.ts +++ b/lib/packages/fabro-api-client/src/api/runs-api.ts @@ -411,7 +411,7 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration) }; }, /** - * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. If a creation is already pending for the run, the response returns that creation unchanged; any different `model` or `force` values in the new request are ignored. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest @@ -1688,7 +1688,7 @@ export const RunsApiFp = function(configuration?: Configuration) { return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, /** - * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. If a creation is already pending for the run, the response returns that creation unchanged; any different `model` or `force` values in the new request are ignored. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest @@ -2145,7 +2145,7 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath? return localVarFp.createRun(runManifest, options).then((request) => request(axios, basePath)); }, /** - * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. If a creation is already pending for the run, the response returns that creation unchanged; any different `model` or `force` values in the new request are ignored. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest @@ -2527,7 +2527,7 @@ export class RunsApi extends BaseAPI { } /** - * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. + * Durably requests creation of a pull request for a completed run. The server generates the pull request content and creates the GitHub pull request after this request returns. Poll the URL in the Location response header until the creation succeeds or fails. If a creation is already pending for the run, the response returns that creation unchanged; any different `model` or `force` values in the new request are ignored. * @summary Create Run Pull Request * @param {string} id Unique run identifier (ULID). * @param {CreateRunPullRequestRequest} createRunPullRequestRequest