Add run retry task, run scheduled task functionality

ui/src/actions/tasksActions.ts

@@ -29,6 +29,8 @@ import {
   runAllRetryTasks,
   runAllScheduledTasks,
   runDeadTask,
+  runRetryTask,
+  runScheduledTask,
 } from "../api";
 import { Dispatch } from "redux";
 
@@ -51,6 +53,12 @@ export const LIST_DEAD_TASKS_ERROR = "LIST_DEAD_TASKS_ERROR";
 export const CANCEL_ACTIVE_TASK_BEGIN = "CANCEL_ACTIVE_TASK_BEGIN";
 export const CANCEL_ACTIVE_TASK_SUCCESS = "CANCEL_ACTIVE_TASK_SUCCESS";
 export const CANCEL_ACTIVE_TASK_ERROR = "CANCEL_ACTIVE_TASK_ERROR";
+export const RUN_SCHEDULED_TASK_BEGIN = "RUN_DEAD_TASK_BEGIN";
+export const RUN_SCHEDULED_TASK_SUCCESS = "RUN_DEAD_TASK_SUCCESS";
+export const RUN_SCHEDULED_TASK_ERROR = "RUN_DEAD_TASK_ERROR";
+export const RUN_RETRY_TASK_BEGIN = "RUN_RETRY_TASK_BEGIN";
+export const RUN_RETRY_TASK_SUCCESS = "RUN_RETRY_TASK_SUCCESS";
+export const RUN_RETRY_TASK_ERROR = "RUN_RETRY_TASK_ERROR";
 export const RUN_DEAD_TASK_BEGIN = "RUN_DEAD_TASK_BEGIN";
 export const RUN_DEAD_TASK_SUCCESS = "RUN_DEAD_TASK_SUCCESS";
 export const RUN_DEAD_TASK_ERROR = "RUN_DEAD_TASK_ERROR";
@@ -215,6 +223,43 @@ interface CancelActiveTaskErrorAction {
   taskId: string;
   error: string;
 }
+interface RunScheduledTaskBeginAction {
+  type: typeof RUN_SCHEDULED_TASK_BEGIN;
+  queue: string;
+  taskKey: string;
+}
+
+interface RunScheduledTaskSuccessAction {
+  type: typeof RUN_SCHEDULED_TASK_SUCCESS;
+  queue: string;
+  taskKey: string;
+}
+
+interface RunScheduledTaskErrorAction {
+  type: typeof RUN_SCHEDULED_TASK_ERROR;
+  queue: string;
+  taskKey: string;
+  error: string;
+}
+
+interface RunRetryTaskBeginAction {
+  type: typeof RUN_RETRY_TASK_BEGIN;
+  queue: string;
+  taskKey: string;
+}
+
+interface RunRetryTaskSuccessAction {
+  type: typeof RUN_RETRY_TASK_SUCCESS;
+  queue: string;
+  taskKey: string;
+}
+
+interface RunRetryTaskErrorAction {
+  type: typeof RUN_RETRY_TASK_ERROR;
+  queue: string;
+  taskKey: string;
+  error: string;
+}
 
 interface RunDeadTaskBeginAction {
   type: typeof RUN_DEAD_TASK_BEGIN;
@@ -522,6 +567,12 @@ export type TasksActionTypes =
   | CancelActiveTaskBeginAction
   | CancelActiveTaskSuccessAction
   | CancelActiveTaskErrorAction
+  | RunScheduledTaskBeginAction
+  | RunScheduledTaskSuccessAction
+  | RunScheduledTaskErrorAction
+  | RunRetryTaskBeginAction
+  | RunRetryTaskSuccessAction
+  | RunRetryTaskErrorAction
   | RunDeadTaskBeginAction
   | RunDeadTaskSuccessAction
   | RunDeadTaskErrorAction
@@ -703,6 +754,42 @@ export function cancelActiveTaskAsync(queue: string, taskId: string) {
   };
 }
 
+export function runScheduledTaskAsync(queue: string, taskKey: string) {
+  return async (dispatch: Dispatch<TasksActionTypes>) => {
+    dispatch({ type: RUN_SCHEDULED_TASK_BEGIN, queue, taskKey });
+    try {
+      await runScheduledTask(queue, taskKey);
+      dispatch({ type: RUN_SCHEDULED_TASK_SUCCESS, queue, taskKey });
+    } catch (error) {
+      console.error("runScheduledTaskAsync: ", error);
+      dispatch({
+        type: RUN_SCHEDULED_TASK_ERROR,
+        error: `Could not run task: ${taskKey}`,
+        queue,
+        taskKey,
+      });
+    }
+  };
+}
+
+export function runRetryTaskAsync(queue: string, taskKey: string) {
+  return async (dispatch: Dispatch<TasksActionTypes>) => {
+    dispatch({ type: RUN_RETRY_TASK_BEGIN, queue, taskKey });
+    try {
+      await runRetryTask(queue, taskKey);
+      dispatch({ type: RUN_RETRY_TASK_SUCCESS, queue, taskKey });
+    } catch (error) {
+      console.error("runRetryTaskAsync: ", error);
+      dispatch({
+        type: RUN_RETRY_TASK_ERROR,
+        error: `Could not run task: ${taskKey}`,
+        queue,
+        taskKey,
+      });
+    }
+  };
+}
+
 export function runDeadTaskAsync(queue: string, taskKey: string) {
   return async (dispatch: Dispatch<TasksActionTypes>) => {
     dispatch({ type: RUN_DEAD_TASK_BEGIN, queue, taskKey });

ui/src/api.ts

@@ -253,6 +253,16 @@ export async function listDeadTasks(
   return resp.data;
 }
 
+export async function runScheduledTask(
+  qname: string,
+  taskKey: string
+): Promise<void> {
+  await axios({
+    method: "post",
+    url: `${BASE_URL}/queues/${qname}/scheduled_tasks/${taskKey}:run`,
+  });
+}
+
 export async function deleteScheduledTask(
   qname: string,
   taskKey: string
@@ -305,6 +315,16 @@ export async function runAllScheduledTasks(qname: string): Promise<void> {
   });
 }
 
+export async function runRetryTask(
+  qname: string,
+  taskKey: string
+): Promise<void> {
+  await axios({
+    method: "post",
+    url: `${BASE_URL}/queues/${qname}/retry_tasks/${taskKey}:run`,
+  });
+}
+
 export async function deleteRetryTask(
   qname: string,
   taskKey: string

ui/src/components/RetryTasksTable.tsx

@@ -29,6 +29,7 @@ import {
   runAllRetryTasksAsync,
   listRetryTasksAsync,
   deleteRetryTaskAsync,
+  runRetryTaskAsync,
 } from "../actions/tasksActions";
 import { AppState } from "../store";
 import TablePaginationActions, {
@@ -63,6 +64,7 @@ const mapDispatchToProps = {
   runAllRetryTasksAsync,
   listRetryTasksAsync,
   deleteRetryTaskAsync,
+  runRetryTaskAsync,
 };
 
 const connector = connect(mapStateToProps, mapDispatchToProps);
@@ -193,6 +195,7 @@ function RetryTasksTable(props: Props & ReduxProps) {
               <Row
                 key={task.id}
                 task={task}
+                allActionPending={props.allActionPending}
                 isSelected={selectedKeys.includes(task.key)}
                 onSelectChange={(checked: boolean) => {
                   if (checked) {
@@ -203,6 +206,9 @@ function RetryTasksTable(props: Props & ReduxProps) {
                     );
                   }
                 }}
+                onRunClick={() => {
+                  props.runRetryTaskAsync(task.queue, task.key);
+                }}
                 onDeleteClick={() => {
                   props.deleteRetryTaskAsync(task.queue, task.key);
                 }}
@@ -246,6 +252,8 @@ interface RowProps {
   isSelected: boolean;
   onSelectChange: (checked: boolean) => void;
   onDeleteClick: () => void;
+  onRunClick: () => void;
+  allActionPending: boolean;
 }
 
 function Row(props: RowProps) {
@@ -281,7 +289,16 @@ function Row(props: RowProps) {
         <TableCell>{task.retried}</TableCell>
         <TableCell>{task.max_retry}</TableCell>
         <TableCell>
-          <Button disabled={task.requestPending} onClick={props.onDeleteClick}>
+          <Button
+            onClick={props.onRunClick}
+            disabled={task.requestPending || props.allActionPending}
+          >
+            Run
+          </Button>
+          <Button
+            disabled={task.requestPending}
+            onClick={props.onDeleteClick || props.allActionPending}
+          >
             Delete
           </Button>
         </TableCell>

ui/src/components/ScheduledTasksTable.tsx

@@ -29,6 +29,7 @@ import {
   runAllScheduledTasksAsync,
   listScheduledTasksAsync,
   deleteScheduledTaskAsync,
+  runScheduledTaskAsync,
 } from "../actions/tasksActions";
 import { AppState } from "../store";
 import TablePaginationActions, {
@@ -63,6 +64,7 @@ const mapDispatchToProps = {
   batchRunScheduledTasksAsync,
   deleteAllScheduledTasksAsync,
   runAllScheduledTasksAsync,
+  runScheduledTaskAsync,
 };
 
 const connector = connect(mapStateToProps, mapDispatchToProps);
@@ -190,6 +192,7 @@ function ScheduledTasksTable(props: Props & ReduxProps) {
               <Row
                 key={task.id}
                 task={task}
+                allActionPending={props.allActionPending}
                 isSelected={selectedKeys.includes(task.key)}
                 onSelectChange={(checked: boolean) => {
                   if (checked) {
@@ -200,6 +203,9 @@ function ScheduledTasksTable(props: Props & ReduxProps) {
                     );
                   }
                 }}
+                onRunClick={() => {
+                  props.runScheduledTaskAsync(queue, task.key);
+                }}
                 onDeleteClick={() => {
                   props.deleteScheduledTaskAsync(queue, task.key);
                 }}
@@ -242,7 +248,9 @@ interface RowProps {
   task: ScheduledTaskExtended;
   isSelected: boolean;
   onSelectChange: (checked: boolean) => void;
+  onRunClick: () => void;
   onDeleteClick: () => void;
+  allActionPending: boolean;
 }
 
 function Row(props: RowProps) {
@@ -275,7 +283,16 @@ function Row(props: RowProps) {
         <TableCell>{task.type}</TableCell>
         <TableCell>{durationBefore(task.next_process_at)}</TableCell>
         <TableCell>
-          <Button onClick={props.onDeleteClick} disabled={task.requestPending}>
+          <Button
+            onClick={props.onRunClick}
+            disabled={task.requestPending || props.allActionPending}
+          >
+            Run
+          </Button>
+          <Button
+            onClick={props.onDeleteClick}
+            disabled={task.requestPending || props.allActionPending}
+          >
             Delete
           </Button>
         </TableCell>

ui/src/reducers/queuesReducer.ts

@@ -35,6 +35,8 @@ import {
   RUN_ALL_RETRY_TASKS_SUCCESS,
   RUN_ALL_SCHEDULED_TASKS_SUCCESS,
   RUN_DEAD_TASK_SUCCESS,
+  RUN_RETRY_TASK_SUCCESS,
+  RUN_SCHEDULED_TASK_SUCCESS,
   TasksActionTypes,
 } from "../actions/tasksActions";
 import { DailyStat, Queue } from "../api";
@@ -163,6 +165,40 @@ function queuesReducer(
       return { ...state, data: newData };
     }
 
+    case RUN_SCHEDULED_TASK_SUCCESS: {
+      const newData = state.data.map((queueInfo) => {
+        if (queueInfo.name !== action.queue) {
+          return queueInfo;
+        }
+        return {
+          ...queueInfo,
+          currentStats: {
+            ...queueInfo.currentStats,
+            pending: queueInfo.currentStats.pending + 1,
+            scheduled: queueInfo.currentStats.scheduled - 1,
+          },
+        };
+      });
+      return { ...state, data: newData };
+    }
+
+    case RUN_RETRY_TASK_SUCCESS: {
+      const newData = state.data.map((queueInfo) => {
+        if (queueInfo.name !== action.queue) {
+          return queueInfo;
+        }
+        return {
+          ...queueInfo,
+          currentStats: {
+            ...queueInfo.currentStats,
+            pending: queueInfo.currentStats.pending + 1,
+            retry: queueInfo.currentStats.retry - 1,
+          },
+        };
+      });
+      return { ...state, data: newData };
+    }
+
     case RUN_DEAD_TASK_SUCCESS: {
       const newData = state.data.map((queueInfo) => {
         if (queueInfo.name !== action.queue) {

ui/src/reducers/snackbarReducer.ts

@@ -19,6 +19,8 @@ import {
   RUN_ALL_RETRY_TASKS_SUCCESS,
   RUN_ALL_SCHEDULED_TASKS_SUCCESS,
   RUN_DEAD_TASK_SUCCESS,
+  RUN_RETRY_TASK_SUCCESS,
+  RUN_SCHEDULED_TASK_SUCCESS,
   TasksActionTypes,
 } from "../actions/tasksActions";
 
@@ -45,6 +47,20 @@ function snackbarReducer(
         isOpen: false,
       };
 
+    case RUN_SCHEDULED_TASK_SUCCESS:
+      return {
+        isOpen: true,
+        // TODO: show only task id
+        message: `Scheduled task ${action.taskKey} is now pending`,
+      };
+
+    case RUN_RETRY_TASK_SUCCESS:
+      return {
+        isOpen: true,
+        // TODO: show only task id
+        message: `Retry task ${action.taskKey} is now pending`,
+      };
+
     case RUN_DEAD_TASK_SUCCESS:
       return {
         isOpen: true,

ui/src/reducers/tasksReducer.ts

@@ -66,6 +66,12 @@ import {
   RUN_ALL_SCHEDULED_TASKS_SUCCESS,
   DELETE_ALL_SCHEDULED_TASKS_BEGIN,
   RUN_ALL_SCHEDULED_TASKS_BEGIN,
+  RUN_RETRY_TASK_BEGIN,
+  RUN_RETRY_TASK_SUCCESS,
+  RUN_RETRY_TASK_ERROR,
+  RUN_SCHEDULED_TASK_BEGIN,
+  RUN_SCHEDULED_TASK_SUCCESS,
+  RUN_SCHEDULED_TASK_ERROR,
 } from "../actions/tasksActions";
 import {
   ActiveTask,
@@ -389,6 +395,7 @@ function tasksReducer(
         },
       };
 
+    case RUN_SCHEDULED_TASK_BEGIN:
     case DELETE_SCHEDULED_TASK_BEGIN:
       return {
         ...state,
@@ -403,6 +410,7 @@ function tasksReducer(
         },
       };
 
+    case RUN_SCHEDULED_TASK_SUCCESS:
     case DELETE_SCHEDULED_TASK_SUCCESS:
       return {
         ...state,
@@ -414,6 +422,7 @@ function tasksReducer(
         },
       };
 
+    case RUN_SCHEDULED_TASK_ERROR:
     case DELETE_SCHEDULED_TASK_ERROR:
       return {
         ...state,
@@ -525,6 +534,7 @@ function tasksReducer(
         },
       };
 
+    case RUN_RETRY_TASK_BEGIN:
     case DELETE_RETRY_TASK_BEGIN:
       return {
         ...state,
@@ -539,6 +549,7 @@ function tasksReducer(
         },
       };
 
+    case RUN_RETRY_TASK_SUCCESS:
     case DELETE_RETRY_TASK_SUCCESS:
       return {
         ...state,
@@ -550,6 +561,7 @@ function tasksReducer(
         },
       };
 
+    case RUN_RETRY_TASK_ERROR:
     case DELETE_RETRY_TASK_ERROR:
       return {
         ...state,