From 251e7180b93855bd6d7be6a41f25cb3df4c3cf8c Mon Sep 17 00:00:00 2001 From: Darcy Rayner Date: Fri, 23 Oct 2026 14:29:42 -0400 Subject: [PATCH 1/3] fix(shutdown): count timeouts and failures whose runtimeDone never arrived After the shutdown drain, send the processor the SHUTDOWN reason. For a TIMEOUT or FAILURE shutdown in On-Demand mode, if the last invoked request never got a platform.runtimeDone, count its error (and timeout), and mark it so a runtimeDone can't count it again. Awaiting the processor's reply also guarantees the earlier fire-and-forget shutdown and OOM commands are handled before the final flush. Also handle platform.runtimeDone events without a metrics field instead of dropping them, which lost their error counts and stalled End-mode flushing. Co-Authored-By: Claude Opus 5.5 --- bottlecap/src/bin/bottlecap/main.rs | 17 +- .../src/lifecycle/invocation/processor.rs | 200 +++++++++++++++++- .../lifecycle/invocation/processor_service.rs | 52 ++++- 3 files changed, 256 insertions(+), 13 deletions(-) diff --git a/bottlecap/src/bin/bottlecap/main.rs b/bottlecap/src/bin/bottlecap/main.rs index f47bcd26f..7d4d5211f 100644 --- a/bottlecap/src/bin/bottlecap/main.rs +++ b/bottlecap/src/bin/bottlecap/main.rs @@ -805,7 +805,11 @@ async fn extension_loop_active( } } - if let NextEventResponse::Shutdown { .. } = maybe_shutdown_event { + if let NextEventResponse::Shutdown { + ref shutdown_reason, + .. + } = maybe_shutdown_event + { // Stop the Telemetry API listener, then let the logs agent drain. The agent sends the // Tombstone only after it has forwarded every telemetry event to the event bus. telemetry_listener_cancel_token.cancel(); @@ -829,6 +833,15 @@ async fn extension_loop_active( ) .await; + // Count a timeout or failure whose runtimeDone never arrived. Awaiting the reply also + // ensures the processor has handled every earlier command before the final flush. + if let Err(e) = invocation_processor_handle + .on_shutdown_reason(shutdown_reason.clone(), chrono::Utc::now().timestamp()) + .await + { + error!("Failed to send shutdown reason to processor: {e}"); + } + // Cancel background services cancel_background_services( api_runtime_proxy_shutdown_signal.as_ref(), @@ -1000,7 +1013,7 @@ async fn handle_event_bus_event( } TelemetryRecord::PlatformRuntimeDone { ref request_id, - metrics: Some(metrics), + metrics, status, ref error_type, .. diff --git a/bottlecap/src/lifecycle/invocation/processor.rs b/bottlecap/src/lifecycle/invocation/processor.rs index 91856ea21..b3944b0cd 100644 --- a/bottlecap/src/lifecycle/invocation/processor.rs +++ b/bottlecap/src/lifecycle/invocation/processor.rs @@ -98,6 +98,9 @@ pub struct Processor { /// Tracks whether if first invocation after init has been received in Managed Instance mode. /// Used to determine if we should search for the empty context on an invocation. awaiting_first_invocation: bool, + /// Request ID of the most recent INVOKE. In On-Demand mode this is the invocation still in + /// flight when Lambda shuts the environment down. + last_invoked_request_id: Option, /// Sender used to forward durable execution context extracted from `aws.lambda` spans to the /// logs agent. Decouples the trace agent from the logs agent: the trace agent sends spans /// to the lifecycle processor, which extracts durable context and relays it here. @@ -183,6 +186,7 @@ impl Processor { dynamic_tags: HashMap::new(), active_invocations: 0, awaiting_first_invocation: false, + last_invoked_request_id: None, durable_context_tx, restore_time: None, init_duration_metric_emitted: false, @@ -290,6 +294,7 @@ impl Processor { } } self.enhanced_metrics.set_invoked_received(); + self.last_invoked_request_id = Some(request_id.clone()); // Both modes: check for buffered UniversalInstrumentationStart by request_id first. // Falls back to FIFO for on-demand when the tracer does not send the header. @@ -580,7 +585,7 @@ impl Processor { pub async fn on_platform_runtime_done( &mut self, request_id: &String, - metrics: RuntimeDoneMetrics, + metrics: Option, status: Status, error_type: Option, tags_provider: Arc, @@ -588,8 +593,10 @@ impl Processor { timestamp: i64, ) { // Set the runtime duration metric - self.enhanced_metrics - .set_runtime_done_metrics(&metrics, timestamp); + if let Some(metrics) = &metrics { + self.enhanced_metrics + .set_runtime_done_metrics(metrics, timestamp); + } if status != Status::Success { // Increment the error metric @@ -614,6 +621,14 @@ impl Processor { self.enhanced_metrics.set_usage_enhanced_metrics(); } + // Without a duration the invocation span would be sent with zero length, so count the + // invocation as done but skip span processing. + let Some(metrics) = metrics else { + if let Some(context) = self.context_buffer.get_mut(request_id) { + context.runtime_done_received = true; + } + return; + }; self.context_buffer .add_runtime_duration(request_id, metrics.duration_ms); @@ -1014,7 +1029,7 @@ impl Processor { if self.context_buffer.get(request_id).is_some() { self.on_platform_runtime_done( request_id, - runtime_done_metrics, + Some(runtime_done_metrics), status, error_type, tags_provider, @@ -1097,6 +1112,42 @@ impl Processor { } } + /// Counts the error, and the timeout, for the invocation that caused a `timeout` or + /// `failure` shutdown when its `platform.runtimeDone` never arrived. Must run after the + /// remaining telemetry has been processed, so a late `runtimeDone` isn't double counted. + /// + /// A `failure` reset that lands within the telemetry buffering window after a successful + /// response, such as another extension crashing, is also counted, because its + /// `runtimeDone` is refused once the listener has shut down. + pub fn on_shutdown_reason(&mut self, reason: &str, timestamp: i64) { + if self.aws_config.is_managed_instance_mode() { + return; + } + let is_timeout = reason.eq_ignore_ascii_case("timeout"); + if !is_timeout && !reason.eq_ignore_ascii_case("failure") { + return; + } + let Some(context) = self + .last_invoked_request_id + .as_ref() + .and_then(|request_id| self.context_buffer.get_mut(request_id)) + else { + return; + }; + if context.runtime_done_received { + return; + } + context.runtime_done_received = true; + debug!( + "Invocation Processor | No runtimeDone for {} before {reason} shutdown, counting error", + context.request_id + ); + self.enhanced_metrics.increment_errors_metric(timestamp); + if is_timeout { + self.enhanced_metrics.increment_timeout_metric(timestamp); + } + } + /// If this method is called, it means that we are operating in a Universally Instrumented /// runtime. Therefore, we need to set the `tracer_detected` flag to `true`. /// @@ -1735,6 +1786,7 @@ mod tests { use super::*; use crate::LAMBDA_RUNTIME_SLUG; use crate::extension::telemetry::events::ManagedInstanceReportMetrics; + use crate::metrics::enhanced::constants::{ERRORS_METRIC, TIMEOUTS_METRIC}; use crate::traces::stats_concentrator_service::StatsConcentratorService; use crate::traces::stats_generator::StatsGenerator; use crate::traces::trace_processor; @@ -3195,10 +3247,10 @@ mod tests { p.on_platform_runtime_done( &request_id, - RuntimeDoneMetrics { + Some(RuntimeDoneMetrics { duration_ms: 100.0, produced_bytes: None, - }, + }), Status::Timeout, None, tags_provider, @@ -3714,4 +3766,140 @@ mod tests { ); } } + + async fn metric_sum(p: &Processor, metric: &str, timestamp: i64) -> Option { + p.enhanced_metrics + .aggr_handle + .get_entry_by_id(metric.into(), None, (timestamp / 10) * 10) + .await + .unwrap() + .map(|entry| { + entry + .value + .get_sketch() + .expect("distribution sketch") + .sum() + .expect("sketch sum") + }) + } + + fn setup_in_flight_invocation(request_id: &str) -> Processor { + let mut p = setup(); + p.context_buffer.start_context(request_id, Span::default()); + p.last_invoked_request_id = Some(request_id.to_string()); + p + } + + #[tokio::test] + async fn test_shutdown_reason_timeout_counts_missing_runtime_done_once() { + let mut p = setup_in_flight_invocation("req-timeout"); + let now = chrono::Utc::now().timestamp(); + + p.on_shutdown_reason("TIMEOUT", now); + p.on_shutdown_reason("TIMEOUT", now); + + assert_eq!(metric_sum(&p, ERRORS_METRIC, now).await, Some(1.0)); + assert_eq!(metric_sum(&p, TIMEOUTS_METRIC, now).await, Some(1.0)); + } + + #[tokio::test] + async fn test_shutdown_reason_resolves_last_invoked_request() { + let request_id = String::from("req-invoked"); + let mut p = setup(); + p.on_invoke_event(request_id.clone()); + + p.on_shutdown_reason("TIMEOUT", chrono::Utc::now().timestamp()); + + assert!( + p.context_buffer + .get(&request_id) + .expect("context") + .runtime_done_received + ); + } + + #[tokio::test] + async fn test_shutdown_reason_ignored_in_managed_instance_mode() { + let request_id = String::from("req-managed"); + let mut p = setup_managed_instance(); + p.on_invoke_event(request_id.clone()); + + p.on_shutdown_reason("TIMEOUT", chrono::Utc::now().timestamp()); + + assert!( + !p.context_buffer + .get(&request_id) + .expect("context") + .runtime_done_received + ); + } + + #[tokio::test] + async fn test_shutdown_reason_failure_counts_error_only() { + let mut p = setup_in_flight_invocation("req-failure"); + let now = chrono::Utc::now().timestamp(); + + p.on_shutdown_reason("FAILURE", now); + + assert_eq!(metric_sum(&p, ERRORS_METRIC, now).await, Some(1.0)); + assert_eq!(metric_sum(&p, TIMEOUTS_METRIC, now).await, None); + } + + #[tokio::test] + async fn test_shutdown_reason_ignored_when_runtime_done_received() { + let request_id = "req-done"; + let mut p = setup_in_flight_invocation(request_id); + p.context_buffer + .get_mut(&request_id.to_string()) + .expect("context") + .runtime_done_received = true; + let now = chrono::Utc::now().timestamp(); + + p.on_shutdown_reason("TIMEOUT", now); + + assert_eq!(metric_sum(&p, ERRORS_METRIC, now).await, None); + } + + #[tokio::test] + async fn test_shutdown_reason_spindown_counts_nothing() { + let mut p = setup_in_flight_invocation("req-spindown"); + let now = chrono::Utc::now().timestamp(); + + p.on_shutdown_reason("SPINDOWN", now); + + assert_eq!(metric_sum(&p, ERRORS_METRIC, now).await, None); + } + + #[tokio::test] + async fn test_runtime_done_without_metrics_counts_timeout_without_span() { + let request_id = String::from("req-no-metrics"); + let mut p = setup_in_flight_invocation(&request_id); + p.tracer_detected = true; + let config = Arc::clone(&p.config); + let tags_provider = Arc::new(provider::Provider::new( + Arc::clone(&config), + LAMBDA_RUNTIME_SLUG.to_string(), + &HashMap::from([("function_arn".to_string(), "test-arn".to_string())]), + )); + let (trace_sender, mut trace_rx) = make_trace_sender_with_rx(config); + let now = chrono::Utc::now().timestamp(); + + p.on_platform_runtime_done( + &request_id, + None, + Status::Timeout, + None, + tags_provider, + Arc::clone(&trace_sender), + now, + ) + .await; + // The runtimeDone was received, so a later TIMEOUT shutdown must not count again. + p.on_shutdown_reason("TIMEOUT", now); + + assert_eq!(metric_sum(&p, ERRORS_METRIC, now).await, Some(1.0)); + assert_eq!(metric_sum(&p, TIMEOUTS_METRIC, now).await, Some(1.0)); + // Without a duration, no zero-length invocation span is sent. + assert!(trace_rx.try_recv().is_err()); + } } diff --git a/bottlecap/src/lifecycle/invocation/processor_service.rs b/bottlecap/src/lifecycle/invocation/processor_service.rs index 093ca8bba..918aeec2e 100644 --- a/bottlecap/src/lifecycle/invocation/processor_service.rs +++ b/bottlecap/src/lifecycle/invocation/processor_service.rs @@ -59,7 +59,7 @@ pub enum ProcessorCommand { }, PlatformRuntimeDone { request_id: String, - metrics: RuntimeDoneMetrics, + metrics: Option, status: Status, error_type: Option, tags_provider: Arc, @@ -127,6 +127,11 @@ pub enum ProcessorCommand { timestamp: i64, }, OnShutdownEvent, + OnShutdownReason { + reason: String, + timestamp: i64, + response: oneshot::Sender<()>, + }, SendCtxSpans { tags_provider: Arc, trace_sender: Arc, @@ -180,7 +185,8 @@ impl InvocationProcessorHandle { let _ = response.send(Ok(None)); } ProcessorCommand::PlatformRuntimeDone { response, .. } - | ProcessorCommand::PlatformReport { response, .. } => { + | ProcessorCommand::PlatformReport { response, .. } + | ProcessorCommand::OnShutdownReason { response, .. } => { let _ = response.send(()); } // Fire-and-forget commands: drop silently. @@ -279,7 +285,7 @@ impl InvocationProcessorHandle { pub async fn on_platform_runtime_done( &self, request_id: String, - metrics: RuntimeDoneMetrics, + metrics: Option, status: Status, error_type: Option, tags_provider: Arc, @@ -508,6 +514,26 @@ impl InvocationProcessorHandle { self.sender.send(ProcessorCommand::OnShutdownEvent).await } + /// Resolves once the processor has handled this and every earlier command. + pub async fn on_shutdown_reason( + &self, + reason: String, + timestamp: i64, + ) -> Result<(), ProcessorError> { + let (tx, rx) = oneshot::channel(); + self.sender + .send(ProcessorCommand::OnShutdownReason { + reason, + timestamp, + response: tx, + }) + .await + .map_err(|e| ProcessorError::ChannelSend(e.to_string()))?; + rx.await + .map_err(|e| ProcessorError::ChannelReceive(e.to_string()))?; + Ok(()) + } + pub async fn send_ctx_spans( &self, tags_provider: &Arc, @@ -746,6 +772,14 @@ impl InvocationProcessorService { ProcessorCommand::OnShutdownEvent => { self.processor.on_shutdown_event(); } + ProcessorCommand::OnShutdownReason { + reason, + timestamp, + response, + } => { + self.processor.on_shutdown_reason(&reason, timestamp); + let _ = response.send(()); + } ProcessorCommand::SendCtxSpans { tags_provider, trace_sender, @@ -878,10 +912,10 @@ mod tests { timeout_dur, handle.on_platform_runtime_done( "rid".to_string(), - RuntimeDoneMetrics { + Some(RuntimeDoneMetrics { duration_ms: 0.0, produced_bytes: None, - }, + }), Status::Success, None, Arc::clone(&tags_provider), @@ -893,6 +927,14 @@ mod tests { .expect("on_platform_runtime_done timed out") .expect("on_platform_runtime_done"); + timeout( + timeout_dur, + handle.on_shutdown_reason("timeout".to_string(), 0), + ) + .await + .expect("on_shutdown_reason timed out") + .expect("on_shutdown_reason"); + timeout( timeout_dur, handle.on_platform_report( From 7af598c4c2c7cf35d61bf68d70134876261c8383 Mon Sep 17 00:00:00 2001 From: Darcy Rayner Date: Fri, 23 Oct 2026 16:11:41 -0400 Subject: [PATCH 2/3] fix(shutdown): track uncounted invocation separately from span readiness runtime_done_received also marks the invocation span ready to send during reparenting, so setting it when counting from the SHUTDOWN reason or a metricless runtimeDone could send an unfinished span. Track the uncounted invocation on the processor instead. Co-Authored-By: Claude Opus 5.5 --- .../src/lifecycle/invocation/processor.rs | 70 +++++++++---------- 1 file changed, 34 insertions(+), 36 deletions(-) diff --git a/bottlecap/src/lifecycle/invocation/processor.rs b/bottlecap/src/lifecycle/invocation/processor.rs index b3944b0cd..28517a9e5 100644 --- a/bottlecap/src/lifecycle/invocation/processor.rs +++ b/bottlecap/src/lifecycle/invocation/processor.rs @@ -98,9 +98,10 @@ pub struct Processor { /// Tracks whether if first invocation after init has been received in Managed Instance mode. /// Used to determine if we should search for the empty context on an invocation. awaiting_first_invocation: bool, - /// Request ID of the most recent INVOKE. In On-Demand mode this is the invocation still in - /// flight when Lambda shuts the environment down. - last_invoked_request_id: Option, + /// Request ID of the most recent INVOKE, until its errors/timeouts are counted from its + /// `platform.runtimeDone` or the SHUTDOWN reason. Kept separate from + /// `Context::runtime_done_received`, which also marks the invocation span ready to send. + uncounted_invocation: Option, /// Sender used to forward durable execution context extracted from `aws.lambda` spans to the /// logs agent. Decouples the trace agent from the logs agent: the trace agent sends spans /// to the lifecycle processor, which extracts durable context and relays it here. @@ -186,7 +187,7 @@ impl Processor { dynamic_tags: HashMap::new(), active_invocations: 0, awaiting_first_invocation: false, - last_invoked_request_id: None, + uncounted_invocation: None, durable_context_tx, restore_time: None, init_duration_metric_emitted: false, @@ -294,7 +295,7 @@ impl Processor { } } self.enhanced_metrics.set_invoked_received(); - self.last_invoked_request_id = Some(request_id.clone()); + self.uncounted_invocation = Some(request_id.clone()); // Both modes: check for buffered UniversalInstrumentationStart by request_id first. // Falls back to FIFO for on-demand when the tracer does not send the header. @@ -598,6 +599,10 @@ impl Processor { .set_runtime_done_metrics(metrics, timestamp); } + if self.uncounted_invocation.as_ref() == Some(request_id) { + self.uncounted_invocation = None; + } + if status != Status::Success { // Increment the error metric self.enhanced_metrics.increment_errors_metric(timestamp); @@ -621,12 +626,9 @@ impl Processor { self.enhanced_metrics.set_usage_enhanced_metrics(); } - // Without a duration the invocation span would be sent with zero length, so count the - // invocation as done but skip span processing. + // Without a duration the invocation span would be sent with zero length, so skip span + // processing. `runtime_done_received` stays false so reparenting doesn't send it either. let Some(metrics) = metrics else { - if let Some(context) = self.context_buffer.get_mut(request_id) { - context.runtime_done_received = true; - } return; }; self.context_buffer @@ -1127,20 +1129,14 @@ impl Processor { if !is_timeout && !reason.eq_ignore_ascii_case("failure") { return; } - let Some(context) = self - .last_invoked_request_id - .as_ref() - .and_then(|request_id| self.context_buffer.get_mut(request_id)) + let Some(request_id) = self + .uncounted_invocation + .take_if(|request_id| self.context_buffer.get(request_id).is_some()) else { return; }; - if context.runtime_done_received { - return; - } - context.runtime_done_received = true; debug!( - "Invocation Processor | No runtimeDone for {} before {reason} shutdown, counting error", - context.request_id + "Invocation Processor | No runtimeDone for {request_id} before {reason} shutdown, counting error" ); self.enhanced_metrics.increment_errors_metric(timestamp); if is_timeout { @@ -3786,7 +3782,7 @@ mod tests { fn setup_in_flight_invocation(request_id: &str) -> Processor { let mut p = setup(); p.context_buffer.start_context(request_id, Span::default()); - p.last_invoked_request_id = Some(request_id.to_string()); + p.uncounted_invocation = Some(request_id.to_string()); p } @@ -3807,11 +3803,14 @@ mod tests { let request_id = String::from("req-invoked"); let mut p = setup(); p.on_invoke_event(request_id.clone()); + assert_eq!(p.uncounted_invocation.as_ref(), Some(&request_id)); p.on_shutdown_reason("TIMEOUT", chrono::Utc::now().timestamp()); + assert_eq!(p.uncounted_invocation, None); + // Counting must not mark the span complete, or reparenting would send it unfinished. assert!( - p.context_buffer + !p.context_buffer .get(&request_id) .expect("context") .runtime_done_received @@ -3826,12 +3825,7 @@ mod tests { p.on_shutdown_reason("TIMEOUT", chrono::Utc::now().timestamp()); - assert!( - !p.context_buffer - .get(&request_id) - .expect("context") - .runtime_done_received - ); + assert_eq!(p.uncounted_invocation, Some(request_id)); } #[tokio::test] @@ -3846,13 +3840,10 @@ mod tests { } #[tokio::test] - async fn test_shutdown_reason_ignored_when_runtime_done_received() { - let request_id = "req-done"; - let mut p = setup_in_flight_invocation(request_id); - p.context_buffer - .get_mut(&request_id.to_string()) - .expect("context") - .runtime_done_received = true; + async fn test_shutdown_reason_ignored_after_context_released() { + let request_id = String::from("req-released"); + let mut p = setup_in_flight_invocation(&request_id); + p.context_buffer.remove(&request_id); let now = chrono::Utc::now().timestamp(); p.on_shutdown_reason("TIMEOUT", now); @@ -3899,7 +3890,14 @@ mod tests { assert_eq!(metric_sum(&p, ERRORS_METRIC, now).await, Some(1.0)); assert_eq!(metric_sum(&p, TIMEOUTS_METRIC, now).await, Some(1.0)); - // Without a duration, no zero-length invocation span is sent. + // Without a duration no zero-length invocation span is sent, and the span isn't marked + // complete, so reparenting won't send it either. assert!(trace_rx.try_recv().is_err()); + assert!( + !p.context_buffer + .get(&request_id) + .expect("context") + .runtime_done_received + ); } } From 44afb591b529081d7b9a19817ff88b88a0932149 Mon Sep 17 00:00:00 2001 From: Darcy Rayner Date: Fri, 23 Oct 2026 16:26:46 -0400 Subject: [PATCH 3/3] fix(shutdown): drop the SHUTDOWN-reason error count The failure-accounting suite showed the count double counting: errors were 3-5 over on the default-strategy function in every attempt, while #1455 alone was exact. A runtimeDone that the closing listener refuses is redelivered to the restarted extension, which counts it, so counting it again from the SHUTDOWN reason was wrong. Keep the rest: platform.runtimeDone without metrics is still counted (without sending a zero-length span), and shutdown now waits on a processor barrier so earlier fire-and-forget commands are handled before the final flush. Co-Authored-By: Claude Opus 5.5 --- bottlecap/src/bin/bottlecap/main.rs | 17 +-- .../src/lifecycle/invocation/processor.rs | 127 +----------------- .../lifecycle/invocation/processor_service.rs | 36 ++--- 3 files changed, 16 insertions(+), 164 deletions(-) diff --git a/bottlecap/src/bin/bottlecap/main.rs b/bottlecap/src/bin/bottlecap/main.rs index 7d4d5211f..969a25fa0 100644 --- a/bottlecap/src/bin/bottlecap/main.rs +++ b/bottlecap/src/bin/bottlecap/main.rs @@ -805,11 +805,7 @@ async fn extension_loop_active( } } - if let NextEventResponse::Shutdown { - ref shutdown_reason, - .. - } = maybe_shutdown_event - { + if let NextEventResponse::Shutdown { .. } = maybe_shutdown_event { // Stop the Telemetry API listener, then let the logs agent drain. The agent sends the // Tombstone only after it has forwarded every telemetry event to the event bus. telemetry_listener_cancel_token.cancel(); @@ -833,13 +829,10 @@ async fn extension_loop_active( ) .await; - // Count a timeout or failure whose runtimeDone never arrived. Awaiting the reply also - // ensures the processor has handled every earlier command before the final flush. - if let Err(e) = invocation_processor_handle - .on_shutdown_reason(shutdown_reason.clone(), chrono::Utc::now().timestamp()) - .await - { - error!("Failed to send shutdown reason to processor: {e}"); + // Make sure the processor has handled every earlier command, including the + // fire-and-forget shutdown and OOM events, before the final flush. + if let Err(e) = invocation_processor_handle.barrier().await { + error!("Failed to wait for the invocation processor: {e}"); } // Cancel background services diff --git a/bottlecap/src/lifecycle/invocation/processor.rs b/bottlecap/src/lifecycle/invocation/processor.rs index 28517a9e5..cfd03535b 100644 --- a/bottlecap/src/lifecycle/invocation/processor.rs +++ b/bottlecap/src/lifecycle/invocation/processor.rs @@ -98,10 +98,6 @@ pub struct Processor { /// Tracks whether if first invocation after init has been received in Managed Instance mode. /// Used to determine if we should search for the empty context on an invocation. awaiting_first_invocation: bool, - /// Request ID of the most recent INVOKE, until its errors/timeouts are counted from its - /// `platform.runtimeDone` or the SHUTDOWN reason. Kept separate from - /// `Context::runtime_done_received`, which also marks the invocation span ready to send. - uncounted_invocation: Option, /// Sender used to forward durable execution context extracted from `aws.lambda` spans to the /// logs agent. Decouples the trace agent from the logs agent: the trace agent sends spans /// to the lifecycle processor, which extracts durable context and relays it here. @@ -187,7 +183,6 @@ impl Processor { dynamic_tags: HashMap::new(), active_invocations: 0, awaiting_first_invocation: false, - uncounted_invocation: None, durable_context_tx, restore_time: None, init_duration_metric_emitted: false, @@ -295,7 +290,6 @@ impl Processor { } } self.enhanced_metrics.set_invoked_received(); - self.uncounted_invocation = Some(request_id.clone()); // Both modes: check for buffered UniversalInstrumentationStart by request_id first. // Falls back to FIFO for on-demand when the tracer does not send the header. @@ -599,10 +593,6 @@ impl Processor { .set_runtime_done_metrics(metrics, timestamp); } - if self.uncounted_invocation.as_ref() == Some(request_id) { - self.uncounted_invocation = None; - } - if status != Status::Success { // Increment the error metric self.enhanced_metrics.increment_errors_metric(timestamp); @@ -1114,36 +1104,6 @@ impl Processor { } } - /// Counts the error, and the timeout, for the invocation that caused a `timeout` or - /// `failure` shutdown when its `platform.runtimeDone` never arrived. Must run after the - /// remaining telemetry has been processed, so a late `runtimeDone` isn't double counted. - /// - /// A `failure` reset that lands within the telemetry buffering window after a successful - /// response, such as another extension crashing, is also counted, because its - /// `runtimeDone` is refused once the listener has shut down. - pub fn on_shutdown_reason(&mut self, reason: &str, timestamp: i64) { - if self.aws_config.is_managed_instance_mode() { - return; - } - let is_timeout = reason.eq_ignore_ascii_case("timeout"); - if !is_timeout && !reason.eq_ignore_ascii_case("failure") { - return; - } - let Some(request_id) = self - .uncounted_invocation - .take_if(|request_id| self.context_buffer.get(request_id).is_some()) - else { - return; - }; - debug!( - "Invocation Processor | No runtimeDone for {request_id} before {reason} shutdown, counting error" - ); - self.enhanced_metrics.increment_errors_metric(timestamp); - if is_timeout { - self.enhanced_metrics.increment_timeout_metric(timestamp); - } - } - /// If this method is called, it means that we are operating in a Universally Instrumented /// runtime. Therefore, we need to set the `tracer_detected` flag to `true`. /// @@ -3779,92 +3739,11 @@ mod tests { }) } - fn setup_in_flight_invocation(request_id: &str) -> Processor { - let mut p = setup(); - p.context_buffer.start_context(request_id, Span::default()); - p.uncounted_invocation = Some(request_id.to_string()); - p - } - - #[tokio::test] - async fn test_shutdown_reason_timeout_counts_missing_runtime_done_once() { - let mut p = setup_in_flight_invocation("req-timeout"); - let now = chrono::Utc::now().timestamp(); - - p.on_shutdown_reason("TIMEOUT", now); - p.on_shutdown_reason("TIMEOUT", now); - - assert_eq!(metric_sum(&p, ERRORS_METRIC, now).await, Some(1.0)); - assert_eq!(metric_sum(&p, TIMEOUTS_METRIC, now).await, Some(1.0)); - } - - #[tokio::test] - async fn test_shutdown_reason_resolves_last_invoked_request() { - let request_id = String::from("req-invoked"); - let mut p = setup(); - p.on_invoke_event(request_id.clone()); - assert_eq!(p.uncounted_invocation.as_ref(), Some(&request_id)); - - p.on_shutdown_reason("TIMEOUT", chrono::Utc::now().timestamp()); - - assert_eq!(p.uncounted_invocation, None); - // Counting must not mark the span complete, or reparenting would send it unfinished. - assert!( - !p.context_buffer - .get(&request_id) - .expect("context") - .runtime_done_received - ); - } - - #[tokio::test] - async fn test_shutdown_reason_ignored_in_managed_instance_mode() { - let request_id = String::from("req-managed"); - let mut p = setup_managed_instance(); - p.on_invoke_event(request_id.clone()); - - p.on_shutdown_reason("TIMEOUT", chrono::Utc::now().timestamp()); - - assert_eq!(p.uncounted_invocation, Some(request_id)); - } - - #[tokio::test] - async fn test_shutdown_reason_failure_counts_error_only() { - let mut p = setup_in_flight_invocation("req-failure"); - let now = chrono::Utc::now().timestamp(); - - p.on_shutdown_reason("FAILURE", now); - - assert_eq!(metric_sum(&p, ERRORS_METRIC, now).await, Some(1.0)); - assert_eq!(metric_sum(&p, TIMEOUTS_METRIC, now).await, None); - } - - #[tokio::test] - async fn test_shutdown_reason_ignored_after_context_released() { - let request_id = String::from("req-released"); - let mut p = setup_in_flight_invocation(&request_id); - p.context_buffer.remove(&request_id); - let now = chrono::Utc::now().timestamp(); - - p.on_shutdown_reason("TIMEOUT", now); - - assert_eq!(metric_sum(&p, ERRORS_METRIC, now).await, None); - } - - #[tokio::test] - async fn test_shutdown_reason_spindown_counts_nothing() { - let mut p = setup_in_flight_invocation("req-spindown"); - let now = chrono::Utc::now().timestamp(); - - p.on_shutdown_reason("SPINDOWN", now); - - assert_eq!(metric_sum(&p, ERRORS_METRIC, now).await, None); - } - #[tokio::test] async fn test_runtime_done_without_metrics_counts_timeout_without_span() { let request_id = String::from("req-no-metrics"); - let mut p = setup_in_flight_invocation(&request_id); + let mut p = setup(); + p.context_buffer.start_context(&request_id, Span::default()); p.tracer_detected = true; let config = Arc::clone(&p.config); let tags_provider = Arc::new(provider::Provider::new( @@ -3885,8 +3764,6 @@ mod tests { now, ) .await; - // The runtimeDone was received, so a later TIMEOUT shutdown must not count again. - p.on_shutdown_reason("TIMEOUT", now); assert_eq!(metric_sum(&p, ERRORS_METRIC, now).await, Some(1.0)); assert_eq!(metric_sum(&p, TIMEOUTS_METRIC, now).await, Some(1.0)); diff --git a/bottlecap/src/lifecycle/invocation/processor_service.rs b/bottlecap/src/lifecycle/invocation/processor_service.rs index 918aeec2e..55b0de954 100644 --- a/bottlecap/src/lifecycle/invocation/processor_service.rs +++ b/bottlecap/src/lifecycle/invocation/processor_service.rs @@ -127,9 +127,7 @@ pub enum ProcessorCommand { timestamp: i64, }, OnShutdownEvent, - OnShutdownReason { - reason: String, - timestamp: i64, + Barrier { response: oneshot::Sender<()>, }, SendCtxSpans { @@ -186,7 +184,7 @@ impl InvocationProcessorHandle { } ProcessorCommand::PlatformRuntimeDone { response, .. } | ProcessorCommand::PlatformReport { response, .. } - | ProcessorCommand::OnShutdownReason { response, .. } => { + | ProcessorCommand::Barrier { response } => { let _ = response.send(()); } // Fire-and-forget commands: drop silently. @@ -515,18 +513,10 @@ impl InvocationProcessorHandle { } /// Resolves once the processor has handled this and every earlier command. - pub async fn on_shutdown_reason( - &self, - reason: String, - timestamp: i64, - ) -> Result<(), ProcessorError> { + pub async fn barrier(&self) -> Result<(), ProcessorError> { let (tx, rx) = oneshot::channel(); self.sender - .send(ProcessorCommand::OnShutdownReason { - reason, - timestamp, - response: tx, - }) + .send(ProcessorCommand::Barrier { response: tx }) .await .map_err(|e| ProcessorError::ChannelSend(e.to_string()))?; rx.await @@ -772,12 +762,7 @@ impl InvocationProcessorService { ProcessorCommand::OnShutdownEvent => { self.processor.on_shutdown_event(); } - ProcessorCommand::OnShutdownReason { - reason, - timestamp, - response, - } => { - self.processor.on_shutdown_reason(&reason, timestamp); + ProcessorCommand::Barrier { response } => { let _ = response.send(()); } ProcessorCommand::SendCtxSpans { @@ -927,13 +912,10 @@ mod tests { .expect("on_platform_runtime_done timed out") .expect("on_platform_runtime_done"); - timeout( - timeout_dur, - handle.on_shutdown_reason("timeout".to_string(), 0), - ) - .await - .expect("on_shutdown_reason timed out") - .expect("on_shutdown_reason"); + timeout(timeout_dur, handle.barrier()) + .await + .expect("barrier timed out") + .expect("barrier"); timeout( timeout_dur,