Add a few safety checks to the new concurrent Queue structures and the promise maps in the transport layer

src/coordinator/coordinator_worker.hpp

@@ -98,6 +98,7 @@ class Queue {
   }
 
   Message Pop() {
+    MG_ASSERT(inner_.use_count() > 0);
     std::unique_lock<std::mutex> lock(inner_->mu);
 
     while (inner_->queue.empty()) {

src/io/local_transport/local_transport_handle.hpp

@@ -160,8 +160,7 @@ class LocalTransportHandle {
       OpaquePromise opaque_promise(std::move(promise).ToUnique());
       DeadlineAndOpaquePromise dop{.requested_at = now, .deadline = deadline, .promise = std::move(opaque_promise)};
 
-      // TODO(tyler) assert not already present
-
+      MG_ASSERT(!promises_.contains(promise_key));
       promises_.emplace(std::move(promise_key), std::move(dop));
     }  // lock dropped
 

src/io/simulator/simulator_handle.hpp

@@ -129,6 +129,9 @@ class SimulatorHandle {
         .deadline = deadline,
         .promise = std::move(opaque_promise),
     };
+
+    MG_ASSERT(!promises_.contains(promise_key));
+
     promises_.emplace(std::move(promise_key), std::move(dop));
 
     stats_.total_messages++;

src/storage/v3/shard_worker.hpp

@@ -102,6 +102,7 @@ class Queue {
   }
 
   Message Pop() {
+    MG_ASSERT(inner_.use_count() > 0);
     std::unique_lock<std::mutex> lock(inner_->mu);
 
     while (inner_->queue.empty()) {