[Alerting V2] Reorganize SO updates and task scheduling in rules_client. (elastic#281214)

Closes elastic/rna-program#779

## Summary

Fixes the ordering of SO update operations and task manager scheduling
in the rules client:
- Delete
    - `bulkDelete(soDescriptors)` first
    - `taskManager.bulkRemove(taskIds)`
- Enable
    - `taskManager.bulkSchedule(...)` first
    - `bulkUpdate(itemsToUpdate)`
- Disable
- The ordering remains the same, but we do not fail silently anymore.

The errors are now logged with `code: 'TASK_MANAGER_DRIFT'`.

Author: adcoelho

x-pack/platform/plugins/shared/alerting_v2/server/lib/errors/error_codes.ts

@@ -51,6 +51,13 @@ export const ALERTING_V2_ERROR_CODES = {
   SCHEDULE_INTERVAL_TOO_SHORT: 'SCHEDULE_INTERVAL_TOO_SHORT',
   /** Scheduling the rule would exceed the configured maximum rule runs per minute. */
   MAX_SCHEDULES_PER_MINUTE_EXCEEDED: 'MAX_SCHEDULES_PER_MINUTE_EXCEEDED',
+  /**
+   * A bulk operation persisted the rule saved object, but the paired Task
+   * Manager call failed, leaving the rule's task state diverged from its saved object.
+   * The saved-object change already committed; this entry flags the drift so the client
+   * can detect and (optionally) retry.
+   */
+  TASK_MANAGER_DRIFT: 'TASK_MANAGER_DRIFT',
   /** A manual "run now" was requested for a disabled rule (it has no executor task to run). */
   RULE_DISABLED: 'RULE_DISABLED',
   /** A manual "run now" was requested for a rule whose executor task is already running. */

x-pack/platform/plugins/shared/alerting_v2/server/lib/rules_client/rules_client.test.ts

@@ -87,6 +87,8 @@ describe('RulesClient', () => {
 
     ensureRuleExecutorTaskScheduledMock.mockResolvedValue({ id: 'task-123' });
     getRuleExecutorTaskIdMock.mockReturnValue('task:fallback');
+
+    taskManager.bulkRemove.mockResolvedValue({ statuses: [] });
   });
 
   afterAll(() => {
@@ -1367,7 +1369,44 @@ describe('RulesClient', () => {
       ]);
     });
 
-    it('continues with deletion even if task removal fails', async () => {
+    it('deletes the saved objects before removing the tasks', async () => {
+      const client = createClient();
+
+      const callOrder: string[] = [];
+      rulesSavedObjectService.bulkDelete.mockImplementationOnce(async () => {
+        callOrder.push('bulkDelete');
+        return [{ id: 'rule-1', success: true }];
+      });
+      taskManager.bulkRemove.mockImplementationOnce(async () => {
+        callOrder.push('bulkRemove');
+        return { statuses: [] };
+      });
+
+      await client.bulkDeleteRules({ ids: ['rule-1'] });
+
+      expect(callOrder).toEqual(['bulkDelete', 'bulkRemove']);
+    });
+
+    it('only removes tasks for rules whose saved object was deleted', async () => {
+      const client = createClient();
+
+      getRuleExecutorTaskIdMock.mockReturnValueOnce('task:rule-1');
+
+      rulesSavedObjectService.bulkDelete.mockResolvedValueOnce([
+        { id: 'rule-1', success: true },
+        {
+          id: 'rule-2',
+          success: false,
+          error: { error: 'Not Found', message: 'Rule not found', statusCode: 404 },
+        },
+      ]);
+
+      await client.bulkDeleteRules({ ids: ['rule-1', 'rule-2'] });
+
+      expect(taskManager.bulkRemove).toHaveBeenCalledWith(['task:rule-1']);
+    });
+
+    it('surfaces TASK_MANAGER_DRIFT errors when task removal fails', async () => {
       const client = createClient();
 
       getRuleExecutorTaskIdMock
@@ -1383,7 +1422,70 @@ describe('RulesClient', () => {
 
       const res = await client.bulkDeleteRules({ ids: ['rule-1', 'rule-2'] });
 
-      expect(res).toEqual({ affected_count: 2, errors: [] });
+      // The saved objects are gone (affected), but the orphan tasks are flagged.
+      expect(res.affected_count).toBe(2);
+      expect(res.errors).toEqual([
+        { id: 'rule-1', error: expect.objectContaining({ code: 'TASK_MANAGER_DRIFT' }) },
+        { id: 'rule-2', error: expect.objectContaining({ code: 'TASK_MANAGER_DRIFT' }) },
+      ]);
+      expect(mockLogger.error).toHaveBeenCalledTimes(1);
+    });
+
+    it('surfaces TASK_MANAGER_DRIFT for per-task failures in the bulkRemove statuses (no throw)', async () => {
+      const client = createClient();
+
+      getRuleExecutorTaskIdMock
+        .mockReturnValueOnce('task:rule-1')
+        .mockReturnValueOnce('task:rule-2');
+
+      rulesSavedObjectService.bulkDelete.mockResolvedValueOnce([
+        { id: 'rule-1', success: true },
+        { id: 'rule-2', success: true },
+      ]);
+
+      taskManager.bulkRemove.mockResolvedValueOnce({
+        statuses: [
+          {
+            id: 'task:rule-1',
+            type: 'task',
+            success: false,
+            error: { error: 'Internal', message: 'boom', statusCode: 500 },
+          },
+          { id: 'task:rule-2', type: 'task', success: true },
+        ],
+      });
+
+      const res = await client.bulkDeleteRules({ ids: ['rule-1', 'rule-2'] });
+
+      expect(res.affected_count).toBe(2);
+      expect(res.errors).toEqual([
+        { id: 'rule-1', error: expect.objectContaining({ code: 'TASK_MANAGER_DRIFT' }) },
+      ]);
+      expect(mockLogger.error).toHaveBeenCalledTimes(1);
+    });
+
+    it('ignores missing tasks (404) in the bulkRemove statuses — the task is already gone', async () => {
+      const client = createClient();
+
+      getRuleExecutorTaskIdMock.mockReturnValueOnce('task:rule-1');
+
+      rulesSavedObjectService.bulkDelete.mockResolvedValueOnce([{ id: 'rule-1', success: true }]);
+
+      taskManager.bulkRemove.mockResolvedValueOnce({
+        statuses: [
+          {
+            id: 'task:rule-1',
+            type: 'task',
+            success: false,
+            error: { error: 'Not Found', message: 'not found', statusCode: 404 },
+          },
+        ],
+      });
+
+      const res = await client.bulkDeleteRules({ ids: ['rule-1'] });
+
+      expect(res).toEqual({ affected_count: 1, errors: [] });
+      expect(mockLogger.error).not.toHaveBeenCalled();
     });
 
     it('returns a zero-affected empty response when ids is an empty array', async () => {
@@ -1439,7 +1541,7 @@ describe('RulesClient', () => {
       expect(res).toEqual({ affected_count: 1, errors: [] });
     });
 
-    it('logs a warning when task scheduling fails but still counts the rule as affected', async () => {
+    it('schedules the tasks before persisting enabled=true', async () => {
       const client = createClient();
 
       const disabledAttrs = createRuleSoAttributes({
@@ -1451,17 +1553,107 @@ describe('RulesClient', () => {
         { id: 'rule-1', attributes: disabledAttrs, version: 'v1' },
       ]);
 
-      rulesSavedObjectService.bulkUpdate.mockResolvedValueOnce([{ id: 'rule-1', success: true }]);
+      const callOrder: string[] = [];
+      taskManager.bulkSchedule.mockImplementationOnce(async () => {
+        callOrder.push('bulkSchedule');
+        return [];
+      });
+      rulesSavedObjectService.bulkUpdate.mockImplementationOnce(async () => {
+        callOrder.push('bulkUpdate');
+        return [{ id: 'rule-1', success: true }];
+      });
+
+      await client.bulkEnableRules({ ids: ['rule-1'] });
+
+      expect(callOrder).toEqual(['bulkSchedule', 'bulkUpdate']);
+    });
+
+    it('leaves rules disabled and surfaces TASK_MANAGER_DRIFT when scheduling fails', async () => {
+      const client = createClient();
+
+      const disabledAttrs = createRuleSoAttributes({
+        metadata: { name: 'disabled-rule' },
+        enabled: false,
+      });
+
+      rulesSavedObjectService.bulkGetByIds.mockResolvedValueOnce([
+        { id: 'rule-1', attributes: disabledAttrs, version: 'v1' },
+      ]);
 
+      getRuleExecutorTaskIdMock.mockReturnValue('task:rule-1');
       taskManager.bulkSchedule.mockRejectedValueOnce(new Error('Failed to grant UIAM API key'));
 
       const res = await client.bulkEnableRules({ ids: ['rule-1'] });
 
-      expect(mockLogger.warn).toHaveBeenCalledTimes(1);
-      expect(mockLogger.warn).toHaveBeenCalledWith(
-        expect.stringContaining('Failed to grant UIAM API key')
+      // Scheduling failed first, so the saved object is never flipped to enabled.
+      expect(rulesSavedObjectService.bulkUpdate).not.toHaveBeenCalled();
+      expect(taskManager.bulkRemove).toHaveBeenCalledWith(['task:rule-1']);
+      expect(res.affected_count).toBe(0);
+      expect(res.errors).toEqual([
+        { id: 'rule-1', error: expect.objectContaining({ code: 'TASK_MANAGER_DRIFT' }) },
+      ]);
+      expect(mockLogger.error).toHaveBeenCalledTimes(1);
+      expect(ruleEventPublisher.emitRuleEnabled).not.toHaveBeenCalled();
+    });
+
+    it('rolls back partially scheduled tasks when bulkSchedule throws', async () => {
+      const client = createClient();
+
+      const disabledAttrs = createRuleSoAttributes({
+        metadata: { name: 'disabled-rule' },
+        enabled: false,
+      });
+
+      rulesSavedObjectService.bulkGetByIds.mockResolvedValueOnce([
+        { id: 'rule-1', attributes: disabledAttrs, version: 'v1' },
+        { id: 'rule-2', attributes: disabledAttrs, version: 'v1' },
+      ]);
+
+      getRuleExecutorTaskIdMock.mockImplementation(
+        ({ ruleId }: { ruleId: string }) => `task:${ruleId}`
       );
-      expect(res).toEqual({ affected_count: 1, errors: [] });
+      taskManager.bulkSchedule.mockRejectedValueOnce(new Error('partial schedule failure'));
+
+      const res = await client.bulkEnableRules({ ids: ['rule-1', 'rule-2'] });
+
+      expect(rulesSavedObjectService.bulkUpdate).not.toHaveBeenCalled();
+      expect(taskManager.bulkRemove).toHaveBeenCalledWith(['task:rule-1', 'task:rule-2']);
+      expect(res.affected_count).toBe(0);
+      expect(res.errors).toEqual([
+        { id: 'rule-1', error: expect.objectContaining({ code: 'TASK_MANAGER_DRIFT' }) },
+        { id: 'rule-2', error: expect.objectContaining({ code: 'TASK_MANAGER_DRIFT' }) },
+      ]);
+    });
+
+    it('cancels the just-scheduled task when the saved object update fails', async () => {
+      const client = createClient();
+
+      const disabledAttrs = createRuleSoAttributes({
+        metadata: { name: 'disabled-rule' },
+        enabled: false,
+      });
+
+      rulesSavedObjectService.bulkGetByIds.mockResolvedValueOnce([
+        { id: 'rule-1', attributes: disabledAttrs, version: 'v1' },
+      ]);
+
+      getRuleExecutorTaskIdMock.mockReturnValue('task:rule-1');
+
+      rulesSavedObjectService.bulkUpdate.mockResolvedValueOnce([
+        {
+          id: 'rule-1',
+          success: false,
+          error: { statusCode: 409, error: 'Conflict', message: 'Version conflict' },
+        },
+      ]);
+
+      const res = await client.bulkEnableRules({ ids: ['rule-1'] });
+
+      expect(taskManager.bulkRemove).toHaveBeenCalledWith(['task:rule-1']);
+      expect(res.affected_count).toBe(0);
+      expect(res.errors).toEqual([
+        { id: 'rule-1', error: { code: 'RULE_VERSION_CONFLICT', message: 'Version conflict' } },
+      ]);
     });
 
     it('counts already-enabled rules as affected without updating them (idempotent)', async () => {
@@ -1620,7 +1812,7 @@ describe('RulesClient', () => {
       ]);
     });
 
-    it('logs a warning and still counts the rule as affected when task removal fails', async () => {
+    it('surfaces TASK_MANAGER_DRIFT errors when the task removal fails', async () => {
       const client = createClient();
 
       const enabledAttrs = createRuleSoAttributes({
@@ -1639,9 +1831,12 @@ describe('RulesClient', () => {
 
       const res = await client.bulkDisableRules({ ids: ['rule-1'] });
 
-      expect(mockLogger.warn).toHaveBeenCalledTimes(1);
-      expect(mockLogger.warn).toHaveBeenCalledWith(expect.stringContaining('task removal failed'));
-      expect(res).toEqual({ affected_count: 1, errors: [] });
+      // The saved object is disabled (affected), but the task drift is flagged.
+      expect(res.affected_count).toBe(1);
+      expect(res.errors).toEqual([
+        { id: 'rule-1', error: expect.objectContaining({ code: 'TASK_MANAGER_DRIFT' }) },
+      ]);
+      expect(mockLogger.error).toHaveBeenCalledTimes(1);
     });
 
     it('returns a zero-affected empty response when ids is an empty array', async () => {