EventLogConcurrencyTest.java
3.94 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
package com.xly.service;
import com.xly.agent.EventLogChatMemory;
import com.fasterxml.jackson.databind.ObjectMapper;
import dev.langchain4j.data.message.AiMessage;
import dev.langchain4j.data.message.ToolExecutionResultMessage;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* 「流式输出中途点【确认】」的并发不变量：对话流（EventLogChatMemory 落模型消息）与确认端点
* （recordOutcome 落 confirm 事件）同时写同一会话时，**所有事件都在、各写者内部有序**。
* 旧 chat:mem 整包读改写在此场景必丢更新；事件日志逐条追加从机制上消除该竞态。
*/
class EventLogConcurrencyTest {
private static final String CONV = "u1:c-conc";
@Test
void concurrentStreamAndConfirmLoseNothing() throws Exception {
InMemoryLedger log = new InMemoryLedger();
EventLogChatMemory mem = new EventLogChatMemory(
CONV, log, new EventProjectionService(new ObjectMapper()), 6000, false);
int writers = 4;
int perWriter = 100;
ExecutorService pool = Executors.newFixedThreadPool(writers);
CountDownLatch start = new CountDownLatch(1);
CountDownLatch done = new CountDownLatch(writers);
for (int w = 0; w < writers; w++) {
final int id = w;
pool.submit(() -> {
try {
start.await();
for (int i = 0; i < perWriter; i++) {
if (id % 2 == 0) { // 对话流：模型消息经记忆适配器落账
if (i % 2 == 0) {
mem.add(AiMessage.from("w" + id + "-answer-" + i));
} else {
mem.add(ToolExecutionResultMessage.from(
"t" + id + "-" + i, "readFormData", "w" + id + "-result-" + i));
}
} else { // 确认端点：按钮事件直接落账
log.append(CONV, "confirm", Map.of(
"opId", "w" + id + "-" + i, "status", "executed",
"msg", "", "description", "op w" + id + "-" + i));
}
}
} catch (InterruptedException ignored) {
} finally {
done.countDown();
}
});
}
start.countDown();
assertTrue(done.await(30, TimeUnit.SECONDS));
pool.shutdown();
List<Map<String, Object>> events = log.events(CONV);
assertEquals(writers * perWriter, events.size(), "并发写者的事件一条不丢");
for (int w = 0; w < writers; w++) {
List<Integer> seq = new ArrayList<>();
for (Map<String, Object> ev : events) {
String marker = String.valueOf(
ev.containsKey("opId") ? ev.get("opId")
: ev.containsKey("tcId") ? ev.get("tcId") : ev.get("text"));
String prefix = marker.startsWith("t" + w + "-") ? "t" + w + "-" : "w" + w + "-";
if (marker.startsWith(prefix)) {
String tail = marker.substring(marker.lastIndexOf('-') + 1);
try {
seq.add(Integer.parseInt(tail));
} catch (NumberFormatException ignore) {
}
}
}
for (int i = 1; i < seq.size(); i++) {
assertTrue(seq.get(i) > seq.get(i - 1), "写者 " + w + " 自身顺序保持");
}
}
}
}