Add button to set failed job to pending

CHANGELOG.unreleased.md

@@ -11,6 +11,7 @@ For upgrade instructions, please check the [migration guide](MIGRATIONS.released
 [Commits](https://github.com/scalableminds/webknossos/compare/25.02.1...HEAD)
 
 ### Added
+- Failed jobs may be retried by super-users. [#8377](https://github.com/scalableminds/webknossos/pull/8377)
 
 ### Changed
 - When using a zarr link to a wk-served data layer as another layer’s source, the user’s token is used to access the data. [#8322](https://github.com/scalableminds/webknossos/pull/8322/)

MIGRATIONS.unreleased.md

@@ -10,5 +10,5 @@ User-facing changes are documented in the [changelog](CHANGELOG.released.md).
 
 
 ### Postgres Evolutions:
-
 - [126-mag-real-paths.sql](conf/evolutions/126-mag-real-paths.sql)
+- [127-job-retried-by-super-user.sql](conf/evolutions/127-job-retried-by-super-user.sql)

app/controllers/JobController.scala

@@ -7,7 +7,7 @@ import com.scalableminds.util.tools.Fox
 import models.dataset.{DataStoreDAO, DatasetDAO, DatasetLayerAdditionalAxesDAO, DatasetService}
 import models.job._
 import models.organization.OrganizationDAO
-import models.user.MultiUserDAO
+import models.user.{MultiUserDAO, UserService}
 import play.api.i18n.Messages
 import play.api.libs.json._
 import play.api.mvc.{Action, AnyContent, PlayBodyParsers}
@@ -50,21 +50,21 @@ object AnimationJobOptions {
   implicit val jsonFormat: OFormat[AnimationJobOptions] = Json.format[AnimationJobOptions]
 }
 
-class JobController @Inject()(
-    jobDAO: JobDAO,
-    sil: Silhouette[WkEnv],
-    datasetDAO: DatasetDAO,
-    datasetService: DatasetService,
-    jobService: JobService,
-    workerService: WorkerService,
-    workerDAO: WorkerDAO,
-    datasetLayerAdditionalAxesDAO: DatasetLayerAdditionalAxesDAO,
-    wkconf: WkConf,
-    multiUserDAO: MultiUserDAO,
-    wkSilhouetteEnvironment: WkSilhouetteEnvironment,
-    slackNotificationService: SlackNotificationService,
-    organizationDAO: OrganizationDAO,
-    dataStoreDAO: DataStoreDAO)(implicit ec: ExecutionContext, playBodyParsers: PlayBodyParsers)
+class JobController @Inject()(jobDAO: JobDAO,
+                              sil: Silhouette[WkEnv],
+                              datasetDAO: DatasetDAO,
+                              datasetService: DatasetService,
+                              jobService: JobService,
+                              workerService: WorkerService,
+                              workerDAO: WorkerDAO,
+                              datasetLayerAdditionalAxesDAO: DatasetLayerAdditionalAxesDAO,
+                              wkconf: WkConf,
+                              multiUserDAO: MultiUserDAO,
+                              wkSilhouetteEnvironment: WkSilhouetteEnvironment,
+                              slackNotificationService: SlackNotificationService,
+                              organizationDAO: OrganizationDAO,
+                              dataStoreDAO: DataStoreDAO,
+                              userService: UserService)(implicit ec: ExecutionContext, playBodyParsers: PlayBodyParsers)
     extends Controller
     with Zarr3OutputHelper {
 
@@ -112,6 +112,16 @@ class JobController @Inject()(
     } yield Ok(js)
   }
 
+  def retry(id: ObjectId): Action[AnyContent] = sil.SecuredAction.async { implicit request =>
+    for {
+      _ <- bool2Fox(wkconf.Features.jobsEnabled) ?~> "job.disabled"
+      _ <- userService.assertIsSuperUser(request.identity) ?~> "notAllowed" ~> FORBIDDEN
+      job <- jobDAO.findOne(id)
+      _ <- jobDAO.retryOne(id)
+      js <- jobService.publicWrites(job)
+    } yield Ok(js)
+  }
+
   // Note that the dataset has to be registered by reserveUpload via the datastore first.
   def runConvertToWkwJob(datasetId: ObjectId, scale: String, unit: Option[String]): Action[AnyContent] =
     sil.SecuredAction.async { implicit request =>

app/models/job/Job.scala

@@ -29,6 +29,7 @@ case class Job(
     _voxelyticsWorkflowHash: Option[String] = None,
     latestRunId: Option[String] = None,
     returnValue: Option[String] = None,
+    retriedBySuperUser: Boolean = false,
     started: Option[Long] = None,
     ended: Option[Long] = None,
     created: Instant = Instant.now,
@@ -126,6 +127,7 @@ class JobDAO @Inject()(sqlClient: SqlClient)(implicit ec: ExecutionContext)
         r._VoxelyticsWorkflowhash,
         r.latestrunid,
         r.returnvalue,
+        r.retriedbysuperuser,
         r.started.map(_.getTime),
         r.ended.map(_.getTime),
         Instant.fromSql(r.created),
@@ -253,6 +255,15 @@ class JobDAO @Inject()(sqlClient: SqlClient)(implicit ec: ExecutionContext)
       _ <- run(q"""UPDATE webknossos.jobs SET manualState = $manualState WHERE _id = $id""".asUpdate)
     } yield ()
 
+  def retryOne(id: ObjectId)(implicit ctx: DBAccessContext): Fox[Unit] =
+    for {
+      _ <- assertUpdateAccess(id)
+      _ <- run(q"""UPDATE webknossos.jobs
+             SET state = 'PENDING', retriedBySuperUser = true
+             WHERE _id = $id
+             AND state = 'FAILURE'""".asUpdate)
+    } yield ()
+
   def updateStatus(jobId: ObjectId, s: JobStatus): Fox[Unit] =
     for {
       _ <- run(q"""UPDATE webknossos.jobs SET

app/models/job/JobService.scala

@@ -63,7 +63,7 @@ class JobService @Inject()(wkConf: WkConf,
         s"Failed job$superUserLabel",
         msg
       )
-      _ = sendFailedEmailNotification(user, jobAfterChange)
+      _ = if (!jobAfterChange.retriedBySuperUser) sendFailedEmailNotification(user, jobAfterChange)
     } yield ()
     ()
   }

conf/evolutions/127-job-retried-by-super-user.sql

@@ -0,0 +1,12 @@
+START TRANSACTION;
+
+do $$ begin ASSERT (select schemaVersion from webknossos.releaseInformation) = 126, 'Previous schema version mismatch'; end; $$ LANGUAGE plpgsql;
+
+DROP VIEW webknossos.jobs_;
+
+ALTER TABLE webknossos.jobs ADD column retriedBySuperUser BOOLEAN NOT NULL DEFAULT false;
+UPDATE webknossos.releaseInformation SET schemaVersion = 127;
+
+CREATE VIEW webknossos.jobs_ AS SELECT * FROM webknossos.jobs WHERE NOT isDeleted;
+
+COMMIT TRANSACTION;

conf/evolutions/reversions/127-job-retried-by-super-user.sql

@@ -0,0 +1,12 @@
+START TRANSACTION;
+
+do $$ begin ASSERT (select schemaVersion from webknossos.releaseInformation) = 127, 'Previous schema version mismatch'; end; $$ LANGUAGE plpgsql;
+
+DROP VIEW webknossos.jobs_;
+
+ALTER TABLE webknossos.jobs DROP column retriedBySuperUser;
+UPDATE webknossos.releaseInformation SET schemaVersion = 126;
+
+CREATE VIEW webknossos.jobs_ AS SELECT * FROM webknossos.jobs WHERE NOT isDeleted;
+
+COMMIT TRANSACTION;

conf/webknossos.latest.routes

@@ -269,6 +269,7 @@ POST          /jobs/run/findLargestSegmentId/:datasetId
 POST          /jobs/run/renderAnimation/:datasetId                                              controllers.JobController.runRenderAnimationJob(datasetId: ObjectId)
 GET           /jobs/:id                                                                         controllers.JobController.get(id: ObjectId)
 PATCH         /jobs/:id/cancel                                                                  controllers.JobController.cancel(id: ObjectId)
+PATCH         /jobs/:id/retry                                                                   controllers.JobController.retry(id: ObjectId)
 POST          /jobs/:id/status                                                                  controllers.WKRemoteWorkerController.updateJobStatus(key: String, id: ObjectId)
 POST          /jobs/:id/attachVoxelyticsWorkflow                                                controllers.WKRemoteWorkerController.attachVoxelyticsWorkflow(key: String, id: ObjectId)
 POST          /jobs/:id/attachDatasetToInference                                                controllers.WKRemoteWorkerController.attachDatasetToInference(key: String, id: ObjectId)

frontend/javascripts/admin/api/jobs.ts

@@ -74,6 +74,12 @@ export async function cancelJob(jobId: string): Promise<APIJob> {
   });
 }
 
+export async function retryJob(jobId: string): Promise<APIJob> {
+  return Request.receiveJSON(`/api/jobs/${jobId}/retry`, {
+    method: "PATCH",
+  });
+}
+
 export async function startConvertToWkwJob(
   datasetId: string,
   scale: Vector3,

frontend/javascripts/admin/job/job_list_view.tsx

@@ -7,22 +7,26 @@ import {
   EyeOutlined,
   InfoCircleOutlined,
   LoadingOutlined,
+  PlayCircleOutlined,
   QuestionCircleTwoTone,
 } from "@ant-design/icons";
 import { PropTypes } from "@scalableminds/prop-types";
-import { cancelJob, getJobs } from "admin/admin_rest_api";
+import { cancelJob, getJobs, retryJob } from "admin/admin_rest_api";
 import { Input, Modal, Spin, Table, Tooltip, Typography } from "antd";
 import { AsyncLink } from "components/async_clickables";
 import FormattedDate from "components/formatted_date";
 import { confirmAsync } from "dashboard/dataset/helper_components";
 import { formatWkLibsNdBBox } from "libs/format_utils";
 import Persistence from "libs/persistence";
 import { useInterval } from "libs/react_helpers";
+import Toast from "libs/toast";
 import * as Utils from "libs/utils";
 import _ from "lodash";
 import { getReadableURLPart } from "oxalis/model/accessors/dataset_accessor";
+import type { OxalisState } from "oxalis/store";
 import type * as React from "react";
 import { useEffect, useState } from "react";
+import { useSelector } from "react-redux";
 import { Link } from "react-router-dom";
 import { type APIJob, APIJobType, type APIUserBase } from "types/api_flow_types";
 
@@ -133,6 +137,7 @@ function JobListView() {
   const [isLoading, setIsLoading] = useState(true);
   const [jobs, setJobs] = useState<APIJob[]>([]);
   const [searchQuery, setSearchQuery] = useState("");
+  const isCurrentUserSuperUser = useSelector((state: OxalisState) => state.activeUser?.isSuperUser);
 
   useEffect(() => {
     fetchData();
@@ -293,11 +298,32 @@ function JobListView() {
               cancelJob(job.id).then(() => fetchData());
             }
           }}
-          icon={<CloseCircleOutlined key="cancel" className="icon-margin-right" />}
+          icon={<CloseCircleOutlined className="icon-margin-right" />}
         >
           Cancel
         </AsyncLink>
       );
+    } else if (job.state === "FAILURE" && isCurrentUserSuperUser) {
+      return (
+        <Tooltip title="Restarts the workflow from the failed task, skipping and reusing artifacts from preceding tasks that were already successful.">
+          <AsyncLink
+            href="#"
+            onClick={async () => {
+              try {
+                await retryJob(job.id);
+                await fetchData();
+                Toast.success("Job is being retried");
+              } catch (e) {
+                console.error("Could not retry job", e);
+                Toast.error("Failed to start retrying the job");
+              }
+            }}
+            icon={<PlayCircleOutlined className="icon-margin-right" />}
+          >
+            Retry
+          </AsyncLink>
+        </Tooltip>
+      );
     } else if (
       job.type === APIJobType.CONVERT_TO_WKW ||
       job.type === APIJobType.COMPUTE_SEGMENT_INDEX_FILE ||

tools/postgres/schema.sql

@@ -20,7 +20,7 @@ CREATE TABLE webknossos.releaseInformation (
   schemaVersion BIGINT NOT NULL
 );
 
-INSERT INTO webknossos.releaseInformation(schemaVersion) values(126);
+INSERT INTO webknossos.releaseInformation(schemaVersion) values(127);
 COMMIT TRANSACTION;
 
 
@@ -475,6 +475,7 @@ CREATE TABLE webknossos.jobs(
   _voxelytics_workflowHash VARCHAR(512),
   latestRunId VARCHAR(1024),
   returnValue Text,
+  retriedBySuperUser BOOLEAN NOT NULL DEFAULT false,
   started TIMESTAMPTZ,
   ended TIMESTAMPTZ,
   created TIMESTAMPTZ NOT NULL DEFAULT NOW(),