fix: enforce ticket assignment mutation fences

This commit is contained in:
2026-08-22 23:06:28 +09:00
parent 8030045602
commit 285763f4ae
2 changed files with 176 additions and 21 deletions
+99 -20
View File
@@ -2713,7 +2713,7 @@ struct CurrentWorkerWorkdirAttachmentResponse {
attached: bool, attached: bool,
} }
#[derive(Debug, Deserialize)] #[derive(Debug, Clone, Deserialize)]
struct ScopedRecordPath { struct ScopedRecordPath {
workspace_id: String, workspace_id: String,
id: String, id: String,
@@ -3698,9 +3698,7 @@ struct BrowserAppendTicketEventRequest {
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
struct BrowserQueueTicketRequest { struct BrowserQueueTicketRequest {}
queued_by: Option<String>,
}
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
@@ -3829,6 +3827,15 @@ async fn scoped_transition_ticket_state(
).into()); ).into());
} }
let current = api.authority.ticket(&path.id)?; let current = api.authority.ticket(&path.id)?;
if request.state == TicketWorkflowState::InProgress
&& current.state != TicketWorkflowState::InProgress.as_str()
{
return Err(Error::TicketAssignmentConflict(
"generic Ticket state mutation cannot enter inprogress; use Queue acceptance or atomic ready-state Coder assignment"
.to_string(),
)
.into());
}
let mut change = TicketStateChange::new( let mut change = TicketStateChange::new(
current.state, current.state,
request.state.as_str(), request.state.as_str(),
@@ -3881,22 +3888,20 @@ async fn scoped_mark_ticket_ready_from_browser(
async fn scoped_queue_ticket( async fn scoped_queue_ticket(
State(api): State<WorkspaceApi>, State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRecordPath>, AxumPath(path): AxumPath<ScopedRecordPath>,
Json(request): Json<BrowserQueueTicketRequest>, Json(_request): Json<BrowserQueueTicketRequest>,
) -> ApiResult<Json<TicketDetail>> { ) -> ApiResult<Json<TicketDetail>> {
validate_workspace_scope(&api, &path.workspace_id)?; validate_workspace_scope(&api, &path.workspace_id)?;
let queued_by = request.queued_by.as_deref().unwrap_or("web"); let _ = execute_ticket_rest_operation(
browser_ticket_backend(&api)?
.queue_ready(TicketIdOrSlug::Id(path.id.clone()), queued_by)
.map_err(Error::from)?;
let Json(ticket) = browser_ticket_detail(&api, &path.id)?;
notify_ticket_recipients(
&api, &api,
&path.workspace_id, &path.workspace_id,
&path.id, HeaderMap::new(),
TicketWorkflowState::Ready.as_str(), TicketBackendOperation::QueueReady {
ticket.state.as_str(), id: TicketIdOrSlug::Id(path.id.clone()),
None, queued_by: "workspace-web".to_string(),
); },
)
.await?;
let Json(ticket) = browser_ticket_detail(&api, &path.id)?;
Ok(Json(ticket)) Ok(Json(ticket))
} }
@@ -15949,6 +15954,68 @@ mod tests {
assert_eq!(invalid_source.status(), StatusCode::BAD_REQUEST); assert_eq!(invalid_source.status(), StatusCode::BAD_REQUEST);
} }
#[tokio::test]
async fn generic_browser_state_mutation_cannot_bypass_assignment_start() {
let dir = tempfile::tempdir().unwrap();
let api = test_api(dir.path()).await;
let backend = browser_ticket_backend(&api).unwrap();
let ticket = backend
.create(ticket::NewTicket::new("Generic transition guard"))
.unwrap();
let path = ScopedRecordPath {
workspace_id: TEST_WORKSPACE_ID.to_string(),
id: ticket.id.clone(),
};
let planning = scoped_transition_ticket_state(
State(api.clone()),
AxumPath(path.clone()),
Json(BrowserTransitionTicketStateRequest {
state: TicketWorkflowState::InProgress,
reason: None,
body: None,
author: None,
}),
)
.await
.unwrap_err()
.into_response();
assert_eq!(planning.status(), StatusCode::CONFLICT);
assert_eq!(
backend
.show(ticket.id.clone().into())
.unwrap()
.meta
.workflow_state,
TicketWorkflowState::Planning
);
let mut ready_input = ticket::NewTicket::new("Ready transition guard");
ready_input.workflow_state = Some(TicketWorkflowState::Ready);
let ready = backend.create(ready_input).unwrap();
let ready_result = scoped_transition_ticket_state(
State(api.clone()),
AxumPath(ScopedRecordPath {
workspace_id: TEST_WORKSPACE_ID.to_string(),
id: ready.id.clone(),
}),
Json(BrowserTransitionTicketStateRequest {
state: TicketWorkflowState::InProgress,
reason: None,
body: None,
author: None,
}),
)
.await
.unwrap_err()
.into_response();
assert_eq!(ready_result.status(), StatusCode::CONFLICT);
assert_eq!(
backend.show(ready.id.into()).unwrap().meta.workflow_state,
TicketWorkflowState::Ready
);
}
#[tokio::test] #[tokio::test]
async fn queue_requires_orchestrator_role_and_records_assignment_fence() { async fn queue_requires_orchestrator_role_and_records_assignment_fence() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
@@ -15962,6 +16029,19 @@ mod tests {
let ticket = backend.create(input).unwrap(); let ticket = backend.create(input).unwrap();
let path = (TEST_WORKSPACE_ID.to_string(), ticket.id.clone()); let path = (TEST_WORKSPACE_ID.to_string(), ticket.id.clone());
let legacy_missing = scoped_queue_ticket(
State(api.clone()),
AxumPath(ScopedRecordPath {
workspace_id: TEST_WORKSPACE_ID.to_string(),
id: ticket.id.clone(),
}),
Json(BrowserQueueTicketRequest {}),
)
.await
.unwrap_err()
.into_response();
assert_eq!(legacy_missing.status(), StatusCode::CONFLICT);
let missing = scoped_queue_ticket_record( let missing = scoped_queue_ticket_record(
State(api.clone()), State(api.clone()),
AxumPath(path.clone()), AxumPath(path.clone()),
@@ -16623,18 +16703,17 @@ mod tests {
.await .await
.unwrap(); .unwrap();
assert_eq!(ready.state, "ready"); assert_eq!(ready.state, "ready");
assign_test_orchestrator(&api, &ticket_id);
let Json(queued) = scoped_queue_ticket( let Json(queued) = scoped_queue_ticket(
State(api.clone()), State(api.clone()),
AxumPath(path()), AxumPath(path()),
Json(BrowserQueueTicketRequest { Json(BrowserQueueTicketRequest {}),
queued_by: Some("browser-user".to_string()),
}),
) )
.await .await
.unwrap(); .unwrap();
assert_eq!(queued.state, "queued"); assert_eq!(queued.state, "queued");
assert_eq!(queued.queued_by.as_deref(), Some("browser-user")); assert_eq!(queued.queued_by.as_deref(), Some("workspace-web"));
let Json(closed) = scoped_close_ticket( let Json(closed) = scoped_close_ticket(
State(api), State(api),
AxumPath(path()), AxumPath(path()),
+77 -1
View File
@@ -3648,6 +3648,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}; };
self.with_conn_mut(|conn| { self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
if let Some(worker) = record.principal.worker() {
ensure_worker_assignment_available(&tx, &record.workspace_id, &worker)?;
}
let existing_operation: Option<(String, Option<String>)> = tx let existing_operation: Option<(String, Option<String>)> = tx
.query_row( .query_row(
@@ -3866,6 +3869,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
self.with_conn_mut(|conn| { self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
if let Some(worker) = record.principal.worker() {
ensure_worker_assignment_available(&tx, &record.workspace_id, &worker)?;
}
let existing_operation: Option<(String, Option<String>)> = tx let existing_operation: Option<(String, Option<String>)> = tx
.query_row( .query_row(
"SELECT request_fingerprint, assignment_id FROM ticket_assignment_operations "SELECT request_fingerprint, assignment_id FROM ticket_assignment_operations
@@ -5374,6 +5380,32 @@ fn read_worker_control_grant_by_operation(
.map_err(Error::from) .map_err(Error::from)
} }
fn ensure_worker_assignment_available(
conn: &Connection,
workspace_id: &str,
worker: &RuntimeWorkerRef,
) -> Result<()> {
let removal_blocks_assignment: bool = conn.query_row(
"SELECT EXISTS(
SELECT 1 FROM worker_removal_operations
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND state IN ('executing', 'failed', 'succeeded')
UNION ALL
SELECT 1 FROM worker_tombstones
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
)",
params![workspace_id, worker.runtime_id, worker.worker_id],
|row| row.get(0),
)?;
if removal_blocks_assignment {
return Err(Error::TicketAssignmentConflict(format!(
"Worker {}/{} is being retained or has been removed",
worker.runtime_id, worker.worker_id
)));
}
Ok(())
}
fn read_ticket_role_assignment_record( fn read_ticket_role_assignment_record(
row: &rusqlite::Row<'_>, row: &rusqlite::Row<'_>,
) -> rusqlite::Result<TicketRoleAssignmentRecord> { ) -> rusqlite::Result<TicketRoleAssignmentRecord> {
@@ -10630,7 +10662,51 @@ INSERT INTO worker_registry (
TicketAssignmentRole::Coder, TicketAssignmentRole::Coder,
) )
.unwrap(), .unwrap(),
Some(coder) Some(coder.clone())
);
assert!(
store
.delete_worker_registry("workspace-role", &worker.worker)
.is_err(),
"active role assignment must prevent Worker removal"
);
assert!(
store
.clear_current_ticket_role_assignment(
"workspace-role",
&ticket.meta.id,
TicketAssignmentRole::Coder,
"coder-manual-1",
"event-clear-coder",
"op-clear-coder",
"user",
"2026-09-01T00:03:00Z",
Some("test removal guard"),
)
.unwrap()
);
assert!(
store
.delete_worker_registry("workspace-role", &worker.worker)
.unwrap()
);
let removed_worker_assignment = TicketRoleAssignmentRecord {
assignment_id: "contributor-removed-worker".to_string(),
role: TicketAssignmentRole::Contributor,
assigned_at: "2026-09-01T00:04:00Z".to_string(),
..coder
};
assert!(
store
.set_current_ticket_role_assignment(
&removed_worker_assignment,
None,
"event-removed-worker",
"op-removed-worker",
false,
)
.is_err(),
"removed/tombstoned Worker cannot become a new role principal"
); );
} }