[app-builder] ensure there is only one process context cleanup task (#510)

Author: loveTsong

app-builder/plugins/waterflow-appfactory/src/main/java/modelengine/fit/waterflow/appfactory/task/CleanFlowInstances.java

@@ -0,0 +1,58 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved.
+ *  This file is a part of the ModelEngine Project.
+ *  Licensed under the MIT License. See License.txt in the project root for license information.
+ *--------------------------------------------------------------------------------------------*/
+
+package modelengine.fit.waterflow.appfactory.task;
+
+import modelengine.fit.waterflow.service.SingleFlowRuntimeService;
+import modelengine.fitframework.annotation.Component;
+import modelengine.fitframework.annotation.Value;
+import modelengine.fitframework.log.Logger;
+import modelengine.fitframework.schedule.annotation.Scheduled;
+import modelengine.fitframework.util.ThreadUtils;
+
+/**
+ * 定时清理流程中已完成的上下文。
+ * <p>包括成功、失败、终止的流程数据。</p>
+ *
+ * @author 杨祥宇
+ * @since 2025-04-02
+ */
+@Component
+public class CleanFlowInstances {
+    private static final Logger LOG = Logger.get(CleanFlowInstances.class);
+    private static final int LIMIT = 1000;
+    private static final int BATCH_INTERNAL = 1000;
+
+    private final int expiredDays;
+    private final SingleFlowRuntimeService singleFlowRuntimeService;
+
+    public CleanFlowInstances(@Value("${jane.flowsEngine.contextExpiredDays}") int expiredDays,
+            SingleFlowRuntimeService singleFlowRuntimeService) {
+        this.expiredDays = expiredDays;
+        this.singleFlowRuntimeService = singleFlowRuntimeService;
+    }
+
+    /**
+     * 每天凌晨 3 点定时清理超指定天数的流程运行数据。
+     * <p>指定天数来源于 {@code ${jane.flowsEngine.contextExpiredDays}} 配置的值。</p>
+     * <p>多实例并发执行分析：会并发执行超期链路信息查询，可能导致重复获取相同 {@code traceIds}，重复删除 {@code traceIds}
+     * 以及上下文数据不会对结果有影响。</p>
+     */
+    @Scheduled(strategy = Scheduled.Strategy.CRON, value = "0 0 3 * * ?")
+    public void cleanContextSchedule() {
+        LOG.info("Starting expired flow instances cleaning");
+        try {
+            while (this.singleFlowRuntimeService.cleanInstances(this.expiredDays, LIMIT)) {
+                ThreadUtils.sleep(BATCH_INTERNAL);
+            }
+        } catch (Exception ex) {
+            LOG.error("Clean expired flow instances error. [errorMessage={}]", ex.getMessage());
+            LOG.error("Exception:", ex);
+        } finally {
+            LOG.info("Finished expired flow instances cleaning.");
+        }
+    }
+}

app-builder/plugins/waterflow-appfactory/src/main/resources/application.yml

@@ -13,5 +13,6 @@ jane:
   flowsEngine:
     retry:
       scheduleRate: 60000
+    contextExpiredDays: 3
 
 distributed-lock-provider: databaseDistributedLockProvider

app-builder/waterflow/java/waterflow-runtime-service/src/main/java/modelengine/fit/waterflow/service/SingleFlowRuntimeService.java

@@ -9,7 +9,6 @@
 import modelengine.fit.waterflow.entity.OperationContext;
 import modelengine.fit.waterflow.entity.FlowStartDTO;
 import modelengine.fit.waterflow.entity.FlowStartInfo;
-import modelengine.fit.waterflow.entity.FlowStartParameter;
 import modelengine.fit.waterflow.entity.JoberErrorInfo;
 
 import java.util.Map;
@@ -71,4 +70,13 @@ void terminateFlows(String flowDefinitionId, String instanceId, Map<String, Obje
      * @param operationContext 操作人，为null则默认为前一个操作人
      */
     void failAsyncJob(String flowDataId, JoberErrorInfo errorInfo, OperationContext operationContext);
+
+    /**
+     * 根据过期时间清理实例上下文数据。
+     *
+     * @param expiredDays 过期天数。
+     * @param limit 最大清理数量。
+     * @return 表示是否有数据被清理。
+     */
+    boolean cleanInstances(int expiredDays, int limit);
 }

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/biz/service/SingleFlowRuntimeServiceImpl.java

@@ -26,6 +26,7 @@
 import modelengine.fitframework.annotation.Component;
 import modelengine.fitframework.inspection.Validation;
 import modelengine.fitframework.log.Logger;
+import modelengine.fitframework.transaction.Transactional;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -127,6 +128,27 @@ public void failAsyncJob(String flowDataId, JoberErrorInfo errorInfo, OperationC
                 new WaterflowException(FLOW_EXECUTE_ASYNC_JOBER_FAILED, errorInfo), operationContext);
     }
 
+    @Override
+    public boolean cleanInstances(int expiredDays, int limit) {
+        List<String> traceIds = this.traceRepo.getExpiredTrace(expiredDays, limit);
+        if (traceIds.isEmpty()) {
+            return false;
+        }
+        this.deleteFlowContext(traceIds);
+        return true;
+    }
+
+    /**
+     * 根据链路唯一标识列表删除链路信息和上下文数据。
+     *
+     * @param traceIds 表示流程链路唯一标识列表的 {@link List}{@code <}{@link String}{@code >}。
+     */
+    @Transactional
+    public void deleteFlowContext(List<String> traceIds) {
+        this.repo.deleteByTraceIdList(traceIds);
+        this.traceRepo.deleteByIdList(traceIds);
+    }
+
     private static boolean isAllowResumeOrFailAsyncJob(FlowContext<FlowData> context) {
         return context != null && FlowNodeStatus.PROCESSING.equals(context.getStatus());
     }

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/biz/service/scheduletasks/CleanContextSchedule.java

@@ -16,7 +16,6 @@ jane:
       scheduleRate: 10000
       maxCount: 0
     isNeedFlowCallbackAdapt: false
-    contextExpiredDays: 1
 
 distributed-lock-provider: databaseDistributedLockProvider