server: add endpoint for a factory to fail a worker

Author: emilyalbini

client/openapi.json

@@ -922,6 +922,42 @@
         }
       }
     },
+    "/0/factory/worker/{worker}/fail": {
+      "post": {
+        "operationId": "factory_worker_fail",
+        "parameters": [
+          {
+            "in": "path",
+            "name": "worker",
+            "required": true,
+            "schema": {
+              "type": "string"
+            }
+          }
+        ],
+        "requestBody": {
+          "content": {
+            "application/json": {
+              "schema": {
+                "$ref": "#/components/schemas/FactoryWorkerFail"
+              }
+            }
+          },
+          "required": true
+        },
+        "responses": {
+          "204": {
+            "description": "resource updated"
+          },
+          "4XX": {
+            "$ref": "#/components/responses/Error"
+          },
+          "5XX": {
+            "$ref": "#/components/responses/Error"
+          }
+        }
+      }
+    },
     "/0/factory/worker/{worker}/flush": {
       "post": {
         "operationId": "factory_worker_flush",
@@ -2943,6 +2979,17 @@
           "target"
         ]
       },
+      "FactoryWorkerFail": {
+        "type": "object",
+        "properties": {
+          "reason": {
+            "type": "string"
+          }
+        },
+        "required": [
+          "reason"
+        ]
+      },
       "FactoryWorkerResult": {
         "type": "object",
         "properties": {

server/src/api/factory.rs

@@ -270,6 +270,53 @@ pub(crate) async fn factory_worker_flush(
     Ok(HttpResponseUpdatedNoContent())
 }
 
+#[derive(Deserialize, Serialize, JsonSchema)]
+pub(crate) struct FactoryWorkerFail {
+    reason: String,
+}
+
+#[endpoint {
+    method = POST,
+    path = "/0/factory/worker/{worker}/fail",
+}]
+pub(crate) async fn factory_worker_fail(
+    rqctx: RequestContext<Arc<Central>>,
+    path: TypedPath<WorkerPath>,
+    body: TypedBody<FactoryWorkerFail>,
+) -> DSResult<HttpResponseUpdatedNoContent> {
+    let c = rqctx.context();
+    let log = &rqctx.log;
+
+    let worker_id = path.into_inner().worker()?;
+    let reason = body.into_inner().reason;
+
+    let factory = c.require_factory(log, &rqctx.request).await?;
+    let worker = c.db.worker(worker_id).or_500()?;
+    factory.owns(log, &worker)?;
+
+    warn!(
+        log, "worker failed!";
+        "id" => worker.id.to_string(), "reason" => &reason,
+    );
+
+    /*
+     * Record in the database that the worker has failed.  This routine will
+     * take care of reporting failure in any assigned jobs, marking the worker
+     * as held, etc.
+     */
+    let failed_jobs = c.db.worker_mark_failed(worker.id, &reason).or_500()?;
+    if !failed_jobs.is_empty() {
+        let jobs = failed_jobs
+            .into_iter()
+            .map(|j| j.to_string())
+            .collect::<Vec<_>>()
+            .join(", ");
+        warn!(log, "worker {} failing caused jobs {jobs} to fail", worker.id);
+    }
+
+    Ok(HttpResponseUpdatedNoContent())
+}
+
 #[derive(Debug, Deserialize, JsonSchema)]
 pub(crate) struct FactoryWorkerAssociate {
     private: String,

server/src/main.rs

@@ -1011,6 +1011,7 @@ async fn main() -> Result<()> {
     ad.register(api::factory::factory_ping)?;
     ad.register(api::factory::factory_worker_create)?;
     ad.register(api::factory::factory_worker_append)?;
+    ad.register(api::factory::factory_worker_fail)?;
     ad.register(api::factory::factory_worker_flush)?;
     ad.register(api::factory::factory_worker_associate)?;
     ad.register(api::factory::factory_worker_destroy)?;