web: fence broadcastAndWait on write completion, not strand post

Replace the two-callback SessionRegistry pattern (SendFn + PostFn)
with a combined SendAndWaitFn whose completion fires after
async_write finishes.  This ensures the shutdown message is on the
wire before stop() tears down the io_context.

Also store the WebLogSink as a WebServer member so it is properly
removed from the Logger on shutdown, and fix a missing
writing_=false reset on the write-error path.

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

Author: maliberty

src/web/include/web/web.h

@@ -25,6 +25,10 @@ namespace sta {
 class dbSta;
 }
 
+namespace spdlog::sinks {
+class sink;
+}
+
 namespace web {
 
 struct Color;
@@ -111,6 +115,7 @@ class WebServer
   int num_threads_ = 0;
   std::shared_ptr<TileGenerator> generator_;
   std::unique_ptr<WebViewerHook> viewer_hook_;
+  std::shared_ptr<spdlog::sinks::sink> log_sink_;
 
   // Background I/O context and worker threads (non-null while running).
   std::unique_ptr<boost::asio::io_context> ioc_;

src/web/src/web.cpp

@@ -310,8 +310,13 @@ class WebSocketSession : public std::enable_shared_from_this<WebSocketSession>
   DRCHandler drc_handler_;
 
   // Write serialization: strand + queue ensures one async_write at a time
+  struct PendingWrite
+  {
+    std::vector<unsigned char> frame;
+    std::function<void()> on_complete;
+  };
   net::strand<net::any_io_executor> strand_;
-  std::deque<std::vector<unsigned char>> write_queue_;
+  std::deque<PendingWrite> write_queue_;
   bool writing_ = false;
 
   // Background search index initialization
@@ -339,7 +344,8 @@ class WebSocketSession : public std::enable_shared_from_this<WebSocketSession>
   void on_accept(beast::error_code ec);
   void do_read();
   void on_read(beast::error_code ec);
-  void queue_response(const WebSocketResponse& resp);
+  void queue_response(const WebSocketResponse& resp,
+                      std::function<void()> on_complete = {});
   void do_write();
 };
 
@@ -435,15 +441,19 @@ void WebSocketSession::on_accept(beast::error_code ec)
           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) {
+        // SendAndWaitFn — queue a JSON push message and invoke the callback
+        // after async_write completes.
+        [weak_self](const std::string& json, 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));
+          WebSocketResponse resp;
+          resp.id = 0;
+          resp.type = 0;  // JSON
+          resp.payload.assign(json.begin(), json.end());
+          self->queue_response(resp, std::move(fn));
         });
 
     // Flush any log output that accumulated before this client
@@ -789,18 +799,23 @@ void WebSocketSession::on_read(beast::error_code ec)
   do_read();
 }
 
-void WebSocketSession::queue_response(const WebSocketResponse& resp)
+void WebSocketSession::queue_response(const WebSocketResponse& resp,
+                                      std::function<void()> on_complete)
 {
   std::vector<unsigned char> frame = serialize_response(resp);
 
   // Post to the strand to serialize write queue access
-  net::post(strand_,
-            [self = shared_from_this(), frame = std::move(frame)]() mutable {
-              self->write_queue_.push_back(std::move(frame));
-              if (!self->writing_) {
-                self->do_write();
-              }
-            });
+  net::post(
+      strand_,
+      [self = shared_from_this(),
+       frame = std::move(frame),
+       on_complete = std::move(on_complete)]() mutable {
+        self->write_queue_.push_back(PendingWrite{
+            .frame = std::move(frame), .on_complete = std::move(on_complete)});
+        if (!self->writing_) {
+          self->do_write();
+        }
+      });
 }
 
 void WebSocketSession::do_write()
@@ -812,19 +827,24 @@ void WebSocketSession::do_write()
   writing_ = true;
   websocket_.binary(true);
   websocket_.async_write(
-      net::buffer(write_queue_.front()),
+      net::buffer(write_queue_.front().frame),
       [self = shared_from_this()](beast::error_code ec, std::size_t) {
         net::post(self->strand_, [self, ec]() {
+          auto on_complete = std::move(self->write_queue_.front().on_complete);
+          self->write_queue_.pop_front();
+          if (on_complete) {
+            on_complete();
+          }
           if (ec) {
             debugPrint(self->logger_,
                        utl::WEB,
                        "websocket",
                        1,
                        "websocket write error: {}",
                        ec.message());
+            self->writing_ = false;
             return;
           }
-          self->write_queue_.pop_front();
           self->do_write();
         });
       });

src/web/src/web_serve.cpp

@@ -140,8 +140,10 @@ void WebServer::serve(int port)
         });
 
     auto log_sink = std::make_shared<WebLogSink>(viewer_hook_.get());
-    logger_->addSink(log_sink);
-    viewer_hook_->setDrainLogsFn([log_sink]() { log_sink->drainToClients(); });
+    log_sink_ = log_sink;
+    logger_->addSink(log_sink_);
+    viewer_hook_->setDrainLogsFn(
+        [log_sink = std::move(log_sink)]() { log_sink->drainToClients(); });
 
     TileGenerator::setDebugOverlayCallback(
         [weak_gen = std::weak_ptr<TileGenerator>(generator_),
@@ -215,9 +217,8 @@ void WebServer::waitForStop()
   lock.unlock();
 
   // Notify connected browsers so they can show "Server stopped" and
-  // 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.
+  // disable auto-reconnect.  broadcastAndWait() waits for the write to
+  // complete before stop() tears down the io_context.
   if (viewer_hook_) {
     constexpr auto kShutdownFlushTimeout = std::chrono::seconds(2);
     viewer_hook_->sessions().broadcastAndWait(R"({"type":"shutdown"})",
@@ -249,6 +250,10 @@ void WebServer::stop()
     }
     gui::Gui::get()->setChartFactory({});
   }
+  if (log_sink_) {
+    logger_->removeSink(log_sink_);
+    log_sink_.reset();
+  }
 
   if (shutdown_listener_) {
     shutdown_listener_();

src/web/src/web_viewer_hook.cpp

@@ -30,13 +30,17 @@ constexpr auto kMaxPauseTimeout = std::chrono::minutes(10);
 // SessionRegistry
 //------------------------------------------------------------------------------
 
-std::size_t SessionRegistry::add(SendFn send, PostFn post)
+std::size_t SessionRegistry::add(SendFn send, SendAndWaitFn send_and_wait)
 {
   std::size_t token;
   {
     std::lock_guard<std::mutex> lock(mutex_);
     token = next_token_++;
-    senders_.emplace(token, SessionCallbacks{std::move(send), std::move(post)});
+    senders_.emplace(token,
+                     SessionCallbacks{
+                         .send = std::move(send),
+                         .send_and_wait = std::move(send_and_wait),
+                     });
   }
   client_cv_.notify_all();
   return token;
@@ -103,10 +107,10 @@ bool SessionRegistry::broadcastAndWait(const std::string& json,
     }
   }
 
-  // Count sessions that support fencing (have a PostFn).
+  // Count sessions that support write-completion fencing.
   std::size_t fence_count = 0;
   for (const auto& cb : to_send) {
-    if (cb.post) {
+    if (cb.send_and_wait) {
       ++fence_count;
     }
   }
@@ -119,8 +123,8 @@ bool SessionRegistry::broadcastAndWait(const std::string& json,
     return true;
   }
 
-  // Shared state for the fence: each session's PostFn decrements the
-  // counter and notifies when all fences have fired.
+  // Shared state for the fence: each session's SendAndWaitFn decrements
+  // the counter and notifies when all writes have completed.
   struct FenceState
   {
     std::mutex mutex;
@@ -131,14 +135,15 @@ bool SessionRegistry::broadcastAndWait(const std::string& json,
   state->remaining = fence_count;
 
   for (const auto& cb : to_send) {
-    cb.send(json);
-    if (cb.post) {
-      cb.post([state]() {
+    if (cb.send_and_wait) {
+      cb.send_and_wait(json, [state]() {
         std::lock_guard<std::mutex> lock(state->mutex);
         if (--state->remaining == 0) {
           state->cv.notify_one();
         }
       });
+    } else {
+      cb.send(json);
     }
   }
 

src/web/src/web_viewer_hook.h

@@ -27,14 +27,15 @@ 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()>)>;
+  // Sends JSON and invokes the callback after the session write completes.
+  using SendAndWaitFn
+      = std::function<void(const std::string& json, std::function<void()>)>;
   using WaitInterruptFn = std::function<bool()>;
 
   // Register a send callback.  Returns a token the caller must pass to
-  // 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 = {});
+  // remove() during teardown.  The optional SendAndWaitFn lets
+  // broadcastAndWait() wait until a write has completed.
+  std::size_t add(SendFn send, SendAndWaitFn send_and_wait = {});
   void remove(std::size_t token);
 
   // True if at least one session is registered.
@@ -51,17 +52,16 @@ 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.
+  // Like broadcast(), but waits until every fenceable session has completed
+  // the queued write (or timeout expires).
   bool broadcastAndWait(const std::string& json,
                         std::chrono::milliseconds timeout);
 
  private:
   struct SessionCallbacks
   {
     SendFn send;
-    PostFn post;  // may be empty for legacy callers
+    SendAndWaitFn send_and_wait;  // may be empty for legacy callers
   };
 
   mutable std::mutex mutex_;

src/web/test/cpp/TestDebugGraphics.cpp

@@ -4,10 +4,12 @@
 #include <atomic>
 #include <chrono>
 #include <cstddef>
+#include <functional>
 #include <memory>
 #include <mutex>
 #include <string>
 #include <thread>
+#include <utility>
 #include <variant>
 #include <vector>
 
@@ -121,12 +123,15 @@ 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).
+  // Register a session whose SendAndWaitFn invokes the fence only after
+  // the simulated write has completed.
   auto token = registry.add(
       [&received](const std::string& json) { received = json; },
-      [](std::function<void()> fence) {
-        std::thread([fence = std::move(fence)]() { fence(); }).detach();
+      [&received](const std::string& json, std::function<void()> fence) {
+        std::thread([&received, json, fence = std::move(fence)]() {
+          received = json;
+          fence();
+        }).detach();
       });
 
   EXPECT_TRUE(registry.broadcastAndWait(R"({"type":"shutdown"})",
@@ -139,10 +144,12 @@ 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 */ });
+  // Register a session whose SendAndWaitFn silently drops the fence.
+  auto token
+      = registry.add([](const std::string&) {},
+                     [](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)));
@@ -159,11 +166,14 @@ TEST(SessionRegistryTest, BroadcastAndWaitDeadSession)
   SessionRegistry registry;
   std::atomic<bool> send_called{false};
 
-  // Register a session whose PostFn calls the fence immediately
+  // Register a session whose SendAndWaitFn 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(); });
+  auto token = registry.add(
+      [&send_called](const std::string&) { send_called = true; },
+      [&send_called](const std::string&, std::function<void()> fence) {
+        send_called = true;
+        fence();
+      });
 
   EXPECT_TRUE(registry.broadcastAndWait("{}", std::chrono::milliseconds(100)));
   EXPECT_TRUE(send_called);
@@ -175,7 +185,7 @@ TEST(SessionRegistryTest, BroadcastAndWaitLegacySendOnly)
   SessionRegistry registry;
   std::string received;
 
-  // Register with SendFn only (no PostFn) — legacy path.
+  // Register with SendFn only (no SendAndWaitFn) — legacy path.
   auto token
       = registry.add([&received](const std::string& json) { received = json; });