From 58de83c496447e70ab9362f2e53240be070af167 Mon Sep 17 00:00:00 2001 From: ausias-armesto Date: Thu, 9 Jul 2026 14:03:44 +0200 Subject: [PATCH] fix(chore): slow bulk alert status changes and prevent duplicate submissions (#6622) Co-authored-by: Claude Sonnet 5 --- .../ui/alert-change-status-modal.tsx | 39 ++++++++++++-- keep/api/routes/alerts.py | 24 +++++---- keep/workflowmanager/workflowmanager.py | 53 ++++++++++++------- tests/deduplication/test_deduplications.py | 2 +- tests/test_workflowstore.py | 17 ++---- 5 files changed, 91 insertions(+), 44 deletions(-) diff --git a/keep-ui/features/alerts/alert-change-status/ui/alert-change-status-modal.tsx b/keep-ui/features/alerts/alert-change-status/ui/alert-change-status-modal.tsx index d1ab516a33..067a720a37 100644 --- a/keep-ui/features/alerts/alert-change-status/ui/alert-change-status-modal.tsx +++ b/keep-ui/features/alerts/alert-change-status/ui/alert-change-status-modal.tsx @@ -42,6 +42,7 @@ export function AlertChangeStatusModal({ const api = useApi(); const [disposeOnNewAlert, setDisposeOnNewAlert] = useState(true); const [selectedStatus, setSelectedStatus] = useState(null); + const [isSubmitting, setIsSubmitting] = useState(false); const revalidateMultiple = useRevalidateMultiple(); const { alertsMutator } = useAlerts(); const presetsMutator = () => revalidateMultiple(["/preset"]); @@ -82,6 +83,7 @@ export function AlertChangeStatusModal({ showErrorToast(new Error("Batch status change should use batch handler.")); return; } + setIsSubmitting(true); try { await api.post( `/alerts/enrich?dispose_on_new_alert=${disposeOnNewAlert}`, @@ -106,14 +108,21 @@ export function AlertChangeStatusModal({ await presetsMutator(); } catch (error) { showErrorToast(error, "Failed to change alert status."); + } finally { + setIsSubmitting(false); } }; const handleChangeStatusBatch = async () => { + if (!selectedStatus) { + showErrorToast(new Error("Please select a new status.")); + return; + } let fingerprints = new Set(); if (Array.isArray(alert)) { alert.forEach((a) => fingerprints.add(a.fingerprint)); } + setIsSubmitting(true); try { await api.post( `/alerts/batch_enrich?dispose_on_new_alert=${disposeOnNewAlert}`, @@ -138,6 +147,8 @@ export function AlertChangeStatusModal({ await presetsMutator(); } catch (error) { showErrorToast(error, "Failed to change alert(s) status."); + } finally { + setIsSubmitting(false); } }; @@ -190,10 +201,20 @@ export function AlertChangeStatusModal({
- -
@@ -248,10 +269,20 @@ export function AlertChangeStatusModal({
- -
diff --git a/keep/api/routes/alerts.py b/keep/api/routes/alerts.py index 890a86b621..71149530c3 100644 --- a/keep/api/routes/alerts.py +++ b/keep/api/routes/alerts.py @@ -1015,15 +1015,21 @@ def batch_enrich_alerts( # @tb add "and session" cuz I saw AttributeError: 'NoneType' object has no attribute 'add'" if should_check_incidents_resolution and session: enrich_alerts_with_incidents(tenant_id=tenant_id, alerts=alerts) - for alert in alerts: - for incident in alert._incidents: - if ( - incident.resolve_on == ResolveOn.ALL.value - and is_all_alerts_resolved(incident=incident, session=session) - ): - incident.status = IncidentStatus.RESOLVED.value - session.add(incident) - session.commit() + # the same incident may be linked to many of the enriched alerts, + # so check each incident only once and commit once at the end + unique_incidents = { + incident.id: incident + for alert in alerts + for incident in alert._incidents + } + for incident in unique_incidents.values(): + if ( + incident.resolve_on == ResolveOn.ALL.value + and is_all_alerts_resolved(incident=incident, session=session) + ): + incident.status = IncidentStatus.RESOLVED.value + session.add(incident) + session.commit() return {"status": "ok"} except HTTPException: diff --git a/keep/workflowmanager/workflowmanager.py b/keep/workflowmanager/workflowmanager.py index 627b033d92..57707cdbe3 100644 --- a/keep/workflowmanager/workflowmanager.py +++ b/keep/workflowmanager/workflowmanager.py @@ -285,25 +285,34 @@ def _convert_filters_to_cel(self, filters: list[dict[str, str]]): raise def insert_events(self, tenant_id, events: typing.List[AlertDto | IncidentDto]): - for event in events: - self.logger.info("Getting all workflows", extra={"tenant_id": tenant_id}) - all_workflow_models = self.workflow_store.get_all_workflows( - tenant_id, exclude_disabled=True - ) - self.logger.info( - "Got all workflows", - extra={ - "num_of_workflows": len(all_workflow_models), - "tenant_id": tenant_id, - }, - ) - for workflow_model in all_workflow_models: - workflow = self._get_workflow_from_store(tenant_id, workflow_model) + if not events: + return - if workflow is None: - # Exception is thrown in _get_workflow_from_store, we don't need to log it here, just continue. - continue + self.logger.info("Getting all workflows", extra={"tenant_id": tenant_id}) + all_workflow_models = self.workflow_store.get_all_workflows( + tenant_id, exclude_disabled=True + ) + self.logger.info( + "Got all workflows", + extra={ + "num_of_workflows": len(all_workflow_models), + "tenant_id": tenant_id, + }, + ) + # Parse each workflow once and reuse it to evaluate triggers for every + # event, instead of re-reading and re-parsing all workflows per event. + parsed_workflows = [] + for workflow_model in all_workflow_models: + workflow = self._get_workflow_from_store(tenant_id, workflow_model) + + if workflow is None: + # Exception is thrown in _get_workflow_from_store, we don't need to log it here, just continue. + continue + + parsed_workflows.append((workflow_model, workflow)) + for event in events: + for workflow_model, workflow in parsed_workflows: for trigger in workflow.workflow_triggers: # If the trigger is not an alert, it's not relevant for this event. if not trigger.get("type") == "alert": @@ -566,10 +575,18 @@ async def enqueue_workflow(): }, ) """ + # Running a workflow mutates its context manager, so every + # queued run gets its own parsed instance instead of the + # shared one used for trigger evaluation. + workflow_instance = self._get_workflow_from_store( + tenant_id, workflow_model + ) + if workflow_instance is None: + continue with self.scheduler.lock: self.scheduler.workflows_to_run.append( { - "workflow": workflow, + "workflow": workflow_instance, "workflow_id": workflow_model.id, "tenant_id": tenant_id, "triggered_by": "alert", diff --git a/tests/deduplication/test_deduplications.py b/tests/deduplication/test_deduplications.py index 2d7fb95f37..508b8b543c 100644 --- a/tests/deduplication/test_deduplications.py +++ b/tests/deduplication/test_deduplications.py @@ -126,7 +126,7 @@ def test_deduplication_sanity(db_session, client, test_app): assert dedup_rule.get("default") -@pytest.mark.timeout(10) +@pytest.mark.timeout(20) @pytest.mark.parametrize( "test_app", [ diff --git a/tests/test_workflowstore.py b/tests/test_workflowstore.py index 3e8ca09be7..8d51278ee3 100644 --- a/tests/test_workflowstore.py +++ b/tests/test_workflowstore.py @@ -550,18 +550,11 @@ def test_get_workflow_run_logs_sorted_by_timestamp(db_session): db_session.commit() db_session.flush() - # Create logs with timestamps in random order - timestamps = [ - datetime.now(tz=timezone.utc) - timedelta(seconds=10), - datetime.now(tz=timezone.utc) + timedelta(seconds=5), - datetime.now(tz=timezone.utc) - timedelta(seconds=3), - datetime.now(tz=timezone.utc) + timedelta(seconds=2), - datetime.now(tz=timezone.utc) - timedelta(seconds=1), - datetime.now(tz=timezone.utc), - datetime.now(tz=timezone.utc) - timedelta(seconds=1), - datetime.now(tz=timezone.utc) - timedelta(seconds=2), - datetime.now(tz=timezone.utc) + timedelta(seconds=3), - ] + # Create logs with timestamps in random order. Anchor on a single "now" and + # use distinct offsets so timestamps can't tie under coarse clock resolution. + now = datetime.now(tz=timezone.utc) + offsets = [-10, 5, -3, 2, -1, 0, 1, -2, 3] + timestamps = [now + timedelta(seconds=offset) for offset in offsets] for i, ts in enumerate(timestamps): workflow_execution_log = WorkflowExecutionLog(