fix(ai): WebSocket 推送静默化,并补重试与静默推送用例(issue 11 / 12)

- WebSocketServer.sendMessage 里两行 System.out.println 删除。该方法被**每个 SSE
  事件**调用一次(大纲单次生成实测 3062 个事件,即一次请求刷 6000+ 行),而且绕过
  日志框架——既不受 logging.level 控制(生产把 com.gxwebsoft 压到 ERROR 也拦不住),
  也不进日志文件,真故障时需要的几行会被顶掉。日志改到低频的 onOpen / onClose;
  同时把 onOpen 里静默吞掉的握手回应异常改成 WARN(那是「连上了却收不到内容」的
  第一现场)。
- 新增用例:AiCreativeRetryTest、AiCreativeRetryIntegrationTest(issue 11,
  后者用本地 HTTP server 扮演 Dify,断言瞬时故障确实重跑一次、凭证失效只请求一次)、
  WebSocketServerSilentPushTest(issue 12,捕获 System.out 断言推送路径零输出)。
This commit is contained in:
2026-10-13 20:00:01 +08:00
parent 8406c90cf3
commit 21e49c6a1a
4 changed files with 532 additions and 88 deletions
@@ -1,88 +1,106 @@
package com.gxwebsoft.ai.config; package com.gxwebsoft.ai.config;
import org.springframework.stereotype.Component; import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.websocket.OnClose; import org.springframework.stereotype.Component;
import javax.websocket.OnOpen;
import javax.websocket.Session; import javax.websocket.OnClose;
import javax.websocket.server.PathParam; import javax.websocket.OnOpen;
import javax.websocket.server.ServerEndpoint; import javax.websocket.Session;
import java.io.IOException; import javax.websocket.server.PathParam;
import java.util.concurrent.ConcurrentHashMap; import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
@ServerEndpoint(value = "/chat/{userId}")
@Component
public class WebSocketServer { @ServerEndpoint(value = "/chat/{userId}")
@Component
/** public class WebSocketServer {
* concurrent包的线程安全Set,用来存放每个客户端对应的MyWebSocket对象。
*/ private static final Logger logger = LoggerFactory.getLogger(WebSocketServer.class);
private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
/** /**
* 与某个客户端的连接会话,需要通过它来给客户端发送数据 * concurrent包的线程安全Set,用来存放每个客户端对应的MyWebSocket对象。
*/ */
private Session session; private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
/** /**
* 接收userId * 与某个客户端的连接会话,需要通过它来给客户端发送数据
*/ */
private String userId = ""; private Session session;
/**
/** * 接收userId
* 连接建立成功调用的方法 */
*/ private String userId = "";
@OnOpen
public void onOpen(Session session, @PathParam("userId") String userId) { /**
this.session = session; * 连接建立成功调用的方法
this.userId = userId; *
if (webSocketMap.containsKey(userId)) { * <p>这里才是值得打日志的位置:一次连接一条,低频且信息量大
webSocketMap.remove(userId); * (排查「大纲/创意生成时端上收不到流式内容」时,第一件事就是确认端到底连没连上)。
webSocketMap.put(userId, this); * 推送路径 {@link #sendMessage(String, String)} 相反,单次请求会被调用上千次,必须静默。
//加入set中 */
} else { @OnOpen
webSocketMap.put(userId, this); public void onOpen(Session session, @PathParam("userId") String userId) {
} this.session = session;
this.userId = userId;
try { if (webSocketMap.containsKey(userId)) {
sendMessage(userId, "连接成功"); webSocketMap.remove(userId);
} catch (IOException e) { webSocketMap.put(userId, this);
//加入set中
} } else {
} webSocketMap.put(userId, this);
}
/** logger.info("websocket connected, userId={}", userId);
* 连接关闭调用的方法
*/ try {
@OnClose sendMessage(userId, "连接成功");
public void onClose() { } catch (IOException e) {
if (webSocketMap.containsKey(userId)) { logger.warn("websocket 握手回应发送失败, userId={}", userId, e);
webSocketMap.remove(userId); }
} }
}
/**
* 连接关闭调用的方法
/** */
* 实现服务器主动推送 @OnClose
*/ public void onClose() {
public void sendMessage(String userId, String message) throws IOException { if (webSocketMap.containsKey(userId)) {
System.out.println("userId:" + userId); webSocketMap.remove(userId);
if (webSocketMap.containsKey(userId)) { logger.info("websocket closed, userId={}", userId);
Session session1 = webSocketMap.get(userId).session; }
System.out.println("session1:" + session1); }
if (session1 != null) session1.getBasicRemote().sendText(message);
}
} /**
* 实现服务器主动推送
*
/** * <p>两条 AI 流式链路(创意助手、活动策划大纲)**每个 SSE 事件**都会走到这里:
* 实现服务器主动推送 * 实测一次大纲生成 3062 个事件,即单次请求就会调 3000+ 次。
*/ * 因此这里**一行日志都不能打**——原先的两行 {@code System.out.println}
public void sendAllMessage(String message) throws IOException { * ({@code userId} 与 {@code session1})在控制台刷出上万行「userId:833」,
ConcurrentHashMap.KeySetView<String, WebSocketServer> userIds = webSocketMap.keySet(); * 既刷屏、又绕过日志级别、还会淹没真正的告警。
for (String userId : userIds) { *
WebSocketServer webSocketServer = webSocketMap.get(userId); * <p>连接与断开另见 {@link #onOpen(Session, String)} / {@link #onClose()},那里是低频的。
webSocketServer.session.getBasicRemote().sendText(message); */
} public void sendMessage(String userId, String message) throws IOException {
} if (webSocketMap.containsKey(userId)) {
Session session1 = webSocketMap.get(userId).session;
} if (session1 != null) {
session1.getBasicRemote().sendText(message);
}
}
}
/**
* 实现服务器主动推送
*/
public void sendAllMessage(String message) throws IOException {
ConcurrentHashMap.KeySetView<String, WebSocketServer> userIds = webSocketMap.keySet();
for (String userId : userIds) {
WebSocketServer webSocketServer = webSocketMap.get(userId);
webSocketServer.session.getBasicRemote().sendText(message);
}
}
}
@@ -0,0 +1,82 @@
package com.gxwebsoft.ai.config;
import org.junit.jupiter.api.Test;
import javax.websocket.RemoteEndpoint;
import javax.websocket.Session;
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.lang.reflect.Proxy;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.Callable;
import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* WebSocket 推送是**高频路径**,不得往标准输出打印。
*
* <p>两条 AI 流式链路(创意助手、活动策划大纲)每个 SSE 事件都会调一次
* {@link WebSocketServer#sendMessage(String, String)}:实测一次大纲生成 3062 个事件,
* 即单次请求就会调用 3000+ 次。最初这里有两行 {@code System.out.println}
* ({@code userId} 与 {@code session1}),于是一次生成就在控制台刷出上万行
* 「userId:833」,既刷屏又绕过日志级别(见 .scratch/ai-assistant/issues/12)。
*
* <p>两个分支(目标已连接 / 未连接)各锁一条:改回 {@code println} 立刻失败。
*/
class WebSocketServerSilentPushTest {
@Test
void 目标未连接时推送不得打印() throws Exception {
String output = captureStdout(() -> {
// 未连接的 userId:走"查不到会话"的分支,也正是刷屏时最常见的那条
new WebSocketServer().sendMessage("silent-push-absent", "{\"answer\":\"测试\"}");
return null;
});
assertEquals("", output, "推送是高频路径,不能刷标准输出");
}
@Test
void 目标已连接时推送也不得打印() throws Exception {
WebSocketServer server = new WebSocketServer();
// onOpen 会打一条 INFO(连接日志,低频且应当保留),故放在捕获之外
server.onOpen(fakeSession(), "silent-push-open");
String output = captureStdout(() -> {
server.sendMessage("silent-push-open", "{\"answer\":\"测试\"}");
return null;
});
assertEquals("", output, "推送是高频路径,不能刷标准输出");
}
/**
* 捕获一段代码往标准输出写的内容。推送是 void 且抛受检异常,故用 {@link Callable}。
*/
private static String captureStdout(Callable<Void> action) throws Exception {
PrintStream original = System.out;
ByteArrayOutputStream captured = new ByteArrayOutputStream();
try {
System.setOut(new PrintStream(captured, true, StandardCharsets.UTF_8.name()));
action.call();
} finally {
System.setOut(original);
}
return captured.toString(StandardCharsets.UTF_8.name());
}
/**
* 一个什么都不做的 {@link Session}:只为让 {@code sendMessage} 走进"已连接"分支。
*/
private static Session fakeSession() {
return (Session) Proxy.newProxyInstance(Session.class.getClassLoader(),
new Class<?>[]{Session.class}, (proxy, method, args) -> {
if ("getBasicRemote".equals(method.getName())) {
return Proxy.newProxyInstance(RemoteEndpoint.Basic.class.getClassLoader(),
new Class<?>[]{RemoteEndpoint.Basic.class},
(endpoint, call, callArgs) -> null);
}
return null;
});
}
}
@@ -0,0 +1,220 @@
package com.gxwebsoft.ai.controller;
import com.alibaba.fastjson.JSONObject;
import com.gxwebsoft.ai.config.DifyProperties;
import com.gxwebsoft.common.core.web.ApiResult;
import com.sun.net.httpserver.HttpServer;
import org.junit.jupiter.api.Test;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* 创意助手自动重试的端到端验证(见 .scratch/ai-assistant/issues/11)。
*
* <p>真起一个本地 HTTP server 扮演 Dify:第一次返回上游容量不足的失败流,
* 第二次返回成功流,然后确认 controller **真的重跑了工作流**并把第二次的产出回给端上。
*
* <p>单元测试只能证明"判定函数返回 true";这条用例证明的是"重试确实发生了、且结果被采用",
* 也就是用户看到的"有时候能出图、有时候失败"是否真的被修掉。
*/
class AiCreativeRetryIntegrationTest {
/**
* 上游容量不足的原文(2026-10-13 内网实测,request_id 为真实值)。
*/
private static final String CAPACITY_ERROR =
"An error occurred in the shaoruidong/volcengine_ai/volcengine_ai, error type:"
+ " ToolProviderCredentialValidationError, error details: b'{\"code\":50500,"
+ "\"data\":{\"algorithm_base_resp\":{\"status_code\":110001,\"status_message\":"
+ "\"no more instances to retry: last error: <nil>\"}},\"message\":\"Internal Error\","
+ "\"request_id\":\"202610131604021580A5E0F3C983DD3D63\",\"status\":50500}'";
/**
* 凭证失效:重试无用,必须只跑一次。
*/
private static final String ACCESS_DENIED_ERROR =
"req_id: dc688258de PluginInvokeError: API request failed with status code 50400,"
+ " message: Access Denied";
@Test
void 上游容量不足_自动重跑一次并把第二次产出回给端上() throws Exception {
AtomicInteger hits = new AtomicInteger();
try (MockDify dify = MockDify.start(exchange -> {
int attempt = hits.incrementAndGet();
String stream = attempt == 1 ? capacityFailureStream() : succeededStream();
MockDify.reply(exchange, stream);
})) {
ApiResult<?> result = executeCreative(dify.baseUrl());
assertEquals(2, hits.get(), "第一次撞上上游容量不足后应自动重跑一次");
assertEquals(0, result.getCode(), "重试成功后对端仍是正常返回");
Map<?, ?> payload = (Map<?, ?>) result.getData();
assertEquals("succeeded", payload.get("status"), "第二次的状态应透出");
assertEquals("迎新晚会海报已生成", payload.get("answer"), "第二次的产出应回给端上");
}
}
@Test
void 凭证失效_不重试且只跑一次() throws Exception {
AtomicInteger hits = new AtomicInteger();
try (MockDify dify = MockDify.start(exchange -> {
hits.incrementAndGet();
MockDify.reply(exchange, failureStream(ACCESS_DENIED_ERROR));
})) {
ApiResult<?> result = executeCreative(dify.baseUrl());
assertEquals(1, hits.get(), "凭证问题重试无用,不应重跑");
Map<?, ?> payload = (Map<?, ?>) result.getData();
assertEquals("failed", payload.get("status"));
assertEquals("生成失败,请稍后重试", payload.get("answer"));
assertTrue(!payload.containsKey("raw"), "失败态仍不得下发整包");
}
}
// ------------------------------------------------------------------ 辅助
private static ApiResult<?> executeCreative(String baseUrl) throws Exception {
DifyProperties properties = new DifyProperties();
properties.setBaseUrl(baseUrl);
properties.getApps().setCreativeAssistant("app-integration-test");
AiController controller = new AiController();
Field propertiesField = AiController.class.getDeclaredField("difyProperties");
propertiesField.setAccessible(true);
propertiesField.set(controller, properties);
Map<String, Object> requestBody = new HashMap<>();
requestBody.put("user_content", "生成一张迎新晚会海报");
Method method = AiController.class.getDeclaredMethod(
"executeCreativeAssistant", Map.class, Integer.class);
method.setAccessible(true);
return (ApiResult<?>) method.invoke(controller, requestBody, 1);
}
/**
* 模拟真实出图分支:前面有个成功的节点,紧接着「即梦AI-图像生成」节点失败。
*/
private static String capacityFailureStream() {
return node("即梦AI-图像生成", "failed", CAPACITY_ERROR)
+ failureBody(CAPACITY_ERROR);
}
private static String failureStream(String error) {
return failureBody(error);
}
private static String failureBody(String error) {
JSONObject data = new JSONObject(true);
data.put("status", "failed");
data.put("outputs", new JSONObject(true));
data.put("error", error);
JSONObject finished = new JSONObject(true);
finished.put("event", "workflow_finished");
finished.put("workflow_run_id", "run-1");
finished.put("task_id", "task-1");
finished.put("data", data);
return sse(finished);
}
private static String succeededStream() {
JSONObject outputs = new JSONObject(true);
outputs.put("photo_message", "迎新晚会海报已生成");
JSONObject data = new JSONObject(true);
data.put("status", "succeeded");
data.put("outputs", outputs);
JSONObject finished = new JSONObject(true);
finished.put("event", "workflow_finished");
finished.put("workflow_run_id", "run-2");
finished.put("task_id", "task-2");
finished.put("data", data);
return node("即梦AI-图像生成", "succeeded", null) + sse(finished);
}
private static String node(String title, String status, String error) {
JSONObject data = new JSONObject(true);
data.put("title", title);
data.put("type", "tool");
data.put("status", status);
if (error != null) {
data.put("error", error);
}
JSONObject node = new JSONObject(true);
node.put("event", "node_finished");
node.put("data", data);
return sse(node);
}
private static String sse(JSONObject payload) {
return "data: " + payload.toJSONString() + "\n\n";
}
/**
* 允许抛 IOException 的单次请求处理器({@code Consumer} 不能抛受检异常)。
*/
@FunctionalInterface
private interface ExchangeHandler {
void handle(com.sun.net.httpserver.HttpExchange exchange) throws java.io.IOException;
}
/**
* 扮演 Dify 的极简 SSE 服务端。
*/
private static final class MockDify implements AutoCloseable {
private final HttpServer server;
private MockDify(HttpServer server) {
this.server = server;
}
private static MockDify start(ExchangeHandler handler) throws Exception {
HttpServer server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
server.createContext("/v1/workflows/run", exchange -> {
drain(exchange);
handler.handle(exchange);
});
server.start();
return new MockDify(server);
}
private String baseUrl() {
return "http://127.0.0.1:" + server.getAddress().getPort() + "/v1";
}
private static void reply(com.sun.net.httpserver.HttpExchange exchange, String body) throws java.io.IOException {
byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
exchange.getResponseHeaders().add("Content-Type", "text/event-stream");
exchange.sendResponseHeaders(200, bytes.length);
exchange.getResponseBody().write(bytes);
exchange.close();
}
private static void drain(com.sun.net.httpserver.HttpExchange exchange) throws java.io.IOException {
byte[] buffer = new byte[1024];
while (exchange.getRequestBody().read(buffer) != -1) {
// 丢掉请求体,确保客户端把请求写完
}
}
@Override
public void close() {
server.stop(0);
}
}
}
@@ -0,0 +1,124 @@
package com.gxwebsoft.ai.controller;
import com.alibaba.fastjson.JSONObject;
import com.gxwebsoft.common.core.utils.JSONUtil;
import org.junit.jupiter.api.Test;
import java.lang.reflect.Method;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* 上游瞬时故障的重试判定与失败文案(见 .scratch/ai-assistant/issues/11)。
*
* <p>背景:内网实例上「即梦AI-图像生成」偶发
* {@code {"code":50500,...,"status_code":110001,"status_message":"no more instances to retry"}},
* 2026-10-13 同一 prompt 连跑 6 次有 2 次失败,整节点 3 秒即返回——是上游算力不足,
* 与内容、参数、凭证都无关,重试一次通常就成。
*
* <p>而凭证失效、模型停用、审核拒绝重试无用,必须排除,
* 否则会把用户的一次失败拖成两次等待。
*/
class AiCreativeRetryTest {
/**
* 2026-10-13 内网实测的上游原文(保留关键结构,request_id 为真实值)。
*/
private static final String CAPACITY_ERROR =
"An error occurred in the shaoruidong/volcengine_ai/volcengine_ai, please contact the"
+ " author of shaoruidong/volcengine_ai/volcengine_ai for help, error type:"
+ " ToolProviderCredentialValidationError, error details: b'{\"code\":50500,"
+ "\"data\":{\"algorithm_base_resp\":{\"status_code\":110001,\"status_message\":"
+ "\"no more instances to retry: last error: <nil>\"},\"binary_data_base64\":[]},"
+ "\"message\":\"Internal Error\","
+ "\"request_id\":\"202610131604021580A5E0F3C983DD3D63\",\"status\":50500}'";
private static final String ACCESS_DENIED_ERROR =
"req_id: dc688258de PluginInvokeError: [models] Error: API request failed with status"
+ " code 50400, message: Access Denied";
private static final String MODEL_DISABLED_ERROR =
"req_id: dc688258de PluginInvokeError: Model disabled.";
private static JSONObject failedEvent(String error) {
JSONObject data = new JSONObject(true);
data.put("status", "failed");
data.put("outputs", new JSONObject(true));
data.put("error", error);
JSONObject event = new JSONObject(true);
event.put("event", "workflow_finished");
event.put("workflow_run_id", "b2e18332-68bd-49e6-90e2-bd9d16d428ff");
event.put("task_id", "0b1fdd88-cc6f-45b8-989d-be4f85bcff74");
event.put("data", data);
return event;
}
private static Map<String, Object> buildFailedResult(String error) throws Exception {
// 直接构造 JSONObject 作为报文体,别绕 JSONUtil.toJSONString——它按值序列化,
// 对字符串不加引号,拼出来的报文非法,parseObject 会静默返回 null。
JSONObject responseData = failedEvent(error);
Method method = AiController.class.getDeclaredMethod(
"buildCreativeAssistantResult", Map.class, String.class);
method.setAccessible(true);
//noinspection unchecked
return (Map<String, Object>) method.invoke(new AiController(), responseData, "");
}
private static boolean retryable(String reason) throws Exception {
Method method = AiController.class.getDeclaredMethod("isRetryableCreativeFailure", String.class);
method.setAccessible(true);
return (Boolean) method.invoke(null, reason);
}
@Test
void 上游容量不足_值得重试() throws Exception {
assertTrue(retryable(CAPACITY_ERROR), "no more instances to retry 属于瞬时故障,应重试");
}
@Test
void 上游容量不足_文案换成服务繁忙() throws Exception {
Map<String, Object> result = buildFailedResult(CAPACITY_ERROR);
assertEquals("failed", result.get("status"), "状态仍须可被判为失败");
assertEquals("生成服务繁忙,请稍后重试", result.get("answer"));
}
@Test
void 上游容量不足_失败态仍不得下发上游原文() throws Exception {
String json = JSONUtil.toJSONString(buildFailedResult(CAPACITY_ERROR));
assertFalse(json.contains("50500"), "不得下发上游错误码:" + json);
assertFalse(json.contains("no more instances"), "不得下发上游错误原文:" + json);
assertFalse(json.contains("request_id"), "不得下发 req_id:" + json);
assertFalse(json.contains("volcengine"), "不得下发上游插件名:" + json);
assertFalse(json.contains("raw"), "失败态不得带 raw:" + json);
}
@Test
void 凭证失效_不重试且文案不变() throws Exception {
assertFalse(retryable(ACCESS_DENIED_ERROR), "凭证问题重试无用");
assertEquals("生成失败,请稍后重试", buildFailedResult(ACCESS_DENIED_ERROR).get("answer"));
}
@Test
void 模型停用_不重试且文案不变() throws Exception {
assertFalse(retryable(MODEL_DISABLED_ERROR), "模型被停用重试无用");
assertEquals("生成失败,请稍后重试", buildFailedResult(MODEL_DISABLED_ERROR).get("answer"));
}
@Test
void 没有错误信息_不重试() throws Exception {
String blank = "";
assertFalse(retryable(null), "错误信息缺失时无从判断,保守不重试");
assertFalse(retryable(blank), "错误信息为空时不重试");
}
@Test
void 网关瞬时故障_值得重试() throws Exception {
assertTrue(retryable("502 Bad Gateway"));
assertTrue(retryable("connection reset by peer"));
assertTrue(retryable("Read timed out"));
}
}