fix: bound and own immediate-mode form dispatch instead of a detached thread

Code review found the detached std::thread used for immediate-mode form
dispatch was unsafe on two counts: nothing joined it, so an ordinary
systemctl restart while a run was in flight would free the controller
under a live thread (use-after-free on scheduler_ via a stale this); and
the form endpoint is public and unauthenticated by design, so a burst of
submissions could spawn threads without limit.

Replace it with a small bounded worker pool owned by WebhookController
(WebhookDispatchConfig: threads=4, queue_capacity=32, both configurable
via server.form_dispatch_threads/form_dispatch_queue_capacity). Workers
are joined in stop()/~WebhookController, called from
WebServerService::stop() right after the HTTP server stops, so nothing
can still be running once the controller is destroyed. A submission that
finds the queue full is answered honestly with 503 and a busy page rather
than a thank-you page for work that will never run; the scheduler slot is
claimed inside the queued task so a refused submission never leaks one.

Verified: both response modes still behave: queue-full path drives 503
then recovers once the pool drains; and shutdown while a 40s run is in
flight blocks stop() until the run actually finishes (built with ASAN,
zero sanitizer reports, clean exit). Details in
.superpowers/sdd/2026-08-09-form-trigger/task-5-report.md.

src/webserver/api/webhook_controller.cpp

@@ -40,13 +40,71 @@ WebhookController::WebhookController(storage::StorageClient& storage,
                                      runners::LoadBalancer& load_balancer,
                                      WebSocketServer& ws_server,
                                      nodes::NodeStore& node_store,
-                                     WorkflowScheduler& scheduler)
+                                     WorkflowScheduler& scheduler,
+                                     DispatchConfig dispatch_config)
     : storage_(storage)
     , registry_(registry)
     , load_balancer_(load_balancer)
     , ws_server_(ws_server)
     , node_store_(node_store)
-    , scheduler_(scheduler) {}
+    , scheduler_(scheduler)
+    , dispatch_queue_capacity_(dispatch_config.queue_capacity) {
+    dispatch_workers_.reserve(dispatch_config.threads);
+    for (std::size_t i = 0; i < dispatch_config.threads; ++i) {
+        dispatch_workers_.emplace_back(&WebhookController::dispatchWorkerLoop, this);
+    }
+}
+
+WebhookController::~WebhookController() {
+    stop();
+}
+
+void WebhookController::stop() {
+    {
+        std::lock_guard<std::mutex> lock(dispatch_mutex_);
+        if (!dispatch_running_) {
+            return;  // already stopped
+        }
+        dispatch_running_ = false;
+    }
+    dispatch_cv_.notify_all();
+    for (auto& worker : dispatch_workers_) {
+        if (worker.joinable()) {
+            worker.join();
+        }
+    }
+    dispatch_workers_.clear();
+}
+
+void WebhookController::dispatchWorkerLoop() {
+    while (true) {
+        std::function<void()> task;
+        {
+            std::unique_lock<std::mutex> lock(dispatch_mutex_);
+            dispatch_cv_.wait(lock, [this] {
+                return !dispatch_running_ || !dispatch_queue_.empty();
+            });
+            if (!dispatch_running_ && dispatch_queue_.empty()) {
+                return;
+            }
+            task = std::move(dispatch_queue_.front());
+            dispatch_queue_.pop();
+        }
+        task();
+    }
+}
+
+bool WebhookController::tryEnqueueDispatch(std::function<void()> task) {
+    {
+        std::lock_guard<std::mutex> lock(dispatch_mutex_);
+        if (!dispatch_running_ || dispatch_queue_.size() >= dispatch_queue_capacity_) {
+            return false;
+        }
+        dispatch_queue_.push(std::move(task));
+    }
+    dispatch_cv_.notify_one();
+    return true;
+}
 
 void WebhookController::registerRoutes(httplib::Server& server) {
     // Match any HTTP method for webhooks
@@ -210,15 +268,14 @@ void WebhookController::handleWebhook(const httplib::Request& req, httplib::Resp
     auto channel = grpc::CreateChannel(runner->address, grpc::InsecureChannelCredentials());
     auto stub = proto::RunnerService::NewStub(channel);
 
-    // A schedule's overlap policy counts what is running, and this call blocks
-    // until the run has finished - so by the time it returns with an execution
-    // id there is nothing left to claim a slot for, and a scheduled tick in the
-    // meantime saw the workflow as idle. Naming the id here rather than letting
-    // the runner pick one is what makes the slot claimable up front; the runner
-    // takes this as the run's id.
+    // A schedule's overlap policy counts what is running, and a synchronous
+    // dispatch blocks until the run has finished - so by the time it returns
+    // with an execution id there is nothing left to claim a slot for, and a
+    // scheduled tick in the meantime saw the workflow as idle. Naming the id
+    // here rather than letting the runner pick one is what makes the slot
+    // claimable up front; the runner takes this as the run's id.
     const std::string execution_id = UUID::generatePrefixed("exec");
     trigger_data["_assignedExecutionId"] = execution_id;
-    scheduler_.notifyExecutionStarted(workflow_id, execution_id);
 
     proto::ExecuteWorkflowRequest grpc_req;
     grpc_req.set_workflow_id(workflow_id);
@@ -226,30 +283,47 @@ void WebhookController::handleWebhook(const httplib::Request& req, httplib::Resp
     // Fired by the outside world, so it runs the published version.
     grpc_req.set_use_published(true);
     grpc_req.set_trigger_data(trigger_data.dump());
-    // A form set to answer immediately must not hold the browser for the run.
-    // The slot claimed above is still released correctly: the runner reports
-    // execution.completed/failed/cancelled/waiting to
-    // /api/v1/internal/execution-event, and execution_controller.cpp:632
-    // releases it there - the same path fire-and-forget scheduled dispatch
-    // relies on today.
     const bool wait_for_run = form_trigger_data.is_null() || form_response_mode != "immediate";
     grpc_req.set_wait_for_completion(wait_for_run);
     grpc_req.set_timeout_ms(wait_for_run ? 30000 : 0);  // 30 second timeout when waiting
 
     if (!wait_for_run) {
         // ExecuteWorkflow is a plain synchronous RPC: the server call runs the
-        // workflow to completion (or until the client's deadline lapses) before
-        // it returns anything, regardless of wait_for_completion - that field
-        // only controls whether the response carries a result. Holding this
-        // handler's thread on that call would hold the browser for the run no
-        // matter how the request is configured, which is exactly what
-        // "immediate" promises not to do. So the call is moved to a detached
-        // thread: the browser gets the thank-you page right away, and the run
-        // proceeds in the background. The scheduler slot claimed above is
-        // still released correctly, the same way a scheduled dispatch's is -
-        // via the runner's execution.completed/failed/cancelled/waiting
-        // report to /api/v1/internal/execution-event.
-        std::thread([this, channel, workflow_id, execution_id, grpc_req]() {
+        // workflow to completion (or until the client's deadline lapses)
+        // before it returns anything - wait_for_completion only controls
+        // whether the response carries a result, it does not make the server
+        // return early. Holding this handler's thread on that call would hold
+        // the browser for the run no matter how the request is configured,
+        // which is exactly what "immediate" promises not to do. So the call
+        // runs on the controller's bounded dispatch pool instead of this
+        // thread or a detached one of its own:
+        //   - a detached thread has no owner, so nothing joins it, and it
+        //     would still be running against `this` after the controller
+        //     that it captured is destroyed - an ordinary
+        //     `systemctl --user restart` while a run is in flight would be a
+        //     use-after-free;
+        //   - the form endpoint is public and unauthenticated by design, so
+        //     an unbounded "one thread per submission" scheme is a resource
+        //     exhaustion vector open to anyone who can reach it.
+        // Bounded and joined in stop()/~WebhookController fixes both: no
+        // task can still be running once the controller that owns the pool
+        // is gone, and a burst of submissions is throttled rather than
+        // spawning without limit.
+        //
+        // The scheduler slot is claimed inside the queued task, not here, so
+        // a submission that is refused for a full queue never claims one -
+        // claiming it up front and then refusing would leak it. Once
+        // claimed, it is released the same way a scheduled dispatch's is:
+        // the runner reports execution.completed/failed/cancelled/waiting to
+        // /api/v1/internal/execution-event, and execution_controller.cpp:632
+        // releases it there. That event-based release is the only part this
+        // shares with scheduled dispatch - scheduled dispatch blocks the
+        // scheduler's own single thread (joined in WorkflowScheduler::stop())
+        // and relies on the client-side deadline lapsing; this runs on a
+        // pool sized and owned for exactly this purpose.
+        bool accepted = tryEnqueueDispatch([this, channel, workflow_id, execution_id, grpc_req]() {
+            scheduler_.notifyExecutionStarted(workflow_id, execution_id);
+
             auto bg_stub = proto::RunnerService::NewStub(channel);
             proto::ExecuteWorkflowResponse bg_res;
             grpc::ClientContext bg_ctx;
@@ -275,7 +349,21 @@ void WebhookController::handleWebhook(const httplib::Request& req, httplib::Resp
                 scheduler_.notifyExecutionFinished(workflow_id, execution_id);
                 scheduler_.notifyExecutionStarted(workflow_id, bg_res.execution_id());
             }
-        }).detach();
+        });
+
+        if (!accepted) {
+            // The queue is full: refuse honestly rather than tell someone
+            // "thanks, received" for a submission that will never run, and
+            // rather than silently drop it. No slot was claimed for this one.
+            LOG_WARN("Immediate-mode dispatch queue is full, refusing submission for workflow {}",
+                     workflow_id);
+            res.status = 503;
+            res.set_content(form_renderer::renderMessage(
+                                form_config_title,
+                                "This form is busy right now. Please try again in a moment."),
+                            "text/html; charset=utf-8");
+            return;
+        }
 
         res.status = 200;
         res.set_content(form_renderer::renderMessage(form_config_title, form_response_message),
@@ -283,6 +371,8 @@ void WebhookController::handleWebhook(const httplib::Request& req, httplib::Resp
         return;
     }
 
+    scheduler_.notifyExecutionStarted(workflow_id, execution_id);
+
     proto::ExecuteWorkflowResponse grpc_res;
     grpc::ClientContext grpc_ctx;
     grpc_ctx.set_deadline(std::chrono::system_clock::now() + std::chrono::seconds(35));

src/webserver/api/webhook_controller.hpp

@@ -2,6 +2,12 @@
 
 #include <httplib.h>
 #include <nlohmann/json.hpp>
+#include <thread>
+#include <vector>
+#include <queue>
+#include <functional>
+#include <condition_variable>
+#include <mutex>
 #include "storage/storage_client.hpp"
 #include "../runners/load_balancer.hpp"
 #include "../websocket_server.hpp"
@@ -10,17 +16,37 @@
 
 namespace smartbotic::webserver::api {
 
+// Bounds the background pool that runs immediate-mode form dispatches.
+// The webhook path is public and unauthenticated by design - a form is
+// meant to be shareable - so nothing about how many are submitted at
+// once is trustworthy, and the pool must stay bounded rather than
+// spawning a thread per submission.
+struct WebhookDispatchConfig {
+    std::size_t threads = 4;
+    std::size_t queue_capacity = 32;
+};
+
 class WebhookController {
 public:
+    using DispatchConfig = WebhookDispatchConfig;
+
     WebhookController(storage::StorageClient& storage,
                      runners::RunnerRegistry& registry,
                      runners::LoadBalancer& load_balancer,
                      WebSocketServer& ws_server,
                      nodes::NodeStore& node_store,
-                     WorkflowScheduler& scheduler);
+                     WorkflowScheduler& scheduler,
+                     DispatchConfig dispatch_config = {});
+    ~WebhookController();
 
     void registerRoutes(httplib::Server& server);
 
+    // Stops accepting new immediate-mode dispatches and joins every worker.
+    // Must run before the controller itself is destroyed - a worker task
+    // captures `this` and calling into a freed controller would be a
+    // use-after-free. Safe to call more than once.
+    void stop();
+
 private:
     void handleWebhook(const httplib::Request& req, httplib::Response& res);
 
@@ -39,6 +65,12 @@ private:
     void sendJson(httplib::Response& res, const nlohmann::json& data, int status = 200);
     void sendError(httplib::Response& res, const std::string& message, int status);
 
+    // Queues a task on the dispatch pool. Returns false, without queuing
+    // anything, once the bounded queue is full - the caller must answer
+    // honestly rather than accept work nobody will run.
+    bool tryEnqueueDispatch(std::function<void()> task);
+    void dispatchWorkerLoop();
+
     storage::StorageClient& storage_;
     runners::RunnerRegistry& registry_;
     runners::LoadBalancer& load_balancer_;
@@ -47,6 +79,17 @@ private:
     // Held so a webhook run counts towards a schedule's overlap policy, the
     // same as any other way of starting the workflow.
     WorkflowScheduler& scheduler_;
+
+    // Bounded pool that runs immediate-mode form dispatches off the request
+    // thread. Fixed-size and joined in stop()/~WebhookController - see the
+    // comment on DispatchConfig for why this cannot be an unbounded thread
+    // per request, and the comment on stop() for why it cannot be detached.
+    std::size_t dispatch_queue_capacity_;
+    std::vector<std::thread> dispatch_workers_;
+    std::queue<std::function<void()>> dispatch_queue_;
+    std::mutex dispatch_mutex_;
+    std::condition_variable dispatch_cv_;
+    bool dispatch_running_ = true;
 };
 
 } // namespace smartbotic::webserver::api

src/webserver/webserver_service.cpp

@@ -203,6 +203,9 @@ WebServerServiceConfig WebServerService::loadConfig(const std::filesystem::path&
         // field cannot fit inside a 16 MB total body once multipart framing
         // is added on top.
         config.max_upload_mb = cfg.getOr<int>("server.max_upload_mb", 32);
+        config.form_dispatch_threads = cfg.getOr<int>("server.form_dispatch_threads", 4);
+        config.form_dispatch_queue_capacity =
+            cfg.getOr<int>("server.form_dispatch_queue_capacity", 32);
 
         // Runner config
         config.runner_config.heartbeat_timeout_sec =
@@ -323,6 +326,12 @@ void WebServerService::stop() {
     LOG_INFO("Stopping WebServer service...");
 
     http_server_->stop();
+    // No new requests can arrive once the HTTP server is stopped, but the
+    // dispatch pool's own workers may still be mid-run - stop() joins them,
+    // so nothing is left running against a controller about to be destroyed.
+    if (webhook_ctrl_) {
+        webhook_ctrl_->stop();
+    }
     ws_server_->stop();
     credential_server_->stop();
     node_sync_server_->stop();
@@ -402,8 +411,12 @@ void WebServerService::setupRoutes() {
         });
     runner_ctrl_->registerRoutes(server);
 
+    api::WebhookController::DispatchConfig form_dispatch_config;
+    form_dispatch_config.threads = config_.form_dispatch_threads;
+    form_dispatch_config.queue_capacity = config_.form_dispatch_queue_capacity;
     webhook_ctrl_ = std::make_unique<api::WebhookController>(
-        *storage_, *runner_registry_, *load_balancer_, *ws_server_, *node_store_, *scheduler_);
+        *storage_, *runner_registry_, *load_balancer_, *ws_server_, *node_store_, *scheduler_,
+        form_dispatch_config);
     webhook_ctrl_->registerRoutes(server);
 
     database_ctrl_ = std::make_unique<api::DatabaseController>(*storage_, *auth_middleware_);

src/webserver/webserver_service.hpp

@@ -66,6 +66,12 @@ struct WebServerServiceConfig {
     std::string database_address = "localhost:9004";
     std::string database_project = "smartbotic-automation";
     int max_upload_mb = 32;
+    // Bounds WebhookController's dispatch pool, which runs immediate-mode
+    // form submissions off the request thread. The form endpoint is public
+    // and unauthenticated by design, so these keep a burst of submissions
+    // from spawning without limit.
+    int form_dispatch_threads = 4;
+    int form_dispatch_queue_capacity = 32;
     runners::RunnerRegistryConfig runner_config;
     runners::LoadBalancerConfig load_balancer_config;
     auth::JwtUtils::Config jwt_config;