[Response Ops][Task Manager] Allow schedule updates when ensureScheduled API is used (#238026)

Resolves #222089

## Summary

Updates `ensureScheduled` API to update task schedule if task already
exists using the existing `bulkUpdateSchedules` method. Currently
`bulkUpdateSchedules` only handle schedule intervals (not rrule
schedules) so this only works for intervals, which I believe is ok
because rrule is not widely used yet.

---------

Co-authored-by: Patrick Mueller <[email protected]>

Author: ymao1

x-pack/platform/plugins/shared/task_manager/README.md

@@ -462,6 +462,8 @@ The danger is that in such a situation, a Task with that same `id` might already
 
 To achieve this you should use the `ensureScheduled` api which has the exact same behavior as `schedule`, except it allows the scheduling of a Task with an `id` that's already in assigned to another Task and it will assume that the existing Task is the one you wished to `schedule`, treating this as a successful operation.
 
+The only exception to this is if you use `ensureScheduled` to schedule a task with a recurring schedule interval. In this case, if a task with the same `id` already exists, the API will attempt to update the schedule of the existing task if it has changed.
+
 #### runSoon
 
 Use `runSoon` to instruct TaskManager to run an existing task as soon as possible by updating the next scheduled run date to be `now`.

x-pack/platform/plugins/shared/task_manager/server/task_scheduling.test.ts

@@ -10,6 +10,7 @@ import moment from 'moment';
 
 import { TaskScheduling } from './task_scheduling';
 import { asOk } from './lib/result_type';
+import type { ConcreteTaskInstance } from './task';
 import { TaskStatus } from './task';
 import { createInitialMiddleware } from './lib/middleware';
 import { taskStoreMock } from './task_store.mock';
@@ -35,6 +36,22 @@ jest.mock('elastic-apm-node', () => ({
   },
 }));
 
+const getTask = (overrides = {}): ConcreteTaskInstance => ({
+  id: 'my-foo-id',
+  taskType: 'foo',
+  params: {},
+  state: {},
+  schedule: { interval: '1m' },
+  scheduledAt: new Date(),
+  attempts: 0,
+  startedAt: new Date(),
+  retryAt: new Date(Date.now() + 5 * 60 * 1000),
+  ownerId: '123',
+  status: TaskStatus.Idle,
+  runAt: new Date(),
+  ...overrides,
+});
+
 describe('TaskScheduling', () => {
   beforeAll(() => {
     fakeTimer = sinon.useFakeTimers();
@@ -149,6 +166,101 @@ describe('TaskScheduling', () => {
     expect(result.id).toEqual('my-foo-id');
   });
 
+  test('tries to updates schedule for tasks that have already been scheduled', async () => {
+    const task = getTask();
+    const taskScheduling = new TaskScheduling(taskSchedulingOpts);
+    const bulkUpdateScheduleSpy = jest
+      .spyOn(taskScheduling, 'bulkUpdateSchedules')
+      .mockResolvedValue({ tasks: [task], errors: [] });
+    mockTaskStore.schedule.mockRejectedValueOnce({
+      statusCode: 409,
+    });
+
+    const result = await taskScheduling.ensureScheduled(task);
+
+    expect(bulkUpdateScheduleSpy).toHaveBeenCalledWith(['my-foo-id'], { interval: '1m' });
+
+    expect(result.id).toEqual('my-foo-id');
+  });
+
+  test('does not try to update schedule for tasks that have already been scheduled if no schedule is provided', async () => {
+    const taskScheduling = new TaskScheduling(taskSchedulingOpts);
+    const bulkUpdateScheduleSpy = jest.spyOn(taskScheduling, 'bulkUpdateSchedules');
+    mockTaskStore.schedule.mockRejectedValueOnce({
+      statusCode: 409,
+    });
+
+    const result = await taskScheduling.ensureScheduled({
+      id: 'my-foo-id',
+      taskType: 'foo',
+      params: {},
+      state: {},
+    });
+
+    expect(bulkUpdateScheduleSpy).not.toHaveBeenCalled();
+
+    expect(result.id).toEqual('my-foo-id');
+  });
+
+  test('propagates error when trying to update schedule for tasks that have already been scheduled', async () => {
+    const task = getTask();
+    const taskScheduling = new TaskScheduling(taskSchedulingOpts);
+    const bulkUpdateScheduleSpy = jest
+      .spyOn(taskScheduling, 'bulkUpdateSchedules')
+      .mockResolvedValue({
+        tasks: [],
+        errors: [
+          {
+            id: 'my-foo-id',
+            type: 'task',
+            error: {
+              error: 'error',
+              message: 'Failed to update schedule for reasons',
+              statusCode: 500,
+            },
+          },
+        ],
+      });
+    mockTaskStore.schedule.mockRejectedValueOnce({
+      statusCode: 409,
+    });
+
+    await expect(taskScheduling.ensureScheduled(task)).rejects.toMatchInlineSnapshot(
+      `[Error: Tried to update schedule for existing task "my-foo-id" but failed with error: Failed to update schedule for reasons]`
+    );
+
+    expect(bulkUpdateScheduleSpy).toHaveBeenCalledWith(['my-foo-id'], { interval: '1m' });
+  });
+
+  test('handles VERSION_CONFLICT_STATUS errors when trying to update schedule for tasks that have already been scheduled', async () => {
+    const task = getTask();
+    const taskScheduling = new TaskScheduling(taskSchedulingOpts);
+    const bulkUpdateScheduleSpy = jest
+      .spyOn(taskScheduling, 'bulkUpdateSchedules')
+      .mockResolvedValue({
+        tasks: [],
+        errors: [
+          {
+            id: 'my-foo-id',
+            type: 'task',
+            error: {
+              error: 'error',
+              message: 'Failed to update schedule due to version conflict',
+              statusCode: 409,
+            },
+          },
+        ],
+      });
+    mockTaskStore.schedule.mockRejectedValueOnce({
+      statusCode: 409,
+    });
+
+    const result = await taskScheduling.ensureScheduled(task);
+
+    expect(bulkUpdateScheduleSpy).toHaveBeenCalledWith(['my-foo-id'], { interval: '1m' });
+    expect(result.id).toEqual('my-foo-id');
+  });
+
   test('doesnt ignore failure to scheduling existing tasks for reasons other than already being scheduled', async () => {
     const taskScheduling = new TaskScheduling(taskSchedulingOpts);
     mockTaskStore.schedule.mockRejectedValueOnce({

x-pack/platform/plugins/shared/task_manager/server/task_scheduling.ts

@@ -284,12 +284,26 @@ export class TaskScheduling {
    */
   public async ensureScheduled(
     taskInstance: TaskInstanceWithId,
-    options?: Record<string, unknown>
+    options?: ScheduleOptions
   ): Promise<TaskInstanceWithId> {
     try {
       return await this.schedule(taskInstance, options);
     } catch (err) {
       if (err.statusCode === VERSION_CONFLICT_STATUS) {
+        // check if task specifies a schedule interval
+        // if so,try to update the just the schedule
+        // only works for interval schedule
+        if (taskInstance.schedule && taskInstance.schedule.interval) {
+          const result = await this.bulkUpdateSchedules([taskInstance.id], taskInstance.schedule);
+          if (
+            result.errors.length &&
+            result.errors[0].error.statusCode !== VERSION_CONFLICT_STATUS
+          ) {
+            throw new Error(
+              `Tried to update schedule for existing task "${taskInstance.id}" but failed with error: ${result.errors[0].error.message}`
+            );
+          }
+        }
         return taskInstance;
       }
       throw err;

x-pack/platform/test/plugin_api_integration/plugins/sample_task_plugin/server/init_routes.ts

@@ -315,12 +315,13 @@ export function initRoutes(
             params: schema.object({}),
             state: schema.maybe(schema.object({})),
             id: schema.maybe(schema.string()),
+            schedule: schema.maybe(schema.object({ interval: schema.string() })),
           }),
         }),
       },
     },
     async function (
-      context: RequestHandlerContext,
+      _: RequestHandlerContext,
       req: KibanaRequest<any, any, any, any>,
       res: KibanaResponseFactory
     ): Promise<IKibanaResponse<any>> {

x-pack/platform/test/plugin_api_integration/test_suites/task_manager/task_management.ts

@@ -208,7 +208,7 @@ export default function ({ getService }: FtrProviderContext) {
         .then((response: { body: BulkUpdateTaskResult }) => response.body);
     }
 
-    function scheduleTaskIfNotExists(task: Partial<ConcreteTaskInstance>) {
+    function ensureTaskScheduled(task: Partial<ConcreteTaskInstance>) {
       return supertest
         .post('/api/sample_tasks/ensure_scheduled')
         .set('kbn-xsrf', 'xxx')
@@ -426,15 +426,15 @@ export default function ({ getService }: FtrProviderContext) {
     });
 
     it('should allow a task with a given ID to be scheduled multiple times', async () => {
-      const result = await scheduleTaskIfNotExists({
+      const result = await ensureTaskScheduled({
         id: 'test-task-to-reschedule-in-task-manager',
         taskType: 'sampleTask',
         params: {},
       });
 
       expect(result.id).to.be('test-task-to-reschedule-in-task-manager');
 
-      const rescheduleResult = await scheduleTaskIfNotExists({
+      const rescheduleResult = await ensureTaskScheduled({
         id: 'test-task-to-reschedule-in-task-manager',
         taskType: 'sampleTask',
         params: {},
@@ -921,6 +921,36 @@ export default function ({ getService }: FtrProviderContext) {
       });
     });
 
+    it('should update schedule for existing task when calling ensureScheduled with a different schedule', async () => {
+      // schedule the task
+      const taskId = 'sample-recurring-task-id';
+      await scheduleTask({
+        id: taskId,
+        taskType: 'sampleRecurringTask',
+        schedule: { interval: '1d' },
+        params: {},
+      });
+
+      await retry.try(async () => {
+        expect((await historyDocs()).length).to.eql(1);
+        const task = await currentTask(taskId);
+        expect(task.schedule?.interval).to.eql('1d');
+      });
+
+      // call ensureScheduled with a different schedule
+      await ensureTaskScheduled({
+        id: taskId,
+        taskType: 'sampleRecurringTask',
+        params: {},
+        schedule: { interval: '5m' },
+      });
+
+      await retry.try(async () => {
+        const task = await currentTask(taskId);
+        expect(task.schedule?.interval).to.eql('5m');
+      });
+    });
+
     function expectReschedule(
       originalRunAt: number,
       task: SerializedConcreteTaskInstance<any, any>,