diff --git a/CLUSTERFLUX_PUBLIC_TREE.json b/CLUSTERFLUX_PUBLIC_TREE.json index f968739..ccdaf0e 100644 --- a/CLUSTERFLUX_PUBLIC_TREE.json +++ b/CLUSTERFLUX_PUBLIC_TREE.json @@ -1,7 +1,7 @@ { "kind": "clusterflux-filtered-public-tree", - "source_commit": "ea887c8f56cd53985a1179b13e5f1b85c485f584", - "release_name": "release-ea887c8f56cd", + "source_commit": "7fcdc75d8eaf4dc03b09086568dbb184f903f6a4", + "release_name": "release-7fcdc75d8eaf", "filtered_out": [ "private/**", "internal/**", diff --git a/crates/clusterflux-coordinator/src/service/logs.rs b/crates/clusterflux-coordinator/src/service/logs.rs index 79e4ade..957e475 100644 --- a/crates/clusterflux-coordinator/src/service/logs.rs +++ b/crates/clusterflux-coordinator/src/service/logs.rs @@ -248,10 +248,19 @@ impl CoordinatorService { self.notify_coordinator_main_waiters(&event); } self.maybe_retire_terminal_process(&event.tenant, &event.project, &event.process)?; + let events_recorded = self + .task_events + .iter() + .filter(|recorded| { + recorded.tenant == event.tenant + && recorded.project == event.project + && recorded.process == event.process + }) + .count(); Ok(CoordinatorResponse::TaskRecorded { process: event.process, task: event.task, - events_recorded: self.task_events.len(), + events_recorded, }) } @@ -613,6 +622,14 @@ impl CoordinatorService { self.coordinator.abort_process(tenant, project, process)?; self.clear_debug_state_for_process(tenant, project, process); self.clear_operator_panel_state(tenant, project, process); + let (pinned, protected_processes) = + self.artifact_retention_guards_for_project(tenant, project); + self.artifact_registry.enforce_project_metadata_limit( + tenant, + project, + &pinned, + &protected_processes, + ); Ok(true) } @@ -623,8 +640,30 @@ impl CoordinatorService { let now_epoch_seconds = self.current_epoch_seconds()?; self.artifact_registry .expire_download_links(now_epoch_seconds); + let tenant = flush.tenant.clone(); + let project = flush.project.clone(); + let (pinned, protected_processes) = + self.artifact_retention_guards_for_project(&tenant, &project); + self.artifact_registry + .flush_metadata_with_protected_processes(flush, &pinned, &protected_processes) + .map(|_| ()) + .map_err(CoordinatorServiceError::Protocol) + } + + fn artifact_retention_guards_for_project( + &self, + tenant: &TenantId, + project: &ProjectId, + ) -> ( + std::collections::BTreeSet, + std::collections::BTreeSet, + ) { let mut pinned = std::collections::BTreeSet::new(); for checkpoint in self.task_restart_checkpoints.values() { + if &checkpoint.assignment.tenant != tenant || &checkpoint.assignment.project != project + { + continue; + } for artifact in &checkpoint.assignment.task_spec.required_artifacts { pinned.insert(ArtifactScopeKey::from_refs( &checkpoint.assignment.tenant, @@ -634,6 +673,9 @@ impl CoordinatorService { } } for pending in &self.pending_task_launches { + if &pending.tenant != tenant || &pending.project != project { + continue; + } for artifact in &pending.task_spec.required_artifacts { pinned.insert(ArtifactScopeKey::from_refs( &pending.tenant, @@ -642,10 +684,13 @@ impl CoordinatorService { )); } } - self.artifact_registry - .flush_metadata_bounded(flush, &pinned) - .map(|_| ()) - .map_err(CoordinatorServiceError::Protocol) + let protected_processes = self + .coordinator + .active_processes_for_project(tenant, project) + .into_iter() + .map(|process| process.id) + .collect(); + (pinned, protected_processes) } fn task_is_known_or_active( diff --git a/crates/clusterflux-coordinator/src/service/nodes.rs b/crates/clusterflux-coordinator/src/service/nodes.rs index b7a7b4e..cedae30 100644 --- a/crates/clusterflux-coordinator/src/service/nodes.rs +++ b/crates/clusterflux-coordinator/src/service/nodes.rs @@ -300,8 +300,8 @@ impl CoordinatorService { node_scope, NodeDescriptor { id: node.clone(), - tenant, - project, + tenant: tenant.clone(), + project: project.clone(), capabilities, cached_environments: cached_environment_digests.into_iter().collect(), dependency_caches: dependency_cache_digests.into_iter().collect(), @@ -311,9 +311,14 @@ impl CoordinatorService { online, }, ); + let node_descriptors = self + .node_descriptors + .values() + .filter(|descriptor| descriptor.tenant == tenant && descriptor.project == project) + .count(); Ok(CoordinatorResponse::NodeCapabilitiesRecorded { node, - node_descriptors: self.node_descriptors.len(), + node_descriptors, }) } diff --git a/crates/clusterflux-coordinator/src/service/process_launch.rs b/crates/clusterflux-coordinator/src/service/process_launch.rs index 31fda49..fc6b69c 100644 --- a/crates/clusterflux-coordinator/src/service/process_launch.rs +++ b/crates/clusterflux-coordinator/src/service/process_launch.rs @@ -545,13 +545,22 @@ impl CoordinatorService { task_spec, wasm_module_base64, }); + let queued_tasks = self + .pending_task_launches + .iter() + .filter(|pending| { + pending.tenant == tenant + && pending.project == project + && pending.process == process + }) + .count(); return Ok(CoordinatorResponse::TaskQueued { process, task, actor, reason, charged_spawns, - queued_tasks: self.pending_task_launches.len(), + queued_tasks, }); } Err(err) => return Err(err.into()), diff --git a/crates/clusterflux-coordinator/src/service/tests.rs b/crates/clusterflux-coordinator/src/service/tests.rs index 3369b5b..fba8638 100644 --- a/crates/clusterflux-coordinator/src/service/tests.rs +++ b/crates/clusterflux-coordinator/src/service/tests.rs @@ -512,6 +512,20 @@ fn service_with_completed_main_and_final_child( public_key: test_node_public_key(node.as_str()), }) .unwrap(); + service + .handle_signed_node_request_auto(CoordinatorRequest::ReportNodeCapabilities { + tenant: tenant.to_string(), + project: project.to_string(), + node: node.to_string(), + capabilities: linux_capabilities(), + cached_environment_digests: Vec::new(), + dependency_cache_digests: Vec::new(), + source_snapshots: Vec::new(), + artifact_locations: Vec::new(), + direct_connectivity: true, + online: true, + }) + .unwrap(); service .handle_request(CoordinatorRequest::StartProcess { launch_attempt: None, @@ -4549,6 +4563,81 @@ fn completed_main_await_operator_blocks_retirement_until_each_resolution() { } } +#[test] +fn completed_main_failed_child_restarted_successfully_retires_with_successful_current_attempt() { + let mut service = service_with_completed_main_and_final_child( + clusterflux_core::TaskFailurePolicy::AwaitOperator, + ); + complete_terminal_matrix_child(&mut service, TaskTerminalState::Failed); + let tenant = TenantId::from("tenant"); + let project = ProjectId::from("project"); + let process = ProcessId::from("terminal-matrix"); + let task = TaskInstanceId::from("final-child"); + + let CoordinatorResponse::TaskRestart { accepted, .. } = service + .handle_request(CoordinatorRequest::RestartTask { + tenant: tenant.to_string(), + project: project.to_string(), + actor_user: "user".to_owned(), + process: process.to_string(), + task: task.to_string(), + replacement_bundle: None, + }) + .unwrap() + else { + panic!("expected task restart"); + }; + assert!(accepted); + + service + .handle_signed_node_request_auto(CoordinatorRequest::TaskCompleted { + tenant: tenant.to_string(), + project: project.to_string(), + process: process.to_string(), + node: "worker".to_owned(), + task: task.to_string(), + terminal_state: Some(TaskTerminalState::Completed), + status_code: Some(0), + stdout_bytes: 2, + stderr_bytes: 0, + stdout_tail: "ok".to_owned(), + stderr_tail: String::new(), + stdout_truncated: false, + stderr_truncated: false, + artifact_path: None, + artifact_digest: None, + artifact_size_bytes: None, + result: Some(TaskBoundaryValue::SmallJson(json!("ok"))), + }) + .unwrap(); + + assert!(service + .coordinator + .active_process(&tenant, &project, &process) + .is_none()); + let attempts = service + .task_attempts + .get(&super::keys::task_restart_key( + &tenant, &project, &process, &task, + )) + .unwrap(); + assert!( + attempts + .iter() + .any(|attempt| !attempt.current + && attempt.state == TaskAttemptState::FailedAwaitingAction) + ); + assert!(attempts + .iter() + .any(|attempt| attempt.current && attempt.state == TaskAttemptState::Completed)); + assert_eq!( + service + .task_join_result(tenant, project, process, task) + .state, + TaskJoinState::Completed + ); +} + #[test] fn completed_main_failed_child_does_not_abort_another_active_child() { let mut service = @@ -5570,6 +5659,29 @@ fn retained_artifact_reverse_stream_is_chunked_below_control_frame_limit() { #[test] fn windows_task_events_share_the_virtual_process_scope() { let mut service = CoordinatorService::new(7); + service.record_task_completion_event(TaskCompletionEvent { + tenant: TenantId::from("other-tenant"), + project: ProjectId::from("other-project"), + process: ProcessId::from("other-process"), + node: NodeId::from("other-node"), + executor: super::TaskExecutor::Node, + task_definition: TaskDefinitionId::from("other-task"), + task: TaskInstanceId::from("other-task"), + attempt_id: None, + placement: None, + terminal_state: TaskTerminalState::Completed, + status_code: Some(0), + stdout_bytes: 0, + stderr_bytes: 0, + stdout_tail: String::new(), + stderr_tail: String::new(), + stdout_truncated: false, + stderr_truncated: false, + artifact_path: None, + artifact_digest: None, + artifact_size_bytes: None, + result: None, + }); service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant".to_owned(), @@ -5670,6 +5782,28 @@ fn windows_task_events_share_the_virtual_process_scope() { #[test] fn service_schedules_task_across_reported_node_descriptors() { let mut service = CoordinatorService::new(7); + service + .handle_request(CoordinatorRequest::AttachNode { + tenant: "other-tenant".to_owned(), + project: "other-project".to_owned(), + node: "other-node".to_owned(), + public_key: test_node_public_key("other-node"), + }) + .unwrap(); + service + .handle_signed_node_request_auto(CoordinatorRequest::ReportNodeCapabilities { + tenant: "other-tenant".to_owned(), + project: "other-project".to_owned(), + node: "other-node".to_owned(), + capabilities: linux_capabilities(), + cached_environment_digests: Vec::new(), + dependency_cache_digests: Vec::new(), + source_snapshots: Vec::new(), + artifact_locations: Vec::new(), + direct_connectivity: false, + online: true, + }) + .unwrap(); for node in ["cold-node", "warm-node"] { service .handle_request(CoordinatorRequest::AttachNode { @@ -6957,6 +7091,52 @@ 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: other_epoch, .. + } = service + .handle_request(CoordinatorRequest::StartProcess { + launch_attempt: None, + tenant: "other-tenant".to_owned(), + project: "other-project".to_owned(), + actor_user: None, + actor_agent: None, + agent_public_key_fingerprint: None, + agent_signature: None, + process: "other-vp-wait".to_owned(), + restart: false, + }) + .unwrap() + else { + panic!("expected unrelated process start"); + }; + let CoordinatorResponse::TaskQueued { + queued_tasks: other_queued_tasks, + .. + } = service + .handle_authorized_test_task_launch(CoordinatorRequest::LaunchTask { + task_spec: test_task_spec( + "other-tenant", + "other-project", + "other-vp-wait", + "other-compile", + other_epoch, + [Capability::Command], + ), + tenant: "other-tenant".to_owned(), + project: "other-project".to_owned(), + actor_user: Some("other-user".to_owned()), + actor_agent: None, + agent_public_key_fingerprint: None, + agent_signature: None, + wait_for_node: true, + artifact_path: "/vfs/artifacts/other-wait-output.txt".to_owned(), + wasm_module_base64: test_wasm_module_base64(), + }) + .unwrap() + else { + panic!("expected unrelated queued task launch"); + }; + assert_eq!(other_queued_tasks, 1); let CoordinatorResponse::ProcessStarted { launch_attempt: None, epoch, diff --git a/crates/clusterflux-core/src/artifact.rs b/crates/clusterflux-core/src/artifact.rs index 240d5b4..bc0e177 100644 --- a/crates/clusterflux-core/src/artifact.rs +++ b/crates/clusterflux-core/src/artifact.rs @@ -10,7 +10,7 @@ use crate::{ const MAX_ISSUED_DOWNLOAD_LINKS_PER_ARTIFACT: usize = 32; const DOWNLOAD_LINK_TOMBSTONE_SECONDS: u64 = 15 * 60; -const MAX_ARTIFACT_METADATA_PER_PROCESS: usize = 256; +const MAX_ARTIFACT_METADATA_PER_PROJECT: usize = 1_024; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -184,53 +184,17 @@ impl ArtifactRegistry { &mut self, flush: ArtifactFlush, pinned: &BTreeSet, + ) -> Result { + self.flush_metadata_with_protected_processes(flush, pinned, &BTreeSet::new()) + } + + pub fn flush_metadata_with_protected_processes( + &mut self, + flush: ArtifactFlush, + pinned: &BTreeSet, + protected_processes: &BTreeSet, ) -> Result { let key = ArtifactScopeKey::from_refs(&flush.tenant, &flush.project, &flush.id); - let replacing_existing = self.artifacts.contains_key(&key); - while !replacing_existing - && self - .artifacts - .values() - .filter(|metadata| { - metadata.tenant == flush.tenant - && metadata.project == flush.project - && metadata.process == flush.process - }) - .count() - >= MAX_ARTIFACT_METADATA_PER_PROCESS - { - let candidate = self - .artifacts - .values() - .filter(|metadata| { - metadata.tenant == flush.tenant - && metadata.project == flush.project - && metadata.process == flush.process - && !pinned.contains(&ArtifactScopeKey::from_refs( - &metadata.tenant, - &metadata.project, - &metadata.id, - )) - && !self.issued_download_links.values().any(|issued| { - issued.link.tenant == metadata.tenant - && issued.link.project == metadata.project - && issued.link.artifact == metadata.id - }) - }) - .min_by_key(|metadata| metadata.flushed_epoch) - .map(|metadata| { - ArtifactScopeKey::from_refs( - &metadata.tenant, - &metadata.project, - &metadata.id, - ) - }) - .ok_or_else(|| { - "artifact metadata retention limit reached and every retained object is pinned by active work, restart state, or a download" - .to_owned() - })?; - self.artifacts.remove(&candidate); - } self.next_epoch += 1; let metadata = ArtifactMetadata { id: flush.id.clone(), @@ -247,9 +211,65 @@ impl ArtifactRegistry { coordinator_has_large_bytes: false, }; self.artifacts.insert(key, metadata.clone()); + let mut protected_processes = protected_processes.clone(); + protected_processes.insert(metadata.process.clone()); + self.enforce_project_metadata_limit( + &metadata.tenant, + &metadata.project, + pinned, + &protected_processes, + ); Ok(metadata) } + pub fn enforce_project_metadata_limit( + &mut self, + tenant: &TenantId, + project: &ProjectId, + pinned: &BTreeSet, + protected_processes: &BTreeSet, + ) -> usize { + let mut evicted = 0; + while self + .artifacts + .values() + .filter(|metadata| &metadata.tenant == tenant && &metadata.project == project) + .count() + > MAX_ARTIFACT_METADATA_PER_PROJECT + { + let candidate = self + .artifacts + .values() + .filter(|metadata| { + &metadata.tenant == tenant + && &metadata.project == project + && !pinned.contains(&ArtifactScopeKey::from_refs( + &metadata.tenant, + &metadata.project, + &metadata.id, + )) + && !protected_processes.contains(&metadata.process) + && metadata.explicit_locations.is_empty() + && !self.issued_download_links.values().any(|issued| { + !issued.revoked + && issued.link.tenant == metadata.tenant + && issued.link.project == metadata.project + && issued.link.artifact == metadata.id + }) + }) + .min_by_key(|metadata| metadata.flushed_epoch) + .map(|metadata| { + ArtifactScopeKey::from_refs(&metadata.tenant, &metadata.project, &metadata.id) + }); + let Some(candidate) = candidate else { + break; + }; + self.artifacts.remove(&candidate); + evicted += 1; + } + evicted + } + pub fn sync_to_explicit_store( &mut self, tenant: &TenantId, @@ -1340,8 +1360,10 @@ mod tests { } #[test] - fn artifact_metadata_is_bounded_without_evicting_pins() { + fn artifact_metadata_is_bounded_per_project_without_evicting_live_or_retained_state() { let mut registry = ArtifactRegistry::default(); + let tenant = TenantId::from("tenant"); + let project = ProjectId::from("project"); registry.flush_metadata(ArtifactFlush { id: ArtifactId::from("artifact-0"), tenant: TenantId::from("other-tenant"), @@ -1352,72 +1374,92 @@ mod tests { digest: Digest::sha256("other-content"), size: 1, }); - let mut pinned = BTreeSet::new(); - for index in 0..MAX_ARTIFACT_METADATA_PER_PROCESS { + for index in 0..MAX_ARTIFACT_METADATA_PER_PROJECT { let id = ArtifactId::new(format!("artifact-{index}")); - pinned.insert(ArtifactScopeKey::new( - TenantId::from("tenant"), - ProjectId::from("project"), - id.clone(), - )); - registry - .flush_metadata_bounded( - ArtifactFlush { - id, - tenant: TenantId::from("tenant"), - project: ProjectId::from("project"), - process: ProcessId::from("process"), - producer_task: TaskInstanceId::new(format!("task-{index}")), - retaining_node: NodeId::from("node"), - digest: Digest::sha256(format!("content-{index}")), - size: 1, - }, - &pinned, - ) - .unwrap(); + registry.flush_metadata(ArtifactFlush { + id, + tenant: tenant.clone(), + project: project.clone(), + process: ProcessId::new(if index == 3 { + "active-process-3".to_owned() + } else { + format!("completed-process-{index}") + }), + producer_task: TaskInstanceId::new(format!("task-{index}")), + retaining_node: NodeId::from("node"), + digest: Digest::sha256(format!("content-{index}")), + size: 1, + }); } assert_eq!( registry.artifact_count(), - MAX_ARTIFACT_METADATA_PER_PROCESS + 1 + MAX_ARTIFACT_METADATA_PER_PROJECT + 1 ); + let pinned = BTreeSet::from([ArtifactScopeKey::new( + tenant.clone(), + project.clone(), + ArtifactId::from("artifact-0"), + )]); + registry + .sync_to_explicit_store( + &tenant, + &project, + &ArtifactId::from("artifact-1"), + "store://retained-export", + ) + .unwrap(); + let context = AuthContext { + tenant: tenant.clone(), + project: project.clone(), + actor: Actor::User(UserId::from("user")), + }; + registry + .create_download_link( + &context, + &ArtifactId::from("artifact-2"), + &DownloadPolicy { max_bytes: 1 }, + "active-download", + 10, + 60, + ) + .unwrap(); + let protected_processes = BTreeSet::from([ + ProcessId::from("active-process-3"), + ProcessId::from("active-process"), + ]); let next = ArtifactFlush { id: ArtifactId::from("artifact-next"), - tenant: TenantId::from("tenant"), - project: ProjectId::from("project"), - process: ProcessId::from("process"), + tenant: tenant.clone(), + project: project.clone(), + process: ProcessId::from("active-process"), producer_task: TaskInstanceId::from("task-next"), retaining_node: NodeId::from("node"), digest: Digest::sha256("next"), size: 1, }; - assert!(registry - .flush_metadata_bounded(next.clone(), &pinned) - .unwrap_err() - .contains("pinned")); - - pinned.remove(&ArtifactScopeKey::new( - TenantId::from("tenant"), - ProjectId::from("project"), - ArtifactId::from("artifact-0"), - )); - registry.flush_metadata_bounded(next, &pinned).unwrap(); + registry + .flush_metadata_with_protected_processes(next, &pinned, &protected_processes) + .unwrap(); assert_eq!( registry.artifact_count(), - MAX_ARTIFACT_METADATA_PER_PROCESS + 1 + MAX_ARTIFACT_METADATA_PER_PROJECT + 1 + ); + for id in ["artifact-0", "artifact-1", "artifact-2", "artifact-3"] { + assert!( + registry + .metadata(&tenant, &project, &ArtifactId::from(id)) + .is_some(), + "{id} was evicted despite being pinned, exported, downloaded, or live" + ); + } + assert!( + registry + .metadata(&tenant, &project, &ArtifactId::from("artifact-4")) + .is_none(), + "the oldest completed unprotected metadata should be evicted" ); assert!(registry - .metadata( - &TenantId::from("tenant"), - &ProjectId::from("project"), - &ArtifactId::from("artifact-0"), - ) - .is_none()); - assert!(registry - .metadata( - &TenantId::from("tenant"), - &ProjectId::from("project"), - &ArtifactId::from("artifact-next"), - ) + .metadata(&tenant, &project, &ArtifactId::from("artifact-next")) .is_some()); assert_eq!( registry diff --git a/crates/clusterflux-dap/src/adapter.rs b/crates/clusterflux-dap/src/adapter.rs index bbf9a28..4e144ab 100644 --- a/crates/clusterflux-dap/src/adapter.rs +++ b/crates/clusterflux-dap/src/adapter.rs @@ -1208,10 +1208,12 @@ fn emit_runtime_outcome( )?; } RuntimeContinuationOutcome::Terminal(record) => { + let exit_code = record.status_code.unwrap_or(1); apply_runtime_record_with_thread_events(writer, state, record)?; if state.last_task_failed { writer.output("stderr", format!("{}\n", state.command_status))?; } + writer.event("exited", json!({ "exitCode": exit_code }))?; writer.event("terminated", json!({}))?; } } diff --git a/crates/clusterflux-dap/src/main.rs b/crates/clusterflux-dap/src/main.rs index 32336b2..5e02bcd 100644 --- a/crates/clusterflux-dap/src/main.rs +++ b/crates/clusterflux-dap/src/main.rs @@ -22,7 +22,7 @@ use breakpoints::{ #[cfg(test)] use dap_protocol::{initialize_capabilities, read_message}; #[cfg(test)] -use runtime_client::{client_user_request, parse_task_restart_response}; +use runtime_client::{client_user_request, parse_task_restart_response, whole_process_status_code}; #[cfg(test)] use variables::variables_response; #[cfg(test)] diff --git a/crates/clusterflux-dap/src/runtime_client.rs b/crates/clusterflux-dap/src/runtime_client.rs index 2cdefaf..233b6bc 100644 --- a/crates/clusterflux-dap/src/runtime_client.rs +++ b/crates/clusterflux-dap/src/runtime_client.rs @@ -1164,6 +1164,8 @@ pub(crate) fn observe_services_runtime( }; if !has_current_runtime_task(&task_snapshots) { if let RuntimeContinuationOutcome::Terminal(record) = &mut outcome { + record.status_code = + whole_process_status_code(record.status_code, &task_snapshots); record.node_report = json!({ "terminal_event": record.node_report.get("terminal_event"), "task_snapshots": task_snapshots, @@ -1368,6 +1370,8 @@ pub(crate) fn observe_services_runtime( }; if !has_current_runtime_task(&task_snapshots) { if let RuntimeContinuationOutcome::Terminal(record) = &mut outcome { + record.status_code = + whole_process_status_code(record.status_code, &task_snapshots); record.node_report = json!({ "terminal_event": record.node_report.get("terminal_event"), "task_snapshots": task_snapshots, @@ -1543,6 +1547,39 @@ fn has_current_runtime_task(task_snapshots: &Value) -> bool { }) } +pub(crate) fn whole_process_status_code( + main_status_code: Option, + task_snapshots: &Value, +) -> Option { + let main_status_code = main_status_code?; + if main_status_code != 0 { + return Some(main_status_code); + } + + let snapshots = task_snapshots.get("snapshots").and_then(Value::as_array)?; + for snapshot in snapshots + .iter() + .filter(|snapshot| snapshot.get("current").and_then(Value::as_bool) == Some(true)) + { + match snapshot.get("state").and_then(Value::as_str) { + Some("completed") => {} + Some("failed" | "cancelled" | "failed_awaiting_action") => { + return Some( + snapshot + .get("status_code") + .and_then(Value::as_i64) + .and_then(|status| i32::try_from(status).ok()) + .filter(|status| *status != 0) + .unwrap_or(1), + ); + } + Some("queued" | "running") => return None, + _ => return Some(1), + } + } + Some(0) +} + fn terminal_runtime_outcome( coordinator: &str, state: &AdapterState, diff --git a/crates/clusterflux-dap/src/tests.rs b/crates/clusterflux-dap/src/tests.rs index d64ac6e..b9d7a36 100644 --- a/crates/clusterflux-dap/src/tests.rs +++ b/crates/clusterflux-dap/src/tests.rs @@ -922,6 +922,98 @@ fn detects_current_failed_attempt_awaiting_operator_action() { ); } +#[test] +fn whole_process_terminal_status_covers_every_current_logical_task_attempt() { + let snapshots = |items: serde_json::Value| json!({ "snapshots": items }); + + assert_eq!( + whole_process_status_code( + Some(0), + &snapshots(json!([ + { + "task": "child", + "current": true, + "state": "completed", + "status_code": 0 + } + ])) + ), + Some(0), + "a successful main and final child must succeed" + ); + for (state, status_code) in [("failed", 23), ("cancelled", 1)] { + assert_eq!( + whole_process_status_code( + Some(0), + &snapshots(json!([ + { + "task": "child", + "current": true, + "state": state, + "status_code": status_code + } + ])) + ), + Some(status_code), + "a terminal child must determine the whole-process result" + ); + } + assert_eq!( + whole_process_status_code( + Some(0), + &snapshots(json!([ + { + "task": "accepted-failure", + "current": true, + "state": "failed", + "command_state": "failure_accepted", + "status_code": 17 + } + ])) + ), + Some(17), + "accepting an AwaitOperator failure releases the process but does not turn failure into success" + ); + assert_eq!( + whole_process_status_code( + Some(0), + &snapshots(json!([ + { + "task": "restarted", + "attempt_id": "attempt-1", + "current": false, + "state": "failed", + "status_code": 7 + }, + { + "task": "restarted", + "attempt_id": "attempt-2", + "current": true, + "state": "completed", + "status_code": 0 + } + ])) + ), + Some(0), + "a successful replacement attempt is authoritative over stale failed attempts" + ); + assert_eq!( + whole_process_status_code( + Some(9), + &snapshots(json!([ + { + "task": "child", + "current": true, + "state": "completed", + "status_code": 0 + } + ])) + ), + Some(9), + "a failed coordinator main cannot be masked by successful children" + ); +} + #[test] fn exact_task_instance_ids_keep_stable_dap_threads_across_snapshot_reordering_and_retry() { let mut state = AdapterState::default();