|
|
@@ -1,6 +1,7 @@
|
|
|
#include "webhook_controller.hpp"
|
|
|
#include "logging/logger.hpp"
|
|
|
#include "common/time_utils.hpp"
|
|
|
+#include "common/uuid.hpp"
|
|
|
#include "proto/runner.grpc.pb.h"
|
|
|
#include <grpcpp/grpcpp.h>
|
|
|
#include <cctype>
|
|
|
@@ -14,12 +15,14 @@ WebhookController::WebhookController(storage::StorageClient& storage,
|
|
|
runners::RunnerRegistry& registry,
|
|
|
runners::LoadBalancer& load_balancer,
|
|
|
WebSocketServer& ws_server,
|
|
|
- nodes::NodeStore& node_store)
|
|
|
+ nodes::NodeStore& node_store,
|
|
|
+ WorkflowScheduler& scheduler)
|
|
|
: storage_(storage)
|
|
|
, registry_(registry)
|
|
|
, load_balancer_(load_balancer)
|
|
|
, ws_server_(ws_server)
|
|
|
- , node_store_(node_store) {}
|
|
|
+ , node_store_(node_store)
|
|
|
+ , scheduler_(scheduler) {}
|
|
|
|
|
|
void WebhookController::registerRoutes(httplib::Server& server) {
|
|
|
// Match any HTTP method for webhooks
|
|
|
@@ -144,6 +147,16 @@ 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.
|
|
|
+ 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);
|
|
|
grpc_req.set_trigger_type("webhook");
|
|
|
@@ -158,11 +171,30 @@ void WebhookController::handleWebhook(const httplib::Request& req, httplib::Resp
|
|
|
auto status = stub->ExecuteWorkflow(&grpc_ctx, grpc_req, &grpc_res);
|
|
|
|
|
|
if (!status.ok()) {
|
|
|
+ // No run means no slot. Left held, this would block every scheduled
|
|
|
+ // tick until the watchdog deadline - up to an hour - because a runner
|
|
|
+ // was briefly unreachable. A deadline is the exception: the call gave
|
|
|
+ // up, the workflow did not, and it will report its own end - releasing
|
|
|
+ // the slot here would say idle while it is still working.
|
|
|
+ if (status.error_code() != grpc::StatusCode::DEADLINE_EXCEEDED) {
|
|
|
+ scheduler_.notifyExecutionFinished(workflow_id, execution_id);
|
|
|
+ }
|
|
|
LOG_ERROR("Webhook execution failed: {}", status.error_message());
|
|
|
sendError(res, "Execution failed: " + status.error_message(), 500);
|
|
|
return;
|
|
|
}
|
|
|
|
|
|
+ // The runner is expected to take the id offered above. If it ever stops
|
|
|
+ // doing so, the slot claimed under our id has nobody to release it and the
|
|
|
+ // schedule would stall until the watchdog deadline, so move it rather than
|
|
|
+ // leave a stuck workflow behind a silent assumption.
|
|
|
+ if (grpc_res.execution_id() != execution_id) {
|
|
|
+ LOG_WARN("Webhook run came back as {} but was dispatched as {} - moving the schedule slot",
|
|
|
+ grpc_res.execution_id(), execution_id);
|
|
|
+ scheduler_.notifyExecutionFinished(workflow_id, execution_id);
|
|
|
+ scheduler_.notifyExecutionStarted(workflow_id, grpc_res.execution_id());
|
|
|
+ }
|
|
|
+
|
|
|
// A workflow that paused mid-run is not done: final_output is still null
|
|
|
// (the body would just be the literal string "null"), and labelling this
|
|
|
// ".completed" is false. Report it honestly - 202 Accepted, a status of
|