crash if the on_connect callback is called twice

src/core/lib/event_engine/windows/windows_engine.cc

@@ -49,6 +49,13 @@
 namespace grpc_event_engine {
 namespace experimental {
 
+namespace {
+EventEngine::OnConnectCallback CreateCrashingOnConnectCallback() {
+  return [](absl::StatusOr<std::unique_ptr<EventEngine::Endpoint>>) {
+    grpc_core::Crash("Internal Error: OnConnect callback called when unset");
+  };
+}
+}  // namespace
 // ---- IOCPWorkClosure ----
 
 WindowsEventEngine::IOCPWorkClosure::IOCPWorkClosure(ThreadPool* thread_pool,
@@ -250,6 +257,7 @@ void WindowsEventEngine::OnConnectCompleted(
     // Connection attempt complete!
     grpc_core::MutexLock lock(&state->mu);
     cb = std::move(state->on_connected_user_callback);
+    state->on_connected_user_callback = CreateCrashingOnConnectCallback();
     state->on_connected = nullptr;
     {
       grpc_core::MutexLock handle_lock(&connection_mu_);
@@ -359,6 +367,8 @@ EventEngine::ConnectionHandle WindowsEventEngine::Connect(
         grpc_core::ReleasableMutexLock lock(&connection_state->mu);
         if (CancelConnectFromDeadlineTimer(connection_state.get())) {
           auto cb = std::move(connection_state->on_connected_user_callback);
+          connection_state->on_connected_user_callback =
+              CreateCrashingOnConnectCallback();
           lock.Release();
           cb(absl::DeadlineExceededError("Connection timed out"));
         }
@@ -385,6 +395,8 @@ EventEngine::ConnectionHandle WindowsEventEngine::Connect(
         {
           grpc_core::MutexLock lock(&connection_state->mu);
           cb = std::move(connection_state->on_connected_user_callback);
+          connection_state->on_connected_user_callback =
+              CreateCrashingOnConnectCallback();
         }
         cb(status);
       });