From a9f6f8b7a9b80c4fd847c23fd1493ce8b60f25e2 Mon Sep 17 00:00:00 2001 From: Clusterflux Release Date: Wed, 29 Jul 2026 18:13:15 +0200 Subject: [PATCH] Publish release-5edbadb22804 filtered source Private source: 5edbadb2280419cf2cb42b5237835304ad7040e0 Public tree identity: sha256:e0d4d5e1c8693959e69448684a71582edfc13e9b18d7f1e733034fea5ce62cda --- crates/clusterflux-client/src/lib.rs | 15 ++ crates/clusterflux-client/src/protocol.rs | 5 + .../tests/client_contract.rs | 28 +- .../tests/fixtures/web_operations.json | 6 + crates/clusterflux-coordinator/src/service.rs | 8 + .../src/service/logs.rs | 241 +++++++++++------- .../src/service/quota.rs | 16 -- .../src/service/summaries.rs | 9 + .../src/service/tests.rs | 156 +++++++++++- .../src/assignment_runner/process_runner.rs | 72 +++--- crates/clusterflux-node/src/command_runner.rs | 28 +- crates/clusterflux-node/src/lib.rs | 12 +- 12 files changed, 430 insertions(+), 166 deletions(-) diff --git a/crates/clusterflux-client/src/lib.rs b/crates/clusterflux-client/src/lib.rs index 12c94bd..93c6feb 100644 --- a/crates/clusterflux-client/src/lib.rs +++ b/crates/clusterflux-client/src/lib.rs @@ -109,6 +109,21 @@ impl ClusterfluxClient { } } + pub async fn cancel_browser_login( + &self, + transaction_id: impl Into, + ) -> Result<(), ClientError> { + match self + .send_login(LoginRequest::CancelWebBrowserLogin { + transaction_id: transaction_id.into(), + }) + .await? + { + WireResponse::WebBrowserLoginCancelled {} => Ok(()), + _ => Err(unexpected_response("web_browser_login_cancelled")), + } + } + pub async fn account_status(&self) -> Result { match self .send_authenticated(AuthenticatedRequest::AuthStatus) diff --git a/crates/clusterflux-client/src/protocol.rs b/crates/clusterflux-client/src/protocol.rs index 084f3ba..a71ecff 100644 --- a/crates/clusterflux-client/src/protocol.rs +++ b/crates/clusterflux-client/src/protocol.rs @@ -122,6 +122,9 @@ pub(crate) enum AuthenticatedRequest { #[serde(tag = "type", rename_all = "snake_case")] pub(crate) enum LoginRequest { BeginWebBrowserLogin {}, + CancelWebBrowserLogin { + transaction_id: String, + }, ExchangeWebLoginHandoff { transaction_id: String, handoff_code: String, @@ -297,6 +300,7 @@ pub(crate) enum WireResponse { authorization_url: String, expires_at_epoch_seconds: u64, }, + WebBrowserLoginCancelled {}, WebBrowserSession { session: WireBrowserSession, }, @@ -334,6 +338,7 @@ mod tests { "abort_process", "auth_status", "begin_web_browser_login", + "cancel_web_browser_login", "cancel_process", "create_artifact_download_link", "create_debug_epoch", diff --git a/crates/clusterflux-client/tests/client_contract.rs b/crates/clusterflux-client/tests/client_contract.rs index 1f3f0ec..8851a54 100644 --- a/crates/clusterflux-client/tests/client_contract.rs +++ b/crates/clusterflux-client/tests/client_contract.rs @@ -6,7 +6,7 @@ use clusterflux_client::{ ClusterfluxClient, ControlTransport, MockTransport, ProjectId, SessionCredential, TenantId, UserId, CLIENT_API_VERSION, }; -use clusterflux_control::CONTROL_API_PATH; +use clusterflux_control::{CONTROL_API_PATH, LOGIN_API_PATH}; use clusterflux_coordinator::service::CoordinatorService; use serde_json::{json, Value}; @@ -64,6 +64,32 @@ async fn structured_errors_retain_machine_fields_and_originating_request_id() { assert!(!error.retryable); } +#[tokio::test] +async fn browser_login_cancellation_uses_the_login_boundary_and_exact_transaction() { + let transport = MockTransport::from_json_responses([ + json!({ "type": "web_browser_login_cancelled" }).to_string(), + ]); + let client = ClusterfluxClient::with_transport(transport.clone()); + + client + .cancel_browser_login("login-transaction") + .await + .unwrap(); + + let requests = transport.requests(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].api_path, LOGIN_API_PATH); + let envelope: Value = serde_json::from_slice(&requests[0].body).unwrap(); + assert_eq!(envelope["operation"], "cancel_web_browser_login"); + assert_eq!( + envelope["payload"], + json!({ + "type": "cancel_web_browser_login", + "transaction_id": "login-transaction" + }) + ); +} + #[tokio::test] async fn a_mismatched_error_request_id_is_rejected_as_a_protocol_error() { let transport = MockTransport::from_json_responses([json!({ diff --git a/crates/clusterflux-client/tests/fixtures/web_operations.json b/crates/clusterflux-client/tests/fixtures/web_operations.json index 16ea235..45b7b23 100644 --- a/crates/clusterflux-client/tests/fixtures/web_operations.json +++ b/crates/clusterflux-client/tests/fixtures/web_operations.json @@ -5,6 +5,12 @@ "request": { "type": "begin_web_browser_login" }, "response": { "type": "web_browser_login_started", "transaction_id": "login-transaction", "authorization_url": "https://auth.clusterflux.lesstuff.com/authorize?state=opaque", "expires_at_epoch_seconds": 1000 } }, + { + "operation": "cancel_web_browser_login", + "boundary": "login", + "request": { "type": "cancel_web_browser_login", "transaction_id": "login-transaction" }, + "response": { "type": "web_browser_login_cancelled" } + }, { "operation": "exchange_web_login_handoff", "boundary": "login", diff --git a/crates/clusterflux-coordinator/src/service.rs b/crates/clusterflux-coordinator/src/service.rs index aef24d0..bab6c13 100644 --- a/crates/clusterflux-coordinator/src/service.rs +++ b/crates/clusterflux-coordinator/src/service.rs @@ -304,6 +304,13 @@ pub struct CoordinatorService { clusterflux_core::TaskInstanceId, String, )>, + recent_log_quota_truncated_streams: BTreeSet<( + TenantId, + ProjectId, + ProcessId, + clusterflux_core::TaskInstanceId, + String, + )>, next_recent_log_sequence: u64, debug_audit_events: VecDeque, debug_epochs: BTreeMap, @@ -579,6 +586,7 @@ impl CoordinatorService { recent_log_dropped_through: BTreeMap::new(), recent_log_accounted_bytes: BTreeMap::new(), recent_log_truncated_streams: BTreeSet::new(), + recent_log_quota_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 c717ad7..1fb575b 100644 --- a/crates/clusterflux-coordinator/src/service/logs.rs +++ b/crates/clusterflux-coordinator/src/service/logs.rs @@ -40,19 +40,7 @@ impl CoordinatorService { validate_task_log_tail("stdout_tail", &stdout_tail)?; validate_task_log_tail("stderr_tail", &stderr_tail)?; 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, unaccounted_bytes, now_epoch_seconds)?; - self.quota - .charge_log_bytes(&tenant, &project, unaccounted_bytes, now_epoch_seconds)?; - self.reconcile_final_log_stream( + let stdout_retained = self.accept_final_log_stream( &tenant, &project, &process, @@ -62,8 +50,8 @@ impl CoordinatorService { &stdout_tail, stdout_truncated, now_epoch_seconds, - ); - self.reconcile_final_log_stream( + )?; + let stderr_retained = self.accept_final_log_stream( &tenant, &project, &process, @@ -73,18 +61,22 @@ impl CoordinatorService { &stderr_tail, stderr_truncated, now_epoch_seconds, - ); + )?; Ok(CoordinatorResponse::TaskLogRecorded { process, task, stdout_bytes, stderr_bytes, - stdout_tail: if stdout_truncated { + stdout_tail: if !stdout_retained { + "[log output truncated at project log quota]".to_owned() + } else if stdout_truncated { format!("{stdout_tail}\n... truncated") } else { stdout_tail }, - stderr_tail: if stderr_truncated { + stderr_tail: if !stderr_retained { + "[log output truncated at project log quota]".to_owned() + } else if stderr_truncated { format!("{stderr_tail}\n... truncated") } else { stderr_tail @@ -150,11 +142,41 @@ impl CoordinatorService { }); } let now_epoch_seconds = self.current_epoch_seconds()?; + if self.recent_log_quota_truncated_streams.contains(&key) { + if end > expected { + self.recent_log_accounted_bytes.insert(key, end); + } + return Ok(CoordinatorResponse::TaskLogChunkRecorded { + process, + task, + sequence: None, + next_offset: end.max(expected), + }); + } 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 self + .quota + .charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds) + .is_err() + { + if end > expected { + self.recent_log_accounted_bytes.insert(key.clone(), end); + } + let sequence = self.mark_log_quota_truncated( + &tenant, + &project, + &process, + &task, + &stream, + now_epoch_seconds, + ); + return Ok(CoordinatorResponse::TaskLogChunkRecorded { + process, + task, + sequence, + next_offset: end.max(expected), + }); + } if offset > expected { self.record_recent_log( tenant.clone(), @@ -334,27 +356,7 @@ impl CoordinatorService { result, }; 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, - unaccounted_bytes, - now_epoch_seconds, - )?; - self.quota.charge_log_bytes( - &event.tenant, - &event.project, - unaccounted_bytes, - now_epoch_seconds, - )?; - self.reconcile_final_log_stream( + let stdout_retained = self.accept_final_log_stream( &event.tenant, &event.project, &event.process, @@ -364,8 +366,8 @@ impl CoordinatorService { &event.stdout_tail, event.stdout_truncated, now_epoch_seconds, - ); - self.reconcile_final_log_stream( + )?; + let stderr_retained = self.accept_final_log_stream( &event.tenant, &event.project, &event.process, @@ -375,7 +377,15 @@ impl CoordinatorService { &event.stderr_tail, event.stderr_truncated, now_epoch_seconds, - ); + )?; + if !stdout_retained { + event.stdout_tail = "[log output truncated at project log quota]".to_owned(); + event.stdout_truncated = true; + } + if !stderr_retained { + event.stderr_tail = "[log output truncated at project log quota]".to_owned(); + event.stderr_truncated = true; + } let task_key = task_control_key( &event.tenant, &event.project, @@ -833,48 +843,94 @@ impl CoordinatorService { } #[allow(clippy::too_many_arguments)] - fn unaccounted_final_log_bytes( - &self, + fn accept_final_log_stream( + &mut self, tenant: &TenantId, project: &ProjectId, process: &ProcessId, task: &TaskInstanceId, - stdout_bytes: u64, - stderr_bytes: u64, - ) -> Result { - let stdout_accounted = self + stream: TaskLogStream, + total_source_bytes: u64, + final_tail: &str, + source_truncated: bool, + now_epoch_seconds: u64, + ) -> Result { + let key = recent_log_offset_key(tenant, project, process, task, &stream); + let accounted = self .recent_log_accounted_bytes - .get(&recent_log_offset_key( + .get(&key) + .copied() + .unwrap_or(0); + let remaining = total_source_bytes.checked_sub(accounted).ok_or_else(|| { + let stream_name = match stream { + TaskLogStream::Stdout => "stdout", + TaskLogStream::Stderr => "stderr", + }; + CoordinatorServiceError::Protocol(format!( + "final {stream_name} byte count {total_source_bytes} is below the {accounted} live bytes already accounted" + )) + })?; + if self.recent_log_quota_truncated_streams.contains(&key) { + self.recent_log_accounted_bytes + .insert(key, total_source_bytes); + return Ok(false); + } + if self + .quota + .charge_log_bytes(tenant, project, remaining, now_epoch_seconds) + .is_err() + { + self.recent_log_accounted_bytes + .insert(key, total_source_bytes); + self.mark_log_quota_truncated( 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) + &stream, + now_epoch_seconds, + ); + return Ok(false); + } + self.reconcile_final_log_stream( + tenant, + project, + process, + task, + stream, + total_source_bytes, + final_tail, + source_truncated, + now_epoch_seconds, + ); + Ok(true) + } + + #[allow(clippy::too_many_arguments)] + fn mark_log_quota_truncated( + &mut self, + tenant: &TenantId, + project: &ProjectId, + process: &ProcessId, + task: &TaskInstanceId, + stream: &TaskLogStream, + now_epoch_seconds: u64, + ) -> Option { + let key = recent_log_offset_key(tenant, project, process, task, stream); + if !self.recent_log_quota_truncated_streams.insert(key.clone()) { + return None; + } + self.recent_log_truncated_streams.insert(key); + Some(self.record_recent_log( + tenant.clone(), + project.clone(), + process.clone(), + task.clone(), + stream.clone(), + "[log output truncated at project log quota]".to_owned(), + true, + now_epoch_seconds, + )) } #[allow(clippy::too_many_arguments)] @@ -1020,6 +1076,14 @@ impl CoordinatorService { || entry_task != task }, ); + self.recent_log_quota_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 { @@ -1260,17 +1324,6 @@ impl CoordinatorService { } } -fn checked_reported_log_bytes( - stdout_bytes: u64, - stderr_bytes: u64, -) -> Result { - stdout_bytes.checked_add(stderr_bytes).ok_or_else(|| { - CoordinatorServiceError::Protocol( - "reported task log byte counts exceed the supported range".to_owned(), - ) - }) -} - fn recent_log_offset_key( tenant: &TenantId, project: &ProjectId, @@ -1305,11 +1358,11 @@ fn bounded_log_tail(mut value: String, truncated: &mut bool) -> String { if value.len() <= MAX_TASK_LOG_TAIL_BYTES { return value; } - let mut boundary = MAX_TASK_LOG_TAIL_BYTES; - while !value.is_char_boundary(boundary) { - boundary -= 1; + let mut boundary = value.len() - MAX_TASK_LOG_TAIL_BYTES; + while boundary < value.len() && !value.is_char_boundary(boundary) { + boundary += 1; } - value.truncate(boundary); + value.drain(..boundary); *truncated = true; value } diff --git a/crates/clusterflux-coordinator/src/service/quota.rs b/crates/clusterflux-coordinator/src/service/quota.rs index 6fcd90f..78b65ae 100644 --- a/crates/clusterflux-coordinator/src/service/quota.rs +++ b/crates/clusterflux-coordinator/src/service/quota.rs @@ -260,22 +260,6 @@ impl CoordinatorQuota { self.charge(tenant, project, LimitKind::ApiCall, 1, now_epoch_seconds) } - pub(super) fn can_charge_log_bytes( - &self, - tenant: &TenantId, - project: &ProjectId, - bytes: u64, - now_epoch_seconds: u64, - ) -> Result<(), LimitError> { - self.can_charge( - tenant, - project, - LimitKind::LogBytes, - bytes, - now_epoch_seconds, - ) - } - pub(super) fn charge_log_bytes( &mut self, tenant: &TenantId, diff --git a/crates/clusterflux-coordinator/src/service/summaries.rs b/crates/clusterflux-coordinator/src/service/summaries.rs index cd54a2d..1e5ad92 100644 --- a/crates/clusterflux-coordinator/src/service/summaries.rs +++ b/crates/clusterflux-coordinator/src/service/summaries.rs @@ -98,6 +98,11 @@ impl CoordinatorService { entry_tenant != tenant || entry_project != project || entry_process != process }, ); + self.recent_log_quota_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); @@ -525,6 +530,10 @@ impl CoordinatorService { .retain(|(tenant, project, process, _, _)| { tenant != &key.0 || project != &key.1 || process != &key.2 }); + self.recent_log_quota_truncated_streams + .retain(|(tenant, project, process, _, _)| { + tenant != &key.0 || project != &key.1 || process != &key.2 + }); self.process_summary_order .retain(|retained| retained != key); } diff --git a/crates/clusterflux-coordinator/src/service/tests.rs b/crates/clusterflux-coordinator/src/service/tests.rs index 659537a..339860d 100644 --- a/crates/clusterflux-coordinator/src/service/tests.rs +++ b/crates/clusterflux-coordinator/src/service/tests.rs @@ -2435,7 +2435,7 @@ fn authenticated_api_calls_are_metered_per_tenant_and_project_before_dispatch() } #[test] -fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() { +fn signed_node_log_ingestion_truncates_at_scoped_quota_without_failing_reports() { let mut limits = ResourceLimits::unlimited(); limits.limits.insert(LimitKind::LogBytes, 4); let quota = CoordinatorQuotaConfiguration::new(limits, [(LimitKind::LogBytes, 60)]).unwrap(); @@ -2494,7 +2494,7 @@ fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() { backpressured: false, }) .unwrap(); - let denied = service + let truncated = service .handle_signed_node_request_auto(CoordinatorRequest::ReportTaskLog { tenant: "tenant".to_owned(), project: "project".to_owned(), @@ -2509,8 +2509,22 @@ fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() { stderr_truncated: false, backpressured: true, }) - .unwrap_err(); - assert!(denied.to_string().contains("LogBytes")); + .unwrap(); + let CoordinatorResponse::TaskLogRecorded { + stdout_tail, + stdout_bytes: 4, + .. + } = truncated + else { + panic!("expected a successful truncated task-log report"); + }; + assert_eq!(stdout_tail, "[log output truncated at project log quota]"); + assert!(service + .recent_logs + .get(&(TenantId::from("tenant"), ProjectId::from("project"))) + .unwrap() + .iter() + .any(|entry| entry.text.contains("project log quota") && entry.truncated)); assert_eq!( service .quota @@ -2519,6 +2533,140 @@ fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() { ); } +#[test] +fn log_quota_exhaustion_cannot_strand_task_completion_or_artifact_publication() { + let mut limits = ResourceLimits::unlimited(); + limits.limits.insert(LimitKind::LogBytes, 4); + let quota = CoordinatorQuotaConfiguration::new(limits, [(LimitKind::LogBytes, 60)]).unwrap(); + let mut service = CoordinatorService::new_with_admin_token_database_url_and_quota( + 7, + "test-admin-token", + None, + quota, + ) + .unwrap(); + service.set_server_time(30); + service + .handle_request(CoordinatorRequest::AttachNode { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + node: "node".to_owned(), + public_key: test_node_public_key("node"), + }) + .unwrap(); + service + .handle_request(CoordinatorRequest::StartProcess { + launch_attempt: None, + tenant: "tenant".to_owned(), + project: "project".to_owned(), + actor_user: None, + actor_agent: None, + agent_public_key_fingerprint: None, + agent_signature: None, + process: "process".to_owned(), + restart: false, + }) + .unwrap(); + service + .handle_signed_node_request_auto(CoordinatorRequest::ReconnectNode { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + node: "node".to_owned(), + process: "process".to_owned(), + epoch: 7, + }) + .unwrap(); + register_test_task_assignment( + &mut service, + "tenant", + "project", + "process", + "node", + "compile", + "compile-one", + 7, + ); + service.record_task_completion_event(TaskCompletionEvent { + tenant: TenantId::from("tenant"), + project: ProjectId::from("project"), + process: ProcessId::from("process"), + node: NodeId::from("coordinator-main"), + executor: TaskExecutor::CoordinatorMain, + task_definition: TaskDefinitionId::from("build"), + task: TaskInstanceId::from("main"), + 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_signed_node_request_auto(CoordinatorRequest::TaskCompleted { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + process: "process".to_owned(), + node: "node".to_owned(), + task: "compile-one".to_owned(), + terminal_state: Some(TaskTerminalState::Completed), + status_code: Some(0), + stdout_bytes: 64, + stderr_bytes: 0, + stdout_tail: "the-real-final-tail".to_owned(), + stderr_tail: String::new(), + stdout_truncated: true, + stderr_truncated: false, + artifact_path: Some("/vfs/artifacts/result.bin".to_owned()), + artifact_digest: Some(Digest::sha256("artifact bytes")), + artifact_size_bytes: Some(14), + result: None, + }) + .unwrap(); + + let event = service + .task_events + .iter() + .find(|event| event.task == TaskInstanceId::from("compile-one")) + .unwrap(); + assert_eq!( + event.stdout_tail, + "[log output truncated at project log quota]" + ); + assert!(event.stdout_truncated); + assert!(!service + .active_tasks + .iter() + .any(|key| key.4 == TaskInstanceId::from("compile-one"))); + assert!(service + .coordinator + .active_process( + &TenantId::from("tenant"), + &ProjectId::from("project"), + &ProcessId::from("process"), + ) + .is_none()); + assert!(matches!( + service + .handle_request(CoordinatorRequest::GetArtifact { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + actor_user: "user".to_owned(), + artifact: "result.bin".to_owned(), + }) + .unwrap(), + CoordinatorResponse::Artifact { .. } + )); +} + #[test] fn service_attaches_node_starts_process_and_records_scoped_task_event() { let mut service = CoordinatorService::new(7); diff --git a/crates/clusterflux-node/src/assignment_runner/process_runner.rs b/crates/clusterflux-node/src/assignment_runner/process_runner.rs index eb81e39..0e2da69 100644 --- a/crates/clusterflux-node/src/assignment_runner/process_runner.rs +++ b/crates/clusterflux-node/src/assignment_runner/process_runner.rs @@ -9,6 +9,26 @@ struct LiveLogChunk { truncated: bool, } +fn append_bounded_tail(tail: &mut Vec, bytes: &[u8], maximum: usize) { + if maximum == 0 { + tail.clear(); + return; + } + if bytes.len() >= maximum { + tail.clear(); + tail.extend_from_slice(&bytes[bytes.len() - maximum..]); + return; + } + let overflow = tail + .len() + .saturating_add(bytes.len()) + .saturating_sub(maximum); + if overflow > 0 { + tail.drain(..overflow); + } + tail.extend_from_slice(bytes); +} + fn redact_safe_live_log_prefix( pending: &[u8], configured_secrets: &[String], @@ -279,9 +299,9 @@ impl CoordinatorControlledProcessRunner { let mut captured = Vec::new(); let mut buffer = [0_u8; 16 * 1024]; let stream_base = source_bytes_total.load(Ordering::Relaxed); + let mut source_bytes_read = 0_u64; let mut pending_offset = stream_base; let mut pending = Vec::new(); - let mut discarded = false; loop { let count = reader .read(&mut buffer) @@ -294,27 +314,21 @@ impl CoordinatorControlledProcessRunner { 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 { - captured.extend_from_slice(&buffer[..retained]); - pending.extend_from_slice(&buffer[..retained]); - if let Some((consumed, text)) = - redact_safe_live_log_prefix(&pending, &configured_secrets, false) - { - let _ = sender.try_send(LiveLogChunk { - stream, - offset: pending_offset, - source_bytes: consumed as u64, - bytes: text.into_bytes(), - truncated: false, - }); - pending.drain(..consumed); - pending_offset = pending_offset.saturating_add(consumed as u64); - } - } - if retained < count { - discarded = true; + source_bytes_read = source_bytes_read.saturating_add(count as u64); + append_bounded_tail(&mut captured, &buffer[..count], maximum); + pending.extend_from_slice(&buffer[..count]); + if let Some((consumed, text)) = + redact_safe_live_log_prefix(&pending, &configured_secrets, false) + { + let _ = sender.try_send(LiveLogChunk { + stream, + offset: pending_offset, + source_bytes: consumed as u64, + bytes: text.into_bytes(), + truncated: false, + }); + pending.drain(..consumed); + pending_offset = pending_offset.saturating_add(consumed as u64); } } if let Some((consumed, text)) = @@ -328,10 +342,10 @@ impl CoordinatorControlledProcessRunner { truncated: false, }); } - if discarded { + if source_bytes_read > maximum as u64 { let _ = sender.try_send(LiveLogChunk { stream, - offset: stream_base.saturating_add(maximum as u64), + offset: stream_base.saturating_add(source_bytes_read), source_bytes: 0, bytes: b"[log output truncated at node capture limit]".to_vec(), truncated: true, @@ -616,11 +630,11 @@ mod tests { } #[test] - fn bounded_capture_retains_the_complete_source_byte_count() { + fn bounded_capture_retains_the_real_tail_and_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]), + Cursor::new(b"0123456789abcdefghijklmnopqrstuv".to_vec()), 8, "stdout", sender, @@ -630,14 +644,14 @@ mod tests { let captured = reader.join().unwrap().unwrap(); let chunks = receiver.into_iter().collect::>(); - assert_eq!(captured, vec![b'x'; 8]); + assert_eq!(captured, b"opqrstuv"); assert_eq!(source_bytes.load(Ordering::Relaxed), 37); assert_eq!( chunks.iter().map(|chunk| chunk.source_bytes).sum::(), - 8 + 32 ); assert_eq!(chunks[0].offset, 5); - assert_eq!(chunks.last().unwrap().offset, 13); + assert_eq!(chunks.last().unwrap().offset, 37); assert!(chunks.iter().any(|chunk| chunk.truncated)); } } diff --git a/crates/clusterflux-node/src/command_runner.rs b/crates/clusterflux-node/src/command_runner.rs index f063043..790b838 100644 --- a/crates/clusterflux-node/src/command_runner.rs +++ b/crates/clusterflux-node/src/command_runner.rs @@ -1,6 +1,6 @@ use clusterflux_core::{ - CommandInvocation, Digest, LogBuffer, NativeCommandPolicy, NodeId, TaskInstanceId, VfsObject, - VfsOverlay, VfsPath, + CommandInvocation, Digest, NativeCommandPolicy, NodeId, TaskInstanceId, VfsObject, VfsOverlay, + VfsPath, }; use serde::{Deserialize, Serialize}; @@ -113,24 +113,20 @@ impl LocalCommandExecutor { } pub(super) fn capture_command_logs( - task: &TaskInstanceId, + _task: &TaskInstanceId, stdout: &[u8], stderr: &[u8], max_log_bytes: usize, ) -> CapturedCommandLogs { - let mut logs = LogBuffer::new(max_log_bytes); - logs.push(task.clone(), stdout); - logs.push(task.clone(), stderr); - let records = logs.records(); - let stdout_record = &records[0]; - let stderr_record = &records[1]; - debug_assert_eq!(&stdout_record.task, task); - debug_assert_eq!(&stderr_record.task, task); + let stdout_truncated = stdout.len() > max_log_bytes; + let stderr_truncated = stderr.len() > max_log_bytes; + let stdout_start = stdout.len().saturating_sub(max_log_bytes); + let stderr_start = stderr.len().saturating_sub(max_log_bytes); CapturedCommandLogs { - stdout: String::from_utf8_lossy(&stdout_record.bytes).into_owned(), - stderr: String::from_utf8_lossy(&stderr_record.bytes).into_owned(), - stdout_truncated: stdout_record.truncated, - stderr_truncated: stderr_record.truncated, - backpressured: logs.backpressured(), + stdout: String::from_utf8_lossy(&stdout[stdout_start..]).into_owned(), + stderr: String::from_utf8_lossy(&stderr[stderr_start..]).into_owned(), + stdout_truncated, + stderr_truncated, + backpressured: stdout_truncated || stderr_truncated, } } diff --git a/crates/clusterflux-node/src/lib.rs b/crates/clusterflux-node/src/lib.rs index 8b100c3..d9db699 100644 --- a/crates/clusterflux-node/src/lib.rs +++ b/crates/clusterflux-node/src/lib.rs @@ -741,7 +741,7 @@ mod tests { } #[test] - fn linux_backend_caps_logs_without_truncating_staged_artifact_bytes() { + fn linux_backend_retains_final_log_tail_without_truncating_staged_artifact_bytes() { let invocation = CommandInvocation { program: "cargo".to_owned(), args: vec!["build".to_owned()], @@ -774,7 +774,7 @@ mod tests { .unwrap(); assert_eq!(output.virtual_thread, TaskInstanceId::from("compile-linux")); - assert_eq!(output.stdout, "abcd"); + assert_eq!(output.stdout, "cdef"); assert!(output.stdout_truncated); assert!(output.log_backpressured); assert_eq!(output.staged_artifact.as_ref().unwrap().size, 6); @@ -992,7 +992,7 @@ mod tests { #[cfg(unix)] #[test] - fn local_command_executor_caps_logs_and_reports_backpressure_by_virtual_thread() { + fn local_command_executor_retains_each_stream_tail_and_reports_backpressure() { let executor = LocalCommandExecutor { node: clusterflux_core::NodeId::from("node"), hosted_control_plane: false, @@ -1021,10 +1021,10 @@ mod tests { .unwrap(); assert_eq!(output.virtual_thread, TaskInstanceId::from("compile-linux")); - assert_eq!(output.stdout, "abcd"); - assert_eq!(output.stderr, ""); + assert_eq!(output.stdout, "cdef"); + assert_eq!(output.stderr, "err"); assert!(output.stdout_truncated); - assert!(output.stderr_truncated); + assert!(!output.stderr_truncated); assert!(output.log_backpressured); }