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

Publish Clusterflux 45bbc21

Browse source
This commit is contained in:
Michel Paulissen 2026-07-20 11:08:26 +02:00
parent eb1077f380
commit 3f88254c70
28 changed files with 896 additions and 167 deletions
Download patch file Download diff file Expand all files Collapse all files

103
crates/clusterflux-cli/src/run.rs
View file

@ -195,12 +195,14 @@ fn coordinator_run_report(plan: RunPlan) -> Result<Value> {
);
}
let process = "vp-current".to_owned();
let launch_attempt = new_launch_attempt_id();
let mut session = JsonLineSession::connect(&coordinator)?;
let mut request = json!({
"type": "start_process",
"tenant": tenant.clone(),
"project": project.clone(),
"process": process.clone(),
"launch_attempt": launch_attempt.clone(),
"restart": false,
});
request = authenticated_human_or_local_trusted_workflow(
@ -219,6 +221,7 @@ fn coordinator_run_report(plan: RunPlan) -> Result<Value> {
&tenant,
&project,
&process,
&launch_attempt,
)?;
let run_start = run_start_summary(&response);
let status = run_start
@ -286,6 +289,7 @@ fn coordinator_run_report(plan: RunPlan) -> Result<Value> {
tenant: &tenant,
project: &project,
process: &process,
launch_attempt: &launch_attempt,
};
let cleanup = rollback_failed_process_launch_reconnecting(
&coordinator,
@ -340,6 +344,7 @@ fn request_process_start_with_rollback(
tenant: &str,
project: &str,
process: &str,
launch_attempt: &str,
) -> Result<Value> {
let rollback = ProcessLaunchRollback {
cli_session,
@ -348,9 +353,25 @@ fn request_process_start_with_rollback(
tenant,
project,
process,
launch_attempt,
};
match session.request_allow_error(request) {
Ok(response) => Ok(response),
Ok(response) => {
if response.get("type").and_then(Value::as_str) == Some("process_started")
&& response.get("launch_attempt").and_then(Value::as_str) != Some(launch_attempt)
{
let ownership_error = anyhow::anyhow!(
"coordinator returned a process-start acknowledgement owned by a different launch attempt"
);
rollback_failed_process_launch_reconnecting(
coordinator,
&rollback,
&json!({ "error": ownership_error.to_string(), "phase": "start_process_ownership" }),
)?;
return Err(ownership_error);
}
Ok(response)
}
Err(start_error) => {
let cleanup = rollback_failed_process_launch_reconnecting(
coordinator,
@ -370,6 +391,17 @@ fn request_process_start_with_rollback(
}
}
static NEXT_LAUNCH_ATTEMPT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
fn new_launch_attempt_id() -> String {
let sequence = NEXT_LAUNCH_ATTEMPT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
format!("launch-{}-{nanos}-{sequence}", std::process::id())
}
struct ProcessLaunchRollback<'a> {
cli_session: &'a CliSession,
fallback_user: &'a str,
@ -377,6 +409,7 @@ struct ProcessLaunchRollback<'a> {
tenant: &'a str,
project: &'a str,
process: &'a str,
launch_attempt: &'a str,
}
fn validate_inline_bundle_size(module_size_bytes: usize) -> Result<()> {
@ -402,6 +435,7 @@ fn rollback_failed_process_launch(
"tenant": context.tenant,
"project": context.project,
"process": context.process,
"launch_attempt": context.launch_attempt,
}),
context.cli_session,
context.fallback_user,
@ -983,6 +1017,10 @@ mod transactional_launch_tests {
payload.get("process").and_then(Value::as_str),
Some("vp-current")
);
assert_eq!(
payload.get("launch_attempt").and_then(Value::as_str),
Some("launch-test")
);
writeln!(
stream,
"{}",
@ -1003,6 +1041,7 @@ mod transactional_launch_tests {
tenant: "tenant-a",
project: "project-a",
process: "vp-current",
launch_attempt: "launch-test",
};
rollback_failed_process_launch(
&mut session,
@ -1024,6 +1063,7 @@ mod transactional_launch_tests {
.read_line(&mut start_line)
.unwrap();
assert!(start_line.contains("\"type\":\"start_process\""));
assert!(start_line.contains("\"launch_attempt\":\"launch-test\""));
drop(stream);
let (mut cleanup_stream, _) = listener.accept().unwrap();
@ -1032,6 +1072,7 @@ mod transactional_launch_tests {
.read_line(&mut cleanup_line)
.unwrap();
assert!(cleanup_line.contains("\"type\":\"abort_process\""));
assert!(cleanup_line.contains("\"launch_attempt\":\"launch-test\""));
writeln!(
cleanup_stream,
"{}",
@ -1043,6 +1084,12 @@ mod transactional_launch_tests {
})
)
.unwrap();
listener.set_nonblocking(true).unwrap();
std::thread::sleep(std::time::Duration::from_millis(25));
assert!(matches!(
listener.accept().unwrap_err().kind(),
std::io::ErrorKind::WouldBlock
));
});
let mut session = JsonLineSession::connect(&address).unwrap();
let error = request_process_start_with_rollback(
@ -1053,6 +1100,7 @@ mod transactional_launch_tests {
"project": "project-a",
"actor_user": "user-a",
"process": "vp-current",
"launch_attempt": "launch-test",
"restart": false
}),
&address,
@ -1062,10 +1110,63 @@ mod transactional_launch_tests {
"tenant-a",
"project-a",
"vp-current",
"launch-test",
)
.unwrap_err()
.to_string();
assert!(error.contains("closed") || error.contains("response"));
server.join().unwrap();
}
#[test]
fn definitive_start_rejection_does_not_abort_existing_process() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut line = String::new();
std::io::BufReader::new(stream.try_clone().unwrap())
.read_line(&mut line)
.unwrap();
writeln!(
stream,
"{}",
json!({
"type": "error",
"message": "project already has active virtual process vp-existing"
})
)
.unwrap();
listener.set_nonblocking(true).unwrap();
std::thread::sleep(std::time::Duration::from_millis(25));
assert_eq!(
listener.accept().unwrap_err().kind(),
std::io::ErrorKind::WouldBlock
);
});
let mut session = JsonLineSession::connect(&address).unwrap();
let response = request_process_start_with_rollback(
&mut session,
json!({
"type": "start_process",
"tenant": "tenant-a",
"project": "project-a",
"actor_user": "user-a",
"process": "vp-current",
"launch_attempt": "launch-rejected",
"restart": false
}),
&address,
&CliSession::Anonymous,
"user-a",
None,
"tenant-a",
"project-a",
"vp-current",
"launch-rejected",
)
.unwrap();
assert_eq!(response.get("type").and_then(Value::as_str), Some("error"));
server.join().unwrap();
}
}

51
crates/clusterflux-cli/src/tests.rs
View file

@ -10,6 +10,15 @@ fn parse(args: &[&str]) -> Cli {
Cli::parse_from(args)
}
fn launch_attempt_from_wire(line: &str) -> String {
let wire: Value = serde_json::from_str(line).unwrap();
wire.pointer("/payload/request/launch_attempt")
.or_else(|| wire.pointer("/payload/launch_attempt"))
.and_then(Value::as_str)
.expect("start request must carry a launch attempt")
.to_owned()
}
fn write_runnable_wasm_project(project: &Path) {
let cli_manifest_dir = Path::new(env!("CARGO_MANIFEST_DIR"));
let sdk = cli_manifest_dir.parent().unwrap().join("clusterflux-sdk");
@ -438,10 +447,11 @@ fn run_with_agent_public_key_sends_attributable_workflow_actor() {
assert!(line.contains(r#""nonce":"agent-signature-"#));
assert!(line.contains(r#""signature":"ed25519:"#));
assert!(!line.contains(r#""actor_user""#));
let launch_attempt = launch_attempt_from_wire(&line);
stream
.write_all(
format!(
r#"{{"type":"process_started","process":"vp-current","epoch":7,"actor":{{"kind":"agent","user":"user-live","agent":"agent-ci","credential_kind":"public_key","public_key_fingerprint":"{expected_fingerprint}","authenticated_without_browser":true,"scopes":["project:read","project:run"]}}}}"#
r#"{{"type":"process_started","process":"vp-current","launch_attempt":"{launch_attempt}","epoch":7,"actor":{{"kind":"agent","user":"user-live","agent":"agent-ci","credential_kind":"public_key","public_key_fingerprint":"{expected_fingerprint}","authenticated_without_browser":true,"scopes":["project:read","project:run"]}}}}"#
)
.as_bytes(),
)
@ -588,7 +598,32 @@ fn run_with_human_session_derives_workflow_scope_from_authenticated_envelope() {
assert!(wire["payload"]["request"].get("tenant").is_none());
assert!(wire["payload"]["request"].get("project").is_none());
assert!(wire["payload"]["request"].get("actor_user").is_none());
stream.write_all(response).unwrap();
if expected_operation == "start_process" {
let launch_attempt = launch_attempt_from_wire(&line);
stream
.write_all(
serde_json::to_string(&json!({
"type": "process_started",
"process": "vp-current",
"launch_attempt": launch_attempt,
"epoch": 7,
"actor": {
"kind": "user",
"user": "user-session",
"agent": null,
"credential_kind": "cli_device_session",
"public_key_fingerprint": null,
"authenticated_without_browser": false,
"scopes": ["project:read", "project:run"]
}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
} else {
stream.write_all(response).unwrap();
}
stream.write_all(b"\n").unwrap();
}
});
@ -661,6 +696,18 @@ fn run_contacts_configured_coordinator_and_reports_active_process_conflicts() {
assert!(line.contains(r#""project":"project-live""#));
assert!(line.contains(r#""process":"vp-current""#));
assert!(line.contains(r#""restart":false"#));
let response = if index == 0 {
let launch_attempt = launch_attempt_from_wire(&line);
serde_json::to_string(&json!({
"type": "process_started",
"process": "vp-current",
"launch_attempt": launch_attempt,
"epoch": 7,
}))
.unwrap()
} else {
response.to_owned()
};
stream.write_all(response.as_bytes()).unwrap();
stream.write_all(b"\n").unwrap();
if index == 0 {

38
crates/clusterflux-coordinator/src/lib.rs
View file

@ -33,6 +33,7 @@ pub use service::{
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ActiveProcess {
pub id: ProcessId,
pub launch_attempt: Option<clusterflux_core::LaunchAttemptId>,
pub tenant: TenantId,
pub project: ProjectId,
pub connected_nodes: BTreeSet<NodeId>,
@ -322,9 +323,20 @@ impl Coordinator {
tenant: TenantId,
project: ProjectId,
id: ProcessId,
) -> ActiveProcess {
self.start_process_for_launch_attempt(tenant, project, id, None)
}
pub fn start_process_for_launch_attempt(
&mut self,
tenant: TenantId,
project: ProjectId,
id: ProcessId,
launch_attempt: Option<clusterflux_core::LaunchAttemptId>,
) -> ActiveProcess {
let process = ActiveProcess {
id: id.clone(),
launch_attempt,
tenant,
project,
connected_nodes: BTreeSet::new(),
@ -531,6 +543,32 @@ impl Coordinator {
.expect("active process was checked immediately before removal"))
}
pub fn abort_process_for_launch_attempt(
&mut self,
tenant: &TenantId,
project: &ProjectId,
process: &ProcessId,
launch_attempt: &clusterflux_core::LaunchAttemptId,
) -> Result<ActiveProcess, CoordinatorError> {
let key = (tenant.clone(), project.clone(), process.clone());
let active = self.active_processes.get(&key).ok_or_else(|| {
CoordinatorError::Unauthorized(
"launch rollback requires an active virtual process".to_owned(),
)
})?;
if active.launch_attempt.as_ref() != Some(launch_attempt) {
return Err(CoordinatorError::Unauthorized(format!(
"launch rollback denied: attempt {} does not own process {}",
launch_attempt.as_str(),
process.as_str()
)));
}
Ok(self
.active_processes
.remove(&key)
.expect("active process was checked immediately before removal"))
}
pub fn active_process_count(&self) -> usize {
self.active_processes.len()
}

38
crates/clusterflux-coordinator/src/service/processes.rs
View file

@ -273,6 +273,7 @@ impl CoordinatorService {
agent_signature: Option<AgentSignedRequest>,
request_payload_digest: Option<&Digest>,
process: String,
launch_attempt: Option<String>,
restart: bool,
) -> Result<CoordinatorResponse, CoordinatorServiceError> {
let tenant = TenantId::new(tenant);
@ -392,10 +393,18 @@ impl CoordinatorService {
self.debug_audit_events.retain(|event| {
event.tenant != tenant || event.project != project || event.process != process
});
self.coordinator
.start_process(tenant, project, process.clone());
let active = self.coordinator.start_process_for_launch_attempt(
tenant,
project,
process.clone(),
launch_attempt.map(clusterflux_core::LaunchAttemptId::new),
);
Ok(CoordinatorResponse::ProcessStarted {
process,
launch_attempt: active
.launch_attempt
.as_ref()
.map(|attempt| attempt.as_str().to_owned()),
epoch: self.coordinator.coordinator_epoch(),
actor,
charged_spawns,
@ -519,6 +528,7 @@ impl CoordinatorService {
project: String,
actor_user: String,
process: String,
launch_attempt: Option<String>,
) -> Result<CoordinatorResponse, CoordinatorServiceError> {
let tenant = TenantId::new(tenant);
let project = ProjectId::new(project);
@ -534,6 +544,17 @@ impl CoordinatorService {
})?;
debug_assert_eq!(active.tenant, tenant);
debug_assert_eq!(active.project, project);
let launch_attempt = launch_attempt.map(clusterflux_core::LaunchAttemptId::new);
if let Some(expected) = launch_attempt.as_ref() {
if active.launch_attempt.as_ref() != Some(expected) {
return Err(CoordinatorError::Unauthorized(format!(
"launch rollback denied: attempt {} does not own process {}",
expected.as_str(),
process.as_str()
))
.into());
}
}
let process_key = process_control_key(&tenant, &project, &process);
self.process_cancellations.remove(&process_key);
@ -575,8 +596,17 @@ impl CoordinatorService {
}
}
self.coordinator
.abort_process(&tenant, &project, &process)?;
if let Some(launch_attempt) = launch_attempt.as_ref() {
self.coordinator.abort_process_for_launch_attempt(
&tenant,
&project,
&process,
launch_attempt,
)?;
} else {
self.coordinator
.abort_process(&tenant, &project, &process)?;
}
let active_restart_tasks = aborted_tasks
.iter()
.map(|target| target.task.clone())

8
crates/clusterflux-coordinator/src/service/protocol.rs
View file

@ -262,6 +262,8 @@ pub enum CoordinatorRequest {
#[serde(default)]
agent_signature: Option<AgentSignedRequest>,
process: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
launch_attempt: Option<String>,
#[serde(default)]
restart: bool,
},
@ -288,6 +290,8 @@ pub enum CoordinatorRequest {
project: String,
actor_user: String,
process: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
launch_attempt: Option<String>,
},
ListProcesses {
tenant: String,
@ -569,6 +573,8 @@ pub enum AuthenticatedCoordinatorRequest {
},
StartProcess {
process: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
launch_attempt: Option<String>,
#[serde(default)]
restart: bool,
},
@ -594,6 +600,8 @@ pub enum AuthenticatedCoordinatorRequest {
},
AbortProcess {
process: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
launch_attempt: Option<String>,
},
ListProcesses,
QuotaStatus,

2
crates/clusterflux-coordinator/src/service/protocol/responses.rs
View file

@ -344,6 +344,8 @@ pub enum CoordinatorResponse {
},
ProcessStarted {
process: ProcessId,
#[serde(default, skip_serializing_if = "Option::is_none")]
launch_attempt: Option<String>,
epoch: u64,
actor: WorkflowActor,
charged_spawns: u64,

39
crates/clusterflux-coordinator/src/service/routing.rs
View file

@ -370,6 +370,7 @@ impl CoordinatorService {
agent_public_key_fingerprint,
agent_signature,
process,
launch_attempt,
restart,
} => self.handle_start_process(
tenant,
@ -380,6 +381,7 @@ impl CoordinatorService {
agent_signature,
Some(&request_payload_digest),
process,
launch_attempt,
restart,
),
CoordinatorRequest::ReconnectNode { .. } => self.reject_unsigned_node_request(),
@ -401,7 +403,8 @@ impl CoordinatorService {
project,
actor_user,
process,
} => self.handle_abort_process(tenant, project, actor_user, process),
launch_attempt,
} => self.handle_abort_process(tenant, project, actor_user, process, launch_attempt),
CoordinatorRequest::ListProcesses {
tenant,
project,
@ -689,18 +692,22 @@ impl CoordinatorService {
actor.as_str().to_owned(),
node,
),
AuthenticatedCoordinatorRequest::StartProcess { process, restart } => self
.handle_start_process(
context.tenant.as_str().to_owned(),
context.project.as_str().to_owned(),
Some(actor.as_str().to_owned()),
None,
None,
None,
None,
process,
restart,
),
AuthenticatedCoordinatorRequest::StartProcess {
process,
launch_attempt,
restart,
} => self.handle_start_process(
context.tenant.as_str().to_owned(),
context.project.as_str().to_owned(),
Some(actor.as_str().to_owned()),
None,
None,
None,
None,
process,
launch_attempt,
restart,
),
request @ AuthenticatedCoordinatorRequest::ScheduleTask { .. } => {
self.handle_authenticated_schedule_task(&context, request)
}
@ -714,11 +721,15 @@ impl CoordinatorService {
actor.as_str().to_owned(),
process,
),
AuthenticatedCoordinatorRequest::AbortProcess { process } => self.handle_abort_process(
AuthenticatedCoordinatorRequest::AbortProcess {
process,
launch_attempt,
} => self.handle_abort_process(
context.tenant.as_str().to_owned(),
context.project.as_str().to_owned(),
actor.as_str().to_owned(),
process,
launch_attempt,
),
AuthenticatedCoordinatorRequest::ListProcesses => self.handle_list_processes(
context.tenant.as_str().to_owned(),

132
crates/clusterflux-coordinator/src/service/tests.rs
View file

@ -96,6 +96,7 @@ fn postgres_backed_runtime_service_survives_restart_without_live_process_state()
.handle_request(CoordinatorRequest::Authenticated {
session_secret: session_secret.to_owned(),
request: AuthenticatedCoordinatorRequest::StartProcess {
launch_attempt: None,
process: "process-ephemeral".to_owned(),
restart: false,
},
@ -433,7 +434,9 @@ fn with_signed_agent_workflow(
let signature = signed_agent_workflow_request(&request, request_kind, process, task, nonce);
match &mut request {
CoordinatorRequest::StartProcess {
agent_signature, ..
launch_attempt: None,
agent_signature,
..
}
| CoordinatorRequest::LaunchTask {
agent_signature, ..
@ -862,10 +865,16 @@ fn authenticated_envelope_derives_user_scope_from_cli_session() {
};
assert_eq!(placement.node, NodeId::from("session-node"));
let CoordinatorResponse::ProcessStarted { process, actor, .. } = service
let CoordinatorResponse::ProcessStarted {
launch_attempt: None,
process,
actor,
..
} = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "cli-session-secret".to_owned(),
request: AuthenticatedCoordinatorRequest::StartProcess {
launch_attempt: None,
process: "vp-session".to_owned(),
restart: false,
},
@ -993,6 +1002,7 @@ fn authenticated_envelope_derives_user_scope_from_cli_session() {
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "victim-cli-session-secret".to_owned(),
request: AuthenticatedCoordinatorRequest::StartProcess {
launch_attempt: None,
process: "vp-victim".to_owned(),
restart: false,
},
@ -1398,6 +1408,7 @@ fn service_reports_and_enforces_public_admin_tenant_suspension() {
let start = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: Some("attempt-a".to_owned()),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -1536,6 +1547,7 @@ fn service_runs_agent_workflows_with_scoped_key_attribution() {
let fingerprint_only = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -1551,6 +1563,7 @@ fn service_runs_agent_workflows_with_scoped_key_attribution() {
let wrong_fingerprint = service
.handle_request(with_signed_agent_workflow(
CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -1592,10 +1605,13 @@ fn service_runs_agent_workflows_with_scoped_key_attribution() {
.unwrap();
let CoordinatorResponse::ProcessStarted {
actor: start_actor, ..
launch_attempt: None,
actor: start_actor,
..
} = service
.handle_request(with_signed_agent_workflow(
CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -1619,6 +1635,7 @@ fn service_runs_agent_workflows_with_scoped_key_attribution() {
let replay = service
.handle_request(with_signed_agent_workflow(
CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -1728,6 +1745,7 @@ fn signed_node_and_agent_requests_reject_body_modification() {
})
.unwrap();
let original_agent_request = CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -1746,6 +1764,7 @@ fn signed_node_and_agent_requests_reject_body_modification() {
);
let mut modified_agent_request = original_agent_request;
let CoordinatorRequest::StartProcess {
launch_attempt: None,
restart,
agent_signature: request_signature,
..
@ -1789,8 +1808,13 @@ fn service_checks_spawn_quota_before_process_or_task_work_starts() {
})
.unwrap();
let CoordinatorResponse::ProcessStarted { charged_spawns, .. } = service
let CoordinatorResponse::ProcessStarted {
launch_attempt: None,
charged_spawns,
..
} = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -1887,6 +1911,7 @@ fn service_checks_spawn_quota_before_process_or_task_work_starts() {
let other_project_process = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "other-project".to_owned(),
actor_user: None,
@ -1899,7 +1924,10 @@ fn service_checks_spawn_quota_before_process_or_task_work_starts() {
.unwrap();
assert!(matches!(
other_project_process,
CoordinatorResponse::ProcessStarted { .. }
CoordinatorResponse::ProcessStarted {
launch_attempt: None,
..
}
));
assert_eq!(
service.quota.used_workflow_spawns(
@ -1935,6 +1963,7 @@ fn project_quota_resets_at_the_configured_window_boundary() {
service.set_server_time(59);
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: Some("attempt-b".to_owned()),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -1947,6 +1976,7 @@ fn project_quota_resets_at_the_configured_window_boundary() {
.unwrap();
service
.handle_request(CoordinatorRequest::AbortProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: "user".to_owned(),
@ -1956,6 +1986,7 @@ fn project_quota_resets_at_the_configured_window_boundary() {
let exhausted = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: Some("attempt-b".to_owned()),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -1971,6 +2002,7 @@ fn project_quota_resets_at_the_configured_window_boundary() {
service.set_server_time(60);
let started = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -1984,6 +2016,7 @@ fn project_quota_resets_at_the_configured_window_boundary() {
assert!(matches!(
started,
CoordinatorResponse::ProcessStarted {
launch_attempt: None,
charged_spawns: 1,
..
}
@ -2073,6 +2106,7 @@ fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -2148,6 +2182,7 @@ fn service_attaches_node_starts_process_and_records_scoped_task_event() {
let started = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -2161,6 +2196,7 @@ fn service_attaches_node_starts_process_and_records_scoped_task_event() {
assert_eq!(
started,
CoordinatorResponse::ProcessStarted {
launch_attempt: None,
process: ProcessId::from("process"),
epoch: 7,
actor: WorkflowActor {
@ -2409,6 +2445,7 @@ fn service_delivers_cancellation_to_connected_node_and_records_terminal_state()
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -2542,6 +2579,7 @@ fn service_authorizes_debug_attach_through_public_api() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -2664,6 +2702,7 @@ fn debug_epoch_commands_are_polled_by_signed_active_task_nodes() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: Some("user".to_owned()),
@ -2991,6 +3030,7 @@ fn service_reports_task_restart_boundary_through_public_api() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -3263,6 +3303,7 @@ fn service_cancels_whole_process_and_blocks_new_task_launches() {
}
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -3382,6 +3423,7 @@ fn service_cancels_whole_process_and_blocks_new_task_launches() {
service
.handle_request(CoordinatorRequest::AbortProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: "user".to_owned(),
@ -3413,6 +3455,7 @@ fn service_rejects_second_active_process_unless_restarting_same_process() {
let mut service = CoordinatorService::new(7);
let started = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -3425,11 +3468,15 @@ fn service_rejects_second_active_process_unless_restarting_same_process() {
.unwrap();
assert!(matches!(
started,
CoordinatorResponse::ProcessStarted { .. }
CoordinatorResponse::ProcessStarted {
launch_attempt: None,
..
}
));
let same_without_restart = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -3446,6 +3493,7 @@ fn service_rejects_second_active_process_unless_restarting_same_process() {
let other_process = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -3460,6 +3508,27 @@ fn service_rejects_second_active_process_unless_restarting_same_process() {
.to_string()
.contains("already has active virtual process"));
let wrong_attempt_abort = service
.handle_request(CoordinatorRequest::AbortProcess {
launch_attempt: Some("attempt-b".to_owned()),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: "user".to_owned(),
process: "process-a".to_owned(),
})
.unwrap_err();
assert!(wrong_attempt_abort
.to_string()
.contains("does not own process process-a"));
assert!(service
.coordinator
.active_process(
&TenantId::from("tenant"),
&ProjectId::from("project"),
&ProcessId::from("process-a")
)
.is_some());
let retired_main_abort = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let process_key = process_control_key(
&TenantId::from("tenant"),
@ -3481,6 +3550,7 @@ fn service_rejects_second_active_process_unless_restarting_same_process() {
);
let restarted = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: Some("attempt-a-restart".to_owned()),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -3494,6 +3564,7 @@ fn service_rejects_second_active_process_unless_restarting_same_process() {
assert_eq!(
restarted,
CoordinatorResponse::ProcessStarted {
launch_attempt: Some("attempt-a-restart".to_owned()),
process: ProcessId::from("process-a"),
epoch: 7,
actor: WorkflowActor {
@ -3517,6 +3588,7 @@ fn quiescent_cooperative_cancel_releases_slot_immediately() {
let mut service = CoordinatorService::new(17);
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -3586,6 +3658,7 @@ fn quiescent_cooperative_cancel_releases_slot_immediately() {
let started = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -3598,7 +3671,10 @@ fn quiescent_cooperative_cancel_releases_slot_immediately() {
.unwrap();
assert!(matches!(
started,
CoordinatorResponse::ProcessStarted { .. }
CoordinatorResponse::ProcessStarted {
launch_attempt: None,
..
}
));
assert!(service.task_events.iter().all(|event| {
event.tenant != TenantId::from("tenant")
@ -3625,6 +3701,7 @@ fn aborted_process_accepts_signed_terminal_event_for_issued_task() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -3655,6 +3732,7 @@ fn aborted_process_accepts_signed_terminal_event_for_issued_task() {
let CoordinatorResponse::ProcessAborted { aborted_tasks, .. } = service
.handle_request(CoordinatorRequest::AbortProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: "user".to_owned(),
@ -3731,6 +3809,7 @@ fn service_download_links_are_scoped_and_streaming_is_metered() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -4127,6 +4206,7 @@ fn download_quota_cannot_be_reset_by_reconnect_retry_or_scope_switch() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -4325,6 +4405,7 @@ fn retained_artifact_reverse_stream_is_chunked_below_control_frame_limit() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -4470,6 +4551,7 @@ fn windows_task_events_share_the_virtual_process_scope() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -4734,8 +4816,13 @@ fn coordinator_side_task_launch_queues_worker_assignment() {
online: true,
})
.unwrap();
let CoordinatorResponse::ProcessStarted { epoch, .. } = service
let CoordinatorResponse::ProcessStarted {
launch_attempt: None,
epoch,
..
} = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -5059,6 +5146,7 @@ fn same_definition_instances_join_correctly_when_they_complete_in_reverse_order(
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: Some("user".to_owned()),
@ -5385,6 +5473,7 @@ fn long_lived_process_state_reaches_a_scoped_bounded_steady_state() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant-soak".to_owned(),
project: "project-soak".to_owned(),
actor_user: Some("user-soak".to_owned()),
@ -5581,6 +5670,7 @@ fn signed_active_wasm_task_can_spawn_and_join_child_in_its_process_only() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: Some("user".to_owned()),
@ -5691,8 +5781,13 @@ fn coordinator_rejects_named_environment_without_requirements() {
online: true,
})
.unwrap();
let CoordinatorResponse::ProcessStarted { epoch, .. } = service
let CoordinatorResponse::ProcessStarted {
launch_attempt: None,
epoch,
..
} = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -5747,6 +5842,7 @@ fn coordinator_side_task_launch_fails_cleanly_without_capable_worker() {
let mut service = CoordinatorService::new(10);
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -5786,8 +5882,13 @@ fn coordinator_side_task_launch_fails_cleanly_without_capable_worker() {
#[test]
fn coordinator_side_task_launch_can_wait_for_capable_worker() {
let mut service = CoordinatorService::new(11);
let CoordinatorResponse::ProcessStarted { epoch, .. } = service
let CoordinatorResponse::ProcessStarted {
launch_attempt: None,
epoch,
..
} = service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
@ -5996,6 +6097,7 @@ fn service_rejects_task_completion_outside_node_scope() {
.unwrap();
service
.handle_request(CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "tenant-b".to_owned(),
project: "project-b".to_owned(),
actor_user: None,
@ -6300,6 +6402,7 @@ fn strict_service_stream_rejects_body_authority_and_accepts_cli_session() {
write_coordinator_wire_request(
&mut stream,
&CoordinatorRequest::StartProcess {
launch_attempt: None,
tenant: "victim-tenant".to_owned(),
project: "victim-project".to_owned(),
actor_user: Some("forged-user".to_owned()),
@ -6326,6 +6429,7 @@ fn strict_service_stream_rejects_body_authority_and_accepts_cli_session() {
&CoordinatorRequest::Authenticated {
session_secret: "strict-stream-session".to_owned(),
request: AuthenticatedCoordinatorRequest::StartProcess {
launch_attempt: None,
process: "vp-authenticated".to_owned(),
restart: false,
},
@ -6335,8 +6439,12 @@ fn strict_service_stream_rejects_body_authority_and_accepts_cli_session() {
line.clear();
reader.read_line(&mut line).unwrap();
let CoordinatorResponse::ProcessStarted { process, actor, .. } =
serde_json::from_str::<CoordinatorResponse>(&line).unwrap()
let CoordinatorResponse::ProcessStarted {
launch_attempt: None,
process,
actor,
..
} = serde_json::from_str::<CoordinatorResponse>(&line).unwrap()
else {
panic!("expected authenticated strict process start");
};

1
crates/clusterflux-core/src/ids.rs
View file

@ -38,6 +38,7 @@ id_type!(AgentId);
id_type!(ArtifactId);
id_type!(NodeId);
id_type!(ProcessId);
id_type!(LaunchAttemptId);
id_type!(ProjectId);
id_type!(TaskDefinitionId);
id_type!(TaskInstanceId);

4
crates/clusterflux-core/src/lib.rs
View file

@ -66,8 +66,8 @@ pub use execution::{
WasmTaskOutcome, WasmTaskResult, MAX_WASM_TASK_ENVELOPE_BYTES, WASM_TASK_ABI_VERSION,
};
pub use ids::{
AgentId, ArtifactId, NodeId, ProcessId, ProjectId, TaskDefinitionId, TaskInstanceId, TenantId,
UserId,
AgentId, ArtifactId, LaunchAttemptId, NodeId, ProcessId, ProjectId, TaskDefinitionId,
TaskInstanceId, TenantId, UserId,
};
pub use limits::{
LargeArgumentPolicy, LimitError, LimitKind, LogBuffer, LogRecord, ResourceLimits,

23
crates/clusterflux-dap/src/adapter.rs
View file

@ -240,6 +240,7 @@ pub(crate) fn run_adapter() -> Result<()> {
}
Err(message) => {
state.breakpoints_installed = false;
emit_unverified_breakpoints(&mut writer, &state, &message)?;
writer.output("stderr", format!("{message}\n"))?;
}
}
@ -1025,6 +1026,28 @@ fn emit_verified_breakpoints(writer: &mut DapWriter, state: &AdapterState) -> Re
Ok(())
}
fn emit_unverified_breakpoints(
writer: &mut DapWriter,
state: &AdapterState,
coordinator_error: &str,
) -> Result<()> {
for (index, line) in state.breakpoints.iter().enumerate() {
writer.event(
"breakpoint",
json!({
"reason": "changed",
"breakpoint": {
"id": index + 1,
"verified": false,
"line": line,
"message": format!("Coordinator breakpoint installation failed: {coordinator_error}"),
}
}),
)?;
}
Ok(())
}
fn emit_thread_lifecycle(
writer: &mut DapWriter,
previous: &BTreeMap<i64, VirtualThread>,

75
crates/clusterflux-dap/src/runtime_client.rs
View file

@ -21,7 +21,7 @@ use debug_protocol::{
};
use local_tools::{child_stderr_suffix, local_tool_command};
pub(crate) use transport::client_user_request;
use transport::{coordinator_request, CoordinatorSession};
use transport::{coordinator_request, coordinator_request_allow_error, CoordinatorSession};
pub(crate) struct LocalRuntimeSession {
coordinator: Option<Child>,
@ -300,7 +300,8 @@ fn launch_services_debug_entrypoint(
) -> Result<RuntimeLaunchRecord> {
let bundle = build_debug_bundle(state, repo)?;
validate_inline_bundle_size(bundle.module_size_bytes)?;
let started = match coordinator_request(
let launch_attempt = new_launch_attempt_id();
let started = match coordinator_request_allow_error(
coordinator,
client_user_request(
state,
@ -310,25 +311,56 @@ fn launch_services_debug_entrypoint(
"project": state.project_id,
"actor_user": state.actor_user,
"process": state.process.as_str(),
"launch_attempt": launch_attempt,
"restart": state.restart_existing,
}),
),
) {
Ok(started) => started,
Err(error) => return Err(debug_launch_error_with_rollback(coordinator, state, error)),
Err(error) => {
return Err(debug_launch_error_with_rollback(
coordinator,
state,
&launch_attempt,
error,
));
}
};
if started.get("type").and_then(Value::as_str) == Some("error") {
return Err(anyhow!(
"{}",
started
.get("message")
.and_then(Value::as_str)
.unwrap_or("coordinator rejected the debug launch")
));
}
if started.get("launch_attempt").and_then(Value::as_str) != Some(launch_attempt.as_str()) {
return Err(debug_launch_error_with_rollback(
coordinator,
state,
&launch_attempt,
anyhow!("coordinator returned a debug launch owned by a different attempt"),
));
}
let epoch = match started.get("epoch").and_then(Value::as_u64) {
Some(epoch) => epoch,
None => {
return Err(debug_launch_error_with_rollback(
coordinator,
state,
&launch_attempt,
anyhow!("coordinator did not report a process epoch"),
));
}
};
if let Err(error) = set_services_debug_breakpoints_at(coordinator, state) {
return Err(debug_launch_error_with_rollback(coordinator, state, error));
return Err(debug_launch_error_with_rollback(
coordinator,
state,
&launch_attempt,
error,
));
}
let launch = match coordinator_request(
coordinator,
@ -368,12 +400,20 @@ fn launch_services_debug_entrypoint(
),
) {
Ok(launch) => launch,
Err(error) => return Err(debug_launch_error_with_rollback(coordinator, state, error)),
Err(error) => {
return Err(debug_launch_error_with_rollback(
coordinator,
state,
&launch_attempt,
error,
));
}
};
if launch.get("type").and_then(Value::as_str) != Some("main_launched") {
return Err(debug_launch_error_with_rollback(
coordinator,
state,
&launch_attempt,
anyhow!(
"coordinator did not start the capless main runtime: {}",
serde_json::to_string(&launch)?
@ -432,6 +472,7 @@ fn validate_inline_bundle_size(module_size_bytes: usize) -> Result<()> {
fn debug_launch_error_with_rollback(
coordinator: &str,
state: &AdapterState,
launch_attempt: &str,
launch_error: anyhow::Error,
) -> anyhow::Error {
let rollback = coordinator_request(
@ -444,6 +485,7 @@ fn debug_launch_error_with_rollback(
"project": state.project_id,
"actor_user": state.actor_user,
"process": state.process.as_str(),
"launch_attempt": launch_attempt,
}),
),
);
@ -460,7 +502,27 @@ fn debug_launch_error_with_rollback(
}
}
static NEXT_LAUNCH_ATTEMPT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
fn new_launch_attempt_id() -> String {
let sequence = NEXT_LAUNCH_ATTEMPT.fetch_add(1, Ordering::Relaxed);
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
format!("dap-launch-{}-{nanos}-{sequence}", std::process::id())
}
fn set_services_debug_breakpoints_at(coordinator: &str, state: &AdapterState) -> Result<()> {
static INJECTED_INSTALL_FAILURE: AtomicBool = AtomicBool::new(false);
if std::env::var_os("CLUSTERFLUX_TEST_DAP_BREAKPOINT_INSTALLATION_FAILURE").is_some()
&& !state.breakpoints.is_empty()
&& !INJECTED_INSTALL_FAILURE.swap(true, Ordering::SeqCst)
{
return Err(anyhow!(
"coordinator breakpoint installation failed: injected live-install failure"
));
}
coordinator_request(
coordinator,
client_user_request(
@ -1267,7 +1329,7 @@ pub(crate) fn observe_services_runtime(
return Ok(());
}
} else {
poll_delay = (poll_delay * 2).min(Duration::from_secs(1));
poll_delay = (poll_delay * 2).min(Duration::from_secs(5));
}
std::thread::sleep(poll_delay);
}
@ -1449,6 +1511,7 @@ mod transactional_launch_tests {
let error = debug_launch_error_with_rollback(
&address,
&state,
"launch-test",
anyhow!("launch acknowledgement was lost"),
);
assert!(error

12
crates/clusterflux-dap/src/runtime_client/transport.rs
View file

@ -33,8 +33,7 @@ impl CoordinatorSession {
}
pub(super) fn request(&mut self, request: Value) -> Result<Value> {
let wire_request = coordinator_wire_request("dap-1", request);
let response = self.session.request(&wire_request)?;
let response = self.request_allow_error(request)?;
if response.get("type").and_then(Value::as_str) == Some("error") {
return Err(anyhow!(
"{}",
@ -46,8 +45,17 @@ impl CoordinatorSession {
}
Ok(response)
}
pub(super) fn request_allow_error(&mut self, request: Value) -> Result<Value> {
let wire_request = coordinator_wire_request("dap-1", request);
Ok(self.session.request(&wire_request)?)
}
}
pub(super) fn coordinator_request(addr: &str, request: Value) -> Result<Value> {
CoordinatorSession::connect(addr)?.request(request)
}
pub(super) fn coordinator_request_allow_error(addr: &str, request: Value) -> Result<Value> {
CoordinatorSession::connect(addr)?.request_allow_error(request)
}