web: replace sleep_for with strand-based fence in shutdown broadcast

The shutdown path used sleep_for(100ms) to give ASIO threads time to
flush the shutdown broadcast before tearing down the io_context.  This
is fragile: 100ms is arbitrary and provides no delivery guarantee.

Add broadcastAndWait() to SessionRegistry which posts a fence lambda
to each session's strand after the write lambda.  Since strands are
FIFO, the fence executes after the write is queued.  A condition
variable collects all fence completions with a 2-second timeout as a
safety valve.

Each session now registers a PostFn alongside its SendFn.  The PostFn
captures the session's strand and calls net::post(strand_, fn).  If
the session is already destroyed (weak_ptr expired), the PostFn calls
the fence immediately so the counter still decrements.

Signed-off-by: Matt Liberty <mliberty@precisioninno.com>

Author: maliberty

src/web/src/web.cpp

@@ -422,18 +422,29 @@ void WebSocketSession::on_accept(beast::error_code ec)
   // browsers reject with "A server must not mask any frames".
   if (viewer_hook_ != nullptr) {
     auto weak_self = std::weak_ptr<WebSocketSession>(shared_from_this());
-    viewer_token_
-        = viewer_hook_->sessions().add([weak_self](const std::string& json) {
-            auto self = weak_self.lock();
-            if (!self) {
-              return;
-            }
-            WebSocketResponse resp;
-            resp.id = 0;
-            resp.type = 0;  // JSON
-            resp.payload.assign(json.begin(), json.end());
-            self->queue_response(resp);
-          });
+    viewer_token_ = viewer_hook_->sessions().add(
+        // SendFn — queue a JSON push message on this session's write queue.
+        [weak_self](const std::string& json) {
+          auto self = weak_self.lock();
+          if (!self) {
+            return;
+          }
+          WebSocketResponse resp;
+          resp.id = 0;
+          resp.type = 0;  // JSON
+          resp.payload.assign(json.begin(), json.end());
+          self->queue_response(resp);
+        },
+        // PostFn — post an arbitrary callable onto this session's strand.
+        // Used by broadcastAndWait() to fence after the write is queued.
+        [weak_self](std::function<void()> fn) {
+          auto self = weak_self.lock();
+          if (!self) {
+            fn();  // session gone — signal fence immediately
+            return;
+          }
+          net::post(self->strand_, std::move(fn));
+        });
   }
 
   // Build search indices in the background; tiles render without shapes

src/web/src/web_serve.cpp

@@ -210,12 +210,13 @@ void WebServer::waitForStop()
   lock.unlock();
 
   // Notify connected browsers so they can show "Server stopped" and
-  // disable auto-reconnect.  broadcast() posts async writes via
-  // net::post() on each session's strand, so we sleep briefly to
-  // let the ASIO threads flush them before stop() tears everything down.
+  // disable auto-reconnect.  broadcastAndWait() posts the message and
+  // then waits for a strand-fence on each session, guaranteeing the
+  // write is queued before stop() tears down the io_context.
   if (viewer_hook_) {
-    viewer_hook_->sessions().broadcast(R"({"type":"shutdown"})");
-    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    constexpr auto kShutdownFlushTimeout = std::chrono::seconds(2);
+    viewer_hook_->sessions().broadcastAndWait(R"({"type":"shutdown"})",
+                                              kShutdownFlushTimeout);
   }
 
   stop();

src/web/src/web_viewer_hook.cpp

@@ -30,13 +30,13 @@ constexpr auto kMaxPauseTimeout = std::chrono::minutes(10);
 // SessionRegistry
 //------------------------------------------------------------------------------
 
-std::size_t SessionRegistry::add(SendFn send)
+std::size_t SessionRegistry::add(SendFn send, PostFn post)
 {
   std::size_t token;
   {
     std::lock_guard<std::mutex> lock(mutex_);
     token = next_token_++;
-    senders_.emplace(token, std::move(send));
+    senders_.emplace(token, SessionCallbacks{std::move(send), std::move(post)});
   }
   client_cv_.notify_all();
   return token;
@@ -75,21 +75,78 @@ void SessionRegistry::notifyClientWaiters()
 
 void SessionRegistry::broadcast(const std::string& json)
 {
-  // Copy the senders out under the lock so we don't hold it while
-  // invoking callbacks (which may take session-level locks of their own).
-  std::vector<SendFn> to_send;
+  // Copy the callbacks out under the lock so we don't hold it while
+  // invoking them (which may take session-level locks of their own).
+  std::vector<SessionCallbacks> to_send;
   {
     std::lock_guard<std::mutex> lock(mutex_);
     to_send.reserve(senders_.size());
-    for (const auto& [_, fn] : senders_) {
-      to_send.push_back(fn);
+    for (const auto& [_, cb] : senders_) {
+      to_send.push_back(cb);
     }
   }
-  for (const auto& fn : to_send) {
-    fn(json);
+  for (const auto& cb : to_send) {
+    cb.send(json);
   }
 }
 
+bool SessionRegistry::broadcastAndWait(const std::string& json,
+                                       std::chrono::milliseconds timeout)
+{
+  // Copy the callbacks out under the lock.
+  std::vector<SessionCallbacks> to_send;
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    to_send.reserve(senders_.size());
+    for (const auto& [_, cb] : senders_) {
+      to_send.push_back(cb);
+    }
+  }
+
+  // Count sessions that support fencing (have a PostFn).
+  std::size_t fence_count = 0;
+  for (const auto& cb : to_send) {
+    if (cb.post) {
+      ++fence_count;
+    }
+  }
+
+  if (fence_count == 0) {
+    // No fenceable sessions — fire-and-forget like broadcast().
+    for (const auto& cb : to_send) {
+      cb.send(json);
+    }
+    return true;
+  }
+
+  // Shared state for the fence: each session's PostFn decrements the
+  // counter and notifies when all fences have fired.
+  struct FenceState
+  {
+    std::mutex mutex;
+    std::condition_variable cv;
+    std::size_t remaining;
+  };
+  auto state = std::make_shared<FenceState>();
+  state->remaining = fence_count;
+
+  for (const auto& cb : to_send) {
+    cb.send(json);
+    if (cb.post) {
+      cb.post([state]() {
+        std::lock_guard<std::mutex> lock(state->mutex);
+        if (--state->remaining == 0) {
+          state->cv.notify_one();
+        }
+      });
+    }
+  }
+
+  std::unique_lock<std::mutex> lock(state->mutex);
+  return state->cv.wait_for(
+      lock, timeout, [&state]() { return state->remaining == 0; });
+}
+
 //------------------------------------------------------------------------------
 // WebViewerHook
 //------------------------------------------------------------------------------

src/web/src/web_viewer_hook.h

@@ -4,6 +4,7 @@
 #pragma once
 
 #include <atomic>
+#include <chrono>
 #include <condition_variable>
 #include <cstddef>
 #include <functional>
@@ -26,11 +27,14 @@ class SessionRegistry
 {
  public:
   using SendFn = std::function<void(const std::string& json)>;
+  // Posts an arbitrary callable onto the session's strand.
+  using PostFn = std::function<void(std::function<void()>)>;
   using WaitInterruptFn = std::function<bool()>;
 
   // Register a send callback.  Returns a token the caller must pass to
-  // remove() during teardown.
-  std::size_t add(SendFn send);
+  // remove() during teardown.  The optional PostFn lets broadcastAndWait()
+  // post a fence lambda onto the session's strand.
+  std::size_t add(SendFn send, PostFn post = {});
   void remove(std::size_t token);
 
   // True if at least one session is registered.
@@ -47,11 +51,23 @@ class SessionRegistry
   // Deliver the JSON string to every currently-registered session.
   void broadcast(const std::string& json);
 
+  // Like broadcast(), but waits until every session's strand has executed
+  // the queued write (or timeout expires).  Returns true if all fences
+  // completed, false on timeout.
+  bool broadcastAndWait(const std::string& json,
+                        std::chrono::milliseconds timeout);
+
  private:
+  struct SessionCallbacks
+  {
+    SendFn send;
+    PostFn post;  // may be empty for legacy callers
+  };
+
   mutable std::mutex mutex_;
   mutable std::condition_variable client_cv_;
   std::size_t next_token_ = 1;
-  std::unordered_map<std::size_t, SendFn> senders_;
+  std::unordered_map<std::size_t, SessionCallbacks> senders_;
 };
 
 // The web viewer's bridge to gui::Gui.  Installed as the Gui's

src/web/test/cpp/TestDebugGraphics.cpp

@@ -109,6 +109,83 @@ static std::size_t registerDummyClient(WebViewerHook& hook)
   return hook.sessions().add([](const std::string&) {});
 }
 
+TEST(SessionRegistryTest, BroadcastAndWaitNoSessions)
+{
+  SessionRegistry registry;
+  // No sessions registered — should return true immediately.
+  EXPECT_TRUE(registry.broadcastAndWait("{}", std::chrono::milliseconds(100)));
+}
+
+TEST(SessionRegistryTest, BroadcastAndWaitFencesComplete)
+{
+  SessionRegistry registry;
+  std::string received;
+
+  // Register a session whose PostFn invokes the fence on a background thread
+  // (simulating a real strand dispatch).
+  auto token = registry.add(
+      [&received](const std::string& json) { received = json; },
+      [](std::function<void()> fence) {
+        std::thread([fence = std::move(fence)]() { fence(); }).detach();
+      });
+
+  EXPECT_TRUE(registry.broadcastAndWait(R"({"type":"shutdown"})",
+                                        std::chrono::seconds(2)));
+  EXPECT_EQ(received, R"({"type":"shutdown"})");
+  registry.remove(token);
+}
+
+TEST(SessionRegistryTest, BroadcastAndWaitTimesOut)
+{
+  SessionRegistry registry;
+
+  // Register a session whose PostFn silently drops the fence.
+  auto token = registry.add(
+      [](const std::string&) {},
+      [](std::function<void()> /*fence*/) { /* never called */ });
+
+  const auto t0 = std::chrono::steady_clock::now();
+  EXPECT_FALSE(registry.broadcastAndWait("{}", std::chrono::milliseconds(200)));
+  const auto elapsed = std::chrono::steady_clock::now() - t0;
+
+  // Should have waited approximately the timeout, not longer.
+  EXPECT_GE(elapsed, std::chrono::milliseconds(150));
+  EXPECT_LT(elapsed, std::chrono::seconds(2));
+  registry.remove(token);
+}
+
+TEST(SessionRegistryTest, BroadcastAndWaitDeadSession)
+{
+  SessionRegistry registry;
+  std::atomic<bool> send_called{false};
+
+  // Register a session whose PostFn calls the fence immediately
+  // (simulating a dead session that signals right away).
+  auto token
+      = registry.add([&send_called](const std::string&) { send_called = true; },
+                     [](std::function<void()> fence) { fence(); });
+
+  EXPECT_TRUE(registry.broadcastAndWait("{}", std::chrono::milliseconds(100)));
+  EXPECT_TRUE(send_called);
+  registry.remove(token);
+}
+
+TEST(SessionRegistryTest, BroadcastAndWaitLegacySendOnly)
+{
+  SessionRegistry registry;
+  std::string received;
+
+  // Register with SendFn only (no PostFn) — legacy path.
+  auto token
+      = registry.add([&received](const std::string& json) { received = json; });
+
+  // Should return true immediately since there are no fenceable sessions.
+  EXPECT_TRUE(registry.broadcastAndWait(R"({"ok":true})",
+                                        std::chrono::milliseconds(100)));
+  EXPECT_EQ(received, R"({"ok":true})");
+  registry.remove(token);
+}
+
 TEST(SessionRegistryTest, WaitForClientCanBeInterrupted)
 {
   SessionRegistry registry;