From cfd4f19da21b244dd85b259ca832df7be8e21b57 Mon Sep 17 00:00:00 2001 From: Clusterflux release Date: Wed, 29 Jul 2026 02:53:56 +0200 Subject: [PATCH] Public release release-e7c2b3ac175d Source commit: e7c2b3ac175db426791c02779841c5df1d0ffdff Public tree identity: sha256:4f0b7cd828932c06d56f2406788f89b2050ef56c1e1a4f76f55341d387d78e46 --- crates/clusterflux-client/src/transport.rs | 101 ++++- .../tests/client_contract.rs | 49 ++- crates/clusterflux-coordinator/src/service.rs | 14 +- .../src/service/logs.rs | 386 +++++++++++++++--- .../src/service/process_launch.rs | 2 + .../src/service/processes.rs | 26 +- .../src/service/summaries.rs | 80 +++- .../src/service/tests.rs | 255 +++++++++++- crates/clusterflux-core/src/artifact.rs | 183 +++++++-- crates/clusterflux-dap/src/runtime_client.rs | 276 +++++++------ .../clusterflux-node/src/assignment_runner.rs | 165 ++++++-- .../src/assignment_runner/process_runner.rs | 50 ++- .../src/assignment_runner/tests.rs | 4 + crates/clusterflux-node/src/command_runner.rs | 6 + crates/clusterflux-node/src/daemon.rs | 12 +- crates/clusterflux-node/src/task_reports.rs | 90 +++- 16 files changed, 1386 insertions(+), 313 deletions(-) diff --git a/crates/clusterflux-client/src/transport.rs b/crates/clusterflux-client/src/transport.rs index 5c440ff..b480c0f 100644 --- a/crates/clusterflux-client/src/transport.rs +++ b/crates/clusterflux-client/src/transport.rs @@ -1,6 +1,7 @@ use std::collections::{BTreeMap, VecDeque}; use std::future::Future; use std::pin::Pin; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -33,13 +34,20 @@ pub trait ClientTransport: Send + Sync + 'static { fn send(&self, request: TransportRequest) -> TransportFuture; } +type ControlSessionPool = Arc>>>; + pub struct ControlTransport { endpoint: String, connect_timeout: Duration, io_timeout: Duration, - sessions: Arc>>, + sessions: Arc>>, + next_session: Arc, } +// A small pool allows independent HTMX reads to progress concurrently while +// keeping connection and blocking-worker use strictly bounded. +const SESSIONS_PER_API_PATH: usize = 4; + impl ControlTransport { pub fn new(endpoint: impl Into) -> Result { Self::with_timeouts(endpoint, Duration::from_secs(10), Duration::from_secs(30)) @@ -58,6 +66,7 @@ impl ControlTransport { connect_timeout, io_timeout, sessions: Arc::new(Mutex::new(BTreeMap::new())), + next_session: Arc::new(AtomicU64::new(0)), }) } } @@ -68,35 +77,89 @@ impl ClientTransport for ControlTransport { let connect_timeout = self.connect_timeout; let io_timeout = self.io_timeout; let sessions = Arc::clone(&self.sessions); + let next_session = Arc::clone(&self.next_session); Box::pin(async move { + let pool = { + let mut sessions = sessions.lock().map_err(|_| { + ClientTransportError::Failed( + "client transport pool lock was poisoned".to_owned(), + ) + })?; + Arc::clone(sessions.entry(request.api_path.clone()).or_insert_with(|| { + Arc::new( + (0..SESSIONS_PER_API_PATH) + .map(|_| Mutex::new(None)) + .collect(), + ) + })) + }; + let slot_index = + next_session.fetch_add(1, Ordering::Relaxed) as usize % SESSIONS_PER_API_PATH; tokio::task::spawn_blocking(move || { let value = serde_json::from_slice(&request.body) .map_err(|error| ClientTransportError::Failed(error.to_string()))?; - let mut sessions = sessions.lock().map_err(|_| { - ClientTransportError::Failed( - "client transport session lock was poisoned".to_owned(), - ) - })?; - if !sessions.contains_key(&request.api_path) { - let session = ControlSession::connect_to_api_path_with_timeouts( - &endpoint, - &request.api_path, - connect_timeout, - io_timeout, - ) - .map_err(|error| ClientTransportError::Failed(error.to_string()))?; - sessions.insert(request.api_path.clone(), session); + let mut selected = None; + for offset in 0..SESSIONS_PER_API_PATH { + let index = (slot_index + offset) % SESSIONS_PER_API_PATH; + match pool[index].try_lock() { + Ok(session) if session.is_some() => { + selected = Some(session); + break; + } + Ok(_) | Err(std::sync::TryLockError::WouldBlock) => {} + Err(std::sync::TryLockError::Poisoned(_)) => { + return Err(ClientTransportError::Failed( + "client transport session lock was poisoned".to_owned(), + )); + } + } } - let response = sessions - .get_mut(&request.api_path) - .expect("session was inserted for the requested API path") + if selected.is_none() { + for offset in 0..SESSIONS_PER_API_PATH { + let index = (slot_index + offset) % SESSIONS_PER_API_PATH; + match pool[index].try_lock() { + Ok(session) => { + selected = Some(session); + break; + } + Err(std::sync::TryLockError::WouldBlock) => {} + Err(std::sync::TryLockError::Poisoned(_)) => { + return Err(ClientTransportError::Failed( + "client transport session lock was poisoned".to_owned(), + )); + } + } + } + } + let mut session = match selected { + Some(session) => session, + None => pool[slot_index].lock().map_err(|_| { + ClientTransportError::Failed( + "client transport session lock was poisoned".to_owned(), + ) + })?, + }; + if session.is_none() { + *session = Some( + ControlSession::connect_to_api_path_with_timeouts( + &endpoint, + &request.api_path, + connect_timeout, + io_timeout, + ) + .map_err(|error| ClientTransportError::Failed(error.to_string()))?, + ); + } + let response = session + .as_mut() + .expect("session was initialized for the requested API path") .request(&value); match response { Ok(response) => serde_json::to_vec(&response) .map(|body| TransportResponse { body }) .map_err(|error| ClientTransportError::Failed(error.to_string())), Err(error) => { - sessions.remove(&request.api_path); + *session = None; Err(ClientTransportError::Failed(error.to_string())) } } diff --git a/crates/clusterflux-client/tests/client_contract.rs b/crates/clusterflux-client/tests/client_contract.rs index 47b5bd7..1f3f0ec 100644 --- a/crates/clusterflux-client/tests/client_contract.rs +++ b/crates/clusterflux-client/tests/client_contract.rs @@ -1,9 +1,10 @@ use std::net::TcpListener; +use std::time::Duration; use clusterflux_client::{ ApiErrorCategory, ApiErrorCode, ArtifactDownloadPoll, ArtifactId, ClientError, - ClusterfluxClient, MockTransport, ProjectId, SessionCredential, TenantId, UserId, - CLIENT_API_VERSION, + ClusterfluxClient, ControlTransport, MockTransport, ProjectId, SessionCredential, TenantId, + UserId, CLIENT_API_VERSION, }; use clusterflux_control::CONTROL_API_PATH; use clusterflux_coordinator::service::CoordinatorService; @@ -181,3 +182,47 @@ async fn typed_client_runs_against_the_real_strict_control_endpoint() { drop(client); server.join().unwrap(); } + +#[tokio::test] +async fn concurrent_requests_expand_the_bounded_pool_without_serializing() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = std::thread::spawn(move || { + let handlers = (0..2) + .map(|_| { + let (stream, _) = listener.accept().unwrap(); + std::thread::spawn(move || { + let mut service = CoordinatorService::new(7); + service + .issue_cli_session( + TenantId::from("tenant-one"), + ProjectId::from("project-one"), + UserId::from("user-one"), + "concurrent-session", + None, + ) + .unwrap(); + service.handle_stream(stream).unwrap(); + }) + }) + .collect::>(); + for handler in handlers { + handler.join().unwrap(); + } + }); + + let transport = ControlTransport::with_timeouts( + format!("clusterflux+tcp://{address}"), + Duration::from_secs(2), + Duration::from_secs(5), + ) + .unwrap(); + let client = ClusterfluxClient::with_transport(transport) + .with_session_credential(&SessionCredential::from_secret("concurrent-session")); + let (first, second) = tokio::join!(client.account_status(), client.account_status()); + assert!(first.unwrap().authenticated); + assert!(second.unwrap().authenticated); + + drop(client); + server.join().unwrap(); +} diff --git a/crates/clusterflux-coordinator/src/service.rs b/crates/clusterflux-coordinator/src/service.rs index d7472bd..aef24d0 100644 --- a/crates/clusterflux-coordinator/src/service.rs +++ b/crates/clusterflux-coordinator/src/service.rs @@ -284,9 +284,10 @@ pub struct CoordinatorService { process_summaries: BTreeMap, process_summary_order: VecDeque, next_process_summary_order: u64, + task_terminal_states: BTreeMap, recent_logs: BTreeMap<(TenantId, ProjectId), VecDeque>, recent_log_dropped_through: BTreeMap, - recent_log_offsets: BTreeMap< + recent_log_accounted_bytes: BTreeMap< ( TenantId, ProjectId, @@ -296,6 +297,13 @@ pub struct CoordinatorService { ), u64, >, + recent_log_truncated_streams: BTreeSet<( + TenantId, + ProjectId, + ProcessId, + clusterflux_core::TaskInstanceId, + String, + )>, next_recent_log_sequence: u64, debug_audit_events: VecDeque, debug_epochs: BTreeMap, @@ -566,9 +574,11 @@ impl CoordinatorService { process_summaries: BTreeMap::new(), process_summary_order: VecDeque::new(), next_process_summary_order: 1, + task_terminal_states: BTreeMap::new(), recent_logs: BTreeMap::new(), recent_log_dropped_through: BTreeMap::new(), - recent_log_offsets: BTreeMap::new(), + recent_log_accounted_bytes: BTreeMap::new(), + recent_log_truncated_streams: BTreeSet::new(), next_recent_log_sequence: 1, debug_audit_events: VecDeque::new(), debug_epochs: BTreeMap::new(), diff --git a/crates/clusterflux-coordinator/src/service/logs.rs b/crates/clusterflux-coordinator/src/service/logs.rs index 1c84a3b..c717ad7 100644 --- a/crates/clusterflux-coordinator/src/service/logs.rs +++ b/crates/clusterflux-coordinator/src/service/logs.rs @@ -39,40 +39,41 @@ impl CoordinatorService { self.authorize_node_for_process_or_termination(&node, &tenant, &project, &process)?; validate_task_log_tail("stdout_tail", &stdout_tail)?; validate_task_log_tail("stderr_tail", &stderr_tail)?; - let reported_bytes = checked_reported_log_bytes(stdout_bytes, stderr_bytes)?; let now_epoch_seconds = self.current_epoch_seconds()?; + let unaccounted_bytes = self.unaccounted_final_log_bytes( + &tenant, + &project, + &process, + &task, + stdout_bytes, + stderr_bytes, + )?; self.quota - .can_charge_log_bytes(&tenant, &project, reported_bytes, now_epoch_seconds)?; + .can_charge_log_bytes(&tenant, &project, unaccounted_bytes, now_epoch_seconds)?; self.quota - .charge_log_bytes(&tenant, &project, reported_bytes, now_epoch_seconds)?; - let stdout_key = - recent_log_offset_key(&tenant, &project, &process, &task, &TaskLogStream::Stdout); - let stderr_key = - recent_log_offset_key(&tenant, &project, &process, &task, &TaskLogStream::Stderr); - if !stdout_tail.is_empty() && !self.recent_log_offsets.contains_key(&stdout_key) { - self.record_recent_log( - tenant.clone(), - project.clone(), - process.clone(), - task.clone(), - TaskLogStream::Stdout, - stdout_tail.clone(), - stdout_truncated, - now_epoch_seconds, - ); - } - if !stderr_tail.is_empty() && !self.recent_log_offsets.contains_key(&stderr_key) { - self.record_recent_log( - tenant.clone(), - project.clone(), - process.clone(), - task.clone(), - TaskLogStream::Stderr, - stderr_tail.clone(), - stderr_truncated, - now_epoch_seconds, - ); - } + .charge_log_bytes(&tenant, &project, unaccounted_bytes, now_epoch_seconds)?; + self.reconcile_final_log_stream( + &tenant, + &project, + &process, + &task, + TaskLogStream::Stdout, + stdout_bytes, + &stdout_tail, + stdout_truncated, + now_epoch_seconds, + ); + self.reconcile_final_log_stream( + &tenant, + &project, + &process, + &task, + TaskLogStream::Stderr, + stderr_bytes, + &stderr_tail, + stderr_truncated, + now_epoch_seconds, + ); Ok(CoordinatorResponse::TaskLogRecorded { process, task, @@ -129,13 +130,18 @@ impl CoordinatorService { let task = TaskInstanceId::new(task); self.authorize_node_for_process_or_termination(&node, &tenant, &project, &process)?; let key = recent_log_offset_key(&tenant, &project, &process, &task, &stream); - let expected = self.recent_log_offsets.get(&key).copied().unwrap_or(0); + let expected = self + .recent_log_accounted_bytes + .get(&key) + .copied() + .unwrap_or(0); let end = offset.checked_add(source_bytes).ok_or_else(|| { CoordinatorServiceError::Protocol( "live log chunk offset exceeds the supported range".to_owned(), ) })?; - if end <= expected { + let state_marker = source_bytes == 0 && truncated; + if end < expected || (end == expected && !state_marker) { return Ok(CoordinatorResponse::TaskLogChunkRecorded { process, task, @@ -144,6 +150,11 @@ impl CoordinatorService { }); } let now_epoch_seconds = self.current_epoch_seconds()?; + let newly_accounted = end.saturating_sub(expected); + self.quota + .can_charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds)?; + self.quota + .charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds)?; if offset > expected { self.record_recent_log( tenant.clone(), @@ -156,26 +167,46 @@ impl CoordinatorService { now_epoch_seconds, ); } else if offset < expected { - return Ok(CoordinatorResponse::TaskLogChunkRecorded { - process, - task, - sequence: None, - next_offset: expected, - }); - } - let sequence = (!text.is_empty()).then(|| { self.record_recent_log( tenant.clone(), project.clone(), process.clone(), task.clone(), - stream, - text, - truncated, + stream.clone(), + format!( + "[log output overlap omitted: {} new source bytes]", + end - expected + ), + true, now_epoch_seconds, - ) - }); - self.recent_log_offsets.insert(key, end); + ); + } + let marker_is_new = if truncated { + self.recent_log_truncated_streams.insert(key.clone()) + } else { + false + }; + let text = if state_marker && text.is_empty() { + "[log output truncated at source]".to_owned() + } else { + text + }; + let sequence = (!text.is_empty() && offset >= expected && (!state_marker || marker_is_new)) + .then(|| { + self.record_recent_log( + tenant.clone(), + project.clone(), + process.clone(), + task.clone(), + stream, + text, + truncated, + now_epoch_seconds, + ) + }); + if end > expected { + self.recent_log_accounted_bytes.insert(key, end); + } Ok(CoordinatorResponse::TaskLogChunkRecorded { process, task, @@ -302,20 +333,49 @@ impl CoordinatorService { artifact_size_bytes, result, }; - let reported_bytes = checked_reported_log_bytes(event.stdout_bytes, event.stderr_bytes)?; let now_epoch_seconds = self.current_epoch_seconds()?; + let unaccounted_bytes = self.unaccounted_final_log_bytes( + &event.tenant, + &event.project, + &event.process, + &event.task, + event.stdout_bytes, + event.stderr_bytes, + )?; self.quota.can_charge_log_bytes( &event.tenant, &event.project, - reported_bytes, + unaccounted_bytes, now_epoch_seconds, )?; self.quota.charge_log_bytes( &event.tenant, &event.project, - reported_bytes, + unaccounted_bytes, now_epoch_seconds, )?; + self.reconcile_final_log_stream( + &event.tenant, + &event.project, + &event.process, + &event.task, + TaskLogStream::Stdout, + event.stdout_bytes, + &event.stdout_tail, + event.stdout_truncated, + now_epoch_seconds, + ); + self.reconcile_final_log_stream( + &event.tenant, + &event.project, + &event.process, + &event.task, + TaskLogStream::Stderr, + event.stderr_bytes, + &event.stderr_tail, + event.stderr_truncated, + now_epoch_seconds, + ); let task_key = task_control_key( &event.tenant, &event.project, @@ -648,6 +708,22 @@ impl CoordinatorService { pub(super) fn record_task_completion_event(&mut self, mut event: TaskCompletionEvent) { event.stdout_tail = bounded_log_tail(event.stdout_tail, &mut event.stdout_truncated); event.stderr_tail = bounded_log_tail(event.stderr_tail, &mut event.stderr_truncated); + match event.executor { + super::TaskExecutor::CoordinatorMain => self.record_main_terminal_state( + &event.tenant, + &event.project, + &event.process, + event.task_definition.clone(), + event.task.clone(), + event.terminal_state.clone(), + ), + super::TaskExecutor::Node => { + self.task_terminal_states.insert( + task_restart_key(&event.tenant, &event.project, &event.process, &event.task), + event.terminal_state.clone(), + ); + } + } let process_scope = ( event.tenant.clone(), event.project.clone(), @@ -756,6 +832,171 @@ impl CoordinatorService { sequence } + #[allow(clippy::too_many_arguments)] + fn unaccounted_final_log_bytes( + &self, + tenant: &TenantId, + project: &ProjectId, + process: &ProcessId, + task: &TaskInstanceId, + stdout_bytes: u64, + stderr_bytes: u64, + ) -> Result { + let stdout_accounted = self + .recent_log_accounted_bytes + .get(&recent_log_offset_key( + tenant, + project, + process, + task, + &TaskLogStream::Stdout, + )) + .copied() + .unwrap_or(0); + let stderr_accounted = self + .recent_log_accounted_bytes + .get(&recent_log_offset_key( + tenant, + project, + process, + task, + &TaskLogStream::Stderr, + )) + .copied() + .unwrap_or(0); + let stdout_remaining = stdout_bytes.checked_sub(stdout_accounted).ok_or_else(|| { + CoordinatorServiceError::Protocol(format!( + "final stdout byte count {stdout_bytes} is below the {stdout_accounted} live bytes already accounted" + )) + })?; + let stderr_remaining = stderr_bytes.checked_sub(stderr_accounted).ok_or_else(|| { + CoordinatorServiceError::Protocol(format!( + "final stderr byte count {stderr_bytes} is below the {stderr_accounted} live bytes already accounted" + )) + })?; + checked_reported_log_bytes(stdout_remaining, stderr_remaining) + } + + #[allow(clippy::too_many_arguments)] + fn reconcile_final_log_stream( + &mut self, + tenant: &TenantId, + project: &ProjectId, + process: &ProcessId, + task: &TaskInstanceId, + stream: TaskLogStream, + total_source_bytes: u64, + final_tail: &str, + source_truncated: bool, + now_epoch_seconds: u64, + ) { + let key = recent_log_offset_key(tenant, project, process, task, &stream); + let accounted = self + .recent_log_accounted_bytes + .get(&key) + .copied() + .unwrap_or(0); + let mut visible_truncation = false; + if total_source_bytes > accounted { + let missing = total_source_bytes - accounted; + if final_tail.is_empty() { + self.record_recent_log( + tenant.clone(), + project.clone(), + process.clone(), + task.clone(), + stream.clone(), + format!("[log output unavailable: {missing} source bytes]"), + true, + now_epoch_seconds, + ); + visible_truncation = true; + } else if (final_tail.len() as u64) <= total_source_bytes { + let tail_source_start = total_source_bytes - final_tail.len() as u64; + if accounted < tail_source_start { + self.record_recent_log( + tenant.clone(), + project.clone(), + process.clone(), + task.clone(), + stream.clone(), + format!( + "[log output lost before final tail: {} source bytes]", + tail_source_start - accounted + ), + true, + now_epoch_seconds, + ); + visible_truncation = true; + } + let source_start = accounted.max(tail_source_start) - tail_source_start; + let mut byte_start = usize::try_from(source_start) + .unwrap_or(final_tail.len()) + .min(final_tail.len()); + while byte_start < final_tail.len() && !final_tail.is_char_boundary(byte_start) { + byte_start += 1; + } + let suffix = &final_tail[byte_start..]; + if !suffix.is_empty() { + self.record_recent_log( + tenant.clone(), + project.clone(), + process.clone(), + task.clone(), + stream.clone(), + suffix.to_owned(), + source_truncated || visible_truncation, + now_epoch_seconds, + ); + visible_truncation |= source_truncated; + } + } else if accounted == 0 { + self.record_recent_log( + tenant.clone(), + project.clone(), + process.clone(), + task.clone(), + stream.clone(), + final_tail.to_owned(), + source_truncated, + now_epoch_seconds, + ); + visible_truncation |= source_truncated; + } else { + self.record_recent_log( + tenant.clone(), + project.clone(), + process.clone(), + task.clone(), + stream.clone(), + format!( + "[{missing} additional source bytes could not be merged without duplicating redacted output]" + ), + true, + now_epoch_seconds, + ); + visible_truncation = true; + } + self.recent_log_accounted_bytes + .insert(key.clone(), total_source_bytes); + } + + let marker_is_new = (source_truncated || visible_truncation) + && self.recent_log_truncated_streams.insert(key); + if marker_is_new && !visible_truncation { + self.record_recent_log( + tenant.clone(), + project.clone(), + process.clone(), + task.clone(), + stream, + "[log output truncated at source]".to_owned(), + true, + now_epoch_seconds, + ); + } + } + pub(super) fn clear_recent_log_offsets_for_task( &mut self, tenant: &TenantId, @@ -763,7 +1004,7 @@ impl CoordinatorService { process: &ProcessId, task: &TaskInstanceId, ) { - self.recent_log_offsets.retain( + self.recent_log_accounted_bytes.retain( |(entry_tenant, entry_project, entry_process, entry_task, _), _| { entry_tenant != tenant || entry_project != project @@ -771,6 +1012,14 @@ impl CoordinatorService { || entry_task != task }, ); + self.recent_log_truncated_streams.retain( + |(entry_tenant, entry_project, entry_process, entry_task, _)| { + entry_tenant != tenant + || entry_project != project + || entry_process != process + || entry_task != task + }, + ); } fn finish_task_attempt(&mut self, event: &mut TaskCompletionEvent) -> bool { @@ -856,13 +1105,11 @@ impl CoordinatorService { return Ok(false); } - let main_completed = self.task_events.iter().rev().any(|event| { - &event.tenant == tenant - && &event.project == project - && &event.process == process - && matches!(event.executor, super::TaskExecutor::CoordinatorMain) - && matches!(event.terminal_state, TaskTerminalState::Completed) - }); + let main_completed = self + .process_summaries + .get(&process_key) + .and_then(|summary| summary.main_terminal_state.as_ref()) + .is_some_and(|state| matches!(state, TaskTerminalState::Completed)); let cancellation_completed = self.process_cancellations.contains(&process_key); if !main_completed && !cancellation_completed { return Ok(false); @@ -879,13 +1126,24 @@ impl CoordinatorService { } let final_result = if cancellation_completed { super::ProcessFinalResult::Cancelled - } else if self.task_events.iter().any(|event| { - &event.tenant == tenant - && &event.project == project - && &event.process == process - && event.terminal_state == TaskTerminalState::Failed - }) { + } else if self.task_terminal_states.iter().any( + |((task_tenant, task_project, task_process, _), terminal_state)| { + task_tenant == tenant + && task_project == project + && task_process == process + && terminal_state == &TaskTerminalState::Failed + }, + ) { super::ProcessFinalResult::Failed + } else if self.task_terminal_states.iter().any( + |((task_tenant, task_project, task_process, _), terminal_state)| { + task_tenant == tenant + && task_project == project + && task_process == process + && terminal_state == &TaskTerminalState::Cancelled + }, + ) { + super::ProcessFinalResult::Cancelled } else { super::ProcessFinalResult::Completed }; diff --git a/crates/clusterflux-coordinator/src/service/process_launch.rs b/crates/clusterflux-coordinator/src/service/process_launch.rs index fc6b69c..81737c3 100644 --- a/crates/clusterflux-coordinator/src/service/process_launch.rs +++ b/crates/clusterflux-coordinator/src/service/process_launch.rs @@ -671,7 +671,9 @@ impl CoordinatorService { )); }; self.task_attempts.remove(&removable); + self.task_terminal_states.remove(&removable); } + self.task_terminal_states.remove(&key); let attempts = self.task_attempts.entry(key).or_default(); for attempt in attempts.iter_mut() { attempt.current = false; diff --git a/crates/clusterflux-coordinator/src/service/processes.rs b/crates/clusterflux-coordinator/src/service/processes.rs index ec6ca29..646c607 100644 --- a/crates/clusterflux-coordinator/src/service/processes.rs +++ b/crates/clusterflux-coordinator/src/service/processes.rs @@ -382,6 +382,10 @@ impl CoordinatorService { || attempt_project != &project || attempt_process != &process }); + self.task_terminal_states + .retain(|(task_tenant, task_project, task_process, _), _| { + task_tenant != &tenant || task_project != &project || task_process != &process + }); self.restart_launches .retain(|(attempt_tenant, attempt_project, attempt_process, _)| { attempt_tenant != &tenant @@ -671,10 +675,18 @@ impl CoordinatorService { .map(|active| { let process_key = process_control_key(&active.tenant, &active.project, &active.id); let main = self.main_runtime.controls.get(&process_key); + let stored = self.process_summaries.get(&process_key); + let stored_main_state = stored + .and_then(|summary| summary.main_terminal_state.as_ref()) + .map(|state| match state { + super::TaskTerminalState::Completed => "completed", + super::TaskTerminalState::Failed => "failed", + super::TaskTerminalState::Cancelled => "cancelled", + }); let state = if self.process_cancellations.contains(&process_key) { "cancelling" } else { - main.map_or("running", |main| main.state.as_str()) + "running" }; let main_wait_state = main.and_then(|main| { if main.state != "running" { @@ -699,9 +711,15 @@ impl CoordinatorService { VirtualProcessStatus { process: active.id, state: state.to_owned(), - main_task_definition: main.map(|main| main.task_definition.clone()), - main_task_instance: main.map(|main| main.task_instance.clone()), - main_state: main.map(|main| main.state.clone()), + main_task_definition: main.map(|main| main.task_definition.clone()).or_else( + || stored.and_then(|summary| summary.main_task_definition.clone()), + ), + main_task_instance: main + .map(|main| main.task_instance.clone()) + .or_else(|| stored.and_then(|summary| summary.main_task_instance.clone())), + main_state: main + .map(|main| main.state.clone()) + .or_else(|| stored_main_state.map(str::to_owned)), main_wait_state, main_debug_epoch: main.and_then(|main| main.debug.requested_epoch()), connected_nodes: active.connected_nodes.into_iter().collect(), diff --git a/crates/clusterflux-coordinator/src/service/summaries.rs b/crates/clusterflux-coordinator/src/service/summaries.rs index c8dc59a..cd54a2d 100644 --- a/crates/clusterflux-coordinator/src/service/summaries.rs +++ b/crates/clusterflux-coordinator/src/service/summaries.rs @@ -2,7 +2,8 @@ use std::collections::BTreeSet; use std::time::Instant; use clusterflux_core::{ - ArtifactId, ArtifactMetadata, NodeId, ProcessId, ProjectId, TenantId, UserId, + ArtifactId, ArtifactMetadata, NodeId, ProcessId, ProjectId, TaskDefinitionId, TaskInstanceId, + TenantId, UserId, }; use super::keys::{process_control_key, ProcessControlKey}; @@ -19,6 +20,9 @@ pub(super) struct StoredProcessSummary { pub(super) ended_at_epoch_seconds: Option, pub(super) final_result: Option, pub(super) connected_nodes: Vec, + pub(super) main_task_definition: Option, + pub(super) main_task_instance: Option, + pub(super) main_terminal_state: Option, pub(super) order: u64, } @@ -84,10 +88,16 @@ impl CoordinatorService { } } self.recent_log_dropped_through.remove(&key); - self.recent_log_offsets - .retain(|(entry_tenant, entry_project, entry_process, _, _), _| { + self.recent_log_accounted_bytes.retain( + |(entry_tenant, entry_project, entry_process, _, _), _| { entry_tenant != tenant || entry_project != project || entry_process != process - }); + }, + ); + self.recent_log_truncated_streams.retain( + |(entry_tenant, entry_project, entry_process, _, _)| { + entry_tenant != tenant || entry_project != project || entry_process != process + }, + ); self.process_summary_order .retain(|retained| retained != &key); self.evict_process_summaries_for_project(tenant, project); @@ -101,6 +111,9 @@ impl CoordinatorService { ended_at_epoch_seconds: None, final_result: None, connected_nodes: Vec::new(), + main_task_definition: None, + main_task_instance: None, + main_terminal_state: None, order, }, ); @@ -133,6 +146,9 @@ impl CoordinatorService { ended_at_epoch_seconds: None, final_result: None, connected_nodes: Vec::new(), + main_task_definition: None, + main_task_instance: None, + main_terminal_state: None, order, } }); @@ -141,6 +157,31 @@ impl CoordinatorService { entry.connected_nodes = connected_nodes; } + pub(super) fn record_main_terminal_state( + &mut self, + tenant: &TenantId, + project: &ProjectId, + process: &ProcessId, + task_definition: TaskDefinitionId, + task_instance: TaskInstanceId, + terminal_state: super::TaskTerminalState, + ) { + let key = process_control_key(tenant, project, process); + if !self.process_summaries.contains_key(&key) { + self.record_process_started( + tenant, + project, + process, + self.liveness_now_epoch_seconds(), + ); + } + if let Some(summary) = self.process_summaries.get_mut(&key) { + summary.main_task_definition = Some(task_definition); + summary.main_task_instance = Some(task_instance); + summary.main_terminal_state = Some(terminal_state); + } + } + pub(super) fn handle_list_process_summaries( &mut self, tenant: String, @@ -449,10 +490,7 @@ impl CoordinatorService { let Some(candidate) = candidate.cloned() else { break; }; - self.process_summaries.remove(&candidate); - self.recent_log_dropped_through.remove(&candidate); - self.process_summary_order - .retain(|retained| retained != &candidate); + self.remove_process_summary_state(&candidate); } } @@ -466,12 +504,30 @@ impl CoordinatorService { let Some(candidate) = candidate.cloned() else { break; }; - self.process_summaries.remove(&candidate); - self.recent_log_dropped_through.remove(&candidate); - self.process_summary_order - .retain(|retained| retained != &candidate); + self.remove_process_summary_state(&candidate); } } + + fn remove_process_summary_state(&mut self, key: &super::ProcessControlKey) { + self.process_summaries.remove(key); + self.recent_log_dropped_through.remove(key); + if let Some(logs) = self.recent_logs.get_mut(&(key.0.clone(), key.1.clone())) { + logs.retain(|entry| entry.process != key.2); + if logs.is_empty() { + self.recent_logs.remove(&(key.0.clone(), key.1.clone())); + } + } + self.recent_log_accounted_bytes + .retain(|(tenant, project, process, _, _), _| { + tenant != &key.0 || project != &key.1 || process != &key.2 + }); + self.recent_log_truncated_streams + .retain(|(tenant, project, process, _, _)| { + tenant != &key.0 || project != &key.1 || process != &key.2 + }); + self.process_summary_order + .retain(|retained| retained != key); + } } fn parse_order_cursor( diff --git a/crates/clusterflux-coordinator/src/service/tests.rs b/crates/clusterflux-coordinator/src/service/tests.rs index 91fc7b3..659537a 100644 --- a/crates/clusterflux-coordinator/src/service/tests.rs +++ b/crates/clusterflux-coordinator/src/service/tests.rs @@ -2501,10 +2501,10 @@ fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() { process: "process".to_owned(), node: "node".to_owned(), task: "task".to_owned(), - stdout_bytes: 1, - stderr_bytes: 0, - stdout_tail: "x".to_owned(), - stderr_tail: String::new(), + stdout_bytes: 4, + stderr_bytes: 1, + stdout_tail: "outx".to_owned(), + stderr_tail: "e".to_owned(), stdout_truncated: false, stderr_truncated: false, backpressured: true, @@ -4474,6 +4474,63 @@ fn completed_main_unpolled_final_assignment_completion_retires_process() { .expect("unpolled terminal completion must release the one-process slot"); } +#[test] +fn completed_main_retires_from_authoritative_state_after_event_history_rotates() { + let mut service = + service_with_completed_main_and_final_child(clusterflux_core::TaskFailurePolicy::FailFast); + let tenant = TenantId::from("tenant"); + let project = ProjectId::from("project"); + let process = ProcessId::from("terminal-matrix"); + + for index in 0..=MAX_TASK_EVENTS_PER_PROCESS { + service.record_task_completion_event(TaskCompletionEvent { + tenant: tenant.clone(), + project: project.clone(), + process: process.clone(), + node: NodeId::from("worker"), + executor: TaskExecutor::Node, + task_definition: TaskDefinitionId::from("historical"), + task: TaskInstanceId::new(format!("historical-{index}")), + 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, + }); + } + assert!( + service.task_events.iter().all(|event| { + event.process != process || event.executor != TaskExecutor::CoordinatorMain + }), + "the regression requires bounded history to have rotated the main event" + ); + + complete_terminal_matrix_child(&mut service, TaskTerminalState::Completed); + + assert!(service + .coordinator + .active_process(&tenant, &project, &process) + .is_none()); + let summary = service + .process_summaries + .get(&process_control_key(&tenant, &project, &process)) + .expect("the terminal process summary must remain authoritative"); + assert_eq!(summary.final_result, Some(ProcessFinalResult::Completed)); + assert_eq!( + summary.main_terminal_state, + Some(TaskTerminalState::Completed) + ); +} + #[test] fn completed_main_await_operator_blocks_retirement_until_each_resolution() { for resolution in [ @@ -4634,6 +4691,14 @@ fn completed_main_failed_child_restarted_successfully_retires_with_successful_cu assert!(attempts .iter() .any(|attempt| attempt.current && attempt.state == TaskAttemptState::Completed)); + assert_eq!( + service + .process_summaries + .get(&process_control_key(&tenant, &project, &process)) + .and_then(|summary| summary.final_result.clone()), + Some(ProcessFinalResult::Completed), + "a successful current retry must override the superseded failed attempt" + ); assert_eq!( service .task_join_result(tenant, project, process, task) @@ -8182,6 +8247,54 @@ fn web_process_summaries_are_scoped_paginated_and_retain_authoritative_terminal_ assert!(oversized.to_string().contains("100")); } +#[test] +fn process_summary_eviction_releases_live_log_accounting_state() { + let mut service = CoordinatorService::new(7); + let tenant = TenantId::from("tenant-summary-bound"); + let project = ProjectId::from("project-summary-bound"); + let task = TaskInstanceId::from("task"); + + for index in 0..MAX_RECENT_PROCESS_SUMMARIES_PER_PROJECT { + let process = ProcessId::new(format!("process-{index:03}")); + service.record_process_started(&tenant, &project, &process, index as u64); + service.record_process_terminal( + &tenant, + &project, + &process, + ProcessFinalResult::Completed, + index as u64 + 1, + ); + let key = ( + tenant.clone(), + project.clone(), + process, + task.clone(), + "stdout".to_owned(), + ); + service.recent_log_accounted_bytes.insert(key.clone(), 10); + service.recent_log_truncated_streams.insert(key); + } + + let evicted = ProcessId::from("process-000"); + service.record_process_started(&tenant, &project, &ProcessId::from("process-next"), 1_000); + + assert!(!service.process_summaries.contains_key(&( + tenant.clone(), + project.clone(), + evicted.clone() + ))); + assert!(!service.recent_log_accounted_bytes.keys().any( + |(entry_tenant, entry_project, process, _, _)| { + entry_tenant == &tenant && entry_project == &project && process == &evicted + } + )); + assert!(!service.recent_log_truncated_streams.iter().any( + |(entry_tenant, entry_project, process, _, _)| { + entry_tenant == &tenant && entry_project == &project && process == &evicted + } + )); +} + #[test] fn web_node_summaries_are_scoped_paginated_and_hard_bounded() { let mut service = CoordinatorService::new(7); @@ -8560,6 +8673,15 @@ fn web_recent_logs_are_signed_scoped_cursor_safe_and_memory_bounded() { else { panic!("expected first live log sequence"); }; + assert_eq!( + service.quota.used_log_bytes( + &TenantId::from("tenant-a"), + &ProjectId::from("project-a"), + 100, + ), + 5, + "live bytes must be charged when accepted" + ); let retry = service .handle_signed_node_request_auto(CoordinatorRequest::ReportTaskLogChunk { tenant: "tenant-a".to_owned(), @@ -8578,6 +8700,15 @@ fn web_recent_logs_are_signed_scoped_cursor_safe_and_memory_bounded() { retry, CoordinatorResponse::TaskLogChunkRecorded { sequence: None, .. } )); + assert_eq!( + service.quota.used_log_bytes( + &TenantId::from("tenant-a"), + &ProjectId::from("project-a"), + 100, + ), + 5, + "a retried chunk must not be charged twice" + ); service .handle_signed_node_request_auto(CoordinatorRequest::ReportTaskLogChunk { tenant: "tenant-a".to_owned(), @@ -8592,6 +8723,15 @@ fn web_recent_logs_are_signed_scoped_cursor_safe_and_memory_bounded() { truncated: false, }) .unwrap(); + assert_eq!( + service.quota.used_log_bytes( + &TenantId::from("tenant-a"), + &ProjectId::from("project-a"), + 100, + ), + 10, + "a gap and the delivered bytes must both count toward source-byte usage" + ); let CoordinatorResponse::RecentLogs { entries, @@ -8633,6 +8773,113 @@ fn web_recent_logs_are_signed_scoped_cursor_safe_and_memory_bounded() { assert_eq!(entries.len(), 1); assert_eq!(entries[0].text, "ok"); + service + .handle_report_task_log( + "tenant-a".to_owned(), + "project-a".to_owned(), + "process-shared".to_owned(), + "node-a".to_owned(), + "task-one".to_owned(), + 12, + 0, + "hello???okZZ".to_owned(), + String::new(), + false, + false, + false, + ) + .unwrap(); + assert_eq!( + service.quota.used_log_bytes( + &TenantId::from("tenant-a"), + &ProjectId::from("project-a"), + 100, + ), + 12, + "the final summary must charge only source bytes not already charged live" + ); + assert_eq!( + service + .recent_logs + .get(&(TenantId::from("tenant-a"), ProjectId::from("project-a"))) + .unwrap() + .back() + .unwrap() + .text, + "ZZ", + "final-tail reconciliation must append only the nonduplicating suffix" + ); + service + .handle_report_task_log( + "tenant-a".to_owned(), + "project-a".to_owned(), + "process-shared".to_owned(), + "node-a".to_owned(), + "task-one".to_owned(), + 12, + 0, + "hello???okZZ".to_owned(), + String::new(), + false, + false, + false, + ) + .unwrap(); + assert_eq!( + service.quota.used_log_bytes( + &TenantId::from("tenant-a"), + &ProjectId::from("project-a"), + 100, + ), + 12, + "replayed final accounting must be idempotent" + ); + + let marker = service + .handle_report_task_log_chunk( + "tenant-a".to_owned(), + "project-a".to_owned(), + "process-shared".to_owned(), + "node-a".to_owned(), + "task-one".to_owned(), + TaskLogStream::Stdout, + 12, + 0, + "[log output truncated at node capture limit]".to_owned(), + true, + ) + .unwrap(); + assert!(matches!( + marker, + CoordinatorResponse::TaskLogChunkRecorded { + sequence: Some(_), + next_offset: 12, + .. + } + )); + let repeated_marker = service + .handle_report_task_log_chunk( + "tenant-a".to_owned(), + "project-a".to_owned(), + "process-shared".to_owned(), + "node-a".to_owned(), + "task-one".to_owned(), + TaskLogStream::Stdout, + 12, + 0, + "[log output truncated at node capture limit]".to_owned(), + true, + ) + .unwrap(); + assert!(matches!( + repeated_marker, + CoordinatorResponse::TaskLogChunkRecorded { + sequence: None, + next_offset: 12, + .. + } + )); + let cross_tenant = service .handle_request(CoordinatorRequest::Authenticated { session_secret: "session-b".to_owned(), diff --git a/crates/clusterflux-core/src/artifact.rs b/crates/clusterflux-core/src/artifact.rs index 42970bd..d9c89a3 100644 --- a/crates/clusterflux-core/src/artifact.rs +++ b/crates/clusterflux-core/src/artifact.rs @@ -195,7 +195,34 @@ impl ArtifactRegistry { protected_processes: &BTreeSet, ) -> Result { let key = ArtifactScopeKey::from_refs(&flush.tenant, &flush.project, &flush.id); - self.next_epoch += 1; + let replacing = self.artifacts.contains_key(&key); + let retained_for_project = self + .artifacts + .values() + .filter(|metadata| metadata.tenant == flush.tenant && metadata.project == flush.project) + .count(); + let eviction = if replacing || retained_for_project < MAX_ARTIFACT_METADATA_PER_PROJECT { + None + } else { + let mut protected_processes = protected_processes.clone(); + protected_processes.insert(flush.process.clone()); + Some( + self.project_metadata_eviction_candidate( + &flush.tenant, + &flush.project, + pinned, + &protected_processes, + ) + .ok_or_else(|| { + format!( + "artifact metadata capacity of {MAX_ARTIFACT_METADATA_PER_PROJECT} is \ + exhausted by active or retained artifacts" + ) + })?, + ) + }; + + self.next_epoch = self.next_epoch.saturating_add(1); let metadata = ArtifactMetadata { id: flush.id.clone(), tenant: flush.tenant, @@ -210,15 +237,10 @@ impl ArtifactRegistry { explicit_locations: Vec::new(), coordinator_has_large_bytes: false, }; + if let Some(eviction) = eviction { + self.artifacts.remove(&eviction); + } 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) } @@ -237,30 +259,12 @@ impl ArtifactRegistry { .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 candidate = self.project_metadata_eviction_candidate( + tenant, + project, + pinned, + protected_processes, + ); let Some(candidate) = candidate else { break; }; @@ -270,6 +274,38 @@ impl ArtifactRegistry { evicted } + fn project_metadata_eviction_candidate( + &self, + tenant: &TenantId, + project: &ProjectId, + pinned: &BTreeSet, + protected_processes: &BTreeSet, + ) -> Option { + 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) + }) + } + pub fn sync_to_explicit_store( &mut self, tenant: &TenantId, @@ -1485,6 +1521,87 @@ mod tests { ); } + #[test] + fn artifact_metadata_capacity_rejects_atomically_when_every_entry_is_protected() { + let mut registry = ArtifactRegistry::default(); + let tenant = TenantId::from("tenant"); + let project = ProjectId::from("project"); + let active_process = ProcessId::from("active-process"); + for index in 0..MAX_ARTIFACT_METADATA_PER_PROJECT { + registry.flush_metadata(ArtifactFlush { + id: ArtifactId::new(format!("artifact-{index}")), + tenant: tenant.clone(), + project: project.clone(), + process: active_process.clone(), + producer_task: TaskInstanceId::new(format!("task-{index}")), + retaining_node: NodeId::from("node"), + digest: Digest::sha256(format!("content-{index}")), + size: 1, + }); + } + + let replacement = ArtifactFlush { + id: ArtifactId::from("artifact-0"), + tenant: tenant.clone(), + project: project.clone(), + process: active_process.clone(), + producer_task: TaskInstanceId::from("replacement-task"), + retaining_node: NodeId::from("node"), + digest: Digest::sha256("replacement"), + size: 2, + }; + registry + .flush_metadata_with_protected_processes( + replacement, + &BTreeSet::new(), + &BTreeSet::from([active_process.clone()]), + ) + .expect("replacement must remain possible at the metadata bound"); + assert_eq!( + registry + .metadata(&tenant, &project, &ArtifactId::from("artifact-0")) + .unwrap() + .digest, + Digest::sha256("replacement") + ); + + let epoch_before_rejection = registry.next_epoch; + let result = registry.flush_metadata_with_protected_processes( + ArtifactFlush { + id: ArtifactId::from("artifact-rejected"), + tenant: tenant.clone(), + project: project.clone(), + process: active_process.clone(), + producer_task: TaskInstanceId::from("rejected-task"), + retaining_node: NodeId::from("node"), + digest: Digest::sha256("rejected"), + size: 3, + }, + &BTreeSet::new(), + &BTreeSet::from([active_process]), + ); + + assert!(result + .unwrap_err() + .contains("exhausted by active or retained artifacts")); + assert_eq!(registry.next_epoch, epoch_before_rejection); + assert_eq!( + registry.metadata_for_project(&tenant, &project).count(), + MAX_ARTIFACT_METADATA_PER_PROJECT + ); + assert!(registry + .metadata(&tenant, &project, &ArtifactId::from("artifact-rejected")) + .is_none()); + assert_eq!( + registry + .metadata(&tenant, &project, &ArtifactId::from("artifact-0")) + .unwrap() + .digest, + Digest::sha256("replacement"), + "rejection must not modify existing metadata" + ); + } + #[test] fn download_stream_accounts_usage_before_and_during_streaming() { let mut registry = registry_with_artifact(); diff --git a/crates/clusterflux-dap/src/runtime_client.rs b/crates/clusterflux-dap/src/runtime_client.rs index 233b6bc..9b483ac 100644 --- a/crates/clusterflux-dap/src/runtime_client.rs +++ b/crates/clusterflux-dap/src/runtime_client.rs @@ -742,14 +742,16 @@ fn fetch_current_process_status( client_user_request( state, json!({ - "type": "list_processes", + "type": "list_process_summaries", "tenant": state.tenant, "project": state.project_id, "actor_user": state.actor_user, + "cursor": null, + "limit": 100, }), ), )?; - let current = statuses + let summary = statuses .get("processes") .and_then(Value::as_array) .and_then(|processes| { @@ -758,6 +760,9 @@ fn fetch_current_process_status( }) }) .cloned(); + let current = merge_active_process_status(state, summary, |request| { + coordinator_request(coordinator, request) + })?; Ok((statuses, current)) } @@ -768,13 +773,15 @@ fn fetch_current_process_status_in( let statuses = session.request(client_user_request( state, json!({ - "type": "list_processes", + "type": "list_process_summaries", "tenant": state.tenant, "project": state.project_id, "actor_user": state.actor_user, + "cursor": null, + "limit": 100, }), ))?; - let current = statuses + let summary = statuses .get("processes") .and_then(Value::as_array) .and_then(|processes| { @@ -783,9 +790,51 @@ fn fetch_current_process_status_in( }) }) .cloned(); + let current = merge_active_process_status(state, summary, |request| session.request(request))?; Ok((statuses, current)) } +fn merge_active_process_status( + state: &AdapterState, + summary: Option, + mut request: F, +) -> Result> +where + F: FnMut(Value) -> Result, +{ + let Some(mut summary) = summary else { + return Ok(None); + }; + if summary.get("lifecycle").and_then(Value::as_str) != Some("active") { + return Ok(Some(summary)); + } + let active_statuses = request(client_user_request( + state, + json!({ + "type": "list_processes", + "tenant": state.tenant, + "project": state.project_id, + "actor_user": state.actor_user, + }), + ))?; + let active = active_statuses + .get("processes") + .and_then(Value::as_array) + .and_then(|processes| { + processes.iter().find(|process| { + process.get("process").and_then(Value::as_str) == Some(state.process.as_str()) + }) + }); + if let (Some(summary), Some(active)) = + (summary.as_object_mut(), active.and_then(Value::as_object)) + { + for (key, value) in active { + summary.entry(key.clone()).or_insert_with(|| value.clone()); + } + } + Ok(Some(summary)) +} + pub(crate) fn relaunch_services_main_runtime(state: &AdapterState) -> Result { let coordinator = crate::view_state::normalize_coordinator_endpoint(&state.coordinator_endpoint); @@ -859,27 +908,7 @@ pub(crate) fn attach_services_runtime(state: &AdapterState) -> Result snapshots, - Err(error) => { - if !emit(RuntimeContinuationOutcome::Diagnostic(format!( - "runtime terminal snapshot observation failed: {error:#}; reconnecting in {} ms", - reconnect_delay.as_millis() - ))) { - return Ok(()); - } - session = None; - std::thread::sleep(reconnect_delay); - reconnect_delay = (reconnect_delay * 2).min(Duration::from_secs(5)); - continue; - } - }; - let process_status_request = - if inject_process_status_failure && !process_status_failure_injected { - process_status_failure_injected = true; - Err(anyhow!("injected process-status transport failure")) - } else { - fetch_current_process_status_in(current_session, state) - }; - let (process_statuses, process_status) = match process_status_request { - Ok(status) => status, - Err(error) => { - if !emit(RuntimeContinuationOutcome::Diagnostic(format!( - "runtime terminal process observation failed: {error:#}; reconnecting in {} ms", - reconnect_delay.as_millis() - ))) { - return Ok(()); - } - session = None; - std::thread::sleep(reconnect_delay); - reconnect_delay = (reconnect_delay * 2).min(Duration::from_secs(5)); - continue; - } - }; - 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, - "process_status": process_status, - "process_statuses": process_statuses, - }); - } - emit(outcome); - return Ok(()); - } - } let snapshot_request = if inject_snapshot_failure && !snapshot_failure_injected { snapshot_failure_injected = true; Err(anyhow!("injected task snapshot transport failure")) @@ -1221,6 +1190,22 @@ pub(crate) fn observe_services_runtime( } }; reconnect_delay = Duration::from_millis(100); + if !has_current_runtime_task(&task_snapshots) { + if let Some(mut outcome) = + terminal_runtime_outcome(&coordinator, state, &events, process_status.as_ref()) + { + if let RuntimeContinuationOutcome::Terminal(record) = &mut outcome { + record.node_report = json!({ + "terminal_event": record.node_report.get("terminal_event"), + "task_snapshots": task_snapshots, + "process_status": process_status, + "process_statuses": process_statuses, + }); + } + emit(outcome); + return Ok(()); + } + } if let Some((failed_task, _attempt_id)) = failed_awaiting_action_snapshot(&task_snapshots) { let failed_task = failed_task.to_owned(); let failed_event = events @@ -1352,7 +1337,9 @@ pub(crate) fn observe_services_runtime( continue; } }; - if let Some(mut outcome) = terminal_runtime_outcome(&coordinator, state, &events) { + if let Some(mut outcome) = + terminal_runtime_outcome(&coordinator, state, &events, process_status.as_ref()) + { let task_snapshots = match fetch_task_snapshots_in(current_session, state) { Ok(snapshots) => snapshots, Err(snapshot_error) => { @@ -1370,8 +1357,6 @@ 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, @@ -1547,6 +1532,7 @@ fn has_current_runtime_task(task_snapshots: &Value) -> bool { }) } +#[cfg(test)] pub(crate) fn whole_process_status_code( main_status_code: Option, task_snapshots: &Value, @@ -1582,66 +1568,75 @@ pub(crate) fn whole_process_status_code( fn terminal_runtime_outcome( coordinator: &str, - state: &AdapterState, + _state: &AdapterState, events: &Value, + process_status: Option<&Value>, ) -> Option { + let final_result = process_status? + .get("final_result") + .and_then(Value::as_str)?; + let status_code = match final_result { + "completed" => 0, + "failed" | "cancelled" => 1, + _ => return None, + }; let event = events .get("events") - .and_then(Value::as_array)? - .get(state.runtime_event_count..)? - .iter() - .rev() - .find(|event| event.get("executor").and_then(Value::as_str) == Some("coordinator_main"))?; - let status_code = event - .get("status_code") - .and_then(Value::as_i64) - .map(|status| status as i32) - .or_else( - || match event.get("terminal_state").and_then(Value::as_str) { - Some("completed") => Some(0), - Some("failed" | "cancelled") => Some(1), - _ => None, - }, - ); + .and_then(Value::as_array) + .and_then(|events| { + events.iter().rev().find(|event| { + event.get("executor").and_then(Value::as_str) == Some("coordinator_main") + }) + }); Some(RuntimeContinuationOutcome::Terminal(RuntimeLaunchRecord { coordinator: coordinator.to_owned(), node: event - .get("node") + .and_then(|event| event.get("node")) .and_then(Value::as_str) + .or_else(|| { + process_status + .and_then(|status| status.get("connected_nodes")) + .and_then(Value::as_array) + .and_then(|nodes| nodes.first()) + .and_then(Value::as_str) + }) .unwrap_or("coordinator-main") .to_owned(), - node_report: json!({ "terminal_event": event }), + node_report: json!({ + "terminal_event": event, + "process_status": process_status, + }), task_events: events.clone(), placed_task_launched: true, - status_code, + status_code: Some(status_code), stdout_bytes: event - .get("stdout_bytes") + .and_then(|event| event.get("stdout_bytes")) .and_then(Value::as_u64) .unwrap_or(0), stderr_bytes: event - .get("stderr_bytes") + .and_then(|event| event.get("stderr_bytes")) .and_then(Value::as_u64) .unwrap_or(0), stdout_tail: event - .get("stdout_tail") + .and_then(|event| event.get("stdout_tail")) .and_then(Value::as_str) .unwrap_or_default() .to_owned(), stderr_tail: event - .get("stderr_tail") + .and_then(|event| event.get("stderr_tail")) .and_then(Value::as_str) .unwrap_or_default() .to_owned(), stdout_truncated: event - .get("stdout_truncated") + .and_then(|event| event.get("stdout_truncated")) .and_then(Value::as_bool) .unwrap_or(false), stderr_truncated: event - .get("stderr_truncated") + .and_then(|event| event.get("stderr_truncated")) .and_then(Value::as_bool) .unwrap_or(false), artifact_path: event - .get("artifact_path") + .and_then(|event| event.get("artifact_path")) .and_then(Value::as_str) .map(str::to_owned), event_count: events @@ -1727,6 +1722,53 @@ mod transactional_launch_tests { assert!(error.contains("no virtual process was created")); } + #[test] + fn durable_process_summary_is_terminal_authority_after_event_rotation() { + let state = AdapterState { + runtime_event_count: 10_000, + ..AdapterState::default() + }; + let completed = terminal_runtime_outcome( + "127.0.0.1:1", + &state, + &json!({ "events": [] }), + Some(&json!({ + "process": state.process.as_str(), + "lifecycle": "recent_terminal", + "final_result": "completed", + "connected_nodes": [] + })), + ) + .expect("the durable summary should terminate observation"); + let RuntimeContinuationOutcome::Terminal(completed) = completed else { + panic!("expected terminal outcome"); + }; + assert_eq!(completed.status_code, Some(0)); + + let failed = terminal_runtime_outcome( + "127.0.0.1:1", + &state, + &json!({ + "events": [{ + "executor": "coordinator_main", + "terminal_state": "completed", + "status_code": 0 + }] + }), + Some(&json!({ + "process": state.process.as_str(), + "lifecycle": "recent_terminal", + "final_result": "failed", + "connected_nodes": [] + })), + ) + .expect("the aggregate summary should override a successful main event"); + let RuntimeContinuationOutcome::Terminal(failed) = failed else { + panic!("expected terminal outcome"); + }; + assert_eq!(failed.status_code, Some(1)); + } + #[test] fn failed_debug_launch_reconnects_and_aborts_the_process() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); diff --git a/crates/clusterflux-node/src/assignment_runner.rs b/crates/clusterflux-node/src/assignment_runner.rs index dc63a24..54f5a00 100644 --- a/crates/clusterflux-node/src/assignment_runner.rs +++ b/crates/clusterflux-node/src/assignment_runner.rs @@ -2,7 +2,7 @@ use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::io::Read; use std::path::PathBuf; use std::process::Stdio; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::thread; use std::time::{Duration, Instant}; @@ -46,6 +46,40 @@ use validation::{ resolve_task_export, task_descriptors, verify_environment_digest, verify_source_snapshot, }; +#[derive(Debug)] +struct AssignmentExecutionError { + message: String, + stdout_source_bytes: u64, + stderr_source_bytes: u64, +} + +impl std::fmt::Display for AssignmentExecutionError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(&self.message) + } +} + +impl std::error::Error for AssignmentExecutionError {} + +fn execution_error_with_log_bytes( + message: impl Into, + stdout_source_bytes: &AtomicU64, + stderr_source_bytes: &AtomicU64, +) -> Box { + Box::new(AssignmentExecutionError { + message: message.into(), + stdout_source_bytes: stdout_source_bytes.load(Ordering::Relaxed), + stderr_source_bytes: stderr_source_bytes.load(Ordering::Relaxed), + }) +} + +pub(crate) fn assignment_error_log_bytes(error: &(dyn std::error::Error + 'static)) -> (u64, u64) { + error + .downcast_ref::() + .map(|error| (error.stdout_source_bytes, error.stderr_source_bytes)) + .unwrap_or((0, 0)) +} + pub(crate) fn run_verified_wasmtime_assignment( args: &Args, task: &RuntimeTask, @@ -95,6 +129,8 @@ pub(crate) fn run_verified_wasmtime_assignment( return Err("Wasm entrypoint assignment omitted its descriptor export".into()); } }; + let command_stdout_source_bytes = Arc::new(AtomicU64::new(0)); + let command_stderr_source_bytes = Arc::new(AtomicU64::new(0)); let (stdout, boundary_result) = match abi { WasmExportAbi::EntrypointV1 | WasmExportAbi::TaskV1 => { let invocation = WasmTaskInvocation::new( @@ -102,18 +138,28 @@ pub(crate) fn run_verified_wasmtime_assignment( task_spec.task_instance.clone(), task_spec.args.clone(), ); - let result = WasmtimeTaskRuntime::new()?.run_task_export_verified_with_task_host( - &module, - expected_bundle_digest, - export, - &invocation, - Box::new(CoordinatorWasmTaskHost::new( - args, - task, - node_private_key, + let result = WasmtimeTaskRuntime::new()? + .run_task_export_verified_with_task_host( &module, - )?), - )?; + expected_bundle_digest, + export, + &invocation, + Box::new(CoordinatorWasmTaskHost::new( + args, + task, + node_private_key, + &module, + Arc::clone(&command_stdout_source_bytes), + Arc::clone(&command_stderr_source_bytes), + )?), + ) + .map_err(|error| { + execution_error_with_log_bytes( + error.to_string(), + &command_stdout_source_bytes, + &command_stderr_source_bytes, + ) + })?; if std::env::var_os("CLUSTERFLUX_DEBUG_CONTROL_TRACE").is_some() { eprintln!( "clusterflux debug control: Wasm assignment returned for task {}", @@ -122,17 +168,35 @@ pub(crate) fn run_verified_wasmtime_assignment( } match result.outcome { WasmTaskOutcome::Completed => { - let boundary = result.result.ok_or("completed Wasm task omitted result")?; + let boundary = result.result.ok_or_else(|| { + execution_error_with_log_bytes( + "completed Wasm task omitted result", + &command_stdout_source_bytes, + &command_stderr_source_bytes, + ) + })?; ( - format!("{}\n", serde_json::to_string(&boundary)?), + format!( + "{}\n", + serde_json::to_string(&boundary).map_err(|error| { + execution_error_with_log_bytes( + error.to_string(), + &command_stdout_source_bytes, + &command_stderr_source_bytes, + ) + })? + ), Some(boundary), ) } WasmTaskOutcome::Failed => { - return Err(result - .error - .unwrap_or_else(|| "Wasm task failed without an error".to_owned()) - .into()) + return Err(execution_error_with_log_bytes( + result + .error + .unwrap_or_else(|| "Wasm task failed without an error".to_owned()), + &command_stdout_source_bytes, + &command_stderr_source_bytes, + )) } } } @@ -140,12 +204,18 @@ pub(crate) fn run_verified_wasmtime_assignment( let task_id = TaskInstanceId::new(task.task.clone()); let artifacts = TaskArtifactStore::new(task_id.clone(), NodeId::new(args.node.clone())); let manifest = artifacts.flush(); + let stdout_source_bytes = command_stdout_source_bytes + .load(Ordering::Relaxed) + .saturating_add(stdout.len() as u64); + let stderr_source_bytes = command_stderr_source_bytes.load(Ordering::Relaxed); Ok(( CommandOutput { virtual_thread: task_id, status_code: Some(0), stdout, stderr: String::new(), + stdout_source_bytes, + stderr_source_bytes, stdout_truncated: false, stderr_truncated: false, log_backpressured: false, @@ -176,6 +246,8 @@ struct CoordinatorWasmTaskHost { next_handle_id: u64, handles: Arc>>, command_status: Arc>>, + command_stdout_source_bytes: Arc, + command_stderr_source_bytes: Arc, cancellation_requested: Arc, abort_requested: Arc, debug_control: Arc, @@ -188,6 +260,8 @@ impl CoordinatorWasmTaskHost { parent: &RuntimeTask, node_private_key: &str, module: &[u8], + command_stdout_source_bytes: Arc, + command_stderr_source_bytes: Arc, ) -> Result> { let task_spec = parent .task_spec @@ -268,6 +342,8 @@ impl CoordinatorWasmTaskHost { next_handle_id: 1, handles, command_status, + command_stdout_source_bytes, + command_stderr_source_bytes, cancellation_requested, abort_requested, debug_control, @@ -293,7 +369,16 @@ impl CoordinatorWasmTaskHost { runtime_task_from_assignment(assignment).map_err(|error| error.to_string())?; let execution = run_verified_wasmtime_assignment(&self.args, &runtime_task, &self.node_private_key); - let (terminal_state, status_code, stdout, stderr, result, retained) = match execution { + let ( + terminal_state, + status_code, + stdout, + stderr, + stdout_source_bytes, + stderr_source_bytes, + result, + retained, + ) = match execution { Ok((output, _manifest, result)) => { let retained = retained_result_artifact( self.args.project_root.as_deref(), @@ -306,20 +391,40 @@ impl CoordinatorWasmTaskHost { output.status_code, output.stdout, output.stderr, + output.stdout_source_bytes, + output.stderr_source_bytes, result, retained, ), - Err(error) => ("failed", Some(1), String::new(), error, None, None), + Err(error) => ( + "failed", + Some(1), + String::new(), + error.clone(), + output.stdout_source_bytes, + output + .stderr_source_bytes + .saturating_add(error.len() as u64), + None, + None, + ), } } - Err(error) => ( - "failed", - Some(1), - String::new(), - error.to_string(), - None, - None, - ), + Err(error) => { + let (stdout_source_bytes, stderr_source_bytes) = + assignment_error_log_bytes(error.as_ref()); + let error = error.to_string(); + ( + "failed", + Some(1), + String::new(), + error.clone(), + stdout_source_bytes, + stderr_source_bytes.saturating_add(error.len() as u64), + None, + None, + ) + } }; let artifact_path = retained .as_ref() @@ -339,8 +444,8 @@ impl CoordinatorWasmTaskHost { "task": runtime_task.task, "terminal_state": terminal_state, "status_code": status_code, - "stdout_bytes": stdout.len(), - "stderr_bytes": stderr.len(), + "stdout_bytes": stdout_source_bytes, + "stderr_bytes": stderr_source_bytes, "stdout_tail": stdout, "stderr_tail": stderr, "stdout_truncated": false, diff --git a/crates/clusterflux-node/src/assignment_runner/process_runner.rs b/crates/clusterflux-node/src/assignment_runner/process_runner.rs index e174722..eb81e39 100644 --- a/crates/clusterflux-node/src/assignment_runner/process_runner.rs +++ b/crates/clusterflux-node/src/assignment_runner/process_runner.rs @@ -66,6 +66,8 @@ pub(super) struct CoordinatorControlledProcessRunner { pub(super) node_private_key: String, pub(super) debug_control: Arc, pub(super) command_status: Arc>>, + pub(super) stdout_source_bytes: Arc, + pub(super) stderr_source_bytes: Arc, pub(super) timeout: Duration, pub(super) configured_secrets: Vec, } @@ -85,6 +87,8 @@ impl CoordinatorControlledProcessRunner { node_private_key: host.node_private_key.clone(), debug_control: Arc::clone(&host.debug_control), command_status: Arc::clone(&host.command_status), + stdout_source_bytes: Arc::clone(&host.command_stdout_source_bytes), + stderr_source_bytes: Arc::clone(&host.command_stderr_source_bytes), timeout, configured_secrets, } @@ -269,12 +273,13 @@ impl CoordinatorControlledProcessRunner { stream: &'static str, sender: SyncSender, configured_secrets: Vec, + source_bytes_total: Arc, ) -> thread::JoinHandle, String>> { thread::spawn(move || { let mut captured = Vec::new(); let mut buffer = [0_u8; 16 * 1024]; - let mut source_offset = 0_u64; - let mut pending_offset = 0_u64; + let stream_base = source_bytes_total.load(Ordering::Relaxed); + let mut pending_offset = stream_base; let mut pending = Vec::new(); let mut discarded = false; loop { @@ -284,6 +289,11 @@ impl CoordinatorControlledProcessRunner { if count == 0 { break; } + let _ = source_bytes_total.fetch_update( + Ordering::Relaxed, + Ordering::Relaxed, + |current| Some(current.saturating_add(count as u64)), + ); let remaining = maximum.saturating_sub(captured.len()); let retained = count.min(remaining); if retained > 0 { @@ -306,7 +316,6 @@ impl CoordinatorControlledProcessRunner { if retained < count { discarded = true; } - source_offset = source_offset.saturating_add(count as u64); } if let Some((consumed, text)) = redact_safe_live_log_prefix(&pending, &configured_secrets, true) @@ -322,7 +331,7 @@ impl CoordinatorControlledProcessRunner { if discarded { let _ = sender.try_send(LiveLogChunk { stream, - offset: maximum as u64, + offset: stream_base.saturating_add(maximum as u64), source_bytes: 0, bytes: b"[log output truncated at node capture limit]".to_vec(), truncated: true, @@ -445,6 +454,7 @@ impl ProcessRunner for CoordinatorControlledProcessRunner { "stdout", live_log_sender.clone(), self.configured_secrets.clone(), + Arc::clone(&self.stdout_source_bytes), ); let stderr = Self::drain_bounded( child @@ -455,6 +465,7 @@ impl ProcessRunner for CoordinatorControlledProcessRunner { "stderr", live_log_sender, self.configured_secrets.clone(), + Arc::clone(&self.stderr_source_bytes), ); let live_log_reporter = self.spawn_live_log_reporter(live_log_receiver); let mut session = match CoordinatorSession::connect(&self.args.coordinator) { @@ -584,7 +595,10 @@ impl ProcessRunner for CoordinatorControlledProcessRunner { #[cfg(test)] mod tests { - use super::redact_safe_live_log_prefix; + use super::{redact_safe_live_log_prefix, CoordinatorControlledProcessRunner}; + use std::io::Cursor; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::sync::{mpsc, Arc}; #[test] fn live_log_redaction_holds_boundaries_until_split_secrets_are_complete() { @@ -600,4 +614,30 @@ mod tests { assert!(!combined.contains("correct-")); assert!(!combined.contains("horse")); } + + #[test] + fn bounded_capture_retains_the_complete_source_byte_count() { + let source_bytes = Arc::new(AtomicU64::new(5)); + let (sender, receiver) = mpsc::sync_channel(8); + let reader = CoordinatorControlledProcessRunner::drain_bounded( + Cursor::new(vec![b'x'; 32]), + 8, + "stdout", + sender, + Vec::new(), + Arc::clone(&source_bytes), + ); + let captured = reader.join().unwrap().unwrap(); + let chunks = receiver.into_iter().collect::>(); + + assert_eq!(captured, vec![b'x'; 8]); + assert_eq!(source_bytes.load(Ordering::Relaxed), 37); + assert_eq!( + chunks.iter().map(|chunk| chunk.source_bytes).sum::(), + 8 + ); + assert_eq!(chunks[0].offset, 5); + assert_eq!(chunks.last().unwrap().offset, 13); + assert!(chunks.iter().any(|chunk| chunk.truncated)); + } } diff --git a/crates/clusterflux-node/src/assignment_runner/tests.rs b/crates/clusterflux-node/src/assignment_runner/tests.rs index 35147d3..d654d1f 100644 --- a/crates/clusterflux-node/src/assignment_runner/tests.rs +++ b/crates/clusterflux-node/src/assignment_runner/tests.rs @@ -48,6 +48,8 @@ fn test_controlled_runner( ), debug_control: Arc::new(WasmDebugControl::default()), command_status: Arc::new(Mutex::new(None)), + stdout_source_bytes: Arc::new(AtomicU64::new(0)), + stderr_source_bytes: Arc::new(AtomicU64::new(0)), timeout, configured_secrets: Vec::new(), } @@ -90,6 +92,8 @@ fn controlled_process_runner_kills_running_group_when_abort_is_polled() { ), debug_control: Arc::new(WasmDebugControl::default()), command_status: Arc::new(Mutex::new(None)), + stdout_source_bytes: Arc::new(AtomicU64::new(0)), + stderr_source_bytes: Arc::new(AtomicU64::new(0)), timeout: Duration::from_secs(30), configured_secrets: Vec::new(), }; diff --git a/crates/clusterflux-node/src/command_runner.rs b/crates/clusterflux-node/src/command_runner.rs index a1ebd33..f063043 100644 --- a/crates/clusterflux-node/src/command_runner.rs +++ b/crates/clusterflux-node/src/command_runner.rs @@ -42,6 +42,8 @@ pub struct CommandOutput { pub status_code: Option, pub stdout: String, pub stderr: String, + pub stdout_source_bytes: u64, + pub stderr_source_bytes: u64, pub stdout_truncated: bool, pub stderr_truncated: bool, pub log_backpressured: bool, @@ -83,6 +85,8 @@ impl LocalCommandExecutor { &output.stderr, max_log_bytes, ); + let stdout_source_bytes = output.stdout.len() as u64; + let stderr_source_bytes = output.stderr.len() as u64; let staged_artifact = if let Some(path) = command.stage_stdout_as { Some(overlay.write( path, @@ -98,6 +102,8 @@ impl LocalCommandExecutor { status_code: output.status.code(), stdout: logs.stdout, stderr: logs.stderr, + stdout_source_bytes, + stderr_source_bytes, stdout_truncated: logs.stdout_truncated, stderr_truncated: logs.stderr_truncated, log_backpressured: logs.backpressured, diff --git a/crates/clusterflux-node/src/daemon.rs b/crates/clusterflux-node/src/daemon.rs index 7a5a858..b223ebc 100644 --- a/crates/clusterflux-node/src/daemon.rs +++ b/crates/clusterflux-node/src/daemon.rs @@ -11,7 +11,7 @@ use clusterflux_core::{ }; use serde_json::{json, Value}; -use crate::assignment_runner::run_verified_wasmtime_assignment; +use crate::assignment_runner::{assignment_error_log_bytes, run_verified_wasmtime_assignment}; #[cfg(test)] use crate::coordinator_session::control_endpoint_identity; use crate::coordinator_session::CoordinatorSession; @@ -464,6 +464,8 @@ fn run_runtime_task( capability_report, debug_command, node_private_key, + 0, + 0, ); } @@ -498,9 +500,13 @@ fn run_runtime_task( debug_command, node_private_key, &error, + output.stdout_source_bytes, + output.stderr_source_bytes, ), }, Err(error) => { + let (stdout_source_bytes, stderr_source_bytes) = + assignment_error_log_bytes(error.as_ref()); let error = error.to_string(); if error.contains("task execution cancelled:") { record_cancelled_task( @@ -512,6 +518,8 @@ fn run_runtime_task( capability_report, debug_command, node_private_key, + stdout_source_bytes, + stderr_source_bytes, ) } else { record_failed_task( @@ -524,6 +532,8 @@ fn run_runtime_task( debug_command, node_private_key, &error, + stdout_source_bytes, + stderr_source_bytes, ) } } diff --git a/crates/clusterflux-node/src/task_reports.rs b/crates/clusterflux-node/src/task_reports.rs index a700e76..afaf0c8 100644 --- a/crates/clusterflux-node/src/task_reports.rs +++ b/crates/clusterflux-node/src/task_reports.rs @@ -47,8 +47,8 @@ pub(crate) fn record_completed_task( "process": &task.process, "node": &args.node, "task": &task.task, - "stdout_bytes": output.stdout.len(), - "stderr_bytes": output.stderr.len(), + "stdout_bytes": output.stdout_source_bytes, + "stderr_bytes": output.stderr_source_bytes, "stdout_tail": &output.stdout, "stderr_tail": &output.stderr, "stdout_truncated": output.stdout_truncated, @@ -85,8 +85,8 @@ pub(crate) fn record_completed_task( "node": &args.node, "task": &task.task, "status_code": output.status_code, - "stdout_bytes": output.stdout.len(), - "stderr_bytes": output.stderr.len(), + "stdout_bytes": output.stdout_source_bytes, + "stderr_bytes": output.stderr_source_bytes, "stdout_tail": &output.stdout, "stderr_tail": &output.stderr, "stdout_truncated": output.stdout_truncated, @@ -123,8 +123,11 @@ pub(crate) fn record_failed_task( debug_command: Value, node_private_key: &str, error: &str, + stdout_source_bytes: u64, + command_stderr_source_bytes: u64, ) -> Result> { let error = bounded_runtime_error(error); + let stderr_source_bytes = command_stderr_source_bytes.saturating_add(error.len() as u64); let log_event = session.request(signed_node_request_json( args, node_private_key, @@ -136,8 +139,8 @@ pub(crate) fn record_failed_task( "process": &task.process, "node": &args.node, "task": &task.task, - "stdout_bytes": 0, - "stderr_bytes": error.len(), + "stdout_bytes": stdout_source_bytes, + "stderr_bytes": stderr_source_bytes, "stdout_tail": "", "stderr_tail": &error, "stdout_truncated": false, @@ -175,8 +178,8 @@ pub(crate) fn record_failed_task( "task": &task.task, "terminal_state": "failed", "status_code": -1, - "stdout_bytes": 0, - "stderr_bytes": error.len(), + "stdout_bytes": stdout_source_bytes, + "stderr_bytes": stderr_source_bytes, "stdout_tail": "", "stderr_tail": &error, "stdout_truncated": false, @@ -189,6 +192,8 @@ pub(crate) fn record_failed_task( Ok(failed_node_report( task, &error, + stdout_source_bytes, + stderr_source_bytes, registration, heartbeat, capability_report, @@ -210,6 +215,8 @@ pub(crate) fn record_cancelled_task( capability_report: Value, debug_command: Value, node_private_key: &str, + stdout_source_bytes: u64, + stderr_source_bytes: u64, ) -> Result> { let recorded = session.request(signed_node_request_json( args, @@ -224,12 +231,12 @@ pub(crate) fn record_cancelled_task( "task": &task.task, "terminal_state": "cancelled", "status_code": null, - "stdout_bytes": 0, - "stderr_bytes": 0, + "stdout_bytes": stdout_source_bytes, + "stderr_bytes": stderr_source_bytes, "stdout_tail": "", "stderr_tail": "", - "stdout_truncated": false, - "stderr_truncated": false, + "stdout_truncated": stdout_source_bytes > 0, + "stderr_truncated": stderr_source_bytes > 0, "artifact_path": null, "artifact_digest": null, "artifact_size_bytes": null, @@ -238,6 +245,8 @@ pub(crate) fn record_cancelled_task( )?)?; Ok(cancelled_node_report( task, + stdout_source_bytes, + stderr_source_bytes, registration, heartbeat, capability_report, @@ -281,8 +290,8 @@ pub(crate) fn completed_node_report( "virtual_thread": output.virtual_thread, "terminal_state": if output.status_code == Some(0) { "completed" } else { "failed" }, "status_code": output.status_code, - "stdout_bytes": output.stdout.len(), - "stderr_bytes": output.stderr.len(), + "stdout_bytes": output.stdout_source_bytes, + "stderr_bytes": output.stderr_source_bytes, "stdout_tail": &output.stdout, "stderr_tail": &output.stderr, "stdout_truncated": output.stdout_truncated, @@ -305,6 +314,8 @@ pub(crate) fn completed_node_report( #[allow(clippy::too_many_arguments)] pub(crate) fn cancelled_node_report( task: &RuntimeTask, + stdout_source_bytes: u64, + stderr_source_bytes: u64, registration_response: Value, heartbeat_response: Value, capability_response: Value, @@ -318,12 +329,12 @@ pub(crate) fn cancelled_node_report( "virtual_thread": &task.task, "terminal_state": "cancelled", "status_code": null, - "stdout_bytes": 0, - "stderr_bytes": 0, + "stdout_bytes": stdout_source_bytes, + "stderr_bytes": stderr_source_bytes, "stdout_tail": "", "stderr_tail": "", - "stdout_truncated": false, - "stderr_truncated": false, + "stdout_truncated": stdout_source_bytes > 0, + "stderr_truncated": stderr_source_bytes > 0, "log_backpressured": false, "staged_artifact": null, "large_bytes_uploaded": false, @@ -343,6 +354,8 @@ pub(crate) fn cancelled_node_report( pub(crate) fn failed_node_report( task: &RuntimeTask, error: &str, + stdout_source_bytes: u64, + stderr_source_bytes: u64, registration_response: Value, heartbeat_response: Value, capability_response: Value, @@ -357,8 +370,8 @@ pub(crate) fn failed_node_report( "virtual_thread": &task.task, "terminal_state": "failed", "status_code": -1, - "stdout_bytes": 0, - "stderr_bytes": error.len(), + "stdout_bytes": stdout_source_bytes, + "stderr_bytes": stderr_source_bytes, "stdout_tail": "", "stderr_tail": error, "stdout_truncated": false, @@ -377,3 +390,40 @@ pub(crate) fn failed_node_report( "coordinator_response": coordinator_response, }) } + +#[cfg(test)] +mod tests { + use super::*; + use clusterflux_core::TaskInstanceId; + + #[test] + fn completed_report_uses_source_counts_instead_of_bounded_tail_lengths() { + let report = completed_node_report( + CommandOutput { + virtual_thread: TaskInstanceId::from("task"), + status_code: Some(0), + stdout: "bounded tail".to_owned(), + stderr: String::new(), + stdout_source_bytes: 519, + stderr_source_bytes: 0, + stdout_truncated: true, + stderr_truncated: false, + log_backpressured: false, + staged_artifact: None, + }, + false, + Value::Null, + Value::Null, + Value::Null, + Value::Null, + Value::Null, + Value::Null, + Value::Null, + Value::Null, + 1, + ); + + assert_eq!(report["stdout_bytes"], 519); + assert_eq!(report["stdout_tail"], "bounded tail"); + } +}