michel/clusterflux-public
1
0
Fork 0
Code Issues Pull requests Projects Releases 21 Packages Wiki Activity Actions

Public dry run dryrun-7143846e2cca

Browse source
This commit is contained in:
Michel Paulissen 2026-07-03 20:19:45 +02:00
parent 13d4d524c4
commit c8526c39ef
6 changed files with 335 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

4
DISASMER_PUBLIC_TREE.json
View file

@ -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/**",

73
crates/disasmer-cli/src/main.rs
View file

@ -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<String>,
#[arg(long)]
task: Option<String>,
#[arg(long)]
yes: bool,
}
@ -1150,12 +1150,11 @@ fn process_cancel_report(args: ProcessCancelArgs) -> Result<Value> {
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<Value> {
"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();

4
crates/disasmer-coordinator/src/lib.rs
View file

@ -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)]

285
crates/disasmer-coordinator/src/service.rs
View file

@ -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<TaskCancellationTarget>,
affected_nodes: Vec<NodeId>,
},
TaskControl {
process: ProcessId,
task: TaskId,
@ -552,7 +571,9 @@ pub struct CoordinatorService {
enrollment_grants: BTreeMap<EnrollmentGrantKey, disasmer_core::EnrollmentGrant>,
task_events: Vec<TaskCompletionEvent>,
task_assignments: BTreeMap<TaskAssignmentKey, VecDeque<TaskAssignment>>,
active_tasks: BTreeSet<TaskControlKey>,
task_cancellations: BTreeSet<TaskControlKey>,
process_cancellations: BTreeSet<ProcessControlKey>,
panel_snapshots: BTreeMap<PanelStopKey, PanelState>,
stopped_panels: BTreeSet<PanelStopKey>,
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);

7
scripts/cli-first-contract-smoke.js
View file

@ -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/],

1
scripts/user-session-token-boundary-smoke.js
View file

@ -54,6 +54,7 @@ for (const variant of [
"StartProcess",
"ReconnectNode",
"CancelTask",
"CancelProcess",
"PollTaskControl",
"TaskCompleted",
]) {