refactor(logs): replace pending_durable_logs vec with held_logs map keyed by request_id

- pending_durable_logs: Vec<IntakeLog> → held_logs: HashMap<String, Vec<IntakeLog>>
  Logs are now grouped by request_id, enabling O(1) lookup and targeted draining.

- try_update_durable_map now returns bool (true = new entry added). The caller is
  responsible for draining held_logs when a new entry arrives, so the method stays
  focused on map maintenance only.

- New drain_held_for_request_id: when a request_id's durable context first becomes
  known, immediately moves all stashed logs for that request_id into ready_logs with
  the appropriate durable tags, without touching the rest of held_logs.

- New resolve_held_logs_on_durable_function_set: called at the exact moment
  is_durable_function transitions from None to Some(...) — inside PlatformInitStart
  and the PlatformStart fallback — rather than on every process() call. For
  Some(false) all held logs are flushed; for Some(true) the held logs are scanned for
  durable context and eligible ones are moved to ready_logs, ineligible ones remain.

- queue_log_after_rules updated: in the None state logs with a request_id go into
  held_logs[request_id]; in the Some(true) state logs with no durable context yet are
  also stashed in held_logs[request_id] instead of being dropped.

- process() no longer needs to drain pending logs on each call; that is now driven
  entirely by the state-transition helpers above.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: lym953

bottlecap/src/logs/lambda/processor.rs

@@ -43,8 +43,13 @@ pub struct LambdaProcessor {
     // Some(true) = durable function; apply durable ID filtering.
     // Some(false) = not a durable function; flush logs normally.
     is_durable_function: Option<bool>,
-    // Logs held while is_durable_function is None (race-condition guard)
-    pending_durable_logs: Vec<IntakeLog>,
+    // Logs held pending resolution, keyed by request_id.
+    // While is_durable_function is None every incoming log is stashed here so
+    // we can decide whether to filter/tag it once the flag is known.
+    // While is_durable_function is Some(true), logs whose request_id has no
+    // durable execution context yet are also stashed here; they are drained
+    // the moment that context arrives.
+    held_logs: HashMap<String, Vec<IntakeLog>>,
     // Maps request_id -> (durable_execution_id, durable_execution_name)
     durable_id_map: HashMap<String, (String, String)>,
     // Insertion order for FIFO eviction when map reaches capacity
@@ -116,7 +121,7 @@ impl LambdaProcessor {
             event_bus,
             is_managed_instance_mode,
             is_durable_function: None,
-            pending_durable_logs: Vec::new(),
+            held_logs: HashMap::new(),
             durable_id_map: HashMap::with_capacity(5),
             durable_id_order: VecDeque::with_capacity(5),
         }
@@ -202,6 +207,7 @@ impl LambdaProcessor {
                 let rv_arn = runtime_version_arn.unwrap_or("?".to_string()); // TODO: check what do containers display
 
                 self.is_durable_function = Some(rv.contains("DurableFunction"));
+                self.resolve_held_logs_on_durable_function_set();
 
                 Ok(Message::new(
                     format!("INIT_START Runtime Version: {rv} Runtime Version ARN: {rv_arn}"),
@@ -234,6 +240,7 @@ impl LambdaProcessor {
                 // processed invocation), treat as non-durable to avoid holding logs forever.
                 if self.is_durable_function.is_none() {
                     self.is_durable_function = Some(false);
+                    self.resolve_held_logs_on_durable_function_set();
                 }
 
                 let version = version.unwrap_or("$LATEST".to_string());
@@ -490,12 +497,15 @@ impl LambdaProcessor {
         }
     }
 
-    /// If the message is a JSON object with `durable_execution_id` and `durable_execution_name`
-    /// fields, inserts a mapping of `request_id` -> `(execution_id, execution_name)` into the
-    /// durable ID map.  Evicts the oldest entry when the map is at capacity (5).
-    fn try_update_durable_map(&mut self, request_id: &str, message: &str) {
+    /// Parses `message` as JSON and, if it contains both `durable_execution_id` and
+    /// `durable_execution_name` fields, inserts or updates the entry for `request_id` in the
+    /// durable ID map (evicting the oldest entry when the map is at capacity 5).
+    ///
+    /// Returns `true` if a brand-new entry was added (the caller may then drain `held_logs`
+    /// for that `request_id`).
+    fn try_update_durable_map(&mut self, request_id: &str, message: &str) -> bool {
         let Ok(serde_json::Value::Object(obj)) = serde_json::from_str(message) else {
-            return;
+            return false;
         };
         let execution_id = obj
             .get("durable_execution_id")
@@ -504,7 +514,8 @@ impl LambdaProcessor {
             .get("durable_execution_name")
             .and_then(serde_json::Value::as_str);
         if let (Some(id), Some(name)) = (execution_id, execution_name) {
-            if !self.durable_id_map.contains_key(request_id) {
+            let is_new = !self.durable_id_map.contains_key(request_id);
+            if is_new {
                 if self.durable_id_order.len() >= 5 {
                     if let Some(oldest) = self.durable_id_order.pop_front() {
                         self.durable_id_map.remove(&oldest);
@@ -514,6 +525,78 @@ impl LambdaProcessor {
             }
             self.durable_id_map
                 .insert(request_id.to_string(), (id.to_string(), name.to_string()));
+            return is_new;
+        }
+        false
+    }
+
+    /// Moves all logs held for `request_id` into `ready_logs`, tagging each with the
+    /// durable execution context that is now known for that request_id.
+    fn drain_held_for_request_id(&mut self, request_id: &str) {
+        let Some(held) = self.held_logs.remove(request_id) else {
+            return;
+        };
+        let tags_suffix = self
+            .durable_id_map
+            .get(request_id)
+            .map(|(id, name)| format!(",durable_execution_id:{id},durable_execution_name:{name}"));
+        // Borrow of durable_id_map is released here (tags_suffix is an owned String).
+        if let Some(suffix) = tags_suffix {
+            for mut log in held {
+                log.tags.push_str(&suffix);
+                if let Ok(s) = serde_json::to_string(&log) {
+                    drop(log);
+                    self.ready_logs.push(s);
+                }
+            }
+        }
+    }
+
+    /// Called once when `is_durable_function` transitions from `None` to `Some(...)`.
+    /// Drains every entry in `held_logs`, routing each batch according to the newly-known flag:
+    /// - `Some(false)` → flush all held logs immediately.
+    /// - `Some(true)`  → try to extract durable context from the held logs; those whose
+    ///                   request_id is now in the durable ID map are flushed with tags; the
+    ///                   rest stay in `held_logs` until their context arrives.
+    fn resolve_held_logs_on_durable_function_set(&mut self) {
+        let held = std::mem::take(&mut self.held_logs);
+        match self.is_durable_function {
+            Some(false) => {
+                for (_, logs) in held {
+                    for log in logs {
+                        if let Ok(s) = serde_json::to_string(&log) {
+                            self.ready_logs.push(s);
+                        }
+                    }
+                }
+            }
+            Some(true) => {
+                for (request_id, logs) in held {
+                    // Try to discover durable context from the held logs themselves.
+                    for log in &logs {
+                        self.try_update_durable_map(&request_id, &log.message.message);
+                    }
+                    let tags_suffix = self.durable_id_map.get(&request_id).map(|(id, name)| {
+                        format!(",durable_execution_id:{id},durable_execution_name:{name}")
+                    });
+                    // Borrow of durable_id_map released here.
+                    match tags_suffix {
+                        Some(suffix) => {
+                            for mut log in logs {
+                                log.tags.push_str(&suffix);
+                                if let Ok(s) = serde_json::to_string(&log) {
+                                    self.ready_logs.push(s);
+                                }
+                            }
+                        }
+                        None => {
+                            // No context yet — put back and wait for it to arrive.
+                            self.held_logs.insert(request_id, logs);
+                        }
+                    }
+                }
+            }
+            None => {} // Should not happen; guard against misuse.
         }
     }
 
@@ -529,29 +612,45 @@ impl LambdaProcessor {
     /// Queues a log that has already had processing rules applied.
     ///
     /// Routing depends on `is_durable_function`:
-    /// - `None`         → hold in `pending_durable_logs` until the flag is resolved.
-    /// - `Some(false)`  → serialize and push straight to `ready_logs`.
-    /// - `Some(true)`   → apply durable-ID filtering: only send if the request_id is
-    ///                    already in the durable ID map, and append the execution tags.
+    /// - `None`        → stash in `held_logs[request_id]`; logs without a request_id are
+    ///                   flushed immediately since they cannot carry durable context.
+    /// - `Some(false)` → serialize and push straight to `ready_logs`.
+    /// - `Some(true)`  → try to update `durable_id_map` from the log; if a new entry was
+    ///                   added, drain `held_logs` for that request_id; then flush this log if
+    ///                   its request_id is in the map, otherwise stash it in `held_logs`.
     fn queue_log_after_rules(&mut self, mut log: IntakeLog) {
         match self.is_durable_function {
             None => {
-                // Not yet known whether this is a durable function — hold the log.
-                self.pending_durable_logs.push(log);
+                match log.message.lambda.request_id.clone() {
+                    Some(rid) => {
+                        self.held_logs.entry(rid).or_default().push(log);
+                    }
+                    None => {
+                        // No request_id — cannot associate with durable context; flush now.
+                        if let Ok(s) = serde_json::to_string(&log) {
+                            drop(log);
+                            self.ready_logs.push(s);
+                        }
+                    }
+                }
             }
             Some(false) => {
                 if let Ok(serialized_log) = serde_json::to_string(&log) {
+                    // explicitly drop log so we don't accidentally re-use it and push
+                    // duplicate logs to the aggregator
                     drop(log);
                     self.ready_logs.push(serialized_log);
                 }
             }
             Some(true) => {
-                // Populate the durable ID map from this log if it carries execution context.
-                if let Some(request_id) = log.message.lambda.request_id.clone() {
-                    self.try_update_durable_map(&request_id, &log.message.message);
+                if let Some(rid) = log.message.lambda.request_id.clone() {
+                    if self.try_update_durable_map(&rid, &log.message.message) {
+                        // New durable context just discovered — drain previously held logs.
+                        self.drain_held_for_request_id(&rid);
+                    }
                 }
 
-                // Only flush logs whose request_id is already in the durable ID map.
+                // Flush this log if its request_id now has durable context; otherwise hold.
                 let durable_tags = log
                     .message
                     .lambda
@@ -562,14 +661,22 @@ impl LambdaProcessor {
                         format!(",durable_execution_id:{id},durable_execution_name:{name}")
                     });
 
-                let Some(extra_tags) = durable_tags else {
-                    return;
-                };
-
-                log.tags.push_str(&extra_tags);
-                if let Ok(serialized_log) = serde_json::to_string(&log) {
-                    drop(log);
-                    self.ready_logs.push(serialized_log);
+                match durable_tags {
+                    Some(extra_tags) => {
+                        log.tags.push_str(&extra_tags);
+                        if let Ok(serialized_log) = serde_json::to_string(&log) {
+                            // explicitly drop log so we don't accidentally re-use it and push
+                            // duplicate logs to the aggregator
+                            drop(log);
+                            self.ready_logs.push(serialized_log);
+                        }
+                    }
+                    None => {
+                        if let Some(rid) = log.message.lambda.request_id.clone() {
+                            self.held_logs.entry(rid).or_default().push(log);
+                        }
+                        // Logs without a request_id cannot match the durable map; drop them.
+                    }
                 }
             }
         }
@@ -579,17 +686,6 @@ impl LambdaProcessor {
         if let Ok(log) = self.make_log(event).await {
             self.process_and_queue_log(log);
 
-            // If is_durable_function was just resolved, drain any logs that were held
-            // while the flag was still None (race-condition guard).
-            if self.is_durable_function.is_some() && !self.pending_durable_logs.is_empty() {
-                let pending = std::mem::take(&mut self.pending_durable_logs);
-                for pending_log in pending {
-                    // Rules were already applied before the log entered pending_durable_logs,
-                    // so go straight to queue_log_after_rules to avoid double-application.
-                    self.queue_log_after_rules(pending_log);
-                }
-            }
-
             // Process orphan logs, since we have a `request_id` now
             let orphan_logs = std::mem::take(&mut self.orphan_logs);
             for mut orphan_log in orphan_logs {