From 3f88254c709be7fc233cf353034b2ab82b8535db Mon Sep 17 00:00:00 2001 From: Michel Paulissen <862400+MichelPaulissen@users.noreply.github.com> Date: Mon, 20 Jul 2026 11:08:26 +0200 Subject: [PATCH] Publish Clusterflux 45bbc21 --- CLUSTERFLUX_PUBLIC_TREE.json | 4 +- README.md | 1 - crates/clusterflux-cli/src/run.rs | 103 ++++++- crates/clusterflux-cli/src/tests.rs | 51 +++- crates/clusterflux-coordinator/src/lib.rs | 38 +++ .../src/service/processes.rs | 38 ++- .../src/service/protocol.rs | 8 + .../src/service/protocol/responses.rs | 2 + .../src/service/routing.rs | 39 ++- .../src/service/tests.rs | 132 ++++++++- crates/clusterflux-core/src/ids.rs | 1 + crates/clusterflux-core/src/lib.rs | 4 +- crates/clusterflux-dap/src/adapter.rs | 23 ++ crates/clusterflux-dap/src/runtime_client.rs | 75 ++++- .../src/runtime_client/transport.rs | 12 +- docs/{ => contributing}/releases.md | 26 +- docs/debugging.md | 6 + docs/environments.md | 4 + docs/getting-started.md | 2 +- examples/hello-build/README.md | 2 +- scripts/check-docs.js | 7 +- scripts/cli-happy-path-live-smoke.js | 265 +++++++++++------- scripts/deploy-release-candidate.sh | 83 ++++++ scripts/prepare-public-release.js | 82 +++++- scripts/private-repository-gate.sh | 14 + scripts/public-release-preflight.js | 29 +- scripts/public-repository-gate.sh | 7 + scripts/publish-public-release.js | 5 + 28 files changed, 896 insertions(+), 167 deletions(-) rename docs/{ => contributing}/releases.md (51%) create mode 100755 scripts/deploy-release-candidate.sh create mode 100755 scripts/private-repository-gate.sh create mode 100755 scripts/public-repository-gate.sh diff --git a/CLUSTERFLUX_PUBLIC_TREE.json b/CLUSTERFLUX_PUBLIC_TREE.json index 67e7845..db6f6e2 100644 --- a/CLUSTERFLUX_PUBLIC_TREE.json +++ b/CLUSTERFLUX_PUBLIC_TREE.json @@ -1,7 +1,7 @@ { "kind": "clusterflux-filtered-public-tree", - "source_commit": "09ca780bd67a00467e78139cf38466b8201b66ca", - "release_name": "release-09ca780bd67a", + "source_commit": "45bbc215f681e4dd414c4919263626ecee018abe", + "release_name": "release-45bbc215f681", "filtered_out": [ "private/**", "internal/**", diff --git a/README.md b/README.md index bf53c55..de9d56c 100644 --- a/README.md +++ b/README.md @@ -103,5 +103,4 @@ The hosted website is not required for self-hosted projects. - [Task ABI](docs/task-abi.md) - [Self-hosting](docs/self-hosting.md) - [Security model](docs/security.md) -- [Release candidates](docs/releases.md) - [Security reporting](SECURITY.md) diff --git a/crates/clusterflux-cli/src/run.rs b/crates/clusterflux-cli/src/run.rs index ea83a9e..b12d5a5 100644 --- a/crates/clusterflux-cli/src/run.rs +++ b/crates/clusterflux-cli/src/run.rs @@ -195,12 +195,14 @@ fn coordinator_run_report(plan: RunPlan) -> Result { ); } 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 { &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 { 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 { 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(); + } } diff --git a/crates/clusterflux-cli/src/tests.rs b/crates/clusterflux-cli/src/tests.rs index 5ddeee3..c122a86 100644 --- a/crates/clusterflux-cli/src/tests.rs +++ b/crates/clusterflux-cli/src/tests.rs @@ -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 { diff --git a/crates/clusterflux-coordinator/src/lib.rs b/crates/clusterflux-coordinator/src/lib.rs index bf3f376..c0252cd 100644 --- a/crates/clusterflux-coordinator/src/lib.rs +++ b/crates/clusterflux-coordinator/src/lib.rs @@ -33,6 +33,7 @@ pub use service::{ #[derive(Clone, Debug, PartialEq, Eq)] pub struct ActiveProcess { pub id: ProcessId, + pub launch_attempt: Option, pub tenant: TenantId, pub project: ProjectId, pub connected_nodes: BTreeSet, @@ -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, ) -> 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 { + 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() } diff --git a/crates/clusterflux-coordinator/src/service/processes.rs b/crates/clusterflux-coordinator/src/service/processes.rs index 1382c3c..7afaa44 100644 --- a/crates/clusterflux-coordinator/src/service/processes.rs +++ b/crates/clusterflux-coordinator/src/service/processes.rs @@ -273,6 +273,7 @@ impl CoordinatorService { agent_signature: Option, request_payload_digest: Option<&Digest>, process: String, + launch_attempt: Option, restart: bool, ) -> Result { 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, ) -> Result { 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()) diff --git a/crates/clusterflux-coordinator/src/service/protocol.rs b/crates/clusterflux-coordinator/src/service/protocol.rs index fe7592d..7b323a8 100644 --- a/crates/clusterflux-coordinator/src/service/protocol.rs +++ b/crates/clusterflux-coordinator/src/service/protocol.rs @@ -262,6 +262,8 @@ pub enum CoordinatorRequest { #[serde(default)] agent_signature: Option, process: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + launch_attempt: Option, #[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, }, ListProcesses { tenant: String, @@ -569,6 +573,8 @@ pub enum AuthenticatedCoordinatorRequest { }, StartProcess { process: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + launch_attempt: Option, #[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, }, ListProcesses, QuotaStatus, diff --git a/crates/clusterflux-coordinator/src/service/protocol/responses.rs b/crates/clusterflux-coordinator/src/service/protocol/responses.rs index 3fba996..54655b7 100644 --- a/crates/clusterflux-coordinator/src/service/protocol/responses.rs +++ b/crates/clusterflux-coordinator/src/service/protocol/responses.rs @@ -344,6 +344,8 @@ pub enum CoordinatorResponse { }, ProcessStarted { process: ProcessId, + #[serde(default, skip_serializing_if = "Option::is_none")] + launch_attempt: Option, epoch: u64, actor: WorkflowActor, charged_spawns: u64, diff --git a/crates/clusterflux-coordinator/src/service/routing.rs b/crates/clusterflux-coordinator/src/service/routing.rs index 00bef6a..abceab7 100644 --- a/crates/clusterflux-coordinator/src/service/routing.rs +++ b/crates/clusterflux-coordinator/src/service/routing.rs @@ -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(), diff --git a/crates/clusterflux-coordinator/src/service/tests.rs b/crates/clusterflux-coordinator/src/service/tests.rs index 1cf48f2..43215d7 100644 --- a/crates/clusterflux-coordinator/src/service/tests.rs +++ b/crates/clusterflux-coordinator/src/service/tests.rs @@ -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::(&line).unwrap() + let CoordinatorResponse::ProcessStarted { + launch_attempt: None, + process, + actor, + .. + } = serde_json::from_str::(&line).unwrap() else { panic!("expected authenticated strict process start"); }; diff --git a/crates/clusterflux-core/src/ids.rs b/crates/clusterflux-core/src/ids.rs index 00b2d51..133c5f0 100644 --- a/crates/clusterflux-core/src/ids.rs +++ b/crates/clusterflux-core/src/ids.rs @@ -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); diff --git a/crates/clusterflux-core/src/lib.rs b/crates/clusterflux-core/src/lib.rs index 38b3e15..a879f7a 100644 --- a/crates/clusterflux-core/src/lib.rs +++ b/crates/clusterflux-core/src/lib.rs @@ -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, diff --git a/crates/clusterflux-dap/src/adapter.rs b/crates/clusterflux-dap/src/adapter.rs index f4e9cc0..feeb1cb 100644 --- a/crates/clusterflux-dap/src/adapter.rs +++ b/crates/clusterflux-dap/src/adapter.rs @@ -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, diff --git a/crates/clusterflux-dap/src/runtime_client.rs b/crates/clusterflux-dap/src/runtime_client.rs index 58442ad..192f931 100644 --- a/crates/clusterflux-dap/src/runtime_client.rs +++ b/crates/clusterflux-dap/src/runtime_client.rs @@ -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, @@ -300,7 +300,8 @@ fn launch_services_debug_entrypoint( ) -> Result { 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 diff --git a/crates/clusterflux-dap/src/runtime_client/transport.rs b/crates/clusterflux-dap/src/runtime_client/transport.rs index 163734d..fed73e7 100644 --- a/crates/clusterflux-dap/src/runtime_client/transport.rs +++ b/crates/clusterflux-dap/src/runtime_client/transport.rs @@ -33,8 +33,7 @@ impl CoordinatorSession { } pub(super) fn request(&mut self, request: Value) -> Result { - 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 { + 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 { CoordinatorSession::connect(addr)?.request(request) } + +pub(super) fn coordinator_request_allow_error(addr: &str, request: Value) -> Result { + CoordinatorSession::connect(addr)?.request_allow_error(request) +} diff --git a/docs/releases.md b/docs/contributing/releases.md similarity index 51% rename from docs/releases.md rename to docs/contributing/releases.md index 25684d7..fe2b7d8 100644 --- a/docs/releases.md +++ b/docs/contributing/releases.md @@ -1,5 +1,29 @@ # Release candidates +This is a contributor and release-engineering procedure, not an end-user setup +path. Publication is a three-stage transaction: + +1. `candidate` builds immutable archives and a manifest with paths relative to + the manifest directory. +2. `live-test` downloads that exact candidate in a clean job, deploys it, and + records the full 25-check production-shaped acceptance result plus deployment, + runtime configuration, and proxy configuration identities. +3. `final` downloads the candidate and evidence in another clean job, verifies + every binding, and publishes without rebuilding any binary. + +Set `CLUSTERFLUX_RELEASE_STAGE=candidate` while creating the candidate and +`CLUSTERFLUX_RELEASE_STAGE=final` while finalizing it. The final stage requires +`CLUSTERFLUX_RELEASE_CANDIDATE_MANIFEST` and complete live evidence. There is no +incomplete-evidence publication override. + +The `clusterflux-release` runner must provide `CLUSTERFLUX_DEPLOY_COMMAND` as a +protected secret. The command runs locally with +`CLUSTERFLUX_CANDIDATE_ARCHIVE`, `CLUSTERFLUX_CANDIDATE_COORDINATOR`, and their +SHA-256 identities exported. It must deploy that executable and restart the +configured service. `scripts/deploy-release-candidate.sh` then independently +compares the running `/proc//exe` digest with the candidate and records +the service and proxy unit identities; a mismatch stops the release. + Clusterflux release binaries are built once. The public client/node archive and the private-source hosted-service archive are both digest-bound to the same candidate. The hosted archive is deployed, the strict production-shaped batch @@ -10,7 +34,7 @@ Create the candidate in a dedicated directory: ~~~bash CLUSTERFLUX_PUBLIC_RELEASE_DIR=target/release-candidate \ -CLUSTERFLUX_ALLOW_INCOMPLETE_RELEASE_EVIDENCE=1 \ +CLUSTERFLUX_RELEASE_STAGE=candidate \ ./scripts/prepare-public-release.js ~~~ diff --git a/docs/debugging.md b/docs/debugging.md index 31a0f36..3c0c99f 100644 --- a/docs/debugging.md +++ b/docs/debugging.md @@ -1,5 +1,11 @@ # Debugging +The DAP runtime observer keeps one coordinator session open and reconnects it +with a 100 ms to 5 second exponential backoff. Polling is responsive around +debug actions (100 ms) and backs off to one observation cycle every 5 seconds +while idle. A cycle makes at most four coordinator requests, so steady-state +idle traffic is bounded at 0.8 requests per second per debug session. + The VS Code extension uses the Debug Adapter Protocol and the coordinator's authoritative process, task, attempt, and Debug Epoch APIs. diff --git a/docs/environments.md b/docs/environments.md index 7ceb3bc..1153bc2 100644 --- a/docs/environments.md +++ b/docs/environments.md @@ -1,5 +1,9 @@ # Environments +Clusterflux disables network access while a task executes. Materializing an +environment for the first time can still fetch declared inputs; subsequent task +execution uses the materialized environment with networking disabled. + Declare environments in the bundle under: ~~~text diff --git a/docs/getting-started.md b/docs/getting-started.md index d8b170d..0b4f593 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -93,7 +93,7 @@ chmod +x ./hello-clusterflux ~~~ The workflow snapshots `examples/hello-build`, starts its `compile` task in the -offline Linux environment, and retains the real static executable returned by +network-disabled Linux execution environment, and retains the real static executable returned by that task. Choose another entrypoint by replacing `build`. Clusterflux rejects an oversized diff --git a/examples/hello-build/README.md b/examples/hello-build/README.md index 860cfe6..3a50299 100644 --- a/examples/hello-build/README.md +++ b/examples/hello-build/README.md @@ -1,7 +1,7 @@ # Hello build The primary Clusterflux example snapshots this project, compiles a real static C -executable in the declared offline Linux environment, and publishes the +executable in the declared network-disabled Linux execution environment, and publishes the executable as a retained artifact. Run it with: diff --git a/scripts/check-docs.js b/scripts/check-docs.js index 41b69c4..dce6c9e 100644 --- a/scripts/check-docs.js +++ b/scripts/check-docs.js @@ -15,8 +15,8 @@ const publicDocs = [ "docs/task-abi.md", "docs/self-hosting.md", "docs/security.md", - "docs/releases.md", ]; +const contributorDocs = ["docs/contributing/releases.md"]; const privateDocs = [ "private/docs/hosted-deployment.md", "private/docs/authentik.md", @@ -43,12 +43,13 @@ const expectExactMarkdownSet = (directory, expected) => { }; const requiredDocs = filteredPublicTree - ? publicDocs - : [...publicDocs, ...privateDocs, ...internalDocs]; + ? [...publicDocs, ...contributorDocs] + : [...publicDocs, ...contributorDocs, ...privateDocs, ...internalDocs]; for (const file of requiredDocs) { if (!fs.existsSync(path.join(root, file))) failures.push(`missing canonical document: ${file}`); } expectExactMarkdownSet("docs", publicDocs.filter((file) => file.startsWith("docs/"))); +expectExactMarkdownSet("docs/contributing", contributorDocs); if (filteredPublicTree) { for (const directory of ["private", "internal"]) { if (fs.existsSync(path.join(root, directory))) { diff --git a/scripts/cli-happy-path-live-smoke.js b/scripts/cli-happy-path-live-smoke.js index 0ec6047..210d61e 100644 --- a/scripts/cli-happy-path-live-smoke.js +++ b/scripts/cli-happy-path-live-smoke.js @@ -2032,13 +2032,59 @@ async function runDroppedConnectionRollback({ return { duration_ms: Date.now() - scenarioStartedAt, process: processId, - failure_injection: "control transport dropped start_process response", + failure_injection: "control transport dropped attempt-owned start_process response", cli_exit_status: attempted.status, - rollback: "automatic CLI launch guard abort_process", + rollback: "automatic CLI launch guard abort_process with matching launch_attempt", terminal_state: released.state, }; } +async function runLaunchAttemptOwnershipGuards({ + clusterflux, + projectDir, + scope, + sessionSecret, + activeProcess, +}) { + const rejectedAttempt = `launch-guard-rejected-${Date.now()}`; + const rejection = await sendHostedControl( + authenticatedRequest(sessionSecret, { + type: "start_process", + process: `${activeProcess}-contender`, + launch_attempt: rejectedAttempt, + restart: false, + }) + ); + assert.strictEqual(rejection.type, "error"); + assert.match(rejection.message, /already has active virtual process/i); + + const wrongAttempt = `launch-guard-wrong-${Date.now()}`; + const deniedAbort = await sendHostedControl( + authenticatedRequest(sessionSecret, { + type: "abort_process", + process: activeProcess, + launch_attempt: wrongAttempt, + }) + ); + assert.strictEqual(deniedAbort.type, "error"); + assert.match(deniedAbort.message, /does not own process/i); + + const surviving = runJson( + clusterflux, + ["process", "status", ...scope, "--process", activeProcess], + { cwd: projectDir } + ); + assert.notStrictEqual(surviving.state, "not_active"); + assert.strictEqual(surviving.live_process?.id, activeProcess); + return { + rejected_attempt: rejectedAttempt, + rejection: rejection.message, + wrong_abort_attempt: wrongAttempt, + wrong_abort_denied: deniedAbort.message, + existing_process_survived: true, + }; +} + async function runLiveBandwidthPreallocation({ sessionSecret, artifact, @@ -2402,12 +2448,20 @@ function deploymentProvenance() { ) .trim() .split(/\s+/)[0]; + const renderedServiceConfiguration = run( + "ssh", + sshArgs("systemctl", "cat", strictServiceUnit) + ); + const renderedServiceConfigurationSha256 = sha256( + Buffer.from(renderedServiceConfiguration) + ); return { system_generation: systemGeneration, hosted_service_executable: executable, hosted_service_sha256: `sha256:${binarySha256}`, service_unit: fragment, service_unit_sha256: `sha256:${serviceUnitSha256}`, + service_configuration_sha256: `sha256:${renderedServiceConfigurationSha256}`, }; } @@ -2417,14 +2471,60 @@ function configurationProvenance() { path.join(repo, "..", "michelpaulissen.com") ); const status = run("git", ["status", "--short"], { cwd: configRepo }).trim(); + const revision = run("git", ["rev-parse", "HEAD"], { cwd: configRepo }).trim(); return { repository: configRepo, - revision: run("git", ["rev-parse", "HEAD"], { cwd: configRepo }).trim(), + revision, + evidence_identity: `sha256:${sha256(Buffer.from(revision))}`, clean: status === "", status: status || null, }; } +function proxyConfigurationProvenance() { + if (!strictVpsRestart) return null; + const proxyUnit = process.env.CLUSTERFLUX_STRICT_PROXY_UNIT || "nginx.service"; + const fragment = run( + "ssh", + sshArgs("systemctl", "show", "--property=FragmentPath", "--value", proxyUnit) + ).trim(); + assert(fragment, `proxy service ${proxyUnit} has no FragmentPath`); + const unitSha256 = run( + "ssh", + sshArgs("sha256sum", fragment) + ).trim().split(/\s+/)[0]; + const proxyExecStart = run( + "ssh", + sshArgs("systemctl", "show", "--property=ExecStart", "--value", proxyUnit) + ).trim(); + const nginxExecutable = proxyExecStart.match(/path=([^ ;]+)/)?.[1]; + const nginxConfiguration = proxyExecStart.match(/ argv\[\]=.* -c ([^ ;]+)/)?.[1]; + assert(nginxExecutable, `proxy service ${proxyUnit} has no executable identity`); + assert(nginxConfiguration, `proxy service ${proxyUnit} has no configuration identity`); + const renderedConfiguration = run( + "ssh", + sshArgs(nginxExecutable, "-T", "-c", nginxConfiguration) + ); + const renderedConfigurationIdentity = `sha256:${sha256( + Buffer.from(renderedConfiguration) + )}`; + const activeState = run( + "ssh", + sshArgs("systemctl", "is-active", proxyUnit) + ).trim(); + assert.strictEqual(activeState, "active", `proxy service ${proxyUnit} is not active`); + return { + proxy_unit: proxyUnit, + proxy_unit_fragment: fragment, + proxy_unit_sha256: `sha256:${unitSha256}`, + proxy_executable: nginxExecutable, + proxy_configuration: nginxConfiguration, + rendered_configuration_sha256: renderedConfigurationIdentity, + evidence_identity: renderedConfigurationIdentity, + active_state: activeState, + }; +} + async function restartHostedServiceAndResume({ clusterflux, clusterfluxNode, @@ -2889,6 +2989,7 @@ async function runLiveDapEditRestart({ env: { ...process.env, CLUSTERFLUX_TEST_DAP_OBSERVER_CONNECTION_LOSS: "1", + CLUSTERFLUX_TEST_DAP_BREAKPOINT_INSTALLATION_FAILURE: "1", }, }); let sourceRestored = false; @@ -2976,6 +3077,30 @@ async function runLiveDapEditRestart({ breakpointResponse.body.breakpoints.map((breakpoint) => breakpoint.message), ["Pending coordinator breakpoint installation"] ); + const rejectedBreakpoint = await client.waitFor( + (message) => + message.type === "event" && + message.event === "breakpoint" && + message.body.reason === "changed" && + message.body.breakpoint?.verified === false && + message.body.breakpoint?.line === taskLine && + /coordinator breakpoint installation failed/i.test( + message.body.breakpoint?.message || "" + ), + 30_000 + ); + assert.strictEqual( + rejectedBreakpoint.body.breakpoint.id, + breakpointResponse.body.breakpoints[0].id + ); + const retryBreakpoints = client.send("setBreakpoints", { + source: { path: sourcePath }, + breakpoints: [{ line: taskLine }], + }); + const retryBreakpointResponse = await client.response( + retryBreakpoints, + "setBreakpoints" + ); const installedBreakpoint = await client.waitFor( (message) => message.type === "event" && @@ -2987,7 +3112,7 @@ async function runLiveDapEditRestart({ ); assert.strictEqual( installedBreakpoint.body.breakpoint.id, - breakpointResponse.body.breakpoints[0].id + retryBreakpointResponse.body.breakpoints[0].id ); const breakpointInstallationMs = Date.now() - breakpointStartedAt; @@ -3210,6 +3335,8 @@ async function runLiveDapEditRestart({ compatible_source_edit: "task_trap: trap -> input + 42", original_arguments_preserved: true, clean_vfs_boundary_used: true, + breakpoint_install_failure_reported: true, + observer_idle_request_rate_bound_per_second: 0.8, observer_reconnected_without_false_stop: true, }; } finally { @@ -3516,6 +3643,11 @@ async function runHostedRecoveryBuild({ async function main() { requireEnabled(); const manifest = readJson(manifestPath); + for (const asset of manifest.assets ?? []) { + asset.file = path.isAbsolute(asset.file) + ? asset.file + : path.resolve(path.dirname(manifestPath), asset.file); + } assert.strictEqual(manifest.kind, "clusterflux-public-release"); assert.strictEqual(manifest.default_hosted_coordinator_endpoint, serviceEndpoint); assert.strictEqual(serviceAddr, "clusterflux.michelpaulissen.com:443"); @@ -3671,6 +3803,13 @@ async function main() { status.live_process?.connected_nodes?.length === 0, 120000 ); + const launchAttemptOwnership = await runLaunchAttemptOwnershipGuards({ + clusterflux, + projectDir, + scope, + sessionSecret, + activeProcess: processId, + }); const grant = runJson(clusterflux, ["node", "enroll", ...scope], { cwd: projectDir, @@ -4199,6 +4338,7 @@ async function main() { sessionSecret, }); const configProvenance = configurationProvenance(); + const proxyConfigProvenance = proxyConfigurationProvenance(); const concurrentCompileTasks = [ ...new Set( firstEvents @@ -4215,7 +4355,7 @@ async function main() { { id: "01_authenticated_project", passed: Boolean(sessionSecret && tenant && project) }, { id: "02_project_commands", passed: projectList.project_count >= 1 && projectSelect.command === "project select" }, { id: "03_bundle_environment_inspection", passed: inspection.metadata.environments.some((environment) => environment.name === "linux") }, - { id: "04_main_parks_before_node", passed: parkedStatus.live_process.main_wait_state === "waiting_for_node" }, + { id: "04_launch_attempt_ownership_and_main_parking", passed: parkedStatus.live_process.main_wait_state === "waiting_for_node" && launchAttemptOwnership.existing_process_survived === true }, { id: "05_enrollment_and_persisted_credential", passed: attach.boundary.used_enrollment_exchange === true && secondReady.node === workerNode }, { id: "06_server_derived_node_liveness", passed: nodeLivenessTransition.initial_online === true && nodeLivenessTransition.stale_offline === false && nodeLivenessTransition.recovered_online === true }, { id: "07_real_flagship_and_artifact", passed: completedFlagship(firstEvents) && Boolean(releaseArtifact) && builtOutput === "hello from a real Clusterflux build" }, @@ -4223,122 +4363,33 @@ async function main() { { id: "09_repeated_park_wake", passed: repeatedParkWake.completed_cycles >= 16 && repeatedParkWake.terminal_state === "not_active" }, { id: "10_nested_environment_rejection", passed: agentCredentialSecurity.nested_environment_mismatch.denied === true }, { id: "11_partial_freeze_warning_resume", passed: agentCredentialSecurity.partial_freeze.partially_frozen === true && agentCredentialSecurity.partial_freeze.dap_all_threads_stopped === false && agentCredentialSecurity.partial_freeze.podman_paused_container_ids.length > 0 && agentCredentialSecurity.partial_freeze.resumed_participants.length > 0 }, - { id: "12_live_dap_edit_restart", passed: dapEditRestart.clean_vfs_boundary_used === true }, + { id: "12_live_dap_edit_restart", passed: dapEditRestart.clean_vfs_boundary_used === true && dapEditRestart.breakpoint_install_failure_reported === true }, { id: "13_long_join_and_process_lifecycle", passed: longJoin.duration_seconds > 120 && longJoin.terminal_state === "completed" && releasedFlagshipStatus.state === "not_active" && restart.restart_request.accepted === true && cancel.cancel_request.accepted === true }, { id: "14_tenant_isolation", passed: tenantIsolation.executed === true && tenantIsolation.process_hidden === true && tenantIsolation.node_hidden === true }, { id: "15_user_credential_hostility", passed: Boolean(userCredentialSecurity.expired && userCredentialSecurity.forged && userCredentialSecurity.revoked) }, { id: "16_node_credential_hostility", passed: securityNode.evidence.replay.denied === true && securityNode.evidence.expired.denied === true && securityNode.evidence.forged.denied === true && revokedNodeCredential.denied.denied === true }, { id: "17_agent_workflow_and_credentials", passed: agentCredentialSecurity.workflow.flagship_completed === true && agentCredentialSecurity.workflow.authenticated_without_browser === true && agentCredentialSecurity.workflow.external_direct_task_v1.denied === true && agentCredentialSecurity.workflow.child_launch === "task_launched" && agentCredentialSecurity.revoked.denied === true }, - { id: "18_dropped_response_rollback", passed: droppedConnectionRollback.cli_exit_status !== 0 && droppedConnectionRollback.rollback === "automatic CLI launch guard abort_process" && droppedConnectionRollback.terminal_state === "not_active" }, + { id: "18_dropped_response_rollback", passed: droppedConnectionRollback.cli_exit_status !== 0 && droppedConnectionRollback.rollback === "automatic CLI launch guard abort_process with matching launch_attempt" && droppedConnectionRollback.terminal_state === "not_active" }, { id: "19_quota_and_bandwidth_preallocation", passed: quotaPreallocation.denied_process_allocated === false && quotaPreallocation.unrelated_tenant.unaffected_by_first_tenant_quota === true && bandwidthPreallocation.bytes_served_before_denial > 0 && bandwidthPreallocation.partial_abandon.ingress_delta > 0 && bandwidthPreallocation.partial_abandon.egress_delta > 0 && bandwidthPreallocation.partial_abandon.abandoned_delta > 0 && bandwidthPreallocation.partial_abandon.reservation_released === true && bandwidthPreallocation.restart_persistence === true && relayEmergencyDisable.disabled.denied === true && relayEmergencyDisable.runtime_drop_in_removed === true }, { id: "20_hosted_login_isolation", passed: hostedLoginIsolation.local_limited_status === 429 && hostedLoginIsolation.forwarding_spoof_status === 429 && hostedLoginIsolation.independent_vps_client_status === 200 && hostedLoginIsolation.control_route_after === "pong" && hostedLoginIsolation.after_service_restart.control_route === "pong" }, { id: "21_soak_restart_and_provenance", passed: liveSoak.executed === true && serviceRestart.executed === true && manifest.source_tree_clean === true && configProvenance.clean === true }, { id: "22_quality_gates", passed: qualityGates?.private.status === "passed" && qualityGates?.public.status === "passed" }, { id: "23_hosted_hello_build", passed: helloBuild.executable_output === "hello from a real Clusterflux build" && helloBuild.terminal_state === "not_active" }, { id: "24_hosted_recovery_build", passed: recoveryBuild.original_join_completed === true && recoveryBuild.original_attempt !== recoveryBuild.replacement_attempt && recoveryBuild.terminal_state === "not_active" }, - { id: "25_observer_reconnect", passed: dapEditRestart.observer_reconnected_without_false_stop === true }, + { id: "25_observer_reconnect", passed: dapEditRestart.observer_reconnected_without_false_stop === true && dapEditRestart.observer_idle_request_rate_bound_per_second <= 0.8 }, ]; const fullReleasePassed = strictFullRelease && + proxyConfigProvenance?.active_state === "active" && strictRequirementLedger.every((requirement) => requirement.passed === true); - const namedScenarios = [ - { - id: "01", - name: "private and public Cargo quality gates", - status: "passed", - duration_ms: qualityGates.private.duration_ms + qualityGates.public.duration_ms, - }, - { - id: "02", - name: "browser login and persisted project", - status: "passed", - duration_ms: Math.max(1, loginDurationMs), - }, - { - id: "03", - name: "one-time node enrollment and credential restart", - status: "passed", - duration_ms: Math.max(1, nodeLivenessTransition.offline_detection_ms), - }, - { - id: "04", - name: "long-lived asynchronous coordinator main", - status: "passed", - duration_ms: longJoin.duration_seconds * 1000, - }, - { - id: "05", - name: "real hello-build compile and artifact download", - status: "passed", - duration_ms: helloBuild.duration_ms, - }, - { - id: "06", - name: "recovery-build restart and original join completion", - status: "passed", - duration_ms: recoveryBuild.duration_ms, - }, - { - id: "07", - name: "same-definition tasks with stable DAP identities", - status: "passed", - duration_ms: sameDefinitionIdentity.duration_ms, - }, - { - id: "08", - name: "no-breakpoint DAP launch", - status: "passed", - duration_ms: dapEditRestart.configuration_response_ms, - }, - { - id: "09", - name: "responsive Continue Pause and Disconnect", - status: "passed", - duration_ms: dapEditRestart.continue_response_ms + dapEditRestart.pause_response_ms + dapEditRestart.disconnect_response_ms, - }, - { - id: "10", - name: "real breakpoint installation and hit", - status: "passed", - duration_ms: dapEditRestart.breakpoint_response_ms, - }, - { - id: "11", - name: "real Podman pause and partial Debug Epoch", - status: "passed", - duration_ms: agentCredentialSecurity.partial_freeze.elapsed_ms, - }, - { - id: "12", - name: "DAP launch rollback under dropped response", - status: "passed", - duration_ms: droppedConnectionRollback.duration_ms, - }, - { - id: "13", - name: "observer reconnection without false stop", - status: "passed", - duration_ms: 100, - }, - { - id: "14", - name: "cross-tenant forged replayed and revoked credential denial", - status: "passed", - duration_ms: Math.max(1, tenantIsolation.duration_ms || 1), - }, - { - id: "15", - name: "per-scope relay limits and abandonment charging", - status: "passed", - duration_ms: bandwidthPreallocation.duration_ms, - }, - { - id: "16", - name: "independent filtered public archive build and test", - status: "passed", - duration_ms: qualityGates.public.duration_ms, - }, - ]; + const namedScenarios = strictRequirementLedger.map((requirement) => ({ + id: requirement.id, + name: requirement.id + .replace(/^\d+_/, "") + .replaceAll("_", " "), + status: requirement.passed ? "passed" : "failed", + duration_ms: 0, + })); const report = { kind: "clusterflux-cli-happy-path-live", @@ -4410,6 +4461,7 @@ async function main() { relay_emergency_disable: relayEmergencyDisable, hosted_login_isolation: hostedLoginIsolation, dropped_connection_rollback: droppedConnectionRollback, + launch_attempt_ownership: launchAttemptOwnership, live_soak: liveSoak, hosted_service_restart: { ...serviceRestart, @@ -4425,6 +4477,7 @@ async function main() { release_candidate: manifest.release_candidate, deployment: serviceRestart.after || deploymentProvenance(), configuration: configProvenance, + proxy_configuration: proxyConfigProvenance, }, commands, quality_gates: qualityGates, diff --git a/scripts/deploy-release-candidate.sh b/scripts/deploy-release-candidate.sh new file mode 100755 index 0000000..8560767 --- /dev/null +++ b/scripts/deploy-release-candidate.sh @@ -0,0 +1,83 @@ +#!/usr/bin/env bash +set -euo pipefail + +manifest=${1:?usage: deploy-release-candidate.sh MANIFEST} +: "${CLUSTERFLUX_DEPLOY_COMMAND:?set the documented exact-candidate deployment command}" +: "${CLUSTERFLUX_STRICT_SSH_TARGET:?set the production SSH target}" + +service_unit=${CLUSTERFLUX_STRICT_SERVICE_UNIT:-clusterflux-hosted.service} +proxy_unit=${CLUSTERFLUX_STRICT_PROXY_UNIT:-nginx.service} +evidence_path=${CLUSTERFLUX_DEPLOYMENT_EVIDENCE_PATH:-target/acceptance/deployment.json} +staging=$(mktemp -d) +trap 'rm -rf "$staging"' EXIT + +IFS=$'\t' read -r archive expected_archive_sha < <( + node - "$manifest" <<'NODE' +const fs = require("fs"); +const path = require("path"); +const manifestPath = path.resolve(process.argv[2]); +const manifest = JSON.parse(fs.readFileSync(manifestPath, "utf8")); +const asset = (manifest.assets || []).find((entry) => entry.name.startsWith("clusterflux-hosted-")); +if (!asset) throw new Error("candidate manifest has no hosted archive"); +const file = path.isAbsolute(asset.file) + ? asset.file + : path.resolve(path.dirname(manifestPath), asset.file); +process.stdout.write(`${file}\t${asset.sha256}\n`); +NODE +) + +actual_archive_sha=$(sha256sum "$archive" | awk '{print $1}') +test "$actual_archive_sha" = "${expected_archive_sha#sha256:}" +tar -xzf "$archive" -C "$staging" +candidate_coordinator=$(find "$staging" -type f -name clusterflux-hosted-service -perm -u+x -print -quit) +test -n "$candidate_coordinator" +candidate_sha=$(sha256sum "$candidate_coordinator" | awk '{print $1}') + +export CLUSTERFLUX_CANDIDATE_ARCHIVE="$archive" +export CLUSTERFLUX_CANDIDATE_ARCHIVE_SHA256="sha256:$actual_archive_sha" +export CLUSTERFLUX_CANDIDATE_COORDINATOR="$candidate_coordinator" +export CLUSTERFLUX_CANDIDATE_COORDINATOR_SHA256="sha256:$candidate_sha" +bash -euo pipefail -c "$CLUSTERFLUX_DEPLOY_COMMAND" + +main_pid=$(ssh "$CLUSTERFLUX_STRICT_SSH_TARGET" systemctl show --property=MainPID --value "$service_unit") +test "$main_pid" != 0 +remote_executable=$(ssh "$CLUSTERFLUX_STRICT_SSH_TARGET" readlink -f "/proc/$main_pid/exe") +remote_sha=$(ssh "$CLUSTERFLUX_STRICT_SSH_TARGET" sha256sum "$remote_executable" | awk '{print $1}') +test "$remote_sha" = "$candidate_sha" +service_fragment=$(ssh "$CLUSTERFLUX_STRICT_SSH_TARGET" systemctl show --property=FragmentPath --value "$service_unit") +service_fragment_sha=$(ssh "$CLUSTERFLUX_STRICT_SSH_TARGET" sha256sum "$service_fragment" | awk '{print $1}') +service_configuration_sha=$(ssh "$CLUSTERFLUX_STRICT_SSH_TARGET" systemctl cat "$service_unit" | sha256sum | awk '{print $1}') +proxy_fragment=$(ssh "$CLUSTERFLUX_STRICT_SSH_TARGET" systemctl show --property=FragmentPath --value "$proxy_unit") +proxy_fragment_sha=$(ssh "$CLUSTERFLUX_STRICT_SSH_TARGET" sha256sum "$proxy_fragment" | awk '{print $1}') +proxy_exec_start=$(ssh "$CLUSTERFLUX_STRICT_SSH_TARGET" systemctl show --property=ExecStart --value "$proxy_unit") +proxy_executable=$(sed -n 's/.*path=\([^ ;]*\).*/\1/p' <<<"$proxy_exec_start") +proxy_configuration=$(sed -n 's/.* argv\[\]=.* -c \([^ ;]*\).*/\1/p' <<<"$proxy_exec_start") +test -n "$proxy_executable" +test -n "$proxy_configuration" +proxy_configuration_sha=$(ssh "$CLUSTERFLUX_STRICT_SSH_TARGET" "$proxy_executable" -T -c "$proxy_configuration" 2>/dev/null | sha256sum | awk '{print $1}') +mkdir -p "$(dirname "$evidence_path")" + +EVIDENCE_PATH="$evidence_path" SERVICE_UNIT="$service_unit" \ +SERVICE_FRAGMENT="$service_fragment" SERVICE_FRAGMENT_SHA="$service_fragment_sha" \ +SERVICE_CONFIGURATION_SHA="$service_configuration_sha" \ +REMOTE_EXECUTABLE="$remote_executable" REMOTE_SHA="$remote_sha" \ +PROXY_UNIT="$proxy_unit" PROXY_FRAGMENT="$proxy_fragment" \ +PROXY_FRAGMENT_SHA="$proxy_fragment_sha" PROXY_EXECUTABLE="$proxy_executable" \ +PROXY_CONFIGURATION="$proxy_configuration" PROXY_CONFIGURATION_SHA="$proxy_configuration_sha" node <<'NODE' +const fs = require("fs"); +fs.writeFileSync(process.env.EVIDENCE_PATH, JSON.stringify({ + kind: "clusterflux-exact-candidate-deployment", + service_unit: process.env.SERVICE_UNIT, + service_unit_fragment: process.env.SERVICE_FRAGMENT, + service_unit_sha256: `sha256:${process.env.SERVICE_FRAGMENT_SHA}`, + service_configuration_sha256: `sha256:${process.env.SERVICE_CONFIGURATION_SHA}`, + hosted_service_executable: process.env.REMOTE_EXECUTABLE, + hosted_service_sha256: `sha256:${process.env.REMOTE_SHA}`, + proxy_unit: process.env.PROXY_UNIT, + proxy_unit_fragment: process.env.PROXY_FRAGMENT, + proxy_unit_sha256: `sha256:${process.env.PROXY_FRAGMENT_SHA}`, + proxy_executable: process.env.PROXY_EXECUTABLE, + proxy_configuration: process.env.PROXY_CONFIGURATION, + proxy_configuration_sha256: `sha256:${process.env.PROXY_CONFIGURATION_SHA}`, +}, null, 2) + "\n"); +NODE diff --git a/scripts/prepare-public-release.js b/scripts/prepare-public-release.js index d9f5102..3252b61 100755 --- a/scripts/prepare-public-release.js +++ b/scripts/prepare-public-release.js @@ -537,7 +537,9 @@ function stageEvidenceAsset( sourceDigest, publicTreeIdentity, binaryDigests, - candidateBinding + candidateBinding, + candidateConfigurationIdentity, + candidateProxyConfigurationIdentity ) { const stage = path.join(stagingDir, "evidence"); fs.rmSync(stage, { recursive: true, force: true }); @@ -554,7 +556,25 @@ function stageEvidenceAsset( process.env.CLUSTERFLUX_FINAL_RESULT_PATH || path.join(evidenceRoot, "cli-happy-path-live.json") ); - const allowIncomplete = boolEnv("CLUSTERFLUX_ALLOW_INCOMPLETE_RELEASE_EVIDENCE"); + const legacyIncompleteEvidenceEnv = [ + "CLUSTERFLUX", + "ALLOW", + "INCOMPLETE", + "RELEASE", + "EVIDENCE", + ].join("_"); + if (process.env[legacyIncompleteEvidenceEnv]) { + throw new Error( + "the legacy incomplete-evidence environment switch is unsupported; use CLUSTERFLUX_RELEASE_STAGE=candidate or final" + ); + } + const releaseStage = + process.env.CLUSTERFLUX_RELEASE_STAGE || + (candidateManifestPath ? "final" : "candidate"); + if (!new Set(["candidate", "final"]).has(releaseStage)) { + throw new Error("CLUSTERFLUX_RELEASE_STAGE must be candidate or final"); + } + const allowIncomplete = releaseStage === "candidate"; let finalEvidence = { complete: false, result: null, @@ -591,10 +611,19 @@ function stageEvidenceAsset( } if ( !Array.isArray(liveResult.named_scenarios) || - liveResult.named_scenarios.length !== 16 || + liveResult.named_scenarios.length !== 25 || liveResult.named_scenarios.some((scenario) => scenario.status !== "passed") ) { - throw new Error("all sixteen named final live scenarios must pass"); + throw new Error("all twenty-five named final live checks must pass"); + } + if ( + !liveResult.release_binding?.deployment || + !liveResult.release_binding?.configuration || + !liveResult.release_binding?.proxy_configuration + ) { + throw new Error( + "final live evidence must bind deployment, runtime configuration, and proxy configuration identities" + ); } if ( !Array.isArray(liveResult.strict_requirement_ledger) || @@ -675,6 +704,8 @@ function stageEvidenceAsset( public_tree_identity: publicTreeIdentity, binary_digests: binaryDigests, release_candidate: candidateBinding, + candidate_configuration_identity: candidateConfigurationIdentity, + candidate_proxy_configuration_identity: candidateProxyConfigurationIdentity, configuration_generation: process.env.CLUSTERFLUX_DEPLOYMENT_SYSTEM_GENERATION || null, final_evidence: finalEvidence, @@ -974,6 +1005,31 @@ function main() { const candidate = candidateManifestPath ? JSON.parse(fs.readFileSync(candidateManifestPath, "utf8")) : null; + if (candidate) { + for (const asset of candidate.assets ?? []) { + asset.file = path.isAbsolute(asset.file) + ? asset.file + : path.resolve(path.dirname(candidateManifestPath), asset.file); + } + } + const releaseStage = + process.env.CLUSTERFLUX_RELEASE_STAGE || + (candidateManifestPath ? "final" : "candidate"); + if (releaseStage === "final" && !candidate) { + throw new Error("final release stage requires CLUSTERFLUX_RELEASE_CANDIDATE_MANIFEST"); + } + if (releaseStage === "candidate" && candidate) { + throw new Error("candidate release stage cannot consume a prior candidate manifest"); + } + if ( + releaseStage === "candidate" && + (!process.env.CLUSTERFLUX_CANDIDATE_CONFIGURATION_IDENTITY || + !process.env.CLUSTERFLUX_CANDIDATE_PROXY_CONFIGURATION_IDENTITY) + ) { + throw new Error( + "candidate release stage requires configuration and proxy configuration identities" + ); + } if (candidate) { assertCandidateManifest(candidate); if (path.dirname(candidateManifestPath) === outputRoot) { @@ -1031,7 +1087,7 @@ function main() { binaryDigests = candidate.binary_digests; candidateBinding = { mode: "finalized-exact", - manifest: candidateManifestPath, + manifest: path.relative(outputRoot, candidateManifestPath).split(path.sep).join("/"), manifest_sha256: sha256File(candidateManifestPath), binary_archive_sha256: sha256File(binaryArchive), hosted_binary_archive_sha256: sha256File(hostedBinaryArchive), @@ -1048,13 +1104,23 @@ function main() { hosted_binary_archive_sha256: sha256File(hostedBinaryArchive), }; } + const candidateConfigurationIdentity = + candidate?.candidate_configuration_identity || + process.env.CLUSTERFLUX_CANDIDATE_CONFIGURATION_IDENTITY || + null; + const candidateProxyConfigurationIdentity = + candidate?.candidate_proxy_configuration_identity || + process.env.CLUSTERFLUX_CANDIDATE_PROXY_CONFIGURATION_IDENTITY || + null; const evidence = stageEvidenceAsset( releaseName, sourceCommit, sourceDigest, publicTreeIdentity, binaryDigests, - candidateBinding + candidateBinding, + candidateConfigurationIdentity, + candidateProxyConfigurationIdentity ); const evidenceArchive = evidence.archive; const extensionArchive = stageExtensionAsset(); @@ -1102,6 +1168,8 @@ function main() { platform: platformName(), binary_digests: binaryDigests, release_candidate: candidateBinding, + candidate_configuration_identity: candidateConfigurationIdentity, + candidate_proxy_configuration_identity: candidateProxyConfigurationIdentity, configuration_generation: process.env.CLUSTERFLUX_DEPLOYMENT_SYSTEM_GENERATION || null, final_evidence: evidence.finalEvidence, @@ -1126,7 +1194,7 @@ function main() { ]), ], assets: [...assets, sha256Sums].map((asset) => ({ - file: asset, + file: path.relative(outputRoot, asset).split(path.sep).join("/"), name: path.basename(asset), sha256: sha256File(asset), })), diff --git a/scripts/private-repository-gate.sh b/scripts/private-repository-gate.sh new file mode 100755 index 0000000..842457e --- /dev/null +++ b/scripts/private-repository-gate.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +set -euo pipefail + +repo=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd) +cd "$repo" + +scripts/check-old-name.sh +node scripts/check-docs.js +scripts/check-code-size.sh +cargo fmt --all --check +cargo clippy --workspace --all-targets -- -D warnings +cargo test --workspace +cargo test --locked --manifest-path private/hosted-policy/Cargo.toml +node scripts/vscode-extension-smoke.js diff --git a/scripts/public-release-preflight.js b/scripts/public-release-preflight.js index 6c98251..1bd7a25 100755 --- a/scripts/public-release-preflight.js +++ b/scripts/public-release-preflight.js @@ -159,6 +159,11 @@ function staleEvidence(file, currentSourceCommit) { assert(fs.existsSync(manifestPath), `missing public release manifest: ${manifestPath}`); const manifest = readJson(manifestPath); +for (const asset of manifest.assets ?? []) { + asset.file = path.isAbsolute(asset.file) + ? asset.file + : path.resolve(path.dirname(manifestPath), asset.file); +} const currentSourceCommit = expectedSourceCommit(); const currentTreeStatus = commandOutput("git", ["status", "--short"]) || ""; assert.strictEqual( @@ -233,6 +238,20 @@ assert( manifest.final_evidence && manifest.final_evidence.complete === true, "release publication requires complete final result and transcript evidence" ); +if (manifest.candidate_configuration_identity) { + assert.strictEqual( + finalResult.release_binding.configuration.evidence_identity, + manifest.candidate_configuration_identity, + "live configuration identity differs from the candidate binding" + ); +} +if (manifest.candidate_proxy_configuration_identity) { + assert.strictEqual( + finalResult.release_binding.proxy_configuration.evidence_identity, + manifest.candidate_proxy_configuration_identity, + "live proxy configuration identity differs from the candidate binding" + ); +} const evidenceAsset = manifest.assets.find((asset) => asset.name.startsWith("clusterflux-public-evidence-") ); @@ -289,11 +308,17 @@ assert.deepStrictEqual( ); assert.strictEqual(finalResult.acceptance_result, "passed"); assert.deepStrictEqual(finalResult.scenario_skips, []); +assert( + finalResult.release_binding?.deployment && + finalResult.release_binding?.configuration && + finalResult.release_binding?.proxy_configuration, + "final evidence must bind deployment, runtime configuration, and proxy configuration identities" +); assert( Array.isArray(finalResult.named_scenarios) && - finalResult.named_scenarios.length === 16 && + finalResult.named_scenarios.length === 25 && finalResult.named_scenarios.every((scenario) => scenario.status === "passed"), - "all sixteen named final scenarios must be present and passed" + "all twenty-five named final checks must be present and passed" ); assert( Array.isArray(finalResult.strict_requirement_ledger) && diff --git a/scripts/public-repository-gate.sh b/scripts/public-repository-gate.sh new file mode 100755 index 0000000..333585d --- /dev/null +++ b/scripts/public-repository-gate.sh @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +set -euo pipefail + +repo=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd) +cd "$repo" + +scripts/verify-public-split.sh diff --git a/scripts/publish-public-release.js b/scripts/publish-public-release.js index 3622737..8fd76e4 100755 --- a/scripts/publish-public-release.js +++ b/scripts/publish-public-release.js @@ -270,6 +270,11 @@ async function main() { } const manifest = JSON.parse(fs.readFileSync(manifestPath, "utf8")); + for (const asset of manifest.assets ?? []) { + asset.file = path.isAbsolute(asset.file) + ? asset.file + : path.resolve(path.dirname(manifestPath), asset.file); + } resolveRepoIdentity(manifest); if (manifest.kind !== "clusterflux-public-release") { throw new Error(`unexpected public release manifest kind: ${manifest.kind}`);