Skip to content

Commit c9c6ee4

Browse files
KDB-WindKDB-Wind
authored
fix(ingestion): 多起点流水线配置执行前失败 (#97)
IngestionEngine 原先对未被引用的节点使用 findFirst() 任选起点，多根节点配置会静默只执行一条链并把任务标记成功。现收集全部根节点并稳定排序，多于一个时在执行任何节点前抛出列明 ID 的 ClientException；无根节点或唯一根节点 ID 为空保持原有错误语义，合法单链行为不变。同时让 IngestionTaskServiceImpl.upload 原样透传 ClientException，避免引擎配置错误被「读取上传文件失败」前缀包装造成误导。
1 parent 8128b8a commit c9c6ee4

4 files changed

Lines changed: 170 additions & 6 deletions

File tree

‎bootstrap/src/main/java/com/nageoffer/ai/ragent/ingestion/engine/IngestionEngine.java‎

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717

1818
package com.nageoffer.ai.ragent.ingestion.engine;
1919

20-
import cn.hutool.core.util.StrUtil;
2120
import com.nageoffer.ai.ragent.framework.exception.ClientException;
2221
import com.nageoffer.ai.ragent.ingestion.domain.context.IngestionContext;
2322
import com.nageoffer.ai.ragent.ingestion.domain.context.NodeLog;
@@ -70,8 +69,15 @@ public IngestionContext execute(PipelineDefinition pipeline, IngestionContext co
7069
validatePipeline(nodeConfigMap);
7170

7271
// 找到起始节点（没有被任何节点引用的节点）
73-
String startNodeId = findStartNode(nodeConfigMap);
74-
if (StrUtil.isBlank(startNodeId)) {
72+
List<String> startNodeIds = findStartNodes(nodeConfigMap);
73+
if (startNodeIds.isEmpty()) {
74+
throw new ClientException("流水线未找到起始节点");
75+
}
76+
if (startNodeIds.size() > 1) {
77+
throw new ClientException("流水线存在多个起始节点: " + String.join(", ", startNodeIds));
78+
}
79+
String startNodeId = startNodeIds.get(0);
80+
if (!StringUtils.hasText(startNodeId)) {
7581
throw new ClientException("流水线未找到起始节点");
7682
}
7783

@@ -140,16 +146,16 @@ private void validatePipeline(Map<String, NodeConfig> nodeConfigMap) {
140146
/**
141147
* 找到起始节点（没有被任何节点引用的节点）
142148
*/
143-
private String findStartNode(Map<String, NodeConfig> nodeConfigMap) {
149+
private List<String> findStartNodes(Map<String, NodeConfig> nodeConfigMap) {
144150
Set<String> referencedNodes = nodeConfigMap.values().stream()
145151
.map(NodeConfig::getNextNodeId)
146152
.filter(StringUtils::hasText)
147153
.collect(Collectors.toSet());
148154

149155
return nodeConfigMap.keySet().stream()
150156
.filter(nodeId -> !referencedNodes.contains(nodeId))
151-
.findFirst()
152-
.orElse(null);
157+
.sorted(Comparator.nullsFirst(Comparator.naturalOrder()))
158+
.toList();
153159
}
154160

155161
/**

‎bootstrap/src/main/java/com/nageoffer/ai/ragent/ingestion/service/impl/IngestionTaskServiceImpl.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,8 @@ public IngestionResult upload(String pipelineId, MultipartFile file) {
127127
IngestionResult result = executeInternal(pipelineId, source, bytes, mimeType, null);
128128
putTaskSnapshot(result);
129129
return result;
130+
} catch (ClientException e) {
131+
throw e;
130132
} catch (Exception e) {
131133
throw new ClientException("读取上传文件失败: " + e.getMessage());
132134
}

‎bootstrap/src/test/java/com/nageoffer/ai/ragent/ingestion/engine/IngestionEngineTest.java‎

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package com.nageoffer.ai.ragent.ingestion.engine;
19+
20+
import com.fasterxml.jackson.databind.ObjectMapper;
21+
import com.nageoffer.ai.ragent.framework.exception.ClientException;
22+
import com.nageoffer.ai.ragent.ingestion.domain.context.IngestionContext;
23+
import com.nageoffer.ai.ragent.ingestion.domain.enums.IngestionStatus;
24+
import com.nageoffer.ai.ragent.ingestion.domain.pipeline.NodeConfig;
25+
import com.nageoffer.ai.ragent.ingestion.domain.pipeline.PipelineDefinition;
26+
import com.nageoffer.ai.ragent.ingestion.domain.result.NodeResult;
27+
import com.nageoffer.ai.ragent.ingestion.node.IngestionNode;
28+
import org.junit.jupiter.api.Test;
29+
30+
import java.util.ArrayList;
31+
import java.util.List;
32+
33+
import static org.junit.jupiter.api.Assertions.assertEquals;
34+
import static org.junit.jupiter.api.Assertions.assertThrows;
35+
36+
class IngestionEngineTest {
37+
38+
@Test
39+
void multipleStartNodesFailBeforeAnyNodeExecutes() {
40+
List<String> executedNodeIds = new ArrayList<>();
41+
IngestionEngine engine = engine(executedNodeIds);
42+
PipelineDefinition pipeline = PipelineDefinition.builder()
43+
.nodes(List.of(
44+
node("z-root", "z-leaf"),
45+
node("a-leaf", null),
46+
node("z-leaf", null),
47+
node("a-root", "a-leaf")))
48+
.build();
49+
50+
ClientException exception = assertThrows(
51+
ClientException.class,
52+
() -> engine.execute(pipeline, IngestionContext.builder().build()));
53+
54+
assertEquals("流水线存在多个起始节点: a-root, z-root", exception.getMessage());
55+
assertEquals(List.of(), executedNodeIds);
56+
}
57+
58+
@Test
59+
void nullStartNodeIdAlongsideNormalRootFailsWithoutNpe() {
60+
List<String> executedNodeIds = new ArrayList<>();
61+
IngestionEngine engine = engine(executedNodeIds);
62+
PipelineDefinition pipeline = PipelineDefinition.builder()
63+
.nodes(List.of(
64+
node("a-root", null),
65+
node(null, null)))
66+
.build();
67+
68+
ClientException exception = assertThrows(
69+
ClientException.class,
70+
() -> engine.execute(pipeline, IngestionContext.builder().build()));
71+
72+
assertEquals("流水线存在多个起始节点: null, a-root", exception.getMessage());
73+
assertEquals(List.of(), executedNodeIds);
74+
}
75+
76+
@Test
77+
void blankStartNodeIdIsRejectedBeforeExecution() {
78+
List<String> executedNodeIds = new ArrayList<>();
79+
IngestionEngine engine = engine(executedNodeIds);
80+
PipelineDefinition pipeline = PipelineDefinition.builder()
81+
.nodes(List.of(node(" ", null)))
82+
.build();
83+
84+
ClientException exception = assertThrows(
85+
ClientException.class,
86+
() -> engine.execute(pipeline, IngestionContext.builder().build()));
87+
88+
assertEquals("流水线未找到起始节点", exception.getMessage());
89+
assertEquals(List.of(), executedNodeIds);
90+
}
91+
92+
@Test
93+
void singleChainStillExecutesEveryNodeInOrder() {
94+
List<String> executedNodeIds = new ArrayList<>();
95+
IngestionEngine engine = engine(executedNodeIds);
96+
PipelineDefinition pipeline = PipelineDefinition.builder()
97+
.nodes(List.of(
98+
node("middle", "leaf"),
99+
node("leaf", null),
100+
node("root", "middle")))
101+
.build();
102+
IngestionContext context = IngestionContext.builder().build();
103+
104+
engine.execute(pipeline, context);
105+
106+
assertEquals(List.of("root", "middle", "leaf"), executedNodeIds);
107+
assertEquals(IngestionStatus.COMPLETED, context.getStatus());
108+
}
109+
110+
private IngestionEngine engine(List<String> executedNodeIds) {
111+
IngestionNode node = new IngestionNode() {
112+
@Override
113+
public String getNodeType() {
114+
return "test";
115+
}
116+
117+
@Override
118+
public NodeResult execute(IngestionContext context, NodeConfig config) {
119+
executedNodeIds.add(config.getNodeId());
120+
return NodeResult.ok();
121+
}
122+
};
123+
return new IngestionEngine(
124+
List.of(node),
125+
new ConditionEvaluator(new ObjectMapper()),
126+
new NodeOutputExtractor());
127+
}
128+
129+
private NodeConfig node(String nodeId, String nextNodeId) {
130+
return NodeConfig.builder()
131+
.nodeId(nodeId)
132+
.nodeType("test")
133+
.nextNodeId(nextNodeId)
134+
.build();
135+
}
136+
}

‎bootstrap/src/test/java/com/nageoffer/ai/ragent/ingestion/service/impl/IngestionTaskServiceImplTest.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,26 +19,33 @@
1919

2020
import com.fasterxml.jackson.databind.ObjectMapper;
2121
import com.nageoffer.ai.ragent.audit.support.BizChangeLogContext;
22+
import com.nageoffer.ai.ragent.framework.exception.ClientException;
2223
import com.nageoffer.ai.ragent.ingestion.controller.vo.IngestionTaskNodeVO;
2324
import com.nageoffer.ai.ragent.ingestion.controller.vo.IngestionTaskVO;
2425
import com.nageoffer.ai.ragent.ingestion.dao.entity.IngestionTaskDO;
2526
import com.nageoffer.ai.ragent.ingestion.dao.entity.IngestionTaskNodeDO;
2627
import com.nageoffer.ai.ragent.ingestion.dao.mapper.IngestionTaskMapper;
2728
import com.nageoffer.ai.ragent.ingestion.dao.mapper.IngestionTaskNodeMapper;
29+
import com.nageoffer.ai.ragent.ingestion.domain.pipeline.PipelineDefinition;
2830
import com.nageoffer.ai.ragent.ingestion.engine.IngestionEngine;
2931
import com.nageoffer.ai.ragent.ingestion.service.IngestionPipelineService;
3032
import org.junit.jupiter.api.BeforeEach;
3133
import org.junit.jupiter.api.Test;
3234
import org.junit.jupiter.api.extension.ExtendWith;
3335
import org.mockito.Mock;
3436
import org.mockito.junit.jupiter.MockitoExtension;
37+
import org.springframework.mock.web.MockMultipartFile;
3538

39+
import java.nio.charset.StandardCharsets;
3640
import java.util.List;
3741
import java.util.Map;
3842

3943
import static org.junit.jupiter.api.Assertions.assertEquals;
44+
import static org.junit.jupiter.api.Assertions.assertSame;
45+
import static org.junit.jupiter.api.Assertions.assertThrows;
4046
import static org.junit.jupiter.api.Assertions.assertTrue;
4147
import static org.mockito.ArgumentMatchers.any;
48+
import static org.mockito.Mockito.mock;
4249
import static org.mockito.Mockito.when;
4350

4451
@ExtendWith(MockitoExtension.class)
@@ -130,4 +137,17 @@ void invalidOrBlankJsonFallsBackToEmptyMaps() {
130137
task.setMetadataJson("null");
131138
assertTrue(service.get("task-1").getMetadata().isEmpty());
132139
}
140+
141+
@Test
142+
void uploadPropagatesEngineClientExceptionWithoutWrapping() {
143+
ClientException engineFailure = new ClientException("流水线存在多个起始节点: a-root, z-root");
144+
when(pipelineService.getDefinition("pipeline-1")).thenReturn(mock(PipelineDefinition.class));
145+
when(engine.execute(any(), any())).thenThrow(engineFailure);
146+
MockMultipartFile file = new MockMultipartFile(
147+
"file", "doc.txt", "text/plain", "hello".getBytes(StandardCharsets.UTF_8));
148+
149+
ClientException thrown = assertThrows(ClientException.class, () -> service.upload("pipeline-1", file));
150+
151+
assertSame(engineFailure, thrown);
152+
}
133153
}

0 commit comments

Comments
 (0)