fix: deadlock in main loop (#696)

Fixes an issue I made in #692. Like all concurrency bugs, this one is
insidious.

At a glance combining `next_event` with `handle_next_invocation` makes a
lot of sense. I wish I could blame it on an LLM, but I had the bright
idea this morning as I reviewed the code.

In the flush_at_end and start cases, it works totally fine.
The problem is that in the high-throughput case where we switch to
periodic mode, we had a tokio::select! block polling over multiple
futures.

In #692 with the combined next_event and handle_next_invocation method,
we end up polling over a future which both waited for the response from
/next_event but then immediately blocked until it could lock the
invocation processor and insert the new request.

Unfortunately this caused a deadlock because inside the tokio::select!
block, we also lock the invocation processor when we need to handle some
event bus events.

If the response from next_event returns and tokio enters that branch, it
may not return if it can't get the lock. But if a request task is
inserting data into the invocation processor, we can deadlock and never
poll that branch again.

![image](https://github.com/user-attachments/assets/cb5795cb-02d3-4298-8f4c-7f28eaf7c7b7)

![image](https://github.com/user-attachments/assets/361ae154-d3b2-478e-b731-e8ada2ba4b8a)

Author: astuyve

bottlecap/src/bin/bottlecap/main.rs

@@ -495,10 +495,11 @@ async fn extension_loop_active(
         "Datadog Next-Gen Extension ready in {:}ms",
         start_time.elapsed().as_millis().to_string()
     );
+    let next_lambda_response = next_event(client, &r.extension_id).await;
     // first invoke we must call next
     let mut pending_flush_handles = PendingFlushHandles::new();
     let mut last_continuous_flush_error = false;
-    handle_next_invocation(client, &r.extension_id, invocation_processor.clone()).await;
+    handle_next_invocation(next_lambda_response, invocation_processor.clone()).await;
     loop {
         let maybe_shutdown_event;
 
@@ -541,8 +542,9 @@ async fn extension_loop_active(
                 &mut race_flush_interval,
             )
             .await;
+            let next_response = next_event(client, &r.extension_id).await;
             maybe_shutdown_event =
-                handle_next_invocation(client, &r.extension_id, invocation_processor.clone()).await;
+                handle_next_invocation(next_response, invocation_processor.clone()).await;
         } else {
             //Periodic flush scenario, flush at top of invocation
             if current_flush_decision == FlushDecision::Continuous && !last_continuous_flush_error {
@@ -592,8 +594,7 @@ async fn extension_loop_active(
             // If we get platform.runtimeDone or platform.runtimeReport
             // That's fine, we still wait to break until we get the response from next
             // and then we break to determine if we'll flush or not
-            let next_lambda_response =
-                handle_next_invocation(client, &r.extension_id, invocation_processor.clone());
+            let next_lambda_response = next_event(client, &r.extension_id);
             tokio::pin!(next_lambda_response);
             'next_invocation: loop {
                 tokio::select! {
@@ -607,7 +608,7 @@ async fn extension_loop_active(
                         race_flush_interval.reset();
                         // Thank you for not removing race_flush_interval.reset();
 
-                        maybe_shutdown_event= next_response;
+                        maybe_shutdown_event = handle_next_invocation(next_response, invocation_processor.clone()).await;
                         // Need to break here to re-call next
                         break 'next_invocation;
                     }
@@ -766,11 +767,9 @@ async fn handle_event_bus_event(
 }
 
 async fn handle_next_invocation(
-    client: &Client,
-    ext_id: &str,
+    next_response: Result<NextEventResponse>,
     invocation_processor: Arc<TokioMutex<InvocationProcessor>>,
 ) -> NextEventResponse {
-    let next_response = next_event(client, ext_id).await;
     match next_response {
         Ok(NextEventResponse::Invoke {
             ref request_id,