From 9a548d2b5ef2a175f2ed6d6a0e859a32d11127c4 Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 18 Aug 2026 08:05:57 +0900 Subject: [PATCH] fix: require observed merge target completion --- crates/merge-request/src/lib.rs | 30 ++++- crates/merge-request/tests/store.rs | 29 ++++- crates/workspace-server/src/repositories.rs | 87 ++------------- crates/workspace-server/src/server.rs | 118 ++++++++++++++------ 4 files changed, 148 insertions(+), 116 deletions(-) diff --git a/crates/merge-request/src/lib.rs b/crates/merge-request/src/lib.rs index ca8821be..ef2d3a5e 100644 --- a/crates/merge-request/src/lib.rs +++ b/crates/merge-request/src/lib.rs @@ -68,6 +68,15 @@ pub struct MergeRequestAuth { pub assignment_id: String, } +impl MergeRequestAuth { + fn actor(&self) -> WorkerIdentity { + WorkerIdentity { + runtime_id: self.runtime_id.clone(), + worker_id: self.worker_id.clone(), + } + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct ReviewRequestedEvent { pub event_id: String, @@ -641,7 +650,7 @@ impl MergeRequestStore { } pub fn validate_completion(&self, i: &CompleteMergeRequest) -> Result<(), MergeRequestError> { let mr = self.get(&i.auth.workspace_id, &i.ticket_id)?; - self.completion_auth(&i.auth, &i.ticket_id, &mr.repository_id)?; + self.repo(&i.auth, &mr.repository_id)?; if let Some(existing) = mr.thread.iter().find_map(|event| match event { MergeRequestThreadEvent::Merge(value) if value.operation_id == i.operation_id => { Some(value) @@ -649,8 +658,12 @@ impl MergeRequestStore { _ => None, }) { if existing.approval_event_id == i.approval_event_id + && existing.approved_source_ref == i.current_subject_ref && existing.target_ref_before == i.target_ref_before && existing.target_ref_after == i.target_ref_after + && existing.strategy == i.strategy + && existing.resolution == i.resolution + && existing.merged_by == i.auth.actor() { return Ok(()); } @@ -658,6 +671,7 @@ impl MergeRequestStore { "operation fingerprint mismatch".into(), )); } + self.completion_auth(&i.auth, &i.ticket_id, &mr.repository_id)?; if mr.state != MergeRequestState::Open { return Err(MergeRequestError::Conflict( "Merge Request is not open".into(), @@ -711,8 +725,12 @@ impl MergeRequestStore { _ => None, }) { if existing.approval_event_id == i.approval_event_id + && existing.approved_source_ref == i.current_subject_ref && existing.target_ref_before == i.target_ref_before && existing.target_ref_after == i.target_ref_after + && existing.strategy == i.strategy + && existing.resolution == i.resolution + && existing.merged_by == i.auth.actor() { return Ok(existing.clone()); } @@ -767,6 +785,16 @@ impl MergeRequestStore { WHERE workspace_id=?1 AND ticket_id=?2 AND workflow_state='inprogress'", params![mr.workspace_id, i.ticket_id, i.now.to_rfc3339()], )?; + let released_assignment = transaction.execute( + "DELETE FROM ticket_current_worker_assignments + WHERE workspace_id=?1 AND ticket_id=?2 AND assignment_id=?3", + params![mr.workspace_id, i.ticket_id, i.auth.assignment_id], + )?; + if released_assignment != 1 { + return Err(MergeRequestError::Unauthorized( + "completion assignment changed while closing Ticket".into(), + )); + } let issued_grants = { let mut statement = transaction.prepare( "SELECT request_event_id,subject_ref,capability_token diff --git a/crates/merge-request/tests/store.rs b/crates/merge-request/tests/store.rs index 33968cad..9be7ba0a 100644 --- a/crates/merge-request/tests/store.rs +++ b/crates/merge-request/tests/store.rs @@ -93,7 +93,7 @@ fn approve(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewEvent { } #[test] fn selectors_thread_and_completion_have_no_revision_or_commit_api() { - let (_d, s) = fixture(); + let (d, s) = fixture(); open(&s); let review = approve(&s, "opaque-source-ref", "token"); let ready = s @@ -122,6 +122,33 @@ fn selectors_thread_and_completion_have_no_revision_or_commit_api() { let mr = s.get("W", "T").unwrap(); assert_eq!(mr.selector_from.as_deref(), Some("work/t")); assert_eq!(mr.state, MergeRequestState::Merged); + let current_assignment: bool = Connection::open(d.path().join("db")) + .unwrap() + .query_row( + "SELECT EXISTS( + SELECT 1 FROM ticket_current_worker_assignments + WHERE workspace_id='W' AND ticket_id='T' + )", + [], + |row| row.get(0), + ) + .unwrap(); + assert!(!current_assignment); + let replayed = s + .complete(CompleteMergeRequest { + ticket_id: "T".into(), + operation_id: "op".into(), + approval_event_id: merged.approval_event_id.clone(), + current_subject_ref: merged.approved_source_ref.clone(), + target_ref_before: merged.target_ref_before.clone(), + target_ref_after: merged.target_ref_after.clone(), + strategy: merged.strategy, + resolution: merged.resolution, + auth: auth(), + now: at(6), + }) + .unwrap(); + assert_eq!(replayed, merged); let json = serde_json::to_string(&mr).unwrap(); for banned in [ "revision_id", diff --git a/crates/workspace-server/src/repositories.rs b/crates/workspace-server/src/repositories.rs index 908bc7a3..21b15634 100644 --- a/crates/workspace-server/src/repositories.rs +++ b/crates/workspace-server/src/repositories.rs @@ -97,38 +97,13 @@ pub struct CommitObservation { #[derive(Debug, Clone, PartialEq, Eq)] pub enum RepositoryLookupError { - UnknownRepository { - id: RepositoryId, - }, - UnsupportedProvider { - id: RepositoryId, - provider: String, - }, - MissingDefaultSelector { - id: RepositoryId, - }, - InvalidSelector { - id: RepositoryId, - selector: String, - }, - CommitNotFound { - id: RepositoryId, - commit: String, - }, - InvalidCommitRelation { - id: RepositoryId, - detail: String, - }, - TargetMoved { - id: RepositoryId, - selector: String, - expected: String, - observed: Option, - }, - ProviderFailure { - id: RepositoryId, - operation: String, - }, + UnknownRepository { id: RepositoryId }, + UnsupportedProvider { id: RepositoryId, provider: String }, + MissingDefaultSelector { id: RepositoryId }, + InvalidSelector { id: RepositoryId, selector: String }, + CommitNotFound { id: RepositoryId, commit: String }, + InvalidCommitRelation { id: RepositoryId, detail: String }, + ProviderFailure { id: RepositoryId, operation: String }, } #[derive(Debug, Clone)] @@ -306,45 +281,6 @@ impl RepositoryRegistryReader { } } - pub fn update_merge_target( - &self, - id: &str, - selector: &str, - expected_target: &str, - result_commit: &str, - ) -> Result<(), RepositoryLookupError> { - let repository = self.merge_repository(id)?; - let target_ref = normalize_target_branch_selector(id, selector)?; - self.observe_commit(id, result_commit)?; - let status = Command::new("git") - .arg("-C") - .arg(&repository.path) - .args([ - "update-ref", - target_ref.as_str(), - result_commit, - expected_target, - ]) - .status() - .map_err(|_| RepositoryLookupError::ProviderFailure { - id: id.into(), - operation: "guarded target update".into(), - })?; - if status.success() { - return Ok(()); - } - let observed = self - .observe_merge_target(id, Some(selector)) - .ok() - .map(|target| target.commit); - Err(RepositoryLookupError::TargetMoved { - id: id.into(), - selector: selector.into(), - expected: expected_target.into(), - observed, - }) - } - fn merge_repository(&self, id: &str) -> Result<&ConfiguredRepository, RepositoryLookupError> { let repository = self .find(id) @@ -755,20 +691,13 @@ mod tests { vec![base.clone()] ); reader.ensure_ancestor("main", &base, &source).unwrap(); - reader - .update_merge_target("main", "main", &base, &source) - .unwrap(); assert_eq!( reader .observe_merge_target("main", Some("refs/heads/main")) .unwrap() .commit, - source + base ); - assert!(matches!( - reader.update_merge_target("main", "refs/heads/main", &base, &base), - Err(RepositoryLookupError::TargetMoved { .. }) - )); assert!(matches!( reader.ensure_ancestor("main", &source, &base), Err(RepositoryLookupError::InvalidCommitRelation { .. }) diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index a843004d..b12b3c14 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -3893,6 +3893,23 @@ fn repository_merge_evidence_error(error: RepositoryLookupError) -> ApiError { .into() } +fn require_completed_target_observation( + observed: &str, + target_ref_before: &str, + target_ref_after: &str, +) -> ApiResult<()> { + if observed == target_ref_after { + return Ok(()); + } + if observed == target_ref_before { + return Err(Error::InvalidInput( + "target selector is still at target_ref_before; push the verified result from the Orchestrator Workdir before MergeRequestComplete".into(), + ) + .into()); + } + Err(Error::InvalidInput("target selector moved outside completion evidence".into()).into()) +} + async fn scoped_show_merge_request( State(api): State, AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>, @@ -4226,19 +4243,55 @@ async fn scoped_complete_merge_request( require_workspace_access(&workspace_id, &api)?; let source = authenticate_worker_mutation_source(&api, &workspace_id, &headers)?; require_online_workspace_orchestrator_source(&api, &source)?; + let store = merge_request_store(&api, &workspace_id)?; + let mr = store.get(&workspace_id, &ticket_id)?; + let repositories = api.repository_reader(); + if let Some(existing) = mr.thread.iter().find_map(|event| match event { + merge_request::MergeRequestThreadEvent::Merge(event) + if event.operation_id == input.operation_id => + { + Some(event) + } + _ => None, + }) { + let observed = repositories + .observe_merge_target(&mr.repository_id, Some(&mr.selector_to)) + .map_err(repository_merge_evidence_error)?; + require_completed_target_observation( + &observed.commit, + &input.target_ref_before, + &input.target_ref_after, + )?; + let replay = merge_request::CompleteMergeRequest { + ticket_id, + operation_id: input.operation_id, + approval_event_id: input.approval_event_id, + current_subject_ref: existing.approved_source_ref.clone(), + target_ref_before: input.target_ref_before, + target_ref_after: input.target_ref_after, + strategy: input.strategy, + resolution: input.resolution, + auth: merge_request::MergeRequestAuth { + workspace_id, + repository_id: mr.repository_id.clone(), + runtime_id: source.runtime_id, + worker_id: source.worker_id, + assignment_id: String::new(), + }, + now: Utc::now(), + }; + return store.complete(replay).map(Json).map_err(Into::into); + } let assignment = api .store .get_current_ticket_worker_assignment(&workspace_id, &ticket_id)? .ok_or_else(|| { Error::TicketAssignmentConflict("Ticket has no current assigned Coder".into()) })?; - let store = merge_request_store(&api, &workspace_id)?; - let mr = store.get(&workspace_id, &ticket_id)?; let selector = mr .selector_from .as_deref() .ok_or_else(|| Error::InvalidInput("selector_from requires repair".into()))?; - let repositories = api.repository_reader(); let current_source_ref = repositories .observe_merge_target(&mr.repository_id, Some(selector)) .map_err(repository_merge_evidence_error)? @@ -4246,12 +4299,11 @@ async fn scoped_complete_merge_request( let observed = repositories .observe_merge_target(&mr.repository_id, Some(&mr.selector_to)) .map_err(repository_merge_evidence_error)?; - if observed.commit != input.target_ref_before && observed.commit != input.target_ref_after { - return Err(Error::InvalidInput( - "target selector moved outside completion evidence".into(), - ) - .into()); - } + require_completed_target_observation( + &observed.commit, + &input.target_ref_before, + &input.target_ref_after, + )?; let completion = merge_request::CompleteMergeRequest { ticket_id, operation_id: input.operation_id, @@ -4271,31 +4323,7 @@ async fn scoped_complete_merge_request( now: Utc::now(), }; store.validate_completion(&completion)?; - let already = observed.commit == input.target_ref_after; - if !already { - repositories - .update_merge_target( - &mr.repository_id, - &mr.selector_to, - &input.target_ref_before, - &input.target_ref_after, - ) - .map_err(repository_merge_evidence_error)? - } - match store.complete(completion) { - Ok(v) => Ok(Json(v)), - Err(e) => { - if !already { - let _ = repositories.update_merge_target( - &mr.repository_id, - &mr.selector_to, - &input.target_ref_after, - &input.target_ref_before, - ); - } - Err(e.into()) - } - } + store.complete(completion).map(Json).map_err(Into::into) } fn reject_non_browser_reopen_auth(headers: &HeaderMap) -> Result<()> { @@ -12443,7 +12471,7 @@ mod tests { assert_eq!(builtin.definition.name, "coder-review"); assert_eq!(builtin.selector.to_string(), "builtin:coder-review"); assert_eq!(builtin.flow_id, "builtin:coder-review"); - assert_eq!(builtin.revision, 2); + assert_eq!(builtin.revision, 3); assert_eq!( api.store .list_flow_sources(&api.config.workspace_id) @@ -12604,6 +12632,26 @@ mod tests { ); } + #[test] + fn merge_request_completion_records_only_an_observed_remote_target_update() { + require_completed_target_observation("after", "before", "after").unwrap(); + + let not_pushed = + require_completed_target_observation("before", "before", "after").unwrap_err(); + assert!(matches!( + not_pushed.error, + Error::InvalidInput(ref message) + if message.contains("push the verified result from the Orchestrator Workdir") + )); + + let moved = require_completed_target_observation("other", "before", "after").unwrap_err(); + assert!(matches!( + moved.error, + Error::InvalidInput(ref message) + if message.contains("moved outside completion evidence") + )); + } + #[test] fn worker_ticket_assignment_projects_coder_intent_and_run_acceptance() { let initial_submit = vec![