From 2892a7eddc7505be32ab3e6aa49cd56b1c2eb4c8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Mon, 27 Jul 2026 19:40:40 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=BC=BA=E5=8C=96=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=B5=81=E4=B8=8E=E6=8F=92=E4=BB=B6=E8=BF=90=E8=A1=8C=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 显式传播工作流缓存故障并补充仓储测试 - 防止状态轮询重入和过期响应，规范执行记录无权限响应 --- .../repository/BaseRepository.java | 54 ++++- .../ChainStateRepositoryImplTest.java | 185 ++++++++++++++++++ .../resource/RequireResourceAccessAspect.java | 2 +- .../views/ai/plugin/PluginRunTestModal.vue | 68 ++++--- .../ai/workflow/components/WorkflowForm.vue | 50 +++-- .../ai/workflow/components/WorkflowForm.vue | 50 +++-- 6 files changed, 351 insertions(+), 58 deletions(-) create mode 100644 easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/repository/ChainStateRepositoryImplTest.java diff --git 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 index 9952201b..6b65e4cc 100644 --- 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 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 缓存值类型 + * @return 命中的缓存值；缓存不存在或已过期时返回 null + * @throws CacheException 缓存读取失败时抛出 + * @throws ClassCastException 缓存值类型与期望类型不一致时抛出 + */ protected T getCache(String key, Class clazz) { - Object value = cache.get(key); + CacheGetResult 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() + ); + } } diff --git 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 new file mode 100644 index 00000000..56950ae1 --- /dev/null +++ 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 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 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 getResult; + private final CacheResult putResult; + private int putCount; + private Object lastPutValue; + + /** + * 创建缓存调用记录器。 + * + * @param getResult 读取操作结果 + * @param putResult 写入操作结果 + */ + private RecordingCache(CacheGetResult getResult, CacheResult putResult) { + this.getResult = getResult; + this.putResult = putResult; + } + + /** + * 创建实现 JetCache 接口的 JDK 动态代理。 + * + * @return JetCache 测试代理 + */ + @SuppressWarnings("unchecked") + private Cache asCache() { + return (Cache) 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; + } + } +} diff --git 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 index 9206080b..bc4dfb08 100644 --- 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, "无权限访问该执行记录"); } } } diff --git a/easyflow-ui-admin/app/src/views/ai/plugin/PluginRunTestModal.vue b/easyflow-ui-admin/app/src/views/ai/plugin/PluginRunTestModal.vue index 94a70613..781b8584 100644 --- 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([]); const executeId = ref(''); const pollingData = ref({ nodes: {} }); const initSignal = ref(false); -const pollingTimer = ref>(null); +const POLLING_INTERVAL_MS = 1000; +const pollingTimer = ref>(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) { diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowForm.vue b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowForm.vue index cd8c4b45..1445f83a 100644 --- 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); +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; } } diff --git a/easyflow-ui-usercenter/app/src/views/ai/workflow/components/WorkflowForm.vue b/easyflow-ui-usercenter/app/src/views/ai/workflow/components/WorkflowForm.vue index 7d712cb4..ef66d235 100644 --- 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); +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; } }