diff --git a/crates/clusterflux-client/src/lib.rs b/crates/clusterflux-client/src/lib.rs index 93c6feb..12c94bd 100644 --- a/crates/clusterflux-client/src/lib.rs +++ b/crates/clusterflux-client/src/lib.rs @@ -109,21 +109,6 @@ 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 a71ecff..084f3ba 100644 --- a/crates/clusterflux-client/src/protocol.rs +++ b/crates/clusterflux-client/src/protocol.rs @@ -122,9 +122,6 @@ 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, @@ -300,7 +297,6 @@ pub(crate) enum WireResponse { authorization_url: String, expires_at_epoch_seconds: u64, }, - WebBrowserLoginCancelled {}, WebBrowserSession { session: WireBrowserSession, }, @@ -338,7 +334,6 @@ 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 8851a54..1f3f0ec 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, LOGIN_API_PATH}; +use clusterflux_control::CONTROL_API_PATH; use clusterflux_coordinator::service::CoordinatorService; use serde_json::{json, Value}; @@ -64,32 +64,6 @@ 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 45b7b23..16ea235 100644 --- a/crates/clusterflux-client/tests/fixtures/web_operations.json +++ b/crates/clusterflux-client/tests/fixtures/web_operations.json @@ -5,12 +5,6 @@ "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 bab6c13..aef24d0 100644 --- a/crates/clusterflux-coordinator/src/service.rs +++ b/crates/clusterflux-coordinator/src/service.rs @@ -304,13 +304,6 @@ 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, @@ -586,7 +579,6 @@ 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 1fb575b..c717ad7 100644 --- a/crates/clusterflux-coordinator/src/service/logs.rs +++ b/crates/clusterflux-coordinator/src/service/logs.rs @@ -40,7 +40,19 @@ 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 stdout_retained = self.accept_final_log_stream( + 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( &tenant, &project, &process, @@ -50,8 +62,8 @@ impl CoordinatorService { &stdout_tail, stdout_truncated, now_epoch_seconds, - )?; - let stderr_retained = self.accept_final_log_stream( + ); + self.reconcile_final_log_stream( &tenant, &project, &process, @@ -61,22 +73,18 @@ impl CoordinatorService { &stderr_tail, stderr_truncated, now_epoch_seconds, - )?; + ); Ok(CoordinatorResponse::TaskLogRecorded { process, task, stdout_bytes, stderr_bytes, - stdout_tail: if !stdout_retained { - "[log output truncated at project log quota]".to_owned() - } else if stdout_truncated { + stdout_tail: if stdout_truncated { format!("{stdout_tail}\n... truncated") } else { stdout_tail }, - stderr_tail: if !stderr_retained { - "[log output truncated at project log quota]".to_owned() - } else if stderr_truncated { + stderr_tail: if stderr_truncated { format!("{stderr_tail}\n... truncated") } else { stderr_tail @@ -142,41 +150,11 @@ 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); - 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), - }); - } + 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(), @@ -356,7 +334,27 @@ impl CoordinatorService { result, }; let now_epoch_seconds = self.current_epoch_seconds()?; - let stdout_retained = self.accept_final_log_stream( + 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( &event.tenant, &event.project, &event.process, @@ -366,8 +364,8 @@ impl CoordinatorService { &event.stdout_tail, event.stdout_truncated, now_epoch_seconds, - )?; - let stderr_retained = self.accept_final_log_stream( + ); + self.reconcile_final_log_stream( &event.tenant, &event.project, &event.process, @@ -377,15 +375,7 @@ 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, @@ -843,94 +833,48 @@ impl CoordinatorService { } #[allow(clippy::too_many_arguments)] - fn accept_final_log_stream( - &mut self, + fn unaccounted_final_log_bytes( + &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, - ) -> Result { - let key = recent_log_offset_key(tenant, project, process, task, &stream); - let accounted = self + stdout_bytes: u64, + stderr_bytes: u64, + ) -> Result { + let stdout_accounted = self .recent_log_accounted_bytes - .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( + .get(&recent_log_offset_key( tenant, project, process, task, - &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, - )) + &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)] @@ -1076,14 +1020,6 @@ 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 { @@ -1324,6 +1260,17 @@ 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, @@ -1358,11 +1305,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 = value.len() - MAX_TASK_LOG_TAIL_BYTES; - while boundary < value.len() && !value.is_char_boundary(boundary) { - boundary += 1; + let mut boundary = MAX_TASK_LOG_TAIL_BYTES; + while !value.is_char_boundary(boundary) { + boundary -= 1; } - value.drain(..boundary); + value.truncate(boundary); *truncated = true; value } diff --git a/crates/clusterflux-coordinator/src/service/quota.rs b/crates/clusterflux-coordinator/src/service/quota.rs index 78b65ae..6fcd90f 100644 --- a/crates/clusterflux-coordinator/src/service/quota.rs +++ b/crates/clusterflux-coordinator/src/service/quota.rs @@ -260,6 +260,22 @@ 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 1e5ad92..cd54a2d 100644 --- a/crates/clusterflux-coordinator/src/service/summaries.rs +++ b/crates/clusterflux-coordinator/src/service/summaries.rs @@ -98,11 +98,6 @@ 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); @@ -530,10 +525,6 @@ 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 339860d..659537a 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_truncates_at_scoped_quota_without_failing_reports() { +fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() { 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_truncates_at_scoped_quota_without_failing_reports() backpressured: false, }) .unwrap(); - let truncated = service + let denied = service .handle_signed_node_request_auto(CoordinatorRequest::ReportTaskLog { tenant: "tenant".to_owned(), project: "project".to_owned(), @@ -2509,22 +2509,8 @@ fn signed_node_log_ingestion_truncates_at_scoped_quota_without_failing_reports() stderr_truncated: false, backpressured: true, }) - .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)); + .unwrap_err(); + assert!(denied.to_string().contains("LogBytes")); assert_eq!( service .quota @@ -2533,140 +2519,6 @@ fn signed_node_log_ingestion_truncates_at_scoped_quota_without_failing_reports() ); } -#[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 0e2da69..eb81e39 100644 --- a/crates/clusterflux-node/src/assignment_runner/process_runner.rs +++ b/crates/clusterflux-node/src/assignment_runner/process_runner.rs @@ -9,26 +9,6 @@ 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], @@ -299,9 +279,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) @@ -314,21 +294,27 @@ impl CoordinatorControlledProcessRunner { Ordering::Relaxed, |current| Some(current.saturating_add(count as u64)), ); - 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); + 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; } } if let Some((consumed, text)) = @@ -342,10 +328,10 @@ impl CoordinatorControlledProcessRunner { truncated: false, }); } - if source_bytes_read > maximum as u64 { + if discarded { let _ = sender.try_send(LiveLogChunk { stream, - offset: stream_base.saturating_add(source_bytes_read), + offset: stream_base.saturating_add(maximum as u64), source_bytes: 0, bytes: b"[log output truncated at node capture limit]".to_vec(), truncated: true, @@ -630,11 +616,11 @@ mod tests { } #[test] - fn bounded_capture_retains_the_real_tail_and_complete_source_byte_count() { + 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(b"0123456789abcdefghijklmnopqrstuv".to_vec()), + Cursor::new(vec![b'x'; 32]), 8, "stdout", sender, @@ -644,14 +630,14 @@ mod tests { let captured = reader.join().unwrap().unwrap(); let chunks = receiver.into_iter().collect::>(); - assert_eq!(captured, b"opqrstuv"); + 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::(), - 32 + 8 ); assert_eq!(chunks[0].offset, 5); - assert_eq!(chunks.last().unwrap().offset, 37); + assert_eq!(chunks.last().unwrap().offset, 13); 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 790b838..f063043 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, NativeCommandPolicy, NodeId, TaskInstanceId, VfsObject, VfsOverlay, - VfsPath, + CommandInvocation, Digest, LogBuffer, NativeCommandPolicy, NodeId, TaskInstanceId, VfsObject, + VfsOverlay, VfsPath, }; use serde::{Deserialize, Serialize}; @@ -113,20 +113,24 @@ impl LocalCommandExecutor { } pub(super) fn capture_command_logs( - _task: &TaskInstanceId, + task: &TaskInstanceId, stdout: &[u8], stderr: &[u8], max_log_bytes: usize, ) -> CapturedCommandLogs { - 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); + 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); CapturedCommandLogs { - 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, + 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(), } } diff --git a/crates/clusterflux-node/src/lib.rs b/crates/clusterflux-node/src/lib.rs index d9db699..8b100c3 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_retains_final_log_tail_without_truncating_staged_artifact_bytes() { + fn linux_backend_caps_logs_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, "cdef"); + assert_eq!(output.stdout, "abcd"); 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_retains_each_stream_tail_and_reports_backpressure() { + fn local_command_executor_caps_logs_and_reports_backpressure_by_virtual_thread() { 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, "cdef"); - assert_eq!(output.stderr, "err"); + assert_eq!(output.stdout, "abcd"); + assert_eq!(output.stderr, ""); assert!(output.stdout_truncated); - assert!(!output.stderr_truncated); + assert!(output.stderr_truncated); assert!(output.log_backpressured); }