fix(ticket): preserve uncertain launch claims on local replay rejection
This commit is contained in:
@@ -4717,13 +4717,23 @@ impl WorkspaceApi {
|
|||||||
&self,
|
&self,
|
||||||
runtime_id: &str,
|
runtime_id: &str,
|
||||||
request: WorkerSpawnRequest,
|
request: WorkerSpawnRequest,
|
||||||
|
) -> ApiResult<WorkerSpawnResult> {
|
||||||
|
self.spawn_workspace_worker_with_claim_admission(runtime_id, request, false)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn spawn_workspace_worker_with_claim_admission(
|
||||||
|
&self,
|
||||||
|
runtime_id: &str,
|
||||||
|
request: WorkerSpawnRequest,
|
||||||
|
fresh_ticket_claim: bool,
|
||||||
) -> ApiResult<WorkerSpawnResult> {
|
) -> ApiResult<WorkerSpawnResult> {
|
||||||
let assignment = request.ticket_assignment.clone();
|
let assignment = request.ticket_assignment.clone();
|
||||||
let mut dispatched = false;
|
let mut dispatched = false;
|
||||||
let mut result = self.spawn_workspace_worker_inner(runtime_id, request, &mut dispatched);
|
let mut result = self.spawn_workspace_worker_inner(runtime_id, request, &mut dispatched);
|
||||||
// A definite pre-dispatch rejection must not strand a Ticket claim. Once
|
// Only a newly inserted durable receipt proves there was no earlier
|
||||||
// dispatch was attempted, only confirmed compensation may settle it.
|
// dispatch. A local rejection on replay says nothing about a prior
|
||||||
if !dispatched && result.is_err() {
|
// unknown outcome; keep its fence until absence is confirmed.
|
||||||
|
if fresh_ticket_claim && !dispatched && result.is_err() {
|
||||||
if let Some(assignment) = assignment.as_ref() {
|
if let Some(assignment) = assignment.as_ref() {
|
||||||
if let Err(error) = self.store.fail_ticket_assignment_operation(
|
if let Err(error) = self.store.fail_ticket_assignment_operation(
|
||||||
&self.config.workspace_id,
|
&self.config.workspace_id,
|
||||||
@@ -27597,10 +27607,11 @@ async fn create_workspace_worker_inner(
|
|||||||
let assignment_fingerprint = crate::hosts::worker_spawn_idempotency(&request)
|
let assignment_fingerprint = crate::hosts::worker_spawn_idempotency(&request)
|
||||||
.map_err(Error::Config)?
|
.map_err(Error::Config)?
|
||||||
.map(|(_, fingerprint)| fingerprint);
|
.map(|(_, fingerprint)| fingerprint);
|
||||||
|
let mut fresh_ticket_claim = false;
|
||||||
if let (Some(assignment), Some(fingerprint)) =
|
if let (Some(assignment), Some(fingerprint)) =
|
||||||
(assignment.as_ref(), assignment_fingerprint.as_deref())
|
(assignment.as_ref(), assignment_fingerprint.as_deref())
|
||||||
{
|
{
|
||||||
api.store.reserve_ticket_assignment_operation(
|
fresh_ticket_claim = api.store.reserve_ticket_assignment_operation(
|
||||||
&api.config.workspace_id,
|
&api.config.workspace_id,
|
||||||
&assignment.operation_id,
|
&assignment.operation_id,
|
||||||
&assignment.ticket_id,
|
&assignment.ticket_id,
|
||||||
@@ -27635,7 +27646,11 @@ async fn create_workspace_worker_inner(
|
|||||||
.into());
|
.into());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
let result = match api.spawn_workspace_worker(&runtime_id, request) {
|
let result = match api.spawn_workspace_worker_with_claim_admission(
|
||||||
|
&runtime_id,
|
||||||
|
request,
|
||||||
|
fresh_ticket_claim,
|
||||||
|
) {
|
||||||
Ok(result) => result,
|
Ok(result) => result,
|
||||||
Err(error) => return Err(error.into()),
|
Err(error) => return Err(error.into()),
|
||||||
};
|
};
|
||||||
@@ -28784,10 +28799,11 @@ async fn create_runtime_worker(
|
|||||||
let requested_worker_name = request.requested_worker_name.clone();
|
let requested_worker_name = request.requested_worker_name.clone();
|
||||||
let spawn_idempotency =
|
let spawn_idempotency =
|
||||||
crate::hosts::worker_spawn_idempotency(&request).map_err(Error::Config)?;
|
crate::hosts::worker_spawn_idempotency(&request).map_err(Error::Config)?;
|
||||||
|
let mut fresh_ticket_claim = false;
|
||||||
if let (Some(assignment), Some((_, fingerprint))) =
|
if let (Some(assignment), Some((_, fingerprint))) =
|
||||||
(lifecycle_assignment.as_ref(), spawn_idempotency.as_ref())
|
(lifecycle_assignment.as_ref(), spawn_idempotency.as_ref())
|
||||||
{
|
{
|
||||||
if let Err(error) = api.store.reserve_ticket_assignment_operation(
|
fresh_ticket_claim = match api.store.reserve_ticket_assignment_operation(
|
||||||
&api.config.workspace_id,
|
&api.config.workspace_id,
|
||||||
&assignment.operation_id,
|
&assignment.operation_id,
|
||||||
&assignment.ticket_id,
|
&assignment.ticket_id,
|
||||||
@@ -28796,9 +28812,12 @@ async fn create_runtime_worker(
|
|||||||
fingerprint,
|
fingerprint,
|
||||||
&Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
|
&Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
|
||||||
) {
|
) {
|
||||||
let diagnostics = cleanup_spawn_created_workdirs(&api, &created_workdir_ids);
|
Ok(fresh) => fresh,
|
||||||
return Err(ApiError::with_diagnostics(error, diagnostics));
|
Err(error) => {
|
||||||
}
|
let diagnostics = cleanup_spawn_created_workdirs(&api, &created_workdir_ids);
|
||||||
|
return Err(ApiError::with_diagnostics(error, diagnostics));
|
||||||
|
}
|
||||||
|
};
|
||||||
}
|
}
|
||||||
if let Some(assignment) = lifecycle_assignment.as_ref()
|
if let Some(assignment) = lifecycle_assignment.as_ref()
|
||||||
&& let Some(worker) = existing_lifecycle_assignment_worker(&api, assignment, &runtime_id)?
|
&& let Some(worker) = existing_lifecycle_assignment_worker(&api, assignment, &runtime_id)?
|
||||||
@@ -28811,7 +28830,8 @@ async fn create_runtime_worker(
|
|||||||
diagnostics: Vec::new(),
|
diagnostics: Vec::new(),
|
||||||
})?));
|
})?));
|
||||||
}
|
}
|
||||||
let result = api.spawn_workspace_worker(&runtime_id, request)?;
|
let result =
|
||||||
|
api.spawn_workspace_worker_with_claim_admission(&runtime_id, request, fresh_ticket_claim)?;
|
||||||
if let Some(worker) = result.worker.as_ref() {
|
if let Some(worker) = result.worker.as_ref() {
|
||||||
let compensation = WorkerSpawnCompensationContext {
|
let compensation = WorkerSpawnCompensationContext {
|
||||||
assignment: lifecycle_assignment.as_ref(),
|
assignment: lifecycle_assignment.as_ref(),
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
// T-719 launch regressions. Parent wiring: mod ticket_worker_launch_tests;
|
// T-719 launch regressions. Parent wiring: mod ticket_worker_launch_tests;
|
||||||
// under server::tests. Runtime execution is in-process; no Worker binary is spawned.
|
// under server::tests. Runtime execution is in-process; no Worker binary is spawned.
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use crate::hosts::{InternalRuntimeSummary, WorkspaceWorkerRuntime};
|
||||||
use worker_runtime::execution::{
|
use worker_runtime::execution::{
|
||||||
WorkerExecutionBackend, WorkerExecutionResult, WorkerExecutionSpawnRequest,
|
WorkerExecutionBackend, WorkerExecutionResult, WorkerExecutionSpawnRequest,
|
||||||
WorkerExecutionSpawnResult, WorkspaceConfigFetchRequest, WorkspaceConfigFetchResult,
|
WorkerExecutionSpawnResult, WorkspaceConfigFetchRequest, WorkspaceConfigFetchResult,
|
||||||
@@ -1051,3 +1052,185 @@ async fn choosing_reviewer_profile_does_not_register_a_reviewer_child_or_grant_m
|
|||||||
.is_err()
|
.is_err()
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Narrow provider fault seam: creation succeeds and the summary is then lost,
|
||||||
|
// while the delete default cannot confirm compensation. No product process runs.
|
||||||
|
struct LostSpawnResponseRuntime {
|
||||||
|
inner: WorkdirlessFixtureRuntime,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl crate::hosts::WorkspaceWorkerRuntime for LostSpawnResponseRuntime {
|
||||||
|
fn runtime_id(&self) -> &str {
|
||||||
|
self.inner.runtime_id()
|
||||||
|
}
|
||||||
|
fn runtime_summary(&self, limit: usize) -> InternalRuntimeSummary {
|
||||||
|
self.inner.runtime_summary(limit)
|
||||||
|
}
|
||||||
|
fn list_hosts(
|
||||||
|
&self,
|
||||||
|
limit: usize,
|
||||||
|
) -> crate::hosts::RuntimeList<crate::hosts::InternalHostSummary> {
|
||||||
|
self.inner.list_hosts(limit)
|
||||||
|
}
|
||||||
|
fn list_workers(&self, limit: usize) -> crate::hosts::RuntimeList<InternalWorkerSummary> {
|
||||||
|
self.inner.list_workers(limit)
|
||||||
|
}
|
||||||
|
fn worker(&self, worker_id: &str) -> crate::hosts::WorkerLookupResult {
|
||||||
|
self.inner.worker(worker_id)
|
||||||
|
}
|
||||||
|
fn working_directory(&self, id: &str) -> crate::hosts::RuntimeWorkingDirectoryResult {
|
||||||
|
self.inner.working_directory(id)
|
||||||
|
}
|
||||||
|
fn spawn_worker(
|
||||||
|
&self,
|
||||||
|
binding: WorkerCreateBinding,
|
||||||
|
request: WorkerSpawnRequest,
|
||||||
|
) -> WorkerSpawnResult {
|
||||||
|
let result = self.inner.spawn_worker(binding, request);
|
||||||
|
assert_eq!(result.state, InternalWorkerOperationState::Accepted);
|
||||||
|
assert!(result.worker.is_some());
|
||||||
|
WorkerSpawnResult {
|
||||||
|
state: InternalWorkerOperationState::Rejected,
|
||||||
|
worker: None,
|
||||||
|
acceptance_evidence: Vec::new(),
|
||||||
|
diagnostics: vec![RuntimeDiagnostic::new(
|
||||||
|
"spawn_response_lost",
|
||||||
|
"error",
|
||||||
|
"Created Worker but lost the response",
|
||||||
|
)],
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn uncertain_launch_replay_after_target_edit_keeps_claim_and_blocks_competing_start() {
|
||||||
|
let fixture = manual_worker_assignment_fixture().await;
|
||||||
|
for id in [&fixture.main_workdir_id, &fixture.docs_workdir_id] {
|
||||||
|
fixture
|
||||||
|
.api
|
||||||
|
.store
|
||||||
|
.detach_worker_workdir(
|
||||||
|
TEST_WORKSPACE_ID,
|
||||||
|
&fixture.worker,
|
||||||
|
Some(id),
|
||||||
|
TEST_CREATED_AT,
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
}
|
||||||
|
sync_runtime_worker_workdir_attachments(&fixture.api, &fixture.worker).unwrap();
|
||||||
|
fixture
|
||||||
|
.api
|
||||||
|
.runtime
|
||||||
|
.register_or_replace(LostSpawnResponseRuntime {
|
||||||
|
inner: fixture.runtime.clone(),
|
||||||
|
});
|
||||||
|
let initial_spawns = fixture.runtime.spawn_requests().len();
|
||||||
|
let payload = json!({
|
||||||
|
"runtime_id": fixture.worker.runtime_id,
|
||||||
|
"display_name": "Uncertain Ticket Worker", "profile": "builtin:companion",
|
||||||
|
"ticket_assignment": {"ticket_id": fixture.ticket_id, "operation_id": "lost-spawn"},
|
||||||
|
"initial_submit": [{"kind":"text", "content":"Research the Ticket"}],
|
||||||
|
"workdir_attachments": [
|
||||||
|
{"alias":"checkout", "working_directory_id":fixture.main_workdir_id},
|
||||||
|
{"alias":"docs", "working_directory_id":fixture.docs_workdir_id}
|
||||||
|
], "feature_connections": {}
|
||||||
|
});
|
||||||
|
let launch = |body: Value| {
|
||||||
|
create_workspace_worker(
|
||||||
|
State(fixture.api.clone()),
|
||||||
|
HeaderMap::new(),
|
||||||
|
Json(serde_json::from_value(body).unwrap()),
|
||||||
|
)
|
||||||
|
};
|
||||||
|
let lost = launch(payload.clone()).await.unwrap_err();
|
||||||
|
assert!(
|
||||||
|
lost.diagnostics
|
||||||
|
.iter()
|
||||||
|
.any(|d| d.code == "worker_spawn_compensation_create_reservation_retained"),
|
||||||
|
"{lost:?}"
|
||||||
|
);
|
||||||
|
let receipt = fixture
|
||||||
|
.api
|
||||||
|
.store
|
||||||
|
.get_ticket_assignment_operation(TEST_WORKSPACE_ID, "lost-spawn")
|
||||||
|
.unwrap()
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(receipt.claim_state, "pending");
|
||||||
|
assert!(receipt.assignment_id.is_none());
|
||||||
|
let original_worker = receipt.worker.clone().unwrap();
|
||||||
|
assert!(
|
||||||
|
fixture
|
||||||
|
.runtime
|
||||||
|
.worker(&original_worker.worker_id)
|
||||||
|
.worker
|
||||||
|
.is_some()
|
||||||
|
);
|
||||||
|
assert_eq!(fixture.runtime.spawn_requests().len(), initial_spawns + 1);
|
||||||
|
|
||||||
|
browser_ticket_backend(&fixture.api)
|
||||||
|
.unwrap()
|
||||||
|
.edit_item(
|
||||||
|
fixture.ticket_id.clone().into(),
|
||||||
|
TicketItemEdit {
|
||||||
|
targets: Some(TicketTargetsEdit::Set {
|
||||||
|
targets: vec![
|
||||||
|
test_ticket_target(
|
||||||
|
"test-repository",
|
||||||
|
"changed-selector",
|
||||||
|
TicketTargetAccess::ReadWrite,
|
||||||
|
),
|
||||||
|
test_ticket_target("docs", "develop", TicketTargetAccess::ReadOnly),
|
||||||
|
],
|
||||||
|
}),
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
let rejected = launch(payload.clone()).await.unwrap_err();
|
||||||
|
assert!(
|
||||||
|
rejected.error.to_string().contains("selector"),
|
||||||
|
"{rejected:?}"
|
||||||
|
);
|
||||||
|
let retained = fixture
|
||||||
|
.api
|
||||||
|
.store
|
||||||
|
.get_ticket_assignment_operation(TEST_WORKSPACE_ID, "lost-spawn")
|
||||||
|
.unwrap()
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(retained.claim_state, "pending");
|
||||||
|
assert_eq!(retained.worker, Some(original_worker.clone()));
|
||||||
|
assert_eq!(retained.request_fingerprint, receipt.request_fingerprint);
|
||||||
|
assert!(retained.failure_reason.is_none());
|
||||||
|
let mut competing = payload;
|
||||||
|
competing["ticket_assignment"]["operation_id"] = json!("competing-spawn");
|
||||||
|
let conflict = launch(competing).await.unwrap_err();
|
||||||
|
assert!(
|
||||||
|
conflict.error.to_string().contains("pending"),
|
||||||
|
"{conflict:?}"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
fixture
|
||||||
|
.api
|
||||||
|
.store
|
||||||
|
.get_ticket_assignment_operation(TEST_WORKSPACE_ID, "competing-spawn")
|
||||||
|
.unwrap()
|
||||||
|
.is_none()
|
||||||
|
);
|
||||||
|
assert_eq!(fixture.runtime.spawn_requests().len(), initial_spawns + 1);
|
||||||
|
assert!(
|
||||||
|
fixture
|
||||||
|
.runtime
|
||||||
|
.worker(&original_worker.worker_id)
|
||||||
|
.worker
|
||||||
|
.is_some()
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
fixture
|
||||||
|
.api
|
||||||
|
.authority
|
||||||
|
.ticket(&fixture.ticket_id)
|
||||||
|
.unwrap()
|
||||||
|
.state,
|
||||||
|
"ready"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user