mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(channel): evict SSE RunState by inactivity, not wall-clock age
This commit is contained in:
parent
7f45b95432
commit
f1d9104422
@ -1464,17 +1464,64 @@ public class ChatStreamTracker {
|
||||
|
||||
/** 已完成的 RunState 保留时间(5 分钟) */
|
||||
private static final long DONE_RETENTION_MS = 5 * 60 * 1000;
|
||||
/** RunState 最大存活时间(30 分钟,防止挂起的流永远占内存) */
|
||||
private static final long MAX_LIFETIME_MS = 30 * 60 * 1000;
|
||||
|
||||
/**
|
||||
* RunState 最长无活动时间。从 wall-clock {@code MAX_LIFETIME_MS=30min}
|
||||
* 切换到 inactivity-based 后默认 30 min — 与 hermes-agent 的
|
||||
* {@code gateway_timeout=1800s} 同口径:只要 agent 还在持续产事件
|
||||
* (tool call / content delta / phase transition / progress_update),
|
||||
* 就一直活下去,墙钟跑 1 小时 2 小时都可以。只有真正"完全静默 ≥ N 分钟"
|
||||
* 才视为卡死并强制清理。
|
||||
*
|
||||
* <p>修复的背景:round-6 的 10-LLM 横评任务实际跑了 47 min,全程都在
|
||||
* 出 tool call,但旧的 wall-clock 30 min 死线在 iter 128 / 8 of 10
|
||||
* 就把 RunState 清掉了 — SSE 流死、UI 空白、用户以为任务挂了。换成
|
||||
* inactivity 后,那种长任务永远不会被误清,而真正卡死的 agent(无活动
|
||||
* 5+ 分钟)会按时清理。可通过 property
|
||||
* {@code mateclaw.sse.idle-timeout-minutes} 调整。
|
||||
*/
|
||||
@org.springframework.beans.factory.annotation.Value("${mateclaw.sse.idle-timeout-minutes:30}")
|
||||
private int idleTimeoutMinutes = 30;
|
||||
|
||||
/**
|
||||
* Test hook — backdates the {@code lastEventAt} timestamp on an
|
||||
* existing RunState so {@link #cleanupStaleRuns()} can be exercised
|
||||
* deterministically without sleeping for minutes. Package-private on
|
||||
* purpose; production callers go through {@link #broadcast} which
|
||||
* stamps the field forward.
|
||||
*/
|
||||
void backdateLastEventForTesting(String conversationId, long lastEventAt) {
|
||||
RunState state = runs.get(conversationId);
|
||||
if (state != null) {
|
||||
state.lastEventAt = lastEventAt;
|
||||
}
|
||||
}
|
||||
|
||||
/** Test hook — true when a RunState row exists for the conversation. */
|
||||
boolean hasRunStateForTesting(String conversationId) {
|
||||
return runs.containsKey(conversationId);
|
||||
}
|
||||
|
||||
/** Test hook — exposes the configurable timeout for assertion. */
|
||||
int idleTimeoutMinutesForTesting() {
|
||||
return idleTimeoutMinutes;
|
||||
}
|
||||
|
||||
/** Test hook — override the timeout in pure-unit tests that bypass Spring. */
|
||||
void setIdleTimeoutMinutesForTesting(int minutes) {
|
||||
this.idleTimeoutMinutes = minutes;
|
||||
}
|
||||
|
||||
/**
|
||||
* 定期清理过期的 RunState,防止内存泄漏。
|
||||
* - 已完成超过 5 分钟的 → 移除
|
||||
* - 存活超过 30 分钟的(无论是否完成)→ 强制移除
|
||||
* - 已完成超过 {@link #DONE_RETENTION_MS} 的 → 移除
|
||||
* - 自 {@link RunState#lastEventAt} 算起静默超过
|
||||
* {@link #idleTimeoutMinutes} 分钟的 → 强制移除(视为卡死)
|
||||
*/
|
||||
@org.springframework.scheduling.annotation.Scheduled(fixedRate = 600_000)
|
||||
public void cleanupStaleRuns() {
|
||||
long now = System.currentTimeMillis();
|
||||
long idleThresholdMs = (long) idleTimeoutMinutes * 60_000L;
|
||||
int evicted = 0;
|
||||
|
||||
var iterator = runs.entrySet().iterator();
|
||||
@ -1482,6 +1529,7 @@ public class ChatStreamTracker {
|
||||
var entry = iterator.next();
|
||||
RunState state = entry.getValue();
|
||||
long age = now - state.createdAt;
|
||||
long idleMs = now - state.lastEventAt;
|
||||
|
||||
boolean shouldEvict = false;
|
||||
String reason = null;
|
||||
@ -1489,9 +1537,11 @@ public class ChatStreamTracker {
|
||||
if (state.done && age > DONE_RETENTION_MS) {
|
||||
shouldEvict = true;
|
||||
reason = "completed and expired";
|
||||
} else if (age > MAX_LIFETIME_MS) {
|
||||
} else if (idleMs > idleThresholdMs) {
|
||||
shouldEvict = true;
|
||||
reason = "exceeded max lifetime (" + (age / 1000) + "s)";
|
||||
reason = "idle for " + (idleMs / 1000) + "s (threshold "
|
||||
+ idleTimeoutMinutes + "min); total wall-clock age "
|
||||
+ (age / 1000) + "s";
|
||||
}
|
||||
|
||||
if (shouldEvict) {
|
||||
|
||||
@ -0,0 +1,86 @@
|
||||
package vip.mate.channel.web;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* Pins {@link ChatStreamTracker#cleanupStaleRuns()} — the switch from
|
||||
* wall-clock {@code MAX_LIFETIME_MS} to inactivity-based eviction.
|
||||
*
|
||||
* <p>Before the fix, a long-running agent that kept producing tool calls
|
||||
* (47-minute LLM-review smoke test, round 6) was killed at the 30-minute
|
||||
* wall-clock mark mid-task. The new behaviour mirrors hermes-agent's
|
||||
* {@code gateway_timeout}: only completely idle runs are evicted, the
|
||||
* actively-producing ones can run as long as they need to.
|
||||
*/
|
||||
class ChatStreamTrackerCleanupTest {
|
||||
|
||||
private static ChatStreamTracker newTracker() {
|
||||
ChatStreamTracker t = new ChatStreamTracker(new ObjectMapper());
|
||||
// Tighten the idle threshold so the test stays at unit-test speed.
|
||||
t.setIdleTimeoutMinutesForTesting(5);
|
||||
return t;
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("Active run (recent lastEventAt) survives cleanup regardless of total age.")
|
||||
void activeRunSurvives() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
tracker.register("conv-active");
|
||||
// lastEventAt was set to "now" inside the RunState constructor —
|
||||
// no backdate, so even a "very old createdAt" would be irrelevant.
|
||||
tracker.cleanupStaleRuns();
|
||||
assertTrue(tracker.hasRunStateForTesting("conv-active"),
|
||||
"actively-producing run must not be evicted");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("Idle run beyond threshold is evicted.")
|
||||
void idleRunEvicted() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
tracker.register("conv-idle");
|
||||
// 6 minutes ago — past the 5-minute threshold set above.
|
||||
tracker.backdateLastEventForTesting("conv-idle", System.currentTimeMillis() - 6 * 60_000L);
|
||||
tracker.cleanupStaleRuns();
|
||||
assertFalse(tracker.hasRunStateForTesting("conv-idle"),
|
||||
"idle run past threshold must be evicted");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("Idle run just inside the threshold survives — no premature eviction.")
|
||||
void idleRunWithinThresholdSurvives() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
tracker.register("conv-borderline");
|
||||
// 4 minutes idle — within 5-minute window.
|
||||
tracker.backdateLastEventForTesting("conv-borderline", System.currentTimeMillis() - 4 * 60_000L);
|
||||
tracker.cleanupStaleRuns();
|
||||
assertTrue(tracker.hasRunStateForTesting("conv-borderline"),
|
||||
"run idle below threshold must survive — would otherwise be a regression of the wall-clock bug");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("Mixed: active + idle runs — only the idle one is evicted.")
|
||||
void mixedRunsSelectiveEviction() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
tracker.register("conv-active");
|
||||
tracker.register("conv-idle");
|
||||
tracker.backdateLastEventForTesting("conv-idle", System.currentTimeMillis() - 10 * 60_000L);
|
||||
tracker.cleanupStaleRuns();
|
||||
assertTrue(tracker.hasRunStateForTesting("conv-active"));
|
||||
assertFalse(tracker.hasRunStateForTesting("conv-idle"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("Default idle timeout from @Value matches the documented 30-minute fallback.")
|
||||
void defaultIdleTimeoutIs30() {
|
||||
// Bypass the @Value injection (no Spring context in this unit test) —
|
||||
// the field initialiser pins the default so a refactor that drops the
|
||||
// = 30 falls over here.
|
||||
ChatStreamTracker tracker = new ChatStreamTracker(new ObjectMapper());
|
||||
org.junit.jupiter.api.Assertions.assertEquals(30, tracker.idleTimeoutMinutesForTesting());
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user