fix: pressing Execute stops reporting failure on a run over 30 seconds

The editor showed no progress and then "Execution failed: timeout", while
the run itself was healthy and finished normally. Reproduced: POST
/workflows/{id}/execute took 30.01 seconds and answered 500 "Deadline
Exceeded", and the browser's own axios timeout is 30000ms, so both gave up
at the same instant.

workflow_controller.cpp already asked for fire-and-forget -
set_wait_for_completion(false) with a 30 second deadline - but that flag
does not do what it looks like. The runner's ExecuteWorkflow handler is
synchronous: it runs the workflow to completion before returning anything,
and the flag only decides whether the reply carries a result. So every
workflow longer than half a minute reported failure while working
perfectly. The same defect was found in the webhook path and named in
a69fa31.

Rather than a second copy of the fix, the bounded dispatch pool built for
webhook immediate-mode is now shared: dispatch_pool.{hpp,cpp}, used by
both controllers. Bounded workers, a bounded queue, workers joined at
shutdown, in-flight RPCs cancelled after a short grace, an honest 503 when
the queue is full rather than a cheerful acceptance of work nobody will
run.

Raising the timeout was the obvious alternative and the wrong one: a
bigger number moves the cliff, and a workflow that takes an hour is
legitimate.

Verified against the real pipeline:

  before   POST /execute   30.01s   500 Deadline Exceeded
  after    POST /execute    0.01s   200 {"executionId": ...}
  the run  completed in 218.9 seconds, 36 node records

Three and a half minutes is the case that could never have worked from the
editor before. 90 passed, 0 failed.

CMakeLists.txt

@@ -184,6 +184,7 @@ add_executable(smartbotic-webserver
     src/webserver/api/auth_controller.cpp
     src/webserver/api/credential_controller.cpp
     src/webserver/api/user_controller.cpp
+    src/webserver/api/dispatch_pool.cpp
     src/webserver/api/workflow_controller.cpp
     src/webserver/api/workflow_group_controller.cpp
     src/webserver/api/execution_controller.cpp

src/webserver/api/dispatch_pool.cpp

@@ -0,0 +1,143 @@
+#include "dispatch_pool.hpp"
+
+#include <grpcpp/grpcpp.h>
+
+#include "common/string_utils.hpp"
+#include "logging/logger.hpp"
+
+namespace smartbotic::webserver::api {
+
+using namespace common;
+
+DispatchPool::DispatchPool(DispatchPoolConfig config)
+    : label_kind_(std::move(config.label_kind))
+    , queue_capacity_(config.queue_capacity) {
+    workers_.reserve(config.threads);
+    for (std::size_t i = 0; i < config.threads; ++i) {
+        workers_.emplace_back(&DispatchPool::workerLoop, this);
+    }
+}
+
+DispatchPool::~DispatchPool() {
+    stop();
+}
+
+void DispatchPool::stop() {
+    std::vector<std::string> discarded_labels;
+    {
+        std::lock_guard<std::mutex> lock(queue_mutex_);
+        if (!running_) {
+            return;  // already stopped
+        }
+        running_ = false;
+
+        // Whatever is still queued never started, so it never claimed
+        // anything a running task would have claimed on its behalf -
+        // dropping it here leaks nothing. But every one of these was
+        // accepted onto the queue already, so silently discarding is worse
+        // than logging it - draining it here would mean shutdown waits for
+        // the entire backlog rather than just what is already running.
+        while (!queue_.empty()) {
+            discarded_labels.push_back(std::move(queue_.front().label));
+            queue_.pop();
+        }
+    }
+    if (!discarded_labels.empty()) {
+        LOG_WARN("Shutting down with {} queued {}(s) never started; discarded for: {}",
+                 discarded_labels.size(), label_kind_, StringUtils::join(discarded_labels, ", "));
+    }
+    queue_cv_.notify_all();
+
+    // A worker that already popped a task is running it outside this lock,
+    // so notify_all above does not reach it - workerLoop only checks
+    // running_ between tasks. That worker still has to reach
+    // registerActiveTask (called right before its blocking call starts)
+    // before it becomes visible here.
+    {
+        std::unique_lock<std::mutex> lock(active_mutex_);
+        // Setting this under the same lock registerActiveTask takes is what
+        // makes the race deterministic rather than timing-dependent: a task
+        // that registers after this point - however late, no matter how
+        // loaded the machine is - sees shutting_down_ already true and
+        // cancels itself right there (see registerActiveTask), instead of
+        // depending on a cancellation pass here having already run by the
+        // time it arrives.
+        shutting_down_ = true;
+        for (const auto& active : active_tasks_) {
+            LOG_WARN("Cancelling in-flight {} for {} at shutdown", label_kind_, active.label);
+            active.context->TryCancel();
+        }
+        // Wait for every active task to unregister, rather than sleeping a
+        // flat interval regardless of whether anything was running at all.
+        // Still bounded - TryCancel is expected to unblock everything well
+        // inside this - so a pathological case does not turn shutdown
+        // unbounded, it just stops waiting and lets the join below find out
+        // how long it actually takes.
+        active_cv_.wait_for(lock, std::chrono::seconds(2),
+                            [this] { return active_tasks_.empty(); });
+    }
+
+    for (auto& worker : workers_) {
+        if (worker.joinable()) {
+            worker.join();
+        }
+    }
+    workers_.clear();
+}
+
+void DispatchPool::workerLoop() {
+    while (true) {
+        QueuedDispatch item;
+        {
+            std::unique_lock<std::mutex> lock(queue_mutex_);
+            queue_cv_.wait(lock, [this] {
+                return !running_ || !queue_.empty();
+            });
+            // Stop pulling from the queue the moment shutdown starts, even
+            // if work remains - stop() itself discards and logs whatever is
+            // left rather than this loop draining it, which would mean
+            // shutdown waits for the whole backlog instead of only what
+            // already started.
+            if (!running_) {
+                return;
+            }
+            item = std::move(queue_.front());
+            queue_.pop();
+        }
+        item.task();
+    }
+}
+
+bool DispatchPool::tryEnqueue(const std::string& label, std::function<void()> task) {
+    {
+        std::lock_guard<std::mutex> lock(queue_mutex_);
+        if (!running_ || queue_.size() >= queue_capacity_) {
+            return false;
+        }
+        queue_.push(QueuedDispatch{label, std::move(task)});
+    }
+    queue_cv_.notify_one();
+    return true;
+}
+
+void DispatchPool::registerActiveTask(::grpc::ClientContext* context, const std::string& label) {
+    std::lock_guard<std::mutex> lock(active_mutex_);
+    active_tasks_.push_back(ActiveTask{context, label});
+    if (shutting_down_) {
+        // stop()'s own cancellation pass already ran by the time this task
+        // reached here - it would otherwise sit uncancelled until the
+        // bounded wait in stop() gives up, or worse, until the task's own
+        // long deadline. Cancel it the moment it becomes visible instead.
+        LOG_WARN("Cancelling in-flight {} for {} at shutdown (registered after shutdown began)",
+                 label_kind_, label);
+        context->TryCancel();
+    }
+}
+
+void DispatchPool::unregisterActiveTask(::grpc::ClientContext* context) {
+    std::lock_guard<std::mutex> lock(active_mutex_);
+    std::erase_if(active_tasks_, [context](const ActiveTask& t) { return t.context == context; });
+    active_cv_.notify_all();
+}
+
+} // namespace smartbotic::webserver::api

src/webserver/api/dispatch_pool.hpp

@@ -0,0 +1,126 @@
+#pragma once
+
+#include <condition_variable>
+#include <cstddef>
+#include <functional>
+#include <mutex>
+#include <queue>
+#include <string>
+#include <thread>
+#include <vector>
+
+namespace grpc {
+class ClientContext;
+}
+
+namespace smartbotic::webserver::api {
+
+// Bounds a background pool that runs a blocking gRPC call whose caller does
+// not want to wait on it. The runner's ExecuteWorkflow RPC is synchronous no
+// matter what wait_for_completion says - it runs the workflow to completion
+// before returning anything - so the only way to answer a caller promptly is
+// to run that blocking call somewhere else and let it finish on its own
+// schedule.
+//
+// Two callers share this: WebhookController's immediate-mode form dispatch
+// (the original implementation this was extracted from) and
+// WorkflowController's editor "Execute" button. Each owns its own pool
+// instance - sized and configured for its own traffic - rather than sharing
+// one, so a burst on one path cannot starve the other's queue. What they
+// share is the policy: fixed-size and joined at shutdown, in-flight calls
+// cancelled with TryCancel rather than waited out, and a full queue answered
+// honestly instead of accepting work nobody will run.
+//
+// Fixed-size and joined in stop()/the destructor - not a thread per request,
+// and not a detached thread either:
+//   - a detached thread has no owner, so nothing joins it, and it would
+//     still be running against whatever it captured after the object that
+//     queued it is destroyed - an ordinary `systemctl --user restart` while
+//     a run is in flight would be a use-after-free;
+//   - the webhook path 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. The editor path is
+//     authenticated, but bounding it costs nothing and avoids a second
+//     policy to reason about.
+struct DispatchPoolConfig {
+    std::size_t threads = 4;
+    std::size_t queue_capacity = 32;
+    // Names what a queued item is, for log lines only - e.g. "immediate-mode
+    // form dispatch" or "editor execute dispatch". Read, never compared.
+    std::string label_kind = "dispatch";
+};
+
+class DispatchPool {
+public:
+    explicit DispatchPool(DispatchPoolConfig config = {});
+    ~DispatchPool();
+
+    DispatchPool(const DispatchPool&) = delete;
+    DispatchPool& operator=(const DispatchPool&) = delete;
+
+    // Stops accepting new work and joins every worker. Must run before the
+    // object that queued tasks capturing itself is destroyed - a queued task
+    // that captures `this` of its owner would otherwise be a use-after-free
+    // the moment that owner is gone. Safe to call more than once
+    // sequentially - it is not safe to call concurrently with itself, since
+    // the running task/join sequence has no lock protecting it from a second
+    // stop() racing the same active-task scan.
+    void stop();
+
+    // Queues a task on the pool. Returns false, without queuing anything,
+    // once the bounded queue is full - the caller must answer honestly
+    // rather than accept work nobody will run. `label` identifies the task
+    // for logging (a workflow id) - it plays no role in scheduling.
+    bool tryEnqueue(const std::string& label, std::function<void()> task);
+
+    // A task that has been popped off the queue but has not yet registered
+    // its gRPC context has already left the queue by the time stop() drains
+    // it, but is not yet visible to the cancellation pass either.
+    // registerActiveTask checks whether stop() has already begun under the
+    // same lock stop() sets it with, so a task that registers late cancels
+    // itself the moment it registers rather than depending on a
+    // cancellation pass that already ran before it got there.
+    void registerActiveTask(::grpc::ClientContext* context, const std::string& label);
+    void unregisterActiveTask(::grpc::ClientContext* context);
+
+private:
+    void workerLoop();
+
+    // A queued but not-yet-started task. Kept alongside its label so a
+    // discard at shutdown can be logged by name, not just counted.
+    struct QueuedDispatch {
+        std::string label;
+        std::function<void()> task;
+    };
+
+    // A task a worker is actively running. context is owned by the worker's
+    // stack frame (the task's local grpc::ClientContext) - stop() only ever
+    // calls TryCancel() on it, never deletes it, and a task unregisters
+    // itself before returning.
+    struct ActiveTask {
+        ::grpc::ClientContext* context;
+        std::string label;
+    };
+
+    std::string label_kind_;
+    std::size_t queue_capacity_;
+    std::vector<std::thread> workers_;
+    std::queue<QueuedDispatch> queue_;
+    std::mutex queue_mutex_;
+    std::condition_variable queue_cv_;
+    bool running_ = true;
+
+    // Tasks currently running their blocking call, so stop() can cancel them
+    // instead of waiting out their deadline. Separate from queue_mutex_:
+    // registration happens from inside a running task, while queue_mutex_
+    // guards the queue a worker is not yet done draining.
+    std::mutex active_mutex_;
+    std::vector<ActiveTask> active_tasks_;
+    // Notified whenever active_tasks_ changes, so stop() can wait for
+    // "no active tasks" instead of a fixed sleep.
+    std::condition_variable active_cv_;
+    // Set once stop() begins, under active_mutex_. See registerActiveTask.
+    bool shutting_down_ = false;
+};
+
+} // namespace smartbotic::webserver::api

src/webserver/api/webhook_controller.cpp

@@ -51,11 +51,8 @@ WebhookController::WebhookController(storage::StorageClient& storage,
     , node_store_(node_store)
     , scheduler_(scheduler)
     , jwt_utils_(jwt_utils)
-    , 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);
-    }
+    , dispatch_pool_(DispatchPoolConfig{dispatch_config.threads, dispatch_config.queue_capacity,
+                                        "immediate-mode form dispatch"}) {
 }
 
 WebhookController::~WebhookController() {
@@ -63,137 +60,27 @@ WebhookController::~WebhookController() {
 }
 
 void WebhookController::stop() {
-    std::vector<std::string> discarded_workflow_ids;
-    {
-        std::lock_guard<std::mutex> lock(dispatch_mutex_);
-        if (!dispatch_running_) {
-            return;  // already stopped
-        }
-        dispatch_running_ = false;
-
-        // Whatever is still queued never started, so it never reached
-        // scheduler_.notifyExecutionStarted (that happens once a worker pops
-        // it, inside the task) - dropping it here leaks no slot. But every
-        // one of these submitters was already told "thanks, your answer was
-        // received", so silently discarding is exactly the kind of promise
-        // this project keeps getting bitten by breaking - log it instead of
-        // draining it. Draining it would mean shutdown waits for the entire
-        // backlog rather than just what is already running, which is what
-        // made the previous version of this able to block for hours.
-        while (!dispatch_queue_.empty()) {
-            discarded_workflow_ids.push_back(std::move(dispatch_queue_.front().workflow_id));
-            dispatch_queue_.pop();
-        }
-    }
-    if (!discarded_workflow_ids.empty()) {
-        LOG_WARN("Shutting down with {} queued immediate-mode form submission(s) never started; "
-                 "discarded for workflows: {}",
-                 discarded_workflow_ids.size(), StringUtils::join(discarded_workflow_ids, ", "));
-    }
-    dispatch_cv_.notify_all();
-
-    // A worker that already popped a task is running it outside this lock,
-    // so notify_all above does not reach it - dispatchWorkerLoop only checks
-    // dispatch_running_ between tasks. That worker still has to reach
-    // registerActiveTask (called right before the blocking gRPC call starts)
-    // before it becomes visible here.
-    {
-        std::unique_lock<std::mutex> lock(active_mutex_);
-        // Setting this under the same lock registerActiveTask takes is what
-        // makes the race deterministic rather than timing-dependent: a task
-        // that registers after this point - however late, no matter how
-        // loaded the machine is - sees shutting_down_ already true and
-        // cancels itself right there (see registerActiveTask), instead of
-        // depending on a cancellation pass here having already run by the
-        // time it arrives.
-        shutting_down_ = true;
-        for (const auto& active : active_tasks_) {
-            LOG_WARN("Cancelling in-flight immediate-mode form dispatch for workflow {} at shutdown",
-                     active.workflow_id);
-            active.context->TryCancel();
-        }
-        // Wait for every active task to unregister (its worker returned from
-        // the gRPC call and is on its way back to dispatchWorkerLoop, which
-        // exits immediately once dispatch_running_ is false), rather than
-        // sleeping a flat 2 seconds regardless of whether anything was
-        // running at all. Still bounded by the same grace period as before -
-        // TryCancel is expected to unblock everything well inside it - so a
-        // pathological case does not turn shutdown unbounded, it just stops
-        // waiting and lets the join below find out how long it actually
-        // takes.
-        active_cv_.wait_for(lock, std::chrono::seconds(2),
-                            [this] { return active_tasks_.empty(); });
-    }
-
-    // TryCancel unblocks this process's own client call quickly regardless
-    // of whether the runner notices - see the comment above registerActiveTask
-    // in the header, and the follow-up note in the task report about whether
-    // the runner treats a cancelled RPC as a reason to report
-    // execution.cancelled/.failed. Either way, nothing can still be running
-    // against this controller once join returns, which is the lifetime
-    // guarantee that matters here.
-    for (auto& worker : dispatch_workers_) {
-        if (worker.joinable()) {
-            worker.join();
-        }
-    }
-    dispatch_workers_.clear();
-}
-
-void WebhookController::dispatchWorkerLoop() {
-    while (true) {
-        QueuedDispatch item;
-        {
-            std::unique_lock<std::mutex> lock(dispatch_mutex_);
-            dispatch_cv_.wait(lock, [this] {
-                return !dispatch_running_ || !dispatch_queue_.empty();
-            });
-            // Stop pulling from the queue the moment shutdown starts, even if
-            // work remains - stop() itself discards and logs whatever is left
-            // rather than this loop draining it, which would mean shutdown
-            // waits for the whole backlog instead of only what already
-            // started.
-            if (!dispatch_running_) {
-                return;
-            }
-            item = std::move(dispatch_queue_.front());
-            dispatch_queue_.pop();
-        }
-        item.task();
-    }
+    // Whatever was still queued never started, so it never reached
+    // scheduler_.notifyExecutionStarted (that happens once a worker pops it,
+    // inside the task) - discarding it leaks no slot. Every in-flight call
+    // is cancelled with TryCancel rather than waited out. See
+    // dispatch_pool.hpp for the full reasoning - this used to be
+    // WebhookController's own private implementation; it now lives there so
+    // WorkflowController's editor "Execute" path can use the same policy
+    // without a second copy to keep in sync.
+    dispatch_pool_.stop();
 }
 
 bool WebhookController::tryEnqueueDispatch(const std::string& workflow_id, 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(QueuedDispatch{workflow_id, std::move(task)});
-    }
-    dispatch_cv_.notify_one();
-    return true;
+    return dispatch_pool_.tryEnqueue(workflow_id, std::move(task));
 }
 
 void WebhookController::registerActiveTask(::grpc::ClientContext* context, const std::string& workflow_id) {
-    std::lock_guard<std::mutex> lock(active_mutex_);
-    active_tasks_.push_back(ActiveTask{context, workflow_id});
-    if (shutting_down_) {
-        // stop()'s own cancellation pass already ran by the time this task
-        // reached here - it would otherwise sit uncancelled until the
-        // bounded wait in stop() gives up, or worse, until the hour-long
-        // deadline set on this context. Cancel it the moment it becomes
-        // visible instead.
-        LOG_WARN("Cancelling in-flight immediate-mode form dispatch for workflow {} at shutdown "
-                 "(registered after shutdown began)", workflow_id);
-        context->TryCancel();
-    }
+    dispatch_pool_.registerActiveTask(context, workflow_id);
 }
 
 void WebhookController::unregisterActiveTask(::grpc::ClientContext* context) {
-    std::lock_guard<std::mutex> lock(active_mutex_);
-    std::erase_if(active_tasks_, [context](const ActiveTask& t) { return t.context == context; });
-    active_cv_.notify_all();
+    dispatch_pool_.unregisterActiveTask(context);
 }
 
 void WebhookController::registerRoutes(httplib::Server& server) {

src/webserver/api/webhook_controller.hpp

@@ -2,18 +2,14 @@
 
 #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"
 #include "../nodes/node_store.hpp"
 #include "../scheduler/workflow_scheduler.hpp"
 #include "../auth/jwt_utils.hpp"
+#include "dispatch_pool.hpp"
 
 namespace grpc {
 class ClientContext;
@@ -47,13 +43,12 @@ public:
 
     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 sequentially - it is not
-    // safe to call concurrently from two threads at once, since the running
-    // task/join sequence has no lock protecting it from another stop()
-    // clearing dispatch_workers_ or racing the same active_tasks_ scan.
+    // Stops accepting new immediate-mode dispatches and joins every worker of
+    // dispatch_pool_. 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 sequentially - see
+    // DispatchPool::stop() for why it is not safe to call concurrently from
+    // two threads at once.
     void stop();
 
     // A signed token rather than the password itself: the password never
@@ -84,19 +79,11 @@ private:
 
     // 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.
+    // honestly rather than accept work nobody will run. Thin forwarders to
+    // dispatch_pool_ - see dispatch_pool.hpp for the pool itself, shared in
+    // implementation (not instance) with WorkflowController's editor
+    // "Execute" path.
     bool tryEnqueueDispatch(const std::string& workflow_id, std::function<void()> task);
-    void dispatchWorkerLoop();
-
-    // A task that has been popped off the queue but has not yet registered
-    // its gRPC context (registerActiveTask has not run) has already left the
-    // queue by the time stop() drains it, but is not yet visible to the
-    // cancellation pass either. registerActiveTask checks shutting_down_
-    // under the same lock stop() sets it with, so this task cancels itself
-    // the moment it registers rather than depending on stop() having already
-    // run its pass - deterministic regardless of how long the task took to
-    // get here, not bounded by the grace period stop() waits before giving
-    // up on the join.
     void registerActiveTask(::grpc::ClientContext* context, const std::string& workflow_id);
     void unregisterActiveTask(::grpc::ClientContext* context);
 
@@ -113,51 +100,12 @@ private:
     // signDetached/verifyDetached rather than duplicated here.
     auth::JwtUtils& jwt_utils_;
 
-    // A queued but not-yet-started dispatch. Kept alongside its workflow id
-    // so a discard at shutdown can be logged by name, not just counted.
-    struct QueuedDispatch {
-        std::string workflow_id;
-        std::function<void()> task;
-    };
-
-    // A dispatch a worker is actively running. context is owned by the
-    // worker's stack frame (the task's local grpc::ClientContext) - stop()
-    // only ever calls TryCancel() on it, never deletes it, and a task
-    // unregisters itself before returning.
-    struct ActiveTask {
-        ::grpc::ClientContext* context;
-        std::string workflow_id;
-    };
-
     // 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<QueuedDispatch> dispatch_queue_;
-    std::mutex dispatch_mutex_;
-    std::condition_variable dispatch_cv_;
-    bool dispatch_running_ = true;
-
-    // Tasks currently running a dispatch's gRPC call, so stop() can cancel
-    // them instead of waiting out their deadline (an hour - see the comment
-    // where it is set). Separate from dispatch_mutex_: registration happens
-    // from inside a running task, while dispatch_mutex_ guards the queue a
-    // worker is not yet done draining.
-    std::mutex active_mutex_;
-    std::vector<ActiveTask> active_tasks_;
-    // Notified whenever active_tasks_ changes, so stop() can wait for
-    // "no active tasks" instead of a fixed sleep - fast when shutdown is
-    // idle, and still bounded by the same grace period when it is not.
-    std::condition_variable active_cv_;
-    // Set once stop() begins, under active_mutex_. A task that pops off the
-    // dispatch queue before dispatch_running_ goes false but has not yet
-    // reached registerActiveTask (see the comment there) checks this and
-    // self-cancels the moment it registers, instead of depending on stop()'s
-    // cancellation pass having already run by the time it gets there - which
-    // a loaded machine could outrun.
-    bool shutting_down_ = false;
+    // thread. Fixed-size and joined in stop()/~WebhookController - see
+    // dispatch_pool.hpp's class comment for why this cannot be an unbounded
+    // thread per request, and its stop() comment for why it cannot be
+    // detached.
+    DispatchPool dispatch_pool_;
 };
 
 } // namespace smartbotic::webserver::api

src/webserver/api/workflow_controller.cpp

@@ -19,7 +19,8 @@ WorkflowController::WorkflowController(storage::StorageClient& storage,
                                        DatabaseWatcher& db_watcher,
                                        auth::AccessControl& access,
                                        nodes::NodeStore& node_store,
-                                       retention::RetentionService& retention)
+                                       retention::RetentionService& retention,
+                                       DispatchConfig dispatch_config)
     : storage_(storage)
     , middleware_(middleware)
     , registry_(registry)
@@ -29,7 +30,12 @@ WorkflowController::WorkflowController(storage::StorageClient& storage,
     , db_watcher_(db_watcher)
     , access_(access)
     , node_store_(node_store)
-    , retention_(retention) {}
+    , retention_(retention)
+    , dispatch_pool_(std::move(dispatch_config)) {}
+
+void WorkflowController::stop() {
+    dispatch_pool_.stop();
+}
 
 
 namespace {
@@ -730,9 +736,33 @@ void WorkflowController::executeWorkflow(const httplib::Request& req, httplib::R
         }
     } catch (...) {}
 
-    // Call runner to execute workflow
+    // The runner's ExecuteWorkflow RPC is a plain synchronous call: it runs
+    // the workflow to completion before returning anything at all -
+    // wait_for_completion only controls whether the response carries a
+    // result, it does not make the server return early (see the comment in
+    // webhook_controller.cpp's immediate-mode branch, which hit this same
+    // fact first). Holding this handler's thread on that call would hold the
+    // editor's "Execute" button for however long the run takes, and the
+    // editor's axios client times out at 30 seconds - a fast build for most
+    // workflows, but a lie against any run that legitimately takes longer,
+    // and a workflow taking an hour is legitimate. So the call runs on the
+    // controller's own bounded dispatch pool instead of this thread, exactly
+    // like the webhook path's immediate-mode form dispatch: nothing is left
+    // running against `this` after a restart (dispatch_pool_.stop() joins
+    // every worker before this controller is destroyed - see
+    // WorkflowController::stop()), and a burst of Execute presses is
+    // throttled rather than spawning threads without limit.
+    //
+    // The scheduler slot is claimed inside the queued task, not here, so a
+    // request that is refused for a full queue never claims one. Once
+    // claimed, it is released the same way any other dispatch's is: the
+    // runner reports execution.completed/failed/cancelled/waiting to
+    // /api/v1/internal/execution-event, and execution_controller.cpp
+    // releases it there.
+    const std::string execution_id = UUID::generatePrefixed("exec");
+    trigger_data["_assignedExecutionId"] = execution_id;
+
     auto channel = load_balancer_.createChannel(runner->address);
-    auto stub = proto::RunnerService::NewStub(channel);
 
     proto::ExecuteWorkflowRequest grpc_req;
     grpc_req.set_workflow_id(workflow_id);
@@ -743,40 +773,69 @@ void WorkflowController::executeWorkflow(const httplib::Request& req, httplib::R
     grpc_req.set_trigger_data(trigger_data.dump());
     grpc_req.set_wait_for_completion(false);
 
-    proto::ExecuteWorkflowResponse grpc_res;
-    grpc::ClientContext grpc_ctx;
-    grpc_ctx.set_deadline(std::chrono::system_clock::now() + std::chrono::seconds(30));
+    const std::string runner_id = runner->id;
+    bool accepted = dispatch_pool_.tryEnqueue(workflow_id,
+            [this, channel, workflow_id, execution_id, runner_id, grpc_req]() {
+        scheduler_.notifyExecutionStarted(workflow_id, execution_id);
+
+        auto bg_stub = proto::RunnerService::NewStub(channel);
+        proto::ExecuteWorkflowResponse bg_res;
+        grpc::ClientContext bg_ctx;
+        // Nobody is waiting on this response under normal operation, so the
+        // deadline only needs to bound a runner that never answers at all.
+        // At shutdown, stop() calls TryCancel() on this context directly
+        // instead of waiting for the deadline.
+        bg_ctx.set_deadline(std::chrono::system_clock::now() + std::chrono::hours(1));
+
+        dispatch_pool_.registerActiveTask(&bg_ctx, workflow_id);
+        auto bg_status = bg_stub->ExecuteWorkflow(&bg_ctx, grpc_req, &bg_res);
+        dispatch_pool_.unregisterActiveTask(&bg_ctx);
+
+        if (!bg_status.ok()) {
+            // Neither a lapsed deadline nor a shutdown-time TryCancel means
+            // the run stopped - both only end this process's own wait. The
+            // runner's ExecuteWorkflow handler never checks whether its
+            // caller cancelled, so the workflow keeps running server-side
+            // regardless and will still report its own
+            // execution.completed/failed/waiting event when it finishes -
+            // releasing the slot here would say idle while it is still
+            // working.
+            if (bg_status.error_code() != grpc::StatusCode::DEADLINE_EXCEEDED &&
+                bg_status.error_code() != grpc::StatusCode::CANCELLED) {
+                scheduler_.notifyExecutionFinished(workflow_id, execution_id);
+            }
+            LOG_ERROR("Editor execution failed for workflow {}: {}",
+                      workflow_id, bg_status.error_message());
+            return;
+        }
 
-    auto status = stub->ExecuteWorkflow(&grpc_ctx, grpc_req, &grpc_res);
+        if (bg_res.execution_id() != execution_id) {
+            LOG_WARN("Editor execution for workflow {} came back as {} but was dispatched as {} - "
+                     "moving the schedule slot",
+                     workflow_id, bg_res.execution_id(), execution_id);
+            scheduler_.notifyExecutionFinished(workflow_id, execution_id);
+            scheduler_.notifyExecutionStarted(workflow_id, bg_res.execution_id());
+        }
+    });
 
-    if (!status.ok()) {
-        LOG_ERROR("Failed to execute workflow: {}", status.error_message());
-        sendError(res, "Failed to execute workflow: " + status.error_message(), 500);
+    if (!accepted) {
+        // The queue is full: refuse honestly rather than tell the editor
+        // "started" for a run that will never happen. No slot was claimed
+        // for this one.
+        LOG_WARN("Editor execute dispatch queue is full, refusing execution for workflow {}",
+                 workflow_id);
+        sendError(res, "Too many executions in progress right now. Try again shortly.", 503);
         return;
     }
 
-    // The scheduler counts what is running so an overlap policy of skip or
-    // queue can hold a tick back. It only ever heard about runs it started
-    // itself, so a run started from the editor was invisible: a workflow set to
-    // never overlap would happily get a second scheduled run on top of the one
-    // someone was watching.
-    scheduler_.notifyExecutionStarted(workflow_id, grpc_res.execution_id());
-
     nlohmann::json response;
-    response["executionId"] = grpc_res.execution_id();
-    response["status"] = grpc_res.status();
-    response["runnerId"] = runner->id;
-
-    // Broadcast execution started
-    ws_server_.broadcast("executions." + grpc_res.execution_id() + ".started", {
-        {"executionId", grpc_res.execution_id()},
-        {"workflowId", workflow_id},
-        {"runnerId", runner->id}
-    });
+    response["executionId"] = execution_id;
+    response["status"] = "running";
+    response["runnerId"] = runner_id;
 
-    LOG_INFO("Workflow {} execution started: {} on runner {}",
-             workflow_id, grpc_res.execution_id(), runner->id);
-    sendJson(res, response, 202);
+    LOG_INFO("Workflow {} execution dispatched: {} on runner {}",
+             workflow_id, execution_id, runner_id);
+    sendJson(res, response, 200);
 }
 
 

src/webserver/api/workflow_controller.hpp

@@ -13,11 +13,14 @@
 #include "../scheduler/database_watcher.hpp"
 #include "../auth/access.hpp"
 #include "proto/runner.grpc.pb.h"
+#include "dispatch_pool.hpp"
 
 namespace smartbotic::webserver::api {
 
 class WorkflowController {
 public:
+    using DispatchConfig = DispatchPoolConfig;
+
     WorkflowController(storage::StorageClient& storage,
                       auth::AuthMiddleware& middleware,
                       runners::RunnerRegistry& registry,
@@ -27,10 +30,18 @@ public:
                       DatabaseWatcher& db_watcher,
                       auth::AccessControl& access,
                       nodes::NodeStore& node_store,
-                      retention::RetentionService& retention);
+                      retention::RetentionService& retention,
+                      DispatchConfig dispatch_config = {});
 
     void registerRoutes(httplib::Server& server);
 
+    // Stops accepting new "Execute" dispatches and joins every worker of the
+    // pool that runs them off the request thread. Must run before this
+    // controller is destroyed, for the same reason WebhookController::stop()
+    // must - a queued or in-flight task captures `this`. Safe to call more
+    // than once sequentially.
+    void stop();
+
 private:
     // CRUD operations
     void listWorkflows(const httplib::Request& req, httplib::Response& res,
@@ -84,6 +95,16 @@ private:
     nodes::NodeStore& node_store_;
     retention::RetentionService& retention_;
 
+    // Bounded pool that runs the editor's "Execute" dispatch off the request
+    // thread. See dispatch_pool.hpp: the runner's ExecuteWorkflow RPC is
+    // synchronous regardless of wait_for_completion, so returning promptly
+    // to the caller means running that blocking call somewhere else. Its own
+    // instance, not shared with WebhookController's pool, so a burst on one
+    // path cannot starve the other's queue - see the class comment on
+    // DispatchPool for why sharing the implementation still made sense even
+    // though the instances stay separate.
+    DispatchPool dispatch_pool_;
+
     // Helper to check for scheduled triggers and register/unregister with scheduler
     void updateScheduledTriggers(const std::string& workflow_id, bool activate);
     int getScheduledInterval(const nlohmann::json& workflow);

src/webserver/webserver_service.hpp

@@ -75,6 +75,14 @@ struct WebServerServiceConfig {
     // from spawning without limit.
     int form_dispatch_threads = 4;
     int form_dispatch_queue_capacity = 32;
+    // Bounds WorkflowController's dispatch pool, which runs the editor's
+    // "Execute" button off the request thread for the same reason the form
+    // pool above exists - the runner's ExecuteWorkflow RPC is synchronous
+    // regardless of wait_for_completion. This path is authenticated, so the
+    // exhaustion risk is lower than the form pool's, but bounding it still
+    // costs nothing.
+    int workflow_dispatch_threads = 4;
+    int workflow_dispatch_queue_capacity = 32;
     runners::RunnerRegistryConfig runner_config;
     runners::LoadBalancerConfig load_balancer_config;
     auth::JwtUtils::Config jwt_config;