2026-08-20 11:36:27 +08:00
6 changed files with 351 additions and 58 deletions
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/repository/BaseRepository.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/repository/BaseRepository.java
@@ -1,27 +1,75 @@
 package tech.easyflow.ai.easyagentsflow.repository;

 import com.alicp.jetcache.Cache;
+import com.alicp.jetcache.CacheException;
+import com.alicp.jetcache.CacheGetResult;
+import com.alicp.jetcache.CacheResult;
+import com.alicp.jetcache.CacheResultCode;

 import javax.annotation.Resource;
 import java.util.concurrent.TimeUnit;

+/**
+ * 工作流运行状态缓存仓储基类。
+ */
 public class BaseRepository {

    @Resource(name = "defaultCache")
    private Cache<String, Object> cache;

    /**
-     * chain 的相关状态缓存三天
+     * 保存工作流运行状态，缓存有效期为三天。
+     *
+     * @param key 缓存键
+     * @param value 缓存值
+     * @throws CacheException 缓存写入失败时抛出
     */
    protected void putCache(String key, Object value) {
-        cache.put(key, value, 3, TimeUnit.DAYS);
+        CacheResult result = cache.PUT(key, value, 3, TimeUnit.DAYS);
+        if (!result.isSuccess()) {
+            throw cacheOperationException("写入", key, result);
+        }
    }

+    /**
+     * 读取并校验工作流运行状态缓存。
+     *
+     * @param key 缓存键
+     * @param clazz 期望的缓存值类型
+     * @param <T> 缓存值类型
+     * @return 命中的缓存值；缓存不存在或已过期时返回 null
+     * @throws CacheException 缓存读取失败时抛出
+     * @throws ClassCastException 缓存值类型与期望类型不一致时抛出
+     */
    protected <T> T getCache(String key, Class<T> clazz) {
-        Object value = cache.get(key);
+        CacheGetResult<Object> result = cache.GET(key);
+        CacheResultCode resultCode = result.getResultCode();
+        if (resultCode == CacheResultCode.NOT_EXISTS || resultCode == CacheResultCode.EXPIRED) {
+            return null;
+        }
+        if (!result.isSuccess()) {
+            throw cacheOperationException("读取", key, result);
+        }
+        Object value = result.getValue();
        if (value == null) {
            return null;
        }
        return clazz.cast(value);
    }
+
+    /**
+     * 构建包含缓存操作上下文的异常。
+     *
+     * @param operation 操作名称
+     * @param key 缓存键
+     * @param result JetCache 操作结果
+     * @return 缓存操作异常
+     */
+    private CacheException cacheOperationException(String operation, String key, CacheResult result) {
+        return new CacheException(
+            "工作流状态缓存" + operation + "失败，key=" + key
+                + "，resultCode=" + result.getResultCode()
+                + "，message=" + result.getMessage()
+        );
+    }
 }
--- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/repository/ChainStateRepositoryImplTest.java
+++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/repository/ChainStateRepositoryImplTest.java
@@ -0,0 +1,185 @@
+package tech.easyflow.ai.easyagentsflow.repository;
+
+import com.alicp.jetcache.Cache;
+import com.alicp.jetcache.CacheException;
+import com.alicp.jetcache.CacheGetResult;
+import com.alicp.jetcache.CacheResult;
+import com.alicp.jetcache.CacheResultCode;
+import com.alicp.jetcache.support.CacheEncodeException;
+import com.easyagents.flow.core.chain.ChainState;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * {@link ChainStateRepositoryImpl} 缓存异常处理回归测试。
+ */
+public class ChainStateRepositoryImplTest {
+
+    /**
+     * 验证缓存解码失败时抛出异常且不创建空工作流状态。
+     *
+     * @throws Exception 缓存依赖注入失败时抛出
+     */
+    @Test
+    public void loadShouldFailWithoutOverwritingStateWhenCacheDecodeFails() throws Exception {
+        String instanceId = "decode-failed-instance";
+        CacheGetResult<Object> failure = new CacheGetResult<>(new CacheEncodeException(
+            "decode error",
+            new ClassNotFoundException("com.alicp.jetcache.CacheValueHolder")
+        ));
+        RecordingCache cache = new RecordingCache(failure, CacheResult.SUCCESS_WITHOUT_MSG);
+        ChainStateRepositoryImpl repository = repository(cache.asCache());
+
+        try {
+            repository.load(instanceId);
+            Assert.fail("cache decode failure should be propagated");
+        } catch (CacheException expected) {
+            Assert.assertTrue(expected.getMessage().contains("工作流状态缓存读取失败"));
+            Assert.assertTrue(expected.getMessage().contains(instanceId));
+        }
+
+        Assert.assertEquals(0, cache.getPutCount());
+    }
+
+    /**
+     * 验证缓存未命中时创建并持久化新的工作流状态。
+     *
+     * @throws Exception 缓存依赖注入失败时抛出
+     */
+    @Test
+    public void loadShouldCreateStateWhenCacheDoesNotExist() throws Exception {
+        String instanceId = "new-instance";
+        RecordingCache cache = new RecordingCache(
+            new CacheGetResult<>(CacheResultCode.NOT_EXISTS, null, null),
+            CacheResult.SUCCESS_WITHOUT_MSG
+        );
+        ChainStateRepositoryImpl repository = repository(cache.asCache());
+
+        ChainState state = repository.load(instanceId);
+
+        Assert.assertEquals(instanceId, state.getInstanceId());
+        Assert.assertEquals(1, cache.getPutCount());
+        Assert.assertSame(state, cache.getLastPutValue());
+    }
+
+    /**
+     * 验证缓存写入失败时不会返回未持久化的工作流状态。
+     *
+     * @throws Exception 缓存依赖注入失败时抛出
+     */
+    @Test
+    public void loadShouldFailWhenNewStateCannotBePersisted() throws Exception {
+        RecordingCache cache = new RecordingCache(
+            new CacheGetResult<>(CacheResultCode.NOT_EXISTS, null, null),
+            new CacheResult(new IllegalStateException("redis unavailable"))
+        );
+        ChainStateRepositoryImpl repository = repository(cache.asCache());
+
+        try {
+            repository.load("write-failed-instance");
+            Assert.fail("cache write failure should be propagated");
+        } catch (CacheException expected) {
+            Assert.assertTrue(expected.getMessage().contains("工作流状态缓存写入失败"));
+        }
+
+        Assert.assertEquals(1, cache.getPutCount());
+    }
+
+    /**
+     * 创建工作流状态仓储并注入缓存。
+     *
+     * @param cache 测试缓存
+     * @return 已完成依赖注入的工作流状态仓储
+     * @throws Exception 反射注入失败时抛出
+     */
+    private ChainStateRepositoryImpl repository(Cache<String, Object> cache) throws Exception {
+        ChainStateRepositoryImpl repository = new ChainStateRepositoryImpl();
+        Field field = BaseRepository.class.getDeclaredField("cache");
+        field.setAccessible(true);
+        field.set(repository, cache);
+        return repository;
+    }
+
+    /**
+     * 仅实现当前仓储测试所需操作的 JetCache 调用记录器。
+     */
+    private static final class RecordingCache implements InvocationHandler {
+
+        private final CacheGetResult<Object> getResult;
+        private final CacheResult putResult;
+        private int putCount;
+        private Object lastPutValue;
+
+        /**
+         * 创建缓存调用记录器。
+         *
+         * @param getResult 读取操作结果
+         * @param putResult 写入操作结果
+         */
+        private RecordingCache(CacheGetResult<Object> getResult, CacheResult putResult) {
+            this.getResult = getResult;
+            this.putResult = putResult;
+        }
+
+        /**
+         * 创建实现 JetCache 接口的 JDK 动态代理。
+         *
+         * @return JetCache 测试代理
+         */
+        @SuppressWarnings("unchecked")
+        private Cache<String, Object> asCache() {
+            return (Cache<String, Object>) Proxy.newProxyInstance(
+                Cache.class.getClassLoader(),
+                new Class<?>[]{Cache.class},
+                this
+            );
+        }
+
+        /**
+         * 处理仓储发起的缓存读写操作。
+         *
+         * @param proxy 代理对象
+         * @param method 被调用方法
+         * @param args 调用参数
+         * @return 预设的缓存操作结果
+         */
+        @Override
+        public Object invoke(Object proxy, Method method, Object[] args) {
+            if ("GET".equals(method.getName())) {
+                return getResult;
+            }
+            if ("PUT".equals(method.getName()) && args != null && args.length == 4) {
+                putCount++;
+                lastPutValue = args[1];
+                Assert.assertEquals(3L, args[2]);
+                Assert.assertEquals(TimeUnit.DAYS, args[3]);
+                return putResult;
+            }
+            throw new UnsupportedOperationException("unsupported cache method: " + method.getName());
+        }
+
+        /**
+         * 获取写入调用次数。
+         *
+         * @return 写入调用次数
+         */
+        private int getPutCount() {
+            return putCount;
+        }
+
+        /**
+         * 获取最后一次写入的缓存值。
+         *
+         * @return 最后一次写入的缓存值
+         */
+        private Object getLastPutValue() {
+            return lastPutValue;
+        }
+    }
+}
--- a/easyflow-modules/easyflow-module-system/src/main/java/tech/easyflow/system/permission/resource/RequireResourceAccessAspect.java
+++ b/easyflow-modules/easyflow-module-system/src/main/java/tech/easyflow/system/permission/resource/RequireResourceAccessAspect.java
@@ -80,7 +80,7 @@ public class RequireResourceAccessAspect {
        LoginAccount loginAccount = SaTokenUtil.getLoginAccount();
        String accountId = loginAccount == null || loginAccount.getId() == null ? null : loginAccount.getId().toString();
        if (!executionOwnerKey.equals(accountId)) {
-            throw new BusinessException("无权限访问该执行记录");
+            throw new BusinessException(403, 403, "无权限访问该执行记录");
        }
    }
 }
--- a/easyflow-ui-admin/app/src/views/ai/plugin/PluginRunTestModal.vue
+++ b/easyflow-ui-admin/app/src/views/ai/plugin/PluginRunTestModal.vue
@@ -44,7 +44,10 @@ const pollingNodes = ref<any[]>([]);
 const executeId = ref('');
 const pollingData = ref<any>({ nodes: {} });
 const initSignal = ref(false);
-const pollingTimer = ref<null | ReturnType<typeof setInterval>>(null);
+const POLLING_INTERVAL_MS = 1000;
+const pollingTimer = ref<null | ReturnType<typeof setTimeout>>(null);
+let pollingActive = false;
+let pollingGeneration = 0;
 const activeIndex = ref('1');
 const dialogContentKey = ref(0);
 const dialogPreparing = ref(false);
@@ -237,42 +240,55 @@ function handleWorkflowSubmit(runParams: any) {

 function startPolling(nextExecuteId: string) {
  stopPolling();
-  pollingTimer.value = setInterval(() => {
-    executePolling(nextExecuteId);
-  }, 1000);
+  pollingActive = true;
+  pollingGeneration += 1;
+  schedulePolling(nextExecuteId, pollingGeneration);
+}
+
+function schedulePolling(nextExecuteId: string, generation: number) {
+  pollingTimer.value = setTimeout(() => {
+    pollingTimer.value = null;
+    void executePolling(nextExecuteId, generation);
+  }, POLLING_INTERVAL_MS);
 }

 function stopPolling() {
+  pollingActive = false;
+  pollingGeneration += 1;
  if (pollingTimer.value) {
-    clearInterval(pollingTimer.value);
+    clearTimeout(pollingTimer.value);
    pollingTimer.value = null;
  }
 }

-function executePolling(nextExecuteId: string) {
-  api
-    .post('/api/v1/pluginItem/testChainStatus', {
+async function executePolling(nextExecuteId: string, generation: number) {
+  try {
+    const res = await api.post('/api/v1/pluginItem/testChainStatus', {
      executeId: nextExecuteId,
      nodes: pollingNodes.value,
-    })
-    .then((res) => {
-      if (res.errorCode !== 0) {
-        return;
-      }
-      const nextData = {
-        ...res.data,
-        nodes: res.data?.nodes || {},
-      };
-      pollingData.value = nextData;
-      runResultResponse.value = nextData;
-      if (nextData.status !== 1) {
-        stopPolling();
-      }
-    })
-    .catch((error) => {
-      stopPolling();
-      runResultResponse.value = buildErrorResult(error);
    });
+    if (!pollingActive || generation !== pollingGeneration) return;
+    if (res.errorCode !== 0) {
+      stopPolling();
+      return;
+    }
+
+    const nextData = {
+      ...res.data,
+      nodes: res.data?.nodes || {},
+    };
+    pollingData.value = nextData;
+    runResultResponse.value = nextData;
+    if (nextData.status !== 1) {
+      stopPolling();
+      return;
+    }
+    schedulePolling(nextExecuteId, generation);
+  } catch (error) {
+    if (!pollingActive || generation !== pollingGeneration) return;
+    stopPolling();
+    runResultResponse.value = buildErrorResult(error);
+  }
 }

 function resumeChain(payload: any) {
--- a/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowForm.vue
+++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowForm.vue
@@ -137,7 +137,10 @@ function submitV2() {
    }
  });
 }
-const timer = ref();
+const POLLING_INTERVAL_MS = 1000;
+const timer = ref<null | ReturnType<typeof setTimeout>>(null);
+let pollingActive = false;
+let pollingGeneration = 0;
 const nodes = ref(
  props.tinyFlowData.nodes.map((node: any) => ({
    nodeId: node.id,
@@ -146,27 +149,46 @@ const nodes = ref(
 );
 // 轮询执行结果
 function startPolling(executeId: any) {
-  if (timer.value) return;
-  timer.value = setInterval(() => executePolling(executeId), 1000);
+  if (pollingActive) return;
+  pollingActive = true;
+  pollingGeneration += 1;
+  schedulePolling(executeId, pollingGeneration);
 }
-function executePolling(executeId: any) {
-  api
-    .post('/api/v1/workflow/getChainStatus', {
+function schedulePolling(executeId: any, generation: number) {
+  timer.value = setTimeout(() => {
+    timer.value = null;
+    void executePolling(executeId, generation);
+  }, POLLING_INTERVAL_MS);
+}
+async function executePolling(executeId: any, generation: number) {
+  try {
+    const res = await api.post('/api/v1/workflow/getChainStatus', {
      executeId,
      nodes: nodes.value,
-    })
-    .then((res) => {
-      // 5 是挂起状态
-      if (res.data.status !== 1 || res.data.status === 5) {
-        stopPolling();
-      }
-      props.onAsyncExecute?.(res.data);
    });
+    if (!pollingActive || generation !== pollingGeneration) return;
+
+    // 5 是挂起状态；所有非运行态都结束当前轮询。
+    if (res.data.status !== 1) {
+      stopPolling();
+    }
+    props.onAsyncExecute?.(res.data);
+
+    if (pollingActive && generation === pollingGeneration) {
+      schedulePolling(executeId, generation);
+    }
+  } catch (error) {
+    if (!pollingActive || generation !== pollingGeneration) return;
+    stopPolling();
+    console.error('工作流状态轮询失败', error);
+  }
 }
 function stopPolling() {
  submitLoading.value = false;
+  pollingActive = false;
+  pollingGeneration += 1;
  if (timer.value) {
-    clearInterval(timer.value);
+    clearTimeout(timer.value);
    timer.value = null;
  }
 }
--- a/easyflow-ui-usercenter/app/src/views/ai/workflow/components/WorkflowForm.vue
+++ b/easyflow-ui-usercenter/app/src/views/ai/workflow/components/WorkflowForm.vue
@@ -136,7 +136,10 @@ function submitV2() {
    }
  });
 }
-const timer = ref();
+const POLLING_INTERVAL_MS = 1000;
+const timer = ref<null | ReturnType<typeof setTimeout>>(null);
+let pollingActive = false;
+let pollingGeneration = 0;
 const nodes = ref(
  props.tinyFlowData.nodes.map((node: any) => ({
    nodeId: node.id,
@@ -145,27 +148,46 @@ const nodes = ref(
 );
 // 轮询执行结果
 function startPolling(executeId: any) {
-  if (timer.value) return;
-  timer.value = setInterval(() => executePolling(executeId), 1000);
+  if (pollingActive) return;
+  pollingActive = true;
+  pollingGeneration += 1;
+  schedulePolling(executeId, pollingGeneration);
 }
-function executePolling(executeId: any) {
-  api
-    .post('/userCenter/workflow/getChainStatus', {
+function schedulePolling(executeId: any, generation: number) {
+  timer.value = setTimeout(() => {
+    timer.value = null;
+    void executePolling(executeId, generation);
+  }, POLLING_INTERVAL_MS);
+}
+async function executePolling(executeId: any, generation: number) {
+  try {
+    const res = await api.post('/userCenter/workflow/getChainStatus', {
      executeId,
      nodes: nodes.value,
-    })
-    .then((res) => {
-      // 5 是挂起状态
-      if (res.data.status !== 1 || res.data.status === 5) {
-        stopPolling();
-      }
-      props.onAsyncExecute?.(res.data);
    });
+    if (!pollingActive || generation !== pollingGeneration) return;
+
+    // 5 是挂起状态；所有非运行态都结束当前轮询。
+    if (res.data.status !== 1) {
+      stopPolling();
+    }
+    props.onAsyncExecute?.(res.data);
+
+    if (pollingActive && generation === pollingGeneration) {
+      schedulePolling(executeId, generation);
+    }
+  } catch (error) {
+    if (!pollingActive || generation !== pollingGeneration) return;
+    stopPolling();
+    console.error('工作流状态轮询失败', error);
+  }
 }
 function stopPolling() {
  submitLoading.value = false;
+  pollingActive = false;
+  pollingGeneration += 1;
  if (timer.value) {
-    clearInterval(timer.value);
+    clearTimeout(timer.value);
    timer.value = null;
  }
 }