feat(logs): [SVLS-8582] Hold logs and add durable context to durable function logs (#1053)

## Summary

If the function is a durable function, then add two attributes to every
log:
- `lambda.durable_function.execution_id`
- `lambda.durable_function.execution_name`

## Background
- In previous PRs
(DataDog/datadog-lambda-python#728,
DataDog/datadog-lambda-js#730), tracer adds
attributes `aws_lambda.durable_function.execution_id` and
`aws_lambda.durable_function.execution_name` to the `aws.lambda` span

## Details

### Data flow
1. `TraceAgent::handle_traces()` detects an `aws.lambda` span carrying
`request_id`, `durable_function_execution_id`, and
`durable_function_execution_name` in its meta tags
2. It sends a `ProcessorCommand::ForwardDurableContext { request_id,
execution_id, execution_name }` to `InvocationProcessorService`
3. `Processor::forward_durable_context()` in the lifecycle processor
relays this as a `DurableContextUpdate` to the logs pipeline via an mpsc
channel, using `send().await` to guarantee delivery
4. `LogsAgent::spin()` receives the update and calls
`LogsProcessor::process_durable_context_update()`, which inserts the
entry into `LambdaProcessor::durable_context_map` and drains any held
logs for that `request_id`

### Log holding and draining
- After cold start, the logs processor holds all incoming logs without
flushing them, because it does not yet know whether this is a durable
function
- Held logs are stored in `held_logs: HashMap<String, Vec<IntakeLog>>`,
keyed by `request_id`
- Logs without a `request_id` (e.g. in managed instance mode) are pushed
directly to `ready_logs` and never held, since they cannot carry durable
context
- `durable_context_map: HashMap<String, DurableExecutionContext>` maps
`request_id` to `(execution_id, execution_name)`. It has a fixed
capacity (500 entries) with FIFO eviction
- When the logs processor receives a `PlatformInitStart` event, it
learns whether the function is a durable function:
- If **not** a durable function: drain all held logs (mark them ready
for aggregation and flush)
- If **durable**: drain held logs whose `request_id` is already in
`durable_context_map` (tag them with
`lambda.durable_function.execution_id` and
`lambda.durable_function.execution_name`); keep the rest held until
their context arrives
- When an entry is inserted into `durable_context_map`, any held logs
for that `request_id` are drained immediately

### Memory safety and resilience
- `held_logs` is capped at **50 keys** (intentionally small — see
below). Insertion order is tracked in `held_logs_order:
VecDeque<String>` for FIFO eviction
- When `held_logs` is at capacity and a new `request_id` arrives, the
**oldest key is evicted**: its logs are serialized and pushed to
`ready_logs` without durable context tags. This ensures logs are always
eventually sent to Datadog even if the tracer is not installed and
context never arrives
- The cap is kept small (50) to limit the size of the batch flushed at
shutdown, reducing the risk of the final flush timing out when held logs
are drained without durable context
- At **shutdown**, after draining the telemetry channel:
1. The `durable_context_rx` channel is drained to apply any pending
context updates, maximising the chance logs are decorated before
flushing
2. All remaining `held_logs` are drained to `ready_logs` without durable
context tags, so no logs are lost

### Types
- `DurableContextUpdate { request_id, execution_id, execution_name }` —
message sent from trace agent through lifecycle processor to logs
pipeline
- `DurableExecutionContext { execution_id, execution_name }` — value
type stored in `durable_context_map`

## Test plan

### Manual test

#### Steps
Build a layer, install it on a function, and invoke it.

#### Result
1. In Datadog, all the logs for this durable execution have the two new
attributes
<img width="734" height="421" alt="image"
src="https://github.com/user-attachments/assets/173e3be2-8bb1-4e08-be97-521c63679bf1"
/>


2. The logs query
> source:lambda
@lambda.arn:"arn:aws:lambda:us-east-2:425362996713:function:yiming-durable-py-custom-tracer"
@lambda.durable_function.execution_name:c949fb3d-a8f5-4ae6-a802-b1458149a4b2

returns all the logs for two invocations of this durable execution. It
returns 98 logs, equal to 49 logs for the first invocation + 49 logs for
the second invocation. ([query
link](https://ddserverless.datadoghq.com/logs?query=source%3Alambda%20%40lambda.arn%3A%22arn%3Aaws%3Alambda%3Aus-east-2%3A425362996713%3Afunction%3Ayiming-durable%22%20%40lambda.durable_execution_name%3A2f492839-75df-4acb-9f2a-30b1b36d5c8f&agg_m=count&agg_m_source=base&agg_t=count&cols=host%2Cservice&fromUser=true&messageDisplay=inline&refresh_mode=paused&storage=flex_tier&stream_sort=time%2Cdesc&viz=stream&from_ts=1772655235033&to_ts=1772655838416&live=false))

### Unit tests
Passed the added unit tests

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: lym953

bottlecap/src/bin/bottlecap/main.rs

@@ -51,6 +51,7 @@ use bottlecap::{
             AggregatorHandle as LogsAggregatorHandle, AggregatorService as LogsAggregatorService,
         },
         flusher::LogsFlusher,
+        lambda::DurableContextUpdate,
     },
     otlp::{agent::Agent as OtlpAgent, should_enable_otlp_agent},
     proxy::{interceptor, should_start_proxy},
@@ -298,15 +299,20 @@ async fn extension_loop_active(
     // and shares the connection pool.
     let shared_client = bottlecap::http::get_client(config);
 
-    let (logs_agent_channel, logs_flusher, logs_agent_cancel_token, logs_aggregator_handle) =
-        start_logs_agent(
-            config,
-            Arc::clone(&api_key_factory),
-            &tags_provider,
-            event_bus_tx.clone(),
-            aws_config.is_managed_instance_mode(),
-            &shared_client,
-        );
+    let (
+        logs_agent_channel,
+        logs_flusher,
+        logs_agent_cancel_token,
+        logs_aggregator_handle,
+        durable_context_tx,
+    ) = start_logs_agent(
+        config,
+        Arc::clone(&api_key_factory),
+        &tags_provider,
+        event_bus_tx.clone(),
+        aws_config.is_managed_instance_mode(),
+        &shared_client,
+    );
 
     let (metrics_flushers, metrics_aggregator_handle, dogstatsd_cancel_token) = start_dogstatsd(
         tags_provider.clone(),
@@ -325,6 +331,7 @@ async fn extension_loop_active(
             Arc::clone(&aws_config),
             metrics_aggregator_handle.clone(),
             Arc::clone(&propagator),
+            durable_context_tx,
         );
     tokio::spawn(async move {
         invocation_processor_service.run().await;
@@ -1039,14 +1046,15 @@ fn start_logs_agent(
     LogsFlusher,
     CancellationToken,
     LogsAggregatorHandle,
+    Sender<DurableContextUpdate>,
 ) {
     let (aggregator_service, aggregator_handle) = LogsAggregatorService::default();
     // Start service in background
     tokio::spawn(async move {
         aggregator_service.run().await;
     });
 
-    let (mut agent, tx) = LogsAgent::new(
+    let (mut agent, tx, durable_context_tx) = LogsAgent::new(
         Arc::clone(tags_provider),
         Arc::clone(config),
         event_bus,
@@ -1068,7 +1076,13 @@ fn start_logs_agent(
         config.clone(),
         client.clone(),
     );
-    (tx, flusher, cancel_token, aggregator_handle)
+    (
+        tx,
+        flusher,
+        cancel_token,
+        aggregator_handle,
+        durable_context_tx,
+    )
 }
 
 #[allow(clippy::type_complexity)]

bottlecap/src/lifecycle/invocation/processor.rs

@@ -16,7 +16,9 @@ use libdd_trace_protobuf::pb::Span;
 use libdd_trace_utils::tracer_header_tags;
 use serde_json::Value;
 use tokio::time::Instant;
-use tracing::{debug, trace, warn};
+use tracing::{debug, error, trace, warn};
+
+use tokio::sync::mpsc;
 
 use crate::{
     config::{self, aws::AwsConfig},
@@ -31,6 +33,7 @@ use crate::{
         span_inferrer::{self, SpanInferrer},
         triggers::get_default_service_name,
     },
+    logs::lambda::DurableContextUpdate,
     metrics::enhanced::lambda::{EnhancedMetricData, Lambda as EnhancedMetrics},
     proc::{
         self, CPUData, NetworkData,
@@ -89,6 +92,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,
+    /// 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.
+    durable_context_tx: mpsc::Sender<DurableContextUpdate>,
 }
 
 impl Processor {
@@ -99,6 +106,7 @@ impl Processor {
         aws_config: Arc<AwsConfig>,
         metrics_aggregator: dogstatsd::aggregator::AggregatorHandle,
         propagator: Arc<DatadogCompositePropagator>,
+        durable_context_tx: mpsc::Sender<DurableContextUpdate>,
     ) -> Self {
         let resource = tags_provider
             .get_canonical_resource_name()
@@ -128,6 +136,7 @@ impl Processor {
             dynamic_tags: HashMap::new(),
             active_invocations: 0,
             awaiting_first_invocation: false,
+            durable_context_tx,
         }
     }
 
@@ -1358,6 +1367,27 @@ impl Processor {
                 .add_tracer_span(request_id, span, client_computed_stats);
         }
     }
+
+    /// Forwards durable execution context extracted from an `aws.lambda` span to the logs
+    /// pipeline so it can release held logs and tag them with durable execution metadata.
+    pub async fn forward_durable_context(
+        &mut self,
+        request_id: &str,
+        execution_id: &str,
+        execution_name: &str,
+    ) {
+        if let Err(e) = self
+            .durable_context_tx
+            .send(DurableContextUpdate {
+                request_id: request_id.to_owned(),
+                execution_id: execution_id.to_owned(),
+                execution_name: execution_name.to_owned(),
+            })
+            .await
+        {
+            error!("Invocation Processor | Failed to forward durable context to logs agent: {e}");
+        }
+    }
 }
 
 #[cfg(test)]
@@ -1403,7 +1433,15 @@ mod tests {
         tokio::spawn(service.run());
 
         let propagator = Arc::new(DatadogCompositePropagator::new(Arc::clone(&config)));
-        Processor::new(tags_provider, config, aws_config, handle, propagator)
+        let (durable_context_tx, _) = tokio::sync::mpsc::channel(1);
+        Processor::new(
+            tags_provider,
+            config,
+            aws_config,
+            handle,
+            propagator,
+            durable_context_tx,
+        )
     }
 
     #[test]
@@ -1940,7 +1978,15 @@ mod tests {
 
         let propagator = Arc::new(DatadogCompositePropagator::new(Arc::clone(&config)));
 
-        let processor = Processor::new(tags_provider, config, aws_config, handle, propagator);
+        let (durable_context_tx, _) = tokio::sync::mpsc::channel(1);
+        let processor = Processor::new(
+            tags_provider,
+            config,
+            aws_config,
+            handle,
+            propagator,
+            durable_context_tx,
+        );
 
         assert!(
             processor.is_managed_instance_mode(),
@@ -2160,12 +2206,14 @@ mod tests {
             AggregatorService::new(EMPTY_TAGS, 1024).expect("failed to create aggregator service");
         tokio::spawn(aggregator_service.run());
         let propagator = Arc::new(DatadogCompositePropagator::new(Arc::clone(&config)));
+        let (durable_context_tx, _) = tokio::sync::mpsc::channel(1);
         let mut p = Processor::new(
             Arc::clone(&tags_provider),
             Arc::clone(&config),
             aws_config,
             aggregator_handle,
             propagator,
+            durable_context_tx,
         );
 
         let (trace_tx, mut trace_rx) = mpsc::channel(10);

bottlecap/src/lifecycle/invocation/processor_service.rs

@@ -18,6 +18,7 @@ use crate::{
         context::{Context, ReparentingInfo},
         processor::Processor,
     },
+    logs::lambda::DurableContextUpdate,
     tags::provider,
     traces::{propagation::DatadogCompositePropagator, trace_processor::SendingTraceProcessor},
 };
@@ -110,6 +111,11 @@ pub enum ProcessorCommand {
         span: Box<Span>,
         client_computed_stats: bool,
     },
+    ForwardDurableContext {
+        request_id: String,
+        execution_id: String,
+        execution_name: String,
+    },
     OnOutOfMemoryError {
         timestamp: i64,
     },
@@ -381,6 +387,21 @@ impl InvocationProcessorHandle {
             .await
     }
 
+    pub async fn forward_durable_context(
+        &self,
+        request_id: String,
+        execution_id: String,
+        execution_name: String,
+    ) -> Result<(), mpsc::error::SendError<ProcessorCommand>> {
+        self.sender
+            .send(ProcessorCommand::ForwardDurableContext {
+                request_id,
+                execution_id,
+                execution_name,
+            })
+            .await
+    }
+
     pub async fn on_out_of_memory_error(
         &self,
         timestamp: i64,
@@ -431,6 +452,7 @@ impl InvocationProcessorService {
         aws_config: Arc<AwsConfig>,
         metrics_aggregator_handle: AggregatorHandle,
         propagator: Arc<DatadogCompositePropagator>,
+        durable_context_tx: mpsc::Sender<DurableContextUpdate>,
     ) -> (InvocationProcessorHandle, Self) {
         let (sender, receiver) = mpsc::channel(1000);
 
@@ -440,6 +462,7 @@ impl InvocationProcessorService {
             aws_config,
             metrics_aggregator_handle,
             propagator,
+            durable_context_tx,
         );
 
         let handle = InvocationProcessorHandle { sender };
@@ -592,6 +615,15 @@ impl InvocationProcessorService {
                 } => {
                     self.processor.add_tracer_span(&span, client_computed_stats);
                 }
+                ProcessorCommand::ForwardDurableContext {
+                    request_id,
+                    execution_id,
+                    execution_name,
+                } => {
+                    self.processor
+                        .forward_durable_context(&request_id, &execution_id, &execution_name)
+                        .await;
+                }
                 ProcessorCommand::OnOutOfMemoryError { timestamp } => {
                     self.processor.on_out_of_memory_error(timestamp);
                 }

bottlecap/src/logs/agent.rs

@@ -12,9 +12,12 @@ use crate::{LAMBDA_RUNTIME_SLUG, config};
 
 const DRAIN_LOG_INTERVAL: Duration = Duration::from_millis(100);
 
+use crate::logs::lambda::DurableContextUpdate;
+
 #[allow(clippy::module_name_repetitions)]
 pub struct LogsAgent {
     rx: mpsc::Receiver<TelemetryEvent>,
+    durable_context_rx: mpsc::Receiver<DurableContextUpdate>,
     processor: LogsProcessor,
     aggregator_handle: AggregatorHandle,
     cancel_token: CancellationToken,
@@ -28,7 +31,7 @@ impl LogsAgent {
         event_bus: Sender<Event>,
         aggregator_handle: AggregatorHandle,
         is_managed_instance_mode: bool,
-    ) -> (Self, Sender<TelemetryEvent>) {
+    ) -> (Self, Sender<TelemetryEvent>, Sender<DurableContextUpdate>) {
         let processor = LogsProcessor::new(
             Arc::clone(&datadog_config),
             tags_provider,
@@ -38,16 +41,18 @@ impl LogsAgent {
         );
 
         let (tx, rx) = mpsc::channel::<TelemetryEvent>(1000);
+        let (durable_context_tx, durable_context_rx) = mpsc::channel::<DurableContextUpdate>(500);
         let cancel_token = CancellationToken::new();
 
         let agent = Self {
             rx,
+            durable_context_rx,
             processor,
             aggregator_handle,
             cancel_token,
         };
 
-        (agent, tx)
+        (agent, tx, durable_context_tx)
     }
 
     pub async fn spin(&mut self) {
@@ -56,10 +61,13 @@ impl LogsAgent {
                 Some(event) = self.rx.recv() => {
                     self.processor.process(event, &self.aggregator_handle).await;
                 }
+                Some(update) = self.durable_context_rx.recv() => {
+                    self.processor.process_durable_context_update(update, &self.aggregator_handle);
+                }
                 () = self.cancel_token.cancelled() => {
                     debug!("LOGS_AGENT | Received shutdown signal, draining remaining events");
 
-                    // Drain remaining events
+                    // Drain remaining telemetry events
                     let mut last_drain_log_time = Instant::now().checked_sub(DRAIN_LOG_INTERVAL).expect("Failed to subtract interval from now");
                     'drain_logs_loop: loop {
                         match self.rx.try_recv() {
@@ -82,6 +90,15 @@ impl LogsAgent {
                         }
                     }
 
+                    // Drain any pending durable context updates before flushing held logs,
+                    // to maximise the chance of decorating logs with execution context.
+                    while let Ok(update) = self.durable_context_rx.try_recv() {
+                        self.processor.process_durable_context_update(update, &self.aggregator_handle);
+                    }
+
+                    // Drain remaining held logs without durable context tags so no logs are lost.
+                    self.processor.drain_held_logs(&self.aggregator_handle);
+
                     break;
                 }
             }

bottlecap/src/logs/aggregator.rs

@@ -106,6 +106,7 @@ mod tests {
                 lambda: Lambda {
                     arn: "arn".to_string(),
                     request_id: Some("request_id".to_string()),
+                    ..Lambda::default()
                 },
                 timestamp: 0,
                 status: "status".to_string(),
@@ -130,6 +131,7 @@ mod tests {
                 lambda: Lambda {
                     arn: "arn".to_string(),
                     request_id: Some("request_id".to_string()),
+                    ..Lambda::default()
                 },
                 timestamp: 0,
                 status: "status".to_string(),
@@ -156,6 +158,7 @@ mod tests {
                 lambda: Lambda {
                     arn: "arn".to_string(),
                     request_id: Some("request_id".to_string()),
+                    ..Lambda::default()
                 },
                 timestamp: 0,
                 status: "status".to_string(),
@@ -196,6 +199,7 @@ mod tests {
                 lambda: Lambda {
                     arn: "arn".to_string(),
                     request_id: Some("request_id".to_string()),
+                    ..Lambda::default()
                 },
                 timestamp: 0,
                 status: "status".to_string(),

bottlecap/src/logs/aggregator_service.rs

@@ -122,6 +122,7 @@ mod tests {
                 lambda: Lambda {
                     arn: "arn".to_string(),
                     request_id: Some("request_id".to_string()),
+                    ..Lambda::default()
                 },
                 timestamp: 0,
                 status: "status".to_string(),