fix(websocket): harden shutdown and runtime integration (avoid abort on exit)

Author: GaspardKirira

include/vix/websocket/App.hpp

@@ -13,8 +13,10 @@
 #ifndef VIX_WEBSOCKET_APP_HPP
 #define VIX_WEBSOCKET_APP_HPP
 
+#include <atomic>
 #include <functional>
 #include <memory>
+#include <mutex>
 #include <string>
 #include <vector>
 
@@ -59,6 +61,8 @@ namespace vix::websocket
         const std::string &configPath,
         std::shared_ptr<vix::executor::RuntimeExecutor> executor);
 
+    ~App() noexcept;
+
     App(const App &) = delete;
     App &operator=(const App &) = delete;
     App(App &&) = delete;
@@ -79,9 +83,11 @@ namespace vix::websocket
     void run_blocking();
 
     /**
-     * @brief Stop the WebSocket server.
+     * @brief Stop the WebSocket app and shared executor.
+     *
+     * Safe to call multiple times.
      */
-    void stop();
+    void stop() noexcept;
 
     /**
      * @brief Access the underlying WebSocket server.
@@ -128,6 +134,14 @@ namespace vix::websocket
      */
     void install_dispatcher();
 
+    /**
+     * @brief Dispatch one typed message to all registered handlers.
+     */
+    void dispatch_typed_message(
+        Session &session,
+        const std::string &type,
+        const vix::json::kvs &payload);
+
   private:
     /** @brief Application configuration source. */
     vix::config::Config config_;
@@ -140,6 +154,12 @@ namespace vix::websocket
 
     /** @brief Registered typed routes. */
     std::vector<Route> routes_;
+
+    /** @brief Protects stop() from concurrent/double shutdown. */
+    mutable std::mutex stopMutex_;
+
+    /** @brief Indicates whether shutdown was already requested. */
+    std::atomic<bool> stopped_{false};
   };
 
 } // namespace vix::websocket

include/vix/websocket/AttachedRuntime.hpp

@@ -15,11 +15,9 @@
 #define VIX_ATTACHED_RUNTIME_HPP
 
 #include <atomic>
-#include <chrono>
 #include <filesystem>
 #include <memory>
 #include <mutex>
-#include <thread>
 #include <utility>
 
 #include <vix/app/App.hpp>
@@ -36,49 +34,54 @@ namespace vix::websocket
   /**
    * @brief Runs a WebSocket server alongside an HTTP app, with shared lifecycle.
    *
-   * Starts the WebSocket server on a dedicated thread and stops it when the HTTP
-   * app requests shutdown or when this object is destroyed.
+   * Starts the WebSocket server immediately and coordinates shutdown with the attached HTTP
+   * application.
+   *
+   * Important lifecycle rule:
+   * - The HTTP shutdown callback only requests an asynchronous stop.
+   * - The final blocking shutdown is executed from an external safe control path.
    */
   class AttachedRuntime
   {
   public:
     /**
      * @brief Attach a WebSocket server to an existing HTTP app.
      *
-     * Starts the WebSocket server in a background thread and registers an app
-     * shutdown callback to stop the WebSocket runtime.
+     * Starts the WebSocket server immediately and registers an app shutdown
+     * callback that requests a non-blocking WebSocket stop.
      *
      * @param app HTTP application instance.
      * @param ws WebSocket server instance.
+     * @param exec Shared runtime executor.
      */
-    AttachedRuntime(vix::App &app, vix::websocket::Server &ws)
-        : app_(app), ws_(ws), wsThread_(), stopped_(false)
+    AttachedRuntime(
+        vix::App &app,
+        vix::websocket::Server &ws,
+        std::shared_ptr<vix::executor::RuntimeExecutor> exec)
+        : app_(app),
+          ws_(ws),
+          exec_(std::move(exec)),
+          stop_requested_(false),
+          finalized_(false)
     {
-      wsThread_ = std::thread(
-          [this]()
-          {
-            vix::utils::console_wait_banner();
-            ws_.start();
-
-            while (!stopped_.load(std::memory_order_relaxed))
-            {
-              std::this_thread::sleep_for(std::chrono::milliseconds(250));
-            }
-          });
+      vix::utils::console_wait_banner();
+      ws_.start();
 
       app_.set_shutdown_callback(
           [this]()
           {
-            stop();
+            request_stop();
           });
     }
 
     /**
-     * @brief Stop the runtime if still running.
+     * @brief Finalize shutdown if still needed.
+     *
+     * This destructor is defensive and idempotent.
      */
-    ~AttachedRuntime()
+    ~AttachedRuntime() noexcept
     {
-      stop();
+      finalize_shutdown();
     }
 
     AttachedRuntime(const AttachedRuntime &) = delete;
@@ -87,23 +90,66 @@ namespace vix::websocket
     AttachedRuntime &operator=(AttachedRuntime &&) = delete;
 
     /**
-     * @brief Stop the WebSocket server and join the worker thread.
+     * @brief Request an asynchronous WebSocket shutdown.
      *
-     * Safe to call multiple times.
+     * This function is idempotent and intentionally non-blocking.
+     * It is safe to call from a shutdown callback running inside a server
+     * worker thread.
      */
-    void stop() noexcept
+    void request_stop() noexcept
     {
       bool expected = false;
-      if (!stopped_.compare_exchange_strong(expected, true))
+      if (!stop_requested_.compare_exchange_strong(
+              expected,
+              true,
+              std::memory_order_acq_rel,
+              std::memory_order_acquire))
       {
         return;
       }
 
-      ws_.stop();
+      try
+      {
+        ws_.stop_async();
+      }
+      catch (...)
+      {
+      }
+    }
+
+    /**
+     * @brief Perform the final blocking shutdown exactly once.
+     *
+     * Order matters:
+     * 1. stop websocket blocking
+     * 2. stop shared executor blocking
+     */
+    void finalize_shutdown() noexcept
+    {
+      std::lock_guard<std::mutex> lock(finalize_mutex_);
 
-      if (wsThread_.joinable())
+      if (finalized_.exchange(true, std::memory_order_acq_rel))
+      {
+        return;
+      }
+
+      try
+      {
+        ws_.stop();
+      }
+      catch (...)
+      {
+      }
+
+      try
+      {
+        if (exec_)
+        {
+          exec_->stop();
+        }
+      }
+      catch (...)
       {
-        wsThread_.join();
       }
     }
 
@@ -114,11 +160,17 @@ namespace vix::websocket
     /** @brief Attached WebSocket server. */
     vix::websocket::Server &ws_;
 
-    /** @brief Background thread driving the WebSocket runtime. */
-    std::thread wsThread_;
+    /** @brief Shared runtime executor. */
+    std::shared_ptr<vix::executor::RuntimeExecutor> exec_;
 
-    /** @brief Idempotent stop flag. */
-    std::atomic<bool> stopped_;
+    /** @brief Idempotent asynchronous stop flag. */
+    std::atomic<bool> stop_requested_;
+
+    /** @brief Ensures final shutdown happens once. */
+    std::atomic<bool> finalized_;
+
+    /** @brief Protects final shutdown sequence. */
+    std::mutex finalize_mutex_;
   };
 
 } // namespace vix::websocket
@@ -148,16 +200,18 @@ namespace vix
   /**
    * @brief Run HTTP and WebSocket servers together in blocking mode.
    *
-   * Registers WebSocket OpenAPI docs once, starts the WebSocket runtime,
-   * then starts the HTTP server and waits until shutdown.
+   * Registers WebSocket OpenAPI docs once, starts the attached WebSocket
+   * runtime, then starts the HTTP server and waits until shutdown.
    *
    * @param app HTTP application instance.
    * @param ws WebSocket server instance.
+   * @param exec Shared runtime executor.
    * @param port HTTP listening port.
    */
   inline void run_http_and_ws(
       vix::App &app,
       vix::websocket::Server &ws,
+      std::shared_ptr<vix::executor::RuntimeExecutor> exec,
       int port = 8080)
   {
     register_ws_openapi_docs_once();
@@ -168,7 +222,7 @@ namespace vix
       vix::openapi::register_openapi_and_docs(*r, "Vix API", "1.33.0");
     }
 
-    vix::websocket::AttachedRuntime runtime{app, ws};
+    vix::websocket::AttachedRuntime runtime{app, ws, exec};
 
     app.listen(
         port,
@@ -192,6 +246,8 @@ namespace vix
 
     app.wait();
     app.close();
+
+    runtime.finalize_shutdown();
   }
 
   /**
@@ -223,7 +279,7 @@ namespace vix
 
     fn(app, ws);
 
-    run_http_and_ws(app, ws, port);
+    run_http_and_ws(app, ws, exec, port);
   }
 
   /**
@@ -240,4 +296,4 @@ namespace vix
 
 } // namespace vix
 
-#endif
+#endif // VIX_ATTACHED_RUNTIME_HPP

include/vix/websocket/server.hpp

@@ -217,14 +217,37 @@ namespace vix::websocket
     }
 
     /**
-     * @brief Stop the WebSocket engine and join its internal threads.
+     * @brief Request an asynchronous stop of the WebSocket engine.
+     *
+     * This function is non-blocking and only signals shutdown to the internal
+     * low-level engine. It does not join worker threads.
      */
-    void stop()
+    void stop_async()
     {
       engine_.stop_async();
+    }
+
+    /**
+     * @brief Join internal WebSocket worker threads.
+     *
+     * This function blocks until the low-level engine threads have exited.
+     */
+    void join()
+    {
       engine_.join_threads();
     }
 
+    /**
+     * @brief Stop the WebSocket engine and join its internal threads.
+     *
+     * This is the blocking shutdown path.
+     */
+    void stop()
+    {
+      stop_async();
+      join();
+    }
+
     /**
      * @brief Start the server and block until stop is requested.
      */

include/vix/websocket/websocket.hpp

@@ -176,6 +176,12 @@ namespace vix::websocket
 
     /** @brief Effective bound port, useful when binding to port 0. */
     std::atomic<int> boundPort_{0};
+
+    /** @brief Protects the join phase against concurrent callers. */
+    mutable std::mutex joinMutex_;
+
+    /** @brief Ensures IO worker threads are joined only once. */
+    std::atomic<bool> threadsJoined_{false};
   };
 
 } // namespace vix::websocket