ILLDEV-456 Filter only finished events and not current event

Author: JanisSaldabols

broker/scheduler/db/repo.go

@@ -22,7 +22,7 @@ type SchedRepo interface {
 	GetScheduledTaskByIdForUpdate(ctx common.ExtendedContext, id string, owners []string) (ScheduledTask, error)
 	HasActiveBatchActionEvents(ctx common.ExtendedContext, taskID string) (bool, error)
 	DeleteBatchActionEvents(ctx common.ExtendedContext, taskID string) error
-	DeleteOldBatchActionRunEvents(ctx common.ExtendedContext, taskID string, retention int32) error
+	DeleteOldBatchActionRunEvents(ctx common.ExtendedContext, currentEventId string, taskID string, retention int32) error
 	DeleteScheduledTask(ctx common.ExtendedContext, id string, owners []string) error
 	GetScheduledTasks(ctx common.ExtendedContext, params GetScheduledTasksParams) ([]ScheduledTask, int64, error)
 }
@@ -128,10 +128,11 @@ func (r *PgSchedRepo) DeleteBatchActionEvents(ctx common.ExtendedContext, taskID
 	return r.eventQueries.DeleteBatchActionEvents(ctx, r.GetConnOrTx(), taskID)
 }
 
-func (r *PgSchedRepo) DeleteOldBatchActionRunEvents(ctx common.ExtendedContext, taskID string, retention int32) error {
+func (r *PgSchedRepo) DeleteOldBatchActionRunEvents(ctx common.ExtendedContext, currentEventId string, taskID string, retention int32) error {
 	return r.eventQueries.DeleteOldBatchActionRunEvents(ctx, r.GetConnOrTx(), events.DeleteOldBatchActionRunEventsParams{
-		TaskID:    taskID,
-		Retention: retention,
+		CurrentEventID: currentEventId,
+		TaskID:         taskID,
+		Retention:      retention,
 	})
 }
 

broker/scheduler/service/batch_action.go

@@ -88,7 +88,7 @@ func (s *BatchActionService) cleanupOldRuns(ctx common.ExtendedContext, event ev
 	if taskID == "" {
 		return
 	}
-	if err := s.schedRepo.DeleteOldBatchActionRunEvents(ctx, taskID, BATCH_ACTION_RUN_RETENTION); err != nil {
+	if err := s.schedRepo.DeleteOldBatchActionRunEvents(ctx, event.ID, taskID, BATCH_ACTION_RUN_RETENTION); err != nil {
 		ctx.Logger().Error("failed to cleanup old batch action runs", "taskId", taskID, "retention", BATCH_ACTION_RUN_RETENTION, "error", err)
 	}
 }

broker/scheduler/service/batch_action_test.go

@@ -42,14 +42,16 @@ type mockBatchActionEventBus struct {
 
 type mockBatchActionCleanupRepo struct {
 	sched_db.PgSchedRepo
-	called       bool
-	gotTaskID    string
-	gotRetention int32
-	err          error
+	called            bool
+	gotCurrentEventId string
+	gotTaskID         string
+	gotRetention      int32
+	err               error
 }
 
-func (m *mockBatchActionCleanupRepo) DeleteOldBatchActionRunEvents(_ common.ExtendedContext, taskID string, retention int32) error {
+func (m *mockBatchActionCleanupRepo) DeleteOldBatchActionRunEvents(_ common.ExtendedContext, currentEventId string, taskID string, retention int32) error {
 	m.called = true
+	m.gotCurrentEventId = currentEventId
 	m.gotTaskID = taskID
 	m.gotRetention = retention
 	return m.err
@@ -165,6 +167,7 @@ func TestBatchAction_CleansOldRunsBeforeDispatch(t *testing.T) {
 	assert.Equal(t, events.EventStatusSuccess, status)
 	assert.NotNil(t, result)
 	assert.True(t, cleanupRepo.called)
+	assert.Equal(t, "batch-event-1", cleanupRepo.gotCurrentEventId)
 	assert.Equal(t, "task-1", cleanupRepo.gotTaskID)
 	assert.Equal(t, int32(5), cleanupRepo.gotRetention)
 	assert.True(t, repo.listCalled)
@@ -175,6 +178,7 @@ func TestBatchAction_CleansOldRunsBeforeDispatch_WhenRetentionIsZero(t *testing.
 	eventBus := &mockBatchActionEventBus{}
 	cleanupRepo := &mockBatchActionCleanupRepo{}
 	svc := NewBatchActionService(eventBus, repo, cleanupRepo, nil)
+	prevValue := BATCH_ACTION_RUN_RETENTION
 	BATCH_ACTION_RUN_RETENTION = 0
 
 	status, result := svc.batchAction(testCtx, requestAgingEvent("cql.allRecords=1", map[string]any{"interval": "24h"}))
@@ -183,7 +187,7 @@ func TestBatchAction_CleansOldRunsBeforeDispatch_WhenRetentionIsZero(t *testing.
 	assert.NotNil(t, result)
 	assert.False(t, cleanupRepo.called)
 	assert.True(t, repo.listCalled)
-	BATCH_ACTION_RUN_RETENTION = 5
+	BATCH_ACTION_RUN_RETENTION = prevValue
 }
 
 func TestBatchAction_CleanupErrorDoesNotBlockDispatch(t *testing.T) {

broker/sqlc/event_query.sql

@@ -115,23 +115,28 @@ WHERE event_name IN ('invoke-batch-action', 'invoke-background-action')
   AND event_data -> 'batchActionData' ->> 'taskId' = sqlc.arg(task_id)::text;
 
 -- name: DeleteOldBatchActionRunEvents :exec
-WITH retained_runs AS (
-    SELECT id
-    FROM event
-    WHERE event_name = 'invoke-batch-action'
-      AND event_data -> 'batchActionData' ->> 'taskId' = sqlc.arg(task_id)::text
-    ORDER BY timestamp DESC, id DESC
-    LIMIT sqlc.arg(retention)::int
-),
-old_runs AS (
-    SELECT id
-    FROM event
-    WHERE event_name = 'invoke-batch-action'
-      AND event_data -> 'batchActionData' ->> 'taskId' = sqlc.arg(task_id)::text
-      AND id NOT IN (SELECT id FROM retained_runs)
-)
-DELETE FROM event
-WHERE id IN (SELECT id FROM old_runs);
+    WITH retained_runs AS (
+        SELECT id
+        FROM event
+        WHERE event_name = 'invoke-batch-action'
+          AND event_status IN ('SUCCESS', 'ERROR', 'PROBLEM')
+          AND id <> sqlc.arg(current_event_id)::text
+          AND event_data -> 'batchActionData' ->> 'taskId' = sqlc.arg(task_id)::text
+        ORDER BY timestamp DESC, id DESC
+        LIMIT sqlc.arg(retention)::int
+    ),
+    old_runs AS (
+        SELECT id
+        FROM event
+        WHERE event_name = 'invoke-batch-action'
+          AND event_status IN ('SUCCESS', 'ERROR', 'PROBLEM')
+          AND id <> sqlc.arg(current_event_id)::text
+          AND event_data -> 'batchActionData' ->> 'taskId' = sqlc.arg(task_id)::text
+          AND id NOT IN (SELECT id FROM retained_runs)
+    )
+    DELETE FROM event
+    WHERE id IN (SELECT id FROM old_runs) OR
+          parent_id IN (SELECT id FROM old_runs);
 
 -- name: UpdateEventLifecycle :one
 UPDATE event SET last_signal = $3, event_status = $2