From c8526c39ef4f83be5c8ab8a190041e1d75e4067c Mon Sep 17 00:00:00 2001 From: Michel Paulissen <862400+MichelPaulissen@users.noreply.github.com> Date: Fri, 3 Jul 2026 20:19:45 +0200 Subject: [PATCH] Public dry run dryrun-7143846e2cca --- DISASMER_PUBLIC_TREE.json | 4 +- crates/disasmer-cli/src/main.rs | 73 +++-- crates/disasmer-coordinator/src/lib.rs | 4 +- crates/disasmer-coordinator/src/service.rs | 285 ++++++++++++++++++- scripts/cli-first-contract-smoke.js | 7 + scripts/user-session-token-boundary-smoke.js | 1 + 6 files changed, 335 insertions(+), 39 deletions(-) diff --git a/DISASMER_PUBLIC_TREE.json b/DISASMER_PUBLIC_TREE.json index 3ce2de1..841a9f8 100644 --- a/DISASMER_PUBLIC_TREE.json +++ b/DISASMER_PUBLIC_TREE.json @@ -1,7 +1,7 @@ { "kind": "disasmer-filtered-public-tree", - "source_commit": "dece97efaef95e21b322410d4ae5dd8ad7824df6", - "release_name": "dryrun-dece97efaef9", + "source_commit": "7143846e2cca94a1e37c12d8dc9b2b3344f1f692", + "release_name": "dryrun-7143846e2cca", "filtered_out": [ "private/**", "experiments/**", diff --git a/crates/disasmer-cli/src/main.rs b/crates/disasmer-cli/src/main.rs index 71eab03..4199690 100644 --- a/crates/disasmer-cli/src/main.rs +++ b/crates/disasmer-cli/src/main.rs @@ -357,10 +357,10 @@ struct ProcessCancelArgs { scope: CliScopeArgs, #[arg(long, default_value = "vp-current")] process: String, - #[arg(long, default_value = "node-local")] - node: String, - #[arg(long, default_value = "compile-linux")] - task: String, + #[arg(long)] + node: Option, + #[arg(long)] + task: Option, #[arg(long)] yes: bool, } @@ -1150,12 +1150,11 @@ fn process_cancel_report(args: ProcessCancelArgs) -> Result { if let Some(coordinator) = &args.scope.coordinator { let mut session = JsonLineSession::connect(coordinator)?; let response = session.request(json!({ - "type": "cancel_task", + "type": "cancel_process", "tenant": args.scope.tenant, "project": args.scope.project, + "actor_user": args.scope.user, "process": args.process, - "node": args.node, - "task": args.task, }))?; let cancel_request = process_cancel_request_summary(&response, !args.yes); return Ok(json!({ @@ -1179,8 +1178,8 @@ fn process_cancel_report(args: ProcessCancelArgs) -> Result { "task": args.task, "cancel_request": { "status": "requires_coordinator", - "operation": "task_scoped_cancel", - "whole_process_cancel_available": false, + "operation": "cancel_virtual_process", + "whole_process_cancel_available": true, "explicit_user_action": true, }, })) @@ -2393,30 +2392,37 @@ fn process_restart_request_summary(response: &Value, requires_confirmation: bool } fn process_cancel_request_summary(response: &Value, requires_confirmation: bool) -> Value { - if response.get("type").and_then(Value::as_str) != Some("task_cancellation_requested") { + if response.get("type").and_then(Value::as_str) != Some("process_cancellation_requested") { return json!({ "status": response.get("type").and_then(Value::as_str).unwrap_or("coordinator_response"), - "operation": "task_scoped_cancel", + "operation": "cancel_virtual_process", "accepted": false, "requires_confirmation": requires_confirmation, "explicit_user_action": true, - "whole_process_cancel_available": false, + "whole_process_cancel_available": true, "error": response.get("message").cloned().unwrap_or(Value::Null), }); } + let cancelled_tasks = response + .get("cancelled_tasks") + .and_then(Value::as_array) + .map(Vec::len) + .unwrap_or(0); json!({ - "status": "task_cancellation_requested", - "operation": "task_scoped_cancel", + "status": "process_cancellation_requested", + "operation": "cancel_virtual_process", "accepted": true, "process": response.get("process").cloned().unwrap_or(Value::Null), - "task": response.get("task").cloned().unwrap_or(Value::Null), - "node": response.get("node").cloned().unwrap_or(Value::Null), + "cancelled_task_count": cancelled_tasks, + "cancelled_tasks": response.get("cancelled_tasks").cloned().unwrap_or_else(|| json!([])), + "affected_nodes": response.get("affected_nodes").cloned().unwrap_or_else(|| json!([])), "requires_confirmation": requires_confirmation, "explicit_user_action": true, "website_required": false, - "whole_process_cancel_available": false, + "whole_process_cancel_available": true, "node_must_poll_task_control": true, + "new_task_launches_blocked": true, "surviving_state_visibility": "task and artifact state remains visible after terminal task events are reported", }) } @@ -4505,8 +4511,8 @@ mod tests { let addr = listener.local_addr().unwrap().to_string(); let server = std::thread::spawn(move || { let restart_response = r#"{"type":"process_started","process":"vp","epoch":42}"#; - let cancel_response = r#"{"type":"task_cancellation_requested","process":"vp","task":"compile-linux","node":"node-a"}"#; - for expected in ["start_process", "cancel_task"] { + let cancel_response = r#"{"type":"process_cancellation_requested","process":"vp","cancelled_tasks":[{"process":"vp","task":"compile-linux","node":"node-a"},{"process":"vp","task":"link-linux","node":"node-b"}],"affected_nodes":["node-a","node-b"]}"#; + for expected in ["start_process", "cancel_process"] { let (mut stream, _) = listener.accept().unwrap(); let mut reader = BufReader::new(stream.try_clone().unwrap()); let mut line = String::new(); @@ -4518,8 +4524,7 @@ mod tests { if expected == "start_process" { stream.write_all(restart_response.as_bytes()).unwrap(); } else { - assert!(line.contains(r#""node":"node-a""#)); - assert!(line.contains(r#""task":"compile-linux""#)); + assert!(line.contains(r#""actor_user":"user""#)); stream.write_all(cancel_response.as_bytes()).unwrap(); } stream.write_all(b"\n").unwrap(); @@ -4542,8 +4547,8 @@ mod tests { let cancel = process_cancel_report(ProcessCancelArgs { scope, process: "vp".to_owned(), - node: "node-a".to_owned(), - task: "compile-linux".to_owned(), + node: None, + task: None, yes: false, }) .unwrap(); @@ -4562,23 +4567,31 @@ mod tests { assert_eq!( cancel["cancel_request"]["status"], - "task_cancellation_requested" + "process_cancellation_requested" + ); + assert_eq!( + cancel["cancel_request"]["operation"], + "cancel_virtual_process" ); - assert_eq!(cancel["cancel_request"]["operation"], "task_scoped_cancel"); assert_eq!(cancel["cancel_request"]["accepted"], true); assert_eq!(cancel["cancel_request"]["process"], "vp"); - assert_eq!(cancel["cancel_request"]["task"], "compile-linux"); - assert_eq!(cancel["cancel_request"]["node"], "node-a"); + assert_eq!(cancel["cancel_request"]["cancelled_task_count"], 2); + assert_eq!( + cancel["cancel_request"]["cancelled_tasks"][0]["task"], + "compile-linux" + ); + assert_eq!(cancel["cancel_request"]["affected_nodes"][1], "node-b"); assert_eq!(cancel["cancel_request"]["requires_confirmation"], true); assert_eq!(cancel["cancel_request"]["website_required"], false); assert_eq!( cancel["cancel_request"]["whole_process_cancel_available"], - false + true ); assert_eq!( cancel["cancel_request"]["node_must_poll_task_control"], true ); + assert_eq!(cancel["cancel_request"]["new_task_launches_blocked"], true); } #[test] @@ -4628,8 +4641,8 @@ mod tests { let cancel = process_cancel_report(ProcessCancelArgs { scope, process: "vp".to_owned(), - node: "node".to_owned(), - task: "compile-linux".to_owned(), + node: None, + task: None, yes: false, }) .unwrap(); diff --git a/crates/disasmer-coordinator/src/lib.rs b/crates/disasmer-coordinator/src/lib.rs index b78118c..b11993d 100644 --- a/crates/disasmer-coordinator/src/lib.rs +++ b/crates/disasmer-coordinator/src/lib.rs @@ -14,8 +14,8 @@ pub use postgres_store::{ }; pub use service::{ CoordinatorRequest, CoordinatorResponse, CoordinatorService, CoordinatorServiceError, - SourcePreparationDisposition, SourcePreparationStatus, TaskAssignment, TaskCompletionEvent, - TaskTerminalState, + SourcePreparationDisposition, SourcePreparationStatus, TaskAssignment, TaskCancellationTarget, + TaskCompletionEvent, TaskTerminalState, }; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] diff --git a/crates/disasmer-coordinator/src/service.rs b/crates/disasmer-coordinator/src/service.rs index 4385889..9b3125a 100644 --- a/crates/disasmer-coordinator/src/service.rs +++ b/crates/disasmer-coordinator/src/service.rs @@ -164,6 +164,12 @@ pub enum CoordinatorRequest { node: String, task: String, }, + CancelProcess { + tenant: String, + project: String, + actor_user: String, + process: String, + }, PollTaskControl { tenant: String, project: String, @@ -297,6 +303,7 @@ type TaskControlKey = (TenantId, ProjectId, ProcessId, NodeId, TaskId); type TaskAssignmentKey = (TenantId, ProjectId, NodeId); type PanelStopKey = (TenantId, ProjectId, ProcessId); type EnrollmentGrantKey = (TenantId, ProjectId, String); +type ProcessControlKey = (TenantId, ProjectId, ProcessId); fn default_download_ttl_seconds() -> u64 { 900 @@ -360,6 +367,13 @@ pub struct TaskAssignment { pub artifact_path: String, } +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct TaskCancellationTarget { + pub process: ProcessId, + pub task: TaskId, + pub node: NodeId, +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum SourcePreparationDisposition { Pending { reason: String }, @@ -457,6 +471,11 @@ pub enum CoordinatorResponse { task: TaskId, node: NodeId, }, + ProcessCancellationRequested { + process: ProcessId, + cancelled_tasks: Vec, + affected_nodes: Vec, + }, TaskControl { process: ProcessId, task: TaskId, @@ -552,7 +571,9 @@ pub struct CoordinatorService { enrollment_grants: BTreeMap, task_events: Vec, task_assignments: BTreeMap>, + active_tasks: BTreeSet, task_cancellations: BTreeSet, + process_cancellations: BTreeSet, panel_snapshots: BTreeMap, stopped_panels: BTreeSet, panel_event_limits: BTreeMap<(TenantId, ProjectId, ProcessId, String), RateLimit>, @@ -575,7 +596,9 @@ impl CoordinatorService { enrollment_grants: BTreeMap::new(), task_events: Vec::new(), task_assignments: BTreeMap::new(), + active_tasks: BTreeSet::new(), task_cancellations: BTreeSet::new(), + process_cancellations: BTreeSet::new(), panel_snapshots: BTreeMap::new(), stopped_panels: BTreeSet::new(), panel_event_limits: BTreeMap::new(), @@ -947,6 +970,16 @@ impl CoordinatorService { ) .into()); } + if self + .process_cancellations + .contains(&process_control_key(&tenant, &project, &process)) + { + return Err(CoordinatorError::Unauthorized( + "task launch is blocked because the virtual process is cancelling" + .to_owned(), + ) + .into()); + } let request = PlacementRequest { tenant: tenant.clone(), project: project.clone(), @@ -976,6 +1009,13 @@ impl CoordinatorService { command_args, artifact_path, }; + self.active_tasks.insert(task_control_key( + &tenant, + &project, + &process, + &placement.node, + &task, + )); self.task_assignments .entry((tenant, project, placement.node.clone())) .or_default() @@ -1121,12 +1161,34 @@ impl CoordinatorService { project, process, } => { + let tenant = TenantId::new(tenant); + let project = ProjectId::new(project); let process = ProcessId::new(process); - self.coordinator.start_process( - TenantId::new(tenant), - ProjectId::new(project), - process.clone(), + self.process_cancellations + .remove(&process_control_key(&tenant, &project, &process)); + self.task_cancellations.retain( + |(task_tenant, task_project, task_process, _, _)| { + task_tenant != &tenant + || task_project != &project + || task_process != &process + }, ); + self.active_tasks + .retain(|(task_tenant, task_project, task_process, _, _)| { + task_tenant != &tenant + || task_project != &project + || task_process != &process + }); + self.task_assignments.retain(|_, assignments| { + assignments.retain(|assignment| { + assignment.tenant != tenant + || assignment.project != project + || assignment.process != process + }); + !assignments.is_empty() + }); + self.coordinator + .start_process(tenant, project, process.clone()); Ok(CoordinatorResponse::ProcessStarted { process, epoch: self.coordinator.coordinator_epoch(), @@ -1177,6 +1239,57 @@ impl CoordinatorService { node, }) } + CoordinatorRequest::CancelProcess { + tenant, + project, + actor_user, + process, + } => { + let tenant = TenantId::new(tenant); + let project = ProjectId::new(project); + let process = ProcessId::new(process); + let actor = UserId::new(actor_user); + self.coordinator + .upsert_user(tenant.clone(), actor, CredentialKind::BrowserSession); + let active = self.coordinator.active_process(&process).ok_or_else(|| { + CoordinatorError::Unauthorized( + "process cancellation requires an active virtual process".to_owned(), + ) + })?; + if active.tenant != tenant || active.project != project { + return Err(CoordinatorError::Unauthorized( + "process cancellation is outside the virtual process tenant/project scope" + .to_owned(), + ) + .into()); + } + self.process_cancellations + .insert(process_control_key(&tenant, &project, &process)); + let mut cancelled_tasks = Vec::new(); + let mut affected_nodes = BTreeSet::new(); + for (task_tenant, task_project, task_process, node, task) in + self.active_tasks.iter() + { + if task_tenant == &tenant + && task_project == &project + && task_process == &process + { + self.task_cancellations + .insert(task_control_key(&tenant, &project, &process, node, task)); + affected_nodes.insert(node.clone()); + cancelled_tasks.push(TaskCancellationTarget { + process: process.clone(), + task: task.clone(), + node: node.clone(), + }); + } + } + Ok(CoordinatorResponse::ProcessCancellationRequested { + process, + cancelled_tasks, + affected_nodes: affected_nodes.into_iter().collect(), + }) + } CoordinatorRequest::PollTaskControl { tenant, project, @@ -1193,7 +1306,10 @@ impl CoordinatorService { .authorize_node_for_process(&node, &tenant, &project, &process)?; let cancel_requested = self .task_cancellations - .contains(&task_control_key(&tenant, &project, &process, &node, &task)); + .contains(&task_control_key(&tenant, &project, &process, &node, &task)) + || self + .process_cancellations + .contains(&process_control_key(&tenant, &project, &process)); Ok(CoordinatorResponse::TaskControl { process, task, @@ -1370,6 +1486,13 @@ impl CoordinatorService { &event.node, &event.task, )); + self.active_tasks.remove(&task_control_key( + &event.tenant, + &event.project, + &event.process, + &event.node, + &event.task, + )); self.task_events.push(event.clone()); Ok(CoordinatorResponse::TaskRecorded { process: event.process, @@ -2011,6 +2134,14 @@ fn task_control_key( ) } +fn process_control_key( + tenant: &TenantId, + project: &ProjectId, + process: &ProcessId, +) -> ProcessControlKey { + (tenant.clone(), project.clone(), process.clone()) +} + fn panel_stop_key(tenant: &TenantId, project: &ProjectId, process: &ProcessId) -> PanelStopKey { (tenant.clone(), project.clone(), process.clone()) } @@ -2428,6 +2559,150 @@ mod tests { ); } + #[test] + fn service_cancels_whole_process_and_blocks_new_task_launches() { + let mut service = CoordinatorService::new(7); + for node in ["node-a", "node-b"] { + let capabilities = if node == "node-a" { + linux_capabilities() + } else { + windows_capabilities() + }; + service + .handle_request(CoordinatorRequest::AttachNode { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + node: node.to_owned(), + public_key: format!("{node}-public-key"), + }) + .unwrap(); + service + .handle_request(CoordinatorRequest::ReportNodeCapabilities { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + node: node.to_owned(), + capabilities, + cached_environment_digests: vec![], + dependency_cache_digests: vec![], + source_snapshots: vec![], + artifact_locations: vec![], + direct_connectivity: true, + online: true, + }) + .unwrap(); + } + service + .handle_request(CoordinatorRequest::StartProcess { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + process: "process".to_owned(), + }) + .unwrap(); + for node in ["node-a", "node-b"] { + service + .handle_request(CoordinatorRequest::ReconnectNode { + node: node.to_owned(), + process: "process".to_owned(), + epoch: 7, + }) + .unwrap(); + } + + for (task, required_capability, preferred_node) in [ + ("compile-linux", Capability::Containers, "node-a"), + ("link-windows", Capability::WindowsCommandDev, "node-b"), + ] { + let response = service + .handle_request(CoordinatorRequest::LaunchTask { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + actor_user: "user".to_owned(), + process: "process".to_owned(), + task: task.to_owned(), + environment: None, + environment_digest: None, + required_capabilities: vec![required_capability], + dependency_cache: None, + source_snapshot: None, + required_artifacts: vec![], + quota_available: true, + policy_allowed: true, + command: "true".to_owned(), + command_args: vec![], + artifact_path: format!("/vfs/artifacts/{task}.txt"), + }) + .unwrap(); + let CoordinatorResponse::TaskLaunched { assignment, .. } = response else { + panic!("expected task launch"); + }; + assert_eq!(assignment.node, NodeId::from(preferred_node)); + } + + let CoordinatorResponse::ProcessCancellationRequested { + process, + cancelled_tasks, + affected_nodes, + } = service + .handle_request(CoordinatorRequest::CancelProcess { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + actor_user: "user".to_owned(), + process: "process".to_owned(), + }) + .unwrap() + else { + panic!("expected process cancellation"); + }; + assert_eq!(process, ProcessId::from("process")); + assert_eq!(cancelled_tasks.len(), 2); + assert_eq!( + affected_nodes, + vec![NodeId::from("node-a"), NodeId::from("node-b")] + ); + + let control = service + .handle_request(CoordinatorRequest::PollTaskControl { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + process: "process".to_owned(), + node: "node-a".to_owned(), + task: "compile-linux".to_owned(), + }) + .unwrap(); + assert_eq!( + control, + CoordinatorResponse::TaskControl { + process: ProcessId::from("process"), + task: TaskId::from("compile-linux"), + cancel_requested: true, + } + ); + + let blocked = service + .handle_request(CoordinatorRequest::LaunchTask { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + actor_user: "user".to_owned(), + process: "process".to_owned(), + task: "package-linux".to_owned(), + environment: None, + environment_digest: None, + required_capabilities: vec![Capability::Command], + dependency_cache: None, + source_snapshot: None, + required_artifacts: vec![], + quota_available: true, + policy_allowed: true, + command: "true".to_owned(), + command_args: vec![], + artifact_path: "/vfs/artifacts/package-linux.txt".to_owned(), + }) + .unwrap_err(); + assert!(blocked + .to_string() + .contains("virtual process is cancelling")); + } + #[test] fn service_download_links_are_scoped_and_streaming_is_metered() { let mut service = CoordinatorService::new(7); diff --git a/scripts/cli-first-contract-smoke.js b/scripts/cli-first-contract-smoke.js index 5980250..798ca94 100644 --- a/scripts/cli-first-contract-smoke.js +++ b/scripts/cli-first-contract-smoke.js @@ -32,6 +32,7 @@ function expect(source, name, pattern) { const criteria = read("cli_acceptance_criteria.md"); const cli = read("crates/disasmer-cli/src/main.rs"); +const coordinator = read("crates/disasmer-coordinator/src/service.rs"); const outputModeSmoke = read("scripts/cli-output-mode-smoke.js"); expect(criteria, "header", /^# Disasmer CLI-First MVP Acceptance Criteria/m); @@ -119,6 +120,12 @@ for (const [name, pattern] of [ expect(cli, name, pattern); } +expect( + coordinator, + "coordinator whole-process cancellation coverage", + /fn service_cancels_whole_process_and_blocks_new_task_launches\(\)/ +); + for (const [name, pattern] of [ ["agent --json flag", /struct AgentEnrollArgs[\s\S]*#\[arg\(long\)\]\s*json: bool/], ["bundle inspect --json flag", /struct BundleInspectArgs[\s\S]*#\[arg\(long\)\]\s*json: bool/], diff --git a/scripts/user-session-token-boundary-smoke.js b/scripts/user-session-token-boundary-smoke.js index c1e652d..0483677 100755 --- a/scripts/user-session-token-boundary-smoke.js +++ b/scripts/user-session-token-boundary-smoke.js @@ -54,6 +54,7 @@ for (const variant of [ "StartProcess", "ReconnectNode", "CancelTask", + "CancelProcess", "PollTaskControl", "TaskCompleted", ]) {